// POST /admin/test-run — operator-side benchmarking flow.
//
// Same end-to-end pipeline as /relay/transcribe-url, but with two
// key differences:
//   1. The operator can OVERRIDE backend + model per call, bypassing
//      planBackend's tier/preference logic. Used by the dashboard's
//      benchmark suite to test specific permutations.
//   2. The audit row is tagged with batch_id + source="admin-test"
//      so test runs are clearly distinguishable from real user
//      traffic in the Jobs tab (and filterable / hideable from view).
//
// Request body (admin-auth-gated by virtue of being under /admin/*):
//   {
//     media_url:           string, required
//     type?:               "youtube" | "podcast"
//     title?:              string
//     transcribe_backend:  "gemini" | "hardware", required
//     transcribe_model?:   string (gemini model id; ignored when hardware)
//     analyze_backend:     "gemini" | "hardware", required
//     analyze_model?:      string
//     batch_id?:           string — groups multiple test runs into one suite
//   }
//
// Response (immediate; job runs in background):
//   { result: { job_id, status: "queued", batch_id } }
// Poll GET /admin/jobs/:id (existing) for status; final transcript +
// analyze result lands in the Jobs table once complete.

import express from "express";
import fs from "fs/promises";
import os from "os";
import path from "path";
import { randomUUID } from "crypto";
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,
  markRunning,
  setProgress,
  markComplete,
  markFailed,
} from "../jobs.js";
import {
  looksLikeYouTube,
  downloadDirect,
  downloadYouTube,
} from "./transcribe-url.js";
import { fetchYouTubeCaptions } from "../youtube-captions.js";
import { saveJobOutput } from "../output-store.js";
import { runChunkedAnalysis } from "../chunked-analyze.js";

// Synthetic install_id used for all test-run audit rows. Keeps them
// out of any real-user aggregations + makes them filterable in the
// Jobs tab via the existing install-id filter.
const TEST_INSTALL_ID = "admin-test";

// ── TX-sharing cache ────────────────────────────────────────────
// The benchmark suite has paired permutations that use the SAME
// transcribe config but differ in their analyze backend:
//   pair 1+6: TX = gemini-3.1-flash-lite (then AN gemini vs hardware)
//   pair 4+5: TX = hardware              (then AN hardware vs gemini)
//   pair 7+8: TX = captions              (then AN gemini vs hardware)
// Without sharing, running both members of a pair re-transcribes
// the same audio twice — wasteful (cost + wall time).
//
// Implementation: an in-memory Map keyed on (mediaUrl, txConfig)
// whose values are PROMISES for the transcript. The first request
// in the pair inserts a pending Promise; subsequent requests with
// the same key await that Promise. Completed entries linger in the
// cache for ~10 minutes so a "rerun last" benchmark within that
// window also dedupes. Cache entries auto-expire to bound memory.
//
// The cache is process-local (single relay process); a relay
// restart clears it. That's fine — benchmark suites are operator-
// initiated and short-lived.
const TX_CACHE_TTL_MS = 10 * 60 * 1000;
const txCache = new Map(); // key → { promise, expiresAt }

function txCacheKey({ mediaUrl, captionsMode, txBackend, txModel }) {
  if (captionsMode === "use") return `captions:${mediaUrl}`;
  return `tx:${txBackend}:${txModel || "(default)"}:${mediaUrl}`;
}

function getOrComputeTx(key, computeFn) {
  const now = Date.now();
  // Evict expired entries opportunistically.
  for (const [k, v] of txCache) {
    if (v.expiresAt < now) txCache.delete(k);
  }
  const existing = txCache.get(key);
  if (existing && existing.expiresAt > now) {
    return {
      promise: existing.promise,
      cached: true,
      startedAt: existing.startedAt,
    };
  }
  const startedAt = Date.now();
  const promise = computeFn();
  txCache.set(key, { promise, expiresAt: now + TX_CACHE_TTL_MS, startedAt });
  // If the compute fails, evict the entry so the next attempt
  // gets a fresh try (don't cache failures).
  promise.catch(() => txCache.delete(key));
  return { promise, cached: false, startedAt };
}

