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 <noreply@paperclip.ing>

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

packages/paperclip-runner/runner/crates/runner-core/src/acpx_provider_backend.rs

@@ -683,6 +683,9 @@ pub struct AcpxCommandExecutor {
     session: Option<AcpxProviderSession>,
     restore_checked: bool,
     restore_error: Option<DurableRunnerError>,
+    persistence_error: Option<DurableRunnerError>,
+    #[cfg(test)]
+    fail_after_state_rename: bool,
     launch_profile: Option<AcpxLaunchProfile>,
 }
 
@@ -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<Vec<PolledEvent>, 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::<AcpxDurableState>(&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");

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<CommandExecution, DurableRunnerError> {
+                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<Vec<PolledEvent>, DurableRunnerError> {
+                self.retained_reads += 1;
+                Err(DurableRunnerError::invalid(
+                    "ACPX persistence latch blocks retained settlement",
+                ))
+            }
+
+            fn poll_events(&mut self) -> Result<Vec<PolledEvent>, 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());

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!({})))