feat(producer): stream parallel screenshot/beginframe capture to the encoder

packages/cli/src/telemetry/events.test.ts

@@ -361,6 +361,22 @@ describe("render telemetry events", () => {
       }),
     );
   });
+
+  it("carries capture_parallel_stream on render_error via the shared payload", () => {
+    trackRenderError({
+      fps: 30,
+      quality: "standard",
+      docker: false,
+      errorMessage: "worker crashed",
+      captureParallelStream: "beginframe",
+    });
+
+    expect(trackEvent).toHaveBeenCalledWith(
+      "render_error",
+      expect.objectContaining({ capture_parallel_stream: "beginframe" }),
+      undefined,
+    );
+  });
 });
 
 describe("trackRenderFeedback", () => {

packages/cli/src/telemetry/events.ts

@@ -53,6 +53,9 @@ export interface RenderObservabilityTelemetryPayload {
   captureDePreRouterWorkers?: number;
   captureDeSelfVerifyFallback?: boolean;
   captureDeFallbackReason?: string;
+  /** Non-DE parallel-streaming router outcome ("screenshot" | "beginframe" —
+   * routed; "eligible_off" — would route but the kill switch is off). */
+  captureParallelStream?: string;
   observabilityExtractVideoCount?: number;
   observabilityExtractedVideoCount?: number;
   observabilityExtractTotalFrames?: number;
@@ -104,6 +107,7 @@ function renderObservabilityEventProperties(props: RenderObservabilityTelemetryP
     de_pre_router_workers: props.captureDePreRouterWorkers,
     de_self_verify_fallback: props.captureDeSelfVerifyFallback,
     de_fallback_reason: props.captureDeFallbackReason,
+    capture_parallel_stream: props.captureParallelStream,
     observability_extract_video_count: props.observabilityExtractVideoCount,
     observability_extracted_video_count: props.observabilityExtractedVideoCount,
     observability_extract_total_frames: props.observabilityExtractTotalFrames,

packages/cli/src/telemetry/renderObservability.test.ts

@@ -82,3 +82,24 @@ describe("renderObservabilityTelemetryPayload — DE inversion/router cohort (fa
     expect(payload.captureDeFallbackReason).toBeUndefined();
   });
 });
+
+describe("renderObservabilityTelemetryPayload — non-DE parallel-stream router", () => {
+  it("maps the router outcome", () => {
+    const payload = renderObservabilityTelemetryPayload(
+      makeSummary({ captureParallelStream: "beginframe" }),
+    );
+    expect(payload.captureParallelStream).toBe("beginframe");
+  });
+
+  it("maps the passive eligible_off cohort-sizing signal", () => {
+    const payload = renderObservabilityTelemetryPayload(
+      makeSummary({ captureParallelStream: "eligible_off" }),
+    );
+    expect(payload.captureParallelStream).toBe("eligible_off");
+  });
+
+  it("stays undefined when the router never fired", () => {
+    const payload = renderObservabilityTelemetryPayload(makeSummary({}));
+    expect(payload.captureParallelStream).toBeUndefined();
+  });
+});

packages/cli/src/telemetry/renderObservability.ts

@@ -46,6 +46,7 @@ export function renderObservabilityTelemetryPayload(
     captureDePreRouterWorkers: capture.dePreRouterWorkers,
     captureDeSelfVerifyFallback: capture.deSelfVerifyFallback,
     captureDeFallbackReason: capture.deFallbackReason,
+    captureParallelStream: capture.captureParallelStream,
     observabilityExtractVideoCount: extraction?.videoCount,
     observabilityExtractedVideoCount: extraction?.extractedVideoCount,
     observabilityExtractTotalFrames: extraction?.totalFramesExtracted,

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

@@ -68,6 +68,16 @@ export interface RenderCaptureObservability {
   deParallelRouter?: "routed" | "reverted";
   /** Worker count the resolver would have used absent the router; undefined if it never fired. */
   dePreRouterWorkers?: number;
+  /**
+   * Non-DE parallel-streaming router outcome (HF_CAPTURE_PARALLEL_STREAM):
+   * "screenshot" | "beginframe" — the render passed every gate AND the kill
+   * switch was on, so it was routed through the interleaved streaming encoder
+   * (the value is the capture mode that streamed); "eligible_off" — the render
+   * passed every gate EXCEPT the kill switch (passive cohort-sizing signal for
+   * the default-off soak: how many renders WOULD route if enabled). Absent =
+   * ineligible regardless of the switch.
+   */
+  captureParallelStream?: "screenshot" | "beginframe" | "eligible_off";
   protocolTimeoutMs?: number;
   pageNavigationTimeoutMs?: number;
   playerReadyTimeoutMs?: number;

packages/producer/src/services/renderOrchestrator.test.ts

@@ -34,6 +34,7 @@ import {
   shouldRetryViaPinnedFallback,
   shouldPreferParallelDrawElement,
   shouldPreferSingleWorkerDrawElement,
+  shouldStreamParallelCapture,
   shouldUseStreamingEncode,
 } from "./renderOrchestrator.js";
 import { ensureFrameWritten } from "./render/stages/captureHdrFrameShared.js";
@@ -2049,3 +2050,43 @@ describe("shouldRetryViaPinnedFallback (widen the self-verify retry to generic c
     ).toBe(false);
   });
 });
