feat(process): report live capture progress through CLI and MCP
Long captures emitted only a final result, so a quiet child looked idle. Pass the shared rate-limited progress reporter through the capture lifecycle and report the observed phase, elapsed time, and collected-unit counts. CLI writes those lines to stderr; MCP forwards them only when the client supplied a progress token. Final Evidence on stdout keeps its shape.
This commit is contained in:
@@ -94,6 +94,12 @@ if retention is unavailable or fails, the diagnostic reports that reason.
|
||||
REA accepts ordinary `tools/call` progress tokens. Updates are monotonic,
|
||||
rate-bounded to at most one intermediate update per 100 ms, and always allow a
|
||||
terminal update. Unknown totals are omitted; REA does not fabricate percentages.
|
||||
`capture_process_scenario` uses that boundary only when the client supplies a
|
||||
progress token. Notifications set `progress` to the count of collected terminal
|
||||
frames, process samples, and interaction events, omit `total`, and put the
|
||||
lifecycle phase, elapsed time, and final cleanup status in `message`. A client
|
||||
that does not send a progress token receives no progress notifications. The
|
||||
tool result is unchanged either way.
|
||||
Provider calls receive the request cancellation signal. Artifact traversal,
|
||||
hashing, version comparisons, Hopper requests, and process capture
|
||||
check the same signal. Cancellation is distinct from timeout. A cleanup failure
|
||||
|
||||
@@ -51,6 +51,32 @@ format is TOON. The file contains the complete capture Evidence record.
|
||||
Choose an output file distinct from the scenario input: shell redirection opens
|
||||
and truncates the output before REA reads the scenario.
|
||||
|
||||
While the capture runs, REA writes live status to stderr as one JSON object per
|
||||
line. Stdout stays the final Evidence document:
|
||||
|
||||
```json
|
||||
{
|
||||
"rea_progress": {
|
||||
"phase": "running",
|
||||
"completed": 3,
|
||||
"total": null,
|
||||
"message": "elapsed_ms=1200 frames=1 samples=2 interactions=0",
|
||||
"sequence": 4
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
`phase` is the observed lifecycle stage: `prepare`, `running`, `settling`, or
|
||||
`cleanup`. `completed` counts collected terminal frames, process samples, and
|
||||
interaction events. `total` stays null; REA does not invent a unit total or a
|
||||
percentage. `message` reports elapsed time and those counts. The final line uses
|
||||
`phase` `cleanup`, sets `terminal` to true, and adds `disposition` (`exited`,
|
||||
`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.
|
||||
|
||||
For example:
|
||||
|
||||
```json
|
||||
|
||||
@@ -15,6 +15,7 @@ import { processTraceSpecificationSchema } from "../../domain/process/processTra
|
||||
import { processScenarioSchema } from "../../domain/process/processScenario.js";
|
||||
import { compareProcessCaptures } from "../../domain/process/processComparison.js";
|
||||
import { parseProcessCapture } from "../../domain/process/processCaptureParsing.js";
|
||||
import { createProgressReporter } from "../ProgressReporter.js";
|
||||
import { captureProcessScenario } from "../../process/capture/ProcessHarness.js";
|
||||
import { PROCESS_PROVIDER } from "../../domain/process/processEvidenceProvider.js";
|
||||
import { createProcessCaptureEvidence } from "./ProcessEvidence.js";
|
||||
@@ -55,11 +56,20 @@ export const captureProcessScenarioFile = async (
|
||||
input,
|
||||
{ cause: parsed.error },
|
||||
);
|
||||
const progress = createProgressReporter(
|
||||
async (update) => {
|
||||
process.stderr.write(`${JSON.stringify({ rea_progress: update })}\n`);
|
||||
},
|
||||
{ minimumIntervalMs: 100 },
|
||||
);
|
||||
const captured = await captureProcessScenario(
|
||||
parsed.data,
|
||||
signal,
|
||||
process.platform,
|
||||
environment,
|
||||
undefined,
|
||||
undefined,
|
||||
progress,
|
||||
);
|
||||
if (!captured.ok) return cliAnalysisError(captured.error);
|
||||
return createProcessCaptureEvidence(parsed.data, captured.value);
|
||||
|
||||
@@ -75,6 +75,7 @@ interface TerminalExitOptions {
|
||||
readonly interactions: InteractionEvent[];
|
||||
readonly dispatchedEventIndexes: ReadonlySet<number>;
|
||||
readonly recordEvent: RecordProcessCaptureEvent;
|
||||
readonly onLiveProgress?: () => void;
|
||||
}
|
||||
|
||||
interface CaptureResultOptions {
|
||||
@@ -601,8 +602,10 @@ export const observeSettlement = async (
|
||||
recordEvent: RecordProcessCaptureEvent = () => undefined,
|
||||
platform: NodeJS.Platform = process.platform,
|
||||
signal?: AbortSignal,
|
||||
onLiveProgress?: () => void,
|
||||
): Promise<ObservedProcessSettlement> => {
|
||||
assertNotCancelled(signal);
|
||||
onLiveProgress?.();
|
||||
if (platform === "win32") {
|
||||
recordEvent("lifecycle", 1);
|
||||
return { state: "unverifiable", elapsed_ms: 0 };
|
||||
@@ -611,6 +614,7 @@ export const observeSettlement = async (
|
||||
let consecutiveEmpty = 0;
|
||||
let deadlineReached = false;
|
||||
while (!deadlineReached) {
|
||||
onLiveProgress?.();
|
||||
assertNotCancelled(signal);
|
||||
const observations = await Promise.all(
|
||||
[...new Set(processGroupIds)].map((processGroupId) =>
|
||||
@@ -651,6 +655,7 @@ export const awaitTerminalExit = async ({
|
||||
interactions,
|
||||
dispatchedEventIndexes,
|
||||
recordEvent,
|
||||
onLiveProgress,
|
||||
}: TerminalExitOptions): Promise<{
|
||||
exitCode: number;
|
||||
signal?: number;
|
||||
@@ -661,6 +666,7 @@ export const awaitTerminalExit = async ({
|
||||
// Keep the initiating lifecycle reason so comparisons distinguish a target
|
||||
// exit from harness-owned timeout, idle-timeout, and cancellation cleanup.
|
||||
let reason: "exited" | "timeout" | "idle_timeout" | "cancelled" = "exited";
|
||||
onLiveProgress?.();
|
||||
terminal.onExit((exit) => {
|
||||
recordEvent("lifecycle", 0);
|
||||
for (const [eventIndex, event] of scenario.events.entries()) {
|
||||
@@ -688,6 +694,7 @@ export const awaitTerminalExit = async ({
|
||||
resolveExit({ ...exit, reason });
|
||||
});
|
||||
const timeout = scheduleProcessInterval(() => {
|
||||
onLiveProgress?.();
|
||||
if (signal?.aborted === true) {
|
||||
reason = "cancelled";
|
||||
terminal.kill("SIGKILL");
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
import type { ProcessCaptureCleanupReport } from "../../domain/process/processCapture.js";
|
||||
|
||||
/**
|
||||
* Progress port accepted by process capture.
|
||||
* Structurally matches the shared application `ProgressReporter`.
|
||||
*/
|
||||
export interface ProcessCaptureProgress {
|
||||
report(update: {
|
||||
readonly phase: string;
|
||||
readonly completed: number;
|
||||
readonly total: null;
|
||||
readonly message: string;
|
||||
readonly terminal?: boolean;
|
||||
}): Promise<void>;
|
||||
}
|
||||
|
||||
/** Observed collection counts. These are not a percentage of an unknown total. */
|
||||
export interface ProcessCaptureProgressCounts {
|
||||
readonly frames: number;
|
||||
readonly samples: number;
|
||||
readonly interactions: number;
|
||||
}
|
||||
|
||||
export type ProcessCaptureProgressPhase =
|
||||
| "prepare"
|
||||
| "running"
|
||||
| "settling"
|
||||
| "cleanup";
|
||||
|
||||
export type ProcessCaptureProgressDisposition =
|
||||
| "exited"
|
||||
| "timeout"
|
||||
| "idle_timeout"
|
||||
| "cancelled"
|
||||
| "failed";
|
||||
|
||||
export interface ProcessCaptureProgressReport {
|
||||
readonly phase: ProcessCaptureProgressPhase;
|
||||
readonly counts: ProcessCaptureProgressCounts;
|
||||
readonly terminal?: boolean;
|
||||
readonly disposition?: ProcessCaptureProgressDisposition;
|
||||
readonly cleanup?: ProcessCaptureCleanupReport;
|
||||
}
|
||||
|
||||
const formatProgressMessage = (
|
||||
elapsedMs: number,
|
||||
counts: ProcessCaptureProgressCounts,
|
||||
disposition: ProcessCaptureProgressDisposition | undefined,
|
||||
cleanup: ProcessCaptureCleanupReport | undefined,
|
||||
): string => {
|
||||
const status = `elapsed_ms=${String(elapsedMs)} frames=${String(counts.frames)} samples=${String(counts.samples)} interactions=${String(counts.interactions)}`;
|
||||
const dispositionText =
|
||||
disposition === undefined ? "" : ` disposition=${disposition}`;
|
||||
const cleanupText =
|
||||
cleanup === undefined
|
||||
? ""
|
||||
: ` owned_process_group=${cleanup.owned_process_group.state} terminal_renderer=${cleanup.terminal_renderer.state} temporary_root=${cleanup.temporary_root.state}`;
|
||||
return `${status}${dispositionText}${cleanupText}`;
|
||||
};
|
||||
|
||||
/** Serialize observed capture status through one rate-limited reporter. */
|
||||
export const createProcessCaptureProgressTracker = (
|
||||
progress: ProcessCaptureProgress | undefined,
|
||||
now: () => number = Date.now,
|
||||
) => {
|
||||
const started = now();
|
||||
let completed = 0;
|
||||
let live = true;
|
||||
let pending = Promise.resolve();
|
||||
return {
|
||||
closeLive(): void {
|
||||
live = false;
|
||||
},
|
||||
report(update: ProcessCaptureProgressReport): Promise<void> {
|
||||
if (progress === undefined) return Promise.resolve();
|
||||
if (update.terminal !== true && !live) return Promise.resolve();
|
||||
const units =
|
||||
update.counts.frames +
|
||||
update.counts.samples +
|
||||
update.counts.interactions;
|
||||
if (units > completed) completed = units;
|
||||
const payload = {
|
||||
phase: update.phase,
|
||||
completed,
|
||||
total: null as null,
|
||||
message: formatProgressMessage(
|
||||
Math.max(0, now() - started),
|
||||
update.counts,
|
||||
update.disposition,
|
||||
update.cleanup,
|
||||
),
|
||||
...(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,
|
||||
);
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
export type ProcessCaptureProgressTracker = ReturnType<
|
||||
typeof createProcessCaptureProgressTracker
|
||||
>;
|
||||
@@ -55,6 +55,13 @@ import {
|
||||
import { makeProcessCaptureEnvironment } from "./ProcessCaptureEnvironment.js";
|
||||
import { classifyFilesystemEffects } from "./ProcessFilesystemEffects.js";
|
||||
import { processCaptureOwnershipUnavailableReason } from "./ProcessCaptureCapability.js";
|
||||
import {
|
||||
createProcessCaptureProgressTracker,
|
||||
type ProcessCaptureProgress,
|
||||
type ProcessCaptureProgressCounts,
|
||||
type ProcessCaptureProgressDisposition,
|
||||
type ProcessCaptureProgressTracker,
|
||||
} from "./ProcessCaptureProgress.js";
|
||||
import type { ProcessOwnershipBaseline } from "../ProcessOwnership.js";
|
||||
|
||||
interface StartedCaptureRuntime {
|
||||
@@ -262,7 +269,10 @@ const finishProcessRun = async (options: {
|
||||
readonly executionFailure: unknown;
|
||||
readonly observations?: ProcessCaptureObservationBuffer;
|
||||
readonly rootPid?: number;
|
||||
readonly progress: ProcessCaptureProgressTracker;
|
||||
readonly progressCounts: () => ProcessCaptureProgressCounts;
|
||||
}): Promise<ProcessCapture> => {
|
||||
options.progress.closeLive();
|
||||
await options.stopSampler();
|
||||
const sampledProcessGroupIds =
|
||||
options.runtime === undefined
|
||||
@@ -281,6 +291,20 @@ const finishProcessRun = async (options: {
|
||||
temporaryRoot: options.temporaryRoot,
|
||||
...(options.cleanupHost === undefined ? {} : { host: options.cleanupHost }),
|
||||
});
|
||||
const observedExit =
|
||||
options.observations?.exit.state === "available"
|
||||
? options.observations.exit.value.reason
|
||||
: options.capture?.exit.reason;
|
||||
await options.progress.report({
|
||||
phase: "cleanup",
|
||||
counts: options.progressCounts(),
|
||||
terminal: true,
|
||||
disposition: captureProgressDisposition(
|
||||
observedExit,
|
||||
options.executionFailure,
|
||||
),
|
||||
cleanup,
|
||||
});
|
||||
const cleanupFailure = cleanupReportFailure(cleanup);
|
||||
let { capture, executionFailure } = options;
|
||||
let verifiedCapture: ProcessCapture | undefined;
|
||||
@@ -336,6 +360,7 @@ const completeCapture = async (options: {
|
||||
readonly eventJournal: readonly ProcessCaptureEventJournalEntry[];
|
||||
readonly observationBuffer: ProcessCaptureObservationBuffer;
|
||||
readonly recordEvent: RecordProcessCaptureEvent;
|
||||
readonly onLiveProgress?: () => void;
|
||||
}): Promise<PendingProcessCapture> => {
|
||||
const unavailableReason = (stage: string, cause: unknown): string =>
|
||||
`${stage} failed: ${cause instanceof Error ? cause.message : "unknown failure"}`;
|
||||
@@ -351,6 +376,7 @@ const completeCapture = async (options: {
|
||||
options.recordEvent,
|
||||
options.hostPlatform,
|
||||
options.signal,
|
||||
options.onLiveProgress,
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
options.observationBuffer.settlement = {
|
||||
@@ -460,6 +486,25 @@ const completeCapture = async (options: {
|
||||
});
|
||||
};
|
||||
|
||||
const captureProgressDisposition = (
|
||||
exitReason: string | undefined,
|
||||
executionFailure: unknown,
|
||||
): ProcessCaptureProgressDisposition => {
|
||||
if (
|
||||
exitReason === "exited" ||
|
||||
exitReason === "timeout" ||
|
||||
exitReason === "idle_timeout" ||
|
||||
exitReason === "cancelled"
|
||||
)
|
||||
return exitReason;
|
||||
if (
|
||||
executionFailure instanceof ProcessCaptureError &&
|
||||
executionFailure.reason === "cancelled"
|
||||
)
|
||||
return "cancelled";
|
||||
return "failed";
|
||||
};
|
||||
|
||||
/** Execute one caller-selected scenario and return bounded observations. */
|
||||
const runProcessScenario = async (
|
||||
scenario: ProcessScenario,
|
||||
@@ -468,12 +513,44 @@ const runProcessScenario = async (
|
||||
hostPlatform: NodeJS.Platform = process.platform,
|
||||
captureSnapshot: typeof snapshotRoots = snapshotRoots,
|
||||
cleanupHost?: ProcessCaptureCleanupHost,
|
||||
progress?: ProcessCaptureProgress,
|
||||
): Promise<ProcessCapture> => {
|
||||
const { temporaryRoot, runId, ownershipBaseline, before } =
|
||||
await prepareProcessCapture(scenario, signal, captureSnapshot);
|
||||
const progressTracker = createProcessCaptureProgressTracker(progress);
|
||||
const frames: TerminalFrame[] = [];
|
||||
const samples: ProcessSample[] = [];
|
||||
const interactions: InteractionEvent[] = [];
|
||||
const progressCounts = (): ProcessCaptureProgressCounts => ({
|
||||
frames: frames.length,
|
||||
samples: samples.length,
|
||||
interactions: interactions.length,
|
||||
});
|
||||
const reportRunning = () => {
|
||||
void progressTracker.report({ phase: "running", counts: progressCounts() });
|
||||
};
|
||||
const reportSettling = () => {
|
||||
void progressTracker.report({
|
||||
phase: "settling",
|
||||
counts: progressCounts(),
|
||||
});
|
||||
};
|
||||
let prepared: Awaited<ReturnType<typeof prepareProcessCapture>>;
|
||||
try {
|
||||
await progressTracker.report({
|
||||
phase: "prepare",
|
||||
counts: progressCounts(),
|
||||
});
|
||||
prepared = await prepareProcessCapture(scenario, signal, captureSnapshot);
|
||||
} catch (cause: unknown) {
|
||||
progressTracker.closeLive();
|
||||
await progressTracker.report({
|
||||
phase: "cleanup",
|
||||
counts: progressCounts(),
|
||||
terminal: true,
|
||||
disposition: signal?.aborted === true ? "cancelled" : "failed",
|
||||
});
|
||||
throw cause;
|
||||
}
|
||||
const { temporaryRoot, runId, ownershipBaseline, before } = prepared;
|
||||
const journal = createProcessCaptureJournal();
|
||||
const { entries: eventJournal, recordEvent } = journal;
|
||||
recordEvent("filesystem_checkpoints", 0);
|
||||
@@ -523,6 +600,7 @@ const runProcessScenario = async (
|
||||
interactions,
|
||||
dispatchedEventIndexes,
|
||||
recordEvent,
|
||||
...(progress === undefined ? {} : { onLiveProgress: reportRunning }),
|
||||
});
|
||||
observations.exit = {
|
||||
state: "available",
|
||||
@@ -549,6 +627,7 @@ const runProcessScenario = async (
|
||||
recordEvent,
|
||||
captureSnapshot,
|
||||
...(signal === undefined ? {} : { signal }),
|
||||
...(progress === undefined ? {} : { onLiveProgress: reportSettling }),
|
||||
});
|
||||
} catch (cause: unknown) {
|
||||
executionFailure = normalizeCaptureFailure(cause, signal);
|
||||
@@ -590,6 +669,8 @@ const runProcessScenario = async (
|
||||
executionFailure,
|
||||
observations,
|
||||
...(actualRootPid === undefined ? {} : { rootPid: actualRootPid }),
|
||||
progress: progressTracker,
|
||||
progressCounts,
|
||||
});
|
||||
};
|
||||
|
||||
@@ -599,8 +680,9 @@ export const captureProcessScenario = async (
|
||||
signal?: AbortSignal,
|
||||
platform: NodeJS.Platform = process.platform,
|
||||
environment: Readonly<Record<string, string | undefined>> = process.env,
|
||||
captureSnapshot: typeof snapshotRoots = snapshotRoots,
|
||||
captureSnapshot?: typeof snapshotRoots,
|
||||
cleanupHost?: ProcessCaptureCleanupHost,
|
||||
progress?: ProcessCaptureProgress,
|
||||
): Promise<Result<ProcessCapture, ProcessCaptureError | AnalysisError>> => {
|
||||
const ownershipReason = processCaptureOwnershipUnavailableReason(platform);
|
||||
if (ownershipReason !== undefined)
|
||||
@@ -624,8 +706,9 @@ export const captureProcessScenario = async (
|
||||
signal,
|
||||
environment,
|
||||
platform,
|
||||
captureSnapshot,
|
||||
captureSnapshot ?? snapshotRoots,
|
||||
cleanupHost,
|
||||
progress,
|
||||
),
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
|
||||
@@ -90,24 +90,20 @@ const registerProcessTools = ({
|
||||
toolRegistrationOptions(captureContract),
|
||||
async (input, context) => {
|
||||
const progress = mcpProgressReporter(context);
|
||||
await progress.report({
|
||||
phase: captureContract.name,
|
||||
completed: 0,
|
||||
total: 1,
|
||||
message: "started",
|
||||
});
|
||||
const captured = await logToolExecution(
|
||||
logger,
|
||||
captureContract.name,
|
||||
() => captureProcessScenario(input, context.mcpReq.signal),
|
||||
() =>
|
||||
captureProcessScenario(
|
||||
input,
|
||||
context.mcpReq.signal,
|
||||
process.platform,
|
||||
process.env,
|
||||
undefined,
|
||||
undefined,
|
||||
progress,
|
||||
),
|
||||
);
|
||||
await progress.report({
|
||||
phase: captureContract.name,
|
||||
completed: 1,
|
||||
total: 1,
|
||||
message: captured.ok ? "completed" : "failed",
|
||||
terminal: true,
|
||||
});
|
||||
if (!captured.ok)
|
||||
return server.delivery.toCallToolResult(captured, captureContract);
|
||||
const evidence = createProcessCaptureEvidence(input, captured.value);
|
||||
|
||||
@@ -0,0 +1,538 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { Client, InMemoryTransport } from "@modelcontextprotocol/client";
|
||||
import { expect, onTestFinished } from "vitest";
|
||||
import { z } from "zod";
|
||||
|
||||
import { analysisCliErrorEnvelopeSchema } from "../../../src/contracts/errorSchemas.js";
|
||||
import { toolContract } from "../../../src/contracts/toolContracts.js";
|
||||
import { parseEvidence } from "../../../src/domain/evidence.js";
|
||||
import { parseProcessCapture } from "../../../src/domain/process/processCaptureParsing.js";
|
||||
import { silentLogger } from "../../../src/logger.js";
|
||||
import { createServer } from "../../../src/server/createServer.js";
|
||||
import { createTestBinarySession } from "../../fixtures/binarySession.js";
|
||||
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
import { itWithCaptureCapability } from "./processCaptureCapability.js";
|
||||
|
||||
const PROGRESS_INTERVAL_MS = 100;
|
||||
|
||||
const progressLineSchema = z.object({
|
||||
rea_progress: z.object({
|
||||
phase: z.string(),
|
||||
completed: z.number().nonnegative(),
|
||||
total: z.null(),
|
||||
message: z.string(),
|
||||
sequence: z.number().int().positive(),
|
||||
terminal: z.boolean().optional(),
|
||||
}),
|
||||
});
|
||||
|
||||
const progressMessageSchema = z.object({
|
||||
elapsed_ms: z.number().int().nonnegative(),
|
||||
frames: z.number().int().nonnegative(),
|
||||
samples: z.number().int().nonnegative(),
|
||||
interactions: z.number().int().nonnegative(),
|
||||
disposition: z
|
||||
.enum(["exited", "timeout", "idle_timeout", "cancelled", "failed"])
|
||||
.optional(),
|
||||
owned_process_group: z
|
||||
.enum(["cleaned", "failed", "unverified", "not_required"])
|
||||
.optional(),
|
||||
terminal_renderer: z
|
||||
.enum(["cleaned", "failed", "unverified", "not_required"])
|
||||
.optional(),
|
||||
temporary_root: z
|
||||
.enum(["cleaned", "failed", "unverified", "not_required"])
|
||||
.optional(),
|
||||
});
|
||||
|
||||
type ProgressLine = z.infer<typeof progressLineSchema>["rea_progress"];
|
||||
type ParsedProgress = ProgressLine & {
|
||||
readonly details: z.infer<typeof progressMessageSchema>;
|
||||
};
|
||||
|
||||
const parseProgressMessage = (message: string) => {
|
||||
const matched =
|
||||
/^elapsed_ms=(?<elapsed_ms>\d+) frames=(?<frames>\d+) samples=(?<samples>\d+) interactions=(?<interactions>\d+)(?: disposition=(?<disposition>exited|timeout|idle_timeout|cancelled|failed))?(?: owned_process_group=(?<owned_process_group>cleaned|failed|unverified|not_required) terminal_renderer=(?<terminal_renderer>cleaned|failed|unverified|not_required) temporary_root=(?<temporary_root>cleaned|failed|unverified|not_required))?$/u.exec(
|
||||
message,
|
||||
);
|
||||
if (matched?.groups === undefined)
|
||||
throw new Error(`Progress message is not capture status: ${message}`);
|
||||
return progressMessageSchema.parse({
|
||||
elapsed_ms: Number(matched.groups.elapsed_ms),
|
||||
frames: Number(matched.groups.frames),
|
||||
samples: Number(matched.groups.samples),
|
||||
interactions: Number(matched.groups.interactions),
|
||||
...(matched.groups.disposition === undefined
|
||||
? {}
|
||||
: { disposition: matched.groups.disposition }),
|
||||
...(matched.groups.owned_process_group === undefined
|
||||
? {}
|
||||
: {
|
||||
owned_process_group: matched.groups.owned_process_group,
|
||||
terminal_renderer: matched.groups.terminal_renderer,
|
||||
temporary_root: matched.groups.temporary_root,
|
||||
}),
|
||||
});
|
||||
};
|
||||
|
||||
interface CliCapture {
|
||||
readonly stdout: string;
|
||||
readonly stderr: string;
|
||||
readonly exitCode: number | null;
|
||||
readonly lines: readonly {
|
||||
readonly update: ParsedProgress;
|
||||
readonly at: number;
|
||||
}[];
|
||||
}
|
||||
|
||||
const writeScenario = async (
|
||||
prefix: string,
|
||||
scenario: Readonly<Record<string, unknown>>,
|
||||
): Promise<string> => {
|
||||
const root = await createTestTempDirectory(prefix);
|
||||
const path = join(root, "scenario.json");
|
||||
await writeFile(path, JSON.stringify(scenario));
|
||||
return path;
|
||||
};
|
||||
|
||||
const runCaptureCli = async (
|
||||
scenarioPath: string,
|
||||
options: { readonly stopOnRunning?: boolean } = {},
|
||||
): Promise<CliCapture> => {
|
||||
const child = spawn(
|
||||
process.execPath,
|
||||
["scripts/rea.mjs", "capture-process", scenarioPath, "--json"],
|
||||
{ stdio: ["ignore", "pipe", "pipe"] },
|
||||
);
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
let pending = "";
|
||||
const lines: Array<{ readonly line: string; readonly at: number }> = [];
|
||||
let sawRunning: (() => void) | undefined;
|
||||
const running = options.stopOnRunning
|
||||
? new Promise<void>((resolve) => {
|
||||
sawRunning = resolve;
|
||||
})
|
||||
: undefined;
|
||||
child.stdout.on("data", (chunk: Buffer) => {
|
||||
stdout += chunk.toString();
|
||||
});
|
||||
child.stderr.on("data", (chunk: Buffer) => {
|
||||
pending += chunk.toString();
|
||||
const parts = pending.split("\n");
|
||||
pending = parts.pop() ?? "";
|
||||
for (const line of parts) {
|
||||
if (line.length === 0) continue;
|
||||
lines.push({ line, at: Date.now() });
|
||||
if (line.includes('"phase":"running"')) sawRunning?.();
|
||||
}
|
||||
});
|
||||
const closed = new Promise<{ code: number | null }>((resolve, reject) => {
|
||||
child.once("error", reject);
|
||||
child.once("close", (code) => resolve({ code }));
|
||||
});
|
||||
if (running !== undefined) {
|
||||
const appeared = await Promise.race([
|
||||
running.then(() => true),
|
||||
new Promise<boolean>((resolve) => {
|
||||
setTimeout(() => resolve(false), 8_000);
|
||||
}),
|
||||
]);
|
||||
if (!appeared) {
|
||||
child.kill("SIGKILL");
|
||||
await closed;
|
||||
throw new Error(
|
||||
`capture produced no running progress on stderr:\n${lines.map(({ line }) => line).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");
|
||||
return {
|
||||
stdout,
|
||||
stderr,
|
||||
exitCode: result.code,
|
||||
lines: lines.map(({ line, at }) => {
|
||||
const update = progressLineSchema.parse(JSON.parse(line)).rea_progress;
|
||||
return {
|
||||
at,
|
||||
update: { ...update, details: parseProgressMessage(update.message) },
|
||||
};
|
||||
}),
|
||||
};
|
||||
};
|
||||
|
||||
const assertMonotonic = (updates: readonly ParsedProgress[]): void => {
|
||||
for (const [index, update] of updates.entries()) {
|
||||
expect(update.total).toBeNull();
|
||||
expect(update.message).not.toMatch(/%|\bpercent/iu);
|
||||
const previous = updates[index - 1];
|
||||
if (previous === undefined) continue;
|
||||
expect(update.completed).toBeGreaterThanOrEqual(previous.completed);
|
||||
expect(update.sequence).toBe(previous.sequence + 1);
|
||||
}
|
||||
};
|
||||
|
||||
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 childScript = (source: string) => ["-e", source];
|
||||
|
||||
const connectProgressClient = async (
|
||||
recordNotifications = false,
|
||||
): Promise<{
|
||||
readonly client: Client;
|
||||
readonly notifications: string[];
|
||||
}> => {
|
||||
const session = createTestBinarySession(() => {
|
||||
throw new Error("Process capture must not launch a binary provider");
|
||||
});
|
||||
const server = createServer(
|
||||
{ kind: "session", session },
|
||||
{ logger: silentLogger },
|
||||
);
|
||||
const client = new Client({
|
||||
name: "process-capture-progress",
|
||||
version: "1",
|
||||
});
|
||||
const notifications: string[] = [];
|
||||
if (recordNotifications) {
|
||||
client.setNotificationHandler("notifications/progress", (notification) => {
|
||||
notifications.push(JSON.stringify(notification));
|
||||
});
|
||||
}
|
||||
const [clientTransport, serverTransport] =
|
||||
InMemoryTransport.createLinkedPair();
|
||||
await server.connect(serverTransport);
|
||||
await client.connect(clientTransport);
|
||||
onTestFinished(async () => {
|
||||
await Promise.allSettled([client.close(), server.close(), session.close()]);
|
||||
});
|
||||
return { client, notifications };
|
||||
};
|
||||
|
||||
itWithCaptureCapability(
|
||||
"keeps capture Evidence on stdout and live status on stderr for a sparse child",
|
||||
async () => {
|
||||
const scenario = await writeScenario("rea-process-progress-sparse-", {
|
||||
executable: process.execPath,
|
||||
arguments: childScript(
|
||||
"process.stdout.write('SPARSE_CHILD_STARTED\\n'); setTimeout(() => process.stdout.write('SPARSE_CHILD_DONE\\n'), 1600);",
|
||||
),
|
||||
timeout_ms: 10_000,
|
||||
idle_timeout_ms: 10_000,
|
||||
});
|
||||
const capture = await runCaptureCli(scenario);
|
||||
expect(capture.exitCode, capture.stderr).toBe(0);
|
||||
expect(capture.stdout).not.toContain("rea_progress");
|
||||
const evidence = parseEvidence(JSON.parse(capture.stdout));
|
||||
const recorded = parseProcessCapture(evidence.normalized_result);
|
||||
expect(evidence).toMatchObject({
|
||||
operation: "capture_process_scenario",
|
||||
predicate_type: "rea.process-capture",
|
||||
});
|
||||
expect(recorded.frames.map(({ data }) => data).join("")).toContain(
|
||||
"SPARSE_CHILD_STARTED",
|
||||
);
|
||||
expect(capture.stderr).not.toContain("SPARSE_CHILD_STARTED");
|
||||
expect(capture.stderr).not.toContain("SPARSE_CHILD_DONE");
|
||||
const updates = capture.lines.map(({ update }) => update);
|
||||
assertMonotonic(updates);
|
||||
const running = updates.filter(({ phase }) => phase === "running");
|
||||
expect(running.length).toBeGreaterThanOrEqual(3);
|
||||
const elapsed = running.map(({ details }) => details.elapsed_ms);
|
||||
expect(Math.max(...elapsed) - Math.min(...elapsed)).toBeGreaterThanOrEqual(
|
||||
1_000,
|
||||
);
|
||||
expect(running.every(({ details }) => details.frames <= 2)).toBe(true);
|
||||
expect(updates.at(-1)).toMatchObject({
|
||||
phase: "cleanup",
|
||||
terminal: true,
|
||||
details: {
|
||||
disposition: "exited",
|
||||
owned_process_group: "cleaned",
|
||||
terminal_renderer: "cleaned",
|
||||
temporary_root: "cleaned",
|
||||
},
|
||||
});
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
itWithCaptureCapability(
|
||||
"bounds stderr progress while a busy child emits many output chunks",
|
||||
async () => {
|
||||
const scenario = await writeScenario("rea-process-progress-busy-", {
|
||||
executable: process.execPath,
|
||||
arguments: childScript(
|
||||
"let count = 0; const timer = setInterval(() => { process.stdout.write('BUSY_CHILD_LINE\\n'); count += 1; if (count >= 50) clearInterval(timer); }, 12);",
|
||||
),
|
||||
timeout_ms: 10_000,
|
||||
idle_timeout_ms: 10_000,
|
||||
});
|
||||
const capture = await runCaptureCli(scenario);
|
||||
expect(capture.exitCode, capture.stderr).toBe(0);
|
||||
const evidence = parseEvidence(JSON.parse(capture.stdout));
|
||||
const recorded = parseProcessCapture(evidence.normalized_result);
|
||||
const occurrences =
|
||||
recorded.frames
|
||||
.map(({ data }) => data)
|
||||
.join("")
|
||||
.split("BUSY_CHILD_LINE").length - 1;
|
||||
expect(occurrences).toBeGreaterThanOrEqual(40);
|
||||
expect(capture.stderr).not.toContain("BUSY_CHILD_LINE");
|
||||
const updates = capture.lines.map(({ update }) => update);
|
||||
assertMonotonic(updates);
|
||||
expect(updates.length).toBeLessThan(occurrences / 2);
|
||||
const nonTerminal = capture.lines.filter(
|
||||
({ update }) => update.terminal !== true,
|
||||
);
|
||||
assertRateBound(nonTerminal.map(({ at }) => at));
|
||||
expect(
|
||||
Math.max(...updates.map(({ details }) => details.frames)),
|
||||
).toBeGreaterThanOrEqual(10);
|
||||
expect(updates.at(-1)?.details.owned_process_group).toBe("cleaned");
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
itWithCaptureCapability(
|
||||
"reports timeout cleanup on stderr without changing Evidence shape",
|
||||
async () => {
|
||||
const scenario = await writeScenario("rea-process-progress-timeout-", {
|
||||
executable: process.execPath,
|
||||
arguments: childScript("setInterval(() => undefined, 1000);"),
|
||||
timeout_ms: 700,
|
||||
idle_timeout_ms: 10_000,
|
||||
});
|
||||
const capture = await runCaptureCli(scenario);
|
||||
expect(capture.exitCode, capture.stderr).toBe(0);
|
||||
const evidence = parseEvidence(JSON.parse(capture.stdout));
|
||||
const recorded = parseProcessCapture(evidence.normalized_result);
|
||||
expect(recorded.exit.reason).toBe("timeout");
|
||||
expect(capture.stdout).not.toContain("rea_progress");
|
||||
const updates = capture.lines.map(({ update }) => update);
|
||||
assertMonotonic(updates);
|
||||
expect(updates.at(-1)).toMatchObject({
|
||||
phase: "cleanup",
|
||||
terminal: true,
|
||||
details: {
|
||||
disposition: "timeout",
|
||||
owned_process_group: "cleaned",
|
||||
},
|
||||
});
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
itWithCaptureCapability(
|
||||
"reports cleanup status on stderr when capture is cancelled",
|
||||
async () => {
|
||||
const scenario = await writeScenario("rea-process-progress-cancel-", {
|
||||
executable: process.execPath,
|
||||
arguments: childScript("setInterval(() => undefined, 1000);"),
|
||||
timeout_ms: 15_000,
|
||||
idle_timeout_ms: 15_000,
|
||||
});
|
||||
const capture = await runCaptureCli(scenario, { stopOnRunning: true });
|
||||
expect(capture.exitCode).toBe(130);
|
||||
expect(
|
||||
analysisCliErrorEnvelopeSchema.parse(JSON.parse(capture.stdout)),
|
||||
).toMatchObject({
|
||||
code: "cancelled",
|
||||
details: { operation: "process_capture", cleanup: "complete" },
|
||||
});
|
||||
expect(capture.stdout).not.toContain("rea_progress");
|
||||
const updates = capture.lines.map(({ update }) => update);
|
||||
assertMonotonic(updates);
|
||||
expect(updates.at(-1)).toMatchObject({
|
||||
phase: "cleanup",
|
||||
terminal: true,
|
||||
details: {
|
||||
disposition: "cancelled",
|
||||
owned_process_group: "cleaned",
|
||||
},
|
||||
});
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
itWithCaptureCapability(
|
||||
"forwards live capture progress only when the MCP client supplies a token",
|
||||
async () => {
|
||||
const silentClient = await connectProgressClient(true);
|
||||
const { client: tokenClient } = await connectProgressClient();
|
||||
const { notifications } = silentClient;
|
||||
const silent = await silentClient.client.callTool({
|
||||
name: "capture_process_scenario",
|
||||
arguments: {
|
||||
executable: process.execPath,
|
||||
arguments: childScript("process.stdout.write('SILENT_CHILD_OK\\n');"),
|
||||
timeout_ms: 10_000,
|
||||
idle_timeout_ms: 10_000,
|
||||
},
|
||||
});
|
||||
expect(silent.isError, JSON.stringify(silent)).not.toBe(true);
|
||||
expect(notifications).toEqual([]);
|
||||
const evidence = toolContract(
|
||||
"capture_process_scenario",
|
||||
).outputSchema.parse(silent.structuredContent);
|
||||
expect(
|
||||
evidence.normalized_result.frames.map((frame) => frame.data).join(""),
|
||||
).toContain("SILENT_CHILD_OK");
|
||||
|
||||
const progress: Array<{
|
||||
readonly progress: number;
|
||||
readonly total: number | undefined;
|
||||
readonly message: string | undefined;
|
||||
readonly at: number;
|
||||
}> = [];
|
||||
const captured = await tokenClient.callTool(
|
||||
{
|
||||
name: "capture_process_scenario",
|
||||
arguments: {
|
||||
executable: process.execPath,
|
||||
arguments: childScript(
|
||||
"process.stdout.write('TOKEN_CHILD_STARTED\\n'); setTimeout(() => process.stdout.write('TOKEN_CHILD_DONE\\n'), 1200);",
|
||||
),
|
||||
timeout_ms: 10_000,
|
||||
idle_timeout_ms: 10_000,
|
||||
},
|
||||
},
|
||||
{
|
||||
onprogress: (update) => {
|
||||
progress.push({
|
||||
progress: update.progress,
|
||||
total: update.total,
|
||||
message: update.message,
|
||||
at: Date.now(),
|
||||
});
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(captured.isError, JSON.stringify(captured)).not.toBe(true);
|
||||
expect(progress.length).toBeGreaterThanOrEqual(3);
|
||||
expect(progress.every(({ total }) => total === undefined)).toBe(true);
|
||||
for (const [index, update] of progress.entries()) {
|
||||
const previous = progress[index - 1];
|
||||
if (previous !== undefined)
|
||||
expect(update.progress).toBeGreaterThanOrEqual(previous.progress);
|
||||
expect(update.message ?? "").not.toContain("TOKEN_CHILD_STARTED");
|
||||
expect(update.message ?? "").not.toContain("TOKEN_CHILD_DONE");
|
||||
}
|
||||
const messages = progress.map(({ message }) => message ?? "");
|
||||
expect(messages.some((message) => message.startsWith("running:"))).toBe(
|
||||
true,
|
||||
);
|
||||
expect(messages.at(-1)).toContain("cleanup:");
|
||||
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));
|
||||
const tokenEvidence = toolContract(
|
||||
"capture_process_scenario",
|
||||
).outputSchema.parse(captured.structuredContent);
|
||||
expect(
|
||||
tokenEvidence.normalized_result.frames
|
||||
.map((frame) => frame.data)
|
||||
.join(""),
|
||||
).toContain("TOKEN_CHILD_STARTED");
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
|
||||
itWithCaptureCapability(
|
||||
"reports timeout and cancellation cleanup through MCP progress",
|
||||
async () => {
|
||||
const { client } = await connectProgressClient();
|
||||
const timeoutProgress: Array<{
|
||||
readonly message: string | undefined;
|
||||
readonly total: number | undefined;
|
||||
}> = [];
|
||||
const timedOut = await client.callTool(
|
||||
{
|
||||
name: "capture_process_scenario",
|
||||
arguments: {
|
||||
executable: process.execPath,
|
||||
arguments: childScript("setInterval(() => undefined, 1000);"),
|
||||
timeout_ms: 600,
|
||||
idle_timeout_ms: 10_000,
|
||||
},
|
||||
},
|
||||
{
|
||||
onprogress: (update) => {
|
||||
timeoutProgress.push({
|
||||
message: update.message,
|
||||
total: update.total,
|
||||
});
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(timedOut.isError, JSON.stringify(timedOut)).not.toBe(true);
|
||||
const timedOutEvidence = toolContract(
|
||||
"capture_process_scenario",
|
||||
).outputSchema.parse(timedOut.structuredContent);
|
||||
expect(timedOutEvidence.normalized_result.exit.reason).toBe("timeout");
|
||||
expect(timeoutProgress.every(({ total }) => total === undefined)).toBe(
|
||||
true,
|
||||
);
|
||||
expect(timeoutProgress.at(-1)?.message).toContain("disposition=timeout");
|
||||
expect(timeoutProgress.at(-1)?.message).toContain(
|
||||
"owned_process_group=cleaned",
|
||||
);
|
||||
|
||||
const cancelling = await connectProgressClient(true);
|
||||
const controller = new AbortController();
|
||||
const stall = setTimeout(() => controller.abort(), 5_000);
|
||||
onTestFinished(() => clearTimeout(stall));
|
||||
const call = cancelling.client.callTool(
|
||||
{
|
||||
name: "capture_process_scenario",
|
||||
arguments: {
|
||||
executable: process.execPath,
|
||||
arguments: childScript("setInterval(() => undefined, 1000);"),
|
||||
timeout_ms: 15_000,
|
||||
idle_timeout_ms: 15_000,
|
||||
},
|
||||
_meta: { progressToken: "capture-cancel" },
|
||||
},
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
const sawRunning = await Promise.race([
|
||||
expect
|
||||
.poll(() =>
|
||||
cancelling.notifications.some((note) => note.includes("running:")),
|
||||
)
|
||||
.toBe(true)
|
||||
.then(() => true),
|
||||
new Promise<boolean>((resolve) => {
|
||||
setTimeout(() => resolve(false), 4_000);
|
||||
}),
|
||||
]);
|
||||
clearTimeout(stall);
|
||||
if (!sawRunning)
|
||||
throw new Error(
|
||||
`MCP cancellation saw no running progress: ${cancelling.notifications.join("\n")}`,
|
||||
);
|
||||
controller.abort();
|
||||
await expect(call).rejects.toThrow(/abort/iu);
|
||||
await expect
|
||||
.poll(() => cancelling.notifications.join("\n"))
|
||||
.toContain("disposition=cancelled");
|
||||
expect(cancelling.notifications.join("\n")).toContain(
|
||||
"owned_process_group=cleaned",
|
||||
);
|
||||
expect(cancelling.notifications.join("\n")).not.toMatch(
|
||||
/setInterval|CHILD/u,
|
||||
);
|
||||
},
|
||||
30_000,
|
||||
);
|
||||
Reference in New Issue
Block a user