fix(render): keep verification fallback storage-aware (#3703)

* fix(render): keep verification fallback storage-aware

* fix(render): honour storage-aware verification fallback for every routing kind

The disk-headroom flag was computed by the orchestrator for both
worker_inversion and parallel_router routings but only honoured by
replanAfterFailure under worker_inversion, so the default-on
parallel_router path still reverted to an sdr_disk plan that the 90%
gate would reject and assertDiskCaptureHeadroom aborted the render.

Give the predicate a single owner: drawElementVerificationFailure
decides when headroom can change the fallback (non-default routing,
sdr_disk preferred fallback, off-disk memory-exhaustion fallback) and
consults the shared inspector only then. replanAfterFailure now honours
the resulting flag for any routing kind, and the orchestrator merges the
two revert log branches keyed on that flag instead of on plan kind,
which also stops OOM retries from being logged as headroom diversions.

inspectDiskCaptureHeadroom returns a discriminated union so the assert
no longer needs a redundant null guard to narrow freeBytes.

Tests: parallel_router headroom diversion and restoration, the failure
constructor across default / streaming-only / disk-only routings, and an
exact-once statfs count.

Co-Authored-By: Miguel Ángel <miguel.sierra@heygen.com>

* test(render): prove non-verification capture failures skip the disk inspector

Extract the streaming drain's failure construction into
`streamingCaptureFailure` so the retry classification is unit-testable,
and assert that a drawElement capture failure or OOM on a routed plan
never calls the disk headroom inspector — only a self-verification
failure may, via `drawElementVerificationFailure`.

Co-Authored-By: Miguel Ángel <miguel.sierra@heygen.com>

---------

Co-authored-by: miga-heygen <miguel.sierra_miga@heygen.com>

packages/producer/src/services/render/capturePlan.test.ts

@@ -1,5 +1,11 @@
 import { describe, expect, it } from "vitest";
-import { createCapturePlan, replanAfterFailure, type CaptureRouting } from "./capturePlan.js";
+import {
+  createCapturePlan,
+  drawElementVerificationFailure,
+  replanAfterFailure,
+  streamingCaptureFailure,
+  type CaptureRouting,
+} from "./capturePlan.js";
 
 function streaming(routing?: CaptureRouting) {
   return createCapturePlan({
@@ -80,7 +86,10 @@ describe("CapturePlan", () => {
         forceParallelStream: false,
       },
     });
-    const next = replanAfterFailure(initial, { kind: "draw_element_verification" });
+    const next = replanAfterFailure(initial, {
+      kind: "draw_element_verification",
+      diskFallbackAvailable: true,
+    });
 
     expect(next).toMatchObject({
       kind: "sdr_disk",
@@ -92,6 +101,191 @@ describe("CapturePlan", () => {
     expect(Object.isFrozen(next.routing)).toBe(true);
   });
 
+  it("keeps verification recovery streaming when the disk fallback lacks headroom", () => {
+    const initial = streaming({
+      kind: "worker_inversion",
+      state: "active",
+      fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
+      memoryExhaustionFallback: {
+        kind: "sdr_streaming",
+        workerCount: 1,
+        forceParallelStream: false,
+      },
+    });
+
+    const next = replanAfterFailure(initial, {
+      kind: "draw_element_verification",
+      diskFallbackAvailable: false,
+    });
+
+    expect(next).toMatchObject({
+      kind: "sdr_streaming",
+      workerCount: 1,
+      forceScreenshot: true,
+      routing: { kind: "worker_inversion", state: "reverted" },
+    });
+    expect(initial).toMatchObject({ workerCount: 1, routing: { state: "active" } });
+    expect(Object.isFrozen(next.routing)).toBe(true);
+  });
+
+  it("keeps parallel-router verification recovery streaming when the disk fallback lacks headroom", () => {
+    const initial = streaming({
+      kind: "parallel_router",
+      state: "active",
+      fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
+      memoryExhaustionFallback: {
+        kind: "sdr_streaming",
+        workerCount: 1,
+        forceParallelStream: false,
+      },
+    });
+
+    const next = replanAfterFailure(initial, {
+      kind: "draw_element_verification",
+      diskFallbackAvailable: false,
+    });
+
+    expect(next).toMatchObject({
+      kind: "sdr_streaming",
+      workerCount: 1,
+      forceScreenshot: true,
+      routing: { kind: "parallel_router", state: "reverted" },
+    });
+    expect(
+      replanAfterFailure(initial, {
+        kind: "draw_element_verification",
+        diskFallbackAvailable: true,
+      }),
+    ).toMatchObject({ kind: "sdr_disk", workerCount: 5 });
+  });
+
+  describe("drawElementVerificationFailure", () => {
+    const diskFallback = { kind: "sdr_disk", workerCount: 5, forceParallelStream: false } as const;
+    const streamFallback = {
+      kind: "sdr_streaming",
+      workerCount: 1,
+      forceParallelStream: false,
+    } as const;
+
+    it("consults disk headroom for every routing whose disk fallback has an off-disk escape", () => {
+      for (const kind of ["worker_inversion", "parallel_router"] as const) {
+        const plan = streaming({
+          kind,
+          state: "active",
+          fallback: diskFallback,
+          memoryExhaustionFallback: streamFallback,
+        });
+        for (const available of [false, true]) {
+          let inspections = 0;
+          const failure = drawElementVerificationFailure(plan, () => {
+            inspections += 1;
+            return available;
+          });
+          expect(failure).toEqual({
+            kind: "draw_element_verification",
+            diskFallbackAvailable: available,
+          });
+          expect(inspections).toBe(1);
+        }
+      }
+    });
+
+    it("skips the disk inspection when headroom cannot change the fallback", () => {
+      const inspect = () => {
+        throw new Error("disk must not be inspected");
+      };
+      const unchanged = { kind: "draw_element_verification", diskFallbackAvailable: undefined };
+
+      expect(drawElementVerificationFailure(streaming(), inspect)).toEqual(unchanged);
+      expect(
+        drawElementVerificationFailure(
+          streaming({
+            kind: "parallel_router",
+            state: "active",
+            fallback: streamFallback,
+            memoryExhaustionFallback: streamFallback,
+          }),
+          inspect,
+        ),
+      ).toEqual(unchanged);
+      expect(
+        drawElementVerificationFailure(
+          streaming({
+            kind: "worker_inversion",
+            state: "active",
+            fallback: diskFallback,
+            memoryExhaustionFallback: { ...diskFallback, workerCount: 1 },
+          }),
+          inspect,
+        ),
+      ).toEqual(unchanged);
+    });
+  });
+
+  describe("streamingCaptureFailure", () => {
+    // The routing shape for which a verification failure DOES consult disk
+    // headroom — so any inspector call below would be on that path alone.
+    const routedPlan = streaming({
+      kind: "worker_inversion",
+      state: "active",
+      fallback: { kind: "sdr_disk", workerCount: 5, forceParallelStream: false },
+      memoryExhaustionFallback: {
+        kind: "sdr_streaming",
+        workerCount: 1,
+        forceParallelStream: false,
+      },
+    });
+    const inspect = () => {
+      throw new Error("disk must not be inspected");
+    };
+
+    // A canvas / paint-record capture error, a renderer stall and an OOM are
+    // not verification failures: each retries as a plain capture failure and
+    // must not touch the disk.
+    it.each([false, true])(
+      "never inspects disk headroom for a non-verification failure (memoryExhaustion: %s)",
+      (isMemoryExhaustion) => {
+        expect(
+          streamingCaptureFailure(
+            routedPlan,
+            { isVerifyError: false, isMemoryExhaustion },
+            inspect,
+          ),
+        ).toEqual({ kind: "capture_failure", memoryExhaustion: isMemoryExhaustion });
+      },
+    );
+
+    it("reverts to the preferred disk fallback on a drawElement capture failure", () => {
+      // The streaming-side capture failure carries no headroom flag, so the
+      // retry takes the routing's preferred fallback without steering.
+      const failure = streamingCaptureFailure(
+        routedPlan,
+        { isVerifyError: false, isMemoryExhaustion: false },
+        inspect,
+      );
+      expect(replanAfterFailure(routedPlan, failure)).toMatchObject({
+        kind: "sdr_disk",
+        workerCount: 5,
+        forceScreenshot: true,
+        routing: { kind: "worker_inversion", state: "reverted" },
+      });
+    });
+
+    it("routes a verification failure through the headroom-aware builder", () => {
+      let inspections = 0;
+      const failure = streamingCaptureFailure(
+        routedPlan,
+        { isVerifyError: true, isMemoryExhaustion: false },
+        () => {
+          inspections += 1;
+          return false;
+        },
+      );
+      expect(failure).toEqual({ kind: "draw_element_verification", diskFallbackAvailable: false });
+      expect(inspections).toBe(1);
+    });
+  });
+
   it("retries an inversion OOM in single-worker screenshot streaming mode", () => {
     const initial = streaming({
       kind: "worker_inversion",

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

@@ -65,10 +65,58 @@ export interface CreateCapturePlanInput {
 
 export type CapturePlanFailure =
   | Readonly<{ kind: "streaming_unavailable" }>
-  | Readonly<{ kind: "draw_element_verification" }>
+  | Readonly<{
+      kind: "draw_element_verification";
+      /**
+       * Set by `drawElementVerificationFailure` only when the routing's
+       * preferred fallback writes frames to disk and its precomputed
+       * low-resource target does not. `false` means that disk route lacks
+       * headroom, so `replanAfterFailure` takes the off-disk target instead.
+       */
+      diskFallbackAvailable?: boolean;
+    }>
   | Readonly<{ kind: "draw_element_capture" }>
   | Readonly<{ kind: "capture_failure"; memoryExhaustion: boolean }>;
 
+/**
+ * Build the verification failure for `plan`, consulting disk headroom only
+ * when the answer can change the fallback: a non-default routing (worker
+ * inversion or parallel router) whose preferred fallback is `sdr_disk` while
+ * its memory-exhaustion fallback is off-disk. Every other plan leaves the flag
+ * `undefined` without calling `hasDiskFallbackHeadroom`.
+ */
+export function drawElementVerificationFailure(
+  plan: CapturePlan,
+  hasDiskFallbackHeadroom: () => boolean,
+): CapturePlanFailure {
+  const { routing } = plan;
+  const needsHeadroom =
+    routing.kind !== "default" &&
+    routing.fallback.kind === "sdr_disk" &&
+    routing.memoryExhaustionFallback.kind === "sdr_streaming";
+  return {
+    kind: "draw_element_verification",
+    diskFallbackAvailable: needsHeadroom ? hasDiskFallbackHeadroom() : undefined,
+  };
+}
+
+/**
+ * Build the failure the streaming drain retries with, from the caller's
+ * classification of the capture error. Only a drawElement self-verification
+ * failure goes through `drawElementVerificationFailure` (and so may consult
+ * disk headroom); every other failure is a plain capture failure and never
+ * calls `hasDiskFallbackHeadroom`.
+ */
+export function streamingCaptureFailure(
+  plan: CapturePlan,
+  classification: Readonly<{ isVerifyError: boolean; isMemoryExhaustion: boolean }>,
+  hasDiskFallbackHeadroom: () => boolean,
+): CapturePlanFailure {
+  return classification.isVerifyError
+    ? drawElementVerificationFailure(plan, hasDiskFallbackHeadroom)
+    : { kind: "capture_failure", memoryExhaustion: classification.isMemoryExhaustion };
+}
+
 function assertWorkerCount(workerCount: number): void {
   if (!Number.isInteger(workerCount) || workerCount < 1) {
     throw new Error(`CapturePlan workerCount must be a positive integer; got ${workerCount}`);
@@ -154,6 +202,12 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur
   }
 
   const isMemoryExhaustion = failure.kind === "capture_failure" && failure.memoryExhaustion;
+  const diskFallbackUnavailable =
+    failure.kind === "draw_element_verification" && failure.diskFallbackAvailable === false;
+  // memoryExhaustionFallback is the routing decision's precomputed
+  // low-resource target. It is also the viable choice when disk, rather than
+  // RAM, makes the preferred fallback impossible — for any routing kind
+  // (see drawElementVerificationFailure for when that flag is populated).
   const fallback =
     plan.routing.kind === "default"
       ? {
@@ -161,7 +215,7 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur
           workerCount: isMemoryExhaustion ? 1 : plan.workerCount,
           forceParallelStream: false,
         }
-      : isMemoryExhaustion
+      : isMemoryExhaustion || diskFallbackUnavailable
         ? plan.routing.memoryExhaustionFallback
         : plan.routing.fallback;
   return createCapturePlan({

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

@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
 import {
   assertDiskCaptureHeadroom,
   estimateDiskCaptureBytes,
+  inspectDiskCaptureHeadroom,
   shouldAllowAdaptiveCaptureRetry,
 } from "./captureStage.js";
 
@@ -33,4 +34,39 @@ describe("disk capture capacity", () => {
       assertDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => 800_000),
     ).toThrow(/may need ~0\.8 MB.*0\.8 MB is free.*--low-memory-mode/s);
   });
+
+  it("exposes the same 90% headroom decision to fallback planning", () => {
+    const estimatedBytes = estimateDiskCaptureBytes(10, captureOptions);
+
+    expect(
+      inspectDiskCaptureHeadroom(
+        "/render/captured-frames",
+        10,
+        captureOptions,
+        () => estimatedBytes / 0.9 - 1,
+      ),
+    ).toEqual({ available: false, estimatedBytes, freeBytes: estimatedBytes / 0.9 - 1 });
+    expect(
+      inspectDiskCaptureHeadroom("/render/captured-frames", 10, captureOptions, () => null)
+        .available,
+    ).toBe(true);
+  });
+
+  it("rejects the reported 5318-frame landscape disk route with 45 GiB free", () => {
+    const landscape = {
+      width: 1920,
+      height: 1080,
+      fps: { num: 30, den: 1 },
+      format: "jpeg" as const,
+    };
+    const headroom = inspectDiskCaptureHeadroom(
+      "/render/captured-frames",
+      5318,
+      landscape,
+      () => 45 * 1024 ** 3,
+    );
+
+    expect(headroom.estimatedBytes).toBe(5318 * 1920 * 1080 * 4);
+    expect(headroom.available).toBe(false);
+  });
 });

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

@@ -147,7 +147,16 @@ export function estimateDiskCaptureBytes(
   return Math.ceil(totalFrames) * outputWidth * outputHeight * 4;
 }
 
-export function assertDiskCaptureHeadroom(
+/**
+ * Unknown free space (`freeBytes: null`) counts as available: the gate can
+ * only reject when it has a measurement.
+ */
+export type DiskCaptureHeadroom =
+  | { available: true; estimatedBytes: number; freeBytes: number | null }
+  | { available: false; estimatedBytes: number; freeBytes: number };
+
+/** Shared 90% disk gate used by both fallback planning and disk execution. */
+export function inspectDiskCaptureHeadroom(
   framesDir: string,
   totalFrames: number,
   captureOptions: CaptureOptions,
@@ -159,14 +168,31 @@ export function assertDiskCaptureHeadroom(
       return null;
     }
   },
-): void {
+): DiskCaptureHeadroom {
   const freeBytes = freeDiskBytes(framesDir);
-  if (freeBytes === null) return;
   const estimatedBytes = estimateDiskCaptureBytes(totalFrames, captureOptions);
-  if (estimatedBytes <= freeBytes * 0.9) return;
+  if (freeBytes === null || estimatedBytes <= freeBytes * 0.9) {
+    return { available: true, estimatedBytes, freeBytes };
+  }
+  return { available: false, estimatedBytes, freeBytes };
+}
+
+export function assertDiskCaptureHeadroom(
+  framesDir: string,
+  totalFrames: number,
+  captureOptions: CaptureOptions,
+  freeDiskBytes?: (path: string) => number | null,
+): void {
+  const headroom = inspectDiskCaptureHeadroom(
+    framesDir,
+    totalFrames,
+    captureOptions,
+    freeDiskBytes,
+  );
+  if (headroom.available) return;
   throw new Error(
-    `Disk capture may need ~${(estimatedBytes / 1e6).toFixed(1)} MB of temporary frame storage, ` +
-      `but only ${(freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` +
+    `Disk capture may need ~${(headroom.estimatedBytes / 1e6).toFixed(1)} MB of temporary frame storage, ` +
+      `but only ${(headroom.freeBytes / 1e6).toFixed(1)} MB is free at ${framesDir}. ` +
       "Re-run with --low-memory-mode to stream frames, raise " +
       "PRODUCER_STREAMING_ENCODE_MAX_DURATION_SECONDS if streaming is supported, " +
       "or free up disk space.",

packages/producer/src/services/renderOrchestrator.ts

@@ -119,6 +119,7 @@ import { ArtifactTransaction } from "./render/artifactTransaction.js";
 import {
   createCapturePlan,
   replanAfterFailure,
+  streamingCaptureFailure,
   type CapturePlan,
   type SdrDiskCapturePlan,
   type CaptureRouting,
@@ -164,7 +165,7 @@ import { runProbeStage } from "./render/stages/probeStage.js";
 import { validateRenderDuration } from "./render/planValidation.js";
 import { runExtractVideosStage } from "./render/stages/extractVideosStage.js";
 import { runAudioStage } from "./render/stages/audioStage.js";
-import { runCaptureStage } from "./render/stages/captureStage.js";
+import { inspectDiskCaptureHeadroom, runCaptureStage } from "./render/stages/captureStage.js";
 import {
   type CaptureStreamingStageResult,
   runCaptureStreamingStage,
@@ -3780,12 +3781,13 @@ async function executeRenderPipeline(input: {
                   : "capture failed; retrying with a fresh screenshot session",
           );
           const failedRouting = capturePlan.routing.kind;
-          capturePlan = replanAfterFailure(
+          const failure = streamingCaptureFailure(
             capturePlan,
-            isVerifyError
-              ? { kind: "draw_element_verification" }
-              : { kind: "capture_failure", memoryExhaustion: isMemoryExhaustion },
+            { isVerifyError, isMemoryExhaustion },
+            () =>
+              inspectDiskCaptureHeadroom(framesDir, totalFrames, buildCaptureOptions()).available,
           );
+          capturePlan = replanAfterFailure(capturePlan, failure);
           syncCapturePlan();
           updateCaptureObservability({
             forceScreenshot: capturePlan.forceScreenshot,
@@ -3809,24 +3811,26 @@ async function executeRenderPipeline(input: {
             probeSession = null;
             await closeOrphanedProbeForRetry(orphaned, closeCaptureSession, log, "streaming");
           }
-          if (failedRouting === "worker_inversion") {
-            // The inversion bet on drawElement and lost — re-render on the
-            // pre-inversion parallel screenshot path instead of single-worker
-            // screenshot streaming (the slowest capture shape for this size).
-            // "reverted" (not cleared) so telemetry keeps the lost-inversion
-            // cohort distinguishable from renders that never inverted.
+          if (failedRouting !== "default") {
+            // The inversion's / router's bet on the pinned path lost. Prefer
+            // the pre-routing parallel screenshot route; a verification
+            // failure alone also retains the already supported low-worker
+            // screenshot stream when that disk route lacks storage headroom
+            // (OOM takes the same target for RAM; other capture failures
+            // revert unconditionally). Routing stays "reverted" (not
+            // cleared) so telemetry keeps the lost-bet cohort distinguishable
+            // from renders that never routed.
+            const route =
+              failedRouting === "worker_inversion" ? "worker inversion" : "parallel router";
+            const diskFallbackLacksHeadroom =
+              failure.kind === "draw_element_verification" &&
+              failure.diskFallbackAvailable === false;
             log.info(
-              `[Render] Reverting worker inversion for the retry: ${capturePlan.workerCount} workers, ` +
-                `plan=${capturePlan.kind}.`,
-            );
-          } else if (failedRouting === "parallel_router") {
-            // The router's bet on verified parallel streaming lost — re-render
-            // on the ordinary (non-DE) parallel path at the pre-router worker
-            // count, same "reverted, not cleared" telemetry contract as the
-            // inversion above.
-            log.info(
-              `[Render] Reverting parallel router for the retry: ${capturePlan.workerCount} workers, ` +
-                `plan=${capturePlan.kind}.`,
+              diskFallbackLacksHeadroom
+                ? `[Render] ${route} disk fallback lacks headroom; retrying with ` +
+                    `${capturePlan.workerCount}-worker screenshot streaming.`
+                : `[Render] Reverting ${route} for the retry: ${capturePlan.workerCount} workers, ` +
+                    `plan=${capturePlan.kind}.`,
             );
           }
           if (capturePlan.kind === "sdr_streaming") {