From f6fe575094658e99ced78d45f385633fbfaaae32 Mon Sep 17 00:00:00 2001 From: Ysopwnable <161784452+N0zoM1z0@users.noreply.github.com> Date: Fri, 9 Oct 2026 19:26:41 +0800 Subject: [PATCH] fix(process): await cleanup on CLI and MCP cancellation Propagate CLI interrupts to capture and interrupt post-exit settlement without cancelling owned-resource cleanup or discarding prior observations. Assisted-by: Kogasa (gensokyo-skills:kogasa-surprise-testing) --- src/application/process/ProcessCli.ts | 3 +- src/cli/processCommands.ts | 13 +- .../ProcessOwnership.validation.test.ts | 35 +++ src/process/ProcessOwnershipObservation.ts | 18 +- .../capture/ProcessCaptureLifecycle.ts | 17 +- src/process/capture/ProcessHarness.ts | 1 + ...processCaptureCancellationAdapters.test.ts | 213 ++++++++++++++++++ tests/fixtures/processCaptureCancellation.mjs | 45 ++++ 8 files changed, 327 insertions(+), 18 deletions(-) create mode 100644 tests/boundary/process/processCaptureCancellationAdapters.test.ts create mode 100644 tests/fixtures/processCaptureCancellation.mjs diff --git a/src/application/process/ProcessCli.ts b/src/application/process/ProcessCli.ts index 560866a59..45b16b953 100644 --- a/src/application/process/ProcessCli.ts +++ b/src/application/process/ProcessCli.ts @@ -47,6 +47,7 @@ export const isProcessCliFailure = ( export const captureProcessScenarioFile = async ( path: string, environment: Readonly> = process.env, + signal?: AbortSignal, ) => { try { const input = await readJson(path); @@ -60,7 +61,7 @@ export const captureProcessScenarioFile = async ( ); const captured = await captureProcessScenario( parsed.data, - undefined, + signal, process.platform, environment, ); diff --git a/src/cli/processCommands.ts b/src/cli/processCommands.ts index af10e95c8..521d46314 100644 --- a/src/cli/processCommands.ts +++ b/src/cli/processCommands.ts @@ -8,6 +8,7 @@ import { import { logCliCommand } from "../cliLogging.js"; import type { Logger } from "../logger.js"; import { CLI_COMMANDS } from "../cliCommandNames.js"; +import { withCommandCancellation } from "./commandCancellation.js"; /** Register direct process capture and comparison commands. */ export const registerProcessCommands = ( @@ -19,11 +20,13 @@ export const registerProcessCommands = ( description: "Capture one caller-selected process scenario", args: z.object({ scenario: z.string().describe("Scenario JSON path") }), run: ({ args }) => - logCliCommand( - logger, - "capture-process", - () => captureProcessScenarioFile(args.scenario, environment), - isProcessCliFailure, + withCommandCancellation((signal) => + logCliCommand( + logger, + "capture-process", + () => captureProcessScenarioFile(args.scenario, environment, signal), + isProcessCliFailure, + ), ), }); cli.command(CLI_COMMANDS.compareProcessCaptures, { diff --git a/src/process/ProcessOwnership.validation.test.ts b/src/process/ProcessOwnership.validation.test.ts index b9c6a789e..763c6182d 100644 --- a/src/process/ProcessOwnership.validation.test.ts +++ b/src/process/ProcessOwnership.validation.test.ts @@ -145,6 +145,41 @@ describe("owned process-group cleanup validation: ownership and lineage", () => }); describe("owned process-group cleanup validation: exited members", () => { + it("preserves cancellation during process-table observation instead of reporting unverifiable ownership", async () => { + const controller = new AbortController(); + const reason = new Error("cancelled process observation"); + const adapter: ProcessOwnershipHost = { + listProcesses: async (signal) => { + controller.abort(reason); + signal?.throwIfAborted(); + return []; + }, + environment: () => Promise.resolve({}), + signalGroup: () => { + throw new Error("Observation must never signal processes"); + }, + }; + await expect( + observeOwnedProcessGroup(ownership, adapter, controller.signal), + ).rejects.toBe(reason); + }); + it("does not inspect a process group after observation was cancelled", async () => { + const controller = new AbortController(); + const reason = new Error("already cancelled observation"); + controller.abort(reason); + const adapter: ProcessOwnershipHost = { + listProcesses: () => { + throw new Error("Cancelled observation must not inspect processes"); + }, + environment: () => Promise.resolve({}), + signalGroup: () => { + throw new Error("Observation must never signal processes"); + }, + }; + await expect( + observeOwnedProcessGroup(ownership, adapter, controller.signal), + ).rejects.toBe(reason); + }); it("ignores exited zombie members during live ownership checks", async () => { const environment = vi.fn((pid: number) => Promise.resolve(pid === 101 ? {} : { REA_PROCESS_RUN_ID: "run-token" }), diff --git a/src/process/ProcessOwnershipObservation.ts b/src/process/ProcessOwnershipObservation.ts index 6dc460d97..661363bfe 100644 --- a/src/process/ProcessOwnershipObservation.ts +++ b/src/process/ProcessOwnershipObservation.ts @@ -19,31 +19,36 @@ import type { export const observeOwnedProcessGroupWithHost = async ( ownership: OwnedProcessGroup, host: ProcessOwnershipHost, + signal?: AbortSignal, ): Promise => { + signal?.throwIfAborted(); let members: readonly ProcessTableEntry[]; try { - members = (await host.listProcesses()).filter( + members = (await host.listProcesses(signal)).filter( ({ processGroupId }) => processGroupId === ownership.processGroupId, ); } catch (cause: unknown) { + signal?.throwIfAborted(); return { state: "unverifiable", reason: `process group could not be inspected: ${errorMessage(cause)}`, }; } + signal?.throwIfAborted(); const liveMembers = liveProcesses(members); if (liveMembers.length === 0) return { state: "empty" }; for (const member of liveMembers) { + signal?.throwIfAborted(); try { - if ( - (await host.environment(member.pid)).REA_PROCESS_RUN_ID !== - ownership.runId - ) + const runId = (await host.environment(member.pid)).REA_PROCESS_RUN_ID; + signal?.throwIfAborted(); + if (runId !== ownership.runId) return { state: "unverifiable", reason: "process ownership did not match", }; } catch (cause: unknown) { + signal?.throwIfAborted(); if (!(await processIsGone(host, member.pid))) { return { state: "unverifiable", @@ -438,8 +443,9 @@ export const prepareProcessOwnershipInspection = async ( export const observeOwnedProcessGroup = async ( ownership: OwnedProcessGroup, host: ProcessOwnershipHost = systemProcessOwnershipHost, + signal?: AbortSignal, ): Promise => - observeOwnedProcessGroupWithHost(ownership, host); + observeOwnedProcessGroupWithHost(ownership, host, signal); /** * Record the live launcher and descendant lineage after run-token validation. diff --git a/src/process/capture/ProcessCaptureLifecycle.ts b/src/process/capture/ProcessCaptureLifecycle.ts index 0e413beb3..d417e9a2f 100644 --- a/src/process/capture/ProcessCaptureLifecycle.ts +++ b/src/process/capture/ProcessCaptureLifecycle.ts @@ -3,6 +3,7 @@ import { constants, statSync, type BigIntStats } from "node:fs"; import { open, mkdtemp, rm, stat } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; import type { IPty } from "@lydell/node-pty"; import type { @@ -530,7 +531,9 @@ export const observeSettlement = async ( settleMs: number, recordEvent: RecordProcessCaptureEvent = () => undefined, platform: NodeJS.Platform = process.platform, + signal?: AbortSignal, ): Promise => { + assertNotCancelled(signal); if (platform === "win32") { recordEvent("lifecycle", 1); return { state: "unverifiable", elapsed_ms: 0 }; @@ -539,15 +542,17 @@ export const observeSettlement = async ( let consecutiveEmpty = 0; let deadlineReached = false; while (!deadlineReached) { + assertNotCancelled(signal); const observations = await Promise.all( [...new Set(processGroupIds)].map((processGroupId) => - observeOwnedProcessGroup({ - runId, - leaderPid: processGroupId, - processGroupId, - }), + observeOwnedProcessGroup( + { runId, leaderPid: processGroupId, processGroupId }, + undefined, + signal, + ), ), ); + assertNotCancelled(signal); if (observations.some(({ state }) => state === "unverifiable")) { recordEvent("lifecycle", 1); return { state: "unverifiable", elapsed_ms: Date.now() - started }; @@ -561,7 +566,7 @@ export const observeSettlement = async ( } else consecutiveEmpty = 0; deadlineReached = Date.now() - started >= settleMs; if (deadlineReached) break; - await new Promise((resolve) => setTimeout(resolve, 50)); + await delay(50, undefined, { signal }); } recordEvent("lifecycle", 1); return { state: "alive_at_deadline", elapsed_ms: Date.now() - started }; diff --git a/src/process/capture/ProcessHarness.ts b/src/process/capture/ProcessHarness.ts index 3d7af9c80..592202fd3 100644 --- a/src/process/capture/ProcessHarness.ts +++ b/src/process/capture/ProcessHarness.ts @@ -370,6 +370,7 @@ const completeCapture = async (options: { scenario.settle_ms, options.recordEvent, options.hostPlatform, + options.signal, ); } catch (cause: unknown) { options.observationBuffer.settlement = { diff --git a/tests/boundary/process/processCaptureCancellationAdapters.test.ts b/tests/boundary/process/processCaptureCancellationAdapters.test.ts new file mode 100644 index 000000000..c3545576f --- /dev/null +++ b/tests/boundary/process/processCaptureCancellationAdapters.test.ts @@ -0,0 +1,213 @@ +import { execFile, spawn } from "node:child_process"; +import { access, readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { Client, InMemoryTransport } from "@modelcontextprotocol/client"; +import pino from "pino"; +import { expect, onTestFinished } from "vitest"; +import { z } from "zod"; +import { cleanupOwnedProcessGroup } from "../../../src/process/ProcessOwnership.js"; +import { createServer } from "../../../src/server/createServer.js"; +import { + createDeferred, + createTestBinarySession, +} from "../../fixtures/binarySession.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; +import { itWithCaptureCapability } from "./processCaptureCapability.js"; + +const fixture = fileURLToPath( + new URL("../../fixtures/processCaptureCancellation.mjs", import.meta.url), +); +const exec = promisify(execFile); +const markerSchema = z.object({ + rootPid: z.number().int().positive(), + childPid: z.number().int().positive().optional(), + runId: z.string().min(1), +}); +type Marker = z.infer; + +const readMarker = async (root: string): Promise => { + const deadline = Date.now() + 10_000; + while (Date.now() < deadline) { + try { + return markerSchema.parse( + JSON.parse(await readFile(join(root, "ready.json"), "utf8")), + ); + } catch (cause: unknown) { + if ( + !(cause instanceof Error && "code" in cause && cause.code === "ENOENT") + ) + throw cause; + } + await delay(25); + } + throw new Error("Owned capture fixture did not become ready"); +}; + +const isRunning = async (pid: number): Promise => { + try { + const { stdout } = await exec("ps", ["-o", "stat=", "-p", String(pid)]); + return stdout.trim().length > 0 && !stdout.trim().startsWith("Z"); + } catch (cause: unknown) { + if (cause instanceof Error && "code" in cause && cause.code === 1) + return false; + throw cause; + } +}; + +const waitForExit = async (pid: number): Promise => { + await expect + .poll(() => isRunning(pid), { timeout: 3_000, interval: 25 }) + .toBe(false); +}; + +const cleanupFixture = async (marker: Marker | undefined): Promise => { + if (marker === undefined) return; + // Teardown uses the fixture's capture run token, never a bare PID kill. + await cleanupOwnedProcessGroup({ + runId: marker.runId, + leaderPid: marker.rootPid, + processGroupId: marker.rootPid, + }); +}; + +itWithCaptureCapability.each(["SIGINT", "SIGTERM"] as const)( + "awaits owned workload cleanup and emits a typed CLI cancellation after %s", + async (signal) => { + const root = await createTestTempDirectory("rea-process-cli-cancel-"); + const scenario = join(root, "scenario.json"); + await writeFile( + scenario, + JSON.stringify({ + executable: process.execPath, + arguments: [fixture, "active", root], + working_directory: root, + timeout_ms: 15_000, + idle_timeout_ms: 15_000, + }), + ); + const cli = spawn( + process.execPath, + ["scripts/rea.mjs", "capture-process", scenario, "--format", "json"], + { stdio: ["ignore", "pipe", "pipe"] }, + ); + let stdout = ""; + cli.stdout.on("data", (chunk: Buffer) => { + stdout += chunk.toString(); + }); + cli.stderr.resume(); + const closed = new Promise<{ + code: number | null; + signal: NodeJS.Signals | null; + }>((resolve, reject) => { + cli.once("error", reject); + cli.once("close", (code, exitSignal) => + resolve({ code, signal: exitSignal }), + ); + }); + let marker: Marker | undefined; + onTestFinished(async () => { + await cleanupFixture(marker); + if (cli.exitCode === null && cli.signalCode === null) cli.kill("SIGKILL"); + await closed; + }); + marker = await readMarker(root); + expect(await isRunning(marker.rootPid)).toBe(true); + cli.kill(signal); + expect(await closed).toEqual({ + code: signal === "SIGINT" ? 130 : 143, + signal: null, + }); + await waitForExit(marker.rootPid); + await expect(access(join(root, "late-write"))).rejects.toMatchObject({ + code: "ENOENT", + }); + expect(JSON.parse(stdout)).toMatchObject({ + code: "cancelled", + details: { operation: "process_capture", cleanup: "complete" }, + }); + }, + 20_000, +); + +itWithCaptureCapability( + "stops the owned descendant when an MCP request is cancelled during post-exit settlement", + async () => { + const root = await createTestTempDirectory( + "rea-process-mcp-settle-cancel-", + ); + const session = createTestBinarySession(() => { + throw new Error("Process capture must not launch a binary provider"); + }); + const completed = createDeferred(); + const completionSchema = z.object({ + tool: z.string(), + status: z.string(), + msg: z.string(), + }); + const logger = pino( + { level: "info" }, + { + write(line) { + const parsed = completionSchema.safeParse(JSON.parse(line)); + if ( + parsed.success && + parsed.data.tool === "capture_process_scenario" && + parsed.data.msg === "MCP tool execution completed" + ) + completed.resolve(parsed.data.status); + }, + }, + ); + const server = createServer(session, session, { logger }); + const client = new Client({ + name: "process-settlement-cancellation", + version: "1", + }); + let marker: Marker | undefined; + onTestFinished(async () => { + await cleanupFixture(marker); + await client.close(); + await server.close(); + await session.close(); + }); + const [clientTransport, serverTransport] = + InMemoryTransport.createLinkedPair(); + await server.connect(serverTransport); + await client.connect(clientTransport); + const controller = new AbortController(); + const call = client.callTool( + { + name: "capture_process_scenario", + arguments: { + executable: process.execPath, + arguments: [fixture, "settlement", root], + working_directory: root, + timeout_ms: 15_000, + idle_timeout_ms: 15_000, + settle_ms: 10_000, + }, + }, + { signal: controller.signal }, + ); + const cancelled = expect(call).rejects.toThrow("cancel settlement fixture"); + marker = await readMarker(root); + if (marker.childPid === undefined) + throw new Error("Settlement fixture did not report its descendant"); + await waitForExit(marker.rootPid); + expect(await isRunning(marker.childPid)).toBe(true); + // The root has exited while the descendant keeps its owned group alive. + await delay(100); + controller.abort(new Error("cancel settlement fixture")); + await cancelled; + await waitForExit(marker.childPid); + expect(await completed.promise).toBe("error"); + await expect(access(join(root, "late-write"))).rejects.toMatchObject({ + code: "ENOENT", + }); + await client.ping(); + }, + 20_000, +); diff --git a/tests/fixtures/processCaptureCancellation.mjs b/tests/fixtures/processCaptureCancellation.mjs new file mode 100644 index 000000000..6bfad03b6 --- /dev/null +++ b/tests/fixtures/processCaptureCancellation.mjs @@ -0,0 +1,45 @@ +import { spawn } from "node:child_process"; +import { writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +const [mode, root] = process.argv.slice(2); + +if (mode === "settlement") { + const child = spawn(process.execPath, [process.argv[1], "workload", root], { + stdio: ["ignore", "ignore", "ignore", "ipc"], + }); + child.on("message", async (message) => { + if (message !== "ready" || child.pid === undefined) return; + await writeFile( + join(root, "ready.json"), + JSON.stringify({ + rootPid: process.pid, + childPid: child.pid, + runId: process.env.REA_PROCESS_RUN_ID, + }), + ); + child.disconnect(); + child.unref(); + process.stdout.write("root exiting; owned workload remains\n"); + process.exit(0); + }); +} else { + // A closed PTY must not masquerade as capture cancellation in this fixture. + process.on("SIGHUP", () => undefined); + setTimeout( + () => void writeFile(join(root, "late-write"), "still running"), + 6_000, + ); + setTimeout(() => process.exit(0), 12_000); + if (mode === "workload") process.send?.("ready"); + else { + await writeFile( + join(root, "ready.json"), + JSON.stringify({ + rootPid: process.pid, + runId: process.env.REA_PROCESS_RUN_ID, + }), + ); + process.stdout.write("owned workload ready\n"); + } +}