// POST /relay/summarize-url — unified transcribe + analyze pipeline.
//
// Why this exists: the older /relay/transcribe-url + /relay/analyze
// pair required the Recap client to do its OWN analyze windowing
// (per-window POST to /relay/analyze, then client-side stitch).
// That round-trips the full transcript back to the client only for
// the client to slice it into 12 windows and ship those slices back
// to the relay, one per HTTP request. Bandwidth + latency tax on long
// content, and the chunking knobs in the operator's Settings tab
// only affected benchmarks — production traffic ignored them because
// Recap had its own hardcoded constants.
//
// This route does both steps server-side in one async job:
//   1. Download audio (yt-dlp / direct HTTP) — same code as transcribe-url
//   2. Transcribe with chunking (Gemini or Hardware) — same backends
//   3. Run chunked-analyze on the transcript using the operator's
//      Settings-tab knobs (window body / overlap / concurrency)
//   4. Per-window section results stream to the client via SSE
//   5. Final transcript + stitched analysis returned in the
//      complete job result
//
// Credit policy: charges ONE credit on full success (same as
// transcribe-url alone — Recap's billing model treats one summarize
// as one credit, regardless of which steps it touches). If transcribe
// fails before analyze starts, no credit is charged. If transcribe
// succeeds but ALL analyze windows fail, the credit IS charged (the
// expensive part — transcribe — completed, and partial-window
// failures are tolerated upstream of this).
//
// Endpoints exposed:
//   POST /relay/summarize-url                 → kick off, returns job_id
//   GET  /relay/summarize-url/:jobId/events   → SSE stream of progress + window events
//   GET  /relay/jobs/:id (existing)           → poll-based fallback

import express from "express";
import fs from "fs/promises";
import os from "os";
import path from "path";
import { getConfigSnapshot } from "../config.js";
import { createGeminiBackend } from "../backends/gemini.js";
import { createHardwareBackend } from "../backends/hardware.js";
import { resolveHardwareConfig } from "../hardware-config.js";
import { recordCall } from "../audit-log.js";
import { calcGeminiCost } from "../pricing.js";
import { getAudioDurationSeconds } from "../audio-meta.js";
import {
  createJob,
  getJob,
  markRunning,
  setProgress,
  markComplete,
  markFailed,
  appendEvent,
  subscribeToJob,
} from "../jobs.js";
import { resolveIdentity, identityTier } from "../identity.js";
import {
  getOrCreateRow,
  planBackend,
  commitCredit,
  licenseFingerprint,
} from "../credits.js";
import { lookupJob, markJobCharged, refundJob } from "../job-credits.js";
import { getTierQuotas } from "../config.js";
import { envelope, errorEnvelope } from "./envelope.js";
import {
  runChunkedAnalysis,
  runPipelinedAnalysis,
  parseBracketedTranscript,
  firstEntryAtOrAfter,
  lastEntryBefore,
  canonicalIndexForOffset,
  stitchAnalysisResults,
} from "../chunked-analyze.js";
import { createChunkBuffer } from "../chunk-buffer.js";
import {
  runNameInference,
  runSummaryPolish,
} from "../post-cluster-polish.js";
import { acquireHardwareSlot } from "../hardware-queue.js";
import { saveJobOutput } from "../output-store.js";
import {
  looksLikeYouTube,
  downloadDirect,
  downloadYouTube,
} from "./transcribe-url.js";
import { reportHealthEvent } from "../spark-control-events.js";

