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 746e0fbc05b..a6a13bda61d 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 @@ -122,6 +122,24 @@ fn acquire_provider_lifetime_fence( }) } +fn turn_start_request(payload: &Value) -> Result<(&str, String), DurableRunnerError> { + let text = payload + .get("text") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.text is required"))?; + let provider_turn_id = payload + .get("turnId") + .and_then(Value::as_str) + .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.turnId is required"))? + .to_owned(); + if !is_stable_id(&provider_turn_id, DURABLE_STABLE_ID_CHARS) { + return Err(DurableRunnerError::invalid( + "turn.start payload.turnId is invalid", + )); + } + Ok((text, provider_turn_id)) +} + fn await_provider_lifetime_fence( candidates: [u16; 3], ) -> Result, DurableRunnerError> { @@ -1838,20 +1856,63 @@ impl AcpxCommandExecutor { } fn start_turn(&mut self, payload: &Value) -> Result { - let text = payload - .get("text") - .and_then(Value::as_str) - .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.text is required"))?; - let requested_provider_turn_id = payload - .get("turnId") - .and_then(Value::as_str) - .ok_or_else(|| DurableRunnerError::invalid("turn.start payload.turnId is required"))?; - let provider_turn_id = requested_provider_turn_id.to_owned(); - if !is_stable_id(&provider_turn_id, DURABLE_STABLE_ID_CHARS) { - return Err(DurableRunnerError::invalid( - "turn.start payload.turnId is invalid", - )); + turn_start_request(payload)?; + let reopenable = self.session.is_none() + && self.state.as_ref().is_some_and(|state| { + state.identity.is_some() + && !state.provider_exit_unconfirmed + && state.descriptor.run_id == self.context.run_id + && matches!( + state.lifecycle.as_str(), + "prepared" | "suspended" | "session_open" + ) + }); + let reopened = if reopenable { + Some(self.open_session()?) + } else { + None + }; + match self.start_turn_in_open_session(payload) { + Ok(mut execution) => { + if let Some(opened) = reopened { + let mut events = opened.events; + events.append(&mut execution.events); + execution.events = events; + } + Ok(execution) + } + Err(error) => { + if let Some(opened) = reopened { + self.retain_reopened_session_events(opened.events)?; + } + Err(error) + } } + } + + fn retain_reopened_session_events( + &mut self, + events: Vec<(String, EventPriority, Value)>, + ) -> Result<(), DurableRunnerError> { + let state = self + .state + .as_mut() + .ok_or_else(|| DurableRunnerError::invalid("ACPX provider is not prepared"))?; + for (event_type, priority, payload) in events { + state.push(NormalizedProviderEvent { + event_type, + priority, + payload, + })?; + } + self.save_state() + } + + fn start_turn_in_open_session( + &mut self, + payload: &Value, + ) -> Result { + let (text, provider_turn_id) = turn_start_request(payload)?; // Retain the old sidecar's read tail before its MCP receipt epoch changes. if self .session 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 168ba6e0a36..e502f1cedab 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 @@ -190,6 +190,123 @@ fn unused_fence_ports() -> [u16; 3] { /// 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 reopens_the_conversation_for_a_turn_after_a_suspension_stop() { + let fixture = + Fixture(std::env::temp_dir().join(format!("acpx-repeat-restore-{}", 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(); + original + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + original + .execute(&command( + 3, + "turn.start", + json!({"turnId": "provider-turn-0", "text": "Do the work."}), + )) + .unwrap(); + original + .execute(&command(4, "turn.stop", json!({}))) + .unwrap(); + assert!(original + .execute(&command( + 5, + "turn.start", + json!({"turnId": "provider-turn-1"}) + )) + .is_err()); + assert!(!original + .poll_events() + .unwrap() + .iter() + .any(|event| event.event_type == "session.resumed")); + let started = original + .execute(&command( + 5, + "turn.start", + json!({"turnId": "provider-turn-1", "text": "Continue."}), + )) + .unwrap(); + assert_eq!(started.result["status"], "accepted"); + assert!(started + .events + .iter() + .any(|event| event.0 == "session.resumed")); + original + .execute(&command(6, "turn.stop", json!({}))) + .unwrap(); + original + .execute(&command(7, "session.close", json!({}))) + .unwrap(); + assert!(original + .execute(&command( + 8, + "turn.start", + json!({"turnId": "provider-turn-2", "text": "Continue."}) + )) + .is_err()); + original.shutdown().unwrap(); +} + #[test] fn restores_the_conversation_after_losing_an_active_turn_to_a_runner_restart() { let fixture =