From aece2e7e162274388ebf02d899a6197594c848c7 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Wed, 11 Nov 2026 17:20:53 +0800 Subject: [PATCH] fix(process): preserve ownership through termination and cleanup (#9891) * fix(process): revalidate ownership before signaling * fix(process): retain cleanup ownership until completion * fix(process): keep omitted live descendants unresolved A non-empty process table that omits a retained live PID cannot authorize a signal or certify cleanup. An empty table can still describe a token-verified owner. * fix(process): read omitted Darwin run tokens by PID A live identity that ps omitted was still validated through Darwin environment(), which rejects any PID absent from a second process table. Ask the native reader for that PID directly so the launch token can bind. --------- Co-authored-by: Cursor Agent --- src/browser/ElectronProcessCleanup.test.ts | 49 ++-- .../DarwinHelperCache.ownership.test.ts | 92 +++++++ src/process/DarwinHelperCache.ts | 22 +- ...ProcessRunTokenReader.finalization.test.ts | 54 +--- src/process/DarwinProcessRunTokenReader.ts | 4 +- src/process/OwnedLease.test.ts | 44 ++++ src/process/OwnedLease.ts | 22 +- src/process/OwnedMcpStdioTransport.ts | 34 ++- src/process/OwnedWorkerRuntime.ts | 17 +- .../ProcessOwnership.exitScope.test.ts | 18 +- src/process/ProcessOwnership.identity.test.ts | 151 ++++++++++- src/process/ProcessOwnership.lineage.test.ts | 246 ++++++++++++------ src/process/ProcessOwnership.test.ts | 7 +- ...ProcessOwnership.tokenAvailability.test.ts | 51 +++- src/process/ProcessOwnership.ts | 98 ++++++- src/process/ProcessOwnershipIdentity.ts | 17 +- src/process/ProcessOwnershipObservation.ts | 46 ++-- src/process/ProcessRunTokenObservations.ts | 62 +++-- src/process/ProcessStartIdentity.test.ts | 104 +++++++- src/process/ProviderProcess.ts | 3 +- src/process/ProviderProcessOutput.test.ts | 35 +++ src/process/TokenOwnedTermination.test.ts | 21 ++ src/process/TokenOwnedTermination.ts | 20 +- src/process/WorkerCleanupRegistry.test.ts | 55 ++++ .../boundary/android/jadxMcpTransport.test.ts | 21 +- .../darwinProcessRunTokenReader.test.ts | 49 +--- 26 files changed, 1032 insertions(+), 310 deletions(-) create mode 100644 src/process/DarwinHelperCache.ownership.test.ts diff --git a/src/browser/ElectronProcessCleanup.test.ts b/src/browser/ElectronProcessCleanup.test.ts index dca446022d..13fed355ed 100644 --- a/src/browser/ElectronProcessCleanup.test.ts +++ b/src/browser/ElectronProcessCleanup.test.ts @@ -61,30 +61,33 @@ describe.skipIf(process.platform === "win32")( processIdentities: (members) => Promise.resolve( new Map( - 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, + 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, + }, }, - relation: { - state: "observed" as const, - parentPid: member.parentPid, - processGroupId: member.processGroupId, - }, - }, - ]), + ]; + }), ), ), signalGroup: vi.fn(() => { diff --git a/src/process/DarwinHelperCache.ownership.test.ts b/src/process/DarwinHelperCache.ownership.test.ts new file mode 100644 index 0000000000..6f51206804 --- /dev/null +++ b/src/process/DarwinHelperCache.ownership.test.ts @@ -0,0 +1,92 @@ +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()), + 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); + }, +); diff --git a/src/process/DarwinHelperCache.ts b/src/process/DarwinHelperCache.ts index 570315b3da..db4f623f79 100644 --- a/src/process/DarwinHelperCache.ts +++ b/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 Promise>(); + const retained = new Map Promise>(); // The first unresolved close failure; later ones depend on it. let cleanupAnchor: ProviderCleanupError | undefined; let key: Promise | undefined; @@ -271,7 +271,7 @@ export const createDarwinHelperCache = (options: { }; const retainClose = (owner: OwnedFileHandle): void => { - retained.set(`file handle ${String(owner.handle.fd)}`, () => owner.close()); + retained.set(owner, () => owner.close()); }; /** Hash the opened descriptor itself; FIFOs and links never block or pass. */ @@ -431,13 +431,19 @@ export const createDarwinHelperCache = (options: { /** Retry every retained close or removal; unresolved ones stay owned. */ const close = async (): Promise => { 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 { - failed.push(resource); + } catch (cause: unknown) { + failed.push( + typeof resource === "string" + ? resource + : `file handle ${String(resource.handle.fd)}`, + ); + causes.push(cause); } } if (failed.length > 0) { @@ -448,7 +454,13 @@ export const createDarwinHelperCache = (options: { reason: "Darwin ownership-helper cache resources remain owned until cleanup succeeds.", }, - { operation: "prepare_owned_process" }, + { + cause: new AggregateError( + causes, + "Darwin helper cache cleanup failed", + ), + operation: "prepare_owned_process", + }, ); // Every returned failure clears once a later retry releases them all. if (cleanupAnchor === undefined) cleanupAnchor = failure; diff --git a/src/process/DarwinProcessRunTokenReader.finalization.test.ts b/src/process/DarwinProcessRunTokenReader.finalization.test.ts index d85fb9f628..29cb3a799a 100644 --- a/src/process/DarwinProcessRunTokenReader.finalization.test.ts +++ b/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 operation = reader.identities([ + const processes = [ { pid: 123, parentPid: 1, @@ -180,7 +180,8 @@ 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( @@ -211,26 +212,17 @@ 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([ - { - pid: 123, - parentPid: 1, - processGroupId: 123, - state: "S", - command: "fixture", - }, - ]) - : reader.identities([ - { - pid: 123, - parentPid: 1, - processGroupId: 123, - state: "S", - command: "fixture", - }, - ]); + mode === "read" ? reader.read(processes) : reader.identities(processes); await command.waitUntilStarted(); let closeSettled = false; @@ -273,25 +265,7 @@ it.each(["read", "identities"] as const)( ]), ); await expect( - mode === "read" - ? reader.read([ - { - pid: 123, - parentPid: 1, - processGroupId: 123, - state: "S", - command: "fixture", - }, - ]) - : reader.identities([ - { - pid: 123, - parentPid: 1, - processGroupId: 123, - state: "S", - command: "fixture", - }, - ]), + mode === "read" ? reader.read(processes) : reader.identities(processes), ).rejects.toThrow("closed"); }, ); diff --git a/src/process/DarwinProcessRunTokenReader.ts b/src/process/DarwinProcessRunTokenReader.ts index c74ea04785..5546fc770a 100644 --- a/src/process/DarwinProcessRunTokenReader.ts +++ b/src/process/DarwinProcessRunTokenReader.ts @@ -735,7 +735,7 @@ export const createDarwinProcessRunTokenReader = ( }; const read = async ( - processes: readonly ProcessTableEntry[], + processes: readonly Pick[], signal?: AbortSignal, ): Promise> => { signal?.throwIfAborted(); @@ -776,7 +776,7 @@ export const createDarwinProcessRunTokenReader = ( }; const identities = async ( - processes: readonly ProcessTableEntry[], + processes: readonly Pick[], signal?: AbortSignal, ): Promise> => { if (processes.length === 0) return new Map(); diff --git a/src/process/OwnedLease.test.ts b/src/process/OwnedLease.test.ts index 123b725814..303e4e21ac 100644 --- a/src/process/OwnedLease.test.ts +++ b/src/process/OwnedLease.test.ts @@ -9,6 +9,7 @@ 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 => { const directory = await mkdtemp(join("/tmp", prefix)); @@ -93,6 +94,49 @@ 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(); + const finish = createDeferred(); + 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((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) => { diff --git a/src/process/OwnedLease.ts b/src/process/OwnedLease.ts index 2dd40c12d5..e6ed45b8a4 100644 --- a/src/process/OwnedLease.ts +++ b/src/process/OwnedLease.ts @@ -240,16 +240,22 @@ const tryAcquireLease = async ( : { acquired: false, owner: existingOwner }; } let released = false; + let releasing: Promise | undefined; const lease: OwnedLease = { socketPath, - 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; + 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; }, }; if (securePath) diff --git a/src/process/OwnedMcpStdioTransport.ts b/src/process/OwnedMcpStdioTransport.ts index ae94ef3413..66800a609b 100644 --- a/src/process/OwnedMcpStdioTransport.ts +++ b/src/process/OwnedMcpStdioTransport.ts @@ -17,6 +17,10 @@ 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 = ( @@ -43,7 +47,10 @@ export class OwnedMcpStdioTransport implements Transport { #bytes = 0; #started = false; #closed = false; - #closePromise: Promise | undefined; + readonly #cleanup = new AnalysisCleanupLease( + () => this.#stop(), + [this.runId], + ); #starting: Promise | undefined; #failure: OwnedMcpTransportFailure | undefined; @@ -139,11 +146,8 @@ export class OwnedMcpStdioTransport implements Transport { /** Verifies owned process exit before declaring the transport closed. */ close(): Promise { - this.#closePromise ??= this.#stop().catch((cause: unknown) => { - this.#closePromise = undefined; - throw cause; - }); - return this.#closePromise; + this.#closed = true; + return this.#cleanup.release(); } /** Keep the interruption reason distinct from errors produced by its teardown. */ @@ -182,7 +186,6 @@ export class OwnedMcpStdioTransport implements Transport { } async #stop(): Promise { - this.#closed = true; this.#frameBuffer = Buffer.alloc(0); this.#frameLength = 0; // A cancellation may race acquisition. Join it before inspecting ownership. @@ -213,11 +216,18 @@ export class OwnedMcpStdioTransport implements Transport { await inputClosed; input?.off("error", this.#error); this.onclose?.(); - if (stopped?.status === "incomplete") - throw new ProviderCleanupError(this.settings.provider, [this.runId], { - reason: stopped.reason, - stderr: this.diagnostics()?.stderr ?? "", - }); + 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; + } } readonly #error = (error: Error): void => { diff --git a/src/process/OwnedWorkerRuntime.ts b/src/process/OwnedWorkerRuntime.ts index 4980e3ceb0..50acd275f3 100644 --- a/src/process/OwnedWorkerRuntime.ts +++ b/src/process/OwnedWorkerRuntime.ts @@ -45,8 +45,21 @@ export class WorkerCleanupRegistry { } async retry(except?: RetainedWorkerCleanup): Promise { - for (const owner of [...this.#pending]) - if (owner !== except) await owner.retryRetainedCleanup(); + 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", + ); } } diff --git a/src/process/ProcessOwnership.exitScope.test.ts b/src/process/ProcessOwnership.exitScope.test.ts index 86b19d9023..49aaf8ffa9 100644 --- a/src/process/ProcessOwnership.exitScope.test.ts +++ b/src/process/ProcessOwnership.exitScope.test.ts @@ -95,7 +95,10 @@ it.each([ ), processIdentities: async (selected) => new Map( - selected.map((process) => { + selected.map(({ pid }) => { + const process = [parent, child].find((entry) => entry.pid === pid); + if (process === undefined) + return [pid, { state: "absent", reason: "not-in-snapshot" }]; if (process.pid === parent.pid && ++parentReads > 2) { reparented = true; return [ @@ -170,10 +173,15 @@ it("shares native exit waiting before signaling healthy groups in one cleanup at processIdentities: async (selected) => { clock += 25; return new Map( - selected.map((process) => [ - process.pid, - identity(process, process.pid !== root.pid), - ]), + 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), + ]; + }), ); }, signalGroup: vi.fn((group) => { diff --git a/src/process/ProcessOwnership.identity.test.ts b/src/process/ProcessOwnership.identity.test.ts index 1efe9d8570..bc35bddaff 100644 --- a/src/process/ProcessOwnership.identity.test.ts +++ b/src/process/ProcessOwnership.identity.test.ts @@ -20,6 +20,126 @@ 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, @@ -39,7 +159,7 @@ it("keeps an unresolved earlier generation unknown when a later unrelated proces entries.map((entry) => [ entry.pid, readable - ? observedIdentity(entry, "later-generation") + ? observedIdentity(row, "later-generation") : { state: "unavailable" as const, reason: "identity unavailable" }, ]), ), @@ -87,7 +207,7 @@ it.each([undefined, "run-token"])( new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, "current-start"), + observedIdentity(row, "current-start"), ]), ), ), @@ -180,7 +300,7 @@ it("validates and cleans an exact-token owner without a capture baseline", async entries.map((entry) => [ entry.pid, live - ? observedIdentity(entry, "owner-start") + ? observedIdentity(process, "owner-start") : { state: "absent" as const, reason: "not-in-snapshot" as const, @@ -239,7 +359,7 @@ it entries.map((entry) => [ entry.pid, { - ...observedIdentity(entry, "reused-current-generation"), + ...observedIdentity(row, "reused-current-generation"), effectiveUid, }, ]), @@ -299,7 +419,7 @@ it entries.map((entry) => [ entry.pid, { - ...observedIdentity(entry, "current-generation"), + ...observedIdentity(row, "current-generation"), effectiveUid: process.getuid!(), ...(environmentRestriction === undefined ? {} @@ -365,7 +485,7 @@ it.each(["omitted", "throws"] as const)( new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, "current-generation"), + observedIdentity(row, "current-generation"), ]), ), ); @@ -415,7 +535,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(entry, replaced ? "new-owner" : "old-owner"), + : observedIdentity(row, replaced ? "new-owner" : "old-owner"), ]), ), runTokens: async (entries) => { @@ -616,7 +736,7 @@ it("requires a scoped current relation before exempting a withheld token", async entries.map((entry) => [ entry.pid, nativeRelation - ? observedIdentity(entry, "opaque-generation") + ? observedIdentity(row, "opaque-generation") : { state: "readable" as const, identity: "opaque-generation", @@ -947,7 +1067,7 @@ describe("capture-local process identity baselines", () => { entry.pid === 902 ? { state: "unavailable", reason: "fixture" } : observedIdentity( - entry, + processes.find(({ pid }) => pid === entry.pid)!, entry.pid === 899 ? "stable-start" : "reused-start", ), ]), @@ -1011,7 +1131,7 @@ describe("capture-local process identity baselines", () => { new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, identity), + observedIdentity(process, identity), ]), ), ); @@ -1248,7 +1368,7 @@ it("revalidates identity after the final environment token read", async () => { new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, identity), + observedIdentity(process, identity), ]), ), ); @@ -1297,7 +1417,10 @@ it("fails closed when candidate PIDs change identity during token reading", asyn new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, identity), + observedIdentity( + processes.find(({ pid }) => pid === entry.pid)!, + identity, + ), ]), ), ); @@ -1361,7 +1484,7 @@ it("uses the launcher's native relation instead of stale process-table fields", members.map((member) => [ member.pid, observedIdentity( - { ...member, parentPid: 1, processGroupId: 100 }, + { ...row, parentPid: 1, processGroupId: 100 }, "start-100", ), ]), @@ -1451,7 +1574,7 @@ it("refuses to signal when PID identity changes after token validation", async ( new Map( entries.map((entry) => [ entry.pid, - observedIdentity(entry, identity), + observedIdentity(process, identity), ]), ), ); diff --git a/src/process/ProcessOwnership.lineage.test.ts b/src/process/ProcessOwnership.lineage.test.ts index 0db4155fa0..78d967e28c 100644 --- a/src/process/ProcessOwnership.lineage.test.ts +++ b/src/process/ProcessOwnership.lineage.test.ts @@ -58,8 +58,17 @@ it.each( runId === undefined ? {} : { REA_PROCESS_RUN_ID: runId }, ...(native ? { - processIdentities: async (rows: readonly ProcessTableEntry[]) => - new Map(rows.map((row) => [row.pid, nativeIdentity(row)])), + processIdentities: async ( + rows: readonly Pick[], + ) => + new Map( + rows.map(({ pid }) => { + const row = [launcher].find( + (candidate) => candidate.pid === pid, + )!; + return [row.pid, nativeIdentity(row)]; + }), + ), } : {}), signalGroup, @@ -152,7 +161,12 @@ it.each( ), processIdentities: async (rows) => new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [ + nativeLauncher, + observing ? foreign : { ...foreign, parentPid: 1 }, + independentlyOwned, + ].find((candidate) => candidate.pid === pid)!; if (!observing && row.pid !== foreign.pid) return [row.pid, { state: "absent", reason: "not-in-snapshot" }]; if ( @@ -217,7 +231,14 @@ it("preserves independently token-owned ancestry when the launcher identity mism ]), ), processIdentities: async (rows) => - new Map(rows.map((row) => [row.pid, nativeIdentity(row)])), + new Map( + rows.map(({ pid }) => { + const row = [foreignLauncher, ownedParent, unreadableChild].find( + (candidate) => candidate.pid === pid, + )!; + return [row.pid, nativeIdentity(row)]; + }), + ), signalGroup: vi.fn(), }; @@ -267,19 +288,24 @@ it.each(["parent", "group"] as const)( processIdentities: async (rows) => { identityReads += 1; return new Map( - 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, - ), - ]), + 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, + ), + ]; + }), ); }, signalGroup: vi.fn(), @@ -348,7 +374,10 @@ it.each([ processIdentities: async (rows) => { identityReads.push(rows.map(({ pid }) => pid)); return new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [launcher, child, grandchild].find( + (candidate) => candidate.pid === pid, + )!; if ( phase === "exited" || (phase === "omitted" && row.pid !== child.pid) @@ -439,12 +468,17 @@ it("retains a live native descendant behind an exited table ancestor", async () }, processIdentities: async (rows) => new Map( - rows.map((row): [number, ProcessIdentityObservation] => [ - row.pid, - row.pid === child.pid - ? { state: "absent", reason: "not-in-snapshot" } - : nativeIdentity(row), - ]), + 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), + ]; + }), ), signalGroup: vi.fn(), }; @@ -493,21 +527,26 @@ it.each(["unavailable-relation", "absent"] as const)( ), processIdentities: async (rows) => new Map( - 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), - ]), + 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), + ]; + }), ), signalGroup: vi.fn(), }; @@ -547,12 +586,17 @@ 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((row): [number, ProcessIdentityObservation] => [ - row.pid, - row.pid === launcher.pid - ? { state: "unavailable", reason: "native identity denied" } - : nativeIdentity(row), - ]), + 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), + ]; + }), ), signalGroup: vi.fn(), }; @@ -588,14 +632,19 @@ it("uses a readable native relation to exclude a stale table descendant", async }, processIdentities: async (rows) => new Map( - rows.map((row) => [ - row.pid, - nativeIdentity( - row.pid === child.pid - ? { ...row, parentPid: 1, processGroupId: 200 } - : row, - ), - ]), + 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, + ), + ]; + }), ), signalGroup: vi.fn(), }; @@ -626,17 +675,25 @@ it("does not retain table scope after a native relation resolves as unrelated", processIdentities: async (rows) => { identityReads += 1; return new Map( - 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 }), - ]), + 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 }), + ]; + }), ); }, signalGroup: vi.fn(), @@ -678,7 +735,10 @@ it.each(["none", "unresolved", "known"] as const)( processIdentities: async (rows) => { identityReads += 1; return new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [launcher, child].find( + (candidate) => candidate.pid === pid, + )!; if (row.pid === launcher.pid) return [ row.pid, @@ -699,7 +759,10 @@ it.each(["none", "unresolved", "known"] as const)( row.pid, { ...nativeIdentity(row), - relation: { state: "unavailable", reason: "relation denied" }, + relation: { + state: "unavailable", + reason: "relation denied", + }, }, ]; return [ @@ -774,7 +837,10 @@ it.each(["replaced", "relation-unavailable-replaced", "unavailable"] as const)( processIdentities: async (rows) => { identityReads += 1; return new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [launcher, child, detached].find( + (candidate) => candidate.pid === pid, + )!; const replacedAncestor = row.pid === child.pid && ancestorGeneration !== "unavailable"; if ( @@ -886,19 +952,24 @@ it("keeps an exiting descendant whose termination cannot be confirmed", async () processIdentities: async (rows) => { identityReads += 1; return new Map( - 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" }, - ]), + 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" }, + ]; + }), ); }, signalGroup, @@ -970,7 +1041,10 @@ it.each( processIdentities: async (rows) => { identityReads += 1; return new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [launcher, child].find( + (candidate) => candidate.pid === pid, + )!; if (row.pid !== stale.pid) return [ row.pid, @@ -1060,13 +1134,19 @@ it.each(["absent", "zombie", "dead"] as const)( processIdentities: async (rows) => { identityReads += 1; return new Map( - rows.map((row): [number, ProcessIdentityObservation] => { + rows.map(({ pid }): [number, ProcessIdentityObservation] => { + const row = [launcher, child, detached].find( + (candidate) => candidate.pid === pid, + )!; if (row.pid === child.pid) return [ row.pid, identityReads === 1 ? exitedIdentity(row, exit) - : { ...nativeIdentity(replacement), identity: "replacement" }, + : { + ...nativeIdentity(replacement), + identity: "replacement", + }, ]; if ( phase === "exited" || diff --git a/src/process/ProcessOwnership.test.ts b/src/process/ProcessOwnership.test.ts index 92b407aee8..f150169140 100644 --- a/src/process/ProcessOwnership.test.ts +++ b/src/process/ProcessOwnership.test.ts @@ -337,7 +337,12 @@ describe("related sweep candidates during cleanup", () => { let childReads = 0; const adapter: ProcessOwnershipHost = { listProcesses: async () => processes, - processIdentities: async (members) => identitiesWithRelations(members), + processIdentities: async (members) => + identitiesWithRelations( + processes.filter(({ pid }) => + members.some((member) => member.pid === pid), + ), + ), environment: async () => ({}), runTokens: async (members) => new Map( diff --git a/src/process/ProcessOwnership.tokenAvailability.test.ts b/src/process/ProcessOwnership.tokenAvailability.test.ts index d71e57f634..88754b209c 100644 --- a/src/process/ProcessOwnership.tokenAvailability.test.ts +++ b/src/process/ProcessOwnership.tokenAvailability.test.ts @@ -3,6 +3,9 @@ import { expect, it, vi } from "vitest"; const source = vi.hoisted(() => ({ present: true, reason: "process_not_in_snapshot", + identity: undefined as string | undefined, + runId: undefined as string | undefined, + reads: [] as number[][], })); vi.mock("./ExecFileOutput.js", () => ({ execFileOutput: async () => ({ @@ -11,12 +14,33 @@ vi.mock("./ExecFileOutput.js", () => ({ })); vi.mock("./DarwinProcessRunTokenReader.js", () => ({ createDarwinProcessRunTokenReader: () => ({ - read: async () => - new Map([[100, { state: "unavailable", reason: source.reason }]]), + read: async (processes: readonly { readonly pid: number }[]) => { + source.reads.push(processes.map(({ pid }) => pid)); + return new Map( + processes.map(({ pid }) => [ + pid, + source.runId === undefined + ? { state: "unavailable" as const, reason: source.reason } + : { state: "readable" as const, runId: source.runId }, + ]), + ); + }, + identities: async (processes: readonly { readonly pid: number }[]) => + new Map( + processes.map(({ pid }) => [ + pid, + source.identity === undefined + ? { state: "absent" as const, reason: "not-in-snapshot" as const } + : { state: "readable" as const, identity: source.identity }, + ]), + ), }), })); -import { createSystemProcessOwnershipHost } from "./ProcessOwnershipObservation.js"; +import { + createSystemProcessOwnershipHost, + observeProcessStartIdentity, +} from "./ProcessOwnershipObservation.js"; import { ProcessRunTokenReadError } from "./ProcessRunTokenObservations.js"; it.each([ @@ -26,7 +50,11 @@ it.each([ ])( "retains native run-token availability at the host adapter: %j", async (state) => { - Object.assign(source, state); + Object.assign(source, state, { + identity: undefined, + runId: undefined, + reads: [], + }); const host = createSystemProcessOwnershipHost("darwin", {}); const failure = await host .environment(100) @@ -39,3 +67,18 @@ it.each([ }); }, ); + +it("binds an omitted Darwin PID through the native token reader", async () => { + Object.assign(source, { + present: false, + reason: "process_not_in_snapshot", + identity: "start-1", + runId: "captured-run", + reads: [], + }); + const host = createSystemProcessOwnershipHost("darwin", {}); + await expect( + observeProcessStartIdentity(100, host, "captured-run"), + ).resolves.toEqual({ state: "readable", identity: "start-1" }); + expect(source.reads).toEqual([[100]]); +}); diff --git a/src/process/ProcessOwnership.ts b/src/process/ProcessOwnership.ts index 98ced55ec9..a719e8eddf 100644 --- a/src/process/ProcessOwnership.ts +++ b/src/process/ProcessOwnership.ts @@ -190,14 +190,21 @@ export interface ProcessOwnershipHost { pid: number, signal?: AbortSignal, ): Promise>>; - /** Read only run-token values in one host operation when the OS supports it. */ + /** + * Read only run-token values in one host operation when the OS supports it. + * Rows carry a PID; they are not process-table facts, so a live PID omitted + * from `ps` can still be read by the native helper. + */ runTokens?( - processes: readonly ProcessTableEntry[], + processes: readonly Pick[], signal?: AbortSignal, ): Promise>; - /** Read stable identities; omit the signal during cleanup so it can finish. */ + /** + * Read stable native identities by PID; caller rows supply no native facts. + * Omit the signal during cleanup so it can finish. + */ processIdentities?( - processes: readonly ProcessTableEntry[], + processes: readonly Pick[], signal?: AbortSignal, /** Linux credential enrichment is selective; omitted means UID and environment restriction remain unknown. */ credentialPids?: ReadonlySet, @@ -595,7 +602,7 @@ const readProcessRelationSnapshot = async ( ownership: OwnedProcessGroup, processes: readonly ProcessTableEntry[], host: ProcessOwnershipHost, -): Promise => { +): Promise => { if (host.processIdentities === undefined) return { currentRelations: processes, @@ -603,12 +610,17 @@ const readProcessRelationSnapshot = async ( failures: [], }; - const identities = await readNativeProcessIdentityMap(processes, host); + const retainedPids = + ownership.evidenceOwner?.candidates.map(({ pid }) => pid) ?? []; + const observationRows = + ownership.evidenceOwner?.rowsForObservation(processes) ?? processes; + const identities = await readNativeProcessIdentityMap(observationRows, host); const possibleDescendants = new Set( descendantsOf([ownership.leaderPid], processes).map(({ pid }) => pid), ); const relevantPids = new Set([ ownership.leaderPid, + ...retainedPids, ...possibleDescendants, ...processes .filter( @@ -620,13 +632,28 @@ const readProcessRelationSnapshot = async ( ]); const currentRelations: ProcessTableEntry[] = []; const failures: ProcessOwnershipValidationFailure[] = []; - for (const process of processes) { + 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) { 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, @@ -641,6 +668,27 @@ 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 }; }; @@ -948,6 +996,7 @@ const createOwnedCleanupPlan = async ( processes, host, ); + if ("cleaned" in relationSnapshot) return relationSnapshot; const sampledProcessGroupIds = new Set( (ownership.sampledProcessGroupIds ?? []).filter( (processGroupId) => @@ -979,8 +1028,14 @@ 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( - tokenOwned.owned.map(({ processGroupId }) => processGroupId), + signalableOwned.map(({ processGroupId }) => processGroupId), ); const launcher = relationSnapshot.currentRelations.find( ({ pid }) => pid === ownership.leaderPid, @@ -988,21 +1043,36 @@ 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 && - tokenOwned.owned.length === 0 && + signalableOwned.length === 0 && escapedProcessGroupIds.size === 0 && !relationSnapshot.currentRelations.some(({ processGroupId }) => sampledProcessGroupIds.has(processGroupId), ) ) - return tokenOwned.failures.length > 0 || - relationSnapshot.failures.length > 0 + return unresolved.length > 0 ? { signalOrder: [], tokenOwnedIdentities: tokenOwned.identities, - unresolved: [...tokenOwned.failures, ...relationSnapshot.failures], + unresolved, unverified: tokenOwned.unverified, } : { @@ -1023,14 +1093,13 @@ const createOwnedCleanupPlan = async ( const processGroupIds = new Set([ownership.processGroupId]); for (const descendant of descendants) processGroupIds.add(descendant.processGroupId); - for (const process of tokenOwned.owned) + for (const process of signalableOwned) processGroupIds.add(process.processGroupId); for (const processGroupId of sampledProcessGroupIds) processGroupIds.add(processGroupId); for (const processGroupId of escapedProcessGroupIds) processGroupIds.add(processGroupId); const signalableGroups = new Set(); - const unresolved = [...tokenOwned.failures, ...relationSnapshot.failures]; const unverifiableSampledGroups = new Set(); for (const processGroupId of processGroupIds) { if (processGroupId === ownership.processGroupId) continue; @@ -1173,6 +1242,7 @@ 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( diff --git a/src/process/ProcessOwnershipIdentity.ts b/src/process/ProcessOwnershipIdentity.ts index 0fb11f4e5f..e0cf77c223 100644 --- a/src/process/ProcessOwnershipIdentity.ts +++ b/src/process/ProcessOwnershipIdentity.ts @@ -6,7 +6,9 @@ import type { /** Compare a live launcher against the identity committed at process launch. */ export const launcherIdentityFailure = ( - launcher: ProcessTableEntry, + launcher: Pick & { + readonly command?: string; + }, ownership: OwnedProcessGroup, ): string | null => { if (launcher.processGroupId !== ownership.processGroupId) @@ -16,10 +18,13 @@ export const launcherIdentityFailure = ( launcher.parentPid !== ownership.expectedParentPid ) return "owned launcher parent identity did not match"; - if ( - ownership.expectedCommand !== undefined && - !matchesOwnedProcessCommand(launcher.command, ownership.expectedCommand) - ) - return `owned launcher command identity did not match (observed=${launcher.command}; expected=${ownership.expectedCommand})`; + 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})`; + } return null; }; diff --git a/src/process/ProcessOwnershipObservation.ts b/src/process/ProcessOwnershipObservation.ts index 4f9a1ca624..5ec980fce0 100644 --- a/src/process/ProcessOwnershipObservation.ts +++ b/src/process/ProcessOwnershipObservation.ts @@ -33,6 +33,7 @@ import { import { darwinHelperCacheDirectory } from "./DarwinHelperCache.js"; import { ProcessRunTokenReadError, + readProcessRunToken, readProcessRunTokens, } from "./ProcessRunTokenObservations.js"; import type { @@ -1662,39 +1663,42 @@ export const observeProcessStartIdentity = async ( host: ProcessOwnershipHost = systemProcessOwnershipHost, expectedRunId?: string, ): Promise => { - 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([process]) + await host.processIdentities([{ pid }]) ).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; - 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; + const process = (await host.listProcesses()).find( + (entry) => entry.pid === pid, + ); + let token: ProcessRunTokenObservation | undefined; + // Darwin's environment() admits a token only after `ps` lists the PID. + // The native reader already observed this live identity by PID, so an + // omitted row uses that reader directly instead of another process table. + if (process === undefined && host.runTokens !== undefined) + token = (await host.runTokens([{ pid }])).get(pid); + if (token === undefined) { + if (process === undefined) token = await readProcessRunToken(host, pid); + else + for await (const { observation } of readProcessRunTokens(host, [process])) + token = observation; } - if (!owned) + if (token === undefined || token.state === "unavailable") return { state: "unavailable", - reason: `process ownership token was not returned for PID ${String(pid)}`, + 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)}`, }; const confirmed = await observeProcessStartIdentity(pid, host); if (confirmed === undefined || confirmed.state !== "readable") diff --git a/src/process/ProcessRunTokenObservations.ts b/src/process/ProcessRunTokenObservations.ts index b33f19a3cc..824ad20f5f 100644 --- a/src/process/ProcessRunTokenObservations.ts +++ b/src/process/ProcessRunTokenObservations.ts @@ -29,6 +29,39 @@ 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 => { + 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, @@ -57,33 +90,12 @@ export async function* readProcessRunTokens( signal?.throwIfAborted(); let observation = batch?.get(process.pid); if (observation === undefined) { - try { - const environment = - signal === undefined - ? await host.environment(process.pid) - : await host.environment(process.pid, signal); + observation = await readProcessRunToken(host, process.pid, signal); + if (observation.state === "unavailable" && batchFailure !== undefined) observation = { - state: "readable", - runId: processRunIdFromEnvironment(environment), + ...observation, + reason: `${observation.reason}; run-token batch failed: ${batchFailure}`, }; - } 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 }; diff --git a/src/process/ProcessStartIdentity.test.ts b/src/process/ProcessStartIdentity.test.ts index 8d2f532618..000e921a16 100644 --- a/src/process/ProcessStartIdentity.test.ts +++ b/src/process/ProcessStartIdentity.test.ts @@ -24,10 +24,112 @@ const host = (identity: string | undefined): ProcessOwnershipHost => ({ ], environment: async () => ({}), processIdentities: async () => - new Map([[42, { state: "readable", identity: identity ?? "" }]]), + new Map([ + [ + 42, + identity === undefined + ? { state: "absent", reason: "not-found-in-caller-pid-namespace" } + : { state: "readable", 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( diff --git a/src/process/ProviderProcess.ts b/src/process/ProviderProcess.ts index 10ce701278..b64f8cc42c 100644 --- a/src/process/ProviderProcess.ts +++ b/src/process/ProviderProcess.ts @@ -642,7 +642,8 @@ export class ProviderProcessSupervisor { stop( options: ProviderProcessStopOptions = {}, ): Promise { - this.#stopRequested = true; + if (this.launch.ownsProcessLifetime && this.#exitObservation === undefined) + this.#stopRequested = true; this.#stopPromise ??= this.#stop(options).then((result) => { if (result.status === "incomplete") this.#stopPromise = undefined; return result; diff --git a/src/process/ProviderProcessOutput.test.ts b/src/process/ProviderProcessOutput.test.ts index 12cd80260a..32553873e9 100644 --- a/src/process/ProviderProcessOutput.test.ts +++ b/src/process/ProviderProcessOutput.test.ts @@ -121,7 +121,9 @@ 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((resolve) => setImmediate(resolve)); @@ -129,6 +131,7 @@ 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", ); @@ -874,6 +877,38 @@ 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, diff --git a/src/process/TokenOwnedTermination.test.ts b/src/process/TokenOwnedTermination.test.ts index 134dec7a14..88ebd37b3c 100644 --- a/src/process/TokenOwnedTermination.test.ts +++ b/src/process/TokenOwnedTermination.test.ts @@ -138,6 +138,27 @@ 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(); diff --git a/src/process/TokenOwnedTermination.ts b/src/process/TokenOwnedTermination.ts index cf2596ea6f..1f2f6950a9 100644 --- a/src/process/TokenOwnedTermination.ts +++ b/src/process/TokenOwnedTermination.ts @@ -82,11 +82,13 @@ const awaitExit = async ( } }; -const send = (deliver: () => void): void => { +const send = (deliver: () => void): boolean => { try { deliver(); + return true; } catch { - // A process that exited first has nothing left to stop. + // Exit or a failed delivery does not establish that SIGTERM was sent. + return false; } }; @@ -113,16 +115,20 @@ export const terminateTokenOwnedProcesses = async ( } const inGroup = (fact: ProcessOwnershipFact) => processGroupOf(fact) === ownership.processGroupId; - const launchGroupSignalled = owned.some(inGroup); - if (launchGroupSignalled) + const launchGroupSignalled = + owned.some(inGroup) && 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"); - }); - await awaitExit(owned, deadline, host.ownership); + }) + ) + signalled.push(fact); + await awaitExit(signalled, deadline, host.ownership); return { launchGroupSignalled }; }; diff --git a/src/process/WorkerCleanupRegistry.test.ts b/src/process/WorkerCleanupRegistry.test.ts index d44dc76474..292e36bae3 100644 --- a/src/process/WorkerCleanupRegistry.test.ts +++ b/src/process/WorkerCleanupRegistry.test.ts @@ -25,3 +25,58 @@ 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); +}); diff --git a/tests/boundary/android/jadxMcpTransport.test.ts b/tests/boundary/android/jadxMcpTransport.test.ts index 8b29cf2420..8b1752f32f 100644 --- a/tests/boundary/android/jadxMcpTransport.test.ts +++ b/tests/boundary/android/jadxMcpTransport.test.ts @@ -5,6 +5,7 @@ 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 { @@ -258,9 +259,27 @@ it("retries an incomplete owned stop while joining concurrent close callers", as await transport.start(); const firstClose = transport.close(); expect(transport.close()).toBe(firstClose); - await expect(firstClose).rejects.toMatchObject({ cleanupIncomplete: true }); + 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); 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(); }); diff --git a/tests/boundary/process/darwinProcessRunTokenReader.test.ts b/tests/boundary/process/darwinProcessRunTokenReader.test.ts index 7729165855..661ede8457 100644 --- a/tests/boundary/process/darwinProcessRunTokenReader.test.ts +++ b/tests/boundary/process/darwinProcessRunTokenReader.test.ts @@ -177,17 +177,9 @@ it("reports an actionable missing Swift compiler without installing it", async ( }, }); try { - await expect( - reader.read([ - { - pid: 1, - parentPid: 0, - processGroupId: 1, - state: "S", - command: "fixture", - }, - ]), - ).rejects.toThrow(/requires the Apple Swift compiler via xcrun/u); + await expect(reader.read([{ pid: 1 }])).rejects.toThrow( + /requires the Apple Swift compiler via xcrun/u, + ); } finally { await reader.close(); } @@ -1129,18 +1121,7 @@ it.skipIf(process.platform === "win32").each(["baseline", "tokens"] as const)( const inspection = kind === "baseline" ? host.captureBaseline?.(controller.signal) - : host.runTokens?.( - [ - { - pid: process.pid, - parentPid: process.ppid, - processGroupId: process.pid, - state: "S", - command: process.execPath, - }, - ], - controller.signal, - ); + : host.runTokens?.([{ pid: process.pid }], controller.signal); const baseline = inspection?.then( () => ({ state: "fulfilled" as const }), @@ -1189,18 +1170,16 @@ it.skipIf(process.platform === "win32")( const baselineSpy = vi .spyOn(systemProcessOwnershipHost, "captureBaseline") .mockImplementation(async (signal) => { - await reader.identities( - [ - { - pid: process.pid, - parentPid: process.ppid, - processGroupId: process.pid, - state: "S", - command: process.execPath, - }, - ], - signal, - ); + const processes = [ + { + pid: process.pid, + parentPid: process.ppid, + processGroupId: process.pid, + state: "S", + command: process.execPath, + }, + ]; + await reader.identities(processes, signal); return []; }); const launchSpy = vi.spyOn(electron, "launch");