From e015c2ef21ef068baf9a1a510cdf711b91e549d1 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 10:15:25 -0600 Subject: [PATCH] fix(runner): keep a Claude session resumable when its turn-end provider restart fails (#20285) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path - #20145 prepares the next Claude provider generation at park, and #20190 restarts the ACPX provider when a turn settles. They move the sidecar respawn off the follow-up's critical path. - When that turn-end restart fails, the runner "fails closed": `restart_session_after_turn` saves the session `closed` and drops it. Within the run, the next turn fails exactly like a failed restart at turn start. That's intended. - Across runs it breaks the conversation: - the next run's `run.attach` refuses a `closed` session ("run.attach requires the same settled ACPX provider profile and session"); - it reconnects for about 10 s; - the controller then records a continuity break and starts a fresh Claude conversation (`run.startup.reuse provider_session: fresh/created`). - A local A/B (`ttft-baseline --adapters claude --samples 3 --phases coldwarm`, on a host with load average about 17–27) confirmed it: | | d40a7494 (before #20145/#20190) | a3a25ce6 (after) | |---|---|---| | Follow-ups that resumed the conversation | 3 of 3 (`verified_reuse`, resume 4.4–8.2 s) | 1 of 3 | | Follow-ups that started a fresh conversation | 0 of 3 | 2 of 3 (bootstrap 16–17 s) | The kept instance shows each lost follow-up's runner received provider state `lifecycle: "closed"`, with `run.attach` refused. - A failed park preparation already recovers safely: it waits for the reaped provider's lifetime fence, then saves `suspended`. The turn-end restart should do the same. ## Linked Issues or Issue Description Refs #20190, #20145 **What happened?** When the post-turn ACPX provider restart failed, the next Claude run could not attach. It spent about 10 s and then started a new conversation, losing the task's conversation context. **Steps to reproduce** 1. Run a Claude task on a busy host so the turn-end provider restart fails (the fixture `reject-open` fault reproduces it). 2. Post a follow-up comment (a new run). 3. The follow-up's `run.attach` is refused, then `run.startup.reuse` reports `provider_session: fresh/created`. **Expected behavior** The follow-up resumes the same Claude conversation. Within the failing run, later turns still fail closed. ## What Changed - `acpx_park_preparation.rs`: - A failed `restart_session_after_turn` now calls the shared `recover_failed_generation_restart`, renamed from `recover_failed_park_preparation`. - That recovery drops the session, marks it `prepared` with the provider exit unconfirmed, waits for the provider-lifetime fence, then saves `suspended`. - If the fence or the save fails, the state stays exit-unconfirmed, which is the safe direction. - `acpx_provider_backend.rs`: - Later turns of the failing run still fail closed: the first with "ACPX turn start failed closed: ", then "ACPX session is not open". That's the same as a failed restart at turn start. - The restriction is saved in the ACPX state as `turnEndRestartFailedRun` (serialized only while set), so it outlasts a Runner process restart inside the same run. Same-run restore doesn't reopen the session while it's set. - Only a `run.attach` from a different run clears it; a same-run attach keeps it. State validation requires it to match the descriptor's run. - The cause is handed to the durable Runner once (`take_turn_end_restart_failure`), which records it as a redacted `turn-end provider restart failed: ` diagnostic in `runner-state.json`. That gives the under-load cause a first recorded sample. - Tests (`tests/support/acpx_turn_end_restart.rs`): - New: `the_next_run_resumes_the_conversation_after_a_failed_turn_end_restart`, red on master (the state is `closed`). It covers the same Runner process with a warm attach, and a new Runner process with a cold attach. The conversation id is preserved, and the next run completes a turn. - New: `a_runner_restart_in_the_same_run_keeps_the_failed_turn_end_restart_closed`, red without the saved field. A new Runner process with the same run id fails `turn.start` without spawning a provider; then a new run cold-attaches and resumes the same conversation. - The existing `a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restart` now also asserts that the turn-end failure saves `suspended` (the turn-start failure stays `closed`). Everything else about the run's failing turns is unchanged. ## Verification - `cargo test -p paperclip-runner-core`: - `--test acpx_park_preparation`: 21 passed. - `--lib acpx`: 93 passed. - `acpx_park_readiness`: 8 passed. - `acpx_warm_attachment`: 4 passed. - `acpx_provider_suspend`: 5 passed. - `acpx_provider_session`: 12 passed. - `acpx_provider_turns`: 29 passed. - `native_provider_backend`: 47 passed. - `durable_recovery`: 3 passed. - Live A/B on this branch (`ttft-baseline --adapters claude --samples 3 --phases coldwarm`): follow-ups resumed 3/3 (1/3 before), follow-up first activity p50 6.2 s (about 20–23 s before). - `cargo clippy --all-targets`: no findings on changed lines. ## Risks - **Late-callback guarantee:** unchanged. The failed restart already reaped the old generation, and the next generation starts only via `run.attach` or a later reopen. That's the same resume path a failed park preparation uses. - **Fence wait:** the failure path now waits for the reaped provider's lifetime fence before saving, matching park recovery. If the fence never clears, the state stays exit-unconfirmed and fails closed. - **Rollback:** `AcpxDurableState` denies unknown fields. The new field is written only between a failed turn-end restart and the next run's attach. A rollback inside that window makes the older Runner refuse the saved state, which still fails closed, as master's `closed` state does today. - **Scope:** the cause of restart failures under load isn't changed here. This makes their failure mode safe. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`, Anthropic), 1M context window, tool use and code execution, via Claude Code. ## 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_park_preparation.rs | 32 ++-- .../runner-core/src/acpx_provider_backend.rs | 75 +++++++-- .../crates/runner-core/src/durable/runner.rs | 10 ++ .../src/native_provider_backend.rs | 7 + .../tests/support/acpx_turn_end_restart.rs | 157 +++++++++++++++++- 5 files changed, 254 insertions(+), 27 deletions(-) diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs index 26d4afb8638..a68e5fce2fa 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs @@ -43,15 +43,18 @@ impl AcpxCommandExecutor { let Err(error) = restarted else { return Ok(()); }; - let state = self - .state - .as_mut() - .expect("ACPX state remains available after a turn-end restart"); - state.lifecycle = "closed".to_owned(); - self.context.provider_turn_id = None; - self.session = None; + // Later turns of this run fail closed, exactly like a failed restart at + // turn start. The saved session stays resumable: the reaped provider is + // proven gone, then it is saved suspended, so the next run resumes the + // same conversation instead of starting a new one. self.failed_turn_end_restart = Some(error.to_string()); - self.save_state() + self.unreported_turn_end_restart_failure = Some(error.to_string()); + let run_id = self.context.run_id.clone(); + self.state + .as_mut() + .expect("ACPX state remains available after a failed turn-end restart") + .turn_end_restart_failed_run = Some(run_id); + self.recover_failed_generation_restart() } pub(super) fn prepare_session_for_park( @@ -70,7 +73,7 @@ impl AcpxCommandExecutor { match prepared { Ok(restarted) => Ok(restarted), Err(error) => { - self.recover_failed_park_preparation()?; + self.recover_failed_generation_restart()?; Err(DurableRunnerError::invalid(format!( "ACPX park preparation failed: {error}" ))) @@ -78,20 +81,23 @@ impl AcpxCommandExecutor { } } - fn recover_failed_park_preparation(&mut self) -> Result<(), DurableRunnerError> { + /// A provider generation restart (at park or after a turn) failed: wait until + /// the reaped provider loses its lifetime fence, then save the session + /// suspended so a later run can resume it. + fn recover_failed_generation_restart(&mut self) -> Result<(), DurableRunnerError> { self.session = None; self.context.provider_turn_id = None; let candidates = { let state = self .state .as_mut() - .expect("ACPX state remains available after park preparation"); + .expect("ACPX state remains available after a failed generation restart"); state.lifecycle = "prepared".to_owned(); state.provider_exit_unconfirmed = true; state .identity .as_ref() - .expect("park preparation requires a provider identity") + .expect("a generation restart requires a provider identity") .provider_lifetime_fence_candidates }; self.save_state()?; @@ -99,7 +105,7 @@ impl AcpxCommandExecutor { let state = self .state .as_mut() - .expect("ACPX state remains available after park preparation"); + .expect("ACPX state remains available after a failed generation restart"); state.provider_exit_unconfirmed = false; state.lifecycle = "suspended".to_owned(); if let Err(error) = self.save_state() { 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 94c33a1ee44..be646761028 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 @@ -627,6 +627,11 @@ struct AcpxDurableState { /// never sees their results; the controller may still deliver them. #[serde(default)] retired_semantic_calls: BTreeSet, + /// The run whose turn-end provider restart failed. Its later turns fail + /// closed, even after a Runner process restart; the next run's attach + /// clears it and resumes the saved session. Written only while set. + #[serde(default, skip_serializing_if = "Option::is_none")] + turn_end_restart_failed_run: Option, } /// The saved launch names the same executable and sidecar files as this @@ -677,6 +682,7 @@ impl AcpxDurableState { next_event_sequence: initial_event_sequence(), open_semantic_calls: BTreeSet::new(), retired_semantic_calls: BTreeSet::new(), + turn_end_restart_failed_run: None, } } @@ -746,6 +752,10 @@ impl AcpxDurableState { .is_some_and(|result| !result.is_object()) || (self.identity.is_none() && !matches!(self.lifecycle.as_str(), "prepared" | "closed")) + || self + .turn_end_restart_failed_run + .as_ref() + .is_some_and(|run_id| *run_id != self.descriptor.run_id) { return Err(DurableRunnerError::invalid( "ACPX durable provider state is malformed or inconsistent", @@ -966,7 +976,10 @@ pub struct AcpxCommandExecutor { fail_after_state_rename: bool, launch_profile: Option, recovery_pending_effects: Option, + /// The failed turn-end restart's cause, reported by the next turn start. failed_turn_end_restart: Option, + /// The failed turn-end restart's cause, until the Runner records it. + unreported_turn_end_restart_failure: Option, } // The durable command's turn and the live provider callback have separate IDs. @@ -1007,6 +1020,7 @@ impl AcpxCommandExecutor { restore_checked: false, recovery_pending_effects: None, failed_turn_end_restart: None, + unreported_turn_end_restart_failure: None, restore_error: None, persistence_error: None, #[cfg(test)] @@ -1216,6 +1230,10 @@ impl AcpxCommandExecutor { if state.descriptor.run_id != self.context.run_id { return Ok(()); } + // This run's turn-end provider restart failed; its turns fail closed. + if state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str()) { + return Ok(()); + } if !matches!( state.lifecycle.as_str(), "session_open" | "turn_starting" | "turn_active" | "suspended" @@ -1738,12 +1756,26 @@ impl AcpxCommandExecutor { state.semantic_result = None; state.pending_events.clear(); state.lifecycle = "suspended".to_owned(); - if cold_candidate_attachment { + // A new run starts a fresh provider generation; the previous run's + // failed turn-end restart no longer applies to it. A same-run attach + // keeps the restriction. + let clears_turn_end_restart_failure = state + .turn_end_restart_failed_run + .as_ref() + .is_some_and(|run_id| *run_id != attachment_context.run_id); + if clears_turn_end_restart_failure { + state.turn_end_restart_failed_run = None; + } + let saved = if cold_candidate_attachment { self.commit_staged_state(state) } else { self.state = Some(state); self.save_state() + }; + if saved.is_ok() && clears_turn_end_restart_failure { + self.failed_turn_end_restart = None; } + saved } fn open_session(&mut self) -> Result { @@ -1756,6 +1788,17 @@ impl AcpxCommandExecutor { "ACPX provider lifetime cleanup is not yet proven", )); } + // No path (session.open, a same-run run.attach) reopens a session whose + // run failed closed at its turn-end provider restart. + if self.session.is_none() + && self.state.as_ref().is_some_and(|state| { + state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str()) + }) + { + return Err(DurableRunnerError::invalid( + "ACPX session failed closed after this run's turn-end provider restart failed", + )); + } if self.session.is_none() { let recovering = self .state @@ -1929,17 +1972,21 @@ impl AcpxCommandExecutor { fn start_turn(&mut self, payload: &Value) -> Result { turn_start_request(payload)?; - if let Some(error) = self.failed_turn_end_restart.take() { - if self.session.is_none() - && self - .state - .as_ref() - .is_some_and(|state| state.lifecycle == "closed") - { - return Err(DurableRunnerError::invalid(format!( - "ACPX turn start failed closed: {error}" - ))); - } + // A failed turn-end restart fails every later turn of this run closed, + // like a failed restart at turn start: the first names the cause, later + // ones find no open session. The restriction is saved, so it outlasts a + // Runner process restart; only the next run's attach clears it. + if self.session.is_none() + && self.state.as_ref().is_some_and(|state| { + state.turn_end_restart_failed_run.as_deref() == Some(self.context.run_id.as_str()) + }) + { + return Err(match self.failed_turn_end_restart.take() { + Some(error) => { + DurableRunnerError::invalid(format!("ACPX turn start failed closed: {error}")) + } + None => DurableRunnerError::invalid("ACPX session is not open"), + }); } let reopenable = self.session.is_none() && self.state.as_ref().is_some_and(|state| { @@ -2927,6 +2974,10 @@ impl CommandExecutor for AcpxCommandExecutor { self.restart_session_after_turn() } + fn take_turn_end_restart_failure(&mut self) -> Option { + self.unreported_turn_end_restart_failure.take() + } + fn execute(&mut self, command: &Command) -> Result { self.restore()?; if self 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 d786da4a324..2386f4218d7 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 @@ -327,6 +327,12 @@ pub trait CommandExecutor { Ok(()) } + /// The cause of a provider restart after a turn that failed and was + /// recovered, once; recorded as a runner diagnostic. + fn take_turn_end_restart_failure(&mut self) -> Option { + None + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -2175,6 +2181,10 @@ fn poll_executor_events_after_controller_ack( poll_executor_events(state, store, config, executor)?; if state.acked_source_seq >= state.highest_source_seq() { executor.prepare_next_turn()?; + if let Some(failure) = executor.take_turn_end_restart_failure() { + state.record_diagnostic(format!("turn-end provider restart failed: {failure}")); + store.save(state)?; + } } Ok(()) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs index 13551eb72b9..d906e5ffa03 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs @@ -351,6 +351,13 @@ impl CommandExecutor for NativeProviderCommandExecutor { } } + fn take_turn_end_restart_failure(&mut self) -> Option { + match self.selected.as_mut() { + Some(SelectedExecutor::Acpx(executor)) => executor.take_turn_end_restart_failure(), + _ => None, + } + } + fn observe_recovery_state( &mut self, state: &crate::durable::DurableState, diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs index 181dc19fd06..55a7a11f534 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs @@ -237,14 +237,22 @@ fn a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restar starts, ) }; - let at_turn_end = outcome(true); + let mut at_turn_end = outcome(true); assert!( at_turn_end.0.starts_with("ACPX turn start failed closed:") && at_turn_end.0.contains("session.open"), "{}", at_turn_end.0 ); - assert_eq!(at_turn_end, outcome(false)); + // The turns of this run fail the same way. Unlike a failed restart at turn + // start, the failed turn-end restart saves a resumable session, so the next + // run resumes the conversation (see the test below). + assert_eq!(at_turn_end.2, "suspended"); + let mut at_turn_start = outcome(false); + assert_eq!(at_turn_start.2, "closed"); + at_turn_end.2 = Value::Null; + at_turn_start.2 = Value::Null; + assert_eq!(at_turn_end, at_turn_start); } #[test] @@ -331,3 +339,148 @@ fn an_active_goal_or_a_request_at_the_terminal_keeps_today_s_restart_timing() { assert_eq!(at_turn_end.3, at_turn_start.3); eprintln!("request at terminal: {at_turn_end:?}"); } + +/// A turn-end restart that fails (for example a slow or loaded host) used to +/// save the session `closed`. The next run's `run.attach` then refused it and +/// the controller started a new conversation. The failed restart now saves a +/// resumable session once the reaped provider is proven gone: later turns of +/// this run still fail closed, and the next run resumes the same conversation, +/// in this Runner process or in a new one. +#[test] +fn the_next_run_resumes_the_conversation_after_a_failed_turn_end_restart() { + for new_runner_process in [false, true] { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + let conversation = harness.opened["providerSessionId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + harness.executor.prepare_next_turn().unwrap(); + assert!(!alive(&old_pid)); + // The Runner records the cause once as a diagnostic. + let reported = harness.executor.take_turn_end_restart_failure(); + assert!( + reported + .as_deref() + .is_some_and(|cause| cause.contains("session.open")), + "{reported:?}" + ); + assert_eq!(harness.executor.take_turn_end_restart_failure(), None); + let saved = harness.fixture.state(); + assert_eq!(saved["lifecycle"], "suspended", "{saved}"); + assert_eq!(saved["providerExitUnconfirmed"], false); + let failed = harness + .executor + .execute(&command( + 90, + "turn.start", + json!({"turnId": "second", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + assert!( + failed.starts_with("ACPX turn start failed closed:"), + "{failed}" + ); + harness.sequence = 90; + harness.fixture.fault(""); + let attached = if new_runner_process { + // A new Runner process for the next run cold-attaches the saved session. + harness.executor.shutdown().unwrap(); + harness.config.run_id = "run-2".to_owned(); + harness.executor = NativeProviderCommandExecutor::with_runner_config( + &harness.fixture.0, + &harness.config, + ); + let payload = harness.attach_payload("run-2", |payload| { + payload + .as_object_mut() + .unwrap() + .remove("paperclipNextAuthority"); + }); + harness.execute("run.attach", payload) + } else { + harness.attach("run-2", |_| {}).0 + }; + assert_eq!( + attached.result["providerSessionId"], conversation, + "new_runner_process={new_runner_process}" + ); + harness.complete_turn("next-run"); + assert_eq!(harness.fixture.state()["descriptor"]["runId"], "run-2"); + harness.executor.shutdown().unwrap(); + } +} + +#[test] +fn a_runner_restart_in_the_same_run_keeps_the_failed_turn_end_restart_closed() { + let mut harness = Harness::new(); + let conversation = harness.opened["providerSessionId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + harness.executor.prepare_next_turn().unwrap(); + harness.fixture.fault(""); + let saved = harness.fixture.state(); + assert_eq!( + saved["turnEndRestartFailedRun"], harness.config.run_id, + "{saved}" + ); + // The Runner process restarts inside the same run: its turns still fail + // closed instead of reopening the saved session. + harness.executor.shutdown().unwrap(); + harness.executor = + NativeProviderCommandExecutor::with_runner_config(&harness.fixture.0, &harness.config); + let starts = harness.fixture.process_starts().len(); + let failed = harness + .executor + .execute(&command( + 90, + "turn.start", + json!({"turnId": "second", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + assert_eq!(failed, "ACPX session is not open"); + assert_eq!(harness.fixture.process_starts().len(), starts); + // Neither session.open nor a same-run attach reopens it. + let reopened = harness + .executor + .execute(&command(91, "session.open", json!({}))) + .unwrap_err() + .to_string(); + assert!(reopened.contains("failed closed"), "{reopened}"); + let same_run_attach = json!({"provider": harness.descriptor.clone()}); + let attach_error = harness + .executor + .execute(&command(92, "run.attach", same_run_attach)) + .unwrap_err() + .to_string(); + assert!(attach_error.contains("failed closed"), "{attach_error}"); + assert_eq!(harness.fixture.process_starts().len(), starts); + assert_eq!( + harness.fixture.state()["turnEndRestartFailedRun"], + harness.config.run_id + ); + assert_eq!(live_providers(&harness), Vec::::new()); + assert_eq!(harness.fixture.state()["lifecycle"], "suspended"); + // The next run still resumes the conversation and clears the restriction. + harness.executor.shutdown().unwrap(); + harness.sequence = 92; + harness.config.run_id = "run-2".to_owned(); + harness.executor = + NativeProviderCommandExecutor::with_runner_config(&harness.fixture.0, &harness.config); + let payload = harness.attach_payload("run-2", |payload| { + payload + .as_object_mut() + .unwrap() + .remove("paperclipNextAuthority"); + }); + let attached = harness.execute("run.attach", payload); + assert_eq!(attached.result["providerSessionId"], conversation); + assert!(harness + .fixture + .state() + .get("turnEndRestartFailedRun") + .is_none()); + harness.complete_turn("next-run"); + harness.executor.shutdown().unwrap(); +}