+
+describe("shouldStreamParallelCapture (non-DE parallel streaming router)", () => {
+  const eligible = {
+    routerEnabled: true,
+    workerCount: 3,
+    useDrawElement: false,
+    outputFormat: "mp4" as const,
+    streamingOk: true,
+    layeredOrEffectRoute: false,
+  };
+
+  it("routes an eligible multi-worker non-drawElement render", () => {
+    expect(shouldStreamParallelCapture(eligible)).toBe(true);
+  });
+
+  it("is disabled by default (kill switch off is the shipped default)", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, routerEnabled: false })).toBe(false);
+  });
+
+  it("never fires for single-worker renders (those already stream)", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, workerCount: 1 })).toBe(false);
+  });
+
+  it("never fires when drawElement will capture (the DE routers own that path)", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, useDrawElement: true })).toBe(false);
+  });
+
+  it("only applies to mp4", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, outputFormat: "webm" })).toBe(false);
+    expect(shouldStreamParallelCapture({ ...eligible, outputFormat: "png-sequence" })).toBe(false);
+  });
+
+  it("respects the streaming-encode config/duration gates", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, streamingOk: false })).toBe(false);
+  });
+
+  it("skips HDR-layered and shader-transition routes", () => {
+    expect(shouldStreamParallelCapture({ ...eligible, layeredOrEffectRoute: true })).toBe(false);
+  });
+});

packages/producer/src/services/renderOrchestrator.ts

