fix(producer): record the manual interleave opt-in on the plan so its retry drops to one worker

The streaming stage interleaves on forceParallelStream OR
HF_DE_PARALLEL_STREAM=true, but the plan only carried the two routers'
flags. A manual opt-in below the DE router's frame floor therefore retried
at N workers with forceParallelStream false, which the env var re-interleaved
on the second attempt, contradicting the single-worker guarantee the retry
comment and log line made. isParallelStreamForced folds the env var into the
plan flag at creation, so the plan, the retry target and the
deParallelStreamForced telemetry agree with the distribution the stage
picks. Found by the adversarial review of this branch.

Also from that review: the parallel-stall log and checkpoint no longer claim
a "single-worker" retry, which router-pinned plans do not make (they take
their precomputed routing fallback; the post-replan log carries the real
count); the fallbackReason doc comments list encoder_death and
parallel_stall; the peer-death test names the engine test that pins the
hook contract it replays.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

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

@@ -51,7 +51,7 @@ export interface RenderCaptureObservability {
    * render re-ran via screenshot. NARROWED semantics since the pinned-fallback
    * retry was widened (review): OOM- and generic-capture-error-triggered
    * fallbacks report FALSE here, with `deFallbackReason` ∈ {oom,
-   * capture_error}. The "any fallback fired" signal is `deFallbackReason`
+   * de_renderer_stall, encoder_death, parallel_stall, capture_error}. The "any fallback fired" signal is `deFallbackReason`
    * being set, NOT this flag — dashboards keyed on `de_self_verify_fallback =
    * true` as any-fallback must migrate to `de_fallback_reason IS NOT NULL`.
    */
