diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 289b51720c8..e0e66ab3a32 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2415,8 +2415,11 @@ run, release its environment, or start another provider. Lease expiry alone does not authorize that stale preparation to take ownership; normal recovery must establish a new claim first. -An exact local Codex restart can restore the conversation after losing its active -turn. Runnerd records this as `provider_turn_lost_on_restore`, with no invented +An exact local Codex, OpenCode, or ACPX restart can restore the conversation after +losing its active turn. ACPX first requires the original provider lifetime to lose +its fence quorum; without that proof, or with an accepted result, unanswered +runtime request, or active goal, it closes the session as before. Cursor Cloud +agents run remotely and are excluded. Runnerd records this as `provider_turn_lost_on_restore`, with no invented task result. The admitted restart attempt may send one continuation in that same conversation. It preserves the workspace and asks the agent to reconcile unfinished commands and external actions before proceeding. It does not resend the task or diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 916a3a6d0a0..29e7669f97e 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -348,13 +348,24 @@ that ledger records `remote_execution_never_claimed` with a `bootstrap_incomplete` disposition. The provider attempt remains zero during classification; the executor increments it before starting provider work. -For a local Codex turn lost while runnerd was stopped, the canonical `turn.failed` +For a local Codex, OpenCode, or ACPX turn lost while runnerd was stopped, the canonical `turn.failed` event retains `error.code: "provider_turn_lost_on_restore"` and `error.recoverable: true`. It records the interrupted turn, not a final task outcome. There is no synthesized result for that interruption. An admitted same-session continuation emits its own `turn.submitted`, accepted turn ID, and terminal events; only the final outcome closes the run. Reconstructed thread history retains the original error so reconciliation and live delivery agree. +An ACPX session records this only after the original provider lifetime loses +its fence quorum and the same conversation is restored. Otherwise its +`turn.failed` carries `error.code: "acpx_active_turn_recovery_closed"` with +`error.recoverable: false`, and the run closes with an honest restart message. +After an ACPX restore, the controller can still deliver results for semantic +calls of the lost turn. Runnerd acknowledges each one as `retired` without +sending it to the restored provider, so the calls settle and live semantic +dispatch resumes for the continuation. A restored ACPX provider is a new process +whose MCP request ids restart, so its semantic `callId` values carry a +per-process prefix (`r:`). They never reuse an earlier call's id in +the same run. ## Native Local Process Stop Evidence diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs index 65647805628..b3ade8b5308 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs @@ -1,7 +1,7 @@ use crate::generated_acpx_profiles::{ acpx_release_profile, ACPX_DRIVER_KIND, QUALIFIED_ACPX_VERSION, }; -use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet, VecDeque}; use std::fs::{self, DirBuilder, File}; use std::io::{Read, Write}; use std::net::TcpListener; @@ -70,6 +70,8 @@ pub const ACPX_PROVIDER_STATE_FILE: &str = "acpx-provider-state.json"; const ACPX_PROVIDER_STATE_SCHEMA: &str = "paperclip.runner.acpx-provider-state.v3"; const MAX_PROVIDER_STATE_BYTES: u64 = 16 * 1024 * 1024; const MAX_PENDING_EVENTS: usize = 8_320; +// Matches the bounded number of pending ACPX tool calls. +const MAX_OPEN_SEMANTIC_CALLS: usize = 4_096; const MAX_EVENTS_PER_POLL: usize = 128; const PROVIDER_LIFETIME_CONFIRMATION_TIMEOUT: Duration = Duration::from_secs(5); const PROVIDER_LIFETIME_CONFIRMATION_RETRY: Duration = Duration::from_millis(10); @@ -597,6 +599,13 @@ struct AcpxDurableState { pending_runtime_requests: BTreeMap, #[serde(default = "initial_event_sequence")] next_event_sequence: u64, + /// Semantic tool calls emitted to the controller without a result yet. + #[serde(default)] + open_semantic_calls: BTreeSet, + /// Open calls of a turn lost to a runner restart. The restored provider + /// never sees their results; the controller may still deliver them. + #[serde(default)] + retired_semantic_calls: BTreeSet, } impl AcpxDurableState { @@ -623,6 +632,8 @@ impl AcpxDurableState { pending_events: VecDeque::new(), pending_runtime_requests: BTreeMap::new(), next_event_sequence: initial_event_sequence(), + open_semantic_calls: BTreeSet::new(), + retired_semantic_calls: BTreeSet::new(), } } @@ -789,6 +800,31 @@ impl AcpxDurableState { if let Some(id) = event.payload.get("requestId").and_then(Value::as_str) { self.pending_runtime_requests.remove(id); } + } else if let Some(call_id) = event + .payload + .pointer("/semantic_tool/callId") + .and_then(Value::as_str) + { + // Remember semantic calls the controller may still answer. A lost + // turn's open calls become retired calls on restore, so a later + // controller result for them settles instead of failing. + match event.event_type.as_str() { + "semantic_tool.input" + if self.open_semantic_calls.len() < MAX_OPEN_SEMANTIC_CALLS => + { + self.open_semantic_calls.insert(call_id.to_owned()); + } + "semantic_tool.result" => { + self.open_semantic_calls.remove(call_id); + } + _ => {} + } + } else if matches!( + event.event_type.as_str(), + "turn.completed" | "turn.failed" | "turn.cancelled" | "turn.interrupted" + ) { + // A provider-observed terminal settles its calls in the live bridge. + self.open_semantic_calls.clear(); } self.next_event_sequence = next_sequence; self.pending_events.push_back(PolledEvent { @@ -1116,6 +1152,9 @@ impl AcpxCommandExecutor { let previous_turn = state.active_turn_id.clone(); if unsafe_active { self.context.provider_turn_id = None; + if self.restore_conversation_after_lost_turn(previous_turn.clone())? { + return Ok(()); + } let state = self .state .as_mut() @@ -1134,6 +1173,13 @@ impl AcpxCommandExecutor { "providerTerminalObserved": false, "code": "acpx_active_turn_recovery_closed", "providerShutdownFailed": true, + // Keep the cause through the controller facade, which + // forwards only the structured error of a failed turn. + "error": { + "code": "acpx_active_turn_recovery_closed", + "recoverable": false, + "message": "The runner restarted during this turn and could not restore the conversation.", + }, }), })?; state.push(NormalizedProviderEvent { @@ -1171,6 +1217,82 @@ impl AcpxCommandExecutor { self.save_state() } + /// Restores the conversation of a turn that was active when the runner + /// died, without inventing its outcome. The original provider lifetime + /// must have lost its fence quorum first, so no second provider can act on + /// the same session. The lost turn is recorded with a recoverable + /// `provider_turn_lost_on_restore` cause and the run stays open, so the + /// controller can send its single same-conversation continuation. Returns + /// false, leaving state untouched, when restoration is not safe. + fn restore_conversation_after_lost_turn( + &mut self, + previous_turn: Option, + ) -> Result { + let Some(state) = self.state.as_ref() else { + return Ok(false); + }; + // An accepted result, an unanswered runtime request, or an active goal + // needs its own reconciliation; a plain continuation cannot settle it. + let Some(identity) = state.identity.as_ref() else { + return Ok(false); + }; + if previous_turn.is_none() + || state.provider_exit_unconfirmed + || state.semantic_result.is_some() + || !state.pending_runtime_requests.is_empty() + || !(state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection)) + { + return Ok(false); + } + match await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates) { + // Release the quorum so the restored provider can inherit it. + Ok(listeners) => drop(listeners), + Err(_) => return Ok(false), + } + let Ok(session) = self.start_session(true) else { + return Ok(false); + }; + let identity = session.identity().clone(); + let process_id = session.process_id(); + let turn_controls = session.turn_control_capabilities(); + let state = self + .state + .as_mut() + .expect("ACPX state remains available during recovery"); + state.lifecycle = "session_open".to_owned(); + state.active_turn_id = None; + state.turn_controls = turn_controls; + // The controller may still answer the lost turn's semantic calls. + // The restored provider never issued them, so settle them on arrival. + let open = std::mem::take(&mut state.open_semantic_calls); + state.retired_semantic_calls.extend(open); + state.push(NormalizedProviderEvent { + event_type: "session.resumed".to_owned(), + priority: EventPriority::P0, + payload: session_event_payload(&state.descriptor, &identity, process_id, turn_controls), + })?; + state.push(NormalizedProviderEvent { + event_type: "turn.failed".to_owned(), + priority: EventPriority::P0, + payload: json!({ + "provider": "acpx", + "providerTurnId": previous_turn, + "status": "failed", + "providerTerminalObserved": false, + "code": "provider_turn_lost_on_restore", + "error": { + "code": "provider_turn_lost_on_restore", + "recoverable": true, + "message": "The runner restored the conversation after process loss, but the previous turn is no longer active.", + }, + }), + })?; + self.recovery_pending_effects = None; + self.session = Some(session); + self.save_state()?; + Ok(true) + } + fn start_session(&self, recovering: bool) -> Result { let state = self .state @@ -2123,6 +2245,20 @@ impl AcpxCommandExecutor { let result: ToolResult = serde_json::from_value(payload.clone()).map_err(|error| { DurableRunnerError::invalid(format!("semantic tool result is invalid: {error}")) })?; + if let Some(state) = self.state.as_ref() { + if state.retired_semantic_calls.contains(&result.call_id) { + // The call belonged to a turn lost to a runner restart. Its + // provider process is gone; acknowledge the controller result + // so the call settles, without sending it to the new provider. + // The retained receipt makes this idempotent: a delivery left + // indeterminate by a crash replays to the same acknowledgement. + return Ok(CommandExecution::result(json!({ + "status": "delivered", + "callId": result.call_id, + "retired": true, + }))); + } + } self.session .as_mut() .ok_or_else(|| DurableRunnerError::invalid("ACPX session is unavailable"))? @@ -2521,6 +2657,18 @@ fn goal_terminal_disposition<'a>( } impl CommandExecutor for AcpxCommandExecutor { + /// A restored session keeps durable receipts for the lost turn's calls. + /// Acknowledging one is idempotent and never reaches the provider or the + /// business operation, so an indeterminate delivery may replay. Live + /// calls have no durable receipt; without retired receipts this stays off. + fn can_reconcile_result_delivery(&mut self) -> Result { + self.restore()?; + Ok(self + .state + .as_ref() + .is_some_and(|state| !state.retired_semantic_calls.is_empty())) + } + fn host_park_ready(&mut self) -> Result { self.ensure_persistence_healthy()?; let Some(state) = self.state.as_ref() else { @@ -5020,6 +5168,14 @@ mod tests { assert_eq!(events[0].payload["replayAllowed"], false); assert_eq!(events[1].event_type, "turn.failed"); assert_eq!(events[1].payload["providerShutdownFailed"], true); + // An unanswered runtime request (and an unproven original lifetime) + // keeps the close path. Its cause survives the controller facade and + // is not offered for the restart continuation. + assert_eq!( + events[1].payload["error"]["code"], + "acpx_active_turn_recovery_closed" + ); + assert_eq!(events[1].payload["error"]["recoverable"], false); assert_eq!(events[2].event_type, "run.terminal"); assert_eq!(events[2].payload["schema"], "paperclip.prp.terminal.v1"); assert_eq!(events[2].payload["turnTerminalState"], "failed"); @@ -5049,6 +5205,74 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn lost_turn_restoration_refuses_unreconciled_state_before_touching_the_provider() { + let directory = temporary_directory("lost-turn-guards"); + let provider_descriptor: AcpxProviderDescriptor = + serde_json::from_value(descriptor("claude")).unwrap(); + let operations = Vec::new(); + let tool_set = AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.to_owned(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + }; + let identity = AcpxProviderSessionIdentity { + kind: "acpx".to_owned(), + normalized_session_id: "session-1".to_owned(), + acpx_record_id: "record-1".to_owned(), + backend_session_id: "backend-1".to_owned(), + agent_session_id: "agent-1".to_owned(), + profile_digest: provider_descriptor.command_digest.clone(), + workspace_digest: format!("sha256:{}", "a".repeat(64)), + requested_model: provider_descriptor.model.clone(), + effective_model: provider_descriptor.model.clone(), + permission_mode: Some(provider_descriptor.permission_mode), + mode: provider_descriptor.mode.clone(), + pi_thinking_level: provider_descriptor.pi_thinking_level, + claude_effort: provider_descriptor.claude_effort.clone(), + grok_effort: provider_descriptor.grok_effort.clone(), + grok_disable_web_search: provider_descriptor.grok_disable_web_search.clone(), + provider_lifetime_fence_candidates: [60_001, 60_002, 60_003], + }; + let guards: [(&str, fn(&mut AcpxDurableState)); 4] = [ + ("provider_exit_unconfirmed", |state| { + state.provider_exit_unconfirmed = true + }), + ("accepted_semantic_result", |state| { + state.semantic_result = Some(json!({"reportedWorkDisposition": "done"})) + }), + ("unanswered_runtime_request", |state| { + state + .pending_runtime_requests + .insert("input-1".to_owned(), pending_input_request("input-1")); + }), + ("active_goal", |state| { + state.goal_projection = json!({"goal": {"status": "active"}}) + }), + ]; + for (label, seed) in guards { + let mut state = + AcpxDurableState::new(provider_descriptor.clone(), tool_set.clone(), "d".into()); + state.lifecycle = "turn_active".to_owned(); + state.identity = Some(identity.clone()); + state.active_turn_id = Some("turn-1".to_owned()); + seed(&mut state); + let mut executor = + AcpxCommandExecutor::with_runner_config(&directory, &test_config(&directory, None)); + executor.state = Some(state.clone()); + assert!( + !executor + .restore_conversation_after_lost_turn(Some("turn-1".to_owned())) + .unwrap(), + "{label} must keep the close path" + ); + assert!(executor.session.is_none(), "{label} started a provider"); + assert_eq!(executor.state, Some(state), "{label} changed saved state"); + } + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn suspension_waits_for_the_original_provider_lifetime_quorum() { let (candidates, original_lifetime_fence) = reserve_provider_lifetime_fence(); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs index 2588c4d049c..6ac3fe4dbbe 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs @@ -421,6 +421,14 @@ fn semantic_result_event( } } +/// Local Codex app-server and OpenCode server restart their own process and +/// restore the same conversation after runner loss, so a turn that vanished +/// meanwhile is a restart interruption the controller may continue once. A +/// Cursor Cloud agent runs remotely; its lost turn is not a process restart. +fn restart_lost_turn_is_continuable(provider: &str) -> bool { + matches!(provider, "codex" | "opencode") +} + fn facade_provider_name(provider: &str) -> &'static str { match provider { "opencode" => "OpenCode", @@ -2693,6 +2701,7 @@ impl CodexCommandExecutor { // trustworthy success notification to replay. Terminate it // conservatively so the controller cannot wait forever or // mistake an unknown outcome for success. + let restart_continuable = restart_lost_turn_is_continuable(&provider_label); let provider_terminal = NormalizedProviderEvent { event_type: "turn.failed".to_owned(), priority: EventPriority::P0, @@ -2701,14 +2710,14 @@ impl CodexCommandExecutor { "providerTurnId": previous_active_turn_id, "status": "failed", "providerTerminalObserved": false, - "error": if provider_label == "codex" { json!({ + "error": if restart_continuable { json!({ "code": "provider_turn_lost_on_restore", "recoverable": true, "message": "The runner restored the conversation after process loss, but the previous turn is no longer active.", }) } else { Value::Null }, }), }; - if provider_label == "codex" && !state.active_terminal_tool_rejected { + if restart_continuable && !state.active_terminal_tool_rejected { // No provider result was observed. Preserve the lost // turn as a fact, but leave the run open for the // controller's bounded same-conversation recovery. @@ -5700,6 +5709,13 @@ impl CommandExecutor for CodexCommandExecutor { #[cfg(test)] mod tests { + #[test] + fn restart_lost_turns_continue_only_for_locally_restored_providers() { + assert!(super::restart_lost_turn_is_continuable("codex")); + assert!(super::restart_lost_turn_is_continuable("opencode")); + assert!(!super::restart_lost_turn_is_continuable("cursor_cloud")); + } + #[test] fn mcp_permission_requests_use_durable_turn_and_item_identity() { let identity = ProviderEventIdentity { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs index 608809e807e..168ba6e0a36 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_warm_attachment.rs @@ -166,3 +166,168 @@ fn checkpoints_the_sidecar_and_rebinds_consecutive_warm_runs_before_accepting_wo assert_eq!(started.result["status"], "accepted"); executor.shutdown().unwrap(); } + +fn unused_fence_ports() -> [u16; 3] { + // Lifetime fence candidates must be dynamic ports (49152 and above). An + // OS-assigned port can be lower, for example on Linux. + let mut listeners = Vec::new(); + for port in (49_152..=u16::MAX).rev() { + if let Ok(listener) = std::net::TcpListener::bind(("127.0.0.1", port)) { + listeners.push(listener); + if listeners.len() == 3 { + break; + } + } + } + let ports = listeners + .iter() + .map(|listener| listener.local_addr().unwrap().port()) + .collect::>(); + [ports[0], ports[1], ports[2]] +} + +/// A runner that dies during an active ACPX turn (for example, when the +/// control plane restarts) must restore the same conversation once the old +/// provider lifetime is proven gone. It records a recoverable restart-loss +/// cause instead of a terminal provider failure. +#[test] +fn restores_the_conversation_after_losing_an_active_turn_to_a_runner_restart() { + let fixture = + Fixture(std::env::temp_dir().join(format!("acpx-restart-{}", uuid::Uuid::new_v4()))); + fs::create_dir_all(&fixture.0).unwrap(); + fs::set_permissions(&fixture.0, fs::Permissions::from_mode(0o700)).unwrap(); + let sidecar = fixture.0.join("sidecar"); + fs::copy(env!("CARGO_BIN_EXE_fake-acpx-sidecar"), &sidecar).unwrap(); + fs::set_permissions(&sidecar, fs::Permissions::from_mode(0o700)).unwrap(); + let ports = unused_fence_ports(); + let args = vec![ + "--mode".to_owned(), + "suspend".to_owned(), + "--profile-digest".to_owned(), + PROFILE_DIGEST.to_owned(), + "--lifetime-fence-ports".to_owned(), + format!("{},{},{}", ports[0], ports[1], ports[2]), + ]; + let config = DurableRunnerConfig { + connect_url: "ws://127.0.0.1/runner".to_owned(), + ca_bundle_path: None, + state_dir: fixture.0.clone(), + runner_instance_id: "runner-1".to_owned(), + environment_lease_id: "lease-1".to_owned(), + run_id: "run-1".to_owned(), + normalized_session_id: "session-1".to_owned(), + turn_id: "turn-1".to_owned(), + item_id: "item-1".to_owned(), + runner_version: "0.0.0".to_owned(), + runner_digest: format!("sha256:{}", "a".repeat(64)), + acpx_launch_profile: Some(AcpxLaunchProfile { + candidate_admission: None, + authority_digest: format!("sha256:{}", "d".repeat(64)), + command: sidecar.clone(), + args: args.clone(), + artifacts: vec![QualifiedLaunchArtifact { + path: sidecar.clone(), + sha256: format!("sha256:{:x}", Sha256::digest(fs::read(&sidecar).unwrap())), + }], + }), + opencode_launch_profile: None, + cursor_cloud_launch_profile: None, + max_outbox_bytes: 1024 * 1024, + p0_reserve_bytes: 64 * 1024, + max_frame_bytes: 1024 * 1024, + reconnect_delay: Duration::from_millis(1), + reconnect_grace: None, + max_runtime: Duration::from_secs(30), + }; + let descriptor = json!({ + "kind": "acpx", "provider": "acpx", "driver": "acpx_runtime", + "providerVersion": "0.13.1", "agent": "codex", "model": "gpt-5.6-sol", + "acpxVersion": "0.13.1", "agentServerPackage": "@agentclientprotocol/codex-acp", + "agentServerVersion": "1.6.2", "agentRuntimePackage": "@openai/codex", + "agentRuntimeVersion": "0.160.0", "commandDigest": PROFILE_DIGEST, + "sidecarCommand": sidecar, "sidecarArgs": args, "runtimeDirectory": fixture.0, + "normalizedSessionId": "session-1", "runId": "run-1", "cwd": fixture.0, + "instructions": "Complete the supplied work.", + "permissionMode": "approve-all", "permissionModePinned": true, + }); + let mut original = AcpxCommandExecutor::with_runner_config(&fixture.0, &config); + original + .execute(&command(1, "run.prepare", json!({"provider": descriptor}))) + .unwrap(); + let opened = original + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + let provider_session = opened.result["providerSessionId"].clone(); + let started = original + .execute(&command( + 3, + "turn.start", + json!({"turnId": "provider-turn-1", "text": "Do the work."}), + )) + .unwrap(); + assert_eq!(started.result["status"], "accepted"); + // The runner process dies with the turn still active, after the turn sent + // the controller a semantic call that has no result yet. + drop(original); + let state_path = fixture.0.join("acpx-provider-state.json"); + let mut saved: Value = serde_json::from_slice(&fs::read(&state_path).unwrap()).unwrap(); + saved["openSemanticCalls"] = json!(["call-lost"]); + fs::write(&state_path, serde_json::to_vec(&saved).unwrap()).unwrap(); + + let mut restored = AcpxCommandExecutor::with_runner_config(&fixture.0, &config); + let snapshot = restored + .execute(&command(4, "session.snapshot", json!({}))) + .unwrap(); + assert_ne!(snapshot.result["status"], "closed"); + assert_eq!(snapshot.result["providerSessionId"], provider_session); + let events = restored.poll_events().unwrap(); + let failed = events + .iter() + .find(|event| event.event_type == "turn.failed") + .expect("the lost turn is terminated"); + assert_eq!(failed.payload["providerTurnId"], "provider-turn-1"); + assert_eq!(failed.payload["providerTerminalObserved"], false); + assert_eq!( + failed.payload["error"]["code"], + "provider_turn_lost_on_restore" + ); + assert_eq!(failed.payload["error"]["recoverable"], true); + // The run stays open for the controller's single same-conversation continuation. + assert!(!events + .iter() + .any(|event| event.event_type == "run.terminal")); + // The controller can still answer the lost call. That result settles the + // call instead of failing against the restored provider, which never saw it. + let lost_result = json!({ + "callId": "call-lost", "operationId": "register_deliverable", + "result": {"ok": true}, "isError": false, + }); + let settled = restored + .execute(&command(5, "semantic_tool.result", lost_result.clone())) + .unwrap(); + assert_eq!(settled.result["retired"], true); + // The receipt is durable and idempotent, so a delivery left indeterminate + // by a crash may replay to the same acknowledgement. + assert!(restored.can_reconcile_result_delivery().unwrap()); + let replayed = restored + .execute(&command(6, "semantic_tool.result", lost_result)) + .unwrap(); + assert_eq!(replayed.result["retired"], true); + // Calls the lost turn never opened still fail closed. + let unknown = json!({ + "callId": "call-unknown", "operationId": "register_deliverable", + "result": {"ok": true}, "isError": false, + }); + assert!(restored + .execute(&command(7, "semantic_tool.result", unknown)) + .is_err()); + let continued = restored + .execute(&command( + 8, + "turn.start", + json!({"turnId": "provider-turn-2", "text": "Continue."}), + )) + .unwrap(); + assert_eq!(continued.result["status"], "accepted"); + restored.shutdown().unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs index 8439dd9642f..e83b70871aa 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs @@ -7,7 +7,7 @@ use std::os::unix::fs::PermissionsExt; use paperclip_runner_core::durable::{ AcpxLaunchProfile, Command, CommandExecutor, CursorCloudLaunchProfile, DurableRunnerConfig, - OpenCodeLaunchProfile, QualifiedLaunchArtifact, PROVIDER_STATE_INCOMPATIBLE, + OpenCodeLaunchProfile, PolledEvent, QualifiedLaunchArtifact, PROVIDER_STATE_INCOMPATIBLE, }; use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; use paperclip_runner_core::provider_bridge::{authorized_tool_catalog_digest, AuthorizedTool}; @@ -579,6 +579,30 @@ fn acpx_runtime_resolution(mode: &str, request_id: &str) -> Value { }) } +/// A turn lost to a runner restart is recorded once as a recoverable loss. +fn assert_restart_lost_turn(events: &[PolledEvent]) { + let lost = events + .iter() + .filter(|event| event.event_type == "turn.failed") + .collect::>(); + assert_eq!(lost.len(), 1); + assert_eq!( + lost[0].payload["error"]["code"], + "provider_turn_lost_on_restore" + ); + assert_eq!(lost[0].payload["error"]["recoverable"], true); + assert!(!events + .iter() + .any(|event| event.event_type == "run.terminal")); +} + +/// Restarting the restored, idle session again replays no turn or request. +fn assert_no_replayed_turn_or_request(events: &[PolledEvent]) { + assert!(events + .iter() + .all(|event| event.event_type == "session.resumed")); +} + fn shutdown_recovered_acpx_fixture(executor: &mut NativeProviderCommandExecutor, directory: &Path) { // The fake sidecar advertises the same three lifetime-fence ports for all // fixtures. Parallel recovered shutdowns hold a quorum through state fsync, @@ -679,20 +703,23 @@ fn acpx_response_delivery_survives_crash_before_journaling_without_replaying_the .execute(&command(6, "request.resolve", resolve.payload)) .is_err()); assert_eq!(recovered.poll_events().unwrap(), events); + // The lost turn's conversation is restored for one continuation. Its + // outcome is a recoverable restart loss; the response is not replayed. assert_eq!( recovered .execute(&command(7, "session.snapshot", json!({}))) .unwrap() .result["status"], - "closed" + "session_open" ); + assert_restart_lost_turn(&events); // Polling is not an ACK; only the journal's ACK retires the event. recovered.acknowledge_events(events.len()).unwrap(); assert!(recovered.poll_events().unwrap().is_empty()); drop(recovered); let mut recovered_again = NativeProviderCommandExecutor::with_runner_config(&directory, &config); - assert!(recovered_again.poll_events().unwrap().is_empty()); + assert_no_replayed_turn_or_request(&recovered_again.poll_events().unwrap()); shutdown_recovered_acpx_fixture(&mut recovered_again, &directory); fs::remove_dir_all(directory).unwrap(); } @@ -864,14 +891,15 @@ fn acpx_failed_settlement_ack_replays_the_same_delivered_event_without_expiry() .execute(&command(6, "request.resolve", response)) .is_err()); assert_eq!(recovered.poll_events().unwrap(), events); + assert_restart_lost_turn(&events); recovered.acknowledge_events(events.len()).unwrap(); assert!(recovered.poll_events().unwrap().is_empty()); shutdown_recovered_acpx_fixture(&mut recovered, &directory); drop(recovered); let mut recovered_again = NativeProviderCommandExecutor::with_runner_config(&directory, &config); - assert!(recovered_again.poll_events().unwrap().is_empty()); - recovered_again.shutdown().unwrap(); + assert_no_replayed_turn_or_request(&recovered_again.poll_events().unwrap()); + shutdown_recovered_acpx_fixture(&mut recovered_again, &directory); fs::remove_dir_all(directory).unwrap(); } } @@ -2278,6 +2306,77 @@ fn rejects_a_mutable_opencode_command_outside_the_runner_launch_profile() { } #[test] +#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")] +fn restart_lost_opencode_turn_records_a_recoverable_cause() { + // A local OpenCode server restarts with runnerd, like Codex app-server: + // the lost turn keeps its recoverable cause and the run stays open for + // the controller's single same-conversation continuation. + let directory = temporary_directory("opencode-restart-lost-turn"); + let mut config = opencode_config(&directory); + let proxy_script = directory.join("qualified-opencode-proxy-script"); + #[cfg(unix)] + fs::set_permissions(&proxy_script, fs::Permissions::from_mode(0o700)).unwrap(); + fs::write( + &proxy_script, + format!( + "#!/bin/sh\nexec '{}' --state-file '{}' --call-log '{}' --require-completion-contract --durable-turn-ids --linger-after-turn-start\n", + env!("CARGO_BIN_EXE_fake-codex-app-server"), + directory.join("fake-opencode-state.json").display(), + directory.join("fake-opencode-calls.log").display(), + ), + ) + .unwrap(); + #[cfg(unix)] + fs::set_permissions(&proxy_script, fs::Permissions::from_mode(0o500)).unwrap(); + config + .opencode_launch_profile + .as_mut() + .unwrap() + .proxy_script = qualified_artifact(proxy_script); + let mut first = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + first + .execute(&command( + 1, + "run.prepare", + opencode_prepare_payload(&directory), + )) + .unwrap(); + first + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + first + .execute(&command(3, "turn.start", json!({"text": "Edit the file."}))) + .unwrap(); + drop(first); + let fake_state = directory.join("fake-opencode-state.json"); + let mut saved: Value = serde_json::from_slice(&fs::read(&fake_state).unwrap()).unwrap(); + saved["activeTurnId"] = Value::Null; + fs::write(&fake_state, serde_json::to_vec(&saved).unwrap()).unwrap(); + + let mut restored = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + restored + .execute(&command(4, "session.snapshot", json!({}))) + .unwrap(); + let events = restored.poll_events().unwrap(); + let lost = events + .iter() + .find(|event| event.event_type == "turn.failed") + .expect("the lost turn is terminated"); + assert_eq!( + lost.payload["error"]["code"], + "provider_turn_lost_on_restore" + ); + assert_eq!(lost.payload["error"]["recoverable"], true); + assert!(!events + .iter() + .any(|event| event.event_type == "run.terminal")); + assert_eq!(opencode_call_count(&directory, "turn/start"), 1); + restored.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + +#[test] +#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")] fn rejects_opencode_launch_profile_drift_across_fresh_recovery() { let directory = temporary_directory("opencode-profile-recovery"); let config = opencode_config(&directory); diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts index 676dabac7cb..db33af5ffd6 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts @@ -467,6 +467,30 @@ describe("qualified ACPX runtime sidecar", () => { } }); + it("keeps a restored provider's restarted call ids distinct from earlier calls in the run", async () => { + // A restored session starts a new provider process whose MCP request ids + // restart. The controller keys calls by run and call id, so a reused id + // would be mistaken for an earlier, already answered call. + const raw = { reportedWorkDisposition: "done", summary: "Complete", evidence: [], verification: [], + completionClaim: { contractRevision: "1", objectiveSatisfied: true, criteria: [], remainingWork: [] } }; + const tools = new Map(), emitted: any[] = []; + const handler = loadWaitForTool({ tools, emitted, validate: validatePrpStructuredRunResult, callIdNamespace: "rgeneration1" }); + const commands = loadActualToolCommands(tools); + const bridge = await startRunnerToolBridge({ handler, captureSemanticReceipt: () => () => {} }); + try { + const response = fetch(bridge.url, { method: "POST", headers: { Authorization: `Bearer ${bridge.secret}`, "Content-Type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/call", params: { name: "paperclip_finish", arguments: raw } }) }); + await vi.waitFor(() => expect(tools.has("rgeneration1:2")).toBe(true)); + expect(emitted).toEqual([expect.objectContaining({ callId: "rgeneration1:2", operationId: "paperclip_finish" })]); + await commands.resolve({ callId: "rgeneration1:2", turnId: "test-turn", result: { accepted: true }, error: null }); + const body = await (await response).json() as ToolCallResponseBody; + expect(body.result.isError).not.toBe(true); + } finally { + for (const pending of tools.values()) pending.cleanup(); + await bridge.close(); + } + }); + it.each([["monitor", true], ["monitor", false], ["same_agent", true], ["same_agent", false]] as const)("resolves an omitted continuation before the production sidecar validator with saved %s authority=%s", async (kind, eligible) => { const raw = { reportedWorkDisposition: "yielded", summary: "The review fix is published.", evidence: [], verification: [], completionClaim: { contractRevision: "1", objectiveSatisfied: false, criteria: [], remainingWork: [{ description: "Wait for checks.", blocksCompletion: true }] } }; @@ -1752,6 +1776,7 @@ function loadWaitForTool(input: { emitted: unknown[]; validate?: typeof validatePrpStructuredRunResult; emit?: (eventType: string, payload: unknown) => boolean; + callIdNamespace?: string | null; }): (call: RunnerToolCall) => Promise { const source = readFileSync( fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url)), @@ -1764,7 +1789,7 @@ function loadWaitForTool(input: { const factory = new Function( "boundedIdentity", "tools", "turnId", "emit", "PRP_COMPLETION_TOOL_NAME", "PRP_BLOCK_TOOL_NAME", "validatePrpStructuredRunResult", "boundedSidecarValue", "record", "MAX_PENDING_TOOLS", "stringifyAcpxSidecarFrame", - "resolveCompletionContinuation", "runId", "normalizingTools", "randomUUID", + "resolveCompletionContinuation", "runId", "normalizingTools", "randomUUID", "callIdNamespace", `return (${functionSource});`, ); return factory( @@ -1786,6 +1811,7 @@ function loadWaitForTool(input: { "test-run", new Set(), randomUUID, + input.callIdNamespace ?? null, ); } diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index 082a9e93ed3..a95d6a4f7a6 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -176,6 +176,12 @@ let openParams: AcpxSidecarOpenParams | null = null; let claudeCompactionCount = 0; let runId: string | null = null; let turnId: string | null = null; +/** + * A restored session runs in a new provider process whose MCP request ids + * restart. The controller keys semantic calls by run and call id, so prefix + * this process's ids to keep them distinct from calls earlier in the run. + */ +let callIdNamespace: string | null = null; let activeCopilotEvidence: CopilotToolEvidence | undefined; let sequence = 0; let requestSequence = 0; @@ -374,6 +380,9 @@ async function dispatch( ); host = openedHost; openParams = params; + callIdNamespace = params.expectedIdentity === undefined + ? null + : `r${randomUUID().replaceAll("-", "").slice(0, 12)}`; emit("runtime.process", { role: "sidecar", pid: process.pid, @@ -821,7 +830,10 @@ async function waitForTool(call: RunnerToolCall): Promise { if (!activeTurnId || !activeRunId || call.signal.aborted) { throw new Error("ACPX tool call is not bound to an active turn"); } - const callId = boundedIdentity(call.callId, "callId"); + const providerCallId = boundedIdentity(call.callId, "callId"); + const callId = callIdNamespace === null + ? providerCallId + : boundedIdentity(`${callIdNamespace}:${providerCallId}`, "callId"); if (tools.has(callId) || normalizingTools.has(callId)) throw new Error("ACPX tool call is duplicated"); const operationId = boundedIdentity(call.tool, "operationId"); if (tools.size + normalizingTools.size >= MAX_PENDING_TOOLS) { diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index aefb325d705..fdb0f7461b7 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -95,11 +95,15 @@ export function isNativeRestartInterruption(error: unknown): boolean { return failure.code === NATIVE_RESTART_INTERRUPTION_CODE && failure.recoverable === true; } +const RESTART_CONTINUATION_DRIVER_KINDS: ReadonlySet = new Set(["codex_app_server", "acpx_runtime", "opencode_server"]); + /** The terminal fingerprint survives a crash between interruption and continuation. */ export function nativeRestartInterruptedTurnId(snapshot: Pick): string | null { - if (snapshot.driverKind !== "codex_app_server" || snapshot.activeTurnId != null || + // Runnerd records this cause for Codex app-server and ACPX sessions after + // proving the old provider lifetime ended and restoring the conversation. + if (!RESTART_CONTINUATION_DRIVER_KINDS.has(snapshot.driverKind ?? "") || snapshot.activeTurnId != null || snapshot.semanticResult != null || snapshot.dispositionOnlyRecoveryConsumed || snapshot.goal != null || snapshot.pendingRuntimeRequests?.length) return null; const terminal = Array.isArray(snapshot.terminalTurns) ? snapshot.terminalTurns.at(-1) : null; diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index de63070094e..7cae1d4d807 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -20,6 +20,7 @@ import { NativeSessionCloseUnrecoverableError, NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, + nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { PROVIDER_STATE_INCOMPATIBLE } from "./contracts/harness-driver.js"; import type { @@ -508,6 +509,71 @@ describe("executeNativeSession recovery", () => { expect(checkpoints.some(snapshot => snapshot.terminalTurns?.some(turn => turn.turnId === "turn-recovery"))).toBe(true); }); + it.each(["acpx_runtime", "opencode_server"] as const)("continues a restart-lost %s turn once in the same conversation", async (driverKind) => { + const qualified = resolveQualifiedAcpxProfile("claude", "claude-sonnet-5"); + const acpxInput: NativeExecutionInputV1 = (driverKind === "acpx_runtime" ? { + ...input, + session: { ...input.session, driverKind }, + provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-5", permissionPolicy: "interactive", profile: { + driverKind: qualified.driverKind, protocolVersion: qualified.protocolVersion, acpxVersion: qualified.acpxVersion, + agent: qualified.agent, agentProfileVersion: qualified.agentProfileVersion, + agentServerPackage: qualified.agentServerPackage, agentServerVersion: qualified.agentServerVersion, + agentRuntimePackage: qualified.agentRuntimePackage, agentRuntimeVersion: qualified.agentRuntimeVersion, + commandDigest: qualified.commandDigest, + } }, + } : { + ...input, + session: { ...input.session, driverKind }, + provider: { kind: "opencode", model: "openrouter/deepseek/deepseek-v4-flash-0731" }, + }) as NativeExecutionInputV1; + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind, sessionId: "record-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-recovery", + terminalTurns: [], semanticResult: null, pendingRuntimeRequests: [], lineage: [], + }; + const interrupted: PersistedNativeSession = { + ...checkpoint, activeTurnId: null, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + }; + const startTurn = vi.fn(async () => ({ turnId: "turn-continuation" })); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + yield runnerEvent(1, "turn.failed", { status: "failed", error: failure }); + if (!startTurn.mock.calls.length) return; + yield { ...runnerEvent(2, "turn.completed", { status: "completed" }), turnId: "turn-continuation" }; + }, + startTurn, + async result() { return startTurn.mock.calls.length ? { result, terminal, turnId: "turn-continuation" } : null; }, + async snapshot() { return structuredClone(interrupted); }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { throw new Error("must preserve the provider session"); }, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event: PrpEvent) { return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + await expect(executeNativeSession({ + input: acpxInput, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + })).resolves.toMatchObject({ result, turnId: "turn-continuation" }); + expect(startTurn).toHaveBeenCalledOnce(); + const submitted = startTurn.mock.calls[0]![0] as Parameters[0]; + expect(submitted.continuation).toBe(true); + expect(submitted.message.text).toContain("Reconcile any unfinished tool or command"); + expect(nativeRestartInterruptedTurnId(interrupted)).toBe("turn-recovery"); + }); + it("never continues task work when a restart loses a compaction turn", async () => { const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; const checkpoint: PersistedNativeSession = { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 273f84ba33a..55fb486d92f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2448,7 +2448,9 @@ export async function executeNativeSession( const resumeInterruptedTurn = async (event: PrpEvent, signal: AbortSignal) => { if (restartContinuationStarted || !recovered || !options.resumeInterruptedTurn || // A compaction-only run never continues task work after a restart. - input.provider.kind !== "codex" || options.sessionCompaction || options.sessionGoalControl || options.resumeSessionGoalHeartbeat || + // Runnerd records a restart-lost turn only for these providers, after + // proving the old process ended and restoring the same conversation. + !["codex", "acpx", "opencode"].includes(input.provider.kind) || options.sessionCompaction || options.sessionGoalControl || options.resumeSessionGoalHeartbeat || recoveredSnapshot.goal || persistedSession?.semanticResult || persistedSession?.pendingRuntimeRequests?.length || event.eventType !== "turn.failed" || !isNativeRestartInterruption(event.payload.error) || event.turnId !== persistedSession?.activeTurnId) return false; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 509213ca403..5241fe2b5b5 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2220,10 +2220,15 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }, ); - it.each([2, 3])("recovers a checkpointed restart interruption within provider budget %s", async (attempt) => { - const fixture = await seedRun(`INTERRUPTED-${attempt}`); + it.each([ + { attempt: 2, driverKind: "codex_app_server" }, + { attempt: 3, driverKind: "codex_app_server" }, + { attempt: 2, driverKind: "acpx_runtime" }, + { attempt: 2, driverKind: "opencode_server" }, + ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind }) => { + const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}`); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { - driverKind: "codex_app_server", + driverKind, sessionId: fixture.native.normalizedSessionId, identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, providerSessionId: "interrupted-provider",