fix(process): coalesce delayed capture progress

Retain one pending intermediate observation and drop stale live status before terminal cleanup. Preserve monotonic counts and final delivery while an observer is slow or rejects an update.

Verify the send rate at its controlled-clock owner and lifecycle notification density at the CLI/MCP boundary, where receive timestamps include buffering and scheduling.

docs/process-capture.md

@@ -74,8 +74,10 @@ percentage. `message` reports elapsed time and those counts. The final line uses
 `timeout`, `idle_timeout`, `cancelled`, or `failed`) plus the owned-process,
 renderer, and temporary-root cleanup states. Updates use the shared progress
 boundary and are limited to one intermediate line per 100 ms; the terminal line
-is always emitted. Progress never copies child output. Child output remains in
-the capture Evidence on stdout.
+is always emitted. If a receiver is slow, intermediate observations are merged
+into the latest status, and stale live observations are discarded before final
+cleanup status. Progress never copies child output. Child output remains in the
+capture Evidence on stdout.
 
 For example:
 

src/application/ProgressReporter.test.ts

@@ -0,0 +1,33 @@
+import { expect, it } from "vitest";
+
+import { createProgressReporter } from "./ProgressReporter.js";
+
+it("limits intermediate sends to 100 ms and always delivers terminal status", async () => {
+  let now = 0;
+  const sent: number[] = [];
+  const reporter = createProgressReporter(
+    async () => {
+      sent.push(now);
+    },
+    { now: () => now },
+  );
+  for (const [completed, elapsed] of [0, 25, 50, 99, 100, 199, 200].entries()) {
+    now = elapsed;
+    await reporter.report({
+      phase: "running",
+      completed,
+      total: null,
+      message: "Observed capture status",
+    });
+  }
+  expect(sent).toEqual([0, 100, 200]);
+  now = 201;
+  await reporter.report({
+    phase: "cleanup",
+    completed: 7,
+    total: null,
+    message: "Owned cleanup complete",
+    terminal: true,
+  });
+  expect(sent).toEqual([0, 100, 200, 201]);
+});

src/process/capture/ProcessCaptureProgress.test.ts

@@ -0,0 +1,113 @@
+import { expect, it } from "vitest";
+
+import {
+  createProcessCaptureProgressTracker,
+  type ProcessCaptureProgress,
+} from "./ProcessCaptureProgress.js";
+
+const blockedProgress = () => {
+  let release: () => void = () => undefined;
+  const blocked = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  const delivered: Parameters<ProcessCaptureProgress["report"]>[0][] = [];
+  const progress: ProcessCaptureProgress = {
+    async report(update) {
+      delivered.push(update);
+      if (delivered.length === 1) await blocked;
+    },
+  };
+  return { progress, delivered, release: () => release() };
+};
+
+it("coalesces intermediate observations while the progress receiver is blocked", async () => {
+  const receiver = blockedProgress();
+  const tracker = createProcessCaptureProgressTracker(receiver.progress);
+  void tracker.report({
+    phase: "running",
+    counts: { frames: 1, samples: 0, interactions: 0 },
+  });
+  await Promise.resolve();
+  let latest: Promise<void> | undefined;
+  for (let frames = 2; frames <= 1_000; frames += 1) {
+    latest = tracker.report({
+      phase: "running",
+      counts: { frames, samples: 0, interactions: 0 },
+    });
+  }
+  expect(receiver.delivered).toHaveLength(1);
+  receiver.release();
+  await latest;
+  expect(receiver.delivered.map((update) => update.completed)).toEqual([
+    1, 1_000,
+  ]);
+  expect(receiver.delivered[1]?.message).toContain("frames=1000");
+});
+
+it("discards stale live observations before delivering final cleanup", async () => {
+  const receiver = blockedProgress();
+  const tracker = createProcessCaptureProgressTracker(receiver.progress);
+  void tracker.report({
+    phase: "running",
+    counts: { frames: 1, samples: 0, interactions: 0 },
+  });
+  await Promise.resolve();
+  void tracker.report({
+    phase: "running",
+    counts: { frames: 2, samples: 0, interactions: 0 },
+  });
+  void tracker.report({
+    phase: "settling",
+    counts: { frames: 2, samples: 1, interactions: 0 },
+  });
+  tracker.closeLive();
+  await tracker.report({
+    phase: "running",
+    counts: { frames: 10, samples: 10, interactions: 10 },
+  });
+  const terminal = tracker.report({
+    phase: "cleanup",
+    counts: { frames: 2, samples: 2, interactions: 0 },
+    disposition: "exited",
+    terminal: true,
+  });
+  receiver.release();
+  await terminal;
+  expect(receiver.delivered.map((update) => update.phase)).toEqual([
+    "running",
+    "cleanup",
+  ]);
+  expect(receiver.delivered.at(-1)).toMatchObject({
+    completed: 4,
+    total: null,
+    terminal: true,
+  });
+});
+
+it("still delivers terminal status after an observer rejects an update", async () => {
+  const delivered: Parameters<ProcessCaptureProgress["report"]>[0][] = [];
+  const tracker = createProcessCaptureProgressTracker({
+    async report(update) {
+      delivered.push(update);
+      if (delivered.length === 1) throw new Error("Observer disconnected");
+    },
+  });
+  await tracker.report({
+    phase: "prepare",
+    counts: { frames: 0, samples: 0, interactions: 0 },
+  });
+  tracker.closeLive();
+  await expect(
+    tracker.report({
+      phase: "cleanup",
+      counts: { frames: 1, samples: 1, interactions: 0 },
+      disposition: "failed",
+      terminal: true,
+    }),
+  ).resolves.toBeUndefined();
+  expect(delivered.at(-1)).toMatchObject({
+    phase: "cleanup",
+    completed: 2,
+    terminal: true,
+  });
+});