// Strip code fences + parse a JSON-formatted analyze response into
// the { sections: [...] } shape Recap's render expects. Returns
// null on parse failure so the saved output can store the raw text
// for forensic review.
function safeParseSections(text) {
  if (!text || typeof text !== "string") return null;
  let jsonStr = text.trim();
  const cb = jsonStr.match(/```(?:json)?\s*([\s\S]*?)```/);
  if (cb) jsonStr = cb[1].trim();
  try {
    const parsed = JSON.parse(jsonStr);
    return parsed && Array.isArray(parsed.sections) ? parsed : null;
  } catch {
    return null;
  }
}

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

  router.post("/test-run", express.json({ limit: "1mb" }), async (req, res) => {
    const {
      media_url: mediaUrl,
      type,
      title,
      transcribe_backend: txBackend,
      transcribe_model: txModel,
      analyze_backend: anBackend,
      analyze_model: anModel,
      batch_id: batchId,
      // When captions_mode === "use", the relay fetches YouTube
      // captions via yt-dlp instead of downloading+transcribing the
      // audio. Transcribe-backend/model are ignored in that case;
      // the captions text feeds straight into analyze. Only works
      // for YouTube URLs (no captions for podcast .mp3 enclosures).
      captions_mode: captionsMode,
    } = req.body || {};

    if (!mediaUrl || typeof mediaUrl !== "string") {
      return res.status(400).json({ error: "missing or non-string media_url" });
    }
    const useCaptions = captionsMode === "use";
    if (!useCaptions && !["gemini", "hardware"].includes(txBackend)) {
      return res.status(400).json({ error: "transcribe_backend must be 'gemini' or 'hardware' (unless captions_mode='use')" });
    }
    if (!["gemini", "hardware"].includes(anBackend)) {
      return res.status(400).json({ error: "analyze_backend must be 'gemini' or 'hardware'" });
    }

    const effectiveBatchId = batchId || randomUUID();
    const job = createJob({
      kind: "admin-test-run",
      installId: TEST_INSTALL_ID,
      metadata: {
        media_url: mediaUrl,
        title,
        transcribe_backend: useCaptions ? "captions" : txBackend,
        analyze_backend: anBackend,
        batch_id: effectiveBatchId,
        captions_mode: captionsMode || null,
      },
    });

    // Hand back the job_id immediately; the dashboard polls for status.
    res.json({
      result: {
        job_id: job.id,
        status: "queued",
        batch_id: effectiveBatchId,
        kind: "admin-test-run",
      },
    });

    // Bundle all the worker-input fields into a ctx object so the
    // worker can be invoked from BOTH this single-perm endpoint AND
    // the /test-run-suite endpoint (which mints jobs upfront then
    // fires the same worker per phase).
    const ctx = {
      mediaUrl, type, title,
      txBackend, txModel, anBackend, anModel,
      batchId: effectiveBatchId, captionsMode, useCaptions,
    };
    executeTestRunWorker(job, ctx).catch((err) => {
      markFailed(job.id, "worker_crashed: " + (err?.message || String(err)));
      console.error(`[admin/test-run ${job.id.slice(0, 8)}] worker crashed:`, err);
    });
  });

  // ── POST /admin/test-run-suite ──────────────────────────────
  // Server-side benchmark runner. Accepts an ARRAY of permutations,
  // mints jobs for all of them upfront (so the client can show the
  // table immediately), and runs the phase-based concurrent
  // execution server-side. Key property: the suite KEEPS RUNNING
  // even if the operator's browser closes / phone sleeps / tab
  // refreshes — the work is in a background loop on the relay
  // process, not in the dashboard's JavaScript.
  //
  // Phases are grouped by TX fingerprint so paired permutations
  // (1+6, 4+5, 7+8) fire concurrently and share TX via the existing
  // in-memory inflight-promise cache.
  router.post("/test-run-suite", express.json({ limit: "10mb" }), async (req, res) => {
    const { media_url: mediaUrl, permutations } = req.body || {};
    if (!mediaUrl || typeof mediaUrl !== "string") {
      return res.status(400).json({ error: "missing or non-string media_url" });
    }
    if (!Array.isArray(permutations) || permutations.length === 0) {
      return res.status(400).json({ error: "permutations must be a non-empty array" });
    }
    const batchId = randomUUID();
    const items = [];
    for (let i = 0; i < permutations.length; i++) {
      const p = permutations[i] || {};
      const ctx = {
        mediaUrl,
        type: p.type,
        title: p.title || `permutation ${i + 1}`,
        txBackend: p.transcribe_backend,
        txModel: p.transcribe_model,
        anBackend: p.analyze_backend,
        anModel: p.analyze_model,
        batchId,
        captionsMode: p.captions_mode,
        useCaptions: p.captions_mode === "use",
      };
      // Validate per-perm — partial failures shouldn't poison the
      // whole batch; mark them so the worker can record the error.
      if (!ctx.useCaptions && !["gemini", "hardware"].includes(ctx.txBackend)) {
        ctx._validationError = `permutation ${i + 1}: transcribe_backend must be 'gemini' or 'hardware'`;
      } else if (!["gemini", "hardware"].includes(ctx.anBackend)) {
        ctx._validationError = `permutation ${i + 1}: analyze_backend must be 'gemini' or 'hardware'`;
      }
      const job = createJob({
        kind: "admin-test-run",
        installId: TEST_INSTALL_ID,
        metadata: {
          media_url: ctx.mediaUrl,
          title: ctx.title,
          transcribe_backend: ctx.useCaptions ? "captions" : ctx.txBackend,
          analyze_backend: ctx.anBackend,
          batch_id: batchId,
          captions_mode: ctx.captionsMode || null,
          suite_position: i + 1,
        },
      });
      items.push({ job, ctx });
    }

    // Respond immediately with the planned IDs so the dashboard can
    // start polling /admin/jobs-history?batch_id=<batchId> without
    // blocking on the actual work.
    res.json({
      result: {
        batch_id: batchId,
        status: "queued",
        job_ids: items.map((it) => it.job.id),
        total: items.length,
        kind: "admin-test-run-suite",
      },
    });

    // ── Background phase runner ──
    // Group items by TX fingerprint into phases. Permutations within
    // a phase fire concurrently (their underlying TX dedupes via the
    // cache); phases themselves run sequentially so we don't overload
    // the transcribe backends. Failures don't abort the suite.
    setImmediate(async () => {
      try {
        const phases = groupItemsByTxFingerprint(items);
        console.log(
          `[admin/test-run-suite] batch=${batchId.slice(0, 8)} ${items.length} perms in ${phases.length} phases`
        );
        for (let pi = 0; pi < phases.length; pi++) {
          const phase = phases[pi];
          console.log(
            `[admin/test-run-suite] batch=${batchId.slice(0, 8)} phase ${pi + 1}/${phases.length}: firing ${phase.length} perm${phase.length === 1 ? "" : "s"}`
          );
          await Promise.allSettled(
            phase.map(async (item) => {
              if (item.ctx._validationError) {
                markFailed(item.job.id, item.ctx._validationError);
                await recordCall({
                  install_id: TEST_INSTALL_ID,
                  tier: "core",
                  pipeline: "transcribe",
                  backend: null,
                  model: null,
                  status: "error",
                  duration_ms: 0,
                  cost_usd: 0,
                  job_id: item.job.id,
                  batch_id: batchId,
                  source: "admin-test",
                  media_url: item.ctx.mediaUrl,
                  title: item.ctx.title,
                  error: item.ctx._validationError,
                });
                return;
              }
              try {
                await executeTestRunWorker(item.job, item.ctx);
              } catch (err) {
                markFailed(item.job.id, "worker_crashed: " + (err?.message || String(err)));
                console.error(
                  `[admin/test-run-suite ${item.job.id.slice(0, 8)}] worker crashed:`,
                  err
                );
              }
            })
          );
        }
        console.log(`[admin/test-run-suite] batch=${batchId.slice(0, 8)} complete`);
      } catch (err) {
        console.error(`[admin/test-run-suite] batch=${batchId.slice(0, 8)} runner crashed:`, err);
      }
    });
  });
  return router;
}

