diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index bddbbe998eb..fb7cdbf5958 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -58,6 +58,9 @@ export interface CodexAppServerTransport { * controller checkpoints the turn; recovery uses this to adopt that turn. */ controllerAcceptedTurnId?(): string | null; + /** The provider turn the latest completed turn.start accepted, live or not; the + * driver adopts it only when provider history shows that exact turn completed. */ + controllerAcceptedCompletedTurnId?(): string | null; /** * Optional provider-neutral resolution path used when a transport has * already normalized a native server request behind another PRP boundary. diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index 09cc8880108..4e77e387541 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -650,8 +650,15 @@ export class CodexAppServerDriver implements HarnessDriver { const turns = Array.isArray(existingThread.turns) ? existingThread.turns.map(record) : []; - const acceptedTurnId = transport.controllerAcceptedTurnId?.() ?? ""; const active = turns.filter((turn) => text(turn.status) === "inProgress"); + // A Codex turn that finished while no controller was attached is no longer live; + // adopt it only when history shows that exact turn completed (a turn the runner + // reports lost on restore keeps the restart continuation). + const completedTurnId = transport.controllerAcceptedCompletedTurnId?.() ?? ""; + const acceptedTurnId = transport.controllerAcceptedTurnId?.() + || (completedTurnId.length > 0 && active.length === 0 + && turns.some((turn) => text(turn.id) === completedTurnId && text(turn.status) === "completed") + ? completedTurnId : ""); // The accepted turn may still run, or it may have finished while no // controller was attached. Either way it must be the only live turn. if ( diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts index 4fc33ec527b..8904234549d 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.recovery.test.ts @@ -483,6 +483,42 @@ describe("Codex app-server Codex driver", () => { }); }); + describe("a Codex turn that completed while no controller was attached", () => { + async function recoverCompleted(status: string) { + const first = new FakeCodexTransport(); + const second = Object.assign(new FakeCodexTransport(), { + controllerAcceptedTurnId: () => null, + controllerAcceptedCompletedTurnId: () => "turn-1", + }); + second.readResponse = { thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "turn-1", status, items: [] }] } }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ runId: "run-completed", normalizedSessionId: "normalized-completed", workingDirectory: WORKSPACE }); + const snapshot = await original.snapshot(); + await original.close({ reason: "controller restarted" }); + const recovery = await driver.recoverSession!(snapshot); + expect(recovery).toMatchObject({ recovered: true }); + second.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } }); + second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] } }); + return recovery.session!; + } + + it("adopts the accepted turn when provider history shows it completed", async () => { + const recovered = await recoverCompleted("completed"); + const events = await collectUntilTerminal(recovered.events()); + expect(events.filter(event => event.eventType.startsWith("turn.")).map(event => [event.eventType, event.turnId])) + .toEqual(expect.arrayContaining([["turn.accepted", "turn-1"], ["turn.completed", "turn-1"]])); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + await recovered.close({ reason: "test complete" }); + }); + + it("leaves a turn the runner reports lost to the restart continuation", async () => { + const recovered = await recoverCompleted("failed"); + expect((await recovered.snapshot()).activeTurnId ?? null).toBeNull(); + await recovered.close({ reason: "test complete" }); + }); + }); + it("rejects a recovered provider thread outside the assigned workspace", async () => { const first = new FakeCodexTransport(); const second = new FakeCodexTransport(); 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 92d14217627..accfce00ee9 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -103,6 +103,36 @@ describe("controllerAcceptedTurnStartId", () => { ])("returns null for %s", (_label, commands) => { expect(controllerAcceptedTurnStartId(commands)).toBeNull(); }); + + // Codex assigns its own provider turn id; only ACPX echoes the requested one. + it("accepts a Codex result whose provider turn id differs from the requested id while that turn is the runner's live turn", () => { + const codex = start({ result: { result: { providerTurnId: "019a-provider-turn", status: "accepted" } } }); + expect(controllerAcceptedTurnStartId([attach, codex], { requireRequestedIdentity: false, runnerActiveTurnId: "019a-provider-turn" })).toBe("019a-provider-turn"); + // A turn the restored runner reports lost (no live turn) is left to the restart continuation. + expect(controllerAcceptedTurnStartId([attach, codex], { requireRequestedIdentity: false, runnerActiveTurnId: null })).toBeNull(); + expect(controllerAcceptedTurnStartId([attach, codex], { requireRequestedIdentity: false, runnerActiveTurnId: "another" })).toBeNull(); + expect(controllerAcceptedTurnStartId([attach, codex], { requireRequestedIdentity: true })).toBeNull(); + }); + + // A restart landed right after turn.start: the runner journaled it, died, and + // settled it as execution_indeterminate; its restored session reconciled the + // ambiguous start to the provider's live replacement turn (t77). + const indeterminate = start({ status: "failed", result: { commandType: "turn.start", controllerSeq: 4, status: "failed", + result: { code: "execution_indeterminate", message: "runner recovered after journaling this command; it will not execute twice" } } }); + + it("names the runner's reconciled active turn for an indeterminate turn.start", () => { + expect(controllerAcceptedTurnStartId([attach, indeterminate], { + requireRequestedIdentity: false, runnerActiveTurnId: "019a-provider-turn" })).toBe("019a-provider-turn"); + }); + + it.each([ + ["no reconciled active turn", [attach, indeterminate], null, false], + ["an ordinary failed turn.start", [attach, start({ status: "failed", result: { result: { code: "provider_turn_failed" } } })], "019a-provider-turn", false], + ["a later run attachment", [{ ...indeterminate, controllerSeq: 1 }, { ...attach, controllerSeq: 2 }], "019a-provider-turn", false], + ["an ACPX turn other than the requested one", [attach, indeterminate], "turn_lab_other", true], + ] as const)("fails closed on an indeterminate start with %s", (_label, commands, active, requireRequestedIdentity) => { + expect(controllerAcceptedTurnStartId([...commands], { requireRequestedIdentity, runnerActiveTurnId: active })).toBeNull(); + }); }); it("attributes opened candidate sessions to the harness biller instead of the model vendor", () => { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index debce74b471..3b261b0e65d 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -609,22 +609,44 @@ function retargetRunAttachPayload( /** * The provider turn accepted by this run's latest durable `turn.start` - * command. Only a completed command whose authenticated result accepted the - * same controller-assigned turn id counts, and it must follow the latest run - * attachment so a prior run's turn on a reused Runner is never claimed. + * command, which must follow the latest run attachment so a prior run's turn + * on a reused Runner is never claimed. Either: + * - the command completed and its authenticated result accepted a provider + * turn. ACPX echoes the requested id, so it must match; Codex and OpenCode + * assign their own, so the turn must instead be the runner's live active + * turn (`runnerActiveTurnId`): a turn the restored runner reports lost is + * left to the restart continuation; or + * - the runner crashed after journaling the command, so it settled as + * `execution_indeterminate`, and its restored session reconciled that + * ambiguous start to the live active turn `runnerActiveTurnId`. + * `runnerActiveTurnId` comes from the runner's authenticated `session.snapshot`. */ -export function controllerAcceptedTurnStartId(commands: readonly unknown[]): string | null { +export function controllerAcceptedTurnStartId(commands: readonly unknown[], options: { + requireRequestedIdentity?: boolean; + runnerActiveTurnId?: string | null; +} = {}): string | null { const ordered = commands.map(record).filter(command => typeof command.controllerSeq === "number") .sort((left, right) => (left.controllerSeq as number) - (right.controllerSeq as number)); const latestStart = ordered.findLast(command => command.type === "turn.start"); - if (!latestStart || latestStart.status !== "completed") return null; + if (!latestStart) return null; const latestAttachment = ordered.findLast(command => command.type === "run.attach" || command.type === "run.prepare"); if (latestAttachment && (latestAttachment.controllerSeq as number) > (latestStart.controllerSeq as number)) return null; + const requestedTurnId = record(latestStart.payload).turnId; + const requireRequested = options.requireRequestedIdentity ?? true; + const active = typeof options.runnerActiveTurnId === "string" && options.runnerActiveTurnId.length > 0 + ? options.runnerActiveTurnId : null; + const bound = (providerTurnId: string) => requireRequested + ? requestedTurnId === undefined || requestedTurnId === providerTurnId + : providerTurnId === active; + if (latestStart.status === "failed") { + const indeterminate = record(record(latestStart.result).result).code === "execution_indeterminate"; + return indeterminate && active !== null && bound(active) ? active : null; + } + if (latestStart.status !== "completed") return null; const accepted = record(record(latestStart.result).result); const providerTurnId = accepted.providerTurnId; if (accepted.status !== "accepted" || typeof providerTurnId !== "string" || providerTurnId.length === 0) return null; - const requestedTurnId = record(latestStart.payload).turnId; - return requestedTurnId === undefined || requestedTurnId === providerTurnId ? providerTurnId : null; + return bound(providerTurnId) ? providerTurnId : null; } function recoveredRunAttachment(state: { @@ -3914,9 +3936,25 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { } controllerAcceptedTurnId(): string | null { - return controllerAcceptedTurnStartId(this.#core?.store.state.commands ?? []); + return controllerAcceptedTurnStartId(this.#core?.store.state.commands ?? [], { + requireRequestedIdentity: this.options.provider === "acpx", + runnerActiveTurnId: this.#snapshotActiveProviderTurnId, + }); } + controllerAcceptedCompletedTurnId(): string | null { + if (this.options.provider === "acpx") return null; + const commands = this.#core?.store.state.commands ?? []; + const latest = [...commands].map(command => command as unknown as Record) + .filter(command => command.type === "turn.start" && command.status === "completed").at(-1); + const accepted = latest ? record(record(latest.result).result).providerTurnId : null; + return typeof accepted === "string" && accepted.length > 0 + ? controllerAcceptedTurnStartId(commands, { requireRequestedIdentity: false, runnerActiveTurnId: accepted }) : null; + } + + /** The active provider turn in the runner's latest authenticated session.snapshot. */ + #snapshotActiveProviderTurnId: string | null = null; + turnControlCapabilities(): NativeTurnControlCapabilities | null { if (this.options.provider !== "acpx") return null; if (this.#closed || this.#failure) return { steering: false, queuedFollowUp: false }; @@ -4315,6 +4353,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { snapshot.activeProviderTurnId.length > 0 ? snapshot.activeProviderTurnId : null; + this.#snapshotActiveProviderTurnId = activeProviderTurnId; if (activeProviderTurnId !== null) this.#turnId = activeProviderTurnId; const recoveredTurns: Array> = activeProviderTurnId === null diff --git a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts index 1e587787853..7a117c991e8 100644 --- a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -422,3 +422,81 @@ it.each(["unread", "read"] as const)("delivers a steered message the restart los await rm(root, { recursive: true, force: true }); } }, 30_000); + +it("adopts the in-flight turn when a restart lands just after turn.start (t77)", async () => { + // The runner journaled turn.start and the provider started the turn, then the + // runner died before confirming it: on restore the command is + // execution_indeterminate. The restored runner reconciles the ambiguous start + // to the provider's live turn, and recovery must adopt that exact turn. + const root = await mkdtemp(join(tmpdir(), "runner-restart-indeterminate-")); + const identity = { runId: "run-indeterminate", sessionId: "session-indeterminate", companyId: "company-indeterminate", issueId: "issue-indeterminate", agentId: "agent-indeterminate" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-indeterminate" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-4", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-indeterminate", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const options = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn"], + stateDirectory: root, + turnStartTimeoutMs: 8_000, + prpIdentity: { runnerInstanceId: "runner-indeterminate", environmentLeaseId: "lease-indeterminate", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-indeterminate", itemId: "item-indeterminate" }, + }; + const first = createRunnerdCodexTransport(options); + let restored: ReturnType | undefined; + const kill = (pid: number | null) => { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } + }; + try { + const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-indeterminate", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + const checkpoint = await session.snapshot(); + await session.startTurn({ message: { role: "user", text: execution.task.prompt } }); + const owner = first.evidence(); + await first.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + expect(checkpoint.activeTurnId ?? null).toBeNull(); + // The state both journals hold when the runner dies after journaling the + // start but before confirming it: the command is still pending. The provider + // keeps the started turn. + const runnerPath = join(root, "runner", "runner-state.json"); + const runnerState = JSON.parse(await readFile(runnerPath, "utf8")); + const startId = Object.values(runnerState.processedCommands as Record) + .find(command => command.commandType === "turn.start")!.commandId; + runnerState.processedCommands[startId] = { ...runnerState.processedCommands[startId], status: "pending", result: null }; + await writeFile(runnerPath, JSON.stringify(runnerState)); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const control = JSON.parse(await readFile(controlPath, "utf8")); + for (const command of control.commands) if (command.commandId === startId) { command.status = "pending"; command.result = null; } + await writeFile(controlPath, JSON.stringify(control)); + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: null }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-indeterminate", transportFactory: () => restored!.transport }); + const recovery = await recovering.recoverSession!(checkpoint, { signal: new AbortController().signal }); + expect(recovery).toMatchObject({ recovered: true }); + const restoredStart = JSON.parse(await readFile(controlPath, "utf8")).commands.find((command: { commandId: string }) => command.commandId === startId); + expect(restoredStart).toMatchObject({ status: "failed", result: { result: { code: "execution_indeterminate" } } }); + expect(restored.transport.controllerAcceptedTurnId?.()).toBe("provider-turn-1"); + expect((await recovery.session!.snapshot()).activeTurnId).toBe("provider-turn-1"); + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(1); + await recovery.session!.close({ reason: "test complete" }).catch(() => undefined); + } finally { + await restored?.transport.close().catch(() => undefined); + for (const bundle of [first, restored]) { + if (!bundle) continue; + const evidence = bundle.evidence(); + kill(evidence.runnerPid); + kill(evidence.providerPid); + } + await rm(root, { recursive: true, force: true }); + } +}, 40_000);