From 6cd5ea0fb282c23cd08ca03e81ae2436b8c0661b Mon Sep 17 00:00:00 2001 From: N0zoM1z0 <161784452+N0zoM1z0@users.noreply.github.com> Date: Sat, 10 Oct 2026 17:33:17 +0800 Subject: [PATCH] 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 | 6 +- src/application/ProgressReporter.test.ts | 33 +++++ .../capture/ProcessCaptureProgress.test.ts | 113 ++++++++++++++++++ src/process/capture/ProcessCaptureProgress.ts | 35 ++++-- .../process/processCaptureProgress.test.ts | 48 +++++--- 5 files changed, 203 insertions(+), 32 deletions(-) create mode 100644 src/application/ProgressReporter.test.ts create mode 100644 src/process/capture/ProcessCaptureProgress.test.ts diff --git a/docs/process-capture.md b/docs/process-capture.md index 0bf6d282f3..723a53561a 100644 --- a/docs/process-capture.md +++ b/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: diff --git a/src/application/ProgressReporter.test.ts b/src/application/ProgressReporter.test.ts new file mode 100644 index 0000000000..0cd44659f4 --- /dev/null +++ b/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]); +}); diff --git a/src/process/capture/ProcessCaptureProgress.test.ts b/src/process/capture/ProcessCaptureProgress.test.ts new file mode 100644 index 0000000000..69349bef9e --- /dev/null +++ b/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((resolve) => { + release = resolve; + }); + const delivered: Parameters[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 | 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[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, + }); +}); diff --git a/src/process/capture/ProcessCaptureProgress.ts b/src/process/capture/ProcessCaptureProgress.ts index d67b1a3c44..83d5a51f5f 100644 --- a/src/process/capture/ProcessCaptureProgress.ts +++ b/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[0] | undefined; + let pending: Promise | undefined; + const flush = async (): Promise => { + 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 { 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; }, }; }; diff --git a/tests/boundary/process/processCaptureProgress.test.ts b/tests/boundary/process/processCaptureProgress.test.ts index cb190e07f9..87c0f7b9d8 100644 --- a/tests/boundary/process/processCaptureProgress.test.ts +++ b/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((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 | undefined; const appeared = await Promise.race([ running.then(() => true), new Promise((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);