fix(engine): self-verify parallel disk drawElement samples

packages/cli/src/commands/render.test.ts

@@ -123,6 +123,20 @@ vi.mock("../telemetry/config.js", () => ({
     configState.cache = { ...configState.disk };
     return { ...configState.disk };
   }),
+  recordRecentRender: vi.fn((id: string, ok: boolean) => {
+    // Mirrors the real ring update (readConfigFresh → append, cap 5 → write)
+    // against the mock's disk state, so a render's recent-renders write is
+    // modeled like every other config mutation here. Fixed timestamp keeps it
+    // deterministic (tests never assert on `at`).
+    const disk = configState.disk as Record<string, unknown>;
+    const ring = [
+      ...((disk.recentRenders as unknown[]) ?? []),
+      { id, at: "2026-01-01T00:00:00Z", ok },
+    ];
+    const next = { ...disk, recentRenders: ring.slice(-5) };
+    configState.disk = next;
+    configState.cache = { ...next };
+  }),
   writeConfig: vi.fn((config: Record<string, unknown>) => {
     configState.writeConfigCalls.push({ ...config });
     if (configState.failWrites > 0) {

packages/engine/src/index.ts

@@ -209,6 +209,8 @@ export type {
 export {
   calculateOptimalWorkers,
   computeWorkerSizing,
+  selectVerifySampleIndicesForTask,
+  verifyDiskDrawElementSamples,
   distributeFrames,
   distributeFramesInterleaved,
   executeParallelCapture,
@@ -273,6 +275,10 @@ export {
 
 export { trackChildProcess, killTrackedProcesses } from "./utils/processTracker.js";
 
+// drawElement self-verify comparison — shared by the streaming drain
+// (producer) and the parallel disk-path verify (parallelCoordinator).
+export { psnrDb, resolveDeVerifyMinDb } from "./utils/psnr.js";
+
 export {
   decodePng,
   decodePngToRgb48le,

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

@@ -6,6 +6,7 @@ import {
   expectedFramesForTask,
   flagSilentWorkerExits,
   formatWorkerFailure,
+  selectVerifySampleIndicesForTask,
   selectWorkerDiagnostics,
   shouldDisableBrowserPoolForParallelWorker,
   shouldVerifyWorkerGpu,
@@ -349,6 +350,35 @@ describe("flagSilentWorkerExits", () => {
   });
 });
 
+describe("selectVerifySampleIndicesForTask", () => {
+  it("keeps only samples inside the task's contiguous range, sorted", () => {
+    // 2-worker split of 3032 frames: worker 1 owns [1516, 3032).
+    expect(
+      selectVerifySampleIndicesForTask([2274, 758, 1516, 3031, 3032], {
+        startFrame: 1516,
+        endFrame: 3032,
+      }),
+    ).toEqual([1516, 2274, 3031]);
+  });
+
+  it("respects the stride lattice for interleaved tasks", () => {
+    // Worker 1 of a 3-way interleave over [1, 30): captures 1, 4, 7, ...
+    expect(
+      selectVerifySampleIndicesForTask([1, 2, 4, 6, 7, 28, 29], {
+        startFrame: 1,
+        endFrame: 30,
+        frameStride: 3,
+      }),
+    ).toEqual([1, 4, 7, 28]);
+  });
+
+  it("returns empty when no samples fall in the range", () => {
+    expect(
+      selectVerifySampleIndicesForTask([0, 10, 20], { startFrame: 100, endFrame: 200 }),
+    ).toEqual([]);
+  });
+});
+
 describe("shouldVerifyWorkerGpu", () => {
   const softwareConfig: Partial<EngineConfig> = { browserGpuMode: "software" };
 

packages/engine/src/services/parallelCoordinator.ts

@@ -7,7 +7,7 @@
 
 import { cpus, freemem } from "os";
 import { existsSync, mkdirSync, readdirSync } from "fs";
-import { copyFile, rename } from "fs/promises";
+import { copyFile, readFile, rename } from "fs/promises";
 import { join } from "path";
 import { getHeapStatistics } from "v8";
 
@@ -19,11 +19,13 @@ import {
   captureFrameToBufferPipelined,
   captureFrameToBuffer,
   getCapturePerfSummary,
+  DrawElementVerificationError,
   type CaptureSession,
   type CaptureOptions,
   type CapturePerfSummary,
   type BeforeCaptureHook,
 } from "./frameCapture.js";
+import { psnrDb, resolveDeVerifyMinDb } from "../utils/psnr.js";
 import { DEFAULT_CONFIG, type EngineConfig } from "../config.js";
 import { assertSwiftShader } from "../utils/assertSwiftShader.js";
 import { readWebGlVendorInfoFromCanvas } from "../utils/readWebGlVendorInfoFromCanvas.js";
@@ -551,6 +553,126 @@ async function captureFrameRange(
   return framesCaptured;
 }
 
+/**
+ * The armed self-verify sample indices this task actually captured: inside
+ * `[startFrame, endFrame)` and on the task's stride lattice. Mirrors the
+ * capture loop in `captureFrameRange` (`i += stride` from `startFrame`).
+ */
+export function selectVerifySampleIndicesForTask(
+  sampleIndices: Iterable<number>,
+  task: Pick<WorkerTask, "startFrame" | "endFrame" | "frameStride">,
+): number[] {
+  const stride = task.frameStride ?? 1;
+  const selected: number[] = [];
+  for (const idx of sampleIndices) {
+    if (idx < task.startFrame || idx >= task.endFrame) continue;
+    if ((idx - task.startFrame) % stride !== 0) continue;
+    selected.push(idx);
+  }
+  return selected.sort((a, b) => a - b);
+}
+
+/**
+ * Disk-path drawElement self-verification (PRINFRA-352). Parallel DISK
+ * workers arm the same pre-injection ground-truth samples as the streaming
+ * path (`resolveParallelDeVerifySamples` even raises the density for
+ * multi-worker capture) — but only the streaming drain ever CHECKED them,
+ * so an explicit `--experimental-fast-capture --workers N` render shipped
+ * unverified drawElement frames. On a 16GB host, two concurrent
+ * hardware-GPU Chrome instances hit the documented compositor-tile-eviction
+ * damage class (frames displaced into vertical strips for one worker's
+ * whole range — reads as "corruption from the exact worker boundary").
+ *
+ * After a worker's range completes, re-read its captured files for the
+ * sampled indices and PSNR-compare against the session's ground truth.
+ * A breach throws `DrawElementVerificationError`, which the orchestrator's
+ * existing pinned-fallback retry converts into a screenshot re-render —
+ * the same recovery the streaming drain gets.
+ */
+/** HF_DE_PAR_DEBUG=1 gated per-worker trace line (message built lazily). */
+function logParDebug(message: () => string): void {
+  if (process.env.HF_DE_PAR_DEBUG === "1") console.log(message());
+}
+
+/**
+ * Throw the verification error for a sample below the PSNR floor; log the
+ * pass otherwise. Split from the sampling loop for the complexity gate.
+ */
+function assertDiskSampleAboveFloor(
+  db: number,
+  verifyMinDb: number,
+  idx: number,
+  workerId: number,
+): void {
+  if (db < verifyMinDb) {
+    // Message keeps the contiguous "drawElement self-verify" phrase —
+    // captureFailure's VERIFICATION_ERROR_PATTERNS classifies on it.
+    throw new DrawElementVerificationError(
+      `drawElement self-verify failed at frame ${idx} (disk path, worker ${workerId}): ` +
+        `${db.toFixed(1)}dB < ${verifyMinDb}dB vs pre-injection screenshot`,
+      { kind: "psnr", frameIndex: idx, failedDb: db, verifyThresholdDb: verifyMinDb },
+    );
+  }
+  console.log(
+    `[Parallel] drawElement disk self-verify passed (worker ${workerId}, frame ${idx}, ` +
+      `${db === Infinity ? "inf" : db.toFixed(1)}dB)`,
+  );
+}
+
+/**
+ * Compare one captured frame file against its ground truth. Returns the
+ * PSNR, or null on infrastructure failure (missing file already surfaces
+ * via the frame completeness check; ffmpeg spawn/tmpdir here) — a skipped
+ * sample is not damage evidence and must not fail the capture.
+ */
+async function psnrForDiskSample(
+  framePath: string,
+  truth: Buffer,
+  workerId: number,
+  idx: number,
+): Promise<number | null> {
+  try {
+    return await psnrDb(await readFile(framePath), truth);
+  } catch (err) {
+    console.warn(
+      `[Parallel] drawElement disk self-verify sample skipped (worker ${workerId}, ` +
+        `frame ${idx}): ${err instanceof Error ? err.message : String(err)}`,
+    );
+    return null;
+  }
+}
+
+// Branches are the gate conditions themselves (mode/armed/streaming guards +
+// per-sample skip/breach) — already decomposed into psnrForDiskSample +
+// assertDiskSampleAboveFloor; further splitting obscures the check.
+// fallow-ignore-next-line complexity
+export async function verifyDiskDrawElementSamples(
+  session: CaptureSession,
+  task: WorkerTask,
+  streaming: boolean,
+): Promise<void> {
+  // Streaming capture verifies every sampled frame in the drain guard already.
+  if (streaming || session.captureMode !== "drawelement") return;
+  const truths = session.deVerifyFrames;
+  if (!truths || truths.size === 0) return;
+  const verifyMinDb = resolveDeVerifyMinDb();
+  const ext = session.options.format === "png" ? "png" : "jpg";
+  const offset = task.outputFrameOffset ?? 0;
+  for (const idx of selectVerifySampleIndicesForTask(truths.keys(), task)) {
+    const truth = truths.get(idx);
+    if (!truth) continue;
+    const framePath = join(task.outputDir, `frame_${String(idx - offset).padStart(6, "0")}.${ext}`);
+    const db = await psnrForDiskSample(framePath, truth, task.workerId, idx);
+    if (db === null) continue;
+    assertDiskSampleAboveFloor(db, verifyMinDb, idx, task.workerId);
+  }
+}
+
+// Inherited worker-lifecycle shape (session create → verify GPU → init →
+// capture → self-verify → perf, with a classifying catch + closing finally);
+// flagged only because the disk self-verify call shifted its line range into
+// the changed-code audit. Not restructured by this PR.
+// fallow-ignore-next-line complexity
 async function executeWorkerTask(
   task: WorkerTask,
   serverUrl: string,
@@ -590,19 +712,16 @@ async function executeWorkerTask(
       createBeforeCaptureHook(),
       workerConfig,
     );
-    if (process.env.HF_DE_PAR_DEBUG === "1") {
-      console.log(`[par:w${task.workerId}] session created`);
-    }
+    logParDebug(() => `[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})`,
-      );
-    }
+    logParDebug(
+      () =>
+        `[par:w${task.workerId}] init done (mode=${session?.captureMode} workerEncode=${session?.workerEncodeEnabled === true})`,
+    );
     framesCaptured = await captureFrameRange(
       session,
       task,
@@ -612,6 +731,8 @@ async function executeWorkerTask(
       onFrameBuffer,
     );
 
+    await verifyDiskDrawElementSamples(session, task, Boolean(onFrameBuffer));
+
     perf = getCapturePerfSummary(session);
     return {
       workerId: task.workerId,

packages/engine/src/utils/psnr.ts

@@ -0,0 +1,45 @@
+import { execFile } from "node:child_process";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { promisify } from "node:util";
+import { getFfmpegBinary } from "./ffmpegBinaries.js";
+
+/**
+ * PSNR (average, dB) between two same-dimension encoded images via ffmpeg.
+ * Infinity means bit-identical pixels. Single source of truth for every
+ * drawElement self-verify comparison (streaming drain + parallel disk path).
+ */
+export async function psnrDb(a: Buffer, b: Buffer): Promise<number> {
+  // promisify(execFile) lazily, not at module load: this module is in the
+  // engine's parallel-capture import chain, and a top-level call to a builtin
+  // crashes any downstream test that partially mocks node:child_process
+  // without an execFile export (vitest surfaces it as a load-time error).
+  const execFileP = promisify(execFile);
+  const dir = await mkdtemp(join(tmpdir(), "hf-de-verify-"));
+  try {
+    const pa = join(dir, "a.jpg");
+    const pb = join(dir, "b.jpg");
+    await Promise.all([writeFile(pa, a), writeFile(pb, b)]);
+    const { stderr } = await execFileP(
+      getFfmpegBinary(),
+      ["-hide_banner", "-i", pa, "-i", pb, "-lavfi", "psnr", "-f", "null", "-"],
+      { maxBuffer: 4 * 1024 * 1024 },
+    );
+    const m = /average:(inf|[\d.]+)/.exec(stderr);
+    if (!m) throw new Error(`psnr parse failed: ${stderr.slice(-300)}`);
+    return m[1] === "inf" ? Infinity : Number(m[1]);
+  } finally {
+    await rm(dir, { recursive: true, force: true }).catch(() => {});
+  }
+}
+
+/**
+ * The drawElement self-verify PSNR floor (dB). HF_DE_VERIFY_MIN_DB overrides,
+ * clamped to [10, 60]; out-of-range or unset falls back to 32 — the threshold
+ * every prior eval used to separate real compositor damage from encoder noise.
+ */
+export function resolveDeVerifyMinDb(): number {
+  const raw = Number(process.env.HF_DE_VERIFY_MIN_DB ?? "32");
+  return Number.isFinite(raw) && raw >= 10 && raw <= 60 ? raw : 32;
+}

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

@@ -140,6 +140,28 @@ describe("CapturePlan", () => {
     });
   });
 
+  it("forces screenshot on a disk-plan drawElement verify failure (PRINFRA-352 recovery)", () => {
+    // Parallel disk workers under the explicit fast-capture opt-in verify their
+    // own captured samples; a breach must re-render the DISK plan on the
+    // screenshot baseline (not throw, and not stay on drawElement).
+    const disk = createCapturePlan({
+      workerCount: 2,
+      forceScreenshot: false,
+      useStreamingEncode: false,
+      useLayeredComposite: false,
+      usePageSideCompositing: false,
+      hasHdrContent: false,
+      needsAlpha: false,
+    });
+    const next = replanAfterFailure(disk, { kind: "draw_element_verification" });
+    expect(next).toMatchObject({
+      kind: "sdr_disk",
+      forceScreenshot: true,
+      forceParallelStream: false,
+      workerCount: 2,
+    });
+  });
+
   it("rejects a streaming transition from a non-streaming plan", () => {
     const disk = createCapturePlan({
       workerCount: 2,

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

@@ -124,6 +124,18 @@ function revertedRouting(routing: CaptureRouting): CaptureRouting {
 
 /** Pure, exhaustive capture fallback transition. The input plan is never mutated. */
 export function replanAfterFailure(plan: CapturePlan, failure: CapturePlanFailure): CapturePlan {
+  // Disk-path drawElement self-verification (parallel disk workers under the
+  // explicit fast-capture opt-in) can also trip — the retry stays on the disk
+  // path but forces the screenshot baseline.
+  if (plan.kind === "sdr_disk" && failure.kind === "draw_element_verification") {
+    return createCapturePlan({
+      ...plan,
+      forceScreenshot: true,
+      useStreamingEncode: false,
+      useLayeredComposite: false,
+      forceParallelStream: false,
+    });
+  }
   if (plan.kind !== "sdr_streaming") {
     throw new Error(`Cannot apply ${failure.kind} to ${plan.kind} capture plan`);
   }

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

@@ -44,6 +44,7 @@ import {
   type EngineConfig,
   captureFrame,
   captureFrameToBufferPipelined,
+  verifyDiskDrawElementSamples,
   writeCapturedFrame,
   closeCaptureSession,
   completeDeferredDrawElementInit,
@@ -331,6 +332,24 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
           reportFrame(i);
         }
       }
+      // Sequential disk drawElement self-verification (PRINFRA-352 follow-up):
+      // the sequential disk path — reachable under the explicit fast-capture
+      // opt-in, including via probe-session reuse — armed ground-truth samples
+      // but never checked them, exactly like the parallel disk workers before
+      // #2749. Same synthetic-task shape the parallel verify uses; a breach
+      // throws DrawElementVerificationError and the orchestrator's disk-stage
+      // retry re-renders via screenshot.
+      await verifyDiskDrawElementSamples(
+        session,
+        {
+          workerId: 0,
+          startFrame: rangeStart,
+          endFrame: rangeEnd,
+          outputDir: framesDir,
+          outputFrameOffset: rangeStart,
+        },
+        false,
+      );
       // Capture the sequential session's static-dedup perf before close (the
       // counters are valid only while the session is live).
       dedupPerfs.push(getCapturePerfSummary(session));