feat(engine,producer): verified interleaved parallel drawElement streaming (opt-in)

Step 2 of the DE engagement plan: multi-worker drawElement capture through
the streaming encoder, with the full runtime self-verification net riding
along — the confinement rule that kept the parallel clamp in place is now
satisfied on this path. Opt-in via HF_DE_PARALLEL_STREAM=true; default
routing (including the #2026 single-worker inversion) is unchanged.

Mechanism:
- distributeFramesInterleaved + WorkerTask.frameStride: worker i captures
  frames i, i+N, i+2N... — seek-based capture makes stride free and the
  ordered writer's reorder window shrinks from totalFrames/N to N (contiguous
  chunks serialize workers behind the writer).
- Depth-2 pipelined worker-encode produce in the parallel worker loop (the
  same shape as the sequential loop; frame k's in-page encode overlaps
  k+stride's produce). HF_DE_PAR_DEBUG=1 traces the first frames per worker.
- Drain guard extracted to createDrainFrameGuard (session-parameterized):
  every parallel frame gets the SAME blank-guard + PSNR self-verify as the
  sequential drain, against its owning worker's pre-injection ground truth
  (all sessions arm identical sample indices from
  CaptureOptions.compositionDurationSeconds).
- FrameReorderBuffer.abort(err): a failed worker (e.g. verification error)
  rejects all parked and future waiters — without this, peers park forever
  in waitForFrame and the pool (which awaits ALL workers before surfacing
  errors) deadlocks. Found by the verify-trip test; unit-tested.
- The typed DrawElementVerificationError is preserved past the pool's
  error-string flattening so the orchestrator's verify-retry recognizes it.
- Static-dedup stride hazard fixed: lastEncodeResult reuse now requires EVERY
  frame in (lastEncodeResultFrame, i] to be predicted-static (sequential
  capture reduces to the old has(i) check).
- Workers get separate browser PROCESSES under the flag: pages co-tenant in
  one browser starve non-active pages of BeginFrames on the paint-wait path
  (measured 86s vs 30s on a 3,245-frame rAF comp).

Validation:
- Happy path W3: verify samples pass across workers (4x inf on the 2,381f
  comp), output vs single-worker DE = 59.3dB (encode noise floor) — the
  interleave + dedup-stride produce identical pixels.
- Verify-trip (marginal comp + HF_DE_VERIFY_MIN_DB=45): fails at frame 649
  (32.2dB < 45), peers abort instead of deadlocking, whole render retries
  via parallel screenshot, RENDER_OK in 42.6s.
- Canary suite 7/7 with the flag off (default paths untouched); producer
  orchestrator tests 99/99; engine suite 909 passed (14 pre-existing main
  failures, stash-A/B verified); reorder-buffer abort unit tests.

Perf note: capture-only parallel speedup measured 1.38x (W2) / 1.52x (W3)
over single-worker DE in the spike; end-to-end numbers on this machine are
currently noisy (separate-browser init overhead + bench load) — clean
benchmarks before any default routing change. The flag stays explicit
opt-in; promoting it into the router replaces the #2026 W=1 pin for the
same cohort.

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

Author: vanceingalls

packages/engine/src/index.ts

@@ -176,6 +176,7 @@ export type {
 export {
   calculateOptimalWorkers,
   distributeFrames,
+  distributeFramesInterleaved,
   executeParallelCapture,
   mergeWorkerFrames,
   getSystemResources,

packages/engine/src/services/frameCapture.ts

@@ -160,6 +160,11 @@ export interface CaptureSession {
    * worker-encode path (mirrors `lastFrameBuffer` on the screenshot path). Only set
    * when static-frame dedup is armed on the drawElement path. */
   lastEncodeResult?: Promise<Buffer>;
+  /** Frame index lastEncodeResult belongs to — static-dedup reuse must verify
+   * every frame in (lastEncodeResultFrame, i] is predicted-static (under the
+   * interleaved parallel stride the "previous" produced frame is i−N, and
+   * reusing it for frame i is only valid when the whole gap is static). */
+  lastEncodeResultFrame?: number;
   /** Per-render self-verification ground truth (ungated-release safety net):
    * K screenshot frames captured at init BEFORE the drawElement canvas is
    * injected (the only window where a page screenshot shows the live DOM, not
@@ -2640,9 +2645,23 @@ export async function captureFrameToBufferPipelined(
   // and skip the seek + drawElement + encode entirely. Same predicate as the serial
   // path; clip-cut frames are excluded from staticFrames so they always capture.
   if (session.staticFrames?.has(frameIndex) && session.lastEncodeResult) {
-    session.staticDedupCount = (session.staticDedupCount ?? 0) + 1;
-    session.capturePerf.frames += 1;
-    return { encodeResult: session.lastEncodeResult, captureTimeMs: Date.now() - startTime };
+    // Reuse is valid only when EVERY frame in (lastEncodeResultFrame, i] is
+    // predicted-static — then all of them (and the reused buffer) share the
+    // same pixels. Sequential capture reduces to has(i) (gap = {i}); the
+    // interleaved parallel stride makes the gap N frames wide.
+    const lastIdx = session.lastEncodeResultFrame ?? frameIndex - 1;
+    let gapStatic = true;
+    for (let j = lastIdx + 1; j <= frameIndex; j++) {
+      if (!session.staticFrames.has(j)) {
+        gapStatic = false;
+        break;
+      }
+    }
+    if (gapStatic) {
+      session.staticDedupCount = (session.staticDedupCount ?? 0) + 1;
+      session.capturePerf.frames += 1;
+      return { encodeResult: session.lastEncodeResult, captureTimeMs: Date.now() - startTime };
+    }
   }
 
   try {
@@ -2677,7 +2696,10 @@ export async function captureFrameToBufferPipelined(
         session.capturePerf.frameMs.push(boundaryMs);
       }
       const boundaryResult = Promise.resolve(buffer);
-      if (session.staticFrames) session.lastEncodeResult = boundaryResult;
+      if (session.staticFrames) {
+        session.lastEncodeResult = boundaryResult;
+        session.lastEncodeResultFrame = frameIndex;
+      }
       return { encodeResult: boundaryResult, captureTimeMs: Date.now() - startTime };
     }
 
@@ -2704,7 +2726,10 @@ export async function captureFrameToBufferPipelined(
     session.capturePerf.frameMs.push(captureTimeMs);
 
     // Task B: retain this encode result so a following static frame can reuse it.
-    if (session.staticFrames) session.lastEncodeResult = encodeResult;
+    if (session.staticFrames) {
+      session.lastEncodeResult = encodeResult;
+      session.lastEncodeResultFrame = frameIndex;
+    }
 
     return { encodeResult, captureTimeMs };
   } catch (captureError) {

packages/engine/src/services/parallelCoordinator.ts

@@ -15,6 +15,7 @@ import {
   initializeSession,
   closeCaptureSession,
   captureFrame,
+  captureFrameToBufferPipelined,
   captureFrameToBuffer,
   getCapturePerfSummary,
   type CaptureSession,
@@ -42,6 +43,13 @@ export interface WorkerTask {
    * calculation still uses the absolute frame index.
    */
   outputFrameOffset?: number;
+  /**
+   * Frame stride for interleaved distribution (HF_DE_PARALLEL_STREAM spike):
+   * the worker captures startFrame, startFrame+stride, … < endFrame. Default 1
+   * (contiguous range). Interleaving keeps the ordered streaming writer's
+   * reorder window at O(workerCount) frames instead of O(totalFrames/N).
+   */
+  frameStride?: number;
 }
 
 export interface WorkerResult {
@@ -234,6 +242,34 @@ export function distributeFrames(
   return tasks;
 }
 
+/**
+ * Interleaved (round-robin) distribution: worker i captures frames
+ * i, i+N, i+2N, …. Seek-based capture makes stride access free (every frame
+ * is an absolute seek), and the streaming reorder window shrinks from
+ * totalFrames/N to N — contiguous chunks serialize workers behind the
+ * ordered writer (worker 1's first frame waits for ALL of worker 0's).
+ * HF_DE_PARALLEL_STREAM spike; disk-path capture keeps contiguous chunks.
+ */
+export function distributeFramesInterleaved(
+  totalFrames: number,
+  workerCount: number,
+  workDir: string,
+  rangeStart: number = 0,
+): WorkerTask[] {
+  const tasks: WorkerTask[] = [];
+  for (let i = 0; i < workerCount && i < totalFrames; i++) {
+    tasks.push({
+      workerId: i,
+      startFrame: rangeStart + i,
+      endFrame: rangeStart + totalFrames,
+      frameStride: workerCount,
+      outputDir: join(workDir, `worker-${i}`),
+      outputFrameOffset: rangeStart,
+    });
+  }
+  return tasks;
+}
+
 /**
  * Decide whether a parallel worker should run the per-worker SwiftShader
  * assertion. Gated to worker 0 only: workers within a chunk share the same
@@ -244,24 +280,77 @@ export function shouldVerifyWorkerGpu(workerId: number, config?: Partial<EngineC
   return config?.browserGpuMode === "software" && workerId === 0;
 }
 
+// fallow-ignore-next-line complexity
 async function captureFrameRange(
   session: CaptureSession,
   task: WorkerTask,
   captureOptions: CaptureOptions,
   signal: AbortSignal | undefined,
   onFrameCaptured: ((workerId: number, frameIndex: number) => void) | undefined,
-  onFrameBuffer: ((frameIndex: number, buffer: Buffer) => Promise<void>) | undefined,
+  onFrameBuffer:
+    | ((frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>)
+    | undefined,
 ): Promise<number> {
   let framesCaptured = 0;
   const outputOffset = task.outputFrameOffset ?? 0;
-  for (let i = task.startFrame; i < task.endFrame; i++) {
+  const stride = task.frameStride ?? 1;
+  // Depth-2 pipelined drawElement produce (HF_DE_PARALLEL_STREAM spike): frame
+  // k's in-page worker encode overlaps frame k+stride's produce phase — the
+  // same shape as the sequential worker-encode loop. Only engaged when the
+  // session's encode worker initialized (drawElement mode) and frames stream
+  // back via onFrameBuffer; the ordered writer's waitForFrame provides the
+  // cross-worker backpressure (each worker runs at most `stride` frames ahead).
+  if (onFrameBuffer && session.workerEncodeEnabled) {
+    const dbg = process.env.HF_DE_PAR_DEBUG === "1";
+    const dbgT0 = Date.now();
+    const dbgWin = 40 * stride;
+    let prev: { idx: number; encodeResult: Promise<Buffer> } | null = null;
+    for (let i = task.startFrame; i < task.endFrame; i += stride) {
+      if (signal?.aborted) throw new Error("Parallel worker cancelled");
+      const time = (i * captureOptions.fps.den) / captureOptions.fps.num;
+      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);
+      if (dbg && i < task.startFrame + dbgWin) {
+        console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms produce ${i} kicked`);
+      }
+      if (prev) {
+        if (dbg && prev.idx < task.startFrame + dbgWin) {
+          console.log(
+            `[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} await-encode`,
+          );
+        }
+        const buf = await prev.encodeResult;
+        if (dbg && prev.idx < task.startFrame + dbgWin) {
+          console.log(
+            `[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} encoded ${buf.length}B`,
+          );
+        }
+        await onFrameBuffer(prev.idx, buf, session);
+        if (dbg && prev.idx < task.startFrame + dbgWin) {
+          console.log(`[par:w${task.workerId}] +${Date.now() - dbgT0}ms drain ${prev.idx} written`);
+        }
+        framesCaptured++;
+        if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx);
+      }
+      prev = { idx: i, encodeResult };
+    }
+    if (prev) {
+      await onFrameBuffer(prev.idx, await prev.encodeResult, session);
+      framesCaptured++;
+      if (onFrameCaptured) onFrameCaptured(task.workerId, prev.idx);
+    }
+    return framesCaptured;
+  }
+  for (let i = task.startFrame; i < task.endFrame; i += stride) {
     if (signal?.aborted) throw new Error("Parallel worker cancelled");
     const time = (i * captureOptions.fps.den) / captureOptions.fps.num;
     const fileFrameIdx = i - outputOffset;
 
     if (onFrameBuffer) {
       const { buffer } = await captureFrameToBuffer(session, fileFrameIdx, time);
-      await onFrameBuffer(i, buffer);
+      await onFrameBuffer(i, buffer, session);
     } else {
       await captureFrame(session, fileFrameIdx, time);
     }
@@ -278,7 +367,7 @@ async function executeWorkerTask(
   createBeforeCaptureHook: () => BeforeCaptureHook | null,
   signal?: AbortSignal,
   onFrameCaptured?: (workerId: number, frameIndex: number) => void,
-  onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise<void>,
+  onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
   config?: Partial<EngineConfig>,
   parallel?: boolean,
 ): Promise<WorkerResult> {
@@ -309,11 +398,19 @@ async function executeWorkerTask(
       createBeforeCaptureHook(),
       workerConfig,
     );
+    if (process.env.HF_DE_PAR_DEBUG === "1") {
+      console.log(`[par:w${task.workerId}] session created`);
+    }
     // Worker-0-only SwiftShader assertion — see `shouldVerifyWorkerGpu` and #955.
     if (shouldVerifyWorkerGpu(task.workerId, workerConfig)) {
       await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas);
     }
     await initializeSession(session);
+    if (process.env.HF_DE_PAR_DEBUG === "1") {
+      console.log(
+        `[par:w${task.workerId}] init done (mode=${session.captureMode} workerEncode=${session.workerEncodeEnabled === true})`,
+      );
+    }
     framesCaptured = await captureFrameRange(
       session,
       task,
@@ -358,7 +455,7 @@ export async function executeParallelCapture(
   createBeforeCaptureHook: () => BeforeCaptureHook | null,
   signal?: AbortSignal,
   onProgress?: (progress: ParallelProgress) => void,
-  onFrameBuffer?: (frameIndex: number, buffer: Buffer) => Promise<void>,
+  onFrameBuffer?: (frameIndex: number, buffer: Buffer, session: CaptureSession) => Promise<void>,
   config?: Partial<EngineConfig>,
 ): Promise<WorkerResult[]> {
   const totalFrames = tasks.reduce((sum, t) => sum + (t.endFrame - t.startFrame), 0);

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

@@ -905,3 +905,23 @@ describe("spawnStreamingEncoder lifecycle and cleanup", () => {
     }
   });
 });
+
+describe("createFrameReorderBuffer abort (interleaved parallel drain)", () => {
+  it("rejects parked waiters so a failed worker cannot deadlock its peers", async () => {
+    const buf = createFrameReorderBuffer(0, 100);
+    const parked = buf.waitForFrame(5);
+    const err = new Error("verify failed");
+    buf.abort(err);
+    await expect(parked).rejects.toThrow("verify failed");
+    // Future waiters reject immediately too.
+    await expect(buf.waitForFrame(6)).rejects.toThrow("verify failed");
+    await expect(buf.waitForAllDone()).rejects.toThrow("verify failed");
+  });
+
+  it("in-order waiters still resolve before an abort", async () => {
+    const buf = createFrameReorderBuffer(0, 10);
+    await expect(buf.waitForFrame(0)).resolves.toBeUndefined();
+    buf.advanceTo(1);
+    await expect(buf.waitForFrame(1)).resolves.toBeUndefined();
+  });
+});

packages/engine/src/services/streamingEncoder.ts

@@ -52,35 +52,48 @@ export interface FrameReorderBuffer {
   waitForFrame: (frame: number) => Promise<void>;
   advanceTo: (frame: number) => void;
   waitForAllDone: () => Promise<void>;
+  /**
+   * Reject every parked and future waiter with `err`. Required by the
+   * interleaved parallel drain: when one worker fails (e.g. drawElement
+   * self-verification), its frames will never be written — peers parked in
+   * waitForFrame would otherwise deadlock the whole capture (the worker pool
+   * awaits ALL workers before surfacing the failure).
+   */
+  abort: (err: Error) => void;
 }
 
 export function createFrameReorderBuffer(startFrame: number, endFrame: number): FrameReorderBuffer {
   let cursor = startFrame;
-  const pending = new Map<number, Array<() => void>>();
+  let aborted: Error | null = null;
+  const pending = new Map<number, Array<{ resolve: () => void; reject: (e: Error) => void }>>();
 
-  const enqueueAt = (frame: number, resolve: () => void): void => {
+  const enqueueAt = (frame: number, resolve: () => void, reject: (e: Error) => void): void => {
     const list = pending.get(frame);
     if (list === undefined) {
-      pending.set(frame, [resolve]);
+      pending.set(frame, [{ resolve, reject }]);
     } else {
-      list.push(resolve);
+      list.push({ resolve, reject });
     }
   };
 
   const flushAt = (frame: number): void => {
     const list = pending.get(frame);
     if (list === undefined) return;
     pending.delete(frame);
-    for (const resolve of list) resolve();
+    for (const waiter of list) waiter.resolve();
   };
 
   const waitForFrame = (frame: number): Promise<void> =>
-    new Promise<void>((resolve) => {
+    new Promise<void>((resolve, reject) => {
+      if (aborted) {
+        reject(aborted);
+        return;
+      }
       if (frame === cursor) {
         resolve();
         return;
       }
-      enqueueAt(frame, resolve);
+      enqueueAt(frame, resolve, reject);
     });
 
   const advanceTo = (frame: number): void => {
@@ -89,15 +102,28 @@ export function createFrameReorderBuffer(startFrame: number, endFrame: number):
   };
 
   const waitForAllDone = (): Promise<void> =>
-    new Promise<void>((resolve) => {
+    new Promise<void>((resolve, reject) => {
+      if (aborted) {
+        reject(aborted);
+        return;
+      }
       if (cursor >= endFrame) {
         resolve();
         return;
       }
-      enqueueAt(endFrame, resolve);
+      enqueueAt(endFrame, resolve, reject);
     });
 
-  return { waitForFrame, advanceTo, waitForAllDone };
+  const abort = (err: Error): void => {
+    if (aborted) return;
+    aborted = err;
+    for (const [frame, list] of pending) {
+      pending.delete(frame);
+      for (const waiter of list) waiter.reject(err);
+    }
+  };
+
+  return { waitForFrame, advanceTo, waitForAllDone, abort };
 }
 
 // ---------------------------------------------------------------------------