fix(producer): retain streaming retry after OOM

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

@@ -74,6 +74,11 @@ describe("CapturePlan", () => {
       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" });
 
@@ -87,11 +92,16 @@ describe("CapturePlan", () => {
     expect(Object.isFrozen(next.routing)).toBe(true);
   });
 
-  it("reduces an OOM fallback to one worker without losing immutable routing state", () => {
+  it("retries an inversion OOM in single-worker screenshot streaming mode", () => {
     const initial = streaming({
-      kind: "parallel_router",
+      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: "capture_failure",
@@ -99,7 +109,31 @@ describe("CapturePlan", () => {
     });
 
     expect(next).toMatchObject({
-      kind: "sdr_disk",
+      kind: "sdr_streaming",
+      workerCount: 1,
+      forceScreenshot: true,
+      routing: { kind: "worker_inversion", state: "reverted" },
+    });
+  });
+
+  it("retries a parallel-router OOM in single-worker screenshot streaming mode", () => {
+    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: "capture_failure",
+      memoryExhaustion: true,
+    });
+
+    expect(next).toMatchObject({
+      kind: "sdr_streaming",
       workerCount: 1,
       forceScreenshot: true,
       routing: { kind: "parallel_router", state: "reverted" },

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

@@ -21,6 +21,7 @@ export type CaptureRouting =
       kind: "worker_inversion" | "parallel_router";
       state: "active" | "reverted";
       fallback: CapturePlanTarget;
+      memoryExhaustionFallback: CapturePlanTarget;
     }>;
 
 interface CapturePlanBase {
@@ -83,7 +84,11 @@ function freezeTarget(target: CapturePlanTarget): CapturePlanTarget {
 
 function freezeRouting(routing: CaptureRouting | undefined): CaptureRouting {
   if (!routing || routing.kind === "default") return Object.freeze({ kind: "default" });
-  return Object.freeze({ ...routing, fallback: freezeTarget(routing.fallback) });
+  return Object.freeze({
+    ...routing,
+    fallback: freezeTarget(routing.fallback),
+    memoryExhaustionFallback: freezeTarget(routing.memoryExhaustionFallback),
+  });
 }
 
 export function createCapturePlan(input: CreateCapturePlanInput): CapturePlan {
@@ -132,15 +137,20 @@ export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailur
     });
   }
 
+  const isMemoryExhaustion = failure.kind === "capture_failure" && failure.memoryExhaustion;
   const fallback =
     plan.routing.kind === "default"
-      ? { kind: plan.kind, workerCount: plan.workerCount, forceParallelStream: false }
-      : plan.routing.fallback;
-  const workerCount =
-    failure.kind === "capture_failure" && failure.memoryExhaustion ? 1 : fallback.workerCount;
+      ? {
+          kind: plan.kind,
+          workerCount: isMemoryExhaustion ? 1 : plan.workerCount,
+          forceParallelStream: false,
+        }
+      : isMemoryExhaustion
+        ? plan.routing.memoryExhaustionFallback
+        : plan.routing.fallback;
   return createCapturePlan({
     ...plan,
-    workerCount,
+    workerCount: fallback.workerCount,
     forceScreenshot: true,
     forceParallelStream: fallback.forceParallelStream,
     useStreamingEncode: fallback.kind === "sdr_streaming",

packages/producer/src/services/renderOrchestrator.ts

@@ -2707,6 +2707,14 @@ async function executeRenderPipeline(input: {
       durationSeconds: job.duration,
       isMemoryExhaustion: false,
     });
+    const inversionMemoryExhaustionFallback = resolveInversionRetryPlan({
+      deWorkerInversion,
+      preInversionWorkerCount: preRoutingWorkerCount,
+      cfg,
+      outputFormat,
+      durationSeconds: job.duration,
+      isMemoryExhaustion: true,
+    });
     const parallelRouterFallback = resolveParallelRouterRetryPlan({
       deParallelRouter,
       preRouterWorkerCount: preRoutingWorkerCount,
@@ -2715,27 +2723,50 @@ async function executeRenderPipeline(input: {
       durationSeconds: job.duration,
       isMemoryExhaustion: false,
     });
-    const captureRouting: CaptureRouting = inversionFallback
-      ? {
-          kind: "worker_inversion",
-          state: "active",
-          fallback: {
-            kind: inversionFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk",
-            workerCount: inversionFallback.workerCount,
-            forceParallelStream: false,
-          },
-        }
-      : parallelRouterFallback
+    const parallelRouterMemoryExhaustionFallback = resolveParallelRouterRetryPlan({
+      deParallelRouter,
+      preRouterWorkerCount: preRoutingWorkerCount,
+      cfg,
+      outputFormat,
+      durationSeconds: job.duration,
+      isMemoryExhaustion: true,
+    });
+    const captureRouting: CaptureRouting =
+      inversionFallback && inversionMemoryExhaustionFallback
         ? {
-            kind: "parallel_router",
+            kind: "worker_inversion",
             state: "active",
             fallback: {
-              kind: parallelRouterFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk",
-              workerCount: parallelRouterFallback.workerCount,
+              kind: inversionFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk",
+              workerCount: inversionFallback.workerCount,
+              forceParallelStream: false,
+            },
+            memoryExhaustionFallback: {
+              kind: inversionMemoryExhaustionFallback.useStreamingEncode
+                ? "sdr_streaming"
+                : "sdr_disk",
+              workerCount: inversionMemoryExhaustionFallback.workerCount,
               forceParallelStream: false,
             },
           }
-        : { kind: "default" };
+        : parallelRouterFallback && parallelRouterMemoryExhaustionFallback
+          ? {
+              kind: "parallel_router",
+              state: "active",
+              fallback: {
+                kind: parallelRouterFallback.useStreamingEncode ? "sdr_streaming" : "sdr_disk",
+                workerCount: parallelRouterFallback.workerCount,
+                forceParallelStream: false,
+              },
+              memoryExhaustionFallback: {
+                kind: parallelRouterMemoryExhaustionFallback.useStreamingEncode
+                  ? "sdr_streaming"
+                  : "sdr_disk",
+                workerCount: parallelRouterMemoryExhaustionFallback.workerCount,
+                forceParallelStream: false,
+              },
+            }
+          : { kind: "default" };
     let capturePlan: CapturePlan = createCapturePlan({
       workerCount,
       forceScreenshot: captureForceScreenshot,