// Group { job, ctx } items by their TX fingerprint into phases.
// Items with the same fingerprint share a phase so they hit the
// TX-share cache. Phase order is preserved from the input array
// (first appearance of a fingerprint wins).
function groupItemsByTxFingerprint(items) {
  const phases = [];
  const seen = new Map();
  for (const item of items) {
    const fp = item.ctx.useCaptions
      ? `captions:${item.ctx.mediaUrl}`
      : `tx:${item.ctx.txBackend}:${item.ctx.txModel || ""}:${item.ctx.mediaUrl}`;
    if (seen.has(fp)) {
      phases[seen.get(fp)].push(item);
    } else {
      seen.set(fp, phases.length);
      phases.push([item]);
    }
  }
  return phases;
}

// Extracted worker — runs the full download / transcribe / analyze
// pipeline for one permutation. Used by both /admin/test-run (one
// permutation) and /admin/test-run-suite (many permutations
// orchestrated server-side in phases). Body is the same flow the
// inline IIFE used previously; ctx replaces what were closure refs.
async function executeTestRunWorker(job, ctx) {
  const {
    mediaUrl, type, title,
    txBackend, txModel, anBackend, anModel,
    batchId: effectiveBatchId,
    captionsMode, useCaptions,
  } = ctx;
  // The legacy body of the IIFE follows verbatim (with `job` already
  // passed in, and the closure vars now destructured from ctx).
  {
      const workerT0 = Date.now();
      markRunning(job.id);

      const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "admin-tr-"));
      const isYT = type === "youtube" || (!type && looksLikeYouTube(mediaUrl));

      // ── Captions fast-path branch ──
      // For YouTube URLs with captions_mode="use", fetch caption track
      // via yt-dlp and skip audio download + Gemini transcribe
      // entirely. The captions text feeds straight into analyze.
      if (useCaptions) {
        if (!isYT) {
          await recordCall({
            install_id: TEST_INSTALL_ID,
            tier: "core",
            pipeline: "transcribe",
            backend: "captions",
            model: null,
            status: "error",
            duration_ms: 0,
            cost_usd: 0,
            job_id: job.id,
            batch_id: effectiveBatchId,
            source: "admin-test",
            media_url: mediaUrl,
            title: title || null,
            error: "captions_mode='use' requires a YouTube URL (no captions for podcast audio)",
          });
          markFailed(job.id, "captions_mode requires YouTube URL");
          try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
          return;
        }
        setProgress(job.id, "fetching captions…");
        const capStart = Date.now();
        let cap;
        let capFromCache = false;
        let capSharedStartedAt = capStart;
        const capKey = txCacheKey({ mediaUrl, captionsMode: "use" });
        try {
          const { promise, cached, startedAt: sharedStartedAt } = getOrComputeTx(capKey, () =>
            fetchYouTubeCaptions({ url: mediaUrl, tmpDir })
          );
          capFromCache = cached;
          capSharedStartedAt = sharedStartedAt || capStart;
          if (cached) setProgress(job.id, "reusing shared captions from paired permutation…");
          cap = await promise;
        } catch (err) {
          await recordCall({
            install_id: TEST_INSTALL_ID,
            tier: "core",
            pipeline: "transcribe",
            backend: "captions",
            model: null,
            status: "error",
            duration_ms: Date.now() - capStart,
            audio_seconds: null,
            cost_usd: 0,
            job_id: job.id,
            batch_id: effectiveBatchId,
            source: "admin-test",
            media_url: mediaUrl,
            title: title || null,
            error: (err?.message || String(err)).slice(0, 300),
          });
          markFailed(job.id, "captions_fetch_failed: " + (err?.message || err));
          try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
          return;
        }
        // Record the captions "transcribe" row. backend="captions"
        // so the dashboard can filter / display it distinctly.
        // When this permutation reused a paired sibling's captions
        // fetch, the wall-time we attribute is the underlying fetch's
        // wall-time (from the cache entry's startedAt) — so the
        // dashboard's per-row TX-rate columns show real numbers on
        // BOTH paired rows, not "—" on the sibling. The `source`
        // flag "admin-test-shared-tx" lets aggregate analytics dedupe.
        await recordCall({
          install_id: TEST_INSTALL_ID,
          tier: "core",
          pipeline: "transcribe",
          backend: "captions",
          model: cap.captions_source === "auto" ? "youtube-auto" : "youtube-manual",
          status: "success",
          duration_ms: Date.now() - capSharedStartedAt,
          audio_seconds: cap.duration_seconds || null,
          audio_bytes: null,        // no audio downloaded
          download_ms: null,        // n/a
          chunk_count: 1,
          cost_usd: 0,
          job_id: job.id,
          batch_id: effectiveBatchId,
          source: capFromCache ? "admin-test-shared-tx" : "admin-test",
          media_url: mediaUrl,
          title: title || null,
        });
        setProgress(job.id, "analyzing topics…");
        const cfg2 = await getConfigSnapshot();
        const hw2 = await resolveHardwareConfig(cfg2);
        let anResultForCaptions = null;
        try {
          anResultForCaptions = await runAnalyzeForTestRun({
            transcriptText: cap.text || "",
            anBackend,
            anModel,
            cfg: cfg2,
            hw: hw2,
            jobId: job.id,
            batchId: effectiveBatchId,
            mediaUrl,
            title,
            audioSeconds: cap.duration_seconds || null,
            audioBytes: null,
          });
        } catch (err) {
          console.warn(`[admin/test-run ${job.id.slice(0, 8)}] analyze failed (captions): ${err?.message || err}`);
        }
        // Save output (test-runs always persist regardless of the
        // save-user-outputs flag).
        await saveJobOutput(job.id, {
          batch_id: effectiveBatchId,
          source: "admin-test",
          transcript: cap.text || "",
          analysis: anResultForCaptions ? safeParseSections(anResultForCaptions.text) : null,
          analysis_raw_text: anResultForCaptions?.text || null,
          meta: {
            title: title || null,
            media_url: mediaUrl,
            audio_seconds: cap.duration_seconds || null,
            audio_bytes: null,
            captions_mode: "use",
            captions_source: cap.captions_source || null,
            transcribe_backend: "captions",
            transcribe_model: cap.captions_source === "auto" ? "youtube-auto" : "youtube-manual",
            analyze_backend: anBackend,
            analyze_model: anResultForCaptions?.model || null,
          },
        });
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
        markComplete(job.id, {
          result: { transcribe_model: "captions", batch_id: effectiveBatchId },
        });
        return;
      }

      // ── Audio download path (no captions) ──
      setProgress(job.id, "downloading media…");
      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);
        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: TEST_INSTALL_ID,
          tier: "core",
          pipeline: "transcribe",
          backend: txBackend,
          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: job.id,
          batch_id: effectiveBatchId,
          source: "admin-test",
          media_url: mediaUrl,
          title: title || null,
          error: "download_failed: " + msg,
        });
        markFailed(job.id, "download_failed: " + msg);
        return;
      }

      // ── Transcription with the operator's chosen backend ──
      // Uses the TX-sharing cache so that paired benchmark
      // permutations (e.g. 1+6 both transcribe with gemini-3.1-flash-
      // lite) only invoke the underlying backend ONCE — the second
      // permutation awaits the first's in-flight promise and reuses
      // its transcript. Cache entries linger ~10 min so a fast
      // "Rerun last" also dedupes.
      const cfg = await getConfigSnapshot();
      const hw = await resolveHardwareConfig(cfg);
      let txResult;
      let txFromCache = false;
      const txStartedAt = Date.now();
      let txSharedStartedAt = txStartedAt;
      const cacheKey = txCacheKey({ mediaUrl, captionsMode: null, txBackend, txModel });
      try {
        const audioBuf = await fs.readFile(audio.filePath);
        const { promise, cached, startedAt: sharedStartedAt } = getOrComputeTx(cacheKey, async () => {
          if (txBackend === "gemini") {
            const backend = createGeminiBackend({
              apiKey: cfg.relay_gemini_api_key,
              transcriptionModel: txModel || 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,
              transcribePromptOverride: cfg.relay_transcribe_prompt || "",
            });
            return await backend.transcribeAudio({
              audio: audioBuf,
              mimeType: audio.mimeType || "audio/mpeg",
              title: title || "",
              offsetSeconds: 0,
            });
          }
          if (!hw.transcribe.url) {
            throw new Error("hardware transcribe URL not configured");
          }
          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,
            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,
          });
          return await backend.transcribeAudio({
            audio: audioBuf,
            mimeType: audio.mimeType || "audio/mpeg",
            offsetSeconds: 0,
          });
        });
        txFromCache = cached;
        txSharedStartedAt = sharedStartedAt || txStartedAt;
        if (cached) {
          setProgress(job.id, "reusing shared TX from paired permutation…");
        }
        txResult = await promise;
      } catch (err) {
        try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}
        const msg = (err?.message || String(err)).slice(0, 400);
        await recordCall({
          install_id: TEST_INSTALL_ID,
          tier: "core",
          pipeline: "transcribe",
          backend: txBackend,
          model: txBackend === "gemini" ? (txModel || cfg.relay_gemini_transcription_model) : (hw.transcribe.model || "(auto)"),
          status: "error",
          credit_charged: 0,
          duration_ms: Date.now() - workerT0,
          download_ms: downloadMs,
          audio_seconds: audio?.seconds || null,
          audio_bytes: audio?.bytes || null,
          cost_usd: 0,
          job_id: job.id,
          batch_id: effectiveBatchId,
          source: "admin-test",
          media_url: mediaUrl,
          title: title || null,
          error: msg,
        });
        markFailed(job.id, "transcribe_failed: " + msg);
        return;
      }

      // Audit the successful transcribe.
      const txCostDetails =
        txBackend === "gemini" && txResult.usage
          ? calcGeminiCost(txResult.model, txResult.usage)
          : { input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0 };
      // Truncation detection — same as the production routes. When
      // any chunk emitted < 80% of its expected audio, mark the
      // benchmark row partial so the operator doesn't compare a
      // truncated TX run against a clean one.
      const txTruncatedChunks = Array.isArray(txResult?.truncated_chunks)
        ? txResult.truncated_chunks
        : [];
      const txWasTruncated = txTruncatedChunks.length > 0;
      const txTruncationError = txWasTruncated
        ? `transcribe: ${txTruncatedChunks.length} chunk(s) truncated — missing ~${txTruncatedChunks.reduce((s, c) => s + (c.missingSec || 0), 0)}s of speech (model: ${txResult.model || "unknown"})`
        : null;
      await recordCall({
        install_id: TEST_INSTALL_ID,
        tier: "core",
        pipeline: "transcribe",
        backend: txBackend,
        model: txResult.model || null,
        status: txWasTruncated ? "partial" : "success",
        credit_charged: 0,
        truncated_chunks: txWasTruncated ? txTruncatedChunks : null,
        error: txTruncationError,
        // When this permutation reused a paired sibling's TX, the
        // attributed duration is the wall-time of the UNDERLYING TX
        // (from when the originating permutation kicked it off until
        // both siblings' awaits resolved) — so the per-row TX rate
        // columns in the Jobs table show real numbers on BOTH paired
        // rows, not "—" on the sibling. Cost is still zero on the
        // sibling (only the originator pays). The "admin-test-shared-tx"
        // source flag lets aggregate analytics dedupe across pairs.
        duration_ms: Date.now() - txSharedStartedAt,
        download_ms: downloadMs,
        audio_bytes: audio.bytes,
        audio_seconds: audio.seconds || null,
        job_id: job.id,
        batch_id: effectiveBatchId,
        source: txFromCache ? "admin-test-shared-tx" : "admin-test",
        media_url: mediaUrl,
        title: title || null,
        attempts: txResult.attempts || null,
        chunk_count: txResult.chunk_count ?? null,
        // Per-chunk wall-times (ms). Aggregator sums this into
        // transcribe_ms_sum so the Jobs table shows BOTH wall-time
        // (from duration_ms) and total backend compute (from sum).
        chunk_durations_ms: txResult.chunk_durations_ms || null,
        ...(txFromCache
          ? { input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0 }
          : txCostDetails),
      });

      // ── Analyze with the operator's chosen backend ──
      // For benchmarking purposes we run the chunked-analyze flow
      // directly here (mirroring Recap's behavior) so the per-window
      // performance is captured in the Jobs table. We build a simple
      // prompt from the transcript text.
      setProgress(job.id, "analyzing topics…");
      let anResult = null;
      try {
        anResult = await runAnalyzeForTestRun({
          transcriptText: txResult.text || "",
          anBackend,
          anModel,
          cfg,
          hw,
          jobId: job.id,
          batchId: effectiveBatchId,
          mediaUrl,
          title,
          audioSeconds: audio.seconds || null,
          audioBytes: audio.bytes,
        });
      } catch (err) {
        // Analyze failure is recorded (inside runAnalyzeForTestRun);
        // we still mark the job complete since transcribe succeeded.
        console.warn(`[admin/test-run ${job.id.slice(0, 8)}] analyze failed: ${err?.message || err}`);
      }

      // Save the transcript + analysis JSON to disk for the
      // dashboard's "View output" feature. Test-run jobs always
      // persist regardless of the save-user-outputs config flag.
      await saveJobOutput(job.id, {
        batch_id: effectiveBatchId,
        source: "admin-test",
        transcript: txResult.text || "",
        analysis: anResult ? safeParseSections(anResult.text) : null,
        analysis_raw_text: anResult?.text || null,
        meta: {
          title: title || null,
          media_url: mediaUrl,
          audio_seconds: audio.seconds || null,
          audio_bytes: audio.bytes,
          captions_mode: null,
          transcribe_backend: txBackend,
          transcribe_model: txResult.model || null,
          analyze_backend: anBackend,
          analyze_model: anResult?.model || null,
        },
      });
      try { await fs.rm(tmpDir, { recursive: true, force: true }); } catch {}

      markComplete(job.id, {
        result: {
          transcribe_model: txResult.model,
          batch_id: effectiveBatchId,
        },
      });
  }
}

