Verify live Claude and Codex ACPX eligibility before parking (#19295)

packages/paperclip-runner/runner/DURABLE_TRANSPORT.md

@@ -157,12 +157,20 @@ for the existing process that completed the current turn. An eligibility check
 never selects or restores a provider. A restored process must complete its own
 authoritative turn before it can be eligible. Reconstructing an accepted finish
 during recovery preserves result delivery but does not qualify that process.
-ACPX remains ineligible in this slice, including after qualified recovery.
 Active turns, goals, startup attempts, tool calls, pending credentials, and
 unsettled events block parking. Known helper threads also block it: their
 diagnostic lineage does not prove that their work finished. The bounded
 completed-turn tail drain retains late accounting and goal changes for durable
 state and normal event delivery.
+
+Claude and Codex through ACPX also need a successful turn in the live sidecar.
+Runner checks its qualified profile, session identity and source run A. It gives
+that sidecar 100 ms to report its goal and session state. If the sidecar is slow
+or declines, Runner stays attached. Runner keeps at most one unanswered read and
+recognizes its late reply. Events received during that read still belong to A
+and are delivered before B attaches. The existing policy still replaces the
+sidecar at the next turn or run attachment; the saved provider identity continues.
+
 The check also inspects the live provider
 after that drain, so a late active-goal notification cannot be hidden by the
 older executor journal. This hook does not detach a controller, stop a process,

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

@@ -2307,6 +2307,11 @@ impl AcpxCommandExecutor {
     }
 
     fn retain_buffered_provider_events(&mut self) -> Result<(), DurableRunnerError> {
+        if let Some(session) = self.session.as_mut() {
+            session.settle_park_observation().map_err(|error| {
+                DurableRunnerError::invalid(format!("ACPX observation settlement failed: {error}"))
+            })?;
+        }
         loop {
             let count = self
                 .session
@@ -2516,6 +2521,59 @@ fn goal_terminal_disposition<'a>(
 }
 
 impl CommandExecutor for AcpxCommandExecutor {
+    fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        self.ensure_persistence_healthy()?;
+        let Some(state) = self.state.as_ref() else {
+            return Ok(false);
+        };
+        if !matches!(state.descriptor.agent.as_str(), "codex" | "claude")
+            || self.session.is_none()
+            || state.lifecycle != "session_open"
+            || state.provider_exit_unconfirmed
+            || state.descriptor.run_id != self.context.run_id
+            || state.identity.is_none()
+            || state.active_turn_id.is_some()
+            || !state.pending_events.is_empty()
+            || !state.pending_runtime_requests.is_empty()
+            || (!state.goal_projection.is_null() && !goal_is_dormant(&state.goal_projection))
+            || !state.descriptor.names_qualified_release()
+            || self.candidate_admission().is_some()
+            || !acpx_release_profile(&state.descriptor.agent)
+                .is_some_and(|profile| !profile.qualification_pending)
+        {
+            return Ok(false);
+        }
+        state.validate(&self.context, &self.launch_profile_digest()?)?;
+        let session = self
+            .session
+            .as_mut()
+            .expect("checked surviving ACPX session");
+        if state.identity.as_ref() != Some(session.identity()) {
+            return Ok(false);
+        }
+        let event_sequence = session.observed_event_sequence();
+        let ready = session
+            .host_park_ready(std::time::Instant::now() + Duration::from_millis(100))
+            .map_err(|error| {
+                DurableRunnerError::invalid(format!("ACPX parking observation failed: {error}"))
+            })?;
+        // Commit new provider observations through the ordinary event path.
+        // A retained prefix must be delivered and ACKed before a later park.
+        self.poll_provider(false)?;
+        Ok(ready
+            && self
+                .session
+                .as_ref()
+                .is_some_and(|session| session.observed_event_sequence() == event_sequence)
+            && self.state.as_ref().is_some_and(|state| {
+                state.pending_events.is_empty()
+                    && state.lifecycle == "session_open"
+                    && state.active_turn_id.is_none()
+                    && !state.provider_exit_unconfirmed
+                    && (state.goal_projection.is_null() || goal_is_dormant(&state.goal_projection))
+            }))
+    }
+
     fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError> {
         self.restore()?;
         if self

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

@@ -1,6 +1,6 @@
 use crate::generated_acpx_profiles::acpx_release_profile;
 use std::path::{Path, PathBuf};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
@@ -351,6 +351,7 @@ pub struct AcpxProviderSession {
     failure: Option<LocalRunnerError>,
     transport_terminated: bool,
     runtime_retired: bool,
+    completed_turn_observed: bool,
 }
 
 pub(crate) fn goal_is_dormant(projection: &Value) -> bool {
@@ -407,6 +408,7 @@ impl AcpxProviderSession {
             failure: None,
             transport_terminated: false,
             runtime_retired: false,
+            completed_turn_observed: false,
         })
     }
 
@@ -446,6 +448,25 @@ impl AcpxProviderSession {
         let snapshot = self
             .transport
             .request(GeneratedAcpxSidecarCommand::SessionSnapshot, json!({}))?;
+        self.validate_dormant_snapshot(&snapshot)?;
+        // The status barrier can queue a newer goal while the snapshot is in flight.
+        // Inspect without consuming facts that ordinary polling still owns.
+        let mut checked_state = self.state.clone();
+        for event in self.transport.buffered_events().skip(prior_events) {
+            if event.event_type != GeneratedAcpxSidecarEventType::RuntimeGoal {
+                continue;
+            }
+            checked_state.accept_event(event)?;
+            if !goal_is_dormant(&event.payload) {
+                return Err(LocalRunnerError::invalid(
+                    "saved dormant ACPX received an active or unknown goal during its snapshot",
+                ));
+            }
+        }
+        Ok(())
+    }
+
+    fn validate_dormant_snapshot(&self, snapshot: &Value) -> Result<(), LocalRunnerError> {
         let identity: AcpxProviderSessionIdentity =
             serde_json::from_value(snapshot["identity"].clone()).map_err(|_| {
                 LocalRunnerError::invalid("saved dormant ACPX snapshot identity is invalid")
@@ -463,23 +484,49 @@ impl AcpxProviderSession {
                 "saved dormant ACPX snapshot has unexpected work or identity",
             ));
         }
-        // The status barrier can queue a newer goal while the snapshot is in flight.
-        // Inspect without consuming facts that ordinary polling still owns.
-        let mut checked_state = self.state.clone();
-        for event in self.transport.buffered_events().skip(prior_events) {
-            if event.event_type != GeneratedAcpxSidecarEventType::RuntimeGoal {
-                continue;
-            }
-            checked_state.accept_event(event)?;
-            if !goal_is_dormant(&event.payload) {
-                return Err(LocalRunnerError::invalid(
-                    "saved dormant ACPX received an active or unknown goal during its snapshot",
-                ));
-            }
-        }
         Ok(())
     }
 
+    pub(crate) fn observed_event_sequence(&self) -> u64 {
+        self.transport.observed_event_sequence()
+    }
+
+    pub(crate) fn settle_park_observation(&mut self) -> Result<(), LocalRunnerError> {
+        self.transport.settle_observation()
+    }
+
+    pub(crate) fn host_park_ready(&mut self, deadline: Instant) -> Result<bool, LocalRunnerError> {
+        if self.closed
+            || self.failure.is_some()
+            || self.transport_terminated
+            || self.runtime_retired
+            || !self.completed_turn_observed
+            || self.state.active_turn_id().is_some()
+            || self.state.has_pending_requests()
+            || self.transport.buffered_events().len() != 0
+        {
+            return Ok(false);
+        }
+        let Some(goal) = self
+            .transport
+            .observe(GeneratedAcpxSidecarCommand::SessionGoalGet, deadline)?
+        else {
+            return Ok(false);
+        };
+        if !goal_is_dormant(&goal) {
+            return Ok(false);
+        }
+        let Some(snapshot) = self
+            .transport
+            .observe(GeneratedAcpxSidecarCommand::SessionSnapshot, deadline)?
+        else {
+            return Ok(false);
+        };
+        // Polling still owns every frame observed during these reads.
+        Ok(self.validate_dormant_snapshot(&snapshot).is_ok()
+            && self.transport.buffered_events().len() == 0)
+    }
+
     /// Read the same live sidecar before exposing its in-memory pending ledger.
     /// A durable file alone cannot authorize a request after process loss.
     pub fn verify_live_request_snapshot(
@@ -661,6 +708,7 @@ impl AcpxProviderSession {
         }
         self.tool_bridge = next_tool_bridge;
         self.reserved_tool_bridge = next_reserved_tool_bridge;
+        self.completed_turn_observed = false;
         Ok(response)
     }
 
@@ -956,6 +1004,16 @@ impl AcpxProviderSession {
                 reconciled_events.push(event);
             }
         }
+        for event in &reconciled_events {
+            if let AcpxProviderStateEvent::TurnTerminal {
+                turn_id, status, ..
+            } = event
+            {
+                self.completed_turn_observed = *status
+                    == crate::acpx_event_payload::AcpxTurnStatus::Completed
+                    && self.state.active_turn_id() == Some(turn_id.as_str());
+            }
+        }
         self.state = next_state;
         self.tool_bridge = next_bridge;
         self.reserved_tool_bridge = next_reserved_bridge;
@@ -2170,6 +2228,7 @@ mod mode_tests {
             failure: None,
             transport_terminated: false,
             runtime_retired: false,
+            completed_turn_observed: false,
         }
     }
 

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

