From d2cea0811af27a32da62cd8ae2a401747e1536e1 Mon Sep 17 00:00:00 2001 From: Diego Rodrigues de Sa e Souza Date: Mon, 24 Aug 2026 02:43:45 -0300 Subject: [PATCH] fix(video): harden result cache identity and bounds --- .../pending-video-bridge-result-cache.md | 1 + .../guardrails/modalityBridge/bridgeCache.ts | 89 +- .../guardrails/modalityBridge/bridgeStats.ts | 6 + src/lib/guardrails/videoBridge.ts | 402 ++++++-- src/lib/guardrails/videoBridgeHelpers.ts | 31 +- src/lib/guardrails/videoBridgeResultCache.ts | 232 +++++ .../guardrails/videoBridgeResultCache.test.ts | 973 ++++++++++++++++++ 7 files changed, 1623 insertions(+), 111 deletions(-) create mode 100644 changelog.d/fixes/pending-video-bridge-result-cache.md create mode 100644 src/lib/guardrails/videoBridgeResultCache.ts create mode 100644 tests/unit/guardrails/videoBridgeResultCache.test.ts diff --git a/changelog.d/fixes/pending-video-bridge-result-cache.md b/changelog.d/fixes/pending-video-bridge-result-cache.md new file mode 100644 index 0000000000..df41cd0a1f --- /dev/null +++ b/changelog.d/fixes/pending-video-bridge-result-cache.md @@ -0,0 +1 @@ +- Fix Video Bridge result caching to fingerprint protected video bytes, coalesce concurrent work, and fail open when its bounded TTL/LRU cache is unavailable or corrupt. diff --git a/src/lib/guardrails/modalityBridge/bridgeCache.ts b/src/lib/guardrails/modalityBridge/bridgeCache.ts index e707e30792..a1bdd4ede2 100644 --- a/src/lib/guardrails/modalityBridge/bridgeCache.ts +++ b/src/lib/guardrails/modalityBridge/bridgeCache.ts @@ -55,6 +55,8 @@ export function bridgeCacheKey( export interface BridgeCacheOptions { maxEntries: number; + /** Aggregate UTF-8 key/value/metadata budget; unlimited when omitted. */ + maxBytes?: number; ttlMs: number; /** Injectable clock for tests. */ now?: () => number; @@ -67,8 +69,37 @@ export interface BridgeCacheEntry { metadata?: Record; } -export class BridgeCache { - private readonly entries = new Map(); +/** Minimal fail-open store contract accepted by complete-result bridge caches. */ +export interface BridgeCacheStore { + delete(key: string): void; + getEntry(key: string): BridgeCacheEntry | undefined; + setEntry(key: string, entry: BridgeCacheEntry): void; +} + +type StoredBridgeCacheEntry = { + bytes: number; + entry: BridgeCacheEntry; + expiresAt: number; +}; + +function cacheEntryBytes(entry: BridgeCacheEntry): number { + try { + const metadata = JSON.stringify({ + metadata: entry.metadata, + producerModel: entry.producerModel, + }); + return Buffer.byteLength(entry.value, "utf8") + Buffer.byteLength(metadata, "utf8"); + } catch (error) { + console.debug("[MODALITY_BRIDGE_CACHE] Entry size calculation failed open", { + errorType: error instanceof Error ? error.name : typeof error, + }); + return Number.POSITIVE_INFINITY; + } +} + +export class BridgeCache implements BridgeCacheStore { + private readonly entries = new Map(); + private totalBytes = 0; constructor(private readonly opts: BridgeCacheOptions) {} @@ -81,7 +112,7 @@ export class BridgeCache { if (!hit) return undefined; const now = (this.opts.now ?? Date.now)(); if (hit.expiresAt <= now) { - this.entries.delete(key); + this.delete(key); return undefined; } // Map preserves insertion order — re-insert to mark as most-recently-used. @@ -96,12 +127,17 @@ export class BridgeCache { setEntry(key: string, entry: BridgeCacheEntry): void { const now = (this.opts.now ?? Date.now)(); - this.entries.delete(key); - this.entries.set(key, { entry, expiresAt: now + this.opts.ttlMs }); - while (this.entries.size > this.opts.maxEntries) { + const bytes = cacheEntryBytes(entry) + Buffer.byteLength(key, "utf8"); + const maxBytes = Math.max(0, this.opts.maxBytes ?? Number.POSITIVE_INFINITY); + const maxEntries = Math.max(0, Math.floor(this.opts.maxEntries)); + this.delete(key); + if (!Number.isFinite(bytes) || bytes > maxBytes || maxEntries === 0) return; + this.entries.set(key, { bytes, entry, expiresAt: now + this.opts.ttlMs }); + this.totalBytes += bytes; + while (this.entries.size > maxEntries || this.totalBytes > maxBytes) { const oldest = this.entries.keys().next().value; if (oldest === undefined) break; - this.entries.delete(oldest); + this.delete(oldest); } } @@ -109,21 +145,52 @@ export class BridgeCache { return this.entries.size; } + /** Current aggregate UTF-8 bytes retained by this cache. */ + get bytes(): number { + return this.totalBytes; + } + delete(key: string): void { + const existing = this.entries.get(key); + if (existing) this.totalBytes = Math.max(0, this.totalBytes - existing.bytes); this.entries.delete(key); } clear(): void { this.entries.clear(); + this.totalBytes = 0; } } /** Process-wide singleton used by the bridges; recreated when config changes. */ -let shared: { cache: BridgeCache; ttlMs: number; maxEntries: number } | null = null; +let shared: { cache: BridgeCache; ttlMs: number; maxBytes: number; maxEntries: number } | null = + null; -export function getSharedBridgeCache(ttlMs: number, maxEntries: number): BridgeCache { - if (!shared || shared.ttlMs !== ttlMs || shared.maxEntries !== maxEntries) { - shared = { cache: new BridgeCache({ maxEntries, ttlMs }), ttlMs, maxEntries }; +/** + * Resolve the process-wide bridge cache, recreating it when any bound changes. + * + * @param ttlMs - Entry lifetime in milliseconds. + * @param maxEntries - Maximum retained entry count. + * @param maxBytes - Aggregate UTF-8 storage budget. + * @returns The process-wide cache for these exact bounds. + */ +export function getSharedBridgeCache( + ttlMs: number, + maxEntries: number, + maxBytes = Number.POSITIVE_INFINITY +): BridgeCache { + if ( + !shared || + shared.ttlMs !== ttlMs || + shared.maxEntries !== maxEntries || + shared.maxBytes !== maxBytes + ) { + shared = { + cache: new BridgeCache({ maxBytes, maxEntries, ttlMs }), + ttlMs, + maxBytes, + maxEntries, + }; } return shared.cache; } diff --git a/src/lib/guardrails/modalityBridge/bridgeStats.ts b/src/lib/guardrails/modalityBridge/bridgeStats.ts index 719a0fd3cd..e3d487fd59 100644 --- a/src/lib/guardrails/modalityBridge/bridgeStats.ts +++ b/src/lib/guardrails/modalityBridge/bridgeStats.ts @@ -19,6 +19,8 @@ export interface BridgeModalityStats { resultCacheBytes: number; resultCacheHits: number; resultCacheLatencyMs: number; + /** Requests that joined an in-flight complete result instead of hitting the persistent cache. */ + resultSingleflightCoalesced: number; failures: number; /** Audio/video fusion runs (video bridge only; 0 for other modalities). */ fusionRuns: number; @@ -47,6 +49,7 @@ function emptyStats(): BridgeModalityStats { resultCacheBytes: 0, resultCacheHits: 0, resultCacheLatencyMs: 0, + resultSingleflightCoalesced: 0, failures: 0, fusionRuns: 0, fusionPartials: 0, @@ -69,6 +72,8 @@ export function recordBridgeUse( resultCacheBytes?: number; resultCacheHit?: boolean; resultCacheLatencyMs?: number; + /** True only when this request joined existing in-flight result work. */ + resultSingleflightCoalesced?: boolean; } = {} ): void { const s = stats[kind]; @@ -104,6 +109,7 @@ export function recordBridgeUse( s.resultCacheLatencyMs += Math.max(0, opts.resultCacheLatencyMs); } } + if (opts.resultSingleflightCoalesced) s.resultSingleflightCoalesced += 1; if (typeof opts.latencyMs === "number" && Number.isFinite(opts.latencyMs)) { s.totalLatencyMs += Math.max(0, opts.latencyMs); s.latencySamples += 1; diff --git a/src/lib/guardrails/videoBridge.ts b/src/lib/guardrails/videoBridge.ts index fd6316d696..b41dbcaf4d 100644 --- a/src/lib/guardrails/videoBridge.ts +++ b/src/lib/guardrails/videoBridge.ts @@ -1,3 +1,5 @@ +import { createHash } from "node:crypto"; + import { fetch as undiciFetch } from "undici"; import { getSettings as defaultGetSettings } from "@/lib/db/settings"; @@ -8,18 +10,34 @@ import { } from "@/shared/constants/modalityBridgeDefaults"; import { BaseGuardrail, type GuardrailContext, type GuardrailResult } from "./base"; -import { bridgeCacheKey, getSharedBridgeCacheFor } from "./modalityBridge/bridgeCache"; +import { + bridgeCacheKey, + getSharedBridgeCacheFor, + type BridgeCacheEntry, + type BridgeCacheStore, +} from "./modalityBridge/bridgeCache"; import { recordBridgeUse } from "./modalityBridge/bridgeStats"; import { describeVideoPart as defaultDescribeVideoPart, extractVideoParts, formatVideoTimestamp, + loadVideoPartBytes, replaceVideoParts, + VIDEO_BRIDGE_MAX_BYTES, type DescribeVideoDependencies, type DescribedVideo, type VideoFusionTelemetry, type VideoPart, } from "./videoBridgeHelpers"; +import { + getSharedVideoResultCacheFor, + runVideoDownloadSingleflight, + runVideoResultSingleflight, + safeDeleteCacheEntry, + safeGetCacheEntry, + safeSetCacheEntry, + videoBridgeAbortError, +} from "./videoBridgeResultCache"; import { callVisionModel as defaultCallVisionModel, type VisionModelConfig, @@ -48,9 +66,59 @@ function safeTranscriptFingerprint(value: unknown): string { } } -const VIDEO_BRIDGE_RESULT_CACHE_VERSION = "v2"; +function waitForVideoBridgePromise(promise: Promise, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(videoBridgeAbortError()); + return new Promise((resolve, reject) => { + let completed = false; + const finish = (callback: () => void): void => { + if (completed) return; + completed = true; + signal.removeEventListener("abort", onAbort); + callback(); + }; + const onAbort = (): void => finish(() => reject(videoBridgeAbortError())); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + promise.then( + (value) => finish(() => resolve(value)), + (error: unknown) => finish(() => reject(error)) + ); + }); +} + +const VIDEO_BRIDGE_RESULT_CACHE_VERSION = "v3"; const VIDEO_BRIDGE_RESULT_CACHE_POLICY = "default"; -const VIDEO_BRIDGE_RESULT_CACHE_KEY_KIND = "video-result-v2"; +const VIDEO_BRIDGE_RESULT_CACHE_KEY_KIND = "video-result-v3"; +const VIDEO_BRIDGE_DOWNLOAD_FLIGHT_VERSION = "v1"; + +function buildVideoDownloadFlightKey( + part: VideoPart, + context: GuardrailContext, + maxBytes: number, + timeoutMs: number +): string { + const rawPrincipalId = context.apiKeyInfo?.id; + const principalId = + typeof rawPrincipalId === "string" || typeof rawPrincipalId === "number" + ? String(rawPrincipalId) + : "local"; + const canonicalIdentity = JSON.stringify({ + container: part.container, + endpoint: context.endpoint ?? null, + maxBytes, + method: context.method ?? null, + model: context.model ?? null, + principalId, + provider: context.provider ?? null, + ref: part.ref, + shape: part.shape, + sourceFormat: context.sourceFormat ?? null, + targetFormat: context.targetFormat ?? null, + timeoutMs, + version: VIDEO_BRIDGE_DOWNLOAD_FLIGHT_VERSION, + }); + return `video-download:${createHash("sha256").update(canonicalIdentity).digest("hex")}`; +} interface VideoResultCacheMetadata { cacheVersion: string; @@ -78,6 +146,74 @@ interface VideoResultCacheMetadata { modelUsed: string; } +type VideoResultCacheIdentity = Pick< + VideoResultCacheMetadata, + | "cacheVersion" + | "extractorVersion" + | "frameCount" + | "maxVideos" + | "model" + | "policyVersion" + | "prompt" + | "strategy" +>; + +const VIDEO_RESULT_CACHE_IDENTITY_KEYS: readonly (keyof VideoResultCacheIdentity)[] = [ + "cacheVersion", + "extractorVersion", + "frameCount", + "maxVideos", + "model", + "policyVersion", + "prompt", + "strategy", +]; + +function createVideoResultCacheIdentity( + runtime: ReturnType, + visionRuntime: ReturnType, + model: string +): VideoResultCacheIdentity { + return { + cacheVersion: VIDEO_BRIDGE_RESULT_CACHE_VERSION, + extractorVersion: VIDEO_BRIDGE_RESULT_CACHE_VERSION, + frameCount: runtime.frameCount, + maxVideos: runtime.maxVideos, + model, + policyVersion: VIDEO_BRIDGE_RESULT_CACHE_POLICY, + prompt: visionRuntime.prompt, + strategy: runtime.samplingPolicy, + }; +} + +function buildVideoResultCacheKey( + contentFingerprint: string, + identity: VideoResultCacheIdentity, + part: VideoPart +): string { + return bridgeCacheKey(contentFingerprint, identity.prompt, identity.model, { + kind: VIDEO_BRIDGE_RESULT_CACHE_KEY_KIND, + extractorVersion: identity.extractorVersion, + policyVersion: identity.policyVersion, + strategy: identity.strategy, + frameCount: identity.frameCount, + maxVideos: identity.maxVideos, + focusEndSeconds: part.focusWindow?.endSeconds ?? null, + focusStartSeconds: part.focusWindow?.startSeconds ?? null, + transcript: safeTranscriptFingerprint(part.transcript), + audioTranscript: safeTranscriptFingerprint(part.audioTranscript), + contactSheet: part.contactSheet ?? false, + version: identity.cacheVersion, + }); +} + +function matchesVideoResultCacheIdentity( + metadata: VideoResultCacheMetadata, + identity: VideoResultCacheIdentity +): boolean { + return VIDEO_RESULT_CACHE_IDENTITY_KEYS.every((key) => metadata[key] === identity[key]); +} + function isFusionTelemetry(value: unknown): value is VideoFusionTelemetry { if (!value || typeof value !== "object") return false; const record = value as Record; @@ -102,6 +238,8 @@ export interface VideoBridgeDependencies { getCapabilities?: (model: string) => { supportsVideo: boolean | null }; describePart?: (part: VideoPart) => Promise; extractFrames?: DescribeVideoDependencies["extractFrames"]; + fetchRemote?: DescribeVideoDependencies["fetchRemote"]; + resultCache?: BridgeCacheStore; selectVisionModel?: (fixedModel?: string) => Promise; callVisionModel?: ( imageDataUri: string, @@ -110,9 +248,46 @@ export interface VideoBridgeDependencies { ) => Promise; } -function isVideoResultCacheMetadata(value: unknown): value is VideoResultCacheMetadata { +function isFiniteNonNegativeNumber(value: unknown): value is number { + return typeof value === "number" && Number.isFinite(value) && value >= 0; +} + +function isFiniteNonNegativeInteger(value: unknown): value is number { + return isFiniteNonNegativeNumber(value) && Number.isInteger(value); +} + +function isVideoResultCacheMetadata( + value: unknown, + expectedCacheBytes: number +): value is VideoResultCacheMetadata { if (!value || typeof value !== "object") return false; const record = value as Record; + if ( + !isFiniteNonNegativeInteger(record.framesRequested) || + !isFiniteNonNegativeInteger(record.framesExtracted) || + !isFiniteNonNegativeInteger(record.framesUsed) || + record.framesExtracted > record.framesRequested || + record.framesUsed > record.framesExtracted + ) { + return false; + } + const dedupDropped = record.dedupDropped ?? 0; + if ( + !isFiniteNonNegativeInteger(dedupDropped) || + record.framesUsed + dedupDropped > record.framesExtracted + ) { + return false; + } + if ( + (record.focusStartSeconds !== undefined && + !isFiniteNonNegativeNumber(record.focusStartSeconds)) || + (record.focusEndSeconds !== undefined && !isFiniteNonNegativeNumber(record.focusEndSeconds)) || + (typeof record.focusStartSeconds === "number" && + typeof record.focusEndSeconds === "number" && + record.focusStartSeconds > record.focusEndSeconds) + ) { + return false; + } return ( typeof record.cacheVersion === "string" && typeof record.policyVersion === "string" && @@ -120,18 +295,14 @@ function isVideoResultCacheMetadata(value: unknown): value is VideoResultCacheMe typeof record.strategy === "string" && typeof record.model === "string" && typeof record.prompt === "string" && - typeof record.frameCount === "number" && - typeof record.maxVideos === "number" && - typeof record.durationSeconds === "number" && - typeof record.framesRequested === "number" && - typeof record.framesExtracted === "number" && - typeof record.framesUsed === "number" && - (record.dedupDropped === undefined || - (typeof record.dedupDropped === "number" && record.dedupDropped >= 0)) && - typeof record.cacheBytes === "number" && + isFiniteNonNegativeInteger(record.frameCount) && + isFiniteNonNegativeInteger(record.maxVideos) && + isFiniteNonNegativeNumber(record.durationSeconds) && + isFiniteNonNegativeInteger(record.cacheBytes) && + record.cacheBytes === expectedCacheBytes && typeof record.modelUsed === "string" && (record.samplingCandidateCount === undefined || - (typeof record.samplingCandidateCount === "number" && record.samplingCandidateCount >= 0)) && + isFiniteNonNegativeInteger(record.samplingCandidateCount)) && (record.samplingPolicyEffective === undefined || record.samplingPolicyEffective === "uniform" || record.samplingPolicyEffective === "scene_aware" || @@ -141,12 +312,22 @@ function isVideoResultCacheMetadata(value: unknown): value is VideoResultCacheMe record.samplingPolicyRequested === "scene_aware" || record.samplingPolicyRequested === "segment_aware") && (record.transcriptCuesApplied === undefined || - (typeof record.transcriptCuesApplied === "number" && record.transcriptCuesApplied >= 0)) && + isFiniteNonNegativeInteger(record.transcriptCuesApplied)) && (record.contactSheetUsed === undefined || typeof record.contactSheetUsed === "boolean") && (record.fusion === undefined || isFusionTelemetry(record.fusion)) ); } +function isVideoResultCacheEntry( + entry: BridgeCacheEntry +): entry is BridgeCacheEntry & { metadata: VideoResultCacheMetadata; value: string } { + if (typeof entry.value !== "string") return false; + return ( + (entry.producerModel === undefined || typeof entry.producerModel === "string") && + isVideoResultCacheMetadata(entry.metadata, Buffer.byteLength(entry.value, "utf8")) + ); +} + export class VideoBridgeGuardrail extends BaseGuardrail { name = "video-bridge"; priority = 7; @@ -188,7 +369,9 @@ export class VideoBridgeGuardrail extends BaseGuardrail { const visionRuntime = resolveVisionBridgeRuntimeSettings(persisted); const configuredModel = runtime.model.trim() || visionRuntime.model.trim(); const routingPlanModel = configuredModel || "auto"; - const cache = runtime.cacheEnabled ? getSharedBridgeCacheFor(runtime) : null; + const cache = runtime.cacheEnabled + ? (this.deps.resultCache ?? getSharedVideoResultCacheFor(runtime)) + : null; const successfulModels = new Set(); let selectedModelPromise: Promise | null = null; const selectVideoModel = (): Promise => { @@ -231,37 +414,62 @@ export class VideoBridgeGuardrail extends BaseGuardrail { if (context.signal?.aborted) throw new Error("Video Bridge processing was aborted"); const part = attemptedParts[index]; const attemptStartedAt = Date.now(); + const timeoutController = new AbortController(); + const attemptTimeout = setTimeout(() => timeoutController.abort(), runtime.timeoutMs); + const attemptSignal = context.signal + ? AbortSignal.any([context.signal, timeoutController.signal]) + : timeoutController.signal; try { - const selectedModel = await selectVideoModel(); - const resultCacheKey = + const selectedModel = await waitForVideoBridgePromise(selectVideoModel(), attemptSignal); + if (attemptSignal.aborted) throw videoBridgeAbortError(); + const shouldLoadVideoBytes = + Boolean(selectedModel) && + (Boolean(cache) || (part.ref.startsWith("https://") && !this.deps.describePart)); + const videoBytes = shouldLoadVideoBytes + ? part.ref.startsWith("https://") + ? await runVideoDownloadSingleflight( + buildVideoDownloadFlightKey( + part, + context, + VIDEO_BRIDGE_MAX_BYTES, + runtime.timeoutMs + ), + attemptSignal, + (downloadSignal) => + loadVideoPartBytes( + part, + VIDEO_BRIDGE_MAX_BYTES, + runtime.timeoutMs, + downloadSignal, + { fetchRemote: this.deps.fetchRemote } + ) + ) + : await loadVideoPartBytes( + part, + VIDEO_BRIDGE_MAX_BYTES, + runtime.timeoutMs, + attemptSignal, + { fetchRemote: this.deps.fetchRemote } + ) + : null; + const contentFingerprint = + cache && videoBytes + ? `sha256:${createHash("sha256").update(videoBytes).digest("hex")}` + : part.ref; + const resultCacheIdentity = cache && selectedModel - ? bridgeCacheKey(part.ref, visionRuntime.prompt, selectedModel, { - kind: VIDEO_BRIDGE_RESULT_CACHE_KEY_KIND, - extractorVersion: VIDEO_BRIDGE_RESULT_CACHE_VERSION, - policyVersion: VIDEO_BRIDGE_RESULT_CACHE_POLICY, - strategy: runtime.samplingPolicy, - frameCount: runtime.frameCount, - maxVideos: runtime.maxVideos, - focusEndSeconds: part.focusWindow?.endSeconds ?? null, - focusStartSeconds: part.focusWindow?.startSeconds ?? null, - transcript: safeTranscriptFingerprint(part.transcript), - audioTranscript: safeTranscriptFingerprint(part.audioTranscript), - contactSheet: part.contactSheet ?? false, - version: VIDEO_BRIDGE_RESULT_CACHE_VERSION, - }) + ? createVideoResultCacheIdentity(runtime, visionRuntime, selectedModel) : null; - const cachedResult = resultCacheKey ? cache.getEntry(resultCacheKey) : null; - if (cachedResult && isVideoResultCacheMetadata(cachedResult.metadata)) { + const resultCacheKey = resultCacheIdentity + ? buildVideoResultCacheKey(contentFingerprint, resultCacheIdentity, part) + : null; + const cachedResult = resultCacheKey + ? safeGetCacheEntry(cache, resultCacheKey, context.log) + : null; + if (cachedResult && isVideoResultCacheEntry(cachedResult)) { const meta = cachedResult.metadata; const matchPolicy = - meta.cacheVersion === VIDEO_BRIDGE_RESULT_CACHE_VERSION && - meta.policyVersion === VIDEO_BRIDGE_RESULT_CACHE_POLICY && - meta.extractorVersion === VIDEO_BRIDGE_RESULT_CACHE_VERSION && - meta.strategy === runtime.samplingPolicy && - meta.frameCount === runtime.frameCount && - meta.maxVideos === runtime.maxVideos && - meta.model === selectedModel && - meta.prompt === visionRuntime.prompt; + resultCacheIdentity && matchesVideoResultCacheIdentity(meta, resultCacheIdentity); if (matchPolicy) { const elapsed = Date.now() - attemptStartedAt; descriptions.push(cachedResult.value); @@ -299,21 +507,60 @@ export class VideoBridgeGuardrail extends BaseGuardrail { }); continue; } - cache.delete(resultCacheKey); + safeDeleteCacheEntry(cache, resultCacheKey, context.log); } else if (cachedResult) { - cache.delete(resultCacheKey); + safeDeleteCacheEntry(cache, resultCacheKey, context.log); } - const cacheStartAt = Date.now(); - const described = this.deps.describePart - ? await this.deps.describePart(part) - : await this.describeWithVisionModel( - part, - runtime, - visionRuntime, - selectedModel, - context.signal + const describeAndCache = async (processingSignal: AbortSignal) => { + const described = this.deps.describePart + ? await this.deps.describePart(part) + : await this.describeWithVisionModel( + part, + runtime, + visionRuntime, + selectedModel, + processingSignal, + videoBytes ?? undefined + ); + if (processingSignal.aborted) throw videoBridgeAbortError(); + const resultCacheBytes = Buffer.byteLength(described.description, "utf8"); + if (resultCacheKey && resultCacheIdentity) { + safeSetCacheEntry( + cache, + resultCacheKey, + { + value: described.description, + producerModel: described.modelUsed ?? resultCacheIdentity.model, + metadata: { + ...resultCacheIdentity, + durationSeconds: described.durationSeconds, + framesRequested: described.framesRequested, + framesExtracted: described.framesExtracted ?? described.framesUsed, + framesUsed: described.framesUsed, + dedupDropped: described.dedupDropped ?? 0, + focusEndSeconds: described.focusWindow?.endSeconds, + focusStartSeconds: described.focusWindow?.startSeconds, + cacheBytes: resultCacheBytes, + modelUsed: described.modelUsed ?? resultCacheIdentity.model, + samplingCandidateCount: described.sampling?.candidateCount ?? 0, + samplingPolicyEffective: described.sampling?.policyEffective ?? "uniform", + samplingPolicyRequested: + described.sampling?.policyRequested ?? runtime.samplingPolicy, + transcriptCuesApplied: described.transcriptCues?.length ?? 0, + contactSheetUsed: described.contactSheetUsed ?? false, + ...(described.fusion ? { fusion: described.fusion } : {}), + }, + }, + context.log ); - if (context.signal?.aborted) throw new Error("Video Bridge processing was aborted"); + } + return described; + }; + const resolved = + resultCacheKey && selectedModel + ? await runVideoResultSingleflight(resultCacheKey, attemptSignal, describeAndCache) + : { coalesced: false, value: await describeAndCache(attemptSignal) }; + const described = resolved.value; if (described.modelUsed) successfulModels.add(described.modelUsed); const videoCacheHits = described.cacheHits ?? 0; const processingLatencyMs = Date.now() - attemptStartedAt; @@ -336,46 +583,12 @@ export class VideoBridgeGuardrail extends BaseGuardrail { } totalCacheHits += videoCacheHits; if (resultCacheKey && selectedModel) { - const resultCacheBytes = Buffer.byteLength(described.description, "utf8"); - const cacheLatencyMs = Date.now() - cacheStartAt; - cache.setEntry(resultCacheKey, { - value: described.description, - producerModel: described.modelUsed ?? selectedModel, - metadata: { - cacheVersion: VIDEO_BRIDGE_RESULT_CACHE_VERSION, - policyVersion: VIDEO_BRIDGE_RESULT_CACHE_POLICY, - extractorVersion: VIDEO_BRIDGE_RESULT_CACHE_VERSION, - strategy: runtime.samplingPolicy, - model: selectedModel, - prompt: visionRuntime.prompt, - frameCount: runtime.frameCount, - maxVideos: runtime.maxVideos, - durationSeconds: described.durationSeconds, - framesRequested: described.framesRequested, - framesExtracted: described.framesExtracted ?? described.framesUsed, - framesUsed: described.framesUsed, - dedupDropped: described.dedupDropped ?? 0, - focusEndSeconds: described.focusWindow?.endSeconds, - focusStartSeconds: described.focusWindow?.startSeconds, - cacheBytes: resultCacheBytes, - modelUsed: described.modelUsed ?? selectedModel, - samplingCandidateCount: described.sampling?.candidateCount ?? 0, - samplingPolicyEffective: described.sampling?.policyEffective ?? "uniform", - samplingPolicyRequested: - described.sampling?.policyRequested ?? runtime.samplingPolicy, - transcriptCuesApplied: described.transcriptCues?.length ?? 0, - contactSheetUsed: described.contactSheetUsed ?? false, - ...(described.fusion ? { fusion: described.fusion } : {}), - }, - }); recordBridgeUse("video", { cacheHits: videoCacheHits, fusionRun: Boolean(described.fusion), fusionPartial: described.fusion?.partial ?? false, latencyMs: processingLatencyMs, - resultCacheBytes, - resultCacheHit: false, - resultCacheLatencyMs: cacheLatencyMs, + resultSingleflightCoalesced: resolved.coalesced, }); } else { recordBridgeUse("video", { @@ -408,6 +621,8 @@ export class VideoBridgeGuardrail extends BaseGuardrail { ? `[Video ${index + 1}]: (unavailable — video could not be described)` : null ); + } finally { + clearTimeout(attemptTimeout); } } @@ -457,7 +672,8 @@ export class VideoBridgeGuardrail extends BaseGuardrail { runtime: ReturnType, visionRuntime: ReturnType, selectedModel: string | null, - signal?: AbortSignal + signal?: AbortSignal, + preloadedBytes?: Uint8Array ): Promise { if (!selectedModel) { throw new Error("No vision-capable provider connected for Video Bridge"); @@ -503,7 +719,11 @@ export class VideoBridgeGuardrail extends BaseGuardrail { if (key && cache) cache.setEntry(key, { value: caption, producerModel }); return caption; }, - { extractFrames: this.deps.extractFrames } + { + extractFrames: this.deps.extractFrames, + fetchRemote: this.deps.fetchRemote, + }, + preloadedBytes ); return { ...described, diff --git a/src/lib/guardrails/videoBridgeHelpers.ts b/src/lib/guardrails/videoBridgeHelpers.ts index eba1d70ba2..c868dd7296 100644 --- a/src/lib/guardrails/videoBridgeHelpers.ts +++ b/src/lib/guardrails/videoBridgeHelpers.ts @@ -372,7 +372,18 @@ export function decodeVideoDataUri( return decode(normalized); } -async function loadVideoBytes( +/** + * Load protected video bytes from an inline data URI or SSRF-guarded HTTPS source. + * + * @param part - Extracted request video part. + * @param maxBytes - Maximum accepted decoded/downloaded size. + * @param timeoutMs - Download deadline passed to the protected fetch boundary. + * @param signal - Caller abort/deadline signal. + * @param deps - Injectable external download boundary. + * @returns Validated video bytes suitable for hashing and extraction. + * @throws When the source, size, deadline, or abort policy rejects the input. + */ +export async function loadVideoPartBytes( part: VideoPart, maxBytes: number, timeoutMs: number, @@ -427,7 +438,8 @@ export async function describeVideoPart( timestampSeconds: number, signal: AbortSignal ) => Promise, - deps: DescribeVideoDependencies = {} + deps: DescribeVideoDependencies = {}, + preloadedBytes?: Uint8Array ): Promise { const timeoutController = new AbortController(); const timeout = setTimeout(() => timeoutController.abort(), options.timeoutMs); @@ -435,13 +447,14 @@ export async function describeVideoPart( ? AbortSignal.any([options.signal, timeoutController.signal]) : timeoutController.signal; try { - const bytes = await loadVideoBytes( - part, - options.maxBytes ?? VIDEO_BRIDGE_MAX_BYTES, - options.timeoutMs, - signal, - deps - ); + const maxBytes = options.maxBytes ?? VIDEO_BRIDGE_MAX_BYTES; + const bytes = preloadedBytes + ? Buffer.isBuffer(preloadedBytes) + ? preloadedBytes + : Buffer.from(preloadedBytes) + : await loadVideoPartBytes(part, maxBytes, options.timeoutMs, signal, deps); + if (signal.aborted) throw new Error("Video Bridge processing timed out or was aborted"); + if (bytes.byteLength > maxBytes) throw new Error("Video exceeds the maximum size"); const extractFrames = deps.extractFrames ?? extractVideoFramesViaBroker; const extracted = await extractFrames(bytes, { focusWindow: options.focusWindow, diff --git a/src/lib/guardrails/videoBridgeResultCache.ts b/src/lib/guardrails/videoBridgeResultCache.ts new file mode 100644 index 0000000000..87e8fa123e --- /dev/null +++ b/src/lib/guardrails/videoBridgeResultCache.ts @@ -0,0 +1,232 @@ +import type { VideoBridgeRuntimeSettings } from "@/shared/constants/modalityBridgeDefaults"; + +import { + BridgeCache, + type BridgeCacheEntry, + type BridgeCacheStore, +} from "./modalityBridge/bridgeCache"; +import type { GuardrailContext } from "./base"; + +/** Aggregate in-memory budget for complete Video Bridge results. */ +export const VIDEO_RESULT_CACHE_MAX_BYTES = 16 * 1024 * 1024; + +let sharedResultCache: { cache: BridgeCache; maxEntries: number; ttlMs: number } | null = null; + +/** + * Resolve the process-wide complete-result cache for Video Bridge settings. + * + * @param settings - Runtime TTL and entry-count bounds. + * @returns A cache isolated from the frame/caption bridge cache. + */ +export function getSharedVideoResultCacheFor( + settings: Pick +): BridgeCache { + const ttlMs = settings.cacheTtlMinutes * 60_000; + if ( + !sharedResultCache || + sharedResultCache.ttlMs !== ttlMs || + sharedResultCache.maxEntries !== settings.cacheMaxEntries + ) { + sharedResultCache = { + cache: new BridgeCache({ + maxBytes: VIDEO_RESULT_CACHE_MAX_BYTES, + maxEntries: settings.cacheMaxEntries, + ttlMs, + }), + maxEntries: settings.cacheMaxEntries, + ttlMs, + }; + } + return sharedResultCache.cache; +} + +interface VideoFlight { + controller: AbortController; + promise: Promise; + settled: boolean; + waiters: number; +} + +const videoDownloadFlights = new Map(); +const videoResultFlights = new Map(); + +/** + * Build the canonical abort error used by Video Bridge waiters. + * + * @returns A sanitized abort error safe to propagate through the guardrail. + */ +export function videoBridgeAbortError(): Error { + return new Error("Video Bridge processing was aborted"); +} + +function waitForVideoFlight(flight: VideoFlight, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(videoBridgeAbortError()); + return new Promise((resolve, reject) => { + let completed = false; + const finish = (callback: () => void): void => { + if (completed) return; + completed = true; + signal.removeEventListener("abort", onAbort); + callback(); + }; + const onAbort = (): void => finish(() => reject(videoBridgeAbortError())); + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + (flight.promise as Promise).then( + (value) => finish(() => resolve(value)), + (error: unknown) => finish(() => reject(error)) + ); + }); +} + +async function runVideoSingleflight( + flights: Map, + key: string, + signal: AbortSignal, + operation: (signal: AbortSignal) => Promise +): Promise<{ coalesced: boolean; value: T }> { + let flight = flights.get(key); + const coalesced = Boolean(flight); + if (!flight) { + const controller = new AbortController(); + flight = { + controller, + promise: Promise.resolve().then(() => operation(controller.signal)), + settled: false, + waiters: 0, + }; + const createdFlight = flight; + flights.set(key, createdFlight); + createdFlight.promise.then( + () => { + createdFlight.settled = true; + if (flights.get(key) === createdFlight) flights.delete(key); + }, + () => { + createdFlight.settled = true; + if (flights.get(key) === createdFlight) flights.delete(key); + } + ); + } + flight.waiters += 1; + try { + return { coalesced, value: await waitForVideoFlight(flight, signal) }; + } finally { + flight.waiters = Math.max(0, flight.waiters - 1); + if (flight.waiters === 0 && !flight.settled) { + flight.controller.abort(); + if (flights.get(key) === flight) flights.delete(key); + } + } +} + +/** + * Coalesce only concurrent protected downloads and release the Buffer after the flight settles. + * + * @param key - Hashed remote-part and request-isolation identity. + * @param signal - Abort signal for this waiter only. + * @param operation - Protected downloader invoked once with a shared producer signal. + * @returns The downloaded value shared by active waiters; it is never retained after settlement. + * @throws When this waiter aborts or the shared producer rejects. + */ +export async function runVideoDownloadSingleflight( + key: string, + signal: AbortSignal, + operation: (signal: AbortSignal) => Promise +): Promise { + return (await runVideoSingleflight(videoDownloadFlights, key, signal, operation)).value; +} + +/** + * Coalesce identical complete-result work while preserving each waiter's abort signal. + * + * @param key - Complete-result cache key. + * @param signal - Abort signal for this waiter only. + * @param operation - Producer invoked once with a shared signal. + * @returns The produced value and whether this waiter joined existing work. + * @throws When this waiter aborts or the shared producer rejects. + */ +export async function runVideoResultSingleflight( + key: string, + signal: AbortSignal, + operation: (signal: AbortSignal) => Promise +): Promise<{ coalesced: boolean; value: T }> { + return runVideoSingleflight(videoResultFlights, key, signal, operation); +} + +type ResultCacheOperation = "delete" | "read" | "write"; + +function logCacheFailure( + log: GuardrailContext["log"], + operation: ResultCacheOperation, + error: unknown +): void { + const message = `Video result cache ${operation} failed open`; + const meta = { errorType: error instanceof Error ? error.name : typeof error }; + if (log?.debug) { + log.debug("VIDEO_BRIDGE_CACHE", message, meta); + } else { + console.debug(`[VIDEO_BRIDGE_CACHE] ${message}`, meta); + } +} + +/** + * Read a complete-result cache entry without allowing cache failure to break video processing. + * + * @param cache - Cache implementation, including caller-supplied adapters. + * @param key - Complete-result key. + * @param log - Optional request logger for fail-open diagnostics. + * @returns The entry, or `undefined` for misses and cache failures. + */ +export function safeGetCacheEntry( + cache: BridgeCacheStore, + key: string, + log?: GuardrailContext["log"] +): BridgeCacheEntry | undefined { + try { + return cache.getEntry(key); + } catch (error) { + logCacheFailure(log, "read", error); + return undefined; + } +} + +/** + * Delete an invalid complete-result entry without breaking video processing. + * + * @param cache - Cache implementation, including caller-supplied adapters. + * @param key - Complete-result key. + * @param log - Optional request logger for fail-open diagnostics. + */ +export function safeDeleteCacheEntry( + cache: BridgeCacheStore, + key: string, + log?: GuardrailContext["log"] +): void { + try { + cache.delete(key); + } catch (error) { + logCacheFailure(log, "delete", error); + } +} + +/** + * Store a computed complete result without allowing cache failure to discard valid output. + * + * @param cache - Cache implementation, including caller-supplied adapters. + * @param key - Complete-result key. + * @param entry - Valid computed description and metadata. + * @param log - Optional request logger for fail-open diagnostics. + */ +export function safeSetCacheEntry( + cache: BridgeCacheStore, + key: string, + entry: BridgeCacheEntry, + log?: GuardrailContext["log"] +): void { + try { + cache.setEntry(key, entry); + } catch (error) { + logCacheFailure(log, "write", error); + } +} diff --git a/tests/unit/guardrails/videoBridgeResultCache.test.ts b/tests/unit/guardrails/videoBridgeResultCache.test.ts new file mode 100644 index 0000000000..c430b69b77 --- /dev/null +++ b/tests/unit/guardrails/videoBridgeResultCache.test.ts @@ -0,0 +1,973 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { VideoBridgeGuardrail } from "../../../src/lib/guardrails/videoBridge.ts"; +import { BridgeCache } from "../../../src/lib/guardrails/modalityBridge/bridgeCache.ts"; +import { getBridgeStats } from "../../../src/lib/guardrails/modalityBridge/bridgeStats.ts"; +import { + getSharedVideoResultCacheFor, + runVideoResultSingleflight, + VIDEO_RESULT_CACHE_MAX_BYTES, +} from "../../../src/lib/guardrails/videoBridgeResultCache.ts"; + +const remoteVideoPayload = () => ({ + model: "example/text-only", + messages: [ + { + role: "user", + content: [ + { + type: "input_video", + video_url: "https://example.test/fu01-content.mp4", + }, + ], + }, + ], +}); + +function resultText(result: Awaited>): string { + const body = result.modifiedPayload as ReturnType; + return String((body.messages[0].content[0] as { text?: string }).text); +} + +test("result cache fingerprints protected bytes instead of trusting a stable HTTPS URL", async () => { + const contents = [Buffer.from("video-a"), Buffer.from("video-b"), Buffer.from("video-b")]; + let fetchedContent = ""; + let fetchCalls = 0; + let describeCalls = 0; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeCacheMaxEntries: 17, + modalityBridgeCacheTtlMinutes: 57, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 content fingerprint", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + fetchRemote: async (url: string) => { + const buffer = contents[Math.min(fetchCalls, contents.length - 1)]; + fetchCalls += 1; + fetchedContent = buffer.toString("utf8"); + return { buffer, contentType: "video/mp4", url }; + }, + describePart: async () => { + describeCalls += 1; + return { + description: `[Video description: ${fetchedContent}]`, + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + + const first = await bridge.preCall(remoteVideoPayload(), {}); + const second = await bridge.preCall(remoteVideoPayload(), {}); + const third = await bridge.preCall(remoteVideoPayload(), {}); + + assert.match(resultText(first), /video-a/); + assert.match(resultText(second), /video-b/); + assert.match(resultText(third), /video-b/); + assert.equal(fetchCalls, 3, "each HTTPS lookup must authenticate the current protected bytes"); + assert.equal(describeCalls, 2, "only identical content may reuse the complete result"); +}); + +test("concurrent requests singleflight extraction and captions for identical content", async () => { + let extractCalls = 0; + let captionCalls = 0; + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeCacheMaxEntries: 19, + modalityBridgeCacheTtlMinutes: 59, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 singleflight", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + extractFrames: async () => { + extractCalls += 1; + await new Promise((resolve) => setTimeout(resolve, 25)); + return { + durationSeconds: 1, + frames: [{ dataUri: "data:image/jpeg;base64,U0lOR0xFRkxJR0hU", timestampSeconds: 0.5 }], + }; + }, + callVisionModel: async () => { + captionCalls += 1; + return "one shared observation"; + }, + }, + }); + const payload = () => ({ + model: "example/text-only", + messages: [ + { + role: "user", + content: [ + { + type: "input_video", + video_url: "data:video/mp4;base64,U0lOR0xFRkxJR0hULVZJREVP", + }, + ], + }, + ], + }); + + const beforeStats = getBridgeStats().video; + const [first, second] = await Promise.all([ + bridge.preCall(payload(), {}), + bridge.preCall(payload(), {}), + ]); + const afterCoalesced = getBridgeStats().video; + + assert.equal( + afterCoalesced.resultCacheHits - beforeStats.resultCacheHits, + 0, + "joining in-flight work is not a persistent cache hit" + ); + assert.equal( + afterCoalesced.resultSingleflightCoalesced - beforeStats.resultSingleflightCoalesced, + 1, + "the joining request must be reported as coalesced work" + ); + assert.equal(afterCoalesced.resultCacheBytes, beforeStats.resultCacheBytes); + assert.equal(afterCoalesced.resultCacheLatencyMs, beforeStats.resultCacheLatencyMs); + + const third = await bridge.preCall(payload(), {}); + const afterPersistentHit = getBridgeStats().video; + + assert.match(resultText(first), /one shared observation/); + assert.match(resultText(second), /one shared observation/); + assert.match(resultText(third), /one shared observation/); + assert.equal(extractCalls, 1, "singleflight and the persistent hit must skip duplicate FFmpeg"); + assert.equal(captionCalls, 1, "singleflight and the persistent hit must skip duplicate captions"); + assert.equal(afterPersistentHit.resultCacheHits - beforeStats.resultCacheHits, 1); + assert.equal( + afterPersistentHit.resultSingleflightCoalesced - beforeStats.resultSingleflightCoalesced, + 1 + ); + assert.ok( + afterPersistentHit.resultCacheBytes > beforeStats.resultCacheBytes, + "only the completed-store hit contributes cached result bytes" + ); +}); + +test("result cache skips entries that exceed its aggregate byte budget", async () => { + const cacheOptions = { maxBytes: 64, maxEntries: 10, ttlMs: 60_000 }; + const resultCache = new BridgeCache(cacheOptions); + let describeCalls = 0; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeCacheMaxEntries: 23, + modalityBridgeCacheTtlMinutes: 63, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 byte budget", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache, + describePart: async () => { + describeCalls += 1; + return { + description: `[Video description: ${"x".repeat(256)}]`, + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const payload = { + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,QllURS1CVURHRVQ=" }], + }, + ], + }; + + assert.ok((await bridge.preCall(structuredClone(payload), {})).modifiedPayload); + assert.ok((await bridge.preCall(structuredClone(payload), {})).modifiedPayload); + assert.equal(describeCalls, 2, "oversized results must fail open without being retained"); +}); + +test("result cache enforces aggregate eviction and the fixed 16 MiB boundary", async (t) => { + await t.test("aggregate bytes evict the least-recently-used entry", () => { + const cache = new BridgeCache({ maxBytes: 140, maxEntries: 10, ttlMs: 60_000 }); + cache.setEntry("a", { value: "a".repeat(80) }); + cache.setEntry("b", { value: "b".repeat(80) }); + + assert.equal(cache.getEntry("a"), undefined); + assert.equal(cache.getEntry("b")?.value, "b".repeat(80)); + assert.ok(cache.bytes <= 140); + }); + + await t.test("the dedicated cache accepts the exact boundary and rejects one byte more", () => { + const cache = getSharedVideoResultCacheFor({ cacheMaxEntries: 2, cacheTtlMinutes: 61 }); + const key = "k".repeat(64); + const storedEnvelopeBytes = Buffer.byteLength(key, "utf8") + Buffer.byteLength("{}", "utf8"); + const exactValue = "x".repeat(VIDEO_RESULT_CACHE_MAX_BYTES - storedEnvelopeBytes); + try { + cache.clear(); + cache.setEntry(key, { value: exactValue }); + assert.equal(cache.size, 1); + assert.equal(cache.bytes, VIDEO_RESULT_CACHE_MAX_BYTES); + + cache.clear(); + cache.setEntry(key, { value: `${exactValue}x` }); + assert.equal(cache.size, 0); + assert.equal(cache.bytes, 0); + } finally { + cache.clear(); + } + }); +}); + +test("result cache expires complete results at its TTL", async () => { + let now = 1_000; + const resultCache = new BridgeCache({ + maxBytes: 4_096, + maxEntries: 10, + now: () => now, + ttlMs: 10, + }); + let describeCalls = 0; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 TTL", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache, + describePart: async () => { + describeCalls += 1; + return { + description: `[Video description: ttl-${describeCalls}]`, + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const payload = { + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,VFRMLVZJREVP" }], + }, + ], + }; + + await bridge.preCall(structuredClone(payload), {}); + await bridge.preCall(structuredClone(payload), {}); + assert.equal(describeCalls, 1, "the unexpired request must hit"); + now = 1_011; + await bridge.preCall(structuredClone(payload), {}); + assert.equal(describeCalls, 2, "the expired request must recompute"); +}); + +test("result cache evicts the least-recently-used content at its entry bound", async () => { + const resultCache = new BridgeCache({ maxBytes: 4_096, maxEntries: 1, ttlMs: 60_000 }); + let describeCalls = 0; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 LRU", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache, + describePart: async () => { + describeCalls += 1; + return { + description: `[Video description: lru-${describeCalls}]`, + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const payload = (base64: string) => ({ + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: `data:video/mp4;base64,${base64}` }], + }, + ], + }); + + await bridge.preCall(payload("TFJVLUE="), {}); + await bridge.preCall(payload("TFJVLUI="), {}); + await bridge.preCall(payload("TFJVLUE="), {}); + assert.equal(describeCalls, 3, "content A must recompute after content B evicts it"); +}); + +test("an unavailable result cache fails open to normal video processing", async () => { + let describeCalls = 0; + const debugMessages: string[] = []; + const unavailableCache = { + delete: () => { + throw new Error("cache unavailable"); + }, + getEntry: () => { + throw new Error("cache unavailable"); + }, + setEntry: () => { + throw new Error("cache unavailable"); + }, + }; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 unavailable cache", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache: unavailableCache, + describePart: async () => { + describeCalls += 1; + return { + description: "[Video description: normal fail-open result]", + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const result = await bridge.preCall( + { + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,VU5BVkFJTEFCTEU=" }], + }, + ], + }, + { + log: { + debug: (_tag, message) => { + debugMessages.push(message); + }, + }, + } + ); + + assert.match(resultText(result), /normal fail-open result/); + assert.equal(describeCalls, 1); + assert.deepEqual(debugMessages, [ + "Video result cache read failed open", + "Video result cache write failed open", + ]); +}); + +test("a corrupt result-cache payload is discarded and recomputed", async () => { + let describeCalls = 0; + const corruptCache = { + delete: () => undefined, + getEntry: () => ({ + value: 42 as unknown as string, + producerModel: "openai/gpt-4o-mini", + metadata: { + cacheVersion: "v3", + policyVersion: "default", + extractorVersion: "v3", + strategy: "uniform", + model: "openai/gpt-4o-mini", + prompt: "FU-01 corrupt cache", + frameCount: 8, + maxVideos: 1, + durationSeconds: 1, + framesRequested: 1, + framesExtracted: 1, + framesUsed: 1, + cacheBytes: 2, + modelUsed: "openai/gpt-4o-mini", + }, + }), + setEntry: () => undefined, + }; + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 corrupt cache", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache: corruptCache, + describePart: async () => { + describeCalls += 1; + return { + description: "[Video description: recomputed after corruption]", + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const result = await bridge.preCall( + { + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,Q09SUlVQVA==" }], + }, + ], + }, + {} + ); + + assert.match(resultText(result), /recomputed after corruption/); + assert.equal(describeCalls, 1); +}); + +test("invalid numeric result-cache metadata is deleted and recomputed", async (t) => { + const cachedValue = "[Video description: cached numeric metadata]"; + const validMetadata = (): Record => ({ + cacheVersion: "v3", + policyVersion: "default", + extractorVersion: "v3", + strategy: "uniform", + model: "openai/gpt-4o-mini", + prompt: "FU-01 numeric cache validation", + frameCount: 8, + maxVideos: 1, + durationSeconds: 3, + framesRequested: 8, + framesExtracted: 6, + framesUsed: 5, + dedupDropped: 1, + cacheBytes: Buffer.byteLength(cachedValue, "utf8"), + modelUsed: "openai/gpt-4o-mini", + }); + const corruptions: Array<{ + name: string; + mutate: (metadata: Record) => void; + }> = [ + { name: "NaN duration", mutate: (metadata) => (metadata.durationSeconds = Number.NaN) }, + { + name: "infinite duration", + mutate: (metadata) => (metadata.durationSeconds = Number.POSITIVE_INFINITY), + }, + { name: "negative duration", mutate: (metadata) => (metadata.durationSeconds = -1) }, + { name: "NaN frame count", mutate: (metadata) => (metadata.framesRequested = Number.NaN) }, + { + name: "infinite frame count", + mutate: (metadata) => (metadata.framesExtracted = Number.POSITIVE_INFINITY), + }, + { name: "negative frame count", mutate: (metadata) => (metadata.framesUsed = -1) }, + { + name: "more extracted than requested", + mutate: (metadata) => (metadata.framesExtracted = 9), + }, + { name: "more used than extracted", mutate: (metadata) => (metadata.framesUsed = 7) }, + { + name: "dedup and used exceed extracted", + mutate: (metadata) => (metadata.dedupDropped = 2), + }, + { name: "NaN cache bytes", mutate: (metadata) => (metadata.cacheBytes = Number.NaN) }, + { + name: "infinite cache bytes", + mutate: (metadata) => (metadata.cacheBytes = Number.POSITIVE_INFINITY), + }, + { name: "negative cache bytes", mutate: (metadata) => (metadata.cacheBytes = -1) }, + { + name: "mismatched cache bytes", + mutate: (metadata) => (metadata.cacheBytes = Buffer.byteLength(cachedValue, "utf8") + 1), + }, + ]; + + for (const corruption of corruptions) { + await t.test(corruption.name, async () => { + const metadata = validMetadata(); + corruption.mutate(metadata); + let deleteCalls = 0; + let describeCalls = 0; + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 numeric cache validation", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache: { + delete: () => { + deleteCalls += 1; + }, + getEntry: () => ({ + value: cachedValue, + producerModel: "openai/gpt-4o-mini", + metadata, + }), + setEntry: () => undefined, + }, + describePart: async () => { + describeCalls += 1; + return { + description: "[Video description: recomputed numeric metadata]", + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }; + }, + }, + }); + + const result = await bridge.preCall( + { + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,TlVNRVJJQw==" }], + }, + ], + }, + {} + ); + + assert.match(resultText(result), /recomputed numeric metadata/); + assert.equal(deleteCalls, 1, "invalid entries must be removed before recomputing"); + assert.equal(describeCalls, 1, "invalid entries must never be served as cache hits"); + }); + } +}); + +test("never-resolving model selection obeys abort and the attempt deadline", async (t) => { + const payload = () => ({ + model: "example/text-only", + messages: [ + { + role: "user", + content: [{ type: "input_video", video_url: "data:video/mp4;base64,U0VMRUNUSU9O" }], + }, + ], + }); + const createBridge = () => + new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVideoTimeout: 1_000, + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: () => new Promise(() => undefined), + }, + }); + + await t.test("request abort rejects without waiting for selection", async () => { + const controller = new AbortController(); + const pending = createBridge().preCall(payload(), { signal: controller.signal }); + setTimeout(() => controller.abort(), 10); + + const outcome = await Promise.race([ + pending.then( + () => "resolved", + (error: unknown) => error + ), + new Promise<"timed out">((resolve) => setTimeout(() => resolve("timed out"), 500)), + ]); + + assert.notEqual(outcome, "timed out", "abort must release model selection promptly"); + assert.match(String(outcome), /aborted/i); + }); + + await t.test("attempt deadline falls back without waiting for selection", async () => { + const outcome = await Promise.race([ + createBridge().preCall(payload(), {}), + new Promise<"timed out">((resolve) => setTimeout(() => resolve("timed out"), 2_500)), + ]); + + assert.notEqual(outcome, "timed out", "deadline must release model selection promptly"); + if (outcome !== "timed out") { + assert.match(resultText(outcome), /unavailable — video could not be described/); + } + }); +}); + +test("concurrent HTTPS requests share one protected download buffer", async () => { + const resultCache = new BridgeCache({ maxBytes: 4_096, maxEntries: 10, ttlMs: 60_000 }); + let fetchCalls = 0; + let extractCalls = 0; + let fetchedBuffer: Buffer | undefined; + let extractedBuffer: Uint8Array | undefined; + let markDownloadStarted: (() => void) | undefined; + let releaseDownload: (() => void) | undefined; + const downloadStarted = new Promise((resolve) => { + markDownloadStarted = resolve; + }); + const downloadGate = new Promise((resolve) => { + releaseDownload = resolve; + }); + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 protected download singleflight", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache, + fetchRemote: async (url: string) => { + fetchCalls += 1; + fetchedBuffer = Buffer.from("one-protected-download"); + markDownloadStarted?.(); + await downloadGate; + return { buffer: fetchedBuffer, contentType: "video/mp4", url }; + }, + extractFrames: async (bytes: Uint8Array) => { + extractCalls += 1; + extractedBuffer = bytes; + return { + durationSeconds: 1, + frames: [{ dataUri: "data:image/jpeg;base64,T05F", timestampSeconds: 0.5 }], + }; + }, + callVisionModel: async () => "one protected observation", + }, + }); + const context = { + apiKeyInfo: { id: "tenant-protected-download" }, + endpoint: "/v1/chat/completions", + sourceFormat: "openai", + targetFormat: "openai", + }; + + const first = bridge.preCall(remoteVideoPayload(), context); + await downloadStarted; + const second = bridge.preCall(remoteVideoPayload(), context); + await new Promise((resolve) => setImmediate(resolve)); + releaseDownload?.(); + + const [firstResult, secondResult] = await Promise.all([first, second]); + assert.match(resultText(firstResult), /one protected observation/); + assert.match(resultText(secondResult), /one protected observation/); + assert.equal(fetchCalls, 1, "concurrent identical requests must allocate one download buffer"); + assert.equal(extractCalls, 1, "complete-result singleflight must extract the shared buffer once"); + assert.strictEqual(extractedBuffer, fetchedBuffer, "the protected buffer must not be copied"); +}); + +test("cache-disabled production requests still share the bounded protected download", async () => { + let fetchCalls = 0; + let fetchedBuffer: Buffer | undefined; + const extractedBuffers: Uint8Array[] = []; + let markDownloadStarted: (() => void) | undefined; + let releaseDownload: (() => void) | undefined; + const downloadStarted = new Promise((resolve) => { + markDownloadStarted = resolve; + }); + const downloadGate = new Promise((resolve) => { + releaseDownload = resolve; + }); + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: false, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 protected download without result cache", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + fetchRemote: async (url: string) => { + fetchCalls += 1; + fetchedBuffer = Buffer.from("bounded-without-result-cache"); + markDownloadStarted?.(); + await downloadGate; + return { buffer: fetchedBuffer, contentType: "video/mp4", url }; + }, + extractFrames: async (bytes: Uint8Array) => { + extractedBuffers.push(bytes); + return { + durationSeconds: 1, + frames: [{ dataUri: "data:image/jpeg;base64,Tk9D", timestampSeconds: 0.5 }], + }; + }, + callVisionModel: async () => "cache-disabled protected observation", + }, + }); + const context = { + apiKeyInfo: { id: "tenant-cache-disabled" }, + endpoint: "/v1/chat/completions", + }; + + const first = bridge.preCall(remoteVideoPayload(), context); + await downloadStarted; + const second = bridge.preCall(remoteVideoPayload(), context); + await new Promise((resolve) => setImmediate(resolve)); + releaseDownload?.(); + + await Promise.all([first, second]); + assert.equal(fetchCalls, 1, "the raw-media budget must not multiply when caching is disabled"); + assert.equal(extractedBuffers.length, 2, "result processing remains independent without cache"); + assert.ok(extractedBuffers.every((bytes) => bytes === fetchedBuffer)); +}); + +test("aborting one singleflight waiter does not cancel another active request", async () => { + const resultCache = new BridgeCache({ maxBytes: 4_096, maxEntries: 10, ttlMs: 60_000 }); + const firstController = new AbortController(); + let fetchCalls = 0; + let extractCalls = 0; + let captionCalls = 0; + let producerSignal: AbortSignal | undefined; + let markDownloadStarted: (() => void) | undefined; + let releaseDownload: (() => void) | undefined; + const downloadStarted = new Promise((resolve) => { + markDownloadStarted = resolve; + }); + const deps = { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 abort waiter", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache, + fetchRemote: async (url: string, options: { enforceHttps: true; signal: AbortSignal }) => { + fetchCalls += 1; + producerSignal = options.signal; + markDownloadStarted?.(); + return new Promise<{ buffer: Buffer; contentType: string; url: string }>( + (resolve, reject) => { + releaseDownload = () => + resolve({ buffer: Buffer.from("shared-video"), contentType: "video/mp4", url }); + const onAbort = () => reject(new Error("protected download producer aborted")); + if (options.signal.aborted) onAbort(); + else options.signal.addEventListener("abort", onAbort, { once: true }); + } + ); + }, + extractFrames: async ( + _bytes: Uint8Array, + options: { signal?: AbortSignal } + ): Promise<{ + durationSeconds: number; + frames: Array<{ dataUri: string; timestampSeconds: number }>; + }> => { + extractCalls += 1; + await new Promise((resolve, reject) => { + const timer = setTimeout(resolve, 50); + const abort = () => { + clearTimeout(timer); + reject(new Error("shared extraction aborted")); + }; + if (options.signal?.aborted) abort(); + else options.signal?.addEventListener("abort", abort, { once: true }); + }); + return { + durationSeconds: 1, + frames: [{ dataUri: "data:image/jpeg;base64,QUJPUlQ=", timestampSeconds: 0.5 }], + }; + }, + callVisionModel: async () => { + captionCalls += 1; + return "surviving waiter result"; + }, + }; + const bridge = new VideoBridgeGuardrail({ deps }); + const context = { + apiKeyInfo: { id: "tenant-abort-waiter" }, + endpoint: "/v1/chat/completions", + }; + + const first = bridge.preCall(remoteVideoPayload(), { + ...context, + signal: firstController.signal, + }); + await downloadStarted; + const second = bridge.preCall(remoteVideoPayload(), context); + await new Promise((resolve) => setTimeout(resolve, 10)); + firstController.abort(); + + await assert.rejects(first, /aborted/i); + assert.equal(producerSignal?.aborted, false, "one waiter must not abort the shared producer"); + releaseDownload?.(); + const surviving = await second; + assert.match(resultText(surviving), /surviving waiter result/); + assert.equal(fetchCalls, 1, "active identical waiters must share the protected download"); + assert.equal(extractCalls, 1, "the active waiter must keep the shared extraction alive"); + assert.equal(captionCalls, 1); +}); + +test("an abandoned protected download flight cannot capture a later request", async () => { + const firstController = new AbortController(); + let fetchCalls = 0; + let abandonedProducerSignal: AbortSignal | undefined; + let markAbandonedStarted: (() => void) | undefined; + const abandonedStarted = new Promise((resolve) => { + markAbandonedStarted = resolve; + }); + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 abandoned protected download", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache: new BridgeCache({ maxBytes: 4_096, maxEntries: 10, ttlMs: 60_000 }), + fetchRemote: async (url: string, options: { enforceHttps: true; signal: AbortSignal }) => { + fetchCalls += 1; + if (fetchCalls === 1) { + abandonedProducerSignal = options.signal; + markAbandonedStarted?.(); + return new Promise(() => undefined); + } + return { buffer: Buffer.from("fresh-download"), contentType: "video/mp4", url }; + }, + describePart: async () => ({ + description: "[Video description: fresh protected download]", + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }), + }, + }); + const context = { + apiKeyInfo: { id: "tenant-abandoned-download" }, + endpoint: "/v1/chat/completions", + }; + + const abandoned = bridge.preCall(remoteVideoPayload(), { + ...context, + signal: firstController.signal, + }); + await abandonedStarted; + firstController.abort(); + await assert.rejects(abandoned, /aborted/i); + assert.equal(abandonedProducerSignal?.aborted, true); + + const replacement = await Promise.race([ + bridge.preCall(remoteVideoPayload(), context), + new Promise<"timed out">((resolve) => setTimeout(() => resolve("timed out"), 500)), + ]); + + assert.notEqual(replacement, "timed out", "the later request must start a fresh download"); + if (replacement !== "timed out") { + assert.match(resultText(replacement), /fresh protected download/); + } + assert.equal(fetchCalls, 2); +}); + +test("protected download flights are isolated by authenticated principal", async () => { + let fetchCalls = 0; + let markBothStarted: (() => void) | undefined; + let releaseDownloads: (() => void) | undefined; + const bothStarted = new Promise((resolve) => { + markBothStarted = resolve; + }); + const downloadGate = new Promise((resolve) => { + releaseDownloads = resolve; + }); + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeCacheEnabled: true, + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "FU-01 tenant download isolation", + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + resultCache: new BridgeCache({ maxBytes: 4_096, maxEntries: 10, ttlMs: 60_000 }), + fetchRemote: async (url: string) => { + fetchCalls += 1; + if (fetchCalls === 2) markBothStarted?.(); + await downloadGate; + return { buffer: Buffer.from("tenant-isolated"), contentType: "video/mp4", url }; + }, + describePart: async () => ({ + description: "[Video description: tenant isolated]", + durationSeconds: 1, + framesRequested: 1, + framesUsed: 1, + }), + }, + }); + const commonContext = { endpoint: "/v1/chat/completions" }; + + const tenantA = bridge.preCall(remoteVideoPayload(), { + ...commonContext, + apiKeyInfo: { id: "tenant-a" }, + }); + const tenantB = bridge.preCall(remoteVideoPayload(), { + ...commonContext, + apiKeyInfo: { id: "tenant-b" }, + }); + await bothStarted; + releaseDownloads?.(); + + await Promise.all([tenantA, tenantB]); + assert.equal(fetchCalls, 2, "different authenticated principals must not share downloads"); +}); + +test("an abandoned flight cannot capture a later request", async () => { + const firstController = new AbortController(); + let releaseAbandoned: ((value: string) => void) | undefined; + let markStarted: (() => void) | undefined; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + const abandoned = runVideoResultSingleflight("abandoned-flight", firstController.signal, () => { + markStarted?.(); + return new Promise((resolve) => { + releaseAbandoned = resolve; + }); + }); + + await started; + firstController.abort(); + await assert.rejects(abandoned, /aborted/i); + + const replacement = await Promise.race([ + runVideoResultSingleflight( + "abandoned-flight", + new AbortController().signal, + async () => "fresh result" + ), + new Promise<"timed out">((resolve) => setTimeout(() => resolve("timed out"), 50)), + ]); + releaseAbandoned?.("stale result"); + + assert.notEqual(replacement, "timed out", "a later request must start a fresh flight"); + if (replacement !== "timed out") { + assert.equal(replacement.coalesced, false); + assert.equal(replacement.value, "fresh result"); + } +});