fix(producer): type video extraction failures

packages/producer/src/server.errorCode.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from "vitest";
+import { extractSafeRenderErrorCode } from "./server.js";
+import { VideoExtractionStageError } from "./services/render/stages/extractVideosStage.js";
+
+describe("extractSafeRenderErrorCode", () => {
+  it("preserves allowlisted typed extraction codes", () => {
+    const deterministic = new VideoExtractionStageError("VIDEO_SOURCE_UNRENDERABLE", false, [
+      { kind: "invalid_media", count: 1 },
+    ]);
+    const exhausted = new VideoExtractionStageError("VIDEO_EXTRACTION_FAILED", true, [
+      { kind: "ffmpeg_timeout", count: 1 },
+    ]);
+
+    expect(extractSafeRenderErrorCode(deterministic)).toBe("VIDEO_SOURCE_UNRENDERABLE");
+    expect(extractSafeRenderErrorCode(exhausted)).toBe("VIDEO_EXTRACTION_FAILED");
+  });
+
+  it("accepts the same bounded structural code across wrapped module boundaries", () => {
+    expect(extractSafeRenderErrorCode({ code: "VIDEO_SOURCE_UNRENDERABLE" })).toBe(
+      "VIDEO_SOURCE_UNRENDERABLE",
+    );
+  });
+
+  it("does not forward arbitrary codes or parse message text", () => {
+    expect(extractSafeRenderErrorCode({ code: "INTERNAL_ERROR" })).toBeUndefined();
+    expect(
+      extractSafeRenderErrorCode(new Error("failed [VIDEO_SOURCE_UNRENDERABLE; secret=/tmp/x]")),
+    ).toBeUndefined();
+  });
+});

packages/producer/src/server.ts

@@ -118,6 +118,23 @@ interface PreparedRenderInput {
 }
 
 const DEFAULT_SERVER_FPS = { num: 30, den: 1 } as const;
