diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index b146dc651a2..040fb5748e2 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -233,6 +233,16 @@ Attaching starts no turn, and the runner admits the new command only for a settled session, so the run continues instead of failing on the old command. Warm transitions and staged attachments keep their exact replay paths. +A restart can kill the runner inside `runner.suspend` in the same way. The +next runner replays it from the welcome as `execution_indeterminate`, sends it +as `failed`, and still treats it as terminal: it records `suspended` and exits +once the controller acknowledges the result. The controller therefore keeps the +terminal fence for that result (it lifts the fence only for a suspension the +runner refused), and its acknowledgement names the runner's journal status +`indeterminate`. Current runners accept either `indeterminate` or `failed`; older +runners accept only `indeterminate` and otherwise exit with "terminal command +result acknowledgement changed its durable identity". + A kill during that attachment's provider launch can also leave a `spawned` startup fence over a `prepared` state that holds a saved thread. The next runner retires that fence once the recorded provider process and its group have exited, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index fb8099b141a..ae1d3bb7aa4 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -1752,13 +1752,18 @@ fn wait_for_terminal_result_ack( "terminal command result acknowledgement payload is required", ) })?; + // The controller acknowledges the status it received. An + // indeterminate journal entry travels as `failed`, so either + // spelling names this exact receipt. + let status = payload.get("status").and_then(Value::as_str); if payload.get("commandId").and_then(Value::as_str) != Some(result.command_id.as_str()) || payload.get("commandType").and_then(Value::as_str) != Some(result.command_type.as_str()) || payload.get("controllerSeq").and_then(Value::as_u64) != Some(result.controller_seq) - || payload.get("status").and_then(Value::as_str) != Some(result.status.as_str()) + || (status != Some(result.status.as_str()) + && status != Some(wire_command_status(result))) { return Err(DurableRunnerError::invalid( "terminal command result acknowledgement changed its durable identity", @@ -2551,18 +2556,24 @@ fn send_outbox( Ok(()) } +/// "indeterminate" is an internal crash-recovery journal state. On the wire it +/// is a failed command with the preserved execution_indeterminate reason so the +/// controller can settle the command and continue replay. +fn wire_command_status(result: &StoredCommandResult) -> &str { + if result.status == "indeterminate" { + "failed" + } else { + result.status.as_str() + } +} + fn command_result_envelope( state: &DurableState, result: &StoredCommandResult, protocol_version: u64, ) -> Value { let mut payload = json!(result); - // "indeterminate" is an internal crash-recovery journal state. On the - // wire it is a failed command with the preserved execution_indeterminate - // reason so the controller can settle the command and continue replay. - if result.status == "indeterminate" { - payload["status"] = json!("failed"); - } + payload["status"] = json!(wire_command_status(result)); json!({ "protocol": PROTOCOL, "version": protocol_version, diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/indeterminate_terminal_ack.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/indeterminate_terminal_ack.rs new file mode 100644 index 00000000000..2fd8d69188b --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/indeterminate_terminal_ack.rs @@ -0,0 +1,142 @@ +/// A runnerd killed inside `runner.suspend` leaves it journaled but unfinished. +/// The next process replays it as `indeterminate`, sends it as `failed`, and the +/// controller acknowledges the `failed` it received. That acknowledgement names +/// the same receipt: the runner finishes the suspension instead of exiting with +/// "terminal command result acknowledgement changed its durable identity". +#[test] +fn indeterminate_suspend_replay_accepts_the_acknowledged_wire_status() { + for acknowledged in ["failed", "indeterminate"] { + indeterminate_suspend_replay_is_acknowledged(acknowledged); + } +} + +fn indeterminate_suspend_replay_is_acknowledged(acknowledged_status: &'static str) { + struct NeverExecutes; + + impl super::super::CommandExecutor for NeverExecutes { + fn execute( + &mut self, + command: &Command, + ) -> Result { + panic!("an indeterminate {} must never execute again", command.command_type); + } + } + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let port = listener.local_addr().unwrap().port(); + let mut config = config(port); + config.max_runtime = Duration::from_secs(5); + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-indeterminate-terminal-ack-{}-{acknowledged_status}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&directory); + config.state_dir = directory.clone(); + let suspend = json!({ + "schema": "paperclip.prp.command.v1", + "commandId": "command_suspend", + "controllerSeq": 1, + "type": "runner.suspend", + "issuedAt": "2026-11-11T01:34:11.000Z", + "payload": {}, + }); + { + // The prior process journaled the suspend before its effect and died. + let store = super::super::DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let command: Command = serde_json::from_value(suspend.clone()).unwrap(); + assert_eq!( + state.begin_command(&command).unwrap(), + super::super::CommandDisposition::Execute + ); + store.save(&state).unwrap(); + } + + let state = test_state(&config); + let expires = current_unix_ms().unwrap() + 60_000; + let server_config = config.clone(); + let server_state = state.clone(); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().unwrap(); + let mut socket = accept(stream).unwrap(); + let mut secure = server_authenticate( + &mut socket, + &server_config, + &server_state, + ServerCredential { + token: "bootstrap-secret", + kind: "bootstrap", + lease_id: None, + expires_at_unix_ms: expires, + revocation_epoch: 0, + }, + ); + send_secure( + &mut socket, + &mut secure, + &server_config, + &welcome( + &server_state, + "connection_1", + Some("lease-secret"), + expires, + 0, + vec![suspend], + ), + ); + let result = receive_secure(&mut socket, &mut secure, &server_config); + assert_eq!(result["kind"], "command_result"); + assert_eq!(result["payload"]["commandId"], "command_suspend"); + assert_eq!(result["payload"]["status"], "failed"); + assert_eq!( + result["payload"]["result"]["code"], + "execution_indeterminate" + ); + send_secure( + &mut socket, + &mut secure, + &server_config, + &control( + &server_state, + "connection_1", + "command_result_ack", + json!({ + "commandId": "command_suspend", + "commandType": "runner.suspend", + "controllerSeq": 1, + "status": acknowledged_status, + }), + ), + ); + // Acknowledge any terminal suffix until the runner closes. + while let Ok(frame) = receive_plain(&mut socket, server_config.max_frame_bytes) { + let message = secure.decrypt(&frame, true).unwrap(); + if message["kind"] == "event" { + send_secure( + &mut socket, + &mut secure, + &server_config, + &control( + &server_state, + "connection_1", + "ack", + json!({"ackedSourceSeq": message["payload"]["sourceSeq"]}), + ), + ); + } + } + }); + + super::super::run_durable_runner( + config.clone(), + BootstrapTicket::new("bootstrap-secret".to_owned()).unwrap(), + NeverExecutes, + ) + .unwrap(); + server.join().unwrap(); + let store = super::super::DurableStateStore::new(&directory).unwrap(); + let (state, _) = store.load_or_create(&config).unwrap(); + assert_eq!(state.lifecycle, "suspended"); + assert!(state.pending_terminal_delivery.is_none()); + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs index e64d60ddf9a..8b245375882 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/transport.rs @@ -1764,6 +1764,7 @@ mod tests { #[cfg(unix)] include!("tests/hosted_retention.rs"); include!("tests/ack_release_poll.rs"); + include!("tests/indeterminate_terminal_ack.rs"); use std::net::{IpAddr, Ipv4Addr, TcpListener}; use std::path::PathBuf; diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index fa77aafb347..698f8a475bc 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -5377,6 +5377,54 @@ describe("DurablePrpControlPlane", () => { } }); + it("acknowledges an indeterminate suspension in the Runner's journal status and keeps its fence", async () => { + // A runnerd killed inside runner.suspend journals it as indeterminate. The + // next runnerd replays it from the welcome as `failed` with + // execution_indeterminate and then exits as suspended. Shipped Runners + // compare the acknowledgement with their journal status and exited with + // "terminal command result acknowledgement changed its durable identity". + const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-indeterminate-suspend-")); + const controlPlane = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + }); + try { + await controlPlane.start(); + const stale = controlPlane.queueCommand("runner.suspend", {}, "command-suspend-stale"); + const next = controlPlane.queueCommand("session.snapshot", {}, "command-snapshot-next"); + const client = await authenticate(controlPlane, controlPlane.issueBootstrapTicket()); + sendSecure(client!, { + protocol: "paperclip.runner", version: 1, kind: "command_result", + payload: { + commandId: stale.commandId, commandType: stale.type, controllerSeq: stale.controllerSeq, + status: "failed", + result: { + code: "execution_indeterminate", + message: "runner recovered after journaling this command; it will not execute twice", + }, + }, + }); + await expect(receiveSecure(client!)).resolves.toMatchObject({ + kind: "command_result_ack", + payload: { + commandId: "command-suspend-stale", + commandType: "runner.suspend", + controllerSeq: stale.controllerSeq, + status: "indeterminate", + }, + }); + expect(controlPlane.getCommand(stale.commandId)?.status).toBe("failed"); + // The Runner is leaving as suspended; nothing may follow on this connection. + expect(controlPlane.store.state.commandDeliveryCounts[next.commandId]).toBeUndefined(); + client?.socket.destroy(); + } finally { + await controlPlane.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + it("acknowledges terminal command results after persisting them", async () => { const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-terminal-ack-")); const controlPlane = new DurablePrpControlPlane({ diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index 6e755c227f0..4a0ca8354a8 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -84,6 +84,15 @@ export function isRunnerRecoveryMaintenanceCommand(type: string): boolean { return ["session.snapshot", "session.goal.get", "runner.drain", "runner.suspend", "runner.shutdown", "turn.stop"].includes(type); } +/** The Runner journaled this command before its effect, then died. It sends + * that `indeterminate` entry as `failed` with `execution_indeterminate`, never + * executes it again, and still treats a lifecycle command as terminal. */ +function isIndeterminateCommandResult(command: DurableRecoveryCoreCommand): boolean { + if (command.status === "indeterminate") return true; + const detail = command.result?.result; + return command.status === "failed" && isRecord(detail) && detail.code === "execution_indeterminate"; +} + const commandTypes = new Set([ "run.prepare", "run.attach", @@ -4102,7 +4111,8 @@ export class DurablePrpControlPlane { * A rejected runner.suspend changes no provider state, and the runner keeps * serving this connection. Lift the terminal-lifecycle fence it installed * so the controller's next command (a fresh suspend) can be delivered. - * Completed suspensions and every shutdown keep the fence. + * Completed suspensions and every shutdown keep the fence. So does an + * indeterminate suspension: the Runner treats it as terminal and exits. */ #releaseServingLifecycle( connection: AuthorityConnection, @@ -4111,6 +4121,7 @@ export class DurablePrpControlPlane { if ( command.type === "runner.suspend" && (command.status === "failed" || command.status === "rejected") && + !isIndeterminateCommandResult(command) && connection.terminalLifecycleCommandId === command.commandId ) { connection.terminalLifecycleCommandId = null; @@ -4159,7 +4170,11 @@ export class DurablePrpControlPlane { commandId: command.commandId, commandType: command.type, controllerSeq: command.controllerSeq, - status: command.status, + // A Runner journals a command it may have started before dying as + // `indeterminate` and sends it as `failed`. Runners compare this + // acknowledgement with their journal, so name that journal status; + // current Runners accept either spelling. + status: isIndeterminateCommandResult(command) ? "indeterminate" : command.status, }, ), );