@@ -59,14 +59,15 @@ export interface RenderCaptureObservability {
   /**
    * Why the capture-stage retry (self-verify OR the pinned-worker-count
    * fallback) fired: "blank"/"psnr" for a real self-verify trip,
-   * "oom"/"capture_error" for the widened generic-failure retry. Set
+   * "oom"/"de_renderer_stall"/"encoder_death"/"parallel_stall"/"capture_error"
+   * for the widened generic-failure retry. Set
    * whenever a fallback is attempted, independent of whether that retry
    * itself later succeeds — so a render that fails AFTER a fallback attempt
    * (perfSummary never built) is still distinguishable in failure-path
    * telemetry from one that never attempted any fallback.
    */
   deFallbackReason?: string;
-  /** The failing PSNR (dB) when `deFallbackReason === "psnr"`; undefined for blank/oom/capture_error (no score exists). */
+  /** The failing PSNR (dB) when `deFallbackReason === "psnr"`; undefined for every other reason (no score exists). */
   deFallbackFailedDb?: number;
   /** Frame index the verification failure was detected at; set for both "psnr" and "blank" fallback reasons. */
   deFallbackFrameIndex?: number;

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

@@ -101,6 +101,9 @@ mock.module("@hyperframes/engine", () => ({
     if (failWorkerTransient) {
       // One worker's Chrome died. The engine reports the ORIGINAL failure
       // through the hook, then the pool rejects with its flattened summary.
+      // That ordering is the engine's contract, pinned without a browser by
+      // parallelCoordinator.test.ts "createPoolFailureHandler"; this mock
+      // only replays it.
       injectedWorkerFailure = Object.assign(
         new Error("Protocol error (Page.captureScreenshot): Target closed"),
         { kind: "transient_browser" },

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

@@ -43,6 +43,7 @@ import {
   isDeRendererStallError,
   isSequentialCaptureStallError,
   isParallelCaptureStallError,
+  isParallelStreamForced,
   fallbackCaptureModeLabel,
   isRetryableEncoderDeath,
   scanElementTags,
@@ -3813,3 +3814,20 @@ describe("fallbackCaptureModeLabel: trace label when no probe session is open",
     ).toBe("drawelement");
   });
 });
+
+describe("isParallelStreamForced: the plan carries the manual interleave opt-in", () => {
+  const flagsOff = { deParallelStreamForced: false, captureParallelStreamForced: false };
+
+  it("folds HF_DE_PARALLEL_STREAM=true into the plan flag so the retry drops to one worker", () => {
+    expect(isParallelStreamForced({ HF_DE_PARALLEL_STREAM: "true" }, flagsOff)).toBe(true);
+  });
+
+  it("otherwise follows the two routers' flags", () => {
+    expect(isParallelStreamForced({}, flagsOff)).toBe(false);
+    expect(isParallelStreamForced({ HF_DE_PARALLEL_STREAM: "false" }, flagsOff)).toBe(false);
+    expect(isParallelStreamForced({}, { ...flagsOff, deParallelStreamForced: true })).toBe(true);
+    expect(isParallelStreamForced({}, { ...flagsOff, captureParallelStreamForced: true })).toBe(
+      true,
+    );
+  });
+});

packages/producer/src/services/renderOrchestrator.ts

@@ -626,7 +626,7 @@ export interface RenderPerfSummary {
      * `fallbackReason` being set is the "any fallback fired" signal.
      */
     selfVerifyFallback: boolean;
-    /** What tripped the fallback retry: psnr | blank | oom | de_renderer_stall | capture_error. */
+    /** What tripped the fallback retry: psnr | blank | oom | de_renderer_stall | encoder_death | parallel_stall | capture_error. */
     fallbackReason?: string;
     /** The failing PSNR (dB) when `fallbackReason === "psnr"`; undefined for every other reason (no score exists). */
     fallbackFailedDb?: number;
@@ -2557,6 +2557,28 @@ export function isCaptureParallelStreamRouterEnabled(
   return true;
 }
 
+/**
+ * Whether this render distributes frames interleaved across its workers. The
+ * two routers set their flags; `HF_DE_PARALLEL_STREAM=true` is the manual
+ * opt-in the streaming stage also honours on its own. Folding all three into
+ * the plan is what keeps the plan, and so the retry target and telemetry, in
+ * agreement with the distribution the stage actually picks: before this, an
+ * env-opt-in render below the DE router's frame floor carried
+ * `forceParallelStream: false`, so its retry kept N workers, which the env
+ * var then re-interleaved instead of dropping to the hardened single-worker
+ * path. Pure; exported for tests.
+ */
+export function isParallelStreamForced(
+  env: Readonly<Record<string, string | undefined>>,
+  flags: { deParallelStreamForced: boolean; captureParallelStreamForced: boolean },
+): boolean {
+  return (
+    flags.deParallelStreamForced ||
+    flags.captureParallelStreamForced ||
+    env.HF_DE_PARALLEL_STREAM === "true"
+  );
+}
+
 /**
  * The duration at which segmented capture is meant to become the default
  * (spec §5 Phase 2d). Not the shipped default: that flip is its own release
@@ -4234,7 +4256,10 @@ async function executeRenderPipeline(input: {
     let capturePlan: CapturePlan = createCapturePlan({
       workerCount,
       forceScreenshot: captureForceScreenshot,
-      forceParallelStream: deParallelStreamForced || captureParallelStreamForced,
+      forceParallelStream: isParallelStreamForced(process.env, {
+        deParallelStreamForced,
+        captureParallelStreamForced,
+      }),
       useStreamingEncode,
       // Segmented capture is opt-in in Phase 2a and single-worker only; the
       // excluded routes are the ones whose concat-copy or capture loop the
@@ -4672,7 +4697,7 @@ async function executeRenderPipeline(input: {
                 : isSequentialStall
                   ? "[Render] sequential capture stalled; retrying on a fresh screenshot session"
                   : isParallelStall
-                    ? "[Render] parallel capture stalled; retrying on a fresh single-worker screenshot session"
+                    ? "[Render] parallel capture stalled; retrying on a fresh screenshot session"
                     : isEncoderDeath
                       ? "[Render] streaming encoder died mid-render; retrying on a fresh screenshot session"
                       : "[Render] capture failed; re-rendering via a fresh screenshot session",
@@ -4687,7 +4712,7 @@ async function executeRenderPipeline(input: {
                 : isSequentialStall
                   ? "sequential capture stalled; retrying with a fresh screenshot session"
                   : isParallelStall
-                    ? "parallel capture stalled; retrying with a fresh single-worker screenshot session"
+                    ? "parallel capture stalled; retrying with a fresh screenshot session"
                     : isEncoderDeath
                       ? "streaming encoder died; retrying with a fresh screenshot session"
                       : "capture failed; retrying with a fresh screenshot session",