@@ -1299,6 +1299,54 @@ export function shouldRetryViaPinnedFallback(args: {
   return args.deWorkerInversion === "inverted" || args.deParallelRouter === "routed";
 }
 
+/**
+ * Parallel-streaming router for NON-drawElement capture (screenshot on
+ * macOS/Windows/forced-screenshot, BeginFrame on Linux): should this
+ * multi-worker render stream captured frame buffers straight into the single
+ * ffmpeg stdin encoder (interleaved distribution + ordered reorder-buffer
+ * writer — the PR #2056 machinery) instead of the parallel disk path (workers
+ * write JPEGs, a separate sequential encode pass reads them back)?
+ *
+ * Measured motivation (2026-07-10, macOS SS W3): the disk path's encode is a
+ * purely additive tail (~27% of wall clock on a 3,600-frame comp). Streaming
+ * overlapped it for 1.29x on a uniform-cost comp and was a wash (not a
+ * regression) on a 39%-static bimodal comp — the interleaved writer's
+ * near-lockstep coupling eats the encode win when frame costs are bimodal.
+ * v1 accepts the wash; static-aware routing is a documented follow-up.
+ *
+ * Unlike the DE router this deliberately does NOT require auto-resolved
+ * workers: streaming doesn't change the worker count, so an explicit
+ * `--workers 3` should benefit too. It requires !useDrawElement
+ * (post-resolveConfig — always true on Linux): DE parallel renders belong to
+ * the DE parallel router (HF_DE_PARALLEL_ROUTER) with its self-verify
+ * machinery; both DE predicates independently require useDrawElement, making
+ * the two routers mutually exclusive by construction.
+ */
+export function shouldStreamParallelCapture(args: {
+  /** HF_CAPTURE_PARALLEL_STREAM === "true" — kill switch, default OFF. */
+  routerEnabled: boolean;
+  workerCount: number;
+  /** cfg.useDrawElement AFTER resolveConfig clamps. */
+  useDrawElement: boolean;
+  outputFormat: NonNullable<RenderConfig["format"]>;
+  /** shouldUseStreamingEncode(cfg, format, 1, duration) at the call site —
+   * carries the enableStreamingEncode/format/duration-cap gates. */
+  streamingOk: boolean;
+  /** HDR layered composite or shader transitions — bespoke pipelines
+   * (including page-side compositing, which only engages when
+   * hasShaderTransitions) that never stream. */
+  layeredOrEffectRoute: boolean;
+}): boolean {
+  return (
+    args.routerEnabled &&
+    args.workerCount > 1 &&
+    !args.useDrawElement &&
+    args.outputFormat === "mp4" &&
+    args.streamingOk &&
+    !args.layeredOrEffectRoute
+  );
+}
+
 export function resolveCaptureForceScreenshotForPageSideCompositing(args: {
   forceScreenshot: boolean;
   usePageSideCompositing: boolean;
@@ -1599,6 +1647,10 @@ export async function executeRenderJob(
     // render already executing in the same process. Threading this as a
     // local instead closes that cross-talk, not just the sequential leak.
     let deParallelStreamForced = false;
+    // Per-render (not process-global) signal that the NON-DE parallel-stream
+    // router fired — same threading discipline as deParallelStreamForced
+    // (see that flag's comment for why this must never be an env mutation).
+    let captureParallelStreamForced = false;
     let deSelfVerifyFallback = false;
     let deFallbackReason: string | undefined;
     let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined;
@@ -2190,6 +2242,50 @@ export async function executeRenderJob(
       deParallelRouter: deParallelRouter ?? "none",
     });
 
+    // Non-DE parallel-streaming router — see shouldStreamParallelCapture.
+    // Mutually exclusive with the DE inversion/router above by construction
+    // (both DE predicates require useDrawElement; this requires its negation).
+    const captureParallelStreamRouterEnabled = process.env.HF_CAPTURE_PARALLEL_STREAM === "true";
+    const captureParallelStreamArgs = {
+      workerCount,
+      useDrawElement: cfg.useDrawElement,
+      outputFormat,
+      streamingOk: shouldUseStreamingEncode(cfg, outputFormat, 1, job.duration),
+      layeredOrEffectRoute: hasHdrContent || compiled.hasShaderTransitions,
+    };
+    const captureParallelStreamEligible = shouldStreamParallelCapture({
+      routerEnabled: captureParallelStreamRouterEnabled,
+      ...captureParallelStreamArgs,
+    });
+    if (captureParallelStreamEligible) {
+      captureParallelStreamForced = true;
+      // Which mode will stream: the engine picks beginframe only on Linux with
+      // headless-shell and no forced screenshot (frameCapture.ts preMode);
+      // everything else is screenshot. Recorded for telemetry cohorting.
+      const captureParallelStream =
+        process.platform === "linux" && !captureForceScreenshot ? "beginframe" : "screenshot";
+      log.info(
+        `[Render] Parallel ${captureParallelStream} capture will stream to the encoder ` +
+          `(interleaved, ${workerCount} workers) instead of the disk path. ` +
+          "Set HF_CAPTURE_PARALLEL_STREAM=false to disable.",
+      );
+      updateCaptureObservability({ captureParallelStream });
+      // NOTE: no string data on the checkpoint — RenderObservationData string
+      // values are dropped unless the key is in observability.ts's
+      // ALLOWED_STRING_DATA_KEYS allow-list. The message carries the detail.
+      observability.checkpoint(
+        "worker_resolution",
+        `parallel ${captureParallelStream} capture routed to streaming`,
+      );
+    } else if (shouldStreamParallelCapture({ routerEnabled: true, ...captureParallelStreamArgs })) {
+      // The kill switch is the ONLY failed gate: emit a passive cohort-sizing
+      // signal (capture_parallel_stream = "eligible_off") so the default-off
+      // soak can measure how many fleet renders WOULD route before anyone
+      // enables the flag. Observability-only — no behavior change, no log
+      // noise on the default path.
+      updateCaptureObservability({ captureParallelStream: "eligible_off" });
+    }
+
     if (workerCount > 1 && probeSession) {
       lastBrowserConsole = probeSession.browserConsoleBuffer;
       await closeCaptureSession(probeSession);
@@ -2206,7 +2302,7 @@ export async function executeRenderJob(
       outputFormat,
       workerCount,
       job.duration,
-      deParallelStreamForced,
+      deParallelStreamForced || captureParallelStreamForced,
     );
     log.info("streaming-encode gate", {
       enabled: useStreamingEncode,
@@ -2450,7 +2546,7 @@ export async function executeRenderJob(
                 workerCount,
                 probeSession,
                 outputFormat,
-                forceParallelStream: deParallelStreamForced,
+                forceParallelStream: deParallelStreamForced || captureParallelStreamForced,
                 streamingEncoderOptions: {
                   fps: job.config.fps,
                   width,