@@ -86,6 +86,7 @@ pub struct AcpxSidecarTransport {
     stderr_tail: BoundedLogBuffer,
     stderr_categories: BTreeSet<&'static str>,
     poisoned: bool,
+    abandoned_observation: Option<u64>,
 }
 
 // A fresh Pi process verifies and copies its native closure before ACP admission.
@@ -387,6 +388,7 @@ impl AcpxSidecarTransport {
             stderr_tail: BoundedLogBuffer::new(32, 8 * 1024),
             stderr_categories: BTreeSet::new(),
             poisoned: false,
+            abandoned_observation: None,
         })
     }
 
@@ -469,6 +471,86 @@ impl AcpxSidecarTransport {
         command: GeneratedAcpxSidecarCommand,
         params: Value,
     ) -> Result<CommandOutcome, LocalRunnerError> {
+        let request_id = self.send_request(command, params)?;
+        self.receive_response(
+            command,
+            request_id,
+            Instant::now() + self.command_timeout(command),
+        )?
+        .ok_or_else(|| self.request_timeout_error(command))
+    }
+
+    /// A park observation may expire without revoking a healthy provider.
+    /// Retain one ID so its late reply cannot masquerade as an ordinary reply.
+    pub(crate) fn observe(
+        &mut self,
+        command: GeneratedAcpxSidecarCommand,
+        deadline: Instant,
+    ) -> Result<Option<Value>, LocalRunnerError> {
+        if self.poisoned {
+            return Err(LocalRunnerError::invalid(
+                "ACPX sidecar transport is unavailable",
+            ));
+        }
+        if !matches!(
+            command,
+            GeneratedAcpxSidecarCommand::SessionGoalGet
+                | GeneratedAcpxSidecarCommand::SessionSnapshot
+        ) {
+            return Err(LocalRunnerError::invalid(
+                "ACPX observation must be read-only",
+            ));
+        }
+        if self.abandoned_observation.is_some() || Instant::now() >= deadline {
+            return Ok(None);
+        }
+        let result = self
+            .send_request(command, json!({}))
+            .and_then(|request_id| {
+                let result = self.receive_response(command, request_id, deadline)?;
+                if result.is_none() {
+                    self.abandoned_observation = Some(request_id);
+                }
+                Ok(result)
+            });
+        match result {
+            Ok(Some(CommandOutcome::Success(value))) => Ok(Some(value)),
+            Ok(Some(CommandOutcome::Rejected(_))) => Ok(None),
+            Ok(None) => Ok(None),
+            Err(error) => {
+                self.poison();
+                Err(error)
+            }
+        }
+    }
+
+    pub(crate) fn observed_event_sequence(&self) -> u64 {
+        self.last_event_sequence
+    }
+
+    /// Ordinary turn admission drains an expired read before replacing its sidecar.
+    pub(crate) fn settle_observation(&mut self) -> Result<(), LocalRunnerError> {
+        let Some(request_id) = self.abandoned_observation.take() else {
+            return Ok(());
+        };
+        let command = GeneratedAcpxSidecarCommand::SessionSnapshot;
+        match self.receive_response(command, request_id, Instant::now() + self.request_timeout) {
+            Ok(Some(_)) => Ok(()),
+            result => {
+                self.poison();
+                Err(match result {
+                    Err(error) => error,
+                    _ => self.request_timeout_error(command),
+                })
+            }
+        }
+    }
+
+    fn send_request(
+        &mut self,
+        command: GeneratedAcpxSidecarCommand,
+        params: Value,
+    ) -> Result<u64, LocalRunnerError> {
         if !params.is_object() {
             return Err(LocalRunnerError::invalid(
                 "ACPX sidecar command params must be an object",
@@ -502,19 +584,30 @@ impl AcpxSidecarTransport {
         })?;
         self.next_request_id = request_id + 1;
 
-        let timeout = self.command_timeout(command);
-        let deadline = Instant::now() + timeout;
+        Ok(request_id)
+    }
+
+    fn receive_response(
+        &mut self,
+        command: GeneratedAcpxSidecarCommand,
+        request_id: u64,
+        deadline: Instant,
+    ) -> Result<Option<CommandOutcome>, LocalRunnerError> {
         loop {
             let remaining = deadline.saturating_duration_since(Instant::now());
             if remaining.is_zero() {
-                return Err(self.request_timeout_error(command));
+                return Ok(None);
             }
             let Some(line) = self.receive_stdout_line(remaining, command.as_str())? else {
-                return Err(self.request_timeout_error(command));
+                return Ok(None);
             };
             match parse_frame(&line)? {
                 ParsedFrame::Event(event) => self.buffer_event(event)?,
                 ParsedFrame::Response(response) => {
+                    if self.abandoned_observation == Some(response.id) {
+                        self.abandoned_observation = None;
+                        continue;
+                    }
                     if response.id != request_id {
                         return Err(LocalRunnerError::invalid(format!(
                             "ACPX sidecar response id mismatch: expected {request_id}, received {}",
@@ -522,20 +615,20 @@ impl AcpxSidecarTransport {
                         )));
                     }
                     if response.ok {
-                        return Ok(CommandOutcome::Success(
+                        return Ok(Some(CommandOutcome::Success(
                             response.result.unwrap_or_else(|| json!({})),
-                        ));
+                        )));
                     }
                     let error = response.error.expect("failed response has validated error");
                     let provider_failure = error.provider_failure.clone();
-                    return Ok(CommandOutcome::Rejected(LocalRunnerError::invalid(
+                    return Ok(Some(CommandOutcome::Rejected(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))));
                 }
             }
         }
@@ -545,18 +638,32 @@ impl AcpxSidecarTransport {
         &mut self,
         timeout: Duration,
     ) -> Result<Option<AcpxSidecarEvent>, LocalRunnerError> {
-        let Some(line) = self.receive_stdout_line(timeout, "event.poll")? else {
-            return Ok(None);
-        };
-        match parse_frame(&line)? {
-            ParsedFrame::Event(event) => {
-                self.validate_event_sequence(event.sequence)?;
-                Ok(Some(event))
+        let deadline = Instant::now() + timeout;
+        loop {
+            let Some(line) = self.receive_stdout_line(
+                deadline.saturating_duration_since(Instant::now()),
+                "event.poll",
+            )?
+            else {
+                return Ok(None);
+            };
+            match parse_frame(&line)? {
+                ParsedFrame::Event(event) => {
+                    self.validate_event_sequence(event.sequence)?;
+                    return Ok(Some(event));
+                }
+                ParsedFrame::Response(response)
+                    if self.abandoned_observation == Some(response.id) =>
+                {
+                    self.abandoned_observation = None;
+                }
+                ParsedFrame::Response(response) => {
+                    return Err(LocalRunnerError::invalid(format!(
+                        "ACPX sidecar emitted response {} without a pending request",
+                        response.id
+                    )))
+                }
             }
-            ParsedFrame::Response(response) => Err(LocalRunnerError::invalid(format!(
-                "ACPX sidecar emitted response {} without a pending request",
-                response.id
-            ))),
         }
     }
 
@@ -1404,3 +1511,7 @@ mod tests {
         );
     }
 }
+
+#[cfg(all(test, unix))]
+#[path = "tests/acpx_park_observation.rs"]
+mod park_observation_tests;

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-acpx-sidecar.rs

@@ -124,6 +124,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     "pendingToolCount":if fault == "tool" || source_pending_completion.is_some() {1} else {0}, "pendingInputCount":0,
                     "pendingRuntimeRequests":[]});
             }
+            if command == "session.snapshot" && fault == "foreign-snapshot" {
+                response["result"]["identity"]["agentSessionId"] = json!("foreign-session");
+            }
             if command == "session.goal.get" {
                 source_goal_revision += 1;
                 response["result"]["providerRevision"] = json!(source_goal_revision);
@@ -153,6 +156,16 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                 )?;
                 next_sequence += 1;
             }
+            if command == "session.snapshot" && fault == "slow-snapshot" {
+                std::thread::sleep(std::time::Duration::from_secs(1));
+                write_json(
+                    &mut stdout,
+                    &json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+                    "sequence":next_sequence,"eventType":"runtime.diagnostic","runId":source_run,"turnId":null,
+                    "payload":{"code":"slow_park_observation","message":"Retained after a slow read."}}),
+                )?;
+                next_sequence += 1;
+            }
             if command == "session.snapshot" && fault == "goal-during-snapshot" {
                 source_goal_revision += 1;
                 let frame = json!({
@@ -172,6 +185,12 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                 response["result"]["sessionGoals"] = json!({"availability":"available", "actions":["set","pause","resume","clear"],
                     "autonomousUpdates":true,"persistentAcrossResume":true,"maxObjectiveChars":4000,"tokenBudgetControl":false,"usageReporting":false});
             }
+            if (command == "session.goal.get" && fault == "reject-goal")
+                || (command == "session.snapshot" && fault == "reject-snapshot")
+            {
+                response = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":id,"ok":false,
+                    "error":{"code":"temporarily_unavailable","message":"Optional read is temporarily unavailable.","retryable":true}});
+            }
             write_json(&mut stdout, &response)?;
             if command == "turn.start" {
                 let turn = request["params"]["turnId"].as_str().unwrap();

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

@@ -332,6 +332,7 @@ impl CommandExecutor for NativeProviderCommandExecutor {
         }
         match self.selected.as_mut() {
             Some(SelectedExecutor::LocalFacade(executor)) => executor.host_park_ready(),
+            Some(SelectedExecutor::Acpx(executor)) => executor.host_park_ready(),
             _ => Ok(false),
         }
     }