export function summarizeUrlRouter() {
  const router = express.Router();

  // POST /relay/summarize-url — kicks off the combined pipeline.
  // Same request shape as /relay/transcribe-url (media_url + optional
  // hints). Same auth (X-Recap-Install-Id header + license proof).
  router.post("/summarize-url", express.json({ limit: "1mb" }), async (req, res) => {
    const summaryJobId = req.header("X-Recap-Job-Id") || null;

    let identity;
    try {
      identity = await resolveIdentity(req);
    } catch (err) {
      const e = await errorEnvelope({
        error: err?.message || "auth_error",
        statusHint: err?.status || 401,
      });
      return res.status(e.statusHint || 401).json(e.body);
    }
    if (identity.kind === "license" && !identity.installId) {
      const e = await errorEnvelope({
        error: "missing X-Recap-Install-Id header",
        statusHint: 400,
      });
      return res.status(400).json(e.body);
    }
    const { creditKey, installId, license } = identity;
    // `title` is `let` rather than `const` because the worker may
    // backfill it from yt-dlp metadata after the download completes
    // (when the client didn't pre-fetch the title — typical for the
    // Recap-app relay-mode branch which submits the URL alone).
    let title;
    // channel/description/chapters are declared with `let` (not const)
    // because the yt-dlp download step below may backfill any of them
    // when the client sent empty values — see the "Fall back to yt-dlp-
    // extracted metadata" block after downloadYouTube returns.
    let channel, description, chapters;
    const {
      media_url: mediaUrl,
      type,
      mime_type: bodyMime,
      title: bodyTitle,
      channel: bodyChannel,
      description: bodyDescription,
      chapters: bodyChapters,
    } = req.body || {};
    channel = bodyChannel;
    description = bodyDescription;
    chapters = bodyChapters;
    // Treat empty strings and the literal sentinel "Untitled" as
    // "no title supplied" so the yt-dlp metadata fallback below
    // (`if (!title && audio.title) title = audio.title`) fires.
    // Older Recap clients (< 0.2.71) pass "Untitled" verbatim when
    // the operator hasn't pre-fetched metadata — without this
    // normalization the relay would echo "Untitled" back as if it
    // were a real title, and the resulting library entry would
    // stay "Untitled" forever despite yt-dlp having the real one.
    const trimmedBodyTitle = typeof bodyTitle === "string" ? bodyTitle.trim() : bodyTitle;
    title = (trimmedBodyTitle === "" || trimmedBodyTitle === "Untitled")
      ? null
      : trimmedBodyTitle;
    if (!mediaUrl || typeof mediaUrl !== "string") {
      const e = await errorEnvelope({
        error: "missing or non-string body.media_url",
        creditKey,
        installId,
        statusHint: 400,
      });
      return res.status(400).json(e.body);
    }

    const row = await getOrCreateRow({ creditKey, installId, license });
    const tier = identityTier(identity, row);
    row.tier_snapshot = tier;
    const licenseFp = identity.kind === "cloud" ? null : licenseFingerprint(license);
    const auditInstall = installId || identity.userId || null;

    const reusedSummaryJob = !!lookupJob({ creditKey, installId, license, jobId: summaryJobId });
    const cfgPlan = await getConfigSnapshot();
    const hw = await resolveHardwareConfig(cfgPlan);
    // Operator-only diagnostic. We do NOT 503 the request here on
    // blocked_reason — that pre-empts the backend-preference choice
    // and would return a hardware-down error even when Gemini was
    // selected as the routing preference (it'd also leak the
    // operator-internal Spark Control / parakeet wording to the
    // client). Instead: log the detail for the operator's eyes,
    // let `hasHardware = !!hw.transcribe.url` go to false, and let
    // planBackend route to Gemini under the normal preference
    // rules. If the operator chose hardware_only mode, planBackend
    // refuses with the generic `hardware_only_not_configured`
    // reason — which is client-safe.
    if (hw.transcribe.blocked_reason) {
      console.warn(
        `[summarize-url] hardware transcribe currently blocked (planBackend will route to Gemini if available): ${hw.transcribe.blocked_reason}`,
      );
    }
    if (hw.analyze.blocked_reason) {
      console.warn(
        `[summarize-url] hardware analyze currently blocked (planBackend will route to Gemini if available): ${hw.analyze.blocked_reason}`,
      );
    }
    const hasHardware = !!hw.transcribe.url;
    const quota = await getTierQuotas();
    const preference =
      cfgPlan.relay_transcribe_backend_preference || "gemini_first";
    const plan = planBackend(row, quota, { hasHardware, preference });
    if (!plan.allowed) {
      await recordCall({
        install_id: auditInstall,
        license_fingerprint: licenseFp,
        tier,
        pipeline: "transcribe",
        backend: null,
        model: null,
        status: "refused",
        credit_charged: 0,
        duration_ms: 0,
        cost_usd: 0,
        job_id: summaryJobId,
        media_url: mediaUrl || null,
        title: title || null,
        error: plan.reason,
      });
      const e = await errorEnvelope({
        error: plan.reason,
        installId,
        license,
        tier,
        statusHint: 402,
      });
      return res.status(402).json(e.body);
    }
    const chosenBackend = plan.backend;

    // The analyze backend follows the operator's analyze-side
    // preference, NOT the transcribe-side preference. They're set
    // independently in StartOS so the operator can route, e.g.,
    // transcribe-to-Gemini + analyze-to-hardware on tight Gemini
    // budgets. planBackend evaluates the same quota/preference rules
    // for analyze and returns a separate decision.
    const analyzePreference =
      cfgPlan.relay_analyze_backend_preference || "gemini_first";
    const hasAnalyzeHardware = !!hw.analyze.url;
    const analyzePlan = planBackend(row, quota, {
      hasHardware: hasAnalyzeHardware,
      preference: analyzePreference,
    });
    if (!analyzePlan.allowed) {
      const e = await errorEnvelope({
        error: analyzePlan.reason,
        installId,
        license,
        tier,
        statusHint: 402,
      });
      return res.status(402).json(e.body);
    }
    const chosenAnalyzeBackend = analyzePlan.backend;

    const job = createJob({
      kind: "summarize-url",
      installId: auditInstall,
      metadata: {
        owner: creditKey, // authorizes the SSE /events stream (per-identity)
        media_url: mediaUrl,
        transcribe_backend: chosenBackend,
        analyze_backend: chosenAnalyzeBackend,
        summary_job_id: summaryJobId,
      },
    });

    // Background worker — same try/catch discipline as transcribe-url.
    // All progress + per-window events go through appendEvent(job.id, ...)
    // so any SSE subscriber gets them in real time.
    (async () => {
      const workerT0 = Date.now();
      markRunning(job.id);
      appendEvent(job.id, "progress", { message: "downloading media…" });
      setProgress(job.id, "downloading media…");
      console.log(
        `[summarize-url ${job.id.slice(0, 8)}] downloading media from ${(mediaUrl || "(no url)").slice(0, 120)}`
      );

      const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "relay-sum-"));
      const isYT = type === "youtube" || (!type && looksLikeYouTube(mediaUrl));
      let audio;
      let downloadMs = 0;
      try {
        const dlStart = Date.now();
        audio = isYT
          ? await downloadYouTube(mediaUrl, tmpDir)
          : await downloadDirect(mediaUrl, tmpDir);
        downloadMs = Date.now() - dlStart;
        audio.seconds = await getAudioDurationSeconds(audio.filePath);
        console.log(
          `[summarize-url ${job.id.slice(0, 8)}] download done in ${(downloadMs / 1000).toFixed(1)}s — ${Math.round((audio.seconds || 0) / 60)} min audio`
        );
        // Fall back to the yt-dlp-extracted title when the client
        // didn't pass one. Recap-app's relay-mode branch doesn't
        // pre-fetch YouTube metadata, so without this the Jobs table
        // shows "Untitled" for every Recap submission.
        if (!title && audio.title) {
          title = audio.title;
        }
        // Same fallback pattern for channel / description / chapters.
        // These power speaker-identification in the transcribe prompt
        // (the model uses them to assign names like "Brandon Karpeles:"
        // and "Matt Hill:" to dialogue turns). Recap-app's direct-to-
        // Gemini path always fetches these via fetchYouTubeMetadata
        // before transcribing; the relay path either gets them from
        // the Recap client (if Recap fetched them and passed them
        // through) OR — when the client sent empty fields — falls back
        // here to whatever yt-dlp extracted during download. Defense-in-
        // depth: works for ANY client that hits /relay/summarize-url,
        // not just Recap, and survives a client upgrade gap where an
        // older Recap is still sending empty channelHint/descriptionHint.
        if (!channel && audio.channel) {
          channel = audio.channel;
        }
        if (!description && audio.description) {
          description = audio.description;
        }
        if ((!Array.isArray(chapters) || chapters.length === 0) && Array.isArray(audio.chapters) && audio.chapters.length > 0) {
          chapters = audio.chapters;
        }
        appendEvent(job.id, "progress", {
          message: `transcribing ${Math.round((audio.seconds || 0) / 60)} min audio…`,
        });
        setProgress(job.id, `transcribing ${Math.round((audio.seconds || 0) / 60)} min audio…`);
      } catch (err) {
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
        const msg = (err?.message || String(err)).slice(0, 300);
        await recordCall({
          install_id: auditInstall,
          license_fingerprint: licenseFp,
          tier,
          pipeline: "transcribe",
          backend: chosenBackend,
          model: null,
          status: "error",
          credit_charged: 0,
          duration_ms: Date.now() - workerT0,
          download_ms: Date.now() - workerT0,
          audio_seconds: null,
          cost_usd: 0,
          job_id: summaryJobId,
          media_url: mediaUrl || null,
          title: title || null,
          error: "download_failed: " + msg,
        });
        markFailed(job.id, "download_failed: " + msg);
        return;
      }

      // ── Hardware FIFO queue acquire ─────────────────────────────
      // Hardware jobs serialize through a single-slot semaphore so
      // the operator's GPU box doesn't get N jobs hammering it at
      // once. Gemini-backend jobs bypass — Google's API handles
      // concurrency at scale. The acquire awaits if there's an
      // active hardware job ahead; we emit a "queued" SSE event so
      // Recap can show "Queued — N jobs ahead" while we wait. See
      // server/hardware-queue.js.
      const usesHardwareForAnything =
        chosenBackend === "hardware" || chosenAnalyzeBackend === "hardware";
      let releaseHardwareSlot = null;
      if (usesHardwareForAnything) {
        releaseHardwareSlot = await acquireHardwareSlot({
          jobId: job.id,
          onWait: ({ position, activeJobId }) => {
            // Tell the SSE client we're waiting so the UI can show
            // a queue position instead of a stalled "Transcribing…"
            // state. Best-effort — appendEvent is fire-and-forget.
            try {
              appendEvent(job.id, "queued", {
                position,
                activeJobId: activeJobId || null,
              });
              setProgress(
                job.id,
                `queued — ${position} job(s) ahead on operator hardware`
              );
            } catch {}
            console.log(
              `[summarize-url ${job.id.slice(0, 8)}] queued for hardware — position ${position}`
            );
          },
        });
      }

      // Wrap from here to end-of-IIFE in try/finally so the hardware
      // slot ALWAYS releases — on success, on early-return failure,
      // and on uncaught exceptions. Finally executes before the
      // throw propagates so the next queued job unblocks promptly.
      try {

      // ── Transcribe ──────────────────────────────────────────────
      // Pipelined-analyze design (v0.2.89): when the hardware backend
      // is transcribing, fire each analyze window's call AS SOON AS
      // its required transcribe chunks have completed — in parallel
      // with later chunks still being transcribed. The total wall
      // time savings are modest (~10s for a 94-min video), but the
      // user-perceived "first topic visible" time drops from
      // ~T=160s to ~T=80s because window 1's sections render while
      // chunks 5-21 are still in-flight. See server/chunk-buffer.js
      // and runPipelinedAnalysis in chunked-analyze.js for the
      // mechanics. Gemini transcribe doesn't expose per-chunk
      // callbacks (single big call) so its path stays sequential.
      const cfg = await getConfigSnapshot();
      let txResult;
      const txPhaseStart = Date.now();

      // Build the analyze backend UP FRONT — we need it before
      // transcribe starts in pipelined mode. Sequential mode would
      // also work with it built here, just doesn't strictly need to.
      const anStart = Date.now();
      let analyzeBackend = null;
      try {
        if (chosenAnalyzeBackend === "gemini") {
          analyzeBackend = createGeminiBackend({
            apiKey: cfg.relay_gemini_api_key,
            transcriptionModel: cfg.relay_gemini_transcription_model,
            analysisModel: cfg.relay_gemini_analysis_model,
            txChunkSeconds: (cfg.relay_gemini_tx_chunk_minutes || 30) * 60,
            txConcurrency: cfg.relay_gemini_tx_concurrency || 12,
            txMaxOutputTokens: cfg.relay_gemini_tx_max_output_tokens || 65536,
            anMaxOutputTokens: cfg.relay_gemini_an_max_output_tokens || 8192,
          });
        } else {
          if (!hw.analyze.url) {
            throw new Error("hardware analyze URL not configured");
          }
          analyzeBackend = createHardwareBackend({
            parakeetBaseURL: hw.transcribe.url || "",
            gemmaBaseURL: hw.analyze.url,
            sparkControlBaseURL: hw.sparkBase || "",
            parakeetModel: hw.transcribe.model || "",
            gemmaModel: hw.analyze.model || "",
            txChunkSeconds: (cfg.relay_hardware_tx_chunk_minutes || 5) * 60,
            txChunkOverlapSeconds:
              cfg.relay_hardware_tx_chunk_overlap_seconds ?? 30,
            diarizationEnabled: !!cfg.relay_hardware_diarization_enabled,
            clusterThresholdPct:
              cfg.relay_hardware_voice_clustering_threshold ?? 70,
            anchorMinSpeakingSec:
              cfg.relay_hardware_anchor_min_speaking_sec ?? 30,
            smallClusterMaxSpeakingSec:
              cfg.relay_hardware_small_cluster_max_speaking_sec ?? 15,
            uncertainMarginPct:
              cfg.relay_hardware_uncertain_margin_pct ?? 10,
            txConcurrency: cfg.relay_hardware_tx_concurrency || 4,
            anMaxTokens: cfg.relay_hardware_an_max_tokens || 16000,
          });
        }
      } catch (err) {
        const msg = (err?.message || String(err)).slice(0, 400);
        await recordCall({
          install_id: auditInstall,
          license_fingerprint: licenseFp,
          tier,
          pipeline: "analyze",
          backend: chosenAnalyzeBackend,
          model: null,
          status: "error",
          duration_ms: 0,
          audio_seconds: 0,
          cost_usd: 0,
          job_id: summaryJobId,
          media_url: mediaUrl || null,
          title: title || null,
          error: "analyze_backend_init_failed: " + msg,
          window_idx: 0,
          window_count: 1,
        });
        markFailed(job.id, "analyze_init_failed: " + msg);
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
        return;
      }

      // Analyze tunables (used by both pipelined + sequential paths).
      const bodyMin = chosenAnalyzeBackend === "gemini"
        ? (cfg.relay_gemini_analyze_window_minutes || 18)
        : (cfg.relay_hardware_analyze_window_minutes || 18);
      const overlapMin = chosenAnalyzeBackend === "gemini"
        ? (cfg.relay_gemini_analyze_overlap_minutes || 2)
        : (cfg.relay_hardware_analyze_overlap_minutes || 2);
      const anConcurrency = chosenAnalyzeBackend === "gemini"
        ? (cfg.relay_gemini_analyze_concurrency || 12)
        : (cfg.relay_hardware_analyze_concurrency || 8);
      const cutoffMin = cfg.relay_analyze_cutoff_minutes || 25;
      const targetTotalsByBucket = {
        under_30: cfg.relay_analyze_total_sections_under_30,
        "30_60": cfg.relay_analyze_total_sections_30_60,
        "60_90": cfg.relay_analyze_total_sections_60_90,
        "90_120": cfg.relay_analyze_total_sections_90_120,
        "120_150": cfg.relay_analyze_total_sections_120_150,
        "150_180": cfg.relay_analyze_total_sections_150_180,
        over_180: cfg.relay_analyze_total_sections_over_180,
      };
      const analyzePromptOverride =
        cfg.relay_analyze_prompt || cfg.relay_analyze_prompt_default || "";
      const computeCostDetails = (model, usage) =>
        chosenAnalyzeBackend === "gemini" && usage
          ? calcGeminiCost(model, usage)
          : { input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0 };

      // Pipelined mode requires hardware transcribe (per-chunk
      // onChunkComplete callback) + a known audio duration. When both
      // hold, we start the analyze workers BEFORE the transcribe
      // promise so they're already waiting on chunkBuffer when the
      // first chunks land.
      const usePipelining =
        chosenBackend === "hardware" && (audio.seconds || 0) > 0;
      const chunkBuffer = usePipelining ? createChunkBuffer() : null;
      const onWindowCompleteHandler = (cb) => {
        appendEvent(job.id, "window_complete", {
          windowIdx: cb.windowIdx,
          totalWindows: cb.totalWindows,
          ownedSections: cb.ownedSections,
          // Pipelined mode: include this window's local entries so
          // the SDK / recap-app handler can render the partial
          // sections WITHOUT waiting for transcribe_complete to
          // populate streamedRelayEntries. The SDK falls back to
          // streamedRelayEntries when windowEntries is absent
          // (sequential mode).
          windowEntries: cb.windowEntries || undefined,
          windowBodySeconds: cb.windowBodySeconds,
          model: cb.model,
          durationMs: cb.durationMs,
        });
        setProgress(job.id, `analyze window ${cb.windowIdx + 1}/${cb.totalWindows} done`);
        const sectionCount = Array.isArray(cb.ownedSections)
          ? cb.ownedSections.length
          : 0;
        console.log(
          `[summarize-url ${job.id.slice(0, 8)}] analyze window ${cb.windowIdx + 1}/${cb.totalWindows} done in ${((cb.durationMs || 0) / 1000).toFixed(1)}s (${sectionCount} sections, model=${cb.model || "?"})`
        );
      };

      // Kick off the pipelined-analyze promise BEFORE we await
      // transcribe. Workers block on chunkBuffer.waitForTime() until
      // their windows' required chunks land. If transcribe throws,
      // we call chunkBuffer.fail() in the catch to release the
      // pending waiters.
      let pipelinedAnalyzePromise = null;
      if (usePipelining) {
        pipelinedAnalyzePromise = runPipelinedAnalysis({
          audioDurationSec: audio.seconds || 0,
          waitForTime: (sec) => chunkBuffer.waitForTime(sec),
          getReadySegments: (s, e) => chunkBuffer.getSegments(s, e),
          bodySeconds: bodyMin * 60,
          overlapSeconds: overlapMin * 60,
          cutoffSeconds: cutoffMin * 60,
          concurrency: anConcurrency,
          backend: analyzeBackend,
          pipelineBackend: chosenAnalyzeBackend,
          jobId: summaryJobId,
          batchId: null,
          mediaUrl,
          title: title || null,
          installId,
          licenseFingerprint: licenseFp,
          source: "summarize-url",
          computeCostDetails,
          analyzePromptOverride,
          targetTotalsByBucket,
          onWindowComplete: onWindowCompleteHandler,
        }).catch((err) => ({ __error: err }));
      }

      try {
        const audioBuf = await fs.readFile(audio.filePath);
        const mimeType = bodyMime || audio.mimeType;
        if (chosenBackend === "gemini") {
          const backend = createGeminiBackend({
            apiKey: cfg.relay_gemini_api_key,
            transcriptionModel: cfg.relay_gemini_transcription_model,
            analysisModel: cfg.relay_gemini_analysis_model,
            txChunkSeconds: (cfg.relay_gemini_tx_chunk_minutes || 30) * 60,
            txConcurrency: cfg.relay_gemini_tx_concurrency || 12,
            // Three-layer prompt resolution: per-session override
            // wins; operator-promoted default is next; gemini.js's
            // hardcoded DEFAULT_TRANSCRIBE_PROMPT_BODY is the
            // factory fallback. Empty string here means "fall back
            // to the code-side default" inside the backend.
            transcribePromptOverride:
              cfg.relay_transcribe_prompt ||
              cfg.relay_transcribe_prompt_default ||
              "",
            txMaxOutputTokens: cfg.relay_gemini_tx_max_output_tokens || 65536,
            anMaxOutputTokens: cfg.relay_gemini_an_max_output_tokens || 8192,
          });
          txResult = await backend.transcribeAudio({
            audio: audioBuf,
            mimeType,
            title: title || "",
            channel: channel || "",
            description: description || "",
            chapters: Array.isArray(chapters) ? chapters : [],
            offsetSeconds: 0,
          });
        } else {
          const backend = createHardwareBackend({
            parakeetBaseURL: hw.transcribe.url || "",
            gemmaBaseURL: hw.analyze.url || "",
            sparkControlBaseURL: hw.sparkBase || "",
            parakeetModel: hw.transcribe.model || "",
            gemmaModel: hw.analyze.model || "",
            txChunkSeconds: (cfg.relay_hardware_tx_chunk_minutes || 5) * 60,
            txChunkOverlapSeconds:
              cfg.relay_hardware_tx_chunk_overlap_seconds ?? 30,
            diarizationEnabled: !!cfg.relay_hardware_diarization_enabled,
            clusterThresholdPct:
              cfg.relay_hardware_voice_clustering_threshold ?? 70,
            anchorMinSpeakingSec:
              cfg.relay_hardware_anchor_min_speaking_sec ?? 30,
            smallClusterMaxSpeakingSec:
              cfg.relay_hardware_small_cluster_max_speaking_sec ?? 15,
            uncertainMarginPct:
              cfg.relay_hardware_uncertain_margin_pct ?? 10,
            txConcurrency: cfg.relay_hardware_tx_concurrency || 4,
            anMaxTokens: cfg.relay_hardware_an_max_tokens || 16000,
            // Pipelined-mode hook: feed each completed chunk into
            // the shared buffer so the analyze workers can unblock.
            onChunkComplete: chunkBuffer
              ? (cd) => chunkBuffer.add(cd)
              : null,
          });
          txResult = await backend.transcribeAudio({
            audio: audioBuf,
            mimeType,
            offsetSeconds: 0,
          });
        }
      } catch (err) {
        if (chunkBuffer) chunkBuffer.fail(err);
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
        if (reusedSummaryJob) await refundJob({ creditKey, installId, license, jobId: summaryJobId });
        const msg = (err?.message || String(err)).slice(0, 400);
        if (chosenBackend === "hardware") {
          reportHealthEvent({
            service: "parakeet",
            ok: false,
            error: msg.slice(0, 280),
            ms: Date.now() - workerT0,
          });
        }
        await recordCall({
          install_id: auditInstall,
          license_fingerprint: licenseFp,
          tier,
          pipeline: "transcribe",
          backend: chosenBackend,
          model: chosenBackend === "gemini"
            ? cfg.relay_gemini_transcription_model
            : hw.transcribe.model || "(auto)",
          status: "error",
          credit_charged: 0,
          duration_ms: Date.now() - txPhaseStart,
          download_ms: downloadMs,
          audio_seconds: audio?.seconds || null,
          audio_bytes: audio?.bytes || null,
          cost_usd: 0,
          job_id: summaryJobId,
          media_url: mediaUrl || null,
          title: title || null,
          error: msg,
        });
        markFailed(job.id, msg);
        return;
      } finally {
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
      }

      // Audit the transcribe row (success path).
      const txCostDetails =
        chosenBackend === "gemini" && txResult.usage
          ? calcGeminiCost(txResult.model, txResult.usage)
          : { input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0 };
      // Detect output-token truncation. The backend returns a
      // truncated_chunks array when any chunk's last emitted
      // timestamp falls short of the chunk's expected duration —
      // a silent failure mode where the model hit its output cap
      // mid-transcript. We mark the audit row status="partial"
      // and stamp an error string so the Jobs table doesn't
      // mislead the operator into thinking the run succeeded.
      const truncatedChunks = Array.isArray(txResult?.truncated_chunks)
        ? txResult.truncated_chunks
        : [];
      const wasTruncated = truncatedChunks.length > 0;
      const truncationError = wasTruncated
        ? `transcribe: ${truncatedChunks.length} chunk(s) truncated — missing ~${truncatedChunks.reduce((s, c) => s + (c.missingSec || 0), 0)}s of speech (model: ${txResult.model || "unknown"}). Likely hit maxOutputTokens.`
        : null;
      await recordCall({
        install_id: auditInstall,
        license_fingerprint: licenseFp,
        tier,
        pipeline: "transcribe",
        backend: chosenBackend,
        model: txResult?.model || null,
        status: wasTruncated ? "partial" : "success",
        credit_charged: 0, // credit committed AFTER analyze succeeds too
        // duration_ms = transcribe phase ONLY (txPhaseStart was
        // stamped after the download completed). Previously this used
        // Date.now() - workerT0 which included the download time too,
        // so the Jobs table's "TX wall time" column appeared inflated
        // by ~50s on long YouTube fetches. download_ms below carries
        // the download phase as a separate field so the operator can
        // see both numbers on the same row.
        duration_ms: Date.now() - txPhaseStart,
        download_ms: downloadMs,
        audio_bytes: audio.bytes,
        audio_seconds: audio.seconds || null,
        job_id: summaryJobId,
        attempts: txResult?.attempts || null,
        media_url: mediaUrl || null,
        title: title || null,
        chunk_count: txResult?.chunk_count ?? null,
        chunk_durations_ms: txResult?.chunk_durations_ms || null,
        truncated_chunks: wasTruncated ? truncatedChunks : null,
        error: truncationError,
        ...txCostDetails,
      });

      // Emit a transcribe_complete event so the SSE client can show
      // the full transcript while analyze is still in flight (useful
      // for a "transcript preview" UI affordance).
      appendEvent(job.id, "transcribe_complete", {
        transcript: txResult.text || "",
        model: txResult.model || null,
        chunk_count: txResult.chunk_count ?? null,
        audio_seconds: audio.seconds || null,
        audio_bytes: audio.bytes || null,
        duration_ms: Date.now() - workerT0,
      });
      setProgress(job.id, "analyzing topics…");
      console.log(
        `[summarize-url ${job.id.slice(0, 8)}] transcribe done in ${((Date.now() - txPhaseStart) / 1000).toFixed(1)}s — starting analyze`
      );

      // ── Analyze ─────────────────────────────────────────────────
      // Pipelined mode: await the promise we kicked off BEFORE
      // transcribe; remap window-local section indices to global
      // canonical indices; stitch.
      // Sequential mode: call runChunkedAnalysis on the full
      // transcript as today.
      let analyzeResult;
      try {
        if (pipelinedAnalyzePromise) {
          const pipelinedRaw = await pipelinedAnalyzePromise;
          if (pipelinedRaw && pipelinedRaw.__error) {
            throw pipelinedRaw.__error;
          }
          // Build canonical entries from the final stitched transcript.
          // The pipelined windows were each analyzed against their own
          // segment-only view; we now map their LOCAL section indices
          // to GLOBAL canonical-entry indices (which is what the
          // stitcher + downstream stitching code expects).
          const canonicalEntries = parseBracketedTranscript(txResult.text || "");
          const syntheticResults = (pipelinedRaw.windowResults || []).map((wr) => {
            if (!wr || !wr.ok) return wr;
            const w = wr.window;
            const globalStartIdx = firstEntryAtOrAfter(canonicalEntries, w.startSec);
            const globalBodyStartIdx = firstEntryAtOrAfter(canonicalEntries, w.bodyStartSec);
            const globalEndIdx = lastEntryBefore(canonicalEntries, w.windowEndSec + 0.5);
            // Edge case: this window's time range produced no canonical
            // entries (audio gap, severe truncation). Mark as failed
            // so the stitcher skips it.
            if (
              globalStartIdx >= canonicalEntries.length ||
              globalEndIdx < globalStartIdx
            ) {
              return {
                window: { startIdx: 0, endIdx: -1, bodyStartIdx: 0 },
                ok: false,
                error: new Error("pipelined window had no canonical entries — audio gap"),
              };
            }
            // Remap each section's window-LOCAL indices to GLOBAL
            // canonical indices via time matching, then express as
            // window-local for the stitcher's offset math (which
            // adds window.startIdx to each section's startIndex).
            const remapped = [];
            for (const s of wr.sections || []) {
              const localStartEntry = wr.windowEntries?.[s.startIndex];
              const localEndEntry = wr.windowEntries?.[s.endIndex];
              if (!localStartEntry || !localEndEntry) continue;
              const globalStart = canonicalIndexForOffset(
                canonicalEntries,
                localStartEntry.offset || 0
              );
              const globalEnd = canonicalIndexForOffset(
                canonicalEntries,
                localEndEntry.offset || 0
              );
              if (globalStart < 0 || globalEnd < 0) continue;
              remapped.push({
                startIndex: globalStart - globalStartIdx,
                endIndex: globalEnd - globalStartIdx,
                title: s.title,
                summary: s.summary,
              });
            }
            return {
              window: {
                startIdx: globalStartIdx,
                endIdx: globalEndIdx,
                bodyStartIdx: globalBodyStartIdx,
              },
              ok: true,
              sections: remapped,
              model: wr.model,
            };
          });
          const stitched = stitchAnalysisResults(syntheticResults);
          analyzeResult = {
            text: JSON.stringify({ sections: stitched }),
            model: pipelinedRaw.dominantModel,
            attempts: pipelinedRaw.attempts,
            usage: null,
          };
        } else {
          analyzeResult = await runChunkedAnalysis({
            transcriptText: txResult.text || "",
            backend: analyzeBackend,
            pipelineBackend: chosenAnalyzeBackend,
            jobId: summaryJobId,
            batchId: null,
            mediaUrl,
            title: title || null,
            installId,
            licenseFingerprint: licenseFp,
            source: "summarize-url",
            computeCostDetails,
            bodySeconds: bodyMin * 60,
            overlapSeconds: overlapMin * 60,
            concurrency: anConcurrency,
            cutoffSeconds: cutoffMin * 60,
            analyzePromptOverride,
            totalAudioSec: audio.seconds || 0,
            targetTotalsByBucket,
            onWindowComplete: ({ windowIdx, totalWindows, ownedSections, windowBodySeconds, model, durationMs }) => {
              appendEvent(job.id, "window_complete", {
                windowIdx,
                totalWindows,
                ownedSections,
                windowBodySeconds,
                model,
                durationMs,
              });
              setProgress(job.id, `analyze window ${windowIdx + 1}/${totalWindows} done`);
              const sectionCount = Array.isArray(ownedSections)
                ? ownedSections.length
                : (typeof ownedSections === "number" ? ownedSections : 0);
              console.log(
                `[summarize-url ${job.id.slice(0, 8)}] analyze window ${windowIdx + 1}/${totalWindows} done in ${(durationMs / 1000).toFixed(1)}s (${sectionCount} sections, model=${model || "?"})`
              );
            },
          });
        }
      } catch (err) {
        // All analyze windows failed (pipelined OR sequential).
        // Refund credit, mark job failed.
        if (reusedSummaryJob) await refundJob({ creditKey, installId, license, jobId: summaryJobId });
        const msg = (err?.message || String(err)).slice(0, 400);
        markFailed(job.id, "analyze_failed: " + msg);
        return;
      }

      // ── Post-cluster polish (Phase 2) ───────────────────────────
      // After diarization + clustering produced speaker IDs AND
      // analyze produced topic sections, run a two-stage LLM pass
      // that (1) infers real speaker names from the labeled
      // transcript + episode metadata, and (2) rewrites section
      // summaries to attribute statements to those speakers.
      // Skipped when:
      //   - Operator disabled via Settings
      //   - Diarization didn't run (no speakers to attribute)
      //   - Fewer than 2 speakers detected (nothing to differentiate)
      //   - Analyze had no sections to polish
      //
      // Failure handling: both stages are best-effort. If either
      // fails entirely, we keep the unpolished analyze output and
      // null speaker names. The user still gets a working result.
      let speakerNames = null;
      const polishEnabled = cfg.relay_post_cluster_polish_enabled !== false;
      const detectedSpeakerCount = Object.keys(txResult?.speakers || {}).length;
      const parsedAnalysisForPolish = tryParseJson(analyzeResult.text);
      const polishableSections = Array.isArray(parsedAnalysisForPolish?.sections)
        ? parsedAnalysisForPolish.sections
        : null;
      if (
        polishEnabled &&
        detectedSpeakerCount >= 2 &&
        Array.isArray(txResult?.segments) &&
        polishableSections &&
        polishableSections.length > 0
      ) {
        const polishStart = Date.now();
        try {
          // Stage 1 — global speaker name inference. Single call,
          // returns map { Speaker_A: "Matt Hill", Speaker_B: null, ... }.
          speakerNames = await runNameInference({
            speakers: txResult.speakers,
            transcriptSegments: txResult.segments,
            channelHint: channel || "",
            titleHint: title || "",
            descriptionHint: description || "",
            // Three-layer prompt resolution — per-session override
            // wins, then operator-promoted default, then the
            // hardcoded default inside post-cluster-polish.js.
            promptOverride:
              cfg.relay_polish_name_inference_prompt ||
              cfg.relay_polish_name_inference_prompt_default ||
              "",
            backend: analyzeBackend,
            pipelineBackend: chosenAnalyzeBackend,
            jobId: summaryJobId,
            batchId: null,
            mediaUrl,
            installId,
            licenseFingerprint: licenseFp,
            source: "summarize-url",
            computeCostDetails,
          });

          // Stage 2 — per-window summary polish (parallel). Rewrites
          // section summaries to attribute statements to speakers.
          // Sections whose window's polish fails keep their
          // original summary. Titles + indices are never modified.
          const canonicalEntriesForPolish = parseBracketedTranscript(txResult.text || "");
          const { planWindowsByDuration } = await import("../chunked-analyze.js");
          const windowsForPolish = planWindowsByDuration({
            totalAudioSec: audio.seconds || 0,
            bodySeconds: bodyMin * 60,
            overlapSeconds: overlapMin * 60,
            cutoffSeconds: cutoffMin * 60,
          });
          const polishedSections = await runSummaryPolish({
            sections: polishableSections,
            canonicalEntries: canonicalEntriesForPolish,
            windows: windowsForPolish,
            transcriptSegments: txResult.segments,
            speakerNames,
            speakerStats: txResult.speakers,
            promptOverride:
              cfg.relay_polish_summary_rewrite_prompt ||
              cfg.relay_polish_summary_rewrite_prompt_default ||
              "",
            backend: analyzeBackend,
            concurrency: anConcurrency,
            pipelineBackend: chosenAnalyzeBackend,
            jobId: summaryJobId,
            batchId: null,
            mediaUrl,
            installId,
            licenseFingerprint: licenseFp,
            source: "summarize-url",
            computeCostDetails,
          });
          // Overwrite analyzeResult.text with the polished sections
          // so downstream code paths (saveJobOutput, markComplete)
          // see the polished version.
          analyzeResult = {
            ...analyzeResult,
            text: JSON.stringify({ sections: polishedSections }),
          };
          const polishMs = Date.now() - polishStart;
          console.log(
            `[summarize-url ${job.id.slice(0, 8)}] post-cluster polish done in ${(polishMs / 1000).toFixed(1)}s — ${Object.values(speakerNames || {}).filter(Boolean).length}/${detectedSpeakerCount} speakers named`
          );
        } catch (polishErr) {
          // Polish failure should never kill the request — keep the
          // unpolished output. Operator can see what went wrong in
          // the audit log; the user gets a working result either way.
          console.warn(
            `[summarize-url ${job.id.slice(0, 8)}] polish pass failed (keeping unpolished output): ${polishErr?.message || polishErr}`
          );
          speakerNames = null;
        }
      } else if (polishEnabled) {
        // Diagnostic — surface why we skipped so operator can see
        // it without grep-ing the source.
        const reason = !Array.isArray(txResult?.segments)
          ? "no transcript segments"
          : detectedSpeakerCount < 2
            ? `only ${detectedSpeakerCount} speaker(s) detected`
            : !polishableSections
              ? "no parseable sections"
              : "no sections to polish";
        console.log(
          `[summarize-url ${job.id.slice(0, 8)}] polish skipped — ${reason}`
        );
      }

      // ── Commit credit + finalize ────────────────────────────────
      // Credit policy: charged ONLY when the full pipeline completes
      // cleanly. "Cleanly" means EVERY analyze window succeeded —
      // any per-window failure voids the charge even if the stitched
      // output is still usable. Rationale: a partial result is
      // surfacing degraded quality to the user; charging full price
      // for that would erode trust. The user keeps the partial
      // output (saved to disk + returned in the result envelope),
      // the operator eats the cost of the compute they paid for on
      // the windows that did succeed.
      const anyWindowFailed = (analyzeResult?.attempts?.failed ?? 0) > 0;
      let creditCharged = 0;
      if (anyWindowFailed) {
        // Don't charge. If a previous endpoint (legacy two-call path)
        // pre-charged on the same X-Recap-Job-Id, refund it.
        if (reusedSummaryJob) await refundJob({ creditKey, installId, license, jobId: summaryJobId });
        console.log(
          `[summarize-url ${job.id.slice(0, 8)}] partial analyze (${analyzeResult.attempts.failed}/${analyzeResult.attempts.windows} windows failed) — credit NOT charged`
        );
      } else if (!reusedSummaryJob) {
        await commitCredit({ creditKey, installId, license, backend: chosenBackend, tier });
        await markJobCharged({ creditKey, installId, license, jobId: summaryJobId, backend: chosenBackend, tier });
        creditCharged = 1;
      }

      // Phase 1D: build the speaker-tagged transcript segments
      // ONCE so both the operator-output store AND the Recap-facing
      // markComplete envelope share the same payload. Each segment
      // carries `start`, `end`, `text`, and (when diarization ran)
      // `speaker` + `speaker_confidence`. Null when diarization was
      // off and txResult.segments isn't an array.
      const transcriptSegments = Array.isArray(txResult?.segments)
        ? txResult.segments.map((s) => ({
            start: s.start || 0,
            end: s.end || 0,
            text: s.text || "",
            speaker: s.speaker || null,
            speaker_confidence: s.speaker_confidence ?? null,
            // Phase 2 — set when the post-cluster suppression pass
            // reassigned this segment's source cluster to an anchor
            // (best-guess attribution). Recap chip will show "?".
            speaker_uncertain: !!s.speaker_uncertain,
          }))
        : null;

      // Save the full transcript+analysis to disk for the dashboard
      // when the operator has opted in via relay_save_user_outputs.
      if (cfg.relay_save_user_outputs) {
        await saveJobOutput(summaryJobId || job.id, {
          batch_id: null,
          source: "summarize-url",
          transcript: txResult?.text || "",
          analysis: tryParseJson(analyzeResult.text),
          analysis_raw_text: analyzeResult.text || null,
          // Persist diarization output too so it shows up in the
          // operator dashboard's per-job viewer. Both keys are null
          // when diarization was off.
          speakers: txResult?.speakers || null,
          // Phase 2 — inferred speaker names from the post-cluster
          // polish pass. Null if polish was skipped or all names
          // came back null. Recap renders these in place of
          // "Speaker A" in the legend.
          speaker_names: speakerNames || null,
          diarization: txResult?.diarization || null,
          transcript_segments: transcriptSegments,
          meta: {
            title: title || null,
            media_url: mediaUrl,
            audio_seconds: audio.seconds || null,
            audio_bytes: audio.bytes,
            captions_mode: null,
            transcribe_backend: chosenBackend,
            transcribe_model: txResult?.model || null,
            analyze_backend: chosenAnalyzeBackend,
            analyze_model: analyzeResult.model || null,
          },
        });
      }

      const wallTimeMs = Date.now() - workerT0;
      markComplete(job.id, {
        result: {
          transcript: txResult.text || "",
          analysis: tryParseJson(analyzeResult.text),
          analysis_raw_text: analyzeResult.text || "",
          // Echo back the resolved title (either operator-supplied via
          // the kickoff body or extracted by yt-dlp during download).
          // Recap-app's relay-mode branch was previously saving every
          // submission as "Untitled" because the title only existed
          // inside the relay's worker — this envelope field is how
          // we hand it back across the SSE done event.
          title: title || null,
          transcribe_model: txResult.model || null,
          analyze_model: analyzeResult.model || null,
          audio_seconds: audio.seconds || null,
          audio_bytes: audio.bytes || null,
          chunk_count: txResult.chunk_count ?? null,
          analyze_windows: analyzeResult.attempts?.windows ?? null,
          analyze_windows_failed: analyzeResult.attempts?.failed ?? null,
          wall_time_ms: wallTimeMs,
          // Phase 1D fields (null when diarization didn't run):
          speakers: txResult?.speakers || null,
          transcript_segments: transcriptSegments,
          // Phase 2 — speaker name map from the post-cluster polish
          // pass. Null when polish was skipped or all names came
          // back null. Recap merges this into the speakers legend
          // (showing "Matt Hill · 24:42" instead of "Speaker A").
          speaker_names: speakerNames || null,
        },
        credit_charged: creditCharged,
        tier,
      });
      console.log(
        `[summarize-url ${job.id.slice(0, 8)}] complete in ${(wallTimeMs / 1000).toFixed(1)}s (TX ${chosenBackend}, AN ${chosenAnalyzeBackend})`
      );
      } finally {
        // Release the hardware FIFO slot so the next queued job can
        // start. Runs on success path, on every early-return inside
        // the wrapped block, and on uncaught exceptions (in which
        // case the throw propagates to the .catch below after this
        // finally executes). releaseHardwareSlot is null when the
        // job was Gemini-only and never acquired a slot — no-op.
        if (releaseHardwareSlot) releaseHardwareSlot();
      }
    })().catch((err) => {
      markFailed(job.id, "worker_crashed: " + (err?.message || String(err)));
      console.error(`[summarize-url ${job.id.slice(0, 8)}] worker crashed:`, err);
    });

    const body = await envelope({
      result: {
        job_id: job.id,
        status: "queued",
        kind: "summarize-url",
      },
      creditKey,
      installId,
      license,
      tier,
    });
    res.json(body);
  });

  // GET /relay/summarize-url/:jobId/events — SSE stream of progress
  // and per-window events. Replays the existing event log on connect
  // (so a client that connects 2s after the kickoff still sees the
  // download + transcribe events), then pushes live events as they
  // arrive until the job reaches a terminal state.
  router.get("/summarize-url/:jobId/events", async (req, res) => {
    let identity;
    try {
      identity = await resolveIdentity(req);
    } catch {
      return res.status(401).json({ error: "unauthorized" });
    }
    const ownerKey = identity.creditKey;
    if (!ownerKey) {
      return res.status(400).json({ error: "missing identity" });
    }
    const job = getJob(req.params.jobId);
    if (!job) return res.status(404).json({ error: "job not found or expired" });
    // New jobs carry metadata.owner = creditKey; older jobs only carry
    // install_id. Authorize by whichever the job has — and never leak the
    // existence of another identity's job.
    const allowed = job.metadata?.owner
      ? job.metadata.owner === ownerKey
      : identity.installId && job.install_id === identity.installId;
    if (!allowed) {
      return res.status(404).json({ error: "job not found or expired" });
    }

    // SSE headers.
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache, no-transform");
    res.setHeader("Connection", "keep-alive");
    res.setHeader("X-Accel-Buffering", "no"); // disable nginx buffering
    res.flushHeaders?.();

    const send = (event) => {
      try {
        res.write(`event: ${event.type}\n`);
        res.write(`data: ${JSON.stringify(event.data)}\n`);
        res.write(`id: ${event.ts}\n\n`);
      } catch (err) {
        // Connection closed mid-write — caller's cleanup handles it.
      }
    };

    // Replay history. The client may have missed events that fired
    // between the POST returning and the GET arriving.
    for (const ev of job.events) {
      send(ev);
    }
    // If the job is already terminal, close immediately.
    if (job.status === "complete" || job.status === "failed") {
      res.end();
      return;
    }

    // Subscribe to live events. Returned function MUST be called on
    // any disconnect path or the subscriber leaks.
    const unsubscribe = subscribeToJob(job.id, send);

    // Heartbeat every 25s. Many proxies (nginx default 60s, others
    // ~30s) close idle SSE connections — a periodic comment frame
    // keeps the conn alive between sparse events without polluting
    // the event stream.
    const heartbeat = setInterval(() => {
      try {
        res.write(": ping\n\n");
      } catch {}
    }, 25_000);

    const cleanup = () => {
      clearInterval(heartbeat);
      if (unsubscribe) unsubscribe();
    };
    req.on("close", cleanup);
    req.on("aborted", cleanup);
    res.on("close", cleanup);
  });

  return router;
}

function tryParseJson(text) {
  if (!text || typeof text !== "string") return null;
  let s = text.trim();
  const cb = s.match(/```(?:json)?\s*([\s\S]*?)```/);
  if (cb) s = cb[1].trim();
  try {
    return JSON.parse(s);
  } catch {
    return null;
  }
}