fix(engine): a stalled BeginFrame capture resumes from the first unwritten frame (#6490)

* fix(engine): a stalled BeginFrame capture resumes from the first unwritten frame

A wedged worker used to throw away every frame already written and restart the
whole render on a screenshot session. Fresh BeginFrame workers now continue from
the first unwritten frame (once; a second stall still takes the screenshot
fallback), and the stall message names each worker's in-flight frame and step.

Force-closing the wedged Chrome no longer crashes the CLI with an unhandled
pipe ECONNRESET.

* chore(engine): cut the stall-resume comments to the comment limit

The explanations live in the PR body.

packages/engine/src/index.ts

@@ -308,6 +308,7 @@ export {
   type WorkerSizingBound,
   type LandedFrame,
   type ParallelProgress,
+  type ParallelWorkerPhaseDiagnostic,
 } from "./services/parallelCoordinator.js";
 
 // ── File server ────────────────────────────────────────────────────────────────

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

@@ -1,6 +1,7 @@
 // fallow-ignore-file code-duplication
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { execFileSync } from "node:child_process";
+import { EventEmitter } from "node:events";
 import { chmodSync, existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -1533,6 +1534,21 @@ describe("forceReleaseBrowser", () => {
     expect(disconnectFn).toHaveBeenCalled();
   });
 
+  it("absorbs the CDP pipes' late ECONNRESET so the killed browser cannot crash the process", () => {
+    const pipe = new EventEmitter();
+    const puppeteerListener = () => {};
+    pipe.on("error", puppeteerListener);
+    const mockBrowser = {
+      process: () => ({ kill: vi.fn(() => true), killed: false, stdio: [null, null, null, pipe] }),
+      // Puppeteer's own pipe listeners go away on disconnect; the pipe still errors afterwards.
+      disconnect: () => pipe.off("error", puppeteerListener),
+    } as any;
+
+    forceReleaseBrowser(mockBrowser);
+
+    expect(() => pipe.emit("error", new Error("read ECONNRESET"))).not.toThrow();
+  });
+
   it("tolerates an already-killed process", () => {
     const killFn = vi.fn();
     const disconnectFn = vi.fn();

packages/engine/src/services/browserManager.ts

@@ -946,9 +946,14 @@ async function launchBrowser(
 function forceCloseBrowserProcess(browser: Browser): void {
   const proc = (
     browser as unknown as {
-      process?: () => { kill: (signal?: NodeJS.Signals) => boolean; killed?: boolean } | null;
+      process?: () => {
+        kill: (signal?: NodeJS.Signals) => boolean;
+        killed?: boolean;
+        stdio?: ReadonlyArray<NodeJS.EventEmitter | null | undefined>;
+      } | null;
     }
   ).process?.();
+  for (const stream of proc?.stdio ?? []) stream?.on("error", () => {});
   if (proc && !proc.killed) {
     try {
       proc.kill("SIGKILL");

packages/engine/src/services/parallelCoordinator.ts

@@ -116,6 +116,7 @@ export interface ParallelWorkerPhaseDiagnostic {
   browserVersion: string;
   canvasDrawElement: boolean | "unknown";
   gpuBackend: string;
+  readStep?: () => string | undefined;
 }
 
 export interface WorkerSizingConfig extends Partial<
@@ -572,6 +573,7 @@ async function captureFrameRange(
   phaseTimeoutMs: number,
   diagnosticFor: (phase: ParallelWorkerPhase, frameIndex?: number) => ParallelWorkerPhaseDiagnostic,
   onWorkerPhase: ((diagnostic: ParallelWorkerPhaseDiagnostic) => void) | undefined,
+  onWorkerOperation?: (diagnostic: ParallelWorkerPhaseDiagnostic) => void,
 ): Promise<number> {
   let framesCaptured = 0;
   const outputOffset = task.outputFrameOffset ?? 0;
@@ -581,13 +583,15 @@ async function captureFrameRange(
     frameIndex: number,
     operation: () => Promise<T>,
   ): Promise<T> => {
-    const diagnostic = diagnosticFor(phase, frameIndex);
+    const diagnostic = { ...diagnosticFor(phase, frameIndex), readStep: () => session.captureStep };
     if (framesCaptured === 0) onWorkerPhase?.(diagnostic);
+    onWorkerOperation?.(diagnostic);
     return withParallelWorkerDeadline(operation(), diagnostic, phaseTimeoutMs, signal);
   };
   const awaitEncode = (frameIndex: number, encodeResult: Promise<Buffer>): Promise<Buffer> => {
     const diagnostic = diagnosticFor("frame_encode", frameIndex);
     if (framesCaptured === 0) onWorkerPhase?.(diagnostic);
+    onWorkerOperation?.(diagnostic);
     return withParallelWorkerDeadline(encodeResult, diagnostic, phaseTimeoutMs, signal);
   };
   // Depth-2 pipelined drawElement produce (HF_DE_PARALLEL_STREAM spike): frame
@@ -847,6 +851,7 @@ async function executeWorkerTask(
   parallel?: boolean,
   onFailure?: (failure: CaptureFailure) => void,
   onWorkerPhase?: (diagnostic: ParallelWorkerPhaseDiagnostic) => void,
+  onWorkerOperation?: (diagnostic: ParallelWorkerPhaseDiagnostic) => void,
 ): Promise<WorkerResult> {
   const startTime = Date.now();
   let framesCaptured = 0;
@@ -932,6 +937,7 @@ async function executeWorkerTask(
       phaseTimeoutMs,
       diagnosticFor,
       onWorkerPhase,
+      onWorkerOperation,
     );
 
     await verifyDiskDrawElementSamples(activeSession, task, Boolean(onFrameBuffer));
@@ -1021,6 +1027,7 @@ export interface ParallelCaptureHooks {
    * frame (the ordered writer) with the original error, not a stall.
    */
   onWorkerFailure?: (failure: CaptureFailure) => void;
+  onWorkerOperation?: (diagnostic: ParallelWorkerPhaseDiagnostic) => void;
 }
 
 function pageErrorLinesOf(session: CaptureSession): string[] {
@@ -1152,6 +1159,7 @@ export async function executeParallelCapture(
         parallel,
         onFailure,
         onWorkerPhase,
+        hooks?.onWorkerOperation,
       ),
     ),
   );

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

@@ -28,6 +28,13 @@ const spawnStreamingEncoder = mock(async () => ({
 let failCaptureFrameToBuffer = false;
 let failInitializeSession = false;
 let hangParallelUntilAbort = false;
+// Parallel capture that writes this many frames, then wedges until aborted: on its first attempt,
+// or on every attempt when stallEveryParallelAttempt is set.
+let stallParallelAfterFrames: number | null = null;
+let stallEveryParallelAttempt = false;
+let parallelAttempts = 0;
+const distributeCalls: Array<{ frames: number; rangeStart: number }> = [];
+const reorderStarts: number[] = [];
 // Which phase the wedged worker reports. frame_capture = past init, wedged in
 // a capture call (the watchdog's job); session_init = still booting (not).
 let hangParallelPhase: "session_init" | "frame_capture" = "frame_capture";
@@ -76,6 +83,51 @@ async function replaySlowCapture(
   return [];
 }
 
+// Writes frames to the ordered writer from the range the stage last asked for, then wedges until
+// aborted on the first attempt (every attempt with stallEveryParallelAttempt).
+async function replayStalledCapture(
+  signal: AbortSignal | undefined,
+  onProgress: ((progress: unknown) => void) | undefined,
+  onFrameBuffer: ((frame: number, buffer: Buffer, session: unknown) => Promise<void>) | undefined,
+): Promise<never[]> {
+  const { frames, rangeStart } = distributeCalls[distributeCalls.length - 1]!;
+  const wedges = ++parallelAttempts === 1 || stallEveryParallelAttempt;
+  const end = rangeStart + (wedges ? stallParallelAfterFrames! : frames);
+  const progress = (capturedFrames: number, extra = {}) =>
+    onProgress?.({
+      totalFrames: frames,
+      capturedFrames,
+      activeWorkers: 1,
+      workerProgress: new Map(),
+      ...extra,
+    });
+  progress(0, {
+    latestWorkerPhase: {
+      workerId: 0,
+      phase: "frame_capture",
+      browserExecutable: "chrome",
+      browserVersion: "Chrome/152.0.7977.30",
+      canvasDrawElement: false,
+      gpuBackend: "swiftshader",
+    },
+  });
+  for (let frame = rangeStart; frame < end; frame++) {
+    await onFrameBuffer?.(frame, Buffer.from(`f${frame}`), { captureMode: "beginframe" });
+    progress(frame - rangeStart + 1);
+  }
+  if (wedges) await rejectWhenAborted(signal);
+  return [];
+}
+
+// Simulates a wedged worker: no further progress, then the pool's generic rejection once aborted.
+function rejectWhenAborted(signal: AbortSignal | undefined): Promise<void> {
+  return new Promise<void>((_resolve, reject) => {
+    const fail = () => reject(new Error("[Parallel] Capture failed: aborted"));
+    if (signal?.aborted) return fail();
+    signal?.addEventListener("abort", fail, { once: true });
+  });
+}
+
 mock.module("@hyperframes/engine", () => ({
   calculateOptimalWorkers: () => 1,
   convertTransfer: () => {},
@@ -141,7 +193,8 @@ mock.module("@hyperframes/engine", () => ({
     },
     warnings: [],
   }),
-  createFrameReorderBuffer: () => ({
+  createFrameReorderBuffer: (startFrame: number) => ({
+    ...(reorderStarts.push(startFrame) && {}),
     waitForFrame: async () => {},
     advanceTo: () => {},
     abort: (reason: unknown) => {
@@ -149,7 +202,10 @@ mock.module("@hyperframes/engine", () => ({
     },
   }),
   distributeFrames: () => [],
-  distributeFramesInterleaved: () => [],
+  distributeFramesInterleaved: (frames: number, _workers: number, _dir: string, rangeStart = 0) => {
+    distributeCalls.push({ frames, rangeStart });
+    return [{ workerId: 0, startFrame: rangeStart, endFrame: rangeStart + frames }];
+  },
   DrawElementVerificationError,
   executeParallelCapture: async (
     _url: string,
@@ -159,10 +215,13 @@ mock.module("@hyperframes/engine", () => ({
     _hook: unknown,
     signal?: AbortSignal,
     onProgress?: (progress: unknown) => void,
-    _onFrameBuffer?: unknown,
+    onFrameBuffer?: (frame: number, buffer: Buffer, session: unknown) => Promise<void>,
     _config?: unknown,
     hooks?: { onWorkerFailure?: (failure: Error) => void },
   ) => {
+    if (stallParallelAfterFrames !== null) {
+      return replayStalledCapture(signal, onProgress, onFrameBuffer);
+    }
     if (failWorkerTransient) {
       // One worker's Chrome died. The engine reports the ORIGINAL failure
       // through the hook, then the pool rejects with its flattened summary.
@@ -198,13 +257,7 @@ mock.module("@hyperframes/engine", () => ({
           gpuBackend: "d3d11/nvidia",
         },
       });
-      // Simulate a wedged worker: make no frame progress, then reject with the
-      // pool's generic string once aborted (by the parent or the watchdog).
-      await new Promise<void>((_resolve, reject) => {
-        const fail = () => reject(new Error("[Parallel] Capture failed: aborted"));
-        if (signal?.aborted) return fail();
-        signal?.addEventListener("abort", fail, { once: true });
-      });
+      await rejectWhenAborted(signal);
     }
     if (slowParallel) return replaySlowCapture(signal, onProgress);
     if (parallelFrames > 0) {
@@ -438,6 +491,64 @@ describe("runCaptureStreamingStage", () => {
     expect(input.abortSignal).toBeUndefined();
   });
 
+  describe("a parallel BeginFrame stall", () => {
+    const run = async (afterFrames: number, everyAttempt = false) => {
+      stallParallelAfterFrames = afterFrames;
+      stallEveryParallelAttempt = everyAttempt;
+      parallelAttempts = 0;
+      distributeCalls.length = 0;
+      reorderStarts.length = 0;
+      writeFrame.mockClear();
+      const prev = process.env.HF_DE_STALL_MS;
+      process.env.HF_DE_STALL_MS = "50";
+      const { runCaptureStreamingStage } = await import("./captureStreamingStage.js");
+      const baseInput = createInput({ forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 });
+      const input = {
+        ...baseInput,
+        totalFrames: 10,
+        plan: { ...baseInput.plan, workerCount: 2, forceParallelStream: true },
+        parallelCaptureLabel: "BeginFrame",
+      };
+      try {
+        return await runCaptureStreamingStage(input).then(
+          (result) => ({ result, caught: undefined as unknown }),
+          (caught: unknown) => ({ result: undefined, caught }),
+        );
+      } finally {
+        stallParallelAfterFrames = null;
+        stallEveryParallelAttempt = false;
+        if (prev === undefined) delete process.env.HF_DE_STALL_MS;
+        else process.env.HF_DE_STALL_MS = prev;
+      }
+    };
+    const written = () => writeFrame.mock.calls.map(([buffer]) => buffer.toString());
+
+    it("resumes from the first unwritten frame, writing every frame once and in order", async () => {
+      const { result, caught } = await run(4);
+      expect(caught).toBeUndefined();
+      expect(result?.success).toBe(true);
+      expect(written()).toEqual(Array.from({ length: 10 }, (_, i) => `f${i}`));
+      expect(distributeCalls).toEqual([
+        { frames: 10, rangeStart: 0 },
+        { frames: 6, rangeStart: 4 },
+      ]);
+      expect(reorderStarts.slice(-2)).toEqual([0, 4]);
+    });
+
+    it("falls back to the typed stall when the resumed workers stall again", async () => {
+      const { caught } = await run(4, true);
+      expect((caught as Error).name).toBe("ParallelCaptureStallError");
+      expect((caught as Error).message).toContain("last frame 8/10");
+      expect(distributeCalls).toHaveLength(2);
+    });
+
+    it("does not resume a stall that wrote nothing new", async () => {
+      const { caught } = await run(0);
+      expect((caught as Error).name).toBe("ParallelCaptureStallError");
+      expect(distributeCalls).toHaveLength(1);
+    });
+  });
+
   it("does not arm the watchdog while every worker is still initialising", async () => {
     // Browsers launching and sub-composition timelines settling can take
     // longer than the stall window on a heavy composition. Counting that as

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

@@ -49,6 +49,7 @@ import {
   type CapturePerfSummary,
   type CaptureSession,
   type EngineConfig,
+  type ParallelWorkerPhaseDiagnostic,
   type StreamingEncoder,
   DrawElementVerificationError,
   captureFrameToBuffer,
@@ -720,9 +721,6 @@ export async function runCaptureStreamingStage(
       // drawElement produce when its session initialized in drawelement mode.
       const deParallelStream =
         forceParallelStream === true || process.env.HF_DE_PARALLEL_STREAM === "true";
-      const tasks = deParallelStream
-        ? distributeFramesInterleaved(totalFrames, workerCount, workDir)
-        : distributeFrames(totalFrames, workerCount, workDir);
 
       // Drain-time safety net for parallel drawElement frames: the SAME
       // blank-guard + PSNR self-verify the sequential worker-encode drain
@@ -756,185 +754,192 @@ export async function runCaptureStreamingStage(
         frameTime: (i: number) => (i * job.config.fps.den) / job.config.fps.num,
       });
       let parallelGuardRan = false;
-      // First guard/write failure aborts the reorder buffer so peer workers
-      // parked in waitForFrame reject instead of deadlocking the pool (the
-      // pool awaits ALL workers before surfacing errors). The original error
-      // is rethrown below so DrawElementVerificationError keeps its type —
-      // executeParallelCapture flattens worker errors to strings.
-      let parallelDrainError: Error | null = null;
-      const onFrameBuffer = async (
-        frameIndex: number,
-        buffer: Buffer,
-        workerSession: CaptureSession,
-      ): Promise<void> => {
-        try {
-          if (deParallelStream && workerSession.captureMode === "drawelement") {
-            parallelGuardRan = true;
-            buffer = await parallelGuard(workerSession, frameIndex, buffer);
+      let writtenFrames = 0;
+      let resumeFrom = 0;
+      let resumedAfterStall = false;
+      let workerResults: Awaited<ReturnType<typeof executeParallelCapture>>;
+      for (;;) {
+        const parallelReorder = createFrameReorderBuffer(resumeFrom, totalFrames);
+        const tasks = deParallelStream
+          ? distributeFramesInterleaved(totalFrames - resumeFrom, workerCount, workDir, resumeFrom)
+          : distributeFrames(totalFrames - resumeFrom, workerCount, workDir, resumeFrom);
+        // The first guard or write failure aborts the reorder buffer so parked peers reject, not
+        // deadlock.
+        let parallelDrainError: Error | null = null;
+        const onFrameBuffer = async (
+          frameIndex: number,
+          buffer: Buffer,
+          workerSession: CaptureSession,
+        ): Promise<void> => {
+          try {
+            if (deParallelStream && workerSession.captureMode === "drawelement") {
+              parallelGuardRan = true;
+              buffer = await parallelGuard(workerSession, frameIndex, buffer);
+            }
+            await parallelReorder.waitForFrame(frameIndex);
+            ensureFrameWritten(await currentEncoder.writeFrame(buffer), frameIndex, currentEncoder);
+            parallelReorder.advanceTo(frameIndex + 1);
+            writtenFrames = frameIndex + 1;
+          } catch (err) {
+            const e = err instanceof Error ? err : new Error(String(err));
+            if (!parallelDrainError) {
+              parallelDrainError = e;
+              parallelReorder.abort(e);
+            }
+            throw e;
           }
-          await reorderBuffer.waitForFrame(frameIndex);
-          ensureFrameWritten(await currentEncoder.writeFrame(buffer), frameIndex, currentEncoder);
-          reorderBuffer.advanceTo(frameIndex + 1);
-        } catch (err) {
-          const e = err instanceof Error ? err : new Error(String(err));
-          if (!parallelDrainError) {
-            parallelDrainError = e;
-            reorderBuffer.abort(e);
-          }
-          throw e;
+        };
+
+        // Stall watchdog on its own controller, so a trip is not read as a cancel; armed by first
+        // progress.
+        // A trip aborts the reorder buffer so peers parked in waitForFrame reject.
+        const captureLabel = input.parallelCaptureLabel ?? "worker";
+        let watchdogArmed = false;
+        const stallController = new AbortController();
+        const forwardParentAbort = () => stallController.abort();
+        if (abortSignal) {
+          if (abortSignal.aborted) stallController.abort();
+          else abortSignal.addEventListener("abort", forwardParentAbort, { once: true });
         }
-      };
-
-      // Progress-stall watchdog. Wired through a SEPARATE controller (linked
-      // to the parent abort) rather than the job's `abortSignal` so a
-      // watchdog trip stays distinguishable from a real cancellation: the
-      // orchestrator gates its pinned fallback on `abortSignal.aborted`, which
-      // stays false here, so a stall re-renders via screenshot instead of
-      // being swallowed as a cancel. On trip we also abort the reorder buffer
-      // so peer workers parked in `waitForFrame` reject instead of deadlocking
-      // the pool (executeParallelCapture awaits ALL workers).
-      //
-      // The clock is ARMED, not just reset, by progress: it starts counting
-      // only once some worker has reached its first capture call or written
-      // a frame. Before that the workers are launching browsers and settling
-      // sub-composition timelines and media, which on a heavy composition
-      // legitimately takes longer than the stall window — the sequential path
-      // starts its clock after initializeSession for the same reason. Init
-      // itself is bounded by the engine (browserTimeout, playerReadyTimeout).
-      const captureLabel = input.parallelCaptureLabel ?? "worker";
-      let watchdogArmed = false;
-      const stallController = new AbortController();
-      const forwardParentAbort = () => stallController.abort();
-      if (abortSignal) {
-        if (abortSignal.aborted) stallController.abort();
-        else abortSignal.addEventListener("abort", forwardParentAbort, { once: true });
-      }
-      const stallTimeoutMs = resolveCaptureStallTimeoutMs();
-      let lastCapturedFrames = 0;
-      let lastProgressAt = Date.now();
-      const workerPhases = new Map<number, string>();
-      const phaseSummary = () =>
-        [...workerPhases.entries()]
-          .sort(([left], [right]) => left - right)
-          .map(([workerId, detail]) => `worker=${workerId} ${detail}`)
-          .join("; ");
-      let stalled = false;
-      const stallTimer = setInterval(
-        () => {
-          if (!watchdogArmed || Date.now() - lastProgressAt <= stallTimeoutMs) return;
-          stalled = true;
-          const stallErr = new ParallelCaptureStallError({
-            variant: "drain",
-            captureLabel,
-            stallTimeoutMs,
-            lastCapturedFrames,
-            totalFrames,
-            phaseSummary: phaseSummary(),
-          });
-          reorderBuffer.abort(stallErr);
-          stallController.abort();
-        },
-        Math.min(stallTimeoutMs, DE_STALL_POLL_MS),
-      );
-
-      let workerResults;
-      try {
-        workerResults = await executeParallelCapture(
-          fileServer.url,
-          workDir,
-          tasks,
-          buildCaptureOptions(),
-          createRenderVideoFrameInjector,
-          stallController.signal,
-          (progress) => {
-            if (progress.latestWorkerPhase) {
-              const phase = progress.latestWorkerPhase;
-              const detail =
-                `phase=${phase.phase} frame=${phase.frameIndex ?? "n/a"} ` +
-                `browser=${phase.browserExecutable} version=${phase.browserVersion} ` +
-                `CanvasDrawElement=${phase.canvasDrawElement} gpu=${phase.gpuBackend}`;
-              workerPhases.set(phase.workerId, detail);
-              if (phase.phase === "frame_capture" && !watchdogArmed) {
-                watchdogArmed = true;
-                lastProgressAt = Date.now();
-              }
-              log.info("[Render] Parallel capture worker phase", {
-                workerId: phase.workerId,
-                phase: phase.phase,
-                frameIndex: phase.frameIndex,
-                browserExecutable: phase.browserExecutable,
-                browserVersion: phase.browserVersion,
-                canvasDrawElement: phase.canvasDrawElement,
-                gpuBackend: phase.gpuBackend,
-              });
-              if (progress.capturedFrames === 0) reportWorkerStartup(job, progress, onProgress);
-              return;
-            }
-            if (progress.capturedFrames > lastCapturedFrames) {
-              lastCapturedFrames = progress.capturedFrames;
-              lastProgressAt = Date.now();
-              watchdogArmed = true;
-            }
-            job.framesRendered = progress.capturedFrames;
-            const frameProgress = progress.capturedFrames / progress.totalFrames;
-            reportFrameProgress(
-              job,
-              `Streaming frame ${progress.capturedFrames}/${progress.totalFrames} (${workerCount} workers)`,
-              Math.round(25 + frameProgress * 55),
-              onProgress,
-              progress.capturedFrames,
-              progress.totalFrames,
-            );
-          },
-          onFrameBuffer,
-          // Interleaved DE workers each need their own browser PROCESS:
-          // pages co-tenant in one browser share one compositor, whose
-          // internal frame scheduling deprioritizes non-active pages — their
-          // canvas `paint` events starve and the drawElement paint-wait slows
-          // ~3x (measured: 86s vs 30s on a 3,245-frame rAF comp). This is
-          // Chromium's internal frame-production scheduling on ALL platforms,
-          // NOT the Linux-only HeadlessExperimental.beginFrame capture mode.
-          deParallelStream ? { ...captureCfg, enableBrowserPool: false } : captureCfg,
-          {
-            // A worker died (Target closed, Page crashed, launch failure).
-            // Its frames are gone and peers are parked in waitForFrame on the
-            // first of them; release them with the ORIGINAL failure so the
-            // orchestrator sees a transient browser error it knows how to
-            // retry, not the stall the watchdog would have raised a minute
-            // later. First failure wins, same as the drain guard above.
-            onWorkerFailure: (failure) => {
-              if (parallelDrainError) return;
-              parallelDrainError = failure;
-              reorderBuffer.abort(failure);
-            },
+        const stallTimeoutMs = resolveCaptureStallTimeoutMs();
+        let lastCapturedFrames = 0;
+        let lastProgressAt = Date.now();
+        const workerPhases = new Map<number, string>();
+        const phaseDetail = (phase: ParallelWorkerPhaseDiagnostic) =>
+          `phase=${phase.phase} frame=${phase.frameIndex ?? "n/a"} step=${phase.readStep?.() ?? "n/a"} ` +
+          `browser=${phase.browserExecutable} version=${phase.browserVersion} ` +
+          `CanvasDrawElement=${phase.canvasDrawElement} gpu=${phase.gpuBackend}`;
+        const phaseSummary = () =>
+          [...workerPhases.entries()]
+            .sort(([left], [right]) => left - right)
+            .map(([workerId, detail]) => `worker=${workerId} ${detail}`)
+            .join("; ");
+        let stalled = false;
+        const stallTimer = setInterval(
+          () => {
+            if (!watchdogArmed || Date.now() - lastProgressAt <= stallTimeoutMs) return;
+            stalled = true;
+            const stallErr = new ParallelCaptureStallError({
+              variant: "drain",
+              captureLabel,
+              stallTimeoutMs,
+              lastCapturedFrames: resumeFrom + lastCapturedFrames,
+              totalFrames,
+              phaseSummary: phaseSummary(),
+            });
+            parallelReorder.abort(stallErr);
+            stallController.abort();
           },
+          Math.min(stallTimeoutMs, DE_STALL_POLL_MS),
         );
-      } catch (err) {
-        // Surface the TYPED drain error (DrawElementVerificationError) so the
-        // orchestrator's verify-retry handler recognizes it — the worker pool
-        // flattens worker errors into a plain message string.
-        if (parallelDrainError) throw poolFailure(parallelDrainError, probeSession);
-        // A watchdog trip aborts the pool via stallController, so the pool
-        // rejects with a generic "[Parallel] Capture failed" string. Replace
-        // it with a clear, non-cancellation stall error (the parent abort did
-        // NOT fire) so the orchestrator's pinned fallback re-renders via
-        // screenshot instead of masking a 5-min hang.
-        if (stalled && abortSignal?.aborted !== true) {
-          throw new ParallelCaptureStallError({
-            variant: "pool",
-            captureLabel,
-            stallTimeoutMs,
-            lastCapturedFrames,
-            totalFrames,
-            phaseSummary: phaseSummary(),
-          });
+
+        try {
+          workerResults = await executeParallelCapture(
+            fileServer.url,
+            workDir,
+            tasks,
+            buildCaptureOptions(),
+            createRenderVideoFrameInjector,
+            stallController.signal,
+            (progress) => {
+              if (progress.latestWorkerPhase) {
+                const phase = progress.latestWorkerPhase;
+                workerPhases.set(phase.workerId, phaseDetail(phase));
+                if (phase.phase === "frame_capture" && !watchdogArmed) {
+                  watchdogArmed = true;
+                  lastProgressAt = Date.now();
+                }
+                log.info("[Render] Parallel capture worker phase", {
+                  workerId: phase.workerId,
+                  phase: phase.phase,
+                  frameIndex: phase.frameIndex,
+                  browserExecutable: phase.browserExecutable,
+                  browserVersion: phase.browserVersion,
+                  canvasDrawElement: phase.canvasDrawElement,
+                  gpuBackend: phase.gpuBackend,
+                });
+                if (progress.capturedFrames === 0) reportWorkerStartup(job, progress, onProgress);
+                return;
+              }
+              if (progress.capturedFrames > lastCapturedFrames) {
+                lastCapturedFrames = progress.capturedFrames;
+                lastProgressAt = Date.now();
+                watchdogArmed = true;
+              }
+              const doneFrames = resumeFrom + progress.capturedFrames;
+              job.framesRendered = doneFrames;
+              const frameProgress = doneFrames / totalFrames;
+              reportFrameProgress(
+                job,
+                `Streaming frame ${doneFrames}/${totalFrames} (${workerCount} workers)`,
+                Math.round(25 + frameProgress * 55),
+                onProgress,
+                doneFrames,
+                totalFrames,
+              );
+            },
+            onFrameBuffer,
+            // Interleaved DE workers each get their own browser process: co-tenant pages starve
+            // each other's paint events.
+            deParallelStream ? { ...captureCfg, enableBrowserPool: false } : captureCfg,
+            {
+              // A worker died: release parked peers with its original failure, not the stall the
+              // watchdog raises later.
+              onWorkerOperation: (phase) => workerPhases.set(phase.workerId, phaseDetail(phase)),
+              onWorkerFailure: (failure) => {
+                if (parallelDrainError) return;
+                parallelDrainError = failure;
+                parallelReorder.abort(failure);
+              },
+            },
+          );
+        } catch (err) {
+          // A wedged worker loses nothing already written: BeginFrame ticks derive from the
+          // absolute frame index, so fresh workers continue the same stream from the first
+          // unwritten frame. A second stall, or one before any new frame landed, falls through.
+          if (
+            stalled &&
+            abortSignal?.aborted !== true &&
+            captureLabel === "BeginFrame" &&
+            !resumedAfterStall &&
+            writtenFrames > resumeFrom
+          ) {
+            resumedAfterStall = true;
+            log.warn(
+              "[Render] Parallel BeginFrame capture stalled; resuming from the first unwritten frame",
+              {
+                resumeFrom: writtenFrames,
+                totalFrames,
+                workerPhases: phaseSummary(),
+              },
+            );
+            resumeFrom = writtenFrames;
+            continue;
+          }
+          // Surface the TYPED drain error (DrawElementVerificationError) so the
+          // orchestrator's verify-retry handler recognizes it — the worker pool
+          // flattens worker errors into a plain message string.
+          if (parallelDrainError) throw poolFailure(parallelDrainError, probeSession);
+          // A watchdog trip rejects the pool with a generic string; replace it with the typed stall
+          // error.
+          if (stalled && abortSignal?.aborted !== true) {
+            throw new ParallelCaptureStallError({
+              variant: "pool",
+              captureLabel,
+              stallTimeoutMs,
+              lastCapturedFrames: resumeFrom + lastCapturedFrames,
+              totalFrames,
+              phaseSummary: phaseSummary(),
+            });
+          }
+          throw poolFailure(err, probeSession);
+        } finally {
+          clearInterval(stallTimer);
+          abortSignal?.removeEventListener("abort", forwardParentAbort);
         }
-        throw poolFailure(err, probeSession);
-      } finally {
-        clearInterval(stallTimer);
-        abortSignal?.removeEventListener("abort", forwardParentAbort);
+        if (parallelDrainError) throw poolFailure(parallelDrainError, probeSession);
+        break;
       }
-      if (parallelDrainError) throw poolFailure(parallelDrainError, probeSession);
       pushWorkerDedupPerfs(workerResults, dedupPerfs);
       const workerPageErrors: string[] = [];
       pushWorkerPageErrors(workerResults, workerPageErrors);