fix(captures): validate packet read scope before paging

src/browser/ElectronProcessCleanup.test.ts

@@ -61,33 +61,30 @@ describe.skipIf(process.platform === "win32")(
         processIdentities: (members) =>
           Promise.resolve(
             new Map(
-              members.map(({ pid }) => {
-                const current = rows.find((entry) => entry.pid === pid);
-                return [
-                  pid,
-                  current === undefined
-                    ? {
-                        state: "absent" as const,
-                        reason:
-                          process.platform === "darwin"
-                            ? ("not-in-snapshot" as const)
-                            : ("not-found-in-caller-pid-namespace" as const),
-                      }
-                    : {
-                        state: "readable" as const,
-                        identity: `start-${String(pid)}`,
-                        lifecycle: {
-                          state: "observed" as const,
-                          lifecycle: "non-zombie" as const,
-                        },
-                        relation: {
-                          state: "observed" as const,
-                          parentPid: current.parentPid,
-                          processGroupId: current.processGroupId,
-                        },
+              members.map((member) => [
+                member.pid,
+                withholdToken && member.pid === 100
+                  ? {
+                      state: "absent" as const,
+                      reason:
+                        process.platform === "darwin"
+                          ? ("not-in-snapshot" as const)
+                          : ("not-found-in-caller-pid-namespace" as const),
+                    }
+                  : {
+                      state: "readable" as const,
+                      identity: `start-${String(member.pid)}`,
+                      lifecycle: {
+                        state: "observed" as const,
+                        lifecycle: "non-zombie" as const,
                       },
-                ];
-              }),
+                      relation: {
+                        state: "observed" as const,
+                        parentPid: member.parentPid,
+                        processGroupId: member.processGroupId,
+                      },
+                    },
+              ]),
             ),
           ),
         signalGroup: vi.fn(() => {

src/captures/TsharkCaptureProvider.ts

@@ -369,7 +369,10 @@ export class TsharkCaptureProvider implements PacketCapturePort {
         resourcePath = "path" in input ? input.path : command;
         // TShark resolves `-e` aliases before emitting canonical field names;
         // the command arguments, not output names, own field selection.
-        const matchingPackets = parsePacketOutput(output.stdout.text);
+        const matchingPackets = parsePacketOutput(
+          output.stdout.text,
+          input.packet_limit,
+        );
         const start = Math.min(input.match_offset, matchingPackets.length);
         const end =
           input.match_limit === null

src/captures/parsePacketOutput.ts

@@ -6,7 +6,7 @@ import { packetFieldsSchema } from "../domain/captures/packetCapture.js";
 const producerSchema = z.array(
   z.object({ _source: z.object({ layers: packetFieldsSchema }) }),
 );
-export const parsePacketOutput = (text: string) => {
+export const parsePacketOutput = (text: string, packetLimit: number | null) => {
   const decoded: unknown = JSON.parse(text);
   const packets = producerSchema
     .parse(decoded)
@@ -19,6 +19,11 @@ export const parsePacketOutput = (text: string) => {
         "inspect_packet_capture",
         "TShark returned unordered frames.",
       );
+    if (packetLimit !== null && frame > BigInt(packetLimit))
+      throw new AnalysisOutputError(
+        "inspect_packet_capture",
+        "TShark returned a frame beyond the selected input packet limit.",
+      );
     previous = frame;
   }
   return packets;

src/process/DarwinHelperCache.ownership.test.ts

@@ -1,92 +0,0 @@
-import { writeFile, type FileHandle } from "node:fs/promises";
-import { join } from "node:path";
-import { expect, it, vi } from "vitest";
-
-import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js";
-import { ProviderCleanupError } from "../domain/providerCleanupError.js";
-import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js";
-
-const fixture = vi.hoisted(() => ({
-  owners: [] as OwnedFileHandle[],
-  next: 0,
-}));
-vi.mock("../filesystem/OwnedRegularFile.js", async (importOriginal) => ({
-  ...(await importOriginal<
-    typeof import("../filesystem/OwnedRegularFile.js")
-  >()),
-  withOwnedRegularFile: async (
-    _path: string,
-    options: { retainFailedClose: (owner: OwnedFileHandle) => void },
-  ) => {
-    const owner = fixture.owners[fixture.next++];
-    if (owner === undefined) throw new Error("Unexpected cache file read");
-    try {
-      await owner.close();
-    } catch (cause: unknown) {
-      options.retainFailedClose(owner);
-      throw cause;
-    }
-    throw new Error("Expected controlled close failure");
-  },
-}));
-vi.mock("./ExecFileOutput.js", async (importOriginal) => ({
-  ...(await importOriginal<typeof import("./ExecFileOutput.js")>()),
-  execFileOutput: async (command: string) => {
-    if (command !== "/bin/ls") throw new Error("Unexpected command");
-    return { stdout: "", stderr: "", stdoutBytes: 0, stderrBytes: 0 };
-  },
-}));
-
-import { createDarwinHelperCache } from "./DarwinHelperCache.js";
-
-it.skipIf(process.platform === "win32")(
-  "retains distinct failed-close owners after both descriptors become invalidated",
-  async () => {
-    const failures = [
-      new Error("first native close failed"),
-      new Error("second native close failed"),
-    ];
-    fixture.owners = failures.map((failure, index) => {
-      const handle = {
-        fd: index + 40,
-        close: async () => {
-          handle.fd = -1;
-          throw failure;
-        },
-      };
-      return new OwnedFileHandle(handle as unknown as FileHandle);
-    });
-    fixture.next = 0;
-    const closes = fixture.owners.map((owner) => vi.spyOn(owner, "close"));
-    const root = await createTestTempDirectory("rea-darwin-cache-owners-");
-    const compiled = join(root, "compiled");
-    await writeFile(compiled, "fixture helper");
-    const cache = createDarwinHelperCache({
-      directory: join(root, "cache"),
-      sources: [],
-      xcrun: "/unused",
-      environment: {},
-    });
-    expect(await cache.publish("a".repeat(64), compiled)).toBeUndefined();
-    expect(await cache.publish("b".repeat(64), compiled)).toBeUndefined();
-    expect(fixture.owners.map(({ handle }) => handle.fd)).toEqual([-1, -1]);
-    const failure = await cache.close().catch((cause: unknown) => cause);
-    if (!(failure instanceof ProviderCleanupError))
-      throw new Error("Expected retained cache cleanup failure");
-    expect(closes.map(({ mock }) => mock.calls.length)).toEqual([2, 2]);
-    expect(failure.cleanupIncomplete).toBe(true);
-    if (!(failure.cause instanceof AggregateError))
-      throw new Error("Expected exact aggregate cleanup causes");
-    expect(failure.cause.errors).toEqual(failures);
-    const retried = await cache.close().catch((cause: unknown) => cause);
-    expect(closes.map(({ mock }) => mock.calls.length)).toEqual([3, 3]);
-    expect(retried).toMatchObject({ cleanupIncomplete: true });
-    if (
-      !(retried instanceof ProviderCleanupError) ||
-      !(retried.cause instanceof AggregateError)
-    )
-      throw new Error("Expected retained aggregate cleanup causes on retry");
-    expect(retried.cause.errors).toEqual(failures);
-    expect(failure.cleanupIncomplete).toBe(true);
-  },
-);

src/process/DarwinHelperCache.ts

@@ -168,7 +168,7 @@ export const createDarwinHelperCache = (options: {
 }) => {
   const uid = process.getuid?.();
   // Failed closes and removals stay owned until close() retries them.
-  const retained = new Map<string | OwnedFileHandle, () => Promise<void>>();
+  const retained = new Map<string, () => Promise<void>>();
   // The first unresolved close failure; later ones depend on it.
   let cleanupAnchor: ProviderCleanupError | undefined;
   let key: Promise<string> | undefined;
@@ -271,7 +271,7 @@ export const createDarwinHelperCache = (options: {
   };
 
   const retainClose = (owner: OwnedFileHandle): void => {
-    retained.set(owner, () => owner.close());
+    retained.set(`file handle ${String(owner.handle.fd)}`, () => owner.close());
   };
 
   /** Hash the opened descriptor itself; FIFOs and links never block or pass. */
@@ -431,19 +431,13 @@ export const createDarwinHelperCache = (options: {
   /** Retry every retained close or removal; unresolved ones stay owned. */
   const close = async (): Promise<void> => {
     const failed: string[] = [];
-    const causes: unknown[] = [];
     // Deleting the visited entry while iterating a Map is well defined.
     for (const [resource, cleanup] of retained) {
       try {
         await cleanup();
         retained.delete(resource);
-      } catch (cause: unknown) {
-        failed.push(
-          typeof resource === "string"
-            ? resource
-            : `file handle ${String(resource.handle.fd)}`,
-        );
-        causes.push(cause);
+      } catch {
+        failed.push(resource);
       }
     }
     if (failed.length > 0) {
@@ -454,13 +448,7 @@ export const createDarwinHelperCache = (options: {
           reason:
             "Darwin ownership-helper cache resources remain owned until cleanup succeeds.",
         },
-        {
-          cause: new AggregateError(
-            causes,
-            "Darwin helper cache cleanup failed",
-          ),
-          operation: "prepare_owned_process",
-        },
+        { operation: "prepare_owned_process" },
       );
       // Every returned failure clears once a later retry releases them all.
       if (cleanupAnchor === undefined) cleanupAnchor = failure;

src/process/DarwinProcessRunTokenReader.finalization.test.ts

@@ -172,7 +172,7 @@ it("decodes a successful native snapshot absence as a distinct identity fact", a
     environment: {},
   });
   try {
-    const processes = [
+    const operation = reader.identities([
       {
         pid: 123,
         parentPid: 1,
@@ -180,8 +180,7 @@ it("decodes a successful native snapshot absence as a distinct identity fact", a
         state: "S",
         command: "fixture",
       },
-    ];
-    const operation = reader.identities(processes);
+    ]);
     await command.waitUntilStarted();
     command.release();
     await expect(operation).resolves.toEqual(
@@ -212,17 +211,26 @@ it.each(["read", "identities"] as const)(
       await Promise.allSettled([operation, closing, reader.close()]);
     });
 
-    const processes = [
-      {
-        pid: 123,
-        parentPid: 1,
-        processGroupId: 123,
-        state: "S",
-        command: "fixture",
-      },
-    ];
     operation =
-      mode === "read" ? reader.read(processes) : reader.identities(processes);
+      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;
@@ -265,7 +273,25 @@ it.each(["read", "identities"] as const)(
       ]),
     );
     await expect(
-      mode === "read" ? reader.read(processes) : reader.identities(processes),
+      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");
   },
 );

src/process/DarwinProcessRunTokenReader.ts

@@ -776,7 +776,7 @@ export const createDarwinProcessRunTokenReader = (
   };
 
   const identities = async (
-    processes: readonly Pick<ProcessTableEntry, "pid">[],
+    processes: readonly ProcessTableEntry[],
     signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessIdentityObservation>> => {
     if (processes.length === 0) return new Map();

src/process/OwnedLease.test.ts

@@ -9,7 +9,6 @@ import { afterEach, describe, expect, it, onTestFinished, vi } from "vitest";
 import { AnalysisCleanupLease } from "../artifacts/AnalysisCleanupLease.js";
 import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
 import { HopperStartError } from "../domain/hopperErrors.js";
-import { createDeferred } from "../../tests/fixtures/binarySession.js";
 // Unix socket paths have a short platform limit; the canonical system temp path can exceed it.
 const createTestTempDirectory = async (prefix: string): Promise<string> => {
   const directory = await mkdtemp(join("/tmp", prefix));
@@ -94,49 +93,6 @@ afterEach(() => {
 describe.skipIf(process.platform === "win32")(
   "accepted lease connections",
   () => {
-    it("joins concurrent release callers until the listening server finishes closing", async () => {
-      const directory = await createTestTempDirectory("rea-release-join-");
-      const acquired = await acquireOwnedLease({
-        identity: { test: "concurrent-release" },
-        directory,
-        runId: "concurrent-release",
-        failure: (failure) => new Error(failure.kind),
-      });
-      if (!acquired.acquired) throw new Error("expected acquired lease");
-      onTestFinished(() => acquired.lease.release());
-      const started = createDeferred<void>();
-      const finish = createDeferred<void>();
-      const close = Server.prototype.close;
-      vi.spyOn(Server.prototype, "close").mockImplementationOnce(function (
-        this: Server,
-        callback,
-      ) {
-        return close.call(this, (cause) => {
-          started.resolve();
-          void finish.promise.then(() => callback?.(cause));
-        });
-      });
-      let firstFinished = false;
-      let secondFinished = false;
-      const first = acquired.lease.release().then(() => {
-        firstFinished = true;
-      });
-      await started.promise;
-      const second = acquired.lease.release().then(() => {
-        secondFinished = true;
-      });
-      try {
-        await new Promise<void>((resolve) => setImmediate(resolve));
-        expect(firstFinished).toBe(false);
-        expect(secondFinished).toBe(false);
-      } finally {
-        finish.resolve();
-        await Promise.all([first, second]);
-      }
-      expect(firstFinished).toBe(true);
-      expect(secondFinished).toBe(true);
-    });
-
     it.each([false, true])(
       "closes a held peer and preserves retry after failure=%s",
       async (failClose) => {

src/process/OwnedLease.ts

@@ -240,22 +240,16 @@ const tryAcquireLease = async (
       : { acquired: false, owner: existingOwner };
   }
   let released = false;
-  let releasing: Promise<void> | undefined;
   const lease: OwnedLease = {
     socketPath,
-    release: () => {
-      if (released) return Promise.resolve();
-      releasing ??= (async () => {
-        // Owner replies do not require peers to close their writable half. End
-        // every accepted connection before awaiting the listening server.
-        for (const socket of connections) socket.destroy();
-        // Node removes the bound socket; never unlink a later owner's socket.
-        await closeServer(server);
-        released = true;
-      })().finally(() => {
-        releasing = undefined;
-      });
-      return releasing;
+    release: async () => {
+      if (released) return;
+      // Owner replies do not require peers to close their writable half. End
+      // every accepted connection before awaiting the listening server.
+      for (const socket of connections) socket.destroy();
+      // Node removes the bound socket; never unlink a later owner's socket.
+      await closeServer(server);
+      released = true;
     },
   };
   if (securePath)

src/process/OwnedMcpStdioTransport.ts

@@ -17,10 +17,6 @@ import {
 } from "./ProviderProcess.js";
 import { cleanupOwnedProcessGroup } from "./ProcessOwnership.js";
 import { ProviderCleanupError } from "../domain/providerCleanupError.js";
-import {
-  AnalysisCleanupLease,
-  AnalysisCleanupLeaseGroup,
-} from "../artifacts/AnalysisCleanupLease.js";
 
 /** Injectable acquisition seam; production owns a POSIX group or Windows job. */
 export type OwnedMcpLauncher = (
@@ -47,10 +43,7 @@ export class OwnedMcpStdioTransport implements Transport {
   #bytes = 0;
   #started = false;
   #closed = false;
-  readonly #cleanup = new AnalysisCleanupLease(
-    () => this.#stop(),
-    [this.runId],
-  );
+  #closePromise: Promise<void> | undefined;
   #starting: Promise<void> | undefined;
   #failure: OwnedMcpTransportFailure | undefined;
 
@@ -146,8 +139,11 @@ export class OwnedMcpStdioTransport implements Transport {
 
   /** Verifies owned process exit before declaring the transport closed. */
   close(): Promise<void> {
-    this.#closed = true;
-    return this.#cleanup.release();
+    this.#closePromise ??= this.#stop().catch((cause: unknown) => {
+      this.#closePromise = undefined;
+      throw cause;
+    });
+    return this.#closePromise;
   }
 
   /** Keep the interruption reason distinct from errors produced by its teardown. */
@@ -186,6 +182,7 @@ export class OwnedMcpStdioTransport implements Transport {
   }
 
   async #stop(): Promise<void> {
+    this.#closed = true;
     this.#frameBuffer = Buffer.alloc(0);
     this.#frameLength = 0;
     // A cancellation may race acquisition. Join it before inspecting ownership.
@@ -216,18 +213,11 @@ export class OwnedMcpStdioTransport implements Transport {
     await inputClosed;
     input?.off("error", this.#error);
     this.onclose?.();
-    if (stopped?.status === "incomplete") {
-      const failure = new ProviderCleanupError(
-        this.settings.provider,
-        [this.runId],
-        {
-          reason: stopped.reason,
-          stderr: this.diagnostics()?.stderr ?? "",
-        },
-      );
-      new AnalysisCleanupLeaseGroup(failure, [this.#cleanup]);
-      throw failure;
-    }
+    if (stopped?.status === "incomplete")
+      throw new ProviderCleanupError(this.settings.provider, [this.runId], {
+        reason: stopped.reason,
+        stderr: this.diagnostics()?.stderr ?? "",
+      });
   }
 
   readonly #error = (error: Error): void => {

src/process/OwnedWorkerRuntime.ts

@@ -45,21 +45,8 @@ export class WorkerCleanupRegistry {
   }
 
   async retry(except?: RetainedWorkerCleanup): Promise<void> {
-    const failures: unknown[] = [];
-    for (const owner of [...this.#pending]) {
-      if (owner === except) continue;
-      try {
-        await owner.retryRetainedCleanup();
-      } catch (cause: unknown) {
-        failures.push(cause);
-      }
-    }
-    if (failures.length === 1) throw failures[0];
-    if (failures.length > 1)
-      throw new AggregateError(
-        failures,
-        "Owned worker cleanup remains incomplete",
-      );
+    for (const owner of [...this.#pending])
+      if (owner !== except) await owner.retryRetainedCleanup();
   }
 }
 

src/process/ProcessOwnership.exitScope.test.ts

@@ -95,10 +95,7 @@ it.each([
         ),
       processIdentities: async (selected) =>
         new Map<number, ProcessIdentityObservation>(
-          selected.map(({ pid }) => {
-            const process = [parent, child].find((entry) => entry.pid === pid);
-            if (process === undefined)
-              return [pid, { state: "absent", reason: "not-in-snapshot" }];
+          selected.map((process) => {
             if (process.pid === parent.pid && ++parentReads > 2) {
               reparented = true;
               return [
@@ -173,15 +170,10 @@ it("shares native exit waiting before signaling healthy groups in one cleanup at
     processIdentities: async (selected) => {
       clock += 25;
       return new Map<number, ProcessIdentityObservation>(
-        selected.map(({ pid }) => {
-          const process = rows.find((entry) => entry.pid === pid);
-          return [
-            pid,
-            process === undefined
-              ? { state: "absent", reason: "not-in-snapshot" }
-              : identity(process, process.pid !== root.pid),
-          ];
-        }),
+        selected.map((process) => [
+          process.pid,
+          identity(process, process.pid !== root.pid),
+        ]),
       );
     },
     signalGroup: vi.fn((group) => {

src/process/ProcessOwnership.identity.test.ts

@@ -20,126 +20,6 @@ const observedIdentity = (entry: ProcessTableEntry, identity: string) => ({
   },
 });
 
-it.each([
-  "live",
-  "absent",
-  "unavailable",
-  "reused-foreign",
-  "omitted-command",
-  "listed-command",
-  "refresh-command",
-] as const)(
-  "revalidates a retained owner with %s observations during non-sweep cleanup",
-  async (nativeState) => {
-    const row: ProcessTableEntry = {
-      pid: 101,
-      parentPid: 1,
-      processGroupId: 101,
-      state: "S",
-      command: "retained-owner",
-    };
-    const evidenceOwner = new ProcessOwnershipEvidenceOwner();
-    const identity = observedIdentity(row, "start-101");
-    evidenceOwner.reconcile(
-      [
-        {
-          process: row,
-          state: "same-generation",
-          identityBefore: identity,
-          identityAfter: identity,
-          token: { state: "readable", runId: "run-token" },
-        },
-      ],
-      {
-        runId: "run-token",
-        listedRows: 1,
-        listedPids: new Set([row.pid]),
-        leaderPid: 100,
-        processGroupIds: new Set([100]),
-      },
-    );
-    expect(evidenceOwner.candidates).toMatchObject([
-      { pid: row.pid, applicability: "token-verified" },
-    ]);
-    let signaled = false;
-    let tableReads = 0;
-    const signalGroup = vi.fn(() => {
-      signaled = true;
-    });
-    const host: ProcessOwnershipHost = {
-      listProcesses: async () => {
-        tableReads += 1;
-        return nativeState === "listed-command" ||
-          (nativeState === "refresh-command" && tableReads === 1)
-          ? [row]
-          : [];
-      },
-      environment: async () => ({
-        REA_PROCESS_RUN_ID:
-          nativeState === "reused-foreign" ? "foreign" : "run-token",
-      }),
-      processIdentities: async (entries) =>
-        new Map(
-          entries.map(({ pid }) => [
-            pid,
-            signaled || nativeState === "absent"
-              ? { state: "absent" as const, reason: "not-in-snapshot" as const }
-              : nativeState === "unavailable"
-                ? {
-                    state: "unavailable" as const,
-                    reason: "native read denied",
-                  }
-                : observedIdentity(
-                    row,
-                    nativeState === "reused-foreign"
-                      ? "replacement"
-                      : "start-101",
-                  ),
-          ]),
-        ),
-      signalGroup,
-    };
-    const result = await cleanupOwnedProcessGroup(
-      {
-        runId: "run-token",
-        leaderPid: 100,
-        processGroupId: 100,
-        evidenceOwner,
-        ...(nativeState === "omitted-command" ||
-        nativeState === "listed-command" ||
-        nativeState === "refresh-command"
-          ? {
-              leaderPid: row.pid,
-              processGroupId: row.processGroupId,
-              expectedCommand: row.command,
-            }
-          : {}),
-      },
-      host,
-    );
-    if (nativeState === "live" || nativeState === "listed-command") {
-      expect(result).toMatchObject({ cleaned: true, signaled: true });
-      expect(signalGroup).toHaveBeenCalledExactlyOnceWith(101, "SIGKILL");
-    } else {
-      expect(signalGroup).not.toHaveBeenCalled();
-      expect(result).toMatchObject(
-        nativeState === "unavailable" ||
-          nativeState === "omitted-command" ||
-          nativeState === "refresh-command"
-          ? { cleaned: false }
-          : { cleaned: true, signaled: false },
-      );
-      expect(evidenceOwner.candidates).toEqual(
-        nativeState === "unavailable" ||
-          nativeState === "omitted-command" ||
-          nativeState === "refresh-command"
-          ? [expect.objectContaining({ pid: row.pid })]
-          : [],
-      );
-    }
-  },
-);
-
 it("keeps an unresolved earlier generation unknown when a later unrelated process has a foreign marker", async () => {
   const row: ProcessTableEntry = {
     pid: 920,
@@ -159,7 +39,7 @@ it("keeps an unresolved earlier generation unknown when a later unrelated proces
         entries.map((entry) => [
           entry.pid,
           readable
-            ? observedIdentity(row, "later-generation")
+            ? observedIdentity(entry, "later-generation")
             : { state: "unavailable" as const, reason: "identity unavailable" },
         ]),
       ),
@@ -207,7 +87,7 @@ it.each([undefined, "run-token"])(
           new Map(
             entries.map((entry) => [
               entry.pid,
-              observedIdentity(row, "current-start"),
+              observedIdentity(entry, "current-start"),
             ]),
           ),
         ),
@@ -300,7 +180,7 @@ it("validates and cleans an exact-token owner without a capture baseline", async
           entries.map((entry) => [
             entry.pid,
             live
-              ? observedIdentity(process, "owner-start")
+              ? observedIdentity(entry, "owner-start")
               : {
                   state: "absent" as const,
                   reason: "not-in-snapshot" as const,
@@ -359,7 +239,7 @@ it
               entries.map((entry) => [
                 entry.pid,
                 {
-                  ...observedIdentity(row, "reused-current-generation"),
+                  ...observedIdentity(entry, "reused-current-generation"),
                   effectiveUid,
                 },
               ]),
@@ -419,7 +299,7 @@ it
               entries.map((entry) => [
                 entry.pid,
                 {
-                  ...observedIdentity(row, "current-generation"),
+                  ...observedIdentity(entry, "current-generation"),
                   effectiveUid: process.getuid!(),
                   ...(environmentRestriction === undefined
                     ? {}
@@ -485,7 +365,7 @@ it.each(["omitted", "throws"] as const)(
           new Map(
             entries.map((entry) => [
               entry.pid,
-              observedIdentity(row, "current-generation"),
+              observedIdentity(entry, "current-generation"),
             ]),
           ),
         );
@@ -535,7 +415,7 @@ it("keeps a matching-token replacement through omission and retries cleanup only
             ? { state: "absent" as const, reason: "not-in-snapshot" as const }
             : identityUnavailable
               ? { state: "unavailable" as const, reason: "native read denied" }
-              : observedIdentity(row, replaced ? "new-owner" : "old-owner"),
+              : observedIdentity(entry, replaced ? "new-owner" : "old-owner"),
         ]),
       ),
     runTokens: async (entries) => {
@@ -736,7 +616,7 @@ it("requires a scoped current relation before exempting a withheld token", async
             entries.map((entry) => [
               entry.pid,
               nativeRelation
-                ? observedIdentity(row, "opaque-generation")
+                ? observedIdentity(entry, "opaque-generation")
                 : {
                     state: "readable" as const,
                     identity: "opaque-generation",
@@ -1067,7 +947,7 @@ describe("capture-local process identity baselines", () => {
               entry.pid === 902
                 ? { state: "unavailable", reason: "fixture" }
                 : observedIdentity(
-                    processes.find(({ pid }) => pid === entry.pid)!,
+                    entry,
                     entry.pid === 899 ? "stable-start" : "reused-start",
                   ),
             ]),
@@ -1131,7 +1011,7 @@ describe("capture-local process identity baselines", () => {
           new Map(
             entries.map((entry) => [
               entry.pid,
-              observedIdentity(process, identity),
+              observedIdentity(entry, identity),
             ]),
           ),
         );
@@ -1368,7 +1248,7 @@ it("revalidates identity after the final environment token read", async () => {
         new Map(
           entries.map((entry) => [
             entry.pid,
-            observedIdentity(process, identity),
+            observedIdentity(entry, identity),
           ]),
         ),
       );
@@ -1417,10 +1297,7 @@ it("fails closed when candidate PIDs change identity during token reading", asyn
         new Map(
           entries.map((entry) => [
             entry.pid,
-            observedIdentity(
-              processes.find(({ pid }) => pid === entry.pid)!,
-              identity,
-            ),
+            observedIdentity(entry, identity),
           ]),
         ),
       );
@@ -1484,7 +1361,7 @@ it("uses the launcher's native relation instead of stale process-table fields",
           members.map((member) => [
             member.pid,
             observedIdentity(
-              { ...row, parentPid: 1, processGroupId: 100 },
+              { ...member, parentPid: 1, processGroupId: 100 },
               "start-100",
             ),
           ]),
@@ -1574,7 +1451,7 @@ it("refuses to signal when PID identity changes after token validation", async (
         new Map(
           entries.map((entry) => [
             entry.pid,
-            observedIdentity(process, identity),
+            observedIdentity(entry, identity),
           ]),
         ),
       );

src/process/ProcessOwnership.lineage.test.ts

@@ -58,17 +58,8 @@ it.each(
         runId === undefined ? {} : { REA_PROCESS_RUN_ID: runId },
       ...(native
         ? {
-            processIdentities: async (
-              rows: readonly Pick<ProcessTableEntry, "pid">[],
-            ) =>
-              new Map(
-                rows.map(({ pid }) => {
-                  const row = [launcher].find(
-                    (candidate) => candidate.pid === pid,
-                  )!;
-                  return [row.pid, nativeIdentity(row)];
-                }),
-              ),
+            processIdentities: async (rows: readonly ProcessTableEntry[]) =>
+              new Map(rows.map((row) => [row.pid, nativeIdentity(row)])),
           }
         : {}),
       signalGroup,
@@ -161,12 +152,7 @@ it.each(
         ),
       processIdentities: async (rows) =>
         new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [
-              nativeLauncher,
-              observing ? foreign : { ...foreign, parentPid: 1 },
-              independentlyOwned,
-            ].find((candidate) => candidate.pid === pid)!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             if (!observing && row.pid !== foreign.pid)
               return [row.pid, { state: "absent", reason: "not-in-snapshot" }];
             if (
@@ -231,14 +217,7 @@ it("preserves independently token-owned ancestry when the launcher identity mism
         ]),
       ),
     processIdentities: async (rows) =>
-      new Map(
-        rows.map(({ pid }) => {
-          const row = [foreignLauncher, ownedParent, unreadableChild].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [row.pid, nativeIdentity(row)];
-        }),
-      ),
+      new Map(rows.map((row) => [row.pid, nativeIdentity(row)])),
     signalGroup: vi.fn(),
   };
 
@@ -288,24 +267,19 @@ it.each(["parent", "group"] as const)(
       processIdentities: async (rows) => {
         identityReads += 1;
         return new Map(
-          rows.map(({ pid }) => {
-            const row = [launcher, detached].find(
-              (candidate) => candidate.pid === pid,
-            )!;
-            return [
-              row.pid,
-              nativeIdentity(
-                row.pid === launcher.pid && identityReads > 1
-                  ? {
-                      ...row,
-                      parentPid: mismatch === "parent" ? 999 : row.parentPid,
-                      processGroupId:
-                        mismatch === "group" ? 999 : row.processGroupId,
-                    }
-                  : row,
-              ),
-            ];
-          }),
+          rows.map((row) => [
+            row.pid,
+            nativeIdentity(
+              row.pid === launcher.pid && identityReads > 1
+                ? {
+                    ...row,
+                    parentPid: mismatch === "parent" ? 999 : row.parentPid,
+                    processGroupId:
+                      mismatch === "group" ? 999 : row.processGroupId,
+                  }
+                : row,
+            ),
+          ]),
         );
       },
       signalGroup: vi.fn(),
@@ -374,10 +348,7 @@ it.each([
       processIdentities: async (rows) => {
         identityReads.push(rows.map(({ pid }) => pid));
         return new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child, grandchild].find(
-              (candidate) => candidate.pid === pid,
-            )!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             if (
               phase === "exited" ||
               (phase === "omitted" && row.pid !== child.pid)
@@ -468,17 +439,12 @@ it("retains a live native descendant behind an exited table ancestor", async ()
     },
     processIdentities: async (rows) =>
       new Map(
-        rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-          const row = [launcher, child, grandchild].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [
-            row.pid,
-            row.pid === child.pid
-              ? { state: "absent", reason: "not-in-snapshot" }
-              : nativeIdentity(row),
-          ];
-        }),
+        rows.map((row): [number, ProcessIdentityObservation] => [
+          row.pid,
+          row.pid === child.pid
+            ? { state: "absent", reason: "not-in-snapshot" }
+            : nativeIdentity(row),
+        ]),
       ),
     signalGroup: vi.fn(),
   };
@@ -527,26 +493,21 @@ it.each(["unavailable-relation", "absent"] as const)(
         ),
       processIdentities: async (rows) =>
         new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child, detached].find(
-              (candidate) => candidate.pid === pid,
-            )!;
-            return [
-              row.pid,
-              (omitted && row.pid === launcher.pid) ||
-              (ancestor === "absent" && row.pid === child.pid)
-                ? { state: "absent", reason: "not-in-snapshot" }
-                : row.pid === child.pid
-                  ? {
-                      ...nativeIdentity(row),
-                      relation: {
-                        state: "unavailable",
-                        reason: "relation denied",
-                      },
-                    }
-                  : nativeIdentity(row),
-            ];
-          }),
+          rows.map((row): [number, ProcessIdentityObservation] => [
+            row.pid,
+            (omitted && row.pid === launcher.pid) ||
+            (ancestor === "absent" && row.pid === child.pid)
+              ? { state: "absent", reason: "not-in-snapshot" }
+              : row.pid === child.pid
+                ? {
+                    ...nativeIdentity(row),
+                    relation: {
+                      state: "unavailable",
+                      reason: "relation denied",
+                    },
+                  }
+                : nativeIdentity(row),
+          ]),
         ),
       signalGroup: vi.fn(),
     };
@@ -586,17 +547,12 @@ it("retains a verified child before rejecting an unreadable launcher identity",
     environment: async () => ({ REA_PROCESS_RUN_ID: "run" }),
     processIdentities: async (rows) =>
       new Map(
-        rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-          const row = [launcher, child].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [
-            row.pid,
-            row.pid === launcher.pid
-              ? { state: "unavailable", reason: "native identity denied" }
-              : nativeIdentity(row),
-          ];
-        }),
+        rows.map((row): [number, ProcessIdentityObservation] => [
+          row.pid,
+          row.pid === launcher.pid
+            ? { state: "unavailable", reason: "native identity denied" }
+            : nativeIdentity(row),
+        ]),
       ),
     signalGroup: vi.fn(),
   };
@@ -632,19 +588,14 @@ it("uses a readable native relation to exclude a stale table descendant", async
     },
     processIdentities: async (rows) =>
       new Map(
-        rows.map(({ pid }) => {
-          const row = [launcher, child].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [
-            row.pid,
-            nativeIdentity(
-              row.pid === child.pid
-                ? { ...row, parentPid: 1, processGroupId: 200 }
-                : row,
-            ),
-          ];
-        }),
+        rows.map((row) => [
+          row.pid,
+          nativeIdentity(
+            row.pid === child.pid
+              ? { ...row, parentPid: 1, processGroupId: 200 }
+              : row,
+          ),
+        ]),
       ),
     signalGroup: vi.fn(),
   };
@@ -675,25 +626,17 @@ it("does not retain table scope after a native relation resolves as unrelated",
     processIdentities: async (rows) => {
       identityReads += 1;
       return new Map(
-        rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-          const row = [launcher, child].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [
-            row.pid,
-            row.pid !== child.pid
-              ? nativeIdentity(row)
-              : identityReads === 1
-                ? {
-                    ...nativeIdentity(row),
-                    relation: {
-                      state: "unavailable",
-                      reason: "relation denied",
-                    },
-                  }
-                : nativeIdentity({ ...row, parentPid: 1, processGroupId: 200 }),
-          ];
-        }),
+        rows.map((row): [number, ProcessIdentityObservation] => [
+          row.pid,
+          row.pid !== child.pid
+            ? nativeIdentity(row)
+            : identityReads === 1
+              ? {
+                  ...nativeIdentity(row),
+                  relation: { state: "unavailable", reason: "relation denied" },
+                }
+              : nativeIdentity({ ...row, parentPid: 1, processGroupId: 200 }),
+        ]),
       );
     },
     signalGroup: vi.fn(),
@@ -735,10 +678,7 @@ it.each(["none", "unresolved", "known"] as const)(
       processIdentities: async (rows) => {
         identityReads += 1;
         return new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child].find(
-              (candidate) => candidate.pid === pid,
-            )!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             if (row.pid === launcher.pid)
               return [
                 row.pid,
@@ -759,10 +699,7 @@ it.each(["none", "unresolved", "known"] as const)(
                 row.pid,
                 {
                   ...nativeIdentity(row),
-                  relation: {
-                    state: "unavailable",
-                    reason: "relation denied",
-                  },
+                  relation: { state: "unavailable", reason: "relation denied" },
                 },
               ];
             return [
@@ -837,10 +774,7 @@ it.each(["replaced", "relation-unavailable-replaced", "unavailable"] as const)(
       processIdentities: async (rows) => {
         identityReads += 1;
         return new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child, detached].find(
-              (candidate) => candidate.pid === pid,
-            )!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             const replacedAncestor =
               row.pid === child.pid && ancestorGeneration !== "unavailable";
             if (
@@ -952,24 +886,19 @@ it("keeps an exiting descendant whose termination cannot be confirmed", async ()
     processIdentities: async (rows) => {
       identityReads += 1;
       return new Map(
-        rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-          const row = [launcher, child].find(
-            (candidate) => candidate.pid === pid,
-          )!;
-          return [
-            row.pid,
-            row.pid === launcher.pid
-              ? omitted
-                ? { state: "absent", reason: "not-in-snapshot" }
-                : nativeIdentity(row)
-              : identityReads === 1
-                ? {
-                    ...nativeIdentity(row),
-                    lifecycle: { state: "observed", lifecycle: "exiting" },
-                  }
-                : { state: "unavailable", reason: "identity denied" },
-          ];
-        }),
+        rows.map((row): [number, ProcessIdentityObservation] => [
+          row.pid,
+          row.pid === launcher.pid
+            ? omitted
+              ? { state: "absent", reason: "not-in-snapshot" }
+              : nativeIdentity(row)
+            : identityReads === 1
+              ? {
+                  ...nativeIdentity(row),
+                  lifecycle: { state: "observed", lifecycle: "exiting" },
+                }
+              : { state: "unavailable", reason: "identity denied" },
+        ]),
       );
     },
     signalGroup,
@@ -1041,10 +970,7 @@ it.each(
       processIdentities: async (rows) => {
         identityReads += 1;
         return new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child].find(
-              (candidate) => candidate.pid === pid,
-            )!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             if (row.pid !== stale.pid)
               return [
                 row.pid,
@@ -1134,19 +1060,13 @@ it.each(["absent", "zombie", "dead"] as const)(
       processIdentities: async (rows) => {
         identityReads += 1;
         return new Map(
-          rows.map(({ pid }): [number, ProcessIdentityObservation] => {
-            const row = [launcher, child, detached].find(
-              (candidate) => candidate.pid === pid,
-            )!;
+          rows.map((row): [number, ProcessIdentityObservation] => {
             if (row.pid === child.pid)
               return [
                 row.pid,
                 identityReads === 1
                   ? exitedIdentity(row, exit)
-                  : {
-                      ...nativeIdentity(replacement),
-                      identity: "replacement",
-                    },
+                  : { ...nativeIdentity(replacement), identity: "replacement" },
               ];
             if (
               phase === "exited" ||

src/process/ProcessOwnership.test.ts

@@ -337,12 +337,7 @@ describe("related sweep candidates during cleanup", () => {
     let childReads = 0;
     const adapter: ProcessOwnershipHost = {
       listProcesses: async () => processes,
-      processIdentities: async (members) =>
-        identitiesWithRelations(
-          processes.filter(({ pid }) =>
-            members.some((member) => member.pid === pid),
-          ),
-        ),
+      processIdentities: async (members) => identitiesWithRelations(members),
       environment: async () => ({}),
       runTokens: async (members) =>
         new Map(

src/process/ProcessOwnership.ts

@@ -195,12 +195,9 @@ export interface ProcessOwnershipHost {
     processes: readonly ProcessTableEntry[],
     signal?: AbortSignal,
   ): Promise<ReadonlyMap<number, ProcessRunTokenObservation>>;
-  /**
-   * Read stable native identities by PID; caller rows supply no native facts.
-   * Omit the signal during cleanup so it can finish.
-   */
+  /** Read stable identities; omit the signal during cleanup so it can finish. */
   processIdentities?(
-    processes: readonly Pick<ProcessTableEntry, "pid">[],
+    processes: readonly ProcessTableEntry[],
     signal?: AbortSignal,
     /** Linux credential enrichment is selective; omitted means UID and environment restriction remain unknown. */
     credentialPids?: ReadonlySet<number>,
@@ -598,7 +595,7 @@ const readProcessRelationSnapshot = async (
   ownership: OwnedProcessGroup,
   processes: readonly ProcessTableEntry[],
   host: ProcessOwnershipHost,
-): Promise<ProcessRelationSnapshot | ProcessCleanupFailure> => {
+): Promise<ProcessRelationSnapshot> => {
   if (host.processIdentities === undefined)
     return {
       currentRelations: processes,
@@ -606,17 +603,12 @@ const readProcessRelationSnapshot = async (
       failures: [],
     };
 
-  const retainedPids =
-    ownership.evidenceOwner?.candidates.map(({ pid }) => pid) ?? [];
-  const observationRows =
-    ownership.evidenceOwner?.rowsForObservation(processes) ?? processes;
-  const identities = await readNativeProcessIdentityMap(observationRows, host);
+  const identities = await readNativeProcessIdentityMap(processes, host);
   const possibleDescendants = new Set(
     descendantsOf([ownership.leaderPid], processes).map(({ pid }) => pid),
   );
   const relevantPids = new Set([
     ownership.leaderPid,
-    ...retainedPids,
     ...possibleDescendants,
     ...processes
       .filter(
@@ -628,28 +620,13 @@ const readProcessRelationSnapshot = async (
   ]);
   const currentRelations: ProcessTableEntry[] = [];
   const failures: ProcessOwnershipValidationFailure[] = [];
-  const listedPids = new Set(processes.map(({ pid }) => pid));
-  // An empty table can still describe a token-verified owner. A table that
-  // lists other processes and omits a live retained PID is incomplete: it
-  // cannot authorize a signal or certify cleanup.
-  const omittedFromListedTable = (pid: number): boolean =>
-    processes.length > 0 && !listedPids.has(pid);
-  for (const process of observationRows) {
+  for (const process of processes) {
     const identity = identities.get(process.pid);
     if (processIdentityHasExited(identity)) continue;
     if (
       identity?.state === "readable" &&
       identity.relation?.state === "observed"
     ) {
-      if (omittedFromListedTable(process.pid)) {
-        failures.push({
-          pid: process.pid,
-          reason: "process-identity-unavailable",
-          diagnostic:
-            "retained process was omitted from the current process table",
-        });
-        continue;
-      }
       currentRelations.push({
         ...process,
         parentPid: identity.relation.parentPid,
@@ -664,27 +641,6 @@ const readProcessRelationSnapshot = async (
         diagnostic: `native process membership could not be read${identity?.state === "unavailable" ? `: ${identity.reason}` : ""}`,
       });
   }
-  const launcher = currentRelations.find(
-    ({ pid }) => pid === ownership.leaderPid,
-  );
-  if (launcher !== undefined) {
-    const listedLauncher = processes.find(
-      ({ pid }) => pid === ownership.leaderPid,
-    );
-    // Retained rows select PIDs, but only this table supplies a current command.
-    // Native parent/group applicability must hold on every pre-signal refresh.
-    const reason = launcherIdentityFailure(
-      {
-        parentPid: launcher.parentPid,
-        processGroupId: launcher.processGroupId,
-        ...(listedLauncher === undefined
-          ? {}
-          : { command: listedLauncher.command }),
-      },
-      ownership,
-    );
-    if (reason !== null) return { cleaned: false, reason };
-  }
   return { currentRelations, identities, failures };
 };
 
@@ -992,7 +948,6 @@ const createOwnedCleanupPlan = async (
     processes,
     host,
   );
-  if ("cleaned" in relationSnapshot) return relationSnapshot;
   const sampledProcessGroupIds = new Set(
     (ownership.sampledProcessGroupIds ?? []).filter(
       (processGroupId) =>
@@ -1024,14 +979,8 @@ const createOwnedCleanupPlan = async (
           exitSettlement,
         )
       : { owned: [], failures: [], unverified: [], identities: new Map() };
-  const listedPids = new Set(processes.map(({ pid }) => pid));
-  const omittedFromListedTable = (pid: number): boolean =>
-    processes.length > 0 && !listedPids.has(pid);
-  const signalableOwned = tokenOwned.owned.filter(
-    ({ pid }) => !omittedFromListedTable(pid),
-  );
   const tokenOwnedGroupIds = new Set(
-    signalableOwned.map(({ processGroupId }) => processGroupId),
+    tokenOwned.owned.map(({ processGroupId }) => processGroupId),
   );
   const launcher = relationSnapshot.currentRelations.find(
     ({ pid }) => pid === ownership.leaderPid,
@@ -1039,36 +988,21 @@ const createOwnedCleanupPlan = async (
   const rootMembers = relationSnapshot.currentRelations.filter(
     ({ processGroupId }) => processGroupId === ownership.processGroupId,
   );
-  const unresolved: ProcessOwnershipValidationFailure[] = [];
-  for (const failure of tokenOwned.failures)
-    mergeProcessOwnershipFailure(unresolved, failure);
-  // The later bracketed token scan retains the current, more useful failure
-  // for a PID whose earlier membership read was also unavailable.
-  for (const failure of relationSnapshot.failures)
-    if (!unresolved.some(({ pid }) => pid === failure.pid))
-      mergeProcessOwnershipFailure(unresolved, failure);
-  for (const process of tokenOwned.owned)
-    if (omittedFromListedTable(process.pid))
-      mergeProcessOwnershipFailure(unresolved, {
-        pid: process.pid,
-        reason: "process-identity-unavailable",
-        diagnostic:
-          "retained process was omitted from the current process table",
-      });
   if (
     launcher === undefined &&
     rootMembers.length === 0 &&
-    signalableOwned.length === 0 &&
+    tokenOwned.owned.length === 0 &&
     escapedProcessGroupIds.size === 0 &&
     !relationSnapshot.currentRelations.some(({ processGroupId }) =>
       sampledProcessGroupIds.has(processGroupId),
     )
   )
-    return unresolved.length > 0
+    return tokenOwned.failures.length > 0 ||
+      relationSnapshot.failures.length > 0
       ? {
           signalOrder: [],
           tokenOwnedIdentities: tokenOwned.identities,
-          unresolved,
+          unresolved: [...tokenOwned.failures, ...relationSnapshot.failures],
           unverified: tokenOwned.unverified,
         }
       : {
@@ -1089,13 +1023,14 @@ const createOwnedCleanupPlan = async (
   const processGroupIds = new Set<number>([ownership.processGroupId]);
   for (const descendant of descendants)
     processGroupIds.add(descendant.processGroupId);
-  for (const process of signalableOwned)
+  for (const process of tokenOwned.owned)
     processGroupIds.add(process.processGroupId);
   for (const processGroupId of sampledProcessGroupIds)
     processGroupIds.add(processGroupId);
   for (const processGroupId of escapedProcessGroupIds)
     processGroupIds.add(processGroupId);
   const signalableGroups = new Set<number>();
+  const unresolved = [...tokenOwned.failures, ...relationSnapshot.failures];
   const unverifiableSampledGroups = new Set<number>();
   for (const processGroupId of processGroupIds) {
     if (processGroupId === ownership.processGroupId) continue;
@@ -1238,7 +1173,6 @@ const revalidateOwnedProcessGroup = async (
         settled.processes,
         host,
       );
-      if ("cleaned" in relationSnapshot) return relationSnapshot;
       if (relationSnapshot.failures.length > 0)
         return cleanupValidationFailure(relationSnapshot.failures);
       members = relationSnapshot.currentRelations.filter(

src/process/ProcessOwnershipIdentity.ts

@@ -6,9 +6,7 @@ import type {
 
 /** Compare a live launcher against the identity committed at process launch. */
 export const launcherIdentityFailure = (
-  launcher: Pick<ProcessTableEntry, "parentPid" | "processGroupId"> & {
-    readonly command?: string;
-  },
+  launcher: ProcessTableEntry,
   ownership: OwnedProcessGroup,
 ): string | null => {
   if (launcher.processGroupId !== ownership.processGroupId)
@@ -18,13 +16,10 @@ export const launcherIdentityFailure = (
     launcher.parentPid !== ownership.expectedParentPid
   )
     return "owned launcher parent identity did not match";
-  if (ownership.expectedCommand !== undefined) {
-    if (launcher.command === undefined)
-      return "owned launcher command identity could not be verified";
-    if (
-      !matchesOwnedProcessCommand(launcher.command, ownership.expectedCommand)
-    )
-      return `owned launcher command identity did not match (observed=${launcher.command}; expected=${ownership.expectedCommand})`;
-  }
+  if (
+    ownership.expectedCommand !== undefined &&
+    !matchesOwnedProcessCommand(launcher.command, ownership.expectedCommand)
+  )
+    return `owned launcher command identity did not match (observed=${launcher.command}; expected=${ownership.expectedCommand})`;
   return null;
 };

src/process/ProcessOwnershipObservation.ts

@@ -33,7 +33,6 @@ import {
 import { darwinHelperCacheDirectory } from "./DarwinHelperCache.js";
 import {
   ProcessRunTokenReadError,
-  readProcessRunToken,
   readProcessRunTokens,
 } from "./ProcessRunTokenObservations.js";
 import type {
@@ -1669,35 +1668,39 @@ export const observeProcessStartIdentity = async (
   host: ProcessOwnershipHost = systemProcessOwnershipHost,
   expectedRunId?: string,
 ): Promise<ProcessIdentityObservation | undefined> => {
+  const process = (await host.listProcesses()).find(
+    (entry) => entry.pid === pid,
+  );
+  if (process === undefined || liveProcesses([process]).length === 0)
+    return undefined;
   if (host.processIdentities === undefined)
     return { state: "unavailable", reason: "process identity is unsupported" };
   const identity: ProcessIdentityObservation = (
-    await host.processIdentities([{ pid }])
+    await host.processIdentities([process])
   ).get(pid) ?? {
     state: "unavailable",
     reason: "process identity was not returned",
   };
-  if (identity.state === "absent" || processIdentityHasExited(identity))
-    return undefined;
   if (expectedRunId === undefined || identity.state !== "readable")
     return identity;
-  const process = (await host.listProcesses()).find(
-    (entry) => entry.pid === pid,
-  );
-  let token: ProcessRunTokenObservation | undefined;
-  if (process === undefined) token = await readProcessRunToken(host, pid);
-  else
-    for await (const { observation } of readProcessRunTokens(host, [process]))
-      token = observation;
-  if (token === undefined || token.state === "unavailable")
+  let owned = false;
+  for await (const { observation } of readProcessRunTokens(host, [process])) {
+    if (observation.state === "unavailable")
+      return {
+        state: "unavailable",
+        reason: `process ownership token could not be read for PID ${String(pid)}: ${observation.reason}`,
+      };
+    if (observation.runId !== expectedRunId)
+      return {
+        state: "unavailable",
+        reason: `process ownership token did not match the captured run for PID ${String(pid)}`,
+      };
+    owned = true;
+  }
+  if (!owned)
     return {
       state: "unavailable",
-      reason: `process ownership token could not be read for PID ${String(pid)}: ${token?.reason ?? "token was not returned"}`,
-    };
-  if (token.runId !== expectedRunId)
-    return {
-      state: "unavailable",
-      reason: `process ownership token did not match the captured run for PID ${String(pid)}`,
+      reason: `process ownership token was not returned for PID ${String(pid)}`,
     };
   const confirmed = await observeProcessStartIdentity(pid, host);
   if (confirmed === undefined || confirmed.state !== "readable")

src/process/ProcessRunTokenObservations.ts

@@ -29,39 +29,6 @@ export const processRunIdFromEnvironment = (
     ? environment.REA_PROCESS_RUN_ID
     : undefined;
 
-/** Read a selected PID's environment token without requiring process-table relations. */
-export const readProcessRunToken = async (
-  host: ProcessOwnershipHost,
-  pid: number,
-  signal?: AbortSignal,
-): Promise<ProcessRunTokenObservation> => {
-  signal?.throwIfAborted();
-  try {
-    const environment =
-      signal === undefined
-        ? await host.environment(pid)
-        : await host.environment(pid, signal);
-    signal?.throwIfAborted();
-    return {
-      state: "readable",
-      runId: processRunIdFromEnvironment(environment),
-    };
-  } catch (cause: unknown) {
-    signal?.throwIfAborted();
-    const code =
-      cause instanceof Error &&
-      "code" in cause &&
-      typeof cause.code === "string"
-        ? cause.code
-        : undefined;
-    return {
-      state: "unavailable",
-      ...(code === undefined ? {} : { code }),
-      reason: errorMessage(cause),
-    };
-  }
-};
-
 /** Read only ownership tokens, using per-process fallback for missing batch rows. */
 export async function* readProcessRunTokens(
   host: ProcessOwnershipHost,
@@ -90,12 +57,33 @@ export async function* readProcessRunTokens(
     signal?.throwIfAborted();
     let observation = batch?.get(process.pid);
     if (observation === undefined) {
-      observation = await readProcessRunToken(host, process.pid, signal);
-      if (observation.state === "unavailable" && batchFailure !== undefined)
+      try {
+        const environment =
+          signal === undefined
+            ? await host.environment(process.pid)
+            : await host.environment(process.pid, signal);
         observation = {
-          ...observation,
-          reason: `${observation.reason}; run-token batch failed: ${batchFailure}`,
+          state: "readable",
+          runId: processRunIdFromEnvironment(environment),
         };
+      } catch (cause: unknown) {
+        signal?.throwIfAborted();
+        const reason = errorMessage(cause);
+        const code =
+          cause instanceof Error &&
+          "code" in cause &&
+          typeof cause.code === "string"
+            ? cause.code
+            : undefined;
+        observation = {
+          state: "unavailable",
+          ...(code === undefined ? {} : { code }),
+          reason:
+            batchFailure === undefined
+              ? reason
+              : `${reason}; run-token batch failed: ${batchFailure}`,
+        };
+      }
     }
     signal?.throwIfAborted();
     yield { process, observation };

src/process/ProcessStartIdentity.test.ts

@@ -24,112 +24,10 @@ const host = (identity: string | undefined): ProcessOwnershipHost => ({
         ],
   environment: async () => ({}),
   processIdentities: async () =>
-    new Map([
-      [
-        42,
-        identity === undefined
-          ? { state: "absent", reason: "not-found-in-caller-pid-namespace" }
-          : { state: "readable", identity },
-      ],
-    ]),
+    new Map([[42, { state: "readable", identity: identity ?? "" }]]),
   signalGroup: () => undefined,
 });
 
-describe("selected native process identity", () => {
-  it.each(["readable", "unavailable"] as const)(
-    "does not infer exit from a table omission when native identity is %s",
-    async (state) => {
-      const omitted: ProcessOwnershipHost = {
-        ...host("start-1"),
-        listProcesses: async () => [],
-        processIdentities: async () =>
-          new Map([
-            [
-              42,
-              state === "readable"
-                ? { state: "readable", identity: "start-1" }
-                : { state: "unavailable", reason: "native inspection denied" },
-            ],
-          ]),
-      };
-      await expect(
-        observeProcessStartIdentityExit(42, "start-1", omitted),
-      ).resolves.toBe(state === "readable" ? "present" : "unverified");
-    },
-  );
-
-  it.each(["absent", "zombie", "dead"] as const)(
-    "uses native %s proof despite a stale live process-table row",
-    async (lifecycle) => {
-      const exited: ProcessOwnershipHost = {
-        ...host("start-1"),
-        processIdentities: async () =>
-          new Map([
-            [
-              42,
-              lifecycle === "absent"
-                ? {
-                    state: "absent",
-                    reason: "not-found-in-caller-pid-namespace",
-                  }
-                : {
-                    state: "readable",
-                    identity: "start-1",
-                    lifecycle: { state: "observed", lifecycle },
-                  },
-            ],
-          ]),
-      };
-      await expect(
-        observeProcessStartIdentityExit(42, "start-1", exited),
-      ).resolves.toBe("exited");
-      const sendSignal = vi.fn();
-      await expect(
-        signalProcessWithStartIdentity(42, "start-1", "SIGKILL", {
-          host: exited,
-          sendSignal,
-        }),
-      ).resolves.toBe("gone");
-      expect(sendSignal).not.toHaveBeenCalled();
-    },
-  );
-
-  it("binds an omitted native PID to its directly observed run token", async () => {
-    const omitted: ProcessOwnershipHost = {
-      ...host("start-1"),
-      listProcesses: async () => [],
-      environment: async () => ({ REA_PROCESS_RUN_ID: "captured-run" }),
-    };
-    await expect(
-      observeProcessStartIdentity(42, omitted, "captured-run"),
-    ).resolves.toMatchObject({ state: "readable", identity: "start-1" });
-  });
-
-  it.each(["unreadable", "foreign", "reused"] as const)(
-    "does not bind an omitted PID with a %s token observation",
-    async (condition) => {
-      let identity = "start-1";
-      const omitted: ProcessOwnershipHost = {
-        ...host("start-1"),
-        listProcesses: async () => [],
-        processIdentities: async () =>
-          new Map([[42, { state: "readable", identity }]]),
-        environment: async () => {
-          if (condition === "unreadable") throw new Error("environment denied");
-          if (condition === "reused") identity = "replacement-start";
-          return {
-            REA_PROCESS_RUN_ID:
-              condition === "foreign" ? "other-run" : "captured-run",
-          };
-        },
-      };
-      await expect(
-        observeProcessStartIdentity(42, omitted, "captured-run"),
-      ).resolves.toMatchObject({ state: "unavailable" });
-    },
-  );
-});
-
 describe("process start identity lease", () => {
   it("captures a readable identity for a live PID", async () => {
     await expect(

src/process/ProviderProcess.ts

@@ -642,8 +642,7 @@ export class ProviderProcessSupervisor {
   stop(
     options: ProviderProcessStopOptions = {},
   ): Promise<ProviderProcessStopResult> {
-    if (this.launch.ownsProcessLifetime && this.#exitObservation === undefined)
-      this.#stopRequested = true;
+    this.#stopRequested = true;
     this.#stopPromise ??= this.#stop(options).then((result) => {
       if (result.status === "incomplete") this.#stopPromise = undefined;
       return result;

src/process/ProviderProcessOutput.test.ts

@@ -121,9 +121,7 @@ it("retries process cleanup with capture attached until output closes", async ()
     },
   });
   await expect(owner.stop()).resolves.toMatchObject({ status: "incomplete" });
-  expect(owner.snapshot().stopRequested).toBe(true);
   producer.stderr.write("retained during retry");
-  producer.emit("exit", 0, null);
   fails = false;
   const retried = owner.stop();
   await new Promise<void>((resolve) => setImmediate(resolve));
@@ -131,7 +129,6 @@ it("retries process cleanup with capture attached until output closes", async ()
   producer.stdout.end();
   producer.emit("close", 0, null);
   await expect(retried).resolves.toEqual({ status: "verified-cleanup" });
-  expect(owner.snapshot().stopRequested).toBe(true);
   expect(owner.snapshot().stderr.text).toBe(
     "retained during retry and final tail",
   );
@@ -877,38 +874,6 @@ it("marks a stop the owner requested before any signal it sends", async () => {
   expect(unowned.snapshot()).toMatchObject({ signal: "SIGKILL" });
   expect(unowned.snapshot()).not.toHaveProperty("stopRequested");
 
-  for (const ownsProcessLifetime of [false, true] as const) {
-    const exited = producer();
-    const finalized = new ProviderProcessSupervisor(
-      ownsProcessLifetime
-        ? {
-            process: exited,
-            ownsProcessLifetime: true,
-            cleanup: async () => ({ cleaned: true, signaled: false }),
-          }
-        : { process: exited, ownsProcessLifetime: false },
-    );
-    exited.emit("exit", null, "SIGKILL");
-    exited.stdout.end();
-    exited.stderr.end();
-    exited.emit("close", null, "SIGKILL");
-    await expect(finalized.stop()).resolves.toEqual({
-      status: ownsProcessLifetime ? "verified-cleanup" : "not-owned",
-    });
-    expect(finalized.snapshot()).toMatchObject({ signal: "SIGKILL" });
-    expect(finalized.snapshot()).not.toHaveProperty("stopRequested");
-  }
-
-  const borrowed = producer();
-  const observer = new ProviderProcessSupervisor({
-    process: borrowed,
-    ownsProcessLifetime: false,
-  });
-  await expect(observer.stop()).resolves.toEqual({ status: "not-owned" });
-  expect(observer.snapshot()).not.toHaveProperty("stopRequested");
-  borrowed.stdout.destroy();
-  borrowed.stderr.destroy();
-
   const stopped = producer();
   const owner = new ProviderProcessSupervisor({
     process: stopped,

src/process/TokenOwnedTermination.test.ts

@@ -138,27 +138,6 @@ describe("terminateTokenOwnedProcesses", () => {
     expect(signalProcess.mock.calls).toEqual([[100, "SIGTERM"]]);
   });
 
-  it("does not report a refused group signal as delivered and still stops other token carriers", async () => {
-    const { host, signalGroup } = fakeHost([
-      row(100, 100, "run"),
-      row(200, 200, "run"),
-    ]);
-    signalGroup.mockImplementation(() => {
-      throw Object.assign(new Error("Signal refused"), { code: "EPERM" });
-    });
-
-    expect(
-      await terminateTokenOwnedProcesses(
-        { runId: "run", processGroupId: 100 },
-        1_000,
-        host,
-      ),
-    ).toEqual({ launchGroupSignalled: false });
-    expect(
-      (await host.ownership.listProcesses()).map(({ pid }) => pid),
-    ).toEqual([100]);
-  });
-
   it("waits for the grace period when a process ignores SIGTERM", async () => {
     const { host } = fakeHost([row(100, 100, "run")], new Set([100]));
     const started = performance.now();

src/process/TokenOwnedTermination.ts

@@ -82,13 +82,11 @@ const awaitExit = async (
   }
 };
 
-const send = (deliver: () => void): boolean => {
+const send = (deliver: () => void): void => {
   try {
     deliver();
-    return true;
   } catch {
-    // Exit or a failed delivery does not establish that SIGTERM was sent.
-    return false;
+    // A process that exited first has nothing left to stop.
   }
 };
 
@@ -115,20 +113,16 @@ export const terminateTokenOwnedProcesses = async (
   }
   const inGroup = (fact: ProcessOwnershipFact) =>
     processGroupOf(fact) === ownership.processGroupId;
-  const launchGroupSignalled =
-    owned.some(inGroup) &&
+  const launchGroupSignalled = owned.some(inGroup);
+  if (launchGroupSignalled)
     send(() => {
       host.ownership.signalGroup(ownership.processGroupId, "SIGTERM");
     });
-  const signalled = launchGroupSignalled ? owned.filter(inGroup) : [];
   for (const fact of owned)
-    if (
-      !inGroup(fact) &&
+    if (!inGroup(fact))
       send(() => {
         host.signalProcess(fact.process.pid, "SIGTERM");
-      })
-    )
-      signalled.push(fact);
-  await awaitExit(signalled, deadline, host.ownership);
+      });
+  await awaitExit(owned, deadline, host.ownership);
   return { launchGroupSignalled };
 };

src/process/WorkerCleanupRegistry.test.ts

@@ -25,58 +25,3 @@ it("retries one host's retained runtime without failing another host", async ()
     await stuck.close();
   }
 });
-
-it("retries every independent retained owner and keeps only exact failures pending", async () => {
-  const registry = new WorkerCleanupRegistry();
-  const physical = new Error("first owner still owns its worker");
-  let failing = true;
-  const calls: string[] = [];
-  const first = {
-    retryRetainedCleanup: async () => {
-      calls.push("first");
-      if (failing) throw physical;
-      registry.delete(first);
-    },
-  };
-  const second = {
-    retryRetainedCleanup: async () => {
-      calls.push("second");
-      registry.delete(second);
-    },
-  };
-  registry.add(first);
-  registry.add(second);
-  await expect(registry.retry()).rejects.toBe(physical);
-  expect(calls).toEqual(["first", "second"]);
-  expect(registry.has(first)).toBe(true);
-  expect(registry.has(second)).toBe(false);
-  failing = false;
-  await registry.retry();
-  expect(calls).toEqual(["first", "second", "first"]);
-  expect(registry.has(first)).toBe(false);
-});
-
-it("retains every failed cleanup cause and skips only the caller-selected owner", async () => {
-  const registry = new WorkerCleanupRegistry();
-  const causes = [
-    new Error("first exact cleanup"),
-    new Error("second exact cleanup"),
-  ];
-  const skipped = {
-    retryRetainedCleanup: async () => {
-      throw new Error("must not retry selected owner");
-    },
-  };
-  const owners = causes.map((cause) => ({
-    retryRetainedCleanup: async () => {
-      throw cause;
-    },
-  }));
-  registry.add(skipped);
-  for (const owner of owners) registry.add(owner);
-  const failure = await registry.retry(skipped).catch((cause) => cause);
-  expect(failure).toBeInstanceOf(AggregateError);
-  expect(failure.errors).toEqual(causes);
-  expect(registry.has(skipped)).toBe(true);
-  for (const owner of owners) expect(registry.has(owner)).toBe(true);
-});

tests/boundary/android/jadxMcpTransport.test.ts

@@ -5,7 +5,6 @@ import {
   type JSONRPCMessage,
 } from "@modelcontextprotocol/client";
 import { expect, it } from "vitest";
-import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
 import { OwnedMcpStdioTransport } from "../../../src/process/OwnedMcpStdioTransport.js";
 import type { ProviderProcessHandle } from "../../../src/process/ProviderProcess.js";
 import {
@@ -259,27 +258,9 @@ it("retries an incomplete owned stop while joining concurrent close callers", as
   await transport.start();
   const firstClose = transport.close();
   expect(transport.close()).toBe(firstClose);
-  const failure: unknown = await firstClose.catch((cause: unknown) => cause);
-  if (!(failure instanceof ProviderCleanupError)) throw failure;
-  expect(failure.cleanupIncomplete).toBe(true);
-  expect(failure.cleanupResources).toEqual([transport.runId]);
-  expect(failure.diagnostics).toMatchObject({
-    reason: "ownership verification unavailable",
-  });
-  const unrelated = new JadxMcpTransport(
-    { command: "fixture", arguments: [] },
-    fixtureLauncher(new FixtureProcess()),
-  );
-  await unrelated.start();
-  await unrelated.close();
-  expect(failure.cleanupIncomplete).toBe(true);
+  await expect(firstClose).rejects.toMatchObject({ cleanupIncomplete: true });
   cleanupFails = false;
   await transport.close();
-  expect(failure.cleanupIncomplete).toBe(false);
-  expect(failure.cleanupResources).toEqual([]);
-  expect(failure.diagnostics).toMatchObject({
-    reason: "ownership verification unavailable",
-  });
   await transport.close();
 });
 

tests/boundary/captures/captureProvider.test.ts

@@ -247,6 +247,26 @@ it("preserves TShark canonical output for a caller-selected field alias", async
   expect(report.tool.arguments).toContain("bootp.type");
 });
 
+it("refuses producer frames outside the selected input read scope before paging", async () => {
+  const f = await fixture("many");
+  const request = { path: f.path, fields: ["ip.addr"], match_limit: 1 };
+  expect(
+    await f.service.inspect({ ...request, packet_limit: 10 }),
+  ).toMatchObject({
+    ok: false,
+    error: { _tag: "AnalysisOutputError", operation: "inspect_packet_capture" },
+  });
+  const admitted = await f.service.inspect({ ...request, packet_limit: 90 });
+  if (!admitted.ok) throw admitted.error;
+  const report = packetCaptureSchema.parse(admitted.value.normalized_result);
+  expect(report.output_window).toEqual({
+    matching_packets: 90,
+    returned_packets: 1,
+    has_more: true,
+  });
+  expect(report.packets[0]?.["frame.number"]).toEqual(["1"]);
+});
+
 it("returns canonical alias output through the public packet-capture CLI", async () => {
   const f = await fixture("alias-canonical");
   const cli = Cli.create("capture-alias-test", { sync: false });

tests/boundary/captures/packetOutput.test.ts

@@ -17,7 +17,7 @@ it("preserves repeated field values and missing fields while discarding export i
     },
     { _source: { layers: { "frame.number": ["7"] } } },
   ]);
-  expect(parsePacketOutput(output)).toEqual([
+  expect(parsePacketOutput(output, 7)).toEqual([
     { "frame.number": ["2"], "ip.addr": ["192.0.2.1", "192.0.2.2"] },
     { "frame.number": ["7"] },
   ]);
@@ -49,7 +49,7 @@ it.each([
 ])("rejects malformed producer fields: $layers", ({ layers, issue }) => {
   let failure: unknown;
   try {
-    parsePacketOutput(JSON.stringify([{ _source: { layers } }]));
+    parsePacketOutput(JSON.stringify([{ _source: { layers } }]), null);
   } catch (error) {
     failure = error;
   }
@@ -72,6 +72,7 @@ it.each([
           _source: { layers: { "frame.number": frame } },
         })),
       ),
+      null,
     );
   } catch (error) {
     failure = error;
@@ -88,7 +89,18 @@ it("preserves canonical field names emitted for a selected alias", () => {
       },
     },
   ]);
-  expect(parsePacketOutput(output)).toEqual([
+  expect(parsePacketOutput(output, null)).toEqual([
     { "frame.number": ["1"], "dhcp.type": ["1"] },
   ]);
 });
+
+it("validates every frame against the input read limit while allowing filtered gaps", () => {
+  const output = JSON.stringify(
+    [2, 7].map((frame) => ({
+      _source: { layers: { "frame.number": [String(frame)] } },
+    })),
+  );
+  expect(parsePacketOutput(output, 7)).toHaveLength(2);
+  expect(() => parsePacketOutput(output, 6)).toThrow(AnalysisOutputError);
+  expect(parsePacketOutput(output, null)).toHaveLength(2);
+});

tests/boundary/process/darwinProcessRunTokenReader.test.ts

@@ -1189,16 +1189,18 @@ it.skipIf(process.platform === "win32")(
     const baselineSpy = vi
       .spyOn(systemProcessOwnershipHost, "captureBaseline")
       .mockImplementation(async (signal) => {
-        const processes = [
-          {
-            pid: process.pid,
-            parentPid: process.ppid,
-            processGroupId: process.pid,
-            state: "S",
-            command: process.execPath,
-          },
-        ];
-        await reader.identities(processes, signal);
+        await reader.identities(
+          [
+            {
+              pid: process.pid,
+              parentPid: process.ppid,
+              processGroupId: process.pid,
+              state: "S",
+              command: process.execPath,
+            },
+          ],
+          signal,
+        );
         return [];
       });
     const launchSpy = vi.spyOn(electron, "launch");