diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md
index 57f40c22111..ea9c726a09d 100644
--- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md
+++ b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md
@@ -1,14 +1,20 @@
 # Durable PRP transport
 
-This layer gives `paperclip-runnerd` a provider-neutral PRP v1 transport. The
-Paperclip server invokes durable mode only for a selected, flag-enabled
-`paperclip_runner` agent or recovery of its persisted native run. Codex is the
-only installed provider; other providers remain unavailable.
+Core uses durable PRP v1 for a selected, flag-enabled `paperclip_runner`
+agent or recovery of its persisted run. On a fresh launch or restart,
+`paperclip-runnerd` selects durable mode from its connection arguments.
+
+- With `--connect-url <url>`, the Runner dials Core.
+- With `--listen-address 0.0.0.0 --listen-path /api/runner/v1/connect/<run-id>`,
+  the Runner listens for Core. `--listen-port` defaults to `43127`.
+
+Choose one form. Both require `PAPERCLIP_RUNNER_BOOTSTRAP_TICKET`.
+Recovery uses the same entry point. The Runner rejects `--source-reconcile`.
 
 ## Trust boundary
 
-- The runner accepts only `ws://` destinations whose complete DNS result is
-  loopback. Resolution happens once and reconnects reuse the pinned addresses.
+- For outbound `ws://`, every resolved address must be loopback. The Runner
+  resolves the destination once and reuses the pinned addresses on reconnect.
 - A bootstrap ticket is read from `PAPERCLIP_RUNNER_BOOTSTRAP_TICKET`, removed
   from the environment immediately, and never sent over the socket. Both peers
   prove possession through HMAC-SHA-256.
@@ -63,13 +69,41 @@ the P0 reserve. When the soft limit is reached, the runner enters backpressure
 and rejects the new non-P0 event rather than silently losing it. Exhausting the
 P0 reserve is an explicit unrecoverable condition.
 
-## Current boundary
+## Recovering Codex between turns
 
-Durable mode is selected only when `paperclip-runnerd` receives
-`--connect-url`. Its executor accepts a Codex app-server descriptor and bound
-completion contract through `run.prepare`, owns the provider process group,
-resumes the persisted Codex thread after runner restart, and translates
-provider notifications to PRP events. The server supplies the bootstrap ticket
-only through the child environment, stores the process identity for bounded
-cancellation, and waits for the durable result and terminal pair. The existing
-local fake-runner mode remains unchanged.
+Suppose run A finishes a turn, its Runner exits, and run B wants to continue
+the conversation. Core uses ordinary recovery for A, then attaches B with
+authenticated `run.attach`. If the original process is alive, its owner can
+verify and adopt it instead.
+
+**Is A idle?** Before launch, the host driver (`sourceRecoveryVersion: 1`)
+checks that A’s process stopped and its saved session was idle. After launch,
+the Runner (`recoveryIntegrityVersion: 1`) verifies that Codex restored an idle
+session when no command effect is uncertain. It reads the goal, then the
+thread. An unexpected turn or active goal blocks recovery, even if the goal
+changes during the thread read. Accepted updates stay queued for delivery.
+
+Active runs use same-run recovery. Indeterminate commands follow the replay
+rules above. A prepared session can continue normally, but this handoff
+requires a completed turn. Reads and cleanup leave a prepared provider unopened.
+
+**Who owns the session?** Core matches A’s Runner instance, environment lease,
+run, normalized session, turn and item. History reads require A’s thread. A
+resume request returns saved state without asking Codex to resume twice. Retained events and usage
+belong to A. The authenticated `authorityRunId` identifies the current owner
+(A for same-run recovery). Losing that owner blocks new commands and dispatch.
+
+If Core loses its cached preparation, the preparer uses A’s frozen
+`recoveryLaunch.requirements`, exact identity and authenticated owner. It
+leaves A’s configuration in place and the cache empty. B inherits A’s recorded
+launch policy on later attachments. Explicit settings must agree after
+provider defaults are applied. Authenticated context and instruction grants
+keep their usual precedence and compatibility checks.
+
+**When can B start work?** After authenticated attachment, Core persists B’s
+fresh snapshot and checkpoint and runs any recovery callback. B can then
+start work. If a write or callback fails, Core can retry the handoff.
+Cancellation cleanup only settles retained work.
+
+Live qualification still needs to verify that restoring a provider starts no
+work, deployed storage honors fsync, and the old process has exited.
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
index fad7b860c79..b97440c45e9 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs
@@ -889,6 +889,7 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
     let pre_response_notification = args
         .iter()
         .any(|value| value == "--notification-before-response");
+    let mut goal_update_on_next_thread_read: Option<String> = None;
     let goal_policy_disabled = args.iter().any(|value| value == "--goal-policy-disabled");
     let goal_autostart = args.iter().any(|value| value == "--goal-autostart");
     let goal_autocontinue = args.iter().any(|value| value == "--goal-autocontinue");
@@ -1457,6 +1458,14 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                 {
                     return Err("thread/read must not hydrate turns".into());
                 }
+                if let Some(status) = goal_update_on_next_thread_read.take() {
+                    state.goal = Some(json!({"objective":"Final read goal", "status":status}));
+                    save_state(&state_path, &state)?;
+                    log_call(call_log.as_deref(), "late-goal-on-thread-read")?;
+                    send(
+                        json!({"method":"thread/goal/updated", "params":{"threadId":state.thread_id,"goal":state.goal}}),
+                    )?;
+                }
                 let turns = state
                     .active_turn_id
                     .as_ref()
