fix(process): settle launcher exit without dropping descendant lineage

Refresh a mismatched launcher snapshot only when a new live process table proves its PID absent. Preserve uncertainty for detached descendants and continue checking ownership of surviving group members before signaling. Index retained lineage membership instead of rescanning it for each refreshed process.

Adapted from #1357. Validation: 71 source tests passed, one Linux-only identity case skipped; 32 real process-boundary tests passed on macOS.

Co-authored-by: You Wish <29498139+Y0UW1SH@users.noreply.github.com>

src/process/ProcessOwnership.ts

@@ -339,11 +339,61 @@ interface OwnedCleanupPlan {
   readonly unverified: readonly OwnershipSweepUnverifiedProcess[];
 }
 
-const createOwnedCleanupPlan = async (
+// A process-table read can race exit and retain the old state with an altered
+// command. Only a fresh table proving that PID is no longer live can settle
+// that mismatch; surviving group members still require normal token checks.
+const settleLauncherExit = async (
   ownership: OwnedProcessGroup,
   processes: readonly ProcessTableEntry[],
   host: ProcessOwnershipHost,
+): Promise<
+  | readonly ProcessTableEntry[]
+  | Extract<ProcessCleanupResult, { readonly cleaned: false }>
+> => {
+  const launcher = processes.find(({ pid }) => pid === ownership.leaderPid);
+  if (launcher === undefined) return processes;
+  const reason = launcherIdentityFailure(launcher, ownership);
+  if (reason === null) return processes;
+  const refreshed = await readLiveProcessTable(host);
+  if (
+    refreshed.available &&
+    !refreshed.processes.some(({ pid }) => pid === launcher.pid)
+  ) {
+    // Refreshing must not discard a surviving detached descendant's lineage,
+    // including a child that changed groups while the launcher exited.
+    const descendants = descendantsOf([launcher.pid], processes);
+    const descendantPids = new Set(descendants.map(({ pid }) => pid));
+    const descendantGroupIds = new Set(
+      descendants.map(({ processGroupId }) => processGroupId),
+    );
+    const refreshedDescendantPids = new Set(
+      descendantsOf(
+        descendants.map(({ pid }) => pid),
+        refreshed.processes,
+      ).map(({ pid }) => pid),
+    );
+    if (
+      refreshed.processes.some(
+        (process) =>
+          process.processGroupId !== ownership.processGroupId &&
+          (refreshedDescendantPids.has(process.pid) ||
+            descendantPids.has(process.pid) ||
+            descendantGroupIds.has(process.processGroupId)),
+      )
+    )
+      return { cleaned: false, reason };
+    return refreshed.processes;
+  }
+  return { cleaned: false, reason };
+};
+
+const createOwnedCleanupPlan = async (
+  ownership: OwnedProcessGroup,
+  initialProcesses: readonly ProcessTableEntry[],
+  host: ProcessOwnershipHost,
 ): Promise<OwnedCleanupPlan | ProcessCleanupResult> => {
+  const processes = await settleLauncherExit(ownership, initialProcesses, host);
+  if ("cleaned" in processes) return processes;
   const sampledProcessGroupIds = new Set(
     (ownership.sampledProcessGroupIds ?? []).filter(
       (processGroupId) =>
@@ -397,10 +447,7 @@ const createOwnedCleanupPlan = async (
         };
   let descendants: readonly ProcessTableEntry[] = [];
   if (launcher !== undefined) {
-    const identityFailure = launcherIdentityFailure(launcher, ownership);
-    if (identityFailure !== null)
-      return { cleaned: false, reason: identityFailure };
-    descendants = descendantsOf(launcher.pid, processes);
+    descendants = descendantsOf([launcher.pid], processes);
   }
   const descendantPids = new Set(descendants.map(({ pid }) => pid));
   const processGroupIds = new Set<number>([ownership.processGroupId]);
@@ -496,7 +543,13 @@ const revalidateOwnedProcessGroup = async (
 ): Promise<{ readonly empty: boolean } | ProcessCleanupResult> => {
   let members: readonly ProcessTableEntry[];
   try {
-    members = liveProcesses(await host.listProcesses()).filter(
+    const processes = await settleLauncherExit(
+      ownership,
+      liveProcesses(await host.listProcesses()),
+      host,
+    );
+    if ("cleaned" in processes) return processes;
+    members = processes.filter(
       ({ processGroupId: observedGroupId }) =>
         observedGroupId === processGroupId,
     );
@@ -513,14 +566,6 @@ const revalidateOwnedProcessGroup = async (
     expectedIdentities,
   );
   if (failures.length > 0) return cleanupValidationFailure(failures);
-  if (processGroupId === ownership.processGroupId) {
-    const launcher = members.find(({ pid }) => pid === ownership.leaderPid);
-    if (launcher !== undefined) {
-      const identityFailure = launcherIdentityFailure(launcher, ownership);
-      if (identityFailure !== null)
-        return { cleaned: false, reason: identityFailure };
-    }
-  }
   if (
     processGroupId !== ownership.processGroupId &&
     members.length > 0 &&
@@ -850,7 +895,7 @@ const scanTokenOwnedProcesses = async (
   const relatedGroupIds = new Set<number>();
   if (relation !== undefined) {
     relatedPids.add(relation.leaderPid);
-    for (const descendant of descendantsOf(relation.leaderPid, processes))
+    for (const descendant of descendantsOf([relation.leaderPid], processes))
       relatedPids.add(descendant.pid);
     relatedGroupIds.add(relation.processGroupId);
     for (const processGroupId of relation.sampledProcessGroupIds ?? [])

src/process/ProcessOwnership.validation.test.ts

@@ -3,6 +3,7 @@ import {
   cleanupOwnedProcessGroup,
   cleanupWindowsProcessTree,
   type ProcessOwnershipHost,
+  type ProcessTableEntry,
   type WindowsProcessTreeHost,
 } from "./ProcessOwnership.js";
 import {
@@ -11,18 +12,20 @@ import {
 } from "./ProcessOwnershipObservation.js";
 import { host, ownership } from "./ProcessOwnership.fixture.js";
 
+const launcher: ProcessTableEntry = {
+  pid: 100,
+  parentPid: 1,
+  processGroupId: 100,
+  state: "S",
+  command: "fixture",
+};
+
 describe("owned process-group cleanup validation: ownership and lineage", () => {
   it("fails closed when a descendant in another process group lacks the token", async () => {
     const adapter: ProcessOwnershipHost = {
       listProcesses: () =>
         Promise.resolve([
-          {
-            pid: 100,
-            parentPid: 1,
-            processGroupId: 100,
-            state: "S",
-            command: "fixture",
-          },
+          launcher,
           {
             pid: 101,
             parentPid: 100,
@@ -211,16 +214,9 @@ describe("owned process-group cleanup liveness rechecks", () => {
   );
 
   it("accepts a member that exits during ownership revalidation", async () => {
-    const liveLauncher = {
-      pid: 100,
-      parentPid: 1,
-      processGroupId: 100,
-      state: "S",
-      command: "fixture",
-    };
     const listProcesses = vi
       .fn<ProcessOwnershipHost["listProcesses"]>()
-      .mockResolvedValueOnce([liveLauncher])
+      .mockResolvedValueOnce([launcher])
       .mockResolvedValue([]);
     const signalGroup = vi.fn();
     const adapter: ProcessOwnershipHost = {
@@ -239,6 +235,181 @@ describe("owned process-group cleanup liveness rechecks", () => {
   });
 });
 
+describe("owned process-group cleanup validation: refreshed descendant lineage", () => {
+  it.each(["planning", "signaling"])(
+    "retains cleanup uncertainty for newly detached descendants during %s",
+    async (phase) => {
+      const child = { ...launcher, pid: 101, parentPid: 100 };
+      const stale = { ...launcher, command: "[MainThread]" };
+      const signalGroup = vi.fn();
+      let reads = 0;
+      const adapter: ProcessOwnershipHost = {
+        listProcesses: () => {
+          reads += 1;
+          if (phase === "signaling" && reads === 1)
+            return Promise.resolve([launcher, child]);
+          if (reads === (phase === "planning" ? 1 : 2))
+            return Promise.resolve([stale, child]);
+          return Promise.resolve([
+            { ...child, parentPid: 1 },
+            { ...child, pid: 102, parentPid: 101 },
+            { ...child, pid: 103, parentPid: 102, processGroupId: 103 },
+          ]);
+        },
+        environment: () =>
+          Promise.resolve({ REA_PROCESS_RUN_ID: ownership.runId }),
+        signalGroup,
+      };
+      await expect(
+        cleanupOwnedProcessGroup(
+          { ...ownership, expectedCommand: "fixture" },
+          adapter,
+        ),
+      ).resolves.toMatchObject({ cleaned: false });
+      expect(signalGroup).not.toHaveBeenCalled();
+    },
+  );
+});
+
+describe("owned process-group cleanup validation: launcher exit races", () => {
+  it.each([101, 999])(
+    "retains cleanup uncertainty for a detached descendant in group %s after launcher exit",
+    async (processGroupId) => {
+      const child = {
+        ...launcher,
+        pid: 101,
+        parentPid: 100,
+        processGroupId: 101,
+      };
+      const signalGroup = vi.fn();
+      const listProcesses = vi
+        .fn<ProcessOwnershipHost["listProcesses"]>()
+        .mockResolvedValueOnce([
+          { ...launcher, command: "[MainThread]" },
+          child,
+        ])
+        .mockResolvedValue([{ ...child, parentPid: 1, processGroupId }]);
+      await expect(
+        cleanupOwnedProcessGroup(
+          {
+            ...ownership,
+            expectedCommand: "fixture",
+          },
+          {
+            listProcesses,
+            environment: () =>
+              Promise.resolve({ REA_PROCESS_RUN_ID: "run-token" }),
+            signalGroup,
+          },
+        ),
+      ).resolves.toMatchObject({ cleaned: false });
+      expect(signalGroup).not.toHaveBeenCalled();
+    },
+  );
+  it.each(["planning", "signaling"])(
+    "settles a stale launcher command during %s without skipping surviving group ownership",
+    async (phase) => {
+      for (const remaining of ["absent", "zombie", "owned", "foreign"]) {
+        const stale = { ...launcher, command: "[MainThread]" };
+        const child = { ...launcher, pid: 101, command: "child" };
+        let reads = 0;
+        const signalGroup = vi.fn();
+        const adapter: ProcessOwnershipHost = {
+          listProcesses: () => {
+            reads += 1;
+            if (phase === "signaling" && reads === 1)
+              return Promise.resolve([launcher]);
+            if (reads === (phase === "planning" ? 1 : 2))
+              return Promise.resolve([stale]);
+            return Promise.resolve(
+              remaining === "absent"
+                ? []
+                : remaining === "zombie"
+                  ? [{ ...stale, state: "Z" }]
+                  : [child],
+            );
+          },
+          environment: () =>
+            Promise.resolve({
+              REA_PROCESS_RUN_ID:
+                remaining === "foreign" && reads > 1
+                  ? "other-run"
+                  : "run-token",
+            }),
+          signalGroup,
+        };
+        const result = await cleanupOwnedProcessGroup(
+          { ...ownership, expectedCommand: "fixture" },
+          adapter,
+        );
+        if (remaining === "foreign")
+          expect(result).toMatchObject({
+            cleaned: false,
+            reason: "process tree contains an unowned or PID-reused process",
+          });
+        else
+          expect(result).toEqual({
+            cleaned: true,
+            signaled: remaining === "owned",
+          });
+        if (remaining === "owned")
+          expect(signalGroup).toHaveBeenCalledWith(100, "SIGKILL");
+        else expect(signalGroup).not.toHaveBeenCalled();
+      }
+    },
+  );
+  it("fails closed when a mismatched launcher's exit cannot be inspected", async () => {
+    const { adapter, signalGroup } = host({
+      100: { REA_PROCESS_RUN_ID: "run-token" },
+    });
+    const listProcesses = vi
+      .fn<ProcessOwnershipHost["listProcesses"]>()
+      .mockResolvedValueOnce(await adapter.listProcesses())
+      .mockRejectedValue(new Error("process table unavailable"));
+    await expect(
+      cleanupOwnedProcessGroup(
+        { ...ownership, expectedCommand: "/owned/hopper" },
+        { ...adapter, listProcesses },
+      ),
+    ).resolves.toMatchObject({
+      cleaned: false,
+      reason: expect.stringContaining("command identity did not match"),
+    });
+    expect(signalGroup).not.toHaveBeenCalled();
+  });
+  it.each(["planning", "signaling"])(
+    "rejects a surviving launcher PID in another group during %s",
+    async (phase) => {
+      let reads = 0;
+      const signalGroup = vi.fn();
+      const adapter: ProcessOwnershipHost = {
+        listProcesses: () => {
+          reads += 1;
+          return Promise.resolve([
+            phase === "signaling" && reads === 1
+              ? launcher
+              : reads === (phase === "planning" ? 1 : 2)
+                ? { ...launcher, command: "[MainThread]" }
+                : { ...launcher, processGroupId: 999 },
+          ]);
+        },
+        environment: () => Promise.resolve({ REA_PROCESS_RUN_ID: "run-token" }),
+        signalGroup,
+      };
+      await expect(
+        cleanupOwnedProcessGroup(
+          { ...ownership, expectedCommand: "fixture" },
+          adapter,
+        ),
+      ).resolves.toMatchObject({
+        cleaned: false,
+        reason: expect.stringContaining("command identity did not match"),
+      });
+      expect(signalGroup).not.toHaveBeenCalled();
+    },
+  );
+});
+
 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();
@@ -293,13 +464,7 @@ describe("owned process-group cleanup validation: exited members", () => {
     const adapter: ProcessOwnershipHost = {
       listProcesses: () =>
         Promise.resolve([
-          {
-            pid: 100,
-            parentPid: 1,
-            processGroupId: 100,
-            state: "S",
-            command: "fixture",
-          },
+          launcher,
           {
             pid: 101,
             parentPid: 100,

src/process/ProcessOwnershipObservation.ts

@@ -116,7 +116,7 @@ export const observeOwnedProcessLineage = async (
   const identityFailure = launcherIdentityFailure(launcher, ownership);
   if (identityFailure !== null)
     return unavailableLineage(ownership, identityFailure);
-  const descendants = descendantsOf(launcher.pid, processes);
+  const descendants = descendantsOf([launcher.pid], processes);
   const verifiedDescendants: ProcessTableEntry[] = [];
   for await (const { process: member, observation } of readProcessRunTokens(
     host,

src/process/ProcessOwnershipProcessTree.ts

@@ -6,9 +6,9 @@ export const liveProcesses = (
 ): readonly ProcessTableEntry[] =>
   members.filter(({ state }) => !state.startsWith("Z"));
 
-/** Return every live-table descendant of a launcher in breadth-first order. */
+/** Return every live-table descendant of the selected roots in breadth-first order. */
 export const descendantsOf = (
-  launcherPid: number,
+  rootPids: readonly number[],
   processes: readonly ProcessTableEntry[],
 ): ProcessTableEntry[] => {
   const childrenByParent = new Map<number, ProcessTableEntry[]>();
@@ -18,8 +18,8 @@ export const descendantsOf = (
     childrenByParent.set(process.parentPid, siblings);
   }
   const descendants: ProcessTableEntry[] = [];
-  const pending = [...(childrenByParent.get(launcherPid) ?? [])];
-  const visited = new Set<number>([launcherPid]);
+  const pending = rootPids.flatMap((pid) => childrenByParent.get(pid) ?? []);
+  const visited = new Set<number>(rootPids);
   for (const process of pending) {
     if (visited.has(process.pid)) continue;
     visited.add(process.pid);