diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index cd88ba60094..abd99258eb3 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md @@ -144,6 +144,12 @@ that arrive during the final status read and keeps them queued for delivery. As with Codex, a prepared session can continue or be cleaned up. This handoff requires a completed turn. Suspended sessions use their usual recovery path. +Before B attaches, Runner saves the events it already received from A's sidecar. +The readiness snapshot stays blocked until Core acknowledges those events. +Another turn within A also keeps its buffered events without waiting for an ACK. +ACPX still replaces the sidecar at the next turn or run attachment. The saved +provider identity continues through that replacement. + ## Host park eligibility `CommandExecutor::host_park_ready` defaults to false. Native Codex opts in only 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 67013b60dc2..3298cd5ceff 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 @@ -1439,6 +1439,7 @@ impl AcpxCommandExecutor { None }; let tool_set = authorized_tool_set(payload)?; + self.retain_buffered_provider_events()?; let state = self .state .as_ref() @@ -1729,6 +1730,14 @@ impl AcpxCommandExecutor { "turn.start payload.turnId is invalid", )); } + // Retain the old sidecar's read tail before its MCP receipt epoch changes. + if self + .session + .as_ref() + .is_some_and(|session| session.state().has_settled_turns()) + { + self.retain_buffered_provider_events()?; + } if self.session.is_none() { return Err(DurableRunnerError::invalid("ACPX session is not open")); } @@ -2294,7 +2303,28 @@ impl AcpxCommandExecutor { { return Ok(()); } - for _ in 0..MAX_EVENTS_PER_POLL { + self.poll_provider_events(MAX_EVENTS_PER_POLL, disconnected) + } + + fn retain_buffered_provider_events(&mut self) -> Result<(), DurableRunnerError> { + loop { + let count = self + .session + .as_ref() + .map_or(0, AcpxProviderSession::buffered_event_count); + if count == 0 { + return Ok(()); + } + self.poll_provider_events(count.min(MAX_EVENTS_PER_POLL), false)?; + } + } + + fn poll_provider_events( + &mut self, + count: usize, + disconnected: bool, + ) -> Result<(), DurableRunnerError> { + for _ in 0..count { let events = self .session .as_mut() @@ -2543,7 +2573,12 @@ impl CommandExecutor for AcpxCommandExecutor { { self.snapshot_live_requests() } - "session.snapshot" => self.snapshot(), + "session.snapshot" => { + if command.payload.get("quiesceForWarmAttach") == Some(&Value::Bool(true)) { + self.retain_buffered_provider_events()?; + } + self.snapshot() + } "session.close" | "session.destroy" => self.close_session(&command.command_type), "runner.suspend" => self.suspend(), "runner.shutdown" => { 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 86adf7dd148..53c1bf8ea3f 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 @@ -531,6 +531,10 @@ impl AcpxProviderSession { self.transport.request(command, payload) } + pub(crate) fn buffered_event_count(&self) -> usize { + self.transport.buffered_events().len() + } + pub fn start_turn( &mut self, turn_id: &str, 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 b12f224226c..42dc16658e6 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 @@ -128,6 +128,31 @@ fn run() -> Result<(), Box> { source_goal_revision += 1; response["result"]["providerRevision"] = json!(source_goal_revision); } + if command == "session.goal.get" + && matches!(fault.as_str(), "read-diagnostic" | "read-burst") + { + let count = if fault == "read-burst" { 200 } else { 1 }; + for _ in 0..count { + write_json( + &mut stdout, + &json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "sequence":next_sequence,"eventType":"runtime.diagnostic","runId":source_run,"turnId":null, + "payload":{"code":"ordinary_read_tail","message":"Retained before an ordinary turn."}}), + )?; + next_sequence += 1; + } + } + if command == "session.goal.get" && fault == "read-invalid-tail" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + &source_run, + "unbound-old-turn", + json!({"type":"text_delta","text":"Rejected stale tail."}), + )?; + next_sequence += 1; + } if command == "session.snapshot" && fault == "goal-during-snapshot" { source_goal_revision += 1; let frame = json!({ 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 new file mode 100644 index 00000000000..26712386365 --- /dev/null +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs @@ -0,0 +1,293 @@ +#![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, CommandExecutor, DurableRunnerConfig, QualifiedLaunchArtifact, +}; +use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor; +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); + } +} + +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 fixture( + agent: &str, +) -> ( + Fixture, + DurableRunnerConfig, + NativeProviderCommandExecutor, + Value, +) { + let catalog: Value = + serde_json::from_str(include_str!("../../../../acpx-profiles.json")).unwrap(); + let profile = &catalog["profiles"][agent]; + let digest = profile["commandDigest"].as_str().unwrap(); + let fixture = Fixture(std::env::temp_dir().join(format!("acpx-warm-{}", uuid::Uuid::new_v4()))); + fs::create_dir_all(&fixture.0).unwrap(); + fs::set_permissions(&fixture.0, fs::Permissions::from_mode(0o700)).unwrap(); + let sidecar = fixture.0.join("sidecar"); + fs::copy(env!("CARGO_BIN_EXE_fake-acpx-sidecar"), &sidecar).unwrap(); + fs::set_permissions(&sidecar, fs::Permissions::from_mode(0o700)).unwrap(); + let native = fixture.0.join("native-sidecar"); + fs::rename(&sidecar, &native).unwrap(); + fs::write(&sidecar, format!("#!/bin/sh\nexec '{}' --mode source-reconcile --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": agent, "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, + }); + 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(); + (fixture, config, executor, opened.result) +} + +fn settle(executor: &mut NativeProviderCommandExecutor) -> Vec { + let deadline = Instant::now() + Duration::from_secs(5); + let mut seen = Vec::new(); + let mut completed = false; + loop { + let events = executor.poll_events().unwrap(); + if events.is_empty() && completed { + return seen; + } + for event in &events { + completed |= event.event_type == "turn.completed"; + seen.push(json!({"type":event.event_type,"payload":event.payload})); + } + executor.acknowledge_events(events.len()).unwrap(); + assert!(Instant::now() < deadline, "no completed provider turn"); + } +} + +#[test] +fn ordinary_acpx_read_tail_survives_immediate_next_turn_without_a_park_probe() { + let (fixture, _config, mut executor, _) = fixture("claude"); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"first", "text":"Complete."}), + )) + .unwrap(); + settle(&mut executor); + fs::write(fixture.0.join("fault"), "read-diagnostic").unwrap(); + executor + .execute(&command(4, "session.goal.get", json!({}))) + .unwrap(); + fs::write(fixture.0.join("fault"), "").unwrap(); + executor + .execute(&command( + 5, + "turn.start", + json!({"turnId":"second", "text":"Continue without a park probe."}), + )) + .unwrap(); + let events = settle(&mut executor); + assert_eq!( + events + .iter() + .filter(|event| event["payload"]["code"] == "ordinary_read_tail") + .count(), + 1 + ); + executor.shutdown().unwrap(); +} + +#[test] +fn ordinary_acpx_replacement_retains_a_multi_batch_tail_behind_existing_ack_debt() { + for attach in [false, true] { + let (fixture, config, mut executor, opened) = fixture("claude"); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"first", "text":"Complete."}), + )) + .unwrap(); + settle(&mut executor); + fs::write(fixture.0.join("fault"), "read-diagnostic").unwrap(); + executor + .execute(&command(4, "session.goal.get", json!({}))) + .unwrap(); + assert_eq!(executor.poll_events().unwrap().len(), 1); // deliberately do not ACK + fs::write(fixture.0.join("fault"), "read-burst").unwrap(); + executor + .execute(&command(5, "session.goal.get", json!({}))) + .unwrap(); + fs::write(fixture.0.join("fault"), "").unwrap(); + let events = if attach { + let state: Value = serde_json::from_slice( + &fs::read(fixture.0.join("acpx-provider-state.json")).unwrap(), + ) + .unwrap(); + let mut descriptor = state["descriptor"].clone(); + descriptor["runId"] = json!("run-2"); + let error = executor.execute(&command(6, "run.attach", json!({"provider":descriptor, + "paperclipNextAuthority":{"identity":{"runId":"run-2","normalizedSessionId":config.normalized_session_id}}}))).unwrap_err(); + assert!(error.to_string().contains("settled ACPX")); + let readiness = executor + .execute(&command( + 7, + "session.snapshot", + json!({"quiesceForWarmAttach":true}), + )) + .unwrap(); + assert_eq!(readiness.result["warmAttachReady"], false); + assert!(std::process::Command::new("kill") + .args(["-0", &opened["processId"].to_string()]) + .status() + .unwrap() + .success()); + let mut retained = Vec::new(); + loop { + let batch = executor.retained_events().unwrap(); + if batch.is_empty() { + break; + } + retained.extend( + batch + .iter() + .map(|event| json!({"type":event.event_type,"payload":event.payload})), + ); + executor.acknowledge_events(batch.len()).unwrap(); + } + let readiness = executor + .execute(&command( + 8, + "session.snapshot", + json!({"quiesceForWarmAttach":true}), + )) + .unwrap(); + assert_eq!(readiness.result["warmAttachReady"], true); + let attached = executor.execute(&command(9, "run.attach", json!({"provider":descriptor, + "paperclipNextAuthority":{"identity":{"runId":"run-2","normalizedSessionId":config.normalized_session_id}}}))).unwrap(); + assert_eq!( + attached.result["providerSessionId"], + opened["providerSessionId"] + ); + retained + } else { + executor + .execute(&command( + 6, + "turn.start", + json!({"turnId":"second","text":"Continue without waiting for old ACKs."}), + )) + .unwrap(); + settle(&mut executor) + }; + assert_eq!( + events + .iter() + .filter(|event| event["payload"]["code"] == "ordinary_read_tail") + .count(), + 201 + ); + executor.shutdown().unwrap(); + } +} + +#[test] +fn rejected_read_tail_returns_a_command_error_without_starting_another_turn() { + let (fixture, _config, mut executor, _) = fixture("claude"); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"first","text":"Complete."}), + )) + .unwrap(); + settle(&mut executor); + fs::write(fixture.0.join("fault"), "read-invalid-tail").unwrap(); + executor + .execute(&command(4, "session.goal.get", json!({}))) + .unwrap(); + fs::write(fixture.0.join("fault"), "").unwrap(); + let error = executor + .execute(&command( + 5, + "turn.start", + json!({"turnId":"second","text":"Must fail without a panic."}), + )) + .unwrap_err(); + assert!(error.to_string().contains("session is not open")); + let journal = fs::read_to_string(fixture.0.join("sidecar.jsonl")).unwrap(); + assert_eq!( + journal + .lines() + .filter(|line| line.contains("\"command\":\"turn.start\"")) + .count(), + 1 + ); + let state: Value = + serde_json::from_slice(&fs::read(fixture.0.join("acpx-provider-state.json")).unwrap()) + .unwrap(); + assert_eq!(state["lifecycle"], "closed"); + assert_eq!(state["activeTurnId"], Value::Null); +} diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts index 90481f91449..bb26e9428b5 100644 --- a/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-acpx-source-reconciliation.test.ts @@ -270,3 +270,27 @@ it.skipIf(process.platform === "win32")("preserves ordinary same-run active goal expect(await f.count("processStart")).toBe(2); expect(await f.count("turn.start")).toBe(1); } finally { await f.cleanup(); } }, 30_000); + +it.skipIf(process.platform === "win32")("settles an ACPX read tail through ordinary transport quiescence before attaching B", async () => { + const f = await fixture("claude"); + const delivered: PrpEvent[] = []; + try { + const source = f.recover({ onRecoveryAccountingEvent: async event => { delivered.push(event); } }); + expect((await source.transport.request("thread/read", {})).thread).toMatchObject(f.thread); + await f.settle(); + await writeFile(f.fault, "read-diagnostic"); + await source.transport.request("thread/goal/get", {}); + await writeFile(f.fault, ""); + const pid = source.evidence().runnerPid; + await source.transport.attachRun!({ runId: "run-target", turnId: "turn-target", itemId: "item-target" }); + await f.settle(); + expect(source.evidence().runnerPid).toBe(pid); + const tail = delivered.filter(event => event.eventType === "harness.diagnostic" && event.payload.code === "ordinary_read_tail"); + expect(tail).toHaveLength(1); + expect(tail[0]!.runId).toBe(f.identity.runId); + await f.complete(source); await f.settle(); + expect((await read(f.providerFile)).identity).toEqual(f.providerIdentity); + expect(await f.count("turn.start")).toBe(2); + expect(await f.count("processStart")).toBe(3); + } finally { await f.cleanup(); } +}, 30_000);