packages/paperclip-runner/runner/crates/runner-core/src/tests/acpx_park_observation.rs

@@ -0,0 +1,94 @@
+use super::*;
+
+fn scripted_transport(reply_id: u64) -> AcpxSidecarTransport {
+    let event = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,
+        "sequence":1,"eventType":"runtime.diagnostic","runId":null,"turnId":null,
+        "payload":{"code":"park_tail","message":"Preserve this exact event."}});
+    let late = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":1,"ok":true,"result":{"late":true}});
+    let ordinary = json!({"protocolVersion":GENERATED_ACPX_SIDECAR_PROTOCOL_VERSION,"id":reply_id,"ok":true,"result":{"ordinary":true}});
+    AcpxSidecarTransport::start(&AcpxSidecarTransportConfig {
+        command: "/bin/sh".into(),
+        args: vec!["-c".to_owned(), format!("read first; sleep 0.1; printf '%s\\n' '{event}' '{late}'; read second; printf '%s\\n' '{ordinary}'; cat >/dev/null")],
+        verified_launch: None,
+        request_timeout: Duration::from_secs(1),
+        shutdown_grace: Duration::from_millis(20),
+    }).unwrap()
+}
+
+#[test]
+fn late_park_observation_reply_preserves_ordinary_commands_and_event_polling() {
+    for poll_first in [false, true] {
+        let mut transport = scripted_transport(2);
+        let start = Instant::now();
+        assert!(transport
+            .observe(
+                GeneratedAcpxSidecarCommand::SessionSnapshot,
+                start + Duration::from_millis(10)
+            )
+            .unwrap()
+            .is_none());
+        assert_eq!(transport.abandoned_observation, Some(1));
+        assert!(transport
+            .observe(
+                GeneratedAcpxSidecarCommand::SessionGoalGet,
+                Instant::now() + Duration::from_millis(10)
+            )
+            .unwrap()
+            .is_none());
+        assert_eq!(transport.next_request_id, 2);
+        let event = if poll_first {
+            let event = transport
+                .poll_event(Duration::from_secs(1))
+                .unwrap()
+                .unwrap();
+            assert!(transport
+                .poll_event(Duration::from_millis(10))
+                .unwrap()
+                .is_none());
+            event
+        } else {
+            assert_eq!(
+                transport
+                    .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({}))
+                    .unwrap(),
+                json!({"ordinary":true})
+            );
+            transport.poll_event(Duration::ZERO).unwrap().unwrap()
+        };
+        assert_eq!(event.sequence, 1);
+        assert_eq!(
+            event.payload,
+            json!({"code":"park_tail","message":"Preserve this exact event."})
+        );
+        assert_eq!(transport.abandoned_observation, None);
+        if poll_first {
+            assert_eq!(
+                transport
+                    .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({}))
+                    .unwrap(),
+                json!({"ordinary":true})
+            );
+        }
+        assert!(!transport.poisoned);
+        transport.shutdown().unwrap();
+    }
+}
+
+#[test]
+fn late_park_observation_does_not_relax_ordinary_reply_identity() {
+    for wrong_id in [1, 99] {
+        let mut transport = scripted_transport(wrong_id);
+        assert!(transport
+            .observe(
+                GeneratedAcpxSidecarCommand::SessionSnapshot,
+                Instant::now() + Duration::from_millis(10)
+            )
+            .unwrap()
+            .is_none());
+        let error = transport
+            .request(GeneratedAcpxSidecarCommand::SessionGoalGet, json!({}))
+            .unwrap_err();
+        assert!(error.to_string().contains("response id mismatch"));
+        assert!(transport.poisoned);
+    }
+}

