diff --git a/src/lib/guardrails/modalityBridge/bridgeStats.ts b/src/lib/guardrails/modalityBridge/bridgeStats.ts index 408b39e6ce..719a0fd3cd 100644 --- a/src/lib/guardrails/modalityBridge/bridgeStats.ts +++ b/src/lib/guardrails/modalityBridge/bridgeStats.ts @@ -20,6 +20,10 @@ export interface BridgeModalityStats { resultCacheHits: number; resultCacheLatencyMs: number; failures: number; + /** Audio/video fusion runs (video bridge only; 0 for other modalities). */ + fusionRuns: number; + /** Fusion runs that completed with one branch failed (partial result). */ + fusionPartials: number; lastUsedAt: string | null; latencySamples: number; successes: number; @@ -44,6 +48,8 @@ function emptyStats(): BridgeModalityStats { resultCacheHits: 0, resultCacheLatencyMs: 0, failures: 0, + fusionRuns: 0, + fusionPartials: 0, lastUsedAt: null, latencySamples: 0, successes: 0, @@ -57,6 +63,8 @@ export function recordBridgeUse( cacheHit?: boolean; cacheHits?: number; failure?: boolean; + fusionRun?: boolean; + fusionPartial?: boolean; latencyMs?: number; resultCacheBytes?: number; resultCacheHit?: boolean; @@ -78,6 +86,8 @@ export function recordBridgeUse( ? 1 : 0; s.cacheHits += cacheHits; + if (opts.fusionRun) s.fusionRuns += 1; + if (opts.fusionPartial) s.fusionPartials += 1; if (opts.resultCacheHit) { s.resultCacheHits += 1; if ( diff --git a/src/lib/guardrails/videoBridge.ts b/src/lib/guardrails/videoBridge.ts index a558818f34..fd6316d696 100644 --- a/src/lib/guardrails/videoBridge.ts +++ b/src/lib/guardrails/videoBridge.ts @@ -17,6 +17,7 @@ import { replaceVideoParts, type DescribeVideoDependencies, type DescribedVideo, + type VideoFusionTelemetry, type VideoPart, } from "./videoBridgeHelpers"; import { @@ -72,10 +73,30 @@ interface VideoResultCacheMetadata { samplingPolicyRequested?: "uniform" | "scene_aware" | "segment_aware"; transcriptCuesApplied?: number; contactSheetUsed?: boolean; + fusion?: VideoFusionTelemetry; cacheBytes: number; modelUsed: string; } +function isFusionTelemetry(value: unknown): value is VideoFusionTelemetry { + if (!value || typeof value !== "object") return false; + const record = value as Record; + if ( + typeof record.audioAvailable !== "boolean" || + typeof record.videoAvailable !== "boolean" || + typeof record.partial !== "boolean" + ) { + return false; + } + if (record.failures === undefined) return true; + if (!record.failures || typeof record.failures !== "object") return false; + return Object.entries(record.failures as Record).every( + ([source, code]) => + (source === "audio" || source === "video") && + (code === "ABORTED" || code === "FAILED" || code === "INVALID") + ); +} + export interface VideoBridgeDependencies { getSettings?: () => Promise>; getCapabilities?: (model: string) => { supportsVideo: boolean | null }; @@ -121,7 +142,8 @@ function isVideoResultCacheMetadata(value: unknown): value is VideoResultCacheMe record.samplingPolicyRequested === "segment_aware") && (record.transcriptCuesApplied === undefined || (typeof record.transcriptCuesApplied === "number" && record.transcriptCuesApplied >= 0)) && - (record.contactSheetUsed === undefined || typeof record.contactSheetUsed === "boolean") + (record.contactSheetUsed === undefined || typeof record.contactSheetUsed === "boolean") && + (record.fusion === undefined || isFusionTelemetry(record.fusion)) ); } @@ -190,6 +212,17 @@ export class VideoBridgeGuardrail extends BaseGuardrail { let focusWindowsApplied = 0; let transcriptCuesApplied = 0; let contactSheetsUsed = 0; + let audioFusionRuns = 0; + let audioFusionPartials = 0; + const audioFusionFailureCodes = new Set(); + const recordFusionTelemetry = (fusion?: VideoFusionTelemetry): void => { + if (!fusion) return; + audioFusionRuns += 1; + if (fusion.partial) audioFusionPartials += 1; + for (const [source, code] of Object.entries(fusion.failures ?? {})) { + audioFusionFailureCodes.add(`${source}:${code}`); + } + }; let samplingPolicyEffective: "uniform" | "scene_aware" | "segment_aware" = "uniform"; let failures = 0; @@ -246,6 +279,7 @@ export class VideoBridgeGuardrail extends BaseGuardrail { totalSamplingCandidateCount += meta.samplingCandidateCount ?? 0; transcriptCuesApplied += meta.transcriptCuesApplied ?? 0; if (meta.contactSheetUsed) contactSheetsUsed += 1; + recordFusionTelemetry(meta.fusion); if (meta.samplingPolicyEffective && meta.samplingPolicyEffective !== "uniform") { samplingPolicyEffective = meta.samplingPolicyEffective; } @@ -256,6 +290,8 @@ export class VideoBridgeGuardrail extends BaseGuardrail { successfulModels.add(meta.modelUsed); } recordBridgeUse("video", { + fusionRun: Boolean(meta.fusion), + fusionPartial: meta.fusion?.partial ?? false, latencyMs: elapsed, resultCacheHit: true, resultCacheBytes: meta.cacheBytes, @@ -289,6 +325,7 @@ export class VideoBridgeGuardrail extends BaseGuardrail { if (described.focusWindow) focusWindowsApplied += 1; transcriptCuesApplied += described.transcriptCues?.length ?? 0; if (described.contactSheetUsed) contactSheetsUsed += 1; + recordFusionTelemetry(described.fusion); totalDurationSeconds += described.durationSeconds; totalSamplingCandidateCount += described.sampling?.candidateCount ?? 0; if ( @@ -328,10 +365,13 @@ export class VideoBridgeGuardrail extends BaseGuardrail { 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, @@ -340,6 +380,8 @@ export class VideoBridgeGuardrail extends BaseGuardrail { } else { recordBridgeUse("video", { cacheHits: videoCacheHits, + fusionRun: Boolean(described.fusion), + fusionPartial: described.fusion?.partial ?? false, latencyMs: processingLatencyMs, }); } @@ -395,6 +437,9 @@ export class VideoBridgeGuardrail extends BaseGuardrail { focusWindowsApplied, transcriptCuesApplied, contactSheetsUsed, + audioFusionRuns, + audioFusionPartials, + audioFusionFailureCodes: [...audioFusionFailureCodes].sort(), samplingCandidateCount: totalSamplingCandidateCount, samplingPolicyEffective, samplingPolicyRequested: runtime.samplingPolicy, diff --git a/src/lib/guardrails/videoBridgeHelpers.ts b/src/lib/guardrails/videoBridgeHelpers.ts index 9af6aba355..eba1d70ba2 100644 --- a/src/lib/guardrails/videoBridgeHelpers.ts +++ b/src/lib/guardrails/videoBridgeHelpers.ts @@ -2,7 +2,7 @@ import { detectMediaParts, type MediaPart } from "@omniroute/open-sse/utils/medi import { fetchRemoteMedia, type RemoteMediaFetchResult } from "@/shared/network/remoteImageFetch"; -import { fuseVideoAndAudio } from "./videoAudioFusion"; +import { fuseVideoAndAudio, type VideoAudioFusionResult } from "./videoAudioFusion"; import { buildVideoContactSheet } from "./videoBridgeContactSheet"; import { extractVideoFramesViaBroker, @@ -238,6 +238,14 @@ export interface DescribeVideoDependencies { ) => Promise; } +/** Observable audio/video fusion outcome: availability per branch plus sanitized failure codes. */ +export interface VideoFusionTelemetry { + audioAvailable: boolean; + videoAvailable: boolean; + partial: boolean; + failures?: VideoAudioFusionResult["failures"]; +} + export interface DescribedVideo { cacheHits?: number; description: string; @@ -251,6 +259,7 @@ export interface DescribedVideo { focusWindow?: VideoFocusWindow; transcriptCues?: VideoTranscriptCue[]; contactSheetUsed?: boolean; + fusion?: VideoFusionTelemetry; } export interface VideoCaptionFrame { @@ -481,11 +490,17 @@ export async function describeVideoPart( if (descriptions.length === 0) { throw new Error("Video frames could not be described"); } + let fusionTelemetry: VideoFusionTelemetry | undefined; if (part.audioTranscript !== undefined) { - const audioCues = normalizeVideoTranscript(part.audioTranscript, extracted.durationSeconds); + // Audio validation runs inside the fusion's audio branch on purpose: an + // invalid audioTranscript must surface as a partial fusion (video kept, + // failures.audio recorded), never fail the whole video description. const fused = await fuseVideoAndAudio({ audio: async () => ({ - observations: audioCues.map((cue) => ({ ...cue, source: "audio" as const })), + observations: normalizeVideoTranscript( + part.audioTranscript, + extracted.durationSeconds + ).map((cue) => ({ ...cue, source: "audio" as const })), }), signal, timeoutMs: options.timeoutMs, @@ -505,6 +520,12 @@ export async function describeVideoPart( })), }), }); + fusionTelemetry = { + audioAvailable: fused.audioAvailable, + videoAvailable: fused.videoAvailable, + partial: fused.partial, + ...(fused.failures ? { failures: fused.failures } : {}), + }; const fusedAudio = fused.observations.filter((observation) => observation.source === "audio"); transcriptCues = [ ...transcriptCues, @@ -529,6 +550,7 @@ export async function describeVideoPart( sampling: extracted.sampling, transcriptCues: transcriptCues.length > 0 ? transcriptCues : undefined, contactSheetUsed: contactSheet?.used || undefined, + fusion: fusionTelemetry, }; } catch (error) { if (signal.aborted) throw new Error("Video Bridge processing timed out or was aborted"); diff --git a/tests/unit/guardrails/videoBridge.test.ts b/tests/unit/guardrails/videoBridge.test.ts index c37799dffb..de0dc3b037 100644 --- a/tests/unit/guardrails/videoBridge.test.ts +++ b/tests/unit/guardrails/videoBridge.test.ts @@ -665,3 +665,52 @@ test("result cache misses when the focus window is added or changed", async () = await bridge.preCall(withFocus({ start: 1, end: 2 }), {}); assert.equal(counter.calls, 3, "a different focus window must invalidate the result cache"); }); + +test("audio/video fusion telemetry reaches guardrail meta, bridge stats, and cache hits", async () => { + const before = getBridgeStats().video; + let describeCalls = 0; + const bridge = new VideoBridgeGuardrail({ + deps: { + getSettings: async () => ({ + modalityBridgeVideoEnabled: true, + modalityBridgeVideoModel: "openai/gpt-4o-mini", + modalityBridgeVisionPrompt: "fusion telemetry", + modalityBridgeCacheEnabled: true, + modalityBridgeCacheTtlMinutes: 60, + modalityBridgeCacheMaxEntries: 50, + }), + getCapabilities: () => ({ supportsVideo: false }), + selectVisionModel: async () => "openai/gpt-4o-mini", + describePart: async () => { + describeCalls += 1; + return { + description: "[Video description: partial fusion observation]", + durationSeconds: 4, + framesRequested: 1, + framesUsed: 1, + fusion: { + audioAvailable: false, + videoAvailable: true, + partial: true, + failures: { audio: "FAILED" as const }, + }, + }; + }, + }, + }); + + const first = await bridge.preCall(payload(), {}); + assert.equal(first.meta?.audioFusionRuns, 1); + assert.equal(first.meta?.audioFusionPartials, 1); + assert.deepEqual(first.meta?.audioFusionFailureCodes, ["audio:FAILED"]); + + const second = await bridge.preCall(payload(), {}); + assert.equal(describeCalls, 1, "the second call must be a result cache hit"); + assert.equal(second.meta?.audioFusionRuns, 1, "cache hits must restore fusion telemetry"); + assert.equal(second.meta?.audioFusionPartials, 1); + assert.deepEqual(second.meta?.audioFusionFailureCodes, ["audio:FAILED"]); + + const after = getBridgeStats().video; + assert.equal(after.fusionRuns - before.fusionRuns, 2); + assert.equal(after.fusionPartials - before.fusionPartials, 2); +}); diff --git a/tests/unit/guardrails/videoBridgeTranscriptProvenance.test.ts b/tests/unit/guardrails/videoBridgeTranscriptProvenance.test.ts index 40d597e998..a8c749a0c0 100644 --- a/tests/unit/guardrails/videoBridgeTranscriptProvenance.test.ts +++ b/tests/unit/guardrails/videoBridgeTranscriptProvenance.test.ts @@ -108,4 +108,41 @@ test("fuses an explicitly supplied audio-bridge track without starting STT", asy assert.equal(captionCalls, 1); assert.equal(described.transcriptCues?.[0]?.source, "audio-bridge"); assert.match(described.description, /audio cue/); + assert.deepEqual(described.fusion, { + audioAvailable: true, + videoAvailable: true, + partial: false, + }); +}); + +test("an invalid audioTranscript degrades to a partial fusion and keeps the visual description", async () => { + const described = await describeVideoPart( + { + container: "messages", + messageIndex: 0, + partIndex: 0, + ref: "data:video/mp4;base64,AA==", + shape: "data_uri_string", + audioTranscript: { + cues: [{ text: "late cue", start: 1, end: 99, source: "audio-bridge" }], + }, + }, + { frameCount: 1, timeoutMs: 1000 }, + async () => "visual cue", + { + extractFrames: async () => ({ + durationSeconds: 5, + frames: [{ dataUri: "data:image/jpeg;base64,AA==", timestampSeconds: 2 }], + }), + } + ); + + assert.match(described.description, /visual cue/); + assert.equal(described.transcriptCues, undefined, "invalid audio must not add transcript cues"); + assert.deepEqual(described.fusion, { + audioAvailable: false, + videoAvailable: true, + partial: true, + failures: { audio: "FAILED" }, + }); });