Merge pull request #2821 from heygen-com/fix/distributed-beginframe-fallback

fix(producer): fallback distributed capture safely

packages/aws-lambda/src/handler.ts

@@ -18,6 +18,7 @@ import { S3Client } from "@aws-sdk/client-s3";
 import {
   assemble,
   type AssembleResult,
+  type ChunkRenderer,
   type ChunkResult,
   type DistributedRenderConfig,
   listPlanV2ArtifactsForTarget,
@@ -77,7 +78,7 @@ export interface HandlerDeps {
   primitives?: {
     plan: typeof plan;
     planV2WithPublisher?: typeof planV2WithPublisher;
-    renderChunk: typeof renderChunk;
+    renderChunk: ChunkRenderer;
     assemble: typeof assemble;
   };
   /** Override the per-invocation `/tmp` workdir root (defaults to Lambda's `/tmp`). */

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

@@ -183,6 +183,16 @@ describe("createVideoFrameInjector cache hygiene against page-side skips", () =>
     return `data:image/png;base64,fake-${framePath}`;
   }
 
+  function makeGpuInjector() {
+    const evaluate = vi.fn(async () => undefined);
+    const page = { evaluate } as unknown as Page;
+    const hook = createVideoFrameInjector(
+      fakeTable({ videoId: "facet", framePath: "/f", frameIndex: 3 }),
+      { frameSrcResolver: inlineResolver },
+    );
+    return { evaluate, page, hook };
+  }
+
   beforeEach(() => {
     injectVideoFramesBatchMock.mockReset();
     syncVideoFrameVisibilityMock.mockReset();
@@ -256,18 +266,33 @@ describe("createVideoFrameInjector cache hygiene against page-side skips", () =>
     expect(injectVideoFramesBatchMock).toHaveBeenCalledTimes(1);
   });
 
+  it("reinjects frame zero when a whole-chunk retry uses a fresh hook and page", async () => {
+    const table = fakeTable({ videoId: "hero", framePath: "/frame-0", frameIndex: 0 });
+    const beginFramePage = { evaluate: async () => undefined } as unknown as Page;
+    const screenshotPage = { evaluate: async () => undefined } as unknown as Page;
+    const beginFrameHook = createVideoFrameInjector(table, {
+      frameSrcResolver: inlineResolver,
+    });
+    const screenshotRetryHook = createVideoFrameInjector(table, {
+      frameSrcResolver: inlineResolver,
+    });
+
+    injectVideoFramesBatchMock.mockResolvedValue(["hero"]);
+    await beginFrameHook!(beginFramePage, 0);
+    await screenshotRetryHook!(screenshotPage, 0);
+
+    expect(injectVideoFramesBatchMock).toHaveBeenCalledTimes(2);
+    expect(injectVideoFramesBatchMock.mock.calls[0]?.[0]).toBe(beginFramePage);
+    expect(injectVideoFramesBatchMock.mock.calls[1]?.[0]).toBe(screenshotPage);
+  });
+
   // Regression: WebGL/WebGPU compositions that sample a <video> as a texture
   // render on `hf-seek` BEFORE frames are injected. After injecting the
   // decoded frames, the hook must re-render the GPU adapters at the same time
   // (window.__hfReseekGpu) so they re-upload their textures from the fresh
   // frames — otherwise the facet flickers / goes black non-deterministically.
   it("re-renders GPU adapters after injecting frames (post-injection reseek)", async () => {
-    const evaluate = vi.fn(async () => undefined);
-    const page = { evaluate } as unknown as Page;
-    const hook = createVideoFrameInjector(
-      fakeTable({ videoId: "facet", framePath: "/f", frameIndex: 3 }),
-      { frameSrcResolver: inlineResolver },
-    );
+    const { evaluate, page, hook } = makeGpuInjector();
 
     injectVideoFramesBatchMock.mockResolvedValueOnce(["facet"]);
     await hook!(page, 1.5);
@@ -284,12 +309,7 @@ describe("createVideoFrameInjector cache hygiene against page-side skips", () =>
   });
 
   it("does not reseek GPU when the page injected no frames", async () => {
-    const evaluate = vi.fn(async () => undefined);
-    const page = { evaluate } as unknown as Page;
-    const hook = createVideoFrameInjector(
-      fakeTable({ videoId: "facet", framePath: "/f", frameIndex: 3 }),
-      { frameSrcResolver: inlineResolver },
-    );
+    const { evaluate, page, hook } = makeGpuInjector();
 
     // Page dropped the video (e.g. hidden host) → nothing injected → no reseek.
     injectVideoFramesBatchMock.mockResolvedValueOnce([]);

packages/gcp-cloud-run/src/server.ts

@@ -26,6 +26,7 @@ import { Hono } from "hono";
 import {
   assemble,
   type AssembleResult,
+  type ChunkRenderer,
   type ChunkResult,
   type DistributedRenderConfig,
   listPlanV2ArtifactsForTarget,
@@ -84,7 +85,7 @@ export interface HandlerDeps {
   primitives?: {
     plan: typeof plan;
     planV2WithPublisher?: typeof planV2WithPublisher;
-    renderChunk: typeof renderChunk;
+    renderChunk: ChunkRenderer;
     assemble: typeof assemble;
   };
   /** Override the per-request workdir root (defaults to the OS tmpdir). */

packages/producer/src/distributed.ts

@@ -89,7 +89,9 @@ export {
   readWebGlVendorInfoFromCanvas,
   renderChunk,
   // Types
+  type ChunkRenderer,
   type ChunkResult,
+  type EffectiveChunkResult,
   // Error codes + classes
   FFMPEG_VERSION_MISMATCH,
   PLAN_HASH_MISMATCH,

packages/producer/src/index.ts

@@ -163,7 +163,9 @@ export {
   renderChunkV2,
   validatePlanV2MaterializedTarget,
   type AssembleResult,
+  type ChunkRenderer,
   type ChunkResult,
+  type EffectiveChunkResult,
   type DistributedRenderCapabilities,
   type DistributedRenderConfig,
   type PlanProtocolConsumerCapabilities,

packages/producer/src/services/distributed/planV2Execution.ts

@@ -3,7 +3,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { assemble, type AssembleResult } from "./assemble.js";
 import { materializePlanV2Target } from "./planV2.js";
-import { renderChunk, type ChunkResult } from "./renderChunk.js";
+import { renderChunk, type EffectiveChunkResult } from "./renderChunk.js";
 
 /**
  * Direct v2 chunk-role entry point. Storage adapters may instead download the
@@ -13,7 +13,7 @@ export async function renderChunkV2(
   planV2Dir: string,
   chunkIndex: number,
   outputChunkPath: string,
-): Promise<ChunkResult> {
+): Promise<EffectiveChunkResult> {
   const workRoot = mkdtempSync(join(tmpdir(), "hf-plan-v2-chunk-"));
   const materializedPlanDir = join(workRoot, "plan");
   try {

packages/producer/src/services/distributed/renderChunk.ts

@@ -42,6 +42,7 @@ import {
   type BeforeCaptureHook,
   BROWSER_GPU_NOT_SOFTWARE,
   calculateOptimalWorkers,
+  classifyCaptureFailure,
   type CaptureOptions,
   type CaptureMode,
   type CapturePerfSummary,
@@ -52,8 +53,10 @@ import {
   createVideoFrameInjector,
   type EngineConfig,
   type ExtractedFrames,
+  type FrameLookupTable,
   getEncoderPreset,
   initializeSession,
+  probeBeginFrameLiveness,
   readWebGlVendorInfoFromCanvas,
   resolveConfig,
 } from "@hyperframes/engine";
@@ -143,9 +146,8 @@ export interface ChunkResult {
    * overhead from frame-proportional work in fleet cost models:
    *
    * - `planHashMs` — full planDir content-hash recomputation (validation).
-   * - `sessionBootMs` — sequential-branch Chrome boot + SwiftShader assert +
-   *   composition warmup. Stays 0 when `workers > 1` (each parallel worker
-   *   boots inside the capture stage instead).
+   * - `sessionBootMs` — Chrome boot + SwiftShader assert + composition warmup
+   *   for the reusable sequential session or parallel BeginFrame preflight.
    * - `captureStageMs` — the capture stage call; includes per-worker session
    *   boots in the parallel branch.
    * - `encodeStageMs` — the encode stage call (single ffmpeg invocation, or
@@ -160,8 +162,14 @@ export interface ChunkResult {
   encodeStageMs: number;
   /** Capture workers used for this chunk (`calculateOptimalWorkers` result). */
   workers: number;
-  /** Effective engine mode used by every worker, after any browser fallback. */
-  captureMode: CaptureMode;
+  /**
+   * Effective engine mode used by every worker, after any browser fallback.
+   *
+   * Current first-party renderers always emit this field. It remains optional
+   * so adapters can accept results from older or injected chunk renderers
+   * without inventing an observed mode that may be false.
+   */
+  captureMode?: CaptureMode;
   /**
    * Path to a sidecar JSON containing per-chunk perf counters. Adapters
    * upload this alongside the chunk so per-chunk regressions are
@@ -170,6 +178,132 @@ export interface ChunkResult {
   perfPath: string;
 }
 
+/** Result returned by the built-in renderer, which always observes its mode. */
+export interface EffectiveChunkResult extends ChunkResult {
+  captureMode: CaptureMode;
+}
+
+/** Compatibility-safe adapter seam for built-in or injected chunk renderers. */
+export type ChunkRenderer = (
+  planDir: string,
+  chunkIndex: number,
+  outputChunkPath: string,
+) => Promise<ChunkResult>;
+
+interface DistributedCaptureSessionDependencies {
+  createCaptureSession: typeof createCaptureSession;
+  assertSwiftShader: typeof assertSwiftShader;
+  initializeSession: typeof initializeSession;
+  closeCaptureSession: typeof closeCaptureSession;
+  readWebGlVendorInfo: typeof readWebGlVendorInfoFromCanvas;
+}
+
+const distributedCaptureSessionDependencies: DistributedCaptureSessionDependencies = {
+  createCaptureSession,
+  assertSwiftShader,
+  initializeSession,
+  closeCaptureSession,
+  readWebGlVendorInfo: readWebGlVendorInfoFromCanvas,
+};
+
+/**
+ * Every browser that can produce distributed frames must pass the software-GL
+ * assertion, including fresh screenshot browsers created after a fallback.
+ */
+export async function createVerifiedDistributedCaptureSession(
+  serverUrl: string,
+  framesDir: string,
+  captureOptions: CaptureOptions,
+  cfg: EngineConfig,
+  dependencies: DistributedCaptureSessionDependencies = distributedCaptureSessionDependencies,
+): Promise<CaptureSession> {
+  const session = await dependencies.createCaptureSession(
+    serverUrl,
+    framesDir,
+    captureOptions,
+    null,
+    cfg,
+  );
+  try {
+    await dependencies.assertSwiftShader(session.page, dependencies.readWebGlVendorInfo);
+    await dependencies.initializeSession(session);
+    return session;
+  } catch (error) {
+    await dependencies.closeCaptureSession(session).catch(() => {});
+    throw error;
+  }
+}
+
+/**
+ * Build the immutable lookup table once, but keep injector state scoped to a
+ * browser session. Reusing one hook across workers or a whole-chunk retry can
+ * incorrectly suppress injection on a fresh page.
+ */
+export function createChunkVideoFrameInjectorFactory(
+  frameLookup: FrameLookupTable | null,
+): () => BeforeCaptureHook | null {
+  return () => createVideoFrameInjector(frameLookup);
+}
+
+function isCaptureMode(value: string): value is CaptureMode {
+  return value === "beginframe" || value === "screenshot" || value === "drawelement";
+}
+
+/**
+ * Only BeginFrame-specific failures are safe to retry in screenshot mode.
+ * Cancellation, memory exhaustion, and unrelated authoring/IO failures must
+ * keep their original classification instead of being hidden by a fallback.
+ */
+export function shouldRetryChunkCaptureWithScreenshot(error: unknown): boolean {
+  const failure = classifyCaptureFailure(error);
+  if (failure.kind === "cancelled" || failure.kind === "memory_exhaustion") return false;
+  return (
+    /HeadlessExperimental\.beginFrame/i.test(failure.message) ||
+    /beginFrame probe timeout/i.test(failure.message) ||
+    /Another frame is pending|Frame still pending/i.test(failure.message)
+  );
+}
+
+/**
+ * Execute capture with at most one whole-chunk screenshot retry. The caller's
+ * reset hook must discard every partial frame and perf record before retrying.
+ */
+export async function runCaptureWithScreenshotFallback<T>(input: {
+  forceScreenshot: boolean;
+  run: (forceScreenshot: boolean) => Promise<T>;
+  resetForScreenshotRetry: () => Promise<void> | void;
+  onFallback?: (error: unknown) => void;
+}): Promise<T> {
+  try {
+    return await input.run(input.forceScreenshot);
+  } catch (error) {
+    if (input.forceScreenshot || !shouldRetryChunkCaptureWithScreenshot(error)) throw error;
+    input.onFallback?.(error);
+    await input.resetForScreenshotRetry();
+    return await input.run(true);
+  }
+}
+
+/**
+ * Probe an initialized distributed session at a monotonic tick between warmup
+ * and frame zero. `true` means the entire chunk should use screenshot mode.
+ */
+export async function beginFrameSessionNeedsScreenshotFallback(
+  session: Pick<
+    CaptureSession,
+    "page" | "launchCaptureMode" | "beginFrameTimeTicks" | "beginFrameIntervalMs"
+  >,
+  probe: typeof probeBeginFrameLiveness = probeBeginFrameLiveness,
+): Promise<boolean> {
+  if (session.launchCaptureMode !== "beginframe") return false;
+  const timeoutMs =
+    Number(process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS) > 0
+      ? Number(process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS)
+      : 30_000;
+  const probeTick = Math.max(0, session.beginFrameTimeTicks - 5 * session.beginFrameIntervalMs);
+  return !(await probe(session.page, timeoutMs, probeTick, session.beginFrameIntervalMs));
+}
+
 /**
  * Rebuild the engine's in-memory `ExtractedFrames[]` from the on-disk
  * planDir layout. `<planDir>/video-frames/<videoId>/` holds the numbered
@@ -339,7 +473,7 @@ export async function renderChunk(
   planDir: string,
   chunkIndex: number,
   outputChunkPath: string,
-): Promise<ChunkResult> {
+): Promise<EffectiveChunkResult> {
   const start = Date.now();
   const log = defaultLogger;
 
@@ -496,25 +630,22 @@ export async function renderChunk(
       forceScreenshotExplicitlyOptedOut: !encoder.forceScreenshot,
     };
 
-    // Build the BeforeCaptureHook that injects pre-extracted video frames
-    // into the page once per chunk and reuse — `runCaptureStage` may
-    // invoke `createRenderVideoFrameInjector` multiple times, and
-    // re-listing `planDir/video-frames/` each call would be wasteful.
-    // Compositions with no video elements produce `null`, matching the
-    // in-process renderer's skip path.
-    const videoInjector: BeforeCaptureHook | null =
+    // Build the immutable frame lookup once. Each browser session/worker gets
+    // its own injector hook because the hook remembers the last injected frame
+    // per video; sharing that state across a fresh retry page can suppress the
+    // first injection and produce a blank/stale frame.
+    const videoFrameLookup =
       planVideos && planVideos.extracted.length > 0
-        ? createVideoFrameInjector(
-            createFrameLookupTable(
-              planVideos.videos,
-              rebuildExtractedFramesFromPlanDir(
-                planDir,
-                planVideos.extracted,
-                v2Manifest === null ? "dense-v1" : "sparse-v2",
-              ),
+        ? createFrameLookupTable(
+            planVideos.videos,
+            rebuildExtractedFramesFromPlanDir(
+              planDir,
+              planVideos.extracted,
+              v2Manifest === null ? "dense-v1" : "sparse-v2",
             ),
           )
         : null;
+    const createChunkVideoFrameInjector = createChunkVideoFrameInjectorFactory(videoFrameLookup);
 
     const videoCaptureBeyondViewport = resolveVideoCaptureBeyondViewport(
       planVideos?.videos.length ?? 0,
@@ -566,15 +697,10 @@ export async function renderChunk(
       lockWarmupTicks: true,
     };
 
-    // Resolve worker count up-front so we can decide whether to bother
-    // pre-warming a probe session at all. The parallel branch
-    // (chunkWorkerCount > 1) closes the probe immediately and creates fresh
-    // per-worker sessions; `executeWorkerTask` runs `assertSwiftShader`
-    // on worker 0 only (gated on `cfg.browserGpuMode === "software"`), so
-    // the safety contract holds without the eager pre-probe and without
-    // every worker concurrently navigating to the GL probe page. See
-    // `heygen-com/hyperframes#955` for the worst-case wall regression that
-    // motivated gating the probe to worker 0.
+    // Resolve worker count up-front. Sequential capture reuses the initialized
+    // session below. Parallel BeginFrame capture pays one bounded preflight
+    // session so composition-specific compositor stalls are detected before
+    // fan-out; the probe is closed before worker sessions start.
     //
     // Capture-cost calibration based on shader transitions / renderModeHints
     // is not threaded through to chunks yet; the in-process renderer's
@@ -596,9 +722,12 @@ export async function renderChunk(
     let encodeStageMs = 0;
     let captureMode: CaptureMode | undefined;
     const capturePerfs: CapturePerfSummary[] = [];
+    const captureAttempts: Parameters<typeof runCaptureStage>[0]["captureAttempts"] = [];
+    let forceScreenshotForChunk = encoder.forceScreenshot;
     try {
-      if (chunkWorkerCount === 1) {
-        // Sequential branch reuses the probe session for the actual capture.
+      if (chunkWorkerCount === 1 || !forceScreenshotForChunk) {
+        // Sequential capture reuses this session. Parallel BeginFrame capture
+        // uses it only for the composition liveness preflight.
         // SwiftShader assertion runs BEFORE initializeSession (which
         // navigates to the composition); on failure we tear down without
         // ever touching the composition URL. We pass
@@ -609,73 +738,152 @@ export async function renderChunk(
         // fact SwiftShader. The canvas + WEBGL_debug_renderer_info probe
         // works on any page (we navigate to about:blank inside the helper).
         const bootStarted = Date.now();
-        session = await createCaptureSession(fileServer.url, framesDir, captureOptions, null, cfg);
-        await assertSwiftShader(session.page, readWebGlVendorInfoFromCanvas);
-        await initializeSession(session);
+        session = await createVerifiedDistributedCaptureSession(
+          fileServer.url,
+          framesDir,
+          captureOptions,
+          cfg,
+        );
         sessionBootMs = Date.now() - bootStarted;
         // `discardWarmupCapture` is intentionally NOT called: every frame
         // seeks fresh DOM, so `lastFrameCache` is never read; priming it
         // would deadlock Chrome's compositor by issuing a second beginFrame
         // at a `frameTimeTicks` it had just advanced to.
+        const browserSelectedScreenshot = session.launchCaptureMode === "screenshot";
+        const beginFrameStalled =
+          !browserSelectedScreenshot && (await beginFrameSessionNeedsScreenshotFallback(session));
+        if (browserSelectedScreenshot) {
+          forceScreenshotForChunk = true;
+          log.warn(
+            "[renderChunk] Browser capability probe selected screenshot capture for the entire chunk",
+            { chunkIndex },
+          );
+          if (chunkWorkerCount > 1) {
+            await closeCaptureSession(session);
+            session = null;
+          }
+        } else if (beginFrameStalled) {
+          forceScreenshotForChunk = true;
+          log.warn(
+            "[renderChunk] BeginFrame liveness probe failed; using screenshot capture for the entire chunk",
+            { chunkIndex },
+          );
+          await closeCaptureSession(session).catch(() => {});
+          session = null;
+          if (chunkWorkerCount === 1) {
+            const screenshotBootStarted = Date.now();
+            const screenshotCfg: EngineConfig = {
+              ...cfg,
+              forceScreenshot: true,
+              forceScreenshotExplicitlyOptedOut: false,
+            };
+            session = await createVerifiedDistributedCaptureSession(
+              fileServer.url,
+              framesDir,
+              captureOptions,
+              screenshotCfg,
+            );
+            sessionBootMs += Date.now() - screenshotBootStarted;
+          }
+        } else if (chunkWorkerCount > 1) {
+          await closeCaptureSession(session);
+          session = null;
+        }
       }
-      // chunkWorkerCount > 1: skip the probe entirely. Each parallel worker
-      // creates its own session and runs `assertSwiftShader` before its
-      // first frame.
 
-      // In the parallel branch (chunkWorkerCount > 1) this stage also boots
-      // one Chrome session per worker, so captureStageMs includes those
-      // boots; sessionBootMs stays 0 there.
       const captureStarted = Date.now();
-      const capturePlan = createCapturePlan({
-        workerCount: chunkWorkerCount,
-        forceScreenshot: encoder.forceScreenshot,
-        useStreamingEncode: false,
-        useLayeredComposite: false,
-        usePageSideCompositing: false,
-        hasHdrContent: false,
-        needsAlpha: plan.dimensions.format !== "mp4",
+      captureMode = await runCaptureWithScreenshotFallback({
+        forceScreenshot: forceScreenshotForChunk,
+        // fallow-ignore-next-line complexity
+        run: async (forceScreenshot) => {
+          const captureCfg: EngineConfig =
+            cfg.forceScreenshot === forceScreenshot
+              ? cfg
+              : {
+                  ...cfg,
+                  forceScreenshot,
+                  forceScreenshotExplicitlyOptedOut: !forceScreenshot,
+                };
+          if (chunkWorkerCount === 1 && session === null) {
+            session = await createVerifiedDistributedCaptureSession(
+              fileServer.url,
+              framesDir,
+              captureOptions,
+              captureCfg,
+            );
+          }
+          const capturePlan = createCapturePlan({
+            workerCount: chunkWorkerCount,
+            forceScreenshot,
+            useStreamingEncode: false,
+            useLayeredComposite: false,
+            usePageSideCompositing: false,
+            hasHdrContent: false,
+            needsAlpha: plan.dimensions.format !== "mp4",
+          });
+          if (capturePlan.kind !== "sdr_disk") {
+            throw new Error(`Distributed chunk requires sdr_disk plan; got ${capturePlan.kind}`);
+          }
+
+          // runCaptureStage owns and closes any probe session it receives,
+          // including error paths. Clear our defensive handle before await.
+          const captureSession = session;
+          session = null;
+          await runCaptureStage({
+            fileServer,
+            workDir,
+            framesDir,
+            job,
+            totalFrames: framesInChunk,
+            cfg: captureCfg,
+            plan: capturePlan,
+            log,
+            probeSession: captureSession,
+            captureAttempts,
+            // This sink records each worker's effective capture mode so a
+            // fallback remains observable to adapters and smoke tests.
+            dedupPerfs: capturePerfs,
+            buildCaptureOptions: () => captureOptions,
+            createRenderVideoFrameInjector: createChunkVideoFrameInjector,
+            abortSignal: undefined,
+            assertNotAborted: () => {},
+            frameRange: { startFrame: slice.startFrame, endFrame: slice.endFrame },
+          });
+
+          const observedModes = new Set(capturePerfs.map((perf) => perf.captureMode));
+          if (observedModes.size !== 1 || ![...observedModes].every(isCaptureMode)) {
+            throw new Error(
+              `[renderChunk] capture workers reported invalid or inconsistent modes: ` +
+                `${[...observedModes].join(",") || "<none>"}`,
+            );
+          }
+          const [observedMode] = observedModes;
+          if (!observedMode || !isCaptureMode(observedMode)) {
+            throw new Error("[renderChunk] capture completed without an observed mode");
+          }
+          return observedMode;
+        },
+        resetForScreenshotRetry: () => {
+          // A mode switch is a whole-chunk retry. Do not allow successfully
+          // captured BeginFrame files or attempt telemetry to mix with the
+          // screenshot result.
+          rmSync(workDir, { recursive: true, force: true, maxRetries: 3, retryDelay: 100 });
+          mkdirSync(framesDir, { recursive: true });
+          capturePerfs.length = 0;
+          captureAttempts.length = 0;
+          job.framesRendered = 0;
+        },
+        onFallback: (error) => {
+          log.warn(
+            "[renderChunk] BeginFrame capture failed; retrying the entire chunk once in screenshot mode",
+            {
+              chunkIndex,
+              error: error instanceof Error ? error.message : String(error),
+            },
+          );
+        },
       });
-      if (capturePlan.kind !== "sdr_disk") {
-        throw new Error(`Distributed chunk requires sdr_disk plan; got ${capturePlan.kind}`);
-      }
-      await runCaptureStage({
-        fileServer,
-        workDir,
-        framesDir,
-        job,
-        totalFrames: framesInChunk,
-        cfg,
-        plan: capturePlan,
-        log,
-        probeSession: session,
-        captureAttempts: [],
-        // This sink also records each worker's effective capture mode. That
-        // makes a BeginFrame → screenshot fallback observable to adapters and
-        // end-to-end smoke tests instead of existing only in stderr.
-        dedupPerfs: capturePerfs,
-        buildCaptureOptions: () => captureOptions,
-        createRenderVideoFrameInjector: () => videoInjector,
-        abortSignal: undefined,
-        assertNotAborted: () => {},
-        frameRange: { startFrame: slice.startFrame, endFrame: slice.endFrame },
-      });
-      // captureStage closes the session it consumed.
       captureStageMs = Date.now() - captureStarted;
-      session = null;
-      const observedModes = new Set(capturePerfs.map((perf) => perf.captureMode));
-      const validModes = new Set<CaptureMode>(["beginframe", "screenshot", "drawelement"]);
-      if (
-        observedModes.size !== 1 ||
-        ![...observedModes].every((mode): mode is CaptureMode =>
-          validModes.has(mode as CaptureMode),
-        )
-      ) {
-        throw new Error(
-          `[renderChunk] capture workers reported invalid or inconsistent modes: ` +
-            `${[...observedModes].join(",") || "<none>"}`,
-        );
-      }
-      captureMode = [...observedModes][0] as CaptureMode;
       framesEncoded = framesInChunk;
 
       // ── Encode the chunk ──

packages/producer/src/services/distributed/renderChunkFallback.test.ts

@@ -0,0 +1,253 @@
+import { afterEach, describe, expect, it } from "bun:test";
+import {
+  type CaptureSession,
+  createFrameLookupTable,
+  readWebGlVendorInfoFromCanvas,
+  resolveConfig,
+} from "@hyperframes/engine";
+import {
+  beginFrameSessionNeedsScreenshotFallback,
+  createChunkVideoFrameInjectorFactory,
+  createVerifiedDistributedCaptureSession,
+  type ChunkResult,
+  runCaptureWithScreenshotFallback,
+  shouldRetryChunkCaptureWithScreenshot,
+} from "./renderChunk.js";
+
+const originalProbeTimeout = process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS;
+
+afterEach(() => {
+  if (originalProbeTimeout === undefined) {
+    delete process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS;
+  } else {
+    process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS = originalProbeTimeout;
+  }
+});
+
+describe("ChunkResult compatibility", () => {
+  it("accepts a legacy or injected result that cannot report capture mode", () => {
+    const result: ChunkResult = {
+      outputPath: "/tmp/chunk.mp4",
+      outputKind: "file",
+      framesEncoded: 30,
+      sha256: "abc",
+      durationMs: 100,
+      planHashMs: 1,
+      sessionBootMs: 2,
+      captureStageMs: 3,
+      encodeStageMs: 4,
+      workers: 1,
+      perfPath: "/tmp/chunk.mp4.perf.json",
+    };
+
+    expect(result.captureMode).toBeUndefined();
+  });
+});
+
+describe("distributed session isolation", () => {
+  const captureOptions = {
+    width: 320,
+    height: 180,
+    fps: { num: 30, den: 1 },
+    format: "jpeg" as const,
+  };
+
+  function fakeSession(): CaptureSession {
+    return {
+      page: undefined as never,
+      launchCaptureMode: "screenshot",
+    } as CaptureSession;
+  }
+
+  it("verifies SwiftShader before initializing every distributed session", async () => {
+    const calls: string[] = [];
+    const session = fakeSession();
+    const result = await createVerifiedDistributedCaptureSession(
+      "http://127.0.0.1:1234",
+      "/tmp/frames",
+      captureOptions,
+      resolveConfig(),
+      {
+        createCaptureSession: async () => {
+          calls.push("create");
+          return session;
+        },
+        assertSwiftShader: async () => {
+          calls.push("assert");
+        },
+        initializeSession: async () => {
+          calls.push("initialize");
+        },
+        closeCaptureSession: async () => {
+          calls.push("close");
+        },
+        readWebGlVendorInfo: readWebGlVendorInfoFromCanvas,
+      },
+    );
+
+    expect(result).toBe(session);
+    expect(calls).toEqual(["create", "assert", "initialize"]);
+  });
+
+  it("closes a replacement session when verification fails", async () => {
+    const calls: string[] = [];
+    await expect(
+      createVerifiedDistributedCaptureSession(
+        "http://127.0.0.1:1234",
+        "/tmp/frames",
+        captureOptions,
+        resolveConfig(),
+        {
+          createCaptureSession: async () => fakeSession(),
+          assertSwiftShader: async () => {
+            throw new Error("not SwiftShader");
+          },
+          initializeSession: async () => {
+            throw new Error("must not initialize");
+          },
+          closeCaptureSession: async () => {
+            calls.push("close");
+          },
+          readWebGlVendorInfo: readWebGlVendorInfoFromCanvas,
+        },
+      ),
+    ).rejects.toThrow("not SwiftShader");
+    expect(calls).toEqual(["close"]);
+  });
+
+  it("creates fresh stateful video injectors for separate workers and retries", () => {
+    const factory = createChunkVideoFrameInjectorFactory(createFrameLookupTable([], []));
+    const firstAttempt = factory();
+    const screenshotRetry = factory();
+
+    expect(firstAttempt).not.toBeNull();
+    expect(screenshotRetry).not.toBeNull();
+    expect(screenshotRetry).not.toBe(firstAttempt);
+  });
+});
+
+describe("shouldRetryChunkCaptureWithScreenshot", () => {
+  it("recognizes BeginFrame protocol and pending-frame failures", () => {
+    for (const message of [
+      "HeadlessExperimental.beginFrame timed out",
+      "Protocol error (HeadlessExperimental.beginFrame): method wasn't found",
+      "[BeginFrame] Frame still pending after 5 retries",
+      "Another frame is pending",
+    ]) {
+      expect(shouldRetryChunkCaptureWithScreenshot(new Error(message))).toBe(true);
+    }
+  });
+
+  it("does not hide cancellation, memory pressure, or unrelated failures", () => {
+    expect(
+      shouldRetryChunkCaptureWithScreenshot(
+        new Error("render_cancelled during HeadlessExperimental.beginFrame"),
+      ),
+    ).toBe(false);
+    expect(shouldRetryChunkCaptureWithScreenshot(new Error("JavaScript heap out of memory"))).toBe(
+      false,
+    );
+    expect(shouldRetryChunkCaptureWithScreenshot(new Error("ffmpeg exited with code 1"))).toBe(
+      false,
+    );
+  });
+});
+
+describe("runCaptureWithScreenshotFallback", () => {
+  it("discards partial state and retries the whole chunk once in screenshot mode", async () => {
+    const calls: string[] = [];
+    const result = await runCaptureWithScreenshotFallback({
+      forceScreenshot: false,
+      run: async (forceScreenshot) => {
+        calls.push(`run:${forceScreenshot ? "screenshot" : "beginframe"}`);
+        if (!forceScreenshot) throw new Error("HeadlessExperimental.beginFrame timed out");
+        return "captured";
+      },
+      resetForScreenshotRetry: () => {
+        calls.push("reset");
+      },
+    });
+
+    expect(result).toBe("captured");
+    expect(calls).toEqual(["run:beginframe", "reset", "run:screenshot"]);
+  });
+
+  it("does not retry a screenshot attempt or retry the fallback twice", async () => {
+    let screenshotCalls = 0;
+    await expect(
+      runCaptureWithScreenshotFallback({
+        forceScreenshot: true,
+        run: async () => {
+          screenshotCalls++;
+          throw new Error("HeadlessExperimental.beginFrame timed out");
+        },
+        resetForScreenshotRetry: () => {
+          throw new Error("reset must not run");
+        },
+      }),
+    ).rejects.toThrow("HeadlessExperimental.beginFrame timed out");
+    expect(screenshotCalls).toBe(1);
+
+    const modes: boolean[] = [];
+    await expect(
+      runCaptureWithScreenshotFallback({
+        forceScreenshot: false,
+        run: async (forceScreenshot) => {
+          modes.push(forceScreenshot);
+          throw new Error("HeadlessExperimental.beginFrame timed out");
+        },
+        resetForScreenshotRetry: () => {},
+      }),
+    ).rejects.toThrow("HeadlessExperimental.beginFrame timed out");
+    expect(modes).toEqual([false, true]);
+  });
+});
+
+describe("beginFrameSessionNeedsScreenshotFallback", () => {
+  function session(
+    launchCaptureMode: "beginframe" | "screenshot",
+  ): Pick<
+    CaptureSession,
+    "page" | "launchCaptureMode" | "beginFrameTimeTicks" | "beginFrameIntervalMs"
+  > {
+    return {
+      page: undefined as never,
+      launchCaptureMode,
+      beginFrameTimeTicks: 100,
+      beginFrameIntervalMs: 10,
+    };
+  }
+
+  it("uses the configured bounded timeout and a monotonic pre-frame tick", async () => {
+    process.env.PRODUCER_BEGINFRAME_PROBE_TIMEOUT_MS = "1234";
+    const args: unknown[][] = [];
+    const fallback = await beginFrameSessionNeedsScreenshotFallback(
+      session("beginframe"),
+      async (...probeArgs) => {
+        args.push(probeArgs);
+        return false;
+      },
+    );
+
+    expect(fallback).toBe(true);
+    expect(args).toHaveLength(1);
+    expect(args[0]?.slice(1)).toEqual([1234, 50, 10]);
+  });
+
+  it("keeps healthy BeginFrame and skips probing an existing screenshot session", async () => {
+    let probes = 0;
+    expect(
+      await beginFrameSessionNeedsScreenshotFallback(session("beginframe"), async () => {
+        probes++;
+        return true;
+      }),
+    ).toBe(false);
+    expect(
+      await beginFrameSessionNeedsScreenshotFallback(session("screenshot"), async () => {
+        probes++;
+        return false;
+      }),
+    ).toBe(false);
+    expect(probes).toBe(1);
+  });
+});

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

@@ -257,12 +257,16 @@ export async function runCaptureStage(input: CaptureStageInput): Promise<Capture
         captureCfg,
       ));
     captureBeyondViewport = session.options.captureBeyondViewport;
-    if (probeSession) {
-      prepareCaptureSessionForReuse(session, framesDir, videoInjector);
-      probeSession = null;
-    }
 
     try {
+      // Reuse preparation can fail while creating/resetting the output
+      // directory (for example EACCES, EROFS, or ENOSPC). Keep it inside the
+      // session-owning try/finally so the borrowed probe browser is closed
+      // even when preparation fails before capture starts.
+      if (probeSession) {
+        prepareCaptureSessionForReuse(session, framesDir, videoInjector);
+        probeSession = null;
+      }
       if (!session.isInitialized) {
         await initializeSession(session);
       } else if (process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE === "true") {

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

@@ -20,6 +20,7 @@ let failInitializeSession = false;
 let hangParallelUntilAbort = false;
 let hangSequentialUntilStall = false;
 let sessionWorkerEncodeEnabled = false;
+let failPrepareCaptureSessionForReuse = false;
 let initializeSessionErrorMessage = "initialize failed";
 const browserConsoleBuffer = ["[FrameCapture:ERROR] page.goto failed"];
 const closeCaptureSession = mock(async () => {});
@@ -100,7 +101,11 @@ mock.module("@hyperframes/engine", () => ({
     pixelFormat: "yuv420p",
   }),
   initTransparentBackground: async () => {},
-  prepareCaptureSessionForReuse: () => {},
+  prepareCaptureSessionForReuse: () => {
+    if (failPrepareCaptureSessionForReuse) {
+      throw new Error("prepare reuse failed: ENOSPC");
+    }
+  },
   recaptureDrawElementFrameForVerify: async () => Buffer.from("frame"),
   spawnStreamingEncoder,
   writeCapturedFrame: async () => {},
@@ -405,6 +410,53 @@ describe("runCaptureStreamingStage", () => {
 });
 
 describe("runCaptureStage", () => {
+  it("closes a reused probe session when reuse preparation fails", async () => {
+    failCaptureFrameToBuffer = false;
+    failInitializeSession = false;
+    failPrepareCaptureSessionForReuse = true;
+    closeCaptureSession.mockClear();
+    const { createCaptureSession } = await import("@hyperframes/engine");
+    const { runCaptureStage } = await import("./captureStage.js");
+    const cfg = { forceScreenshot: false, ffmpegStreamingTimeout: 3_600_000 };
+    const probeSession = await createCaptureSession(
+      "http://127.0.0.1:4173",
+      "/tmp/hf-test-frames",
+      {},
+      null,
+      cfg,
+    );
+
+    let caught: unknown;
+    try {
+      await runCaptureStage({
+        ...createInput(cfg),
+        plan: createCapturePlan({
+          workerCount: 1,
+          forceScreenshot: false,
+          useStreamingEncode: false,
+          useLayeredComposite: false,
+          usePageSideCompositing: false,
+          hasHdrContent: false,
+          needsAlpha: false,
+        }),
+        probeSession,
+        videoOnlyPath: undefined,
+        outputFormat: undefined,
+        streamingEncoderOptions: undefined,
+        captureAttempts: [],
+      });
+    } catch (error) {
+      caught = error;
+    } finally {
+      failPrepareCaptureSessionForReuse = false;
+    }
+
+    expect(caught).toBeInstanceOf(Error);
+    expect((caught as Error).message).toContain("prepare reuse failed: ENOSPC");
+    expect(closeCaptureSession).toHaveBeenCalledTimes(1);
+    expect(closeCaptureSession).toHaveBeenCalledWith(probeSession);
+  });
+
   it("wraps sequential capture failures with the browser console buffer", async () => {
     failCaptureFrameToBuffer = false;
     failInitializeSession = true;