feat(video): derive embedded subtitle provenance in the protected broker (#11659) (#12011)

FU-05 subtitle adapter (Refs #11659 — deliberately not Closes: the adapter is not yet wired into the live describeVideoPart path, that composition point is sibling #12009 which just landed): server-owned, loopback-only ffprobe/ffmpeg subtitle extraction that legitimately earns the "embedded" provenance label, mirroring the existing frame-extraction lifecycle. Broker route now also serves ?subtitles=1, stamped with the shared broker fingerprint so the client-side adapter can verify the payload actually came from the trusted process. Bounded, ReDoS-safe WebVTT parser.

changelog.d/features/11659-video-bridge-embedded-subtitle-provenance.md

@@ -0,0 +1 @@
+- **feat(video bridge):** "embedded" transcript provenance can now be legitimately earned instead of merely asserted — a bounded, allowlisted (`mov_text`/`subrip`/`webvtt`) subtitle probe runs through the loopback-only Video Bridge broker (at most 2 streams, 10s subdeadline bounded by the request deadline, 256 KiB output, 4096-code-unit lines), normalized through a bounded, ReDoS-safe WebVTT parser and Zod-validated end to end. The adapter always resolves to an explicit `success`/`absent`/`transient_failure` outcome — a subtitle failure never breaks the visual description path, and only a fingerprint-verified broker response (never a caller-declared label) can produce embedded cues (#11659).

src/app/api/modality-bridge/video/extract/route.ts

@@ -1,5 +1,6 @@
 import { createErrorResponse } from "@/lib/api/errorResponse";
 import {
+  currentVideoBridgeBrokerFingerprint,
   VIDEO_BRIDGE_BROKER_PATH,
   isVideoBridgeBrokerInternalRequest,
 } from "@/lib/guardrails/videoBridgeBrokerAuth";
@@ -13,6 +14,10 @@ import {
   type VideoFocusBounds,
   type VideoSamplingPolicy,
 } from "@/lib/guardrails/videoBridgeRuntime";
+import {
+  extractVideoSubtitlesFromBytes,
+  VIDEO_SUBTITLE_SUBDEADLINE_MS,
+} from "@/lib/guardrails/videoBridgeSubtitleRuntime";
 import { resolveModelSyncInternalBaseUrl } from "@/shared/services/modelSyncScheduler";
 import { VIDEO_BRIDGE_TIMEOUT_MAX_MS } from "@/shared/constants/modalityBridgeDefaults";
 import { createLogger } from "@/shared/utils/logger";
@@ -37,6 +42,11 @@ function invalid(message: string, status = 400, headers?: Record<string, string>
   return response;
 }
 
+/** The extract route serves two mutually exclusive operations behind one authenticated path. */
+function isSubtitleProbeRequest(url: URL): boolean {
+  return url.searchParams.get("subtitles") === "1";
+}
+
 function parseFrameCount(url: URL): number | null {
   if (
     [...url.searchParams.keys()].some(
@@ -117,9 +127,107 @@ export async function readBoundedVideoBrokerBody(
 interface VideoExtractionBrokerRouteDependencies {
   deadlineSignal?: AbortSignal;
   extractFrames?: typeof extractVideoFramesFromBytes;
+  extractSubtitles?: typeof extractVideoSubtitlesFromBytes;
   queue?: VideoExtractionQueue;
 }
 
+/** Shared by both operations: declared-length pre-check, bounded read, actual-size check. */
+async function readValidatedVideoBrokerBytes(request: Request): Promise<Buffer | Response> {
+  const declaredHeader = request.headers.get("content-length");
+  const declaredLength = declaredHeader === null ? null : Number(declaredHeader);
+  if (
+    declaredLength !== null &&
+    (!Number.isFinite(declaredLength) || declaredLength < 1 || declaredLength > MAX_INPUT_BYTES)
+  ) {
+    await request.body?.cancel("Video Bridge input exceeds the byte limit");
+    return invalid("Video Bridge input exceeds the byte limit", 413);
+  }
+  let bytes: Buffer;
+  try {
+    bytes = await readBoundedVideoBrokerBody(request);
+  } catch (error) {
+    if (error instanceof Error && error.message === "VIDEO_INPUT_TOO_LARGE") {
+      return invalid("Video Bridge input exceeds the byte limit", 413);
+    }
+    return invalid("Video Bridge input could not be read");
+  }
+  if (
+    bytes.byteLength < 1 ||
+    bytes.byteLength > MAX_INPUT_BYTES ||
+    (declaredLength !== null && bytes.byteLength !== declaredLength)
+  ) {
+    return invalid("Video Bridge input exceeds the byte limit", 413);
+  }
+  return bytes;
+}
+
+/**
+ * Bounded, allowlisted-codec subtitle probe (#11659). Stamps the response with the shared
+ * broker fingerprint so the client-side adapter can refuse anything not actually produced by
+ * this trusted loopback process before treating it as "embedded" provenance.
+ */
+async function handleSubtitleProbeBrokerRequest(
+  request: Request,
+  url: URL,
+  dependencies: VideoExtractionBrokerRouteDependencies
+): Promise<Response> {
+  if ([...url.searchParams.keys()].some((key) => key !== "subtitles")) {
+    return invalid("Video Bridge subtitle probe accepts no other parameters");
+  }
+  const bytesOrResponse = await readValidatedVideoBrokerBytes(request);
+  if (bytesOrResponse instanceof Response) return bytesOrResponse;
+  const bytes = bytesOrResponse;
+
+  const deadline =
+    dependencies.deadlineSignal ?? AbortSignal.timeout(VIDEO_SUBTITLE_SUBDEADLINE_MS);
+  const signal = AbortSignal.any([request.signal, deadline]);
+  const queue = dependencies.queue ?? extractionQueue;
+  const extractSubtitles = dependencies.extractSubtitles ?? extractVideoSubtitlesFromBytes;
+  try {
+    const result = await queue.run(
+      bytes.byteLength,
+      () =>
+        extractSubtitles(bytes, {
+          maxDurationSeconds: MAX_DURATION_SECONDS,
+          signal,
+          timeoutMs: VIDEO_SUBTITLE_SUBDEADLINE_MS,
+        }),
+      signal
+    );
+    return Response.json(
+      { ...result, fingerprint: currentVideoBridgeBrokerFingerprint() },
+      { headers: { "Cache-Control": "no-store" } }
+    );
+  } catch (error) {
+    const queueCapacity =
+      error instanceof VideoExtractionQueueError && error.code === "QUEUE_CAPACITY";
+    const clientAborted = request.signal.aborted;
+    const deadlineExceeded = !clientAborted && deadline.aborted;
+    log.warn(
+      {
+        aborted: clientAborted,
+        code: clientAborted
+          ? "CLIENT_ABORTED"
+          : queueCapacity
+            ? "QUEUE_CAPACITY"
+            : deadlineExceeded
+              ? "DEADLINE_EXCEEDED"
+              : "EXTRACTION_FAILED",
+        inputBytes: bytes.byteLength,
+      },
+      "Video Bridge broker subtitle probe failed"
+    );
+    if (clientAborted) return invalid("Video extraction was aborted", 499);
+    if (deadlineExceeded) return invalid("Video extraction deadline exceeded", 504);
+    if (queueCapacity) {
+      return invalid("Video extraction capacity is temporarily unavailable", 503, {
+        "Retry-After": "1",
+      });
+    }
+    return invalid("Video extraction failed", 422);
+  }
+}
+
 export async function handleVideoExtractionBrokerRequest(
   request: Request,
   dependencies: VideoExtractionBrokerRouteDependencies = {}
@@ -134,6 +242,9 @@ export async function handleVideoExtractionBrokerRequest(
   if (request.headers.get("content-type")?.toLowerCase() !== "application/octet-stream") {
     return invalid("Video Bridge broker requires application/octet-stream");
   }
+  if (isSubtitleProbeRequest(url)) {
+    return handleSubtitleProbeBrokerRequest(request, url, dependencies);
+  }
   const frameCount = parseFrameCount(url);
   if (!frameCount) return invalid("Video Bridge frame count must be between 1 and 16");
   const samplingPolicy = parseSamplingPolicy(url);

src/lib/guardrails/videoBridgeBrokerAuth.ts

@@ -22,6 +22,23 @@ export function buildVideoBridgeBrokerHeaders(): Record<string, string> {
   return { [VIDEO_BRIDGE_BROKER_AUTH_HEADER]: brokerToken() };
 }
 
+/**
+ * The same process-local secret used to authenticate requests INTO the broker (#11659).
+ * The broker route stamps this into its JSON response body so a client-side adapter can
+ * prove the payload actually came from this trusted loopback process — never from a
+ * caller-declared label — before it is allowed to construct "embedded" provenance.
+ */
+export function currentVideoBridgeBrokerFingerprint(): string {
+  return brokerToken();
+}
+
+export function verifyVideoBridgeBrokerFingerprint(candidate: unknown): boolean {
+  if (typeof candidate !== "string" || candidate.length === 0) return false;
+  const expected = brokerToken();
+  if (candidate.length !== expected.length) return false;
+  return timingSafeEqual(Buffer.from(candidate, "utf8"), Buffer.from(expected, "utf8"));
+}
+
 function normalizeVideoBridgePrincipalId(value: string | null): string | null {
   if (!value || value.length > 256) return null;
   for (let index = 0; index < value.length; index += 1) {

src/lib/guardrails/videoBridgeBrokerClient.ts

@@ -1,3 +1,5 @@
+import { z } from "zod";
+
 import {
   fetchModelSyncInternal,
   resolveModelSyncInternalBaseUrl,
@@ -13,6 +15,7 @@ import type {
   VideoSamplingMetadata,
   VideoSamplingPolicy,
 } from "./videoBridgeRuntime";
+import { VIDEO_SUBTITLE_CODEC_ALLOWLIST } from "./videoBridgeSubtitleRuntime";
 
 export {
   VIDEO_BRIDGE_BROKER_PATH,
@@ -179,3 +182,88 @@ export async function extractVideoFramesViaBroker(
     options.frameCount
   );
 }
+
+// ─── Subtitle probe (#11659) ──────────────────────────────────────────────
+// The transport/shape half of the adapter: send bounded bytes to the same loopback-only
+// broker path, and Zod-validate the raw envelope before any provenance/text normalization
+// happens. Bounded WebVTT parsing and the success/absent/transient_failure outcome contract
+// live in `videoBridgeSubtitleProbe.ts`, which is the caller of this function.
+
+export const VIDEO_SUBTITLE_MAX_STREAMS = 2;
+// Mirrors the broker's own 256 KiB output cap as a client-side defense-in-depth bound —
+// never trust the broker (same process, but still an HTTP hop) to have enforced it.
+const VIDEO_SUBTITLE_MAX_RAW_TEXT_CODE_UNITS = 300_000;
+
+const VideoSubtitleBrokerStreamSchema = z
+  .object({
+    codecName: z.enum(VIDEO_SUBTITLE_CODEC_ALLOWLIST),
+    streamIndex: z.number().int().nonnegative(),
+    webvtt: z.string().max(VIDEO_SUBTITLE_MAX_RAW_TEXT_CODE_UNITS),
+  })
+  .strict();
+
+const VideoSubtitleBrokerResponseSchema = z
+  .object({
+    durationSeconds: z.number().positive(),
+    fingerprint: z.string().uuid(),
+    formatName: z.string().regex(/^[a-z0-9,_]{1,64}$/i),
+    streams: z.array(VideoSubtitleBrokerStreamSchema).max(VIDEO_SUBTITLE_MAX_STREAMS),
+  })
+  .strict();
+
+export type VideoSubtitleBrokerResponse = z.infer<typeof VideoSubtitleBrokerResponseSchema>;
+
+export interface BrokerSubtitleExtractionOptions {
+  signal?: AbortSignal;
+  timeoutMs: number;
+}
+
+/**
+ * Requests a bounded, allowlisted-codec subtitle probe from the loopback broker. Only the
+ * transport + envelope shape are validated here (safe container metadata, bounded stream
+ * count, canonical codec names, a well-formed fingerprint) — never trust this alone for
+ * provenance: the caller must still compare `fingerprint` against
+ * `currentVideoBridgeBrokerFingerprint()` before treating anything as "embedded".
+ */
+export async function extractVideoSubtitlesViaBroker(
+  bytes: Uint8Array,
+  options: BrokerSubtitleExtractionOptions,
+  dependencies: { fetchImpl?: typeof fetch; maxResponseBytes?: number } = {}
+): Promise<VideoSubtitleBrokerResponse> {
+  if (options.signal?.aborted) throw new Error("Video extraction request aborted");
+  const baseUrl = resolveVideoBridgeBrokerBaseUrl();
+  const url = new URL(`${baseUrl}${VIDEO_BRIDGE_BROKER_PATH}`);
+  url.searchParams.set("subtitles", "1");
+  const fetchImpl = dependencies.fetchImpl ?? fetchModelSyncInternal;
+  const timeoutSignal = AbortSignal.timeout(options.timeoutMs);
+  const signal = options.signal ? AbortSignal.any([options.signal, timeoutSignal]) : timeoutSignal;
+  let response: Response;
+  try {
+    response = await fetchImpl(url, {
+      method: "POST",
+      body: Buffer.from(bytes),
+      headers: {
+        "Content-Type": "application/octet-stream",
+        ...buildVideoBridgeBrokerHeaders(),
+      },
+      redirect: "error",
+      signal,
+    });
+  } catch {
+    if (signal.aborted) throw new Error("Video extraction request aborted");
+    throw new Error("Video extraction broker is unavailable");
+  }
+  if (!response.ok) {
+    throw new Error(`Video extraction broker failed (${response.status})`);
+  }
+  const maxResponseBytes = Math.min(
+    MAX_BROKER_RESPONSE_BYTES,
+    dependencies.maxResponseBytes ?? MAX_BROKER_RESPONSE_BYTES
+  );
+  const raw = await readBoundedResponse(response, maxResponseBytes);
+  const parsed = VideoSubtitleBrokerResponseSchema.safeParse(raw);
+  if (!parsed.success) {
+    throw new Error("Video extraction broker returned invalid subtitle metadata");
+  }
+  return parsed.data;
+}

src/lib/guardrails/videoBridgeRuntime.ts

@@ -133,7 +133,7 @@ const SAFE_FORMATS = new Set([
   "webm",
 ]);
 const SAFE_FORMAT_WHITELIST = [...SAFE_FORMATS].join(",");
-const defaultRunner: VideoCommandRunner = async (executable, args, options) => {
+export const defaultRunner: VideoCommandRunner = async (executable, args, options) => {
   const result = await execFileAsync(executable, [...args], {
     encoding: "utf8",
     maxBuffer: 1024 * 1024,
@@ -143,7 +143,7 @@ const defaultRunner: VideoCommandRunner = async (executable, args, options) => {
   });
   return { stdout: String(result.stdout), stderr: String(result.stderr) };
 };
-function assertLocalPath(filePath: string): void {
+export function assertLocalPath(filePath: string): void {
   if (!isAbsolute(filePath) || filePath.includes("\0") || filePath.includes("://")) {
     throw new Error("Video runtime requires a local path");
   }

src/lib/guardrails/videoBridgeSubtitleProbe.ts

@@ -0,0 +1,242 @@
+/**
+ * Client-side adapter that legitimately EARNS "embedded" transcript provenance (#11659,
+ * FU-05). "Embedded" must mean subtitles the server actually extracted and verified from the
+ * real container via the loopback-only Video Bridge broker — never a label the caller
+ * asserted. See `videoBridgeHelpers.ts::normalizeVideoTranscript` for the caller-declared
+ * transcript path this composes with; the client-forgery boundary and cross-source
+ * reconciliation for that seam are #11652/#12009's scope, not this file's.
+ *
+ * Contract:
+ *  - `probeEmbeddedVideoSubtitles` NEVER rejects for an expected failure mode (broker error,
+ *    timeout, malformed/oversized/forged response, unusable content) — it always resolves to
+ *    one of three explicit outcomes so a subtitle failure can never break the visual
+ *    description path (fail-open). It only rejects when the CALLER's own `signal` aborted.
+ *  - `outcome: "transient_failure"` always carries `cacheable: false` — callers must not
+ *    persist it as a stable "no subtitles" fact (a retry may behave differently).
+ *  - `outcome: "success"` is only ever reached after the broker response's fingerprint has
+ *    been verified against this process's own broker secret (`videoBridgeBrokerAuth.ts`) —
+ *    a forged or unverifiable response can never produce embedded cues.
+ */
+import { z } from "zod";
+
+import { verifyVideoBridgeBrokerFingerprint } from "./videoBridgeBrokerAuth";
+import {
+  extractVideoSubtitlesViaBroker,
+  type BrokerSubtitleExtractionOptions,
+  type VideoSubtitleBrokerResponse,
+} from "./videoBridgeBrokerClient";
+import type { VideoTranscriptCue } from "./videoBridgeHelpers";
+
+/** Extraction never runs longer than this, regardless of how much request budget remains. */
+export const VIDEO_SUBTITLE_SUBDEADLINE_MS = 10_000;
+export const VIDEO_SUBTITLE_MAX_LINE_CODE_UNITS = 4096;
+const VIDEO_SUBTITLE_MAX_CUES = 4_000;
+
+/** The 10s subtitle subdeadline is itself bounded by whatever remains of the request deadline. */
+export function resolveVideoSubtitleSubdeadlineMs(requestDeadlineRemainingMs: number): number {
+  if (!Number.isFinite(requestDeadlineRemainingMs)) return VIDEO_SUBTITLE_SUBDEADLINE_MS;
+  return Math.max(0, Math.min(VIDEO_SUBTITLE_SUBDEADLINE_MS, requestDeadlineRemainingMs));
+}
+
+// ─── Bounded, ReDoS-safe WebVTT parsing ────────────────────────────────────
+// Subtitle bytes are fully untrusted. Every regex below uses strictly bounded,
+// non-overlapping quantifiers (repo ReDoS rule) and the heavy lifting (block splitting,
+// timestamp arithmetic) is plain string/array work, never regex.
+
+const REPLACEMENT_CHARACTER = "<22>";
+// HH:MM:SS.mmm or MM:SS.mmm, each group bounded — no nested unbounded quantifiers.
+const TIMESTAMP_LINE_PATTERN =
+  /^(\d{1,2}(?::\d{2}){1,2}\.\d{1,3})[ \t]+-->[ \t]+(\d{1,2}(?::\d{2}){1,2}\.\d{1,3})/;
+const TAG_PATTERN = /<[^>]{0,200}>/g;
+
+interface CandidateCue {
+  endSeconds: number;
+  startSeconds: number;
+  text: string;
+}
+
+function parseWebVttTimestamp(value: string): number | null {
+  const segments = value.split(":");
+  if (segments.length !== 2 && segments.length !== 3) return null;
+  const secondsPart = segments[segments.length - 1];
+  const minutesPart = segments[segments.length - 2];
+  const hoursPart = segments.length === 3 ? segments[0] : "0";
+  const [wholeSecondsRaw, millisRaw = "0"] = secondsPart.split(".");
+  const hours = Number(hoursPart);
+  const minutes = Number(minutesPart);
+  const wholeSeconds = Number(wholeSecondsRaw);
+  const millis = Number(millisRaw.padEnd(3, "0").slice(0, 3));
+  if (
+    !Number.isInteger(hours) ||
+    hours < 0 ||
+    !Number.isInteger(minutes) ||
+    minutes < 0 ||
+    minutes > 59 ||
+    !Number.isInteger(wholeSeconds) ||
+    wholeSeconds < 0 ||
+    wholeSeconds > 59 ||
+    !Number.isInteger(millis) ||
+    millis < 0
+  ) {
+    return null;
+  }
+  return hours * 3_600 + minutes * 60 + wholeSeconds + millis / 1_000;
+}
+
+function splitIntoBlocks(lines: readonly string[]): string[][] {
+  const blocks: string[][] = [];
+  let current: string[] = [];
+  for (const line of lines) {
+    if (line.trim().length === 0) {
+      if (current.length > 0) blocks.push(current);
+      current = [];
+    } else {
+      current.push(line);
+    }
+  }
+  if (current.length > 0) blocks.push(current);
+  return blocks;
+}
+
+function extractTimingLine(block: readonly string[]): { textLines: string[]; timingLine: string } | null {
+  if (TIMESTAMP_LINE_PATTERN.test(block[0])) {
+    return { textLines: block.slice(1), timingLine: block[0] };
+  }
+  if (block.length > 1 && TIMESTAMP_LINE_PATTERN.test(block[1])) {
+    return { textLines: block.slice(2), timingLine: block[1] };
+  }
+  return null;
+}
+
+function buildCandidateCue(block: readonly string[], durationSeconds: number): CandidateCue | null {
+  const extracted = extractTimingLine(block);
+  if (!extracted) return null;
+  const match = TIMESTAMP_LINE_PATTERN.exec(extracted.timingLine);
+  if (!match) return null;
+  const startSeconds = parseWebVttTimestamp(match[1]);
+  const endSeconds = parseWebVttTimestamp(match[2]);
+  if (startSeconds === null || endSeconds === null) return null;
+  if (!(endSeconds > startSeconds) || startSeconds > durationSeconds + 1) return null;
+  if (extracted.textLines.length === 0) return null;
+  for (const line of extracted.textLines) {
+    // Individual lines beyond the bound, or containing the UTF-8 replacement character
+    // (this content was not valid text), invalidate the whole cue rather than truncating
+    // or silently dropping bytes.
+    if (line.length > VIDEO_SUBTITLE_MAX_LINE_CODE_UNITS) return null;
+    if (line.includes(REPLACEMENT_CHARACTER)) return null;
+  }
+  const text = extracted.textLines
+    .map((line) => line.replace(TAG_PATTERN, "").trim())
+    .filter((line) => line.length > 0)
+    .join(" ")
+    .trim();
+  if (!text) return null;
+  return { endSeconds, startSeconds, text };
+}
+
+/** Bounded WebVTT -> candidate cue list. Never throws; unparseable input yields `[]`. */
+export function parseBoundedWebVtt(rawText: string, durationSeconds: number): CandidateCue[] {
+  if (typeof rawText !== "string" || rawText.length === 0) return [];
+  if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) return [];
+  const normalized = rawText.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+  const lines = normalized.split("\n");
+  if (!lines[0]?.trim().startsWith("WEBVTT")) return [];
+  const candidates: CandidateCue[] = [];
+  for (const block of splitIntoBlocks(lines.slice(1))) {
+    if (candidates.length >= VIDEO_SUBTITLE_MAX_CUES) break;
+    const candidate = buildCandidateCue(block, durationSeconds);
+    if (candidate) candidates.push(candidate);
+  }
+  return candidates.sort((left, right) => left.startSeconds - right.startSeconds);
+}
+
+const CandidateCueSchema = z
+  .object({
+    endSeconds: z.number().finite().nonnegative(),
+    startSeconds: z.number().finite().nonnegative(),
+    text: z.string().min(1).max(VIDEO_SUBTITLE_MAX_LINE_CODE_UNITS),
+  })
+  .strict()
+  .refine((cue) => cue.endSeconds > cue.startSeconds, {
+    message: "Video subtitle cue end must be after its start",
+  });
+const CandidateCueListSchema = z.array(CandidateCueSchema).max(VIDEO_SUBTITLE_MAX_CUES);
+
+function selectCuesFromStreams(
+  streams: VideoSubtitleBrokerResponse["streams"],
+  durationSeconds: number
+): VideoTranscriptCue[] | null {
+  for (const stream of streams) {
+    const candidates = parseBoundedWebVtt(stream.webvtt, durationSeconds);
+    const validated = CandidateCueListSchema.safeParse(candidates);
+    if (!validated.success || validated.data.length === 0) continue;
+    return validated.data.map((cue) => ({
+      confidence: 1,
+      endSeconds: cue.endSeconds,
+      source: "embedded" as const,
+      startSeconds: cue.startSeconds,
+      text: cue.text,
+    }));
+  }
+  return null;
+}
+
+// ─── Outcome contract ───────────────────────────────────────────────────────
+
+export type VideoSubtitleProbeOutcome =
+  | { cacheable: true; cues: VideoTranscriptCue[]; outcome: "success" }
+  | { cacheable: true; outcome: "absent" }
+  | { cacheable: false; outcome: "transient_failure"; reason: string };
+
+export interface VideoSubtitleProbeOptions {
+  /** Remaining budget of the overall request; the internal 10s subdeadline is clamped to it. */
+  requestDeadlineRemainingMs: number;
+  signal?: AbortSignal;
+}
+
+export interface VideoSubtitleProbeDependencies {
+  probeBroker?: (
+    bytes: Uint8Array,
+    options: BrokerSubtitleExtractionOptions
+  ) => Promise<VideoSubtitleBrokerResponse>;
+}
+
+function transientFailure(reason: string): VideoSubtitleProbeOutcome {
+  return { cacheable: false, outcome: "transient_failure", reason };
+}
+
+/**
+ * Probes a video's embedded subtitle streams through the loopback-only broker and returns an
+ * explicit success/absent/transient_failure outcome. Resolves for every expected failure mode;
+ * only a genuine `options.signal` abort propagates as a rejection.
+ */
+export async function probeEmbeddedVideoSubtitles(
+  bytes: Uint8Array,
+  options: VideoSubtitleProbeOptions,
+  dependencies: VideoSubtitleProbeDependencies = {}
+): Promise<VideoSubtitleProbeOutcome> {
+  if (options.signal?.aborted) throw new Error("Video subtitle probe request aborted");
+  const subdeadlineMs = resolveVideoSubtitleSubdeadlineMs(options.requestDeadlineRemainingMs);
+  if (subdeadlineMs <= 0) {
+    return transientFailure("Video subtitle probe deadline already exceeded");
+  }
+  const probeBroker = dependencies.probeBroker ?? extractVideoSubtitlesViaBroker;
+  let response: VideoSubtitleBrokerResponse;
+  try {
+    response = await probeBroker(bytes, { signal: options.signal, timeoutMs: subdeadlineMs });
+  } catch (error) {
+    if (options.signal?.aborted) throw error;
+    return transientFailure(
+      error instanceof Error ? error.message : "Video subtitle probe failed"
+    );
+  }
+  if (!verifyVideoBridgeBrokerFingerprint(response.fingerprint)) {
+    return transientFailure("Video subtitle broker response fingerprint mismatch");
+  }
+  if (response.streams.length === 0) {
+    return { cacheable: true, outcome: "absent" };
+  }
+  const cues = selectCuesFromStreams(response.streams, response.durationSeconds);
+  if (!cues) return { cacheable: true, outcome: "absent" };
+  return { cacheable: true, cues, outcome: "success" };
+}

src/lib/guardrails/videoBridgeSubtitleRuntime.ts

@@ -0,0 +1,224 @@
+/**
+ * Loopback-only, server-owned subtitle extraction runtime for the Video Bridge broker
+ * (#11659, FU-05). Mirrors the temp-file/ffprobe/ffmpeg lifecycle already established by
+ * `videoBridgeRuntime.ts::extractVideoFramesFromBytes` (container safety probe, mkdtemp,
+ * bounded read, guaranteed cleanup) but scoped to allowlisted embedded subtitle streams.
+ *
+ * This module never trusts caller-declared provenance: it only reports what the broker
+ * itself extracted from the real container via ffprobe/ffmpeg. The HTTP-facing broker route
+ * (`src/app/api/modality-bridge/video/extract/route.ts`) is the only caller and stamps the
+ * response with the shared broker fingerprint so the client-side adapter
+ * (`videoBridgeSubtitleProbe.ts`) can refuse anything that didn't come from this process.
+ */
+import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import {
+  assertLocalPath,
+  defaultRunner,
+  probeLocalVideo,
+  type VideoCommandRunner,
+} from "./videoBridgeRuntime";
+
+/** mov_text (MP4), subrip (SRT/MKV) and webvtt are the only codecs this probe will touch. */
+export const VIDEO_SUBTITLE_CODEC_ALLOWLIST = ["mov_text", "subrip", "webvtt"] as const;
+export type VideoSubtitleCodec = (typeof VIDEO_SUBTITLE_CODEC_ALLOWLIST)[number];
+const SUBTITLE_CODEC_SET: ReadonlySet<string> = new Set(VIDEO_SUBTITLE_CODEC_ALLOWLIST);
+
+export const VIDEO_SUBTITLE_MAX_STREAMS = 2;
+export const VIDEO_SUBTITLE_MAX_OUTPUT_BYTES = 256 * 1024;
+/** Subtitle extraction never runs longer than this, regardless of the request deadline. */
+export const VIDEO_SUBTITLE_SUBDEADLINE_MS = 10_000;
+
+export interface VideoSubtitleStreamProbe {
+  codecName: VideoSubtitleCodec;
+  streamIndex: number;
+}
+
+export interface VideoSubtitleCommandOptions {
+  runner?: VideoCommandRunner;
+  signal?: AbortSignal;
+  timeoutMs: number;
+}
+
+interface FfprobeStreamEntry {
+  codec_name?: unknown;
+  codec_type?: unknown;
+  index?: unknown;
+}
+
+/** Lists only the allowlisted embedded subtitle streams, ordered by container index. */
+export async function probeSubtitleStreams(
+  inputPath: string,
+  options: VideoSubtitleCommandOptions
+): Promise<VideoSubtitleStreamProbe[]> {
+  assertLocalPath(inputPath);
+  const runner = options.runner ?? defaultRunner;
+  const result = await runner(
+    "ffprobe",
+    [
+      "-v",
+      "error",
+      "-protocol_whitelist",
+      "file",
+      "-threads",
+      "1",
+      "-show_entries",
+      "stream=index,codec_type,codec_name",
+      "-of",
+      "json",
+      inputPath,
+    ],
+    { signal: options.signal, timeoutMs: options.timeoutMs }
+  );
+  let streams: FfprobeStreamEntry[] = [];
+  try {
+    const parsed = JSON.parse(result.stdout) as { streams?: FfprobeStreamEntry[] };
+    streams = Array.isArray(parsed.streams) ? parsed.streams : [];
+  } catch {
+    return [];
+  }
+  const probes: VideoSubtitleStreamProbe[] = [];
+  for (const stream of streams) {
+    if (stream.codec_type !== "subtitle") continue;
+    const codecName = typeof stream.codec_name === "string" ? stream.codec_name : "";
+    if (!SUBTITLE_CODEC_SET.has(codecName)) continue;
+    const streamIndex = Number(stream.index);
+    if (!Number.isInteger(streamIndex) || streamIndex < 0) continue;
+    probes.push({ codecName: codecName as VideoSubtitleCodec, streamIndex });
+  }
+  return probes
+    .sort((left, right) => left.streamIndex - right.streamIndex)
+    .slice(0, VIDEO_SUBTITLE_MAX_STREAMS);
+}
+
+/** Remuxes a single subtitle stream to WebVTT text on disk. Never touches other streams. */
+export async function extractSubtitleStreamToFile(
+  inputPath: string,
+  outputPath: string,
+  streamIndex: number,
+  options: VideoSubtitleCommandOptions
+): Promise<void> {
+  assertLocalPath(inputPath);
+  assertLocalPath(outputPath);
+  if (!Number.isInteger(streamIndex) || streamIndex < 0) {
+    throw new Error("Video subtitle stream index is invalid");
+  }
+  const runner = options.runner ?? defaultRunner;
+  await runner(
+    "ffmpeg",
+    [
+      "-nostdin",
+      "-hide_banner",
+      "-loglevel",
+      "error",
+      "-protocol_whitelist",
+      "file",
+      "-threads",
+      "1",
+      "-i",
+      inputPath,
+      "-map",
+      `0:${streamIndex}`,
+      "-c:s",
+      "webvtt",
+      "-f",
+      "webvtt",
+      "-y",
+      outputPath,
+    ],
+    { signal: options.signal, timeoutMs: options.timeoutMs }
+  );
+}
+
+/** Bounded read: rejects anything empty, oversized, or that changed mid-read. */
+export async function readBoundedSubtitleOutput(
+  path: string,
+  maxBytes: number = VIDEO_SUBTITLE_MAX_OUTPUT_BYTES
+): Promise<string> {
+  assertLocalPath(path);
+  const metadata = await stat(path);
+  if (!metadata.isFile() || metadata.size < 1) {
+    throw new Error("Video subtitle extraction produced no output");
+  }
+  if (metadata.size > maxBytes) {
+    throw new Error("Video subtitle extraction output exceeded its byte limit");
+  }
+  const bytes = await readFile(path);
+  if (bytes.byteLength !== metadata.size) {
+    throw new Error("Video subtitle output changed before it could be read");
+  }
+  return bytes.toString("utf8");
+}
+
+export interface VideoSubtitleExtractionResult {
+  durationSeconds: number;
+  formatName: string;
+  streams: Array<{ codecName: VideoSubtitleCodec; streamIndex: number; webvtt: string }>;
+}
+
+/**
+ * Full bytes -> bounded-WebVTT-text pipeline for the broker route. Attempts at most
+ * `VIDEO_SUBTITLE_MAX_STREAMS` allowlisted streams within the caller-supplied subdeadline
+ * (`options.timeoutMs`, already clamped by the route to <= 10s and to the request deadline).
+ * One bad stream (timeout, oversized, empty) never prevents trying the next candidate.
+ * The temp directory is always removed, including on abort/timeout/probe failure.
+ */
+export async function extractVideoSubtitlesFromBytes(
+  bytes: Uint8Array,
+  options: {
+    maxDurationSeconds: number;
+    runner?: VideoCommandRunner;
+    signal?: AbortSignal;
+    timeoutMs: number;
+  }
+): Promise<VideoSubtitleExtractionResult> {
+  const temporaryDirectory = await mkdtemp(join(tmpdir(), "omniroute-video-subtitle-"));
+  try {
+    if (options.signal?.aborted) throw new Error("Video subtitle extraction request aborted");
+    const inputPath = join(temporaryDirectory, "input.video");
+    await writeFile(inputPath, bytes, { mode: 0o600 });
+    const probeTimeoutMs = Math.min(options.timeoutMs, 5_000);
+    const containerMetadata = await probeLocalVideo(inputPath, {
+      maxDurationSeconds: options.maxDurationSeconds,
+      runner: options.runner,
+      signal: options.signal,
+      timeoutMs: probeTimeoutMs,
+    });
+    const streamProbes = await probeSubtitleStreams(inputPath, {
+      runner: options.runner,
+      signal: options.signal,
+      timeoutMs: probeTimeoutMs,
+    });
+    const streams: VideoSubtitleExtractionResult["streams"] = [];
+    const deadlineAt = Date.now() + options.timeoutMs;
+    for (const probe of streamProbes) {
+      if (options.signal?.aborted) throw new Error("Video subtitle extraction request aborted");
+      const remainingMs = deadlineAt - Date.now();
+      if (remainingMs <= 0) break;
+      const outputPath = join(temporaryDirectory, `subtitle-${probe.streamIndex}.vtt`);
+      try {
+        await extractSubtitleStreamToFile(inputPath, outputPath, probe.streamIndex, {
+          runner: options.runner,
+          signal: options.signal,
+          timeoutMs: remainingMs,
+        });
+        const webvtt = await readBoundedSubtitleOutput(outputPath);
+        streams.push({ codecName: probe.codecName, streamIndex: probe.streamIndex, webvtt });
+      } catch (error) {
+        if (options.signal?.aborted) throw new Error("Video subtitle extraction request aborted");
+        if (error instanceof Error && error.message.includes("aborted")) throw error;
+        // A single unusable stream (timeout, empty, oversized) must not block the next candidate.
+        continue;
+      }
+    }
+    return {
+      durationSeconds: containerMetadata.durationSeconds,
+      formatName: containerMetadata.formatName,
+      streams,
+    };
+  } finally {
+    await rm(temporaryDirectory, { force: true, recursive: true });
+  }
+}

tests/unit/guardrails/videoBridgeSubtitleProbe.test.ts

@@ -0,0 +1,317 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  parseBoundedWebVtt,
+  probeEmbeddedVideoSubtitles,
+  resolveVideoSubtitleSubdeadlineMs,
+  VIDEO_SUBTITLE_SUBDEADLINE_MS,
+} from "../../../src/lib/guardrails/videoBridgeSubtitleProbe.ts";
+import { currentVideoBridgeBrokerFingerprint } from "../../../src/lib/guardrails/videoBridgeBrokerAuth.ts";
+import type { VideoSubtitleBrokerResponse } from "../../../src/lib/guardrails/videoBridgeBrokerClient.ts";
+
+const VALID_WEBVTT = "WEBVTT\n\n00:00:01.000 --> 00:00:04.000\nHello world\n";
+
+function brokerResponse(
+  overrides: Partial<VideoSubtitleBrokerResponse> = {}
+): VideoSubtitleBrokerResponse {
+  return {
+    durationSeconds: 10,
+    fingerprint: currentVideoBridgeBrokerFingerprint(),
+    formatName: "mp4",
+    streams: [{ codecName: "webvtt", streamIndex: 3, webvtt: VALID_WEBVTT }],
+    ...overrides,
+  };
+}
+
+// ─── resolveVideoSubtitleSubdeadlineMs ─────────────────────────────────────
+
+test("the 10s subtitle subdeadline is clamped by the remaining request deadline", () => {
+  assert.equal(resolveVideoSubtitleSubdeadlineMs(60_000), VIDEO_SUBTITLE_SUBDEADLINE_MS);
+  assert.equal(resolveVideoSubtitleSubdeadlineMs(2_000), 2_000);
+  assert.equal(resolveVideoSubtitleSubdeadlineMs(-5), 0);
+  assert.equal(resolveVideoSubtitleSubdeadlineMs(Number.NaN), VIDEO_SUBTITLE_SUBDEADLINE_MS);
+});
+
+// ─── parseBoundedWebVtt ─────────────────────────────────────────────────────
+
+test("parses a well-formed WebVTT cue, stripping tags and joining wrapped lines", () => {
+  const cues = parseBoundedWebVtt(
+    "WEBVTT\n\n1\n00:00:01.500 --> 00:00:03.250\n<b>Hello</b>\nworld\n",
+    10
+  );
+  assert.deepEqual(cues, [{ endSeconds: 3.25, startSeconds: 1.5, text: "Hello world" }]);
+});
+
+test("returns no cues for text missing the WEBVTT magic header", () => {
+  assert.deepEqual(parseBoundedWebVtt("00:00:01.000 --> 00:00:02.000\nnot really vtt\n", 10), []);
+});
+
+test("drops a cue whose end does not exceed its start, or whose start is beyond the duration", () => {
+  const backwards = parseBoundedWebVtt("WEBVTT\n\n00:00:05.000 --> 00:00:02.000\ntext\n", 10);
+  assert.deepEqual(backwards, []);
+  const beyondDuration = parseBoundedWebVtt("WEBVTT\n\n00:00:20.000 --> 00:00:22.000\ntext\n", 10);
+  assert.deepEqual(beyondDuration, []);
+});
+
+test("drops a cue containing the UTF-8 replacement character (invalid encoding)", () => {
+  const cues = parseBoundedWebVtt(
+    "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\ncorrupted <20> text\n",
+    10
+  );
+  assert.deepEqual(cues, []);
+});
+
+test("drops a cue with an individual line beyond the 4096 code-unit bound", () => {
+  const oversizedLine = "a".repeat(4_097);
+  const cues = parseBoundedWebVtt(`WEBVTT\n\n00:00:01.000 --> 00:00:02.000\n${oversizedLine}\n`, 10);
+  assert.deepEqual(cues, []);
+});
+
+test("multiple cues come back sorted by start time regardless of source order", () => {
+  const cues = parseBoundedWebVtt(
+    "WEBVTT\n\n00:00:05.000 --> 00:00:06.000\nsecond\n\n00:00:01.000 --> 00:00:02.000\nfirst\n",
+    10
+  );
+  assert.deepEqual(
+    cues.map((cue) => cue.text),
+    ["first", "second"]
+  );
+});
+
+test("a cue identifier line before the timing line is tolerated", () => {
+  const cues = parseBoundedWebVtt("WEBVTT\n\ncue-1\n00:00:01.000 --> 00:00:02.000\ntext\n", 10);
+  assert.equal(cues.length, 1);
+});
+
+// ─── probeEmbeddedVideoSubtitles: outcome contract ─────────────────────────
+
+test("valid: a single verified stream with usable cues yields a cacheable success outcome", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse() }
+  );
+  assert.equal(outcome.outcome, "success");
+  assert.equal(outcome.cacheable, true);
+  if (outcome.outcome === "success") {
+    assert.deepEqual(outcome.cues, [
+      { confidence: 1, endSeconds: 4, source: "embedded", startSeconds: 1, text: "Hello world" },
+    ]);
+  }
+});
+
+test("absent: the broker finds zero allowlisted subtitle streams", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse({ streams: [] }) }
+  );
+  assert.deepEqual(outcome, { cacheable: true, outcome: "absent" });
+});
+
+test("absent: a structurally read stream that yields zero usable cues after parsing", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    {
+      probeBroker: async () =>
+        brokerResponse({
+          streams: [{ codecName: "subrip", streamIndex: 2, webvtt: "WEBVTT\n\n" }],
+        }),
+    }
+  );
+  assert.deepEqual(outcome, { cacheable: true, outcome: "absent" });
+});
+
+test("malformed: a single stream whose WebVTT never yields a valid cue is not cacheable success/absent confusion", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    {
+      probeBroker: async () =>
+        brokerResponse({
+          streams: [{ codecName: "webvtt", streamIndex: 1, webvtt: "not webvtt at all" }],
+        }),
+    }
+  );
+  // Structurally garbage content is treated the same as "no usable subtitles" (stable fact,
+  // safe to cache) rather than a transient/infra failure — it will not resolve on retry.
+  assert.deepEqual(outcome, { cacheable: true, outcome: "absent" });
+});
+
+test("oversized: a broker response exceeding the client-side WebVTT length bound is a transient failure", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    {
+      probeBroker: async () => {
+        throw new Error("Video extraction broker returned invalid subtitle metadata");
+      },
+    }
+  );
+  assert.equal(outcome.outcome, "transient_failure");
+  assert.equal(outcome.cacheable, false);
+});
+
+test("invalid encoding: a corrupted cue is dropped but a clean cue in the same stream still succeeds", async () => {
+  const webvtt =
+    "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\ncorrupted <20> text\n\n00:00:03.000 --> 00:00:04.000\nclean text\n";
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse({ streams: [{ codecName: "webvtt", streamIndex: 1, webvtt }] }) }
+  );
+  assert.equal(outcome.outcome, "success");
+  if (outcome.outcome === "success") {
+    assert.equal(outcome.cues.length, 1);
+    assert.equal(outcome.cues[0].text, "clean text");
+  }
+});
+
+test("timeout: the broker never resolves within the (shrunk) subdeadline, resolving to transient_failure", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 10 },
+    {
+      // Simulates what extractVideoSubtitlesViaBroker does internally with
+      // AbortSignal.timeout(options.timeoutMs): the real transport rejects once its own
+      // internal timer elapses, independent of whether the caller's own signal ever fires.
+      probeBroker: (_bytes, options) =>
+        new Promise((_resolve, reject) => {
+          const timer = setTimeout(
+            () => reject(new Error("Video extraction request aborted")),
+            options.timeoutMs
+          );
+          options.signal?.addEventListener("abort", () => {
+            clearTimeout(timer);
+            reject(new Error("Video extraction request aborted"));
+          });
+        }),
+    }
+  );
+  assert.equal(outcome.outcome, "transient_failure");
+  assert.equal(outcome.cacheable, false);
+});
+
+test("abort: a caller-driven abort propagates as a rejection instead of a resolved outcome", async () => {
+  const controller = new AbortController();
+  const pending = probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000, signal: controller.signal },
+    {
+      probeBroker: (_bytes, options) =>
+        new Promise((_resolve, reject) => {
+          options.signal?.addEventListener("abort", () => reject(new Error("aborted")));
+        }),
+    }
+  );
+  controller.abort();
+  await assert.rejects(() => pending);
+});
+
+test("abort: an already-aborted signal rejects before the broker is ever called", async () => {
+  const controller = new AbortController();
+  controller.abort();
+  let called = false;
+  await assert.rejects(() =>
+    probeEmbeddedVideoSubtitles(
+      Buffer.from("video"),
+      { requestDeadlineRemainingMs: 30_000, signal: controller.signal },
+      {
+        probeBroker: async () => {
+          called = true;
+          return brokerResponse();
+        },
+      }
+    )
+  );
+  assert.equal(called, false);
+});
+
+test("multi-stream selection: the first stream without usable cues is skipped in favor of the second", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    {
+      probeBroker: async () =>
+        brokerResponse({
+          streams: [
+            { codecName: "webvtt", streamIndex: 1, webvtt: "WEBVTT\n\n" },
+            {
+              codecName: "subrip",
+              streamIndex: 4,
+              webvtt: "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nsecond stream wins\n",
+            },
+          ],
+        }),
+    }
+  );
+  assert.equal(outcome.outcome, "success");
+  if (outcome.outcome === "success") {
+    assert.equal(outcome.cues[0].text, "second stream wins");
+  }
+});
+
+test("broker forgery: a response with a mismatched fingerprint never yields embedded provenance", async () => {
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse({ fingerprint: "00000000-0000-0000-0000-000000000000" }) }
+  );
+  assert.equal(outcome.outcome, "transient_failure");
+  assert.equal(outcome.cacheable, false);
+  assert.equal("cues" in outcome, false);
+});
+
+test("broker forgery: a response missing the fingerprint field entirely is rejected the same way", async () => {
+  const forged = brokerResponse() as unknown as Record<string, unknown>;
+  delete forged.fingerprint;
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => forged as unknown as VideoSubtitleBrokerResponse }
+  );
+  assert.equal(outcome.outcome, "transient_failure");
+});
+
+test("cache poisoning guard: every transient_failure carries cacheable:false; success/absent carry true", async () => {
+  const success = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse() }
+  );
+  const absent = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    { probeBroker: async () => brokerResponse({ streams: [] }) }
+  );
+  const transient = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 30_000 },
+    {
+      probeBroker: async () => {
+        throw new Error("broker unavailable");
+      },
+    }
+  );
+  assert.equal(success.cacheable, true);
+  assert.equal(absent.cacheable, true);
+  assert.equal(transient.cacheable, false);
+});
+
+test("deadline already exhausted before the broker is called resolves to a transient failure, not a hang", async () => {
+  let called = false;
+  const outcome = await probeEmbeddedVideoSubtitles(
+    Buffer.from("video"),
+    { requestDeadlineRemainingMs: 0 },
+    {
+      probeBroker: async () => {
+        called = true;
+        return brokerResponse();
+      },
+    }
+  );
+  assert.equal(outcome.outcome, "transient_failure");
+  assert.equal(called, false);
+});

