Merge pull request #2095 from heygen-com/feat/de-parallel-router

feat(producer): default-off router for verified parallel drawElement

packages/cli/src/commands/render.ts

@@ -1688,6 +1688,8 @@ function trackRenderMetrics(
     deClampReason: perf?.drawElement?.clampReason,
     deWorkerInversion: perf?.drawElement?.workerInversion,
     dePreInversionWorkers: perf?.drawElement?.preInversionWorkers,
+    deParallelRouter: perf?.drawElement?.parallelRouter,
+    dePreRouterWorkers: perf?.drawElement?.preRouterWorkers,
     deGateReason: perf?.drawElement?.gateReason,
     deWorkerEncode: perf?.drawElement?.workerEncode,
     deVerifyArmed: perf?.drawElement?.verifyArmed,

packages/cli/src/telemetry/events.ts

@@ -126,6 +126,8 @@ export function trackRenderComplete(
     deClampReason?: string;
     deWorkerInversion?: string;
     dePreInversionWorkers?: number;
+    deParallelRouter?: string;
+    dePreRouterWorkers?: number;
     deGateReason?: string;
     deWorkerEncode?: boolean;
     deVerifyArmed?: number;
@@ -204,6 +206,8 @@ export function trackRenderComplete(
       de_clamp_reason: props.deClampReason,
       de_worker_inversion: props.deWorkerInversion,
       de_pre_inversion_workers: props.dePreInversionWorkers,
+      de_parallel_router: props.deParallelRouter,
+      de_pre_router_workers: props.dePreRouterWorkers,
       de_gate_reason: props.deGateReason,
       de_worker_encode: props.deWorkerEncode,
       de_verify_armed: props.deVerifyArmed,

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

@@ -44,6 +44,8 @@ export interface RenderCaptureObservability {
   deSelfVerifyFallback?: boolean;
   /** Auto-parallel inversion outcome: "inverted" (fired, held) | "reverted" (fired, self-verify retry rolled back). */
   deWorkerInversion?: "inverted" | "reverted";
+  /** DE parallel-router outcome: "routed" (fired, held) | "reverted" (fired, self-verify retry rolled back). */
+  deParallelRouter?: "routed" | "reverted";
   protocolTimeoutMs?: number;
   pageNavigationTimeoutMs?: number;
   playerReadyTimeoutMs?: number;

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

@@ -67,6 +67,9 @@ export interface DrawElementPerfInput {
   workerInversion?: "inverted" | "reverted";
   /** Auto-resolved worker count before the inversion pinned it to 1 (set only when the inversion fired). */
   preInversionWorkers?: number;
+  parallelRouter?: "routed" | "reverted";
+  /** Auto-resolved worker count before the router pinned it to 3 (set only when the router fired). */
+  preRouterWorkers?: number;
   selfVerifyFallback: boolean;
   fallbackReason?: string;
   drainStats?: {
@@ -96,6 +99,8 @@ function aggregateDrawElement(
     clampReason: de.clampReason,
     workerInversion: de.workerInversion ?? "none",
     preInversionWorkers: de.preInversionWorkers,
+    parallelRouter: de.parallelRouter ?? "none",
+    preRouterWorkers: de.preRouterWorkers,
     gateReason: gateReasons.length > 0 ? gateReasons.join("|") : undefined,
     workerEncode: perfs.some((p) => p.deWorkerEncode),
     verifyArmed: perfs.reduce((sum, p) => sum + (p.deVerifyArmed ?? 0), 0),

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

@@ -113,6 +113,14 @@ export interface CaptureStreamingStageInput {
   log: ProducerLogger;
   workerCount: number;
   probeSession: CaptureSession | null;
+  /**
+   * Per-render override from the DE parallel router — see
+   * deParallelStreamForced's declaration in renderOrchestrator.ts. Distinct
+   * from the `HF_DE_PARALLEL_STREAM` manual opt-in (still read directly by
+   * this stage) because the router's decision must not leak across
+   * concurrently-running renders sharing this process via a global env var.
+   */
+  forceParallelStream?: boolean;
   /** For the spawn-failure log message context only. */
   outputFormat: string;
   /** Pre-built encoder options; passed straight to `spawnStreamingEncoder`. */
@@ -456,6 +464,7 @@ export async function runCaptureStreamingStage(
     assertNotAborted,
     onProgress,
     dedupPerfs,
+    forceParallelStream,
   } = input;
   let { workerCount, probeSession } = input;
   let lastBrowserConsole: string[] = [];
@@ -507,13 +516,14 @@ export async function runCaptureStreamingStage(
 
     if (workerCount > 1) {
       // Parallel capture → streaming encode
-      // HF_DE_PARALLEL_STREAM (opt-in): interleaved distribution — worker i
-      // takes frames i, i+N, i+2N… so the ordered writer's reorder window is
-      // N frames and workers run in lockstep instead of serializing behind
-      // contiguous ranges (see distributeFramesInterleaved). Each worker runs
-      // the depth-2 pipelined drawElement produce when its session initialized
-      // in drawelement mode.
-      const deParallelStream = process.env.HF_DE_PARALLEL_STREAM === "true";
+      // HF_DE_PARALLEL_STREAM (manual opt-in) / forceParallelStream (router):
+      // interleaved distribution — worker i takes frames i, i+N, i+2N… so the
+      // ordered writer's reorder window is N frames and workers run in
+      // lockstep instead of serializing behind contiguous ranges (see
+      // distributeFramesInterleaved). Each worker runs the depth-2 pipelined
+      // drawElement produce when its session initialized in drawelement mode.
+      const deParallelStream =
+        forceParallelStream === true || process.env.HF_DE_PARALLEL_STREAM === "true";
       const tasks = deParallelStream
         ? distributeFramesInterleaved(totalFrames, workerCount, workDir)
         : distributeFrames(totalFrames, workerCount, workDir);

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

@@ -29,6 +29,8 @@ import {
   resolveCaptureForceScreenshotForPageSideCompositing,
   shouldDiscardProbeSessionForPageSideCompositing,
   resolveInversionRetryPlan,
+  resolveParallelRouterRetryPlan,
+  shouldPreferParallelDrawElement,
   shouldPreferSingleWorkerDrawElement,
   shouldUseStreamingEncode,
 } from "./renderOrchestrator.js";
@@ -402,6 +404,14 @@ describe("shouldUseStreamingEncode", () => {
     expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 2, 240)).toBe(false);
   });
 
+  it("forceParallelStream overrides the parallel-capture clamp for verified multi-worker streaming", () => {
+    expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 3, 240, true)).toBe(true);
+    expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 3, 240, false)).toBe(false);
+    expect(shouldUseStreamingEncode(streamingEnabledConfig, "png-sequence", 3, 240, true)).toBe(
+      false,
+    );
+  });
+
   it("keeps renders over the configured max duration on normal encoding", () => {
     expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 1, 240)).toBe(true);
     expect(shouldUseStreamingEncode(streamingEnabledConfig, "mp4", 1, 240.001)).toBe(false);
@@ -1702,3 +1712,123 @@ describe("resolveInversionRetryPlan (self-verify retry rollback)", () => {
     });
   });
 });
