fix(process): await owned observation commands through finalization

Supervise sampler and ownership-query commands through physical exit on
cancellation. Share Darwin helper command ownership, reject new commands
after close starts, and drain admitted commands before removing the root.
Keep failed root cleanup retryable without rerunning completed queries.

Fixes #5199
Fixes #5202

docs/process-capture.md

@@ -38,6 +38,10 @@ supervision on macOS. REA removes that directory when it exits. REA does not
 install Xcode, Command Line Tools, or other software. Capability checks prepare
 this helper before reporting macOS capture as available.
 
+Stopping process sampling or cancelling an ownership query waits for its issued
+inspection command to exit. Closing a macOS ownership reader rejects new queries
+and waits for admitted helper commands before removing their executable directory.
+
 ## Capture a command
 
 For a small first capture, save this as `scenario.json`:

src/process/DarwinProcessRunTokenReader.finalization.test.ts

@@ -0,0 +1,328 @@
+import { basename, join } from "node:path";
+import { tmpdir } from "node:os";
+import { access } from "node:fs/promises";
+
+import { expect, it, onTestFinished, vi } from "vitest";
+
+import { ProviderCleanupError } from "../domain/providerCleanupError.js";
+
+const command = vi.hoisted(() => {
+  let helperMode: "read" | "identities" = "read";
+  let helperSessions: Array<{
+    readonly pid: number;
+    readonly failed: boolean;
+    readonly gate: Promise<void>;
+    readonly release: () => void;
+  }> = [];
+  let expectedHelpers = 1;
+  let startedHelpers = 0;
+  let failRootRemovals = 0;
+  let startedResolve: (() => void) | undefined;
+  let started: Promise<void> = Promise.resolve();
+  const events: string[] = [];
+
+  const makeGate = () => {
+    let release = (): void => {};
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    return { gate, release };
+  };
+
+  return {
+    configure(mode: "read" | "identities") {
+      helperMode = mode;
+      events.length = 0;
+      expectedHelpers = 1;
+      startedHelpers = 0;
+      failRootRemovals = 0;
+      const session = makeGate();
+      helperSessions = [{ pid: 123, failed: false, ...session }];
+      started = new Promise<void>((resolve) => {
+        startedResolve = resolve;
+      });
+    },
+    configureMixed() {
+      helperMode = "read";
+      events.length = 0;
+      expectedHelpers = 2;
+      startedHelpers = 0;
+      failRootRemovals = 1;
+      helperSessions = [
+        { pid: 123, failed: false, ...makeGate() },
+        { pid: 124, failed: true, ...makeGate() },
+      ];
+      started = new Promise<void>((resolve) => {
+        startedResolve = resolve;
+      });
+    },
+    waitUntilStarted() {
+      return started;
+    },
+    release() {
+      for (const session of helperSessions) session.release();
+    },
+    releaseHelper(pid: number) {
+      helperSessions.find((session) => session.pid === pid)?.release();
+    },
+    async remove(path: string) {
+      const ownedPath = String(path);
+      if (ownedPath.includes("rea-process-token-")) {
+        events.push(`root-remove:${ownedPath}`);
+        if (failRootRemovals > 0) {
+          failRootRemovals -= 1;
+          throw new Error("fixture root removal failure");
+        }
+      }
+    },
+    get events() {
+      return [...events];
+    },
+    async run(commandPath: string, arguments_: readonly string[]) {
+      if (basename(commandPath) === "fixture-swiftc") {
+        const outputIndex = arguments_.indexOf("-o");
+        const output = arguments_[outputIndex + 1];
+        if (output === undefined) throw new Error("Missing helper output path");
+        const fs = await import("node:fs/promises");
+        await fs.writeFile(output, "fixture helper");
+        return { stdout: "", stderr: "", stdoutBytes: 0, stderrBytes: 0 };
+      }
+      const session = helperSessions[startedHelpers];
+      if (session === undefined)
+        throw new Error("Unexpected helper invocation");
+      startedHelpers += 1;
+      events.push(`helper-start:${session.pid}`);
+      if (startedHelpers === expectedHelpers) startedResolve?.();
+      try {
+        await session.gate;
+        if (session.failed) throw new Error("fixture helper failure");
+        const stdout = JSON.stringify({
+          results: [
+            helperMode === "read"
+              ? { pid: session.pid, state: "readable", run_id: "run" }
+              : {
+                  pid: session.pid,
+                  state: "readable",
+                  identity: "start:456",
+                },
+          ],
+        });
+        return {
+          stdout,
+          stderr: "",
+          stdoutBytes: Buffer.byteLength(stdout),
+          stderrBytes: 0,
+        };
+      } finally {
+        events.push(`helper-finalized:${session.pid}`);
+      }
+    },
+  };
+});
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const original = await importOriginal<typeof import("node:fs/promises")>();
+  return {
+    ...original,
+    rm: async (
+      path: Parameters<typeof original.rm>[0],
+      options?: Parameters<typeof original.rm>[1],
+    ) => {
+      const ownedPath = String(path);
+      await command.remove(ownedPath);
+      return original.rm(path, options);
+    },
+  };
+});
+
+vi.mock("./ExecFileOutput.js", async (importOriginal) => {
+  const original = await importOriginal<typeof import("./ExecFileOutput.js")>();
+  return {
+    ...original,
+    execFileOutput: (commandPath: string, arguments_: readonly string[]) =>
+      command.run(commandPath, arguments_),
+  };
+});
+
+import { createDarwinProcessRunTokenReader } from "./DarwinProcessRunTokenReader.js";
+
+it.each(["read", "identities"] as const)(
+  "waits for an active helper %s before removing its executable root",
+  async (mode) => {
+    command.configure(mode);
+    const reader = createDarwinProcessRunTokenReader({
+      compiler: {
+        kind: "swiftc",
+        executable: join(tmpdir(), "fixture-swiftc"),
+        sdkRoot: tmpdir(),
+      },
+      environment: {},
+    });
+    let operation: Promise<unknown> | undefined;
+    let closing: Promise<void> | undefined;
+    onTestFinished(async () => {
+      command.release();
+      await Promise.allSettled([operation, closing, reader.close()]);
+    });
+
+    operation =
+      mode === "read"
+        ? reader.read([
+            {
+              pid: 123,
+              parentPid: 1,
+              processGroupId: 123,
+              state: "S",
+              command: "fixture",
+            },
+          ])
+        : reader.identities([
+            {
+              pid: 123,
+              parentPid: 1,
+              processGroupId: 123,
+              state: "S",
+              command: "fixture",
+            },
+          ]);
+    await command.waitUntilStarted();
+
+    let closeSettled = false;
+    closing = reader.close().then(() => {
+      closeSettled = true;
+    });
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    const closeWaitedForHelper = !closeSettled;
+    command.release();
+    const result = await operation;
+    await closing;
+
+    expect(closeWaitedForHelper).toBe(true);
+    const events = command.events;
+    const helperFinalizedAt = events.findIndex((event) =>
+      event.startsWith("helper-finalized:"),
+    );
+    const rootRemovedAt = events.findIndex((event) =>
+      event.startsWith("root-remove:"),
+    );
+    expect(helperFinalizedAt).toBeGreaterThanOrEqual(0);
+    expect(rootRemovedAt).toBeGreaterThan(helperFinalizedAt);
+    expect(result).toEqual(
+      new Map([
+        [
+          123,
+          mode === "read"
+            ? { state: "readable", runId: "run" }
+            : { state: "readable", identity: "start:456" },
+        ],
+      ]),
+    );
+    await expect(
+      mode === "read"
+        ? reader.read([
+            {
+              pid: 123,
+              parentPid: 1,
+              processGroupId: 123,
+              state: "S",
+              command: "fixture",
+            },
+          ])
+        : reader.identities([
+            {
+              pid: 123,
+              parentPid: 1,
+              processGroupId: 123,
+              state: "S",
+              command: "fixture",
+            },
+          ]),
+    ).rejects.toThrow("closed");
+  },
+);
+
+it("drains mixed helper outcomes before cleanup and retries failed root removal", async () => {
+  command.configureMixed();
+  const reader = createDarwinProcessRunTokenReader({
+    compiler: {
+      kind: "swiftc",
+      executable: join(tmpdir(), "fixture-swiftc"),
+      sdkRoot: tmpdir(),
+    },
+    environment: {},
+  });
+  const entry = (pid: number) => ({
+    pid,
+    parentPid: 1,
+    processGroupId: pid,
+    state: "S",
+    command: "fixture",
+  });
+  let successfulRead: Promise<unknown> | undefined;
+  let failedRead: Promise<unknown> | undefined;
+  let closing: Promise<void> | undefined;
+  onTestFinished(async () => {
+    command.release();
+    await Promise.allSettled([
+      successfulRead,
+      failedRead,
+      closing,
+      reader.close(),
+    ]);
+  });
+
+  successfulRead = reader.read([entry(123)]);
+  failedRead = reader.read([entry(124)]);
+  await command.waitUntilStarted();
+
+  closing = reader.close();
+  command.releaseHelper(124);
+  const failedOutcome = await failedRead.then(
+    () => undefined,
+    (cause: unknown) => cause,
+  );
+  expect(failedOutcome).toBeInstanceOf(Error);
+  expect(command.events.some((event) => event.startsWith("root-remove:"))).toBe(
+    false,
+  );
+
+  command.releaseHelper(123);
+  await expect(successfulRead).resolves.toEqual(
+    new Map([[123, { state: "readable", runId: "run" }]]),
+  );
+  const closeFailure = await closing.then(
+    () => undefined,
+    (cause: unknown) => cause,
+  );
+  if (!(closeFailure instanceof ProviderCleanupError))
+    throw new Error("Expected retained helper-root cleanup failure");
+  expect(closeFailure.cleanupIncomplete).toBe(true);
+  expect(closeFailure.cleanupResources).toHaveLength(1);
+  const [retainedRoot] = closeFailure.cleanupResources;
+  if (retainedRoot === undefined)
+    throw new Error("Expected retained helper root");
+  await expect(access(retainedRoot)).resolves.toBeUndefined();
+
+  const events = command.events;
+  const finalHelperAt = Math.max(
+    events.indexOf("helper-finalized:123"),
+    events.indexOf("helper-finalized:124"),
+  );
+  const removalEvents = events.filter((event) =>
+    event.startsWith("root-remove:"),
+  );
+  expect(finalHelperAt).toBeGreaterThanOrEqual(0);
+  expect(removalEvents).toHaveLength(1);
+  expect(events.indexOf(removalEvents[0]!)).toBeGreaterThan(finalHelperAt);
+
+  await reader.close();
+  await expect(access(retainedRoot)).rejects.toMatchObject({ code: "ENOENT" });
+  expect(closeFailure.cleanupIncomplete).toBe(false);
+  expect(
+    command.events.filter((event) => event.startsWith("helper-start:")),
+  ).toHaveLength(2);
+  expect(
+    command.events.filter((event) => event.startsWith("root-remove:")),
+  ).toHaveLength(2);
+  await expect(reader.read([entry(123)])).rejects.toThrow("closed");
+});