tests/unit/guardrails/videoBridgeSubtitleRuntime.test.ts

@@ -0,0 +1,196 @@
+import assert from "node:assert/strict";
+import { access, writeFile } from "node:fs/promises";
+import test from "node:test";
+
+import {
+  extractSubtitleStreamToFile,
+  extractVideoSubtitlesFromBytes,
+  probeSubtitleStreams,
+  readBoundedSubtitleOutput,
+  VIDEO_SUBTITLE_MAX_OUTPUT_BYTES,
+} from "../../../src/lib/guardrails/videoBridgeSubtitleRuntime.ts";
+import { type VideoCommandRunner } from "../../../src/lib/guardrails/videoBridgeRuntime.ts";
+
+const CONTAINER_PROBE_STDOUT = JSON.stringify({
+  format: { duration: "12.0", format_name: "mov,mp4,m4a,3gp,3g2,mj2" },
+  streams: [{ index: 0, codec_type: "video", width: 640, height: 360 }],
+});
+
+function isSubtitleStreamProbe(args: readonly string[]): boolean {
+  return args.some((arg) => arg.includes("codec_name"));
+}
+
+test("lists only allowlisted subtitle streams, ordered by index and capped at two", async () => {
+  const runner: VideoCommandRunner = async (executable, args) => {
+    assert.equal(executable, "ffprobe");
+    assert.ok(isSubtitleStreamProbe(args));
+    return {
+      stdout: JSON.stringify({
+        streams: [
+          { index: 4, codec_type: "subtitle", codec_name: "ass" },
+          { index: 3, codec_type: "subtitle", codec_name: "webvtt" },
+          { index: 1, codec_type: "subtitle", codec_name: "subrip" },
+          { index: 2, codec_type: "subtitle", codec_name: "mov_text" },
+          { index: 0, codec_type: "audio", codec_name: "aac" },
+        ],
+      }),
+      stderr: "",
+    };
+  };
+
+  const probes = await probeSubtitleStreams("/tmp/input.video", { runner, timeoutMs: 5_000 });
+
+  assert.deepEqual(probes, [
+    { codecName: "subrip", streamIndex: 1 },
+    { codecName: "mov_text", streamIndex: 2 },
+  ]);
+});
+
+test("an unparseable ffprobe response yields zero subtitle streams instead of throwing", async () => {
+  const runner: VideoCommandRunner = async () => ({ stdout: "not json", stderr: "" });
+  const probes = await probeSubtitleStreams("/tmp/input.video", { runner, timeoutMs: 5_000 });
+  assert.deepEqual(probes, []);
+});
+
+test("extracts a single allowlisted stream to bounded WebVTT text", async () => {
+  const runner: VideoCommandRunner = async (executable, args) => {
+    if (executable === "ffmpeg") {
+      await writeFile(args.at(-1) ?? "", "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nhi\n");
+    }
+    return { stdout: "", stderr: "" };
+  };
+  const outputPath = "/tmp/omniroute-subtitle-runtime-test.vtt";
+  await extractSubtitleStreamToFile("/tmp/input.video", outputPath, 3, {
+    runner,
+    timeoutMs: 5_000,
+  });
+  const text = await readBoundedSubtitleOutput(outputPath);
+  assert.match(text, /WEBVTT/);
+});
+
+test("readBoundedSubtitleOutput rejects output above the byte cap", async () => {
+  const outputPath = "/tmp/omniroute-subtitle-runtime-oversized.vtt";
+  await writeFile(outputPath, "x".repeat(VIDEO_SUBTITLE_MAX_OUTPUT_BYTES + 1));
+  await assert.rejects(() => readBoundedSubtitleOutput(outputPath), /byte limit/);
+});
+
+test("readBoundedSubtitleOutput rejects an empty extraction result", async () => {
+  const outputPath = "/tmp/omniroute-subtitle-runtime-empty.vtt";
+  await writeFile(outputPath, "");
+  await assert.rejects(() => readBoundedSubtitleOutput(outputPath), /no output/);
+});
+
+test("full bytes-to-WebVTT pipeline probes the container once, extracts up to two streams, and always cleans up", async () => {
+  const calls: string[] = [];
+  let capturedInputPath = "";
+  const runner: VideoCommandRunner = async (executable, args) => {
+    calls.push(executable);
+    if (capturedInputPath === "") capturedInputPath = String(args.at(-1));
+    if (executable === "ffprobe") {
+      if (isSubtitleStreamProbe(args)) {
+        return {
+          stdout: JSON.stringify({
+            streams: [
+              { index: 2, codec_type: "subtitle", codec_name: "webvtt" },
+              { index: 3, codec_type: "subtitle", codec_name: "subrip" },
+            ],
+          }),
+          stderr: "",
+        };
+      }
+      return { stdout: CONTAINER_PROBE_STDOUT, stderr: "" };
+    }
+    await writeFile(args.at(-1) ?? "", "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nhello\n");
+    return { stdout: "", stderr: "" };
+  };
+
+  const result = await extractVideoSubtitlesFromBytes(Buffer.from("fake-video-bytes"), {
+    maxDurationSeconds: 600,
+    runner,
+    timeoutMs: 5_000,
+  });
+
+  assert.equal(result.durationSeconds, 12);
+  assert.equal(result.streams.length, 2);
+  assert.deepEqual(
+    result.streams.map((stream) => stream.streamIndex),
+    [2, 3]
+  );
+  assert.equal(calls.filter((name) => name === "ffmpeg").length, 2);
+
+  assert.ok(capturedInputPath.includes("omniroute-video-subtitle-"));
+  const temporaryDirectory = capturedInputPath.slice(0, capturedInputPath.lastIndexOf("/"));
+  await assert.rejects(() => access(temporaryDirectory));
+});
+
+test("one unusable stream (extraction throws) does not block trying the next allowlisted stream", async () => {
+  let ffmpegCalls = 0;
+  const runner: VideoCommandRunner = async (executable, args) => {
+    if (executable === "ffprobe") {
+      if (isSubtitleStreamProbe(args)) {
+        return {
+          stdout: JSON.stringify({
+            streams: [
+              { index: 2, codec_type: "subtitle", codec_name: "webvtt" },
+              { index: 5, codec_type: "subtitle", codec_name: "mov_text" },
+            ],
+          }),
+          stderr: "",
+        };
+      }
+      return { stdout: CONTAINER_PROBE_STDOUT, stderr: "" };
+    }
+    ffmpegCalls += 1;
+    if (ffmpegCalls === 1) throw new Error("simulated ffmpeg failure");
+    await writeFile(args.at(-1) ?? "", "WEBVTT\n\n00:00:01.000 --> 00:00:02.000\nok\n");
+    return { stdout: "", stderr: "" };
+  };
+
+  const result = await extractVideoSubtitlesFromBytes(Buffer.from("fake-video-bytes"), {
+    maxDurationSeconds: 600,
+    runner,
+    timeoutMs: 5_000,
+  });
+
+  assert.equal(result.streams.length, 1);
+  assert.equal(result.streams[0].streamIndex, 5);
+});
+
+test("temp directory is removed even when the container probe itself fails", async () => {
+  let capturedInputPath = "";
+  const runner: VideoCommandRunner = async (_executable, args) => {
+    capturedInputPath = String(args.at(-1));
+    throw new Error("ffprobe unavailable");
+  };
+  await assert.rejects(() =>
+    extractVideoSubtitlesFromBytes(Buffer.from("fake-video-bytes"), {
+      maxDurationSeconds: 600,
+      runner,
+      timeoutMs: 5_000,
+    })
+  );
+  assert.ok(capturedInputPath.includes("omniroute-video-subtitle-"));
+  const temporaryDirectory = capturedInputPath.slice(0, capturedInputPath.lastIndexOf("/"));
+  await assert.rejects(() => access(temporaryDirectory));
+});
+
+test("an already-aborted signal is rejected before any command runs", async () => {
+  const controller = new AbortController();
+  controller.abort();
+  let ran = false;
+  const runner: VideoCommandRunner = async () => {
+    ran = true;
+    return { stdout: "", stderr: "" };
+  };
+  await assert.rejects(
+    () =>
+      extractVideoSubtitlesFromBytes(Buffer.from("fake-video-bytes"), {
+        maxDurationSeconds: 600,
+        runner,
+        signal: controller.signal,
+        timeoutMs: 5_000,
+      }),
+    /aborted/
+  );
+  assert.equal(ran, false);
+});

