fix(producer): type video extraction failures (#2776)

## Summary
- classify per-source video download/probe/decode/extraction failures with a bounded taxonomy and safe producer-facing summaries
- add candidate-only, at-most-one transient retry with cleanup and retry telemetry
- preserve default engine/producer behavior when the policy is off
- carry allowlisted extraction error codes through blocking JSON and SSE responses

## Stack
Depends on #2774 for atomic remote downloads and its single owned download retry. This PR is intentionally based on `fix/atomic-video-download-retry`; rebase/change the base to `main` after #2774 merges.

## Default compatibility
`HF_VIDEO_EXTRACTION_FAILURE_MODE` defaults to `off` and forces `maxTransientRetries=0`.

With the feature off:
- metadata probe failures keep the legacy Promise rejection
- grouped extraction keeps the existing grouped-to-direct fallback
- no new producer failure gate is enforced
- render-plan schema, Plan v1 artifacts, chunk routing, and distributed execution are unchanged

Typed metadata aggregation is explicit and enabled only by the candidate enforce lane.

## Retry ownership
- remote downloads: exactly one retry owned by #2774
- metadata/FFmpeg extraction: at most one retry only when `HF_VIDEO_EXTRACTION_MAX_RETRIES=1`
- invalid, missing, rejected, out-of-range, zero-output, cancellation, and unknown/internal failures do not retry
- non-finite or invalid runtime retry budgets fail closed to zero
- the superset optimization is never retried; on failure it preserves direct-member fallback, and only the individual ranges can use the bounded retry
- retry counters increment when a retry is scheduled, including exhausted retries

The internal sidecar and Experiment Framework must treat both exhausted stage codes as workflow-terminal after the producer-local budget. Candidate enforcement must not be enabled until those companion mappings are deployed, or Temporal can multiply producer attempts.

## Failure contract
- `VIDEO_SOURCE_UNRENDERABLE`: at least one deterministic/unknown source failure
- `VIDEO_EXTRACTION_FAILED`: all source failures are transient but the producer-local budget is exhausted

Only the allowlisted code and kind/count summaries cross JSON/SSE. Raw diagnostics remain engine-local because they may contain signed URLs or local paths.

## Rollout
1. merge and deploy with stable/candidate both `off`
2. candidate `observe`, retries 0
3. candidate `observe`, retries 1
4. deploy internal + EF terminal transport mappings
5. candidate `enforce`, retries 1
6. keep stable off until success delta, retry counts, extraction latency, CPU/disk, and queue backlog are acceptable

## Validation
- engine focused suites: 105 passed
- producer focused suites: 15 passed
- full engine suite: 1,176 passed, 3 skipped
- full producer unit lane: 32 Vitest files / 393 tests plus all classified Bun unit tests
- engine and producer typechecks passed
- oxlint, oxfmt, Fallow, tracked-artifact, and commit hooks passed
- independent review: approved for merge default-off; candidate enforcement held on companion transport rollout

Author: jrusso1020

packages/engine/src/index.ts

@@ -184,12 +184,18 @@ export {
   createFrameLookupTable,
   FrameLookupTable,
   analyzeClipMediaFit,
+  classifyVideoExtractionError,
+  isVideoSourceExtractionError,
+  runVideoExtractionWithRetry,
+  VideoSourceExtractionError,
   type VideoElement,
   type ImageElement,
   type ExtractedFrames,
   type ExtractionOptions,
   type ExtractionResult,
   type ExtractionPhaseBreakdown,
+  type VideoExtractionFailure,
+  type VideoExtractionFailureKind,
   type VideoFrameFormat,
   VIDEO_FRAME_FORMATS,
   isVideoFrameFormat,

packages/engine/src/services/videoFrameExtractor.errorClassification.test.ts

@@ -0,0 +1,20 @@
+import { describe, expect, it } from "vitest";
+import { classifyFfmpegSpawnError } from "./videoFrameExtractor.js";
+
+describe("classifyFfmpegSpawnError", () => {
+  it.each(["ENOENT", "EACCES", "ENOEXEC", "UNKNOWN"])(
+    "keeps deterministic launch failure %s terminal",
+    (code) => {
+      expect(classifyFfmpegSpawnError(Object.assign(new Error(code), { code }))).toMatchObject({
+        retryable: false,
+      });
+    },
+  );
+
+  it.each(["EAGAIN", "EMFILE", "ENFILE"])("retries known transient launch failure %s", (code) => {
+    expect(classifyFfmpegSpawnError(Object.assign(new Error(code), { code }))).toMatchObject({
+      kind: "ffmpeg_transient",
+      retryable: true,
+    });
+  });
+});

packages/engine/src/services/videoFrameExtractor.test.ts

@@ -1,3 +1,4 @@
+// fallow-ignore-file code-duplication
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import {
   existsSync,
@@ -25,6 +26,9 @@ import {
   decoderForCodec,
   getFrameAtTime,
   analyzeClipMediaFit,
+  classifyVideoExtractionError,
+  runVideoExtractionWithRetry,
+  VideoSourceExtractionError,
   type VideoElement,
   type ExtractedFrames,
   type ExtractionResult,
@@ -41,6 +45,120 @@ import { COMPLETE_SENTINEL, GC_MARKER, SCHEMA_PREFIX } from "./extractionCache.j
 // synthesized VFR fixture.
 const HAS_FFMPEG = spawnSync("ffmpeg", ["-version"]).status === 0;
 
+describe("video extraction failure taxonomy and bounded retry", () => {
+  it("classifies missing and transient HTTP sources without exposing retry ambiguity", () => {
+    expect(classifyVideoExtractionError(new Error("HTTP 404: Not Found"))).toMatchObject({
+      kind: "download_not_found",
+      retryable: false,
+    });
+    expect(classifyVideoExtractionError(new Error("HTTP 503: Service Unavailable"))).toMatchObject({
+      kind: "download_transient",
+      retryable: true,
+    });
+  });
+
+  it("retries one transient failure, cleaning partial output before the retry", async () => {
+    const retryDir = mkdtempSync(join(tmpdir(), "hf-extract-retry-"));
+    const partialPath = join(retryDir, "frame-00001.jpg");
+    let attempts = 0;
+    try {
+      const outcome = await runVideoExtractionWithRetry(
+        async () => {
+          attempts += 1;
+          if (attempts === 1) {
+            writeFileSync(partialPath, "partial");
+            throw new VideoSourceExtractionError(
+              "ffmpeg_timeout",
+              true,
+              "Video frame extraction timed out",
+            );
+          }
+          expect(existsSync(partialPath)).toBe(false);
+          return "frames";
+        },
+        {
+          maxTransientRetries: 1,
+          onRetry: () => {
+            rmSync(retryDir, { recursive: true, force: true });
+            mkdirSync(retryDir, { recursive: true });
+          },
+        },
+      );
+
+      expect(outcome).toEqual({ result: "frames", retries: 1 });
+      expect(attempts).toBe(2);
+    } finally {
+      rmSync(retryDir, { recursive: true, force: true });
+    }
+  });
+
+  it("does not retry deterministic or caller-aborted failures", async () => {
+    let deterministicAttempts = 0;
+    await expect(
+      runVideoExtractionWithRetry(async () => {
+        deterministicAttempts += 1;
+        throw new VideoSourceExtractionError(
+          "zero_output",
+          false,
+          "Video source produced no decodable frames",
+        );
+      }),
+    ).rejects.toMatchObject({ kind: "zero_output", retryable: false });
+    expect(deterministicAttempts).toBe(1);
+
+    const controller = new AbortController();
+    controller.abort();
+    let abortedAttempts = 0;
+    await expect(
+      runVideoExtractionWithRetry(
+        async () => {
+          abortedAttempts += 1;
+          throw new VideoSourceExtractionError(
+            "download_transient",
+            true,
+            "Video source download failed transiently",
+          );
+        },
+        { signal: controller.signal },
+      ),
+    ).rejects.toMatchObject({ kind: "cancelled", retryable: false });
+    expect(abortedAttempts).toBe(0);
+  });
+
+  it("does not retry transient extraction failures unless the caller opts in", async () => {
+    let attempts = 0;
+    await expect(
+      runVideoExtractionWithRetry(async () => {
+        attempts += 1;
+        throw new VideoSourceExtractionError(
+          "ffmpeg_timeout",
+          true,
+          "Video frame extraction timed out",
+        );
+      }),
+    ).rejects.toMatchObject({ kind: "ffmpeg_timeout", retryable: true });
+    expect(attempts).toBe(1);
+  });
+
+  it("fails closed to zero retries for a non-finite runtime retry budget", async () => {
+    let attempts = 0;
+    await expect(
+      runVideoExtractionWithRetry(
+        async () => {
+          attempts += 1;
+          throw new VideoSourceExtractionError(
+            "ffmpeg_timeout",
+            true,
+            "Video frame extraction timed out",
+          );
+        },
+        { maxTransientRetries: Number.NaN },
+      ),
+    ).rejects.toMatchObject({ kind: "ffmpeg_timeout", retryable: true });
+    expect(attempts).toBe(1);
+  });
+});
+
 // Codec-based alpha defaulting replaces tag-based detection (the
 // alpha_mode/ALPHA_MODE case bug — see ffprobe.test.ts for the regression
 // pin on that). The extractor uses these helpers for two decisions:
@@ -966,6 +1084,45 @@ describe.skipIf(!HAS_FFMPEG)("extractAllVideoFrames on a VFR source", () => {
     return src;
   }
 
+  it("rejects a media start beyond source duration before invoking FFmpeg", async () => {
+    const src = await synthCfrClip("zero-output-src.mp4", 1);
+    const outputDir = join(FIXTURE_DIR, "out-zero-output");
+    await expect(
+      extractVideoFramesRange(src, "past-eof", 2, 1, { fps: 30, outputDir }),
+    ).rejects.toMatchObject({
+      kind: "media_start_out_of_range",
+      retryable: false,
+    });
+  }, 60_000);
+
+  it("preserves legacy metadata rejection unless typed aggregation is explicitly enabled", async () => {
+    const src = join(FIXTURE_DIR, "invalid-probe.mp4");
+    writeFileSync(src, "not a media container");
+    const video = cfrClipElement("invalid-probe", src, 1);
+
+    await expect(
+      extractAllVideoFrames([video], FIXTURE_DIR, {
+        fps: 30,
+        outputDir: join(FIXTURE_DIR, "out-invalid-probe-legacy"),
+      }),
+    ).rejects.toThrow();
+
+    const collected = await extractAllVideoFrames([video], FIXTURE_DIR, {
+      fps: 30,
+      outputDir: join(FIXTURE_DIR, "out-invalid-probe-typed"),
+      collectProbeFailures: true,
+    });
+    expect(collected.success).toBe(false);
+    expect(collected.extracted).toEqual([]);
+    expect(collected.errors).toEqual([
+      expect.objectContaining({
+        videoId: "invalid-probe",
+        kind: "invalid_media",
+        retryable: false,
+      }),
+    ]);
+  }, 60_000);
+
   async function synthHdrTaggedClip(name: string, durationSeconds: number): Promise<string> {
     const src = join(FIXTURE_DIR, name);
     const synth = await runFfmpeg([