mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(runner): keep an accepted result when ACPX reports its turn lost on restore (#19691)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native Codex, OpenCode, and ACPX runs survive control-plane restarts: runnerd restores the conversation, and the control plane continues the task from its checkpoint > - #19658 made recovery keep a durable result when a Codex restore loses the turn that proposed it. It covered the recovery read path only > - ACPX reports the lost turn differently. After the restored conversation reopens, a live `turn.failed` arrives with `provider_turn_lost_on_restore` > - That live terminal still turned an accepted `done` result into a failed run. The run ended `failed` (`adapter_failed`), and the finished work needed a Retry > - This pull request moves the rule into the shared terminal mapping. A restore-lost turn with a held result for that exact turn settles as completed, whichever path reports it > - The benefit is that ACPX and OpenCode tasks that already finished keep their accepted result after a restart, as Codex tasks now do ## Linked Issues or Issue Description Refs #19658, #19525, #19501, #19539 **What happened?** In a restart chaos wave, an ACPX Claude task (workload A1) wrote its brief and proposed a `done` result (`run.result.proposed`, then a completed `paperclip_finish` tool call). Then a K1 and a K6 crash loop hit the control plane, and `resume_dead_runner` ran at controller generations 2 and 3. The restored ACPX conversation emitted `turn.failed` with `provider_turn_lost_on_restore` (`recoverable: true`) for the checkpointed turn. The control plane then emitted `run.result.accepted` with the `done` result, but `run.terminal` came back `runTerminalState: failed` / `turnTerminalState: failed` / `reportedWorkDisposition: done`. The run ended `failed` / `adapter_failed`, "Native session failed". **Steps to reproduce** 1. Start a native ACPX (or OpenCode) task through runnerd. 2. Let the agent call `paperclip_finish`, and let the control plane persist the proposal. 3. Kill the control plane and runner before the turn ends. 4. Let restart recovery resume the run. The restored runner reports the turn lost on restore as a live terminal. **Expected behavior** Recovery finalizes the accepted result, and the run succeeds. **Actual behavior** ACPX and OpenCode use the same Codex harness session as Codex (`codex-native-backend.ts`). Recovery restored the held result for the turn from the durable proposal. But the live `turn/completed` notification with `status: failed` and `provider_turn_lost_on_restore` went through `mapTerminalTurn`, which emitted `turn.failed`. The backend maps `turn.failed` to `runTerminalState: failed`. #19658 handled only the `thread/read` reconciliation path. ## What Changed - `codex-session-terminal.ts`: a new `settleRestoreLostTurnWithHeldResult` helper. When a terminal is `failed` with `provider_turn_lost_on_restore` and the session holds the semantic result that this exact turn proposed, the turn settles as `completed`. `mapTerminalTurn` applies it first, so live notifications and recovery reads follow one rule. - `codex-session-terminal.ts`: the helper also covers `acpx_active_turn_recovery_closed`. Runnerd's ACPX backend (`acpx_provider_backend.rs`, `restore_session_if_needed`) emits that code instead of restoring the conversation when its own state already recorded the accepted result. It is the same restart loss, with the result already accepted. #19525 already keeps accepted results when close cannot prove suspension, and session close still handles `provider_exit_unconfirmed`. - `codex-harness-session.ts`: `reconcile()` keeps only the missing-turn case from #19658. The restore-lost read case and the prior-terminal check now use the shared helper. - `doc/execution-semantics.md`: the restart paragraph now states that the rule covers Codex, OpenCode, and ACPX, and both report paths. ## Verification - Red before the fix, with the real Codex native backend (`createCodexNativeSessionBackend`) and a fake transport through `executeNativeSession`: `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/native-session-runtime.test.ts -t "journaled completion proposal"`. The new `restore-lost live` scenario failed. Green after the fix: the run completes with the journaled result on `turn-1` and a succeeded terminal, and no `turn/start` is sent. The `restore-lost read` scenario passes on both (#19658 already covered it). - Driver red/green: `src/drivers/codex/codex-app-server-driver.recovery.test.ts -t "live restore-lost"`. With a held result, the live lost terminal was `turn.failed` before the fix and is `turn.completed` after it. With no held result, it stays `turn.failed` with the lost cause, so the #19501 continuation is unchanged. - The driver case was extended to `acpx_active_turn_recovery_closed` (red with a held result before the extension, then green). It was found in a live replay where the run ended `failed`/`adapter_failed` with an accepted `done` result. - Suites: `src/drivers/codex/` and `src/native-session-runtime.test.ts` (580 tests), plus `src/backends/` and `src/live/runnerd-restart-continuation.test.ts`, pass. `typecheck:typescript` passes. - Live chaos replays (ACPX Claude): results and a separate new signature are in a comment below. ## Risks - The rule needs a held result whose turn id equals the lost turn. Recovery restores that result only from the durable `run.result.proposed` for the checkpointed turn, so a result for another turn does not settle it. - A restore-lost turn without a held result keeps its failure and the bounded same-conversation continuation. - Real provider failures and other terminal states are unchanged. - No schema, profile, pack contract, or database changes. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
8a34c1b62c
commit
ece09c156a
5 files changed
+169
-24
No files matched your search
@@ -2576,8 +2576,11 @@ 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
|
||||
result that exact turn proposed, recovery also settles the turn with that result.
|
||||
This rule applies to Codex, OpenCode, and ACPX, both when the recovery read reports
|
||||
the lost turn and when the restored conversation reports it live. 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.
|
||||
|
||||
Each admitted restart attempt that loses a turn this way gets its own continuation.
|
||||
|
||||
@@ -43,6 +43,7 @@ import {
|
||||
type PrpStructuredRunResult,
|
||||
} from "./codex-app-server-driver.test-support.js";
|
||||
import { NativeSessionProtocolIntegrityError, nativeRestartInterruptedTurnId } from "../../contracts/native-session-backend.js";
|
||||
import { canonicalJson } from "./codex-driver-values.js";
|
||||
|
||||
function cursorProviderIdentity(mode: unknown) {
|
||||
return {
|
||||
@@ -1324,6 +1325,93 @@ describe("Codex app-server Codex driver", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ held: true, lost: { code: "provider_turn_lost_on_restore", recoverable: true } },
|
||||
{ held: false, lost: { code: "provider_turn_lost_on_restore", recoverable: true } },
|
||||
{ held: true, lost: { code: "acpx_active_turn_recovery_closed", recoverable: false } },
|
||||
{ held: false, lost: { code: "acpx_active_turn_recovery_closed", recoverable: false } },
|
||||
])("settles a live restart-lost terminal from its durable result only when one is held ($lost.code, held=$held)", async ({ held, lost }) => {
|
||||
// ACPX reports the lost turn as a live notification after the session
|
||||
// reopens, not through thread/read. When runnerd already recorded the
|
||||
// accepted result, it closes the turn instead of restoring it. An accepted
|
||||
// result for that exact turn must still finalize as a completed turn.
|
||||
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: "inProgress", items: [] }] },
|
||||
};
|
||||
const driver = makeDriver([first, second]);
|
||||
const original = await driver.openSession({
|
||||
runId: `run-live-lost-${held}`,
|
||||
normalizedSessionId: `normalized-live-lost-${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 collecting = collectUntilTerminal(recovery!.session!.events());
|
||||
second.push("turn/completed", { threadId: "thread-1", turn: { id: "turn-1", status: "failed", error: lost, items: [] } });
|
||||
const events = await collecting;
|
||||
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 });
|
||||
if (held) expect(events.filter((event) => event.eventType === "run.result.proposed")).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("accepts a stored restart-lost failure for a held result without a replay conflict", async () => {
|
||||
// A runner from before this rule recorded the restart-lost turn as failed
|
||||
// with its held result. Replaying that same terminal is not a conflict.
|
||||
// (A held result plus a committed terminal makes the session terminal, so
|
||||
// the recovery read is the path that can replay 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-stored-lost",
|
||||
normalizedSessionId: "normalized-stored-lost",
|
||||
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 stored = canonicalJson({ terminalState: "failed", error: lost, result });
|
||||
const recovery = await driver.recoverSession?.({
|
||||
...snapshot,
|
||||
activeTurnId: null,
|
||||
terminalTurns: [{ turnId: "turn-1", fingerprint: stored }],
|
||||
});
|
||||
expect(recovery).toMatchObject({ recovered: true });
|
||||
await recovery!.session!.reconcile?.();
|
||||
const events: string[] = [];
|
||||
const draining = (async () => { for await (const event of recovery!.session!.events()) events.push(event.eventType); })();
|
||||
await new Promise((resolveWait) => setTimeout(resolveWait, 50));
|
||||
await recovery!.session!.close({ reason: "test complete" });
|
||||
await draining;
|
||||
expect(events).not.toContain("session.failed");
|
||||
expect(events).not.toContain("harness.protocol_failed");
|
||||
expect((await recovery!.session!.snapshot().catch((error: unknown) => error)) instanceof Error).toBe(false);
|
||||
});
|
||||
|
||||
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.
|
||||
|
||||
@@ -2,7 +2,7 @@ import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from
|
||||
import type { DurableRecoveryIdentity } from "../../contracts/durable-recovery.js";
|
||||
import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { isNativeRestartInterruption, NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";
|
||||
import { NativeProviderTerminalFailure, NativeSessionProtocolIntegrityError } from "../../contracts/native-session-backend.js";
|
||||
|
||||
import type {
|
||||
HarnessGoalOperation,
|
||||
@@ -53,7 +53,8 @@ import { pumpNotifications } from "./codex-session-notifications.js";
|
||||
import { handleServerRequest } from "./codex-session-server-requests.js";
|
||||
import {
|
||||
mapTerminalTurn,
|
||||
terminalReplayConflict,
|
||||
settleRestoreLostTurnWithHeldResult,
|
||||
settledTerminalReplayConflict,
|
||||
} from "./codex-session-terminal.js";
|
||||
import { boundedText, record, text, userInput } from "./codex-driver-values.js";
|
||||
|
||||
@@ -800,15 +801,14 @@ export class CodexHarnessSession
|
||||
const expectedTurn = turns.find(
|
||||
(turn) => text(turn.id) === expectedTurnId,
|
||||
);
|
||||
const holdsTurnResult = this.result !== null && this.resultTurnId === expectedTurnId;
|
||||
if (holdsTurnResult && (expectedTurn === undefined || (
|
||||
text(expectedTurn.status) === "failed" && isNativeRestartInterruption(expectedTurn.error)))) {
|
||||
if (expectedTurn === undefined && this.result !== null && this.resultTurnId === expectedTurnId) {
|
||||
// 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.
|
||||
// terminal was flushed. 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.
|
||||
// (A turn reported lost on restore settles the same way in
|
||||
// mapTerminalTurn.)
|
||||
reconciledTerminalTurnId = expectedTurnId;
|
||||
mapTerminalTurn(this, { id: expectedTurnId, status: "completed", items: [] }, expectedTurnId, true);
|
||||
} else if (expectedTurn === undefined) {
|
||||
@@ -833,12 +833,11 @@ export class CodexHarnessSession
|
||||
}
|
||||
|
||||
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;
|
||||
const reported = turns.find((turn) => text(turn.id) === turnId);
|
||||
if (reported === undefined) continue;
|
||||
// A turn settled from its held result keeps that outcome when the
|
||||
// restored runner later reports the same turn lost on restore.
|
||||
const observed = settleRestoreLostTurnWithHeldResult(this, reported, turnId);
|
||||
if (
|
||||
!["completed", "failed", "interrupted", "cancelled"].includes(
|
||||
text(observed.status),
|
||||
@@ -848,7 +847,7 @@ export class CodexHarnessSession
|
||||
`previously terminal turn ${turnId} is no longer terminal in thread/read`,
|
||||
);
|
||||
}
|
||||
const conflict = terminalReplayConflict(this, observed, fingerprint);
|
||||
const conflict = settledTerminalReplayConflict(this, reported, observed, fingerprint);
|
||||
if (conflict !== null) {
|
||||
throw new HarnessReconciliationError(
|
||||
`previously terminal turn ${turnId} ${conflict.message}`,
|
||||
|
||||
@@ -101,6 +101,53 @@ function finalize(
|
||||
state.terminal = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* ACPX closes, rather than restores, a turn that a restart interrupted after
|
||||
* the runner had recorded its accepted semantic result.
|
||||
*/
|
||||
const ACPX_RESTART_CLOSED_TURN_CODE = "acpx_active_turn_recovery_closed";
|
||||
|
||||
/**
|
||||
* The restored runner reports a turn lost on restore when it cannot prove the
|
||||
* turn's outcome itself, and ACPX closes the turn when its runner already
|
||||
* recorded the accepted result. In both cases a restart lost only the provider
|
||||
* turn. When Paperclip already holds the semantic result this exact turn
|
||||
* proposed, that result is the turn's outcome: the turn settles as completed
|
||||
* instead of failing the run and rerunning finished work. Live notifications
|
||||
* and recovery reads use the same rule. Any other terminal, or a lost turn
|
||||
* without a held result, is returned unchanged.
|
||||
*/
|
||||
export function settleRestoreLostTurnWithHeldResult(
|
||||
state: CodexSessionState,
|
||||
turn: Record<string, unknown>,
|
||||
turnId = text(turn.id),
|
||||
): Record<string, unknown> {
|
||||
const restartLost = isNativeRestartInterruption(turn.error) ||
|
||||
record(turn.error).code === ACPX_RESTART_CLOSED_TURN_CODE;
|
||||
if (text(turn.status) !== "failed" || !restartLost ||
|
||||
state.result === null || turnId.length === 0 || state.resultTurnId !== turnId) {
|
||||
return turn;
|
||||
}
|
||||
return { ...turn, status: "completed", error: null };
|
||||
}
|
||||
|
||||
/**
|
||||
* A replay of an already committed terminal matches when either the terminal
|
||||
* as reported or its settled form matches. A checkpoint from before the
|
||||
* restart-loss rule can hold the reported failure with the same held result;
|
||||
* that is the same terminal, not a conflicting one.
|
||||
*/
|
||||
export function settledTerminalReplayConflict(
|
||||
state: CodexSessionState,
|
||||
reported: Record<string, unknown>,
|
||||
settled: Record<string, unknown>,
|
||||
expectedFingerprint: string,
|
||||
): TerminalReplayConflict | null {
|
||||
const conflict = terminalReplayConflict(state, settled, expectedFingerprint);
|
||||
if (conflict === null || settled === reported) return conflict;
|
||||
return terminalReplayConflict(state, reported, expectedFingerprint) === null ? null : conflict;
|
||||
}
|
||||
|
||||
export function mapTerminalTurn(
|
||||
state: CodexSessionState,
|
||||
turn: Record<string, unknown>,
|
||||
@@ -108,10 +155,12 @@ export function mapTerminalTurn(
|
||||
reconciling = false,
|
||||
): void {
|
||||
const turnId = text(turn.id, fallbackTurnId || state.activeTurnId || "");
|
||||
const reported = turn;
|
||||
turn = settleRestoreLostTurnWithHeldResult(state, turn, turnId);
|
||||
const status = text(turn.status, "completed");
|
||||
const previous = state.terminalTurns.get(turnId);
|
||||
if (previous !== undefined) {
|
||||
const conflict = terminalReplayConflict(state, turn, previous);
|
||||
const conflict = settledTerminalReplayConflict(state, reported, turn, previous);
|
||||
if (conflict !== null) {
|
||||
if (reconciling) {
|
||||
throw new HarnessReconciliationError(
|
||||
|
||||
@@ -338,7 +338,7 @@ describe("executeNativeSession recovery", () => {
|
||||
|
||||
it.each([
|
||||
"completed", "pending", "duplicate", "conflicting", "other turn", "newer turn",
|
||||
"wrong session", "invalid",
|
||||
"wrong session", "invalid", "restore-lost live", "restore-lost read",
|
||||
])("recovers a journaled completion proposal after restart: %s", async (scenario) => {
|
||||
const originalTransport = new FakeCodexTransport();
|
||||
const recoveredTransport = new FakeCodexTransport();
|
||||
@@ -388,7 +388,9 @@ describe("executeNativeSession recovery", () => {
|
||||
await original.close({ reason: "controller disconnected after persisting the proposal" });
|
||||
recoveredTransport.readResponse = { thread: {
|
||||
id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE,
|
||||
turns: [{ id: "turn-1", status: scenario === "pending" ? "inProgress" : "completed", items: [
|
||||
turns: [{ id: "turn-1", ...(scenario === "restore-lost read"
|
||||
? { status: "failed", error: { code: "provider_turn_lost_on_restore", recoverable: true } }
|
||||
: { status: ["pending", "restore-lost live"].includes(scenario) ? "inProgress" : "completed" }), items: [
|
||||
{ id: "answer", type: "agentMessage", text: "The review was published." },
|
||||
] }],
|
||||
} };
|
||||
@@ -430,11 +432,15 @@ describe("executeNativeSession recovery", () => {
|
||||
expect(recoveredTransport.calls.filter(call => call.method === "turn/start")).toEqual([]);
|
||||
return;
|
||||
}
|
||||
if (scenario === "pending") {
|
||||
if (scenario === "pending" || scenario === "restore-lost live") {
|
||||
await vi.waitFor(() => expect(checkpoints[0]?.semanticResult).toEqual(result));
|
||||
expect(completeRun).not.toHaveBeenCalled();
|
||||
// ACPX reports a turn lost on restore as a live terminal once the
|
||||
// restored runner reopens the conversation.
|
||||
recoveredTransport.push("turn/completed", {
|
||||
threadId: "thread-1", turn: { id: "turn-1", status: "completed", items: [] },
|
||||
threadId: "thread-1", turn: scenario === "pending"
|
||||
? { id: "turn-1", status: "completed", items: [] }
|
||||
: { id: "turn-1", status: "failed", error: { code: "provider_turn_lost_on_restore", recoverable: true }, items: [] },
|
||||
});
|
||||
}
|
||||
await expect(execution).resolves.toMatchObject({ result, turnId: "turn-1", terminal });
|
||||
|
||||
Reference in new issue
Block a user