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 6ee022d89e9..82a056efbff 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 { resolveRunnerdCodexSkillInputs, withCodexCollaborationRuntimeInstructions, controllerAcceptedTurnStartId, + committedEventWindowComplete, } from "./runnerd-codex-transport.js"; import { defaultCapabilityRunnerdBinary, @@ -78,6 +79,18 @@ import { assignedRuntimeContext, } from "../../test/support/runnerd-transport.js"; +describe("committedEventWindowComplete", () => { + const events = (...seqs: number[]) => seqs.map(sourceSeq => ({ sourceSeq })); + it("is complete only when every acknowledged event from the first one is retained", () => { + expect(committedEventWindowComplete({ committedEvents: events(1, 2, 3), ackedSourceSeq: 3 })).toBe(true); + expect(committedEventWindowComplete({ committedEvents: [], ackedSourceSeq: 0 })).toBe(true); + expect(committedEventWindowComplete({ committedEvents: events(1, 3, 4), ackedSourceSeq: 4 })).toBe(false); + expect(committedEventWindowComplete({ committedEvents: events(2, 3), ackedSourceSeq: 3 })).toBe(false); + expect(committedEventWindowComplete({ committedEvents: events(1, 2, 2), ackedSourceSeq: 3 })).toBe(false); + expect(committedEventWindowComplete(undefined)).toBe(false); + }); +}); + describe("controllerAcceptedTurnStartId", () => { // Shapes copied from the RC-59 durable controller journal. const attach = { commandId: "command_prp_00000001", controllerSeq: 1, type: "run.attach", status: "completed", result: {} }; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index d8ac94c8ef6..d657075010b 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -802,6 +802,14 @@ async function awaitProviderDrainBarrier(input: { return false; } +export function committedEventWindowComplete(state: { committedEvents: readonly { sourceSeq: number }[]; ackedSourceSeq: number } | undefined): boolean { + if (!state || state.committedEvents.length !== state.ackedSourceSeq) return false; + const seen = new Set(state.committedEvents.map(event => event.sourceSeq)); + if (seen.size !== state.ackedSourceSeq) return false; + for (let seq = 1; seq <= state.ackedSourceSeq; seq += 1) if (!seen.has(seq)) return false; + return true; +} + function providerTurnIsActiveFromCommittedEvents( events: readonly { eventType: string }[], ): boolean { @@ -4319,7 +4327,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { const item = record(rehydrateRunnerdItemNotification(payload, this.#threadId, observedTurn).item); if (typeof item.id === "string") items.set(item.id, { turnId: observedTurn, item }); } - if (!observedStart) throw new Error("codex_history_incomplete: requested turn start is outside the retained runner event window"); + if (!observedStart && !committedEventWindowComplete(this.#core?.store.state)) { + throw new Error("codex_history_incomplete: requested turn start is outside the retained runner event window"); + } data = [...items.values()]; // Runner authority wins over schema-shaped prose in an assistant item. if (semanticResultItem) data.push(semanticResultItem); 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 269f5c68d9b..4ffa1c0bb84 100644 --- a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -597,3 +597,90 @@ it("starts a new turn when the lost turn's start reached the runner only after t await rm(root, { recursive: true, force: true }); } }, 45_000); + +it("starts a new turn when the checkpoint names an accepted turn whose start was never committed", async () => { + const root = await mkdtemp(join(tmpdir(), "runner-restart-uncommitted-start-")); + const identity = { runId: "run-uncommitted-start", sessionId: "session-uncommitted-start", companyId: "company-uncommitted-start", issueId: "issue-uncommitted-start", agentId: "agent-uncommitted-start" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-uncommitted-start" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-6", 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-uncommitted-start", 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", "--turn-started-gate", join(root, "release-turn-started")], + stateDirectory: root, + turnStartTimeoutMs: 8_000, + prpIdentity: { runnerInstanceId: "runner-uncommitted-start", environmentLeaseId: "lease-uncommitted-start", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-uncommitted-start", itemId: "item-uncommitted-start" }, + }; + const first = createRunnerdCodexTransport(options); + let restored: ReturnType | undefined; + const alive = (pid: number | null) => { + if (!pid) return false; + try { process.kill(pid, 0); return true; } catch { return false; } + }; + 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-uncommitted-start", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + const opened = await session.snapshot(); + const starting = session.startTurn({ message: { role: "user", text: execution.task.prompt } }).catch((error: unknown) => error); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + await vi.waitFor(async () => { + const control = JSON.parse(await readFile(controlPath, "utf8")); + expect(control.commands.find((command: { type: string }) => command.type === "turn.start")).toMatchObject({ status: "completed" }); + }, { timeout: 8_000, interval: 25 }); + const owner = first.evidence(); + const checkpoint = { ...opened, activeTurnId: "provider-turn-1" }; + await first.detachControllerForRestart(); + await starting; + expect(await readFile(join(root, "runner", "runner-state.json"), "utf8")).not.toContain("turn.started"); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + await writeFile(join(root, "release-turn-started"), ""); + const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8")); + providerState.activeTurnId = null; + await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState)); + restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: "provider-turn-1" }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-uncommitted-start", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const completion = await executeNativeSession({ + input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-uncommitted-start", controlPlaneInstanceId: "controller-uncommitted-start", timeoutMs: 15_000, + controlPlane: { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); events.push(event); return { cursor: events.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = events.filter(event => event.sourceInstanceId === request.sourceInstanceId); + return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 }; + }, + async completeRun() {}, + }, + }); + expect(completion).toMatchObject({ providerSessionId: checkpoint.providerSessionId, turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } }); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(2); + } 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 }); + } +}, 45_000);