fix(engine): a worker death is pool-fatal on the streaming path

On the disk path a transient worker death is deliberately non-fatal: the
orchestrator's adaptive retry re-captures that worker's missing frames. The
streaming path has no per-worker retry and the dead worker's frames are
gone, yet it inherited the same gating — so its peers parked in the ordered
writer waiting for a frame that would never come, and sixty seconds later
the no-progress watchdog relabelled the death as a stall. That is the
`stuck at k/N` field signature.

isPoolFatalWorkerFailure splits the two: every non-cancelled failure ends a
streaming pool, and a new onWorkerFailure hook delivers the ORIGINAL failure
to the caller before peers are aborted, so the ordered writer can be released
with it rather than with a stall. Mutation-checked in both directions,
including that `cancelled` never propagates (the pool is already aborting).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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

@@ -7,6 +7,7 @@ import {
   flagSilentWorkerExits,
   formatWorkerFailure,
   isFfmpegInfrastructureFailure,
+  isPoolFatalWorkerFailure,
   selectVerifySampleIndicesForTask,
   selectWorkerDiagnostics,
   shouldDisableBrowserPoolForParallelWorker,
@@ -17,6 +18,7 @@ import {
   type WorkerResult,
 } from "./parallelCoordinator.js";
 import type { EngineConfig } from "../config.js";
+import { CaptureFailure } from "./captureFailure.js";
 
 describe("parallel worker phase deadline", () => {
   it("fails a wedged operation with phase and browser diagnostics before the aggregate watchdog", async () => {
@@ -512,3 +514,25 @@ describe("isFfmpegInfrastructureFailure", () => {
     expect(isFfmpegInfrastructureFailure(42)).toBe(false);
   });
 });
+
+describe("isPoolFatalWorkerFailure", () => {
+  const failure = (kind: CaptureFailure["kind"]) => new CaptureFailure({ kind, message: kind });
+
+  it("keeps the disk path's tolerance: transient deaths are retried per worker, not pool-fatal", () => {
+    expect(isPoolFatalWorkerFailure(failure("transient_browser"), false)).toBe(false);
+    expect(isPoolFatalWorkerFailure(failure("protocol_timeout"), false)).toBe(false);
+    expect(isPoolFatalWorkerFailure(failure("cancelled"), false)).toBe(false);
+    expect(isPoolFatalWorkerFailure(failure("authoring"), false)).toBe(true);
+    expect(isPoolFatalWorkerFailure(failure("io"), false)).toBe(true);
+  });
+
+  it("makes every non-cancelled failure pool-fatal on the streaming path", () => {
+    // No per-worker retry exists there; the dead worker's frames are gone and
+    // its peers would otherwise park until the watchdog relabels the death.
+    expect(isPoolFatalWorkerFailure(failure("transient_browser"), true)).toBe(true);
+    expect(isPoolFatalWorkerFailure(failure("protocol_timeout"), true)).toBe(true);
+    expect(isPoolFatalWorkerFailure(failure("authoring"), true)).toBe(true);
+    // Cancelled means the pool is already aborting: nothing to propagate.
+    expect(isPoolFatalWorkerFailure(failure("cancelled"), true)).toBe(false);
+  });
+});

packages/engine/src/services/parallelCoordinator.ts

@@ -983,6 +983,31 @@ export function resolveParallelDeVerifySamples(
   return Math.min(8, 4 + 2 * (workerCount - 1));
 }
 
+/**
+ * Whether one worker's failure ends the whole pool. On the disk path a
+ * transient death (Target closed, Page crashed) is not fatal: the
+ * orchestrator's adaptive retry re-captures that worker's missing frames. On
+ * the streaming path (`onFrameBuffer` present) there is no per-worker retry
+ * and the dead worker's frames are gone, so its peers would park in the
+ * ordered writer waiting for a frame that never comes until the producer's
+ * no-progress watchdog relabelled the death as a stall a minute later.
+ * Every non-cancelled failure is therefore pool-fatal there; `cancelled`
+ * means the pool was already aborted and there is nothing left to propagate.
+ */
+export function isPoolFatalWorkerFailure(failure: CaptureFailure, streaming: boolean): boolean {
+  if (streaming) return failure.kind !== "cancelled";
+  return isFatalCaptureFailure(failure);
+}
+
+export interface ParallelCaptureHooks {
+  /**
+   * The first pool-fatal worker failure, delivered BEFORE peers are aborted
+   * so a streaming caller can release anything parked on the dead worker's
+   * frame (the ordered writer) with the original error, not a stall.
+   */
+  onWorkerFailure?: (failure: CaptureFailure) => void;
+}
+
 export async function executeParallelCapture(
   serverUrl: string,
   workDir: string,
@@ -993,6 +1018,7 @@ export async function executeParallelCapture(
   onProgress?: (progress: ParallelProgress) => void,
   onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
   config?: Partial<EngineConfig>,
+  hooks?: ParallelCaptureHooks,
 ): Promise<WorkerResult[]> {
   // `endFrame - startFrame` is the correct per-task frame count for contiguous
   // tasks (stride 1), but for interleaved tasks (stride = workerCount) each
@@ -1050,8 +1076,9 @@ export async function executeParallelCapture(
     : peerController.signal;
   let firstFatalFailure: CaptureFailure | undefined;
   const onFailure = (failure: CaptureFailure): void => {
-    if (firstFatalFailure || !isFatalCaptureFailure(failure)) return;
+    if (firstFatalFailure || !isPoolFatalWorkerFailure(failure, Boolean(onFrameBuffer))) return;
     firstFatalFailure = failure;
+    hooks?.onWorkerFailure?.(failure);
     peerController.abort(failure);
   };
   const results = await Promise.all(