From b0429189067feb1722869473b899db64099eac8f Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 12 Sep 2026 21:08:18 -0500 Subject: [PATCH] Fence native runner admission on durable process ownership Cancel pending launches and contain verified processes before reporting failed ownership. Keep unresolved checkpoints recoverable. Co-Authored-By: Paperclip --- .../durable-prp-control-plane.ts | 35 ++- .../src/live/runnerd-codex-transport.test.ts | 231 ++++++++++++++++- .../src/live/runnerd-codex-transport.ts | 237 +++++++++++++++--- .../native-session-executor.test.ts | 114 +++++++++ .../native-runtime/native-session-executor.ts | 190 ++++++++++++-- 5 files changed, 746 insertions(+), 61 deletions(-) diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index fadddfc957..e4937dd386 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -272,6 +272,16 @@ export interface RunnerProcessHandle { kill(signal?: NodeJS.Signals | number): boolean; }; completion: Promise; + /** Async launch identity AND launcher-side ownership persistence. Consumers + * must await this before admission and must not publish ownership a second time. + * Reject only after attempted containment and observe rejection immediately. */ + ready?: Promise; + /** Latch cancellation before async launch/readiness; prevent a later dispatch + * or contain an already dispatched process through launcher-owned authority. */ + cancelPendingLaunch?(): void; + /** Exact ready/completion rejection and the launcher's verified containment + * outcome. Never infer containment from a rejected lifetime promise alone. */ + ownershipFailure?: { error: unknown; containment: "confirmed" | "unconfirmed" }; processGroupId?: number | null; startedAt?: string; /** Relaunches the same immutable process specification with a fresh ticket. */ @@ -3513,14 +3523,23 @@ export function spawnRunner(options: { const command = options.runnerBinaryPath ?? runnerBinary; const environment = runnerEnvironment(options.ticket, options.environment); - const withRestart = (handle: RunnerProcessHandle): RunnerProcessHandle => ({ - ...handle, - // Remote launchers resolve process identity after returning the handle. - get startedAt() { - return handle.startedAt; - }, - restart: (ticket) => spawnRunner({ ...options, ticket }), - }); + const withRestart = (handle: RunnerProcessHandle): RunnerProcessHandle => { + // Consumers await the original promises, but a remote launch can reject + // before that await is installed. Observe without replacing their results. + void handle.ready?.catch(() => undefined); + void handle.completion.catch(() => undefined); + return { + ...handle, + // Remote launchers resolve process identity after returning the handle. + get startedAt() { + return handle.startedAt; + }, + get ownershipFailure() { + return handle.ownershipFailure; + }, + restart: (ticket) => spawnRunner({ ...options, ticket }), + }; + }; if (options.processLauncher !== undefined) { return withRestart( options.processLauncher({ command, args, cwd: packageRoot, environment }), diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 69532d1918..0d4e152cee 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -2085,9 +2085,178 @@ it("publishes spawned runner ownership before waiting for provider startup", asy } }); -it.each([false, true])("persists each recovered runner identity before declaring recovery complete (save fails: %s)", async (failOwnershipSave) => { +it.each([ + { phase: "initial", rejectSave: true, stop: "none" }, + { phase: "replacement", rejectSave: true, stop: "none" }, + { phase: "initial", rejectSave: false, stop: "none" }, + { phase: "initial", rejectSave: false, stop: "close" }, + { phase: "replacement", rejectSave: false, stop: "deadline" }, +])("fences asynchronous $phase ownership admission (save rejects: $rejectSave, stop: $stop)", async ({ phase, rejectSave, stop }) => { + const stateDirectory = await mkdtemp(join(tmpdir(), "runner-async-ownership-")); + const handles: durableControlPlane.RunnerProcessHandle[] = []; + let rejectOwnership!: () => void; + const barrier = new Promise((resolve) => { rejectOwnership = resolve; }); + let readyRejected = false; + const launcherOwnershipSave = vi.fn(); + const transportOwnershipSave = vi.fn(); + const realSpawn = durableControlPlane.spawnRunner; + const wrap = (handle: durableControlPlane.RunnerProcessHandle): durableControlPlane.RunnerProcessHandle => { + handles.push(handle); + const selected = handles.length === (phase === "initial" ? 1 : 2); + if (!selected) return { ...handle, restart: handle.restart ? (ticket) => wrap(handle.restart!(ticket)) : undefined }; + let publishedPid: number | undefined; + let ownershipFailure: durableControlPlane.RunnerProcessHandle["ownershipFailure"]; + let cancelLaunch!: (error: Error) => void; + const cancelled = new Promise((_resolve, reject) => { cancelLaunch = reject; }); + void cancelled.catch(() => undefined); + const ready = (async () => { + try { await Promise.race([barrier, cancelled]); } + catch (error) { + handle.child.kill("SIGKILL"); + await handle.completion; + ownershipFailure = { error, containment: "confirmed" }; + throw error; + } + publishedPid = handle.child.pid; + launcherOwnershipSave({ pid: publishedPid }); + if (!rejectSave) return; + // Simulate remote ownership persistence rejecting after its own process + // cleanup. The transport must independently retire its authority route. + handle.child.kill("SIGKILL"); + await handle.completion; + readyRejected = true; + const error = new Error("fixture async ownership unavailable"); + ownershipFailure = { error, containment: "confirmed" }; + throw error; + })(); + void ready.catch(() => undefined); + const completion = ready.then(() => handle.completion); + void completion.catch(() => undefined); + return { + ...handle, ready, completion, processGroupId: null, + ...(stop === "deadline" ? { cancelPendingLaunch: () => cancelLaunch(new Error("fixture launch cancelled")) } : {}), + get ownershipFailure() { return ownershipFailure; }, + child: { get pid() { return publishedPid; }, get exitCode() { return handle.child.exitCode; }, kill: (signal) => handle.child.kill(signal) }, + restart: handle.restart ? (ticket) => wrap(handle.restart!(ticket)) : undefined, + }; + }; + const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn(options))); + const diagnostics: string[] = []; + const releaseRoute = vi.fn(); + const activate = vi.fn(); + let authority!: DurablePrpControlPlane; + let connectionAttempts = 0; + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerReconnectGraceMs: stop === "deadline" ? 500 : 5_000, + closeGraceMs: 1_000, + onSpawn: transportOwnershipSave, + onDiagnostic: (message) => diagnostics.push(message), + controlPlaneRegistration: async (core) => { + authority = core; + const upgrade = core.handleUpgrade.bind(core); + vi.spyOn(core, "handleUpgrade").mockImplementation((...args) => { connectionAttempts += 1; return upgrade(...args); }); + await core.start(); return { activate, release: releaseRoute }; + }, + }); + let openingSettled = false; + const opening = bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() }); + void opening.then(() => { openingSettled = true; }, () => { openingSettled = true; }); + try { + if (phase === "replacement") { + await opening; + handles[0]!.child.kill("SIGKILL"); + authority.queueCommand("runner.drain", {}, "pending-before-ownership"); + } + const expectedHandles = phase === "initial" ? 1 : 2; + await vi.waitFor(() => expect(handles).toHaveLength(expectedHandles)); + // Real runner handshakes arrive while PID publication/persistence is + // pending. They must not consume bootstrap credentials or replay work. + await vi.waitFor(() => expect(connectionAttempts).toBeGreaterThanOrEqual(expectedHandles)); + await new Promise((resolve) => setTimeout(resolve, 75)); + expect(readyRejected).toBe(false); + expect(authority.store.state.connectionCount).toBe(expectedHandles - 1); + if (phase === "replacement") expect(authority.store.state.commands.find((command) => command.commandId === "pending-before-ownership")?.status).toBe("pending"); + else expect(authority.store.state.commands.every((command) => command.status === "pending")).toBe(true); + if (phase === "initial") { expect(openingSettled).toBe(false); expect(activate).not.toHaveBeenCalled(); } + expect(diagnostics).not.toContain("runner process restored its durable PRP session"); + if (stop !== "none") { + if (stop === "close") { + const closing = bundle.transport.close().catch(() => undefined); + await vi.waitFor(() => expect(openingSettled).toBe(true), { timeout: 1_000 }); + await closing; + } else { + await vi.waitFor(() => expect(diagnostics.some((message) => message.includes("runner_ownership_admission_deadline_exceeded"))).toBe(true), { timeout: 4_000 }); + } + rejectOwnership(); // A late successful readiness must not revive admission. + await new Promise((resolve) => setTimeout(resolve, 100)); + expect(activate).toHaveBeenCalledTimes(phase === "initial" ? 0 : 1); + expect(authority.activeRunnerConnectionCount()).toBe(0); + expect(diagnostics).not.toContain("runner process restored its durable PRP session"); + expect(handles).toHaveLength(expectedHandles); + if (stop === "deadline") expect(diagnostics.some((message) => message.includes("cleanup incomplete"))).toBe(false); + expect(() => process.kill(handles[expectedHandles - 1]!.child.pid!, 0)).toThrow(); + return; + } + rejectOwnership(); + if (!rejectSave) { + await opening; + expect(launcherOwnershipSave).toHaveBeenCalledExactlyOnceWith({ pid: handles[0]!.child.pid }); + expect(transportOwnershipSave).not.toHaveBeenCalled(); + expect(activate).toHaveBeenCalledOnce(); + expect(authority.activeRunnerConnectionCount()).toBe(1); + return; + } + await vi.waitFor(() => expect(diagnostics.some((message) => message.includes("native_runner_process_ownership_failed: fixture async ownership unavailable"))).toBe(true)); + expect(launcherOwnershipSave).toHaveBeenCalledOnce(); + expect(transportOwnershipSave).toHaveBeenCalledTimes(phase === "initial" ? 0 : 1); + expect(diagnostics).toContain("native_runner_process_ownership_failed: fixture async ownership unavailable"); + expect(diagnostics.some((message) => message.includes("cleanup incomplete"))).toBe(false); + expect(authority.activeRunnerConnectionCount()).toBe(0); + expect(() => authority.connectUrl).toThrow("not listening"); + expect(releaseRoute).toHaveBeenCalledOnce(); + expect(handles).toHaveLength(expectedHandles); + expect((await stat(join(stateDirectory, "control-plane"))).isDirectory()).toBe(true); + await expect(bundle.transport.request("thread/start", { cwd: tmpdir() })).rejects.toThrow("native_runner_process_ownership_failed"); + } finally { + rejectOwnership(); + await opening.catch(() => undefined); + await bundle.transport.close().catch(() => undefined); + for (const handle of handles) { + if (process.platform !== "win32" && handle.processGroupId) { try { process.kill(-handle.processGroupId, "SIGKILL"); } catch {} } + else handle.child.kill("SIGKILL"); + await handle.completion.catch(() => undefined); + } + spawnSpy.mockRestore(); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 15_000); + +it.each([ + { failOwnershipSave: false, parentExitsBeforeSave: false }, + { failOwnershipSave: true, parentExitsBeforeSave: false }, + { failOwnershipSave: true, parentExitsBeforeSave: true }, +])("persists each recovered runner identity before declaring recovery complete (save fails: $failOwnershipSave, parent exited: $parentExitsBeforeSave)", async ({ failOwnershipSave, parentExitsBeforeSave }) => { const stateDirectory = await mkdtemp(join(tmpdir(), "runner-replacement-ownership-")); const handles: durableControlPlane.RunnerProcessHandle[] = []; + const descendantFile = join(stateDirectory, "runner-descendants.txt"); + const runnerWrapper = join(stateDirectory, "runner-with-descendant.sh"); + const shellQuote = (value: string) => "'" + value.replaceAll("'", "'\\''") + "'"; + if (process.platform !== "win32") { + await writeFile(runnerWrapper, `#!/bin/sh +sleep 600 & +printf '%s:%s\\n' "$$" "$!" >> ${shellQuote(descendantFile)} +exec ${shellQuote(defaultCapabilityRunnerdBinary())} "$@" +`, { mode: 0o700 }); + } + const descendantRunning = (pid: number) => { + try { + process.kill(pid, 0); + const state = execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim(); + return state.length > 0 && !state.startsWith("Z"); + } catch { return false; } + }; const realSpawn = durableControlPlane.spawnRunner; const wrap = (handle: durableControlPlane.RunnerProcessHandle): durableControlPlane.RunnerProcessHandle => { const wrapped = { @@ -2098,7 +2267,10 @@ it.each([false, true])("persists each recovered runner identity before declaring handles.push(wrapped); return wrapped; }; - const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn(options))); + const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn({ + ...options, + ...(process.platform !== "win32" ? { runnerBinaryPath: runnerWrapper } : {}), + }))); let releaseOwnership!: () => void; const ownershipBarrier = new Promise((resolve) => { releaseOwnership = resolve; }); let durableOwner: { pid: number; processGroupId: number | null; startedAt: string } | null = null; @@ -2110,12 +2282,19 @@ it.each([false, true])("persists each recovered runner identity before declaring durableOwner = structuredClone(owner); }); const diagnostics: string[] = []; + let recoveryAuthority!: DurablePrpControlPlane; + const releaseRoute = vi.fn(); const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, codexArgs: fakeCodexArgs(stateDirectory), stateDirectory, lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerReconnectGraceMs: 5_000, onSpawn, onDiagnostic: (message) => diagnostics.push(message), + controlPlaneRegistration: async (authority) => { + recoveryAuthority = authority; + await authority.start(); + return { release: releaseRoute }; + }, }); try { await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() }); @@ -2124,6 +2303,23 @@ it.each([false, true])("persists each recovered runner identity before declaring await vi.waitFor(() => expect(onSpawn).toHaveBeenCalledTimes(2)); expect(durableOwner).toEqual(originalOwner); expect(diagnostics).not.toContain("runner process restored its durable PRP session"); + let replacementDescendant: number | null = null; + if (process.platform !== "win32") { + await vi.waitFor(async () => { + const rows = (await readFile(descendantFile, "utf8")).trim().split("\n"); + const row = rows.find((line) => line.startsWith(`${handles[1]!.child.pid}:`)); + expect(row).toBeDefined(); + replacementDescendant = Number(row!.split(":")[1]); + expect(descendantRunning(replacementDescendant)).toBe(true); + }); + } + if (parentExitsBeforeSave) { + handles[1]!.child.kill("SIGKILL"); + await handles[1]!.completion; + // The detached group can outlive its leader. Rejection must still fence + // that group's descendant even though completion is already resolved. + if (replacementDescendant !== null) expect(descendantRunning(replacementDescendant)).toBe(true); + } releaseOwnership(); if (failOwnershipSave) { await vi.waitFor(() => expect(diagnostics).toContain("native_runner_process_ownership_failed: fixture durable ownership unavailable")); @@ -2131,6 +2327,14 @@ it.each([false, true])("persists each recovered runner identity before declaring expect(diagnostics).not.toContain("runner process restored its durable PRP session"); await expect(bundle.transport.request("thread/start", { cwd: tmpdir() })).rejects.toThrow("native_runner_process_ownership_failed"); expect(handles).toHaveLength(2); + // Assert fencing before the test's finally calls transport.close(): a + // terminal failure alone must not leave an unowned runner authorized. + expect(() => process.kill(handles[1]!.child.pid!, 0)).toThrow(); + if (replacementDescendant !== null) expect(descendantRunning(replacementDescendant)).toBe(false); + expect(recoveryAuthority.activeRunnerConnectionCount()).toBe(0); + expect(() => recoveryAuthority.connectUrl).toThrow("not listening"); + expect(releaseRoute).toHaveBeenCalledOnce(); + expect((await stat(join(stateDirectory, "control-plane"))).isDirectory()).toBe(true); return; } await vi.waitFor(() => expect(diagnostics).toContain("runner process restored its durable PRP session")); @@ -2148,7 +2352,9 @@ it.each([false, true])("persists each recovered runner identity before declaring releaseOwnership(); await bundle.transport.close().catch(() => undefined); for (const handle of handles) { - if (handle.child.exitCode === null) handle.child.kill("SIGKILL"); + if (process.platform !== "win32" && handle.processGroupId) { + try { process.kill(-handle.processGroupId, "SIGKILL"); } catch { /* Already exited. */ } + } else if (handle.child.exitCode === null) handle.child.kill("SIGKILL"); await handle.completion.catch(() => undefined); } spawnSpy.mockRestore(); @@ -2156,6 +2362,25 @@ it.each([false, true])("persists each recovered runner identity before declaring } }, 15_000); +it("delegates unpersisted custom runner containment without host-signalling its reported process group", () => { + const kill = vi.fn(() => true); + const hostSignal = vi.spyOn(process, "kill").mockImplementation(() => { + throw new Error("remote PID must not be signalled on this host"); + }); + try { + const handle = { + child: { pid: 12345, kill }, + processGroupId: 12345, + completion: Promise.resolve({ code: 0, signal: null }), + } as unknown as durableControlPlane.RunnerProcessHandle; + runnerdRecoveryInternals.signalUnpersistedRunner(handle, true); + expect(kill).toHaveBeenCalledExactlyOnceWith("SIGKILL"); + expect(hostSignal).not.toHaveBeenCalled(); + } finally { + hostSignal.mockRestore(); + } +}); + it("launches runnerd with its production durable outbox limits", () => { expect(runnerdLaunchProfileInternals.maxOutboxBytes).toBe(16 * 1024 * 1024); expect(runnerdLaunchProfileInternals.p0ReserveBytes).toBe(1024 * 1024); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index b97e566316..51094355a4 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3337,6 +3337,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { readonly #failureSignal: Promise; #rejectFailureSignal!: (error: Error) => void; #runnerRecoveryInProgress = false; + #runnerOwnershipAdmission: Promise | null = null; + #runnerOwnershipPendingHandle: RunnerProcessHandle | null = null; + readonly #runnerOwnershipAbort = new AbortController(); + readonly #cancelledRunnerLaunches = new WeakSet(); #startupComplete = false; #startupFailureCode = "native_runner_process_exited"; #controlPlaneCheckpoint: @@ -3935,20 +3939,158 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }; } - async #publishSpawnedProcess(handle: RunnerProcessHandle): Promise { - this.#evidence.runnerPid = handle.child.pid ?? null; - this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null; - this.#publish(); - if (handle.child.pid !== undefined) { - await this.options.onSpawn?.({ - pid: handle.child.pid, - processGroupId: handle.processGroupId ?? null, - // Persist the same original process identity exposed to recovery. - startedAt: this.processInfo().startedAt, - }); + async #awaitRunnerOwnershipAdmission(core: DurablePrpControlPlane): Promise { + try { + await this.#runnerOwnershipAdmission; + return true; + } catch { + // Close pending handshakes before returning to the core's post-admission + // credential recheck. Do not await the outer fence here: it joins these + // callbacks after stopping ingress and would otherwise wait on itself. + await core.stop(); + return false; } } + #cancelPendingRunnerLaunch(handle: RunnerProcessHandle): void { + if (this.#cancelledRunnerLaunches.has(handle)) return; + this.#cancelledRunnerLaunches.add(handle); + if (handle.cancelPendingLaunch) { + handle.cancelPendingLaunch(); + } else if (handle.ready) { + // Compatibility for custom launchers without a cancellation latch: a late + // PID must still be signalled through its own handle, never on this host. + void handle.ready.then(() => { + if (!this.#controllerDetachedForRestart) signalUnpersistedRunner(handle, this.options.runnerProcessLauncher !== undefined); + }).catch(() => undefined); + } + } + + async #publishSpawnedProcess(handle: RunnerProcessHandle, deadline?: number): Promise { + void handle.completion.catch(() => undefined); + this.#runnerOwnershipPendingHandle = handle; + const assertActive = () => { + if (this.#closed) throw new Error("PRP Codex transport is closed"); + this.#throwIfFailed(); + if (deadline !== undefined && Date.now() >= deadline) throw new Error("runner_ownership_admission_deadline_exceeded"); + }; + let stopped!: () => void; + const closeSignal = new Promise((_resolve, reject) => { + stopped = () => reject(new Error("PRP Codex transport is closed")); + this.#runnerOwnershipAbort.signal.addEventListener("abort", stopped, { once: true }); + if (this.#runnerOwnershipAbort.signal.aborted) stopped(); + }); + let deadlineTimer: NodeJS.Timeout | undefined; + const deadlineSignal = new Promise((_resolve, reject) => { + if (deadline !== undefined) deadlineTimer = setTimeout(() => reject(new Error("runner_ownership_admission_deadline_exceeded")), Math.max(0, deadline - Date.now())); + }); + const ownership = (async () => { + await handle.ready; + assertActive(); + if (handle.ready && handle.child.pid === undefined) { + throw new Error("runner_ready_without_process_identity"); + } + this.#evidence.runnerPid = handle.child.pid ?? null; + this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null; + this.#publish(); + if (handle.child.pid !== undefined && handle.ready === undefined) { + await this.options.onSpawn?.({ + pid: handle.child.pid, + processGroupId: handle.processGroupId ?? null, + startedAt: this.processInfo().startedAt, + }); + } + assertActive(); + })(); + const admission = Promise.race([ownership, closeSignal, deadlineSignal, this.#failureSignal]); + this.#runnerOwnershipAdmission = admission; + try { + await admission; + assertActive(); + } catch (error) { + if (this.#closed) throw error; // close/detach owns route and process disposition. + throw await this.#fenceFailedRunnerOwnership(handle, error); + } finally { + this.#runnerOwnershipAbort.signal.removeEventListener("abort", stopped); + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); + if (this.#runnerOwnershipPendingHandle === handle) this.#runnerOwnershipPendingHandle = null; + } + } + + async #fenceFailedRunnerOwnership(handle: RunnerProcessHandle, error: unknown): Promise { + // The process already holds a valid bootstrap ticket. Fence its + // route and process before exposing failure; caller.close() may never + // run. Keep the durable files/checkpoint and sandbox lease for recovery. + const core = this.#core; + core?.disconnectActiveRunner(); + const releaseRoute = this.#controlPlaneRelease; + const cleanup = Promise.allSettled([ + Promise.resolve().then(async () => { + this.#cancelPendingRunnerLaunch(handle); + const ownershipFailure = handle.ownershipFailure; + if (handle.ready && ownershipFailure && ownershipFailure.error === error) { + if (ownershipFailure.containment !== "confirmed") throw error; + // The launcher already joined verified remote containment before + // rejecting readiness/lifetime. Do not signal again or misclassify + // that same expected rejection as a new cleanup failure. + this.#evidence.runnerExited = true; + this.#evidence.runnerExitCode = handle.child.exitCode; + this.#evidence.runnerSignal = handle.child.signalCode ?? null; + return; + } + // Default local launches own a detached process group. Custom + // launchers (including sandboxes) own their signalling callback; + // their reported PIDs must never be signalled on this host. + signalUnpersistedRunner(handle, this.options.runnerProcessLauncher !== undefined); + // Signal dispatch is not exit proof. Join the original handle's + // completion before reporting the replacement itself as exited. + let result; + try { + result = await handle.completion; + } catch (completionError) { + // Cancellation can finish containment after the initial getter read. + // Preserve the original admission error while recognizing only this + // exact producer-confirmed lifetime rejection as successful cleanup. + const completedFailure = handle.ownershipFailure; + if (!handle.ready || !completedFailure || completedFailure.error !== completionError || completedFailure.containment !== "confirmed") throw completionError; + this.#evidence.runnerExited = true; + this.#evidence.runnerExitCode = handle.child.exitCode; + this.#evidence.runnerSignal = handle.child.signalCode ?? null; + return; + } + this.#evidence.runnerExited = true; + this.#evidence.runnerExitCode = result.code; + this.#evidence.runnerSignal = result.signal as NodeJS.Signals | null; + }), + Promise.resolve().then(async () => { + await releaseRoute?.(); + if (this.#controlPlaneRelease === releaseRoute) this.#controlPlaneRelease = null; + }), + Promise.resolve().then(() => core?.stop()).then(() => core?.drainPendingConnectionProcessing()), + ]); + let cleanupDetail = ""; + let cleanupTimer: NodeJS.Timeout | undefined; + try { + const results = await Promise.race([ + cleanup, + new Promise((_resolve, reject) => { + cleanupTimer = setTimeout(() => reject(new Error("ownership failure cleanup timed out")), 2_000); + }), + ]); + const rejected = results.find((result) => result.status === "rejected"); + if (rejected?.status === "rejected") throw rejected.reason; + } catch (cleanupError) { + cleanupDetail = `; cleanup incomplete: ${cleanupError instanceof Error ? cleanupError.message : String(cleanupError)}`; + } finally { + if (cleanupTimer !== undefined) clearTimeout(cleanupTimer); + } + const failure = new Error( + `native_runner_process_ownership_failed: ${error instanceof Error ? error.message : String(error)}${cleanupDetail}`, + ); + this.#failTransport(failure); + return failure; + } + async #readDurableRunnerState(): Promise> { if (this.options.readRunnerState) return this.options.readRunnerState(); return record( @@ -4100,6 +4242,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (this.#closed) return; this.#controllerDetachedForRestart = true; this.#closed = true; + this.#runnerOwnershipAbort.abort(); this.#turnStartAdmission?.resolve(false); if (this.#pump !== null) clearInterval(this.#pump); this.#pump = null; @@ -4128,6 +4271,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #closeOnce(): Promise { this.#closed = true; + this.#runnerOwnershipAbort.abort(); + if (this.#runnerOwnershipPendingHandle) { + try { this.#cancelPendingRunnerLaunch(this.#runnerOwnershipPendingHandle); } + catch (error) { this.#evidence.diagnostics.push(`pending launch cancellation failed: ${error instanceof Error ? error.message : String(error)}`); } + } this.#turnStartAdmission?.resolve(false); const adoptedRunner = this.options.adoptExistingRunner; // `settled` is a durable-state assertion, not merely the absence of a @@ -4330,6 +4478,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { identity, expectedRunnerVersion: runnerArtifact.version, expectedRunnerDigest: runnerArtifact.digest, + beforeAuthenticatedConnection: async () => { await this.#awaitRunnerOwnershipAdmission(core); }, onProtocolIntegrityError: (error) => this.#failTransport(error), onSemanticToolInput: (call) => this.#handleSemanticToolInput(call), connectionLeaseTtlMs: 60 * 60 * 1_000, @@ -4706,8 +4855,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { processLauncher: this.options.runnerProcessLauncher, }); this.#handle = handle; - this.#watchRunner(handle); await this.#publishSpawnedProcess(handle); + if (this.#closed) throw new Error("PRP Codex transport is closed"); + this.#throwIfFailed(); + this.#watchRunner(handle); await registration?.activate?.(); if (registration?.failure) { void registration.failure.catch((error: unknown) => { @@ -5056,25 +5207,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { expectedRunnerDigest: runnerArtifact.digest, onProtocolIntegrityError: (error) => this.#failTransport(error), onSemanticToolInput: (call) => this.#handleSemanticToolInput(call), - ...(warmRecovery - ? { - beforeAuthenticatedConnection: async (admission) => { - // Core policy already validated the current lease and tuple. - // Receipt replay still needs the recovery claim, including the - // completed-core/lost-final-ACK window. Ordinary post-ACK lease - // reconnects no longer depend on that historical claim. - if ( - core.store.state.warmTransition?.receipt.transitionId === - warmRecovery.transitionId || - admission.warmTransitionId === warmRecovery.transitionId - ) { - await this.options.authorizeWarmTransitionRecovery?.( - "before_authentication", - ); - } - }, - } - : {}), + beforeAuthenticatedConnection: async (admission) => { + if (!await this.#awaitRunnerOwnershipAdmission(core)) return; + // Preserve warm-transition recovery admission after process ownership. + if (warmRecovery && ( + core.store.state.warmTransition?.receipt.transitionId === warmRecovery.transitionId || + admission.warmTransitionId === warmRecovery.transitionId + )) { + await this.options.authorizeWarmTransitionRecovery?.("before_authentication"); + } + }, connectionLeaseTtlMs: 60 * 60 * 1_000, }); this.#core = core; @@ -5337,8 +5479,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }); if (handle) { this.#handle = handle; - this.#watchRunner(handle); await this.#publishSpawnedProcess(handle); + if (this.#closed) throw new Error("PRP Codex transport is closed"); + this.#throwIfFailed(); + this.#watchRunner(handle); } if (oldTransitionRegistration && newTransitionRegistration) { await oldTransitionRegistration.activate?.(); @@ -6486,11 +6630,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }, ); try { - await this.#publishSpawnedProcess(recoveredHandle); + await this.#publishSpawnedProcess(recoveredHandle, deadline); } catch (error) { - this.#failTransport(new Error( - `native_runner_process_ownership_failed: ${error instanceof Error ? error.message : String(error)}`, - )); + // Publication fenced the route and process before rejecting. A + // failed ownership save is terminal, never a fresh recovery attempt. return; } if (this.#closed || this.#failure !== null) return; @@ -6608,7 +6751,29 @@ export const runnerdLaunchProfileInternals = Object.freeze({ p0ReserveBytes: RUNNERD_P0_RESERVE_BYTES, }); +// A locally spawned detached group can outlive its runner. Unlike waiting for +// completion with a timeout, unconditional containment also reaches descendants +// when the group leader exited while ownership persistence was pending. +function signalUnpersistedRunner(handle: RunnerProcessHandle, customLauncher: boolean): void { + const groupId = handle.processGroupId; + if ( + !customLauncher && process.platform !== "win32" && + Number.isSafeInteger(groupId) && (groupId ?? 0) > 0 && groupId === handle.child.pid + ) { + try { + process.kill(-groupId!, "SIGKILL"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; + } + return; + } + // A custom handle may refer to a remote PID. Only its launcher-provided + // callback has authority to signal it; never address that PID on this host. + handle.child.kill("SIGKILL"); +} + export const runnerdRecoveryInternals = Object.freeze({ + signalUnpersistedRunner, completedMaintenanceTerminalReceipt, completedMaintenanceTerminalReplayMatches, awaitProviderDrainBarrier, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 5fd3848306..2f2bf11afc 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -299,6 +299,116 @@ beforeEach(() => { }); describe("remote runner process supervision", () => { + it.runIf(process.platform === "linux" || !!process.env.PAPERCLIP_REMOTE_PROCESS_TEST_CONTAINER).each([ + "live-parent", "exited-parent", "replaced-marker", "unverified-member", "unavailable-containment", + "cancel-pending-ownership", "cancel-pending-ownership-late-rejection", + ])("contains ownership-save failure without signalling unrelated remote processes (%s)", async (scenario) => { + const root = await mkdtemp(join(process.env.PAPERCLIP_REMOTE_PROCESS_TEST_ROOT ?? tmpdir(), "remote-ownership-")); + const container = process.env.PAPERCLIP_REMOTE_PROCESS_TEST_CONTAINER; + const execute = async (command: { command?: string; args?: string[]; cwd?: string; timeoutMs?: number }) => { + const result = spawnSync(container ? "docker" : command.command!, container + ? ["exec", container, command.command!, ...(command.args ?? [])] + : command.args ?? [], { cwd: command.cwd, timeout: command.timeoutMs ?? 10_000, encoding: "utf8" }); + let stdout = result.stdout ?? ""; + if (scenario === "unavailable-containment" && command.args?.[2] === "paperclip-runner-process-identity") { + stdout = stdout.split("\n").slice(0, 4).join("\n") + "\n"; + } + return { exitCode: result.status, signal: result.signal, timedOut: result.error?.message.includes("ETIMEDOUT") ?? false, + stdout, stderr: result.stderr ?? "" }; + }; + const script = join(root, "runner.sh"), pidsPath = join(root, "pids"), unrelatedPath = join(root, "unrelated"); + await writeFile(script, `#!/bin/sh +trap '' TERM +${scenario === "unverified-member" ? "env -u PAPERCLIP_RUNNER_PROCESS_NONCE " : ""}sh -c 'trap "" TERM; while :; do sleep 1; done' --runner-id owned-child & +printf '%s %s\\n' "$$" "$!" > '${pidsPath}' +while :; do sleep 1; done +`); + const marker = join(root, "runner-process.identity"), checkpoint = join(root, "runner-state.json"); + await writeFile(checkpoint, '{"keep":"original-checkpoint"}'); + let pids: number[] = [], unrelated = 0, expectedMarker = ""; + const cancelling = scenario.startsWith("cancel-pending-ownership"); + const confirmed = scenario === "live-parent" || scenario === "exited-parent" || cancelling; + let cancelLaunch!: () => void; + let resolveOwnership!: () => void, rejectOwnership!: (error: Error) => void; + const pendingOwnership = new Promise((resolve, reject) => { resolveOwnership = resolve; rejectOwnership = reject; }); + try { + await execute({ command: "sh", args: ["-c", `nohup setsid sh -c 'echo $$ > "${unrelatedPath}"; exec sleep 300' /dev/null 2>&1 &`] }); + await vi.waitFor(async () => { unrelated = Number((await readFile(unrelatedPath, "utf8")).trim()); expect(unrelated).toBeGreaterThan(1); }); + const saveFailure = new Error("ownership-save-rejected"); + const onSpawn = vi.fn(async () => { + await vi.waitFor(async () => { pids = (await readFile(pidsPath, "utf8")).trim().split(/\s+/).map(Number); expect(pids).toHaveLength(2); }); + expectedMarker = await readFile(marker, "utf8"); + if (scenario === "exited-parent") await execute({ command: "sh", args: ["-c", `kill -KILL ${pids[0]}`] }); + if (scenario === "replaced-marker") { + const lines = expectedMarker.split("\n"); lines[1] = String(unrelated); expectedMarker = lines.join("\n"); + await writeFile(marker, expectedMarker); + } + if (cancelling) { + cancelLaunch(); + await pendingOwnership; + return; + } + throw saveFailure; + }); + const launcher = createRemoteRunnerProcessLauncher({ + target: { kind: "remote", transport: "sandbox", environmentId: "test", leaseId: "test", remoteCwd: root }, + runner: { execute } as never, remoteBinary: "/bin/sh", processIdentityPath: marker, + stateDirectory: root, diagnosticsDirectory: join(root, "diagnostics"), runnerInstanceId: "runner-ownership-test", onSpawn, + }); + const handle = launcher({ command: "/bin/sh", args: [script, "--runner-id", "runner-ownership-test"], cwd: root, environment: {} }); + cancelLaunch = () => handle.cancelPendingLaunch?.(); + const failure = await handle.completion.catch(error => error); + expect(failure).toBeInstanceOf(Error); + if (cancelling) expect(failure.message).toBe("runner_remote_process_launch_cancelled"); + else if (confirmed) expect(failure).toBe(saveFailure); + else expect(failure.message).toContain("cleanup incomplete"); + await expect(handle.ready).rejects.toBe(failure); + expect(handle.ownershipFailure).toEqual({ error: failure, containment: confirmed ? "confirmed" : "unconfirmed" }); + expect(onSpawn).toHaveBeenCalledOnce(); + if (scenario === "cancel-pending-ownership-late-rejection") rejectOwnership(new Error("late ownership save failure")); + else resolveOwnership(); + await new Promise(resolve => setTimeout(resolve, 0)); + await expect(handle.ready).rejects.toBe(failure); + handle.cancelPendingLaunch?.(); // A classified failure is immutable. + expect(handle.ownershipFailure).toEqual({ error: failure, containment: confirmed ? "confirmed" : "unconfirmed" }); + for (const pid of pids) { + const observed = await execute({ command: "sh", args: ["-c", `test ! -r /proc/${pid}/stat || test "$(awk '{print $3}' /proc/${pid}/stat)" = Z`] }); + expect(observed.exitCode, `remote owned process ${pid}`).toBe(confirmed ? 0 : 1); + } + expect((await execute({ command: "sh", args: ["-c", `kill -0 ${unrelated}`] })).exitCode).toBe(0); + expect(await readFile(checkpoint, "utf8")).toBe('{"keep":"original-checkpoint"}'); + expect(await readFile(marker, "utf8")).toBe(expectedMarker); + } finally { + for (const pid of [...pids, unrelated].filter(pid => Number.isSafeInteger(pid) && pid > 1)) { + await execute({ command: "sh", args: ["-c", `kill -KILL ${pid} 2>/dev/null || true`] }); + } + await rm(root, { recursive: true, force: true }); + } + }, 30_000); + it.each([false, true])("does not dispatch after cancellation while staging is pending (trace=%s)", async (traced) => { + let finishStaging!: () => void; + const staging = new Promise(resolve => { finishStaging = resolve; }); + const execute = vi.fn(), onSpawn = vi.fn(), onRunnerProcessSpawned = vi.fn(); + const launcher = createRemoteRunnerProcessLauncher({ + target: { kind: "remote", transport: "sandbox", environmentId: "test", leaseId: "test", remoteCwd: "/workspace" }, + runner: { execute } as never, remoteBinary: "/runtime/runnerd", processIdentityPath: "/runtime/identity", + stateDirectory: "/runtime", diagnosticsDirectory: "/runtime/diagnostics", runnerInstanceId: "runner-cancel", + ensureArtifact: () => staging, onSpawn, onRunnerProcessSpawned, + trace: traced ? { measure: (_name: string, action: () => Promise) => action() } as never : undefined, + }); + const handle = launcher({ command: "/runtime/runnerd", args: [], cwd: "/workspace", environment: {} }); + handle.cancelPendingLaunch?.(); + const failure = await handle.completion.catch(error => error); + expect(failure.message).toBe("runner_remote_process_launch_cancelled"); + await expect(handle.ready).rejects.toBe(failure); + expect(handle.ownershipFailure).toEqual({ error: failure, containment: "confirmed" }); + finishStaging(); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(execute).not.toHaveBeenCalled(); + expect(onSpawn).not.toHaveBeenCalled(); + expect(onRunnerProcessSpawned).not.toHaveBeenCalled(); + }); + it.each([false, true])("supervises detached runnerd and observes signal failures (%s)", async (signalFails) => { let launchNonce = ""; const execute = vi.fn( @@ -405,6 +515,10 @@ describe("remote runner process supervision", () => { stderr: "paperclip-runnerd: provider transport closed", }); + await expect(handle.ready).resolves.toBeUndefined(); + handle.cancelPendingLaunch?.(); + expect(handle.ownershipFailure).toBeUndefined(); + const launch = execute.mock.calls.find( ([input]) => input.args?.[2] === "paperclip-runner-launch", )?.[0]; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 8f32fcad84..a73b3aba4d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9729,7 +9729,101 @@ const REMOTE_RUNNER_IDENTITY_CHECK_SCRIPT = 'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; expected_pid=$4; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test "$pid" = "$expected_pid" && test -n "$started_at" || exit 4; kill -0 "$pid" 2>/dev/null || exit 3; if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi'; const REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT = - 'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"'; + 'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; export PAPERCLIP_RUNNER_PROCESS_NONCE="$identity_nonce"; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"'; + +// Linux proof is optional: older/non-Linux transports still launch normally. +// Capture it in the existing marker read, before publishing durable ownership. +type RemoteRunnerContainment = { pid: number; startTicks: string; sessionId: number }; +const REMOTE_RUNNER_CONTAINMENT_PROBE = String.raw` +import json, os, pathlib, sys +try: + marker = pathlib.Path(sys.argv[1]).read_text().splitlines() + pid = int(marker[1]) + fields = pathlib.Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split() + command = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0") + environment = pathlib.Path(f"/proc/{pid}/environ").read_bytes().split(b"\0") + expected = ("PAPERCLIP_RUNNER_PROCESS_NONCE=" + marker[0]).encode() + if expected not in environment or not any(command[i:i+2] == [b"--runner-id", marker[3].encode()] for i in range(len(command))): + print(json.dumps({"pending": True})) + elif int(fields[2]) == pid and int(fields[3]) == pid: + print(json.dumps({"pid": pid, "startTicks": fields[19], "sessionId": pid})) + else: + print("null") +except (OSError, ValueError, IndexError): + print("null") +`; + +// Open pidfds before checking process birth/nonce. Signals then address those +// kernel process objects, never a numeric PID that may have been recycled. +// An exited leader is safe only while every surviving session member carries +// this launch's inherited nonce. Unknown members preserve all recovery files. +const REMOTE_RUNNER_OWNERSHIP_FAILURE_CLEANUP = String.raw` +import json, os, pathlib, signal, sys, time +marker_path, nonce, runner_id, raw_pid, started_at, raw_proof = sys.argv[1:] +proof = json.loads(raw_proof) +pid = int(raw_pid) +expected_marker = [nonce, str(pid), started_at, runner_id] +expected_environment = ("PAPERCLIP_RUNNER_PROCESS_NONCE=" + nonce).encode() +if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"): + sys.exit(4) +def check_marker(): + if os.path.islink(marker_path) or pathlib.Path(marker_path).read_text().splitlines() != expected_marker: + raise RuntimeError("runner marker changed") +def members(): + found = [] + try: + check_marker() + for entry in pathlib.Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + try: + fields = (entry / "stat").read_text().rsplit(")", 1)[1].split() + if int(fields[3]) != proof["sessionId"] or fields[0] == "Z": + continue + member_pid = int(entry.name) + fd = os.pidfd_open(member_pid) + found.append(fd) + current = (entry / "stat").read_text().rsplit(")", 1)[1].split() + if fields[19] != current[19] or int(current[3]) != proof["sessionId"]: + raise RuntimeError("process identity changed") + if member_pid == pid and current[19] != proof["startTicks"]: + raise RuntimeError("runner pid reused") + if expected_environment not in (entry / "environ").read_bytes().split(b"\0"): + raise RuntimeError("unverified session member") + except (FileNotFoundError, ProcessLookupError): + continue + return found + except BaseException: + for fd in found: + os.close(fd) + raise +try: + if proof["pid"] != pid or proof["sessionId"] != pid: + raise RuntimeError("invalid containment") + deadline = time.monotonic() + 8 + kill_after = time.monotonic() + 2 + while True: + fds = members() + if not fds: + check_marker() + sys.exit(0) + try: + requested = signal.SIGKILL if time.monotonic() >= kill_after else signal.SIGTERM + for fd in fds: + try: + signal.pidfd_send_signal(fd, requested) + except ProcessLookupError: + pass + finally: + for fd in fds: + os.close(fd) + if time.monotonic() >= deadline: + sys.exit(5) + time.sleep(0.1) +except (OSError, ValueError, KeyError, RuntimeError) as error: + print(str(error), file=sys.stderr) + sys.exit(4) +`; const REMOTE_RUNNER_FAILED_IDENTITY_CLEANUP_SCRIPT = 'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; marker_wait=0; while { test ! -f "$identity_path" || test -L "$identity_path"; } && test "$marker_wait" -lt 50; do marker_wait=$((marker_wait + 1)); sleep 0.1; done; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test -n "$started_at" || exit 4; case "$pid" in ""|*[!0-9]*) exit 4 ;; esac; test "$pid" -gt 0 || exit 4; if kill -0 "$pid" 2>/dev/null; then if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi; signal_target=$pid; if command -v ps >/dev/null 2>&1; then session_id=$(ps -o sid= -p "$pid" 2>/dev/null | tr -d " ") || true; if test "$session_id" = "$pid"; then signal_target="-$pid"; fi; fi; kill -TERM -- "$signal_target" 2>/dev/null || kill -TERM "$pid" 2>/dev/null || true; term_wait=0; while kill -0 "$pid" 2>/dev/null && test "$term_wait" -lt 50; do term_wait=$((term_wait + 1)); sleep 0.1; done; if kill -0 "$pid" 2>/dev/null; then kill -KILL -- "$signal_target" 2>/dev/null || kill -KILL "$pid" 2>/dev/null || true; kill_wait=0; while kill -0 "$pid" 2>/dev/null && test "$kill_wait" -lt 50; do kill_wait=$((kill_wait + 1)); sleep 0.1; done; fi; kill -0 "$pid" 2>/dev/null && exit 5; fi; test -f "$identity_path" && test ! -L "$identity_path" || exit 4; { IFS= read -r final_nonce; IFS= read -r final_pid; IFS= read -r final_started_at; IFS= read -r final_runner_id; } < "$identity_path"; test "$final_nonce" = "$nonce" && test "$final_pid" = "$pid" && test "$final_started_at" = "$started_at" && test "$final_runner_id" = "$runner_id" || exit 4; rm -f -- "$identity_path"'; @@ -9761,7 +9855,7 @@ async function waitForRemoteRunnerProcessIdentity(input: { identityPath: string; nonce: string; runnerInstanceId: string; -}): Promise<{ pid: number; startedAt: string }> { +}): Promise<{ pid: number; startedAt: string; containment: RemoteRunnerContainment | null }> { const deadline = Date.now() + REMOTE_RUNNER_PROCESS_IDENTITY_WAIT_MS; while (Date.now() < deadline) { const result = await input.runner @@ -9769,19 +9863,30 @@ async function waitForRemoteRunnerProcessIdentity(input: { command: "sh", args: [ "-c", - 'test -f "$1" && test ! -L "$1" && cat -- "$1"', + 'test -f "$1" && test ! -L "$1" && cat -- "$1"; if command -v python3 >/dev/null 2>&1; then python3 -c "$2" "$1"; fi', "paperclip-runner-process-identity", input.identityPath, + REMOTE_RUNNER_CONTAINMENT_PROBE, ], bypassSession: true, timeoutMs: 2_000, }) .catch(() => null); + const lines = result?.stdout.trim().split("\n") ?? []; const identity = result && result.exitCode === 0 && !result.timedOut - ? parseRemoteRunnerProcessIdentity(result.stdout, input) + ? parseRemoteRunnerProcessIdentity(lines.slice(0, 4).join("\n"), input) : null; - if (identity) return identity; + if (identity) { + let proof: Record | null = null; + try { proof = JSON.parse(lines[4] ?? "null"); } catch { /* optional platform proof */ } + if (proof?.pending !== true) { + const containment = proof?.pid === identity.pid && proof?.sessionId === identity.pid + && typeof proof.startTicks === "string" && /^\d+$/.test(proof.startTicks) + ? proof as RemoteRunnerContainment : null; + return { ...identity, containment }; + } + } await new Promise((resolve) => setTimeout(resolve, 100)); } throw new Error("runner_remote_process_identity_unavailable"); @@ -9835,7 +9940,25 @@ export function createRemoteRunnerProcessLauncher(input: { nonce: string; pid: number; startedAt: string; + containment: RemoteRunnerContainment | null; } | null = null; + let ownershipFailure: { error: unknown; containment: "confirmed" | "unconfirmed" } | undefined; + let resolveReady!: () => void; + let rejectReady!: (error: unknown) => void; + const ready = new Promise((resolve, reject) => { resolveReady = resolve; rejectReady = reject; }); + void ready.catch(() => undefined); + let launchCancelled = false; + let launchDispatched = false; + let readySucceeded = false; + const cancelledError = new Error("runner_remote_process_launch_cancelled"); + let rejectCancellation!: (error: Error) => void; + const cancellation = new Promise((_resolve, reject) => { rejectCancellation = reject; }); + void cancellation.catch(() => undefined); + const cancelPendingLaunch = () => { + if (readySucceeded || ownershipFailure || launchCancelled) return; + launchCancelled = true; + rejectCancellation(cancelledError); + }; const child: RunnerProcessHandle["child"] = { pid: undefined, exitCode: null, @@ -9885,15 +10008,16 @@ export function createRemoteRunnerProcessLauncher(input: { const completion = (async () => { if (input.ensureArtifact) { if (input.trace) { - await input.trace.measure( + await Promise.race([input.trace.measure( "runner.runtime.stage", input.ensureArtifact, { parentName: "runner.session.startup" }, - ); + ), cancellation]); } else { - await input.ensureArtifact(); + await Promise.race([input.ensureArtifact(), cancellation]); } } + if (launchCancelled) throw cancelledError; const launchStartedAtMs = Date.now(); // The provider's onSpawn callback is optional and some sandbox command // runners cannot report a remote pid until after the command has begun @@ -9924,6 +10048,10 @@ export function createRemoteRunnerProcessLauncher(input: { // into its own session instead; its own bounded diagnostics directory and // durable PRP state remain the authorities, and the controller monitors // the exact persisted process identity below. + if (launchCancelled) throw cancelledError; + launchDispatched = true; + // After dispatch, cancellation must still adopt the exact identity before + // containment. Abandoning the RPC here could strand an unknown process. const launchResult = await runner.execute({ command: "sh", args: [ @@ -9951,7 +10079,7 @@ export function createRemoteRunnerProcessLauncher(input: { : "runner_remote_process_launch_failed", ); } - let identity: { pid: number; startedAt: string }; + let identity: { pid: number; startedAt: string; containment: RemoteRunnerContainment | null }; try { identity = await waitForRemoteRunnerProcessIdentity({ runner, @@ -9983,11 +10111,36 @@ export function createRemoteRunnerProcessLauncher(input: { } launchedIdentity = { nonce: identityNonce, ...identity }; child.pid = identity.pid; - await input.onSpawn?.({ - pid: identity.pid, - processGroupId: null, - startedAt: identity.startedAt, - }); + try { + if (launchCancelled) throw cancelledError; + await Promise.race([Promise.resolve().then(() => { + if (launchCancelled) throw cancelledError; + return input.onSpawn?.({ + pid: identity.pid, + processGroupId: null, + startedAt: identity.startedAt, + }); + }), cancellation]); + if (launchCancelled) throw cancelledError; + } catch (error) { + const cleanup = identity.containment ? await runner.execute({ + command: "python3", + args: ["-c", REMOTE_RUNNER_OWNERSHIP_FAILURE_CLEANUP, input.processIdentityPath, + identityNonce, input.runnerInstanceId, String(identity.pid), identity.startedAt, + JSON.stringify(identity.containment)], + bypassSession: true, + timeoutMs: 12_000, + }).catch(() => null) : null; + if (cleanup?.exitCode !== 0 || cleanup.timedOut) { + const failure = new Error(`${error instanceof Error ? error.message : String(error)}; cleanup incomplete: remote runner ownership unverified`, { cause: error }); + ownershipFailure = { error: failure, containment: "unconfirmed" }; + throw failure; + } + ownershipFailure = { error, containment: "confirmed" }; + throw error; + } + readySucceeded = true; + resolveReady(); await input.trace?.record({ name: "runner.process.launch", parentName: "runner.session.startup", @@ -10067,9 +10220,18 @@ export function createRemoteRunnerProcessLauncher(input: { }; } })(); + void completion.catch(error => { + if (!launchDispatched && error === cancelledError) { + ownershipFailure = { error, containment: "confirmed" }; + } + rejectReady(error); + }); return { child, completion, + ready, + cancelPendingLaunch, + get ownershipFailure() { return ownershipFailure; }, get startedAt() { return launchedIdentity?.startedAt; }, }; };