perf(process): batch liveness rechecks within ownership phases

Recheck unreadable token and identity observations against one fresh live process snapshot per phase. Preserve token mismatches, fail closed on inspection errors, and retain liveness failure diagnostics.

Validated with focused process ownership, native capture, CLI and MCP boundaries. An eight-member injected-unreadable workload using native ps reduced snapshots from nine to two with equivalent failures; real permission-denied provider behavior was not exercised.

src/process/ProcessOwnership.identity.test.ts

@@ -377,7 +377,7 @@ describe("process identity during token validation", () => {
     expect(reusedDuringFinalRead).toBe(true);
   });
 
-  it("fails closed when a candidate PID changes identity during token reading", async () => {
+  it("fails closed when candidate PIDs change identity during token reading", async () => {
     let identityReads = 0;
     const process: ProcessTableEntry = {
       pid: 911,
@@ -386,8 +386,10 @@ describe("process identity during token validation", () => {
       state: "S",
       command: "reused-during-token-read",
     };
+    const processes = [process, { ...process, pid: 912, processGroupId: 912 }];
+    const listProcesses = vi.fn(async () => processes);
     const host: ProcessOwnershipHost = {
-      listProcesses: () => Promise.resolve([process]),
+      listProcesses,
       environment: () => Promise.resolve({}),
       processIdentities: (entries) => {
         identityReads += 1;
@@ -414,21 +416,24 @@ describe("process identity during token validation", () => {
     };
 
     await expect(
-      verifyNoTokenOwnedProcesses("run-token", host, [
-        { pid: process.pid, identity: "old-start" },
-      ]),
+      verifyNoTokenOwnedProcesses(
+        "run-token",
+        host,
+        processes.map(({ pid }) => ({ pid, identity: "old-start" })),
+      ),
     ).resolves.toEqual({
       cleaned: false,
       reason: "process identity could not be revalidated",
-      failures: [
-        {
-          pid: process.pid,
-          reason: "process-identity-unavailable",
-          diagnostic:
-            "process identity changed or became unavailable during token validation",
-        },
-      ],
+      failures: processes.map(({ pid }) => ({
+        pid,
+        reason: "process-identity-unavailable",
+        diagnostic:
+          "process identity changed or became unavailable during token validation",
+      })),
     });
+    // Initial observation, final token classification, then one fresh recheck
+    // shared by both changed identities.
+    expect(listProcesses).toHaveBeenCalledTimes(3);
   });
 });
 

src/process/ProcessOwnership.ts

@@ -628,6 +628,7 @@ const processOwnershipFailures = async (
       expectedIdentities.has(pid),
     );
     const observed = await host.processIdentities(expectedMembers);
+    const identityFailures: ProcessOwnershipValidationFailure[] = [];
     for (const member of expectedMembers) {
       const expected = expectedIdentities.get(member.pid);
       const current = observed.get(member.pid);
@@ -636,8 +637,7 @@ const processOwnershipFailures = async (
         current?.state !== "readable" ||
         current.identity !== expected
       ) {
-        if (await processIsGone(host, member.pid)) continue;
-        failures.push({
+        identityFailures.push({
           pid: member.pid,
           reason: "process-identity-unavailable",
           diagnostic:
@@ -645,6 +645,7 @@ const processOwnershipFailures = async (
         });
       }
     }
+    failures.push(...(await recheckOwnershipFailures(identityFailures, host)));
   }
   failures.push(...(await processRunTokenFailures(members, runId, host)));
   if (
@@ -656,6 +657,7 @@ const processOwnershipFailures = async (
       expectedIdentities.has(pid),
     );
     const observed = await host.processIdentities(expectedMembers);
+    const identityFailures: ProcessOwnershipValidationFailure[] = [];
     for (const member of expectedMembers) {
       const expected = expectedIdentities.get(member.pid);
       const current = observed.get(member.pid);
@@ -665,14 +667,14 @@ const processOwnershipFailures = async (
         current.identity === expected
       )
         continue;
-      if (await processIsGone(host, member.pid)) continue;
-      failures.push({
+      identityFailures.push({
         pid: member.pid,
         reason: "process-identity-unavailable",
         diagnostic:
           "process identity changed or became unavailable during token validation",
       });
     }
+    failures.push(...(await recheckOwnershipFailures(identityFailures, host)));
   }
   return failures;
 };
@@ -692,24 +694,43 @@ const processRunTokenFailures = async (
         failures.push({ pid: member.pid, reason: "run-token-mismatch" });
       continue;
     }
-    try {
-      const live = liveProcesses(await host.listProcesses());
-      if (!live.some(({ pid }) => pid === member.pid)) continue;
-    } catch (recheckCause: unknown) {
-      failures.push({
-        pid: member.pid,
-        reason: "environment-unreadable",
-        diagnostic: `${observation.reason}; process liveness recheck failed: ${errorMessage(recheckCause)}`,
-      });
-      continue;
-    }
     failures.push({
       pid: member.pid,
       reason: "environment-unreadable",
       diagnostic: observation.reason,
     });
   }
-  return failures;
+  return recheckOwnershipFailures(failures, host);
+};
+
+/**
+ * Take a fresh liveness snapshot after each ownership-read phase. Only an
+ * absent PID resolves an unreadable observation; a readable token mismatch
+ * remains a failure. Never reuse this snapshot across identity/token reads.
+ */
+const recheckOwnershipFailures = async (
+  failures: readonly ProcessOwnershipValidationFailure[],
+  host: ProcessOwnershipHost,
+): Promise<readonly ProcessOwnershipValidationFailure[]> => {
+  if (!failures.some(({ reason }) => reason !== "run-token-mismatch"))
+    return failures;
+  try {
+    const live = new Set(
+      liveProcesses(await host.listProcesses()).map(({ pid }) => pid),
+    );
+    return failures.filter(
+      ({ pid, reason }) => reason === "run-token-mismatch" || live.has(pid),
+    );
+  } catch (cause: unknown) {
+    return failures.map((failure) =>
+      failure.reason === "run-token-mismatch"
+        ? failure
+        : {
+            ...failure,
+            diagnostic: `${failure.diagnostic}; process liveness recheck failed: ${errorMessage(cause)}`,
+          },
+    );
+  }
 };
 
 interface TokenOwnedProcessScan {
@@ -785,18 +806,19 @@ const scanTokenOwnedProcesses = async (
   const failures: ProcessOwnershipValidationFailure[] = [];
   if (captureBaseline !== undefined && host.processIdentities !== undefined) {
     const stableCandidates: ProcessTableEntry[] = [];
+    const identityFailures: ProcessOwnershipValidationFailure[] = [];
     for (const process of candidates) {
       if (candidateIdentities.get(process.pid)?.state === "readable") {
         stableCandidates.push(process);
         continue;
       }
-      if (!(await processIsGone(host, process.pid)))
-        failures.push({
-          pid: process.pid,
-          reason: "process-identity-unavailable",
-          diagnostic: "process identity was unavailable during token scan",
-        });
+      identityFailures.push({
+        pid: process.pid,
+        reason: "process-identity-unavailable",
+        diagnostic: "process identity was unavailable during token scan",
+      });
     }
+    failures.push(...(await recheckOwnershipFailures(identityFailures, host)));
     candidates = stableCandidates;
   }
   const owned: ProcessTableEntry[] = [];
@@ -882,6 +904,7 @@ const scanTokenOwnedProcesses = async (
   if (candidates.length > 0 && host.processIdentities !== undefined) {
     const afterRead = await host.processIdentities(candidates);
     const stillOwned: ProcessTableEntry[] = [];
+    const identityFailures: ProcessOwnershipValidationFailure[] = [];
     const ownedPids = new Set(owned.map(({ pid }) => pid));
     for (const process of candidates) {
       const before = candidateIdentities.get(process.pid);
@@ -894,14 +917,14 @@ const scanTokenOwnedProcesses = async (
         if (ownedPids.has(process.pid)) stillOwned.push(process);
         continue;
       }
-      if (await processIsGone(host, process.pid)) continue;
-      failures.push({
+      identityFailures.push({
         pid: process.pid,
         reason: "process-identity-unavailable",
         diagnostic:
           "process identity changed or became unavailable during token validation",
       });
     }
+    failures.push(...(await recheckOwnershipFailures(identityFailures, host)));
     return {
       owned: stillOwned,
       failures,
@@ -963,17 +986,3 @@ const settleUnreadableTokenCandidates = async (
   }
   return remaining;
 };
-
-const processIsGone = async (
-  host: ProcessOwnershipHost,
-  pid: number,
-): Promise<boolean> => {
-  try {
-    return !liveProcesses(await host.listProcesses()).some(
-      (process) => process.pid === pid,
-    );
-  } catch (cause: unknown) {
-    void cause;
-    return false;
-  }
-};

src/process/ProcessOwnership.validation.test.ts

@@ -115,6 +115,101 @@ describe("owned process-group cleanup validation: ownership and lineage", () =>
     });
     expect(signalGroup).not.toHaveBeenCalled();
   });
+});
+
+describe("owned process-group cleanup liveness rechecks", () => {
+  it("rechecks all unreadable members after the token phase in one fresh snapshot", async () => {
+    const members = [100, 101, 102, 103].map((pid) => ({
+      pid,
+      parentPid: pid === 100 ? 1 : 100,
+      processGroupId: 100,
+      state: "S",
+      command: "fixture",
+    }));
+    const tokenReads: number[] = [];
+    const listProcesses = vi.fn(async () => {
+      if (tokenReads.length > 0)
+        expect(tokenReads).toEqual([100, 101, 102, 103]);
+      return members;
+    });
+    const signalGroup = vi.fn();
+    const adapter: ProcessOwnershipHost = {
+      listProcesses,
+      environment: async (pid) => {
+        tokenReads.push(pid);
+        throw new Error(`unreadable ${pid}`);
+      },
+      signalGroup,
+    };
+    await expect(cleanupOwnedProcessGroup(ownership, adapter)).resolves.toEqual(
+      {
+        cleaned: false,
+        reason:
+          "process ownership token could not be read for 4 live process(es): other_unavailable=4; live candidates 100=other_unavailable, 101=other_unavailable, 102=other_unavailable, 103=other_unavailable",
+        failures: members.map(({ pid }) => ({
+          pid,
+          reason: "environment-unreadable",
+          diagnostic: `unreadable ${pid}`,
+        })),
+      },
+    );
+    expect(listProcesses).toHaveBeenCalledTimes(2);
+    expect(signalGroup).not.toHaveBeenCalled();
+  });
+
+  it.each([false, true])(
+    "retains mismatches and fails closed when the liveness recheck fails: %s",
+    async (recheckFails) => {
+      const members = [100, 101, 102].map((pid) => ({
+        pid,
+        parentPid: pid === 100 ? 1 : 100,
+        processGroupId: 100,
+        state: "S",
+        command: "fixture",
+      }));
+      const listProcesses = vi
+        .fn<ProcessOwnershipHost["listProcesses"]>()
+        .mockResolvedValueOnce(members);
+      if (recheckFails)
+        listProcesses.mockRejectedValue(new Error("process table denied"));
+      else listProcesses.mockResolvedValue([]);
+      const signalGroup = vi.fn();
+      const adapter: ProcessOwnershipHost = {
+        listProcesses,
+        environment: async (pid) => {
+          if (pid === 101) return { REA_PROCESS_RUN_ID: "other-run" };
+          throw new Error(`unreadable ${pid}`);
+        },
+        signalGroup,
+      };
+      await expect(
+        cleanupOwnedProcessGroup(ownership, adapter),
+      ).resolves.toEqual({
+        cleaned: false,
+        reason: "process tree contains an unowned or PID-reused process",
+        failures: recheckFails
+          ? [
+              {
+                pid: 100,
+                reason: "environment-unreadable",
+                diagnostic:
+                  "unreadable 100; process liveness recheck failed: process table denied",
+              },
+              { pid: 101, reason: "run-token-mismatch" },
+              {
+                pid: 102,
+                reason: "environment-unreadable",
+                diagnostic:
+                  "unreadable 102; process liveness recheck failed: process table denied",
+              },
+            ]
+          : [{ pid: 101, reason: "run-token-mismatch" }],
+      });
+      expect(listProcesses).toHaveBeenCalledTimes(2);
+      expect(signalGroup).not.toHaveBeenCalled();
+    },
+  );
+
   it("accepts a member that exits during ownership revalidation", async () => {
     const liveLauncher = {
       pid: 100,