fix(producer): wire video frame injector into renderChunk

The chunk worker passed `createRenderVideoFrameInjector: () => null` to
`runCaptureStage`, leaving the page's `<video>` elements to decode the
source mp4 against the virtual clock. Chrome's native video pipeline
seeks ±1 frame off what the in-process renderer captures (which uses
pre-extracted frames injected as images via createVideoFrameInjector).
That ±1 frame drift produced the PSNR gap on sub-composition-video and
style-1-prod against the in-process baselines.

Two pieces:

1. `plan()` now persists the engine's `VideoElement[]` (composition.videos)
   and a serialized form of `extractionResult.extracted` (videoId,
   srcPath, framePattern, fps, totalFrames, metadata — paths omitted) to
   `<planDir>/meta/videos.json`. This is the data renderChunk needs to
   reconstruct a `FrameLookupTable` without re-running the extract stage.

2. `plan()` no longer calls `frameLookup.cleanup()` after extraction.
   That cleanup was rm-rf-ing each video's outputDir, which for the
   in-process orchestrator is a scratch tree the renderer owns — but for
   plan() that "scratch" IS `compiledDir/__hyperframes_video_frames/<videoId>/`,
   the source material that the subsequent rename moves into
   `planDir/video-frames/`. Cleaning it up before the rename left
   planDir/video-frames/ with only the `_downloads/` subdirectory and no
   actual frame files. Both `style-1-prod` and `sub-composition-video`
   reproduced this on every distributed-simulated run; both pass after
   the cleanup is dropped.

3. `renderChunk` reads `meta/videos.json`, rebuilds `ExtractedFrames[]`
   by re-listing `planDir/video-frames/<videoId>/` for each video, calls
   `createFrameLookupTable(videos, extracted)`, and wraps the result in
   `createVideoFrameInjector` — the same hook the in-process renderer
   uses. The rebuilt entries set `ownedByLookup: false` so any later
   cleanup() call from the engine doesn't rm the planDir bytes another
   worker may still be reading.

Validated in `docker:test --mode=distributed-simulated`:
  font-variant-numeric:           PASSED
  many-cuts:                      PASSED
  gsap-letters-render-compat:     PASSED
  style-1-prod:                   PASSED (was: 15 frames at 26-29 dB)
  sub-composition-video:          PASSED (was: most frames at 21-25 dB)

In-process unchanged; 54 distributed unit tests still pass in Docker.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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

@@ -24,10 +24,24 @@
  * never have to handle them.
  */
 
-import { cpSync, existsSync, mkdirSync, readdirSync, renameSync, rmSync, statSync } from "node:fs";
+import {
+  cpSync,
+  existsSync,
+  mkdirSync,
+  readdirSync,
+  renameSync,
+  rmSync,
+  statSync,
+  writeFileSync,
+} from "node:fs";
 import { join } from "node:path";
 import { type CanvasResolution } from "@hyperframes/core";
-import { type EngineConfig, resolveConfig } from "@hyperframes/engine";
+import {
+  type EngineConfig,
+  type ExtractedFrames,
+  resolveConfig,
+  type VideoElement,
+} from "@hyperframes/engine";
 import { defaultLogger, type ProducerLogger } from "../../logger.js";
 import { runAudioStage } from "../render/stages/audioStage.js";
 import { runCompileStage } from "../render/stages/compileStage.js";
@@ -419,6 +433,42 @@ function buildLockedRenderConfig(input: {
   };
 }
 