tests/unit/video-bridge-subtitle-broker.test.ts

@@ -0,0 +1,175 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import {
+  handleVideoExtractionBrokerRequest,
+} from "../../src/app/api/modality-bridge/video/extract/route.ts";
+import {
+  buildVideoBridgeBrokerHeaders,
+  currentVideoBridgeBrokerFingerprint,
+} from "../../src/lib/guardrails/videoBridgeBrokerAuth.ts";
+import { createVideoExtractionQueue } from "../../src/lib/guardrails/videoBridgeBrokerQueue.ts";
+import { extractVideoSubtitlesViaBroker } from "../../src/lib/guardrails/videoBridgeBrokerClient.ts";
+import { AUTHZ_HEADER_PEER_LOCALITY } from "../../src/server/authz/headers.ts";
+
+const EXTRACT_PATH = "/api/modality-bridge/video/extract";
+
+function trustedSubtitleRequest(signal?: AbortSignal): Request {
+  return new Request(`http://localhost${EXTRACT_PATH}?subtitles=1`, {
+    method: "POST",
+    headers: {
+      ...buildVideoBridgeBrokerHeaders(),
+      [AUTHZ_HEADER_PEER_LOCALITY]: "loopback",
+      "Content-Type": "application/octet-stream",
+    },
+    body: Buffer.from("video"),
+    signal,
+  });
+}
+
+test("subtitle probe route requires the same loopback broker identity as the frame path", async () => {
+  const response = await handleVideoExtractionBrokerRequest(
+    new Request(`http://localhost${EXTRACT_PATH}?subtitles=1`, {
+      method: "POST",
+      headers: { "Content-Type": "application/octet-stream" },
+      body: Buffer.from("video"),
+    })
+  );
+  assert.equal(response.status, 403);
+});
+
+test("subtitle probe route rejects any extra query parameter", async () => {
+  const response = await handleVideoExtractionBrokerRequest(
+    new Request(`http://localhost${EXTRACT_PATH}?subtitles=1&frames=2`, {
+      method: "POST",
+      headers: {
+        ...buildVideoBridgeBrokerHeaders(),
+        [AUTHZ_HEADER_PEER_LOCALITY]: "loopback",
+        "Content-Type": "application/octet-stream",
+      },
+      body: Buffer.from("video"),
+    })
+  );
+  assert.equal(response.status, 400);
+});
+
+test("subtitle probe route stamps the shared broker fingerprint on a successful extraction", async () => {
+  const response = await handleVideoExtractionBrokerRequest(trustedSubtitleRequest(), {
+    extractSubtitles: async () => ({ durationSeconds: 5, formatName: "mp4", streams: [] }),
+  });
+  assert.equal(response.status, 200);
+  const body = await response.json();
+  assert.equal(body.fingerprint, currentVideoBridgeBrokerFingerprint());
+  assert.deepEqual(body.streams, []);
+});
+
+test("subtitle probe route maps queue capacity and client-abort to the same statuses as the frame path", async () => {
+  const neverExtract = async () => {
+    throw new Error("extractor must not run");
+  };
+
+  const capacity = await handleVideoExtractionBrokerRequest(trustedSubtitleRequest(), {
+    queue: createVideoExtractionQueue({ concurrency: 1, maxPending: 0, maxQueuedBytes: 1 }),
+    extractSubtitles: neverExtract,
+  });
+  assert.equal(capacity.status, 503);
+  assert.equal(capacity.headers.get("Retry-After"), "1");
+
+  const clientController = new AbortController();
+  clientController.abort();
+  const clientAbort = await handleVideoExtractionBrokerRequest(
+    trustedSubtitleRequest(clientController.signal),
+    { extractSubtitles: neverExtract }
+  );
+  assert.equal(clientAbort.status, 499);
+
+  const deadline = await handleVideoExtractionBrokerRequest(trustedSubtitleRequest(), {
+    deadlineSignal: AbortSignal.abort(),
+    extractSubtitles: neverExtract,
+  });
+  assert.equal(deadline.status, 504);
+});
+
+// ─── extractVideoSubtitlesViaBroker: transport + envelope Zod validation ───
+
+test("client broker call sends the subtitles=1 marker and no other query parameters", async () => {
+  let requestedUrl = "";
+  await extractVideoSubtitlesViaBroker(
+    Buffer.from("video"),
+    { timeoutMs: 5_000 },
+    {
+      fetchImpl: async (input) => {
+        requestedUrl = String(input);
+        return Response.json({
+          durationSeconds: 4,
+          fingerprint: currentVideoBridgeBrokerFingerprint(),
+          formatName: "mp4",
+          streams: [],
+        });
+      },
+    }
+  );
+  const parsed = new URL(requestedUrl);
+  assert.equal(parsed.searchParams.get("subtitles"), "1");
+  assert.equal([...parsed.searchParams.keys()].length, 1);
+});
+
+test("client broker call rejects an envelope with an unsupported codec name", async () => {
+  await assert.rejects(
+    () =>
+      extractVideoSubtitlesViaBroker(
+        Buffer.from("video"),
+        { timeoutMs: 5_000 },
+        {
+          fetchImpl: async () =>
+            Response.json({
+              durationSeconds: 4,
+              fingerprint: currentVideoBridgeBrokerFingerprint(),
+              formatName: "mp4",
+              streams: [{ codecName: "ass", streamIndex: 0, webvtt: "WEBVTT\n\n" }],
+            }),
+        }
+      ),
+    /invalid subtitle metadata/
+  );
+});
+
+test("client broker call rejects a response missing the fingerprint field", async () => {
+  await assert.rejects(
+    () =>
+      extractVideoSubtitlesViaBroker(
+        Buffer.from("video"),
+        { timeoutMs: 5_000 },
+        {
+          fetchImpl: async () =>
+            Response.json({ durationSeconds: 4, formatName: "mp4", streams: [] }),
+        }
+      ),
+    /invalid subtitle metadata/
+  );
+});
+
+test("client broker call rejects more than two streams", async () => {
+  const streams = Array.from({ length: 3 }, (_unused, index) => ({
+    codecName: "webvtt" as const,
+    streamIndex: index,
+    webvtt: "WEBVTT\n\n",
+  }));
+  await assert.rejects(
+    () =>
+      extractVideoSubtitlesViaBroker(
+        Buffer.from("video"),
+        { timeoutMs: 5_000 },
+        {
+          fetchImpl: async () =>
+            Response.json({
+              durationSeconds: 4,
+              fingerprint: currentVideoBridgeBrokerFingerprint(),
+              formatName: "mp4",
+              streams,
+            }),
+        }
+      ),
+    /invalid subtitle metadata/
+  );
+});