src/process/DarwinProcessRunTokenReader.ts

@@ -402,6 +402,7 @@ export const createDarwinProcessRunTokenReader = (
   let closePromise: Promise<void> | undefined;
   let closed = false;
   let exitCleanupInstalled = false;
+  const activeCommands = new Set<Promise<string>>();
 
   const cleanupAtExit = () => {
     if (root !== undefined) rmSync(root.path, { recursive: true, force: true });
@@ -600,19 +601,36 @@ export const createDarwinProcessRunTokenReader = (
     }
   };
 
+  const runHelper = (
+    arguments_: readonly string[],
+    signal?: AbortSignal,
+  ): Promise<string> => {
+    if (closed)
+      return Promise.reject(new Error("Darwin process token reader is closed"));
+    const pending = (async () => {
+      const binary = await compile(signal);
+      const { stdout } = await execFileOutput(binary, arguments_, {
+        timeout: 15_000,
+        stopSignal: "SIGTERM",
+        ...(signal === undefined ? {} : { signal }),
+      });
+      signal?.throwIfAborted();
+      return stdout;
+    })();
+    activeCommands.add(pending);
+    return pending.finally(() => {
+      activeCommands.delete(pending);
+    });
+  };
+
   const read = async (
     processes: readonly ProcessTableEntry[],
     signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessRunTokenObservation>> => {
     signal?.throwIfAborted();
     if (processes.length === 0) return new Map();
-    const binary = await compile(signal);
     const pids = processes.map(({ pid }) => String(pid));
-    const { stdout } = await execFileOutput(binary, pids, {
-      timeout: 15_000,
-      ...(signal === undefined ? {} : { signal }),
-    });
-    signal?.throwIfAborted();
+    const stdout = await runHelper(pids, signal);
     const decoded = safeParseJson(stdout);
     if (!decoded.ok)
       throw new Error(
@@ -657,12 +675,8 @@ export const createDarwinProcessRunTokenReader = (
     signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessIdentityObservation>> => {
     if (processes.length === 0) return new Map();
-    const binary = await compile(signal);
     const pids = processes.map(({ pid }) => String(pid));
-    const { stdout } = await execFileOutput(binary, ["--identities", ...pids], {
-      timeout: 15_000,
-      ...(signal === undefined ? {} : { signal }),
-    });
+    const stdout = await runHelper(["--identities", ...pids], signal);
     const decoded = safeParseJson(stdout);
     if (!decoded.ok)
       throw new Error(
@@ -708,6 +722,7 @@ export const createDarwinProcessRunTokenReader = (
         operation.controller.abort();
         await operation.promise;
       }
+      await Promise.allSettled(activeCommands);
       await cleanupRoot();
     })().catch((cause: unknown) => {
       if (closePromise === pendingClose) closePromise = undefined;

src/process/ProcessOwnershipObservation.ts

@@ -259,7 +259,11 @@ export const createSystemProcessOwnershipHost = (
     const { stdout } = await execFileOutput(
       "ps",
       ["-axo", "pid=,ppid=,pgid=,uid=,stat=,command="],
-      { env: hostEnvironment, ...(signal === undefined ? {} : { signal }) },
+      {
+        env: hostEnvironment,
+        stopSignal: "SIGTERM",
+        ...(signal === undefined ? {} : { signal }),
+      },
     );
     return stdout
       .split("\n")
@@ -412,6 +416,7 @@ export const createSystemProcessOwnershipHost = (
         ["eww", "-p", String(pid)],
         {
           env: hostEnvironment,
+          stopSignal: "SIGTERM",
           ...(signal === undefined ? {} : { signal }),
         },
       );

src/process/capture/ProcessSampling.ts

@@ -130,7 +130,7 @@ const readPsRowsRaw = async (
   const { stdout } = await execFileOutput(
     "ps",
     ["-axo", "pid=,ppid=,pgid=,sess=,command="],
-    { signal },
+    { signal, stopSignal: "SIGTERM" },
   );
   return stdout
     .split("\n")

tests/process-global/processObservationCommandFinalization.test.ts

@@ -0,0 +1,274 @@
+import { chmod, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { expect, it, onTestFinished } from "vitest";
+
+import { createDarwinProcessRunTokenReader } from "../../src/process/DarwinProcessRunTokenReader.js";
+import { createSystemProcessOwnershipHost } from "../../src/process/ProcessOwnershipObservation.js";
+import { startProcessSampler } from "../../src/process/capture/ProcessSampling.js";
+
+const waitForFile = async (path: string, timeoutMs = 4_000): Promise<void> => {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    try {
+      await readFile(path);
+      return;
+    } catch (cause) {
+      if (
+        typeof cause !== "object" ||
+        cause === null ||
+        !("code" in cause) ||
+        cause.code !== "ENOENT"
+      )
+        throw cause;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 5));
+  }
+  throw new Error(`Timed out waiting for ${path}`);
+};
+
+const waitForCommandOrFailure = async (
+  readyPath: string,
+  pending: Promise<unknown>,
+): Promise<void> => {
+  const outcome = await Promise.race([
+    waitForFile(readyPath).then(() => undefined),
+    pending.then(
+      () => new Error("The command completed before its ready marker"),
+      (cause: unknown) => cause,
+    ),
+  ]);
+  if (outcome !== undefined) throw outcome;
+};
+
+const delayedCommandSource = (
+  readyPath: string,
+  donePath: string,
+  stdout = "123 1 123 123 501 S fixture-process",
+): string =>
+  [
+    `const fs = require("node:fs");`,
+    `process.on("SIGTERM", () => { setTimeout(() => { fs.writeFileSync(${JSON.stringify(donePath)}, "done"); process.exit(0); }, 150); });`,
+    `fs.writeFileSync(${JSON.stringify(readyPath)}, "ready");`,
+    `process.stdout.write(${JSON.stringify(`${stdout}\n`)});`,
+    `setInterval(() => {}, 1000);`,
+    "",
+  ].join("\n");
+
+const createDelayedCommand = async (directory: string) => {
+  const readyPath = join(directory, "ready");
+  const donePath = join(directory, "done");
+  const executable = join(directory, "ps");
+  const runtime = `#!${process.execPath}\n${delayedCommandSource(readyPath, donePath)}`;
+  await writeFile(executable, runtime);
+  await chmod(executable, 0o755);
+  return { readyPath, donePath, executable };
+};
+
+const expectOwnedCommandToFinishOnAbort = async (
+  run: (signal: AbortSignal, directory: string) => Promise<unknown>,
+): Promise<void> => {
+  const directory = await mkdtemp(join(tmpdir(), "rea-process-command-"));
+  const command = await createDelayedCommand(directory);
+  const controller = new AbortController();
+  const reason = new Error("cancel owned process command");
+  let pending: Promise<unknown> | undefined;
+  onTestFinished(async () => {
+    try {
+      controller.abort(reason);
+      await pending?.catch(() => {});
+      if (
+        await readFile(command.readyPath).then(
+          () => true,
+          () => false,
+        )
+      )
+        await waitForFile(command.donePath);
+    } finally {
+      await rm(directory, { recursive: true, force: true });
+    }
+  });
+
+  pending = run(controller.signal, directory);
+  void pending.catch(() => {});
+  await waitForCommandOrFailure(command.readyPath, pending);
+  controller.abort(reason);
+  const failure = await pending.then(
+    () => undefined,
+    (cause: unknown) => cause,
+  );
+
+  if (!(failure instanceof Error))
+    throw new Error("The cancelled process command did not return an error");
+  expect(failure.cause).toBe(reason);
+  await expect(readFile(command.donePath, "utf8")).resolves.toBe("done");
+};
+
+const createFakeDarwinReader = async (
+  directory: string,
+  command: Awaited<ReturnType<typeof createDelayedCommand>>,
+) => {
+  const compiler = join(directory, "swiftc");
+  const generatedRuntime = `#!${process.execPath}\n${delayedCommandSource(command.readyPath, command.donePath)}`;
+  const compilerSource = [
+    `const fs = require("node:fs");`,
+    `const args = process.argv.slice(2);`,
+    `const outputIndex = args.indexOf("-o");`,
+    `const output = args[outputIndex + 1];`,
+    `fs.writeFileSync(output, ${JSON.stringify(generatedRuntime)});`,
+    `fs.chmodSync(output, 0o755);`,
+  ].join("\n");
+  await writeFile(compiler, `#!${process.execPath}\n${compilerSource}\n`);
+  await chmod(compiler, 0o755);
+  const reader = createDarwinProcessRunTokenReader({
+    compiler: { kind: "swiftc", executable: compiler, sdkRoot: directory },
+    compileTimeoutMs: 2_000,
+    moduleCachePath: join(directory, "module-cache"),
+  });
+  return reader;
+};
+
+const processEntry = {
+  pid: 123,
+  parentPid: 1,
+  processGroupId: 123,
+  uid: 501,
+  state: "S",
+  command: "fixture-process",
+} as const;
+
+it.skipIf(process.platform === "win32")(
+  "waits for the owned ps command to exit when sampling stops",
+  async () => {
+    const directory = await mkdtemp(join(tmpdir(), "rea-process-sampler-"));
+    const readyPath = join(directory, "ready");
+    const donePath = join(directory, "done");
+    const psPath = join(directory, "ps");
+    const originalPath = process.env.PATH;
+    const platformDescriptor = Object.getOwnPropertyDescriptor(
+      process,
+      "platform",
+    );
+    if (platformDescriptor === undefined)
+      throw new Error("process.platform descriptor is unavailable");
+
+    let stopSampler: (() => Promise<unknown>) | undefined;
+    let restoreProcessGlobals = (): void => {};
+    onTestFinished(async () => {
+      try {
+        await stopSampler?.();
+        if (
+          await readFile(readyPath).then(
+            () => true,
+            () => false,
+          )
+        )
+          await waitForFile(donePath);
+      } finally {
+        restoreProcessGlobals();
+        await rm(directory, { recursive: true, force: true });
+      }
+    });
+
+    await writeFile(
+      psPath,
+      `#!${process.execPath}\n${delayedCommandSource(readyPath, donePath, "100 1 100 100 fixture-ps-root")}`,
+    );
+    await chmod(psPath, 0o755);
+
+    process.env.PATH = `${directory}${originalPath === undefined ? "" : `:${originalPath}`}`;
+    Object.defineProperty(process, "platform", {
+      ...platformDescriptor,
+      value: "darwin",
+    });
+    restoreProcessGlobals = () => {
+      if (originalPath === undefined) delete process.env.PATH;
+      else process.env.PATH = originalPath;
+      Object.defineProperty(process, "platform", platformDescriptor);
+    };
+
+    stopSampler = startProcessSampler({
+      rootPid: 100,
+      runId: "fixture-run-token",
+      started: 0,
+      limit: 20,
+      samples: [],
+    });
+    await waitForFile(readyPath);
+
+    await stopSampler();
+
+    await expect(readFile(donePath, "utf8")).resolves.toBe("done");
+  },
+);
+
+it
+  .skipIf(process.platform === "win32")
+  .each(["darwin-list", "freebsd-environment"] as const)(
+  "waits for the %s ownership command to exit after cancellation",
+  async (kind) => {
+    await expectOwnedCommandToFinishOnAbort(async (signal, directory) => {
+      const platform = kind === "darwin-list" ? "darwin" : "freebsd";
+      const host = createSystemProcessOwnershipHost(platform, {
+        PATH: directory,
+      });
+      return kind === "darwin-list"
+        ? host.listProcesses(signal)
+        : host.environment(123, signal);
+    });
+  },
+);
+
+const expectDarwinReaderCommandToFinishOnAbort = async (
+  operation: "read" | "identities",
+): Promise<void> => {
+  const directory = await mkdtemp(join(tmpdir(), `rea-darwin-${operation}-`));
+  const command = await createDelayedCommand(directory);
+  const reader = await createFakeDarwinReader(directory, command);
+  const controller = new AbortController();
+  const reason = new Error(`cancel ${operation}`);
+  let pending: Promise<unknown> | undefined;
+  onTestFinished(async () => {
+    try {
+      controller.abort(reason);
+      await pending?.catch(() => {});
+      if (
+        await readFile(command.readyPath).then(
+          () => true,
+          () => false,
+        )
+      )
+        await waitForFile(command.donePath);
+      await reader.close();
+    } finally {
+      await rm(directory, { recursive: true, force: true });
+    }
+  });
+
+  pending =
+    operation === "read"
+      ? reader.read([processEntry], controller.signal)
+      : reader.identities([processEntry], controller.signal);
+  void pending.catch(() => {});
+  await waitForCommandOrFailure(command.readyPath, pending);
+  controller.abort(reason);
+  const failure = await pending.then(
+    () => undefined,
+    (cause: unknown) => cause,
+  );
+
+  if (!(failure instanceof Error))
+    throw new Error(
+      `The cancelled ${operation} command did not return an error`,
+    );
+  expect(failure.cause).toBe(reason);
+  await expect(readFile(command.donePath, "utf8")).resolves.toBe("done");
+};
+
+it.skipIf(process.platform === "win32").each(["read", "identities"] as const)(
+  "waits for the compiled Darwin %s command to exit after cancellation",
+  async (operation) => {
+    await expectDarwinReaderCommandToFinishOnAbort(operation);
+  },
+);