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);