+
+describe("shouldPreferParallelDrawElement (DE parallel router)", () => {
+  const eligible = {
+    workerCount: 5,
+    requestedWorkers: "auto" as const,
+    useDrawElement: true,
+    deCompileGate: undefined,
+    forceScreenshot: false,
+    outputFormat: "mp4" as const,
+    totalFrames: 2381,
+    minFrames: 2000,
+    layeredOrEffectRoute: false,
+    supersampling: false,
+    probeDeGated: false,
+    experimentalParallelDeOptIn: false,
+    routerEnabled: true,
+  };
+
+  it("routes an auto-resolved multi-worker render for an eligible long comp", () => {
+    expect(shouldPreferParallelDrawElement(eligible)).toBe(true);
+  });
+
+  it("is disabled by default (routerEnabled: false is the shipped default)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, routerEnabled: false })).toBe(false);
+  });
+
+  it("honors explicitly requested workers", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, requestedWorkers: 3 })).toBe(false);
+  });
+
+  it("routes for requestedWorkers undefined — the value production actually passes for auto", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, requestedWorkers: undefined })).toBe(
+      true,
+    );
+  });
+
+  it("skips comps routed to layered/HDR/shader paths (drawElement never runs there)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, layeredOrEffectRoute: true })).toBe(
+      false,
+    );
+  });
+
+  it("skips supersampled renders (engine init-time DE gate)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, supersampling: true })).toBe(false);
+  });
+
+  it("skips when the probe session already shows DE gated out", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, probeDeGated: true })).toBe(false);
+  });
+
+  it("honors the explicit experimental parallel-DE opt-in (already parallel, router is a no-op)", () => {
+    expect(
+      shouldPreferParallelDrawElement({ ...eligible, experimentalParallelDeOptIn: true }),
+    ).toBe(false);
+  });
+
+  it("skips below the amortization threshold (benchmark: real-work comps clear 1.25x at ~2,000+ frames)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, totalFrames: 915 })).toBe(false);
+    expect(shouldPreferParallelDrawElement({ ...eligible, totalFrames: 2000 })).toBe(true);
+  });
+
+  it("is disabled by minFrames <= 0 (HF_DE_PARALLEL_MIN_FRAMES=0 kill switch)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, minFrames: 0 })).toBe(false);
+    expect(shouldPreferParallelDrawElement({ ...eligible, minFrames: -1 })).toBe(false);
+  });
+
+  it("requires drawElement to be enabled and ungated", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, useDrawElement: false })).toBe(false);
+    expect(shouldPreferParallelDrawElement({ ...eligible, deCompileGate: "3d" })).toBe(false);
+    expect(shouldPreferParallelDrawElement({ ...eligible, forceScreenshot: true })).toBe(false);
+  });
+
+  it("only applies to mp4 (the benchmarked configuration)", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, outputFormat: "webm" })).toBe(false);
+  });
+
+  it("is a no-op when workers already resolved to 1", () => {
+    expect(shouldPreferParallelDrawElement({ ...eligible, workerCount: 1 })).toBe(false);
+  });
+});
+
+describe("resolveParallelRouterRetryPlan (self-verify retry rollback)", () => {
+  const cfg = { enableStreamingEncode: true, streamingEncodeMaxDurationSeconds: 240 };
+
+  it("returns null when the render was never router-routed", () => {
+    expect(
+      resolveParallelRouterRetryPlan({
+        deParallelRouter: undefined,
+        preRouterWorkerCount: 5,
+        cfg,
+        outputFormat: "mp4",
+        durationSeconds: 80,
+      }),
+    ).toBe(null);
+    expect(
+      resolveParallelRouterRetryPlan({
+        deParallelRouter: "reverted",
+        preRouterWorkerCount: 5,
+        cfg,
+        outputFormat: "mp4",
+        durationSeconds: 80,
+      }),
+    ).toBe(null);
+  });
+
+  it("restores the pre-router worker count and routes multi-worker retries to disk", () => {
+    const plan = resolveParallelRouterRetryPlan({
+      deParallelRouter: "routed",
+      preRouterWorkerCount: 5,
+      cfg,
+      outputFormat: "mp4",
+      durationSeconds: 80,
+    });
+    expect(plan).toEqual({
+      workerCount: 5,
+      useStreamingEncode: false,
+      deParallelRouter: "reverted",
+    });
+  });
+});