packages/paperclip-runner/runner/crates/runner-core/tests/acpx_park_readiness.rs

@@ -125,6 +125,169 @@ fn settle(executor: &mut NativeProviderCommandExecutor) -> Vec<serde_json::Value
     }
 }
 
+#[test]
+fn hosted_park_slow_acpx_observation_preserves_the_provider_and_next_turn() {
+    for agent in ["codex", "claude"] {
+        let (fixture, _config, mut executor, opened) = fixture(agent);
+        assert!(
+            !executor.host_park_ready().unwrap(),
+            "opening is not completed work"
+        );
+        executor
+            .execute(&command(
+                3,
+                "turn.start",
+                json!({"turnId":"first", "text":"Complete first work."}),
+            ))
+            .unwrap();
+        settle(&mut executor);
+        assert!(
+            executor.host_park_ready().unwrap(),
+            "completed surviving ACPX must become eligible"
+        );
+        fs::write(fixture.0.join("fault"), "slow-snapshot").unwrap();
+        let start = Instant::now();
+        assert!(!executor.host_park_ready().unwrap());
+        assert!(
+            start.elapsed() < Duration::from_millis(500),
+            "park probe blocked ordinary work"
+        );
+        assert!(
+            !executor.host_park_ready().unwrap(),
+            "an unfinished observation cannot prove readiness"
+        );
+        assert!(std::process::Command::new("kill")
+            .args(["-0", &opened["processId"].to_string()])
+            .status()
+            .unwrap()
+            .success());
+        fs::write(fixture.0.join("fault"), "").unwrap();
+        let second = executor
+            .execute(&command(
+                5,
+                "turn.start",
+                json!({"turnId":"second", "text":"Continue on the healthy session."}),
+            ))
+            .unwrap();
+        let events = settle(&mut executor);
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| event["payload"]["code"] == "slow_park_observation")
+                .count(),
+            1
+        );
+        assert!(events
+            .iter()
+            .any(|event| event["payload"]["message"] == "Retained after a slow read."));
+        let snapshot = executor
+            .execute(&command(6, "session.snapshot", json!({})))
+            .unwrap();
+        assert_eq!(
+            snapshot.result["providerSessionId"],
+            opened["providerSessionId"]
+        );
+        // Existing MCP receipt epoch isolation replaces the sidecar on the next turn.
+        assert!(second.result["status"] == "accepted");
+        assert!(executor.host_park_ready().unwrap());
+        let journal = fs::read_to_string(fixture.0.join("sidecar.jsonl")).unwrap();
+        assert_eq!(
+            journal
+                .lines()
+                .filter(|line| line.contains("processStart"))
+                .count(),
+            2
+        );
+        assert_eq!(
+            journal
+                .lines()
+                .filter(|line| line.contains("\"command\":\"turn.start\""))
+                .count(),
+            2
+        );
+        executor.shutdown().unwrap();
+    }
+}
+
+#[test]
+fn hosted_park_acpx_preserves_late_goals_and_rejects_foreign_or_busy_observations() {
+    for agent in ["codex", "claude"] {
+        let (fixture, _config, mut executor, _) = fixture(agent);
+        executor
+            .execute(&command(
+                3,
+                "turn.start",
+                json!({"turnId":"first", "text":"Complete."}),
+            ))
+            .unwrap();
+        settle(&mut executor);
+        for fault in [
+            "active",
+            "tool",
+            "goal",
+            "foreign-snapshot",
+            "reject-goal",
+            "reject-snapshot",
+        ] {
+            fs::write(fixture.0.join("fault"), fault).unwrap();
+            assert!(
+                !executor.host_park_ready().unwrap(),
+                "{agent} admitted {fault}"
+            );
+        }
+        fs::write(fixture.0.join("fault"), "").unwrap();
+        assert!(
+            executor.host_park_ready().unwrap(),
+            "a refused optional read must leave the provider healthy"
+        );
+        fs::write(fixture.0.join("fault"), "goal-during-snapshot").unwrap();
+        assert!(!executor.host_park_ready().unwrap());
+        let retained = executor.retained_events().unwrap();
+        assert!(retained.iter().any(|event| event
+            .payload
+            .pointer("/goal/status")
+            .and_then(Value::as_str)
+            == Some("active")));
+        assert_eq!(executor.retained_events().unwrap(), retained);
+        assert!(!executor.host_park_ready().unwrap());
+        executor.shutdown().unwrap();
+    }
+}
+
+#[test]
+fn hosted_park_acpx_requires_completion_in_this_lifetime_and_exact_authority() {
+    let (fixture, mut config, mut executor, _) = fixture("claude");
+    executor
+        .execute(&command(
+            3,
+            "turn.start",
+            json!({"turnId":"first", "text":"Complete."}),
+        ))
+        .unwrap();
+    settle(&mut executor);
+    assert!(executor.host_park_ready().unwrap());
+    config.run_id = "other-run".to_owned();
+    executor.rotate_authority(&config);
+    assert!(!executor.host_park_ready().unwrap());
+    config.run_id = "run-1".to_owned();
+    drop(executor);
+    let before = fs::read_to_string(fixture.0.join("sidecar.jsonl")).unwrap();
+    let mut recovered = NativeProviderCommandExecutor::with_runner_config(&fixture.0, &config);
+    assert!(!recovered.host_park_ready().unwrap());
+    assert_eq!(
+        fs::read_to_string(fixture.0.join("sidecar.jsonl")).unwrap(),
+        before
+    );
+    recovered
+        .execute(&command(4, "session.snapshot", json!({})))
+        .unwrap();
+    assert!(
+        !recovered.host_park_ready().unwrap(),
+        "restored identity is not a completed current-generation turn"
+    );
+    recovered.shutdown().unwrap();
+}
+
 #[test]
 fn ordinary_acpx_read_tail_survives_immediate_next_turn_without_a_park_probe() {
     let (fixture, _config, mut executor, _) = fixture("claude");
@@ -291,3 +454,18 @@ fn rejected_read_tail_returns_a_command_error_without_starting_another_turn() {
     assert_eq!(state["lifecycle"], "closed");
     assert_eq!(state["activeTurnId"], Value::Null);
 }
+
+#[test]
+fn hosted_park_acpx_does_not_enable_an_unqualified_park_provider() {
+    let (_fixture, _config, mut executor, _) = fixture("grok");
+    executor
+        .execute(&command(
+            3,
+            "turn.start",
+            json!({"turnId":"first", "text":"Complete on an otherwise qualified release."}),
+        ))
+        .unwrap();
+    settle(&mut executor);
+    assert!(!executor.host_park_ready().unwrap());
+    executor.shutdown().unwrap();
+}