From 533e5c6ec72f036fdad0d84e415b7f134bc81907 Mon Sep 17 00:00:00 2001 From: Xiangzhe Date: Tue, 18 Aug 2026 06:07:38 -0300 Subject: [PATCH] feat(video): surface audio/video fusion telemetry and degrade invalid audio to partial The fusion result's availability, partial and failure fields now reach DescribedVideo.fusion, the guardrail meta (audioFusionRuns/Partials/ FailureCodes), the result-cache metadata and bridge stats. Audio transcript validation moved inside the fusion's audio branch, so an invalid audioTranscript records failures.audio and keeps the visual description instead of failing the whole video. --- .../guardrails/modalityBridge/bridgeStats.ts | 10 ++++ src/lib/guardrails/videoBridge.ts | 47 +++++++++++++++++- src/lib/guardrails/videoBridgeHelpers.ts | 28 +++++++++-- tests/unit/guardrails/videoBridge.test.ts | 49 +++++++++++++++++++ .../videoBridgeTranscriptProvenance.test.ts | 37 ++++++++++++++ 5 files changed, 167 insertions(+), 4 deletions(-) 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" }, + }); });