packages/producer/src/services/renderOrchestrator.ts

@@ -387,6 +387,10 @@ export interface RenderPerfSummary {
     workerInversion?: string;
     /** Worker count the auto-resolution chose BEFORE the inversion pinned it to 1 — the parallel counterfactual for speedup math. Only set when the inversion fired. */
     preInversionWorkers?: number;
+    /** DE parallel-router outcome: "routed" (fired, held), "reverted" (fired, self-verify retry rolled back), "none". Mutually exclusive with workerInversion. */
+    parallelRouter?: string;
+    /** Worker count the auto-resolution chose BEFORE the router pinned it to 3 — the single-worker-inversion counterfactual. Only set when the router fired. */
+    preRouterWorkers?: number;
     /** Engine init-time gate: swiftshader | css_effect:* | at_risk_timeline | 3d_init_failed | supersampling | render_mode_hint. */
     gateReason?: string;
     /** Worker-encode drain (the verified path) was active. */
@@ -957,19 +961,23 @@ export function shouldUseStreamingEncode(
   workerCount: number,
   // Composition timeline duration in seconds.
   durationSeconds: number,
+  // Per-render override (set by the DE parallel router) — see
+  // deParallelStreamForced's declaration in executeRenderJob for why this is
+  // a parameter instead of an env-var read.
+  forceParallelStream = false,
 ): boolean {
   if (!cfg.enableStreamingEncode) return false;
   if (outputFormat === "png-sequence") return false;
   if (outputFormat === "gif") return false;
   if (!Number.isFinite(durationSeconds) || durationSeconds <= 0) return false;
   if (durationSeconds > cfg.streamingEncodeMaxDurationSeconds) return false;
-  // SPIKE (HF_DE_PARALLEL_STREAM): allow multi-worker streaming for the
-  // interleaved drawElement produce experiment. Contiguous-chunk parallel
-  // streaming stalls (worker k+1's first frame waits for ALL of worker k's),
-  // so this only makes sense with the interleaved distribution the capture
-  // stage selects under the same flag. Explicit opt-in, unverified — do not
-  // ship default-on without threading guardFrame through onFrameBuffer.
-  if (process.env.HF_DE_PARALLEL_STREAM === "true") return true;
+  // HF_DE_PARALLEL_STREAM (manual opt-in) / forceParallelStream (router):
+  // allow multi-worker streaming for the interleaved drawElement produce
+  // path. Contiguous-chunk parallel streaming stalls (worker k+1's first
+  // frame waits for ALL of worker k's), so this only makes sense with the
+  // interleaved distribution the capture stage selects under the same
+  // condition.
+  if (forceParallelStream || process.env.HF_DE_PARALLEL_STREAM === "true") return true;
   return workerCount === 1;
 }
 
@@ -1065,6 +1073,95 @@ export function resolveInversionRetryPlan(args: {
   };
 }
 
+/**
+ * DE parallel-router predicate: should an AUTO-resolved multi-worker render
+ * use VERIFIED PARALLEL drawElement streaming (HF_DE_PARALLEL_STREAM) instead
+ * of the #2026 single-worker inversion?
+ *
+ * Benchmarked 2026-07-08 (clean, quiet-machine re-run): par3/single 1.16–1.36x
+ * on real-work comps ≥2,000 frames (2,381f GSAP graphics 1.36x, 3,245f rAF
+ * high-variance 1.29x, 915f crossover probe 1.27x); the one comp that didn't
+ * clear 1.25x (3,600f, 39% static/dedup-heavy) still didn't LOSE to single-
+ * worker (1.16x) — dedup already skips the capture work parallelism would
+ * split, so there's mechanically less headroom, not a regression. No comp
+ * anywhere showed par3 < single. Default-off (HF_DE_PARALLEL_ROUTER): this
+ * promotes the opt-in mechanism from #2056 into the auto-routing decision,
+ * but the decision itself stays gated behind its own flag pending the
+ * telemetry soak (revert rate, de_verify_min_db distribution) on real wild
+ * traffic — there is currently none, since nothing routes here by default.
+ * Takes priority over the single-worker inversion when both would fire (a
+ * higher minFrames than HF_DE_SINGLE_MIN_FRAMES is the intended shape: this
+ * only picks up the long tail the inversion's own benchmark didn't cover).
+ */
+export function shouldPreferParallelDrawElement(args: {
+  workerCount: number;
+  /** job.config.workers — a number means the user explicitly chose. */
+  requestedWorkers: number | "auto" | undefined;
+  useDrawElement: boolean;
+  deCompileGate: string | undefined;
+  forceScreenshot: boolean;
+  outputFormat: NonNullable<RenderConfig["format"]>;
+  totalFrames: number;
+  /** Amortization threshold; <=0 disables the router. */
+  minFrames: number;
+  layeredOrEffectRoute: boolean;
+  supersampling: boolean;
+  probeDeGated: boolean;
+  experimentalParallelDeOptIn: boolean;
+  /** HF_DE_PARALLEL_ROUTER === "true" — the router's own kill switch, default off. */
+  routerEnabled: boolean;
+}): boolean {
+  return (
+    args.routerEnabled &&
+    args.workerCount > 1 &&
+    typeof args.requestedWorkers !== "number" &&
+    args.useDrawElement &&
+    !args.deCompileGate &&
+    !args.forceScreenshot &&
+    args.outputFormat === "mp4" &&
+    args.minFrames > 0 &&
+    args.totalFrames >= args.minFrames &&
+    !args.layeredOrEffectRoute &&
+    !args.supersampling &&
+    !args.probeDeGated &&
+    !args.experimentalParallelDeOptIn
+  );
+}
+
+/**
+ * Plan the self-verify retry for a router-routed render: the bet on verified
+ * parallel drawElement streaming lost, so the re-render falls back to the
+ * pre-router worker count on the ordinary (non-DE) parallel path. Unlike
+ * `resolveInversionRetryPlan`, the caller must also clear the router's
+ * `deParallelStreamForced` local BEFORE calling this — `shouldUseStreamingEncode`
+ * takes it as a direct argument, so a stale `true` would keep resolving to
+ * the parallel-streaming shape on the retry instead of the well-tested
+ * parallel-disk fallback. Returns null when the render was not router-routed.
+ */
+export function resolveParallelRouterRetryPlan(args: {
+  deParallelRouter: "routed" | "reverted" | undefined;
+  preRouterWorkerCount: number;
+  cfg: Pick<EngineConfig, "enableStreamingEncode" | "streamingEncodeMaxDurationSeconds">;
+  outputFormat: NonNullable<RenderConfig["format"]>;
+  durationSeconds: number;
+}): {
+  workerCount: number;
+  useStreamingEncode: boolean;
+  deParallelRouter: "reverted";
+} | null {
+  if (args.deParallelRouter !== "routed") return null;
+  return {
+    workerCount: args.preRouterWorkerCount,
+    useStreamingEncode: shouldUseStreamingEncode(
+      args.cfg,
+      args.outputFormat,
+      args.preRouterWorkerCount,
+      args.durationSeconds,
+    ),
+    deParallelRouter: "reverted",
+  };
+}
+
 export function resolveCaptureForceScreenshotForPageSideCompositing(args: {
   forceScreenshot: boolean;
   usePageSideCompositing: boolean;
@@ -1217,6 +1314,9 @@ export async function executeRenderJob(
   // between declaration and the try-block (currently impossible, but
   // defensible if more setup ever lands here) can't leak the interval.
   let memSampler: MemorySampler | null = null;
+  // "routed" = the parallel router fired and held; "reverted" = fired but
+  // the self-verify retry rolled back; undefined = never fired.
+  let deParallelRouter: "routed" | "reverted" | undefined;
 
   try {
     memSampler = createMemorySampler();
@@ -1341,6 +1441,20 @@ export async function executeRenderJob(
     // "inverted" = fired and held; "reverted" = fired but the self-verify
     // retry rolled back to the parallel path; undefined = never fired.
     let deWorkerInversion: "inverted" | "reverted" | undefined;
+    // deParallelRouter is mutually exclusive with deWorkerInversion — the
+    // router takes priority when both would be eligible (see
+    // shouldPreferParallelDrawElement).
+    //
+    // Per-render (not process-global) signal that the router wants parallel
+    // drawElement streaming. `HF_DE_PARALLEL_STREAM` env var stays as the
+    // manual opt-in for local testing (read directly by
+    // shouldUseStreamingEncode / the capture stage), but the router itself
+    // must NOT mutate process.env: the producer server runs concurrent
+    // renders in one process (PRODUCER_MAX_CONCURRENT_RENDERS), and a global
+    // flag set by one render's router decision would leak into an unrelated
+    // 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;
     let deSelfVerifyFallback = false;
     let deFallbackReason: string | undefined;
     let deDrainStats: import("./render/stages/captureStreamingStage.js").DeDrainStats | undefined;
@@ -1718,12 +1832,46 @@ export async function executeRenderJob(
         // Verified parallel DE streaming (opt-in) wants its parallelism kept.
         process.env.HF_DE_PARALLEL_STREAM === "true",
     });
+    // DE parallel-router eligibility — see shouldPreferParallelDrawElement.
+    // Default-off (HF_DE_PARALLEL_ROUTER); HF_DE_PARALLEL_MIN_FRAMES defaults
+    // higher than the single-worker inversion's threshold since it targets
+    // the long tail the inversion's own benchmark didn't cover.
+    const deParallelRouterEnabled = process.env.HF_DE_PARALLEL_ROUTER === "true";
+    const deParallelMinFramesRaw = process.env.HF_DE_PARALLEL_MIN_FRAMES;
+    const deParallelMinFramesNum =
+      deParallelMinFramesRaw === undefined || deParallelMinFramesRaw.trim() === ""
+        ? 2000
+        : Number(deParallelMinFramesRaw);
+    const deParallelMinFrames = Number.isFinite(deParallelMinFramesNum)
+      ? deParallelMinFramesNum
+      : 2000;
+    const deParallelRouterEligible = shouldPreferParallelDrawElement({
+      workerCount: WOULD_RESOLVE_MULTI_WORKER,
+      requestedWorkers: job.config.workers,
+      useDrawElement: cfg.useDrawElement,
+      deCompileGate,
+      forceScreenshot: captureForceScreenshot,
+      outputFormat,
+      totalFrames,
+      minFrames: deParallelMinFrames,
+      layeredOrEffectRoute: hasHdrContent || compiled.hasShaderTransitions,
+      supersampling: deviceScaleFactor > 1,
+      probeDeGated:
+        probeSession !== null &&
+        probeSession.captureMode !== "drawelement" &&
+        !probeSession.deInitDeferred,
+      experimentalParallelDeOptIn:
+        process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE === "true" ||
+        process.env.HF_DE_PARALLEL_STREAM === "true",
+      routerEnabled: deParallelRouterEnabled,
+    });
     if (
       job.config.workers === undefined &&
       totalFrames >= 60 &&
       !htmlInCanvasDetected &&
       !cfg.lowMemoryMode &&
-      !deInversionEligible
+      !deInversionEligible &&
+      !deParallelRouterEligible
     ) {
       const outcome = await observeRenderStage(
         observability,
@@ -1763,6 +1911,7 @@ export async function executeRenderJob(
         htmlInCanvasDetected,
         lowMemoryMode: Boolean(cfg.lowMemoryMode),
         deInversionEligible,
+        deParallelRouterEligible,
       });
     }
 
@@ -1782,10 +1931,36 @@ export async function executeRenderJob(
     // INIT-time gate at capture (css-effects / at-risk, ~1.5% of local
     // renders) render single-worker screenshot streaming — slower than
     // parallel would have been, accepted for the routing win everywhere else.
-    // `preInversionWorkerCount` lets the self-verify retry return to the
-    // parallel path when the drawElement bet loses.
-    const preInversionWorkerCount = workerCount;
-    if (deInversionEligible && workerCount > 1) {
+    // `preRoutingWorkerCount` lets the self-verify retry return to the
+    // parallel path when the drawElement bet loses — shared by both the
+    // inversion and the router below, whichever fires (mutually exclusive).
+    const preRoutingWorkerCount = workerCount;
+    // Router takes priority over the single-worker inversion when both would
+    // fire — its higher frame threshold means this only ever picks up long-
+    // tail comps the inversion's own benchmark didn't cover (see
+    // shouldPreferParallelDrawElement). Pins to a fixed worker count exactly
+    // like the inversion pins to 1 — calibration is skipped for both (see
+    // the capture_calibration gate above), so this deliberately overrides
+    // whatever a calibrated resolution would have chosen (e.g. 2, on a
+    // resource-constrained host): the benchmark validated par3 specifically,
+    // not "whatever calibration picks above 1", and the self-verify retry is
+    // the safety net if 3 workers tips a given host over.
+    if (deParallelRouterEligible && workerCount > 1) {
+      deParallelRouter = "routed";
+      // Fixed at 3, not calibration-derived: the benchmark validated exactly
+      // this worker count (par3 beat par2 consistently; W4/W5 unmeasured for
+      // this path), same shape as the single-worker inversion pinning to a
+      // fixed 1 rather than a calibrated count.
+      const ROUTER_WORKER_COUNT = 3;
+      log.info(
+        "[Render] Fast capture: verified parallel drawElement streaming preferred over " +
+          `single-worker inversion (${totalFrames} frames >= ${deParallelMinFrames}; ` +
+          "benchmark-validated at 3 workers, pinned regardless of calibration). " +
+          "Set HF_DE_PARALLEL_ROUTER=false or --workers N to override.",
+      );
+      workerCount = ROUTER_WORKER_COUNT;
+      deParallelStreamForced = true;
+    } else if (deInversionEligible && workerCount > 1) {
       deWorkerInversion = "inverted";
       log.info(
         "[Render] Fast capture: single-worker drawElement streaming preferred over " +
@@ -1795,10 +1970,11 @@ export async function executeRenderJob(
       );
       workerCount = 1;
     }
-    updateCaptureObservability({ workerCount, deWorkerInversion });
+    updateCaptureObservability({ workerCount, deWorkerInversion, deParallelRouter });
     observability.checkpoint("worker_resolution", "resolved", {
       workerCount,
       deWorkerInversion: deWorkerInversion ?? "none",
+      deParallelRouter: deParallelRouter ?? "none",
     });
 
     if (workerCount > 1 && probeSession) {
@@ -1812,7 +1988,13 @@ export async function executeRenderJob(
     // let auto-parallel renders use disk frames: the current ordered streaming
     // writer would otherwise stall later workers behind earlier frame ranges.
     // png-sequence has no encoded video output, so streaming is always bypassed.
-    let useStreamingEncode = shouldUseStreamingEncode(cfg, outputFormat, workerCount, job.duration);
+    let useStreamingEncode = shouldUseStreamingEncode(
+      cfg,
+      outputFormat,
+      workerCount,
+      job.duration,
+      deParallelStreamForced,
+    );
     log.info("streaming-encode gate", {
       enabled: useStreamingEncode,
       configFlag: cfg.enableStreamingEncode,
@@ -1831,7 +2013,9 @@ export async function executeRenderJob(
     // drain guard), so the confinement rule is satisfied and the parallel
     // clamp does not apply. The disk path stays clamped.
     const deParallelStreamVerified =
-      process.env.HF_DE_PARALLEL_STREAM === "true" && useStreamingEncode && workerCount > 1;
+      (deParallelStreamForced || process.env.HF_DE_PARALLEL_STREAM === "true") &&
+      useStreamingEncode &&
+      workerCount > 1;
     if (
       cfg.useDrawElement &&
       process.env.PRODUCER_EXPERIMENTAL_FAST_CAPTURE !== "true" &&
@@ -2052,6 +2236,7 @@ export async function executeRenderJob(
                 workerCount,
                 probeSession,
                 outputFormat,
+                forceParallelStream: deParallelStreamForced,
                 streamingEncoderOptions: {
                   fps: job.config.fps,
                   width,
@@ -2101,9 +2286,21 @@ export async function executeRenderJob(
             deSelfVerifyFallback: true,
           });
           probeSession = null;
+          // Must clear BEFORE resolveParallelRouterRetryPlan recomputes
+          // useStreamingEncode, or shouldUseStreamingEncode would keep
+          // resolving to the parallel-streaming shape on the retry instead
+          // of the well-tested parallel-disk fallback.
+          if (deParallelRouter === "routed") deParallelStreamForced = false;
           const inversionRetryPlan = resolveInversionRetryPlan({
             deWorkerInversion,
-            preInversionWorkerCount,
+            preInversionWorkerCount: preRoutingWorkerCount,
+            cfg,
+            outputFormat,
+            durationSeconds: job.duration,
+          });
+          const parallelRouterRetryPlan = resolveParallelRouterRetryPlan({
+            deParallelRouter,
+            preRouterWorkerCount: preRoutingWorkerCount,
             cfg,
             outputFormat,
             durationSeconds: job.duration,
@@ -2126,6 +2323,23 @@ export async function executeRenderJob(
               `[Render] Reverting worker inversion for the retry: ${workerCount} workers, ` +
                 `streaming=${useStreamingEncode}.`,
             );
+          } else if (parallelRouterRetryPlan) {
+            // The router's bet on verified parallel streaming lost — re-render
+            // on the ordinary (non-DE) parallel path at the pre-router worker
+            // count, same "reverted, not cleared" telemetry contract as the
+            // inversion above.
+            deParallelRouter = parallelRouterRetryPlan.deParallelRouter;
+            workerCount = parallelRouterRetryPlan.workerCount;
+            useStreamingEncode = parallelRouterRetryPlan.useStreamingEncode;
+            updateCaptureObservability({
+              workerCount,
+              useStreamingEncode,
+              deParallelRouter,
+            });
+            log.info(
+              `[Render] Reverting parallel router for the retry: ${workerCount} workers, ` +
+                `streaming=${useStreamingEncode}.`,
+            );
           }
           if (useStreamingEncode) {
             streamingRes = await invokeStreaming();
@@ -2337,7 +2551,9 @@ export async function executeRenderJob(
         compileGate: deCompileGate,
         clampReason: deClampReason,
         workerInversion: deWorkerInversion,
-        preInversionWorkers: deWorkerInversion ? preInversionWorkerCount : undefined,
+        preInversionWorkers: deWorkerInversion ? preRoutingWorkerCount : undefined,
+        parallelRouter: deParallelRouter,
+        preRouterWorkers: deParallelRouter ? preRoutingWorkerCount : undefined,
         selfVerifyFallback: deSelfVerifyFallback,
         fallbackReason: deFallbackReason,
         drainStats: deDrainStats,