diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index 4dc35b2c37c..e84d343e00e 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -1781,6 +1781,17 @@ fn run() -> Result<(), Box> { "id": id, "result": {"turn": {"id": provider_turn_id, "status": "inProgress"}} }))?; + // A loaded host can accept a turn long before it starts it. + if let Some(delay) = argument(&args, "--turn-started-delay-ms") { + thread::sleep(Duration::from_millis(delay.parse()?)); + } + // An accepted turn can be lost before it starts. + if args + .iter() + .any(|value| value == "--exit-before-turn-started") + { + return Err("configured exit before turn started".into()); + } send(json!({ "method": "turn/started", "params": {"turn": {"id": provider_turn_id}} diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts index 06cd0ae3735..aa67fc52484 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts @@ -24,6 +24,62 @@ function fakeCodexArgs(stateDirectory: string, ...args: string[]): string[] { return ["--state-file", join(stateDirectory, "fake-codex-state.json"), ...args]; } +it("adopts a durably accepted turn whose start notification outlives the turn-start bound", async () => { + // A loaded host can accept a turn (durable command result with its provider + // turn id) and start it only after the turn-start bound. The provider still + // runs that turn, so failing the start would abandon a live turn. + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-late-turn-started-")); + const callsPath = join(stateDirectory, "calls.log"); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory, "--turn-started-delay-ms", "12000", "--fail-turn-immediately", "--call-log", callsPath), + stateDirectory, turnStartTimeoutMs: 8_000, + }); + try { + await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + const started = await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Start late." }] }); + expect(started).toMatchObject({ turn: { id: "provider-turn-1" } }); + const methods: string[] = []; + for await (const notification of bundle.transport.notifications()) { + methods.push(notification.method); + if (notification.method === "turn/completed") { + expect(notification.params).toMatchObject({ turnId: "provider-turn-1" }); + break; + } + } + expect(methods).toContain("turn/started"); + const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); + expect(calls.filter(call => call === "turn/start")).toHaveLength(1); + } finally { + await bundle.transport.close().catch(() => undefined); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 60_000); + +it("does not adopt an accepted turn the runner no longer reports active", async () => { + // The provider accepts the turn, then its process ends before the turn + // starts. The accepted turn is not live, so the start fails within its bound. + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-lost-turn-start-")); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory, "--exit-before-turn-started"), + stateDirectory, turnStartTimeoutMs: 2_000, + }); + try { + await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + const started = Date.now(); + // The runner may still list the turn after relaunching its provider; the + // process that accepted it is gone, so the turn is not adopted. + await expect(bundle.transport.request("turn/start", { input: [{ type: "text", text: "Lose this turn." }] })) + .rejects.toThrow("runnerd did not report the provider turn identity"); + // The adoption check stays inside the configured bound (plus scheduling slack). + expect(Date.now() - started).toBeLessThan(2_000 + 1_500); + } finally { + await bundle.transport.close().catch(() => undefined); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 60_000); + it("keeps a quiet active Codex turn in the same process across connection lease expiry", async () => { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-renew-active-")); const callsPath = join(stateDirectory, "calls.log"); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 41663b41523..23e28c7bd03 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -864,6 +864,9 @@ function terminalMayBelongToPendingTurnStart(input: { ); } +/** Part of the turn-start bound kept for the accepted-turn adoption check. */ +const TURN_START_ADOPTION_CHECK_RESERVE_MS = 5_000; + function turnStartNotificationDisposition(input: { responsePending: boolean; expectedProviderTurnId: string | null; @@ -6756,6 +6759,35 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return this.options.measureOperation ? this.options.measureOperation("runner.transport.turn_start", () => this.#startTurnWork(params), {inputBytes:Buffer.byteLength(JSON.stringify(params)), tokenDataAvailable:false}) : this.#startTurnWork(params); } + /** + * Live-turn proof for adopting an accepted turn whose start event is late: + * the runner still reports it active, the runner attributed no provider + * launch to a command after this turn.start, and, when the accepting provider is a + * local process, that exact process (pid and start time) is still alive. + */ + #acceptedTurnProvablyLive(input: { + snapshot: Record | null; + expectedProviderTurnId: string; + acceptingProviderPid: number | null; + acceptingProviderStartedAt: string | null; + acceptingCommandSeq: number; + }): boolean { + if (input.snapshot?.activeProviderTurnId !== input.expectedProviderTurnId) return false; + if (!Number.isSafeInteger(input.acceptingCommandSeq)) return false; + const relaunched = (this.#core?.store.state.committedEvents ?? []).some((event) => { + if (event.eventType !== "harness.diagnostic") return false; + const payload = record(record(event.envelope.payload).payload); + return payload.code === "provider_startup_ownership" && + Number(record(record(payload.startup).command).controllerSeq) > input.acceptingCommandSeq; + }); + if (relaunched) return false; + if (input.acceptingProviderPid === null || this.#evidence.providerPid !== input.acceptingProviderPid) return false; + // A remote provider's pid names no local process; the runner's own + // launch reports above are its replacement evidence. + if (input.acceptingProviderStartedAt === null) return true; + return readLocalProcessStartedAt(input.acceptingProviderPid) === input.acceptingProviderStartedAt; + } + async #startTurnWork( params: Record, ): Promise> { @@ -6848,6 +6880,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw error; } this.#expectedProviderTurnId = expectedProviderTurnId; + // The exact local provider process that accepted this turn. Adoption + // below requires that same process to still be alive. + const acceptingProviderPid = this.#evidence.providerPid; + const acceptingProviderStartedAt = this.#evidence.providerProcessStartedAt; + // The durable controller sequence of this turn.start command. A provider + // launch the runner attributes to a later command replaced the provider. + const acceptingCommandSeq = Number(record([...(this.#core?.store.state.commands ?? [])] + .reverse().find((command) => command.type === "turn.start")?.result).controllerSeq); // Command completion only means runnerd accepted the command. Bind the // provider turn from the subsequent turn/started event before answering // the strict driver. Correlate that event with the exact identity in this @@ -6865,8 +6905,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { expectedProviderTurnId, boundTurnId: this.#turnId, }); + // Keep the adoption check inside the configured bound: stop waiting for + // the start event early enough to read the runner's snapshot before it. + const adoptionCheckAt = deadline - Math.min( + TURN_START_ADOPTION_CHECK_RESERVE_MS, + Math.floor(Math.max(0, deadline - Date.now()) / 4), + ); const waitForStarted = async () => { - while (!providerTurnStarted() && Date.now() < deadline) { + while (!providerTurnStarted() && Date.now() < adoptionCheckAt) { this.#throwIfFailed(); this.#pumpEvents(); if (providerTurnStarted()) break; @@ -6874,8 +6920,48 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { throw new Error("runnerd exited before provider turn startup"); await new Promise((resolveWait) => setTimeout(resolveWait, 10)); } - if (!providerTurnStarted()) - throw new Error("runnerd did not report the provider turn identity"); + if (!providerTurnStarted()) { + // The durable command result already carries the provider's own + // acceptance and turn id. A loaded provider can start that turn after + // this bound, and it still runs it. Failing would abandon a live turn + // that no controller owns, so its tools and close could never settle. + // Bind the exact accepted identity when the runner proves it is live; + // its turn/started is then accepted in order, and the turn's own + // terminal, timeout and Stop paths own it from here. + this.#throwIfFailed(); + if (await this.#runnerHasExited()) + throw new Error("runnerd exited before provider turn startup"); + const snapshot = await this.#commandResult( + "session.snapshot", {}, deadline, + ).catch(() => null); + // Ingest any runner-reported provider launch committed before it. A + // pump that rejected a stale start has failed the transport; never + // report a successful start on it. + this.#pumpEventsSafely(); + this.#throwIfFailed(); + if (!providerTurnStarted() && this.#acceptedTurnProvablyLive({ + snapshot, expectedProviderTurnId, acceptingProviderPid, + acceptingProviderStartedAt, acceptingCommandSeq, + })) { + this.#turnId = expectedProviderTurnId; + this.#observedTurnStartEpoch = responseEpoch; + this.#diagnostic( + "provider turn start outlived its bound; adopting the durably accepted active provider turn", + ); + } + // Not provably live: keep the original contract and wait for the + // start event until the configured bound, then fail as before. + while (!providerTurnStarted() && Date.now() < deadline) { + this.#throwIfFailed(); + this.#pumpEvents(); + if (providerTurnStarted()) break; + if (await this.#runnerHasExited()) + throw new Error("runnerd exited before provider turn startup"); + await new Promise((resolveWait) => setTimeout(resolveWait, 10)); + } + if (!providerTurnStarted()) + throw new Error("runnerd did not report the provider turn identity"); + } }; if (this.options.measureOperation) await this.options.measureOperation("runner.transport.started_event", waitForStarted); else await waitForStarted();