src/process/capture/ProcessCaptureProgress.ts

@@ -66,10 +66,28 @@ export const createProcessCaptureProgressTracker = (
   const started = now();
   let completed = 0;
   let live = true;
-  let pending = Promise.resolve();
+  let next: Parameters<ProcessCaptureProgress["report"]>[0] | undefined;
+  let pending: Promise<void> | undefined;
+  const flush = async (): Promise<void> => {
+    try {
+      while (next !== undefined) {
+        const update = next;
+        next = undefined;
+        try {
+          await progress?.report(update);
+        } catch {
+          // Progress delivery is observational; capture and cleanup still finish.
+          continue;
+        }
+      }
+    } finally {
+      pending = undefined;
+    }
+  };
   return {
     closeLive(): void {
       live = false;
+      if (next?.terminal !== true) next = undefined;
     },
     report(update: ProcessCaptureProgressReport): Promise<void> {
       if (progress === undefined) return Promise.resolve();
@@ -79,7 +97,8 @@ export const createProcessCaptureProgressTracker = (
         update.counts.samples +
         update.counts.interactions;
       if (units > completed) completed = units;
-      const payload = {
+      // A slow observer needs the latest observation, not a queue of old states.
+      next = {
         phase: update.phase,
         completed,
         total: null as null,
@@ -91,16 +110,8 @@ export const createProcessCaptureProgressTracker = (
         ),
         ...(update.terminal === true ? { terminal: true as const } : {}),
       };
-      const send = () => progress.report(payload);
-      const scheduled = pending.then(send, send);
-      pending = scheduled.then(
-        () => undefined,
-        () => undefined,
-      );
-      return scheduled.then(
-        () => undefined,
-        () => undefined,
-      );
+      pending ??= Promise.resolve().then(flush);
+      return pending;
     },
   };
 };

tests/boundary/process/processCaptureProgress.test.ts

@@ -84,7 +84,6 @@ interface CliCapture {
   readonly exitCode: number | null;
   readonly lines: readonly {
     readonly update: ParsedProgress;
-    readonly at: number;
   }[];
 }
 
@@ -110,7 +109,7 @@ const runCaptureCli = async (
   let stdout = "";
   let stderr = "";
   let pending = "";
-  const lines: Array<{ readonly line: string; readonly at: number }> = [];
+  const lines: string[] = [];
   let sawRunning: (() => void) | undefined;
   const running = options.stopOnRunning
     ? new Promise<void>((resolve) => {
@@ -126,7 +125,7 @@ const runCaptureCli = async (
     pending = parts.pop() ?? "";
     for (const line of parts) {
       if (line.length === 0) continue;
-      lines.push({ line, at: Date.now() });
+      lines.push(line);
       if (line.includes('"phase":"running"')) sawRunning?.();
     }
   });
@@ -135,32 +134,34 @@ const runCaptureCli = async (
     child.once("close", (code) => resolve({ code }));
   });
   if (running !== undefined) {
+    let timer: ReturnType<typeof setTimeout> | undefined;
     const appeared = await Promise.race([
       running.then(() => true),
       new Promise<boolean>((resolve) => {
-        setTimeout(() => resolve(false), 8_000);
+        timer = setTimeout(() => resolve(false), 8_000);
       }),
-    ]);
+    ]).finally(() => {
+      clearTimeout(timer);
+    });
     if (!appeared) {
       child.kill("SIGKILL");
       await closed;
       throw new Error(
-        `capture produced no running progress on stderr:\n${lines.map(({ line }) => line).join("\n")}`,
+        `capture produced no running progress on stderr:\n${lines.join("\n")}`,
       );
     }
     child.kill("SIGINT");
   }
   const result = await closed;
-  if (pending.length > 0) lines.push({ line: pending, at: Date.now() });
-  stderr = lines.map(({ line }) => line).join("\n");
+  if (pending.length > 0) lines.push(pending);
+  stderr = lines.join("\n");
   return {
     stdout,
     stderr,
     exitCode: result.code,
-    lines: lines.map(({ line, at }) => {
+    lines: lines.map((line) => {
       const update = progressLineSchema.parse(JSON.parse(line)).rea_progress;
       return {
-        at,
         update: { ...update, details: parseProgressMessage(update.message) },
       };
     }),
@@ -178,10 +179,16 @@ const assertMonotonic = (updates: readonly ParsedProgress[]): void => {
   }
 };
 
-const assertRateBound = (times: readonly number[]): void => {
-  const gaps = times.slice(1).map((time, index) => time - (times[index] ?? 0));
-  expect(gaps.length).toBeGreaterThan(0);
-  expect(Math.min(...gaps)).toBeGreaterThanOrEqual(PROGRESS_INTERVAL_MS);
+const assertRateBound = (
+  intermediateCount: number,
+  elapsedMs: number,
+): void => {
+  // Pipe/transport buffering can deliver correctly spaced sends together.
+  // Exact send spacing belongs to the reporter's controlled-clock test.
+  expect(intermediateCount).toBeGreaterThan(1);
+  expect(intermediateCount).toBeLessThanOrEqual(
+    Math.floor(elapsedMs / PROGRESS_INTERVAL_MS) + 1,
+  );
 };
 
 const childScript = (source: string) => ["-e", source];
@@ -295,7 +302,10 @@ itWithCaptureCapability(
     const nonTerminal = capture.lines.filter(
       ({ update }) => update.terminal !== true,
     );
-    assertRateBound(nonTerminal.map(({ at }) => at));
+    assertRateBound(
+      nonTerminal.length,
+      updates.at(-1)?.details.elapsed_ms ?? 0,
+    );
     expect(
       Math.max(...updates.map(({ details }) => details.frames)),
     ).toBeGreaterThanOrEqual(10);
@@ -393,7 +403,6 @@ itWithCaptureCapability(
       readonly progress: number;
       readonly total: number | undefined;
       readonly message: string | undefined;
-      readonly at: number;
     }> = [];
     const captured = await tokenClient.callTool(
       {
@@ -413,7 +422,6 @@ itWithCaptureCapability(
             progress: update.progress,
             total: update.total,
             message: update.message,
-            at: Date.now(),
           });
         },
       },
@@ -436,7 +444,11 @@ itWithCaptureCapability(
     expect(messages.at(-1)).toContain("owned_process_group=cleaned");
     expect(messages.at(-1)).toContain("disposition=exited");
     const nonTerminal = progress.slice(0, -1);
-    assertRateBound(nonTerminal.map(({ at }) => at));
+    assertRateBound(
+      nonTerminal.length,
+      parseProgressMessage((messages.at(-1) ?? "").slice("cleanup: ".length))
+        .elapsed_ms,
+    );
     const tokenEvidence = toolContract(
       "capture_process_scenario",
     ).outputSchema.parse(captured.structuredContent);