From 8b1a647bd9b83a59f0fff3f43d211131f9d3655f Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Sat, 15 Aug 2026 01:13:29 -0300 Subject: [PATCH] feat(video-bridge): add safe frame extraction runtime --- src/lib/guardrails/videoBridgeHelpers.ts | 232 +++++++++++++++++ src/lib/guardrails/videoBridgeRuntime.ts | 204 +++++++++++++++ src/lib/guardrails/visionBridgeHelpers.ts | 25 +- src/shared/network/remoteImageFetch.ts | 19 +- .../guardrails/videoBridgeHelpers.test.ts | 244 ++++++++++++++++++ .../guardrails/videoBridgeRuntime.test.ts | 125 +++++++++ ...isionBridgeHelpers.callVisionModel.test.ts | 44 ++++ tests/unit/remote-media-fetch.test.ts | 36 +++ 8 files changed, 922 insertions(+), 7 deletions(-) create mode 100644 src/lib/guardrails/videoBridgeHelpers.ts create mode 100644 src/lib/guardrails/videoBridgeRuntime.ts create mode 100644 tests/unit/guardrails/videoBridgeHelpers.test.ts create mode 100644 tests/unit/guardrails/videoBridgeRuntime.test.ts create mode 100644 tests/unit/remote-media-fetch.test.ts diff --git a/src/lib/guardrails/videoBridgeHelpers.ts b/src/lib/guardrails/videoBridgeHelpers.ts new file mode 100644 index 0000000000..f990a94753 --- /dev/null +++ b/src/lib/guardrails/videoBridgeHelpers.ts @@ -0,0 +1,232 @@ +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { detectMediaParts, type MediaPart } from "@omniroute/open-sse/utils/mediaParts"; + +import { fetchRemoteMedia, type RemoteMediaFetchResult } from "@/shared/network/remoteImageFetch"; + +import { + extractFramesFromLocalVideo, + probeLocalVideo, + type VideoCommandRunner, +} from "./videoBridgeRuntime"; + +export const VIDEO_BRIDGE_MAX_BYTES = 50 * 1024 * 1024; +export const VIDEO_BRIDGE_MAX_DURATION_SECONDS = 600; + +type VideoContainer = "messages" | "input"; +type VideoMessage = { role?: string; content?: unknown }; +type VideoRequestBody = { + messages?: VideoMessage[]; + input?: VideoMessage[]; + [key: string]: unknown; +}; + +export interface VideoPart { + container: VideoContainer; + messageIndex: number; + partIndex: number; + ref: string; + shape: "input_video" | "video_url" | "video_source" | "data_uri_string"; +} + +const REPLACEABLE_VIDEO_SHAPES: ReadonlySet = new Set([ + "input_video", + "video_url", + "video_source", + "data_uri_string", +]); + +export function extractVideoParts(body: VideoRequestBody): VideoPart[] { + const container: VideoContainer | null = Array.isArray(body.messages) + ? "messages" + : Array.isArray(body.input) + ? "input" + : null; + if (!container) return []; + return detectMediaParts(body[container]) + .filter( + (part) => + part.kind === "video" && + !part.nested && + part.ref.length > 0 && + REPLACEABLE_VIDEO_SHAPES.has(part.shape) + ) + .map((part) => ({ + container, + messageIndex: part.messageIndex, + partIndex: part.partIndex, + ref: part.ref, + shape: part.shape as VideoPart["shape"], + })); +} + +export function replaceVideoParts( + body: TBody, + parts: readonly VideoPart[], + descriptions: readonly (string | null)[] +): TBody { + const result = structuredClone(body); + for (let index = 0; index < parts.length && index < descriptions.length; index++) { + const description = descriptions[index]; + if (description === null) continue; + const part = parts[index]; + const content = result[part.container]?.[part.messageIndex]?.content; + if (!Array.isArray(content) || part.partIndex >= content.length) continue; + content[part.partIndex] = { + type: part.container === "input" ? "input_text" : "text", + text: description, + }; + } + return result; +} + +export interface DescribeVideoOptions { + frameCount: number; + maxBytes?: number; + maxDurationSeconds?: number; + timeoutMs: number; + signal?: AbortSignal; +} + +export interface DescribeVideoDependencies { + fetchRemote?: (url: string, options: { signal: AbortSignal }) => Promise; + runner?: VideoCommandRunner; +} + +export interface DescribedVideo { + cacheHits?: number; + description: string; + durationSeconds: number; + framesRequested: number; + framesUsed: number; +} + +function decodeVideoDataUri(ref: string): Buffer | null { + const match = /^data:video\/[A-Za-z0-9.+-]+;base64,([A-Za-z0-9+/=\s]+)$/i.exec(ref); + return match ? Buffer.from(match[1].replace(/\s/g, ""), "base64") : null; +} + +async function loadVideoBytes( + part: VideoPart, + maxBytes: number, + timeoutMs: number, + signal: AbortSignal, + deps: DescribeVideoDependencies +): Promise { + const dataBytes = decodeVideoDataUri(part.ref); + let bytes: Buffer; + if (dataBytes) { + bytes = dataBytes; + } else { + if (!part.ref.startsWith("https://")) { + throw new Error("Video Bridge accepts only HTTPS URLs or video data URIs"); + } + const fetchRemote = + deps.fetchRemote ?? + ((url: string, options: { signal: AbortSignal }) => + fetchRemoteMedia(url, { + guard: "public-only", + maxBytes, + pinDns: true, + signal: options.signal, + timeoutMs, + })); + bytes = (await fetchRemote(part.ref, { signal })).buffer; + } + if (bytes.byteLength > maxBytes) { + throw new Error("Video exceeds the maximum size"); + } + return bytes; +} + +export function formatVideoTimestamp(timestampSeconds: number): string { + const totalMilliseconds = Math.max(0, Math.round(timestampSeconds * 1000)); + const minutes = Math.floor(totalMilliseconds / 60_000); + const seconds = Math.floor((totalMilliseconds % 60_000) / 1000); + const milliseconds = totalMilliseconds % 1000; + return `${String(minutes).padStart(2, "0")}:${String(seconds).padStart(2, "0")}.${String(milliseconds).padStart(3, "0")}`; +} + +export async function describeVideoPart( + part: VideoPart, + options: DescribeVideoOptions, + captionFrame: ( + frameDataUri: string, + timestampSeconds: number, + signal: AbortSignal + ) => Promise, + deps: DescribeVideoDependencies = {} +): Promise { + const timeoutController = new AbortController(); + const timeout = setTimeout(() => timeoutController.abort(), options.timeoutMs); + const signal = options.signal + ? AbortSignal.any([options.signal, timeoutController.signal]) + : timeoutController.signal; + const temporaryDirectory = await mkdtemp(join(tmpdir(), "omniroute-video-bridge-")); + try { + const bytes = await loadVideoBytes( + part, + options.maxBytes ?? VIDEO_BRIDGE_MAX_BYTES, + options.timeoutMs, + signal, + deps + ); + const inputPath = join(temporaryDirectory, "input.video"); + const framesDirectory = join(temporaryDirectory, "frames"); + await mkdir(framesDirectory, { mode: 0o700 }); + await writeFile(inputPath, bytes, { mode: 0o600 }); + + const metadata = await probeLocalVideo(inputPath, { + maxDurationSeconds: options.maxDurationSeconds ?? VIDEO_BRIDGE_MAX_DURATION_SECONDS, + runner: deps.runner, + signal, + timeoutMs: Math.min(options.timeoutMs, 30_000), + }); + const frames = await extractFramesFromLocalVideo(inputPath, framesDirectory, { + durationSeconds: metadata.durationSeconds, + frameCount: options.frameCount, + runner: deps.runner, + signal, + timeoutMs: options.timeoutMs, + }); + + const descriptions: string[] = []; + for (const frame of frames) { + try { + const jpeg = await readFile(frame.path); + const caption = ( + await captionFrame( + `data:image/jpeg;base64,${jpeg.toString("base64")}`, + frame.timestampSeconds, + signal + ) + ).trim(); + if (caption) { + descriptions.push(`frame@t=${formatVideoTimestamp(frame.timestampSeconds)} ${caption}`); + } + } catch { + if (signal.aborted) { + throw new Error("Video Bridge processing timed out or was aborted"); + } + // Partial frame failures are omitted. An all-frame failure is handled below. + } + } + if (descriptions.length === 0) { + throw new Error("Video frames could not be described"); + } + return { + description: `[Video description: ${descriptions.join("; ")}]`, + durationSeconds: metadata.durationSeconds, + framesRequested: frames.length, + framesUsed: descriptions.length, + }; + } catch (error) { + if (signal.aborted) throw new Error("Video Bridge processing timed out or was aborted"); + throw error; + } finally { + clearTimeout(timeout); + await rm(temporaryDirectory, { force: true, recursive: true }); + } +} diff --git a/src/lib/guardrails/videoBridgeRuntime.ts b/src/lib/guardrails/videoBridgeRuntime.ts new file mode 100644 index 0000000000..b332a63af2 --- /dev/null +++ b/src/lib/guardrails/videoBridgeRuntime.ts @@ -0,0 +1,204 @@ +import { execFile } from "node:child_process"; +import { isAbsolute, join } from "node:path"; +import { promisify } from "node:util"; + +const execFileAsync = promisify(execFile); + +export interface VideoCommandOptions { + timeoutMs: number; + signal?: AbortSignal; +} + +export type VideoCommandRunner = ( + executable: "ffmpeg" | "ffprobe", + args: readonly string[], + options: VideoCommandOptions +) => Promise<{ stdout: string; stderr: string }>; + +export interface VideoRuntimeStatus { + available: boolean; + ffmpegVersion: string | null; + ffprobeVersion: string | null; + reason?: string; +} + +export interface VideoFrameFile { + path: string; + timestampSeconds: 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) }; +}; + +function assertLocalPath(filePath: string): void { + if (!isAbsolute(filePath) || filePath.includes("\0") || filePath.includes("://")) { + throw new Error("Video runtime requires a local path"); + } +} + +function parseVersion(output: string): string | null { + const version = /\bversion\s+([^\s]+)/i.exec(output)?.[1]; + return version ? version.slice(0, 80).replace(/[^A-Za-z0-9._+-]/g, "_") : null; +} + +let runtimeProbeCache: { expiresAt: number; value: VideoRuntimeStatus } | null = null; + +export function resetVideoRuntimeProbeCacheForTests(): void { + runtimeProbeCache = null; +} + +export async function probeVideoRuntime( + options: { + cacheTtlMs?: number; + runner?: VideoCommandRunner; + signal?: AbortSignal; + timeoutMs?: number; + } = {} +): Promise { + const now = Date.now(); + if (runtimeProbeCache && runtimeProbeCache.expiresAt > now) { + return structuredClone(runtimeProbeCache.value); + } + + const runner = options.runner ?? defaultRunner; + const commandOptions = { + signal: options.signal, + timeoutMs: options.timeoutMs ?? 5_000, + }; + let value: VideoRuntimeStatus; + try { + const [ffmpeg, ffprobe] = await Promise.all([ + runner("ffmpeg", ["-version"], commandOptions), + runner("ffprobe", ["-version"], commandOptions), + ]); + const ffmpegVersion = parseVersion(ffmpeg.stdout); + const ffprobeVersion = parseVersion(ffprobe.stdout); + value = + ffmpegVersion && ffprobeVersion + ? { available: true, ffmpegVersion, ffprobeVersion } + : { + available: false, + ffmpegVersion, + ffprobeVersion, + reason: "FFmpeg and ffprobe versions could not be verified", + }; + } catch { + value = { + available: false, + ffmpegVersion: null, + ffprobeVersion: null, + reason: "FFmpeg and ffprobe are not available on PATH", + }; + } + + runtimeProbeCache = { + expiresAt: now + (options.cacheTtlMs ?? 30_000), + value, + }; + return structuredClone(value); +} + +export function calculateFrameTimestamps( + durationSeconds: number, + requestedFrameCount: number +): number[] { + if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) { + throw new Error("Video duration must be positive"); + } + if ( + !Number.isInteger(requestedFrameCount) || + requestedFrameCount < 1 || + requestedFrameCount > 16 + ) { + throw new Error("Video frame count must be between 1 and 16"); + } + const frameCount = Math.min(requestedFrameCount, Math.max(1, Math.floor(durationSeconds))); + return Array.from( + { length: frameCount }, + (_unused, index) => ((index + 0.5) * durationSeconds) / frameCount + ); +} + +export async function probeLocalVideo( + inputPath: string, + options: { + maxDurationSeconds?: number; + runner?: VideoCommandRunner; + signal?: AbortSignal; + timeoutMs?: number; + } = {} +): Promise<{ durationSeconds: number }> { + assertLocalPath(inputPath); + const result = await (options.runner ?? defaultRunner)( + "ffprobe", + ["-v", "error", "-show_entries", "format=duration", "-of", "json", inputPath], + { signal: options.signal, timeoutMs: options.timeoutMs ?? 30_000 } + ); + let durationSeconds = Number.NaN; + try { + const parsed = JSON.parse(result.stdout) as { format?: { duration?: unknown } }; + durationSeconds = Number(parsed.format?.duration); + } catch { + // The stable error below deliberately excludes raw ffprobe output. + } + if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) { + throw new Error("Video runtime returned invalid duration metadata"); + } + if (durationSeconds > (options.maxDurationSeconds ?? 600)) { + throw new Error("Video exceeds the maximum duration"); + } + return { durationSeconds }; +} + +export async function extractFramesFromLocalVideo( + inputPath: string, + outputDirectory: string, + options: { + durationSeconds: number; + frameCount: number; + runner?: VideoCommandRunner; + signal?: AbortSignal; + timeoutMs?: number; + } +): Promise { + assertLocalPath(inputPath); + assertLocalPath(outputDirectory); + const timestamps = calculateFrameTimestamps(options.durationSeconds, options.frameCount); + const runner = options.runner ?? defaultRunner; + const frames: VideoFrameFile[] = []; + + for (let index = 0; index < timestamps.length; index++) { + const timestampSeconds = timestamps[index]; + const outputPath = join(outputDirectory, `frame-${String(index + 1).padStart(2, "0")}.jpg`); + await runner( + "ffmpeg", + [ + "-nostdin", + "-hide_banner", + "-loglevel", + "error", + "-ss", + timestampSeconds.toFixed(3), + "-i", + inputPath, + "-frames:v", + "1", + "-q:v", + "2", + "-y", + outputPath, + ], + { signal: options.signal, timeoutMs: options.timeoutMs ?? 120_000 } + ); + frames.push({ path: outputPath, timestampSeconds }); + } + return frames; +} diff --git a/src/lib/guardrails/visionBridgeHelpers.ts b/src/lib/guardrails/visionBridgeHelpers.ts index e8b0b7e3bc..fbc8e06847 100644 --- a/src/lib/guardrails/visionBridgeHelpers.ts +++ b/src/lib/guardrails/visionBridgeHelpers.ts @@ -346,6 +346,8 @@ export interface VisionModelConfig { prompt: string; timeoutMs: number; maxImages: number; + /** Optional parent deadline/abort propagated by multi-step media bridges. */ + signal?: AbortSignal; /** Injectable fetch (tests). Defaults to undici fetch to bypass the runtime's hooked global fetch. */ fetchImpl?: typeof fetch; } @@ -375,6 +377,10 @@ export async function callVisionModel( routerConfig?: Partial, deps?: import("./visionBridgeRouter").VisionBridgeRouterDeps ): Promise { + if (config.signal?.aborted) { + throw new Error("Vision model call aborted"); + } + // Auto-select the best vision model. `deps` is the router's existing // injectable credential-check seam — without forwarding it, tests (and any // embedder) cannot keep model selection away from the live connections DB. @@ -398,6 +404,9 @@ export async function callVisionModel( const maxAttempts = Math.min(modelsToTry.length, routerConfig?.maxFallbackAttempts ?? 3); for (let attempt = 0; attempt < maxAttempts; attempt++) { + if (config.signal?.aborted) { + throw lastError ?? new Error("Vision model call aborted"); + } const currentModel = modelsToTry[attempt]; const attemptStart = Date.now(); try { @@ -411,6 +420,9 @@ export async function callVisionModel( } catch (error) { recordLatency(currentModel, Date.now() - attemptStart, false); lastError = error instanceof Error ? error : new Error(String(error)); + if (config.signal?.aborted) { + throw lastError; + } // Continue to next model on failure } } @@ -620,6 +632,9 @@ async function callVisionModelSingle( ): Promise { const controller = new AbortController(); const timeoutId = setTimeout(() => controller.abort(), config.timeoutMs); + const signal = config.signal + ? AbortSignal.any([config.signal, controller.signal]) + : controller.signal; // Resolve API key based on provider const resolvedApiKey = resolveProviderApiKey(config.model, apiKey); @@ -642,7 +657,7 @@ async function callVisionModelSingle( const normalizedImageInput = await normalizeVisionImageInput( imageDataUri, requiresBase64, - controller.signal, + signal, fetchImpl ); @@ -664,7 +679,7 @@ async function callVisionModelSingle( response = await fetchImpl(`${anthropicBaseUrl}/v1/messages`, { method: "POST", - signal: controller.signal, + signal, headers: { "x-api-key": resolvedApiKey, "anthropic-version": "2023-06-01", @@ -748,7 +763,7 @@ async function callVisionModelSingle( response = await fetchImpl(`${baseUrl}/chat/completions`, { method: "POST", - signal: controller.signal, + signal, headers, body: JSON.stringify({ model: requestModel, @@ -813,7 +828,9 @@ async function callVisionModelSingle( clearTimeout(timeoutId); if (error instanceof Error && error.name === "AbortError") { - throw new Error("Vision model call timed out"); + throw new Error( + config.signal?.aborted ? "Vision model call aborted" : "Vision model call timed out" + ); } throw error; diff --git a/src/shared/network/remoteImageFetch.ts b/src/shared/network/remoteImageFetch.ts index a77955b92a..18f5917059 100644 --- a/src/shared/network/remoteImageFetch.ts +++ b/src/shared/network/remoteImageFetch.ts @@ -44,6 +44,11 @@ export interface RemoteImageFetchResult { url: string; } +/** Generic aliases for non-image callers that need the same SSRF/bounds policy. */ +export type RemoteMediaLookup = RemoteImageLookup; +export type RemoteMediaFetchOptions = RemoteImageFetchOptions; +export type RemoteMediaFetchResult = RemoteImageFetchResult; + function validateRemoteImageUrl(input: string | URL, guard: OutboundUrlGuardMode) { return guard === "public-only" ? parseAndValidatePublicUrl(input) : parseOutboundUrl(input); } @@ -166,10 +171,10 @@ async function readResponseBuffer(response: Response, maxBytes: number) { return Buffer.concat(chunks, totalBytes); } -export async function fetchRemoteImage( +export async function fetchRemoteMedia( input: string | URL, - options: RemoteImageFetchOptions = {} -): Promise { + options: RemoteMediaFetchOptions = {} +): Promise { const injectedFetch = options.fetchImpl; // Default off: production callers that need connection pinning opt in. This keeps // globalThis.fetch mockable for image-generation tests and preserves the previous @@ -222,3 +227,11 @@ export async function fetchRemoteImage( throw new Error(`Remote image exceeded ${maxRedirects} redirect limit`); } + +/** Backward-compatible image-specific entry point. */ +export async function fetchRemoteImage( + input: string | URL, + options: RemoteImageFetchOptions = {} +): Promise { + return fetchRemoteMedia(input, options); +} diff --git a/tests/unit/guardrails/videoBridgeHelpers.test.ts b/tests/unit/guardrails/videoBridgeHelpers.test.ts new file mode 100644 index 0000000000..c0e6e3347a --- /dev/null +++ b/tests/unit/guardrails/videoBridgeHelpers.test.ts @@ -0,0 +1,244 @@ +import assert from "node:assert/strict"; +import { access, writeFile } from "node:fs/promises"; +import test from "node:test"; + +import { + describeVideoPart, + extractVideoParts, + replaceVideoParts, +} from "../../../src/lib/guardrails/videoBridgeHelpers.ts"; +import type { VideoCommandRunner } from "../../../src/lib/guardrails/videoBridgeRuntime.ts"; + +test("extracts and replaces video parts in Chat and Responses payloads without shifting siblings", () => { + const chatBody = { + messages: [ + { + role: "user", + content: [ + { type: "text", text: "before" }, + { type: "input_video", video_url: "data:video/mp4;base64,QUJD" }, + { type: "text", text: "after" }, + ], + }, + ], + }; + const chatParts = extractVideoParts(chatBody); + assert.equal(chatParts.length, 1); + assert.equal(chatParts[0].container, "messages"); + assert.deepEqual( + replaceVideoParts(chatBody, chatParts, ["[Video description: frame@t=00:01.000 demo]"]) + .messages[0].content, + [ + { type: "text", text: "before" }, + { type: "text", text: "[Video description: frame@t=00:01.000 demo]" }, + { type: "text", text: "after" }, + ] + ); + + const responsesBody = { + input: [ + { + role: "user", + content: [{ type: "video_url", video_url: { url: "https://example.test/a.mp4" } }], + }, + ], + }; + const responseParts = extractVideoParts(responsesBody); + assert.equal(responseParts[0].container, "input"); + assert.deepEqual( + replaceVideoParts(responsesBody, responseParts, ["description"]).input[0].content, + [{ type: "input_text", text: "description" }] + ); +}); + +test("downloads bytes before ffmpeg, captions frames sequentially, and removes temporary files", async () => { + const commandArgs: string[][] = []; + let temporaryInput = ""; + const runner: VideoCommandRunner = async (executable, args) => { + commandArgs.push([...args]); + if (executable === "ffprobe") { + temporaryInput = args.at(-1) ?? ""; + return { stdout: JSON.stringify({ format: { duration: "4" } }), stderr: "" }; + } + await writeFile(args.at(-1) ?? "", Buffer.from(`jpeg-${commandArgs.length}`)); + return { stdout: "", stderr: "" }; + }; + const captionOrder: string[] = []; + const result = await describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "https://example.test/private.mp4", + shape: "video_url", + }, + { + frameCount: 2, + maxBytes: 1024, + maxDurationSeconds: 600, + timeoutMs: 20_000, + }, + async (frame, timestampSeconds) => { + captionOrder.push(`${timestampSeconds}:${frame.slice(0, 20)}`); + return timestampSeconds < 2 ? "first frame" : "second frame"; + }, + { + fetchRemote: async () => ({ + buffer: Buffer.from("downloaded-video"), + contentType: "video/mp4", + url: "https://example.test/private.mp4", + }), + runner, + } + ); + + assert.equal( + result.description, + "[Video description: frame@t=00:01.000 first frame; frame@t=00:03.000 second frame]" + ); + assert.equal(result.framesUsed, 2); + assert.deepEqual( + captionOrder.map((entry) => entry.split(":", 1)[0]), + ["1", "3"] + ); + assert.equal( + commandArgs.every((args) => !args.some((arg) => arg.includes("example.test"))), + true + ); + await assert.rejects(() => access(temporaryInput)); +}); + +test("rejects oversized video data before invoking the process boundary", async () => { + let called = false; + await assert.rejects( + () => + describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "data:video/mp4;base64,QUJDRA==", + shape: "input_video", + }, + { frameCount: 1, maxBytes: 2, maxDurationSeconds: 600, timeoutMs: 5_000 }, + async () => "unused", + { + runner: async () => { + called = true; + return { stdout: "", stderr: "" }; + }, + } + ), + /maximum size/ + ); + assert.equal(called, false); +}); + +test("keeps successful captions after a partial frame failure and still cleans up", async () => { + let temporaryInput = ""; + const runner: VideoCommandRunner = async (executable, args) => { + if (executable === "ffprobe") { + temporaryInput = args.at(-1) ?? ""; + return { stdout: JSON.stringify({ format: { duration: "4" } }), stderr: "" }; + } + await writeFile(args.at(-1) ?? "", Buffer.from("jpeg")); + return { stdout: "", stderr: "" }; + }; + let captionCalls = 0; + const result = await describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "data:video/mp4;base64,QUJD", + shape: "input_video", + }, + { frameCount: 2, timeoutMs: 5_000 }, + async () => { + captionCalls += 1; + if (captionCalls === 1) throw new Error("one frame failed"); + return "usable second frame"; + }, + { runner } + ); + + assert.equal(result.description, "[Video description: frame@t=00:03.000 usable second frame]"); + assert.equal(result.framesRequested, 2); + assert.equal(result.framesUsed, 1); + await assert.rejects(() => access(temporaryInput)); +}); + +test("propagates abort as a sanitized error and removes the temporary tree", async () => { + const controller = new AbortController(); + controller.abort(); + let temporaryInput = ""; + await assert.rejects( + () => + describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "data:video/mp4;base64,QUJD", + shape: "input_video", + }, + { frameCount: 1, signal: controller.signal, timeoutMs: 5_000 }, + async () => "unused", + { + runner: async (_executable, args, options) => { + temporaryInput = args.at(-1) ?? ""; + assert.equal(options.signal?.aborted, true); + throw new Error("private process detail"); + }, + } + ), + /processing timed out or was aborted/ + ); + await assert.rejects(() => access(temporaryInput)); +}); + +test("aborts an in-flight caption at the total video deadline without starting later frames", async () => { + let temporaryInput = ""; + let captionCalls = 0; + const runner: VideoCommandRunner = async (executable, args) => { + if (executable === "ffprobe") { + temporaryInput = args.at(-1) ?? ""; + return { stdout: JSON.stringify({ format: { duration: "4" } }), stderr: "" }; + } + await writeFile(args.at(-1) ?? "", Buffer.from("jpeg")); + return { stdout: "", stderr: "" }; + }; + + await assert.rejects( + () => + describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "data:video/mp4;base64,QUJD", + shape: "input_video", + }, + { frameCount: 2, timeoutMs: 25 }, + async (_frame, _timestampSeconds, signal) => { + captionCalls += 1; + await new Promise((_resolve, reject) => { + signal.addEventListener( + "abort", + () => { + const error = new Error("private caption transport detail"); + error.name = "AbortError"; + reject(error); + }, + { once: true } + ); + }); + }, + { runner } + ), + /processing timed out or was aborted/ + ); + + assert.equal(captionCalls, 1, "the shared deadline must stop sequential frame captioning"); + await assert.rejects(() => access(temporaryInput)); +}); diff --git a/tests/unit/guardrails/videoBridgeRuntime.test.ts b/tests/unit/guardrails/videoBridgeRuntime.test.ts new file mode 100644 index 0000000000..947dad6133 --- /dev/null +++ b/tests/unit/guardrails/videoBridgeRuntime.test.ts @@ -0,0 +1,125 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + calculateFrameTimestamps, + extractFramesFromLocalVideo, + probeLocalVideo, + probeVideoRuntime, + resetVideoRuntimeProbeCacheForTests, + type VideoCommandRunner, +} from "../../../src/lib/guardrails/videoBridgeRuntime.ts"; + +test("calculates uniform midpoint timestamps", () => { + assert.deepEqual(calculateFrameTimestamps(8, 4), [1, 3, 5, 7]); + assert.deepEqual(calculateFrameTimestamps(0.4, 8), [0.2]); +}); + +test("probes and extracts a local video using shell-free bounded commands", async () => { + const calls: Array<{ executable: string; args: string[]; timeoutMs: number }> = []; + const runner: VideoCommandRunner = async (executable, args, options) => { + calls.push({ executable, args: [...args], timeoutMs: options.timeoutMs }); + if (executable === "ffprobe") { + return { stdout: JSON.stringify({ format: { duration: "8.0" } }), stderr: "" }; + } + return { stdout: "", stderr: "" }; + }; + + const metadata = await probeLocalVideo("/tmp/input.mp4", { + maxDurationSeconds: 600, + runner, + timeoutMs: 5_000, + }); + const frames = await extractFramesFromLocalVideo("/tmp/input.mp4", "/tmp/frames", { + durationSeconds: metadata.durationSeconds, + frameCount: 4, + runner, + timeoutMs: 10_000, + }); + + assert.equal(metadata.durationSeconds, 8); + assert.deepEqual( + frames.map((frame) => frame.timestampSeconds), + [1, 3, 5, 7] + ); + assert.deepEqual(calls[0], { + executable: "ffprobe", + args: ["-v", "error", "-show_entries", "format=duration", "-of", "json", "/tmp/input.mp4"], + timeoutMs: 5_000, + }); + assert.equal( + calls.slice(1).every((call) => call.executable === "ffmpeg"), + true + ); + assert.equal( + calls.slice(1).every((call) => call.args.includes("-nostdin")), + true + ); + assert.equal( + calls.slice(1).every((call) => !call.args.some((arg) => arg.includes("://"))), + true + ); +}); + +test("rejects remote process inputs and videos beyond the duration bound", async () => { + const runner: VideoCommandRunner = async () => ({ + stdout: JSON.stringify({ format: { duration: "601" } }), + stderr: "private upstream details", + }); + await assert.rejects( + () => probeLocalVideo("https://example.test/video.mp4", { runner }), + /local path/ + ); + await assert.rejects( + () => probeLocalVideo("/tmp/input.mp4", { maxDurationSeconds: 600, runner }), + /maximum duration/ + ); +}); + +test("runtime status exposes sanitized versions and a sanitized unavailable reason", async () => { + resetVideoRuntimeProbeCacheForTests(); + const ready = await probeVideoRuntime({ + cacheTtlMs: 0, + runner: async (executable) => ({ + stdout: + executable === "ffmpeg" ? "ffmpeg version 6.1.1 secret" : "ffprobe version 6.1.1 secret", + stderr: "", + }), + }); + assert.deepEqual(ready, { + available: true, + ffmpegVersion: "6.1.1", + ffprobeVersion: "6.1.1", + }); + + resetVideoRuntimeProbeCacheForTests(); + const unavailable = await probeVideoRuntime({ + cacheTtlMs: 0, + runner: async () => { + throw new Error("spawn /private/operator/path ENOENT"); + }, + }); + assert.deepEqual(unavailable, { + available: false, + ffmpegVersion: null, + ffprobeVersion: null, + reason: "FFmpeg and ffprobe are not available on PATH", + }); +}); + +test("runtime probe uses its short cache instead of spawning on every status read", async () => { + resetVideoRuntimeProbeCacheForTests(); + let calls = 0; + const runner: VideoCommandRunner = async (executable) => { + calls += 1; + return { + stdout: `${executable} version 7.0`, + stderr: "", + }; + }; + + const first = await probeVideoRuntime({ cacheTtlMs: 30_000, runner }); + const second = await probeVideoRuntime({ cacheTtlMs: 30_000, runner }); + assert.deepEqual(second, first); + assert.equal(calls, 2, "one ffmpeg + one ffprobe process should serve both reads"); +}); diff --git a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts index 6954be9f18..bdbef94807 100644 --- a/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts +++ b/tests/unit/guardrails/visionBridgeHelpers.callVisionModel.test.ts @@ -337,3 +337,47 @@ test("callVisionModel fetches remote images before Anthropic requests", async () globalThis.fetch = originalFetch; } }); + +test("callVisionModel propagates an external abort to fetch and stops before fallback", async () => { + const controller = new AbortController(); + let fetchCalls = 0; + let fetchSignal: AbortSignal | null = null; + + globalThis.fetch = async (_url: URL | RequestInfo, init?: RequestInit) => { + fetchCalls += 1; + fetchSignal = init?.signal instanceof AbortSignal ? init.signal : null; + controller.abort(); + const error = new Error("private aborted request detail"); + error.name = "AbortError"; + throw error; + }; + + try { + const config: VisionModelConfig = { + model: "openai/gpt-4o-mini", + prompt: "Describe this image", + timeoutMs: 30_000, + maxImages: 10, + signal: controller.signal, + }; + + await assert.rejects( + () => + callVisionModel( + "data:image/png;base64,iVBORw0KGgo", + config, + "sk-test", + { maxFallbackAttempts: 2 }, + { + hasUsableCredentials: async (model) => + model === "openai/gpt-4o-mini" || model.startsWith("anthropic/"), + } + ), + /timed out|aborted/i + ); + assert.equal(fetchCalls, 1, "an aborted parent request must not try a fallback model"); + assert.equal(fetchSignal?.aborted, true, "the parent abort must reach the active fetch"); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/tests/unit/remote-media-fetch.test.ts b/tests/unit/remote-media-fetch.test.ts new file mode 100644 index 0000000000..bee0c11b40 --- /dev/null +++ b/tests/unit/remote-media-fetch.test.ts @@ -0,0 +1,36 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { fetchRemoteMedia } from "../../src/shared/network/remoteImageFetch.ts"; + +test("generic remote-media fetch reuses the public-only bounded download policy", async () => { + const result = await fetchRemoteMedia("https://cdn.example.test/video.mp4", { + fetchImpl: async () => + new Response(Buffer.from("video-bytes"), { + headers: { "content-type": "video/mp4" }, + }), + guard: "public-only", + lookup: async () => [{ address: "203.0.113.10", family: 4 }], + maxBytes: 1024, + }); + + assert.equal(result.buffer.toString(), "video-bytes"); + assert.equal(result.contentType, "video/mp4"); +}); + +test("generic remote-media fetch rejects private DNS answers before downloading", async () => { + let fetched = false; + await assert.rejects( + () => + fetchRemoteMedia("https://cdn.example.test/video.mp4", { + fetchImpl: async () => { + fetched = true; + return new Response("unexpected"); + }, + guard: "public-only", + lookup: async () => [{ address: "127.0.0.1", family: 4 }], + }), + /blocked private address/ + ); + assert.equal(fetched, false); +});