diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs index 58fffdfb2e4..fb77b4db2f9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs @@ -822,9 +822,18 @@ impl DurableState { .checked_add(1) .ok_or_else(|| DurableRunnerError::invalid("controller sequence exhausted"))?; if command.controller_seq != expected { - return Err(DurableRunnerError::invalid(format!( - "controller sequence must be contiguous: expected {expected}, received {}", - command.controller_seq + return Ok(CommandDisposition::Reject(command_result( + command, + "rejected", + json!({ + "code": "controller_sequence_mismatch", + "message": format!( + "controller sequence must be contiguous: expected {expected}, received {}", + command.controller_seq + ), + "expectedControllerSeq": expected, + "receivedControllerSeq": command.controller_seq, + }), ))); } @@ -3287,7 +3296,11 @@ mod tests { let mut unknown = command("command_unknown", 1); unknown.command_type = "future.required.command".to_owned(); assert!(state.begin_command(&unknown).is_err()); - assert!(state.begin_command(&command("command_2", 2)).is_err()); + assert!(matches!( + state.begin_command(&command("command_2", 2)).unwrap(), + CommandDisposition::Reject(result) if result.result["code"] == "controller_sequence_mismatch" + )); + assert_eq!(state.last_controller_command_seq, 0); let first = command("command_1", 1); state.begin_command(&first).unwrap(); state.complete_command(&first, json!({})).unwrap(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs index 1ff0cd61168..32c63e148e4 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/durable_recovery.rs @@ -189,3 +189,52 @@ fn pre_fingerprint_journal_recovers_without_reexecuting_old_commands() { fs::remove_dir_all(directory).expect("remove integration-test state"); } + +#[test] +fn an_out_of_order_new_command_is_rejected_without_changing_the_journal() { + let directory = temporary_directory(); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).expect("create private state store"); + let (mut state, _) = store.load_or_create(&config).expect("create durable state"); + state.last_controller_command_seq = 12; + + let stale = Command { + command_id: "command_stale".to_owned(), + controller_seq: 11, + ..command() + }; + let disposition = state + .begin_command(&stale) + .expect("an out-of-order command is not fatal"); + assert!(matches!( + &disposition, + CommandDisposition::Reject(result) + if result.status == "rejected" + && result.result["code"] == "controller_sequence_mismatch" + && result.result["expectedControllerSeq"] == 13 + && result.result["receivedControllerSeq"] == 11 + )); + let gap = Command { + command_id: "command_gap".to_owned(), + controller_seq: 15, + ..command() + }; + assert!(matches!( + state.begin_command(&gap).expect("a sequence gap is not fatal"), + CommandDisposition::Reject(result) if result.result["code"] == "controller_sequence_mismatch" + )); + assert_eq!(state.last_controller_command_seq, 12); + assert!(!state.processed_commands.contains_key("command_stale")); + + let next = Command { + command_id: "command_next".to_owned(), + controller_seq: 13, + ..command() + }; + assert_eq!( + state.begin_command(&next).expect("the contiguous command"), + CommandDisposition::Execute + ); + + fs::remove_dir_all(directory).expect("remove integration-test state"); +} diff --git a/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts b/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts index 3941f32ea80..fe240d52a1c 100644 --- a/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts @@ -31,7 +31,7 @@ function kill(pid: number | null | undefined) { try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } } -it.each(["stale", "current"] as const)("a follow-up controller resumes the predecessor's suspended runner from a %s journal", async journal => { +it.each(["stale", "current"] as const)("a follow-up controller on the predecessor's suspended runner with a %s journal keeps the runner and gets a precise answer", async journal => { const root = await mkdtemp(join(tmpdir(), "runnerd-stale-journal-")); const controlPath = join(root, "control-plane", "control-plane-state.json"); const predecessor = transport(root); @@ -50,10 +50,11 @@ it.each(["stale", "current"] as const)("a follow-up controller resumes the prede const outcome = await follower.transport.request("thread/read", {}) .then(() => "resumed", (error: unknown) => String(error)); if (journal === "stale") { - expect(outcome).not.toBe("resumed"); - expect(outcome).toContain("controller sequence must be contiguous"); - await follower.transport.close().catch(() => undefined); - expect(follower.transport.runnerExitConfirmed?.()).toBe(true); + expect(outcome).toContain('"code":"controller_sequence_mismatch"'); + expect(outcome).not.toMatch(/runnerd exited/); + expect(follower.evidence().runnerExited).toBe(false); + const after = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")) as { lastControllerCommandSeq: number }; + expect(after.lastControllerCommandSeq).toBe(closed.commands.length); } else { expect(outcome).toBe("resumed"); expect(outcome).not.toContain("controller sequence must be contiguous"); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 9422dedaf6c..9f3ce96b9d1 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -449,6 +449,7 @@ import { NATIVE_PROVIDER_STARTUP_REJECTED, nativeRemoteStartupRejectionCode, nativeRunnerExitedBeforeSessionOpen, + NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER, nativeRepeatedStartupFailureAction, nativePreviousStartupFailureSignature, nativeStartupFailureSignature, @@ -14489,6 +14490,14 @@ describe("native session bounded recovery", () => { }); }); + it("rests a run whose Runner refused its commands as out of sequence with a precise code and a Retry", () => { + const refused = new Error(`PRP command session.open rejected: {"commandId":"c","controllerSeq":11,"commandType":"session.open","status":"rejected","result":{"code":"controller_sequence_mismatch","message":"controller sequence must be contiguous: expected 13, received 11","expectedControllerSeq":13,"receivedControllerSeq":11}}`); + expect(nativeSessionFailureSourceCode(refused)).toBe(NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER); + expect(nativeSessionFailureSourceCode(new Error(`native_session_recovery_failed: ${refused}`))).toBe(NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER); + expect(nativeSessionFailureDisposition(1, new Date(), NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER)).toMatchObject({ + phase: "terminal_failure", failureCode: NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER }); + }); + it("records a runner proven to have exited before answering session.open as a failed fresh start", () => { const proven = (message: string) => Object.defineProperty(new Error(message), "runnerExitConfirmed", { value: true }); expect(nativeRunnerExitedBeforeSessionOpen(new Error("native_session_recovery_failed", { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 8b07b87646f..0d23246118f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -8575,7 +8575,8 @@ export function nativeSessionFailureDisposition( sourceFailureCode === "native_adopted_runner_authentication_timeout" || sourceFailureCode === "native_provider_usage_limit" || sourceFailureCode === NATIVE_PROVIDER_SESSION_CLOSED || - sourceFailureCode === NATIVE_PROVIDER_STARTUP_REJECTED; + sourceFailureCode === NATIVE_PROVIDER_STARTUP_REJECTED || + sourceFailureCode === NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER; const permissionDeclined = sourceFailureCode === "native_permission_declined"; const exhausted = permanentFailure || permissionDeclined || attempt >= 3; return { @@ -8628,6 +8629,9 @@ export const NATIVE_PROVIDER_SESSION_CLOSED_ACTION = "The provider session closed after a turn start that may have reached the provider, and this run cannot reopen it. Retry starts a new run with a fresh provider conversation and the saved task context."; const RUNNER_PROVIDER_SESSION_CLOSED = "provider_session_closed"; +export const NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER = "native_controller_journal_behind_runner"; +export const NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER_ACTION = + "The saved controller journal was behind its Runner, so the Runner refused this run's commands and no provider work began. Retry starts a new run, which reconciles the journal from the saved checkpoint before it opens the session, or stops with a precise error if it cannot."; export const NATIVE_PROVIDER_STARTUP_REJECTED = "native_provider_startup_rejected"; export const NATIVE_PROVIDER_STARTUP_REJECTED_ACTION = "The provider refused to start a session, so no provider work began. Read the provider message and fix its cause in the agent's environment or AI connection if needed, then Retry. Retry starts a new run with a fresh provider session."; @@ -8735,6 +8739,7 @@ export function nativeSessionFailureSourceCode( | "native_provider_usage_limit" | typeof NATIVE_PROVIDER_SESSION_CLOSED | typeof NATIVE_PROVIDER_STARTUP_REJECTED + | typeof NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER | "native_session_cleanup_quarantined" | "native_adopted_runner_authentication_timeout" | "runner_remote_provider_artifact_incompatible" @@ -8801,6 +8806,7 @@ export function nativeSessionFailureSourceCode( } if (runnerClosedTurnStartSession(error)) return NATIVE_PROVIDER_SESSION_CLOSED; const message = error instanceof Error ? error.message : String(error); + if (/"code":"controller_sequence_mismatch"/.test(message)) return NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER; if (/^(?:runner_ingress_unavailable: runnerd exited unexpectedly with code 1: )?runner_task_admission_rejected: (session_state_unavailable|account_binding_mismatch)$/.test(message)) { return "runner_task_admission_rejected"; } @@ -12733,6 +12739,8 @@ async function executePaperclipNativeSessionAttempt( ? NATIVE_PROVIDER_SESSION_CLOSED_ACTION : sourceFailureCode === NATIVE_PROVIDER_STARTUP_REJECTED ? startupRejectedAction + : sourceFailureCode === NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER + ? NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER_ACTION : sourceFailureCode === "native_session_cleanup_quarantined" ? nativeCleanupRefusalAction(completedTurnCleanup) : sourceFailureCode === "runner_remote_provider_verification_unavailable" @@ -13012,6 +13020,8 @@ async function executePaperclipNativeSessionAttempt( ? NATIVE_PROVIDER_SESSION_CLOSED_ACTION : sourceFailureCode === NATIVE_PROVIDER_STARTUP_REJECTED ? startupRejectedAction + : sourceFailureCode === NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER + ? NATIVE_CONTROLLER_JOURNAL_BEHIND_RUNNER_ACTION : sourceFailureCode === "native_session_cleanup_quarantined" ? nativeCleanupRefusalAction(completedTurnCleanup) : sourceFailureCode === "runner_task_admission_rejected"