From 463ebb62661d81ef1fbd79ae0a7b43ece68830f8 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 12 Oct 2026 19:16:15 -0500 Subject: [PATCH] fix(runner): drain owned startup before certifying cleanup Co-Authored-By: Paperclip --- maps/agent-access-map/runs-adapters.md | 8 + maps/feature-map/runs-adapters.md | 8 + .../src/live/runnerd-codex-transport.test.ts | 145 ++++++++++++++++++ .../src/live/runnerd-codex-transport.ts | 59 ++++++- 4 files changed, 213 insertions(+), 7 deletions(-) diff --git a/maps/agent-access-map/runs-adapters.md b/maps/agent-access-map/runs-adapters.md index 62a99f42b3a..daee072a911 100644 --- a/maps/agent-access-map/runs-adapters.md +++ b/maps/agent-access-map/runs-adapters.md @@ -146,3 +146,11 @@ retain existing execution on unrelated edits and use recorded run configuration for active/recovered runs. Explicit runner changes use revisions and session invalidation. Verify saved configuration, current versus historical run identity, restart and follow-up without granting session-reset authority to an agent. + +Scoped source-only evidence: the +[native transport](../../packages/paperclip-runner/src/live/runnerd-codex-transport.ts) +joins owned startup before certifying cleanup. Unconfirmed preparation or +registration retains recovery state; late registration may retire only its own +route and cannot launch a provider after close. This does not grant an agent +account-setup, cleanup-override or session-reset authority. Verify the no-provider +ownership controls independently from installed/live provider acceptance. diff --git a/maps/feature-map/runs-adapters.md b/maps/feature-map/runs-adapters.md index 453e2f81c63..36873e6bcd9 100644 --- a/maps/feature-map/runs-adapters.md +++ b/maps/feature-map/runs-adapters.md @@ -80,3 +80,11 @@ different observations, both required in the evidence. - The run log is instance database data; it is not the opt-out first-party Telemetry system or operator-configured OpenTelemetry tracing. - Authentication/model discovery does not establish tool, image, steering, or resumption capability parity. - Use [recovery](./recovery.md) for interrupted runs and [execution environments](./execution-environments.md) for local/SSH/sandbox differences. + +Native setup retirement must join owned startup before cleanup can be confirmed. +[Transport ownership controls](../../packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts) +hold credential preparation or registration without launching a provider, then +verify bounded close, no later launch, and retention of unconfirmed state. A +primary setup timeout remains a failure even when teardown also fails. These +controls cover the lifecycle boundary; final packaged provider execution is +separate evidence. 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 5ff210ad59b..6332a402749 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -69,6 +69,7 @@ import { codexSemanticToolSpecs, } from "../drivers/codex/codex-app-server-driver.js"; import { releaseMaterializedNativeRuntimeSkills } from "../drivers/runtime-context-materializer.js"; +import * as runtimeMaterializer from "../drivers/runtime-context-materializer.js"; import { RUNNERD_CANONICAL_ITEM } from "../drivers/codex/codex-driver-values.js"; import { @@ -2811,6 +2812,150 @@ function fakeCodexArgs(stateDirectory: string, ...args: string[]): string[] { ]; } +it("joins owned credential preparation before close returns and never launches after close", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-owned-startup-")); + const artifact = join(root, "unexecuted-runner-artifact"); + await writeFile(artifact, "fixture artifact bytes; never executed"); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let entered!: () => void; + const started = new Promise(resolve => { entered = resolve; }); + const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async ({ codexHome }) => { + entered(); + await gate; + await mkdir(codexHome, { recursive: true }); + await writeFile(join(codexHome, "owned-write.txt"), "completed owned preparation"); + }); + const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} })); + const launch = vi.fn(() => { throw new Error("unexecuted late launch"); }); + const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: 1000, + controlPlaneRegistration: registration, runnerProcessLauncher: launch }); + let opening: Promise | undefined; + let closing: Promise | undefined; + try { + opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] }); + void opening.catch(() => {}); + await started; + let closeReturned = false; + closing = bundle.transport.close("owned preparation test"); + void closing.then(() => { closeReturned = true; }, () => {}); + await new Promise(resolve => setTimeout(resolve, 25)); + expect(closeReturned).toBe(false); + release(); + await expect(opening).rejects.toThrow(/closed/); + await closing; + expect(await readFile(join(root, "codex-home", "owned-write.txt"), "utf8")).toBe("completed owned preparation"); + expect(registration).not.toHaveBeenCalled(); + expect(launch).not.toHaveBeenCalled(); + } finally { + release(); + await Promise.allSettled([opening, closing]); + preparation.mockRestore(); + await rm(root, { recursive: true, force: true }); + } +}); + +it("retains owned state when credential preparation outlives bounded close", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-retained-startup-")); + const artifact = join(root, "unexecuted-artifact"); + await writeFile(artifact, "never executed"); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async () => { entered(); await gate; }); + const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} })); + const launch = vi.fn(() => { throw new Error("unexpected launch"); }); + const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: 100, + controlPlaneRegistration: registration, runnerProcessLauncher: launch }); + const opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] }); + void opening.catch(() => {}); + try { + await started; + await expect(bundle.transport.close("bounded preparation")).rejects.toThrow("startup cleanup is unconfirmed"); + expect((await stat(root)).isDirectory()).toBe(true); + release(); + await expect(opening).rejects.toThrow(/closed/); + expect(registration).not.toHaveBeenCalled(); + expect(launch).not.toHaveBeenCalled(); + } finally { + release(); await opening.catch(() => {}); preparation.mockRestore(); + await rm(root, { recursive: true, force: true }); + } +}); + +it.each(["settled", "late", "late-release-error"] as const)("retires owned startup registration without post-close launch (%s)", async mode => { + const root = await mkdtemp(join(tmpdir(), "runnerd-startup-registration-")); + const artifact = join(root, "unexecuted-artifact"); + await writeFile(artifact, "never executed"); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const registered = new Promise(resolve => { entered = resolve; }); + const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockResolvedValue(undefined); + const retire = vi.fn(async () => { if (mode === "late-release-error") throw new Error("owned registration release failed"); }); + const registration = vi.fn(async () => { entered(); await gate; return { connectUrl: "ws://127.0.0.1:1", release: retire }; }); + const launch = vi.fn(() => { throw new Error("unexpected launch"); }); + const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: mode === "settled" ? 1000 : 100, + controlPlaneRegistration: registration, runnerProcessLauncher: launch }); + const opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] }); + void opening.catch(() => {}); + let closing: Promise | undefined; + try { + await registered; + let closeReturned = false; + closing = bundle.transport.close("owned registration"); + void closing.then(() => { closeReturned = true; }, () => {}); + if (mode === "settled") { + await new Promise(resolve => setTimeout(resolve, 25)); + expect(closeReturned).toBe(false); + } else { + await expect(closing).rejects.toThrow("startup cleanup is unconfirmed"); + expect((await stat(root)).isDirectory()).toBe(true); + } + release(); + await expect(opening).rejects.toThrow(mode === "late-release-error" ? "owned registration release failed" : /closed/); + if (mode === "settled") await closing; + expect(retire).toHaveBeenCalledTimes(1); + expect(launch).not.toHaveBeenCalled(); + } finally { + release(); await Promise.allSettled([opening, closing]); preparation.mockRestore(); + await rm(root, { recursive: true, force: true }); + } +}); + +it("does not certify probe cleanup while owned credential preparation is unconfirmed", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-probe-owned-preparation-")); + const artifact = join(root, "unexecuted-artifact"); + await writeFile(artifact, "never executed"); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async () => { entered(); await gate; }); + const cleanup = vi.fn(async () => {}); + const launch = vi.fn(() => { throw new Error("unexpected launch"); }); + const probe = probeNativeRunnerEnvironment({ runtimeDirectory: root, provider: "codex", model: "gpt-test", timeoutMs: 100, + environment: { OPENAI_API_KEY: "fixture" }, onCleanupConfirmed: cleanup, + transportOptions: { runnerBinary: artifact, codexCommand: artifact, runnerProcessLauncher: launch, + controlPlaneRegistration: async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} }) }, + }); + const result = probe.catch(error => error); + try { + await started; + const error = await result; + expect(cleanup).not.toHaveBeenCalled(); + expect(error).toBeInstanceOf(AggregateError); + expect(error.message).toContain("hello probe timed out"); + expect(error.message).toContain("startup cleanup is unconfirmed"); + expect((await stat(root)).isDirectory()).toBe(true); + release(); + await new Promise(resolve => setTimeout(resolve, 25)); + expect(launch).not.toHaveBeenCalled(); + } finally { + release(); await result; preparation.mockRestore(); + await new Promise(resolve => setTimeout(resolve, 25)); + await rm(root, { recursive: true, force: true }); + } +}, 15_000); + it.each(["complete", "api-key", "default-model", "mismatched-model", "model-refresh-error", "model-cleanup-error", "missing-model", "missing-runtime", "missing-completion", "late-bootstrap", "remote-complete", "remote-mismatched-model", "late-remote-preparation", "late-remote-release-error"] as const)("native Codex setup requires its daemon/app-server hello (%s)", async mode => { const root = await mkdtemp(join(tmpdir(), "runnerd-native-setup-")); const runtimeDirectory = join(root, "native-state"); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index c7eaf5d6f94..4d0b8f36268 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3602,6 +3602,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #runAttachTemplate: Record | null = null; #closed = false; #closePromise: Promise | null = null; + #startupOperation: Promise | null = null; + #startupDrainTimedOut = false; #controllerDetachedForRestart = false; #failure: Error | null = null; readonly #failureSignal: Promise; @@ -3715,7 +3717,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw new Error("native_runner_warm_transition_completion_pending"); } if (method === "initialize") return { user: {} }; - if (method === "thread/start") return this.#start(params); + if (method === "thread/start") return this.#trackStartup(this.#start(params)); if (method === "collaborationMode/list") { // runnerd negotiates the real Codex preset or the provider-proxy-owned // planning contract during session.open. This transport-level mask @@ -3815,7 +3817,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (method === "thread/read") { if (this.#core === null) { this.#recoveryTurnBindingPending = true; - await this.#resume(); + await this.#trackStartup(this.#resume()); } // Ask the authenticated runner for its live provider snapshot rather // than reading its filesystem. This both supports remote process owners @@ -4455,8 +4457,33 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); } + #trackStartup(operation: Promise): Promise { + this.#startupOperation = operation; + return operation; + } + + #assertStartupOpen(): void { + if (this.#closed) throw new Error("runnerd transport closed during provider startup"); + } + async #closeOnce(): Promise { this.#closed = true; + const { preparationDeadline, closeDeadline } = runnerCloseDeadlines(Date.now(), runnerCloseGraceMs(this.options)); + let startupConfirmed = true; + if (this.#startupOperation !== null) { + let timer: ReturnType | undefined; + try { + await Promise.race([ + this.#startupOperation.catch(() => undefined), + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error("Native startup drain timed out")), Math.max(0, preparationDeadline - Date.now())); + }), + ]); + } catch { + startupConfirmed = false; + this.#startupDrainTimedOut = true; + } finally { clearTimeout(timer); } + } this.#turnStartAdmission?.resolve(false); const adoptedRunner = this.options.adoptExistingRunner; // `settled` is a durable-state assertion, not merely the absence of a @@ -4481,10 +4508,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // A terminal provider frame can become visible one control loop before // its durable provider suffix is ACKed. Drain it before suspension so a // fresh run authority never inherits the prior run's pending events. - const { preparationDeadline, closeDeadline } = runnerCloseDeadlines( - Date.now(), - runnerCloseGraceMs(this.options), - ); + if (!(await this.#runnerHasExited())) { // Let an already-admitted tool result reach its original provider // before turn.stop can retire that tool-call identity. This shares the @@ -4611,7 +4635,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // an incomplete or identity-conflicting state remains fail-closed. const finalProviderState = this.#providerDrainState(); const runnerSettled = - runnerSuspended && + startupConfirmed && runnerSuspended && providerDrained && this.#core?.semanticToolResultsSettled() === true && (finalProviderState === null || @@ -4663,6 +4687,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } throw new NativeSessionCloseUnrecoverableError(settlement); } + if (!startupConfirmed) { + throw new Error("Native runner startup cleanup is unconfirmed; recovery state retained."); + } if (this.#ownsRoot && !adoptedRunner) { rmSync(this.#root, { recursive: true, force: true }); } @@ -4756,6 +4783,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { nativeMcp: nativeMcpLaunchBinding(this.options.environment), }); } + this.#assertStartupOpen(); const opencodeProxyPath = this.options.opencodeProxyPath ?? (provider === "opencode" && !this.options.runnerFilesystemRoot @@ -5056,6 +5084,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#controlPlaneCheckpoint = registration.checkpoint ?? null; this.#controlPlaneRelease = registration.release; } + if (this.#closed && this.#startupDrainTimedOut) { + const release = this.#controlPlaneRelease; + this.#controlPlaneRelease = null; + await release?.(); + } + this.#assertStartupOpen(); const admissionDeadline = this.#coldAdmissionDeadline(); const handle = spawnRunner({ connection: registration?.connection ?? { @@ -5098,7 +5132,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#handle = handle; this.#watchRunner(handle); await this.#publishSpawnedProcess(handle); + this.#assertStartupOpen(); await registration?.activate?.(); + this.#assertStartupOpen(); if (registration?.failure) { void registration.failure.catch((error: unknown) => { this.#failTransport( @@ -5107,6 +5143,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }); } await this.#awaitRegistrationReady(registration?.ready); + this.#assertStartupOpen(); this.#evidence.runnerPid = handle.child.pid ?? null; this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null; this.#publish(); @@ -5633,6 +5670,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } : registration.release; } + if (this.#closed && this.#startupDrainTimedOut) { + const release = this.#controlPlaneRelease; + this.#controlPlaneRelease = null; + await release?.(); + } + this.#assertStartupOpen(); // Route and immutable endpoint admission precedes every launch-material // write. A refused transition leaves its retained home/context untouched. if (runtimeContext !== null) { @@ -5664,6 +5707,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (warmRecovery) { await this.options.authorizeWarmTransitionRecovery?.("before_bootstrap"); } + this.#assertStartupOpen(); const bootstrapTicket = adoptedRunner ? null : warmRecovery @@ -5679,6 +5723,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await this.options.authorizeWarmTransitionRecovery?.("before_spawn"); } // A replacement executor restores its cold provider before acknowledging recovery. + this.#assertStartupOpen(); const admissionDeadline = this.#coldAdmissionDeadline(adoptedRunner === undefined); const handle = adoptedRunner ? null