diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 8ede5fa339a..6c8d070b092 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2499,6 +2499,15 @@ results, intentional stops, and historical unmarked failures keep their existing terminal behavior. This continuation uses the already charged restart attempt and does not reset the provider-attempt budget. +A restart can also land after the agent proposed its result but before the +provider turn ended. When the restored runner proves that turn ended with the +result Paperclip accepted, it records a reconciliation and the run terminal, +which settle that turn. When provider history no longer contains the checkpointed +turn, or the runner reports it lost on restore, but Paperclip durably holds the +result that exact turn proposed, recovery also settles the turn with that result. Either way, recovery finalizes the +accepted result instead of failing and rerunning finished work. A missing turn +without such a result still fails closed, because its outcome is unknown. + Provider execution and control-plane finalization have different clocks. A healthy provider can think or execute a long tool without output. Once execution settles, recovery and finalization control steps have a 60-second deadline, checked on startup and every 15 seconds. With a healthy database and scheduler, an abandoned transition must be repaired or surfaced within 90 seconds. Terminal persistence must not wait on provider cleanup or publication; a late finalizer cannot change a reassigned or closed task or release another run's locks. In particular, a done report from a run whose task now belongs to another agent, a user or nobody is kept as an accepted result but records `completion_owner_superseded` and preserves the task status for the current owner. Historical ambiguous runs are never automatically replayed after an upgrade. When that deadline fails a native run with a prepared workspace, the same 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 b56ffc364bc..4dc35b2c37c 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 @@ -734,6 +734,11 @@ fn run() -> Result<(), Box> { let exit_after_tool_call_completion = args .iter() .any(|value| value == "--exit-after-tool-call-completion"); + // Keep the turn open after an accepted completion report, so a test can + // kill the runner between the durable proposal and the provider terminal. + let hold_after_terminal_result = args + .iter() + .any(|value| value == "--hold-after-terminal-result"); let emit_tool_call_on_resume = args .iter() .any(|value| value == "--emit-tool-call-on-resume"); @@ -1112,6 +1117,9 @@ fn run() -> Result<(), Box> { "terminal-result:rejected" }, )?; + if success && hold_after_terminal_result { + continue; + } finish_turn( &state_path, &mut state, 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 f1b967f161a..8b1aca6caef 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 @@ -1224,6 +1224,110 @@ describe("Codex app-server Codex driver", () => { } }); + it("settles a missing persisted turn with the durable result it already proposed", async () => { + // Provider history lost the turn (for example, it was killed before the + // turn terminal was flushed), but Paperclip durably holds the result that + // turn proposed. That result is the turn's outcome: finalize it instead of + // failing recovery and re-running finished work. + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + second.readResponse = { + thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [] }, + }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-missing-proposed-turn", + normalizedSessionId: "normalized-missing-proposed-turn", + workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + expect( + await first.invoke({ + id: "call-before-kill", + method: "item/tool/call", + params: { threadId: "thread-1", turnId: "turn-1", callId: "call-before-kill", tool: "paperclip_finish", arguments: result }, + }), + ).toMatchObject({ success: true }); + const snapshot = await original.snapshot(); + expect(snapshot).toMatchObject({ activeTurnId: "turn-1", semanticResult: { turnId: "turn-1", result } }); + await original.close({ reason: "killed before the turn terminal" }); + + const recovery = await driver.recoverSession?.(snapshot); + expect(recovery).toMatchObject({ recovered: true }); + const events = await collectUntilTerminal(recovery!.session!.events()); + expect(events.filter((event) => event.eventType === "turn.completed")).toHaveLength(1); + expect(events.filter((event) => event.eventType === "run.result.proposed")).toHaveLength(0); + expect(await recovery!.session!.snapshot()).toMatchObject({ + activeTurnId: null, + semanticResult: { turnId: "turn-1", result }, + terminalTurns: [expect.objectContaining({ turnId: "turn-1" })], + }); + }); + + it.each([true, false])("settles a restore-lost turn from its durable result only when one is held (held=%s)", async (held) => { + // The runner can report the checkpointed turn lost on restore when it + // cannot prove the accepted result itself, while Paperclip already holds it. + const lost = { code: "provider_turn_lost_on_restore", recoverable: true }; + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + second.readResponse = { + thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, + turns: [{ id: "turn-1", status: "failed", error: lost, items: [] }] }, + }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: `run-lost-proposed-${held}`, + normalizedSessionId: `normalized-lost-proposed-${held}`, + workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + if (held) { + await first.invoke({ + id: "call-before-kill", + method: "item/tool/call", + params: { threadId: "thread-1", turnId: "turn-1", callId: "call-before-kill", tool: "paperclip_finish", arguments: result }, + }); + } + const snapshot = await original.snapshot(); + await original.close({ reason: "killed before the turn terminal" }); + const recovery = await driver.recoverSession?.(snapshot); + expect(recovery).toMatchObject({ recovered: true }); + const events = await collectUntilTerminal(recovery!.session!.events()); + const terminal = events.find((event) => event.eventType.startsWith("turn.")); + expect(terminal?.eventType).toBe(held ? "turn.completed" : "turn.failed"); + if (!held) expect(terminal?.payload).toMatchObject({ error: lost }); + }); + + it("does not settle a missing persisted turn with a result another turn proposed", async () => { + const first = new FakeCodexTransport(); + const second = new FakeCodexTransport(); + second.readResponse = { + thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [] }, + }; + const driver = makeDriver([first, second]); + const original = await driver.openSession({ + runId: "run-missing-other-turn", + normalizedSessionId: "normalized-missing-other-turn", + workingDirectory: WORKSPACE, + }); + await original.startTurn({ message: { role: "user", text: "Complete." } }); + await first.invoke({ + id: "call-before-kill", + method: "item/tool/call", + params: { threadId: "thread-1", turnId: "turn-1", callId: "call-before-kill", tool: "paperclip_finish", arguments: result }, + }); + const snapshot = await original.snapshot(); + await original.close({ reason: "killed before the turn terminal" }); + const recovery = await driver.recoverSession?.({ + ...snapshot, + semanticResult: { ...snapshot.semanticResult!, turnId: "turn-older" }, + }); + expect(recovery).toMatchObject({ + recovered: false, + reason: expect.stringContaining("persisted active turn turn-1 is missing"), + }); + }); + it("reconciles a turn completed during transport loss without replacing either session identity", async () => { const first = new FakeCodexTransport( "driver-thread-loss", diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 6bbbe3e7585..813b69452d0 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -1,7 +1,7 @@ import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from "./codex-history.js"; import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js"; import { randomUUID } from "node:crypto"; -import { NativeProviderTerminalFailure } from "../../contracts/native-session-backend.js"; +import { isNativeRestartInterruption, NativeProviderTerminalFailure } from "../../contracts/native-session-backend.js"; import type { HarnessGoalOperation, @@ -791,29 +791,45 @@ export class CodexHarnessSession const expectedTurn = turns.find( (turn) => text(turn.id) === expectedTurnId, ); - if (expectedTurn === undefined) { + const holdsTurnResult = this.result !== null && this.resultTurnId === expectedTurnId; + if (holdsTurnResult && (expectedTurn === undefined || ( + text(expectedTurn.status) === "failed" && isNativeRestartInterruption(expectedTurn.error)))) { + // Provider history can lose a turn whose process was killed before its + // terminal was flushed, or the restored runner can report it lost on + // restore. Paperclip already holds the semantic result this exact turn + // proposed, so that durable result is the turn's outcome: settle it + // rather than fail recovery and rerun finished work. Without such a + // result the outcome stays unknown and recovery fails closed. + reconciledTerminalTurnId = expectedTurnId; + mapTerminalTurn(this, { id: expectedTurnId, status: "completed", items: [] }, expectedTurnId, true); + } else if (expectedTurn === undefined) { throw new HarnessReconciliationError( `persisted active turn ${expectedTurnId} is missing from thread/read`, ); - } - const status = text(expectedTurn.status); - if (status === "inProgress") { - this.activeTurnId = expectedTurnId; - } else if ( - ["completed", "failed", "interrupted", "cancelled"].includes(status) - ) { - reconciledTerminalTurnId = expectedTurnId; - mapTerminalTurn(this, expectedTurn, expectedTurnId, true); } else { - throw new HarnessReconciliationError( - `persisted active turn ${expectedTurnId} has unreconcilable status ${status || "missing"}`, - ); + const status = text(expectedTurn.status); + if (status === "inProgress") { + this.activeTurnId = expectedTurnId; + } else if ( + ["completed", "failed", "interrupted", "cancelled"].includes(status) + ) { + reconciledTerminalTurnId = expectedTurnId; + mapTerminalTurn(this, expectedTurn, expectedTurnId, true); + } else { + throw new HarnessReconciliationError( + `persisted active turn ${expectedTurnId} has unreconcilable status ${status || "missing"}`, + ); + } } } for (const [turnId, fingerprint] of this.terminalTurns) { const observed = turns.find((turn) => text(turn.id) === turnId); if (observed === undefined) continue; + // A turn settled from its durable result keeps that outcome when the + // restored runner reports the same turn lost on restore. + if (this.result !== null && this.resultTurnId === turnId && + text(observed.status) === "failed" && isNativeRestartInterruption(observed.error)) continue; if ( !["completed", "failed", "interrupted", "cancelled"].includes( text(observed.status), 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 c2a8f8faa14..b16668d84f1 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1505,6 +1505,26 @@ it("does not hold a pending turn start behind another turn's restore-lost termin params: { turn: { id: "provider-turn-raw", status: "failed" } } }] })).toBe("provider-turn-raw"); }); +it("settles a restored turn from the runner's reconciled result terminal", () => { + const { runnerdRecoveredTurnTerminal: recovered } = runnerdRecoveryInternals; + const event = (eventType: string, payload: Record) => ({ eventType, envelope: { payload: { payload } } }); + const reconciled = event("session.reconciled", { previousProviderTurnId: "turn-1", activeProviderTurnId: null }); + const runTerminal = event("run.terminal", { turnTerminalState: "completed", runTerminalState: "succeeded" }); + // The runner recovered turn-1's accepted result: no provider turn terminal, + // only the reconciliation and the run terminal. + expect(recovered([event("turn.started", { providerTurnId: "turn-1" }), reconciled, runTerminal], "turn-1")) + .toEqual({ status: "completed", error: null }); + // An explicit provider terminal keeps its cause. + const lost = { code: "provider_turn_lost_on_restore", recoverable: true }; + expect(recovered([reconciled, event("turn.failed", { providerTurnId: "turn-1", error: lost }), runTerminal], "turn-1")) + .toEqual({ status: "failed", error: lost }); + // A run terminal settles only the turn the runner reconciled as ended. + expect(recovered([event("session.reconciled", { previousProviderTurnId: "turn-0", activeProviderTurnId: null }), runTerminal], "turn-1")).toBeNull(); + expect(recovered([event("session.reconciled", { previousProviderTurnId: "turn-1", activeProviderTurnId: "turn-1" }), runTerminal], "turn-1")).toBeNull(); + expect(recovered([reconciled, event("turn.started", { providerTurnId: "turn-2" }), runTerminal], "turn-1")).toBeNull(); + expect(recovered([runTerminal], "turn-1")).toBeNull(); +}); + it("requires ACPX command results to preserve the requested turn identity", () => { expect( runnerdRecoveryInternals.turnStartCommandResultValid({ diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 2b3aeae27b4..41663b41523 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -788,6 +788,52 @@ function turnStartResponseReady(input: { ); } +type RunnerdRecoveredTurnTerminal = { + status: "completed" | "failed" | "interrupted" | "cancelled"; + error: unknown; +}; + +const RUNNERD_TURN_TERMINAL_STATUSES = new Set(["completed", "failed", "interrupted", "cancelled"]); + +/** + * The terminal the runner recorded for a provider turn that is no longer + * active after restore. A provider terminal names the turn explicitly. When the + * runner restores a conversation whose turn already ended with the semantic + * result Paperclip accepted, it records `session.reconciled` (naming the ended + * turn as previous, with no active turn) followed by the run terminal, and no + * provider turn terminal. That run terminal settles the reconciled turn. + */ +function runnerdRecoveredTurnTerminal( + events: readonly Pick[], + turnId: string, +): RunnerdRecoveredTurnTerminal | null { + let terminal: RunnerdRecoveredTurnTerminal | null = null; + let reconciledEndedTurn = false; + for (const event of events) { + const payload = record(record(event.envelope.payload).payload); + const providerTurnId = payload.providerTurnId ?? payload.turnId ?? record(payload.turn).id; + if (event.eventType === "turn.started") { + reconciledEndedTurn = false; + } else if (event.eventType === "session.reconciled") { + reconciledEndedTurn = payload.previousProviderTurnId === turnId && payload.activeProviderTurnId == null; + } else if (event.eventType.startsWith("turn.") && RUNNERD_TURN_TERMINAL_STATUSES.has(event.eventType.slice("turn.".length))) { + if (providerTurnId !== turnId) continue; + // Reconciliation must retain the cause, including the runner's explicit + // process-loss marker, rather than inventing error:null. + terminal = { status: event.eventType.slice("turn.".length) as RunnerdRecoveredTurnTerminal["status"], + error: payload.error ?? record(payload.turn).error ?? null }; + reconciledEndedTurn = false; + } else if (event.eventType === "run.terminal" && reconciledEndedTurn) { + const status = String(payload.turnTerminalState); + if (RUNNERD_TURN_TERMINAL_STATUSES.has(status)) { + terminal = { status: status as RunnerdRecoveredTurnTerminal["status"], error: null }; + } + reconciledEndedTurn = false; + } + } + return terminal; +} + /** The provider turn a durable terminal names, when it names one explicitly. */ function runnerdTerminalProviderTurnId(eventType: string, payload: unknown): string | null { const named = (value: unknown) => typeof value === "string" && value.length > 0 ? value : null; @@ -4130,48 +4176,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : [{ id: activeProviderTurnId, status: "inProgress" }]; if (activeProviderTurnId === null && this.#turnId.length > 0) { const recoveryDeadline = Date.now() + 5_000; - let terminal = (this.#core?.store.state.committedEvents ?? []).find( - () => false, - ); + let terminal: RunnerdRecoveredTurnTerminal | null = null; while (Date.now() < recoveryDeadline) { this.#throwIfFailed(); this.#pumpEvents(); - terminal = [...(this.#core?.store.state.committedEvents ?? [])] - .reverse() - .find((event) => { - if ( - event.eventType !== "turn.completed" && - event.eventType !== "turn.failed" && - event.eventType !== "turn.interrupted" && - event.eventType !== "turn.cancelled" - ) - return false; - const payload = record(record(event.envelope.payload).payload); - const providerTurnId = - payload.providerTurnId ?? - payload.turnId ?? - record(payload.turn).id; - return providerTurnId === this.#turnId; - }); - if (terminal !== undefined) break; + terminal = runnerdRecoveredTurnTerminal( + this.#core?.store.state.committedEvents ?? [], + this.#turnId, + ); + if (terminal !== null) break; await new Promise((resolveWait) => setTimeout(resolveWait, 10)); } - if (terminal !== undefined) { - const payload = record(record(terminal.envelope.payload).payload); - recoveredTurns.push({ - id: this.#turnId, - // Reconciliation must retain the cause, including the runner's - // explicit process-loss marker, rather than inventing error:null. - error: payload.error ?? record(payload.turn).error ?? null, - status: - terminal.eventType === "turn.completed" - ? "completed" - : terminal.eventType === "turn.interrupted" - ? "interrupted" - : terminal.eventType === "turn.cancelled" - ? "cancelled" - : "failed", - }); + if (terminal !== null) { + recoveredTurns.push({ id: this.#turnId, ...terminal }); } } if (restoreRuntimeRequests && activeProviderTurnId !== null) { @@ -8360,6 +8377,7 @@ export const runnerdRecoveryInternals = Object.freeze({ rotatedRunAttachPayload, localRunnerStateArchive, rotateAuthorityEpoch, + runnerdRecoveredTurnTerminal, runnerdTerminalProviderTurnId, terminalMayBelongToPendingTurnStart, turnStartCommandResultValid, 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 1b31aa945cd..b18758b83cf 100644 --- a/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts @@ -208,3 +208,104 @@ it("starts a new turn when the restored runner reports a lost turn the checkpoin await rm(root, { recursive: true, force: true }); } }, 40_000); + +it("finalizes the durable proposed result when the runner and provider die before the turn terminal", async () => { + // The provider proposed a complete result and Paperclip durably recorded it. + // Both processes were then killed before the provider turn ended. The + // restored runner recovers that turn's authoritative result: recovery must + // finalize it instead of failing reconciliation or re-running the task. + const root = await mkdtemp(join(tmpdir(), "runner-restart-proposed-")); + const identity = { runId: "run-proposed", sessionId: "session-proposed", companyId: "company-proposed", issueId: "issue-proposed", agentId: "agent-proposed" }; + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-proposed" }, + session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + task: { identifier: "TEST-3", title: "Finish checks", description: null, prompt: "Run the checks and report the result.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + provider: { kind: "codex", model: null }, + completionContract: { id: "contract-proposed", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: { + revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }], + } }, + interactionResponses: [], credentialBindings: [], + }; + const report = { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "The checks pass.", + completionClaim: { contractRevision: "1", objectiveSatisfied: true, criteria: [{ criterionId: "checks", status: "satisfied", evidenceRefs: [] }], remainingWork: [] }, + evidence: [], verification: [], attentionRequests: [], artifacts: [] }; + await writeFile(join(root, "report.json"), JSON.stringify(report)); + 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", + "--terminal-result-file", join(root, "report.json"), "--hold-after-terminal-result"], + stateDirectory: root, + prpIdentity: { runnerInstanceId: "runner-proposed", environmentLeaseId: "lease-proposed", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-proposed", itemId: "item-proposed" }, + }; + 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-proposed", transportFactory: () => first.transport }); + const session = await backend.openSession({ identity, workingDirectory: root }); + await session.startTurn({ message: { role: "user", text: execution.task.prompt } }); + // The controller's last checkpoint names the active turn but not the result. + const checkpoint = await session.snapshot(); + expect(checkpoint.activeTurnId).toBe("provider-turn-1"); + expect(checkpoint.semanticResult ?? null).toBeNull(); + // The control plane durably appended the proposal before the kill. + const durableEvents: PrpEvent[] = []; + const initialEvents = session.events()[Symbol.asyncIterator](); + while (!durableEvents.some(event => event.eventType === "run.result.proposed")) { + const event = await initialEvents.next(); + if (event.done) throw new Error("initial event history ended before the proposal"); + durableEvents.push(event.value); + } + await vi.waitFor(async () => expect(await readFile(join(root, "calls.log"), "utf8")).toContain("terminal-result:accepted")); + const owner = first.evidence(); + await first.detachControllerForRestart(); + kill(owner.runnerPid); + kill(owner.providerPid); + await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); }); + 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: checkpoint.activeTurnId }); + const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-proposed", transportFactory: () => restored!.transport }); + const events: PrpEvent[] = []; + const completion = await executeNativeSession({ + input: execution, backend: recovering, persistedSession: checkpoint, resumeInterruptedTurn: true, + runnerInstanceId: "runner-proposed", controlPlaneInstanceId: "controller-proposed", 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); durableEvents.push(event); return { cursor: durableEvents.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { + const source = durableEvents.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-1", + result: { reportedWorkDisposition: "done", summary: "The checks pass." }, terminal: { runTerminalState: "succeeded" } }); + expect(events.find(event => event.eventType === "session.failed")).toBeUndefined(); + expect(events.filter(event => event.eventType === "turn.failed")).toHaveLength(0); + // The finished work is not re-run. + expect((await readFile(join(root, "calls.log"), "utf8")).match(/^turn\/start$/gm)).toHaveLength(1); + } 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);