From 2d934130850bd396647537f308aed154f7804521 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sun, 8 Nov 2026 21:38:57 -0800 Subject: [PATCH] perf(runner): prepare the next Claude provider generation at park (#20145) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip runs agents through Paperclip Runner, and a host can park an idle Runner between runs and resume it for the next run on the same task. > - For Claude (ACPX), a follow-up run on a parked Runner pays about 1.5–2.3 s at `run.attach` before the provider can start the turn. The attachment closes the Claude sidecar and spawns a new one: initialize, `session.open` with `session/load`, `run.attach`, then a dormant verify. > - The respawn is required. The tool-bridge MCP endpoint lasts for the whole session and can't tell which turn a late callback came from, so a new turn must not run in a process that ran an earlier one. > - Nothing requires the respawn to happen at the *start* of the next run. A turn that ends 2 s later is a better experience than one that starts 2 s later. > - This pull request does the respawn while parking. The next run's `run.attach` then only rebinds the run id on a fresh, never-used generation. > - Follow-up Claude runs on a parked Runner reach the provider about 1.5–2 s sooner, and the late-callback guarantees are unchanged. ## Linked Issues or Issue Description No public issue exists. The issue is described inline. ### What happened? On a hosted deployment with parked Runners, Claude follow-up runs reached the provider about 1.8 s later than Codex follow-ups. `runner.session.resume` medians were 5.2 s for Claude and 3.4 s for Codex. The extra time is the ACPX respawn inside `run.attach`. Codex keeps its process in place. ### Expected behavior The respawn happens while the Runner parks, so the next run starts on a prepared generation. ### Steps to reproduce 1. Run a Claude (ACPX) task on a host that parks Runners between runs. 2. Let the run finish and the Runner park. 3. Send a follow-up. `run.attach` closes the provider and starts, initializes, opens and attaches a new sidecar before `turn.start`. The Claude CLI itself starts at the first prompt on either path. ### Paperclip version or commit `master` at e194e193. ### Deployment mode Hosted, with parked Runner tasks. ## What Changed - **`durable/runner.rs`:** - New `CommandExecutor::prepare_for_park(deadline)` hook, a no-op by default. The runner passes a deadline 20 s after the park request; the host waits 30 s for the reply. - The Park host-control arm calls it after `host_park_ready()`, re-proves readiness, and only then replies `parked`. - A failure records a diagnostic and replies `conflict`, the existing park-failure path. - **ACPX, park side** (`acpx_park_preparation.rs`, `acpx_provider_generation.rs`): - For Claude, with the session open, nothing pending and nothing queued: reap the old generation, using the same suspend-and-reap sequence a second turn uses. - Start the replacement with the current run attached, and prove it dormant. - Roll the tool receipt epochs and the turn-identity ledger over. - Record a digest of the session configuration it was opened with, ignoring the MCP binding id and the aggregate digest. - Every sidecar request during preparation is capped at the park deadline. Running out of time fails like any other preparation failure. - **ACPX, attach side:** - A prepared, never-used generation whose digest matches the next run is rebound: the sidecar gets `run.attach` with the new run id, the reducer and tool bridges are reset, and the snapshot is checked for the new run id. - Anything else takes today's cold path: a different tool set, skill, prompt or launch; a buffered callback; or any rebind error. - **ACPX, turn start:** the provider restarts when its generation has run a turn, instead of when the run has settled turns. `has_settled_turns` is removed. - **ACPX, polling:** polling skips a prepared generation until it is rebound or starts a turn, so nothing it emits is attributed to the parked run. - **Sidecar** (`acpx-runtime-sidecar.ts`): rejects `run.attach` once a turn has started in that process. - **Fake sidecar:** reports live predecessors at process start, and adds the faults `reject-open`, `late-tool-after-suspend` and `pre-turn-tool`. ## Verification - `cargo test -p paperclip-runner-core`: 1029 passed (debug). - New `tests/acpx_park_preparation.rs` (10 tests). These prove: - park reaps the old pid before spawning the new one, and acks only after the dormant verify; - a late callback from the old generation after park is rejected, and so is a pre-turn call from the new generation; - a compatible attach keeps the pid, and the sidecar and snapshot carry the new run id; - changed instructions, tools, skills or runtime context take the cold path; - a used generation is never rebound; - a prepare failure produces a park conflict and a recoverable state; - Retire works after a prepared park; - a replacement whose `session.open` stalls fails at the park deadline, leaves the session suspended, and the next run attaches and completes a turn. - `cargo fmt --check` and `cargo check --workspace --locked`: clean. - `vitest` `acpx-runtime-sidecar.test.ts` and `acpx-sidecar-input.test.ts`: 126 passed. The new re-attach guard test fails if the guard is removed. - `vitest` `plugin-runner-release-resume.integration.test.ts` and `plugin-runner-eviction.integration.test.ts` with the branch's Runner: 48 passed. Both now expect the provider process to be replaced at park instead of at attach. - Fake-sidecar timing (release build, 3 runs): `run.attach` took 230–260 ms cold and 40–80 ms after a prepared park. - Real Claude (claude-agent-acp 0.73.0, release build on a hosted test VM, 3 runs each way plus one with a 180 s cgroup freeze between park and attach): | | before | this PR | |---|---|---| | `run.attach` | 1691 ms mean | 212 ms mean | | `run.attach` start → follow-up `turn.started` | 2565 ms mean | 1065 ms mean | | `turn.start` → first provider output | 2261 ms mean | 2232 ms mean | | park preparation added to park | — | about 1450 ms | Session recall, an authorized tool call after the rebind, and the absence of any tool call or event from the old run all passed in every run, including after the freeze. The prepared generation holds only the sidecar: the Claude CLI starts at the follow-up prompt on both paths, so a parked Runner also holds one node process instead of the whole Claude tree. - Hit rate: on a hosted test deployment, all 15 consecutive Claude runs across two tasks carried identical content-addressed instruction and skill bundle paths, so the prepared digest would have matched. - Not run: the hosted park, claim and attach loop with real Claude, a native MCP binding or skills with real Claude, `cargo test --release`, smoke tests. - Pre-existing flake: `unconfirmed_suspension_state_becomes_recoverable_only_after_cleanup_and_suspend` fails intermittently on clean `master` when the `hosted acpx durable::` subset runs in parallel. It passed in both full runs. ## Risks - **Park takes about 1.5 s longer** before it replies with real Claude. Preparation is capped at 20 s; the host's park wait is 30 s. - **Configuration changes:** a change between park and the next run (registered asset paths, tools, prompt) takes the cold path. That's safe, but saves nothing. - **Stale environment:** a prepared process was opened with the previous run's environment and disk state. If credential or configuration files change on disk while the Runner is parked, the prepared process won't see them, whereas the cold path would. - **Failed prepare:** a park that fails after the reap leaves no live provider. The next turn or `run.attach` starts Claude cold, as today. - **Scope:** only ACPX Claude prepares. Other providers park exactly as before. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, git, GitHub CLI) and subagents. ## 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 - [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_event_scope.rs | 4 - .../runner-core/src/acpx_park_preparation.rs | 114 ++++ .../runner-core/src/acpx_provider_backend.rs | 46 +- .../src/acpx_provider_generation.rs | 203 ++++++ .../runner-core/src/acpx_provider_session.rs | 27 +- .../runner-core/src/acpx_provider_state.rs | 4 - .../runner-core/src/acpx_sidecar_transport.rs | 18 +- .../runner-core/src/bin/fake-acpx-sidecar.rs | 56 +- .../crates/runner-core/src/durable/runner.rs | 149 +++++ .../src/durable/tests/hosted_retention.rs | 9 + .../src/native_provider_backend.rs | 7 + .../tests/acpx_park_preparation.rs | 594 ++++++++++++++++++ .../runner-core/tests/acpx_park_readiness.rs | 10 + .../runner-core/tests/acpx_provider_turns.rs | 145 ++++- .../src/cli/acpx-runtime-sidecar.test.ts | 44 ++ .../src/cli/acpx-runtime-sidecar.ts | 3 + ...plugin-runner-eviction.integration.test.ts | 4 +- ...-runner-release-resume.integration.test.ts | 2 +- 18 files changed, 1402 insertions(+), 37 deletions(-) create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs create mode 100644 packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs create mode 100644 packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs index 70a8adb2fad..414715f168d 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_event_scope.rs @@ -38,10 +38,6 @@ impl AcpxEventScope { self.active_turn_id.as_deref() } - pub(crate) fn has_settled_turns(&self) -> bool { - !self.settled_turn_ids.is_empty() - } - pub fn bind_turn(&mut self, turn_id: impl Into) -> Result<(), LocalRunnerError> { let turn_id = turn_id.into(); validate_scope_id(&turn_id, "turn", DURABLE_STABLE_ID_CHARS)?; 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 new file mode 100644 index 00000000000..1b6f91c5bfa --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_park_preparation.rs @@ -0,0 +1,114 @@ +use std::time::Instant; + +use super::{ + await_provider_lifetime_fence, 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() + && self.state.as_ref().is_some_and(|state| { + state.descriptor.agent == "claude" + && state.lifecycle == "session_open" + && state.descriptor.run_id == self.context.run_id + && state.identity.is_some() + && state.active_turn_id.is_none() + && !state.provider_exit_unconfirmed + && state.pending_events.is_empty() + && state.pending_runtime_requests.is_empty() + }) + && self + .session + .as_ref() + .is_some_and(AcpxProviderSession::can_prepare_next_generation); + if !eligible { + return Ok(false); + } + let prepared = self + .session + .as_mut() + .expect("checked ACPX session") + .prepare_next_generation(deadline); + match prepared { + Ok(()) => Ok(true), + Err(error) => { + self.recover_failed_park_preparation()?; + Err(DurableRunnerError::invalid(format!( + "ACPX park preparation failed: {error}" + ))) + } + } + } + + fn recover_failed_park_preparation(&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"); + state.lifecycle = "prepared".to_owned(); + state.provider_exit_unconfirmed = true; + state + .identity + .as_ref() + .expect("park preparation requires a provider identity") + .provider_lifetime_fence_candidates + }; + self.save_state()?; + let _provider_lifetime_fence = await_provider_lifetime_fence(candidates)?; + let state = self + .state + .as_mut() + .expect("ACPX state remains available after park preparation"); + state.provider_exit_unconfirmed = false; + state.lifecycle = "suspended".to_owned(); + if let Err(error) = self.save_state() { + let state = self + .state + .as_mut() + .expect("ACPX state remains available after save failure"); + state.provider_exit_unconfirmed = true; + state.lifecycle = "prepared".to_owned(); + return Err(error); + } + Ok(()) + } + + pub(super) fn rebind_prepared_session( + &mut self, + descriptor: &AcpxProviderDescriptor, + tool_set: &AuthorizedToolSet, + next_candidate: bool, + ) -> bool { + if next_candidate + || self.candidate_admission().is_some() + || !self + .session + .as_ref() + .is_some_and(AcpxProviderSession::awaits_run_rebind) + { + return false; + } + let Some(identity) = self.state.as_ref().and_then(|state| state.identity.clone()) else { + return false; + }; + let Ok(config) = descriptor.session_config( + tool_set.clone(), + Some(identity), + self.launch_profile.as_ref(), + ) else { + return false; + }; + self.session + .as_mut() + .is_some_and(|session| session.rebind_run(&config).unwrap_or(false)) + } +} 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 e16c4f3e7c2..9063b2da0e1 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 @@ -39,6 +39,9 @@ use crate::provider_events::{ }; use crate::qualified_launch::{verify_executable_launch_artifact, verify_launch_artifact}; +#[path = "acpx_park_preparation.rs"] +mod park_preparation; + fn bounded_turn_timing(response: &Value, elapsed: Duration) -> Value { let mut receipt = serde_json::Map::new(); for key in [ @@ -1703,16 +1706,18 @@ impl AcpxCommandExecutor { "run.attach requires the same settled ACPX provider profile and session", )); } - if let Some(session) = self.session.as_mut() { - session - .shutdown("run authority rotation") - .map_err(|error| { - DurableRunnerError::invalid(format!( - "failed to checkpoint ACPX before attaching a new run: {error}" - )) - })?; + if !self.rebind_prepared_session(&descriptor, &tool_set, next_candidate.is_some()) { + if let Some(session) = self.session.as_mut() { + session + .shutdown("run authority rotation") + .map_err(|error| { + DurableRunnerError::invalid(format!( + "failed to checkpoint ACPX before attaching a new run: {error}" + )) + })?; + } + self.session = None; } - self.session = None; if let Some(admission) = next_candidate { self.launch_profile .as_mut() @@ -1982,7 +1987,7 @@ impl AcpxCommandExecutor { if self .session .as_ref() - .is_some_and(|session| session.state().has_settled_turns()) + .is_some_and(AcpxProviderSession::generation_has_run_turn) { self.retain_buffered_provider_events()?; } @@ -1992,7 +1997,10 @@ impl AcpxCommandExecutor { let (provider_process_will_be_replaced, previous_process_id) = self .session .as_ref() - .map(|session| (session.state().has_settled_turns(), session.process_id())) + .map(|session| match session.prepared_from_process_id() { + Some(process_id) => (true, process_id), + None => (session.generation_has_run_turn(), session.process_id()), + }) .expect("ACPX session exists after availability check"); { let state = self @@ -2600,7 +2608,10 @@ impl AcpxCommandExecutor { .state .as_ref() .is_some_and(|state| !state.pending_events.is_empty() || state.lifecycle == "closed") - || self.session.is_none() + || self + .session + .as_ref() + .is_none_or(AcpxProviderSession::awaits_run_rebind) { return Ok(()); } @@ -2608,6 +2619,13 @@ impl AcpxCommandExecutor { } fn retain_buffered_provider_events(&mut self) -> Result<(), DurableRunnerError> { + if self + .session + .as_ref() + .is_some_and(AcpxProviderSession::awaits_run_rebind) + { + return Ok(()); + } if let Some(session) = self.session.as_mut() { session.settle_park_observation().map_err(|error| { DurableRunnerError::invalid(format!("ACPX observation settlement failed: {error}")) @@ -2887,6 +2905,10 @@ impl CommandExecutor for AcpxCommandExecutor { })) } + fn prepare_for_park(&mut self, deadline: Instant) -> Result { + self.prepare_session_for_park(deadline) + } + 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 new file mode 100644 index 00000000000..cd826f6df07 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_generation.rs @@ -0,0 +1,203 @@ +use std::time::Instant; + +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use super::{ + reserved_terminal_tool_bridge, sidecar_run_tool_operations, AcpxProviderSession, + AcpxProviderSessionConfig, +}; +use crate::acpx_provider_state::AcpxProviderState; +use crate::generated_acpx_sidecar_contract::{ + GeneratedAcpxSidecarCommand, GeneratedAcpxSidecarEventType, +}; +use crate::local_runner::LocalRunnerError; +use crate::provider_bridge::ProviderToolBridge; + +fn generation_digest(config: &AcpxProviderSessionConfig) -> String { + let mut runtime_context = config.runtime_context.clone(); + let mcp_bound = runtime_context + .pointer("/mcp/bindingId") + .is_some_and(|binding| !binding.is_null()); + if let Some(object) = runtime_context.as_object_mut() { + object.remove("mcp"); + object.remove("aggregateDigest"); + } + let shape = json!({ + "candidateAdmission": config.candidate_admission, + "command": config.transport.command, + "args": config.transport.args, + "agent": config.agent, + "model": config.model, + "catalogRevision": config.catalog_revision, + "runtimeDirectory": config.runtime_directory, + "normalizedSessionId": config.normalized_session_id, + "workingDirectory": config.working_directory, + "permissionMode": config.permission_mode, + "mode": config.mode, + "piThinkingLevel": config.pi_thinking_level, + "claudeEffort": config.claude_effort, + "grokEffort": config.grok_effort, + "grokDisableWebSearch": config.grok_disable_web_search, + "permissionModePinned": config.permission_mode_pinned, + "providerPolicy": config.provider_policy, + "systemInstructions": config.system_instructions, + "runtimeContext": runtime_context, + "mcpBound": mcp_bound, + "toolSet": config.tool_set, + }); + format!("sha256:{:x}", Sha256::digest(shape.to_string().as_bytes())) +} + +impl AcpxProviderSession { + pub fn generation_has_run_turn(&self) -> bool { + self.generation_has_run_turn + } + + 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(super) fn park_tail_is_quiet(&self) -> bool { + if self.prepared_generation.is_none() { + return self.transport.buffered_events().len() == 0; + } + self.transport.buffered_events().all(|event| { + event.turn_id.is_none() + && matches!( + event.event_type, + GeneratedAcpxSidecarEventType::RuntimeProcess + | GeneratedAcpxSidecarEventType::RuntimeDiagnostic + ) + }) + } + + pub fn can_prepare_next_generation(&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 + } + + pub fn prepare_next_generation(&mut self, deadline: Instant) -> Result<(), LocalRunnerError> { + 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() + { + return Err(self.fail_closed(error)); + } + } + 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(()) + } + + pub fn rebind_run( + &mut self, + next: &AcpxProviderSessionConfig, + ) -> Result { + let Some((prepared, _)) = self.prepared_generation.as_ref() else { + return Ok(false); + }; + if self.closed + || self.failure.is_some() + || self.transport_terminated + || self.runtime_retired + || self.generation_has_run_turn + || self.state.active_turn_id().is_some() + || self.state.has_pending_requests() + || next.run_id == self.config.run_id + || next.expected_identity.as_ref() != Some(&self.identity) + || &generation_digest(next) != prepared + || next.validate().is_err() + { + return Ok(false); + } + let state = AcpxProviderState::new(&next.run_id)?; + let mut tool_bridge = ProviderToolBridge::default(); + tool_bridge + .prepare(next.tool_set.clone()) + .map_err(|error| { + LocalRunnerError::invalid(format!("ACPX authorized tools are invalid: {error}")) + })?; + let reserved_tool_bridge = reserved_terminal_tool_bridge()?; + let previous_run_id = self.config.run_id.clone(); + let attached = match self.transport.request( + GeneratedAcpxSidecarCommand::RunAttach, + json!({ + "runId": next.run_id, + "catalogRevision": next.catalog_revision, + "tools": sidecar_run_tool_operations(&next.tool_set), + }), + ) { + Ok(attached) => attached, + Err(error) => return Err(self.fail_closed(error)), + }; + if attached.get("runId").and_then(Value::as_str) != Some(next.run_id.as_str()) + || attached.get("catalogRevision").and_then(Value::as_u64) + != Some(next.catalog_revision) + { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX sidecar did not confirm the rebound run", + ))); + } + if self.transport.buffered_events().any(|event| { + event.turn_id.is_some() || event.run_id.as_deref() == Some(previous_run_id.as_str()) + }) { + return Err(self.fail_closed(LocalRunnerError::invalid( + "ACPX prepared provider emitted work bound to its previous run", + ))); + } + self.state = state; + self.tool_bridge = tool_bridge; + self.reserved_tool_bridge = reserved_tool_bridge; + self.config = next.clone(); + self.catalog_revision = next.catalog_revision; + self.working_directory = next.working_directory.clone(); + self.prepared_generation = None; + if let Err(error) = self.verify_dormant_session() { + return Err(self.fail_closed(error)); + } + Ok(true) + } +} 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 bb2045e82ba..a81d3db3742 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 @@ -22,6 +22,9 @@ use crate::provider_bridge::{ use crate::question_response::validate_question_response; use crate::stable_identity::{is_stable_id, DURABLE_STABLE_ID_CHARS, SHORT_STABLE_ID_CHARS}; +#[path = "acpx_provider_generation.rs"] +mod generation; + const MAX_ID_CHARS: usize = 240; const MAX_MODEL_CHARS: usize = 240; pub(crate) const MAX_SYSTEM_INSTRUCTIONS_BYTES: usize = 1024 * 1024; @@ -353,6 +356,8 @@ pub struct AcpxProviderSession { rejected_start_resumable: bool, runtime_retired: bool, completed_turn_observed: bool, + generation_has_run_turn: bool, + prepared_generation: Option<(String, u32)>, } pub(crate) fn goal_is_dormant(projection: &Value) -> bool { @@ -411,6 +416,8 @@ impl AcpxProviderSession { rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, + generation_has_run_turn: false, + prepared_generation: None, }) } @@ -506,10 +513,10 @@ impl AcpxProviderSession { || self.failure.is_some() || self.transport_terminated || self.runtime_retired - || !self.completed_turn_observed + || !(self.completed_turn_observed || self.prepared_generation.is_some()) || self.state.active_turn_id().is_some() || self.state.has_pending_requests() - || self.transport.buffered_events().len() != 0 + || !self.park_tail_is_quiet() { return Ok(false); } @@ -529,8 +536,7 @@ impl AcpxProviderSession { return Ok(false); }; // Polling still owns every frame observed during these reads. - Ok(self.validate_dormant_snapshot(&snapshot).is_ok() - && self.transport.buffered_events().len() == 0) + Ok(self.validate_dormant_snapshot(&snapshot).is_ok() && self.park_tail_is_quiet()) } /// Read the same live sidecar before exposing its in-memory pending ledger. @@ -654,9 +660,9 @@ 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.state.has_settled_turns(); + let provider_restarted = self.generation_has_run_turn; if provider_restarted { - if let Err(error) = self.restart_idle_provider() { + if let Err(error) = self.restart_idle_provider(None) { return Err(self.fail_closed(error)); } } @@ -693,6 +699,8 @@ impl AcpxProviderSession { "ACPX reserved tool receipt rotation failed: {error}" )))); } + self.generation_has_run_turn = true; + self.prepared_generation = None; let response = match self.transport.request_turn_start( json!({"turnId":turn_id,"message":message,"performance":performance}), ) { @@ -1307,7 +1315,8 @@ impl AcpxProviderSession { Ok(()) } - fn restart_idle_provider(&mut self) -> Result<(), LocalRunnerError> { + fn restart_idle_provider(&mut self, deadline: Option) -> Result<(), LocalRunnerError> { + self.transport.bound_requests_by(deadline); let suspended = self.transport.request( GeneratedAcpxSidecarCommand::SessionSuspend, json!({"reason":"ACPX turn receipt epoch rotation"}), @@ -1322,6 +1331,7 @@ impl AcpxProviderSession { &restart_config.transport, &restart_config.agent, )?; + replacement.bound_requests_by(deadline); let (replacement_identity, _, turn_controls) = match bootstrap(&mut replacement, &restart_config) { Ok(value) => value, @@ -1340,6 +1350,7 @@ impl AcpxProviderSession { self.transport = replacement; self.turn_controls = turn_controls; self.transport_terminated = false; + self.generation_has_run_turn = false; Ok(()) } @@ -2244,6 +2255,8 @@ mod mode_tests { rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, + generation_has_run_turn: false, + prepared_generation: None, } } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs index c55718e3453..58f537c07a8 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_state.rs @@ -155,10 +155,6 @@ impl AcpxProviderState { self.scope.active_turn_id() } - pub(crate) fn has_settled_turns(&self) -> bool { - self.scope.has_settled_turns() - } - pub(crate) fn validate_new_turn_identity(&self, turn_id: &str) -> Result<(), LocalRunnerError> { self.scope.validate_new_turn_identity(turn_id) } diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs index be77e3be7ce..631a61b3c9b 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_sidecar_transport.rs @@ -85,6 +85,7 @@ pub struct AcpxSidecarTransport { process: SupervisedProcess, request_timeout: Duration, session_open_timeout: Duration, + request_deadline: Option, next_request_id: u64, last_event_sequence: u64, buffered_events: VecDeque, @@ -389,6 +390,7 @@ impl AcpxSidecarTransport { process, request_timeout: config.request_timeout, session_open_timeout: config.request_timeout, + request_deadline: None, next_request_id: 1, last_event_sequence: 0, buffered_events: VecDeque::new(), @@ -407,6 +409,10 @@ impl AcpxSidecarTransport { } } + pub(crate) fn bound_requests_by(&mut self, deadline: Option) { + self.request_deadline = deadline; + } + pub fn process_id(&self) -> u32 { self.process.id() } @@ -508,13 +514,13 @@ impl AcpxSidecarTransport { command: GeneratedAcpxSidecarCommand, params: Value, ) -> Result { + let timeout_at = Instant::now() + self.command_timeout(command); + let deadline = self + .request_deadline + .map_or(timeout_at, |bound| bound.min(timeout_at)); let request_id = self.send_request(command, params)?; - self.receive_response( - command, - request_id, - Instant::now() + self.command_timeout(command), - )? - .ok_or_else(|| self.request_timeout_error(command)) + self.receive_response(command, request_id, deadline)? + .ok_or_else(|| self.request_timeout_error(command)) } /// A park observation may expire without revoking a healthy provider. 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 49b2d151dc9..76a6a309f09 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 @@ -74,11 +74,27 @@ fn run() -> Result<(), Box> { .windows(2) .find(|pair| pair[0] == "--journal") .ok_or("source journal is missing")?; + let live_predecessors = std::fs::read_to_string(&path[1]) + .unwrap_or_default() + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter_map(|line| line["processStart"].as_u64()) + .filter(|pid| { + std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .stderr(std::process::Stdio::null()) + .status() + .is_ok_and(|status| status.success()) + }) + .collect::>(); let mut file = std::fs::OpenOptions::new() .create(true) .append(true) .open(&path[1])?; - write_json(&mut file, &json!({"processStart": std::process::id()}))?; + write_json( + &mut file, + &json!({"processStart": std::process::id(), "livePredecessors": live_predecessors}), + )?; if args.iter().any(|arg| arg == "--spawn-owned-child") { let child = std::process::Command::new("/bin/sleep") .arg("120") @@ -224,7 +240,45 @@ fn run() -> Result<(), Box> { response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false, "error":{"code":"temporarily_unavailable","message":"Optional read is temporarily unavailable.","retryable":true}}); } + if command == "session.open" && fault == "slow-open" { + std::thread::sleep(std::time::Duration::from_secs(10)); + } + if command == "session.open" && fault == "reject-open" { + response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false, + "error":{"code":"session_open_failed","message":"The replacement provider could not open.","retryable":false}}); + } write_json(&mut stdout, &response)?; + if command == "session.open" && args.iter().any(|arg| arg == "--emit-process") { + write_json( + &mut stdout, + &json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":next_sequence,"eventType":"runtime.process","runId":null,"turnId":null, + "payload":{"role":"sidecar","pid":std::process::id(),"processGroupId":null,"startedAt":"2026-01-01T00:00:00.000Z"}}), + )?; + next_sequence += 1; + } + if command == "session.suspend" && fault == "late-tool-after-suspend" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + &source_run, + "late-turn", + json!({"callId":"late-call","operationId":"issues.delete","input":{"source":"late-old-generation"}}), + )?; + next_sequence += 1; + } + if command == "run.attach" && fault == "pre-turn-tool" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.tool_called", + &source_run, + "pre-turn", + json!({"callId":"pre-turn-call","operationId":"issues.read","input":{"id":"issue-1"}}), + )?; + next_sequence += 1; + } if command == "turn.start" { let turn = request["params"]["turnId"].as_str().unwrap(); if args.iter().any(|arg| arg == "--wait-completion-feedback") { 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 fb0040aec08..5a45df27369 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 @@ -47,6 +47,7 @@ enum CommandLifecycle { } const TERMINAL_RESULT_ACK_TIMEOUT: Duration = Duration::from_secs(2); +const HOST_PARK_PREPARATION_BUDGET: Duration = Duration::from_secs(20); const CUMULATIVE_ACK_PERSIST_INTERVAL: usize = 16; fn reconnect_delay(base: Duration, max_delay: Duration, attempt: &mut u32) -> Duration { @@ -318,6 +319,10 @@ pub trait CommandExecutor { Ok(false) } + fn prepare_for_park(&mut self, _deadline: Instant) -> Result { + Ok(false) + } + fn execute(&mut self, command: &Command) -> Result; fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> { @@ -1451,6 +1456,20 @@ fn process_host_control( ) && executor.host_park_ready()? => { + let ready = + match executor.prepare_for_park(Instant::now() + HOST_PARK_PREPARATION_BUDGET) { + Ok(false) => true, + Ok(true) => executor.host_park_ready()?, + Err(error) => { + state.record_diagnostic(format!("host park preparation failed: {error}")); + store.save(state)?; + false + } + }; + if !ready { + control.channel.reply(&request, "conflict")?; + return Ok(HostAction::None); + } control.park_id = Some(id.clone()); control.channel.reply(&request, "parked")?; Ok(HostAction::Park) @@ -3216,6 +3235,136 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + #[cfg(unix)] + fn hosted_park_reproves_a_prepared_provider_before_replying() { + use std::io::{BufRead, BufReader, Write}; + use std::os::unix::net::UnixStream; + struct PreparingExecutor { + calls: Vec<&'static str>, + prepared: Option, + ready_after_prepare: bool, + prepared_once: bool, + } + impl CommandExecutor for PreparingExecutor { + fn execute(&mut self, _: &Command) -> Result { + panic!("private control cannot execute PRP commands"); + } + fn host_park_ready(&mut self) -> Result { + self.calls.push("proof"); + Ok(!self.prepared_once || self.ready_after_prepare) + } + fn prepare_for_park(&mut self, _deadline: Instant) -> Result { + self.calls.push("prepare"); + self.prepared_once = true; + self.prepared + .ok_or_else(|| DurableRunnerError::invalid("replacement provider failed")) + } + fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + self.calls.push("shutdown"); + Ok(()) + } + } + for (prepared, ready_after_prepare, status, calls) in [ + ( + Some(true), + true, + "parked", + vec!["proof", "prepare", "proof"], + ), + ( + Some(true), + false, + "conflict", + vec!["proof", "prepare", "proof"], + ), + (None, true, "conflict", vec!["proof", "prepare"]), + (Some(false), false, "parked", vec!["proof", "prepare"]), + ] { + let directory = std::env::temp_dir() + .join(format!("paperclip-host-prepare-{}", uuid::Uuid::new_v4())); + let config = config(directory.clone()); + let store = DurableStateStore::new(&config.state_dir).unwrap(); + let mut state = DurableState::new(&config); + let (socket, mut host) = UnixStream::pair().unwrap(); + let mut reader = BufReader::new(host.try_clone().unwrap()); + let mut control = HostedRunnerControl::new(HostControl::new(socket).unwrap()); + let mut executor = PreparingExecutor { + calls: Vec::new(), + prepared, + ready_after_prepare, + prepared_once: false, + }; + let identity = json!({ + "runnerInstanceId":state.runner_instance_id,"environmentLeaseId":state.environment_lease_id, + "runId":state.run_id,"normalizedSessionId":state.normalized_session_id,"turnId":state.turn_id,"itemId":state.item_id, + }); + let mut exchange = |request: Value, + state: &mut DurableState, + control: &mut HostedRunnerControl, + executor: &mut PreparingExecutor, + parked: bool| { + writeln!(host, "{request}").unwrap(); + let action = process_host_control( + control, + state, + &store, + executor, + parked, + Some(Duration::from_secs(1)), + ) + .unwrap(); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + (action, serde_json::from_str::(&line).unwrap()) + }; + let (action, reply) = exchange( + json!({"op":"park","id":"park-1","identity":identity, + "lastControllerCommandSeq":0,"ackedSourceSeq":0}), + &mut state, + &mut control, + &mut executor, + false, + ); + assert_eq!(reply["status"], status); + assert_eq!(executor.calls, calls); + assert_eq!(matches!(action, HostAction::Park), status == "parked"); + assert_eq!(control.park_id.is_some(), status == "parked"); + assert_eq!( + state + .diagnostics + .iter() + .any(|message| message.contains("host park preparation failed")), + prepared.is_none() + ); + if prepared == Some(true) && status == "parked" { + let (action, _) = exchange( + json!({"op":"claim","id":"claim-1","parkId":"park-1","bootstrapTicket":"fresh-private-ticket"}), + &mut state, + &mut control, + &mut executor, + true, + ); + assert!(matches!(action, HostAction::Claim(_))); + let (action, reply) = exchange( + json!({"op":"retire","id":"retire-claim","parkId":"park-1","claimId":"claim-1", + "identity":identity,"lastControllerCommandSeq":0,"ackedSourceSeq":0}), + &mut state, + &mut control, + &mut executor, + false, + ); + assert_eq!(reply["status"], "retired"); + assert!(matches!(action, HostAction::Retire)); + assert_eq!( + executor.calls, + ["proof", "prepare", "proof", "proof", "shutdown"] + ); + } + fs::remove_dir_all(directory).unwrap(); + } + } + #[test] #[cfg(unix)] fn hosted_park_blocks_without_provider_polls_and_retires_on_eof_or_lifetime() { diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs index 6f25a381c8c..5a55c968543 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/tests/hosted_retention.rs @@ -10,6 +10,7 @@ use std::os::unix::net::UnixStream; struct RetentionExecutor { polls: Arc, shutdowns: Arc, + prepares: Arc, } impl CommandExecutor for RetentionExecutor { fn execute(&mut self, _: &Command) -> Result { @@ -18,6 +19,10 @@ impl CommandExecutor for RetentionExecutor { fn host_park_ready(&mut self) -> Result { Ok(true) } + fn prepare_for_park(&mut self, _deadline: std::time::Instant) -> Result { + self.prepares.fetch_add(1, Ordering::SeqCst); + Ok(true) + } fn poll_events(&mut self) -> Result, DurableRunnerError> { self.polls.fetch_add(1, Ordering::SeqCst); Ok(vec![]) @@ -174,9 +179,11 @@ fn exercise_hosted_loop(blocked_work: Option) { }); let polls = Arc::new(AtomicUsize::new(0)); let shutdowns = Arc::new(AtomicUsize::new(0)); + let prepares = Arc::new(AtomicUsize::new(0)); let executor = RetentionExecutor { polls: polls.clone(), shutdowns: shutdowns.clone(), + prepares: prepares.clone(), }; let (socket, mut host) = UnixStream::pair().unwrap(); host.set_read_timeout(Some(Duration::from_secs(5))).unwrap(); @@ -201,6 +208,7 @@ fn exercise_hosted_loop(blocked_work: Option) { "parked" ); let parked_polls = polls.load(Ordering::SeqCst); + assert_eq!(prepares.load(Ordering::SeqCst), 1); // Outlast the ordinary reconnect grace without any polling or shutdown. thread::sleep(Duration::from_millis(80)); assert_eq!(polls.load(Ordering::SeqCst), parked_polls); @@ -245,6 +253,7 @@ fn exercise_hosted_loop(blocked_work: Option) { finish_tx.send(()).unwrap(); server.join().unwrap(); assert_eq!(shutdowns.load(Ordering::SeqCst), 1); + assert_eq!(prepares.load(Ordering::SeqCst), 1); assert_eq!( DurableStateStore::new(&directory) .unwrap() 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 7db36c3ea39..f1f593634d9 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 @@ -337,6 +337,13 @@ impl CommandExecutor for NativeProviderCommandExecutor { } } + fn prepare_for_park(&mut self, deadline: Instant) -> Result { + match self.selected.as_mut() { + Some(SelectedExecutor::Acpx(executor)) => executor.prepare_for_park(deadline), + _ => Ok(false), + } + } + 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 new file mode 100644 index 00000000000..0f7fcf96fc7 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_preparation.rs @@ -0,0 +1,594 @@ +#![cfg(unix)] + +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::PathBuf; +use std::time::{Duration, Instant}; + +use paperclip_runner_core::durable::{ + AcpxLaunchProfile, Command, CommandExecution, CommandExecutor, DurableRunnerConfig, + QualifiedLaunchArtifact, +}; +use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; +use paperclip_runner_core::provider_bridge::{authorized_tool_catalog_digest, AuthorizedTool}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +struct Fixture(PathBuf); +impl Drop for Fixture { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +impl Fixture { + fn fault(&self, fault: &str) { + fs::write(self.0.join("fault"), fault).unwrap(); + } + + fn journal(&self) -> Vec { + fs::read_to_string(self.0.join("sidecar.jsonl")) + .unwrap() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() + } + + fn process_starts(&self) -> Vec { + self.journal() + .into_iter() + .filter(|line| line.get("processStart").is_some()) + .collect() + } + + fn state(&self) -> Value { + serde_json::from_slice(&fs::read(self.0.join("acpx-provider-state.json")).unwrap()).unwrap() + } +} + +struct Harness { + fixture: Fixture, + config: DurableRunnerConfig, + executor: NativeProviderCommandExecutor, + descriptor: Value, + opened: Value, + sequence: u64, +} + +impl Harness { + fn new() -> Self { + let catalog: Value = + serde_json::from_str(include_str!("../../../../acpx-profiles.json")).unwrap(); + let profile = &catalog["profiles"]["claude"]; + let digest = profile["commandDigest"].as_str().unwrap(); + let fixture = + Fixture(std::env::temp_dir().join(format!("acpx-park-prep-{}", uuid::Uuid::new_v4()))); + fs::create_dir_all(&fixture.0).unwrap(); + fs::set_permissions(&fixture.0, fs::Permissions::from_mode(0o700)).unwrap(); + let native = fixture.0.join("native-sidecar"); + fs::copy(env!("CARGO_BIN_EXE_fake-acpx-sidecar"), &native).unwrap(); + fs::set_permissions(&native, fs::Permissions::from_mode(0o700)).unwrap(); + let sidecar = fixture.0.join("sidecar"); + fs::write( + &sidecar, + format!( + "#!/bin/sh\nexec '{}' --mode source-reconcile --emit-process --journal '{}' --source-fault '{}' \"$@\"\n", + native.display(), + fixture.0.join("sidecar.jsonl").display(), + fixture.0.join("fault").display() + ), + ) + .unwrap(); + fs::set_permissions(&sidecar, fs::Permissions::from_mode(0o700)).unwrap(); + let args = vec!["--profile-digest".to_owned(), digest.to_owned()]; + 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": "claude", "model": "explicit-model", + "acpxVersion": "0.13.1", "agentServerPackage": profile["agentServerPackage"], + "agentServerVersion": profile["agentServerVersion"], "agentRuntimePackage": profile["agentRuntimePackage"], + "agentRuntimeVersion": profile["agentRuntimeVersion"], "commandDigest": 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, + "runtimeContext": { + "aggregateDigest": "a".repeat(64), + "skills": [{"key": "skill-1", "bundle": {"digest": "d".repeat(64), "rootPath": "/skills/one"}}], + "mcp": {"assignmentSetId": "assignment-1", "digest": "e".repeat(64), "bindingId": "binding-run-1"}, + }, + }); + let mut executor = NativeProviderCommandExecutor::with_runner_config(&fixture.0, &config); + executor + .execute(&command(1, "run.prepare", json!({"provider": descriptor}))) + .unwrap(); + let opened = executor + .execute(&command(2, "session.open", json!({}))) + .unwrap() + .result; + let mut harness = Self { + fixture, + config, + executor, + descriptor, + opened, + sequence: 2, + }; + harness.drain(); + harness + } + + fn execute(&mut self, kind: &str, payload: Value) -> CommandExecution { + self.sequence += 1; + self.executor + .execute(&command(self.sequence, kind, payload)) + .unwrap() + } + + fn drain(&mut self) -> Vec { + let mut seen = Vec::new(); + loop { + let events = self.executor.poll_events().unwrap(); + if events.is_empty() { + return seen; + } + seen.extend( + events + .iter() + .map(|event| json!({"type":event.event_type,"payload":event.payload})), + ); + self.executor.acknowledge_events(events.len()).unwrap(); + } + } + + fn complete_turn(&mut self, turn: &str) -> Vec { + let started = self.execute( + "turn.start", + json!({"turnId": turn, "text": "Complete the work."}), + ); + let deadline = Instant::now() + Duration::from_secs(5); + let mut seen = started + .events + .iter() + .map(|(kind, _, payload)| json!({"type":kind,"payload":payload})) + .collect::>(); + while !seen + .iter() + .any(|event: &Value| event["type"] == "turn.completed") + { + assert!(Instant::now() < deadline, "no completed provider turn"); + seen.extend(self.drain()); + } + seen.extend(self.drain()); + seen + } + + fn attach_payload(&self, run: &str, mutate: impl FnOnce(&mut Value)) -> Value { + let mut descriptor = self.descriptor.clone(); + descriptor["runId"] = json!(run); + descriptor["runtimeContext"]["mcp"]["bindingId"] = json!(format!("binding-{run}")); + descriptor["runtimeContext"]["aggregateDigest"] = json!(format!("{run}-aggregate")); + let mut payload = json!({ + "provider": descriptor, + "paperclipNextAuthority": {"identity": { + "runId": run, "normalizedSessionId": self.config.normalized_session_id, + }}, + }); + mutate(&mut payload); + payload + } + + fn attach( + &mut self, + run: &str, + mutate: impl FnOnce(&mut Value), + ) -> (CommandExecution, Duration) { + let payload = self.attach_payload(run, mutate); + let started = Instant::now(); + let attached = self.execute("run.attach", payload); + let elapsed = started.elapsed(); + self.config.run_id = run.to_owned(); + self.executor.rotate_authority(&self.config); + (attached, elapsed) + } + + fn quiesce(&mut self) { + let ready = self.execute("session.snapshot", json!({"quiesceForWarmAttach": true})); + assert_eq!(self.drain(), Vec::::new()); + assert_eq!(ready.result["warmAttachReady"], true); + } +} + +fn command(sequence: u64, kind: &str, payload: Value) -> Command { + Command { + schema: "paperclip.prp.command.v1".to_owned(), + command_id: format!("command-{sequence}"), + controller_seq: sequence, + command_type: kind.to_owned(), + issued_at: "2026-09-21T00:00:00.000Z".to_owned(), + deadline_at: None, + precondition: None, + payload, + } +} + +fn alive(pid: &Value) -> bool { + std::process::Command::new("kill") + .args(["-0", &pid.to_string()]) + .stderr(std::process::Stdio::null()) + .status() + .unwrap() + .success() +} + +fn commands_after(journal: &[Value], start: usize) -> Vec { + journal[start..] + .iter() + .filter_map(|line| line["command"].as_str().map(str::to_owned)) + .collect() +} + +fn last_process_start(journal: &[Value]) -> usize { + journal + .iter() + .rposition(|line| line.get("processStart").is_some()) + .unwrap() +} + +fn prepared_park(harness: &mut Harness) -> Value { + harness.complete_turn("first"); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.process_starts().last().unwrap()["processStart"].clone() +} + +#[test] +fn park_reaps_the_old_generation_before_spawning_and_proves_the_new_one_dormant() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + let before = harness.fixture.journal().len(); + assert!(harness.executor.host_park_ready().unwrap()); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert!(spawned > before); + assert_eq!( + commands_after(&journal[..spawned], before), + ["session.goal.get", "session.snapshot", "session.suspend"] + ); + assert_eq!(journal[spawned]["livePredecessors"], json!([])); + assert_ne!(journal[spawned]["processStart"], old_pid); + assert_eq!( + commands_after(&journal, spawned), + [ + "initialize", + "session.open", + "run.attach", + "session.goal.get", + "session.snapshot" + ] + ); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-1"); + assert!(!alive(&old_pid)); + assert!(alive(&journal[spawned]["processStart"])); + assert!( + harness.executor.host_park_ready().unwrap(), + "a prepared dormant generation is park-ready" + ); + assert_eq!(harness.drain(), Vec::::new()); + assert_eq!(harness.fixture.state()["lifecycle"], "session_open"); + assert!( + !harness.executor.prepare_for_park(park_deadline()).unwrap(), + "a prepared generation is not prepared twice" + ); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); + assert!(!alive(&journal[spawned]["processStart"])); +} + +#[test] +fn compatible_attach_keeps_the_prepared_provider_and_rebinds_its_run() { + let mut cold = Harness::new(); + cold.complete_turn("first"); + cold.quiesce(); + let (_, cold_attach) = cold.attach("run-2", |_| {}); + cold.executor.shutdown().unwrap(); + + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + let before = harness.fixture.journal().len(); + let (attached, warm_attach) = harness.attach("run-2", |_| {}); + eprintln!("run.attach cold={cold_attach:?} prepared={warm_attach:?}"); + assert_eq!(attached.result["status"], "resumed"); + assert_eq!(attached.result["processId"], prepared_pid); + assert_eq!( + attached.result["providerSessionId"], + harness.opened["providerSessionId"] + ); + let journal = harness.fixture.journal(); + assert_eq!(harness.fixture.process_starts().len(), 2); + assert_eq!( + commands_after(&journal, before), + [ + "run.attach", + "session.goal.get", + "session.snapshot", + "session.goal.get" + ] + ); + assert_eq!(journal[before]["params"]["runId"], "run-2"); + let state = harness.fixture.state(); + assert_eq!(state["descriptor"]["runId"], "run-2"); + assert_eq!( + state["descriptor"]["runtimeContext"]["mcp"]["bindingId"], + "binding-run-2" + ); + let events = harness.complete_turn("second"); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == prepared_pid)); + assert_eq!( + harness.fixture.process_starts().len(), + 2, + "the first turn of the rebound run reuses the prepared provider" + ); + let snapshot = harness.execute("session.snapshot", json!({})); + assert_eq!(snapshot.result["warmAttachReady"], true); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn changed_tools_or_skills_take_the_cold_path() { + let operations = vec![AuthorizedTool { + operation_id: "issues.read".to_owned(), + version: 1, + description: "Read an issue.".to_owned(), + input_schema: json!({"type":"object"}), + response_schema: json!({"type":"object"}), + }]; + let tools = json!({ + "schema": "paperclip.runner.authorized-tools.v1", "schemaVersion": 1, + "catalogDigest": authorized_tool_catalog_digest(&operations).unwrap(), + "operations": operations, + }); + let changes: [(&str, Box); 2] = [ + ( + "tools", + Box::new(move |payload| payload["authorizedTools"] = tools.clone()), + ), + ( + "skills", + Box::new(|payload| { + payload["provider"]["runtimeContext"]["skills"][0]["bundle"]["rootPath"] = + json!("/skills/two") + }), + ), + ]; + for (name, change) in changes { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + let (attached, _) = harness.attach("run-2", |payload| change(payload)); + assert_eq!(attached.result["status"], "resumed", "{name}"); + assert_ne!(attached.result["processId"], prepared_pid, "{name}"); + assert!(!alive(&prepared_pid), "{name}"); + assert_eq!(harness.fixture.process_starts().len(), 3, "{name}"); + let journal = harness.fixture.journal(); + let spawned = last_process_start(&journal); + assert_eq!(journal[spawned]["livePredecessors"], json!([]), "{name}"); + assert_eq!(journal[spawned + 3]["params"]["runId"], "run-2", "{name}"); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); + } +} + +#[test] +fn a_generation_that_ran_a_turn_is_never_rebound() { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + harness.attach("run-2", |_| {}); + harness.complete_turn("second"); + harness.quiesce(); + let (attached, _) = harness.attach("run-3", |_| {}); + assert_ne!(attached.result["processId"], prepared_pid); + assert!(!alive(&prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn late_and_pre_turn_callbacks_never_reach_the_next_run() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("late-tool-after-suspend"); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.fault(""); + let (attached, _) = harness.attach("run-2", |_| {}); + let events = harness.complete_turn("second"); + assert!(!format!("{events:?}").contains("late-call")); + assert!(!format!("{events:?}").contains("late-old-generation")); + let reused_pid = attached.result["processId"].clone(); + assert_eq!(harness.fixture.process_starts().len(), 2); + harness.executor.shutdown().unwrap(); + assert!(!alive(&reused_pid)); + + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("pre-turn-tool"); + assert!(harness.executor.prepare_for_park(park_deadline()).unwrap()); + harness.fixture.fault(""); + let prepared_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!( + !harness.executor.host_park_ready().unwrap(), + "a turn-bound callback before any turn blocks the park" + ); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_ne!(attached.result["processId"], prepared_pid); + assert!(!alive(&prepared_pid)); + let events = harness.complete_turn("second"); + assert!(!format!("{events:?}").contains("pre-turn-call")); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn failed_preparation_conflicts_and_leaves_a_cold_recoverable_session() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + let error = harness + .executor + .prepare_for_park(park_deadline()) + .unwrap_err(); + assert!( + error.to_string().contains("park preparation failed"), + "{error}" + ); + harness.fixture.fault(""); + assert!(!alive(&old_pid)); + let failed_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(!alive(&failed_pid)); + let state = harness.fixture.state(); + assert_eq!(state["lifecycle"], "suspended"); + assert_eq!(state["providerExitUnconfirmed"], false); + assert!(state["identity"].is_object()); + assert!(!harness.executor.host_park_ready().unwrap()); + assert!(!harness.executor.prepare_for_park(park_deadline()).unwrap()); + assert_eq!(harness.drain(), Vec::::new()); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["status"], "resumed"); + assert_eq!( + attached.result["providerSessionId"], + harness.opened["providerSessionId"] + ); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn a_slow_replacement_open_fails_at_the_park_deadline_and_leaves_the_runner_responsive() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + harness.complete_turn("first"); + harness.fixture.fault("slow-open"); + let started = Instant::now(); + let error = harness + .executor + .prepare_for_park(Instant::now() + Duration::from_secs(1)) + .unwrap_err(); + let elapsed = started.elapsed(); + harness.fixture.fault(""); + assert!( + error.to_string().contains("timed out at session.open"), + "{error}" + ); + assert!( + elapsed < Duration::from_secs(5), + "park preparation took {elapsed:?}" + ); + assert!(!alive(&old_pid)); + let slow_pid = harness.fixture.process_starts().last().unwrap()["processStart"].clone(); + assert!(!alive(&slow_pid)); + let state = harness.fixture.state(); + assert_eq!(state["lifecycle"], "suspended"); + assert_eq!(state["providerExitUnconfirmed"], false); + assert!(!harness.executor.host_park_ready().unwrap()); + let (attached, _) = harness.attach("run-2", |_| {}); + assert_eq!(attached.result["status"], "resumed"); + harness.complete_turn("second"); + harness.executor.shutdown().unwrap(); +} + +#[test] +fn failed_preparation_recovers_the_same_run_on_its_next_turn() { + let mut harness = Harness::new(); + harness.complete_turn("first"); + harness.fixture.fault("reject-open"); + assert!(harness.executor.prepare_for_park(park_deadline()).is_err()); + harness.fixture.fault(""); + 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.shutdown().unwrap(); +} + +#[test] +fn prepared_park_retires_through_ordinary_shutdown() { + let mut harness = Harness::new(); + let prepared_pid = prepared_park(&mut harness); + assert!(harness.executor.host_park_ready().unwrap()); + harness.executor.shutdown().unwrap(); + assert!(!alive(&prepared_pid)); + let journal = harness.fixture.journal(); + assert_eq!( + journal.last().unwrap()["command"], + "session.close", + "retirement closes the prepared provider" + ); + assert_eq!(harness.fixture.state()["descriptor"]["runId"], "run-1"); +} + +#[test] +fn a_conflicted_park_keeps_the_prepared_provider_for_the_same_run() { + let mut harness = Harness::new(); + let old_pid = harness.opened["processId"].clone(); + let prepared_pid = prepared_park(&mut harness); + let events = harness.complete_turn("second"); + let reconciled = events + .iter() + .find(|event| event["type"] == "session.reconciled") + .expect("the turn reports the replaced provider process"); + assert_eq!(reconciled["payload"]["previousProcessId"], old_pid); + assert_eq!(reconciled["payload"]["processId"], prepared_pid); + assert!(events + .iter() + .any(|event| event["type"] == "harness.diagnostic" + && event["payload"]["details"]["pid"] == prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 2); + let events = harness.complete_turn("third"); + assert!(events + .iter() + .any(|event| event["type"] == "session.reconciled" + && event["payload"]["previousProcessId"] == prepared_pid)); + assert_eq!(harness.fixture.process_starts().len(), 3); + harness.executor.shutdown().unwrap(); +} + +fn park_deadline() -> Instant { + Instant::now() + Duration::from_secs(20) +} 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 9303af23960..4b56a5abc4c 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 @@ -266,6 +266,13 @@ fn hosted_park_acpx_requires_completion_in_this_lifetime_and_exact_authority() { .unwrap(); settle(&mut executor); assert!(executor.host_park_ready().unwrap()); + assert!(executor + .prepare_for_park(Instant::now() + Duration::from_secs(20)) + .unwrap()); + assert!( + executor.host_park_ready().unwrap(), + "a prepared dormant generation inherits its park proof" + ); config.run_id = "other-run".to_owned(); executor.rotate_authority(&config); assert!(!executor.host_park_ready().unwrap()); @@ -467,5 +474,8 @@ fn hosted_park_acpx_does_not_enable_an_unqualified_park_provider() { .unwrap(); settle(&mut executor); assert!(!executor.host_park_ready().unwrap()); + assert!(!executor + .prepare_for_park(Instant::now() + Duration::from_secs(20)) + .unwrap()); 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 e4485a9f7bb..1294ea1adf5 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 @@ -1,5 +1,5 @@ use std::path::PathBuf; -use std::time::Duration; +use std::time::{Duration, Instant}; use paperclip_runner_core::acpx_provider_session::{ AcpxPermissionMode, AcpxProviderSession, AcpxProviderSessionConfig, @@ -791,3 +791,146 @@ fn forced_cancellation_retirement_preserves_terminal_polling_but_rejects_live_re session.shutdown("test complete").unwrap(); } } + +fn reusable_config(directory: &std::path::Path) -> AcpxProviderSessionConfig { + let mut config = config("source-reconcile"); + config.transport.args = vec![ + "--mode".to_owned(), + "source-reconcile".to_owned(), + "--journal".to_owned(), + directory.join("journal.jsonl").display().to_string(), + "--source-fault".to_owned(), + directory.join("fault").display().to_string(), + ]; + config.runtime_context = json!({ + "aggregateDigest": "run-1-aggregate", + "skills": [{"key": "skill-1", "bundle": {"rootPath": "/skills/one"}}], + "mcp": {"bindingId": "binding-run-1"}, + }); + config +} + +fn settle_turn(session: &mut AcpxProviderSession, turn_id: &str) -> Vec { + session + .start_turn(turn_id, "Please continue", &std::env::temp_dir()) + .unwrap(); + let mut seen = Vec::new(); + loop { + let events = session.poll_event(Duration::from_secs(1)).unwrap().unwrap(); + let settled = events.iter().any(|event| { + matches!(event, AcpxProviderStateEvent::TurnTerminal { turn_id: settled, .. } if settled == turn_id) + }); + seen.extend(events); + if settled { + return seen; + } + } +} + +fn process_starts(directory: &std::path::Path) -> usize { + std::fs::read_to_string(directory.join("journal.jsonl")) + .unwrap() + .lines() + .filter(|line| line.contains("processStart")) + .count() +} + +#[test] +fn prepared_generation_rebinds_only_an_identical_unused_configuration() { + let directory = std::env::temp_dir().join(format!("acpx-generation-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&directory).unwrap(); + let config = reusable_config(&directory); + let mut session = AcpxProviderSession::start(&config).unwrap(); + assert!(!session.can_prepare_next_generation()); + assert!(session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .is_err()); + settle_turn(&mut session, "turn-1"); + let first_pid = session.process_id(); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + let prepared_pid = session.process_id(); + assert_ne!(prepared_pid, first_pid); + assert!(session.awaits_run_rebind() && !session.generation_has_run_turn()); + let mut next = config.clone(); + next.run_id = "run-2".to_owned(); + next.expected_identity = Some(session.identity().clone()); + next.runtime_context["mcp"]["bindingId"] = json!("binding-run-2"); + next.runtime_context["aggregateDigest"] = json!("run-2-aggregate"); + let changes: [fn(&mut AcpxProviderSessionConfig); 7] = [ + |config| config.system_instructions.push_str(" Changed."), + |config| config.tool_set.operations.clear(), + |config| config.runtime_context["skills"][0]["bundle"]["rootPath"] = json!("/skills/two"), + |config| config.runtime_context["mcp"]["bindingId"] = serde_json::Value::Null, + |config| config.permission_mode = AcpxPermissionMode::DenyAll, + |config| config.run_id = "run-1".to_owned(), + |config| config.expected_identity = None, + ]; + for change in changes { + let mut changed = next.clone(); + change(&mut changed); + assert!(!session.rebind_run(&changed).unwrap()); + assert_eq!(session.process_id(), prepared_pid); + assert!(session.awaits_run_rebind()); + } + assert!(session.rebind_run(&next).unwrap()); + assert_eq!(session.process_id(), prepared_pid); + assert_eq!(session.state().run_id(), "run-2"); + assert!(!session.awaits_run_rebind()); + let attached_runs = std::fs::read_to_string(directory.join("journal.jsonl")) + .unwrap() + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|line| line["command"] == "run.attach") + .map(|line| line["params"]["runId"].clone()) + .collect::>(); + assert_eq!( + attached_runs, + [json!("run-1"), json!("run-1"), json!("run-2")] + ); + settle_turn(&mut session, "turn-1"); + assert_eq!(session.process_id(), prepared_pid); + assert!(session.generation_has_run_turn()); + let mut later = next.clone(); + later.run_id = "run-3".to_owned(); + assert!( + !session.rebind_run(&later).unwrap(), + "a used generation is never rebound" + ); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + settle_turn(&mut session, "turn-2"); + assert!( + !session.rebind_run(&later).unwrap(), + "a prepared generation that ran a turn is never rebound" + ); + assert_eq!(process_starts(&directory), 3); + session.shutdown("test complete").unwrap(); + std::fs::remove_dir_all(directory).unwrap(); +} + +#[test] +fn prepared_generation_never_admits_the_reaped_generation_late_callback() { + let directory = std::env::temp_dir().join(format!("acpx-generation-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(&directory).unwrap(); + let mut session = AcpxProviderSession::start(&reusable_config(&directory)).unwrap(); + settle_turn(&mut session, "turn-1"); + std::fs::write(directory.join("fault"), "late-tool-after-suspend").unwrap(); + session + .prepare_next_generation(Instant::now() + Duration::from_secs(20)) + .unwrap(); + std::fs::write(directory.join("fault"), "").unwrap(); + let events = settle_turn(&mut session, "turn-2"); + assert!(!events + .iter() + .any(|event| matches!(event, AcpxProviderStateEvent::ToolCall { .. }))); + assert_eq!( + process_starts(&directory), + 2, + "the prepared turn does not restart again" + ); + session.shutdown("test complete").unwrap(); + std::fs::remove_dir_all(directory).unwrap(); +} diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts index db33af5ffd6..784b98e8c26 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts @@ -1404,6 +1404,50 @@ describe("qualified ACPX runtime sidecar", () => { expect(attachedRunId).toBe("run-1"); }); + it("rebinds a prepared sidecar to another run only before its first turn", async () => { + const root = mkdtempSync(join(tmpdir(), "acpx-run-rebind-")); + const loader = join(root, "provider-loader.mjs"), host = join(root, "rebind-host.mjs"); + const baseHost = new URL("../test-support/acpx-blocked-output-host.mjs", import.meta.url).href; + writeFileSync(host, `import { AcpxRuntimeHost as BaseHost } from ${JSON.stringify(baseHost)}; + export class AcpxRuntimeHost extends BaseHost { + static async open(options) { const host = await BaseHost.open(options); Object.setPrototypeOf(host, this.prototype); return host; } + startTurn({ requestId }) { return { requestId, promptStarted: Promise.resolve(), + result: Promise.resolve({ status: "completed" }), events: (async function* () {})() }; } + }`); + writeFileSync(loader, `import { registerHooks } from "node:module"; + import { pathToFileURL } from "node:url"; + registerHooks({ resolve(specifier, context, next) { + if (context.parentURL?.includes("/acpx-runtime-sidecar.") && specifier.endsWith("/runtime-host.js")) + return { url: pathToFileURL(${JSON.stringify(host)}).href, shortCircuit: true }; + return next(specifier, context); + } });`); + const sidecar = new SidecarProcess({loader, environment:{PAPERCLIP_TEST_OUTPUT_ROOT:root}}); + children.add(sidecar); + const send = (id:number, command:string, params:Record) => sidecar.write({protocolVersion:ACPX_SIDECAR_PROTOCOL_VERSION,id,command,params}); + try { + sidecar.write(initializeRequest(1,"codex")); + expect(await sidecar.next(frame=>frame.id === 1)).toMatchObject({ok:true}); + send(2,"session.open",{agent:"codex",model:"gpt-5.6-sol",runtimeDirectory:root,workingDirectory:root, + normalizedSessionId:"rebind-session",permissionMode:"deny-all",systemInstructions:"Complete the local fixture turn.",tools:[]}); + expect(await sidecar.next(frame=>frame.id === 2)).toMatchObject({ok:true}); + send(3,"run.attach",{runId:"parked-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 3)).toMatchObject({ok:true,result:{runId:"parked-run"}}); + send(4,"run.attach",{runId:"next-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 4)).toMatchObject({ok:true,result:{runId:"next-run"}}); + send(5,"session.snapshot",{}); + expect(await sidecar.next(frame=>frame.id === 5)).toMatchObject({ok:true,result:{runId:"next-run",turnId:null}}); + send(6,"turn.start",{turnId:"next-turn",message:"Complete the fixture turn."}); + expect(await sidecar.next(frame=>frame.id === 6)).toMatchObject({ok:true}); + expect(await sidecar.next(frame=>frame.eventType === "runtime.turn_terminal")) + .toMatchObject({runId:"next-run",turnId:"next-turn",payload:{status:"completed"}}); + send(7,"run.attach",{runId:"later-run",catalogRevision:1,tools:[]}); + expect(await sidecar.next(frame=>frame.id === 7)).toMatchObject({ok:false, + error:{message:"cannot attach another run after this sidecar started a turn"}}); + send(8,"session.snapshot",{}); + expect(await sidecar.next(frame=>frame.id === 8)).toMatchObject({ok:true,result:{runId:"next-run",turnId:null}}); + } finally {await sidecar.close();children.delete(sidecar);rmSync(root,{recursive:true,force:true});} + }, 15_000); + it("recovers after malformed input and reports its qualified Codex profile", async () => { const sidecar = startSidecar(); sidecar.write({ diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index a95d6a4f7a6..941ae867f17 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -176,6 +176,7 @@ let openParams: AcpxSidecarOpenParams | null = null; let claudeCompactionCount = 0; let runId: string | null = null; let turnId: string | null = null; +let turnStartedInProcess = false; /** * A restored session runs in a new provider process whose MCP request ids * restart. The controller keys semantic calls by run and call id, so prefix @@ -402,6 +403,7 @@ async function dispatch( if (request.command === "run.attach") { requireHost(); if (turnId) throw new Error("cannot attach a run during an active turn"); + if (turnStartedInProcess) throw new Error("cannot attach another run after this sidecar started a turn"); const attachedTools = parseTools(request.params.tools); if ( canonicalJson(attachedTools) !== canonicalJson(openParams?.tools ?? []) @@ -424,6 +426,7 @@ async function dispatch( const currentTurnId = boundedIdentity(request.params.turnId, "turnId"); const activeAgent = openParams!.agent; turnId = currentTurnId; + turnStartedInProcess = true; turnControls.begin(currentTurnId); let runtimeTurn: AcpxRuntimeTurn; const extensions = bindAcpxExtensionTurn({ diff --git a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts index c10017c71e9..837787940cc 100644 --- a/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-eviction.integration.test.ts @@ -296,7 +296,9 @@ describeEmbeddedPostgres("plugin Runner eviction through the Core execution entr expect((await state()).lifecycle).toBe("suspended"); } else { const providerPids: number[] = harness === "codex" ? [Number(await readFile(providerPidFile, "utf8"))] - : (await readFile(calls, "utf8")).trim().split(/\r?\n/).flatMap(line => { const entry = JSON.parse(line); return [entry.processStart, entry.childPid].filter(Number.isInteger); }); + : (await readFile(calls, "utf8")).trim().split(/\r?\n/).flatMap(line => { const entry = JSON.parse(line); return [entry.processStart, entry.childPid].filter(Number.isInteger); }) + .filter(pid => { try { process.kill(pid, 0); return true; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ESRCH") return false; throw error; } }); + expect(providerPids.length).toBeGreaterThan(0); const nativePid = handles.get(taskA)!.child.pid!; for (const pid of providerPids) { const [parent, group] = execFileSync("ps", ["-p", String(pid), "-o", "ppid=", "-o", "pgid="], { encoding: "utf8" }).trim().split(/\s+/).map(Number); diff --git a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts index 4da3feb7710..fed25e693ab 100644 --- a/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts +++ b/server/src/services/native-runtime/plugin-runner-release-resume.integration.test.ts @@ -298,7 +298,7 @@ describeEmbeddedPostgres("plugin Runner release/resume through the Core executio expect(methods.filter(method => method === "turn/start")).toHaveLength(adopted ? 3 : 2); } else { const journal = methods.map(line => JSON.parse(line)); - expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 3 : 2); + expect(journal.filter(entry => entry.processStart)).toHaveLength(adopted ? 4 : 3); expect(journal.filter(entry => entry.command === "turn.start")).toHaveLength(adopted ? 3 : 2); } } catch (error) {