Runner: verify provider eligibility before parking (#19133)

## Thinking Path > - Paperclip coordinates AI agents and their execution. > - Hosted Runners can retain a provider process between turns. > - The host must prove that process is idle before allowing its VM to sleep. > - A completed turn alone does not rule out an active goal or trailing provider work. > - This PR adds a conservative park-readiness hook for the existing Codex process. > - The later host-control consumer can use this proof with its identity and sequence fences. ## Linked Issues or Issue Description **Current behavior** There is no provider eligibility hook for host parking. The earlier implementation in #19060 also missed an active goal that arrived during its tail drain. **Proposed behavior** Default every executor to ineligible. Permit the surviving native Codex process only when its completed-turn authority still matches and it has no live or unsettled work. **Reason and benefit** Keep active goals and unresolved provider obligations awake. Review the provider proof separately from socket handling and the native park/reclaim loop. ## What Changed - Add `host_park_ready` with a default false result. - Refresh the configured credential observer before checking pending delivery. Refuse parking during credential delivery, without selecting or restoring a provider. - Verify the current Codex process and completed-turn identity; reject startup, active turns/goals/tools, and unsettled event queues. - Require completion actually observed by that process. Accepted-finish reconciliation still preserves ordinary result authority without qualifying a restored process to park. - Refuse known helper lineage, whose diagnostics do not prove that helper work has finished. - Drain bounded trailing output through the existing accounting path. Retain goal updates and clears in durable state and normal event delivery, then recheck live goal state. - Add process tests for normal eligibility and the late-goal race. Assert that a restored process is ineligible until it completes an authoritative turn in its new generation. This PR is based on master, which contains the merged Codex (#19132) and ACPX (#19153) recovery foundations. The private socket codec in #19129 is merged; #19140 consumes this proof in the shared native loop. This PR does not advertise parking support. A separate ACPX readiness change must prove its live sidecar/session has no active or unsettled work. Both Codex App Server and qualified ACPX are required in the remaining park/reclaim, Core restart, eviction, and live sleep/wake acceptance phases. ## Verification - `cargo test --offline --locked --manifest-path packages/paperclip-runner/runner/Cargo.toml -p paperclip-runner-core --test codex_provider hosted_park` — 5 passed. - `ordinary_recovery_restores_a_completed_idle_session` passes and asserts that the genuinely restored provider cannot park before it completes an authoritative turn in its new generation. The fixture proves two provider starts and only one admitted turn. - The initial live-goal recheck mutation failed as expected. Review then identified that goal changes also needed durable retention. The new activation and clear cases verify saved goal state, retained events, and eligibility after delivery. - Existing goal, active-goal attachment, warm-tail, and late-accounting process regressions passed after sharing the goal event path. - The initial late-goal fixture timed out because it omitted `--durable-turn-ids`, which enables its existing first-turn gate. Added that fixture switch; both park tests then passed. - The credential-observation regression fails before the fix and passes afterward. Three focused credential cases prove durable replay and acknowledgement, fresh ACPX startup, and readiness without provider selection. The three provider parking checks also pass after the fix. - Full regression testing is left to GitHub CI. - Changed lines scanned for secrets, personal data, and private deployment references before push. Current head `28d41b083f58ee93db4305b1217719796bf3171d` adds one scoped review-fix commit to readiness `e8ba82374e0f72c82ac3195d4b1b3798c9072bab`, based on master `3d0d178ff7051aa44e07cd873fd4281357e0d815` without duplicating either recovery foundation. The PR remains eight parking-specific files. Review identified quiet helper work and reconstructed accepted-finish authority that could incorrectly qualify. The provider now records its actually observed terminal authority separately in memory, and parking refuses known helper lineage in both the saved state and the live inventory after draining. No durable field, lifecycle, public API, or ordinary recovery policy changed. Both process regressions fail when the parking guards are removed and pass when restored. Seventeen focused checks pass, including ordinary recovery, helper requests, receipt replay, credentials, and ACPX ineligibility. Rustfmt, diff and added-line scans passed. All exact-head CI is green: [PR run](https://github.com/paperclipai/paperclip/actions/runs/42087773928), [Runner Windows](https://github.com/paperclipai/paperclip/actions/runs/42087773720), and [Docker Runner](https://github.com/paperclipai/paperclip/actions/runs/42087773712) passed; Storybook was skipped. Standard review 15 scored 5/5 before the authorized follow-up [Apex review](https://github.com/paperclipai/paperclip/runs/127471966630). Apex review 16 also scored [5/5 on this exact head](https://github.com/paperclipai/paperclip/pull/19133#issuecomment-6645371152), with zero new comments and zero unresolved threads. On this current head, the first server shard 11/12 [job 127458190155](https://github.com/paperclipai/paperclip/actions/runs/42087773928/job/127458190155) failed the unchanged decision-model concurrency test: after releasing the database lock, its `waitFor` observed one provider call instead of two. The identical file passed all 37 tests on the prior reviewed head. One targeted unchanged-head retry [job 127466020677](https://github.com/paperclipai/paperclip/actions/runs/42087773928/job/127466020677) passed all 1,779 tests, including those 37 decision-model cases; 13 tests were skipped. No code, test timeout, or CI configuration changed between attempts. The original failure's cause remains unestablished. On prior reviewed head `e8ba82374e0f72c82ac3195d4b1b3798c9072bab`, the first `General tests (workspaces-c)` job [127389771745](https://github.com/paperclipai/paperclip/actions/runs/42068371976/job/127389771745) failed the unchanged database SIGTERM test at `packages/db/src/test-postgres-command.test.ts:44`: `JSON.parse` read empty readiness-file content immediately after an existence check. The fixture writes directly to that path, leaving a create-before-content race. One targeted unchanged-head retry [127398676711](https://github.com/paperclipai/paperclip/actions/runs/42068371976/job/127398676711) passed all **313 tests in 72 files**, including SIGTERM and SIGINT. No source or timeout changed between attempts. On the superseded `ca088abd552b` head, one Cloud sign-in recovery case returned to the expected task URL but missed its heading within the five-second assertion window; the other 40 cases passed. One targeted unchanged-head retry (job 127136265280) passed all **41 cases**. No test timeout or application code changed between attempts, and the first failure's cause remains unestablished. The superseded c7 parent failed eight retained-cleanup settlement cases because prepared restoration started a provider during a drain-only epoch. The corrected parent preserves ordinary prepared behavior; the new completed-source path excludes prepared providers. The parent reports all 19 settlement rows passing. This PR adds no prepared-session proof or restoration policy. ## Risks Known helpers remain conservatively ineligible even after their progress diagnostics: this slice has no authoritative helper-completion ledger. The eligibility check drains the existing bounded provider tail and can persist late accounting. That accounting must settle before parking succeeds. This hook alone does not prove VM suspension or vendor goal behavior across a process restart. Live sleep/wake qualification remains a later rollout check. ## Model Used OpenAI GPT-6.1 Sol (`gpt-6.1-sol`), High reasoning, through Codex with tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-08 01:57:31 +01:00 · 2026-11-05 11:24:36 -08:00
1 parent 377e98f1a2
commit a6f786479e
@@ -120,3 +120,22 @@ and goal with A’s saved idle state. After reading the goal, it checks updates
 that arrive during the final status read and keeps them queued for delivery.
 As with Codex, a prepared session can continue or be cleaned up. This handoff
 requires a completed turn. Suspended sessions use their usual recovery path.
+
+## Host park eligibility
+
+`CommandExecutor::host_park_ready` defaults to false. Native Codex opts in only
+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.
+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,
+or advertise hosted parking support. The host-control consumer must separately
+validate the exact Runner identity and acknowledged sequence cut.
@@ -1788,6 +1788,17 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                    )?;
                    continue;
                }
