From ce23716503a2c9158b5a7ab74ab61571ff4ee7a3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 12 Oct 2026 19:59:20 -0500 Subject: [PATCH] fix(runner): retain startup ownership through cancellation Reuse the shared startup repair with duplicate admission, shutdown-aware readiness, and resume fences. Preserve absolute cleanup deadlines and unconfirmed recovery state. Co-Authored-By: Paperclip --- .../src/live/runnerd-codex-transport.test.ts | 298 +++++++++++++++++- .../src/live/runnerd-codex-transport.ts | 106 ++++++- 2 files changed, 392 insertions(+), 12 deletions(-) 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..f9cd2051838 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 { @@ -395,6 +396,9 @@ it("publishes spawned runner ownership before waiting for provider startup", asy }); const opening = bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() }); try { + // Same-tick recovery must not mutate an already admitted fresh startup. + await expect(bundle.transport.request("thread/read", { threadId: "rejected-duplicate" })) + .rejects.toThrow("PRP provider thread is already started"); await vi.waitFor(() => expect(onSpawn).toHaveBeenCalledOnce()); expect(onSpawn).toHaveBeenCalledWith({ pid: expect.any(Number), processGroupId: expect.any(Number), startedAt: expect.any(String) }); expect(bundle.transport.processInfo?.().pid).toBeGreaterThan(0); @@ -402,13 +406,17 @@ it("publishes spawned runner ownership before waiting for provider startup", asy releaseOwnership(); await opening; expect(activate).toHaveBeenCalledOnce(); + const turnStarted = expectTurnStarted(bundle.transport.notifications()[Symbol.asyncIterator]()); + void turnStarted.catch(() => undefined); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Verify fixture turn delivery." }] }); + await turnStarted; } finally { releaseOwnership(); await opening.catch(() => undefined); await bundle.transport.close(); await rm(stateDirectory, { recursive: true, force: true }); } -}); +}, 30_000); it("launches runnerd with its production durable outbox limits", () => { expect(runnerdLaunchProfileInternals.maxOutboxBytes).toBe(16 * 1024 * 1024); @@ -2176,6 +2184,115 @@ it.each(["installed", "explicit"] as const)("records the selected %s Codex comma } }); +it("does not invoke deferred startup preparation after immediate close", async () => { + const prepare = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockResolvedValue(undefined); + const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: () => undefined })); + const launch = vi.fn(() => { throw new Error("unexpected launch"); }); + const { transport } = createCapabilityRunnerdCodexTransport({ + runnerBinary: resolve(import.meta.dirname, "../../package.json"), codexCommand: "/unused-fixture-command", sourceCodexHome: "", + controlPlaneRegistration: registration, runnerProcessLauncher: launch, + }); + const opening = transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + void opening.catch(() => undefined); + try { + await transport.close(); + await expect(opening).rejects.toThrow(/closed/); + expect(prepare).not.toHaveBeenCalled(); + expect(registration).not.toHaveBeenCalled(); + expect(launch).not.toHaveBeenCalled(); + } finally { + await opening.catch(() => undefined); + await transport.close().catch(() => undefined); + prepare.mockRestore(); + } +}); + +it("owns blocked Codex home preparation through driver abort without later registration or launch", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-blocked-home-preparation-")); + const runnerBinary = join(root, "runnerd"); + const command = join(root, "codex"); + await writeFile(runnerBinary, "unexecuted fixture artifact", { mode: 0o700 }); + await writeFile(command, "#!/bin/sh\nexit 99\n", { mode: 0o700 }); + let releasePreparation!: () => void; + const preparationReleased = new Promise(resolve => { releasePreparation = resolve; }); + let markPreparationStarted!: () => void; + const preparationStarted = new Promise(resolve => { markPreparationStarted = resolve; }); + let closeSettled = false; + let preparationContinuedAfterClose = false; + const prepare = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async input => { + markPreparationStarted(); + await preparationReleased; + preparationContinuedAfterClose = closeSettled; + // A synthetic, credential-free write makes the owned preparation visible. + await mkdir(input.codexHome, { mode: 0o700 }); + await writeFile(join(input.codexHome, "preparation-marker"), "fixture preparation drained\n", { mode: 0o600 }); + }); + const registration = vi.fn(async () => ({ + connection: { mode: "connect" as const, connectUrl: "ws://127.0.0.1:1/fixture-never-connected" }, + release: () => undefined, + })); + const launch = vi.fn(() => { throw new Error("fixture_stop_before_runner_launch"); }); + const { transport } = createCapabilityRunnerdCodexTransport({ + provider: "codex", runnerBinary, codexCommand: command, stateDirectory: root, + sourceCodexHome: "", environment: {}, controlPlaneRegistration: registration, runnerProcessLauncher: launch, + }); + const request = transport.request.bind(transport); + let bootstrapOperation: Promise> | undefined; + const requestSpy = vi.spyOn(transport, "request").mockImplementation((method, params) => { + const operation = request(method, params); + if (method === "thread/start") bootstrapOperation = operation; + return operation; + }); + const closeTransport = transport.close.bind(transport); + let closeOperation: Promise | undefined; + const closeSpy = vi.spyOn(transport, "close").mockImplementation(reason => { + closeOperation ??= closeTransport(reason).then(() => { closeSettled = true; }); + return closeOperation; + }); + const controller = new AbortController(); + const cancelled = new Error("fixture Codex bootstrap cancelled"); + const driver = new CodexAppServerDriver({ + taskEnvelope: { + schema: "paperclip.skillless_task.v1", objective: "Credential-free bootstrap ownership fixture.", + completionContract: { revision: "blocked-preparation-v1", criteria: [] }, + constraints: [], expectedResultSchema: "paperclip.run_result.v1", + }, + conversationMode: "direct", environment: {}, transportFactory: () => transport, + }); + let openingSettled = false; + const opening = driver.openSession({ runId: "fixture-blocked-home", normalizedSessionId: "fixture-blocked-home", workingDirectory: root, signal: controller.signal }) + .then(value => { openingSettled = true; return { value }; }, error => { openingSettled = true; return { error }; }); + try { + await preparationStarted; + // A rejected duplicate must not replace the first admitted startup owner. + await expect(transport.request("thread/start", { cwd: root, dynamicTools: [] })) + .rejects.toThrow("PRP provider thread is already started"); + controller.abort(cancelled); + await new Promise(resolve => setImmediate(resolve)); + expect.soft(closeOperation).toBeDefined(); + expect.soft(closeSettled).toBe(false); + expect.soft(openingSettled).toBe(false); + releasePreparation(); + const outcome = await opening; + await closeOperation; + await bootstrapOperation?.catch(() => undefined); + expect.soft(preparationContinuedAfterClose).toBe(false); + expect.soft(registration.mock.calls.length).toBe(0); + expect.soft(launch.mock.calls.length).toBe(0); + expect.soft(outcome).toHaveProperty("error"); + expect.soft(outcome).toHaveProperty("error", cancelled); + } finally { + releasePreparation(); + await opening; + await bootstrapOperation?.catch(() => undefined); + await transport.close().catch(() => undefined); + requestSpy.mockRestore(); + closeSpy.mockRestore(); + prepare.mockRestore(); + await rm(root, { recursive: true, force: true }); + } +}); + it("rejects remote Codex without a guest executable before resolving controller dependencies", async () => { const root = await mkdtemp(join(tmpdir(), "paperclip-runner-remote-codex-command-")); const runnerBinary = join(root, "runnerd"); @@ -2811,6 +2928,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"); @@ -7864,6 +8125,41 @@ it("keeps an adopted runner alive when the detached controller finalizer closes" it("adopts an opening session without a checkpoint instead of bootstrapping a duplicate provider", () => verifyLiveRunnerAdoption(false, false, false, false, true), 30_000); +it("cancels startup readiness before releasing the route that settles its ready callback", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-ready-release-cycle-")); + let releaseReady!: () => void; + const readyReleased = new Promise(resolve => { releaseReady = resolve; }); + let markReadyStarted!: () => void; + const readyStarted = new Promise(resolve => { markReadyStarted = resolve; }); + const release = vi.fn(async () => { releaseReady(); }); + let core!: DurablePrpControlPlane; + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root), stateDirectory: root, sourceCodexHome: "", environment: {}, closeGraceMs: 2_000, + controlPlaneRegistration: async authority => { + core = authority; + await authority.start(); + return { ready: () => { markReadyStarted(); return readyReleased; }, release }; + }, + }); + const opening = bundle.transport.request("thread/start", { cwd: root, dynamicTools: [] }); + void opening.catch(() => undefined); + try { + await readyStarted; + await vi.waitFor(() => expect(core.store.state.commands.find(command => command.type === "session.open")?.status).toBe("completed")); + await expect(bundle.transport.close("cancel pending ingress authentication")).resolves.toBeUndefined(); + await expect(opening).rejects.toThrow(/closed/); + expect(release).toHaveBeenCalledOnce(); + expect(bundle.evidence().runnerExited).toBe(true); + expect(bundle.evidence().diagnostics.join("\n")).not.toContain("startup cleanup is unconfirmed"); + } finally { + releaseReady(); + await opening.catch(() => undefined); + await bundle.transport.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}, 30_000); + it("surfaces a runner exit while provider-ingress readiness is still pending", async () => { const neverReady = new Promise(() => undefined); const bundle = createCapabilityRunnerdCodexTransport({ diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 20c94e11cc6..f099283b738 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3598,7 +3598,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #authorizedTools: Record | null = null; #runAttachTemplate: Record | null = null; #closed = false; + readonly #closedSignal: Promise; + #resolveClosedSignal!: () => void; #closePromise: Promise | null = null; + #startupOperation: Promise | null = null; + #startupDrainTimedOut = false; #controllerDetachedForRestart = false; #failure: Error | null = null; readonly #failureSignal: Promise; @@ -3635,6 +3639,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#failureSignal = new Promise((_resolve, reject) => { this.#rejectFailureSignal = reject; }); + this.#closedSignal = new Promise(resolve => { + this.#resolveClosedSignal = resolve; + }); // Failure is also observed by request/notification paths. Register an // internal handler so a process exit after the owner has closed the // session cannot become an unhandled process-level rejection. @@ -3711,7 +3718,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 @@ -3810,8 +3817,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } if (method === "thread/read") { if (this.#core === null) { - this.#recoveryTurnBindingPending = true; - await this.#resume(); + await this.#trackStartup(() => { + this.#recoveryTurnBindingPending = true; + return this.#resume(); + }); } // Ask the authenticated runner for its live provider snapshot rather // than reading its filesystem. This both supports remote process owners @@ -4425,6 +4434,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (this.#closed) return; this.#controllerDetachedForRestart = true; this.#closed = true; + this.#resolveClosedSignal(); this.#turnStartAdmission?.resolve(false); if (this.#pump !== null) clearInterval(this.#pump); this.#pump = null; @@ -4451,8 +4461,43 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); } + #trackStartup(start: () => Promise): Promise { + // Admit one owner before invoking startup. A duplicate request's rejection + // must never replace the preparation that close is responsible for joining. + if (this.#startupOperation !== null) { + return Promise.reject(new Error("PRP provider thread is already started")); + } + const operation = Promise.resolve().then(() => { + this.#assertStartupOpen(); + return start(); + }); + 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; + this.#resolveClosedSignal(); + 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 @@ -4477,10 +4522,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 @@ -4607,7 +4649,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 || @@ -4659,6 +4701,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 }); } @@ -4752,6 +4797,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { nativeMcp: nativeMcpLaunchBinding(this.options.environment), }); } + this.#assertStartupOpen(); const opencodeProxyPath = this.options.opencodeProxyPath ?? (provider === "opencode" && !this.options.runnerFilesystemRoot @@ -5051,6 +5097,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 ?? { @@ -5093,7 +5145,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( @@ -5102,6 +5156,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(); @@ -5556,6 +5611,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { >; let oldTransitionRegistration: RecoveryRegistration | null = null; let newTransitionRegistration: RecoveryRegistration | null = null; + this.#assertStartupOpen(); if (warmRecovery && this.options.controlPlaneRegistration) { try { oldTransitionRegistration = await this.options.controlPlaneRegistration( @@ -5574,6 +5630,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { "native_runner_warm_transition_registered_endpoint_mismatch", ); } + this.#assertStartupOpen(); newTransitionRegistration = await this.options.controlPlaneRegistration( core, warmRecovery.proof.receipt.newIdentity, @@ -5628,6 +5685,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) { @@ -5643,6 +5706,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await releaseMaterializedNativeRuntimeSkills( resolve(localCodexHome, "skills"), ); + this.#assertStartupOpen(); await prepareIsolatedCodexHome({ context: sourceRuntimeContext, codexHome: localCodexHome, @@ -5659,6 +5723,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (warmRecovery) { await this.options.authorizeWarmTransitionRecovery?.("before_bootstrap"); } + this.#assertStartupOpen(); const bootstrapTicket = adoptedRunner ? null : warmRecovery @@ -5674,6 +5739,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 @@ -5719,10 +5785,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#watchRunner(handle); await this.#publishSpawnedProcess(handle); } + this.#assertStartupOpen(); if (oldTransitionRegistration && newTransitionRegistration) { await oldTransitionRegistration.activate?.(); + this.#assertStartupOpen(); await newTransitionRegistration.activate?.(); } else await registration?.activate?.(); + this.#assertStartupOpen(); if (registration?.failure) { void registration.failure.catch((error: unknown) => { this.#failTransport( @@ -5731,6 +5800,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }); } if (!adoptedRunner) await this.#awaitRegistrationReady(registration?.ready); + this.#assertStartupOpen(); if (adoptedRunner) { this.#evidence.runnerPid = adoptedRunner.pid; this.#evidence.runnerProcessGroupId = adoptedRunner.processGroupId; @@ -5747,6 +5817,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { !recoveryIdentityMatches(core.store.state.identity, desiredIdentity) || core.store.state.warmTransition !== undefined ) { + this.#assertStartupOpen(); if (Date.now() >= deadline || (await this.#runnerHasExited())) { throw new Error("native_runner_warm_transition_recovery_pending"); } @@ -5761,11 +5832,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#deferredTurnStartEvents = []; if (oldTransitionRegistration && newTransitionRegistration) { await oldTransitionRegistration.release(); + this.#assertStartupOpen(); this.#controlPlaneRelease = newTransitionRegistration.release; this.#controlPlaneCheckpoint = newTransitionRegistration.checkpoint ?? null; } } + this.#assertStartupOpen(); this.#pump = setInterval(() => this.#pumpEventsSafely(), 5); if (adoptedRunner) { await this.#awaitAdoptedRunnerConnection( @@ -6803,8 +6876,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await awaitAdoptedRunnerAuthentication({ activeConnectionCount: () => core.activeRunnerConnectionCount(), isAlive: () => adoptedRunner.isAlive(), - throwIfFailed: () => this.#throwIfFailed(), - failure: this.#failureSignal, + throwIfFailed: () => { this.#assertStartupOpen(); this.#throwIfFailed(); }, + failure: Promise.race([ + this.#failureSignal, + this.#closedSignal.then(() => { throw new Error("runnerd transport closed during provider startup"); }), + ]), ready, timeoutMs: this.options.runnerReconnectGraceMs ?? 30_000, }); @@ -6813,6 +6889,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#failTransport(failure); throw failure; } + this.#assertStartupOpen(); this.#adoptedRunnerAuthenticated = true; this.#diagnostic( `adopted runner ${adoptedRunner.pid} authenticated to its durable PRP authority`, @@ -7017,11 +7094,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #awaitRegistrationReady( ready: (() => Promise) | undefined, ): Promise { + this.#assertStartupOpen(); if (ready === undefined) { this.#throwIfFailed(); return; } - await Promise.race([ready(), this.#failureSignal]); + await Promise.race([ + ready(), this.#failureSignal, + // Readiness is a notification wait, not launch-material preparation. + // Its route owner is released by close; joining it first would cycle. + this.#closedSignal.then(() => { this.#assertStartupOpen(); }), + ]); + this.#assertStartupOpen(); this.#throwIfFailed(); }