diff --git a/packages/paperclip-runner/docs/durable-recovery.md b/packages/paperclip-runner/docs/durable-recovery.md index b2748091865..8bd198be3d4 100644 --- a/packages/paperclip-runner/docs/durable-recovery.md +++ b/packages/paperclip-runner/docs/durable-recovery.md @@ -83,6 +83,22 @@ a scripted harness, removes the controller's launch files, then recovers the same process and provider session and completes another turn. This exercises the real transport; it does not qualify a hosted OpenCode or ACPX deployment. +## ACPX checkpoints after a rejected turn start + +A matching `turn.start` response with `SESSION_RESUME_REQUIRED` and +`retryable: true` can retain the exact saved conversation. The session must have +no active turn, pending request or buffered event. The saved identity must still +match, with no retained work event or active goal. A pending `session.resumed` +notification is retained for the normal drain; it does not start provider work. The Runner first reaps +the sidecar, then confirms the provider lifetime fence. It stores this boundary +as `prepared`, with exit confirmation, before the normal drain and suspension +make it reusable. The scheduled recovery keeps its existing attempt limit. + +Unknown acceptance, malformed replies, other rejection codes, pending work and +unconfirmed cleanup do not qualify. Error messages cannot authorize recovery. +An arbitrary `closed` provider state stays closed. This path does not establish +recovery for a Runner killed during an accepted turn. + ## Settled ACPX sessions across a Runner upgrade Saved ACPX provider state records the digest of the launch profile that wrote 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 b3ade8b5308..746e0fbc05b 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 @@ -1900,15 +1900,54 @@ impl AcpxCommandExecutor { .expect("ACPX session exists before turn start") .start_turn_timed(&provider_turn_id, text, &working_directory, measure); if let Err(error) = &start_result { + let rejected_checkpoint = self + .session + .as_ref() + .and_then(AcpxProviderSession::rejected_start_checkpoint) + .filter(|identity| { + self.state.as_ref().is_some_and(|state| { + state.identity.as_ref() == Some(*identity) + && !state.provider_exit_unconfirmed + && state.pending_runtime_requests.is_empty() + && state + .pending_events + .iter() + .all(|event| event.event_type == "session.resumed") + && (state.goal_projection.is_null() + || goal_is_dormant(&state.goal_projection)) + }) + }) + .cloned(); let state = self .state .as_mut() .expect("ACPX state remains available after failed turn start"); - state.lifecycle = "closed".to_owned(); + state.lifecycle = if rejected_checkpoint.is_some() { + "prepared" + } else { + "closed" + } + .to_owned(); + state.provider_exit_unconfirmed = rejected_checkpoint.is_some(); state.active_turn_id = None; self.context.provider_turn_id = None; self.session = None; self.save_state()?; + if let Some(identity) = rejected_checkpoint { + let _provider_lifetime_fence = + await_provider_lifetime_fence(identity.provider_lifetime_fence_candidates)?; + self.state + .as_mut() + .expect("ACPX rejected checkpoint remains available") + .provider_exit_unconfirmed = false; + if let Err(error) = self.save_state() { + self.state + .as_mut() + .expect("ACPX rejected checkpoint remains available") + .provider_exit_unconfirmed = true; + return Err(error); + } + } return Err(DurableRunnerError::invalid(format!( "ACPX turn start failed closed: {error}" ))); diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs b/packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_session.rs index 7c3c07a0f02..bb2045e82ba 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 @@ -350,6 +350,7 @@ pub struct AcpxProviderSession { closed: bool, failure: Option, transport_terminated: bool, + rejected_start_resumable: bool, runtime_retired: bool, completed_turn_observed: bool, } @@ -407,6 +408,7 @@ impl AcpxProviderSession { closed: false, failure: None, transport_terminated: false, + rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, }) @@ -424,6 +426,10 @@ impl AcpxProviderSession { &self.identity } + pub(crate) fn rejected_start_checkpoint(&self) -> Option<&AcpxProviderSessionIdentity> { + self.rejected_start_resumable.then_some(&self.identity) + } + pub fn turn_control_capabilities(&self) -> AcpxTurnControlCapabilities { self.turn_controls } @@ -598,6 +604,7 @@ impl AcpxProviderSession { working_directory: &Path, performance: bool, ) -> Result { + self.rejected_start_resumable = false; self.ensure_open()?; if self.runtime_retired { return Err(LocalRunnerError::invalid( @@ -686,12 +693,19 @@ impl AcpxProviderSession { "ACPX reserved tool receipt rotation failed: {error}" )))); } - let response = match self.transport.request( - GeneratedAcpxSidecarCommand::TurnStart, + let response = match self.transport.request_turn_start( json!({"turnId":turn_id,"message":message,"performance":performance}), ) { Ok(response) => response, - Err(error) => return Err(self.fail_closed(error)), + Err(failure) => { + let resumable = failure.resume_required + && self.state.active_turn_id().is_none() + && !self.state.has_pending_requests() + && self.transport.buffered_events().len() == 0; + let error = self.fail_closed(failure.error); + self.rejected_start_resumable = resumable && self.transport_terminated; + return Err(error); + } }; if response.get("turnId").and_then(Value::as_str) != Some(turn_id) { return Err(self.fail_closed(LocalRunnerError::invalid( @@ -2227,6 +2241,7 @@ mod mode_tests { closed: false, failure: None, transport_terminated: false, + rejected_start_resumable: false, runtime_retired: false, completed_turn_observed: false, } 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 8db06a56c36..be77e3be7ce 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 @@ -76,6 +76,11 @@ pub struct AcpxSidecarEvent { pub payload: Value, } +pub(crate) struct AcpxTurnStartFailure { + pub error: LocalRunnerError, + pub resume_required: bool, +} + pub struct AcpxSidecarTransport { process: SupervisedProcess, request_timeout: Duration, @@ -415,6 +420,37 @@ impl AcpxSidecarTransport { command: GeneratedAcpxSidecarCommand, params: Value, ) -> Result { + match self.request_outcome(command, params)? { + CommandOutcome::Success(value) => Ok(value), + CommandOutcome::Rejected { error, .. } => Err(error), + } + } + + pub(crate) fn request_turn_start( + &mut self, + params: Value, + ) -> Result { + match self.request_outcome(GeneratedAcpxSidecarCommand::TurnStart, params) { + Ok(CommandOutcome::Success(value)) => Ok(value), + Ok(CommandOutcome::Rejected { + error, + resume_required, + }) => Err(AcpxTurnStartFailure { + error, + resume_required, + }), + Err(error) => Err(AcpxTurnStartFailure { + error, + resume_required: false, + }), + } + } + + fn request_outcome( + &mut self, + command: GeneratedAcpxSidecarCommand, + params: Value, + ) -> Result { if self.poisoned { return Err(LocalRunnerError::invalid( "ACPX sidecar transport is unavailable after a protocol failure", @@ -422,8 +458,7 @@ impl AcpxSidecarTransport { } let result = self.request_inner(command, params); match result { - Ok(CommandOutcome::Success(value)) => Ok(value), - Ok(CommandOutcome::Rejected(error)) => Err(error), + Ok(outcome) => Ok(outcome), Err(error) => { self.poison(); Err(error) @@ -517,7 +552,7 @@ impl AcpxSidecarTransport { }); match result { Ok(Some(CommandOutcome::Success(value))) => Ok(Some(value)), - Ok(Some(CommandOutcome::Rejected(_))) => Ok(None), + Ok(Some(CommandOutcome::Rejected { .. })) => Ok(None), Ok(None) => Ok(None), Err(error) => { self.poison(); @@ -623,14 +658,17 @@ impl AcpxSidecarTransport { } let error = response.error.expect("failed response has validated error"); let provider_failure = error.provider_failure.clone(); - return Ok(Some(CommandOutcome::Rejected(LocalRunnerError::invalid( + let resume_required = command == GeneratedAcpxSidecarCommand::TurnStart + && error.retryable + && error.code == "SESSION_RESUME_REQUIRED"; + return Ok(Some(CommandOutcome::Rejected { error: LocalRunnerError::invalid( format!( "ACPX sidecar command {} was rejected (retryable={}, classification={})", command.as_str(), error.retryable, response_error_classification(&error), ), - ).with_provider_failure(provider_failure)))); + ).with_provider_failure(provider_failure), resume_required })); } } } @@ -823,7 +861,10 @@ impl AcpxSidecarTransport { enum CommandOutcome { Success(Value), - Rejected(LocalRunnerError), + Rejected { + error: LocalRunnerError, + resume_required: bool, + }, } enum ParsedFrame { @@ -1154,6 +1195,30 @@ fn response_error_classification(error: &ResponseError) -> &'static str { #[cfg(test)] mod tests { + #[test] + #[cfg(unix)] + fn a_timed_out_turn_start_never_establishes_a_resumable_rejection() { + let mut transport = AcpxSidecarTransport::start(&AcpxSidecarTransportConfig { + command: "/bin/sh".into(), + args: vec!["-c".to_owned(), "read request; sleep 10".to_owned()], + verified_launch: None, + request_timeout: Duration::from_millis(30), + shutdown_grace: Duration::from_millis(20), + }) + .unwrap(); + let failure = transport + .request_turn_start(json!({"turnId":"turn-1"})) + .unwrap_err(); + assert!(!failure.resume_required); + assert!(failure.error.to_string().contains("timed out")); + let reuse = transport + .request_turn_start(json!({"turnId":"turn-2"})) + .unwrap_err(); + assert!(!reuse.resume_required); + assert!(reuse.error.to_string().contains("unavailable")); + transport.shutdown().unwrap(); + } + #[test] fn pi_credentials_require_a_bounded_binding_without_process_control_variables() { assert!(pi_credential_environment_keys(None).unwrap().is_empty()); 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 280adde3421..49b2d151dc9 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 @@ -99,6 +99,39 @@ fn run() -> Result<(), Box> { .get("command") .and_then(Value::as_str) .ok_or("request command is missing")?; + if command == "turn.start" { + if let Some(fault) = args + .windows(2) + .find(|pair| pair[0] == "--reject-turn-start") + { + let fault = fault[1].as_str(); + if fault == "event" { + write_turn_event( + &mut stdout, + next_sequence, + "runtime.event", + "run-1", + request["params"]["turnId"].as_str().unwrap(), + json!({"type":"text_delta","text":"Work may have started."}), + )?; + next_sequence += 1; + } + let code = match fault { + "unknown" | "message-only" => "UNKNOWN", + "lowercase" => "session_resume_required", + _ => "SESSION_RESUME_REQUIRED", + }; + let mut error = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION, + "id":if fault == "wrong-id" {id + 1} else {id},"ok":false, + "error":{"code":code,"message":"SESSION_RESUME_REQUIRED retryable=true classification=session_resume_required", + "retryable":fault != "permanent"}}); + if fault == "malformed" { + error["error"]["retryable"] = json!("true"); + } + write_json(&mut stdout, &error)?; + continue; + } + } if let Some(journal) = source_journal.as_mut() { write_json(journal, &request)?; if command == "run.attach" { diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs index 888a21f57b3..bf9c2ffc267 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs @@ -3099,6 +3099,141 @@ fn idle_pi_stop_cannot_publish_a_reusable_identity_while_its_lifetime_fence_is_h fs::remove_dir_all(directory).unwrap(); } +fn rejected_turn_start_fixture(directory: &Path, fault: &str) -> (DurableRunnerConfig, Value) { + let mut config = acpx_config(directory, "bootstrap"); + config + .acpx_launch_profile + .as_mut() + .unwrap() + .args + .extend(["--reject-turn-start".to_owned(), fault.to_owned()]); + let mut payload = prepare_payload_with_mode(directory, "codex", "bootstrap"); + payload["provider"]["sidecarArgs"] = json!(config.acpx_launch_profile.as_ref().unwrap().args); + (config, payload) +} + +#[test] +#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")] +fn rejected_turn_start_retains_only_the_definite_retryable_checkpoint() { + for fault in [ + "resume", + "permanent", + "unknown", + "lowercase", + "message-only", + "event", + "wrong-id", + "malformed", + ] { + let directory = temporary_directory(&format!("rejected-start-{fault}")); + let (config, payload) = rejected_turn_start_fixture(&directory, fault); + let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + executor + .execute(&command(1, "run.prepare", payload)) + .unwrap(); + executor + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + let path = directory.join("acpx-provider-state.json"); + let opened: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"rejected-turn", "text":"Continue saved work."}), + )) + .unwrap_err(); + let rejected: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + assert_eq!( + rejected["lifecycle"], + if fault == "resume" { + "prepared" + } else { + "closed" + }, + "{fault}" + ); + assert_eq!(rejected["providerExitUnconfirmed"], false, "{fault}"); + assert_eq!(rejected["identity"], opened["identity"], "{fault}"); + assert_eq!(rejected["activeTurnId"], Value::Null, "{fault}"); + executor + .execute(&command(4, "runner.drain", json!({}))) + .unwrap(); + executor + .execute(&command(5, "runner.suspend", json!({}))) + .unwrap(); + let suspended: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + assert_eq!( + suspended["lifecycle"], + if fault == "resume" { + "suspended" + } else { + "closed" + }, + "{fault}" + ); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } +} + +#[test] +#[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")] +fn rejected_turn_start_cannot_publish_a_checkpoint_while_its_provider_lifetime_is_held() { + let directory = temporary_directory("rejected-start-held-lifetime"); + let fence = (61_000..=u16::MAX) + .filter_map(|port| std::net::TcpListener::bind(("127.0.0.1", port)).ok()) + .take(3) + .collect::>(); + assert_eq!(fence.len(), 3); + let ports = fence + .iter() + .map(|listener| listener.local_addr().unwrap().port().to_string()) + .collect::>() + .join(","); + let (mut config, mut payload) = rejected_turn_start_fixture(&directory, "resume"); + config + .acpx_launch_profile + .as_mut() + .unwrap() + .args + .extend(["--lifetime-fence-ports".to_owned(), ports]); + payload["provider"]["sidecarArgs"] = json!(config.acpx_launch_profile.as_ref().unwrap().args); + let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + executor + .execute(&command(1, "run.prepare", payload)) + .unwrap(); + executor + .execute(&command(2, "session.open", json!({}))) + .unwrap(); + let error = executor + .execute(&command( + 3, + "turn.start", + json!({"turnId":"rejected-turn", "text":"Continue saved work."}), + )) + .unwrap_err(); + assert!( + error + .to_string() + .contains("original provider lifetime remains active"), + "{error}" + ); + let path = directory.join("acpx-provider-state.json"); + let state: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + assert_eq!(state["lifecycle"], "prepared"); + assert_eq!(state["providerExitUnconfirmed"], true); + assert!(executor + .execute(&command(4, "runner.suspend", json!({}))) + .is_err()); + assert!(executor.poll_events().unwrap().is_empty()); + let after_poll: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + assert_eq!(after_poll, state); + drop(fence); + executor.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); +} + #[test] #[cfg_attr(windows, ignore = "verified launch is not supported on Windows yet")] fn publishes_only_changed_pi_controls_before_the_new_turn() { diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 887f5b1c5f9..c88ac3d35f2 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -191,12 +191,21 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); bundles.push(bundle); return bundle; }; - const expectCompleted = async (bundle: ReturnType) => { + const expectCompleted = async (bundle: ReturnType, expectedTurnId: string) => { + expect(expectedTurnId).toEqual(expect.any(String)); + expect(expectedTurnId).not.toBe(""); let timer: ReturnType | undefined; try { await Promise.race([ (async () => { - for await (const event of bundle.transport.notifications()) if (event.method === "turn/completed") return; + for await (const event of bundle.transport.notifications()) { + if (event.method !== "turn/completed") continue; + const turn = event.params?.turn as { id?: string; status?: string; error?: unknown } | undefined; + if (turn?.id !== expectedTurnId) continue; + expect(turn).toMatchObject({ id: expectedTurnId, status: "completed" }); + expect(turn?.error ?? null).toBeNull(); + return; + } throw new Error("Fixture turn ended without completion"); })(), new Promise((_, reject) => { timer = setTimeout(() => reject(new Error("Fixture turn completion timed out")), 5_000); }), @@ -208,8 +217,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); const seed = create(seedDirectory); const opened = await seed.transport.request("thread/start", { cwd: root, model: "fixture-model", dynamicTools: [] }); const thread = opened.thread as { id: string; sessionId: string }; - await seed.transport.request("turn/start", { input: [{ type: "text", text: "Save the provider checkpoint." }] }); - await expectCompleted(seed); + const seedStarted = await seed.transport.request("turn/start", { input: [{ type: "text", text: "Save the provider checkpoint." }] }); + await expectCompleted(seed, (seedStarted.turn as { id: string }).id); await seed.transport.close(); const provider = JSON.parse(await readFile(join(seedDirectory, "runner", "acpx-provider-state.json"), "utf8")); const resume = { driverSessionId: thread.id, providerSessionId: thread.sessionId, providerIdentity: provider.identity }; @@ -230,6 +239,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); `fixture_nonce\n${pid}\n${new Date().toISOString()}\n${identity.runnerInstanceId}\n${fingerprint}\n`); await expect(fresh.transport.request("turn/start", { input: [{ type: "text", text: "Resume saved work." }] })) .rejects.toThrow("classification=session_resume_required"); + const rejectedCheckpoint = JSON.parse(await readFile(join(stateDirectory, "runner", "acpx-provider-state.json"), "utf8")); + expect(rejectedCheckpoint).toMatchObject({ lifecycle: "prepared", providerExitUnconfirmed: false, identity: provider.identity }); await fresh.transport.close(); await vi.waitFor(() => expect(child.exitCode).not.toBeNull(), { timeout: 5_000 }); const resumedControl = JSON.parse(await readFile(join(stateDirectory, "control-plane", "control-plane-state.json"), "utf8")); @@ -239,6 +250,9 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(saved.pendingTerminalDelivery ?? null).toBeNull(); expect(saved.pendingProviderCleanup ?? null).toBeNull(); expect(saved.warmTransition ?? null).toBeNull(); + const suspendedCheckpoint = JSON.parse(await readFile(join(stateDirectory, "runner", "acpx-provider-state.json"), "utf8")); + expect(suspendedCheckpoint).toMatchObject({ lifecycle: "suspended", providerExitUnconfirmed: false, + activeTurnId: null, pendingEvents: [], pendingRuntimeRequests: {}, identity: provider.identity }); const runner = { execute: async (request: { command: string; args: string[]; timeoutMs?: number }) => new Promise<{ stdout: string; stderr: string; exitCode: number; timedOut: boolean }>(resolveResult => { execFile(request.command, request.args, { timeout: request.timeoutMs }, (error, stdout, stderr) => resolveResult({ @@ -256,6 +270,16 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); const retried = create(stateDirectory, resume, adopted); await retried.transport.request("thread/read", { threadId: thread.id, includeTurns: false }); await retried.transport.request("thread/resume", { threadId: thread.id, cwd: root, model: "fixture-model", dynamicTools: [] }); + const retryStarted = await retried.transport.request("turn/start", { input: [{ type: "text", text: "Continue after the bounded retry." }] }); + const retryTurnId = (retryStarted.turn as { id: string }).id; + await expectCompleted(retried, retryTurnId); + const calls = (await readFile(journal, "utf8")).trim().split("\n").map(line => JSON.parse(line)); + const turnStarts = calls.filter(call => call.command === "turn.start"); + expect(turnStarts).toHaveLength(3); + expect(new Set(turnStarts.map(call => call.params.turnId)).size).toBe(3); + expect(turnStarts.at(-1).params.turnId).toBe(retryTurnId); + expect(calls.filter(call => call.command === "session.open").map(call => call.params.expectedIdentity)) + .toEqual([null, provider.identity, provider.identity]); expect(launches.mock.calls).toEqual([[undefined], [undefined], [undefined]]); expect(retried.evidence().runnerPid).toBeGreaterThan(0); expect(retried.evidence().runnerPid).not.toBe(pid);