+const SAFE_RENDER_ERROR_CODES = new Set([
+  "VIDEO_SOURCE_UNRENDERABLE",
+  "VIDEO_EXTRACTION_FAILED",
+] as const);
+
+/**
+ * Preserve only bounded producer error codes across JSON/SSE. Never derive a
+ * code from the message: it may contain local paths or signed source URLs.
+ */
+export function extractSafeRenderErrorCode(error: unknown): string | undefined {
+  if (typeof error !== "object" || error === null || !("code" in error)) return undefined;
+  const code = (error as { code?: unknown }).code;
+  return typeof code === "string" &&
+    SAFE_RENDER_ERROR_CODES.has(code as "VIDEO_SOURCE_UNRENDERABLE" | "VIDEO_EXTRACTION_FAILED")
+    ? code
+    : undefined;
+}
 
 function parseServerFps(value: unknown): RenderInput["fps"] {
   if (typeof value !== "number" && typeof value !== "string") return DEFAULT_SERVER_FPS;
@@ -524,6 +541,7 @@ async function writeRenderStreamFailure(input: {
     return;
   }
   const errorMsg = error instanceof Error ? error.message : String(error);
+  const errorCode = extractSafeRenderErrorCode(error);
   const elapsedMs = Date.now() - startedAtMs;
   log.error("render-stream failed", {
     requestId,
@@ -536,6 +554,7 @@ async function writeRenderStreamFailure(input: {
       type: "error",
       requestId,
       error: errorMsg,
+      errorCode,
       stage: job.currentStage,
       elapsedMs,
       errorDetails: job.errorDetails ?? null,
@@ -684,6 +703,7 @@ export function createRenderHandlers(options: HandlerOptions = {}): RenderHandle
     } catch (error) {
       const durationMs = Date.now() - t0;
       const errorMsg = error instanceof Error ? error.message : String(error);
+      const errorCode = extractSafeRenderErrorCode(error);
       log.error("render failed", {
         requestId,
         durationMs,
@@ -695,6 +715,7 @@ export function createRenderHandlers(options: HandlerOptions = {}): RenderHandle
           success: false,
           requestId,
           error: errorMsg,
+          errorCode,
           stage: job.currentStage,
           durationMs,
           errorDetails: job.errorDetails ?? null,

packages/producer/src/services/distributed/plan.ts

@@ -996,6 +996,7 @@ export async function plan(
     assertNotAborted,
     materializeSymlinks: true,
   });
+  if (extractResult.failureToEnforce) throw extractResult.failureToEnforce;
   // Skip `extractResult.frameLookup.cleanup()`: it would rm-rf each
   // video's outputDir, but in `plan()` those directories ARE the source
   // material the renames below move into `planDir/video-frames/`.

packages/producer/src/services/render/observability.ts

@@ -109,6 +109,8 @@ export interface RenderExtractionObservability {
   vfrPreflightCount?: number;
   cacheHits?: number;
   cacheMisses?: number;
+  /** Per-source transient download/metadata/FFmpeg retries performed during extraction. */
+  transientRetries?: number;
   /**
    * Per-clip captured-vs-expected-frame gauges. Emitted by the parity gate
    * at extract finalization (see `videoFrameCoverage.ts`). Undefined when

packages/producer/src/services/render/stages/extractVideosStage.test.ts

@@ -1,6 +1,17 @@
 import { describe, expect, it } from "vitest";
-import { appendAutoDetectedVideoAudio, shouldCopyExtractedFrames } from "./extractVideosStage.js";
-import type { ExtractedFrames, VideoElement } from "@hyperframes/engine";
+import type {
+  ExtractedFrames,
+  ExtractionResult,
+  VideoElement,
+  VideoExtractionFailure,
+} from "@hyperframes/engine";
+import {
+  appendAutoDetectedVideoAudio,
+  assertVideoExtractionSucceeded,
+  resolveVideoExtractionPolicy,
+  shouldCopyExtractedFrames,
+  VideoExtractionStageError,
+} from "./extractVideosStage.js";
 
 function makeVideo(overrides: Partial<VideoElement> = {}): VideoElement {
   return {
@@ -35,6 +46,33 @@ function makeExtracted(videoId: string, fileHasAudio: boolean): ExtractedFrames
   } as ExtractedFrames;
 }
 
+function extractionResult(errors: VideoExtractionFailure[]): ExtractionResult {
+  return {
+    success: errors.length === 0,
+    errors,
+    extracted: [],
+    totalFramesExtracted: 0,
+    durationMs: 1,
+    phaseBreakdown: {
+      resolveMs: 0,
+      cachePublishFailures: 0,
+      cacheGcEvictions: 0,
+      cacheGcBytesFreed: 0,
+      cacheAgedPartialsCleared: 0,
+      hdrProbeMs: 0,
+      hdrPreflightMs: 0,
+      hdrPreflightCount: 0,
+      vfrProbeMs: 0,
+      vfrPreflightMs: 0,
+      vfrPreflightCount: 0,
+      extractMs: 0,
+      cacheHits: 0,
+      cacheMisses: 0,
+      transientRetries: 0,
+    },
+  };
+}
+
 describe("appendAutoDetectedVideoAudio", () => {
   it("adds audio for an audible video whose file has an audio track", () => {
     const composition = { videos: [makeVideo()], audios: [] as never[] };
@@ -92,3 +130,114 @@ describe("shouldCopyExtractedFrames", () => {
     expect(shouldCopyExtractedFrames("linux")).toBe(false);
   });
 });
+
+describe("resolveVideoExtractionPolicy", () => {
+  it("preserves stable behavior by default", () => {
+    expect(resolveVideoExtractionPolicy({})).toEqual({
+      failureMode: "off",
+      maxTransientRetries: 0,
+    });
+  });
+
+  it("allows only the bounded candidate rollout values", () => {
+    expect(
+      resolveVideoExtractionPolicy({
+        HF_VIDEO_EXTRACTION_FAILURE_MODE: "observe",
+        HF_VIDEO_EXTRACTION_MAX_RETRIES: "1",
+      }),
+    ).toEqual({ failureMode: "observe", maxTransientRetries: 1 });
+    expect(
+      resolveVideoExtractionPolicy({
+        HF_VIDEO_EXTRACTION_FAILURE_MODE: "unexpected",
+        HF_VIDEO_EXTRACTION_MAX_RETRIES: "1",
+      }),
+    ).toEqual({ failureMode: "off", maxTransientRetries: 0 });
+  });
+});
+
+describe("assertVideoExtractionSucceeded", () => {
+  it("accepts a complete extraction", () => {
+    expect(() => assertVideoExtractionSucceeded(extractionResult([]))).not.toThrow();
+  });
+
+  it("fails deterministic media errors without forwarding paths or signed URLs", () => {
+    const result = extractionResult([
+      {
+        videoId: "narrator",
+        kind: "zero_output",
+        retryable: false,
+        error:
+          "FFmpeg failed for /tmp/render/secret.mp4 from https://cdn.example/x?Signature=secret",
+      },
+      {
+        videoId: "missing",
+        kind: "source_missing",
+        retryable: false,
+        error: "Video file not found: /tmp/private/input.mp4",
+      },
+    ]);
+
+    let caught: unknown;
+    try {
+      assertVideoExtractionSucceeded(result);
+    } catch (error) {
+      caught = error;
+    }
+
+    expect(caught).toBeInstanceOf(VideoExtractionStageError);
+    expect(caught).toMatchObject({
+      code: "VIDEO_SOURCE_UNRENDERABLE",
+      retryable: false,
+      failures: [
+        { kind: "source_missing", count: 1 },
+        { kind: "zero_output", count: 1 },
+      ],
+    });
+    expect((caught as Error).message).not.toContain("/tmp/");
+    expect((caught as Error).message).not.toContain("Signature");
+  });
+
+  it("keeps exhausted transient failures retryable and collapses duplicate kinds", () => {
+    const result = extractionResult([
+      {
+        videoId: "a",
+        kind: "download_transient",
+        retryable: true,
+        error: "HTTP 503",
+      },
+      {
+        videoId: "b",
+        kind: "download_transient",
+        retryable: true,
+        error: "HTTP 503",
+      },
+    ]);
+
+    expect(() => assertVideoExtractionSucceeded(result)).toThrow(
+      expect.objectContaining({
+        code: "VIDEO_EXTRACTION_FAILED",
+        retryable: true,
+        failures: [{ kind: "download_transient", count: 2 }],
+      }),
+    );
+  });
+
+  it("fails closed for legacy failures without a kind or retryability", () => {
+    expect(() =>
+      assertVideoExtractionSucceeded(
+        extractionResult([
+          {
+            videoId: "legacy",
+            error: "legacy extraction error",
+          },
+        ]),
+      ),
+    ).toThrow(
+      expect.objectContaining({
+        code: "VIDEO_SOURCE_UNRENDERABLE",
+        retryable: false,
+        failures: [{ kind: "internal", count: 1 }],
+      }),
+    );
+  });
+});

packages/producer/src/services/render/stages/extractVideosStage.ts

@@ -34,15 +34,19 @@ import {
   type CaptureVideoMetadataHint,
   type EngineConfig,
   type ExtractedFrames,
+  type ExtractionResult,
   type FrameLookupTable,
   type HdrTransfer,
+  type VideoExtractionFailureKind,
   type VideoColorSpace,
+  classifyVideoExtractionError,
   createFrameLookupTable,
   detectTransfer,
   extractAllVideoFrames,
   extractMediaMetadata,
   isHdrColorSpace,
   resolveProjectRelativeSrc,
+  runVideoExtractionWithRetry,
 } from "@hyperframes/engine";
 import { fpsToNumber } from "@hyperframes/core";
 import {
@@ -94,6 +98,11 @@ export interface ExtractVideosStageResult {
   imageColorSpaces: (VideoColorSpace | null)[];
   /** Wall-clock ms for the video extraction phase. */
   videoExtractMs: number;
+  /**
+   * Candidate-only typed failure gate. Callers throw this only after their
+   * extraction telemetry checkpoint has been emitted.
+   */
+  failureToEnforce: VideoExtractionStageError | null;
 }
 
 /**
@@ -108,6 +117,98 @@ export function shouldCopyExtractedFrames(platform: NodeJS.Platform): boolean {
   return platform === "win32";
 }
 
+export type VideoExtractionStageErrorCode = "VIDEO_SOURCE_UNRENDERABLE" | "VIDEO_EXTRACTION_FAILED";
+
+export interface VideoExtractionStageFailureSummary {
+  kind: VideoExtractionFailureKind;
+  count: number;
+}
+
+export type VideoExtractionFailureMode = "off" | "observe" | "enforce";
+
+export interface VideoExtractionPolicy {
+  failureMode: VideoExtractionFailureMode;
+  maxTransientRetries: 0 | 1;
+}
+
+/**
+ * Candidate-lane rollout controls. Stable behavior remains unchanged unless
+ * explicitly enabled in the producer environment.
+ */
+export function resolveVideoExtractionPolicy(
+  env: Readonly<Record<string, string | undefined>> = process.env,
+): VideoExtractionPolicy {
+  const rawMode = env.HF_VIDEO_EXTRACTION_FAILURE_MODE?.trim().toLowerCase();
+  const failureMode: VideoExtractionFailureMode =
+    rawMode === "observe" || rawMode === "enforce" ? rawMode : "off";
+  const maxTransientRetries =
+    failureMode !== "off" && env.HF_VIDEO_EXTRACTION_MAX_RETRIES?.trim() === "1" ? 1 : 0;
+  return { failureMode, maxTransientRetries };
+}
+
+/**
+ * Producer-safe terminal error for per-source extraction failures.
+ *
+ * `ExtractionResult.errors[].error` intentionally retains local diagnostics
+ * and can contain signed URLs or filesystem paths. This error carries only a
+ * bounded taxonomy/count summary so the HTTP/Temporal boundary can transport
+ * the cause without leaking those values.
+ */
+export class VideoExtractionStageError extends Error {
+  constructor(
+    readonly code: VideoExtractionStageErrorCode,
+    readonly retryable: boolean,
+    readonly failures: readonly VideoExtractionStageFailureSummary[],
+  ) {
+    const total = failures.reduce((sum, failure) => sum + failure.count, 0);
+    const breakdown = failures.map((failure) => `${failure.kind}=${failure.count}`).join(",");
+    super(`Video extraction failed for ${total} source(s) [${code}; ${breakdown}]`);
+    this.name = "VideoExtractionStageError";
+  }
+}
+
+export function assertVideoExtractionSucceeded(result: ExtractionResult): void {
+  const error = buildVideoExtractionStageError(result);
+  if (error) throw error;
+}
+
+function buildVideoExtractionStageError(
+  result: ExtractionResult,
+): VideoExtractionStageError | null {
+  if (result.success && result.errors.length === 0) return null;
+  const counts = new Map<VideoExtractionFailureKind, number>();
+  for (const failure of result.errors) {
+    const kind = failure.kind ?? "internal";
+    counts.set(kind, (counts.get(kind) ?? 0) + 1);
+  }
+  const failures = Array.from(counts, ([kind, count]) => ({ kind, count })).sort((a, b) =>
+    a.kind.localeCompare(b.kind),
+  );
+  const retryable =
+    result.errors.length > 0 && result.errors.every((failure) => failure.retryable === true);
+  return new VideoExtractionStageError(
+    retryable ? "VIDEO_EXTRACTION_FAILED" : "VIDEO_SOURCE_UNRENDERABLE",
+    retryable,
+    failures,
+  );
+}
+
+function applyVideoExtractionFailurePolicy(
+  result: ExtractionResult,
+  policy: VideoExtractionPolicy,
+  log?: ProducerLogger,
+): VideoExtractionStageError | null {
+  const error = buildVideoExtractionStageError(result);
+  if (!error || policy.failureMode === "off") return null;
+  log?.warn("Video extraction produced typed source failures", {
+    mode: policy.failureMode,
+    code: error.code,
+    retryable: error.retryable,
+    failures: error.failures,
+  });
+  return policy.failureMode === "enforce" ? error : null;
+}
+
 export async function runExtractVideosStage(
   input: ExtractVideosStageInput,
 ): Promise<ExtractVideosStageResult> {
@@ -124,9 +225,11 @@ export async function runExtractVideosStage(
   } = input;
 
   const stage2Start = Date.now();
+  const extractionPolicy = resolveVideoExtractionPolicy();
 
   let frameLookup: FrameLookupTable | null = null;
   let extractionResult: Awaited<ReturnType<typeof extractAllVideoFrames>> | null = null;
+  let failureToEnforce: VideoExtractionStageError | null = null;
   let videoReadinessSkipIds: string[] = [];
   let videoMetadataHints: CaptureVideoMetadataHint[] = [];
 
@@ -136,6 +239,7 @@ export async function runExtractVideosStage(
   // avoid ffprobe overhead when the user has explicitly opted out.
   const nativeHdrVideoIds = new Set<string>();
   const videoTransfers = new Map<string, HdrTransfer>();
+  let hdrProbeTransientRetries = 0;
   if (job.config.hdrMode !== "force-sdr" && composition.videos.length > 0) {
     log?.info("Probing video color spaces...", { videoCount: composition.videos.length });
     await Promise.all(
@@ -149,10 +253,42 @@ export async function runExtractVideosStage(
           ? v.src
           : resolveProjectRelativeSrc(v.src, projectDir, compiledDir);
         if (!existsSync(videoPath)) return;
-        const meta = await extractMediaMetadata(videoPath);
-        if (isHdrColorSpace(meta.colorSpace)) {
-          nativeHdrVideoIds.add(v.id);
-          videoTransfers.set(v.id, detectTransfer(meta.colorSpace));
+        try {
+          // Retries are separately opt-in from the failure gate. With the
+          // default zero budget this remains the exact legacy single probe.
+          const attempted =
+            extractionPolicy.maxTransientRetries === 0
+              ? { result: await extractMediaMetadata(videoPath), retries: 0 }
+              : await runVideoExtractionWithRetry(() => extractMediaMetadata(videoPath), {
+                  signal: abortSignal,
+                  maxTransientRetries: extractionPolicy.maxTransientRetries,
+                  onRetry: () => {
+                    hdrProbeTransientRetries += 1;
+                  },
+                });
+          const meta = attempted.result;
+          if (isHdrColorSpace(meta.colorSpace)) {
+            nativeHdrVideoIds.add(v.id);
+            videoTransfers.set(v.id, detectTransfer(meta.colorSpace));
+          }
+        } catch (error) {
+          if (extractionPolicy.failureMode !== "off") {
+            const classified = classifyVideoExtractionError(error);
+            log?.warn("Video HDR metadata probe failed", {
+              mode: extractionPolicy.failureMode,
+              kind: classified.kind,
+              retryable: classified.retryable,
+              transientRetries: hdrProbeTransientRetries,
+            });
+            if (extractionPolicy.failureMode === "enforce") {
+              throw new VideoExtractionStageError(
+                classified.retryable ? "VIDEO_EXTRACTION_FAILED" : "VIDEO_SOURCE_UNRENDERABLE",
+                classified.retryable,
+                [{ kind: classified.kind, count: 1 }],
+              );
+            }
+          }
+          throw error;
         }
       }),
     );
@@ -207,12 +343,17 @@ export async function runExtractVideosStage(
         fps: fpsToNumber(job.config.fps),
         outputDir: join(compiledDir, "__hyperframes_video_frames"),
         format: job.config.videoFrameFormat ?? "auto",
+        maxTransientRetries: extractionPolicy.maxTransientRetries,
+        collectProbeFailures: extractionPolicy.failureMode === "enforce",
       },
       abortSignal,
       { extractCacheDir: cfg.extractCacheDir, extractCacheMaxBytes: cfg.extractCacheMaxBytes },
       compiledDir,
     );
+    extractionResult.phaseBreakdown.transientRetries =
+      (extractionResult.phaseBreakdown.transientRetries ?? 0) + hdrProbeTransientRetries;
     assertNotAborted();
+    failureToEnforce = applyVideoExtractionFailurePolicy(extractionResult, extractionPolicy, log);
 
     materializeExtractedFramesForCompiledDir(extractionResult.extracted, compiledDir, {
       materializeSymlinks,
@@ -243,6 +384,7 @@ export async function runExtractVideosStage(
     hdrImageSrcPaths,
     imageColorSpaces,
     videoExtractMs,
+    failureToEnforce,
   };
 }
 

packages/producer/src/services/renderOrchestrator.ts

@@ -231,6 +231,7 @@ function summarizeExtractionObservability(
     vfrPreflightCount: phaseBreakdown?.vfrPreflightCount,
     cacheHits: phaseBreakdown?.cacheHits,
     cacheMisses: phaseBreakdown?.cacheMisses,
+    transientRetries: phaseBreakdown?.transientRetries,
     ...coverageGauges,
     authoredTimedClipCount,
   };
@@ -2110,6 +2111,7 @@ async function executeRenderPipeline(input: {
       imageTransfers,
       hdrImageSrcPaths,
       imageColorSpaces,
+      failureToEnforce,
     } = extractResult;
     perfStages.videoExtractMs = extractResult.videoExtractMs;
 
@@ -2145,9 +2147,11 @@ async function executeRenderPipeline(input: {
       vfrPreflightMs: extractionObservability.vfrPreflightMs ?? null,
       cacheHits: extractionObservability.cacheHits ?? null,
       cacheMisses: extractionObservability.cacheMisses ?? null,
+      transientRetries: extractionObservability.transientRetries ?? null,
       minVideoFrameCoverageRatio: extractionObservability.minVideoFrameCoverageRatio ?? null,
       authoredTimedClipCount: extractionObservability.authoredTimedClipCount ?? null,
     });
+    if (failureToEnforce) throw failureToEnforce;
     // Gate AFTER the checkpoint so a coverage-failed render still emits
     // the observability row (partial telemetry is still worth having).
     // `assertVideoFrameCoverage` no-ops on an empty report list AND on a