From 9b3c53923c4f4f26e642bf2fef861895a45c28ae Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 29 Sep 2026 08:27:09 -0500 Subject: [PATCH] fix(runner): stop settlement publication after persistence failure Publish staged ACP resolutions only after their atomic state write succeeds. Latch persistence failures across commands, polls, retained event drains and acknowledgements; stop owned processes without overwriting uncertain snapshots. Cover failed rename and post-rename writes, mixed requests, acknowledgement failure and indeterminate command replay. Co-Authored-By: Paperclip --- .../runner-rich-acp-capabilities.md | 5 + .../runner-core/src/acpx_provider_backend.rs | 251 +++++++++++++++++- .../crates/runner-core/src/durable/runner.rs | 119 +++++++++ .../tests/native_provider_backend.rs | 195 +++++++++++++- 4 files changed, 546 insertions(+), 24 deletions(-) diff --git a/doc/architecture/runner-rich-acp-capabilities.md b/doc/architecture/runner-rich-acp-capabilities.md index 5fedbb2be24..5d1f7c6ff7f 100644 --- a/doc/architecture/runner-rich-acp-capabilities.md +++ b/doc/architecture/runner-rich-acp-capabilities.md @@ -116,6 +116,11 @@ state save that removes the pending request. A restart before journal delivery retains that resolution for normal event replay and acknowledgment. A restart before this state save instead expires the unresolved request; neither path sends the provider response again. +If persistence fails, the current executor stops accepting commands and exposing +or acknowledging retained events. Cleanup still terminates its owned provider, +but leaves the uncertain snapshot untouched. A fresh executor reads the complete +atomic snapshot that survived; it cannot publish an in-memory resolution that +conflicts with a later recovery expiry. Standard ACP does not acknowledge application of a permission reply; a lost transport acknowledgment is not proof of exactly-once external effects. A replacement provider process cannot inherit an old approval promise. A bounded durable ledger expires pending requests after provider loss or unsafe restart, including requests whose creation events were already acknowledged. No tool 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 e725b7524b0..c2408ec731a 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 @@ -683,6 +683,9 @@ pub struct AcpxCommandExecutor { session: Option, restore_checked: bool, restore_error: Option, + persistence_error: Option, + #[cfg(test)] + fail_after_state_rename: bool, launch_profile: Option, } @@ -701,6 +704,9 @@ impl AcpxCommandExecutor { session: None, restore_checked: false, restore_error: None, + persistence_error: None, + #[cfg(test)] + fail_after_state_rename: false, launch_profile: config.acpx_launch_profile.clone(), } } @@ -721,6 +727,7 @@ impl AcpxCommandExecutor { } fn restore(&mut self) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; if self.restore_checked { return Ok(()); } @@ -876,11 +883,46 @@ impl AcpxCommandExecutor { Ok(session) } - fn save_state(&self) -> Result<(), DurableRunnerError> { - let state = self + fn ensure_persistence_healthy(&self) -> Result<(), DurableRunnerError> { + match self.persistence_error.as_ref() { + Some(error) => Err(error.clone()), + None => Ok(()), + } + } + + fn record_state_write( + &mut self, + result: Result<(), DurableRunnerError>, + ) -> Result<(), DurableRunnerError> { + if let Err(error) = result.as_ref() { + // A rename may have succeeded before directory fsync or protection + // failed. Neither rolling back nor continuing from memory can prove + // which state survived. Stop publication and mutations until a fresh + // executor reloads the complete atomic snapshot; never resend input. + self.persistence_error = Some(error.clone()); + } + result + } + + fn save_state(&mut self) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; + let result = self .state .as_ref() - .ok_or_else(|| DurableRunnerError::invalid("ACPX provider state is unavailable"))?; + .ok_or_else(|| DurableRunnerError::invalid("ACPX provider state is unavailable")) + .and_then(|state| self.write_state(state)); + self.record_state_write(result) + } + + fn commit_staged_state(&mut self, state: AcpxDurableState) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; + let result = self.write_state(&state); + self.record_state_write(result)?; + self.state = Some(state); + Ok(()) + } + + fn write_state(&self, state: &AcpxDurableState) -> Result<(), DurableRunnerError> { let launch_profile_digest = self.launch_profile_digest()?; state.validate(&self.context, &launch_profile_digest)?; secure_directory(&self.state_dir, "provider state")?; @@ -899,6 +941,10 @@ impl AcpxCommandExecutor { file.sync_all()?; drop(file); fs::rename(&temporary, &path)?; + #[cfg(test)] + if self.fail_after_state_rename { + return Err(std::io::Error::other("injected post-rename sync failure")); + } #[cfg(unix)] File::open(&self.state_dir)?.sync_all()?; Ok(()) @@ -1535,6 +1581,19 @@ impl AcpxCommandExecutor { DurableRunnerError::invalid("request.resolve requires resolution or response") })?; let permission = session.state().pending_permission(request_id).is_some(); + // Admit the event before sending the response, but keep its request + // retirement and outbox entry private until the state write succeeds. + let mut staged = self + .state + .clone() + .expect("ACPX response has an active durable turn"); + staged.push(NormalizedProviderEvent { + event_type: "runtime_request.resolved".to_owned(), + priority: EventPriority::P0, + payload: json!({"provider": "acpx", "requestId": request_id, "status": "delivered", + "requestKind": if permission {"permission_approval"} else {"runtime"}, + "turnId":self.context.turn_id,"itemId":self.context.item_id,"action":resolution.get("action")}), + })?; (if permission { session.resolve_permission(request_id, &turn_id, &resolution) } else { @@ -1547,17 +1606,7 @@ impl AcpxCommandExecutor { // Retain that fact in the same atomic state write that retires the // pending request. A crash before this write expires the old request; // a crash after it replays this event, never the provider response. - self.state - .as_mut() - .expect("ACPX response has an active durable turn") - .push(NormalizedProviderEvent { - event_type: "runtime_request.resolved".to_owned(), - priority: EventPriority::P0, - payload: json!({"provider": "acpx", "requestId": request_id, "status": "delivered", - "requestKind": if permission {"permission_approval"} else {"runtime"}, - "turnId":self.context.turn_id,"itemId":self.context.item_id,"action":resolution.get("action")}), - })?; - self.save_state()?; + self.commit_staged_state(staged)?; Ok(CommandExecution::result( json!({"status": "delivered", "requestId": request_id}), )) @@ -1935,6 +1984,7 @@ impl CommandExecutor for AcpxCommandExecutor { } fn retained_events(&mut self) -> Result, DurableRunnerError> { + self.ensure_persistence_healthy()?; // Explicit drain runs while control traffic suppresses provider polling. // Expose the already-retained suffix so runnerd can commit and ACK it // before suspension, without restoring or advancing the provider. @@ -1948,6 +1998,7 @@ impl CommandExecutor for AcpxCommandExecutor { } fn acknowledge_events(&mut self, count: usize) -> Result<(), DurableRunnerError> { + self.ensure_persistence_healthy()?; if count == 0 { return Ok(()); } @@ -1965,6 +2016,20 @@ impl CommandExecutor for AcpxCommandExecutor { } fn shutdown(&mut self) -> Result<(), DurableRunnerError> { + if let Some(error) = self.persistence_error.clone() { + // Reap our owned process even when durable cleanup is uncertain, + // but do not overwrite the last atomic snapshot or claim success. + if let Some(mut session) = self.session.take() { + session + .shutdown("ACPX provider state persistence failed") + .map_err(|cleanup| { + DurableRunnerError::invalid(format!( + "{error}; failed to stop ACPX provider: {cleanup}" + )) + })?; + } + return Err(error); + } // A replacement durable runner may reach terminal reconciliation // before any provider command or event poll. Restore the persisted // session first so cleanup cannot succeed merely because this process @@ -2349,6 +2414,164 @@ mod tests { } } + #[test] + fn post_rename_failure_blocks_settlement_publication_until_fresh_recovery() { + for event_type in [ + "runtime_request.resolved", + "runtime_request.expired", + "runtime_request.cancelled", + ] { + let directory = temporary_directory("post-rename-settlement"); + let marker = directory.join("must-not-start-provider"); + let command_path = directory.join("sidecar"); + write_artifact( + &command_path, + format!("#!/bin/sh\ntouch '{}'\n", marker.display()).as_bytes(), + true, + ); + let launch_profile = AcpxLaunchProfile { + authority_digest: format!("sha256:{}", "d".repeat(64)), + command: command_path.clone(), + args: Vec::new(), + artifacts: vec![artifact(&command_path)], + }; + let mut value = descriptor("codex"); + value["sidecarCommand"] = json!(command_path); + value["sidecarArgs"] = json!([]); + let descriptor: AcpxProviderDescriptor = serde_json::from_value(value).unwrap(); + let identity = AcpxProviderSessionIdentity { + kind: "acpx".into(), + normalized_session_id: "session-1".into(), + acpx_record_id: "record-1".into(), + backend_session_id: "backend-1".into(), + agent_session_id: "agent-1".into(), + profile_digest: descriptor.command_digest.clone(), + workspace_digest: format!("sha256:{}", "a".repeat(64)), + requested_model: descriptor.model.clone(), + effective_model: descriptor.model.clone(), + permission_mode: Some(descriptor.permission_mode), + provider_lifetime_fence_candidates: [60_001, 60_002, 60_003], + }; + let operations = Vec::new(); + let mut state = AcpxDurableState::new( + descriptor, + AuthorizedToolSet { + schema: TOOL_SET_SCHEMA.into(), + schema_version: 1, + catalog_digest: authorized_tool_catalog_digest(&operations).unwrap(), + operations, + }, + launch_profile.canonical_digest().unwrap(), + ); + state.identity = Some(identity); + state.lifecycle = "turn_active".into(); + state.active_turn_id = Some("turn-1".into()); + for request_id in ["input-1", "input-2"] { + state + .push(NormalizedProviderEvent { + event_type: "runtime_request.created".into(), + priority: EventPriority::P0, + payload: json!({"request":pending_input_request(request_id)}), + }) + .unwrap(); + } + state.pending_events.clear(); + let config = test_config(&directory, Some(launch_profile)); + let mut original = AcpxCommandExecutor::with_runner_config(&directory, &config); + original.restore_checked = true; + original.state = Some(state.clone()); + original.save_state().unwrap(); + let mut staged = state.clone(); + staged + .push(NormalizedProviderEvent { + event_type: event_type.into(), + priority: EventPriority::P0, + payload: json!({"requestId":"input-1"}), + }) + .unwrap(); + original.fail_after_state_rename = true; + let error = if event_type == "runtime_request.resolved" { + let error = original.commit_staged_state(staged.clone()).unwrap_err(); + assert_eq!(original.state.as_ref(), Some(&state)); + error + } else { + // Existing cancellation/expiry paths mutate their working + // state before save; the same latch must quarantine them. + original.state = Some(staged.clone()); + original.save_state().unwrap_err() + }; + assert!(error + .to_string() + .contains("injected post-rename sync failure")); + original.fail_after_state_rename = false; + let disk = fs::read(original.state_path()).unwrap(); + assert_eq!( + serde_json::from_slice::(&disk).unwrap(), + staged + ); + for result in [ + original.retained_events().map(|_| ()), + original.poll_events().map(|_| ()), + original.acknowledge_events(0), + original.acknowledge_events(1), + original.restore(), + original.save_state(), + original.shutdown(), + ] { + assert_eq!(result.unwrap_err().to_string(), error.to_string()); + } + for command_type in ["request.resolve", "runner.drain", "runner.suspend"] { + let blocked = original.execute(&Command { + schema: "paperclip.prp.command.v1".into(), + command_id: "blocked-command".into(), + controller_seq: 1, + command_type: command_type.into(), + issued_at: "2026-09-01T00:00:00.000Z".into(), + deadline_at: None, + precondition: None, + payload: json!({"requestId":"input-1", "resolution":{"action":"submit"}}), + }); + assert_eq!(blocked.unwrap_err().to_string(), error.to_string()); + } + assert_eq!(fs::read(original.state_path()).unwrap(), disk); + drop(original); + + let mut recovered = AcpxCommandExecutor::with_runner_config(&directory, &config); + let events = recovered.poll_events().unwrap(); + let settlements: Vec<_> = events + .iter() + .filter(|event| event.event_type.starts_with("runtime_request.")) + .collect(); + assert_eq!(settlements.len(), 2); + assert_eq!(settlements[0].event_type, event_type); + assert_eq!(settlements[0].payload["requestId"], "input-1"); + assert_eq!( + settlements[0].executor_event_id, + staged.pending_events[0].executor_event_id + ); + assert_eq!(settlements[1].event_type, "runtime_request.expired"); + assert_eq!(settlements[1].payload["requestId"], "input-2"); + assert_eq!(settlements[1].payload["replayAllowed"], false); + assert!( + event_sequence(&settlements[0].executor_event_id).unwrap() + < event_sequence(&settlements[1].executor_event_id).unwrap() + ); + assert!(recovered + .resolve_request(&json!({"requestId":"input-1", "resolution":{"action":"submit"}})) + .is_err()); + assert_eq!(recovered.poll_events().unwrap(), events); + recovered.acknowledge_events(events.len()).unwrap(); + drop(recovered); + let mut recovered_again = AcpxCommandExecutor::with_runner_config(&directory, &config); + assert!(recovered_again.poll_events().unwrap().is_empty()); + assert!( + !marker.exists(), + "recovery must not restart a provider or replay input" + ); + fs::remove_dir_all(directory).unwrap(); + } + } + #[test] fn turn_control_rejects_nonstring_mode_before_provider_access() { let directory = temporary_directory("control-mode"); 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 8f4eab51cab..0a1d9bdf5cc 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 @@ -1996,6 +1996,125 @@ mod tests { } } + #[test] + fn response_persistence_failure_with_blocked_drain_remains_indeterminate_on_replay() { + struct PersistenceFailedResponseExecutor { + delivered_responses: usize, + retained_reads: usize, + } + impl CommandExecutor for PersistenceFailedResponseExecutor { + fn execute( + &mut self, + command: &Command, + ) -> Result { + assert_eq!(command.command_type, "request.resolve"); + assert_eq!(command.payload["requestId"], "input-1"); + // Model a real pipe delivery ACK followed by an uncertain + // provider-state write. Its staged settlement is not durable. + self.delivered_responses += 1; + Err(DurableRunnerError::invalid( + "ACPX response state persistence failed after delivery", + )) + } + + fn retained_events(&mut self) -> Result, DurableRunnerError> { + self.retained_reads += 1; + Err(DurableRunnerError::invalid( + "ACPX persistence latch blocks retained settlement", + )) + } + + fn poll_events(&mut self) -> Result, DurableRunnerError> { + panic!("failed response settlement must not poll or restore the provider") + } + + fn acknowledge_events(&mut self, _: usize) -> Result<(), DurableRunnerError> { + panic!("an uncertain settlement must not receive a journal ACK") + } + } + + let directory = std::env::temp_dir().join(format!( + "paperclip-response-persistence-failure-{}", + uuid::Uuid::new_v4() + )); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = PersistenceFailedResponseExecutor { + delivered_responses: 0, + retained_reads: 0, + }; + let mut resolve = command("request.resolve"); + resolve.payload = json!({ + "requestId": "input-1", + "resolution": {"action": "submit", "response": {"answer": "accepted"}}, + }); + + let failure = + process_command(&mut state, &store, &config, &mut executor, &resolve).unwrap_err(); + assert!(failure + .to_string() + .starts_with("ACPX response state persistence failed after delivery")); + assert!(failure + .to_string() + .contains("retained failure evidence remains uncommitted")); + assert!(failure + .to_string() + .contains("ACPX persistence latch blocks retained settlement")); + assert_eq!(executor.delivered_responses, 1); + assert_eq!(executor.retained_reads, 1); + assert_eq!( + state.processed_commands[&resolve.command_id].status, + "pending" + ); + assert!(state.outbox.is_empty()); + assert_eq!(state.highest_source_seq(), 0); + + // Inspect the actual journal before recovery changes pending into + // indeterminate. Neither a failed/completed result nor a resolved event + // may replace the pre-effect marker while the provider snapshot is unsure. + let journal: DurableState = + serde_json::from_slice(&fs::read(store.path()).unwrap()).unwrap(); + assert_eq!( + journal.processed_commands[&resolve.command_id].status, + "pending" + ); + assert!(journal.outbox.is_empty()); + assert_eq!(journal.highest_source_seq(), 0); + + let pending = process_command(&mut state, &store, &config, &mut executor, &resolve) + .unwrap() + .0; + assert_eq!(pending.status, "pending"); + assert_eq!(executor.delivered_responses, 1); + assert_eq!(executor.retained_reads, 1); + + let (mut recovered, existed) = store.load_or_create(&config).unwrap(); + assert!(existed); + assert_eq!( + recovered.processed_commands[&resolve.command_id].status, + "indeterminate" + ); + let mut replacement = CountingExecutor { calls: 0 }; + let replay = process_command(&mut recovered, &store, &config, &mut replacement, &resolve) + .unwrap() + .0; + assert_eq!(replay.status, "indeterminate"); + assert_eq!(replay.result["code"], "execution_indeterminate"); + assert_eq!( + replacement.calls, 0, + "recovery must never resend the response" + ); + assert!(recovered.outbox.is_empty()); + let (reloaded_again, _) = store.load_or_create(&config).unwrap(); + assert_eq!( + reloaded_again.processed_commands[&resolve.command_id], + replay + ); + assert!(reloaded_again.outbox.is_empty()); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn unlimited_lifetime_keeps_attempts_bounded_after_weeks_of_work() { let mut config = config(std::env::temp_dir()); 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 e768abaacc8..9699a80278c 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 @@ -242,7 +242,12 @@ fn pi_prepare_payload(directory: &Path, mode: &str) -> Value { fn pending_acpx_runtime_request( directory: &Path, mode: &str, -) -> (NativeProviderCommandExecutor, DurableRunnerConfig, String) { +) -> ( + NativeProviderCommandExecutor, + DurableRunnerConfig, + String, + u32, +) { let mut config = acpx_config(directory, mode); let mut payload = prepare_payload_with_mode(directory, "codex", mode); if mode.starts_with("permissions-") { @@ -280,9 +285,10 @@ fn pending_acpx_runtime_request( executor .execute(&command(1, "run.prepare", payload)) .unwrap(); - executor + let opened = executor .execute(&command(2, "session.open", json!({}))) .unwrap(); + let process_id = opened.result["processId"].as_u64().unwrap() as u32; executor .execute(&command( 3, @@ -301,7 +307,7 @@ fn pending_acpx_runtime_request( .map(str::to_owned); executor.acknowledge_events(events.len()).unwrap(); if let Some(request_id) = request_id { - return (executor, config, request_id); + return (executor, config, request_id, process_id); } assert!( std::time::Instant::now() < deadline, @@ -369,7 +375,7 @@ fn acpx_response_delivery_survives_crash_before_journaling_without_replaying_the "resolutions-projected-id", ] { let directory = temporary_directory(mode); - let (mut executor, config, request_id) = pending_acpx_runtime_request(&directory, mode); + let (mut executor, config, request_id, _) = pending_acpx_runtime_request(&directory, mode); let resolve = command( 4, "request.resolve", @@ -440,7 +446,7 @@ fn acpx_response_delivery_survives_crash_before_journaling_without_replaying_the fn acpx_delivered_response_is_acknowledged_once_before_session_close() { for mode in ["permissions-interactive", "resolutions"] { let directory = temporary_directory("response-ack"); - let (mut executor, _, request_id) = pending_acpx_runtime_request(&directory, mode); + let (mut executor, _, request_id, _) = pending_acpx_runtime_request(&directory, mode); let response = acpx_runtime_resolution(mode, &request_id); executor .execute(&command(4, "request.resolve", response.clone())) @@ -469,7 +475,8 @@ fn acpx_delivered_response_is_acknowledged_once_before_session_close() { fn acpx_uncommitted_delivery_expires_after_failed_state_write_without_replay() { for mode in ["permissions-interactive", "resolutions"] { let directory = temporary_directory("response-save-failure"); - let (mut executor, config, request_id) = pending_acpx_runtime_request(&directory, mode); + let (mut executor, config, request_id, process_id) = + pending_acpx_runtime_request(&directory, mode); let state_path = directory.join("acpx-provider-state.json"); let previous_path = directory.join("previous-state.json"); let previous = fs::read(&state_path).unwrap(); @@ -485,12 +492,44 @@ fn acpx_uncommitted_delivery_expires_after_failed_state_write_without_replay() { .to_string() .contains("atomically replace ACPX provider state")); assert_eq!(fs::read(&previous_path).unwrap(), previous); - assert_eq!( - executor.retained_events().unwrap()[0].event_type, - "runtime_request.resolved" + // Neither ordinary polling nor the explicit drain/ACK path may publish + // a delivered response whose settlement failed to persist. + for result in [ + executor.retained_events().map(|_| ()), + executor.poll_events().map(|_| ()), + executor.acknowledge_events(0), + executor.acknowledge_events(1), + executor + .execute(&command(5, "runner.drain", json!({}))) + .map(|_| ()), + executor + .execute(&command(6, "request.resolve", response.clone())) + .map(|_| ()), + executor.shutdown(), + ] { + assert_eq!(result.unwrap_err().to_string(), error.to_string()); + } + #[cfg(unix)] + assert!( + !std::process::Command::new("kill") + .args(["-0", &process_id.to_string()]) + .env_clear() + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .unwrap() + .success(), + "failed durable cleanup must still reap the owned sidecar" ); + assert_eq!(fs::read(&previous_path).unwrap(), previous); fs::remove_dir(&state_path).unwrap(); fs::rename(&previous_path, &state_path).unwrap(); + // Repairing the file path does not authorize this uncertain executor + // to resume, retry an approval, or overwrite the recovery snapshot. + assert!(executor.poll_events().is_err()); + assert!(executor.retained_events().is_err()); + assert!(executor.shutdown().is_err()); + assert_eq!(fs::read(&state_path).unwrap(), previous); drop(executor); let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); @@ -513,6 +552,142 @@ fn acpx_uncommitted_delivery_expires_after_failed_state_write_without_replay() { } } +#[test] +fn acpx_failed_settlement_ack_replays_the_same_delivered_event_without_expiry() { + for mode in ["permissions-interactive", "resolutions"] { + let directory = temporary_directory("response-ack-save-failure"); + let (mut executor, config, request_id, _) = pending_acpx_runtime_request(&directory, mode); + let response = acpx_runtime_resolution(mode, &request_id); + executor + .execute(&command(4, "request.resolve", response.clone())) + .unwrap(); + let committed = executor.poll_events().unwrap(); + assert_eq!(committed.len(), 1); + assert_eq!(committed[0].event_type, "runtime_request.resolved"); + let state_path = directory.join("acpx-provider-state.json"); + let previous_path = directory.join("previous-state.json"); + let previous = fs::read(&state_path).unwrap(); + fs::rename(&state_path, &previous_path).unwrap(); + fs::create_dir(&state_path).unwrap(); + // The journal has committed this executorEventId, but retiring the + // provider's retained prefix must not escape a failed state write. + let error = executor.acknowledge_events(1).unwrap_err(); + assert!(error + .to_string() + .contains("atomically replace ACPX provider state")); + for result in [ + executor.retained_events().map(|_| ()), + executor.poll_events().map(|_| ()), + executor.acknowledge_events(0), + executor + .execute(&command(5, "request.resolve", response.clone())) + .map(|_| ()), + executor.shutdown(), + ] { + assert_eq!(result.unwrap_err().to_string(), error.to_string()); + } + assert_eq!(fs::read(&previous_path).unwrap(), previous); + fs::remove_dir(&state_path).unwrap(); + fs::rename(&previous_path, &state_path).unwrap(); + drop(executor); + + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + let events = recovered.poll_events().unwrap(); + assert_eq!(events[0], committed[0]); + assert_eq!( + events + .iter() + .filter(|event| event.event_type.starts_with("runtime_request.")) + .count(), + 1 + ); + assert!(recovered + .execute(&command(6, "request.resolve", response)) + .is_err()); + assert_eq!(recovered.poll_events().unwrap(), events); + recovered.acknowledge_events(events.len()).unwrap(); + assert!(recovered.poll_events().unwrap().is_empty()); + shutdown_recovered_acpx_fixture(&mut recovered, &directory); + drop(recovered); + let mut recovered_again = + NativeProviderCommandExecutor::with_runner_config(&directory, &config); + assert!(recovered_again.poll_events().unwrap().is_empty()); + recovered_again.shutdown().unwrap(); + fs::remove_dir_all(directory).unwrap(); + } +} + +#[test] +fn acpx_uncommitted_expiry_and_cancellation_cannot_escape_through_poll_or_drain() { + for (mode, close) in [ + ("permissions-wrong-ack", false), + ("permissions-interactive", true), + ("resolutions", true), + ] { + let directory = temporary_directory("expiry-save-failure"); + let (mut executor, config, request_id, _) = pending_acpx_runtime_request(&directory, mode); + let state_path = directory.join("acpx-provider-state.json"); + let previous_path = directory.join("previous-state.json"); + let previous = fs::read(&state_path).unwrap(); + fs::rename(&state_path, &previous_path).unwrap(); + fs::create_dir(&state_path).unwrap(); + let response = acpx_runtime_resolution(mode, &request_id); + let error = if close { + executor + .execute(&command(4, "session.close", json!({}))) + .unwrap_err() + } else { + assert!(executor + .execute(&command(4, "request.resolve", response.clone())) + .unwrap_err() + .to_string() + .contains("did not confirm permission resolution")); + executor.poll_events().unwrap_err() + }; + assert!(error + .to_string() + .contains("atomically replace ACPX provider state")); + for result in [ + executor.retained_events().map(|_| ()), + executor.poll_events().map(|_| ()), + executor.acknowledge_events(1), + executor + .execute(&command(5, "runner.drain", json!({}))) + .map(|_| ()), + executor + .execute(&command(6, "request.resolve", response.clone())) + .map(|_| ()), + executor.shutdown(), + ] { + assert_eq!(result.unwrap_err().to_string(), error.to_string()); + } + assert_eq!(fs::read(&previous_path).unwrap(), previous); + fs::remove_dir(&state_path).unwrap(); + fs::rename(&previous_path, &state_path).unwrap(); + assert!(executor.shutdown().is_err()); + assert_eq!(fs::read(&state_path).unwrap(), previous); + drop(executor); + + let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config); + let events = recovered.poll_events().unwrap(); + let settlements: Vec<_> = events + .iter() + .filter(|event| event.event_type.starts_with("runtime_request.")) + .collect(); + assert_eq!(settlements.len(), 1); + assert_eq!(settlements[0].event_type, "runtime_request.expired"); + assert_eq!(settlements[0].payload["requestId"], request_id); + assert_eq!(settlements[0].payload["replayAllowed"], false); + assert!(recovered + .execute(&command(7, "request.resolve", response)) + .is_err()); + assert_eq!(recovered.poll_events().unwrap(), events); + recovered.acknowledge_events(events.len()).unwrap(); + shutdown_recovered_acpx_fixture(&mut recovered, &directory); + fs::remove_dir_all(directory).unwrap(); + } +} + #[test] fn acpx_failed_or_cancelled_responses_retain_one_durable_settlement() { for (mode, close) in [ @@ -521,7 +696,7 @@ fn acpx_failed_or_cancelled_responses_retain_one_durable_settlement() { ("resolutions", true), ] { let directory = temporary_directory("response-expiry"); - let (mut executor, config, request_id) = pending_acpx_runtime_request(&directory, mode); + let (mut executor, config, request_id, _) = pending_acpx_runtime_request(&directory, mode); if close { executor .execute(&command(4, "session.close", json!({})))