@@ -1478,6 +1487,8 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
             }))?,
             "thread/goal/get" => {
                 send(json!({"id": id, "result": {"goal": state.goal}}))?;
+                goal_update_on_next_thread_read = argument(&args, "--goal-on-thread-read-file")
+                    .and_then(|path| fs::read_to_string(path).ok());
                 if args
                     .iter()
                     .any(|value| value == "--idle-protocol-failure-on-goal-probe")
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs
index 0362b8a554a..9b0d056156d 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs
@@ -121,6 +121,7 @@ fn build_metadata() -> serde_json::Value {
         "packageName": "@paperclipai/paperclip-runner",
         "packageVersion": env!("CARGO_PKG_VERSION"),
         "binaryContractVersion": 2,
+        "recoveryIntegrityVersion": 1,
         "durableSessionCapabilities": ["unlimited_runtime", "connection_lease_renewal"],
         "externalProviderCapabilities": ["openai_dot_mcp", "grok_bot_mcp", "muse_external_v1"],
         "nativeExecutionVersion": 1,
@@ -521,6 +522,11 @@ fn run(args: &[String]) -> Result<(), LocalRunnerError> {
         println!("{}", metadata_with_provider_packs(args)?);
         return Ok(());
     }
+    if args.iter().any(|arg| arg == "--source-reconcile") {
+        return Err(LocalRunnerError::invalid(
+            "--source-reconcile is no longer supported; use ordinary durable recovery",
+        ));
+    }
     apply_oom_score_adj(args)?;
     apply_task_memory_cgroup(args)?;
     if args.iter().any(|argument| argument == "--connect-url")
@@ -652,9 +658,22 @@ mod tests {
         assert!(durable_connect_url(&mixed, "run_1").is_err());
     }
 
+    #[test]
+    fn legacy_source_mode_is_rejected_before_any_startup() {
+        let error = run(&[
+            "--listen-port".to_owned(),
+            "1234".to_owned(),
+            "--source-reconcile".to_owned(),
+        ])
+        .unwrap_err();
+        assert!(error.to_string().contains("no longer supported"));
+    }
+
     #[test]
     fn build_metadata_advertises_the_remote_transport_contract() {
         let metadata = build_metadata();
+        assert_eq!(metadata["recoveryIntegrityVersion"], 1);
+        assert!(metadata.get("sourceRecoveryVersion").is_none());
         assert_eq!(metadata["schema"], RUNNERD_BUILD_METADATA_SCHEMA);
         assert_eq!(metadata["binaryContractVersion"], 2);
         assert_eq!(
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
index a5bd41f72d0..db5efd7226d 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/codex_provider.rs
@@ -2239,6 +2239,28 @@ impl CodexProvider {
         Ok(())
     }
 
+    pub(crate) fn read_thread_after_recovery_goal(
+        &mut self,
+        goal_is_dormant: fn(&Value) -> bool,
+    ) -> Result<Value, LocalRunnerError> {
+        // The preceding goal reply supersedes earlier notifications. Inspect only
+        // frames received by this read, and leave them queued for normal delivery.
+        let first = self.pending_messages.len();
+        let snapshot = self.read_thread()?;
+        for message in self.pending_messages.iter().skip(first) {
+            if message.value["method"] == "thread/goal/updated" {
+                let params = &message.value["params"];
+                validate_notification_binding(&self.thread_id, None, params)?;
+                if !params["goal"].is_object() || !goal_is_dormant(params) {
+                    return Err(LocalRunnerError::invalid(
+                        "saved dormant Codex session resumed an active or unknown goal",
+                    ));
+                }
+            }
+        }
+        Ok(snapshot)
+    }
+
     pub fn read_thread(&mut self) -> Result<Value, LocalRunnerError> {
         // Probing provider state does not supersede an authoritative terminal.
         // A later replacement turn must still establish its own identity.
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs
index 97bf9644745..98f9967e400 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs
@@ -308,6 +308,10 @@ pub enum TerminalDeliveryReconciliation {
 pub trait CommandExecutor {
     fn execute(&mut self, command: &Command) -> Result<CommandExecution, DurableRunnerError>;
 
+    fn observe_recovery_state(&mut self, _state: &DurableState) -> Result<(), DurableRunnerError> {
+        Ok(())
+    }
+
     /// Opt in only when semantic result delivery validates an exact durable
     /// call receipt and cannot dispatch the underlying business operation.
     /// Ordinary commands and providers without this proof remain indeterminate.
@@ -462,6 +466,9 @@ pub fn run_durable_runner<E: CommandExecutor>(
     {
         return Ok(());
     }
+    if recovered {
+        executor.observe_recovery_state(&state)?;
+    }
     if recovered && state.warm_transition.is_none() {
         state.reconnect_count = state.reconnect_count.saturating_add(1);
         state.record_diagnostic("runner restored its durable identity after process recovery");
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs
index 1741d687db2..bf25c7cce29 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/state.rs
@@ -359,6 +359,24 @@ pub struct DurableState {
 }
 
 impl DurableState {
+    pub(crate) fn recovery_pending_effects(&self) -> usize {
+        self.processed_commands
+            .values()
+            .filter(|command| {
+                matches!(command.status.as_str(), "pending" | "indeterminate")
+                    && !matches!(
+                        command.command_type.as_str(),
+                        "session.snapshot"
+                            | "session.goal.get"
+                            | "runner.drain"
+                            | "runner.suspend"
+                            | "runner.shutdown"
+                            | "turn.stop"
+                    )
+            })
+            .count()
+    }
+
     pub(crate) fn new(config: &DurableRunnerConfig) -> Self {
         Self {
             identity_output: IdentityOutputBuffer::default(),
@@ -1541,7 +1559,7 @@ fn sensitive_key(key: &str, value: &Value) -> bool {
     if normalized == "tokenbudgetcontrol" {
         return !value.is_boolean();
     }
-    if normalized == "tokenbudget" {
+    if matches!(normalized.as_str(), "tokenbudget" | "tokensused") {
         return !value.is_number() && !value.is_null();
     }
     if matches!(
@@ -1558,7 +1576,6 @@ fn sensitive_key(key: &str, value: &Value) -> bool {
             | "totaltokens"
             | "pretokens"
             | "posttokens"
-            | "tokensused"
     ) {
         return !value.is_number();
     }
@@ -3690,6 +3707,13 @@ mod tests {
         assert_eq!(sanitized["tokenBudget"], json!(4096));
         assert_eq!(sanitized["tokensUsed"], json!(128));
         assert_eq!(sanitized["accessToken"], json!("[REDACTED]"));
+        let unknown = sanitize_value(&json!({"tokenBudget":null, "tokensUsed":null}));
+        assert!(unknown["tokenBudget"].is_null());
+        assert!(unknown["tokensUsed"].is_null());
+        assert_eq!(
+            sanitize_value(&json!({"tokensUsed":"synthetic credential"}))["tokensUsed"],
+            json!("[REDACTED]")
+        );
     }
 
     #[test]
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs
index 39b6d7400a6..2c856c2b68f 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/native_provider_backend.rs
@@ -278,6 +278,20 @@ impl NativeProviderCommandExecutor {
 }
 
 impl CommandExecutor for NativeProviderCommandExecutor {
+    fn observe_recovery_state(
+        &mut self,
+        state: &crate::durable::DurableState,
+    ) -> Result<(), DurableRunnerError> {
+        if state.warm_transition.is_some() {
+            return Ok(());
+        }
+        self.select_recovery()?;
+        if let Some(SelectedExecutor::LocalFacade(executor)) = self.selected.as_mut() {
+            executor.observe_recovery_state(state);
+        }
+        Ok(())
+    }
+
     fn can_reconcile_result_delivery(&mut self) -> Result<bool, DurableRunnerError> {
         self.select_recovery()?;
         self.selected
diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
index 82a22d33c90..ef0b777c2e2 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/src/provider_backend.rs
@@ -1102,6 +1102,15 @@ fn normalize_codex_goal(value: &Value, working_now: bool) -> Option<SessionGoalS
     })
 }
 
+fn codex_goal_is_dormant(value: &Value) -> bool {
+    [value.get("workingNow"), value.pointer("/goal/workingNow")]
+        .into_iter()
+        .flatten()
+        .all(|working| working.as_bool() == Some(false))
+        && (value.get("goal") == Some(&Value::Null)
+            || normalize_codex_goal(value, false).is_some_and(|goal| goal.status != "active"))
+}
+
 fn goal_event_payload(
     goal: Option<&SessionGoalSnapshot>,
     capability: Option<&SessionGoalCapability>,
@@ -1856,6 +1865,7 @@ pub struct CodexCommandExecutor {
     cursor_cloud_launch_profile: Option<CursorCloudLaunchProfile>,
     startup_command: Option<ProviderStartupCommand>,
     startup_evidence_error: Option<DurableRunnerError>,
+    recovery_pending_effects: Option<usize>,
 }
 
 impl CodexCommandExecutor {
@@ -1871,6 +1881,7 @@ impl CodexCommandExecutor {
             cursor_cloud_launch_profile: None,
             startup_command: None,
             startup_evidence_error: None,
+            recovery_pending_effects: None,
         }
     }
 
@@ -1882,6 +1893,10 @@ impl CodexCommandExecutor {
         executor
     }
 
+    pub(crate) fn observe_recovery_state(&mut self, state: &crate::durable::DurableState) {
+        self.recovery_pending_effects = Some(state.recovery_pending_effects());
+    }
+
     fn bind_opencode_launch_profile(
         &self,
         config: &CodexProviderConfig,
@@ -2221,6 +2236,21 @@ impl CodexCommandExecutor {
         {
             return Ok(());
         }
+        let expected_dormant = state.config.provider == "codex"
+            && self
+                .recovery_pending_effects
+                .as_ref()
+                .is_some_and(|count| *count == 0)
+            && state.completed_turn_authoritative
+            && state.active_provider_turn_id.is_none()
+            && !state.ambiguous_turn_start_pending
+            && state.startup_attempt.is_none()
+            && !state.receipt_limit_interrupt_pending
+            && state.tool_bridge.pending_calls().next().is_none()
+            && state
+                .goal
+                .as_ref()
+                .is_none_or(|goal| goal.status != "active" && !goal.working_now);
         let provider_label = state.config.provider.clone();
         let provider_name = facade_provider_name(&provider_label);
         let provider_had_exited = state.lifecycle == "provider_exited";
@@ -2306,6 +2336,39 @@ impl CodexCommandExecutor {
                 return Ok(());
             }
             let recovered_active_turn_id = provider.active_provider_turn_id().map(str::to_owned);
+            if expected_dormant && recovered_active_turn_id.is_some() {
+                return Err(DurableRunnerError::invalid(
+                    "saved dormant Codex session resumed unexpected active work",
+                ));
+            }
+            if expected_dormant {
+                let live_goal = provider.get_goal().map_err(|error| {
+                    DurableRunnerError::invalid(format!(
+                        "dormant recovery goal probe failed: {error}"
+                    ))
+                })?;
+                if !codex_goal_is_dormant(&live_goal) {
+                    return Err(DurableRunnerError::invalid(
+                        "saved dormant Codex session resumed an active or unknown goal",
+                    ));
+                }
+                let live_thread = provider
+                    .read_thread_after_recovery_goal(codex_goal_is_dormant)
+                    .map_err(|error| {
+                        DurableRunnerError::invalid(format!(
+                            "dormant recovery thread probe failed: {error}"
+                        ))
+                    })?;
+                if live_thread
+                    .pointer("/thread/status/type")
+                    .and_then(Value::as_str)
+                    != Some("idle")
+                {
+                    return Err(DurableRunnerError::invalid(
+                        "saved dormant Codex session resumed unexpected active work",
+                    ));
+                }
+            }
             let recovered_turn_ended_with_result = active_provider_result_authoritative
                 && previous_active_turn_id.is_some()
                 && recovered_active_turn_id.is_none();
@@ -2610,6 +2673,7 @@ impl CodexCommandExecutor {
         {
             self.fail_started_provider(&mut provider);
         } else {
+            self.recovery_pending_effects = None;
             self.provider = Some(provider);
         }
         admission
@@ -5295,6 +5359,7 @@ impl CommandExecutor for CodexCommandExecutor {
     }
 
     fn rotate_authority(&mut self, config: &DurableRunnerConfig) {
+        self.recovery_pending_effects = None;
         self.event_identity = Some(ProviderEventIdentity::from_config(config));
     }
 
diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
index 840883c58f1..11b3dd1f575 100644
--- a/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
+++ b/packages/paperclip-runner/runner/crates/runner-core/tests/codex_provider.rs
@@ -7,8 +7,8 @@ use paperclip_runner_core::codex_provider::{
     CodexProvider, CodexProviderConfig, CodexProviderEvent,
 };
 use paperclip_runner_core::durable::{
-    Command, CommandExecutor, DurableRunnerConfig, DurableRunnerError, PolledEvent,
-    TerminalDeliveryReconciliation,
+    Command, CommandDisposition, CommandExecutor, DurableRunnerConfig, DurableRunnerError,
+    DurableStateStore, EventPriority, PolledEvent, TerminalDeliveryReconciliation,
 };
 use paperclip_runner_core::native_provider_backend::NativeProviderCommandExecutor;
 use paperclip_runner_core::provider_backend::CodexCommandExecutor;
@@ -76,6 +76,311 @@ fn provider_config(directory: &Path, switches: &[&str]) -> CodexProviderConfig {
     }
 }
 
+fn dormant_source_fixture(directory: &Path) -> (DurableRunnerConfig, Value) {
+    dormant_source_fixture_with_switches(directory, &[])
+}
+
+fn dormant_source_fixture_with_switches(
+    directory: &Path,
+    switches: &[&str],
+) -> (DurableRunnerConfig, Value) {
+    let mut switches_with_defaults = vec!["--record-process-start", "--durable-turn-ids"];
+    switches_with_defaults.extend_from_slice(switches);
+    let config = durable_config(directory);
+    let mut executor = CodexCommandExecutor::with_runner_config(directory, &config);
+    executor
+        .execute(&command(
+            "prepare-source",
+            1,
+            "run.prepare",
+            json!({"provider":provider_config(directory, &switches_with_defaults)}),
+        ))
+        .unwrap();
+    executor
+        .execute(&command("open-source", 2, "session.open", json!({})))
+        .unwrap();
+    executor
+        .execute(&command(
+            "source-turn",
+            3,
+            "turn.start",
+            json!({"text":"Complete the source turn."}),
+        ))
+        .unwrap();
+    wait_for_executor_event(&mut executor, "turn.completed");
+    while !poll_and_ack(&mut executor).unwrap().is_empty() {}
+    executor.shutdown().unwrap();
+    let provider =
+        serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap())
+            .unwrap();
+    (config, provider)
+}
+
+#[test]
+fn ordinary_recovery_restores_a_completed_idle_session() {
+    let directory = temporary_directory("dormant-recovery");
+    let (config, _) = dormant_source_fixture(&directory);
+    let runner = DurableStateStore::new(&directory)
+        .unwrap()
+        .load_or_create(&config)
+        .unwrap()
+        .0;
+    let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+    executor.observe_recovery_state(&runner).unwrap();
+    executor
+        .execute(&command("restore", 1, "session.snapshot", json!({})))
+        .unwrap();
+    while !poll_and_ack(&mut executor).unwrap().is_empty() {}
+    let snapshot = executor
+        .execute(&command("snapshot", 2, "session.snapshot", json!({})))
+        .unwrap();
+    assert_eq!(
+        snapshot.result["warmAttachReady"], true,
+        "{}",
+        snapshot.result
+    );
+    assert_eq!(call_count(&directory, "process-start"), 2);
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    executor.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn ordinary_recovery_checks_saved_dormant_state_against_live_work() {
+    for fault in ["active", "goal"] {
+        let directory = temporary_directory("dormant-live-recovery");
+        let (config, _) = dormant_source_fixture(&directory);
+        let path = directory.join("fake-state.json");
+        let mut live: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
+        if fault == "active" {
+            live["activeTurnId"] = json!("unexpected-provider-turn");
+        } else {
+            live["goal"] = json!({"objective":"Unexpected autonomous work", "status":"active"});
+        }
+        fs::write(&path, serde_json::to_vec(&live).unwrap()).unwrap();
+        let runner = DurableStateStore::new(&directory)
+            .unwrap()
+            .load_or_create(&config)
+            .unwrap()
+            .0;
+        let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+        recovered.observe_recovery_state(&runner).unwrap();
+        assert_eq!(call_count(&directory, "process-start"), 1);
+        let error = recovered
+            .execute(&command("snapshot", 1, "session.snapshot", json!({})))
+            .unwrap_err();
+        assert!(
+            error.to_string().contains("saved dormant Codex"),
+            "{fault}: {error}"
+        );
+        assert_eq!(call_count(&directory, "process-start"), 2);
+        drop(recovered);
+        fs::remove_dir_all(directory).unwrap();
+    }
+}
+
+#[test]
+fn dormant_recovery_rejects_goal_activation_during_the_final_thread_read() {
+    let directory = temporary_directory("dormant-goal-during-read");
+    let trigger = directory.join("activate-goal");
+    let (config, _) = dormant_source_fixture_with_switches(
+        &directory,
+        &["--goal-on-thread-read-file", trigger.to_str().unwrap()],
+    );
+    fs::write(&trigger, b"active").unwrap();
+    let runner = DurableStateStore::new(&directory)
+        .unwrap()
+        .load_or_create(&config)
+        .unwrap()
+        .0;
+    let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+    recovered.observe_recovery_state(&runner).unwrap();
+    let result = recovered.execute(&command("snapshot", 1, "session.snapshot", json!({})));
+    assert_eq!(call_count(&directory, "late-goal-on-thread-read"), 1);
+    let error = result.expect_err("late goal activation must refuse dormant recovery");
+    assert!(error.to_string().contains("saved dormant Codex"), "{error}");
+    let state: Value =
+        serde_json::from_slice(&fs::read(directory.join("codex-provider-state.json")).unwrap())
+            .unwrap();
+    assert_eq!(state["startupAttempt"]["directChildExitObserved"], true);
+    #[cfg(unix)]
+    assert!(
+        !std::process::Command::new("kill")
+            .args([
+                "-0",
+                &state["startupAttempt"]["processId"]
+                    .as_u64()
+                    .unwrap()
+                    .to_string()
+            ])
+            .output()
+            .unwrap()
+            .status
+            .success(),
+        "provider must be reaped before refusal returns"
+    );
+    assert_eq!(call_count(&directory, "process-start"), 2);
+    assert_eq!(call_count(&directory, "turn/start"), 1);
+    drop(recovered);
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn dormant_recovery_delivers_a_paused_goal_observed_during_the_final_thread_read() {
+    let directory = temporary_directory("paused-goal-during-read");
+    let trigger = directory.join("paused-goal");
+    let (config, _) = dormant_source_fixture_with_switches(
+        &directory,
+        &["--goal-on-thread-read-file", trigger.to_str().unwrap()],
+    );
+    fs::write(&trigger, b"paused").unwrap();
+    let runner = DurableStateStore::new(&directory)
+        .unwrap()
+        .load_or_create(&config)
+        .unwrap()
+        .0;
+    let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+    recovered.observe_recovery_state(&runner).unwrap();
+    recovered
+        .execute(&command("snapshot", 1, "session.snapshot", json!({})))
+        .unwrap();
+    assert_eq!(call_count(&directory, "late-goal-on-thread-read"), 1);
+    let update = wait_for_executor_event(&mut recovered, "session.goal.updated");
+    assert_eq!(update.payload["goal"]["status"], "paused");
+    drop(recovered);
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn ordinary_recovery_preserves_pending_same_run_goal_effects() {
+    for status in ["pending", "indeterminate"] {
+        let directory = temporary_directory("pending-goal-recovery");
+        let (config, _) = dormant_source_fixture(&directory);
+        let path = directory.join("fake-state.json");
+        let mut live: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
+        live["goal"] = json!({"objective":"Already authorized same-run goal", "status":"active"});
+        fs::write(&path, serde_json::to_vec(&live).unwrap()).unwrap();
+        let store = DurableStateStore::new(&directory).unwrap();
+        let mut runner = store.load_or_create(&config).unwrap().0;
+        runner
+            .begin_command(&command(
+                "goal-in-flight",
+                1,
+                "session.goal.set",
+                json!({"objective":"Already authorized same-run goal", "status":"active"}),
+            ))
+            .unwrap();
+        if status == "indeterminate" {
+            store.save(&runner).unwrap();
+            runner = store.load_or_create(&config).unwrap().0;
+        }
+        assert_eq!(runner.processed_commands["goal-in-flight"].status, status);
+        let mut recovered = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+        recovered.observe_recovery_state(&runner).unwrap();
+        recovered
+            .execute(&command("snapshot", 2, "session.snapshot", json!({})))
+            .unwrap();
+        let goal = recovered
+            .execute(&command("goal-read", 3, "session.goal.get", json!({})))
+            .unwrap();
+        assert_eq!(goal.result["goal"]["status"], "active");
+        assert_eq!(call_count(&directory, "turn/start"), 1);
+        recovered.shutdown().unwrap();
+        fs::remove_dir_all(directory).unwrap();
+    }
+}
+
+#[test]
+fn ordinary_recovery_does_not_restore_prepared_during_reads_or_cleanup() {
+    let directory = temporary_directory("prepared-journal-recovery");
+    let (config, mut provider) = dormant_source_fixture(&directory);
+    provider["lifecycle"] = json!("prepared");
+    provider["completedTurnAuthoritative"] = json!(false);
+    provider["completedTurnProcessGeneration"] = Value::Null;
+    provider["completedProviderTurnId"] = Value::Null;
+    fs::write(
+        directory.join("codex-provider-state.json"),
+        serde_json::to_vec(&provider).unwrap(),
+    )
+    .unwrap();
+    let runner = DurableStateStore::new(&directory)
+        .unwrap()
+        .load_or_create(&config)
+        .unwrap()
+        .0;
+    let mut executor = NativeProviderCommandExecutor::with_runner_config(&directory, &config);
+    executor.observe_recovery_state(&runner).unwrap();
+    let snapshot = executor
+        .execute(&command("snapshot", 1, "session.snapshot", json!({})))
+        .unwrap();
+    assert_eq!(snapshot.result["warmAttachReady"], false);
+    executor
+        .execute(&command("drain", 2, "runner.drain", json!({})))
+        .unwrap();
+    executor
+        .execute(&command("stop", 3, "turn.stop", json!({})))
+        .unwrap();
+    assert_eq!(call_count(&directory, "process-start"), 1);
+    executor.shutdown().unwrap();
+    fs::remove_dir_all(directory).unwrap();
+}
+
+#[test]
+fn ordinary_recovery_preserves_command_replay_and_collision_checks() {
+    let directory = temporary_directory("recovery-command-replay");
+    let config = durable_config(&directory);
+    let mut runner = DurableStateStore::new(&directory)
+        .unwrap()
+        .load_or_create(&config)
+        .unwrap()
+        .0;
+    let recorded = command(
+        "recorded-attach",
+        1,
+        "run.attach",
+        json!({"paperclipNextAuthority":{}}),
+    );
+    assert!(matches!(
+        runner.begin_command(&recorded).unwrap(),
+        CommandDisposition::Execute
+    ));
+    runner
+        .complete_command(&recorded, json!({"status":"attached"}))
+        .unwrap();
+    runner
+        .enqueue_event(
+            &config,
+            "runner.diagnostic",
+            EventPriority::P0,
+            json!({"message":"unacknowledged fact"}),
+        )
+        .unwrap();
+    assert!(
+        matches!(runner.begin_command(&recorded).unwrap(), CommandDisposition::Replay(result) if result.status == "completed")
+    );
+    let mut collision = recorded.clone();
+    collision.payload["changed"] = json!(true);
+    assert!(runner
+        .begin_command(&collision)
+        .unwrap_err()
+        .to_string()
+        .contains("different command data"));
+    let pending = command(
+        "pending-goal",
+        2,
+        "session.goal.set",
+        json!({"objective":"Original effect"}),
+    );
+    assert!(matches!(
+        runner.begin_command(&pending).unwrap(),
+        CommandDisposition::Execute
+    ));
+    assert!(
+        matches!(runner.begin_command(&pending).unwrap(), CommandDisposition::Replay(result) if result.status == "pending")
+    );
+    fs::remove_dir_all(directory).unwrap();
+}
+
 #[test]
 fn delegates_command_isolation_to_an_explicit_external_sandbox() {
     let directory = temporary_directory("external-sandbox");
@@ -190,7 +495,12 @@ fn durable_config(directory: &Path) -> DurableRunnerConfig {
 
 fn command(id: &str, sequence: u64, command_type: &str, payload: Value) -> Command {
     Command {
-        schema: "paperclip.prp.command.v1".to_owned(),
+        schema: if command_type.starts_with("session.goal.") {
+            "paperclip.prp.command.v2"
+        } else {
+            "paperclip.prp.command.v1"
+        }
+        .to_owned(),
         command_id: id.to_owned(),
         controller_seq: sequence,
         command_type: command_type.to_owned(),
@@ -428,7 +738,7 @@ fn recorded_tool_responses(directory: &Path) -> Vec<String> {
 }
 
 fn poll_and_ack(
-    executor: &mut CodexCommandExecutor,
+    executor: &mut impl CommandExecutor,
 ) -> Result<Vec<PolledEvent>, DurableRunnerError> {
     let events = executor.poll_events()?;
     executor.acknowledge_events(events.len())?;
@@ -472,7 +782,7 @@ fn wait_for_provider_error(provider: &mut CodexProvider) -> String {
 }
 
 fn wait_for_executor_event(
-    executor: &mut CodexCommandExecutor,
+    executor: &mut impl CommandExecutor,
     expected_event_type: &str,
 ) -> PolledEvent {
     let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts
index dc2ccd7140e..0e858fa1955 100644
--- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts
+++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts
@@ -59,6 +59,40 @@ const identity: DurableRecoveryIdentity = {
 const expectedRunnerVersion = "0.3.0";
 const expectedRunnerDigest = `sha256:${"a".repeat(64)}`;
 
+it("admits only source maintenance while the authenticated owner differs", async () => {
+  const root = mkdtempSync(resolve(tmpdir(), "runner-authority-admission-"));
+  let owner = identity.runId;
+  const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
+    get authorityRunId() { return owner; } });
+  try {
+    const prior = core.queueCommand("turn.start", { text: "Owned same-run work" }, "same-run-command");
+    expect(prior.type).toBe("turn.start");
+    owner = "new-owner";
+    const count = core.store.state.commands.length;
+    for (const type of ["turn.start", "session.goal.set", "semantic_tool.result", "session.open"]) {
+      expect(() => core.queueCommand(type, {})).toThrow("source_attachment_pending");
+    }
+    expect(core.store.state.commands).toHaveLength(count);
+    expect(core.queueCommand("session.snapshot").type).toBe("session.snapshot");
+    expect(core.queueCommand("turn.stop").type).toBe("turn.stop");
+    expect(() => core.enableSemanticInputDispatch(identity)).toThrow("source_attachment_pending");
+  } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); }
+});
+
+it("keeps already-dispatching target work independent of predecessor completion", async () => {
+  const root = mkdtempSync(resolve(tmpdir(), "runner-target-dispatch-"));
+  const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
+    authorityRunId: identity.runId });
+  try {
+    core.queueCommand("turn.start", { text: "Current target work" });
+    const settled = vi.spyOn(core, "semanticToolResultsSettled").mockReturnValue(false);
+    expect(() => core.enableSemanticInputDispatch(identity)).not.toThrow();
+    expect(settled).not.toHaveBeenCalled();
+    expect(core.queueCommand("session.goal.set", { status: "active" }).type).toBe("session.goal.set");
+    expect(() => core.enableSemanticInputDispatch({ ...identity, runId: "foreign" })).toThrow("source_attachment_pending");
+  } finally { await core.stop(); rmSync(root, { recursive: true, force: true }); }
+});
+
 it("recognizes a complete unclaimed bootstrap through cold checkpoint restoration", async () => {
   const root = mkdtempSync(resolve(tmpdir(), "runner-unclaimed-bootstrap-"));
   const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest });
@@ -595,6 +629,30 @@ it("persists authentication and commands before the runner can observe them", as
   }
 });
 
+it("does not dispatch after the authenticated owner changes during checkpoint persistence", async () => {
+  const root = mkdtempSync(resolve(tmpdir(), "runner-tool-owner-race-"));
+  let owner = identity.runId, entered = false, release!: () => void;
+  const barrier = new Promise<void>(resolveWrite => { release = resolveWrite; });
+  const dispatch = vi.fn(async () => ({ result: "must not run" }));
+  const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest,
+    get authorityRunId() { return owner; }, onSemanticToolInput: dispatch });
+  core.setCheckpointPersistence(async checkpoint => {
+    if (JSON.parse(checkpoint.records.get("cursor")!).ackedSourceSeq > 0) { entered = true; await barrier; }
+  });
+  try {
+    await core.start();
+    const client = (await authenticate(core, core.issueBootstrapTicket()))!;
+    sendSecure(client, semanticInputEvent());
+    await vi.waitFor(() => expect(entered).toBe(true));
+    owner = "replacement-owner"; release();
+    const ack = await receiveSecure(client);
+    expect(dispatch).not.toHaveBeenCalled();
+    expect(ack!.kind).toBe("ack");
+    expect(core.store.state.commands).toEqual([]);
+    client.socket.destroy();
+  } finally { release(); await core.stop(); rmSync(root, { recursive: true, force: true }); }
+});
+
 it("does not dispatch a semantic effect when external journal persistence fails", async () => {
   const root = mkdtempSync(resolve(tmpdir(), "runner-external-journal-failure-"));
   const dispatch = vi.fn(async () => ({ result: "written" }));
@@ -804,8 +862,9 @@ it.each([true, false])("reconciles a saved semantic input only from an authorita
   }
 });
 
-it("resolves a saved semantic input only once across overlapping recovery calls", async () => {
+it.each([false, true])("resolves a saved semantic input only once across overlapping recovery calls (owner changed: %s)", async changedOwner => {
   const root = mkdtempSync(resolve(tmpdir(), "runner-saved-semantic-race-"));
+  let owner = identity.runId;
   const options = { stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest };
   const original = new DurablePrpControlPlane({ ...options,
     onSemanticToolInput: () => new Promise<{ result: unknown }>(() => undefined) });
@@ -821,17 +880,18 @@ it("resolves a saved semantic input only once across overlapping recovery calls"
     expect((await receiveSecure(client))!.kind).toBe("ack");
     client.socket.destroy();
     await original.stop();
-    replacement = new DurablePrpControlPlane({ ...options, recoverSemanticToolInput: recover });
+    replacement = new DurablePrpControlPlane({ ...options, get authorityRunId() { return owner; }, recoverSemanticToolInput: recover });
     pending.push(replacement.recoverPendingSemanticInputs());
     await vi.waitFor(() => expect(recover).toHaveBeenCalledOnce());
     pending.push(replacement.recoverPendingSemanticInputs());
     await new Promise<void>(resolveTurn => setImmediate(resolveTurn));
     expect(recover).toHaveBeenCalledOnce();
+    if (changedOwner) owner = "replacement-owner";
     resolveOutcome({ result: { receipt: "one-authoritative-outcome" } });
     await Promise.all(pending);
     await replacement.recoverPendingSemanticInputs();
     expect(recover).toHaveBeenCalledOnce();
-    expect(replacement.store.state.commands.filter(command => command.type === "semantic_tool.result")).toHaveLength(1);
+    expect(replacement.store.state.commands.filter(command => command.type === "semantic_tool.result")).toHaveLength(changedOwner ? 0 : 1);
   } finally {
     resolveOutcome({ result: { receipt: "one-authoritative-outcome" } });
     await Promise.allSettled(pending);
diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
index 218ea36aeec..cd29daad233 100644
--- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
+++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts
@@ -76,6 +76,10 @@ const maxStateBytes = 256 * 1024 * 1024;
 const authChallengeTtlMs = 5_000;
 const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/;
 const runnerDigestPattern = /^sha256:[0-9a-f]{64}$/;
+export function isRunnerRecoveryMaintenanceCommand(type: string): boolean {
+  return ["session.snapshot", "session.goal.get", "runner.drain", "runner.suspend", "runner.shutdown", "turn.stop"].includes(type);
+}
+
 const commandTypes = new Set([
   "run.prepare",
   "run.attach",
@@ -269,6 +273,7 @@ export class SemanticToolNotDispatchedError extends Error {
 export interface DurablePrpControlPlaneOptions {
   stateDirectory: string;
   identity: DurableRecoveryIdentity;
+  authorityRunId?: string;
   expectedRunnerVersion: string;
   expectedRunnerDigest: string;
   /** Complete caller-owned admission before consuming a credential or releasing commands. */
@@ -2010,6 +2015,8 @@ export class DurablePrpControlPlane {
   }> = [];
   #port: number | null = null;
   #semanticInputMode: "dispatch" | "retain_only";
+  #semanticInputSourceRunId: string | null;
+  readonly #authorityRunId: () => string | undefined;
   #onSemanticToolInput?: DurablePrpControlPlaneOptions["onSemanticToolInput"];
   #onCommittedEvent?: DurablePrpControlPlaneOptions["onCommittedEvent"];
   #recoverSemanticToolInput?: DurablePrpControlPlaneOptions["recoverSemanticToolInput"];
@@ -2064,7 +2071,10 @@ export class DurablePrpControlPlane {
         "Warm run transition requires its exact approved runner artifact.",
       );
     }
-    this.#semanticInputMode = options.semanticInputMode ?? "dispatch";
+    this.#authorityRunId = () => options.authorityRunId;
+    this.#semanticInputMode = options.authorityRunId !== undefined && options.authorityRunId !== options.identity.runId
+      ? "retain_only" : options.semanticInputMode ?? "dispatch";
+    this.#semanticInputSourceRunId = this.#semanticInputMode === "retain_only" ? options.identity.runId : null;
     this.#onSemanticToolInput = options.onSemanticToolInput;
     this.#onCommittedEvent = options.onCommittedEvent;
     this.#recoverSemanticToolInput = options.recoverSemanticToolInput;
@@ -2215,10 +2225,30 @@ export class DurablePrpControlPlane {
     );
   }
 
+  #retainsSourceInputs(): boolean {
+    const owner = this.#authorityRunId();
+    return this.#semanticInputMode === "retain_only" || (owner !== undefined && owner !== this.#identity.runId);
+  }
+
+  enableSemanticInputDispatch(identity: DurableRecoveryIdentity): void {
+    if ((this.#authorityRunId() !== undefined && this.#authorityRunId() !== identity.runId)
+      || canonicalJson(this.#identity) !== canonicalJson(identity) || this.#store.state.warmTransition) {
+      throw new Error("native_runner_source_attachment_pending");
+    }
+    if (this.#semanticInputMode === "dispatch") return;
+    if (this.#identity.runId === this.#semanticInputSourceRunId
+      || this.activeRunnerConnectionCount() !== 1 || !this.semanticToolResultsSettled()) {
+      throw new Error("native_runner_source_attachment_pending");
+    }
+    this.#semanticInputMode = "dispatch";
+    this.#semanticInputSourceRunId = null;
+  }
+
   async recoverPendingSemanticInputs(): Promise<void> {
-    if (this.#semanticInputMode === "retain_only" || !this.#recoverSemanticToolInput) return;
+    if (this.#retainsSourceInputs() || !this.#recoverSemanticToolInput) return;
     const identity = this.#identity;
     for (const saved of [...this.#store.state.committedEvents]) {
+      if (this.#retainsSourceInputs() || this.#identity !== identity || this.#semanticCallbacksRetired) return;
       if (!unsettledSemanticInput(saved, this.#store.state)) continue;
       const event = saved.envelope.payload as PrpEvent;
       const semantic = (event.payload as Record<string, unknown>).semantic_tool as Record<string, unknown>;
@@ -2233,7 +2263,7 @@ export class DurablePrpControlPlane {
           correlation: { runId: identity.runId, normalizedSessionId: identity.normalizedSessionId,
             turnId: identity.turnId, itemId: event.itemId ?? identity.itemId } };
         const outcome = await this.#recoverSemanticToolInput(call);
-        if (this.#identity !== identity || this.#semanticCallbacksRetired) return;
+        if (this.#identity !== identity || this.#semanticCallbacksRetired || this.#retainsSourceInputs()) return;
         if (!outcome || this.#getCommand(commandId)) continue;
         const { input, ...metadata } = call;
         this.queueCommand("semantic_tool.result", { ...metadata, inputDigest: canonicalDigest(input),
@@ -2467,6 +2497,10 @@ export class DurablePrpControlPlane {
     ) {
       throw new Error("Durable PRP run identity rotation is invalid.");
     }
+    if (this.#authorityRunId() !== undefined && this.#authorityRunId() !== this.#identity.runId) {
+      this.#semanticInputMode = "retain_only";
+      this.#semanticInputSourceRunId ??= this.#identity.runId;
+    }
     this.disconnectActiveRunner();
     const leases = Object.fromEntries(
       Object.entries(this.#store.state.leases).map(([key, lease]) => [
@@ -2687,6 +2721,17 @@ export class DurablePrpControlPlane {
     deliverImmediately = false,
   ): DurableRecoveryCoreCommand {
     this.#store.assertWritable();
+    const authorityRunId = this.#authorityRunId();
+    const retainedSource = this.#semanticInputSourceRunId ?? this.#identity.runId;
+    const prior = commandId ? this.#getCommand(commandId) : undefined;
+    if (authorityRunId !== undefined && retainedSource !== authorityRunId
+      && !isRunnerRecoveryMaintenanceCommand(type)
+      && !(type === "run.attach" && isRecord(payload.paperclipNextAuthority)
+        && isRecord(payload.paperclipNextAuthority.identity)
+        && payload.paperclipNextAuthority.identity.runId === authorityRunId)
+      && !(prior && prior.status !== "pending" && prior.type === type && canonicalJson(prior.payload) === canonicalJson(payload))) {
+      throw new Error("native_runner_source_attachment_pending");
+    }
     const staged = this.#store.state.stagedRunAttachment;
     const retireStaged = stagedTargetActivated(this.#store.state) && (type === "turn.start"
       || (this.#stagedTargetCheckpointPublished === canonicalJson(this.#store.state.identity)
@@ -3974,7 +4019,7 @@ export class DurablePrpControlPlane {
       eventType === "semantic_tool.input" || eventType === "mcp_app.tool_input";
     if (
       isSemanticInput &&
-      ((this.#onSemanticToolInput === undefined && this.#semanticInputMode !== "retain_only") ||
+      ((this.#onSemanticToolInput === undefined && !this.#retainsSourceInputs()) ||
         semantic?.phase !== "input" ||
         typeof semantic.callId !== "string" ||
         typeof semantic.operationId !== "string" ||
@@ -4114,7 +4159,7 @@ export class DurablePrpControlPlane {
     }
     if (!isCurrentOwner()) return;
 
-    const liveCall = this.#semanticInputMode !== "retain_only" && isSemanticInput && this.#onSemanticToolInput && semantic !== undefined
+    const liveCall = !this.#retainsSourceInputs() && isSemanticInput && this.#onSemanticToolInput && semantic !== undefined
       && typeof semantic.callId === "string" && typeof semantic.operationId === "string"
       ? { callId: semantic.callId, operationId: semantic.operationId, input: semantic.input,
           sourceEventId, sourceEventType: eventType,
@@ -4132,7 +4177,7 @@ export class DurablePrpControlPlane {
     try {
       await this.flushCheckpoint();
       if (this.#eventOwner !== connection || this.#identity !== identity) return;
-      if (ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) {
+      if (!this.#retainsSourceInputs() && ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) {
         const call = liveCall;
         const commandId = liveCommandId;
         const recordFailure = (stage: "persist_result" | "dispatch", error: unknown) => {
diff --git a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts
index 526d628d9f7..681bf7234fb 100644
--- a/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts
+++ b/packages/paperclip-runner/src/control-plane/runner-execution-driver.ts
@@ -9,6 +9,10 @@ import type { DurableRecoveryIdentity } from "../contracts/durable-recovery.js";
  * specifications stay private to the driver; plugins receive versioned requirements.
  */
 export interface RunnerExecutionDriver {
+  /** Attests stopped ownership and dormant prelaunch admission for the selected provider. */
+  sourceRecoveryVersion?: 1;
+  /** Current authenticated lease/checkpoint-writer owner; may change during warm reuse. */
+  authorityRunId?: string;
   /** Own provider configuration and file preparation in the execution environment. */
   prepareLaunch?: RunnerLaunchPreparer;
   /** Reauthorize retained resources before adoption without preparing them again.
diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts
index 1c60274c6af..e3d46a188bd 100644
--- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts
+++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts
@@ -60,6 +60,8 @@ export class CodexHarnessSession
   extends CodexSessionState
   implements HarnessSession
 {
+  #deferredNotificationPump = false;
+
   constructor(input: CodexSessionStateInput) {
     super(input);
     this.transport.setServerRequestHandler((request) =>
@@ -77,6 +79,7 @@ export class CodexHarnessSession
       ));
     }
     if (this.terminal) {
+      this.#deferredNotificationPump = true;
       this.eventQueue.close();
     } else {
       void pumpNotifications(this);
@@ -150,6 +153,11 @@ export class CodexHarnessSession
       this.protocolFailed = false;
       this.protocolFailureCode = null;
       this.protocolFailureMessage = null;
+      if (this.#deferredNotificationPump) {
+        this.eventQueue.reopen();
+        this.#deferredNotificationPump = false;
+        void pumpNotifications(this);
+      }
       this.emit("run.attached", { runId: input.runId, sameSession: true });
     } finally {
       this.pendingRunAttachment = null;
diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts
index f2bc252abe0..3113d182f7e 100644
--- a/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts
+++ b/packages/paperclip-runner/src/drivers/codex/codex-session-state.ts
@@ -56,6 +56,11 @@ class AsyncQueue<T> implements AsyncIterable<T> {
       waiter.resolve({ value: undefined, done: true });
   }
 
+  reopen(): void {
+    if (this.#failure !== null) throw this.#failure;
+    this.#closed = false;
+  }
+
   fail(error: Error): void {
     this.#failure ??= error;
     this.#closed = true;
diff --git a/packages/paperclip-runner/src/live/runner-provider-launch.test.ts b/packages/paperclip-runner/src/live/runner-provider-launch.test.ts
index d5c1fd22aa0..62183f9a5e8 100644
--- a/packages/paperclip-runner/src/live/runner-provider-launch.test.ts
+++ b/packages/paperclip-runner/src/live/runner-provider-launch.test.ts
@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { createRunnerProviderLaunch, preparedRunnerLaunchWorkingDirectory, runnerLaunchRequirementsDigest, validatePreparedRunnerLaunch,
+import { createRunnerProviderLaunch, preparedRunnerLaunchWorkingDirectory, runnerLaunchRequirementsDigest, runnerLaunchSettingsMatch, validatePreparedRunnerLaunch,
   runnerProviderLaunchRequirements, type RunnerProviderLaunchEnvironment, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js";
 
 const requirements: RunnerProviderLaunchRequirements = {
@@ -19,6 +19,14 @@ const environment: RunnerProviderLaunchEnvironment = {
 };
 
 describe("built-in provider launch preparation", () => {
+  it("compares effective ACPX defaults without changing the saved requirements", () => {
+    const frozen = { ...requirements, provider: "acpx" as const, model: "gpt-5.4", settings: {} };
+    expect(runnerLaunchSettingsMatch(frozen, { acpxAgent: "codex", acpxPermissionMode: "approve-all", acpxPermissionModePinned: true })).toBe(true);
+    expect(runnerLaunchSettingsMatch(frozen, { acpxPermissionMode: "approve-reads" })).toBe(false);
+    expect(runnerLaunchSettingsMatch(frozen, { acpxPermissionModePinned: false })).toBe(false);
+    expect(frozen.settings).toEqual({});
+  });
+
   it("preserves probe project-instruction suppression through a prepared launch", () => {
     const selected = runnerProviderLaunchRequirements({ provider: "codex", disableCodexProjectInstructions: true,
       disableCodexWebSearch: true }, requirements.identity,
diff --git a/packages/paperclip-runner/src/live/runner-provider-launch.ts b/packages/paperclip-runner/src/live/runner-provider-launch.ts
index a7afb0b0f5f..cff4b9a4154 100644
--- a/packages/paperclip-runner/src/live/runner-provider-launch.ts
+++ b/packages/paperclip-runner/src/live/runner-provider-launch.ts
@@ -100,6 +100,18 @@ export function runnerLaunchRequirementsDigest(requirements: RunnerProviderLaunc
   return `sha256:${createHash("sha256").update(canonical).digest("hex")}`;
 }
 
+function runnerLaunchPolicy(requirements: RunnerProviderLaunchRequirements): Record<string, unknown> {
+  return createRunnerProviderLaunch(requirements, {
+    options: { codexArgs: [] }, root: "/", cwd: "/", codexHome: "/",
+    providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "", codexCommand: "",
+  });
+}
+
+/** Compare effective provider policy, including the factory's defaults. */
+export function runnerLaunchSettingsMatch(requirements: RunnerProviderLaunchRequirements, settings: RunnerProviderLaunchRequirements["settings"]): boolean {
+  return JSON.stringify(runnerLaunchPolicy(requirements)) === JSON.stringify(runnerLaunchPolicy({ ...requirements, settings }));
+}
+
 /** The environment may resolve paths, but it cannot change Core's selected policy. */
 export function validatePreparedRunnerLaunch(
   requirements: RunnerProviderLaunchRequirements,
@@ -138,10 +150,7 @@ export function validatePreparedRunnerLaunch(
       || descriptor.runId !== requirements.identity.runId
       || descriptor.normalizedSessionId !== requirements.identity.normalizedSessionId) reject();
   }
-  const expected = createRunnerProviderLaunch(requirements, {
-    options: { codexArgs: [] }, root: "/", cwd: "/", codexHome: "/",
-    providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "", codexCommand: "",
-  });
+  const expected = runnerLaunchPolicy(requirements);
   const hostFields = new Set(["command", "args", "sidecarCommand", "sidecarArgs", "runtimeDirectory", "cwd",
     "preconfiguredHome", "externallySandboxed", "runtimeContext", "instructions"]);
   for (const [key, value] of Object.entries(expected)) {
diff --git a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts
index 4f38d7edf40..4800a50f5e4 100644
--- a/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-acpx-permission-ordering.test.ts
@@ -124,16 +124,18 @@ describe("permission and tool order through the actual transport and harness", (
   });
 });
 
-async function bareTransport() {
+async function bareTransport(authorityRunId?: () => string | undefined) {
   const root = mkdtempSync(join(tmpdir(), "permission-queue-lifecycle-"));
   const binary = join(root, "unexecuted-runner");
   writeFileSync(binary, "not executable: process boundary is mocked");
   const bundle = createCapabilityRunnerdCodexTransport({ runnerBinary: binary, stateDirectory: root,
+    get authorityRunId() { return authorityRunId?.(); },
     sourceCodexHome: join(root, "absent-source-home"), environment: { HOME: root, PATH: "/usr/bin:/bin" },
   });
   await bundle.transport.request("initialize", {});
   await bundle.transport.request("thread/start", { cwd: root });
-  await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Fixture" }] });
+  const started = await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Fixture" }] });
+  const turnId = (started.turn as { id: string }).id;
   const iterator = bundle.transport.notifications()[Symbol.asyncIterator]();
   expect((await iterator.next()).value?.method).toBe("turn/started");
   const emitPermission = (requestId = "permission-7", prompt = "Run command") => peer.current!.emit("runtime_request.created", { request: {
@@ -142,9 +144,39 @@ async function bareTransport() {
     origin: { adapter: "acpx-runtime-sidecar", provider: "cursor", method: "session/request_permission" },
   } });
   const emitMarker = () => peer.current!.emit("item.delta", { kind: "agentMessage", text: "marker" });
-  return { bundle, iterator, emitPermission, emitMarker, async close() { await bundle.detachControllerForRestart(); rmSync(root, { recursive: true, force: true }); } };
+  return { bundle, iterator, turnId, emitPermission, emitMarker, async close() { await bundle.detachControllerForRestart(); rmSync(root, { recursive: true, force: true }); } };
 }
 
+it.each(["same-owner", "rebound-before-queue", "rebound-before-delivery"] as const)("fences deferred permission presentation and preserves settlement: %s", async scenario => {
+  let owner: string | undefined;
+  const fixture = await bareTransport(() => owner); const handler = vi.fn(async () => ({}));
+  fixture.bundle.transport.setServerRequestHandler(handler);
+  const sourceRunId = peer.current!.store.state.identity.runId;
+  owner = scenario === "rebound-before-queue" ? "replacement-owner" : sourceRunId;
+  try {
+    fixture.emitMarker(); fixture.emitPermission(); fixture.emitMarker();
+    expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta");
+    const request = structuredClone(peer.current!.store.state.committedEvents.find(event => event.eventType === "runtime_request.created"));
+    expect(request).toBeDefined();
+    expect(handler).not.toHaveBeenCalled();
+    if (scenario === "rebound-before-delivery") owner = "replacement-owner";
+    expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta");
+    expect(handler).toHaveBeenCalledTimes(scenario === "same-owner" ? 1 : 0);
+    const resolve = () => fixture.bundle.transport.resolveRuntimeRequest!({requestId:"permission-7",turnId:fixture.turnId,resolution:{action:"cancel"}});
+    if (scenario !== "same-owner") await expect(resolve()).rejects.toThrow("source_attachment_pending");
+    expect(peer.current!.store.state.committedEvents.find(event => event.eventType === "runtime_request.created")).toEqual(request);
+    // Suppressing presentation must retain the pending request. Its source
+    // owner can still settle it; the later durable receipt retires the binding.
+    owner = sourceRunId;
+    await resolve();
+    expect(peer.current!.store.state.commands.filter(command => command.type === "request.resolve")).toHaveLength(1);
+    peer.current!.emit("runtime_request.resolved", {requestId:"permission-7"}); fixture.emitMarker();
+    expect((await fixture.iterator.next()).value?.method).toBe("item/agentMessage/delta");
+    await expect(resolve()).rejects.toThrow("no longer pending");
+    expect(handler).toHaveBeenCalledTimes(scenario === "same-owner" ? 1 : 0);
+  } finally { await fixture.close(); }
+});
+
 it("discards a queued permission on close before the notification consumer asks for next", async () => {
   const fixture = await bareTransport(); const handler = vi.fn(async () => ({}));
   fixture.bundle.transport.setServerRequestHandler(handler);
diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts
index 72b95a8206c..d672d9bb2e3 100644
--- a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts
@@ -84,8 +84,10 @@ it.each([
     "wrong_thread",
     "callback_failure",
     "callback_async_failure",
+    "callback_checkpoint_failure",
+    "callback_checkpoint_owner_change",
   ].map((ordinaryFollowup) => ({
-    lossPoint: "after-confirmation",
+    lossPoint: ordinaryFollowup.startsWith("callback_") ? "after-result" : "after-confirmation",
     routed: true,
     recoveryFault: "none",
     ordinaryFollowup,
@@ -116,6 +118,7 @@ it.each([
     let recovering = false;
     let recoveryClaimCurrent = true;
     let recoveryFenceActive = true;
+    let recoveredOwner: string;
     const completionSnapshotIds: string[] = [];
     let rejectHeldCompletion: ((error: Error) => void) | undefined;
     let markHeldCompletionEntered!: () => void;
@@ -183,6 +186,26 @@ it.each([
       ) {
         const core = new OriginalCore(options);
         cores.push(core);
+        if (recovering && (ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change")) {
+          const getCommand = core.getCommand.bind(core);
+          let injected = false;
+          vi.spyOn(core, "getCommand").mockImplementation(id => {
+            const command = getCommand(id);
+            if (!injected && command?.type === "session.snapshot" && command.status === "completed") {
+              injected = true;
+              if (ordinaryFollowup === "callback_checkpoint_failure") {
+                vi.spyOn(core, "flushCheckpoint").mockRejectedValueOnce(new Error("fixture recovery completion callback failed"));
+              } else {
+                const flush = core.flushCheckpoint.bind(core);
+                vi.spyOn(core, "flushCheckpoint").mockImplementationOnce(async () => {
+                  await flush();
+                  recoveredOwner = "replacement-owner";
+                });
+              }
+            }
+            return command;
+          });
+        }
         if (recovering && snapshotFault) {
           const getCommand = core.getCommand.bind(core);
           snapshotObserverSpy = vi
@@ -694,8 +717,10 @@ it.each([
       const beforeProviderBytes = await readFile(
         join(stateDirectory, "runner", "codex-provider-state.json"),
       );
+      recoveredOwner = desired.runId;
       resumed = createCapabilityRunnerdCodexTransport({
         ...options,
+        get authorityRunId() { return ordinaryFollowup ? recoveredOwner : undefined; },
         ...(externalArtifact ? {
           runnerBinary: "/controller-missing-runner",
           runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary),
@@ -891,7 +916,9 @@ it.each([
       if (
         snapshotFault ||
         ordinaryFollowup === "callback_failure" ||
-        ordinaryFollowup === "callback_async_failure"
+        ordinaryFollowup === "callback_async_failure" ||
+        ordinaryFollowup === "callback_checkpoint_failure" ||
+        ordinaryFollowup === "callback_checkpoint_owner_change"
       ) {
         const firstRead = within(resumed.transport.request("thread/read", {}));
         void firstRead.catch(() => undefined);
@@ -932,10 +959,10 @@ it.each([
             ? ordinaryFollowup === "rejected_snapshot"
               ? "snapshot observation rejected"
               : "completion_unproven"
-            : "recovery completion callback failed",
+            : ordinaryFollowup === "callback_checkpoint_owner_change" ? "source_attachment_pending" : "recovery completion callback failed",
         );
         expect(recoveryFenceActive).toBe(true);
-        expect(completionSnapshotIds).toHaveLength(snapshotFault ? 0 : 1);
+        expect(completionSnapshotIds).toHaveLength(snapshotFault || ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change" ? 0 : 1);
         const beforeDeniedWork = cores[1]!.store.state.commands.map(
           (command) => command.commandId,
         );
@@ -962,18 +989,21 @@ it.each([
           beforeDeniedProviderCalls,
         );
         snapshotObserverSpy?.mockRestore();
+        recoveredOwner = desired.runId;
         const providerCallsBeforeRetry = await readFile(callsPath, "utf8");
         expect(
           (await within(resumed.transport.request("thread/read", {}))).thread,
         ).toMatchObject(thread);
         expect(recoveryFenceActive).toBe(false);
-        expect(completionSnapshotIds).toHaveLength(snapshotFault ? 1 : 2);
+        expect(completionSnapshotIds).toHaveLength(snapshotFault || ordinaryFollowup === "callback_checkpoint_failure" || ordinaryFollowup === "callback_checkpoint_owner_change" ? 1 : 2);
         expect(new Set(completionSnapshotIds).size).toBe(
           completionSnapshotIds.length,
         );
         expect(await readFile(callsPath, "utf8")).toBe(
           providerCallsBeforeRetry,
         );
+        await within(resumed.transport.request("turn/start", { input: [{ type: "text", text: "Work after completion retry" }] }));
+        for await (const event of resumed.transport.notifications()) if (event.method === "turn/completed") break;
         return;
       }
       const read = await within(
@@ -1016,6 +1046,7 @@ it.each([
           (await within(resumed.transport.request("thread/read", {}))).thread,
         ).toMatchObject(thread);
         const thirdRunId = randomUUID();
+        recoveredOwner = thirdRunId;
         await within(
           resumed.transport.attachRun!({
             runId: thirdRunId,
diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts
index 4db05f6b55a..fdcb5a2cc54 100644
--- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts
@@ -1173,7 +1173,7 @@ it("installs required accounting acceptance before the Runner acknowledgement ca
   }
 });
 
-it("uses the same API result envelope for live dispatch and recovered receipts", async () => {
+it.each([false, true])("uses the same API result envelope and fences handler publication (owner changed: %s)", async changedOwner => {
   const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-recovered-api-"));
   const OriginalCore = durableControlPlane.DurablePrpControlPlane;
   let callbacks!: ConstructorParameters<typeof OriginalCore>[0];
@@ -1185,17 +1185,31 @@ it("uses the same API result envelope for live dispatch and recovered receipts",
   let outcome: { result: unknown; isError?: boolean } | null = {
     result: { ok: true, apiOperationId: "POST /api/example", data: { saved: true } }, isError: false,
   };
+  let owner: string | undefined;
   const bundle = createCapabilityRunnerdCodexTransport({
+    get authorityRunId() { return owner; },
     runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
     codexArgs: fakeCodexArgs(stateDirectory), stateDirectory,
     recoverSemanticToolInput: async () => outcome,
   });
-  bundle.transport.setServerRequestHandler(async () => ({ success: !outcome?.isError,
-    contentItems: [{ type: "inputText", text: JSON.stringify(outcome?.result) }] }));
+  const handler = vi.fn(async () => ({ success: !outcome?.isError,
+    contentItems: [{ type: "inputText" as const, text: JSON.stringify(outcome?.result) }] }));
+  if (!changedOwner) bundle.transport.setServerRequestHandler(handler);
   try {
     await bundle.transport.request("thread/start", { cwd: tmpdir() });
     const call = { callId: "recovered-api", operationId: "call_api", input: { operationId: "POST /api/example" },
       sourceEventId: "saved-event", sourceEventType: "semantic_tool.input", correlation: core.store.state.identity };
+    if (changedOwner) {
+      owner = core.store.state.identity.runId;
+      const pending = callbacks.onSemanticToolInput!(call);
+      await new Promise<void>(resolveTurn => setImmediate(resolveTurn));
+      owner = "replacement-owner";
+      bundle.transport.setServerRequestHandler(handler);
+      await expect(pending).rejects.toBeInstanceOf(durableControlPlane.SemanticToolNotDispatchedError);
+      expect(handler).not.toHaveBeenCalled();
+      expect(await callbacks.recoverSemanticToolInput!(call)).toBeNull();
+      return;
+    }
     const live = await callbacks.onSemanticToolInput!(call);
     expect(live).toMatchObject({ isError: false, result: { ok: true, operationId: "call_api", callId: call.callId, result: outcome!.result } });
     expect(await callbacks.recoverSemanticToolInput!(call)).toMatchObject({ result: live.result, isError: live.isError });
diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
index f699208da3d..7452a2e5944 100644
--- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
+++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts
@@ -4,7 +4,7 @@ import { resolveGrokDisableWebSearch, admittedGrokWebSearchPolicy } from "../dri
 import { PRP_COMPLETION_TOOL_NAME, PRP_COMPLETION_RESULT_BRIDGE_INPUT_SCHEMA } from "../contracts/completion-result.js";
 import { RunnerLaunchPreparationStore } from "./runner-launch-preparation-store.js";
 import type { PreparedRunnerProviderLaunch, RunnerProviderLaunchRequirements } from "./runner-provider-launch.js";
-import { preparedRunnerLaunchWorkingDirectory, validatePreparedRunnerLaunch, runnerProviderLaunchRequirements, createRunnerProviderLaunch, createRunnerdCodexAppServerArgs, trustedRuntimeReadOnlyRoots } from "./runner-provider-launch.js";
+import { preparedRunnerLaunchWorkingDirectory, validatePreparedRunnerLaunch, runnerLaunchSettingsMatch, runnerProviderLaunchRequirements, createRunnerProviderLaunch, createRunnerdCodexAppServerArgs, trustedRuntimeReadOnlyRoots } from "./runner-provider-launch.js";
 export { createRunnerdCodexAppServerArgs, resolveRunnerdAcpxPermissionMode, trustedRuntimeReadOnlyRoots, withCodexCollaborationRuntimeInstructions } from "./runner-provider-launch.js";
 import { applyProviderObservabilityEnvironment, redactObservabilityDiagnostic } from "../drivers/agent-observability.js";
 import type { RunnerExecutionDriver } from "../control-plane/runner-execution-driver.js";
@@ -74,6 +74,7 @@ import {
   SemanticToolNotDispatchedError,
   durableRecoveryInternals,
   inspectWarmRunTransition,
+  isRunnerRecoveryMaintenanceCommand,
   spawnRunner,
   waitForProcess,
   type RunnerProcessHandle,
@@ -1255,6 +1256,8 @@ export interface CapabilityRunnerdCodexTransportOptions extends RunnerExecutionD
   cursorCloudProxyPath?: string;
   cursorCloudProxySha256?: string;
   recoveryPurpose?: "execution" | "stop_cleanup";
+  /** Cacheless preparation requires an authenticated authorityRunId throughout this transport's lifetime. */
+  recoveryLaunch?: { requirements: RunnerProviderLaunchRequirements };
   opencodePermissionMode?: NativeOpenCodePermissionMode;
   acpxAgent?: QualifiedAcpxAgent;
   /** Explicit evaluation-only candidate selection, never derived from persisted session input. */
@@ -3660,6 +3663,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   #rejectFailureSignal!: (error: Error) => void;
   #runnerRecoveryInProgress = false;
   #startupComplete = false;
+  readonly #authorityRunId: () => string | undefined;
   #startupFailureCode = "native_runner_process_exited";
   #controlPlaneCheckpoint:
     ((settlement: "settled" | "unsettled") => Promise<void> | void) | null =
@@ -3674,6 +3678,19 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   readonly #bridgedRuntimeInputs = new Map<string, { durableTurnId: string; permission?: boolean }>();
 
   constructor(readonly options: CapabilityRunnerdCodexTransportOptions) {
+    this.#authorityRunId = options.recoveryLaunch ? () => {
+      const owner = options.authorityRunId;
+      if (typeof owner !== "string" || !owner.trim()) throw new Error("native_runner_recovery_launch_requirements_invalid");
+      return owner;
+    } : () => options.authorityRunId;
+    if (options.recoveryLaunch) this.#authorityRunId();
+    const requirements = options.recoveryLaunch?.requirements;
+    if (options.recoveryLaunch && (!options.prepareLaunch || !options.resumeProviderSession?.driverSessionId
+      || requirements?.schema !== "paperclip.runner.launch-requirements.v1"
+      || requirements.provider !== (options.provider ?? "codex") || !options.prpIdentity
+      || commandDigest(requirements.identity) !== commandDigest(options.prpIdentity))) {
+      throw new Error("native_runner_recovery_launch_requirements_invalid");
+    }
     if (options.adoptExistingRunner && options.authorizeRunnerAdoption && !options.prepareLaunch) {
       throw new Error("runner_adoption_authorization_requires_launch_preparation");
     }
@@ -3811,6 +3828,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     };
   }
 
+  #retainsSourceInputs(): boolean {
+    const identity = this.#eventIdentity ?? this.options.prpIdentity;
+    return this.#authorityRunId() !== undefined && identity !== undefined && identity !== null
+      && this.#authorityRunId() !== identity.runId;
+  }
+
   evidence(): CapabilityRunnerdProcessEvidence {
     return structuredClone(this.#evidence);
   }
@@ -3821,6 +3844,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   ): Promise<Record<string, unknown>> {
     if (this.#closed) throw new Error("PRP Codex transport is closed");
     this.#throwIfFailed();
+    const expectedThread = this.#threadId || this.options.resumeProviderSession?.driverSessionId;
+    if (expectedThread && params.threadId !== undefined && params.threadId !== expectedThread
+      && ["thread/read", "thread/resume", "thread/turns/list", "thread/items/list", "thread/goal/get"].includes(method)) {
+      throw new Error("codex_history_identity_mismatch");
+    }
     if (this.#sourceRecoveryIdentity && !["initialize", "collaborationMode/list"].includes(method)) {
       throw new Error("native_runner_source_attachment_pending");
     }
@@ -4119,6 +4147,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
 
   takeRestoredRuntimeRequests(): CodexRpcServerRequest[] {
     this.#throwIfFailed();
+    if (this.#retainsSourceInputs()) return [];
     const requests = this.#restoredRuntimeRequests;
     this.#restoredRuntimeRequests = [];
     return requests.filter(request => this.#bridgedRuntimeInputs.has(String(request.id))
@@ -4133,7 +4162,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   }
 
   #recoverPendingSemanticInputsWhenReady(): void {
-    if (this.options.recoveryPurpose !== "stop_cleanup" && this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) {
+    if (!this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup" && this.#startupComplete && this.#handlerInstalled && this.#sourceRecoveryIdentity === null && this.#pendingWarmRecoveryCompletion === null) {
       void Promise.resolve().then(() => this.#core?.recoverPendingSemanticInputs())
         .catch(error => this.#failTransport(error instanceof Error ? error : new Error(String(error))));
     }
@@ -4238,11 +4267,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     if (this.#pendingWarmRecoveryCompletion !== null) {
       throw new Error("native_runner_warm_transition_completion_pending");
     }
+    const retainingSource = this.#retainsSourceInputs();
     const core = this.#core;
     if (!core || !this.#startupComplete) {
       throw new Error("native_runner_prp_run_rotation_unavailable");
     }
+    if (retainingSource && this.#bridgedRuntimeInputs.size > 0) {
+      throw new Error("native_runner_source_runtime_reconciliation_required");
+    }
     await this.#awaitWarmRunAttachmentReady();
+    if (retainingSource && !core.semanticToolResultsSettled()) {
+      throw new Error("native_runner_source_semantic_reconciliation_required");
+    }
     this.#assertStartupOpen();
     const currentRunGrant = input.currentRunGrant
       ? projectCopilotRunGrant(input.currentRunGrant, this.options.runnerFilesystemRoot)
@@ -4295,11 +4331,30 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
           );
       if (this.options.prepareLaunch && !staged) {
         const priorProvider = record(runAttachTemplate.provider);
-        const priorRequirements = new RunnerLaunchPreparationStore(this.#root).load(prior)?.requirements;
-        await this.#prepareOwnedProvider(runAttachTemplate, desired,
-          currentRunGrant ? currentRunGrant.runtimeContext : priorRequirements?.runtimeContext
-            ?? priorProvider.runtimeContext as NativeRuntimeContextSnapshot | null,
-          currentRunGrant?.instructions ?? priorRequirements?.instructions ?? String(priorProvider.instructions ?? ""));
+        const sourceRequirements = this.options.recoveryLaunch?.requirements;
+        const priorRequirements = new RunnerLaunchPreparationStore(this.#root).load(prior)?.requirements
+          ?? (sourceRequirements && recoveryIdentityMatches(sourceRequirements.identity, prior) ? sourceRequirements : undefined);
+        const runtimeContext = currentRunGrant ? currentRunGrant.runtimeContext : priorRequirements?.runtimeContext
+          ?? priorProvider.runtimeContext as NativeRuntimeContextSnapshot | null;
+        const instructions = currentRunGrant?.instructions ?? priorRequirements?.instructions ?? String(priorProvider.instructions ?? "");
+        let recordedRequirements: RunnerProviderLaunchRequirements | undefined;
+        if (sourceRequirements && priorRequirements
+          && priorRequirements.provider === sourceRequirements.provider
+          && prior.runnerInstanceId === sourceRequirements.identity.runnerInstanceId
+          && prior.environmentLeaseId === sourceRequirements.identity.environmentLeaseId
+          && prior.normalizedSessionId === sourceRequirements.identity.normalizedSessionId) {
+          // A recovered session keeps its frozen policy through later warm attaches.
+          const selected = runnerProviderLaunchRequirements(this.options, desired, {}, runtimeContext, instructions);
+          const settings = { ...priorRequirements.settings };
+          for (const key of Object.keys({ ...settings, ...selected.settings }) as Array<keyof RunnerProviderLaunchRequirements["settings"]>) {
+            if (this.options[key] !== undefined) Object.assign(settings, { [key]: selected.settings[key] });
+          }
+          recordedRequirements = { ...priorRequirements, identity: desired, runtimeContext, instructions };
+          if (!runnerLaunchSettingsMatch(recordedRequirements, settings)) {
+            throw new Error("native_runner_recovery_launch_policy_changed");
+          }
+        }
+        await this.#prepareOwnedProvider(runAttachTemplate, desired, runtimeContext, instructions, undefined, recordedRequirements);
       }
       this.#assertStartupOpen();
       this.#runAttachTemplate = structuredClone(runAttachTemplate);
@@ -4375,6 +4430,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
         await new Promise<void>((resolveWait) => setTimeout(resolveWait, 10));
       }
       this.#assertStartupOpen();
+      if (retainingSource) await core.flushCheckpoint();
       this.#eventIdentity = structuredClone(desired);
       this.#eventSourceSeq = 0;
       this.#deferredTurnStartEvents = [];
@@ -4382,6 +4438,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       if (!previousReleased) await previousRelease?.();
       previousReleased = true;
       this.#controlPlaneRelease = registration?.release ?? null;
+      if (retainingSource) {
+        core.enableSemanticInputDispatch(desired);
+        this.#recoverPendingSemanticInputsWhenReady();
+      }
     } catch (error) {
       const failure = error instanceof Error ? error : new Error(String(error));
       // The future route is ours from registration onward, including failures
@@ -4412,6 +4472,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     turnId: string;
     resolution: HarnessRuntimeRequestResolution;
   }): Promise<void> {
+    if (this.#retainsSourceInputs()) throw new Error("native_runner_source_attachment_pending");
     if (this.#pendingWarmRecoveryCompletion !== null) {
       throw new Error("native_runner_warm_transition_completion_pending");
     }
@@ -5040,8 +5101,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     this.#recordPreparedWorkingDirectory(prepared);
     this.#assertStartupOpen();
     template.provider = prepared.provider;
-    new RunnerLaunchPreparationStore(this.#root).save(requirements, prepared, this.#core!.store.state.identity);
-    this.#core!.persistPreparedLaunchConfiguration(identity, prepared.configurationId);
+    const sourceLaunch = this.options.recoveryLaunch
+      && recoveryIdentityMatches(identity, this.options.recoveryLaunch.requirements.identity);
+    if (!sourceLaunch) {
+      new RunnerLaunchPreparationStore(this.#root).save(requirements, prepared, this.#core!.store.state.identity);
+      this.#core!.persistPreparedLaunchConfiguration(identity, prepared.configurationId);
+    }
     return prepared;
   }
 
@@ -5320,19 +5385,22 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     const dynamicTools = Array.isArray(params.dynamicTools)
       ? params.dynamicTools.map(record)
       : [];
+    const authorityRunId = this.#authorityRunId;
     const core = new DurablePrpControlPlane({
       stateDirectory: resolve(this.#root, "control-plane"),
       identity,
       expectedRunnerVersion: runnerArtifact.version,
       expectedRunnerDigest: runnerArtifact.digest,
       onProtocolIntegrityError: (error) => this.#failTransport(error),
-      semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
+      get authorityRunId() { return authorityRunId(); },
+      semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
       onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
       recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call),
       onCommittedEvent: async (event) => {
-        if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) {
+        const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined);
+        if (source && event.runId === source.runId) {
           if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable");
-          await this.options.onRecoveryAccountingEvent?.(event, this.#sourceRecoveryIdentity);
+          await this.options.onRecoveryAccountingEvent?.(event, source);
         } else await this.options.onAccountingEvent?.(event);
         if (event.eventType === "auth.credentials.updated") {
           await this.options.onCredentialsUpdated?.(event);
@@ -5526,6 +5594,21 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     let identity = controlPlaneState
       ? controlPlaneIdentity(controlPlaneState)
       : desiredIdentity;
+    const predecessor = this.#authorityRunId() !== undefined
+      && this.#authorityRunId() !== identity.runId && recoveryIdentityMatches(identity, desiredIdentity);
+    const lostPredecessor = predecessor && !this.options.adoptExistingRunner;
+    if (predecessor && (!controlPlaneState || !recoveryIdentityMatches(identity, desiredIdentity)
+      || controlPlaneState.warmTransition || controlPlaneState.stagedRunAttachment
+      || (Array.isArray(controlPlaneState.commands) && controlPlaneState.commands.some(entry => {
+        const command = record(entry);
+        return command.status === "pending" && !isRunnerRecoveryMaintenanceCommand(String(command.type));
+      })))) {
+      throw new Error("native_runner_source_reconciliation_requires_exact_authority");
+    }
+    if (lostPredecessor && (this.options.provider ?? "codex") !== "codex") throw new Error("native_runner_recovery_integrity_unsupported");
+    if (lostPredecessor && this.options.sourceRecoveryVersion !== 1) {
+      throw new Error("native_runner_recovery_integrity_unsupported");
+    }
     const hasControlPlaneWarmBoundary =
       controlPlaneState?.warmTransition !== undefined ||
       controlPlaneState?.schema ===
@@ -5541,7 +5624,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
         );
       });
     if (
-      this.options.readRunnerState &&
+      !lostPredecessor && this.options.readRunnerState &&
       controlPlaneState &&
       (hasControlPlaneWarmBoundary ||
         controlPlaneState.completedWarmTransition !== undefined ||
@@ -5834,19 +5917,22 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
             opencodeExecutable!,
           )
         : undefined;
+    const authorityRunId = this.#authorityRunId;
     const core = new DurablePrpControlPlane({
       stateDirectory: controlPlaneDirectory,
       identity,
       expectedRunnerVersion: runnerArtifact.version,
       expectedRunnerDigest: runnerArtifact.digest,
       onProtocolIntegrityError: (error) => this.#failTransport(error),
-      semanticInputMode: this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
+      get authorityRunId() { return authorityRunId(); },
+      semanticInputMode: this.#retainsSourceInputs() || this.options.recoveryPurpose === "stop_cleanup" ? "retain_only" : "dispatch",
       onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
       recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call),
       onCommittedEvent: async (event) => {
-        if (this.#sourceRecoveryIdentity && event.runId === this.#sourceRecoveryIdentity.runId) {
+        const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined);
+        if (source && event.runId === source.runId) {
           if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable");
-          await this.options.onRecoveryAccountingEvent?.(event, this.#sourceRecoveryIdentity);
+          await this.options.onRecoveryAccountingEvent?.(event, source);
         } else await this.options.onAccountingEvent?.(event);
         if (event.eventType === "auth.credentials.updated") {
           await this.options.onCredentialsUpdated?.(event);
@@ -5877,7 +5963,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     // Reusing resources still requires current environment authorization. The
     // optional hook avoids re-preparation after controller storage is lost.
     // Existing consumers retain the preparation callback's authorization path.
-    const authorizeAdoption = this.options.adoptExistingRunner && this.options.authorizeRunnerAdoption;
+    const authorizeAdoption = !this.options.recoveryLaunch
+      && this.options.adoptExistingRunner && this.options.authorizeRunnerAdoption;
     if (this.options.prepareLaunch && authorizeAdoption) {
       const configurationId = core.preparedLaunchConfigurationId(desiredIdentity);
       if (!configurationId) throw new Error("runner_launch_recovery_configuration_unavailable");
@@ -5886,7 +5973,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       this.#recordPreparedWorkingDirectory({ provider: record(record(controlPlaneState!.runAttachTemplate).provider) });
     }
     let environmentPrepared: PreparedRunnerProviderLaunch | undefined;
-    if (this.options.prepareLaunch && !authorizeAdoption) {
+    if (this.options.recoveryLaunch) {
+      const requirements = this.options.recoveryLaunch.requirements;
+      const template = rotatedRunAttachPayload(controlPlaneState!, desiredIdentity, this.#authorizedTools,
+        this.options.resumeCompletionContract);
+      environmentPrepared = await this.#prepareOwnedProvider(template, desiredIdentity,
+        requirements.runtimeContext, requirements.instructions, undefined, requirements);
+      this.#runAttachTemplate = structuredClone(template);
+    } else if (this.options.prepareLaunch && !authorizeAdoption) {
       const preparations = new RunnerLaunchPreparationStore(this.#root);
       const saved = preparations.load(desiredIdentity);
       const previous = preparations.load(controlPlaneIdentity(controlPlaneState!));
@@ -6035,6 +6129,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
           : "restored provider identity from the exact durable checkpoint; awaiting live confirmation",
       );
     }
+    if (queuedSourceAttachment && this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup"
+      && this.#checkpointProviderIdentityExpectation === null) {
+      if (!this.#threadId || !this.#sessionId) throw new Error("native_runner_warm_transition_completion_unproven");
+      const saved = this.options.resumeProviderSession;
+      this.#checkpointProviderIdentityExpectation = {
+        driverSessionId: saved?.driverSessionId ?? this.#threadId,
+        providerSessionId: saved?.providerSessionId ?? this.#sessionId,
+        providerIdentity: saved?.providerIdentity === undefined ? this.#providerIdentity : record(saved.providerIdentity),
+      };
+      this.#confirmCheckpointProviderIdentity({ threadId: this.#threadId, sessionId: this.#sessionId,
+        providerIdentity: this.#providerIdentity }, "committed queued attachment identity");
+    }
     if (staged && !sourceAttachmentTarget) {
       await this.options.onStagedRunAttachment?.(core, { sourceIdentity: staged.sourceIdentity,
         targetIdentity: staged.targetIdentity, templateDigest: staged.templateDigest });
@@ -6080,7 +6186,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       ? recoveryIdentityMatches(warmRecovery?.runnerIdentity ?? queuedSourceAttachment!.sourceIdentity, recoverySource!)
         ? oldTransitionRegistration : newTransitionRegistration
       : this.options.controlPlaneRegistration
-        ? await this.options.controlPlaneRegistration(core, this.#sourceRecoveryIdentity ?? undefined) : null;
+        ? await this.options.controlPlaneRegistration(core, this.#retainsSourceInputs() ? identity : this.#sourceRecoveryIdentity ?? undefined) : null;
     this.#startupFailureCode =
       registration?.startupFailureCode ?? "runner_local_connect_failed";
     if (registration === null) {
@@ -6285,6 +6391,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       if (core.getCommand(warmRecovery?.commandId ?? queuedSourceAttachment!.commandId!)?.status !== "completed") {
         throw new Error("native_runner_warm_transition_result_unproven");
       }
+      if (queuedSourceAttachment && this.#retainsSourceInputs() && this.options.recoveryPurpose !== "stop_cleanup") {
+        const receipt = core.store.state.completedWarmTransition?.receipt;
+        if (!receipt || receipt.commandId !== queuedSourceAttachment.commandId
+          || !recoveryIdentityMatches(receipt.oldIdentity, queuedSourceAttachment.sourceIdentity)
+          || !recoveryIdentityMatches(receipt.newIdentity, recoveryTarget!)) {
+          throw new Error("native_runner_warm_transition_completion_unproven");
+        }
+        this.#pendingWarmRecoveryCompletion = { transitionId: receipt.transitionId, identity: structuredClone(recoveryTarget!) };
+      }
       desiredIdentity = recoveryTarget!;
       this.#durableTurnId = desiredIdentity.turnId;
       this.#eventIdentity = structuredClone(desiredIdentity);
@@ -6363,6 +6478,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     }
 
   async #recoverSemanticToolInput(call: Parameters<NonNullable<DurablePrpControlPlaneOptions["onSemanticToolInput"]>>[0]) {
+    if (this.#retainsSourceInputs() || (this.#authorityRunId() !== undefined && call.correlation.runId !== this.#authorityRunId())) return null;
     if (this.#sourceRecoveryIdentity) throw new Error("native_runner_source_semantic_reconciliation_required");
     if (call.operationId === "paperclip_finish" || call.operationId === "paperclip_block") {
       return this.#handleSemanticToolInput(call);
@@ -6414,6 +6530,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
     }
     if (
       !accepted ||
+      this.#retainsSourceInputs() ||
       this.#closed ||
       epoch !== this.#turnStartResponseEpoch ||
       threadId !== this.#threadId ||
@@ -6709,14 +6826,24 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       // final activation ACK. Callback failure retains the completion gate;
       // retry observes a fresh snapshot, never repeats a provider turn.
       const completing = (this.#warmRecoveryCompletionInFlight ??=
-        Promise.resolve().then(() =>
-          this.options.onWarmTransitionRecoveryCompleted?.({
+        Promise.resolve().then(async () => {
+          this.#assertStartupOpen();
+          await core.flushCheckpoint();
+          this.#assertStartupOpen();
+          const owner = this.#authorityRunId();
+          if (owner !== undefined && owner !== completion.identity.runId) throw new Error("native_runner_source_attachment_pending");
+          if (!recoveryIdentityMatches(core.store.state.identity, completion.identity) || core.store.state.warmTransition !== undefined) {
+            throw new Error("native_runner_warm_transition_completion_unproven");
+          }
+          await this.options.onWarmTransitionRecoveryCompleted?.({
             transitionId: completion.transitionId,
-          }),
-        ));
+          });
+        }));
       try {
         await completing;
+        this.#assertStartupOpen();
         if (this.#pendingWarmRecoveryCompletion === completion) {
+          if (this.options.recoveryPurpose !== "stop_cleanup") core.enableSemanticInputDispatch(completion.identity);
           this.#pendingWarmRecoveryCompletion = null;
           this.#recoverPendingSemanticInputsWhenReady();
         }
@@ -6985,10 +7112,14 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
             },
           };
           const threadId = this.#threadId, turnId = this.#turnId, handler = this.#handler;
+          const runId = core.store.state.identity.runId;
           this.#queue.pushControl(request, () => {
-            // A detached controller, replaced turn or settled durable request
-            // cannot acquire approval authority when a slow consumer resumes.
+            // Recheck authority when the consumer resumes. Retain the durable
+            // fact and pending binding while stale source presentation is held.
+            const owner = this.#authorityRunId();
             if (this.#closed || this.#failure || this.#core !== core || this.#handler !== handler
+              || this.#retainsSourceInputs()
+              || core.store.state.identity.runId !== runId || owner !== undefined && owner !== runId
               || this.#threadId !== threadId || this.#turnId !== turnId
               || this.#bridgedRuntimeInputs.get(requestId) !== binding) return;
             try {
diff --git a/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts
new file mode 100644
index 00000000000..603f9123d63
--- /dev/null
+++ b/packages/paperclip-runner/src/live/runnerd-source-reconciliation.test.ts
@@ -0,0 +1,333 @@
+import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { createHash } from "node:crypto";
+import { spawn } from "node:child_process";
+import { createServer } from "node:http";
+import type { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js";
+import { expect, it, vi } from "vitest";
+import { createCapabilityRunnerdCodexTransport, type CapabilityRunnerdCodexTransportOptions } from "./runnerd-codex-transport.js";
+import { createRunnerProviderLaunch, runnerLaunchRequirementsDigest, runnerProviderLaunchRequirements, type RunnerProviderLaunchRequirements } from "./runner-provider-launch.js";
+import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js";
+import { createNativeSessionBackend } from "../backends/native-backend-factory.js";
+import type { NativeExecutionInput } from "../contracts/native-execution.js";
+import type { PersistedNativeSession } from "../contracts/native-session-backend.js";
+
+type Bundle = ReturnType<typeof createCapabilityRunnerdCodexTransport>;
+const read = async (path: string) => JSON.parse(await readFile(path, "utf8"));
+const absent = (pid: number) => { try { process.kill(pid, 0); return false; } catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; } };
+const kill = (pid: number | null) => { if (pid) { try { process.kill(-pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } } };
+
+async function sourceFixture(registration?: CapabilityRunnerdCodexTransportOptions["controlPlaneRegistration"], nativeSource = false, homeMode: "isolated" | "preconfigured" = "isolated", disableCodexWebSearch = false) {
+  const root = await mkdtemp(join(tmpdir(), "runner-source-reconciliation-"));
+  const home = join(root, "empty-codex-home"); await mkdir(home);
+  const report = join(root, "completion.json");
+  if (nativeSource) await writeFile(report, JSON.stringify({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed fake work",
+    completionClaim: { contractRevision: "1", objectiveSatisfied: true, criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [] }], remainingWork: [] },
+    evidence: [], verification: [], attentionRequests: [], artifacts: [] }));
+  const identity = { runnerInstanceId: "runner-source", environmentLeaseId: "lease-source", runId: "run-source", normalizedSessionId: "session-source", turnId: "turn-source", itemId: "item-source" };
+  const options = {
+    runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
+    codexArgs: fakeCodexArgs(root, "--call-log", join(root, "calls"), "--record-process-start", "--durable-turn-ids", ...(nativeSource ? ["--terminal-result-file", report] : [])),
+    stateDirectory: root, sourceCodexHome: home, prpIdentity: identity, homeMode,
+    ...(disableCodexWebSearch ? { disableCodexWebSearch: true } : {}),
+    ...(homeMode === "preconfigured" ? { environment: { ...process.env, HOME: home, CODEX_HOME: home } } : {}),
+    lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 }, runnerProcessReconnectGraceMs: null, runnerReconnectGraceMs: 3_000,
+    ...(registration ? { controlPlaneRegistration: registration } : {}),
+  };
+  const bundles: Bundle[] = [];
+  const first = createCapabilityRunnerdCodexTransport(options); bundles.push(first);
+  const runnerFile = join(root, "runner", "runner-state.json"), providerFile = join(root, "runner", "codex-provider-state.json"), coreFile = join(root, "control-plane", "control-plane-state.json");
+  const calls = async (method: string) => (await readFile(join(root, "calls"), "utf8")).split("\n").filter(line => line === method).length;
+  const settled = async (completed = true) => vi.waitFor(async () => {
+    const runner = await read(runnerFile), provider = await read(providerFile), core = await read(coreFile);
+    expect(runner.outbox).toEqual([]); expect(runner.ackedSourceSeq).toBe(runner.nextSourceSeq - 1);
+    expect(provider.pendingEvents).toEqual([]); expect(provider.queuedEvents).toEqual([]);
+    expect(provider.activeProviderTurnId).toBeNull(); expect(provider.completedTurnAuthoritative).toBe(completed);
+    expect(core.commands.some((command: { status: string }) => command.status === "pending")).toBe(false);
+  }, { timeout: 5_000, interval: 25 });
+  const complete = async (bundle: Bundle, text: string) => {
+    await bundle.transport.request("turn/start", { input: [{ type: "text", text }] });
+    for await (const notification of bundle.transport.notifications()) if (notification.method === "turn/completed") return;
+    throw new Error("missing terminal turn");
+  };
+  const dispose = async () => {
+    for (const bundle of bundles.reverse()) { await bundle.transport.close().catch(() => undefined); const evidence = bundle.evidence(); kill(evidence.runnerPid); kill(evidence.codexPid); }
+    await rm(root, { recursive: true, force: true });
+  };
+  const nativeIdentity = { companyId: "company", agentId: "agent", issueId: "issue", runId: identity.runId, sessionId: identity.normalizedSessionId };
+  const execution: NativeExecutionInput = { schema: "paperclip.native-execution-input.v1",
+    binding: { companyId: nativeIdentity.companyId, agentId: nativeIdentity.agentId, issueId: nativeIdentity.issueId,
+      runId: identity.runId, executionWorkspaceId: "workspace" },
+    task: { identifier: "TASK-1", title: "Complete fake work", description: null, prompt: "Complete fake work", workMode: "standard" },
+    workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
+    session: { normalizedSessionId: identity.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1 },
+    provider: { kind: "codex", model: null, approvalPolicy: "never" },
+    completionContract: { id: "contract", sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.completion-contract.v1",
+      contract: { revision: "1", objective: "Complete fake work", criteria: [{ id: "objective", requirement: "Complete fake work" }] } }, interactionResponses: [], credentialBindings: [] };
+  try {
+    let nativeSnapshot: PersistedNativeSession | undefined;
+    let thread: { id: string; sessionId: string };
+    if (nativeSource) {
+      const session = await createNativeSessionBackend(execution, { codexTransportFactory: () => first.transport,
+        environment: { PAPERCLIP_WORKSPACE_CWD: root }, dynamicTools: [] }).openSession({ identity: nativeIdentity, workingDirectory: root });
+      await session.startTurn({ message: { role: "user", text: "Complete before abrupt loss." } });
+      for await (const event of session.events()) if (event.eventType === "turn.completed") break;
+      nativeSnapshot = await session.snapshot();
+      expect(nativeSnapshot.terminal?.runTerminalState).toBe("succeeded");
+      thread = { id: nativeSnapshot.sessionId, sessionId: nativeSnapshot.providerSessionId! };
+    } else {
+      const opened = await first.transport.request("thread/start", { cwd: root, dynamicTools: [] });
+      const originalThread = opened.thread as { id: string; sessionId: string };
+      thread = { id: originalThread.id, sessionId: originalThread.sessionId };
+      await complete(first, "Complete before abrupt loss.");
+    }
+    await settled();
+    const evidence = first.evidence();
+    await first.detachControllerForRestart();
+    expect(absent(evidence.runnerPid!)).toBe(false); expect(absent(evidence.codexPid!)).toBe(false);
+    kill(evidence.runnerPid); kill(evidence.codexPid);
+    await vi.waitFor(() => { expect(absent(evidence.runnerPid!)).toBe(true); expect(absent(evidence.codexPid!)).toBe(true); });
+    expect((await read(runnerFile)).lifecycle).toBe("ready");
+    const recover = (overrides: Partial<CapabilityRunnerdCodexTransportOptions> = {}) => {
+      const bundle = createCapabilityRunnerdCodexTransport({ ...options, sourceRecoveryVersion: 1, resumeProviderSession: { driverSessionId: thread.id, providerSessionId: thread.sessionId }, onRecoveryAccountingEvent: async () => {}, ...overrides, get authorityRunId() { return Object.hasOwn(overrides, "authorityRunId") ? overrides.authorityRunId : "run-target"; } });
+      bundles.push(bundle); return bundle;
+    };
+    return { root, options, identity, thread, runnerFile, providerFile, coreFile, calls, complete, settled, recover, dispose, nativeIdentity, execution, nativeSnapshot };
+  } catch (error) { await dispose(); throw error; }
+}
+
+it.each(["isolated", "preconfigured"] as const)("recovers the real native backend before attaching the next run with %s home", async homeMode => {
+  const f = await sourceFixture(undefined, true, homeMode);
+  try {
+    expect(f.nativeSnapshot!.sessionId).toBe(f.thread.id);
+    let owner = "run-target";
+    const source = f.recover({ ...(homeMode === "isolated" ? { environment: { ...process.env, PAPERCLIP_RUNNER_HOME_MODE: "preconfigured" } } : {}), get authorityRunId() { return owner; } });
+    await expect(source.transport.request("thread/read", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch");
+    const recovered = await createNativeSessionBackend(f.execution, { codexTransportFactory: () => source.transport,
+      environment: { PAPERCLIP_WORKSPACE_CWD: f.root }, dynamicTools: [] }).recoverSession!(f.nativeSnapshot!, { signal: new AbortController().signal });
+    expect(recovered.recovered).toBe(true);
+    expect(await f.calls("turn/start")).toBe(1); expect(await f.calls("thread/resume")).toBe(1);
+    const before = source.evidence();
+    await expect(source.transport.request("thread/resume", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch");
+    await expect(source.transport.request("thread/turns/list", { threadId: "foreign-thread" })).rejects.toThrow("history_identity_mismatch");
+    await expect(recovered.session!.startTurn({ message: { role: "user", text: "Not yet authorized" } })).rejects.toThrow();
+    expect(await f.calls("turn/start")).toBe(1);
+    await expect(source.transport.request("thread/goal/set", { objective: "Unowned source work", status: "active" })).rejects.toThrow("source_attachment_pending");
+    expect((await read(f.coreFile)).commands.some((command: { type: string }) => command.type === "session.goal.set")).toBe(false);
+    for (const runId of ["run-target", "run-warm-third"]) {
+      owner = runId;
+      await recovered.session!.attachRun!({ identity: { ...f.nativeIdentity, runId } });
+      expect(source.evidence().runnerPid).toBe(before.runnerPid);
+      await recovered.session!.startTurn({ message: { role: "user", text: "Continue under the current run." } });
+      let completed = false;
+      for await (const event of recovered.session!.events()) {
+        if (event.eventType === "turn.completed" && event.runId === runId) { completed = true; break; }
+      }
+      expect(completed).toBe(true); await f.settled();
+      const snapshot = await recovered.session!.snapshot();
+      expect(snapshot.sessionId).toBe(f.thread.id);
+      expect(snapshot.terminal).toMatchObject({ runTerminalState: "succeeded" });
+      expect(snapshot.semanticResult?.reportedWorkDisposition).toBe("done");
+    }
+    expect(await f.calls("turn/start")).toBe(3); expect(await f.calls("process-start")).toBe(2);
+    expect(await f.calls("thread/resume")).toBe(1);
+    await recovered.session!.close({ reason: "test complete" });
+  } finally { await f.dispose(); }
+}, 30_000);
+
+it.each(["attach", "suspend"] as const)("reconciles killed source then uses ordinary %s without a cold receipt", async mode => {
+  const fixture = await sourceFixture();
+  try {
+    const resumed = fixture.recover();
+    expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread);
+    await fixture.settled();
+    for (const method of ["turn/start", "turn/steer", "thread/goal/set"]) {
+      await expect(resumed.transport.request(method, method === "turn/steer" ? { input: [{ type: "text", text: "Unowned" }] } : {})).rejects.toThrow("source_attachment_pending");
+    }
+    expect(await fixture.calls("turn/start")).toBe(1);
+    const target = { ...fixture.identity, runId: "run-target", turnId: "turn-target", itemId: "item-target" };
+    if (mode === "suspend") {
+      await resumed.transport.close();
+      expect((await read(fixture.runnerFile)).lifecycle).toBe("suspended");
+      expect(await fixture.calls("process-start")).toBe(2);
+    } else {
+      const before = resumed.evidence();
+      await resumed.transport.attachRun!(target);
+      expect(resumed.evidence()).toMatchObject({ runnerPid: before.runnerPid, codexPid: before.codexPid });
+      await fixture.complete(resumed, "Continue under new authority."); await fixture.settled();
+      expect((await read(fixture.coreFile)).identity).toEqual(target);
+      expect((await read(fixture.providerFile)).threadId).toBe(fixture.thread.id);
+      expect(await fixture.calls("turn/start")).toBe(2);
+      expect(await fixture.calls("process-start")).toBe(2);
+    }
+  } finally { await fixture.dispose(); }
+}, 30_000);
+
+it.each([["launch", "default"], ["adopt", "default"], ["adopt-denied", "default"], ["launch", "saved"], ["launch", "conflict"]] as const)("prepares a cacheless source %s with %s policy without replacing the ordinary source configuration", async (mode, policy) => {
+  let authority: DurablePrpControlPlane | null = null;
+  const server = createServer();
+  server.on("upgrade", (request, socket, head) => {
+    if (authority) authority.handleUpgrade(request, socket, "/runner", head);
+    else socket.destroy();
+  });
+  await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (!address || typeof address === "string") throw new Error("missing source listener");
+  const registeredIdentities: unknown[] = [];
+  const registration: NonNullable<CapabilityRunnerdCodexTransportOptions["controlPlaneRegistration"]> = async (next, identity) => {
+    registeredIdentities.push(identity);
+    authority = next;
+    return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: () => { if (authority === next) authority = null; } };
+  };
+  let cleanup = async () => {};
+  try {
+    const fixture = await sourceFixture(registration, false, "isolated", policy !== "default"); cleanup = fixture.dispose;
+    const before = await read(fixture.coreFile), provider = before.runAttachTemplate.provider;
+    before.preparedLaunchConfigurations = [{ identity: fixture.identity, configurationId: "old-source-configuration" }];
+    await writeFile(fixture.coreFile, JSON.stringify(before));
+    const requirements = runnerProviderLaunchRequirements(fixture.options, fixture.identity, {
+      model: provider.model, approvalPolicy: provider.approvalPolicy,
+      permissions: provider.readOnly ? "paperclip-runner-workspace-read-only" : "paperclip-runner-workspace-only",
+      conversationMode: provider.conversationMode,
+      config: { include_collaboration_mode_instructions: provider.includeCollaborationModeInstructions !== false },
+    }, null, "You are a Paperclip agent.");
+    const prepare = vi.fn(async (input: RunnerProviderLaunchRequirements) => ({ configurationId: input.identity.runId === fixture.identity.runId ? "new-source-incarnation" : "target-configuration", requirementsDigest: runnerLaunchRequirementsDigest(input),
+      provider: createRunnerProviderLaunch(input, { options: fixture.options, root: fixture.root, cwd: provider.cwd,
+        codexHome: join(fixture.root, "codex-home"), codexCommand: fakeCodex, providerNodeCommand: "", opencodeProxyPath: "", acpxSidecarPath: "" }) }));
+    expect((await prepare(requirements)).provider).toEqual(provider); prepare.mockClear();
+    const binary = fixture.options.runnerBinary;
+    let owner: string | undefined = "prepared-target";
+    const sourceOptions: Partial<CapabilityRunnerdCodexTransportOptions> = { sourceRecoveryVersion: 1, get authorityRunId() { return owner; },
+      recoveryLaunch: { requirements }, prepareLaunch: prepare,
+      disableCodexWebSearch: policy === "conflict" ? false : undefined,
+      runnerArtifact: { version: "0.3.0", digest: `sha256:${createHash("sha256").update(await readFile(binary)).digest("hex")}` },
+      controlPlaneRegistration: registration,
+      runnerProcessLauncher: spec => {
+        expect(spec.nativeRunner).not.toHaveProperty("recoveryPurpose");
+        expect(spec.args).not.toContain("--source-reconcile");
+        const child = spawn(binary, [...spec.args], { env: spec.environment, stdio: "ignore", detached: process.platform !== "win32" });
+        return { child, completion: new Promise((resolve, reject) => {
+          child.once("error", reject); child.once("exit", (code, signal) => resolve({ code, signal, stdout: "", stderr: "" }));
+        }) };
+      },
+    };
+    if (mode === "launch") {
+      const launch = { ...fixture.options, ...sourceOptions,
+        resumeProviderSession: { driverSessionId: fixture.thread.id, providerSessionId: fixture.thread.sessionId } };
+      for (const authorityRunId of [undefined, "", "  "]) {
+        expect(() => createCapabilityRunnerdCodexTransport({ ...launch, authorityRunId }))
+          .toThrow("recovery_launch_requirements_invalid");
+      }
+      expect(() => createCapabilityRunnerdCodexTransport({ ...launch, authorityRunId: fixture.identity.runId })).not.toThrow();
+      expect(prepare).not.toHaveBeenCalled();
+    }
+    let resumed = fixture.recover(sourceOptions);
+    expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread);
+    expect(prepare).toHaveBeenCalledExactlyOnceWith(requirements, undefined);
+    expect(registeredIdentities.at(-1)).toEqual(fixture.identity);
+    expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toEqual(before.preparedLaunchConfigurations);
+    await expect(readFile(join(fixture.root, "launch-preparation.json"))).rejects.toHaveProperty("code", "ENOENT");
+    if (mode !== "launch") {
+      const evidence = resumed.evidence(), startedAt = resumed.transport.processInfo().startedAt;
+      await resumed.detachControllerForRestart();
+      prepare.mockClear();
+      if (mode === "adopt-denied") prepare.mockRejectedValueOnce(new Error("source preparation authorization revoked"));
+      const authorizeOldConfiguration = vi.fn(async () => { throw new Error("old configuration must not authorize source adoption"); });
+      const launch = vi.fn(() => { throw new Error("duplicate source process"); });
+      const adoption = { pid: evidence.runnerPid, processGroupId: evidence.runnerPid, startedAt,
+        isAlive: () => !absent(evidence.runnerPid!), signal: () => false };
+      resumed = fixture.recover({ ...sourceOptions, runnerProcessLauncher: launch, authorizeRunnerAdoption: authorizeOldConfiguration,
+        adoptExistingRunner: adoption });
+      if (mode === "adopt-denied") {
+        await expect(resumed.transport.request("thread/read", {})).rejects.toThrow("source preparation authorization revoked");
+        expect(absent(evidence.runnerPid!)).toBe(false);
+        expect(authorizeOldConfiguration).not.toHaveBeenCalled(); expect(launch).not.toHaveBeenCalled();
+        expect(await fixture.calls("turn/start")).toBe(1); expect(await fixture.calls("process-start")).toBe(2);
+        return;
+      }
+      expect((await resumed.transport.request("thread/read", {})).thread).toMatchObject(fixture.thread);
+      expect(resumed.evidence().runnerPid).toBe(evidence.runnerPid);
+      expect(registeredIdentities.at(-1)).toEqual(fixture.identity);
+      expect(prepare).toHaveBeenCalledExactlyOnceWith(requirements, undefined);
+      expect(authorizeOldConfiguration).not.toHaveBeenCalled(); expect(launch).not.toHaveBeenCalled();
+      expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toEqual(before.preparedLaunchConfigurations);
+      await expect(readFile(join(fixture.root, "launch-preparation.json"))).rejects.toHaveProperty("code", "ENOENT");
+    }
+    const target = { ...fixture.identity, runId: "prepared-target", turnId: "prepared-turn", itemId: "prepared-item" };
+    const attachedAuthority = authority!;
+    const grant = policy === "saved" ? { runtimeContext: null, instructions: requirements.instructions } : undefined;
+    if (policy === "conflict") {
+      const preparations = prepare.mock.calls.length;
+      await expect(resumed.transport.attachRun!(target)).rejects.toThrow("recovery_launch_policy_changed");
+      expect(prepare).toHaveBeenCalledTimes(preparations);
+      expect(attachedAuthority.store.state.identity).toEqual(fixture.identity);
+      expect(await fixture.calls("turn/start")).toBe(1);
+      return;
+    }
+    await resumed.transport.attachRun!({ ...target, ...(grant ? { currentRunGrant: grant } : {}) });
+    expect(prepare.mock.calls.at(-1)?.[0].instructions).toBe(grant?.instructions ?? requirements.instructions);
+    if (policy === "saved") expect(prepare.mock.calls.at(-1)?.[0].settings.disableCodexWebSearch).toBe(true);
+    expect((await read(fixture.coreFile)).preparedLaunchConfigurations).toContainEqual({ identity: target, configurationId: "target-configuration" });
+    expect((await read(join(fixture.root, "launch-preparation.json"))).entries).toContainEqual(expect.objectContaining({ prepared: expect.objectContaining({ configurationId: "target-configuration" }) }));
+    if (mode === "launch") {
+      expect(attachedAuthority.store.state.identity).toEqual(target);
+      const commands = attachedAuthority.store.state.commands.length;
+      owner = undefined;
+      await expect(resumed.transport.request("turn/start", { input: [{ type: "text", text: "Unowned work" }] }))
+        .rejects.toThrow("recovery_launch_requirements_invalid");
+      expect(() => attachedAuthority.queueCommand("turn.start", { text: "Unowned work" }))
+        .toThrow("recovery_launch_requirements_invalid");
+      expect(attachedAuthority.store.state.commands).toHaveLength(commands);
+      expect(await fixture.calls("turn/start")).toBe(1);
+      owner = "prepared-target";
+    }
+    await fixture.complete(resumed, "Continue after source preparation."); await fixture.settled();
+    expect(await fixture.calls("turn/start")).toBe(2); expect(await fixture.calls("process-start")).toBe(2);
+    if (policy === "saved") {
+      owner = "warm-third";
+      await resumed.transport.attachRun!({ runId: owner, turnId: "third-turn", itemId: "third-item", currentRunGrant: { runtimeContext: null, instructions: requirements.instructions } });
+      expect(prepare.mock.calls.at(-1)?.[0]).toMatchObject({ identity: { runId: owner }, settings: { disableCodexWebSearch: true }, instructions: requirements.instructions });
+      await fixture.complete(resumed, "Continue with frozen launch policy."); await fixture.settled();
+      expect(await fixture.calls("turn/start")).toBe(3); expect(await fixture.calls("process-start")).toBe(2);
+    }
+    await resumed.transport.close();
+    expect((await read(fixture.runnerFile)).lifecycle).toBe("suspended");
+    expect(() => fixture.recover({ recoveryLaunch: { requirements: { ...requirements, identity: { ...fixture.identity, runId: "foreign-source" } } }, prepareLaunch: prepare })).toThrow("recovery_launch_requirements_invalid");
+    expect(() => fixture.recover({ recoveryLaunch: { requirements }, prepareLaunch: undefined })).toThrow("recovery_launch_requirements_invalid");
+  } finally { await cleanup(); await new Promise<void>((resolve, reject) => server.close(error => error ? reject(error) : resolve())); }
+}, 30_000);
+
+it("rejects unexpected live goal after ordinary dormant recovery", async () => {
+  const fixture = await sourceFixture();
+  try {
+    const file = join(fixture.root, "fake-codex-state.json");
+    const actual = await read(file);
+    actual.goal = { objective: "Unexpected autonomous work", status: "active", tokenBudget: null, tokensUsed: 0, elapsedSeconds: 0, iterations: 0, createdAt: null, updatedAt: null, workingNow: false };
+    await writeFile(file, JSON.stringify(actual));
+    await expect(fixture.recover().transport.request("thread/read", {})).rejects.toThrow("saved dormant Codex session resumed an active or unknown goal");
+    expect(await fixture.calls("process-start")).toBe(2); expect(await fixture.calls("turn/start")).toBe(1);
+  } finally { await fixture.dispose(); }
+}, 30_000);
+
+it("retains replayed source tool inputs without dispatching business work or attaching the target", async () => {
+  const fixture = await sourceFixture();
+  try {
+    const provider = await read(fixture.providerFile);
+    provider.pendingEvents.push({ executorEventId: `codex_provider_${String(provider.nextProviderEventSeq++).padStart(16, "0")}`, eventType: "semantic_tool.input", priority: "p0", payload: { semantic_tool: {
+      schema: "paperclip.prp.semantic_tool.v1", schemaVersion: 1, phase: "input", callId: "retained-source-call", operationId: "get_task_context", input: {}, idempotencyKey: null,
+      correlation: { runId: fixture.identity.runId, normalizedSessionId: fixture.identity.normalizedSessionId, turnId: fixture.identity.turnId, itemId: fixture.identity.itemId },
+      content: { digest: `sha256:${createHash("sha256").update("{}").digest("hex")}`, redactionDisposition: "digest_only", references: [] },
+    } } });
+    await writeFile(fixture.providerFile, JSON.stringify(provider));
+    const resumed = fixture.recover(), dispatch = vi.fn(async () => ({ success: true, contentItems: [] }));
+    resumed.transport.setServerRequestHandler(dispatch);
+    await resumed.transport.request("thread/read", {});
+    await vi.waitFor(async () => expect((await read(fixture.coreFile)).committedEvents.some((event: { eventType: string }) => event.eventType === "semantic_tool.input")).toBe(true));
+    expect(dispatch).not.toHaveBeenCalled();
+    await expect(resumed.transport.attachRun!({ runId: "run-target", turnId: "turn-target", itemId: "item-target" })).rejects.toThrow();
+    expect(dispatch).not.toHaveBeenCalled();
+    expect((await read(fixture.coreFile)).commands.some((command: { type: string }) => command.type === "run.attach")).toBe(false);
+    expect(await fixture.calls("turn/start")).toBe(1);
+  } finally { await fixture.dispose(); }
+}, 30_000);
diff --git a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts
index 7d7d89ecd80..7749ce4d927 100644
--- a/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts
+++ b/packages/paperclip-runner/src/live/runnerd-staged-handoff.test.ts
@@ -99,7 +99,7 @@ it("delivers a WebSocket message coalesced with the real HTTP upgrade response",
 });
 
 // Real encrypted PRP and real Runner/provider processes; no model or server imports.
-it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every durable boundary without another process", async mode => {
+it.each([["connect", "owned"], ["listen", "owned"], ["connect", "legacy"], ["listen", "legacy"], ["connect", "cleanup"]] as const)("recovers a staged %s %s handoff at every durable boundary without another process", async (mode, policy) => {
   const root = await mkdtemp(join(tmpdir(), "runner-staged-handoff-"));
   const shared = join(root, "shared"), calls = join(root, "calls.log");
   const source = { runnerInstanceId: "runner-staged", environmentLeaseId: "lease-staged", runId: "source-run",
@@ -141,6 +141,7 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d
       const registration = async (core: DurablePrpControlPlane, identity = core.store.state.identity) => {
         currentCore = core;
         if (!registered.has(core)) {
+          if (policy === "cleanup") vi.spyOn(core, "enableSemanticInputDispatch");
           registered.add(core); cores.add(core);
           const epoch = generation;
           core.setCheckpointPersistence(async saved => {
@@ -208,7 +209,8 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d
         checkpoint.restoreSessionDirectory(next);
         const noSpawn = vi.fn(() => { throw new Error("Duplicate Runner launch"); });
         const recoveryOptions: Parameters<typeof createCapabilityRunnerdCodexTransport>[0] = { ...options, stateDirectory: next,
-          prpIdentity: target, runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary),
+          authorityRunId: policy === "legacy" ? undefined : target.runId,
+          recoveryPurpose: policy === "cleanup" ? "stop_cleanup" : undefined, prpIdentity: target, runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary),
           runnerStateDirectory: join(directory, "runner"), readRunnerState: async () => JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8")),
           prepareExternalRunnerState: async () => { throw new Error("Verified live handoff must not restore a backup"); },
           runnerProcessLauncher: noSpawn, warmTransitionRegistrationMode: mode === "listen" ? "routed_same_endpoint" : "routed_connect",
@@ -216,6 +218,13 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d
           adoptExistingRunner: { pid: runnerPid, processGroupId: runnerPid, startedAt,
             isAlive: () => !dead(runnerPid!), signal: signal => { if (dead(runnerPid!)) return true; process.kill(-runnerPid!, signal); return true; } } };
         if (boundary === "queued") {
+          if (policy === "owned") {
+            const wrongProvider = createCapabilityRunnerdCodexTransport({ ...recoveryOptions,
+              resumeProviderSession: { ...recoveryOptions.resumeProviderSession!, driverSessionId: "foreign-thread" } });
+            await expect(wrongProvider.transport.request("thread/read", {})).rejects.toThrow("provider_identity_mismatch");
+            await wrongProvider.detachControllerForRestart().catch(() => undefined);
+            checkpoint.restoreSessionDirectory(next);
+          }
           const noProof = createCapabilityRunnerdCodexTransport({ ...recoveryOptions, adoptExistingRunner: undefined });
           await expect(noProof.transport.request("thread/read", {})).rejects.toThrow("queued_attachment_requires_verified_live_source");
           await noProof.detachControllerForRestart().catch(() => undefined);
@@ -250,11 +259,16 @@ it.each(["connect", "listen"] as const)("recovers a staged %s handoff at every d
         await currentCore.flushCheckpoint(); expect(checkpoint).toBe(beforeMark);
         await within(replacement.transport.request("thread/goal/get", {}), "post-checkpoint read");
         expect(currentCore.store.state.stagedRunAttachment).toBeUndefined();
+        if (policy === "cleanup") expect(currentCore.enableSemanticInputDispatch).not.toHaveBeenCalled();
+        else {
+          await within(replacement.transport.request("turn/start", { input: [{ type: "text", text: "Work after recovered attachment" }] }), "target turn");
+          for await (const event of replacement.transport.notifications()) if (event.method === "turn/completed") break;
+        }
         expect(noSpawn).not.toHaveBeenCalled();
         const effects = (await readFile(callLog, "utf8")).trim().split(/\r?\n/);
         expect(effects.filter(line => line === "process-start")).toHaveLength(1);
         expect(effects.filter(line => line === "thread/start")).toHaveLength(1);
-        expect(effects.filter(line => line === "turn/start")).toHaveLength(0);
+        expect(effects.filter(line => line === "turn/start")).toHaveLength(policy === "cleanup" ? 0 : 1);
       } catch (error) {
         const physical = JSON.parse(await readFile(join(directory, "runner/runner-state.json"), "utf8").catch(() => "{}"));
         throw new Error(`${mode}/${boundary}: ${String(error)}; ${JSON.stringify({