From 60dc2421785e8fd2a6af49cdf8f7b5db0c7de01d Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Sat, 29 Aug 2026 19:33:27 -0300 Subject: [PATCH] feat(video): derive embedded subtitle provenance in the protected broker (#11659) (#12011) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- ...deo-bridge-embedded-subtitle-provenance.md | 1 + .../modality-bridge/video/extract/route.ts | 111 ++++++ src/lib/guardrails/videoBridgeBrokerAuth.ts | 17 + src/lib/guardrails/videoBridgeBrokerClient.ts | 88 +++++ src/lib/guardrails/videoBridgeRuntime.ts | 4 +- .../guardrails/videoBridgeSubtitleProbe.ts | 242 +++++++++++++ .../guardrails/videoBridgeSubtitleRuntime.ts | 224 +++++++++++++ .../videoBridgeSubtitleProbe.test.ts | 317 ++++++++++++++++++ .../videoBridgeSubtitleRuntime.test.ts | 196 +++++++++++ .../unit/video-bridge-subtitle-broker.test.ts | 175 ++++++++++ 10 files changed, 1373 insertions(+), 2 deletions(-) create mode 100644 changelog.d/features/11659-video-bridge-embedded-subtitle-provenance.md create mode 100644 src/lib/guardrails/videoBridgeSubtitleProbe.ts create mode 100644 src/lib/guardrails/videoBridgeSubtitleRuntime.ts create mode 100644 tests/unit/guardrails/videoBridgeSubtitleProbe.test.ts create mode 100644 tests/unit/guardrails/videoBridgeSubtitleRuntime.test.ts create mode 100644 tests/unit/video-bridge-subtitle-broker.test.ts diff --git a/changelog.d/features/11659-video-bridge-embedded-subtitle-provenance.md b/changelog.d/features/11659-video-bridge-embedded-subtitle-provenance.md new file mode 100644 index 0000000000..67eba61b77 --- /dev/null +++ b/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). diff --git a/src/app/api/modality-bridge/video/extract/route.ts b/src/app/api/modality-bridge/video/extract/route.ts index 9a9e89396e..f05cf94f8c 100644 --- a/src/app/api/modality-bridge/video/extract/route.ts +++ b/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 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 { + 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 { + 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); diff --git a/src/lib/guardrails/videoBridgeBrokerAuth.ts b/src/lib/guardrails/videoBridgeBrokerAuth.ts index c1096d2dde..5cc3aad19a 100644 --- a/src/lib/guardrails/videoBridgeBrokerAuth.ts +++ b/src/lib/guardrails/videoBridgeBrokerAuth.ts @@ -22,6 +22,23 @@ export function buildVideoBridgeBrokerHeaders(): Record { 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) { diff --git a/src/lib/guardrails/videoBridgeBrokerClient.ts b/src/lib/guardrails/videoBridgeBrokerClient.ts index 95aaf5abc8..5a815d5dfd 100644 --- a/src/lib/guardrails/videoBridgeBrokerClient.ts +++ b/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; + +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 { + 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; +} diff --git a/src/lib/guardrails/videoBridgeRuntime.ts b/src/lib/guardrails/videoBridgeRuntime.ts index 736d240c19..7099acd231 100644 --- a/src/lib/guardrails/videoBridgeRuntime.ts +++ b/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"); } diff --git a/src/lib/guardrails/videoBridgeSubtitleProbe.ts b/src/lib/guardrails/videoBridgeSubtitleProbe.ts new file mode 100644 index 0000000000..4c0ab6b14a --- /dev/null +++ b/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 = "�"; +// 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; +} + +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 { + 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" }; +} diff --git a/src/lib/guardrails/videoBridgeSubtitleRuntime.ts b/src/lib/guardrails/videoBridgeSubtitleRuntime.ts new file mode 100644 index 0000000000..75798f60be --- /dev/null +++ b/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 = 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 { + 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 { + 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 { + 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 { + 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 }); + } +} diff --git a/tests/unit/guardrails/videoBridgeSubtitleProbe.test.ts b/tests/unit/guardrails/videoBridgeSubtitleProbe.test.ts new file mode 100644 index 0000000000..af244b7818 --- /dev/null +++ b/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 { + 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\nHello\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 � 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 � 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; + 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); +}); diff --git a/tests/unit/guardrails/videoBridgeSubtitleRuntime.test.ts b/tests/unit/guardrails/videoBridgeSubtitleRuntime.test.ts new file mode 100644 index 0000000000..3af6bce432 --- /dev/null +++ b/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); +}); diff --git a/tests/unit/video-bridge-subtitle-broker.test.ts b/tests/unit/video-bridge-subtitle-broker.test.ts new file mode 100644 index 0000000000..79e0fa35c2 --- /dev/null +++ b/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/ + ); +});