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

@@ -47,6 +47,7 @@ export const isProcessCliFailure = (
 export const captureProcessScenarioFile = async (
   path: string,
   environment: Readonly<Record<string, string | undefined>> = 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,
     );

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, {

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" }),

src/process/ProcessOwnershipObservation.ts

@@ -19,31 +19,36 @@ import type {
 export const observeOwnedProcessGroupWithHost = async (
   ownership: OwnedProcessGroup,
   host: ProcessOwnershipHost,
+  signal?: AbortSignal,
 ): Promise<ProcessGroupObservation> => {
+  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<ProcessGroupObservation> =>
-  observeOwnedProcessGroupWithHost(ownership, host);
+  observeOwnedProcessGroupWithHost(ownership, host, signal);
 
 /**
  * Record the live launcher and descendant lineage after run-token validation.

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<ObservedProcessSettlement> => {
+  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<void>((resolve) => setTimeout(resolve, 50));
+    await delay(50, undefined, { signal });
   }
   recordEvent("lifecycle", 1);
   return { state: "alive_at_deadline", elapsed_ms: Date.now() - started };

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 = {

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<typeof markerSchema>;
+
+const readMarker = async (root: string): Promise<Marker> => {
+  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<boolean> => {
+  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<void> => {
+  await expect
+    .poll(() => isRunning(pid), { timeout: 3_000, interval: 25 })
+    .toBe(false);
+};
+
+const cleanupFixture = async (marker: Marker | undefined): Promise<void> => {
+  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<string>();
+    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,
+);

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");
+  }
+}