diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts index 8a18b200122..c13ffcb1596 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts @@ -245,13 +245,30 @@ describe("managed Codex credentials", () => { } }); + it("prepares two occupied candidates without disturbing a foreign collision", async () => { + const foreign = await silentPrimaryQuorumFixture(); + const nextFixture = vi.fn().mockResolvedValueOnce(foreign.fixture).mockImplementation(credentialFixture); + let prepared: Awaited> | undefined; + try { + prepared = await silentPrimaryQuorumFixture(nextFixture, "quorum"); + expect(nextFixture.mock.calls.length).toBeGreaterThanOrEqual(2); + expect(nextFixture.mock.calls.length).toBeLessThanOrEqual(8); + expect(prepared.fixture.home).not.toBe(foreign.fixture.home); + await expect(listenSilently(credentialLeasePorts(foreign.fixture.home)[0])).rejects.toMatchObject({ code: "EADDRINUSE" }); + const destination = join(prepared.fixture.home, "auth.json"); + await writeFile(destination, '{"sentinel":true}', { mode: 0o600 }); + await expect(stageManagedCodexCredential({ agentHomeDirectory: prepared.fixture.home, environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}' } })).rejects.toThrow("already has an active lease"); + await expect(readFile(destination, "utf8")).resolves.toBe('{"sentinel":true}'); + } finally { + await prepared?.occupied.close(); + await foreign.occupied.close(); + } + }); + it("fails before auth mutation when two quorum candidates are occupied", async () => { - const fixture = await credentialFixture(); + const { fixture, occupied } = await silentPrimaryQuorumFixture(credentialFixture, "quorum"); const destination = join(fixture.home, "auth.json"); await writeFile(destination, '{"sentinel":true}', { mode: 0o600 }); - const ports = credentialLeasePorts(await realpath(fixture.home)); - const first = await listenSilently(ports[0]); - const second = await listenSilently(ports[1]); try { await expect( stageManagedCodexCredential({ @@ -265,7 +282,7 @@ describe("managed Codex credentials", () => { '{"sentinel":true}', ); } finally { - await Promise.all([first.close(), second.close()]); + await occupied.close(); } }); @@ -1654,7 +1671,7 @@ async function credentialFixture(): Promise<{ root: string; home: string }> { async function silentPrimaryQuorumFixture( nextFixture = credentialFixture, - occupiedIndex: 0 | 1 = 0, + occupiedIndex: 0 | 1 | "quorum" = 0, ) { for (let attempt = 0; attempt < 8; attempt += 1) { const fixture = await nextFixture(); @@ -1674,22 +1691,30 @@ async function silentPrimaryQuorumFixture( continue; throw error; } - // Only fixture preparation may retry. Release the two free candidates + const occupiedIndices = occupiedIndex === "quorum" ? [0, 1] : [occupiedIndex]; + const occupied = { + close: async () => { + const closed = await Promise.allSettled(owned.filter((_, index) => occupiedIndices.includes(index)).map(listener => listener.close())); + const failure = closed.find(result => result.status === "rejected"); + if (failure?.status === "rejected") throw failure.reason; + }, + }; + // Only fixture preparation may retry. Release the unselected candidates // immediately before the caller stages once. A foreign bind racing this // handoff remains a visible production-call failure, never a hidden retry. const released = await Promise.allSettled( owned - .filter((_, index) => index !== occupiedIndex) + .filter((_, index) => !occupiedIndices.includes(index)) .map((listener) => listener.close()), ); const releaseFailure = released.find( (result) => result.status === "rejected", ); if (releaseFailure?.status === "rejected") { - await owned[occupiedIndex]!.close(); + await occupied.close(); throw releaseFailure.reason; } - return { fixture, occupied: owned[occupiedIndex]! }; + return { fixture, occupied }; } throw new Error("Silent-primary credential fixture reservation exhausted."); } 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 6332a402749..f9cd2051838 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -396,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); @@ -403,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); @@ -2177,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"); @@ -8009,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 4d0b8f36268..471686d4d43 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3601,6 +3601,8 @@ 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; @@ -3641,6 +3643,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. @@ -3717,7 +3722,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.#trackStartup(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 @@ -3816,8 +3821,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } if (method === "thread/read") { if (this.#core === null) { - this.#recoveryTurnBindingPending = true; - await this.#trackStartup(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 @@ -4431,6 +4438,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; @@ -4457,7 +4465,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { ); } - #trackStartup(operation: Promise): Promise { + #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; } @@ -4468,6 +4485,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { async #closeOnce(): Promise { this.#closed = true; + this.#resolveClosedSignal(); const { preparationDeadline, closeDeadline } = runnerCloseDeadlines(Date.now(), runnerCloseGraceMs(this.options)); let startupConfirmed = true; if (this.#startupOperation !== null) { @@ -5598,6 +5616,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( @@ -5616,6 +5635,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { "native_runner_warm_transition_registered_endpoint_mismatch", ); } + this.#assertStartupOpen(); newTransitionRegistration = await this.options.controlPlaneRegistration( core, warmRecovery.proof.receipt.newIdentity, @@ -5691,6 +5711,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { await releaseMaterializedNativeRuntimeSkills( resolve(localCodexHome, "skills"), ); + this.#assertStartupOpen(); await prepareIsolatedCodexHome({ context: sourceRuntimeContext, codexHome: localCodexHome, @@ -5769,10 +5790,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( @@ -5781,6 +5805,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; @@ -5797,6 +5822,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"); } @@ -5811,11 +5837,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( @@ -6855,8 +6883,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, }); @@ -6865,6 +6896,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`, @@ -7069,11 +7101,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(); }