From 5fcd39bd6fd31e5c1efdbda59ad1729c07496a6d Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Sat, 29 Aug 2026 19:34:57 -0300 Subject: [PATCH] feat(video): orchestrate Audio Bridge STT with one-download budgets (FU-06, #11654) (#12012) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FU-06 (Audio Bridge STT orchestration): one download, two extractions — takes already-downloaded video bytes and extracts bounded mono 16kHz PCM WAV via the loopback broker's new mode=audio operation, sharing the exact same queue/deadline/byte budgets as the frame path. Dual opt-in (operator setting default false + per-request), only reaches the STT call when both are on. Rebased onto the tip after sibling #12011 (subtitle mode) landed first, both touching the same broker route/client — combined additively so frames/audio/subtitles all share the one extractionQueue singleton. Re-validated: 59/59 focused tests pass. --- ...54-video-audio-bridge-stt-orchestration.md | 1 + .../modality-bridge/video/extract/route.ts | 181 ++++++--- src/lib/guardrails/audioBridgeHelpers.ts | 218 +++++++--- .../guardrails/videoBridgeAudioExtraction.ts | 124 ++++++ .../videoBridgeAudioOrchestration.ts | 246 +++++++++++ src/lib/guardrails/videoBridgeBrokerClient.ts | 79 ++++ .../constants/modalityBridgeDefaults.ts | 27 ++ src/shared/validation/settingsSchemas.ts | 4 + .../guardrails/audioBridgeHelpers.test.ts | 97 +++++ .../videoBridgeAudioExtraction.test.ts | 140 +++++++ .../videoBridgeAudioOrchestration.test.ts | 381 ++++++++++++++++++ tests/unit/video-bridge-broker.test.ts | 153 +++++++ tests/unit/video-bridge-settings.test.ts | 20 + 13 files changed, 1565 insertions(+), 106 deletions(-) create mode 100644 changelog.d/features/11654-video-audio-bridge-stt-orchestration.md create mode 100644 src/lib/guardrails/videoBridgeAudioExtraction.ts create mode 100644 src/lib/guardrails/videoBridgeAudioOrchestration.ts create mode 100644 tests/unit/guardrails/videoBridgeAudioExtraction.test.ts create mode 100644 tests/unit/guardrails/videoBridgeAudioOrchestration.test.ts diff --git a/changelog.d/features/11654-video-audio-bridge-stt-orchestration.md b/changelog.d/features/11654-video-audio-bridge-stt-orchestration.md new file mode 100644 index 0000000000..78edac2797 --- /dev/null +++ b/changelog.d/features/11654-video-audio-bridge-stt-orchestration.md @@ -0,0 +1 @@ +- **feat(video):** orchestrate optional Video Bridge audio extraction and Audio Bridge STT behind a dual opt-in (operator setting AND per-request signal) — a new loopback-only broker `mode=audio` operation shares the frame path's exact process queue, deadline, AbortSignal, and byte budgets to extract a bounded mono 16 kHz PCM WAV from the same already-downloaded video, then reuses the existing Audio Bridge transcription boundary; provider segment timing is preserved when available and marked coarse otherwise, and every failure degrades to a visual-only-safe partial instead of throwing (#11654). diff --git a/src/app/api/modality-bridge/video/extract/route.ts b/src/app/api/modality-bridge/video/extract/route.ts index f05cf94f8c..9bd14d6904 100644 --- a/src/app/api/modality-bridge/video/extract/route.ts +++ b/src/app/api/modality-bridge/video/extract/route.ts @@ -9,6 +9,10 @@ import { type VideoExtractionQueue, VideoExtractionQueueError, } from "@/lib/guardrails/videoBridgeBrokerQueue"; +import { + extractVideoAudioFromBytes, + type ExtractedVideoAudio, +} from "@/lib/guardrails/videoBridgeAudioExtraction"; import { extractVideoFramesFromBytes, type VideoFocusBounds, @@ -42,15 +46,35 @@ function invalid(message: string, status = 400, headers?: Record return response; } -/** The extract route serves two mutually exclusive operations behind one authenticated path. */ +/** + * The extract route serves up to three mutually exclusive operations behind one authenticated + * path: subtitle probe (`?subtitles=1`), audio extraction (`?mode=audio`), and frame sampling + * (default / `?mode=frames`). + */ function isSubtitleProbeRequest(url: URL): boolean { return url.searchParams.get("subtitles") === "1"; } +export type VideoBrokerMode = "audio" | "frames"; + +const FRAME_ONLY_QUERY_KEYS = ["frames", "samplingPolicy", "start", "end"]; + +/** `mode` is opt-in and defaults to the pre-existing frame path (back-compat). */ +function parseBrokerMode(url: URL): VideoBrokerMode | null { + const raw = url.searchParams.get("mode"); + if (raw === null) return "frames"; + return raw === "frames" || raw === "audio" ? raw : null; +} + +/** Audio mode shares this route/queue but takes no frame-shaping parameters (v1). */ +function hasFrameOnlyParams(url: URL): boolean { + return FRAME_ONLY_QUERY_KEYS.some((key) => url.searchParams.has(key)); +} + function parseFrameCount(url: URL): number | null { if ( [...url.searchParams.keys()].some( - (key) => !["frames", "samplingPolicy", "start", "end"].includes(key) + (key) => !["mode", ...FRAME_ONLY_QUERY_KEYS].includes(key) ) ) { return null; @@ -126,12 +150,13 @@ export async function readBoundedVideoBrokerBody( interface VideoExtractionBrokerRouteDependencies { deadlineSignal?: AbortSignal; + extractAudio?: typeof extractVideoAudioFromBytes; extractFrames?: typeof extractVideoFramesFromBytes; extractSubtitles?: typeof extractVideoSubtitlesFromBytes; queue?: VideoExtractionQueue; } -/** Shared by both operations: declared-length pre-check, bounded read, actual-size check. */ +/** Shared by all three 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); @@ -161,6 +186,56 @@ async function readValidatedVideoBrokerBytes(request: Request): Promise + extractAudio(bytes, { + maxDurationSeconds: MAX_DURATION_SECONDS, + signal, + timeoutMs: BROKER_TIMEOUT_MS, + }), + signal + ); + return Response.json(serializeAudioResult(result), { + headers: { "Cache-Control": "no-store" }, + }); + } catch (error) { + return mapBrokerExtractionError(error, { + deadline, + inputBytes: bytes.byteLength, + mode, + request, + }); + } + } + const frameCount = parseFrameCount(url); if (!frameCount) return invalid("Video Bridge frame count must be between 1 and 16"); const samplingPolicy = parseSamplingPolicy(url); @@ -252,32 +366,8 @@ export async function handleVideoExtractionBrokerRequest( if (focusWindow === null && (url.searchParams.has("start") || url.searchParams.has("end"))) { return invalid("Video Bridge focus window bounds are invalid"); } - 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); - } + const bytes = await readValidatedVideoBrokerBytes(request); + if (bytes instanceof Response) return bytes; const deadline = dependencies.deadlineSignal ?? AbortSignal.timeout(BROKER_TIMEOUT_MS); const signal = AbortSignal.any([request.signal, deadline]); @@ -299,38 +389,7 @@ export async function handleVideoExtractionBrokerRequest( ); return Response.json(result, { headers: { "Cache-Control": "no-store" } }); } catch (error) { - const unavailable = - error && typeof error === "object" && "code" in error && error.code === "ENOENT"; - 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" - : unavailable - ? "RUNTIME_UNAVAILABLE" - : "EXTRACTION_FAILED", - frameCount, - inputBytes: bytes.byteLength, - }, - "Video Bridge broker extraction 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", - }); - } - if (unavailable) return invalid("Video extraction runtime is unavailable", 503); - return invalid("Video extraction failed", 422); + return mapBrokerExtractionError(error, { deadline, inputBytes: bytes.byteLength, mode, request }); } } diff --git a/src/lib/guardrails/audioBridgeHelpers.ts b/src/lib/guardrails/audioBridgeHelpers.ts index ab4c6f71a3..2d09576508 100644 --- a/src/lib/guardrails/audioBridgeHelpers.ts +++ b/src/lib/guardrails/audioBridgeHelpers.ts @@ -24,6 +24,19 @@ export interface AudioTranscriptionConfig { timeoutMs: number; } +export interface AudioTranscriptionSegment { + confidence?: number; + endSeconds: number; + startSeconds: number; + text: string; +} + +export interface AudioTranscriptionTimedResult { + /** Present only when the provider actually returned per-segment timing. */ + segments?: AudioTranscriptionSegment[]; + text: string; +} + export interface AudioTranscriptionDependencies { fetchImpl?: typeof fetch; fetchRemote?: ( @@ -163,39 +176,94 @@ export async function selectAudioBridgeModel( return null; } -/** Send one audio part through OmniRoute's existing multipart transcription route. */ -export async function callAudioTranscription( +/** Resolve one audio part's raw bytes + best-guess MIME, decoding/fetching its `ref`. */ +async function resolveAudioBytes( part: AudioPart, config: AudioTranscriptionConfig, - deps: AudioTranscriptionDependencies = {} -): Promise { + deps: AudioTranscriptionDependencies, + signal: AbortSignal +): Promise<{ bytes: Buffer; mime?: string }> { + const dataUri = /^data:([^;,]+);base64,(.+)$/is.exec(part.ref); + if (dataUri) { + return { bytes: Buffer.from(dataUri[2], "base64"), mime: dataUri[1].toLowerCase() }; + } + if (/^https?:\/\//i.test(part.ref)) { + const fetchRemote = + deps.fetchRemote ?? + ((url: string, options: { signal: AbortSignal }) => + fetchRemoteImage(url, { + guard: "public-only", + maxBytes: 25 * 1024 * 1024, + pinDns: true, + signal: options.signal, + timeoutMs: config.timeoutMs, + })); + const remote = await fetchRemote(part.ref, { signal }); + return { bytes: remote.buffer, mime: remote.contentType.split(";", 1)[0]?.trim().toLowerCase() }; + } + return { bytes: Buffer.from(part.ref, "base64") }; +} + +/** Build the exact multipart body OmniRoute's own transcription route expects. */ +function buildTranscriptionMultipartBody( + bytes: Buffer, + format: string, + safeMime: string, + model: string, + extraFields: Readonly> + // Buffer (what Buffer.concat actually returns), not the wider + // Buffer: only the former is assignable to fetch's BodyInit under the + // open-sse tsconfig, and this body is passed straight to fetch() below (#11654). +): { body: Buffer; boundary: string } { + const fileName = `audio.${format.replace(/[^a-z0-9]/g, "") || "wav"}`; + const boundary = `----OmniRouteAudioBridge${randomUUID().replace(/-/g, "")}`; + const CRLF = "\r\n"; + const extraFieldParts = Object.entries(extraFields).map(([name, value]) => + Buffer.from( + `--${boundary}${CRLF}` + + `Content-Disposition: form-data; name="${name}"${CRLF}${CRLF}` + + `${value}${CRLF}` + ) + ); + const body = Buffer.concat([ + Buffer.from( + `--${boundary}${CRLF}` + + `Content-Disposition: form-data; name="file"; filename="${fileName}"${CRLF}` + + `Content-Type: ${safeMime}${CRLF}${CRLF}` + ), + bytes, + Buffer.from( + `${CRLF}--${boundary}${CRLF}` + + `Content-Disposition: form-data; name="model"${CRLF}${CRLF}` + + `${model}${CRLF}` + ), + ...extraFieldParts, + Buffer.from(`--${boundary}--${CRLF}`), + ]); + return { body, boundary }; +} + +/** + * Send one audio part through OmniRoute's existing multipart transcription + * route (the one Audio Bridge transcription boundary) and return the parsed + * JSON response. `extraFields` lets callers request provider extras (e.g. + * `response_format=verbose_json`) without duplicating this HTTP client. + */ +async function sendAudioTranscriptionRequest( + part: AudioPart, + config: AudioTranscriptionConfig, + deps: AudioTranscriptionDependencies, + extraFields: Readonly> +): Promise> { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), config.timeoutMs); try { - let bytes: Buffer; - let detectedMime: string | undefined; - const dataUri = /^data:([^;,]+);base64,(.+)$/is.exec(part.ref); - if (dataUri) { - detectedMime = dataUri[1].toLowerCase(); - bytes = Buffer.from(dataUri[2], "base64"); - } else if (/^https?:\/\//i.test(part.ref)) { - const fetchRemote = - deps.fetchRemote ?? - ((url: string, options: { signal: AbortSignal }) => - fetchRemoteImage(url, { - guard: "public-only", - maxBytes: 25 * 1024 * 1024, - pinDns: true, - signal: options.signal, - timeoutMs: config.timeoutMs, - })); - const remote = await fetchRemote(part.ref, { signal: controller.signal }); - bytes = remote.buffer; - detectedMime = remote.contentType.split(";", 1)[0]?.trim().toLowerCase(); - } else { - bytes = Buffer.from(part.ref, "base64"); - } - + const { bytes, mime: detectedMime } = await resolveAudioBytes( + part, + config, + deps, + controller.signal + ); const configuredFormat = part.format?.trim().toLowerCase(); const format = configuredFormat || (detectedMime ? AUDIO_MIME_FORMAT[detectedMime] : undefined) || "wav"; @@ -206,22 +274,13 @@ export async function callAudioTranscription( const safeMime = /^[a-z0-9][a-z0-9.+-]*\/[a-z0-9][a-z0-9.+-]*$/i.test(mime) ? mime : "application/octet-stream"; - const fileName = `audio.${format.replace(/[^a-z0-9]/g, "") || "wav"}`; - const boundary = `----OmniRouteAudioBridge${randomUUID().replace(/-/g, "")}`; - const CRLF = "\r\n"; - const multipartBody = Buffer.concat([ - Buffer.from( - `--${boundary}${CRLF}` + - `Content-Disposition: form-data; name="file"; filename="${fileName}"${CRLF}` + - `Content-Type: ${safeMime}${CRLF}${CRLF}` - ), + const { body: multipartBody, boundary } = buildTranscriptionMultipartBody( bytes, - Buffer.from( - `${CRLF}--${boundary}${CRLF}` + - `Content-Disposition: form-data; name="model"${CRLF}${CRLF}` + - `${config.model}${CRLF}--${boundary}--${CRLF}` - ), - ]); + format, + safeMime, + config.model, + extraFields + ); const port = (deps.getPort ?? (() => getRuntimePorts().port))(); const bearer = (deps.getBearer ?? resolveSelfLoopBearer)(); @@ -241,11 +300,11 @@ export async function callAudioTranscription( if (!response.ok) { throw new Error(`Audio transcription failed (${response.status})`); } - const data = (await response.json()) as { text?: unknown }; + const data = (await response.json()) as Record; if (typeof data.text !== "string") { throw new Error("Audio transcription returned an invalid response"); } - return data.text.trim(); + return data; } catch (error) { if (error instanceof Error && error.name === "AbortError") { throw new Error("Audio transcription timed out"); @@ -255,3 +314,72 @@ export async function callAudioTranscription( clearTimeout(timeout); } } + +/** Send one audio part through OmniRoute's existing multipart transcription route. */ +export async function callAudioTranscription( + part: AudioPart, + config: AudioTranscriptionConfig, + deps: AudioTranscriptionDependencies = {} +): Promise { + const data = await sendAudioTranscriptionRequest(part, config, deps, {}); + return (data.text as string).trim(); +} + +function normalizeTranscriptionSegment(raw: unknown): AudioTranscriptionSegment | null { + if (!raw || typeof raw !== "object") return null; + const segment = raw as Record; + const startSeconds = + typeof segment.start === "number" + ? segment.start + : typeof segment.startSeconds === "number" + ? segment.startSeconds + : Number.NaN; + const endSeconds = + typeof segment.end === "number" + ? segment.end + : typeof segment.endSeconds === "number" + ? segment.endSeconds + : Number.NaN; + const text = typeof segment.text === "string" ? segment.text.trim() : ""; + if ( + !text || + !Number.isFinite(startSeconds) || + !Number.isFinite(endSeconds) || + startSeconds < 0 || + endSeconds <= startSeconds + ) { + return null; + } + const confidence = + typeof segment.confidence === "number" && Number.isFinite(segment.confidence) + ? Math.min(1, Math.max(0, segment.confidence)) + : undefined; + return { + endSeconds, + startSeconds, + text, + ...(confidence === undefined ? {} : { confidence }), + }; +} + +/** + * Same Audio Bridge transcription boundary as {@link callAudioTranscription}, + * requesting `verbose_json` so a provider that supports it can return + * per-segment timing. Callers must treat a result with no `segments` as + * coarse (whole-clip) timing only — the provider did not supply detail. + */ +export async function callAudioTranscriptionTimed( + part: AudioPart, + config: AudioTranscriptionConfig, + deps: AudioTranscriptionDependencies = {} +): Promise { + const data = await sendAudioTranscriptionRequest(part, config, deps, { + response_format: "verbose_json", + }); + const text = (data.text as string).trim(); + const rawSegments = Array.isArray(data.segments) ? data.segments : []; + const segments = rawSegments + .map(normalizeTranscriptionSegment) + .filter((segment): segment is AudioTranscriptionSegment => segment !== null); + return segments.length > 0 ? { segments, text } : { text }; +} diff --git a/src/lib/guardrails/videoBridgeAudioExtraction.ts b/src/lib/guardrails/videoBridgeAudioExtraction.ts new file mode 100644 index 0000000000..264b33c833 --- /dev/null +++ b/src/lib/guardrails/videoBridgeAudioExtraction.ts @@ -0,0 +1,124 @@ +/** + * Video Bridge — bounded mono 16 kHz PCM WAV audio extraction (FU-06, #11654). + * + * Runs strictly local, array-argument `ffmpeg` (never a shell string — Hard + * Rule #13) against a video file already probed by + * {@link probeLocalVideo}, so the safe-container/format/dimension validation + * that path already enforces for frame extraction stays the single source of + * truth. This module owns only the audio-specific extraction step and its + * own output byte cap; it does not duplicate probing. + */ +import { execFile } from "node:child_process"; +import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { promisify } from "node:util"; + +import { probeLocalVideo, type VideoCommandRunner } from "./videoBridgeRuntime"; + +const execFileAsync = promisify(execFile); + +/** Bounded output — a 10-minute mono 16 kHz 16-bit WAV tops out well under this. */ +export const VIDEO_AUDIO_MAX_BYTES = 25 * 1024 * 1024; +export const VIDEO_AUDIO_SAMPLE_RATE_HZ = 16_000; +export const VIDEO_AUDIO_CHANNELS = 1; + +export interface ExtractedVideoAudio { + channels: number; + dataUri: string; + durationSeconds: number; + sampleRateHz: number; +} + +export interface VideoAudioExtractionOptions { + maxDurationSeconds: number; + maxOutputBytes?: number; + runner?: VideoCommandRunner; + signal?: AbortSignal; + timeoutMs: number; +} + +const defaultRunner: VideoCommandRunner = async (executable, args, options) => { + const result = await execFileAsync(executable, [...args], { + encoding: "utf8", + maxBuffer: 1024 * 1024, + signal: options.signal, + timeout: options.timeoutMs, + windowsHide: true, + }); + return { stdout: String(result.stdout), stderr: String(result.stderr) }; +}; + +/** Extract the first audio stream of `bytes` as a bounded mono 16 kHz PCM WAV. */ +export async function extractVideoAudioFromBytes( + bytes: Uint8Array, + options: VideoAudioExtractionOptions +): Promise { + const temporaryDirectory = await mkdtemp(join(tmpdir(), "omniroute-video-audio-broker-")); + try { + if (options.signal?.aborted) throw new Error("Video audio extraction request aborted"); + const runner = options.runner ?? defaultRunner; + const inputPath = join(temporaryDirectory, "input.video"); + const outputPath = join(temporaryDirectory, "audio.wav"); + await writeFile(inputPath, bytes, { mode: 0o600 }); + // Reuses the frame path's container/format/dimension safety checks — this + // module adds no parallel validation of its own. + const metadata = await probeLocalVideo(inputPath, { + maxDurationSeconds: options.maxDurationSeconds, + runner, + signal: options.signal, + timeoutMs: Math.min(options.timeoutMs, 30_000), + }); + await runner( + "ffmpeg", + [ + "-nostdin", + "-hide_banner", + "-loglevel", + "error", + "-protocol_whitelist", + "file", + "-threads", + "1", + "-i", + inputPath, + "-vn", + "-map", + "0:a:0", + "-ac", + String(VIDEO_AUDIO_CHANNELS), + "-ar", + String(VIDEO_AUDIO_SAMPLE_RATE_HZ), + "-sample_fmt", + "s16", + "-t", + metadata.durationSeconds.toFixed(3), + "-f", + "wav", + "-y", + outputPath, + ], + { signal: options.signal, timeoutMs: options.timeoutMs } + ); + const outputStats = await stat(outputPath); + const maxOutputBytes = options.maxOutputBytes ?? VIDEO_AUDIO_MAX_BYTES; + if (!outputStats.isFile() || outputStats.size < 1) { + throw new Error("Video audio extraction produced no output"); + } + if (outputStats.size > maxOutputBytes) { + throw new Error("Extracted video audio byte limit exceeded"); + } + const audioBytes = await readFile(outputPath); + if (audioBytes.byteLength !== outputStats.size) { + throw new Error("Extracted video audio changed before it could be read"); + } + return { + channels: VIDEO_AUDIO_CHANNELS, + dataUri: `data:audio/wav;base64,${audioBytes.toString("base64")}`, + durationSeconds: metadata.durationSeconds, + sampleRateHz: VIDEO_AUDIO_SAMPLE_RATE_HZ, + }; + } finally { + await rm(temporaryDirectory, { force: true, recursive: true }); + } +} diff --git a/src/lib/guardrails/videoBridgeAudioOrchestration.ts b/src/lib/guardrails/videoBridgeAudioOrchestration.ts new file mode 100644 index 0000000000..704ca22bd2 --- /dev/null +++ b/src/lib/guardrails/videoBridgeAudioOrchestration.ts @@ -0,0 +1,246 @@ +/** + * Video Bridge — Audio Bridge STT orchestration (FU-06, #11654). + * + * Composes three already-independent seams behind one dual opt-in gate: + * - the SAME video bytes the frame path already downloaded — this module + * has no download path of its own, so it structurally cannot fetch the + * video a second time; + * - the loopback-only broker's bounded mono 16 kHz WAV operation + * ({@link extractVideoAudioViaBroker}), which shares the frame path's + * process queue, deadline, AbortSignal, and byte budgets because it is + * the exact same broker route (`mode=audio` on the frame route); + * - the existing Audio Bridge transcription boundary + * ({@link callAudioTranscriptionTimed}) — no new STT client. + * + * "No paid transcription without dual opt-in" is enforced by two early + * returns, both before the broker or the transcription boundary is ever + * touched. Every failure path returns `track: null` instead of throwing, so + * a caller can always fall back to a visual-only description. + */ +import { + callAudioTranscriptionTimed, + selectAudioBridgeModel, + type AudioCredentialCheck, + type AudioPart, + type AudioTranscriptionTimedResult, +} from "./audioBridgeHelpers"; +import { bridgeCacheKey, type BridgeCacheStore } from "./modalityBridge/bridgeCache"; +import type { FusionObservation, FusionTrack } from "./videoAudioFusion"; +import { + extractVideoAudioViaBroker, + type BrokerAudioExtractionResult, +} from "./videoBridgeBrokerClient"; + +export type VideoAudioOrchestrationReason = + | "ABORTED" + | "EXTRACTION_FAILED" + | "OPERATOR_OPT_OUT" + | "PROVIDER_UNAVAILABLE" + | "REQUEST_OPT_OUT" + | "TIMEOUT" + | "TRANSCRIPTION_FAILED"; + +export interface VideoAudioOrchestrationOptions { + /** Result cache; omit (or omit `cacheKeyRef`) to skip caching entirely. */ + cache?: BridgeCacheStore | null; + /** Stable identity for the ORIGINAL video part `ref` — never the raw bytes. */ + cacheKeyRef?: string; + extractAudio?: typeof extractVideoAudioViaBroker; + hasUsableCredentials?: AudioCredentialCheck; + /** Configured STT model — `"auto"` or a fixed `provider/model` string. */ + model: string; + /** Operator half of the dual opt-in (settings). */ + operatorOptIn: boolean; + /** Request half of the dual opt-in (a per-request signal from the caller). */ + requestOptIn: boolean; + selectModel?: typeof selectAudioBridgeModel; + signal?: AbortSignal; + /** The SAME budget the sibling frame extraction for this video used. */ + timeoutMs: number; + transcribe?: typeof callAudioTranscriptionTimed; + /** The SAME already-downloaded video bytes used for frame extraction. */ + videoBytes: Uint8Array; +} + +export interface VideoAudioOrchestrationResult { + /** False only for the two opt-out reasons — nothing was attempted at all. */ + attempted: boolean; + reason?: VideoAudioOrchestrationReason; + sttModel: string | null; + /** "coarse" means the provider gave no segment timing (whole-clip span only). */ + timingPrecision?: "coarse" | "exact"; + track: FusionTrack | null; +} + +interface CachedOrchestration { + timingPrecision: "coarse" | "exact"; + track: FusionTrack; +} + +type StepOutcome = { ok: true; value: T } | { ok: false; result: VideoAudioOrchestrationResult }; + +function optedOut(reason: "OPERATOR_OPT_OUT" | "REQUEST_OPT_OUT"): VideoAudioOrchestrationResult { + return { attempted: false, reason, sttModel: null, track: null }; +} + +function failed( + reason: Exclude, + sttModel: string | null +): VideoAudioOrchestrationResult { + return { attempted: true, reason, sttModel, track: null }; +} + +/** Caller-signal cancellation always wins; else a timeout-shaped message is a shared-budget timeout. */ +function classifyStepFailure( + error: unknown, + signal: AbortSignal | undefined, + timeoutPattern: RegExp +): "ABORTED" | "TIMEOUT" | "FAILED" { + if (signal?.aborted) return "ABORTED"; + if (error instanceof Error && timeoutPattern.test(error.message)) return "TIMEOUT"; + return "FAILED"; +} + +function cacheKeyFor(ref: string, sttModel: string): string { + return bridgeCacheKey(ref, "video-audio-transcription", sttModel); +} + +function readCache( + options: VideoAudioOrchestrationOptions, + sttModel: string +): CachedOrchestration | null { + if (!options.cache || !options.cacheKeyRef) return null; + const entry = options.cache.getEntry(cacheKeyFor(options.cacheKeyRef, sttModel)); + if (!entry) return null; + try { + const parsed = JSON.parse(entry.value) as Partial; + if (!parsed.track || !Array.isArray(parsed.track.observations)) return null; + if (parsed.timingPrecision !== "coarse" && parsed.timingPrecision !== "exact") return null; + return { timingPrecision: parsed.timingPrecision, track: parsed.track }; + } catch { + return null; + } +} + +function writeCache( + options: VideoAudioOrchestrationOptions, + sttModel: string, + value: CachedOrchestration +): void { + if (!options.cache || !options.cacheKeyRef) return; + options.cache.setEntry(cacheKeyFor(options.cacheKeyRef, sttModel), { + value: JSON.stringify(value), + }); +} + +async function runExtraction( + options: VideoAudioOrchestrationOptions, + sttModel: string +): Promise> { + const extractAudio = options.extractAudio ?? extractVideoAudioViaBroker; + try { + const value = await extractAudio(options.videoBytes, { + signal: options.signal, + timeoutMs: options.timeoutMs, + }); + return { ok: true, value }; + } catch (error) { + const outcome = classifyStepFailure(error, options.signal, /abort|timed out/i); + return { + ok: false, + result: failed(outcome === "FAILED" ? "EXTRACTION_FAILED" : outcome, sttModel), + }; + } +} + +async function runTranscription( + options: VideoAudioOrchestrationOptions, + sttModel: string, + extraction: BrokerAudioExtractionResult +): Promise> { + const transcribe = options.transcribe ?? callAudioTranscriptionTimed; + const audioPart: AudioPart = { + format: "wav", + messageIndex: -1, + partIndex: -1, + ref: extraction.audio.dataUri, + shape: "input_audio", + }; + try { + const value = await transcribe(audioPart, { model: sttModel, timeoutMs: options.timeoutMs }); + return { ok: true, value }; + } catch (error) { + const outcome = classifyStepFailure(error, options.signal, /timed out/i); + return { + ok: false, + result: failed(outcome === "FAILED" ? "TRANSCRIPTION_FAILED" : outcome, sttModel), + }; + } +} + +/** Server-minted observations only — never trusts a caller-declared source. */ +function buildObservations( + extraction: BrokerAudioExtractionResult, + transcription: AudioTranscriptionTimedResult +): { observations: FusionObservation[]; timingPrecision: "coarse" | "exact" } { + if (transcription.segments && transcription.segments.length > 0) { + return { + observations: transcription.segments.map((segment) => ({ + confidence: segment.confidence ?? 1, + endSeconds: segment.endSeconds, + source: "audio", + startSeconds: segment.startSeconds, + text: segment.text, + })), + timingPrecision: "exact", + }; + } + const text = transcription.text.trim(); + if (!text) return { observations: [], timingPrecision: "coarse" }; + return { + observations: [ + { + confidence: 1, + endSeconds: Math.max(extraction.durationSeconds, 0.001), + source: "audio", + startSeconds: 0, + text, + }, + ], + timingPrecision: "coarse", + }; +} + +/** + * Orchestrate server-side Video Bridge audio extraction + Audio Bridge STT, + * gated on BOTH the operator and the request opt-in. Never throws: every + * outcome — opt-out, no provider, extraction/transcription failure, timeout, + * abort — comes back as `track: null` so the caller's visual-only fallback + * always has a well-defined result to check against. + */ +export async function orchestrateVideoAudioTranscription( + options: VideoAudioOrchestrationOptions +): Promise { + if (!options.operatorOptIn) return optedOut("OPERATOR_OPT_OUT"); + if (!options.requestOptIn) return optedOut("REQUEST_OPT_OUT"); + + const selectModel = options.selectModel ?? selectAudioBridgeModel; + const sttModel = await selectModel(options.model, options.hasUsableCredentials); + if (!sttModel) return failed("PROVIDER_UNAVAILABLE", null); + + const cached = readCache(options, sttModel); + if (cached) { + return { attempted: true, sttModel, timingPrecision: cached.timingPrecision, track: cached.track }; + } + + const extraction = await runExtraction(options, sttModel); + if (!extraction.ok) return extraction.result; + + const transcription = await runTranscription(options, sttModel, extraction.value); + if (!transcription.ok) return transcription.result; + + const { observations, timingPrecision } = buildObservations(extraction.value, transcription.value); + const track: FusionTrack = { observations }; + writeCache(options, sttModel, { timingPrecision, track }); + return { attempted: true, sttModel, timingPrecision, track }; +} diff --git a/src/lib/guardrails/videoBridgeBrokerClient.ts b/src/lib/guardrails/videoBridgeBrokerClient.ts index 5a815d5dfd..8f85a9dfb6 100644 --- a/src/lib/guardrails/videoBridgeBrokerClient.ts +++ b/src/lib/guardrails/videoBridgeBrokerClient.ts @@ -218,6 +218,17 @@ export interface BrokerSubtitleExtractionOptions { timeoutMs: number; } +export interface BrokerAudioExtractionResult { + audio: { channels: number; dataUri: string; sampleRateHz: number }; + durationSeconds: number; +} + +export interface BrokerAudioExtractionOptions { + maxDurationSeconds?: number; + 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 @@ -267,3 +278,71 @@ export async function extractVideoSubtitlesViaBroker( } return parsed.data; } + +function parseBrokerAudioResult(value: unknown): BrokerAudioExtractionResult { + const record = value && typeof value === "object" ? (value as Record) : null; + const durationSeconds = Number(record?.durationSeconds); + const audio = + record?.audio && typeof record.audio === "object" + ? (record.audio as Record) + : null; + const dataUri = typeof audio?.dataUri === "string" ? audio.dataUri : ""; + const sampleRateHz = Number(audio?.sampleRateHz); + const channels = Number(audio?.channels); + if ( + !Number.isFinite(durationSeconds) || + durationSeconds <= 0 || + !/^data:audio\/wav;base64,[A-Za-z0-9+/=]+$/.test(dataUri) || + !Number.isInteger(sampleRateHz) || + sampleRateHz <= 0 || + !Number.isInteger(channels) || + channels <= 0 + ) { + throw new Error("Video audio extraction broker returned invalid metadata"); + } + return { audio: { channels, dataUri, sampleRateHz }, durationSeconds }; +} + +/** + * The loopback-only broker's mono 16 kHz PCM WAV extraction operation. Shares + * the exact route, queue, deadline, and byte budgets as + * {@link extractVideoFramesViaBroker} — only the `mode=audio` query flag and + * response shape differ. + */ +export async function extractVideoAudioViaBroker( + bytes: Uint8Array, + options: BrokerAudioExtractionOptions, + dependencies: { fetchImpl?: typeof fetch; maxResponseBytes?: number } = {} +): Promise { + if (options.signal?.aborted) throw new Error("Video audio extraction request aborted"); + const baseUrl = resolveVideoBridgeBrokerBaseUrl(); + const url = new URL(`${baseUrl}${VIDEO_BRIDGE_BROKER_PATH}`); + url.searchParams.set("mode", "audio"); + 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 audio 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 + ); + return parseBrokerAudioResult(await readBoundedResponse(response, maxResponseBytes)); +} diff --git a/src/shared/constants/modalityBridgeDefaults.ts b/src/shared/constants/modalityBridgeDefaults.ts index 3fa8faadad..9deefe0ec0 100644 --- a/src/shared/constants/modalityBridgeDefaults.ts +++ b/src/shared/constants/modalityBridgeDefaults.ts @@ -33,6 +33,12 @@ export const MODALITY_BRIDGE_DEFAULTS = { videoSamplingPolicy: "uniform" as VideoSamplingPolicy, videoMaxVideos: 1, videoTimeoutMs: 120000, + // Server-orchestrated Audio Bridge STT over Video Bridge audio extraction + // (FU-06, #11654) spends provider credit on the operator's behalf, so it + // stays OFF by default — Hard Rule #20's "never spend by default" spirit. + // Every transcription attempt additionally requires a per-request opt-in; + // this flag alone never triggers a call. + videoAudioTranscriptionEnabled: false, } as const; export interface VisionBridgeRuntimeSettings { @@ -59,6 +65,11 @@ export interface AudioBridgeRuntimeSettings { cacheMaxEntries: number; } +export interface VideoAudioTranscriptionRuntimeSettings { + /** Operator opt-in only — a request still needs its own opt-in (FU-06, #11654). */ + enabled: boolean; +} + export interface VideoBridgeRuntimeSettings { enabled: boolean; model: string; @@ -142,6 +153,22 @@ export function resolveAudioBridgeRuntimeSettings( }; } +/** + * Resolve the operator half of the FU-06 dual opt-in for Video Bridge audio + * transcription. The request-side opt-in is a separate, per-request signal — + * this settings flag alone never authorizes a transcription call. + */ +export function resolveVideoAudioTranscriptionRuntimeSettings( + settings: Record | null | undefined +): VideoAudioTranscriptionRuntimeSettings { + const s = settings ?? {}; + return { + enabled: + pickBoolean(s.modalityBridgeVideoAudioTranscriptionEnabled) ?? + MODALITY_BRIDGE_DEFAULTS.videoAudioTranscriptionEnabled, + }; +} + /** Resolve persisted Video Bridge settings with safe, bounded defaults. */ export function resolveVideoBridgeRuntimeSettings( settings: Record | null | undefined diff --git a/src/shared/validation/settingsSchemas.ts b/src/shared/validation/settingsSchemas.ts index 21550c2a8b..0c33cda5c5 100644 --- a/src/shared/validation/settingsSchemas.ts +++ b/src/shared/validation/settingsSchemas.ts @@ -464,6 +464,10 @@ export const updateSettingsSchema = z.object({ .min(VIDEO_BRIDGE_TIMEOUT_MIN_MS) .max(VIDEO_BRIDGE_TIMEOUT_MAX_MS) .optional(), + // Operator half of the FU-06 dual opt-in (#11654) for server-orchestrated + // Audio Bridge STT over Video Bridge audio extraction — defaults false + // (Hard Rule #20). A request-side opt-in is required in addition to this. + modalityBridgeVideoAudioTranscriptionEnabled: z.boolean().optional(), modalityBridgeCacheEnabled: z.boolean().optional(), modalityBridgeCacheTtlMinutes: z.number().int().min(1).max(1440).optional(), modalityBridgeCacheMaxEntries: z.number().int().min(10).max(5000).optional(), diff --git a/tests/unit/guardrails/audioBridgeHelpers.test.ts b/tests/unit/guardrails/audioBridgeHelpers.test.ts index 82fcefdb1d..6eee88e2a7 100644 --- a/tests/unit/guardrails/audioBridgeHelpers.test.ts +++ b/tests/unit/guardrails/audioBridgeHelpers.test.ts @@ -3,6 +3,7 @@ import test from "node:test"; import { callAudioTranscription, + callAudioTranscriptionTimed, extractAudioParts, replaceAudioParts, selectAudioBridgeModel, @@ -217,3 +218,99 @@ test("remote audio_url uses the guarded remote fetch before self-loop upload", a assert.equal(fetchedUrl, "https://media.example.test/clip.ogg"); assertMultipartFile(uploaded, "audio.ogg", "audio/ogg", Buffer.from("OggS remote audio")); }); + +test("timed transcription requests verbose_json and preserves provider-supplied segment timing", async () => { + let uploaded: RequestInit | undefined; + const result = await callAudioTranscriptionTimed( + { + messageIndex: 0, + partIndex: 0, + ref: Buffer.from("RIFF timed audio").toString("base64"), + shape: "input_audio", + format: "wav", + }, + { model: "deepgram/nova-3", timeoutMs: 1_000 }, + { + fetchImpl: async (_input, init) => { + uploaded = init; + return Response.json({ + text: "hello world", + segments: [ + { start: 0, end: 1.2, text: "hello", confidence: 0.95 }, + { start: 1.2, end: 2.4, text: "world" }, + ], + }); + }, + getPort: () => 3210, + getBearer: () => "internal-test-key", + } + ); + + const contentType = new Headers(uploaded?.headers).get("content-type"); + const boundary = contentType?.split("boundary=", 2)[1]; + const body = uploaded?.body as Buffer; + assert.ok( + body.includes( + Buffer.from( + `--${boundary}\r\n` + + 'Content-Disposition: form-data; name="response_format"\r\n\r\n' + + `verbose_json\r\n` + ) + ), + "verbose_json must be requested from the same transcription boundary" + ); + assert.equal(result.text, "hello world"); + assert.deepEqual(result.segments, [ + { startSeconds: 0, endSeconds: 1.2, text: "hello", confidence: 0.95 }, + { startSeconds: 1.2, endSeconds: 2.4, text: "world" }, + ]); +}); + +test("timed transcription reports no segments when the provider ignores verbose_json", async () => { + const result = await callAudioTranscriptionTimed( + { + messageIndex: 0, + partIndex: 0, + ref: Buffer.from("RIFF coarse audio").toString("base64"), + shape: "input_audio", + format: "wav", + }, + { model: "deepgram/nova-3", timeoutMs: 1_000 }, + { + fetchImpl: async () => Response.json({ text: "coarse only" }), + getPort: () => 3210, + getBearer: () => "internal-test-key", + } + ); + + assert.equal(result.text, "coarse only"); + assert.equal(result.segments, undefined); +}); + +test("timed transcription drops malformed provider segments instead of trusting them", async () => { + const result = await callAudioTranscriptionTimed( + { + messageIndex: 0, + partIndex: 0, + ref: Buffer.from("RIFF bad segments").toString("base64"), + shape: "input_audio", + format: "wav", + }, + { model: "deepgram/nova-3", timeoutMs: 1_000 }, + { + fetchImpl: async () => + Response.json({ + text: "text", + segments: [ + { start: 5, end: 2, text: "backwards" }, + { start: 0, end: 1, text: "" }, + { start: "nope", end: 1, text: "not a number" }, + ], + }), + getPort: () => 3210, + getBearer: () => "internal-test-key", + } + ); + + assert.equal(result.segments, undefined); +}); diff --git a/tests/unit/guardrails/videoBridgeAudioExtraction.test.ts b/tests/unit/guardrails/videoBridgeAudioExtraction.test.ts new file mode 100644 index 0000000000..4920b446e7 --- /dev/null +++ b/tests/unit/guardrails/videoBridgeAudioExtraction.test.ts @@ -0,0 +1,140 @@ +import assert from "node:assert/strict"; +import { writeFile } from "node:fs/promises"; +import test from "node:test"; + +import { + VIDEO_AUDIO_CHANNELS, + VIDEO_AUDIO_SAMPLE_RATE_HZ, + extractVideoAudioFromBytes, +} from "../../../src/lib/guardrails/videoBridgeAudioExtraction.ts"; +import type { VideoCommandRunner } from "../../../src/lib/guardrails/videoBridgeRuntime.ts"; + +test("extracts bounded mono 16kHz WAV bytes and reuses the probed duration", async () => { + const calls: Array<{ executable: string; args: string[] }> = []; + const runner: VideoCommandRunner = async (executable, args) => { + calls.push({ executable, args: [...args] }); + if (executable === "ffprobe") { + return { + stdout: JSON.stringify({ + format: { duration: "4.5", format_name: "mp4" }, + streams: [{ index: 0, codec_type: "video", width: 640, height: 360 }], + }), + stderr: "", + }; + } + await writeFile(args.at(-1) ?? "", Buffer.from("RIFF....WAVEfmt ")); + return { stdout: "", stderr: "" }; + }; + + const result = await extractVideoAudioFromBytes(Buffer.from("fake video bytes"), { + maxDurationSeconds: 600, + runner, + timeoutMs: 5_000, + }); + + assert.equal(result.durationSeconds, 4.5); + assert.equal(result.channels, VIDEO_AUDIO_CHANNELS); + assert.equal(result.sampleRateHz, VIDEO_AUDIO_SAMPLE_RATE_HZ); + assert.match(result.dataUri, /^data:audio\/wav;base64,/); + assert.deepEqual( + Buffer.from(result.dataUri.split(",", 2)[1], "base64"), + Buffer.from("RIFF....WAVEfmt ") + ); + + const ffmpegCall = calls.find((call) => call.executable === "ffmpeg"); + assert.ok(ffmpegCall, "ffmpeg must be invoked"); + assert.deepEqual(ffmpegCall!.args.slice(ffmpegCall!.args.indexOf("-map"), ffmpegCall!.args.indexOf("-map") + 2), [ + "-map", + "0:a:0", + ]); + assert.deepEqual(ffmpegCall!.args.slice(ffmpegCall!.args.indexOf("-ac"), ffmpegCall!.args.indexOf("-ac") + 2), [ + "-ac", + "1", + ]); + assert.deepEqual(ffmpegCall!.args.slice(ffmpegCall!.args.indexOf("-ar"), ffmpegCall!.args.indexOf("-ar") + 2), [ + "-ar", + "16000", + ]); + // No shell string ever built — every arg is a discrete array element (Hard Rule #13). + assert.equal( + ffmpegCall!.args.some((arg) => arg.includes(";") || arg.includes("&&")), + false + ); +}); + +test("propagates the shared probe's duration cap instead of re-validating it", async () => { + const runner: VideoCommandRunner = async (executable) => { + if (executable === "ffprobe") { + return { + stdout: JSON.stringify({ + format: { duration: "700", format_name: "mp4" }, + streams: [{ index: 0, codec_type: "video", width: 640, height: 360 }], + }), + stderr: "", + }; + } + throw new Error("ffmpeg must not run once the shared duration cap already rejected the file"); + }; + + await assert.rejects( + () => + extractVideoAudioFromBytes(Buffer.from("fake video bytes"), { + maxDurationSeconds: 600, + runner, + timeoutMs: 5_000, + }), + /maximum duration/ + ); +}); + +test("rejects a WAV output that exceeds the configured byte cap", async () => { + const runner: VideoCommandRunner = async (executable, args) => { + if (executable === "ffprobe") { + return { + stdout: JSON.stringify({ + format: { duration: "4", format_name: "mp4" }, + streams: [{ index: 0, codec_type: "video", width: 640, height: 360 }], + }), + stderr: "", + }; + } + await writeFile(args.at(-1) ?? "", Buffer.alloc(16)); + return { stdout: "", stderr: "" }; + }; + + await assert.rejects( + () => + extractVideoAudioFromBytes(Buffer.from("fake video bytes"), { + maxDurationSeconds: 600, + maxOutputBytes: 8, + runner, + timeoutMs: 5_000, + }), + /byte limit exceeded/ + ); +}); + +test("propagates ffmpeg's own failure when the container has no audio stream", async () => { + const runner: VideoCommandRunner = async (executable) => { + if (executable === "ffprobe") { + return { + stdout: JSON.stringify({ + format: { duration: "4", format_name: "mp4" }, + streams: [{ index: 0, codec_type: "video", width: 640, height: 360 }], + }), + stderr: "", + }; + } + throw new Error("Stream map '0:a:0' matches no streams"); + }; + + await assert.rejects( + () => + extractVideoAudioFromBytes(Buffer.from("fake video bytes"), { + maxDurationSeconds: 600, + runner, + timeoutMs: 5_000, + }), + /matches no streams/ + ); +}); diff --git a/tests/unit/guardrails/videoBridgeAudioOrchestration.test.ts b/tests/unit/guardrails/videoBridgeAudioOrchestration.test.ts new file mode 100644 index 0000000000..73a6b82f66 --- /dev/null +++ b/tests/unit/guardrails/videoBridgeAudioOrchestration.test.ts @@ -0,0 +1,381 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { BridgeCache } from "../../../src/lib/guardrails/modalityBridge/bridgeCache.ts"; +import { orchestrateVideoAudioTranscription } from "../../../src/lib/guardrails/videoBridgeAudioOrchestration.ts"; + +const VIDEO_BYTES = Buffer.from("fake video bytes"); + +function baseOptions(overrides: Partial[0]> = {}) { + return { + extractAudio: async () => { + throw new Error("extractAudio must not be called in this test"); + }, + hasUsableCredentials: async () => true, + model: "deepgram/nova-3", + operatorOptIn: true, + requestOptIn: true, + timeoutMs: 5_000, + transcribe: async () => { + throw new Error("transcribe must not be called in this test"); + }, + videoBytes: VIDEO_BYTES, + ...overrides, + }; +} + +function extraction(overrides: Partial<{ durationSeconds: number; dataUri: string }> = {}) { + return { + audio: { + channels: 1, + dataUri: overrides.dataUri ?? "data:audio/wav;base64,UklGRg==", + sampleRateHz: 16_000, + }, + durationSeconds: overrides.durationSeconds ?? 4, + }; +} + +// --- Dual opt-in permutations ------------------------------------------------- + +test("neither opt-in: no extraction and no transcription call happens", async () => { + let extractCalled = false; + let transcribeCalled = false; + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + operatorOptIn: false, + requestOptIn: false, + extractAudio: async () => { + extractCalled = true; + return extraction(); + }, + transcribe: async () => { + transcribeCalled = true; + return { text: "should never happen" }; + }, + }) + ); + + assert.equal(result.attempted, false); + assert.equal(result.reason, "OPERATOR_OPT_OUT"); + assert.equal(result.track, null); + assert.equal(extractCalled, false); + assert.equal(transcribeCalled, false); +}); + +test("operator opt-in only: request opt-out still blocks every call", async () => { + let extractCalled = false; + let transcribeCalled = false; + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + operatorOptIn: true, + requestOptIn: false, + extractAudio: async () => { + extractCalled = true; + return extraction(); + }, + transcribe: async () => { + transcribeCalled = true; + return { text: "should never happen" }; + }, + }) + ); + + assert.equal(result.attempted, false); + assert.equal(result.reason, "REQUEST_OPT_OUT"); + assert.equal(extractCalled, false); + assert.equal(transcribeCalled, false); +}); + +test("request opt-in only: operator opt-out still blocks every call", async () => { + let extractCalled = false; + let transcribeCalled = false; + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + operatorOptIn: false, + requestOptIn: true, + extractAudio: async () => { + extractCalled = true; + return extraction(); + }, + transcribe: async () => { + transcribeCalled = true; + return { text: "should never happen" }; + }, + }) + ); + + assert.equal(result.attempted, false); + assert.equal(result.reason, "OPERATOR_OPT_OUT"); + assert.equal(extractCalled, false); + assert.equal(transcribeCalled, false); +}); + +test("both opt-ins present: this is the only permutation that transcribes", async () => { + let extractCalled = false; + let transcribeCalled = false; + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + operatorOptIn: true, + requestOptIn: true, + extractAudio: async (bytes) => { + extractCalled = true; + assert.equal(bytes, VIDEO_BYTES, "must reuse the exact same already-downloaded bytes"); + return extraction(); + }, + transcribe: async () => { + transcribeCalled = true; + return { text: "hello world" }; + }, + }) + ); + + assert.equal(result.attempted, true); + assert.equal(extractCalled, true); + assert.equal(transcribeCalled, true); + assert.ok(result.track); + assert.equal(result.track?.observations[0]?.text, "hello world"); +}); + +// --- Core scenarios ------------------------------------------------------------ + +test("audio extraction success feeds STT and returns a source-owned observation", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => extraction({ durationSeconds: 6 }), + transcribe: async () => ({ text: "spoken words" }), + }) + ); + + assert.equal(result.attempted, true); + assert.equal(result.sttModel, "deepgram/nova-3"); + assert.equal(result.timingPrecision, "coarse"); + assert.deepEqual(result.track, { + observations: [ + { confidence: 1, endSeconds: 6, source: "audio", startSeconds: 0, text: "spoken words" }, + ], + }); +}); + +test("STT success with provider segments preserves exact per-segment timing", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => extraction({ durationSeconds: 10 }), + transcribe: async () => ({ + segments: [ + { confidence: 0.9, endSeconds: 2, startSeconds: 0, text: "hello" }, + { endSeconds: 4, startSeconds: 2, text: "world" }, + ], + text: "hello world", + }), + }) + ); + + assert.equal(result.timingPrecision, "exact"); + assert.deepEqual(result.track?.observations, [ + { confidence: 0.9, endSeconds: 2, source: "audio", startSeconds: 0, text: "hello" }, + { confidence: 1, endSeconds: 4, source: "audio", startSeconds: 2, text: "world" }, + ]); +}); + +test("unavailable STT provider: extraction never runs and the result is a visual-only-safe partial", async () => { + let extractCalled = false; + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + hasUsableCredentials: async () => false, + extractAudio: async () => { + extractCalled = true; + return extraction(); + }, + }) + ); + + assert.equal(result.attempted, true); + assert.equal(result.reason, "PROVIDER_UNAVAILABLE"); + assert.equal(result.sttModel, null); + assert.equal(result.track, null); + assert.equal(extractCalled, false, "no broker call once no provider is usable"); +}); + +test("extraction timeout is reported distinctly from a generic extraction failure", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => { + throw new Error("Video audio extraction request aborted"); + }, + }) + ); + + assert.equal(result.reason, "TIMEOUT"); + assert.equal(result.track, null); +}); + +test("caller abort during extraction is reported as ABORTED, not TIMEOUT", async () => { + const controller = new AbortController(); + controller.abort(); + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + signal: controller.signal, + extractAudio: async () => { + throw new Error("Video audio extraction request aborted"); + }, + }) + ); + + assert.equal(result.reason, "ABORTED"); +}); + +test("caller abort during transcription is reported as ABORTED", async () => { + const controller = new AbortController(); + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + signal: controller.signal, + extractAudio: async () => extraction(), + transcribe: async () => { + controller.abort(); + throw new Error("Audio transcription timed out"); + }, + }) + ); + + assert.equal(result.reason, "ABORTED"); +}); + +test("transcription-boundary timeout is reported distinctly from a generic transcription failure", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => extraction(), + transcribe: async () => { + throw new Error("Audio transcription timed out"); + }, + }) + ); + + assert.equal(result.reason, "TIMEOUT"); +}); + +test("a generic transcription failure is reported as TRANSCRIPTION_FAILED, not confused with a timeout", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => extraction(), + transcribe: async () => { + throw new Error("Audio transcription failed (500)"); + }, + }) + ); + + assert.equal(result.reason, "TRANSCRIPTION_FAILED"); +}); + +test("a generic extraction failure (no audio track) is reported as EXTRACTION_FAILED", async () => { + const result = await orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => { + throw new Error("Stream map '0:a:0' matches no streams"); + }, + }) + ); + + assert.equal(result.reason, "EXTRACTION_FAILED"); +}); + +test("every failure path stays a well-defined partial — never throws, always track: null", async () => { + for (const scenario of [ + () => + orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => { + throw new Error("boom"); + }, + }) + ), + () => + orchestrateVideoAudioTranscription( + baseOptions({ + extractAudio: async () => extraction(), + transcribe: async () => { + throw new Error("boom"); + }, + }) + ), + () => orchestrateVideoAudioTranscription(baseOptions({ hasUsableCredentials: async () => false })), + ]) { + const result = await scenario(); + assert.equal(result.track, null, "visual-only fallback must remain available"); + } +}); + +// --- Cache ----------------------------------------------------------------- + +test("a cache hit skips both the broker call and the transcription boundary", async () => { + const cache = new BridgeCache({ maxEntries: 10, ttlMs: 60_000 }); + let extractCalls = 0; + let transcribeCalls = 0; + const options = baseOptions({ + cache, + cacheKeyRef: "video-ref-1", + extractAudio: async () => { + extractCalls += 1; + return extraction(); + }, + transcribe: async () => { + transcribeCalls += 1; + return { text: "cached transcript" }; + }, + }); + + const first = await orchestrateVideoAudioTranscription(options); + assert.equal(extractCalls, 1); + assert.equal(transcribeCalls, 1); + + const second = await orchestrateVideoAudioTranscription(options); + assert.equal(extractCalls, 1, "second call must not re-extract"); + assert.equal(transcribeCalls, 1, "second call must not re-transcribe — no repeat paid STT call"); + assert.deepEqual(second.track, first.track); + assert.equal(second.attempted, true); +}); + +test("a different cache key (different video) does not share another video's cache entry", async () => { + const cache = new BridgeCache({ maxEntries: 10, ttlMs: 60_000 }); + let extractCalls = 0; + const run = (cacheKeyRef: string) => + orchestrateVideoAudioTranscription( + baseOptions({ + cache, + cacheKeyRef, + extractAudio: async () => { + extractCalls += 1; + return extraction(); + }, + transcribe: async () => ({ text: "distinct transcript" }), + }) + ); + + await run("video-a"); + await run("video-b"); + assert.equal(extractCalls, 2, "each distinct video ref must be extracted independently"); +}); + +// --- Shared budgets ---------------------------------------------------------- + +test("the same timeoutMs and signal thread through both the extraction and transcription steps", async () => { + const controller = new AbortController(); + const seen: Array<{ signal?: AbortSignal; timeoutMs: number }> = []; + await orchestrateVideoAudioTranscription( + baseOptions({ + signal: controller.signal, + timeoutMs: 42_000, + extractAudio: async (_bytes, options) => { + seen.push({ signal: options.signal, timeoutMs: options.timeoutMs }); + return extraction(); + }, + transcribe: async (_part, config) => { + seen.push({ timeoutMs: config.timeoutMs }); + return { text: "ok" }; + }, + }) + ); + + assert.equal(seen[0].timeoutMs, 42_000); + assert.equal(seen[0].signal, controller.signal); + assert.equal(seen[1].timeoutMs, 42_000, "the transcription boundary must reuse the same shared budget"); +}); diff --git a/tests/unit/video-bridge-broker.test.ts b/tests/unit/video-bridge-broker.test.ts index 761cf65e9d..cab5337788 100644 --- a/tests/unit/video-bridge-broker.test.ts +++ b/tests/unit/video-bridge-broker.test.ts @@ -5,12 +5,19 @@ import test from "node:test"; import { VIDEO_BRIDGE_BROKER_PATH, buildVideoBridgeBrokerHeaders, + extractVideoAudioViaBroker, extractVideoFramesViaBroker, isVideoBridgeBrokerInternalRequest, resolveVideoBridgeBrokerBaseUrl, } from "../../src/lib/guardrails/videoBridgeBrokerClient.ts"; import { createVideoExtractionQueue } from "../../src/lib/guardrails/videoBridgeBrokerQueue.ts"; import { AUTHZ_HEADER_PEER_LOCALITY } from "../../src/server/authz/headers.ts"; +import { + BROKER_TIMEOUT_MS, + handleVideoExtractionBrokerRequest, +} from "../../src/app/api/modality-bridge/video/extract/route.ts"; + +const EXTRACT_PATH = "/api/modality-bridge/video/extract"; test("broker origin is pinned to the active loopback listener and ignores client-controlled origins", () => { const previousPort = process.env.PORT; @@ -211,6 +218,152 @@ test("broker queue bounds pending jobs and queued bytes", async () => { await Promise.all([active, pending]); }); +test("audio broker client requests mode=audio on the exact same pinned route", async () => { + let requestedUrl = ""; + let requestedInit: RequestInit | undefined; + const response = await extractVideoAudioViaBroker( + Buffer.from("safe-video"), + { timeoutMs: 5_000 }, + { + fetchImpl: async (input, init) => { + requestedUrl = String(input); + requestedInit = init; + return Response.json({ + audio: { channels: 1, dataUri: "data:audio/wav;base64,UklGRg==", sampleRateHz: 16000 }, + durationSeconds: 4, + }); + }, + } + ); + + assert.match(requestedUrl, /\/api\/modality-bridge\/video\/extract\?mode=audio$/); + assert.equal(new URL(requestedUrl).hostname, "127.0.0.1"); + assert.equal(requestedInit?.method, "POST"); + assert.deepEqual(Buffer.from(requestedInit?.body as Uint8Array), Buffer.from("safe-video")); + assert.equal(response.durationSeconds, 4); + assert.deepEqual(response.audio, { + channels: 1, + dataUri: "data:audio/wav;base64,UklGRg==", + sampleRateHz: 16000, + }); +}); + +test("audio broker client rejects a malformed or non-WAV response instead of trusting it", async () => { + await assert.rejects( + () => + extractVideoAudioViaBroker( + Buffer.from("safe-video"), + { timeoutMs: 5_000 }, + { + fetchImpl: async () => + Response.json({ + audio: { channels: 1, dataUri: "data:image/jpeg;base64,QQ==", sampleRateHz: 16000 }, + durationSeconds: 4, + }), + } + ), + /invalid metadata/ + ); +}); + +test("broker route enforces the exact same queue-capacity contract for the audio operation as for frames", async () => { + // maxPending: 0 rejects every run() as capacity-exceeded from the first call — + // this is the identical VideoExtractionQueueError path frames already relies on + // (see "broker route maps queue capacity..." above), now exercised for mode=audio. + const queue = createVideoExtractionQueue({ concurrency: 1, maxPending: 0, maxQueuedBytes: 100 }); + + const audioResponse = await handleVideoExtractionBrokerRequest( + new Request(`http://localhost${EXTRACT_PATH}?mode=audio`, { + method: "POST", + headers: { + ...buildVideoBridgeBrokerHeaders(), + [AUTHZ_HEADER_PEER_LOCALITY]: "loopback", + "Content-Type": "application/octet-stream", + }, + body: Buffer.from("video"), + }), + { + queue, + extractAudio: async () => { + throw new Error("audio extractor must not run once the shared queue reports capacity"); + }, + } + ); + + assert.equal(audioResponse.status, 503); + assert.equal(audioResponse.headers.get("Retry-After"), "1"); +}); + +test("the module-level broker singleton queue is the same object for every dispatch branch", async () => { + // Neither branch of the route is given a `queue` dependency here, so all + // three must fall back to the identical `extractionQueue` singleton — + // proving production traffic for frames, audio, and subtitles (#11659, + // merged alongside #11654 in the same /merge-batch) shares one process + // queue instead of each operation racing its own. + const routeSource = await import("node:fs/promises").then((fs) => + fs.readFile( + new URL("../../src/app/api/modality-bridge/video/extract/route.ts", import.meta.url), + "utf8" + ) + ); + const queueReferences = routeSource.match(/dependencies\.queue \?\? extractionQueue/g) ?? []; + assert.equal( + queueReferences.length, + 3, + "the frame, audio, and subtitle branches must all default to the one extractionQueue singleton" + ); +}); + +test("broker route rejects frame-only parameters when mode=audio", async () => { + const response = await handleVideoExtractionBrokerRequest( + new Request(`http://localhost${EXTRACT_PATH}?mode=audio&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("broker route runs the injected audio extractor with the shared deadline and byte budget", async () => { + let receivedTimeoutMs = 0; + let receivedByteLength = 0; + const response = await handleVideoExtractionBrokerRequest( + new Request(`http://localhost${EXTRACT_PATH}?mode=audio`, { + method: "POST", + headers: { + ...buildVideoBridgeBrokerHeaders(), + [AUTHZ_HEADER_PEER_LOCALITY]: "loopback", + "Content-Type": "application/octet-stream", + }, + body: Buffer.from("video-bytes"), + }), + { + extractAudio: async (bytes, options) => { + receivedByteLength = bytes.byteLength; + receivedTimeoutMs = options.timeoutMs; + return { + channels: 1, + dataUri: "data:audio/wav;base64,UklGRg==", + durationSeconds: 3, + sampleRateHz: 16000, + }; + }, + } + ); + + assert.equal(response.status, 200); + assert.equal(receivedByteLength, Buffer.from("video-bytes").byteLength); + assert.equal(receivedTimeoutMs, BROKER_TIMEOUT_MS); + const body = (await response.json()) as { audio: { dataUri: string }; durationSeconds: number }; + assert.equal(body.durationSeconds, 3); + assert.equal(body.audio.dataUri, "data:audio/wav;base64,UklGRg=="); +}); + test("broker queue removes an aborted pending item and never executes it", async () => { const queue = createVideoExtractionQueue({ concurrency: 1, maxPending: 2, maxQueuedBytes: 16 }); let release!: () => void; diff --git a/tests/unit/video-bridge-settings.test.ts b/tests/unit/video-bridge-settings.test.ts index b7c73d09bf..943bb0f864 100644 --- a/tests/unit/video-bridge-settings.test.ts +++ b/tests/unit/video-bridge-settings.test.ts @@ -7,6 +7,7 @@ import { } from "../../src/lib/modelCapabilityModalities.ts"; import { MODALITY_BRIDGE_DEFAULTS, + resolveVideoAudioTranscriptionRuntimeSettings, resolveVideoBridgeRuntimeSettings, } from "../../src/shared/constants/modalityBridgeDefaults.ts"; import { updateSettingsSchema } from "../../src/shared/validation/settingsSchemas.ts"; @@ -95,3 +96,22 @@ test("persisted segment-aware policy remains an explicit opt-in", () => { "segment_aware" ); }); + +test("the operator half of the FU-06 audio-transcription dual opt-in defaults OFF (#11654)", () => { + assert.deepEqual(resolveVideoAudioTranscriptionRuntimeSettings({}), { enabled: false }); + assert.deepEqual(resolveVideoAudioTranscriptionRuntimeSettings(undefined), { enabled: false }); + assert.deepEqual( + resolveVideoAudioTranscriptionRuntimeSettings({ + modalityBridgeVideoAudioTranscriptionEnabled: true, + }), + { enabled: true } + ); + assert.equal( + updateSettingsSchema.safeParse({ modalityBridgeVideoAudioTranscriptionEnabled: true }).success, + true + ); + assert.equal( + updateSettingsSchema.safeParse({ modalityBridgeVideoAudioTranscriptionEnabled: "yes" }).success, + false + ); +});