From 92a18bb0f3ecd81dbd13599ed0231926e35ba500 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sun, 8 Nov 2026 22:34:26 -0800 Subject: [PATCH] perf(runner): restart the ACPX provider when a turn settles (#20190) 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. > - Paperclip Runner drives Claude through the ACPX sidecar. The sidecar's MCP endpoint lasts for the whole session, so it cannot tell which turn sent a late tool callback. > - Because of this, the Runner reaps the sidecar and provider of a generation that ran a turn before it releases call-ID tombstones and rotates receipt epochs. Today this restart runs at the start of the next turn. > - Every follow-up turn in a run pays that restart on its critical path: about 1.5 s with real Claude. #20145 moves the restart to park, but a second turn in the same run, after the first turn that follows an attach, still restarts cold. > - The restart does not have to wait for the next turn. When a turn settles and the host has its terminal result, the Runner is idle. > - This pull request restarts the provider at that point. The next turn, or the next park, then finds a fresh generation. > - The benefit: same-run follow-up turns start without a restart, park does no restart after a settled turn, and the late-callback guarantee is unchanged. ## Linked Issues or Issue Description Refs #20145, which moved the restart to park. This pull request builds on it. No public issue exists, so the problem is described here. **Subsystem affected** Paperclip Runner, ACPX provider (Claude). **Current behavior** `AcpxProviderSession::start_turn_timed` restarts the provider before any turn whose generation already ran a turn. A follow-up turn in the same run waits for `session.suspend`, sidecar shutdown, a new sidecar, `initialize`, `session.open` and `run.attach` before `turn.start`. Since #20145, park does the restart instead, but only for the first turn after an attach. **Proposed behavior** When a turn settles and the controller has ACKed every committed event, the Runner restarts the provider. The next `turn.start` uses the fresh generation and only rotates receipts. Park reuses the fresh generation and does no second restart. **Reason and benefit** The restart leaves the turn's critical path, both for same-run follow-ups and for the turns after a park. The order of reap and rotation stays the same, so a late callback from the reaped generation still cannot reach a later turn. ## What Changed - **Runner loop** (`durable/runner.rs`): new `CommandExecutor::prepare_next_turn` hook, a no-op by default. The runner calls it after the idle provider poll, only when no control frame is waiting and the controller has ACKed every committed event (`acked_source_seq >= highest_source_seq`). So the restart always runs after the terminal result reaches the host. - **ACPX turn-end restart** (`acpx_park_preparation.rs`, `acpx_provider_generation.rs`): - `restart_session_after_turn` restarts a Claude session that has a completed turn, nothing pending, no buffered events, a dormant goal and its whole event tail ACKed. It first settles an abandoned park observation. If that settlement reads more events, it skips this round, and the next idle round restarts after those events are ACKed. - `AcpxProviderSession::restart_after_turn` does the same suspend → reap → start → bootstrap → identity check as the turn-start restart. It records the reaped process id in `restarted_after_turn_from`. This field is the explicit "fresh because of a turn-end restart" marker. - The session does not poll a fresh generation until its next turn or rebind, the same as a prepared one. Its startup output is published with the next turn, as today. Polling it early would add events after the host computed its park cut. - **Next turn** (`acpx_provider_session.rs`): `start_turn_timed` restarts only when the generation ran a turn. When `restarted_after_turn_from` is set, it does the post-restart receipt and ledger rotation without a second restart. `session.reconciled` reports the reaped process as `previousProcessId`. - **Park**: `prepare_next_generation` restarts (with the 20 s park deadline) only if no turn-end restart ran. Otherwise it marks the generation for `run.attach` rebind and returns `false`, so the runner does not ask for a second park proof. The park-time receipt rotation moves to the next turn start, where the turn-end path does it too. - **Restart failure**: the executor sets the same state that a failed turn-start restart leaves (`closed`, no session, provider exit confirmed). The next `turn.start` fails with the same `ACPX turn start failed closed: …` error that the turn-start restart gives today. - **Fake sidecar**: new `pending-input` fault, which opens an input request in a turn that then completes. ### Races | Race | Handling | | --- | --- | | A turn or steer arrives during the turn-end restart | The runner loop is single-threaded. The command waits in the socket until the restart returns, then starts on the fresh generation without a restart. Only one generation is live. | | A turn arrives before the hook runs (for example, in the same read as the ACK) | Control frames go first, so the turn restarts at its start, as today. The hook then restarts after that turn. | | Park arrives during the turn-end restart | Host control is read at the top of the loop, so park waits. Park then reuses the fresh generation. No event is published in between, so the host's park cut still matches. | | Park arrives before the hook runs | Park restarts with the 20 s deadline, as in #20145. The hook then finds a fresh generation and does nothing. | | Eviction, `runner.suspend` or retire after the restart | They reap the fresh generation through the ordinary paths. | | The restart fails | The session is `closed`. The next turn fails with the same error and the same durable state as a failed turn-start restart. Park returns conflict, as it does for a closed session today. | | Cancellation | `run.cancel`, `turn.interrupt` and `turn.stop` after the restart return `already_settled`, as after any settled turn. A cancellation that retires the runtime closes the session, so no restart runs. | | Interrupted, cancelled or failed turn | `completed_turn_observed` is false, so no turn-end restart runs. The next turn restarts at its start, as today. | | Pending permission or input requests at the terminal | The reducer ends them at the terminal, so the restart runs. A later `request.resolve` fails with `no active turn`, with or without the restart. | | Active session goal at the terminal | No restart, because the goal can keep working after its prompt. | ### Removed as redundant, and what stays - **Removed:** the park-time restart and dormant verify when a turn-end restart already made the generation fresh. The runner's `host_park_ready` proves dormancy just before `prepare_for_park`, and nothing runs in between. - **Removed:** the runner's second park proof in that case, because the provider did not change. - **Removed:** the park-time receipt and ledger rotation. The next turn start does it once, for both paths. - **Kept: the turn-start restart.** Without it, a turn queued before the hook, a turn after an interrupted or cancelled turn, a turn after an active goal, and any non-Claude ACPX turn would run in the generation that ran the previous turn. Tests cover the queued-turn and interrupted cases. - **Kept: the park-time restart.** Without it, a park that arrives before the hook would leave the next run to restart at attach. A test covers this. - **Kept: the settlement of an abandoned park observation.** Without it, the restart drops events that the old generation sent before the abandoned response. A test covers this, and it fails with the settlement removed. - **Kept: the goal-dormancy check.** Without it, the restart kills an active goal's autonomous work. A test covers this, and it fails with the check removed. - **Kept: the ACK gate.** Without it, the restart can run before the host has the terminal result. A runner test covers this. - **Kept: deferred polling of the fresh generation.** Without it, its startup event lands after the host computed its park cut, and the park conflicts. - **Kept: the kept failure message.** Without it, the next turn fails with `ACPX session is not open`, and the restart error is lost. ## Verification - `cargo test -p paperclip-runner-core --lib` and every `tests/acpx_*.rs`, `native_provider_backend` and `durable_recovery` suite: all pass (debug). - New tests (`tests/support/acpx_turn_end_restart.rs`, run from `acpx_park_preparation`) prove these points: - A settled turn reaps the old pid before it spawns the new one, and no predecessor is still live. A same-run follow-up then sends only `turn.start` and reports the reaped pid in `session.reconciled`. - Park after a turn-end restart sends no sidecar command and returns `false`. Attach rebinds the same pid. The run's second turn restarts at its end, and the third turn starts without a restart. - The reaped generation's late callback never reaches the next turn. - A turn that arrives before the hook restarts at its start. A park before the hook restarts once. - A failed turn-end restart gives the same next-turn error, retry error, durable state and process count as a failed turn-start restart. - Cancellation after the restart is `already_settled`. `runner.suspend` reaps the fresh generation. - An active goal skips the restart. An input request open at the terminal behaves the same with or without the restart. - `acpx_provider_turns`: a session-level turn-end restart drops the old generation's late `issues.delete` callback, which reuses the previous turn's call ID. The next turn then admits that call ID from the fresh provider. Interrupted and retired turns do not restart at turn end. - `acpx_park_readiness`: a turn-end-restarted generation is park-ready. An abandoned park observation tail is delivered before the restart. - `durable::runner`: the hook runs only after the terminal is ACKed and when control is idle. It never runs during ACK debt, a waiting control frame or a disconnect. - Server integration tests (`plugin-runner-release-resume`, `plugin-runner-eviction` and `plugin-runner-task-warm`) with this branch's Runner: 49 passed. The provider process counts do not change: each released run still adds one process start, now at turn end instead of at park. - `cargo fmt --check`: clean. - Fake-sidecar timing (release build, 7 runs, medians): | | before (`master` with #20145) | this PR | | --- | --- | --- | | same-run follow-up `turn.start` → `turn.started` | 51.5 ms | 22.0 ms | | second turn after an attach, `turn.start` → `turn.started` | 270.3 ms | 23.5 ms | | turn-end restart, after the terminal is ACKed | — | 29.0 ms | | park preparation after a settled turn | 34.1 ms | 0.03 ms | - Real Claude: not run for this pull request. With real Claude, the restart took about 1.5 s on the turn-start path, and this change takes it off that path. That number is not measured on this branch. ## Risks - A run that ends after one turn without a park now starts one extra sidecar that no turn uses. The Claude CLI itself starts only at the first prompt, so the cost is the sidecar's start and bootstrap. - The restart blocks the Runner loop for its length (about 1.5 s with real Claude), as the turn-start restart does today. It has no deadline, as before. A park that arrives during it waits, inside the host's 30 s park wait. - The fresh generation is not polled until its next turn. If it exits while idle, the next turn finds out at `turn.start` and fails closed. Park finds out at its observation. - The turn-end restart is for Claude only, the same as park preparation. Other ACPX agents keep restarting at turn start. - If the Runner crashes during the restart, the durable state still says `session_open`. Recovery is the same as for a crash while idle. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), in Claude Code with tool use (shell, file edits, local test runs). ## 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 - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] 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 | 58 ++- .../runner-core/src/acpx_provider_backend.rs | 24 +- .../src/acpx_provider_generation.rs | 82 +++-- .../runner-core/src/acpx_provider_session.rs | 11 +- .../runner-core/src/bin/fake-acpx-sidecar.rs | 14 + .../crates/runner-core/src/durable/runner.rs | 102 +++++- .../src/native_provider_backend.rs | 7 + .../tests/acpx_park_preparation.rs | 6 +- .../runner-core/tests/acpx_park_readiness.rs | 50 ++- .../runner-core/tests/acpx_provider_turns.rs | 80 +++++ .../tests/support/acpx_turn_end_restart.rs | 333 ++++++++++++++++++ 11 files changed, 707 insertions(+), 60 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs 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 1b6f91c5bfa..26d4afb8638 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 @@ -1,18 +1,15 @@ use std::time::Instant; use super::{ - await_provider_lifetime_fence, AcpxCommandExecutor, AcpxProviderDescriptor, AcpxProviderSession, + await_provider_lifetime_fence, goal_is_dormant, AcpxCommandExecutor, AcpxProviderDescriptor, + AcpxProviderSession, }; use crate::durable::DurableRunnerError; use crate::provider_bridge::AuthorizedToolSet; impl AcpxCommandExecutor { - pub(super) fn prepare_session_for_park( - &mut self, - deadline: Instant, - ) -> Result { - self.ensure_persistence_healthy()?; - let eligible = self.candidate_admission().is_none() + fn generation_restartable(&self, session_ready: fn(&AcpxProviderSession) -> bool) -> bool { + self.candidate_admission().is_none() && self.state.as_ref().is_some_and(|state| { state.descriptor.agent == "claude" && state.lifecycle == "session_open" @@ -22,12 +19,47 @@ impl AcpxCommandExecutor { && !state.provider_exit_unconfirmed && state.pending_events.is_empty() && state.pending_runtime_requests.is_empty() + && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection)) }) - && self - .session - .as_ref() - .is_some_and(AcpxProviderSession::can_prepare_next_generation); - if !eligible { + && self.session.as_ref().is_some_and(session_ready) + } + + pub(super) fn restart_session_after_turn(&mut self) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; + if !self.generation_restartable(AcpxProviderSession::can_restart_after_turn) { + return Ok(()); + } + let session = self.session.as_mut().expect("checked ACPX session"); + if session.settle_park_observation().is_err() + || !self.generation_restartable(AcpxProviderSession::can_restart_after_turn) + { + return Ok(()); + } + let restarted = self + .session + .as_mut() + .expect("checked ACPX session") + .restart_after_turn(None); + 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; + self.failed_turn_end_restart = Some(error.to_string()); + self.save_state() + } + + pub(super) fn prepare_session_for_park( + &mut self, + deadline: Instant, + ) -> Result { + self.ensure_persistence_healthy()?; + if !self.generation_restartable(AcpxProviderSession::can_prepare_next_generation) { return Ok(false); } let prepared = self @@ -36,7 +68,7 @@ impl AcpxCommandExecutor { .expect("checked ACPX session") .prepare_next_generation(deadline); match prepared { - Ok(()) => Ok(true), + Ok(restarted) => Ok(restarted), Err(error) => { self.recover_failed_park_preparation()?; Err(DurableRunnerError::invalid(format!( 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 9063b2da0e1..94c33a1ee44 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 @@ -966,6 +966,7 @@ pub struct AcpxCommandExecutor { fail_after_state_rename: bool, launch_profile: Option, recovery_pending_effects: Option, + failed_turn_end_restart: Option, } // The durable command's turn and the live provider callback have separate IDs. @@ -1005,6 +1006,7 @@ impl AcpxCommandExecutor { session: None, restore_checked: false, recovery_pending_effects: None, + failed_turn_end_restart: None, restore_error: None, persistence_error: None, #[cfg(test)] @@ -1927,6 +1929,18 @@ 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}" + ))); + } + } let reopenable = self.session.is_none() && self.state.as_ref().is_some_and(|state| { state.identity.is_some() @@ -1997,7 +2011,7 @@ impl AcpxCommandExecutor { let (provider_process_will_be_replaced, previous_process_id) = self .session .as_ref() - .map(|session| match session.prepared_from_process_id() { + .map(|session| match session.restarted_after_turn_from() { Some(process_id) => (true, process_id), None => (session.generation_has_run_turn(), session.process_id()), }) @@ -2611,7 +2625,7 @@ impl AcpxCommandExecutor { || self .session .as_ref() - .is_none_or(AcpxProviderSession::awaits_run_rebind) + .is_none_or(AcpxProviderSession::awaits_next_turn) { return Ok(()); } @@ -2622,7 +2636,7 @@ impl AcpxCommandExecutor { if self .session .as_ref() - .is_some_and(AcpxProviderSession::awaits_run_rebind) + .is_some_and(AcpxProviderSession::awaits_next_turn) { return Ok(()); } @@ -2909,6 +2923,10 @@ impl CommandExecutor for AcpxCommandExecutor { self.prepare_session_for_park(deadline) } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + self.restart_session_after_turn() + } + fn execute(&mut self, command: &Command) -> Result { self.restore()?; if self diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs index cd826f6df07..6d7cd78c6ba 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs @@ -54,18 +54,20 @@ impl AcpxProviderSession { self.generation_has_run_turn } + pub fn restarted_after_turn_from(&self) -> Option { + self.restarted_after_turn_from + } + pub fn awaits_run_rebind(&self) -> bool { self.prepared_generation.is_some() } - pub fn prepared_from_process_id(&self) -> Option { - self.prepared_generation - .as_ref() - .map(|(_, process_id)| *process_id) + pub fn awaits_next_turn(&self) -> bool { + self.prepared_generation.is_some() || self.restarted_after_turn_from.is_some() } pub(super) fn park_tail_is_quiet(&self) -> bool { - if self.prepared_generation.is_none() { + if !self.awaits_next_turn() { return self.transport.buffered_events().len() == 0; } self.transport.buffered_events().all(|event| { @@ -78,65 +80,70 @@ impl AcpxProviderSession { }) } - pub fn can_prepare_next_generation(&self) -> bool { + fn at_settled_turn_boundary(&self) -> bool { !self.closed && self.failure.is_none() && !self.transport_terminated && !self.runtime_retired - && self.generation_has_run_turn && self.completed_turn_observed && self.prepared_generation.is_none() && self.config.candidate_admission.is_none() && self.state.active_turn_id().is_none() && !self.state.has_pending_requests() && !self.state.has_pending_tools() - && self.transport.buffered_events().len() == 0 + && self.park_tail_is_quiet() } - pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result<(), LocalRunnerError> { + pub fn can_restart_after_turn(&self) -> bool { + self.generation_has_run_turn && self.at_settled_turn_boundary() + } + + pub fn can_prepare_next_generation(&self) -> bool { + (self.generation_has_run_turn || self.restarted_after_turn_from.is_some()) + && self.at_settled_turn_boundary() + } + + pub fn restart_after_turn( + &mut self, + deadline: Option, + ) -> Result<(), LocalRunnerError> { + if !self.can_restart_after_turn() { + return Err(LocalRunnerError::invalid( + "ACPX provider is not at a settled turn boundary", + )); + } + let previous_process_id = self.process_id(); + if let Err(error) = self.restart_idle_provider(deadline) { + return Err(self.fail_closed(error)); + } + self.restarted_after_turn_from = Some(previous_process_id); + Ok(()) + } + + pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result { if !self.can_prepare_next_generation() { return Err(LocalRunnerError::invalid( "ACPX provider is not at a reusable generation boundary", )); } - let previous_process_id = self.process_id(); - let mut tool_bridge = self.tool_bridge.clone(); - let mut reserved_tool_bridge = self.reserved_tool_bridge.clone(); - tool_bridge - .prepare_turn_after_provider_restart() - .and_then(|()| reserved_tool_bridge.prepare_turn_after_provider_restart()) - .map_err(|error| { - LocalRunnerError::invalid(format!( - "ACPX tool receipt rotation before park failed: {error}" - )) - })?; - if let Err(error) = self.restart_idle_provider(Some(deadline)) { - return Err(self.fail_closed(error)); - } - if self.state.settled_turn_identity_capacity_reached() { - if let Err(error) = self - .state - .rotate_settled_turn_identities_after_provider_restart() - { + let restarted = self.generation_has_run_turn; + if restarted { + self.restart_after_turn(Some(deadline))?; + if let Err(error) = self.verify_dormant_session() { return Err(self.fail_closed(error)); } + self.transport.bound_requests_by(None); } - self.tool_bridge = tool_bridge; - self.reserved_tool_bridge = reserved_tool_bridge; self.completed_turn_observed = false; - if let Err(error) = self.verify_dormant_session() { - return Err(self.fail_closed(error)); - } - self.transport.bound_requests_by(None); - self.prepared_generation = Some((generation_digest(&self.config), previous_process_id)); - Ok(()) + self.prepared_generation = Some(generation_digest(&self.config)); + Ok(restarted) } pub fn rebind_run( &mut self, next: &AcpxProviderSessionConfig, ) -> Result { - let Some((prepared, _)) = self.prepared_generation.as_ref() else { + let Some(prepared) = self.prepared_generation.as_ref() else { return Ok(false); }; if self.closed @@ -194,6 +201,7 @@ impl AcpxProviderSession { self.config = next.clone(); self.catalog_revision = next.catalog_revision; self.working_directory = next.working_directory.clone(); + self.restarted_after_turn_from = None; self.prepared_generation = None; if let Err(error) = self.verify_dormant_session() { return Err(self.fail_closed(error)); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index a81d3db3742..e588b0a4d68 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs @@ -357,7 +357,8 @@ pub struct AcpxProviderSession { runtime_retired: bool, completed_turn_observed: bool, generation_has_run_turn: bool, - prepared_generation: Option<(String, u32)>, + restarted_after_turn_from: Option, + prepared_generation: Option, } pub(crate) fn goal_is_dormant(projection: &Value) -> bool { @@ -417,6 +418,7 @@ impl AcpxProviderSession { runtime_retired: false, completed_turn_observed: false, generation_has_run_turn: false, + restarted_after_turn_from: None, prepared_generation: None, }) } @@ -660,12 +662,13 @@ impl AcpxProviderSession { // provider turn originated a late HTTP callback. Reap the old sidecar // and provider before releasing its call-ID tombstones, then resume // the same verified persistent session in a fresh process generation. - let provider_restarted = self.generation_has_run_turn; - if provider_restarted { + let restart_now = self.generation_has_run_turn; + if restart_now { if let Err(error) = self.restart_idle_provider(None) { return Err(self.fail_closed(error)); } } + let provider_restarted = restart_now || self.restarted_after_turn_from.is_some(); if rotate_turn_identity_ledger { if let Err(error) = self .state @@ -700,6 +703,7 @@ impl AcpxProviderSession { )))); } self.generation_has_run_turn = true; + self.restarted_after_turn_from = None; self.prepared_generation = None; let response = match self.transport.request_turn_start( json!({"turnId":turn_id,"message":message,"performance":performance}), @@ -2256,6 +2260,7 @@ mod mode_tests { runtime_retired: false, completed_turn_observed: false, generation_has_run_turn: false, + restarted_after_turn_from: None, prepared_generation: None, } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs index 76a6a309f09..cfba5cc3ba9 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs @@ -299,6 +299,20 @@ fn run() -> Result<(), Box> { source_pending_completion = Some(turn.to_owned()); continue; } + if fault == "pending-input" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.input_requested", + &source_run, + turn, + json!({"requestId":"input-at-terminal","questionSet":{ + "schema":"paperclip.question_set.v1", + "questions":[{"id":"target","prompt":"Which target?","required":true, + "answerMode":"single_select","options":[{"id":"first","label":"First"}]}]}}), + )?; + next_sequence += 1; + } write_turn_event( &mut stdout, next_sequence, 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 5a45df27369..d786da4a324 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 @@ -323,6 +323,10 @@ pub trait CommandExecutor { Ok(false) } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + Ok(()) + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -2168,7 +2172,11 @@ fn poll_executor_events_after_controller_ack( if state.acked_source_seq < sent_source_seq { return executor.maintain_backpressured_provider(); } - poll_executor_events(state, store, config, executor) + poll_executor_events(state, store, config, executor)?; + if state.acked_source_seq >= state.highest_source_seq() { + executor.prepare_next_turn()?; + } + Ok(()) } fn poll_executor_events_when_control_idle( @@ -5075,6 +5083,98 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn turn_boundary_preparation_waits_until_the_controller_acks_the_terminal_event() { + struct BoundaryExecutor { + events: VecDeque, + calls: Vec<&'static str>, + } + impl CommandExecutor for BoundaryExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("the poll gate cannot execute a command") + } + fn poll_events(&mut self) -> Result, DurableRunnerError> { + self.calls.push("poll"); + Ok(self.events.iter().cloned().collect()) + } + fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> { + self.events.drain(..count); + Ok(()) + } + fn maintain_backpressured_provider(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("maintain"); + Ok(()) + } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("prepare"); + Ok(()) + } + } + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-turn-boundary-{}", + uuid::Uuid::new_v4() + )); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = BoundaryExecutor { + events: VecDeque::from([PolledEvent { + executor_event_id: "provider-terminal".to_owned(), + event_type: "turn.completed".to_owned(), + priority: EventPriority::P0, + payload: json!({"status": "completed"}), + }]), + calls: Vec::new(), + }; + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 0) + .unwrap(); + assert_eq!(executor.calls, ["poll"]); + assert_eq!(state.highest_source_seq(), 1); + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 0) + .unwrap(); + assert_eq!( + executor.calls, + ["poll", "poll"], + "an unsent terminal is not out" + ); + poll_executor_events_after_controller_ack(&mut state, &store, &config, &mut executor, 1) + .unwrap(); + assert_eq!(executor.calls[2..], ["maintain"]); + state.apply_ack(1, 2).unwrap(); + store.save(&state).unwrap(); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Ok(Some(json!({"kind": "command"}))), + ) + .unwrap(); + assert_eq!(executor.calls[3..], ["maintain"], "control frames go first"); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Err(DurableRunnerError::invalid("disconnected")), + ) + .unwrap(); + assert_eq!(executor.calls.len(), 4); + poll_executor_events_when_control_idle( + &mut state, + &store, + &config, + &mut executor, + 1, + &Ok(None), + ) + .unwrap(); + assert_eq!(executor.calls[4..], ["poll", "prepare"]); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn controller_ack_debt_runs_cleanup_without_moving_event_or_ack_cursors() { struct MaintenanceExecutor { 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 f1f593634d9..13551eb72b9 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 @@ -344,6 +344,13 @@ impl CommandExecutor for NativeProviderCommandExecutor { } } + fn prepare_next_turn(&mut self) -> Result<(), DurableRunnerError> { + match self.selected.as_mut() { + Some(SelectedExecutor::Acpx(executor)) => executor.prepare_next_turn(), + _ => Ok(()), + } + } + fn observe_recovery_state( &mut self, state: &crate::durable::DurableState, diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs index 0f7fcf96fc7..f9d9a3072ca 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs @@ -14,6 +14,9 @@ use paperclip_runner_core::provider_bridge::{authorized_tool_catalog_digest, Aut use serde_json::{json, Value}; use sha2::{Digest, Sha256}; +#[path = "support/acpx_turn_end_restart.rs"] +mod turn_end_restart; + struct Fixture(PathBuf); impl Drop for Fixture { fn drop(&mut self) { @@ -267,8 +270,9 @@ fn last_process_start(journal: &[Value]) -> usize { fn prepared_park(harness: &mut Harness) -> Value { harness.complete_turn("first"); + harness.executor.prepare_next_turn().unwrap(); assert!(harness.executor.host_park_ready().unwrap()); - assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); harness.fixture.process_starts().last().unwrap()["processStart"].clone() } diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs index 4b56a5abc4c..f1994f3b21a 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs @@ -265,8 +265,12 @@ fn hosted_park_acpx_requires_completion_in_this_lifetime_and_exact_authority() { )) .unwrap(); settle(&mut executor); - assert!(executor.host_park_ready().unwrap()); - assert!(executor + executor.prepare_next_turn().unwrap(); + assert!( + executor.host_park_ready().unwrap(), + "a generation restarted at the turn boundary is park-ready" + ); + assert!(!executor .prepare_for_park(Instant::now() + Duration::from_secs(20)) .unwrap()); assert!( @@ -479,3 +483,45 @@ fn hosted_park_acpx_does_not_enable_an_unqualified_park_provider() { .unwrap()); executor.shutdown().unwrap(); } + +#[test] +fn a_turn_end_restart_waits_until_an_abandoned_park_observation_tail_is_delivered() { + let (fixture, _config, mut executor, opened) = fixture("claude"); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"first", "text":"Complete first work."}), + )) + .unwrap(); + settle(&mut executor); + fs::write(fixture.0.join("fault"), "slow-snapshot").unwrap(); + assert!(!executor.host_park_ready().unwrap()); + fs::write(fixture.0.join("fault"), "").unwrap(); + let starts = || { + fs::read_to_string(fixture.0.join("sidecar.jsonl")) + .unwrap() + .lines() + .filter(|line| line.contains("processStart")) + .count() + }; + executor.prepare_next_turn().unwrap(); + assert_eq!( + starts(), + 1, + "the old generation's tail is not yet delivered" + ); + assert!(std::process::Command::new("kill") + .args(["-0", &opened["processId"].to_string()]) + .status() + .unwrap() + .success()); + let events = executor.poll_events().unwrap(); + assert!(events + .iter() + .any(|event| event.payload["code"] == "slow_park_observation")); + executor.acknowledge_events(events.len()).unwrap(); + executor.prepare_next_turn().unwrap(); + assert_eq!(starts(), 2); + executor.shutdown().unwrap(); +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs index 1294ea1adf5..f31bb5e1da5 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_provider_turns.rs @@ -934,3 +934,83 @@ fn prepared_generation_never_admits_the_reaped_generation_late_callback() { session.shutdown("test complete").unwrap(); std::fs::remove_dir_all(directory).unwrap(); } + +#[test] +fn a_turn_end_restart_reaps_the_late_callback_and_releases_receipts_at_the_next_turn() { + let mut session = AcpxProviderSession::start(&config("turns-late-tool-after-suspend")).unwrap(); + session + .start_turn("turn-1", "Please continue", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(!session.can_restart_after_turn()); + session + .deliver_tool_result(&late_result("call-reused")) + .unwrap(); + let first_pid = session.process_id(); + session.restart_after_turn(None).unwrap(); + let fresh_pid = session.process_id(); + assert_ne!(fresh_pid, first_pid); + assert_eq!(session.restarted_after_turn_from(), Some(first_pid)); + assert!(!session.generation_has_run_turn() && !session.can_restart_after_turn()); + assert!(session + .poll_event(Duration::from_millis(200)) + .unwrap() + .is_none()); + + session + .start_turn("turn-2", "Use the fresh provider", &std::env::temp_dir()) + .unwrap(); + assert_eq!(session.process_id(), fresh_pid); + assert_eq!(session.restarted_after_turn_from(), None); + let fresh = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + &fresh[0], + AcpxProviderStateEvent::ToolCall { call_id, operation_id, input } + if call_id == "call-reused" + && operation_id == "issues.read" + && input["id"] == "issue-1" + )); + let terminal = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(matches!( + terminal.last().unwrap(), + AcpxProviderStateEvent::TurnTerminal { turn_id, .. } if turn_id == "turn-2" + )); + session.shutdown("test complete").unwrap(); +} + +#[test] +fn an_interrupted_or_retired_turn_restarts_at_the_next_turn_start() { + let mut session = AcpxProviderSession::start(&config("turns")).unwrap(); + session + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(!session.can_restart_after_turn()); + session + .interrupt_turn("turn-1", "Paperclip interruption") + .unwrap(); + session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + let interrupted_pid = session.process_id(); + assert!(!session.can_restart_after_turn()); + assert!(session.restart_after_turn(None).is_err()); + assert_eq!(session.process_id(), interrupted_pid); + session + .start_turn("turn-2", "Start again", &std::env::temp_dir()) + .unwrap(); + assert_ne!(session.process_id(), interrupted_pid); + session.shutdown("test complete").unwrap(); + + let mut retired = AcpxProviderSession::start(&config("turns-retired")).unwrap(); + retired + .start_turn("turn-1", "Please help", &std::env::temp_dir()) + .unwrap(); + retired + .interrupt_turn("turn-1", "Paperclip cancellation") + .unwrap(); + retired.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + assert!(retired.state().active_turn_id().is_none()); + assert!(retired.runtime_retired()); + assert!(!retired.can_restart_after_turn()); + retired.shutdown("test complete").unwrap(); +} 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 new file mode 100644 index 00000000000..181dc19fd06 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/support/acpx_turn_end_restart.rs @@ -0,0 +1,333 @@ +use super::*; + +const WARM_TURN: [&str; 2] = ["turn.start", "session.goal.get"]; + +fn live_providers(harness: &Harness) -> Vec { + harness + .fixture + .process_starts() + .into_iter() + .map(|line| line["processStart"].clone()) + .filter(alive) + .collect() +} + +fn settle(harness: &mut Harness, turn: &str) -> Vec { + let events = harness.complete_turn(turn); + harness.executor.prepare_next_turn().unwrap(); + events +} + +fn start_follow_up(harness: &mut Harness, turn: &str) -> (Vec, Vec, Duration) { + let before = harness.fixture.journal().len(); + let started = Instant::now(); + let execution = harness.execute( + "turn.start", + json!({"turnId": turn, "text": "Complete the work."}), + ); + let elapsed = started.elapsed(); + let mut events = execution + .events + .iter() + .map(|(kind, _, payload)| json!({"type":kind,"payload":payload})) + .collect::>(); + let deadline = Instant::now() + Duration::from_secs(5); + while !events + .iter() + .any(|event: &Value| event["type"] == "turn.completed") + { + assert!(Instant::now() < deadline, "no completed provider turn"); + events.extend(harness.drain()); + } + events.extend(harness.drain()); + let commands = commands_after(&harness.fixture.journal(), before); + (events, commands, elapsed) +} + +fn reconciled(events: &[Value]) -> &Value { + &events + .iter() + .find(|event| event["type"] == "session.reconciled") + .expect("the turn reports the replaced provider process")["payload"] +} + +#[test] +fn a_settled_turn_restarts_its_provider_and_the_next_turn_starts_without_one() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let before = harness.fixture.journal().len(); + let started = Instant::now(); + harness.executor.prepare_next_turn().unwrap(); + let restart = started.elapsed(); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert_eq!( + commands_after(&journal[..spawned], before), + ["session.suspend"] + ); + assert_eq!(journal[spawned]["livePredecessors"], json!([])); + assert_eq!( + commands_after(&journal, spawned), + ["initialize", "session.open", "run.attach"] + ); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-1"); + let fresh_pid = journal[spawned]["processStart"].clone(); + assert!(!alive(&old_pid)); + assert_eq!(live_providers(&harness), [fresh_pid.clone()]); + assert_eq!( + harness.drain(), + Vec::::new(), + "the fresh generation's output waits for its first turn" + ); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 2); + + let (events, commands, turn_start) = start_follow_up(&mut harness, "second"); + eprintln!("turn-end restart={restart:?} follow-up turn.start={turn_start:?}"); + assert_eq!(commands, WARM_TURN); + assert_eq!(reconciled(&events)["previousProcessId"], old_pid); + assert_eq!(reconciled(&events)["processId"], fresh_pid); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == fresh_pid)); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 3); + assert_eq!(live_providers(&harness).len(), 1); + let (_, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + harness.executor.shutdown().unwrap(); + assert_eq!(live_providers(&harness), Vec::::new()); +} + +#[test] +fn park_reuses_the_generation_a_settled_turn_restarted() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + let fresh_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(harness.executor.host_park_ready().unwrap()); + let before = harness.fixture.journal().len(); + let started = Instant::now(); + assert!( + !harness.executor.prepare_for_park(park_deadline()).unwrap(), + "park neither restarts nor asks for a second proof" + ); + eprintln!( + "park preparation after a turn-end restart={:?}", + started.elapsed() + ); + assert_eq!( + commands_after(&harness.fixture.journal(), before), + Vec::::new() + ); + assert!(harness.executor.host_park_ready().unwrap()); + assert_eq!(harness.fixture.process_starts().len(), 2); + + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["processId"], fresh_pid); + let events = settle(&mut harness, "second"); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == fresh_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + let (events, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + assert_eq!(reconciled(&events)["previousProcessId"], fresh_pid); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn the_reaped_generation_late_callback_never_reaches_the_next_turn() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("late-tool-after-suspend"); + harness.executor.prepare_next_turn().unwrap(); + harness.fixture.fault(""); + assert_eq!(harness.drain(), Vec::::new()); + let (events, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands, WARM_TURN); + assert!(!format!("{events:?}").contains("late-call")); + assert!(!format!("{events:?}").contains("late-old-generation")); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_turn_that_arrives_before_the_turn_end_restart_restarts_at_its_start() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let (events, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands[..2], ["session.suspend", "initialize"]); + assert_eq!(reconciled(&events)["previousProcessId"], old_pid); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!(live_providers(&harness).len(), 1); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 3); + assert_eq!(live_providers(&harness).len(), 1); + let (_, commands, _) = start_follow_up(&mut harness, "third"); + assert_eq!(commands, WARM_TURN); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_park_before_the_turn_end_restart_restarts_once() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!(live_providers(&harness).len(), 1); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["processId"], live_providers(&harness)[0]); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_failed_turn_end_restart_fails_the_next_turn_like_a_failed_turn_start_restart() { + let outcome = |turn_end: bool| { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + if turn_end { + harness.executor.prepare_next_turn().unwrap(); + assert!(!alive(&old_pid)); + assert!(!harness.executor.host_park_ready().unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert_eq!(harness.drain(), Vec::::new()); + harness.executor.prepare_next_turn().unwrap(); + } + let first = harness + .executor + .execute(&command( + 90, + "turn.start", + json!({"turnId": "second", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + harness.fixture.fault(""); + let retry = harness + .executor + .execute(&command( + 91, + "turn.start", + json!({"turnId": "third", "text": "Complete the work."}), + )) + .unwrap_err() + .to_string(); + assert_eq!(live_providers(&harness), Vec::::new()); + let state = harness.fixture.state(); + let starts = harness.fixture.process_starts().len(); + harness.executor.shutdown().unwrap(); + ( + first, + retry, + state["lifecycle"].clone(), + state["providerExitUnconfirmed"].clone(), + state["activeTurnId"].clone(), + state["pendingEvents"].clone(), + starts, + ) + }; + let 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)); +} + +#[test] +fn cancellation_after_a_turn_end_restart_is_already_settled() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + for kind in ["run.cancel", "turn.interrupt", "turn.stop"] { + let result = harness.execute(kind, json!({})).result; + assert_eq!(result["status"], "already_settled", "{kind}"); + } + let (_, commands, _) = start_follow_up(&mut harness, "second"); + assert_eq!(commands, WARM_TURN); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn suspension_and_retirement_after_a_turn_end_restart_reap_the_fresh_generation() { + let mut harness = Harness::new(); + settle(&mut harness, "first"); + let fresh_pid = live_providers(&harness)[0].clone(); + harness.execute("runner.suspend", json!({})); + assert!(!alive(&fresh_pid)); + assert_eq!(harness.fixture.state()["lifecycle"], "suspended"); + let events = harness.complete_turn("second"); + assert!(events + .iter() + .any(|event| event["type"] == "session.resumed")); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.prepare_next_turn().unwrap(); + let fresh_pid = live_providers(&harness)[0].clone(); + assert_eq!(harness.fixture.process_starts().len(), 4); + harness.executor.shutdown().unwrap(); + assert!(!alive(&fresh_pid)); +} + +#[test] +fn an_active_goal_or_a_request_at_the_terminal_keeps_today_s_restart_timing() { + let mut harness = Harness::new(); + harness.fixture.fault("goal"); + harness.complete_turn("first"); + harness.executor.prepare_next_turn().unwrap(); + assert_eq!( + harness.fixture.process_starts().len(), + 1, + "an active goal can keep working after its prompt" + ); + harness.executor.shutdown().unwrap(); + + let resolve = |turn_end: bool| { + let mut harness = Harness::new(); + harness.fixture.fault("pending-input"); + let events = harness.complete_turn("first"); + harness.fixture.fault(""); + if turn_end { + harness.executor.prepare_next_turn().unwrap(); + } + let resolved = harness + .executor + .execute(&command( + 90, + "request.resolve", + json!({"requestId": "input-at-terminal", "answers": {"target": "first"}}), + )) + .map(|execution| execution.result) + .map_err(|error| error.to_string()); + let (_, commands, _) = start_follow_up(&mut harness, "second"); + let starts = harness.fixture.process_starts().len(); + harness.executor.shutdown().unwrap(); + ( + events + .iter() + .map(|event| event["type"].clone()) + .collect::>(), + resolved, + commands.contains(&"session.suspend".to_owned()), + starts, + ) + }; + let at_turn_end = resolve(true); + let at_turn_start = resolve(false); + assert_eq!(at_turn_end.0, at_turn_start.0); + assert_eq!(at_turn_end.1, at_turn_start.1); + assert_eq!(at_turn_end.3, at_turn_start.3); + eprintln!("request at terminal: {at_turn_end:?}"); +}