+/**
+ * Persisted shape of the video-extraction outputs from `runExtractVideosStage`,
+ * written to `<planDir>/meta/videos.json` for `renderChunk` to reconstruct
+ * a `FrameLookupTable` from. The in-process renderer keeps these structures
+ * in memory; the distributed pipeline writes them out so a separate chunk
+ * worker process can rebuild the video-frame injector without re-running
+ * the extract stage.
+ *
+ * `ExtractedFrames` from the engine carries an absolute `outputDir`,
+ * an open file descriptor in some paths, and a `framePaths` map — none of
+ * those survive a serialize → re-deserialize round trip across processes.
+ * The serialized form keeps only what plan-time produced and lets
+ * `renderChunk` re-derive `outputDir` (always `<planDir>/video-frames/<videoId>`)
+ * and `framePaths` (re-listed from that directory).
+ */
+interface PlanVideosJson {
+  /**
+   * Composition's `<video>` elements in document order. Identical to
+   * `composition.videos` from `compileForRender` — the chunk worker uses
+   * this to drive `createFrameLookupTable`'s time/index math.
+   */
+  videos: VideoElement[];
+  /**
+   * Per-video extraction outputs. `outputDir` is omitted — `renderChunk`
+   * reconstructs it from `<planDir>/video-frames/<videoId>/`.
+   */
+  extracted: Array<{
+    videoId: string;
+    srcPath: string;
+    framePattern: string;
+    fps: number;
+    totalFrames: number;
+    metadata: ExtractedFrames["metadata"];
+  }>;
+}
+
 /**
  * Per-format encoder + pixel-format + preset triple. Distributed mode is
  * SDR-only: H.264 8-bit for mp4, ProRes 4444 for mov, raw RGBA for
@@ -602,7 +652,15 @@ export async function plan(
     assertNotAborted,
     materializeSymlinks: true,
   });
-  if (extractResult.frameLookup) extractResult.frameLookup.cleanup();
+  // DO NOT call `extractResult.frameLookup.cleanup()` here. cleanup()
+  // rm-rfs each video's outputDir, which for the in-process renderer is
+  // a scratch tree the orchestrator owns. In plan(), that "scratch" tree
+  // (`compiledDir/__hyperframes_video_frames/<videoId>/`) IS the source
+  // material for `planDir/video-frames/`; cleanup before the rename
+  // leaves the planDir with only the `_downloads/` subdirectory and no
+  // actual per-video frame files, and the chunk worker can't reconstruct
+  // the BeforeCaptureHook without them. The renames below move the tree
+  // into its final planDir location.
 
   // ── Audio ──
   const audioResult = await runAudioStage({
@@ -633,6 +691,31 @@ export async function plan(
   if (existsSync(finalCompiledDir)) rmSync(finalCompiledDir, { recursive: true, force: true });
   renameSync(compiledDir, finalCompiledDir);
 
+  // Persist the video-extraction outputs alongside the rest of the planDir
+  // metadata so `renderChunk` can rebuild the BeforeCaptureHook that
+  // injects pre-extracted frames into the page. Without this file the
+  // chunk worker has no way to recover the per-video frame timing and
+  // falls back to letting Chrome's native `<video>` element decode the
+  // source mp4 on the fly — which produces ±1-frame drift relative to
+  // the in-process renderer's pre-extracted injection (the source of
+  // every committed baseline). Writing this file is the contract that
+  // makes distributed renders pixel-comparable to in-process renders for
+  // compositions with video sources.
+  const planVideosJson: PlanVideosJson = {
+    videos: composition.videos,
+    extracted: (extractResult.extractionResult?.extracted ?? []).map((ext) => ({
+      videoId: ext.videoId,
+      srcPath: ext.srcPath,
+      framePattern: ext.framePattern,
+      fps: ext.fps,
+      totalFrames: ext.totalFrames,
+      metadata: ext.metadata,
+    })),
+  };
+  const metaDir = join(planDir, "meta");
+  if (!existsSync(metaDir)) mkdirSync(metaDir, { recursive: true });
+  writeFileSync(join(metaDir, "videos.json"), JSON.stringify(planVideosJson, null, 2), "utf-8");
+
   const planAudioPath = join(planDir, "audio.aac");
   if (audioResult.hasAudio && existsSync(audioResult.audioOutputPath)) {
     renameSync(audioResult.audioOutputPath, planAudioPath);

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

@@ -48,10 +48,15 @@ import {
   type CaptureSession,
   closeCaptureSession,
   createCaptureSession,
+  createFrameLookupTable,
+  createVideoFrameInjector,
   type EngineConfig,
+  type ExtractedFrames,
   getEncoderPreset,
   initializeSession,
   resolveConfig,
+  type VideoElement,
+  type VideoMetadata,
 } from "@hyperframes/engine";
 import { defaultLogger } from "../../logger.js";
 import { runEncodeStage } from "../render/stages/encodeStage.js";
@@ -126,6 +131,90 @@ export interface ChunkResult {
   perfPath: string;
 }
 
+/**
+ * On-disk shape of `<planDir>/meta/videos.json` written by `plan()`.
+ * Mirrored from `plan.ts`'s internal `PlanVideosJson` — duplicated here so
+ * `renderChunk` doesn't import from `plan.ts` (which would create a cycle).
+ */
+interface PlanVideosJson {
+  videos: VideoElement[];
+  extracted: Array<{
+    videoId: string;
+    srcPath: string;
+    framePattern: string;
+    fps: number;
+    totalFrames: number;
+    metadata: VideoMetadata;
+  }>;
+}
+
+/**
+ * Rebuild the engine's in-memory `ExtractedFrames[]` from the on-disk
+ * planDir layout. `plan()` wrote a serialized manifest of which videos
+ * exist + per-video frame counts; `<planDir>/video-frames/<videoId>/`
+ * holds the actual numbered frame files. This walks each video's output
+ * directory, populates the `framePaths` Map keyed by 1-based frame index
+ * (the convention `videoFrameInjector` and `FrameLookupTable` both rely
+ * on), and returns the structures ready to feed back into
+ * `createFrameLookupTable`.
+ *
+ * Pure: no Chrome involvement, no engine launch. Used by `renderChunk`
+ * once per chunk to recreate the BeforeCaptureHook the in-process
+ * renderer produces from a live extraction stage.
+ */
+function rebuildExtractedFramesFromPlanDir(
+  planDir: string,
+  videos: PlanVideosJson["extracted"],
+): ExtractedFrames[] {
+  const result: ExtractedFrames[] = [];
+  for (const v of videos) {
+    const outputDir = join(planDir, "video-frames", v.videoId);
+    if (!existsSync(outputDir)) {
+      throw new Error(
+        `[renderChunk] planDir missing extracted video frames for ${JSON.stringify(v.videoId)}: ` +
+          `${outputDir} not present. plan() should have written frames here; the planDir is malformed.`,
+      );
+    }
+    // The framePattern from extractAllVideoFrames looks like
+    // `frame_%06d.jpg`. We can't sprintf at runtime, so list the dir and
+    // index by sorted name — the encoder writes frame_NNNNNN.<ext> in
+    // monotonic order, so sorted-by-name is also sorted-by-frame-index.
+    const ext = v.framePattern.includes(".")
+      ? v.framePattern.slice(v.framePattern.lastIndexOf(".")).toLowerCase()
+      : ".jpg";
+    const frames = readdirSync(outputDir)
+      .filter((name) => name.toLowerCase().endsWith(ext))
+      .sort();
+    const framePaths = new Map<number, string>();
+    for (let i = 0; i < frames.length; i++) {
+      const frameName = frames[i];
+      if (!frameName) continue;
+      // FrameLookupTable / videoFrameInjector index frames 1-based; the
+      // extractor writes frame_000001, frame_000002, ... so the on-disk
+      // names are already 1-based but we use the sorted-list position to
+      // be tolerant of gaps.
+      framePaths.set(i + 1, join(outputDir, frameName));
+    }
+    result.push({
+      videoId: v.videoId,
+      srcPath: v.srcPath,
+      outputDir,
+      framePattern: v.framePattern,
+      fps: v.fps,
+      totalFrames: v.totalFrames,
+      metadata: v.metadata,
+      framePaths,
+      // The chunk worker doesn't own the planDir's video-frames/ directory
+      // (the controller does — adapters that fan out chunks across machines
+      // share the planDir as read-only). Mark ownership as false so the
+      // injector's eventual cleanup doesn't rm bytes another worker may
+      // still be reading.
+      ownedByLookup: false,
+    });
+  }
+  return result;
+}
+
 /** Plan-time JSON manifest written by `freezePlan`. */
 interface PlanJson {
   planHash: string;
@@ -259,6 +348,23 @@ export async function renderChunk(
   const encoder = JSON.parse(readFileSync(encoderJsonPath, "utf-8")) as LockedRenderConfig;
   const chunks = JSON.parse(readFileSync(chunksJsonPath, "utf-8")) as ChunkSliceJson[];
 
+  // Optional: `meta/videos.json` is present whenever the composition has
+  // `<video>` elements. Absence is fine — compositions without video skip
+  // the frame-extraction stage entirely and don't need an injector.
+  // Presence drives the BeforeCaptureHook below.
+  const videosJsonPath = join(planDir, "meta", "videos.json");
+  let planVideos: PlanVideosJson | null = null;
+  if (existsSync(videosJsonPath)) {
+    try {
+      planVideos = JSON.parse(readFileSync(videosJsonPath, "utf-8")) as PlanVideosJson;
+    } catch (err) {
+      throw new RenderChunkValidationError(
+        MISSING_PLAN_ARTIFACT,
+        `[renderChunk] failed to parse ${videosJsonPath}: ${err instanceof Error ? err.message : String(err)}`,
+      );
+    }
+  }
+
   if (chunkIndex < 0 || chunkIndex >= chunks.length) {
     throw new RenderChunkValidationError(
       CHUNK_INDEX_OUT_OF_RANGE,
@@ -464,7 +570,20 @@ export async function renderChunk(
         needsAlpha: plan.dimensions.format !== "mp4",
         captureAttempts: [],
         buildCaptureOptions: () => captureOptions,
-        createRenderVideoFrameInjector: () => null,
+        // Rebuild the in-process renderer's video-frame injector from the
+        // planDir's `meta/videos.json` + `video-frames/<id>/`. Without
+        // this, the chunk worker's page falls back to letting Chrome's
+        // native `<video>` element decode the source mp4 against the
+        // virtual clock, which produces ±1-frame drift relative to the
+        // in-process renderer that pre-extracts frames and injects them
+        // as images. Compositions with no `<video>` elements get a
+        // null injector (no overhead, same as in-process).
+        createRenderVideoFrameInjector: () => {
+          if (!planVideos || planVideos.extracted.length === 0) return null;
+          const extracted = rebuildExtractedFramesFromPlanDir(planDir, planVideos.extracted);
+          const frameLookup = createFrameLookupTable(planVideos.videos, extracted);
+          return createVideoFrameInjector(frameLookup);
+        },
         abortSignal: undefined,
         assertNotAborted: () => {},
         frameRange: { startFrame: slice.startFrame, endFrame: slice.endFrame },