// Run chunked analyze over the just-transcribed text using the same
// windowing strategy Recap's client uses (~18 min window body, 2 min
// overlap, N windows in flight). Each window emits its own audit row
// via recordCall (handled inside runChunkedAnalysis), so the Jobs
// table sees:
//   - analyze_windows_total = N
//   - analyze_ms            = sum of per-window duration_ms (total backend work)
//   - wall_time_ms          = elapsed from first window start → last window end
//                              (computed by job-stats.js from row timestamps)
// Per-window `audio_seconds` is the window body length (not total audio),
// so per-row rate columns (s/audio-min) divide by the right denominator.
async function runAnalyzeForTestRun({
  transcriptText,
  anBackend,
  anModel,
  cfg,
  hw,
  jobId,
  batchId,
  mediaUrl,
  title,
  audioSeconds, // unused — chunked-analyze uses per-window seconds
  audioBytes,   // unused
}) {
  // Build the right backend, then hand to runChunkedAnalysis which
  // handles per-window prompt building, parallelism, audit logging,
  // and stitching. Construction errors (missing apiKey, missing
  // hardware URL) are audited as a single failed analyze row so the
  // Jobs table shows what happened — runChunkedAnalysis only writes
  // rows once it has a backend to call.
  let backend;
  let resolvedModel;
  let computeCostDetails;
  try {
    if (anBackend === "gemini") {
      backend = createGeminiBackend({
        apiKey: cfg.relay_gemini_api_key,
        transcriptionModel: cfg.relay_gemini_transcription_model,
        analysisModel: anModel || cfg.relay_gemini_analysis_model,
        // tx knobs are unused on the analyze path but the factory
        // accepts them anyway — pass for consistency.
        txChunkSeconds: (cfg.relay_gemini_tx_chunk_minutes || 30) * 60,
        txConcurrency: cfg.relay_gemini_tx_concurrency || 12,
      });
      resolvedModel = anModel || cfg.relay_gemini_analysis_model;
      computeCostDetails = (model, usage) =>
        usage ? calcGeminiCost(model, usage) : {
          input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0,
        };
    } else {
      if (!hw.analyze.url) {
        throw new Error("hardware analyze URL not configured");
      }
      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,
      });
      resolvedModel = hw.analyze.model || null;
      computeCostDetails = () => ({
        input_tokens: 0, output_tokens: 0, thinking_tokens: 0, cost_usd: 0,
      });
    }
  } catch (err) {
    await recordCall({
      install_id: TEST_INSTALL_ID,
      tier: "core",
      pipeline: "analyze",
      backend: anBackend,
      model: anBackend === "gemini"
        ? (anModel || cfg.relay_gemini_analysis_model)
        : (hw.analyze.model || "(auto)"),
      status: "error",
      duration_ms: 0,
      audio_seconds: 0,
      cost_usd: 0,
      job_id: jobId,
      batch_id: batchId,
      source: "admin-test",
      media_url: mediaUrl,
      title: title || null,
      error: (err?.message || String(err)).slice(0, 400),
      window_idx: 0,
      window_count: 1,
    });
    throw err;
  }

  // Pull windowing tunables from config (Settings tab).
  const bodyMin = anBackend === "gemini"
    ? (cfg.relay_gemini_analyze_window_minutes || 18)
    : (cfg.relay_hardware_analyze_window_minutes || 18);
  const overlapMin = anBackend === "gemini"
    ? (cfg.relay_gemini_analyze_overlap_minutes || 2)
    : (cfg.relay_hardware_analyze_overlap_minutes || 2);
  const concurrency = anBackend === "gemini"
    ? (cfg.relay_gemini_analyze_concurrency || 12)
    : (cfg.relay_hardware_analyze_concurrency || 8);
  const cutoffMin = cfg.relay_analyze_cutoff_minutes || 25;

  const result = await runChunkedAnalysis({
    transcriptText,
    backend,
    pipelineBackend: anBackend,
    jobId,
    batchId,
    mediaUrl,
    title,
    installId: TEST_INSTALL_ID,
    source: "admin-test",
    computeCostDetails,
    bodySeconds: bodyMin * 60,
    overlapSeconds: overlapMin * 60,
    concurrency,
    cutoffSeconds: cutoffMin * 60,
    analyzePromptOverride: cfg.relay_analyze_prompt || "",
    // Section-count target wiring (matches the summarize-url path).
    // Without these, buildWindowPrompt falls back to "1 section" —
    // works defensively but means test-run benchmarks don't reflect
    // production segmentation density.
    totalAudioSec: audioSeconds || 0,
    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,
    },
  });

  return {
    text: result.text || "",
    model: result.model || resolvedModel,
    attempts: result.attempts,
  };
}