From a8bbdf237c2dc1e99e5652e0af34e2e146b08a71 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:46:11 -0600 Subject: [PATCH] fix(runner): reopen the ACPX session at turn.start when it can be safely reopened (#20110) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native Claude ACPX run must survive restarts and suspensions. A turn stopped for suspension drops the live ACPX session and leaves the saved lifecycle `prepared` > - The next `turn.start` checked for a live session before anything else, so it failed with "ACPX session is not open". That happened even though the saved session was safe to reopen > - In restart chaos this made a Claude approval task (G1 t03) need a Retry after repeated restores > - This pull request makes `turn.start` reopen a reopenable session before it starts the turn > - The benefit is that a suspended or restored Claude run keeps going on its own, with no Retry ## Linked Issues or Issue Description Refs #20089 **What happened?** In a local restart-chaos G1 (approval) wave, Claude ACPX task t03 failed after repeated restores with `ACPX session is not open` and needed a Retry. The Retry overwrote the session directory, so the exact sequence could not be read back. The same error reproduces directly with `turn.stop` (suspension) followed by `turn.start` on the same session: the stop clears the live session and sets the lifecycle to `prepared`, and `turn.start` never reopens it. **Expected behavior** When the saved session can be reopened, `turn.start` reopens it, the same way `session.open` does, and then starts the turn. "Can be reopened" means: the conversation identity is recorded, the provider exit is confirmed, the session belongs to this run, and the lifecycle is `prepared`, `suspended` or `session_open`. A session that was closed on purpose, or whose provider exit is unconfirmed, still fails. **Steps to reproduce** 1. `run.prepare`, `session.open`, then `turn.start`. 2. `turn.stop` (suspension stop). 3. `turn.start` again. Before this change, it fails with "ACPX session is not open". **Paperclip version or commit** master at f789f3e2af. Branch rebased on current master. ## What Changed - `acpx_provider_backend.rs`: `turn.start` checks whether there is no live session and the saved state is reopenable. If so, it reopens through `open_session` first, and the reopen's `session.resumed` event comes before the turn's events. The previous body moved unchanged into `start_turn_in_open_session`. - `tests/acpx_warm_attachment.rs`: a new test, `reopens_the_conversation_for_a_turn_after_a_suspension_stop`. It drives prepare, open, turn, stop, turn and expects the second turn to be accepted with `session.resumed`. It also checks that after `session.close`, `turn.start` still fails. ## Verification - Red/green: the new test fails on master with "ACPX session is not open" and passes with the change. `cargo test -p runner-core --test acpx_warm_attachment`: 3/3 passed. - I also ran the full `runner-core` crate suite. Five tests failed: four `task_memory` monitor tests and `unconfirmed_suspension_state...`. Each of them passes on an unchanged master checkout, and the ACPX one also passes when rerun in this branch, so they are timing-dependent and not caused by this change. ## Risks - Low risk. A reopen happens only under the same conditions in which `session.open` would already succeed for this run. A closed session and an unconfirmed provider exit stay fail-closed. The order of events changes only by adding `session.resumed` ahead of the turn's events. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test execution). ## 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 --- .../runner-core/src/acpx_provider_backend.rs | 87 +++++++++++-- .../runner-core/tests/acpx_warm_attachment.rs | 117 ++++++++++++++++++ 2 files changed, 191 insertions(+), 13 deletions(-) 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 =