+                if args.iter().any(|value| value == "--quiet-helper-turn") {
+                    send(json!({"method":"item/completed","params":{
+                        "threadId":state.thread_id,"turnId":provider_turn_id,
+                        "item":{"id":"spawn-quiet-helper","type":"collabAgentToolCall",
+                            "tool":"spawnAgent","status":"completed",
+                            "senderThreadId":state.thread_id,"receiverThreadIds":["descendant-0"]}
+                    }}))?;
+                    send(json!({"method":"turn/started","params":{
+                        "threadId":"descendant-0","turnId":"quiet-helper-turn"
+                    }}))?;
+                }
                if descendant_notifications {
                    // Real Codex announces a child's MCP startup before its
                    // thread/started notification establishes parent lineage.
@@ -2024,6 +2035,8 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                            let turn_id = provider_turn_id.clone();
                            let unsafe_tail =
                                args.iter().any(|arg| arg == "--unsafe-after-late-usage");
+                            let active_goal = args.iter().any(|arg| arg == "--late-active-goal");
+                            let clear_goal = args.iter().any(|arg| arg == "--late-goal-cleared");
                            let duplicate_burst = args
                                .iter()
                                .any(|arg| arg == "--late-accounting-duplicate-burst");
@@ -2033,6 +2046,23 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                                    thread::sleep(Duration::from_millis(1));
                                }
                                if gate.is_file() {
+                                    if active_goal || clear_goal {
+                                        let method = if clear_goal {
+                                            "thread/goal/cleared"
+                                        } else {
+                                            "thread/goal/updated"
+                                        };
+                                        let goal = if clear_goal {
+                                            Value::Null
+                                        } else {
+                                            json!({"objective":"Continue work","status":"active"})
+                                        };
+                                        let _ = send(
+                                            json!({"method":method,"params":{"threadId":thread_id,"goal":goal}}),
+                                        );
+                                        let _ = fs::write(gate.with_extension("sent"), "sent");
+                                        return;
+                                    }
                                    if duplicate_burst {
                                        for _ in 0..33 {
                                            let _ = send(
@@ -781,6 +781,7 @@ pub struct CodexProvider {
    expected_shutdown: bool,
    process_generation: u64,
    completed_turn_authority: Option<CompletedTurnAuthority>,
+    observed_completed_turn_authority: Option<CompletedTurnAuthority>,
    active_turn_result_authoritative: bool,
    completion_reconciliation_pending: bool,
    goal_allows_autonomous_turns: bool,
@@ -1201,6 +1202,7 @@ impl CodexProvider {
            expected_shutdown: false,
            process_generation,
            completed_turn_authority: None,
+            observed_completed_turn_authority: None,
            active_turn_result_authoritative: false,
            completion_reconciliation_pending: false,
            goal_allows_autonomous_turns: false,
@@ -1518,7 +1520,10 @@ impl CodexProvider {
                        }
                        if matches!(
                            method.as_str(),
-                            "thread/tokenUsage/updated" | "paperclip/resumeUsageSnapshot"
+                            "thread/tokenUsage/updated"
+                                | "paperclip/resumeUsageSnapshot"
+                                | "thread/goal/updated"
+                                | "thread/goal/cleared"
                        ) {
                            accounting.push((method, params));
                            continue;
@@ -1702,6 +1707,21 @@ impl CodexProvider {
        })
    }

+    pub(crate) fn observed_completed_turn_authority(&self) -> Option<(u64, &str)> {
+        self.observed_completed_turn_authority
+            .as_ref()
+            .map(|authority| {
+                (
+                    authority.process_generation,
+                    authority.provider_turn_id.as_str(),
+                )
+            })
+    }
+
+    pub(crate) fn has_known_descendants(&self) -> bool {
+        !self.descendant_thread_ids.is_empty()
+    }
+
    pub(crate) fn mark_active_turn_result_authoritative(&mut self) -> Result<(), LocalRunnerError> {
        if self.active_provider_turn_id.is_none() || self.ambiguous_turn_start_pending {
            return Err(LocalRunnerError::invalid(
@@ -1819,6 +1839,10 @@ impl CodexProvider {
        self.restart_idle_identity_epoch()
    }

+    pub(crate) fn has_active_goal(&self) -> bool {
+        self.goal_allows_autonomous_turns
+    }
+
    pub fn get_goal(&mut self) -> Result<Value, LocalRunnerError> {
        let result = self.request("thread/goal/get", json!({"threadId": self.thread_id}))?;
        self.goal_allows_autonomous_turns =
@@ -3214,6 +3238,8 @@ impl CodexProvider {
                self.active_turn_result_authoritative = false;
                self.expected_shutdown = true;
                self.completed_turn_authority = completed_turn_authority;
+                // Restored result authority cannot prove this process saw a terminal.
+                self.observed_completed_turn_authority = self.completed_turn_authority.clone();
                self.completion_reconciliation_pending =
                    terminal_event_type == "turn.completed" || result_authoritative;
                // The provider terminal is authoritative once received. Clear
@@ -306,6 +306,11 @@ pub enum TerminalDeliveryReconciliation {
 }

 pub trait CommandExecutor {
+    /// Prove the existing provider is idle without restoring or starting it.
+    fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        Ok(false)
+    }
+
    fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError>;

    fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> {
@@ -278,6 +278,18 @@ impl NativeProviderCommandExecutor {
 }

 impl CommandExecutor for NativeProviderCommandExecutor {
+    fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        // Observe credentials without selecting or restoring a provider.
+        self.credentials.poll(true)?;
+        if self.credentials.pending().is_some() {
+            return Ok(false);
+        }
+        match self.selected.as_mut() {
+            Some(SelectedExecutor::LocalFacade(executor)) => executor.host_park_ready(),
+            _ => Ok(false),
+        }
+    }
+
    fn observe_recovery_state(
        &mut self,
        state: &crate::durable::DurableState,
@@ -761,6 +761,35 @@ fn terminal_completion_rejected(state: &CodexProviderState) -> bool {
    state.active_terminal_tool_rejected && state.active_provider_result_fingerprint.is_none()
 }

+fn retain_goal_notification(
+    state: &mut CodexProviderState,
+    method: &str,
+    params: &Value,
+    working_now: bool,
+) -> Result<(), DurableRunnerError> {
+    let event_type = match method {
+        "thread/goal/updated" => {
+            state.goal = normalize_codex_goal(params, working_now);
+            "session.goal.updated"
+        }
+        "thread/goal/cleared" => {
+            state.goal = None;
+            "session.goal.cleared"
+        }
+        _ => return Ok(()),
+    };
+    state.goal_revision = state.goal_revision.saturating_add(1);
+    state.push_event(NormalizedProviderEvent {
+        event_type: event_type.to_owned(),
+        priority: EventPriority::P0,
+        payload: goal_event_payload(
+            state.goal.as_ref(),
+            state.goal_capability.as_ref(),
+            state.goal_revision,
+        ),
+    })
+}
+
 fn goal_continues_after_turn(state: &CodexProviderState, goal_active: bool) -> bool {
    goal_active && !terminal_completion_rejected(state)
 }
@@ -2966,6 +2995,7 @@ impl CodexCommandExecutor {
                DurableRunnerError::invalid("Codex provider has not been prepared")
            })?;
            for (method, params) in tail {
+                retain_goal_notification(state, &method, &params, false)?;
                let events = normalize_provider_notification(state, &method, &params)?;
                state.extend_events(events)?;
            }
@@ -5035,31 +5065,7 @@ impl CodexCommandExecutor {
                            });
                        }
                    }
-                    if method == "thread/goal/updated" {
-                        state.goal = normalize_codex_goal(&params, working_now);
-                        state.goal_revision = state.goal_revision.saturating_add(1);
-                        state.push_event(NormalizedProviderEvent {
-                            event_type: "session.goal.updated".to_owned(),
-                            priority: EventPriority::P0,
-                            payload: goal_event_payload(
-                                state.goal.as_ref(),
-                                state.goal_capability.as_ref(),
-                                state.goal_revision,
-                            ),
-                        })?;
-                    } else if method == "thread/goal/cleared" {
-                        state.goal = None;
-                        state.goal_revision = state.goal_revision.saturating_add(1);
-                        state.push_event(NormalizedProviderEvent {
-                            event_type: "session.goal.cleared".to_owned(),
-                            priority: EventPriority::P0,
-                            payload: goal_event_payload(
-                                None,
-                                state.goal_capability.as_ref(),
-                                state.goal_revision,
-                            ),
-                        })?;
-                    }
+                    retain_goal_notification(state, &method, &params, working_now)?;
                    if let Some(reconciliation) = goal_reconciliation {
                        match reconciliation {
                            Ok(snapshot) => {
@@ -5272,6 +5278,61 @@ impl CodexCommandExecutor {
 }

 impl CommandExecutor for CodexCommandExecutor {
+    fn host_park_ready(&mut self) -> Result<bool, DurableRunnerError> {
+        let Some(state) = self.state.as_ref() else {
+            return Ok(false);
+        };
+        if state.config.provider != "codex"
+            || state.config.driver != "codex_app_server"
+            || self.provider.is_none()
+            || !state.completed_turn_authoritative
+            || state.completed_turn_process_generation != Some(state.provider_process_generation)
+            || state.completed_provider_turn_id.is_none()
+            || state.active_provider_turn_id.is_some()
+            || state.ambiguous_turn_start_pending
+            || state.startup_attempt.is_some()
+            || state.receipt_limit_interrupt_pending
+            || state
+                .goal
+                .as_ref()
+                .is_some_and(|goal| goal.status == "active" || goal.working_now)
+            || state.tool_bridge.pending_calls().next().is_some()
+            || !state.descendant_thread_ids.is_empty()
+            || !state.pending_events.is_empty()
+            || !state.queued_events.is_empty()
+            || state.lifecycle == "reconciliation_required"
+        {
+            return Ok(false);
+        }
+        self.retain_completed_accounting_tail()?;
+        let state = self
+            .state
+            .as_ref()
+            .expect("surviving native provider state");
+        if !state.pending_events.is_empty() || !state.queued_events.is_empty() {
+            return Ok(false);
+        }
+        let provider = self
+            .provider
+            .as_mut()
+            .expect("checked surviving native provider");
+        // Tail draining can consume a late goal update before the executor
+        // journal sees it. Check the live provider after that drain.
+        if provider.has_active_goal()
+            || provider.has_known_descendants()
+            || provider.observed_completed_turn_authority() != provider.completed_turn_authority()
+            || provider.completed_turn_authority()
+                != state
+                    .completed_turn_process_generation
+                    .zip(state.completed_provider_turn_id.as_deref())
+        {
+            return Ok(false);
+        }
+        Ok(provider
+            .warm_run_attachment_blockers()
+            .map_err(|_| DurableRunnerError::invalid("native Codex parking proof failed"))?
+            .is_empty())
+    }
    fn can_reconcile_result_delivery(&mut self) -> Result<bool, DurableRunnerError> {
        // ProviderToolBridge persists the exact call/result and rejects changed
        // receipts. Delivery cannot execute a server-side semantic operation.
@@ -147,6 +147,9 @@ fn ordinary_recovery_restores_a_completed_idle_session() {
    );
    assert_eq!(call_count(&directory, "process-start"), 2);
    assert_eq!(call_count(&directory, "turn/start"), 1);
+    // Warm attachment readiness does not prove this new provider generation
+    // completed its own authoritative turn.
+    assert!(!executor.host_park_ready().unwrap());
    executor.shutdown().unwrap();
    fs::remove_dir_all(directory).unwrap();
 }
@@ -525,6 +528,323 @@ fn call_count(directory: &Path, method: &str) -> usize {
        .count()
 }

+#[test]
+fn hosted_park_proof_requires_the_surviving_settled_native_codex_process() {
+    let directory = temporary_directory("hosted-park-proof");
+    let config = provider_config(&directory, &["--record-process-start"]);
+    let mut executor =
+        CodexCommandExecutor::with_runner_config(&directory, &durable_config(&directory));
+    assert!(!executor.host_park_ready().unwrap());
+    executor
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider":config}),
+        ))
+        .unwrap();
+    assert!(!executor.host_park_ready().unwrap());
+    executor
+        .execute(&command("open", 2, "session.open", json!({})))
+        .unwrap();
+    assert!(!executor.host_park_ready().unwrap());
+    executor
+        .execute(&command(
+            "turn",
+            3,
+            "turn.start",
+            json!({"text":"Complete the admitted native turn."}),
+        ))
+        .unwrap();
+    assert!(!executor.host_park_ready().unwrap());
+    wait_for_executor_event(&mut executor, "turn.completed");
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    while !executor.host_park_ready().unwrap() {
+        assert!(std::time::Instant::now() < deadline);
+        poll_and_ack(&mut executor).unwrap();
+    }
+    assert_eq!(call_count(&directory, "process-start"), 1);
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    executor
+        .execute(&command(
+            "goal",
+            4,
+            "session.goal.set",
+            json!({"objective":"Continue autonomous work.","status":"active"}),
+        ))
+        .unwrap();
+    for _ in 0..8 {
+        poll_and_ack(&mut executor).unwrap();
+    }
+    assert!(!executor.host_park_ready().unwrap());
+    executor.shutdown().unwrap();
+    assert!(!executor.host_park_ready().unwrap());
+    assert_eq!(call_count(&directory, "process-start"), 1);
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn hosted_park_rejects_a_quiet_helper_after_parent_completion() {
+    let directory = temporary_directory("hosted-park-helper");
+    let config = provider_config(&directory, &["--quiet-helper-turn"]);
+    let mut executor =
+        CodexCommandExecutor::with_runner_config(&directory, &durable_config(&directory));
+    executor
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider":config}),
+        ))
+        .unwrap();
+    executor
+        .execute(&command("open", 2, "session.open", json!({})))
+        .unwrap();
+    executor
+        .execute(&command(
+            "turn",
+            3,
+            "turn.start",
+            json!({"text":"Complete the parent while its helper keeps working."}),
+        ))
+        .unwrap();
+    wait_for_executor_event(&mut executor, "turn.completed");
+    while !poll_and_ack(&mut executor).unwrap().is_empty() {}
+    let state: Value =
+        serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap())
+            .unwrap();
+    assert_eq!(state["completedTurnAuthoritative"], true);
+    assert!(state["descendantThreadIds"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .any(|id| id == "descendant-0"));
+    assert!(!executor.host_park_ready().unwrap());
+    executor.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn hosted_park_rejects_reconciled_finish_without_a_live_completion() {
+    let directory = temporary_directory("hosted-park-reconciled-finish");
+    let result_schema: Value = serde_json::from_slice(
+        &fs::read(
+            PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+                .join("../../../protocol/schemas/result.schema.json"),
+        )
+        .unwrap(),
+    )
+    .unwrap();
+    let input = json!({"schema":"paperclip.run_result.v1", "reportedWorkDisposition":"done", "summary":"Completed once",
+        "completionClaim":{"contractRevision":"contract-1", "objectiveSatisfied":true,
+            "criteria":[{"criterionId":"output", "status":"satisfied", "evidenceRefs":[]}], "remainingWork":[]},
+        "evidence":[], "verification":[], "attentionRequests":[], "artifacts":[]});
+    let input_path = directory.join("finish-input.json");
+    fs::write(&input_path, serde_json::to_vec(&input).unwrap()).unwrap();
+    let config = provider_config(
+        &directory,
+        &[
+            "--record-process-start",
+            "--durable-turn-ids",
+            "--terminal-result-file",
+            input_path.to_str().unwrap(),
+        ],
+    );
+    let operation = AuthorizedTool {
+        operation_id: "paperclip_finish".to_owned(),
+        version: 1,
+        description: "Report completed work.".to_owned(),
+        input_schema: result_schema,
+        response_schema: json!({"type":"object"}),
+    };
+    let tools = AuthorizedToolSet {
+        schema: TOOL_SET_SCHEMA.to_owned(),
+        schema_version: 1,
+        catalog_digest: authorized_tool_catalog_digest(std::slice::from_ref(&operation)).unwrap(),
+        operations: vec![operation],
+    };
+    let runner_config = durable_config(&directory);
+    let mut first = CodexCommandExecutor::with_runner_config(&directory, &runner_config);
+    first
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider":config, "authorizedTools":tools,
+        "completionContract":{"revision":"contract-1", "criterionIds":["output"]}}),
+        ))
+        .unwrap();
+    first
+        .execute(&command("open", 2, "session.open", json!({})))
+        .unwrap();
+    first
+        .execute(&command(
+            "turn",
+            3,
+            "turn.start",
+            json!({"text":"Finish before the completion is journaled."}),
+        ))
+        .unwrap();
+    wait_for_executor_event(&mut first, "semantic_tool.input");
+    first
+        .execute(&command(
+            "accept",
+            4,
+            "semantic_tool.result",
+            json!({"callId":"terminal-result-call",
+        "operationId":"paperclip_finish", "result":{"accepted":true}, "isError":false}),
+        ))
+        .unwrap();
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    loop {
+        let ended = fs::read(directory.join("fake-state.json"))
+            .ok()
+            .and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
+            .is_some_and(|live| live.get("activeTurnId") == Some(&Value::Null));
+        if ended {
+            break;
+        }
+        assert!(std::time::Instant::now() < deadline);
+        std::thread::sleep(std::time::Duration::from_millis(1));
+    }
+    let state: Value =
+        serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap())
+            .unwrap();
+    assert!(state["activeProviderResultFingerprint"].is_string());
+    assert_eq!(state["completedTurnAuthoritative"], false);
+    drop(first);
+    let mut recovered = CodexCommandExecutor::with_runner_config(&directory, &runner_config);
+    recovered
+        .execute(&command("restore", 5, "session.snapshot", json!({})))
+        .unwrap();
+    let mut events = Vec::new();
+    loop {
+        let next = poll_and_ack(&mut recovered).unwrap();
+        if next.is_empty() {
+            break;
+        }
+        events.extend(next);
+    }
+    assert_eq!(
+        events
+            .iter()
+            .filter(|event| event.event_type == "run.terminal")
+            .count(),
+        1
+    );
+    let snapshot = recovered
+        .execute(&command("snapshot", 6, "session.snapshot", json!({})))
+        .unwrap();
+    assert_eq!(snapshot.result["warmAttachReady"], true);
+    assert_eq!(call_count(&directory, "process-start"), 2);
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    assert!(!recovered.host_park_ready().unwrap());
+    recovered.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn hosted_park_preserves_goal_activation_during_tail_drain() {
+    verify_late_goal_retention(false);
+}
+
+#[test]
+fn hosted_park_preserves_goal_clear_during_tail_drain() {
+    verify_late_goal_retention(true);
+}
+
+fn verify_late_goal_retention(clear_goal: bool) {
+    let directory = temporary_directory("hosted-park-late-goal");
+    let gate = directory.join("release-goal");
+    let gate_arg = gate.to_string_lossy().into_owned();
+    let config = provider_config(
+        &directory,
+        &[
+            "--record-process-start",
+            "--durable-turn-ids",
+            "--late-accounting-tail-gate",
+            &gate_arg,
+            if clear_goal {
+                "--late-goal-cleared"
+            } else {
+                "--late-active-goal"
+            },
+        ],
+    );
+    let mut executor =
+        CodexCommandExecutor::with_runner_config(&directory, &durable_config(&directory));
+    executor
+        .execute(&command(
+            "prepare",
+            1,
+            "run.prepare",
+            json!({"provider":config}),
+        ))
+        .unwrap();
+    executor
+        .execute(&command("open", 2, "session.open", json!({})))
+        .unwrap();
+    executor
+        .execute(&command(
+            "turn",
+            3,
+            "turn.start",
+            json!({"text":"Complete one turn."}),
+        ))
+        .unwrap();
+    wait_for_executor_event(&mut executor, "turn.completed");
+    if clear_goal {
+        executor
+            .execute(&command(
+                "paused-goal",
+                4,
+                "session.goal.set",
+                json!({"objective":"Wait for work", "status":"paused"}),
+            ))
+            .unwrap();
+    }
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
+    while !executor.host_park_ready().unwrap() {
+        assert!(std::time::Instant::now() < deadline);
+        poll_and_ack(&mut executor).unwrap();
+    }
+    fs::write(&gate, "release").unwrap();
+    while !gate.with_extension("sent").is_file() {
+        assert!(std::time::Instant::now() < deadline);
+        std::thread::sleep(std::time::Duration::from_millis(1));
+    }
+    // Do not poll the executor: the goal must first be observed by the park
+    // probe's tail drain, after its initial saved-state eligibility check.
+    assert!(!executor.host_park_ready().unwrap());
+    assert!(!executor.host_park_ready().unwrap());
+    let saved: Value =
+        serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap())
+            .unwrap();
+    let expected_status = if clear_goal {
+        Value::Null
+    } else {
+        json!("active")
+    };
+    let expected_event = if clear_goal {
+        "session.goal.cleared"
+    } else {
+        "session.goal.updated"
+    };
+    assert_eq!(saved["goal"]["status"], expected_status);
+    let events = poll_and_ack(&mut executor).unwrap();
+    assert!(events.iter().any(|event| event.event_type == expected_event
+        && event.payload["goal"]["status"] == expected_status));
+    assert_eq!(
+        executor.host_park_ready().unwrap(),
+        clear_goal,
+        "only the cleared goal becomes parkable after event delivery"
+    );
+    assert_eq!(call_count(&directory, "process-start"), 1);
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    executor.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
 #[test]
 fn failed_provider_startup_is_persistently_fenced_before_another_process_can_resume() {
    let directory = temporary_directory("failed-startup-fence");
@@ -62,6 +62,17 @@ fn credential_observation_allows_fresh_acpx_session_startup() {
 #[test]
 #[cfg(unix)]
 fn credential_updates_replay_through_the_native_executor_until_durable_ack() {
+    exercise_credential_updates(false);
+}
+
+#[test]
+#[cfg(unix)]
+fn hosted_park_refreshes_credentials_without_selecting_a_provider() {
+    exercise_credential_updates(true);
+}
+
+#[cfg(unix)]
+fn exercise_credential_updates(parking: bool) {
    let root = temporary_directory("credential-updates");
    let root = fs::canonicalize(root).unwrap();
    let auth = root.join("auth.json");
@@ -88,16 +99,21 @@ fn credential_updates_replay_through_the_native_executor_until_durable_ack() {
    let mut executor = NativeProviderCommandExecutor::with_runner_config(&root, &runner_config);
    assert!(executor.poll_events().unwrap().is_empty());
    write_auth("refreshed");
-    let _ = executor.execute(&Command {
-        schema: "paperclip.prp.command.v1".to_owned(),
-        command_id: "close-unselected".to_owned(),
-        controller_seq: 1,
-        command_type: "session.close".to_owned(),
-        issued_at: "2026-10-15T12:00:00.000Z".to_owned(),
-        deadline_at: None,
-        precondition: None,
-        payload: json!({}),
-    });
+    if parking {
+        assert!(!executor.host_park_ready().unwrap());
+    } else {
+        let _ = executor.execute(&Command {
+            schema: "paperclip.prp.command.v1".to_owned(),
+            command_id: "close-unselected".to_owned(),
+            controller_seq: 1,
+            command_type: "session.close".to_owned(),
+            issued_at: "2026-10-15T12:00:00.000Z".to_owned(),
+            deadline_at: None,
+            precondition: None,
+            payload: json!({}),
+        });
+    }
+    assert!(!root.join("codex-provider-state.json").exists());
    let updates = executor.retained_events().unwrap();
    assert_eq!(updates.len(), 1);
    assert_eq!(updates[0].event_type, "auth.credentials.updated");
@@ -1521,6 +1537,7 @@ fn resumes_an_idle_acpx_session_in_a_cold_replacement_runner() {
    let original = executor
        .execute(&command(2, "session.open", json!({})))
        .unwrap();
+    assert!(!executor.host_park_ready().unwrap());
    executor
        .execute(&command(
            3,
@@ -1530,6 +1547,8 @@ fn resumes_an_idle_acpx_session_in_a_cold_replacement_runner() {
        .unwrap();
    let events = executor.poll_events().unwrap();
    executor.acknowledge_events(events.len()).unwrap();
+    // Generic ACPX recovery does not opt this provider into host parking.
+    assert!(!executor.host_park_ready().unwrap());
    executor
        .execute(&command(4, "runner.suspend", json!({})))
        .unwrap();
@@ -1547,6 +1566,7 @@ fn resumes_an_idle_acpx_session_in_a_cold_replacement_runner() {
        .execute(&command(1, "run.attach", payload))
        .unwrap();
    assert_eq!(resumed.result["status"], "resumed");
+    assert!(!replacement.host_park_ready().unwrap());
    assert_eq!(
        resumed.result["providerSessionId"],
        original.result["providerSessionId"]