fix(producer): trip DE parallel-router circuit breaker on stalls and hangs

packages/cli/src/commands/render.ts

@@ -1314,8 +1314,9 @@ function persistDeParallelRouterTrialFired(): boolean {
 
 /**
  * After a trial-armed render, persist that the router's OWN bet actually
- * failed — its self-verify/generic-failure safety net fired
- * (`deParallelRouter === "reverted"`) — or that the render-count backstop
+ * failed — its self-verify/generic-failure safety net fired (recorded as
+ * anything other than a clean `"routed"`, e.g. `"reverted"`, or a stall/hang
+ * outcome — heygen-com/hyperframes#3441) — or that the render-count backstop
  * (`DE_PARALLEL_ROUTER_TRIAL_MAX_RENDERS`) was reached, so it's never
  * enabled again for this install. A clean "routed" (the render succeeded
  * with no fallback) does NOT consume the trial by itself — the whole point
@@ -1352,11 +1353,19 @@ function maybeConsumeDeParallelRouterTrial(
   const config = readConfigFresh();
   const renderCount = (config.deParallelRouterTrialRenderCount ?? 0) + 1;
   config.deParallelRouterTrialRenderCount = renderCount;
-  // Trip ONLY on an actual fallback. The old trial also tripped at a
-  // 25-render exposure cap, which was sampling logic: bound how long an
-  // experiment force-enables itself. Under a shipped default that would
-  // switch the feature off behind the user's back after 25 good renders.
-  const fired = outcome === "reverted";
+  // Trip on any recorded non-success, not only the literal string
+  // "reverted". The old trial also tripped at a 25-render exposure cap,
+  // which was sampling logic: bound how long an experiment force-enables
+  // itself. Under a shipped default that would switch the feature off
+  // behind the user's back after 25 good renders — so a clean "routed" (no
+  // fallback needed) must NOT trip. But narrowing the positive check to the
+  // single string "reverted" (heygen-com/hyperframes#3441) meant any other
+  // non-success signal the observability layer might ever record — a stall,
+  // a timeout, a future outcome value — would silently fall through to "not
+  // fired" instead of tripping. `outcome` is `undefined`-filtered above, so
+  // by this point it is a real recorded outcome; the only one that means
+  // "no fallback happened" is "routed" itself.
+  const fired = outcome !== "routed";
   if (fired) {
     config.deParallelRouterTrialFired = true;
     // Latch BEFORE attempting persistence — the decision holds for this

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

@@ -61,4 +61,57 @@ describe("executeParallelCapture peer abort", () => {
       rmSync(root, { recursive: true, force: true });
     }
   });
+
+  // heygen-com/hyperframes#3441: a worker wedged INSIDE a native capture call
+  // (WSL2 hangs the very first drawElement/BeginFrame call at frame 0 with no
+  // error) must actually be unstuck once the caller's `signal` aborts — e.g.
+  // the DE parallel-router stall watchdog in `captureStreamingStage.ts` firing
+  // after HF_DE_STALL_MS. Before this fix, `captureFrameRange` only checked
+  // `signal.aborted` BEFORE starting each frame's capture call, which is a
+  // no-op for a call that is already in flight and never settles — the abort
+  // had no way to reach it, so `executeParallelCapture`'s `Promise.all` (and
+  // therefore the whole render) hung forever, and the CLI circuit breaker
+  // (which only runs after `executeRenderJob` settles) never got a chance to
+  // trip.
+  it("rejects promptly when the signal aborts while a worker is wedged inside a capture call that never settles", async () => {
+    const root = mkdtempSync(join(tmpdir(), "hf-stall-abort-"));
+    // Simulates the native capture call hanging indefinitely (never resolves,
+    // never rejects) — exactly the WSL2 shape from the field report.
+    const captureFrame = vi.fn(() => new Promise<void>(() => {}));
+    const closeCaptureSession = vi.fn().mockResolvedValue(undefined);
+    vi.doMock("./frameCapture.js", () => ({
+      createCaptureSession: vi.fn(
+        async () => ({ workerId: 0, browserConsoleBuffer: [] }) as unknown as CaptureSession,
+      ),
+      initializeSession: vi.fn(async () => {}),
+      captureFrame,
+      captureFrameToBuffer: vi.fn(),
+      captureFrameToBufferPipelined: vi.fn(),
+      closeCaptureSession,
+      getCapturePerfSummary: vi.fn(() => ({ frames: 0 })),
+    }));
+
+    try {
+      const { executeParallelCapture } = await import("./parallelCoordinator.js");
+      const controller = new AbortController();
+      const result = executeParallelCapture(
+        "http://127.0.0.1",
+        root,
+        [{ workerId: 0, startFrame: 0, endFrame: 3, outputDir: join(root, "worker-0") }],
+        { width: 320, height: 180, fps: { num: 30, den: 1 } },
+        () => null,
+        controller.signal,
+      );
+
+      // Give the worker a tick to actually enter the (never-resolving)
+      // capture call before simulating the watchdog's abort.
+      await new Promise((resolve) => setTimeout(resolve, 10));
+      controller.abort();
+
+      await expect(result).rejects.toMatchObject({ name: "CaptureFailure" });
+      expect(captureFrame).toHaveBeenCalledTimes(1);
+    } finally {
+      rmSync(root, { recursive: true, force: true });
+    }
+  });
 });

packages/engine/src/services/parallelCoordinator.ts

@@ -461,6 +461,43 @@ export function shouldVerifyWorkerGpu(workerId: number, config?: Partial<EngineC
   return config?.browserGpuMode === "software" && workerId === 0;
 }
 
+/**
+ * Race a single in-flight capture call against `signal` actually firing.
+ *
+ * `captureFrame`/`captureFrameToBuffer`/`captureFrameToBufferPipelined` take
+ * no abort signal of their own — a native browser call wedged inside one of
+ * them (WSL2 hangs the very first drawElement/BeginFrame capture at frame 0
+ * with no error, heygen-com/hyperframes#3441) cannot be cancelled, only
+ * raced. Without this, the `signal?.aborted` checks at the top of the
+ * `captureFrameRange` loop are a no-op the moment a worker is already
+ * awaiting a hung call: nothing revisits that check until the await settles,
+ * which on a genuine hang is never. The DE parallel-router's stall watchdog
+ * (`captureStreamingStage.ts`) does fire `stallController.abort()` after
+ * `HF_DE_STALL_MS`, but until this call actually observes the signal, that
+ * abort has no effect on an already-wedged worker — `executeParallelCapture`'s
+ * `Promise.all` waits forever, so the render hangs indefinitely and the CLI's
+ * circuit breaker (which only runs after `executeRenderJob` settles) never
+ * gets a chance to trip.
+ */
+function raceAgainstAbort<T>(promise: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
+  if (!signal) return promise;
+  if (signal.aborted) return Promise.reject(new Error("Parallel worker cancelled"));
+  return new Promise<T>((resolve, reject) => {
+    const onAbort = () => reject(new Error("Parallel worker cancelled"));
+    signal.addEventListener("abort", onAbort, { once: true });
+    promise.then(
+      (value) => {
+        signal.removeEventListener("abort", onAbort);
+        resolve(value);
+      },
+      (error) => {
+        signal.removeEventListener("abort", onAbort);
+        reject(error);
+      },
+    );
+  });
+}
+
 // fallow-ignore-next-line complexity
 async function captureFrameRange(
   session: CaptureSession,
@@ -498,7 +535,10 @@ async function captureFrameRange(
       if (dbg && i < task.startFrame + dbgWin) {
         console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} start`);
       }
-      const { encodeResult } = await captureFrameToBufferPipelined(session, i - outputOffset, time);
+      const { encodeResult } = await raceAgainstAbort(
+        captureFrameToBufferPipelined(session, i - outputOffset, time),
+        signal,
+      );
       // Marks the promise "handled" for Node's unhandled-rejection detector
       // without affecting the real `await prev.encodeResult` below — if a
       // later iteration throws (abort, downstream writeFrame failure) before
@@ -542,10 +582,13 @@ async function captureFrameRange(
     const fileFrameIdx = i - outputOffset;
 
     if (onFrameBuffer) {
-      const { buffer } = await captureFrameToBuffer(session, fileFrameIdx, time);
+      const { buffer } = await raceAgainstAbort(
+        captureFrameToBuffer(session, fileFrameIdx, time),
+        signal,
+      );
       await onFrameBuffer(i, buffer, session);
     } else {
-      await captureFrame(session, fileFrameIdx, time);
+      await raceAgainstAbort(captureFrame(session, fileFrameIdx, time), signal);
     }
     framesCaptured++;
     if (onFrameCaptured) onFrameCaptured(task.workerId, i);