Merge pull request #2510 from heygen-com/via/parallel-capture-observability

feat(producer): calibration-aware heartbeat + worker-death terminal-error contract

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

@@ -2,10 +2,12 @@ import { describe, it, expect } from "vitest";
 import {
   calculateOptimalWorkers,
   distributeFrames,
+  expectedFramesForTask,
   formatWorkerFailure,
   selectWorkerDiagnostics,
   shouldDisableBrowserPoolForParallelWorker,
   shouldVerifyWorkerGpu,
+  synthesizeSilentWorkerExitError,
   resolveParallelDeVerifySamples,
 } from "./parallelCoordinator.js";
 import type { EngineConfig } from "../config.js";
@@ -161,6 +163,91 @@ describe("worker failure diagnostics", () => {
       "Worker 1: Navigation timeout of 60000 ms exceeded; diagnostics: [FrameCapture:ERROR] page.goto failed mode=screenshot timeoutMs=60000",
     );
   });
+
+  // Field signal ts=1784042064: a Windows render hard-exited during video
+  // frame extraction without emitting a terminal error string; the parent
+  // treated the result as success because `error` was falsy. The synthesized
+  // string surfaces the shortfall so downstream telemetry can classify the
+  // failure instead of it disappearing silently.
+  it("synthesizes a terminal error when a worker exits with no error string", () => {
+    const message = formatWorkerFailure({
+      workerId: 3,
+      framesCaptured: 12,
+      startFrame: 0,
+      endFrame: 60,
+      durationMs: 180_000,
+    });
+    expect(message).toContain("Worker 3: worker 3 exited without terminal error string");
+    expect(message).toContain("framesCaptured=12");
+    expect(message).toContain("expected=60");
+    expect(message).toContain("range=[0, 60)");
+    expect(message).toContain("ts=1784042064");
+    expect(message).toContain("--workers=1");
+  });
+
+  it("prefers an explicit error string over the synthesized silent-exit message", () => {
+    const message = formatWorkerFailure({
+      workerId: 3,
+      framesCaptured: 12,
+      startFrame: 0,
+      endFrame: 60,
+      durationMs: 180_000,
+      error: "Target closed",
+    });
+    expect(message).toBe("Worker 3: Target closed");
+    expect(message).not.toContain("ts=1784042064");
+    expect(message).not.toContain("exited without terminal error string");
+  });
+
+  it("treats an empty error string as a silent exit for the message contract", () => {
+    const message = formatWorkerFailure({
+      workerId: 3,
+      framesCaptured: 12,
+      startFrame: 0,
+      endFrame: 60,
+      durationMs: 180_000,
+      error: "",
+    });
+    expect(message).toContain("exited without terminal error string");
+    expect(message).toContain("ts=1784042064");
+  });
+});
+
+describe("expectedFramesForTask", () => {
+  it("returns the range length for contiguous tasks", () => {
+    expect(expectedFramesForTask({ startFrame: 0, endFrame: 30 })).toBe(30);
+    expect(expectedFramesForTask({ startFrame: 10, endFrame: 25 })).toBe(15);
+  });
+
+  it("divides by stride for interleaved tasks", () => {
+    // Worker 0 in a 3-way interleave over 30 frames captures 0, 3, 6, ..., 27 → 10 frames.
+    expect(expectedFramesForTask({ startFrame: 0, endFrame: 30, frameStride: 3 })).toBe(10);
+  });
+
+  it("rounds up so the last-stride-remainder frame is expected", () => {
+    // Worker 0 in a 3-way interleave over 31 frames captures 0, 3, ..., 30 → 11 frames.
+    expect(expectedFramesForTask({ startFrame: 0, endFrame: 31, frameStride: 3 })).toBe(11);
+  });
+
+  it("returns zero for empty ranges", () => {
+    expect(expectedFramesForTask({ startFrame: 10, endFrame: 10 })).toBe(0);
+    expect(expectedFramesForTask({ startFrame: 30, endFrame: 10 })).toBe(0);
+  });
+});
+
+describe("synthesizeSilentWorkerExitError", () => {
+  it("names the field signal and reruns hint so operators can classify the failure", () => {
+    const message = synthesizeSilentWorkerExitError(
+      { workerId: 7, framesCaptured: 40, startFrame: 60, endFrame: 120 },
+      60,
+    );
+    expect(message).toContain("worker 7 exited without terminal error string");
+    expect(message).toContain("framesCaptured=40");
+    expect(message).toContain("expected=60");
+    expect(message).toContain("range=[60, 120)");
+    expect(message).toContain("Field signal ts=1784042064");
+    expect(message).toContain("--workers=1");
+  });
 });
 
 describe("shouldVerifyWorkerGpu", () => {

packages/engine/src/services/parallelCoordinator.ts

@@ -147,8 +147,49 @@ function compactDiagnosticLine(line: string): string {
   return line.replace(/\s+/g, " ").trim();
 }
 
+/**
+ * Expected frame count for a worker task, honoring its stride. Contiguous
+ * tasks (stride 1) expect `endFrame - startFrame`; interleaved tasks
+ * (stride > 1) expect `ceil((endFrame - startFrame) / stride)`, matching
+ * the loop shape in `captureFrameRange`.
+ */
+export function expectedFramesForTask(task: {
+  startFrame: number;
+  endFrame: number;
+  frameStride?: number;
+}): number {
+  const stride = task.frameStride ?? 1;
+  return Math.max(0, Math.ceil((task.endFrame - task.startFrame) / stride));
+}
+
+/**
+ * Synthetic terminal-error message for a worker whose exit didn't produce
+ * an explicit error string but under-captured its expected frame range.
+ * Field signal ts=1784042064: a 1292s Windows render hard-exited during
+ * capture with no final error string, leaving the operator with no
+ * actionable trace. This message surfaces the shortfall + reruns hint
+ * so downstream telemetry (and operators grepping logs) can classify the
+ * failure instead of it disappearing silently.
+ */
+export function synthesizeSilentWorkerExitError(
+  result: Pick<WorkerResult, "workerId" | "framesCaptured" | "startFrame" | "endFrame">,
+  expectedFrames: number,
+): string {
+  return (
+    `worker ${result.workerId} exited without terminal error string ` +
+    `(framesCaptured=${result.framesCaptured}, expected=${expectedFrames}, ` +
+    `range=[${result.startFrame}, ${result.endFrame})). ` +
+    `Field signal ts=1784042064 — this class of failure has been reported; ` +
+    `consider re-run with --workers=1 to isolate.`
+  );
+}
+
 export function formatWorkerFailure(result: WorkerResult): string {
-  const base = `Worker ${result.workerId}: ${result.error ?? "unknown error"}`;
+  const errorText =
+    result.error && result.error.length > 0
+      ? result.error
+      : synthesizeSilentWorkerExitError(result, expectedFramesForTask(result));
+  const base = `Worker ${result.workerId}: ${errorText}`;
   if (!result.diagnostics || result.diagnostics.length === 0) return base;
 
   const diagnostics = result.diagnostics.map(compactDiagnosticLine).join(" | ");
@@ -541,6 +582,17 @@ export async function executeParallelCapture(
     ),
   );
 
+  // A worker may return without an error string yet with framesCaptured
+  // below the task's expected count — that's the silent-exit shape field
+  // signal ts=1784042064 called out. Synthesize a terminal error string
+  // in-place so the filter below treats it as a failure (and so the
+  // caller's failure message actually names what went wrong).
+  for (const r of results) {
+    if (!r.error && r.framesCaptured < expectedFramesForTask(r)) {
+      r.error = synthesizeSilentWorkerExitError(r, expectedFramesForTask(r));
+    }
+  }
+
   const errors = results.filter((r) => r.error);
   if (errors.length > 0) {
     const errorMessages = errors.map(formatWorkerFailure).join("; ");

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

@@ -209,6 +209,98 @@ describe("RenderObservabilityRecorder", () => {
     vi.useRealTimers();
   });
 
+  // Field signal ts=1784019503: a healthy ~64s browser calibration
+  // pre-capture emitted heartbeats saying "stage still running /
+  // framesCompleted: 0" and read as broken to downstream consumers. The
+  // calibration call sites now pass `heartbeatMessage: "browser calibrating
+  // (frames not started)"` so operator-facing logs and metrics can
+  // distinguish healthy calibration waits from actual zero-frame stalls
+  // once capture is meant to be underway.
+  it("uses a custom heartbeat message during calibration stages", async () => {
+    vi.useFakeTimers();
+    const log = makeLog();
+    const recorder = new RenderObservabilityRecorder({
+      pipelineStartMs: Date.now(),
+      log,
+      renderJobId: "render-calibrating",
+    });
+    let resolveStage: (() => void) | undefined;
+    const stage = observeRenderStage(
+      recorder,
+      "capture_calibration",
+      { stagePhase: "calibrating", framesCompleted: 0, totalFrames: 900 },
+      () =>
+        new Promise<void>((resolve) => {
+          resolveStage = resolve;
+        }),
+      { heartbeatMessage: "browser calibrating (frames not started)" },
+    );
+
+    await vi.advanceTimersByTimeAsync(30_000);
+    const heartbeatCalls = log.info.mock.calls.filter(
+      ([message, meta]) =>
+        message === "[Render:trace]" &&
+        meta?.phase === "capture_calibration" &&
+        meta?.status === "checkpoint",
+    );
+    expect(heartbeatCalls).toHaveLength(1);
+    expect(heartbeatCalls[0]?.[1]).toEqual(
+      expect.objectContaining({
+        message: "browser calibrating (frames not started)",
+        stagePhase: "calibrating",
+        framesCompleted: 0,
+        heartbeatIndex: 1,
+      }),
+    );
+    // No "stage still running" for this stage — the calibration override
+    // must fully replace the default, not sit alongside it.
+    expect(
+      log.info.mock.calls.some(
+        ([message, meta]) =>
+          message === "[Render:trace]" && meta?.message === "stage still running",
+      ),
+    ).toBe(false);
+
+    resolveStage?.();
+    await stage;
+    vi.useRealTimers();
+  });
+
+  it("keeps the default heartbeat message when no override is passed", async () => {
+    vi.useFakeTimers();
+    const log = makeLog();
+    const recorder = new RenderObservabilityRecorder({
+      pipelineStartMs: Date.now(),
+      log,
+      renderJobId: "render-default",
+    });
+    let resolveStage: (() => void) | undefined;
+    const stage = observeRenderStage(
+      recorder,
+      "capture_streaming",
+      { stagePhase: "capturing", framesCompleted: 42, totalFrames: 900 },
+      () =>
+        new Promise<void>((resolve) => {
+          resolveStage = resolve;
+        }),
+    );
+
+    await vi.advanceTimersByTimeAsync(30_000);
+    const heartbeatCalls = log.info.mock.calls.filter(
+      ([message, meta]) => message === "[Render:trace]" && meta?.message === "stage still running",
+    );
+    expect(heartbeatCalls).toHaveLength(1);
+    expect(heartbeatCalls[0]?.[1]).toEqual(
+      expect.objectContaining({
+        stagePhase: "capturing",
+        framesCompleted: 42,
+      }),
+    );
+    resolveStage?.();
+    await stage;
+    vi.useRealTimers();
+  });
+
   it("clears pending heartbeats when a stage rejects", async () => {
     vi.useFakeTimers();
     const log = makeLog();

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

@@ -164,6 +164,12 @@ const ALLOWED_STRING_DATA_KEYS = new Set([
   "renderJobId",
   "requestedHdrMode",
   "requestedWorkers",
+  // "calibrating" during pre-capture browser warm-up / calibration stages;
+  // "capturing" during capture_disk / capture_streaming / capture_hdr_*
+  // stages. Distinguishes healthy 0-frame heartbeats (browser starting up)
+  // from actual zero-frame stalls once capture is meant to be underway.
+  // Field signal ts=1784019503.
+  "stagePhase",
 ]);
 const RESERVED_LOG_KEYS = new Set([
   "data",
@@ -416,12 +422,40 @@ function heartbeatTargetMs(index: number): number {
   return HEARTBEAT_RAMP_END_MS + overflow * HEARTBEAT_REPEAT_MS;
 }
 
+/**
+ * Options for `observeRenderStage` heartbeat behavior.
+ *
+ * The default heartbeat message is "stage still running", chosen for the
+ * capture stages where a live frame count in the observation data already
+ * communicates progress. Stages that run BEFORE any frame count is
+ * meaningful — the ~64s browser calibration path in particular — inherit
+ * that default and confusingly report "stage still running / framesCompleted:
+ * 0" to downstream consumers. Field signal ts=1784019503 captured exactly
+ * that read-as-broken shape on a healthy 64s calibration.
+ *
+ * `heartbeatMessage` lets the calibration call sites override the message
+ * to "browser calibrating" so operator-facing logs and downstream metrics
+ * can distinguish healthy pre-capture waits from actual zero-frame stalls
+ * mid-capture. The `data` payload also flows through (callers pass a
+ * `stagePhase: "calibrating" | "capturing"` field) so structured consumers
+ * don't have to string-match on the message.
+ */
+export interface ObserveRenderStageOptions {
+  /**
+   * Message to attach to each heartbeat checkpoint for this stage.
+   * Defaults to "stage still running".
+   */
+  heartbeatMessage?: string;
+}
+
 export async function observeRenderStage<T>(
   recorder: RenderObservabilityRecorder,
   phase: string,
   data: RenderObservationData | undefined,
   fn: () => Promise<T>,
+  options: ObserveRenderStageOptions = {},
 ): Promise<T> {
+  const heartbeatMessage = options.heartbeatMessage ?? "stage still running";
   const startedAt = recorder.stageStart(phase, data);
   let heartbeatCount = 0;
   let lastFiredAtMs = 0;
@@ -431,7 +465,7 @@ export async function observeRenderStage<T>(
     heartbeatTimer = setTimeout(() => {
       lastFiredAtMs = targetMs;
       heartbeatCount += 1;
-      recorder.checkpoint(phase, "stage still running", {
+      recorder.checkpoint(phase, heartbeatMessage, {
         ...data,
         heartbeatIndex: heartbeatCount,
         stageElapsedMs: Date.now() - startedAt,

packages/producer/src/services/renderOrchestrator.ts

@@ -1862,7 +1862,7 @@ export async function executeRenderJob(
     const probeResult = await observeRenderStage(
       observability,
       "browser_probe",
-      { forceScreenshot: captureForceScreenshot },
+      { forceScreenshot: captureForceScreenshot, stagePhase: "calibrating" },
       () =>
         runProbeStage({
           projectDir,
@@ -1879,6 +1879,10 @@ export async function executeRenderJob(
           needsAlpha,
           deviceScaleFactor,
         }),
+      // Browser probe is pre-capture; report `browser calibrating` so a
+      // slow probe (~64s SwiftShader warm-up on Windows was the reported
+      // shape) doesn't read as a zero-frame stall. Field signal ts=1784019503.
+      { heartbeatMessage: "browser calibrating (frames not started)" },
     );
     compiled = probeResult.compiled;
     compositionHash = computeCompositionObservabilityHash(compiled.html);
@@ -2277,9 +2281,15 @@ export async function executeRenderJob(
     // captureStageObservationData can close over it for the calibration
     // stage itself — reads as undefined until resolveRenderWorkerCount runs.
     let workerCount: number;
+    // Default `stagePhase` — spread FIRST so a caller can override via
+    // `extra` (the calibration call site passes `stagePhase: "calibrating"`
+    // to distinguish healthy pre-capture waits from actual zero-frame stalls
+    // during capture; heartbeats in `capture_calibration` otherwise emit
+    // `framesCompleted: 0` and read as broken). Field signal ts=1784019503.
     const captureStageObservationData = (
       extra: RenderObservationData = {},
     ): RenderObservationData => ({
+      stagePhase: "capturing",
       ...extra,
       get workerCount() {
         return workerCount;
@@ -2329,7 +2339,14 @@ export async function executeRenderJob(
       const outcome = await observeRenderStage(
         observability,
         "capture_calibration",
-        captureStageObservationData({ forceScreenshot: captureForceScreenshot }),
+        captureStageObservationData({
+          forceScreenshot: captureForceScreenshot,
+          // Override the default `capturing` — calibration writes probe
+          // frames only, not `job.framesRendered`, so heartbeats reporting
+          // `framesCompleted: 0` misread as broken. Field signal
+          // ts=1784019503.
+          stagePhase: "calibrating",
+        }),
         () =>
           runCaptureCalibration({
             cfg,
@@ -2344,6 +2361,7 @@ export async function executeRenderJob(
             createRenderVideoFrameInjector,
             assertNotAborted,
           }),
+        { heartbeatMessage: "browser calibrating (frames not started)" },
       );
       captureCalibration = outcome.calibration;
       captureForceScreenshot = outcome.forceScreenshot;