diff --git a/crates/freshell-runtime-protocol/src/lib.rs b/crates/freshell-runtime-protocol/src/lib.rs
index 3e8e71e0f..2fe93202b 100644
--- a/crates/freshell-runtime-protocol/src/lib.rs
+++ b/crates/freshell-runtime-protocol/src/lib.rs
@@ -3453,6 +3453,13 @@ pub enum HostCommand {
         fresh_agent: Option<Box<FreshAgentLaunchSpec>>,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         resume_spec: Option<Box<ResumeSpec>>,
+        /// How long the host may spend dispatching this grant before it must
+        /// reply. The supervisor waits a little longer than this for the
+        /// reply, so a host that runs out refuses the grant with a typed error
+        /// naming the stage it was in, instead of being abandoned mid-dispatch.
+        /// Absent means the host applies no bound of its own.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        dispatch_budget_ms: Option<u64>,
     },
     Stop {
         incarnation_id: IncarnationId,
diff --git a/crates/freshell-session-host/src/main.rs b/crates/freshell-session-host/src/main.rs
index 3826e695c..08b359d7c 100644
--- a/crates/freshell-session-host/src/main.rs
+++ b/crates/freshell-session-host/src/main.rs
@@ -378,6 +378,7 @@ async fn dispatch(
                     terminal,
                     fresh_agent,
                     resume_spec,
+                    dispatch_budget_ms,
                 } => {
                     ensure_incarnation(&incarnation_id, state)?;
                     if host_boot_id != state.host_boot_id {
@@ -398,6 +399,7 @@ async fn dispatch(
                             terminal: terminal.map(|terminal| *terminal),
                             fresh_agent: fresh_agent.map(|agent| *agent),
                             resume_spec: resume_spec.map(|resume_spec| *resume_spec),
+                            dispatch_budget: dispatch_budget_ms.map(Duration::from_millis),
                         },
                     )
                     .await
@@ -938,11 +940,134 @@ struct GrantExecutionRequest {
     terminal: Option<TerminalLaunchSpec>,
     fresh_agent: Option<FreshAgentLaunchSpec>,
     resume_spec: Option<ResumeSpec>,
+    /// The supervisor's bound on this dispatch; see [`GrantStageTimeline`].
+    dispatch_budget: Option<Duration>,
+}
+
+/// Wall-clock breakdown of one execution grant, and its dispatch budget.
+///
+/// The supervisor sends the budget with the grant and waits a fixed reply
+/// allowance longer than it for the reply. The host, not the supervisor's
+/// wait, therefore ends a dispatch that runs long: every awaited stage runs
+/// inside the budget, the native dispatch boundary is never crossed once the
+/// budget is spent (the grant stays unconsumed), and the refusal names the
+/// stage the budget ran out before. A synchronous stage (the provider
+/// projection, the provider-owned config refresh) cannot be interrupted; the
+/// host notices when it returns, before the next stage starts.
+///
+/// The breakdown names the stage that spent the time, for every outcome, so
+/// a slow dispatch is diagnosable after the container is gone.
+struct GrantStageTimeline {
+    started: std::time::Instant,
+    stage_started: std::time::Instant,
+    stages: Vec<(&'static str, u128)>,
+    budget: Option<Duration>,
+    exhausted_before: Option<&'static str>,
+}
+
+impl GrantStageTimeline {
+    fn start(budget: Option<Duration>) -> Self {
+        let now = std::time::Instant::now();
+        Self {
+            started: now,
+            stage_started: now,
+            stages: Vec::new(),
+            budget,
+            exhausted_before: None,
+        }
+    }
+
+    fn deadline(&self) -> Option<std::time::Instant> {
+        self.budget.map(|budget| self.started + budget)
+    }
+
+    fn budget_exhausted(&mut self, stage: &'static str) -> RuntimeError {
+        self.exhausted_before = Some(stage);
+        RuntimeError::new(
+            RuntimeErrorCode::HostUnreachable,
+            format!(
+                "execution grant spent its {}ms dispatch budget before {stage} completed",
+                self.budget.unwrap_or_default().as_millis()
+            ),
+        )
+    }
+
+    /// Refuse to begin `stage` once the dispatch budget is spent.
+    fn ensure_budget(&mut self, stage: &'static str) -> Result<(), RuntimeError> {
+        match self.deadline() {
+            Some(deadline) if std::time::Instant::now() >= deadline => {
+                Err(self.budget_exhausted(stage))
+            }
+            _ => Ok(()),
+        }
+    }
+
+    /// Run one awaited dispatch stage inside the budget and record it.
+    async fn run<T>(
+        &mut self,
+        stage: &'static str,
+        work: impl std::future::Future<Output = Result<T, RuntimeError>>,
+    ) -> Result<T, RuntimeError> {
+        self.ensure_budget(stage)?;
+        let result = match self.deadline() {
+            Some(deadline) => {
+                match tokio::time::timeout_at(tokio::time::Instant::from_std(deadline), work).await
+                {
+                    Ok(result) => result,
+                    Err(_) => Err(self.budget_exhausted(stage)),
+                }
+            }
+            None => work.await,
+        };
+        if result.is_ok() {
+            self.completed(stage);
+        }
+        result
+    }
+
+    fn completed(&mut self, stage: &'static str) {
+        let now = std::time::Instant::now();
+        self.stages
+            .push((stage, now.duration_since(self.stage_started).as_millis()));
+        self.stage_started = now;
+    }
+
+    fn event(&self, grant_id: &GrantId, outcome: Result<(), &RuntimeError>) -> serde_json::Value {
+        serde_json::json!({
+            "severity": if outcome.is_ok() { "info" } else { "warn" },
+            "event": "host.grant_timing",
+            "grantId": grant_id,
+            "outcome": if outcome.is_ok() { "accepted" } else { "refused" },
+            "errorCode": outcome.err().map(|error| error.code),
+            "totalMs": self.started.elapsed().as_millis(),
+            "budgetMs": self.budget.map(|budget| budget.as_millis()),
+            "budgetExhaustedBefore": self.exhausted_before,
+            "stages": self
+                .stages
+                .iter()
+                .map(|(stage, ms)| serde_json::json!({"stage": stage, "ms": ms}))
+                .collect::<Vec<_>>(),
+        })
+    }
 }
 
 async fn grant_execution(
     state: &Arc<HostState>,
     request: GrantExecutionRequest,
+) -> Result<HostResult, RuntimeError> {
+    let grant_id = request.grant_id.clone();
+    let mut timeline = GrantStageTimeline::start(request.dispatch_budget);
+    let result = grant_execution_stages(state, request, &mut timeline).await;
+    let event = timeline.event(&grant_id, result.as_ref().map(|_| ()));
+    emit_environment_event(event.clone());
+    append_event_with_secret(&state.state_dir, &state.secret, "host.grant_timing", event).ok();
+    result
+}
+
+async fn grant_execution_stages(
+    state: &Arc<HostState>,
+    request: GrantExecutionRequest,
+    timeline: &mut GrantStageTimeline,
 ) -> Result<HostResult, RuntimeError> {
     let GrantExecutionRequest {
         soul_id,
@@ -954,6 +1079,7 @@ async fn grant_execution(
         terminal,
         fresh_agent,
         resume_spec,
+        dispatch_budget: _,
     } = request;
     let provider = terminal
         .as_ref()
@@ -1008,6 +1134,7 @@ async fn grant_execution(
             "host boot already consumed a different grant",
         ));
     }
+    timeline.completed("admit");
 
     // Resolve every mounted grant once before consuming the execution grant.
     // A changed source selection remains a pre-dispatch preparation failure.
@@ -1043,29 +1170,34 @@ async fn grant_execution(
                 format!("resolve managed provider credentials: {error}"),
             )
         })?;
+        timeline.completed("resolve_secrets");
         if matches!(provider, "opencode" | "codex" | "claude" | "amplifier") {
-            let generated = provider_input_preparation::prepare_for_identity(
-                context,
-                &mut secrets,
-                &provider_path_inputs,
-                &state
-                    .provider_config_paths
-                    .private_mount
-                    .join("user-provider"),
-                &state.provider_config_paths.provider_home,
-                identity,
-            )
-            .await
-            .map_err(|error| {
-                RuntimeError::new(
-                    if error.starts_with("configuration reference selection changed") {
-                        RuntimeErrorCode::ProviderInputsChanged
-                    } else {
-                        RuntimeErrorCode::HostUnreachable
-                    },
-                    error,
-                )
-            })?;
+            let generated = timeline
+                .run("prepare_provider_inputs", async {
+                    provider_input_preparation::prepare_for_identity(
+                        context,
+                        &mut secrets,
+                        &provider_path_inputs,
+                        &state
+                            .provider_config_paths
+                            .private_mount
+                            .join("user-provider"),
+                        &state.provider_config_paths.provider_home,
+                        identity,
+                    )
+                    .await
+                    .map_err(|error| {
+                        RuntimeError::new(
+                            if error.starts_with("configuration reference selection changed") {
+                                RuntimeErrorCode::ProviderInputsChanged
+                            } else {
+                                RuntimeErrorCode::HostUnreachable
+                            },
+                            error,
+                        )
+                    })
+                })
+                .await?;
             secrets.environment.extend(generated);
         } else if !provider_path_inputs.is_empty() {
             return Err(RuntimeError::new(
@@ -1129,6 +1261,7 @@ async fn grant_execution(
                         .arg(&resume.provider_session.native_session_id);
                 }
             }
+            timeline.ensure_budget("consume_grant")?;
             consume_grant(&mut persisted)?;
             let child = command.spawn().map_err(|e| {
                 RuntimeError::new(
@@ -1143,6 +1276,7 @@ async fn grant_execution(
                 )
             })?;
             *state.child.lock().await = Some(child);
+            timeline.completed("spawn_worker");
             let evidence = if fixture == FixtureKind::SecurityProbe {
                 security_probe(&state.state_dir)
             } else {
@@ -1201,52 +1335,73 @@ async fn grant_execution(
                 )
                 .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?;
             }
-            let prepared = providers::prepare_terminal(
-                terminal,
-                &resolved_secrets.environment,
-                resolved_secrets.opencode_plan.as_ref(),
-                resolved_secrets.claude_settings_argument.as_deref(),
-            )
-            .await
-            .map_err(|error| {
-                RuntimeError::new(
-                    RuntimeErrorCode::HostUnreachable,
-                    format!("prepare managed provider: {error}"),
+            timeline.completed("prepare_provider_state");
+            let prepared = timeline
+                .run("prepare_terminal", async {
+                    providers::prepare_terminal(
+                        terminal,
+                        &resolved_secrets.environment,
+                        resolved_secrets.opencode_plan.as_ref(),
+                        resolved_secrets.claude_settings_argument.as_deref(),
+                    )
+                    .await
+                    .map_err(|error| {
+                        RuntimeError::new(
+                            RuntimeErrorCode::HostUnreachable,
+                            format!("prepare managed provider: {error}"),
+                        )
+                    })
+                })
+                .await?;
+            timeline
+                .run(
+                    "verify_claude_sources",
+                    claude_source_observation::verify_final(
+                        prepared.terminal.provider_launch_context.as_ref(),
+                        &state
+                            .provider_config_paths
+                            .private_mount
+                            .join("user-provider"),
+                        Path::new(&prepared.terminal.cwd),
+                        &state.provider_config_paths.provider_home,
+                        prepared.terminal.run_as_uid,
+                        prepared.terminal.run_as_gid,
+                        freshell_platform::claude_inputs::NativeSourceRuntime::Terminal,
+                    ),
                 )
-            })?;
-            claude_source_observation::verify_final(
-                prepared.terminal.provider_launch_context.as_ref(),
-                &state
-                    .provider_config_paths
-                    .private_mount
-                    .join("user-provider"),
-                Path::new(&prepared.terminal.cwd),
-                &state.provider_config_paths.provider_home,
-                prepared.terminal.run_as_uid,
-                prepared.terminal.run_as_gid,
-                freshell_platform::claude_inputs::NativeSourceRuntime::Terminal,
-            )
-            .await?;
+                .await?;
+            timeline.ensure_budget("consume_grant")?;
             consume_grant(&mut persisted)?;
-            let prepared = providers::start_terminal_sidecar(prepared)
-                .await
-                .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?;
+            timeline.completed("consume_grant");
+            let prepared = timeline
+                .run("start_sidecar", async {
+                    providers::start_terminal_sidecar(prepared)
+                        .await
+                        .map_err(|error| {
+                            RuntimeError::new(RuntimeErrorCode::HostUnreachable, error)
+                        })
+                })
+                .await?;
             let terminal = prepared.terminal;
-            let hosted = HostedPty::spawn(
-                &state.state_dir,
-                state.incarnation_id.clone(),
-                &terminal,
-                prepared.child_env,
-                prepared.codex,
-                prepared.claude_settings,
-            )
-            .await
-            .map_err(|e| {
-                RuntimeError::new(
-                    RuntimeErrorCode::HostUnreachable,
-                    format!("spawn managed terminal: {e}"),
-                )
-            })?;
+            let hosted = timeline
+                .run("spawn_worker", async {
+                    HostedPty::spawn(
+                        &state.state_dir,
+                        state.incarnation_id.clone(),
+                        &terminal,
+                        prepared.child_env,
+                        prepared.codex,
+                        prepared.claude_settings,
+                    )
+                    .await
+                    .map_err(|e| {
+                        RuntimeError::new(
+                            RuntimeErrorCode::HostUnreachable,
+                            format!("spawn managed terminal: {e}"),
+                        )
+                    })
+                })
+                .await?;
             let pid = hosted.pid();
             let evidence = serde_json::json!({
                 "workload":"terminal",
@@ -1352,37 +1507,51 @@ async fn grant_execution(
                 )
                 .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?;
             }
-            providers::prepare_fresh_agent_child_environment(
-                &launch,
-                &mut resolved_secrets.environment,
-                resolved_secrets.opencode_plan.as_ref(),
-            )
-            .await
-            .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?;
-            claude_source_observation::verify_final(
-                launch.provider_launch_context.as_ref(),
-                &state
-                    .provider_config_paths
-                    .private_mount
-                    .join("user-provider"),
-                Path::new(&launch.cwd),
-                &state.provider_config_paths.provider_home,
-                launch.run_as_uid,
-                launch.run_as_gid,
-                freshell_platform::claude_inputs::NativeSourceRuntime::Sdk,
-            )
-            .await?;
+            timeline
+                .run("prepare_provider_state", async {
+                    providers::prepare_fresh_agent_child_environment(
+                        &launch,
+                        &mut resolved_secrets.environment,
+                        resolved_secrets.opencode_plan.as_ref(),
+                    )
+                    .await
+                    .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))
+                })
+                .await?;
+            timeline
+                .run(
+                    "verify_claude_sources",
+                    claude_source_observation::verify_final(
+                        launch.provider_launch_context.as_ref(),
+                        &state
+                            .provider_config_paths
+                            .private_mount
+                            .join("user-provider"),
+                        Path::new(&launch.cwd),
+                        &state.provider_config_paths.provider_home,
+                        launch.run_as_uid,
+                        launch.run_as_gid,
+                        freshell_platform::claude_inputs::NativeSourceRuntime::Sdk,
+                    ),
+                )
+                .await?;
+            timeline.ensure_budget("consume_grant")?;
             consume_grant(&mut persisted)?;
+            timeline.completed("consume_grant");
             freshell_platform::managed_child_secrets::install(resolved_secrets.environment)
                 .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?;
-            let actor = providers::open_hosted_fresh_agent(&state.state_dir, launch.clone())
-                .await
-                .map_err(|error| {
-                    RuntimeError::new(
-                        RuntimeErrorCode::HostUnreachable,
-                        format!("hosted fresh-agent provider failed to start: {error}"),
-                    )
-                })?;
+            let actor = timeline
+                .run("start_fresh_agent", async {
+                    providers::open_hosted_fresh_agent(&state.state_dir, launch.clone())
+                        .await
+                        .map_err(|error| {
+                            RuntimeError::new(
+                                RuntimeErrorCode::HostUnreachable,
+                                format!("hosted fresh-agent provider failed to start: {error}"),
+                            )
+                        })
+                })
+                .await?;
             let profile = actor.profile().await;
             let native_session_id = profile.native_session_id.clone();
             let evidence = serde_json::json!({
@@ -3228,12 +3397,33 @@ fn sync_provider_projection_directory(path: &Path) -> Result<(), String> {
 }
 
 fn sync_provider_projection_tree(root: &Path) -> Result<(), String> {
-    sync_provider_projection_tree_with(root, &mut |path| std::fs::File::open(path)?.sync_all())
+    sync_provider_projection_tree_with(root, &|path| std::fs::File::open(path)?.sync_all())
 }
 
+/// Concurrent flushes per projection durability barrier. Every flush is one
+/// journal commit when issued alone; overlapping flushes of independent nodes
+/// let the filesystem commit them together, so a selected environment with
+/// thousands of files (an installed plugin's node_modules) costs a handful of
+/// commits instead of one per file. Measured on the qualification host's
+/// ext4 volume, 4,200 files took 13.4 s one at a time and 1.4 s at 16.
+const PROVIDER_PROJECTION_SYNC_WORKERS: usize = 16;
+
+/// Make every node of a staged projection durable before the transaction
+/// advances. Symlinks are never followed. Each node is still flushed exactly
+/// once with a full `fsync`; only their order is relaxed, which the
+/// transaction does not depend on because it waits for the whole barrier.
 fn sync_provider_projection_tree_with(
     root: &Path,
-    sync_node: &mut impl FnMut(&Path) -> std::io::Result<()>,
+    sync_node: &(impl Fn(&Path) -> std::io::Result<()> + Sync),
+) -> Result<(), String> {
+    let mut nodes = Vec::new();
+    collect_provider_projection_sync_nodes(root, &mut nodes)?;
+    flush_provider_projection_nodes(&nodes, sync_node)
+}
+
+fn collect_provider_projection_sync_nodes(
+    root: &Path,
+    nodes: &mut Vec<PathBuf>,
 ) -> Result<(), String> {
     let metadata = std::fs::symlink_metadata(root).map_err(|error| {
         format!(
@@ -3251,15 +3441,63 @@ fn sync_provider_projection_tree_with(
             let entry = entry.map_err(|error| {
                 format!("provider projection sync read {}: {error}", root.display())
             })?;
-            sync_provider_projection_tree_with(&entry.path(), sync_node)?;
+            collect_provider_projection_sync_nodes(&entry.path(), nodes)?;
         }
     }
-    sync_node(root).map_err(|error| {
-        format!(
-            "provider projection flush failed for {}: {error}",
-            root.display()
-        )
-    })
+    nodes.push(root.to_path_buf());
+    Ok(())
+}
+
+fn flush_provider_projection_nodes(
+    nodes: &[PathBuf],
+    sync_node: &(impl Fn(&Path) -> std::io::Result<()> + Sync),
+) -> Result<(), String> {
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+    let next = AtomicUsize::new(0);
+    let failed = AtomicBool::new(false);
+    let failure = std::sync::Mutex::new(None::<String>);
+    let flush = || {
+        while !failed.load(Ordering::Acquire) {
+            let Some(path) = nodes.get(next.fetch_add(1, Ordering::Relaxed)) else {
+                return;
+            };
+            if let Err(error) = sync_node(path) {
+                failed.store(true, Ordering::Release);
+                failure
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner)
+                    .get_or_insert_with(|| {
+                        format!(
+                            "provider projection flush failed for {}: {error}",
+                            path.display()
+                        )
+                    });
+                return;
+            }
+        }
+    };
+    std::thread::scope(|scope| {
+        // The calling thread always takes part, so the barrier completes even
+        // when the container's pid budget refuses additional threads.
+        for _ in 1..PROVIDER_PROJECTION_SYNC_WORKERS.min(nodes.len()) {
+            if std::thread::Builder::new()
+                .name("projection-sync".into())
+                .spawn_scoped(scope, flush)
+                .is_err()
+            {
+                break;
+            }
+        }
+        flush();
+    });
+    match failure
+        .into_inner()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+    {
+        Some(error) => Err(error),
+        None => Ok(()),
+    }
 }
 
 fn sync_provider_projection_file(path: &Path) -> Result<(), String> {
@@ -6566,12 +6804,13 @@ mod tests {
         std::fs::write(&outside_file, "outside state").unwrap();
         symlink(&outside_file, nested.join("outside-link.json")).unwrap();
 
-        let mut synced = Vec::new();
-        super::sync_provider_projection_tree_with(tree.path(), &mut |path| {
-            synced.push(path.to_path_buf());
+        let synced = std::sync::Mutex::new(Vec::new());
+        super::sync_provider_projection_tree_with(tree.path(), &|path| {
+            synced.lock().unwrap().push(path.to_path_buf());
             Ok(())
         })
         .unwrap();
+        let synced = synced.into_inner().unwrap();
 
         assert!(synced.contains(&file));
         assert!(synced.contains(&nested));
@@ -6582,6 +6821,77 @@ mod tests {
         );
     }
 
+    /// A selected provider environment can hold thousands of files (an
+    /// installed plugin's node_modules). Each flush is one journal commit, so
+    /// flushing them one at a time inside the execution grant made a healthy
+    /// 4,200-file projection take 8-13 s and miss the grant reply. Flushes of
+    /// independent nodes must overlap, and every node is still flushed once.
+    #[cfg(unix)]
+    #[test]
+    fn provider_projection_sync_walk_overlaps_flushes_of_independent_nodes() {
+        use std::collections::BTreeSet;
+        use std::sync::atomic::{AtomicUsize, Ordering};
+
+        let tree = tempfile::tempdir().unwrap();
+        let entries = tree.path().join("node_modules/parity-installed/entries");
+        std::fs::create_dir_all(&entries).unwrap();
+        for index in 0..32 {
+            std::fs::write(
+                entries.join(format!("entry-{index}.cjs")),
+                "module.exports=1\n",
+            )
+            .unwrap();
+        }
+        let in_flight = AtomicUsize::new(0);
+        let peak = AtomicUsize::new(0);
+        let synced = std::sync::Mutex::new(Vec::new());
+        super::sync_provider_projection_tree_with(tree.path(), &|path| {
+            let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
+            peak.fetch_max(now, Ordering::SeqCst);
+            std::thread::sleep(std::time::Duration::from_millis(25));
+            synced.lock().unwrap().push(path.to_path_buf());
+            in_flight.fetch_sub(1, Ordering::SeqCst);
+            Ok(())
+        })
+        .unwrap();
+
+        let synced = synced.into_inner().unwrap();
+        let unique = synced.iter().collect::<BTreeSet<_>>();
+        assert_eq!(synced.len(), 32 + 4, "32 files, 3 directories and the root");
+        assert_eq!(
+            unique.len(),
+            synced.len(),
+            "each node is flushed exactly once"
+        );
+        assert!(
+            peak.load(Ordering::SeqCst) > 1,
+            "independent flushes must overlap instead of running one at a time"
+        );
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn provider_projection_sync_walk_reports_the_node_whose_flush_failed() {
+        let tree = tempfile::tempdir().unwrap();
+        for index in 0..16 {
+            std::fs::write(tree.path().join(format!("file-{index}")), "state").unwrap();
+        }
+        let failing = tree.path().join("file-7");
+        let error = super::sync_provider_projection_tree_with(tree.path(), &|path| {
+            if path == failing {
+                Err(std::io::Error::other("synthetic flush failure"))
+            } else {
+                Ok(())
+            }
+        })
+        .unwrap_err();
+        assert!(
+            error.contains(&failing.display().to_string())
+                && error.contains("synthetic flush failure"),
+            "{error}"
+        );
+    }
+
     fn copy_provider_config_references(
         context: &freshell_runtime_protocol::ProviderLaunchContext,
         provider: &str,
@@ -7180,6 +7490,55 @@ mod tests {
         }
         .expect("owned executable fixture directory")
     }
+
+    /// Run a fixture executable a test has just written, retrying the
+    /// transient ETXTBSY a concurrently forking test thread can cause (see the
+    /// test below). Every other outcome, and a refusal that outlasts the short
+    /// deadline, is returned as is.
+    fn execute_fixture_with_transient_text_file_busy_retry(
+        path: &Path,
+    ) -> std::io::Result<std::process::Output> {
+        let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
+        loop {
+            match std::process::Command::new(path).output() {
+                Err(error)
+                    if error.raw_os_error() == Some(libc::ETXTBSY)
+                        && std::time::Instant::now() < deadline =>
+                {
+                    std::thread::sleep(std::time::Duration::from_millis(5));
+                }
+                result => return result,
+            }
+        }
+    }
+
+    /// Another test thread that forks while a fixture is still open for
+    /// writing hands its child a copy of that write descriptor until the child
+    /// execs, and exec of a file open for writing fails with ETXTBSY. Running
+    /// a fixture must outlast that transient refusal.
+    #[test]
+    fn fixture_execution_outlasts_a_transient_text_file_busy() {
+        use std::os::unix::fs::PermissionsExt;
+        let root = executable_fixture();
+        let script = root.path().join("run");
+        std::fs::write(&script, "#!/bin/sh\nprintf busy-nonce\n").unwrap();
+        std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o700)).unwrap();
+        let writer = std::fs::OpenOptions::new()
+            .write(true)
+            .open(&script)
+            .unwrap();
+        let release = std::thread::spawn(move || {
+            std::thread::sleep(std::time::Duration::from_millis(50));
+            drop(writer);
+        });
+
+        let output = execute_fixture_with_transient_text_file_busy_retry(&script)
+            .expect("the fixture runs once no write descriptor remains");
+
+        release.join().unwrap();
+        assert!(output.status.success(), "{output:?}");
+        assert_eq!(output.stdout, b"busy-nonce");
+    }
     #[test]
     fn provider_config_precreated_home_needs_no_empty_bootstrap_handoff() {
         use super::*;
@@ -7916,7 +8275,7 @@ mod tests {
     #[cfg(unix)]
     #[test]
     fn provider_config_two_stage_script_is_runnable() {
-        use std::{os::unix::fs::PermissionsExt, process::Command};
+        use std::os::unix::fs::PermissionsExt;
         let root = executable_fixture();
         let source = root.path().join("source");
         std::fs::create_dir(&source).unwrap();
@@ -7927,8 +8286,7 @@ mod tests {
         let home = root.path().join("home");
         copy_provider_config_entry(&source, &source, &stage, &mut Vec::new(), &mut 0, 0).unwrap();
         copy_provider_config_entry(&stage, &stage, &home, &mut Vec::new(), &mut 0, 0).unwrap();
-        let output = Command::new(home.join("run"))
-            .output()
+        let output = execute_fixture_with_transient_text_file_busy_retry(&home.join("run"))
             .expect("two-stage executable spawn");
         assert!(output.status.success());
         assert_eq!(output.stdout, b"resource-nonce");
@@ -9371,9 +9729,9 @@ mod tests {
                 ".codex/skills/fixture/scripts/run",
                 ".codex/skills/freshell-discovered-agents/fixture/scripts/run",
             ] {
-                let output = std::process::Command::new(home.join(relative))
-                    .output()
-                    .unwrap();
+                let output =
+                    execute_fixture_with_transient_text_file_busy_retry(&home.join(relative))
+                        .unwrap();
                 assert!(output.status.success(), "{output:?}");
                 assert_eq!(output.stdout, b"resource-nonce\n");
             }
@@ -10366,6 +10724,7 @@ mod tests {
                 terminal: Some(terminal),
                 fresh_agent: None,
                 resume_spec: None,
+                dispatch_budget: None,
             },
         )
         .await;
@@ -10449,6 +10808,7 @@ mod tests {
                         terminal: terminal.clone(),
                         fresh_agent: fresh.clone(),
                         resume_spec: Some(resume.clone()),
+                        dispatch_budget: None,
                     },
                 )
                 .await
@@ -10521,6 +10881,7 @@ mod tests {
                         terminal: Some(terminal.clone()),
                         fresh_agent: None,
                         resume_spec: Some(resume.clone()),
+                        dispatch_budget: None,
                     },
                 )
                 .await
@@ -10581,6 +10942,143 @@ mod tests {
         assert!(state.child.lock().await.is_none());
     }
 
+    fn grant_timing_events(state_dir: &Path) -> Vec<serde_json::Value> {
+        journal_events(state_dir)
+            .into_iter()
+            .filter(|event| event["event"] == "host.grant_timing")
+            .map(|event| event["data"].clone())
+            .collect()
+    }
+
+    /// A refused grant still records how long the host held it and why, so a
+    /// supervisor-side timeout can be told apart from a host-side refusal.
+    #[tokio::test]
+    async fn refused_grant_records_its_timing_and_refusal_code() {
+        let state = test_host_state();
+        {
+            let mut persisted = state.persisted.lock().await;
+            persisted.max_control_epoch = 7;
+        }
+        let grant_id = GrantId::new();
+        let envelope = authenticated_host_envelope(
+            &state,
+            HostCommand::GrantExecution {
+                incarnation_id: state.incarnation_id.clone(),
+                soul_id: SoulId::new(),
+                host_boot_id: state.host_boot_id.clone(),
+                control_epoch: 1,
+                execution_generation: 1,
+                grant_id: grant_id.clone(),
+                fixture: Some(FixtureKind::Heartbeat),
+                provider_path_inputs: Default::default(),
+                terminal: None,
+                fresh_agent: None,
+                resume_spec: None,
+                dispatch_budget_ms: None,
+            },
+        );
+        let error = dispatch(envelope, &state).await.unwrap_err();
+        assert_eq!(error.code, RuntimeErrorCode::StaleExecutionGrant);
+        let timing = grant_timing_events(&state.state_dir);
+        assert_eq!(timing.len(), 1, "{timing:?}");
+        assert_eq!(timing[0]["grantId"], serde_json::json!(grant_id));
+        assert_eq!(timing[0]["outcome"], "refused");
+        assert_eq!(timing[0]["severity"], "warn");
+        assert_eq!(timing[0]["errorCode"], "STALE_EXECUTION_GRANT");
+        assert_eq!(timing[0]["stages"], serde_json::json!([]));
+    }
+
+    /// The supervisor stops waiting shortly after the dispatch budget it sent.
+    /// A host whose budget is already spent must not cross the native dispatch
+    /// boundary: it refuses typed, names where the budget ran out, and leaves
+    /// the grant unconsumed and no worker behind.
+    #[tokio::test]
+    async fn grant_whose_dispatch_budget_is_spent_is_refused_before_consumption() {
+        let state = test_host_state();
+        let grant_id = GrantId::new();
+        let envelope = authenticated_host_envelope(
+            &state,
+            HostCommand::GrantExecution {
+                incarnation_id: state.incarnation_id.clone(),
+                soul_id: SoulId::new(),
+                host_boot_id: state.host_boot_id.clone(),
+                control_epoch: 1,
+                execution_generation: 1,
+                grant_id: grant_id.clone(),
+                fixture: Some(FixtureKind::Heartbeat),
+                provider_path_inputs: Default::default(),
+                terminal: None,
+                fresh_agent: None,
+                resume_spec: None,
+                dispatch_budget_ms: Some(0),
+            },
+        );
+
+        let error = dispatch(envelope, &state).await.unwrap_err();
+
+        assert_eq!(error.code, RuntimeErrorCode::HostUnreachable);
+        assert!(
+            error.message.contains("0ms dispatch budget")
+                && error.message.contains("consume_grant"),
+            "{}",
+            error.message
+        );
+        let persisted = state.persisted.lock().await.clone();
+        assert!(persisted.grant_id.is_none(), "the grant stays replayable");
+        assert_eq!(persisted.worker_launch_count, 0);
+        assert!(state.child.lock().await.is_none());
+        let timing = grant_timing_events(&state.state_dir);
+        assert_eq!(timing.len(), 1, "{timing:?}");
+        assert_eq!(timing[0]["grantId"], serde_json::json!(grant_id));
+        assert_eq!(timing[0]["outcome"], "refused");
+        assert_eq!(timing[0]["errorCode"], "HOST_UNREACHABLE");
+        assert_eq!(timing[0]["budgetMs"], 0);
+        assert_eq!(timing[0]["budgetExhaustedBefore"], "consume_grant");
+    }
+
+    /// An awaited stage (a provider start that keeps retrying, a source
+    /// reader) is cut at the dispatch deadline rather than allowed to run past
+    /// the supervisor's wait, and the refusal names that stage.
+    #[tokio::test]
+    async fn awaited_stage_is_cut_at_the_dispatch_deadline() {
+        let mut timeline = GrantStageTimeline::start(Some(Duration::from_millis(300)));
+        timeline
+            .run("prepare_terminal", async { Ok(()) })
+            .await
+            .expect("a stage inside the budget completes");
+        let started = std::time::Instant::now();
+
+        let error = timeline
+            .run("start_sidecar", async {
+                sleep(Duration::from_secs(30)).await;
+                Ok(())
+            })
+            .await
+            .unwrap_err();
+
+        assert!(
+            started.elapsed() < Duration::from_secs(5),
+            "the stage was cut at the deadline, not allowed to finish"
+        );
+        assert_eq!(error.code, RuntimeErrorCode::HostUnreachable);
+        assert!(
+            error.message.contains("300ms dispatch budget")
+                && error.message.contains("start_sidecar"),
+            "{}",
+            error.message
+        );
+        let event = timeline.event(&GrantId::new(), Err(&error));
+        assert_eq!(event["budgetMs"], 300);
+        assert_eq!(event["budgetExhaustedBefore"], "start_sidecar");
+        let stages = event["stages"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|stage| stage["stage"].as_str().unwrap().to_string())
+            .collect::<Vec<_>>();
+        assert_eq!(stages, ["prepare_terminal"], "a cut stage is not completed");
+    }
+
     #[tokio::test]
     async fn grant_from_another_host_boot_is_rejected_before_consumption() {
         let state = test_host_state();
@@ -10598,6 +11096,7 @@ mod tests {
                 terminal: None,
                 fresh_agent: None,
                 resume_spec: None,
+                dispatch_budget_ms: None,
             },
         );
         let error = dispatch(envelope, &state).await.unwrap_err();
@@ -10780,6 +11279,7 @@ mod tests {
                 terminal: Some(Box::new(terminal)),
                 fresh_agent: None,
                 resume_spec: Some(Box::new(resume_spec)),
+                dispatch_budget_ms: Some(60_000),
             },
         );
 
@@ -10795,6 +11295,33 @@ mod tests {
         ));
         assert_eq!(state.persisted.lock().await.worker_launch_count, 1);
         assert!(state.pty.lock().await.is_some());
+        let timing = grant_timing_events(&state.state_dir);
+        assert_eq!(timing.len(), 1, "one dispatch records one timing event");
+        assert_eq!(timing[0]["outcome"], "accepted");
+        let stages = timing[0]["stages"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|stage| stage["stage"].as_str().unwrap().to_string())
+            .collect::<Vec<_>>();
+        assert_eq!(
+            stages,
+            [
+                "admit",
+                "resolve_secrets",
+                "prepare_provider_inputs",
+                "prepare_provider_state",
+                "prepare_terminal",
+                "verify_claude_sources",
+                "consume_grant",
+                "start_sidecar",
+                "spawn_worker",
+            ],
+            "the accepted terminal dispatch names every stage it spent time in"
+        );
+        assert!(timing[0]["totalMs"].as_u64().is_some());
+        assert_eq!(timing[0]["budgetMs"], 60_000);
+        assert_eq!(timing[0]["budgetExhaustedBefore"], serde_json::Value::Null);
         let mut observed = None;
         for _ in 0..100 {
             if let Ok(value) = std::fs::read(&child_output) {
diff --git a/crates/freshell-supervisor/src/backend/claude_source_observation.rs b/crates/freshell-supervisor/src/backend/claude_source_observation.rs
index f46efee7f..3672cac20 100644
--- a/crates/freshell-supervisor/src/backend/claude_source_observation.rs
+++ b/crates/freshell-supervisor/src/backend/claude_source_observation.rs
@@ -657,7 +657,13 @@ impl DockerEngineBackend {
             )
             .await?;
         if absent.status != 404 {
-            return Err(fail("Claude candidate scratch name is already present"));
+            // Only an answered lookup proves presence; any other refusal
+            // leaves absence unconfirmed. Either way nothing is created.
+            return Err(fail(if absent.status == 200 {
+                "Claude candidate scratch name is already present"
+            } else {
+                "Claude candidate scratch absence could not be confirmed"
+            }));
         }
         // Retain ownership on a lost acknowledgement. A conflict is foreign.
         scratch.attempted = true;
diff --git a/crates/freshell-supervisor/src/claude_source_observation.rs b/crates/freshell-supervisor/src/claude_source_observation.rs
index b431bc06b..716955273 100644
--- a/crates/freshell-supervisor/src/claude_source_observation.rs
+++ b/crates/freshell-supervisor/src/claude_source_observation.rs
@@ -4,23 +4,79 @@ use super::*;
 use crate::backend::ClaudeSourceOwner;
 use freshell_runtime_protocol::{ClaudeSourceAuthority, ObserveClaudeSourcesRequest};
 
+/// A refused or failed observation with the exact check that stopped it. The
+/// web caller receives only the fixed, sanitized `RuntimeError`; the reason is
+/// recorded in the supervisor lifecycle log so a blocked Claude creation or
+/// recovery names its cause.
+struct ObservationFailure {
+    error: RuntimeError,
+    reason: String,
+}
+
+impl From<RuntimeError> for ObservationFailure {
+    fn from(error: RuntimeError) -> Self {
+        Self {
+            reason: error.message.clone(),
+            error,
+        }
+    }
+}
+
 impl Supervisor {
     pub(super) async fn observe_claude_sources(
         &self,
         request: ObserveClaudeSourcesRequest,
     ) -> Result<freshell_platform::claude_inputs::observation::ObservationReply, RuntimeError> {
+        let soul_id = request.soul_id.clone();
+        let provider = request.provider.clone();
+        let candidate = request.candidate.is_some();
+        let authority = match &request.authority {
+            ClaudeSourceAuthority::InitialReservation { .. } => "initial_reservation",
+            ClaudeSourceAuthority::RecoveryAttempt { .. } => "recovery_attempt",
+        };
+        self.observe_claude_sources_checked(request)
+            .await
+            .map_err(|failure| {
+                append_event(
+                    &self.config.lifecycle_log,
+                    "supervisor.claude_source_observation.failed",
+                    serde_json::json!({
+                        "severity": "warn",
+                        "soulId": soul_id,
+                        "provider": provider,
+                        "authority": authority,
+                        "candidate": candidate,
+                        "errorCode": failure.error.code,
+                        "reason": failure.reason,
+                    }),
+                );
+                failure.error
+            })
+    }
+
+    async fn observe_claude_sources_checked(
+        &self,
+        request: ObserveClaudeSourcesRequest,
+    ) -> Result<freshell_platform::claude_inputs::observation::ObservationReply, ObservationFailure>
+    {
         request.validate()?;
-        let denied = || {
-            RuntimeError::new(
+        let denied = |reason: &str| {
+            ObservationFailure {
+            error: RuntimeError::new(
                 RuntimeErrorCode::InvalidRequest,
                 "Claude source observation requires its current creation or recovery attempt; retain inputs and reprepare",
-            )
+            ),
+            reason: reason.to_string(),
+        }
         };
-        let unavailable = || {
-            RuntimeError::new(
+        let unavailable = |reason: String| {
+            ObservationFailure {
+            error: RuntimeError::new(
                 RuntimeErrorCode::HostUnreachable,
                 "Claude native source evidence is unavailable; retain inputs and repair the source before dispatch",
-            )
+            ),
+            reason,
+        }
         };
         let mut recovery = None;
         let owner = match &request.authority {
@@ -30,22 +86,26 @@ impl Supervisor {
                     .initial_input_reservation(request.soul_id.clone(), reservation_id.clone())
                     .await
                     .map_err(map_registry)?
-                    .ok_or_else(denied)?;
+                    .ok_or_else(|| denied("initial input reservation is not recorded"))?;
                 if reservation.provider != request.provider
                     || reservation.image_ref != self.config.image_ref
                     || reservation.binary != self.config.host_binary_path
                 {
-                    return Err(denied());
+                    return Err(denied(
+                        "initial input reservation belongs to another provider or artifact",
+                    ));
                 }
                 let observation = tokio::time::timeout(
                     Duration::from_secs(10),
                     self.backend.observe_initial_auth_volume(&reservation),
                 )
                 .await
-                .map_err(|_| unavailable())?
-                .map_err(|_| unavailable())?;
+                .map_err(|_| unavailable("initial auth volume observation timed out".into()))?
+                .map_err(|error| {
+                    unavailable(format!("initial auth volume observation failed: {error}"))
+                })?;
                 if observation != expected {
-                    return Err(denied());
+                    return Err(denied("initial auth volume observation changed"));
                 }
                 ClaudeSourceOwner::Initial {
                     reservation,
@@ -64,7 +124,7 @@ impl Supervisor {
                         .get(attempt_id)
                         .filter(|authority| authority.expires > Instant::now())
                         .cloned()
-                        .ok_or_else(denied)?
+                        .ok_or_else(|| denied("recovery attempt scope is absent or expired"))?
                 };
                 if authority.start.context.soul_id != request.soul_id
                     || authority.start.context.provider != request.provider
@@ -77,7 +137,7 @@ impl Supervisor {
                         .await
                         .map_err(map_registry)?
                 {
-                    return Err(denied());
+                    return Err(denied("recovery attempt is no longer current"));
                 }
                 let handle = authority.start.context.prior_handle.clone();
                 let recovery_source = if request.candidate.is_some() {
@@ -89,7 +149,7 @@ impl Supervisor {
                                 .fresh_agent()
                                 .and_then(|spec| spec.provider_launch_context.as_ref())
                         })
-                        .ok_or_else(denied)?;
+                        .ok_or_else(|| denied("retained owner has no provider launch context"))?;
                     Some(self.claude_recovery_source_binding(
                         &authority.start,
                         attempt_id,
@@ -109,24 +169,26 @@ impl Supervisor {
             &request.observation_request,
             freshell_platform::claude_inputs::observation::MAX_REQUEST_BYTES,
         )
-        .map_err(|_| denied())?;
+        .map_err(|_| denied("observation request exceeds its metadata profile"))?;
         let digest = format!("{:x}", Sha256::digest(&raw));
         // The backend owns a 25-second operation and five-second finite cleanup.
         // This outer deadline permits that cleanup to finish before returning.
         let reply = tokio::time::timeout(
-            Duration::from_secs(35),
+            CLAUDE_SOURCE_OBSERVATION_BUDGET,
             self.backend
                 .observe_claude_sources(&owner, &request, &self.config.test_run_id),
         )
         .await
-        .map_err(|_| unavailable())?
-        .map_err(|_| unavailable())?;
+        .map_err(|_| unavailable("source reader timed out".into()))?
+        .map_err(|error| unavailable(format!("source reader failed: {error}")))?;
         reply
             .validate_for_request(&request.observation_request, &digest)
-            .map_err(|_| unavailable())?;
+            .map_err(|_| unavailable("source reader reply does not answer this request".into()))?;
         let observed = &reply.observation;
         if (observed.uid, observed.euid, observed.gid, observed.egid) != (65534, 65534, 0, 0) {
-            return Err(unavailable());
+            return Err(unavailable(
+                "source reader ran under another identity".into(),
+            ));
         }
         match &request.authority {
             ClaudeSourceAuthority::InitialReservation { reservation_id } => {
@@ -135,21 +197,25 @@ impl Supervisor {
                     .initial_input_reservation(request.soul_id.clone(), reservation_id.clone())
                     .await
                     .map_err(map_registry)?
-                    .ok_or_else(denied)?;
+                    .ok_or_else(|| {
+                        denied("initial input reservation disappeared during observation")
+                    })?;
                 let ClaudeSourceOwner::Initial {
                     reservation,
                     observation,
                 } = &owner
                 else {
-                    return Err(denied());
+                    return Err(denied("observation owner changed kind"));
                 };
                 if current.0.binary_sha256 != reservation.binary_sha256 || &current.1 != observation
                 {
-                    return Err(denied());
+                    return Err(denied(
+                        "initial input reservation changed during observation",
+                    ));
                 }
             }
             ClaudeSourceAuthority::RecoveryAttempt { attempt_id } => {
-                let start = recovery.ok_or_else(denied)?;
+                let start = recovery.ok_or_else(|| denied("observation owner changed kind"))?;
                 let live_scope = {
                     let authorities = self
                         .auth_lookup_authorities
@@ -168,7 +234,7 @@ impl Supervisor {
                         .await
                         .map_err(map_registry)?
                 {
-                    return Err(denied());
+                    return Err(denied("recovery attempt ended during observation"));
                 }
             }
         }
@@ -395,7 +461,7 @@ mod tests {
             .unwrap();
             let lock = supervisor.lifecycle_lock(&soul).await;
             let _held = lock.lock().await;
-            let scope = supervisor.auth_lookup_scope(&start);
+            let scope = supervisor.auth_lookup_scope(&start, CLAUDE_RECOVERY_CALLBACK_BUDGET);
             if matches!(
                 mode,
                 "dropped" | "expired" | "candidate-dropped" | "candidate-expired"
@@ -477,8 +543,32 @@ mod tests {
                 matches!(mode, "live" | "candidate"),
                 "{mode}: changed scoped authority after IPC cannot qualify source evidence: {result:?}"
             );
+            let failures = std::fs::read_to_string(owned.path().join("lifecycle.jsonl"))
+                .unwrap_or_default()
+                .lines()
+                .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
+                .filter(|event| event["event"] == "supervisor.claude_source_observation.failed")
+                .map(|event| event["data"].clone())
+                .collect::<Vec<_>>();
             if let Err(error) = result {
                 assert_eq!(error.code, RuntimeErrorCode::InvalidRequest);
+                // The web caller sees only the fixed refusal; the lifecycle log
+                // names the exact check that stopped the observation.
+                let expected_reason = if mode == "candidate-unrecorded-prior" {
+                    "retained owner has no provider launch context"
+                } else {
+                    "recovery attempt ended during observation"
+                };
+                assert_eq!(failures.len(), 1, "{mode}: {failures:?}");
+                assert_eq!(failures[0]["severity"], "warn");
+                assert_eq!(failures[0]["soulId"], serde_json::json!(soul));
+                assert_eq!(failures[0]["authority"], "recovery_attempt");
+                assert_eq!(failures[0]["candidate"], mode.starts_with("candidate"));
+                assert_eq!(failures[0]["errorCode"], "INVALID_REQUEST");
+                assert_eq!(failures[0]["reason"], expected_reason, "{mode}");
+                assert!(!error.message.contains(expected_reason));
+            } else {
+                assert!(failures.is_empty(), "{mode}: {failures:?}");
             }
             assert!(std::fs::read(&prior).unwrap() == b"immutable accepted prior generation");
         }
diff --git a/crates/freshell-supervisor/src/native_auth_lookup_tests.rs b/crates/freshell-supervisor/src/native_auth_lookup_tests.rs
index a44453d71..851522421 100644
--- a/crates/freshell-supervisor/src/native_auth_lookup_tests.rs
+++ b/crates/freshell-supervisor/src/native_auth_lookup_tests.rs
@@ -548,7 +548,7 @@ async fn provider_native_auth_nested_callback_uses_current_attempt_without_reloc
     .unwrap();
     let lock = supervisor.lifecycle_lock(&soul).await;
     let _held = lock.lock().await;
-    let scope = supervisor.auth_lookup_scope(&start);
+    let scope = supervisor.auth_lookup_scope(&start, PROVIDER_CALLBACK_BUDGET);
     let key = scope.key.clone();
     let request = |soul: SoulId, key: String| {
         let mut envelope = Envelope::new(
@@ -734,7 +734,7 @@ async fn provider_native_auth_nested_callback_uses_current_attempt_without_reloc
         .await
         .result
         .is_err());
-    let scope = supervisor.auth_lookup_scope(&start);
+    let scope = supervisor.auth_lookup_scope(&start, PROVIDER_CALLBACK_BUDGET);
     supervisor
         .registry
         .begin_stop_expected(soul.clone(), None)
diff --git a/crates/freshell-supervisor/src/provider_callback_budget_tests.rs b/crates/freshell-supervisor/src/provider_callback_budget_tests.rs
new file mode 100644
index 000000000..8179ab2cf
--- /dev/null
+++ b/crates/freshell-supervisor/src/provider_callback_budget_tests.rs
@@ -0,0 +1,235 @@
+//! How long the supervisor waits for a provider-preparation callback to the
+//! web server, and how long the recovery scope that callback carries lives.
+
+use super::*;
+use crate::registry::*;
+use freshell_runtime_protocol::*;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+struct RecoveryFixture {
+    _root: tempfile::TempDir,
+    supervisor: Supervisor,
+    soul: SoulId,
+    start: RecoveryStart,
+    context: ProviderLaunchContext,
+}
+
+/// A retained Claude terminal with one accepted prepared-input generation,
+/// and a recovery attempt that begins from it.
+async fn claude_recovery(budgets: ProviderCallbackBudgets) -> RecoveryFixture {
+    let root = tempfile::tempdir().unwrap();
+    let binary = root.path().join("host");
+    std::fs::write(&binary, b"synthetic helper artifact").unwrap();
+    let registry = Registry::open(root.path().join("registry"), None).unwrap();
+    let soul = SoulId::new();
+    let generation = format!("grant-{}", uuid::Uuid::new_v4());
+    let context: ProviderLaunchContext = serde_json::from_value(serde_json::json!({
+        "preparation":{"claude":{"mcp_args":[]}},"config":[],
+        "mcpCapability":{"grantId":generation,"endpoint":"http://host.docker.internal:4000","providerRelativePath":".freshell/mcp-capability.json"},
+        "preparedInputs":{"generationId":generation,"manifest":{"relativePath":".freshell/input-manifest.json","sha256":"a".repeat(64),"sourceCount":0,"aliasCount":0,"recipeCount":0}}
+    }))
+    .unwrap();
+    let terminal: TerminalLaunchSpec = serde_json::from_value(serde_json::json!({
+        "terminalId":"callback-budget","streamId":"callback-budget-stream","mode":"claude","program":"claude","args":[],
+        "cwd":root.path(),"workspacePath":root.path(),"runAsUid":65534,"runAsGid":0,"cols":80,"rows":24,
+        "projectKey":"callback-budget-workspace","providerLaunchContext":&context
+    }))
+    .unwrap();
+    let prepared = registry
+        .prepare_launch(LaunchPreparation {
+            soul_id: soul.clone(),
+            provider: "claude".into(),
+            provider_store_id: "synthetic-store".into(),
+            native_session_id: None,
+            creation_seed_ref: "synthetic-seed".into(),
+            request_id: RequestId::new(),
+            payload_digest: "synthetic-digest".into(),
+            requested_limits: RuntimeLimits {
+                cpu_milli: 500,
+                memory_bytes: 64 * 1024 * 1024,
+                swap_bytes: 0,
+                pids_max: 32,
+            },
+            profile: RuntimeProfile::Custom,
+            project_key: "callback-budget-workspace".into(),
+            fixture: None,
+            terminal: Some(terminal),
+            fresh_agent: None,
+            view_intent: None,
+            admission: crate::admission::AdmissionPolicy::default(),
+        })
+        .await
+        .unwrap();
+    let image = format!("sha256:{}", "a".repeat(64));
+    registry
+        .commit_created(
+            prepared.incarnation_id,
+            BackendCreatedRecord {
+                daemon_id: DockerDaemonId::new(),
+                container_id: "callback-budget-container".into(),
+                image_ref: image.clone(),
+                runtime_dir: root.path().into(),
+                host_binary_path: binary.clone(),
+                immutable_config_digest: "synthetic-digest".into(),
+            },
+        )
+        .await
+        .unwrap();
+    let start = registry
+        .begin_recovery(
+            soul.clone(),
+            RecoveryTrigger::ManualRetry,
+            RecoveryPath::NativeResume,
+        )
+        .await
+        .unwrap();
+    let mut supervisor = Supervisor::new(
+        registry,
+        default_backend(root.path().join("absent-docker.sock")),
+        SupervisorConfig {
+            runtime_root: root.path().join("runtime"),
+            control_socket_path: root.path().join("control.sock"),
+            host_binary_path: binary,
+            image_ref: image,
+            test_run_id: "provider-callback-budget".into(),
+            control_secret: "synthetic-control-secret".into(),
+            lifecycle_log: root.path().join("lifecycle.jsonl"),
+            admission: crate::admission::AdmissionPolicy::default(),
+        },
+    )
+    .unwrap();
+    supervisor.provider_callback_budgets = budgets;
+    RecoveryFixture {
+        _root: root,
+        supervisor,
+        soul,
+        start,
+        context,
+    }
+}
+
+/// A synthetic web server callback socket: it reads one callback, holds it
+/// for `delay`, runs `inspect` on the request and the instant it arrived,
+/// then answers `reply`.
+fn slow_callback_server(
+    supervisor: &Supervisor,
+    delay: Duration,
+    reply: &'static [u8],
+    inspect: impl FnOnce(&serde_json::Value, Instant) + Send + 'static,
+) -> tokio::task::JoinHandle<()> {
+    let listener = tokio::net::UnixListener::bind(
+        supervisor
+            .config
+            .control_socket_path
+            .with_file_name("mcp.sock"),
+    )
+    .unwrap();
+    tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.unwrap();
+        let received = Instant::now();
+        let mut bytes = Vec::new();
+        stream.read_to_end(&mut bytes).await.unwrap();
+        tokio::time::sleep(delay).await;
+        inspect(&serde_json::from_slice(&bytes).unwrap(), received);
+        let _ = stream.write_all(reply).await;
+    })
+}
+
+fn budgets(ordinary_ms: u64, claude_recovery_ms: u64) -> ProviderCallbackBudgets {
+    ProviderCallbackBudgets {
+        ordinary: Duration::from_millis(ordinary_ms),
+        claude_recovery: Duration::from_millis(claude_recovery_ms),
+    }
+}
+
+/// The Claude recovery callback makes two source observations back to back
+/// through this supervisor before it answers. It may take longer than an
+/// ordinary callback, and the recovery scope those observations present must
+/// still be live when they finish.
+#[tokio::test]
+async fn claude_recovery_callback_outlasts_an_ordinary_callback_with_a_live_scope() {
+    // Longer than the old fixed 30-second scope; the test never waits it out.
+    let claude_wait = Duration::from_secs(40);
+    let fixture = claude_recovery(budgets(200, claude_wait.as_millis() as u64)).await;
+    let lock = fixture.supervisor.lifecycle_lock(&fixture.soul).await;
+    let _held = lock.lock().await;
+    let authorities = fixture.supervisor.auth_lookup_authorities.clone();
+    let server = slow_callback_server(
+        &fixture.supervisor,
+        Duration::from_millis(600),
+        b"not a prepared context",
+        move |request, received| {
+            assert!(
+                request["claudeRecoverySource"].is_object(),
+                "the callback carries the Claude recovery source binding"
+            );
+            let key = request["authLookupAuthority"].as_str().unwrap();
+            let expires = authorities
+                .lock()
+                .unwrap()
+                .get(key)
+                .map(|authority| authority.expires)
+                .expect("the recovery scope is still registered");
+            // The scope is created just before the callback starts, so it
+            // may end a moment before the wait does, never sooner.
+            assert!(
+                expires + Duration::from_secs(1) >= received + claude_wait,
+                "the scope lives as long as the supervisor waits for this callback"
+            );
+        },
+    );
+
+    let error = fixture
+        .supervisor
+        .issue_replacement_mcp_capability(
+            &fixture.start,
+            "claude",
+            &fixture.context,
+            &[],
+            "/fixture/instance",
+            "/fixture/workspace",
+        )
+        .await
+        .unwrap_err();
+
+    server.await.expect("the server saw a live scope");
+    assert!(
+        error.message.contains("invalid scoped grant reference"),
+        "the callback was answered, not timed out: {}",
+        error.message
+    );
+}
+
+/// Every other callback keeps the ordinary wait.
+#[tokio::test]
+async fn ordinary_callback_keeps_the_ordinary_wait() {
+    let fixture = claude_recovery(budgets(200, 5_000)).await;
+    let server = slow_callback_server(
+        &fixture.supervisor,
+        Duration::from_millis(600),
+        b"ok",
+        |_, _| {},
+    );
+
+    let error = fixture
+        .supervisor
+        .notify_mcp_capability(
+            "activate",
+            "http://host.docker.internal:4000",
+            &fixture.soul,
+            &IncarnationId::new(),
+            "grant-callback-budget",
+        )
+        .await
+        .unwrap_err();
+
+    assert_eq!(error.code, RuntimeErrorCode::HostUnreachable);
+    assert!(
+        error
+            .message
+            .contains("server callback timed out after 200ms"),
+        "{}",
+        error.message
+    );
+    server.abort();
+}
diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs
index 98309481c..03a17f45e 100644
--- a/crates/freshell-supervisor/src/service.rs
+++ b/crates/freshell-supervisor/src/service.rs
@@ -84,6 +84,9 @@ mod claude_source_observation;
 #[path = "native_auth_lookup_tests.rs"]
 mod native_auth_lookup_tests;
 #[cfg(test)]
+#[path = "provider_callback_budget_tests.rs"]
+mod provider_callback_budget_tests;
+#[cfg(test)]
 #[path = "provider_preparation_tests.rs"]
 mod provider_preparation_tests;
 impl Drop for AuthLookupScope {
@@ -117,6 +120,9 @@ pub struct Supervisor {
     pub(crate) live_resume_evidence_attempts: Arc<Mutex<HashMap<LiveResumeEvidenceKey, Instant>>>,
     auth_lookup_authorities: Arc<StdMutex<HashMap<String, AuthLookupAuthority>>>,
     provider_path_inputs: Arc<StdMutex<ProviderPathInputBindings>>,
+    /// Reply budgets for supervisor→session-host commands, read once.
+    pub(crate) host_command_budgets: HostCommandBudgets,
+    pub(crate) provider_callback_budgets: ProviderCallbackBudgets,
 }
 
 impl Supervisor {
@@ -166,6 +172,8 @@ impl Supervisor {
             live_resume_evidence_attempts: Arc::new(Mutex::new(HashMap::new())),
             auth_lookup_authorities: Arc::new(StdMutex::new(HashMap::new())),
             provider_path_inputs: Arc::new(StdMutex::new(HashMap::new())),
+            host_command_budgets: HostCommandBudgets::from_env(),
+            provider_callback_budgets: ProviderCallbackBudgets::default(),
         })
     }
 
@@ -679,7 +687,14 @@ impl Supervisor {
         }
     }
 
-    fn auth_lookup_scope(&self, start: &crate::registry::RecoveryStart) -> AuthLookupScope {
+    /// Register a recovery scope for nested lookups made inside one
+    /// provider-preparation callback. It lives as long as the supervisor waits
+    /// for that callback (`lifetime`) and is removed when the callback ends.
+    fn auth_lookup_scope(
+        &self,
+        start: &crate::registry::RecoveryStart,
+        lifetime: Duration,
+    ) -> AuthLookupScope {
         let key = uuid::Uuid::new_v4().to_string();
         let mut authorities = self
             .auth_lookup_authorities
@@ -690,7 +705,7 @@ impl Supervisor {
             key.clone(),
             AuthLookupAuthority {
                 start: start.clone(),
-                expires: Instant::now() + Duration::from_secs(30),
+                expires: Instant::now() + lifetime,
             },
         );
         AuthLookupScope {
@@ -2512,10 +2527,13 @@ impl Supervisor {
             }
             _ => None,
         };
-        let scope = self.auth_lookup_scope(start);
-        let claude_recovery_source = if provider == "claude"
-            && matches!(start.context.provider.as_str(), "claude" | "kilroy")
-        {
+        let claude_recovery =
+            provider == "claude" && matches!(start.context.provider.as_str(), "claude" | "kilroy");
+        let scope = self.auth_lookup_scope(
+            start,
+            self.provider_callback_budgets.for_callback(claude_recovery),
+        );
+        let claude_recovery_source = if claude_recovery {
             Some(self.claude_recovery_source_binding(start, &scope.key, context)?)
         } else {
             None
@@ -2602,12 +2620,18 @@ impl Supervisor {
             }
             Ok::<_, std::io::Error>(response)
         };
-        let response = tokio::time::timeout(Duration::from_secs(35), exchange)
+        let budget = self
+            .provider_callback_budgets
+            .for_callback(request.claude_recovery_source.is_some());
+        let response = tokio::time::timeout(budget, exchange)
             .await
             .map_err(|_| {
                 RuntimeError::new(
                     RuntimeErrorCode::HostUnreachable,
-                    "capability_pending: server callback timed out",
+                    format!(
+                        "capability_pending: server callback timed out after {}ms",
+                        budget.as_millis()
+                    ),
                 )
             })?
             .map_err(|error| {
@@ -2795,7 +2819,7 @@ impl Supervisor {
         // yet. Once connected, the reply itself is bounded by the same budget
         // every other host command uses: a host that accepts and then goes
         // silent is HostUnreachable, never an unbounded wait.
-        let budget = host_command_budget(std::env::var(HOST_COMMAND_TIMEOUT_ENV).ok().as_deref());
+        let budget = self.host_command_budgets.control;
         loop {
             match UnixStream::connect(&socket).await {
                 Ok(mut stream) => {
@@ -2884,6 +2908,7 @@ impl Supervisor {
             terminal: workload.terminal.clone().map(Box::new),
             fresh_agent: workload.fresh_agent.clone().map(Box::new),
             resume_spec: workload.resume_spec.clone().map(Box::new),
+            dispatch_budget_ms: None,
         };
         match self
             .send_authenticated_host_command(incarnation_id, handle.runtime_dir(), host, command)
@@ -2982,7 +3007,8 @@ impl Supervisor {
         host: &AuthenticatedHost,
         command: HostCommand,
     ) -> Result<HostResult, RuntimeError> {
-        let is_slow_fresh_agent_command = uses_fresh_agent_command_budget(&command);
+        let budget = self.host_command_budgets.for_command(&command);
+        let command = with_host_dispatch_budget(command, budget);
         let mut envelope = Envelope::new(RequestId::new(), ControlRole::Supervisor, command);
         envelope.auth = Some(host_proof(
             &host.secret,
@@ -2990,15 +3016,6 @@ impl Supervisor {
             &host.host_boot_id,
             &incarnation_id,
         ));
-        let budget = if is_slow_fresh_agent_command {
-            fresh_agent_command_budget(
-                std::env::var(FRESH_AGENT_COMMAND_TIMEOUT_ENV)
-                    .ok()
-                    .as_deref(),
-            )
-        } else {
-            host_command_budget(std::env::var(HOST_COMMAND_TIMEOUT_ENV).ok().as_deref())
-        };
         let reply = request_host_reply(&runtime_dir.join("host.sock"), &envelope, budget).await?;
         reply.result
     }
@@ -3178,6 +3195,26 @@ pub(crate) const DEFAULT_FRESH_AGENT_COMMAND_TIMEOUT_MS: u64 = 60_000;
 pub(crate) const FRESH_AGENT_COMMAND_TIMEOUT_ENV: &str =
     "FRESHELL_RUNTIME_FRESH_AGENT_COMMAND_TIMEOUT_MS";
 
+/// Default wait for a terminal execution grant's reply.
+///
+/// A terminal grant is not a control round trip: it is the native dispatch
+/// boundary. Before the host can reply it projects the selected provider
+/// inputs and starts the provider, and some of those steps have long bounds of
+/// their own. One Codex app-server start may take up to 45 s
+/// (`freshell_codex::launch_lifecycle::SIDECAR_START_BUDGET`) and a terminal
+/// start retries it up to five times (about 251 s worst case); an exact
+/// resume first runs the provider-owned config refresh (25 s, plus up to 32 s
+/// of projection recovery). No fixed wait covers every such sum, so this is a
+/// policy bound on one dispatch, roughly ten times a healthy one (4-6 s
+/// measured with a 4,200-file selected plugin). The host is told its share of
+/// it with the grant ([`host_dispatch_budget`]) and stops there itself,
+/// refusing typed and naming the stage it was in, so the supervisor never
+/// abandons a dispatch the host is still working on; the recovery policy then
+/// decides whether a fresh incarnation tries again.
+pub(crate) const DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS: u64 = 60_000;
+pub(crate) const TERMINAL_DISPATCH_TIMEOUT_ENV: &str =
+    "FRESHELL_RUNTIME_TERMINAL_DISPATCH_TIMEOUT_MS";
+
 /// Resolve the round-trip budget. Absent, unparseable, or zero configuration
 /// falls back to the default: the failure direction is always a bounded wait,
 /// never an unbounded one.
@@ -3199,6 +3236,133 @@ pub(crate) fn fresh_agent_command_budget(configured: Option<&str>) -> Duration {
         ))
 }
 
+pub(crate) fn terminal_dispatch_budget(configured: Option<&str>) -> Duration {
+    configured
+        .and_then(|value| value.trim().parse::<u64>().ok())
+        .filter(|ms| *ms > 0)
+        .map(Duration::from_millis)
+        .unwrap_or(Duration::from_millis(DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS))
+}
+
+/// The part of an execution grant's wait the supervisor reserves for the
+/// reply itself: the frame round trip, plus a synchronous host stage (the
+/// provider projection, the provider-owned refresh) that cannot be cut at the
+/// deadline and is only noticed when it returns.
+pub(crate) const GRANT_REPLY_ALLOWANCE: Duration = Duration::from_secs(5);
+
+/// How long the host may spend dispatching a grant the supervisor waits
+/// `wait` for: the wait less [`GRANT_REPLY_ALLOWANCE`], and never less than
+/// half of it, so a short configured wait still leaves the host time to work.
+pub(crate) fn host_dispatch_budget(wait: Duration) -> Duration {
+    wait.saturating_sub(GRANT_REPLY_ALLOWANCE).max(wait / 2)
+}
+
+/// Tell the host the dispatch budget of a grant from the wait that bounds its
+/// reply, so the host refuses typed before the supervisor gives up on it.
+fn with_host_dispatch_budget(mut command: HostCommand, wait: Duration) -> HostCommand {
+    if let HostCommand::GrantExecution {
+        dispatch_budget_ms, ..
+    } = &mut command
+    {
+        *dispatch_budget_ms =
+            Some(u64::try_from(host_dispatch_budget(wait).as_millis()).unwrap_or(u64::MAX));
+    }
+    command
+}
+
+/// How long the supervisor waits for a session host's reply, by what
+/// answering the command makes the host do. Resolved once from the
+/// environment when the supervisor starts.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct HostCommandBudgets {
+    /// One control round trip: hello, status, stop, terminal I/O, probes.
+    pub(crate) control: Duration,
+    /// A terminal execution grant: provider preparation plus native dispatch.
+    pub(crate) terminal_dispatch: Duration,
+    /// A fresh-agent execution grant or turn operation.
+    pub(crate) fresh_agent: Duration,
+}
+
+impl HostCommandBudgets {
+    pub(crate) fn from_env() -> Self {
+        Self::from_lookup(|name| std::env::var(name).ok())
+    }
+
+    pub(crate) fn from_lookup(lookup: impl Fn(&str) -> Option<String>) -> Self {
+        Self {
+            control: host_command_budget(lookup(HOST_COMMAND_TIMEOUT_ENV).as_deref()),
+            terminal_dispatch: terminal_dispatch_budget(
+                lookup(TERMINAL_DISPATCH_TIMEOUT_ENV).as_deref(),
+            ),
+            fresh_agent: fresh_agent_command_budget(
+                lookup(FRESH_AGENT_COMMAND_TIMEOUT_ENV).as_deref(),
+            ),
+        }
+    }
+
+    pub(crate) fn for_command(&self, command: &HostCommand) -> Duration {
+        if uses_fresh_agent_command_budget(command) {
+            self.fresh_agent
+        } else if matches!(
+            command,
+            HostCommand::GrantExecution {
+                terminal: Some(_),
+                ..
+            }
+        ) {
+            self.terminal_dispatch
+        } else {
+            self.control
+        }
+    }
+}
+
+/// Bound on one Claude source observation: the fixed reader's 25-second
+/// operation and 5-second owned cleanup, with slack for the reply.
+pub(crate) const CLAUDE_SOURCE_OBSERVATION_BUDGET: Duration = Duration::from_secs(35);
+
+/// Wait for one provider-preparation callback to the web server, covering the
+/// web server's own work for it (selecting and copying the provider's inputs).
+pub(crate) const PROVIDER_CALLBACK_BUDGET: Duration = Duration::from_secs(35);
+
+/// Wait for a Claude recovery's `issue` callback. Before it answers, the web
+/// server observes the current Claude sources and then the candidate
+/// projection, back to back, each through this supervisor and each bounded by
+/// [`CLAUDE_SOURCE_OBSERVATION_BUDGET`]. The callback's wait covers its own
+/// work plus both observations, so an observation that finishes inside its
+/// own bound is never cut off by the callback around it.
+pub(crate) const CLAUDE_RECOVERY_CALLBACK_BUDGET: Duration = Duration::from_secs(
+    PROVIDER_CALLBACK_BUDGET.as_secs() + 2 * CLAUDE_SOURCE_OBSERVATION_BUDGET.as_secs(),
+);
+
+/// How long the supervisor waits for a provider-preparation callback. The
+/// recovery scope a callback carries lives exactly as long, so a nested
+/// lookup made inside the callback's wait never finds its scope expired.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct ProviderCallbackBudgets {
+    pub(crate) ordinary: Duration,
+    pub(crate) claude_recovery: Duration,
+}
+
+impl Default for ProviderCallbackBudgets {
+    fn default() -> Self {
+        Self {
+            ordinary: PROVIDER_CALLBACK_BUDGET,
+            claude_recovery: CLAUDE_RECOVERY_CALLBACK_BUDGET,
+        }
+    }
+}
+
+impl ProviderCallbackBudgets {
+    pub(crate) fn for_callback(&self, claude_recovery: bool) -> Duration {
+        if claude_recovery {
+            self.claude_recovery
+        } else {
+            self.ordinary
+        }
+    }
+}
+
 /// One bounded framed request/reply against a session host's control socket.
 /// Connect, write, and read all share the single budget so a host that accepts
 /// the connection and then goes silent cannot extend the wait.
@@ -3414,7 +3578,8 @@ mod host_actor_storage_tests {
 mod host_ipc_timeout_tests {
     use super::{
         fresh_agent_command_budget, host_command_budget, request_host_reply,
-        DEFAULT_FRESH_AGENT_COMMAND_TIMEOUT_MS, DEFAULT_HOST_COMMAND_TIMEOUT_MS,
+        terminal_dispatch_budget, DEFAULT_FRESH_AGENT_COMMAND_TIMEOUT_MS,
+        DEFAULT_HOST_COMMAND_TIMEOUT_MS, DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS,
     };
     use freshell_runtime_protocol::{
         ControlRole, Envelope, HostCommand, IncarnationId, RequestId, RuntimeErrorCode,
@@ -3538,6 +3703,354 @@ mod host_ipc_timeout_tests {
             fresh_agent_command_budget(Some("750")),
             Duration::from_millis(750)
         );
+        assert_eq!(
+            terminal_dispatch_budget(None),
+            Duration::from_millis(DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS)
+        );
+        assert_eq!(
+            terminal_dispatch_budget(Some("0")),
+            Duration::from_millis(DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS)
+        );
+        assert_eq!(
+            terminal_dispatch_budget(Some("900")),
+            Duration::from_millis(900)
+        );
+    }
+}
+
+#[cfg(test)]
+mod host_command_budget_tests {
+    use super::{
+        default_backend, host_dispatch_budget, AuthenticatedHost, HostCommandBudgets, Supervisor,
+        SupervisorConfig, FRESH_AGENT_COMMAND_TIMEOUT_ENV, HOST_COMMAND_TIMEOUT_ENV,
+        TERMINAL_DISPATCH_TIMEOUT_ENV,
+    };
+    use crate::registry::Registry;
+    use freshell_runtime_protocol::{
+        read_frame, write_frame, Envelope, GrantId, HostBootId, HostCommand, HostReply, HostResult,
+        IncarnationId, RequestId, RuntimeErrorCode, RuntimeLimits, SoulId, TerminalLaunchSpec,
+    };
+    use std::path::Path;
+    use std::sync::{Arc, Mutex as StdMutex};
+    use std::time::{Duration, Instant};
+    use tokio::net::UnixListener;
+
+    fn limits() -> RuntimeLimits {
+        RuntimeLimits {
+            cpu_milli: 500,
+            memory_bytes: 64 * 1024 * 1024,
+            swap_bytes: 0,
+            pids_max: 32,
+        }
+    }
+
+    /// Distinct budgets per command class, so a test can tell which one a
+    /// command was held to.
+    fn budgets(
+        control_ms: &'static str,
+        terminal_ms: &'static str,
+        fresh_ms: &'static str,
+    ) -> HostCommandBudgets {
+        HostCommandBudgets::from_lookup(move |name| {
+            match name {
+                HOST_COMMAND_TIMEOUT_ENV => Some(control_ms),
+                TERMINAL_DISPATCH_TIMEOUT_ENV => Some(terminal_ms),
+                FRESH_AGENT_COMMAND_TIMEOUT_ENV => Some(fresh_ms),
+                _ => None,
+            }
+            .map(str::to_string)
+        })
+    }
+
+    fn supervisor(root: &Path, budgets: HostCommandBudgets) -> Supervisor {
+        let registry = Registry::open(root.join("registry"), None).unwrap();
+        let host_binary_path = root.join("synthetic-host");
+        std::fs::write(&host_binary_path, "synthetic-host-artifact").unwrap();
+        let mut supervisor = Supervisor::new(
+            registry,
+            default_backend(root.join("absent-docker.sock")),
+            SupervisorConfig {
+                runtime_root: root.join("runtime"),
+                control_socket_path: root.join("control.sock"),
+                host_binary_path,
+                image_ref: format!("sha256:{}", "a".repeat(64)),
+                test_run_id: "host-command-budgets".into(),
+                control_secret: "synthetic-control-secret".into(),
+                lifecycle_log: root.join("lifecycle.jsonl"),
+                admission: crate::admission::AdmissionPolicy::default(),
+            },
+        )
+        .unwrap();
+        supervisor.host_command_budgets = budgets;
+        supervisor
+    }
+
+    fn terminal_grant(incarnation_id: &IncarnationId, root: &Path) -> HostCommand {
+        let terminal: TerminalLaunchSpec = serde_json::from_value(serde_json::json!({
+            "terminalId":"terminal-dispatch-budget","streamId":"stream-dispatch-budget",
+            "mode":"codex","program":"codex","args":[],"env":{"HOME":"/home/freshell/provider"},
+            "cwd":root,"runAsUid":65534,"runAsGid":0,"cols":80,"rows":24,
+            "projectKey":"project-dispatch-budget","workspacePath":root,
+        }))
+        .unwrap();
+        HostCommand::GrantExecution {
+            incarnation_id: incarnation_id.clone(),
+            soul_id: SoulId::new(),
+            host_boot_id: HostBootId::new(),
+            control_epoch: 1,
+            execution_generation: 1,
+            grant_id: GrantId::new(),
+            fixture: None,
+            provider_path_inputs: Default::default(),
+            terminal: Some(Box::new(terminal)),
+            fresh_agent: None,
+            resume_spec: None,
+            dispatch_budget_ms: None,
+        }
+    }
+
+    type Received = Arc<StdMutex<Vec<HostCommand>>>;
+
+    /// A synthetic host that answers every command after `delay` (or never),
+    /// recording each command it received.
+    fn slow_host(
+        runtime_dir: &Path,
+        delay: Option<Duration>,
+    ) -> (tokio::task::JoinHandle<()>, Received) {
+        let listener = UnixListener::bind(runtime_dir.join("host.sock")).unwrap();
+        let received = Received::default();
+        let record = received.clone();
+        let task = tokio::spawn(async move {
+            loop {
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let record = record.clone();
+                tokio::spawn(async move {
+                    let envelope: Envelope<HostCommand> = read_frame(&mut stream).await.unwrap();
+                    record.lock().unwrap().push(envelope.body.clone());
+                    let Some(delay) = delay else {
+                        tokio::time::sleep(Duration::from_secs(3_600)).await;
+                        return;
+                    };
+                    tokio::time::sleep(delay).await;
+                    let result = match envelope.body {
+                        HostCommand::GrantExecution { host_boot_id, .. } => {
+                            HostResult::GrantAccepted {
+                                host_boot_id,
+                                worker_pid: 4321,
+                                worker_launch_count: 1,
+                                fixture_evidence: serde_json::json!({}),
+                                native_session_id: None,
+                            }
+                        }
+                        _ => HostResult::Stopped,
+                    };
+                    let _ = write_frame(
+                        &mut stream,
+                        &HostReply {
+                            request_id: envelope.request_id,
+                            result: Ok(result),
+                        },
+                    )
+                    .await;
+                });
+            }
+        });
+        (task, received)
+    }
+
+    fn authenticated() -> AuthenticatedHost {
+        AuthenticatedHost {
+            host_boot_id: HostBootId::new(),
+            secret: b"synthetic-host-secret".to_vec(),
+            effective_limits: limits(),
+        }
+    }
+
+    /// A terminal grant is the native dispatch boundary: the host prepares
+    /// provider inputs and starts the provider (Codex's app-server alone may
+    /// take up to its 45-second start budget) before it can reply. The
+    /// supervisor must not abandon that dispatch at the short control
+    /// round-trip bound, while ordinary control commands keep that bound.
+    #[tokio::test]
+    async fn terminal_grant_waits_for_dispatch_while_control_commands_stay_short() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_dir = root.path().join("incarnation");
+        std::fs::create_dir_all(&runtime_dir).unwrap();
+        let supervisor = supervisor(root.path(), budgets("200", "5000", "9000"));
+        let (host, _) = slow_host(&runtime_dir, Some(Duration::from_millis(700)));
+        let incarnation_id = IncarnationId::new();
+
+        let granted = supervisor
+            .send_authenticated_host_command(
+                incarnation_id.clone(),
+                &runtime_dir,
+                &authenticated(),
+                terminal_grant(&incarnation_id, root.path()),
+            )
+            .await;
+        assert!(
+            matches!(granted, Ok(HostResult::GrantAccepted { .. })),
+            "a terminal dispatch slower than one control round trip must be accepted: {granted:?}"
+        );
+
+        let status = supervisor
+            .send_authenticated_host_command(
+                incarnation_id.clone(),
+                &runtime_dir,
+                &authenticated(),
+                HostCommand::Status {
+                    incarnation_id: incarnation_id.clone(),
+                },
+            )
+            .await
+            .expect_err("a control round trip keeps its short bound");
+        assert_eq!(status.code, RuntimeErrorCode::HostUnreachable);
+        assert!(
+            status.message.contains("timed out after 200ms"),
+            "{}",
+            status.message
+        );
+        host.abort();
+    }
+
+    /// The longer dispatch bound is still a bound: a host that accepts a
+    /// terminal grant and goes silent cannot hold the activation open.
+    #[tokio::test]
+    async fn silent_host_cannot_hold_a_terminal_grant_past_its_dispatch_budget() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_dir = root.path().join("incarnation");
+        std::fs::create_dir_all(&runtime_dir).unwrap();
+        let supervisor = supervisor(root.path(), budgets("100", "400", "9000"));
+        let (host, _) = slow_host(&runtime_dir, None);
+        let incarnation_id = IncarnationId::new();
+        let started = Instant::now();
+        let error = supervisor
+            .send_authenticated_host_command(
+                incarnation_id.clone(),
+                &runtime_dir,
+                &authenticated(),
+                terminal_grant(&incarnation_id, root.path()),
+            )
+            .await
+            .expect_err("a silent host must not hold a terminal grant forever");
+        assert_eq!(error.code, RuntimeErrorCode::HostUnreachable);
+        assert!(
+            error.message.contains("timed out after 400ms"),
+            "{}",
+            error.message
+        );
+        assert!(started.elapsed() < Duration::from_secs(5));
+        host.abort();
+    }
+
+    /// Terminal grants, fresh-agent operations, and control round trips are
+    /// each held to their own configured budget: none borrows another's.
+    #[tokio::test]
+    async fn each_command_class_is_held_to_its_own_budget() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_dir = root.path().join("incarnation");
+        std::fs::create_dir_all(&runtime_dir).unwrap();
+        let supervisor = supervisor(root.path(), budgets("100", "300", "700"));
+        let (host, _) = slow_host(&runtime_dir, None);
+        let incarnation_id = IncarnationId::new();
+        let host_auth = authenticated();
+        let send = |command| {
+            supervisor.send_authenticated_host_command(
+                incarnation_id.clone(),
+                &runtime_dir,
+                &host_auth,
+                command,
+            )
+        };
+
+        let terminal = send(terminal_grant(&incarnation_id, root.path()))
+            .await
+            .expect_err("a silent host times out");
+        let fresh_agent = send(HostCommand::FreshAgentSend {
+            incarnation_id: incarnation_id.clone(),
+            request_id: RequestId::new(),
+            text: "synthetic turn".into(),
+            settings: None,
+            checkpoint_logical_attempt_id: None,
+        })
+        .await
+        .expect_err("a silent host times out");
+        let control = send(HostCommand::Status {
+            incarnation_id: incarnation_id.clone(),
+        })
+        .await
+        .expect_err("a silent host times out");
+
+        for (error, budget) in [(&terminal, 300), (&fresh_agent, 700), (&control, 100)] {
+            assert_eq!(error.code, RuntimeErrorCode::HostUnreachable);
+            assert!(
+                error
+                    .message
+                    .contains(&format!("timed out after {budget}ms")),
+                "{}",
+                error.message
+            );
+        }
+        host.abort();
+    }
+
+    /// A grant tells the host how long it has to dispatch, derived from the
+    /// wait that bounds its reply, so the host refuses before the supervisor
+    /// gives up on it.
+    #[tokio::test]
+    async fn terminal_grant_tells_the_host_its_dispatch_budget() {
+        let root = tempfile::tempdir().unwrap();
+        let runtime_dir = root.path().join("incarnation");
+        std::fs::create_dir_all(&runtime_dir).unwrap();
+        let supervisor = supervisor(root.path(), budgets("100", "300", "700"));
+        let (host, received) = slow_host(&runtime_dir, Some(Duration::from_millis(10)));
+        let incarnation_id = IncarnationId::new();
+
+        supervisor
+            .send_authenticated_host_command(
+                incarnation_id.clone(),
+                &runtime_dir,
+                &authenticated(),
+                terminal_grant(&incarnation_id, root.path()),
+            )
+            .await
+            .expect("the grant is accepted");
+
+        let granted_budget = received
+            .lock()
+            .unwrap()
+            .iter()
+            .find_map(|command| match command {
+                HostCommand::GrantExecution {
+                    dispatch_budget_ms, ..
+                } => Some(*dispatch_budget_ms),
+                _ => None,
+            })
+            .expect("the host received the grant");
+        assert_eq!(
+            granted_budget,
+            Some(150),
+            "the host is told the terminal wait less its reply allowance"
+        );
+        host.abort();
+    }
+
+    /// The host's dispatch budget leaves the supervisor a reply allowance of
+    /// five seconds, never more than half of the wait.
+    #[test]
+    fn host_dispatch_budget_leaves_a_reply_allowance() {
+        assert_eq!(
+            host_dispatch_budget(Duration::from_secs(60)),
+            Duration::from_secs(55)
+        );
+        assert_eq!(
+            host_dispatch_budget(Duration::from_secs(10)),
+            Duration::from_secs(5)
+        );
+        assert_eq!(
+            host_dispatch_budget(Duration::from_millis(400)),
+            Duration::from_millis(200)
+        );
     }
 }
 
diff --git a/docs/development/managed-runtime.md b/docs/development/managed-runtime.md
index eb8692fc8..0815e5656 100644
--- a/docs/development/managed-runtime.md
+++ b/docs/development/managed-runtime.md
@@ -127,6 +127,55 @@ handshake included — is bounded by
 typed `HostUnreachable`, after which the caller proceeds through the
 Docker-level path it already owns.
 
+An execution grant is not a control round trip, so it has its own bound. The
+host answers a grant only after it has prepared the selected provider inputs
+and started the provider. The supervisor waits
+`FRESHELL_RUNTIME_TERMINAL_DISPATCH_TIMEOUT_MS` for a terminal grant and
+`FRESHELL_RUNTIME_FRESH_AGENT_COMMAND_TIMEOUT_MS` for a fresh-agent grant or
+turn operation (both default 60s). That is a policy bound on one dispatch,
+roughly ten times a healthy one, not the sum of the host's inner limits: one
+Codex app-server start may take up to 45s and a terminal start retries it up
+to five times (about 251s worst case), and an exact resume first runs the
+provider-owned config refresh (25s, plus up to 32s of recovery).
+
+So the grant carries its own dispatch budget (`dispatch_budget_ms`): the
+supervisor's wait less a five-second reply allowance, never less than half the
+wait (55s by default). The host stops there itself. Every awaited dispatch
+stage runs inside the budget and is cut at the deadline; the native dispatch
+boundary is never crossed once the budget is spent, so the grant stays
+unconsumed; and the refusal is a typed `HostUnreachable` that names the stage
+the budget ran out before. A synchronous stage (the provider projection, the
+provider-owned refresh) cannot be cut; the host notices when it returns, and
+the reply allowance covers that overrun. The supervisor therefore never
+abandons a dispatch the host is still working on, and its recovery policy
+decides whether a fresh incarnation tries again.
+
+Every grant records a `host.grant_timing` event (in the incarnation's
+`host-lifecycle.jsonl` and on the host's stderr) naming how long each dispatch
+stage took, the budget it had (`budgetMs`) and, when it ran out, the stage it
+ran out before (`budgetExhaustedBefore`), for accepted and refused grants
+alike.
+
+Recovery also calls back to the web server to prepare the replacement's
+provider inputs. That callback waits 35s for the web server's own work; a
+Claude recovery's callback waits 105s, because before it answers the web
+server observes the current Claude sources and then the candidate projection,
+back to back, through the supervisor, and each observation is bounded at 35s
+(the fixed reader's 25s operation plus 5s cleanup). The recovery scope those
+nested lookups present lives exactly as long as the callback wait, so an
+observation that finishes inside its own bound never finds its scope expired.
+
+A launch or recovery holds the soul's lifecycle lock for its whole
+activation, callback and grant included, and `stop` takes the same lock
+before it commits the stop intent. A stop issued during activation therefore
+waits for the activation to settle — up to the dispatch wait (60s by default)
+when the host is slow or silent, plus the callback wait during a recovery.
+
+The provider projection inside a grant makes every staged file durable before
+it is published. Those flushes overlap (`PROVIDER_PROJECTION_SYNC_WORKERS`) so
+the filesystem journal commits them together; one at a time, a selected
+plugin with a 4,200-file `node_modules` spent 8-13s of the grant flushing.
+
 This matters most for `stop`: the durable stop intent commits first, then the
 graceful host stop is *attempted*. A host that accepts the control connection
 and then goes silent used to hold the authoritative stop open forever. It is
diff --git a/scripts/testing/runtime-test-broker.ts b/scripts/testing/runtime-test-broker.ts
index 41996af57..0ae038c99 100644
--- a/scripts/testing/runtime-test-broker.ts
+++ b/scripts/testing/runtime-test-broker.ts
@@ -495,7 +495,7 @@ export class RestrictedDockerBroker {
     }
     if (method === 'GET' && candidateVolume && !candidate
       && /^freshell-claude-candidate-[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/.test(candidateVolume)) {
-      const rejection = this.initialClaudePublicationRejection(candidateVolume.slice('freshell-claude-candidate-'.length))
+      const rejection = this.claudePublicationRejection(candidateVolume.slice('freshell-claude-candidate-'.length))
       if (rejection) {
         this.block(response, method, url, 403, `candidate absence qualification unavailable at ${rejection}`, false)
         return
@@ -1224,53 +1224,104 @@ export class RestrictedDockerBroker {
   // stages exist before Docker receives their first request. Register that
   // immutable handoff here; the ordinary Docker validators still decide every
   // request. Explicit native fixture handoffs continue to use the same APIs.
-  private initialClaudePublicationRejection(requestId: string): string | undefined {
+  private claudePublicationRejection(requestId: string): string | undefined {
     let phase = 'physical_stage'
     try {
       const root = path.dirname(this.policy.runtimeRootPrefix)
       for (const scenario of this.fixtureDirectories(path.join(root, 'scenarios'), 32)) {
         const control = path.join(root, 'scenarios', scenario, 'control')
         for (const grant of this.fixtureDirectories(path.join(control, 'provider-roots'), 128)) {
-          if (!/^grant-[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/.test(grant)) continue
-          const stageRoot = path.join(control, 'provider-roots', grant)
-          const bindingFile = path.join(stageRoot, '.freshell/claude-candidate-binding.json')
-          if (!fs.existsSync(bindingFile)) continue
-          phase = 'owner_capability'
-          const binding = this.fixtureMetadata(bindingFile).value
-          const capability = this.fixtureMetadata(path.join(control, 'mcp-capabilities', `${grant}.json`), 128 * 1024).value
-          const owner = [...this.initialProviderVolumes.values()].find(owner => owner.soulId === binding.soulId && owner.soulId === capability.soulId)
-          if (!owner || capability.incarnationId != null || binding.version !== 1 || binding.provider !== 'claude'
-            || binding.generationId !== grant || binding.intentRevision !== 1 || binding.registryAttemptId != null
-            || !/^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/.test(binding.initialReservationId ?? '')) continue
-          phase = 'prepared_metadata'
-          const context = this.fixtureMetadata(path.join(stageRoot, '.freshell/prepared-context.json'))
-          const result = this.fixtureMetadata(path.join(stageRoot, '.freshell/prepared-result.json'))
-          const inputs = context.value.preparedInputs
-          if (!context.value.preparation?.claude || !inputs || inputs.generationId !== grant
-            || context.value.mcpCapability?.grantId !== grant || inputs.initialAuthReservation !== binding.initialReservationId
-            || JSON.stringify(result.value.providerLaunchContext) !== JSON.stringify(context.value)
-            || !Array.isArray(result.value.providerSecretReferences) || inputs.manifest?.relativePath !== '.freshell/input-manifest.json'
-            || sha256Hex(context.bytes) !== binding.preparedContextSha256 || sha256Hex(result.bytes) !== binding.preparedResultSha256) continue
-          phase = 'authority_binding'
-          if (sha256Hex(Buffer.from(JSON.stringify({ kind: 'initial_reservation', reservationId: binding.initialReservationId }))) !== binding.authoritySha256) continue
-          phase = 'manifest_artifact'
-          const manifest = this.fixtureMetadata(path.join(stageRoot, inputs.manifest.relativePath), 64 * 1024 * 1024)
-          const binary = fs.lstatSync(owner.hostBinaryPath)
-          if (sha256Hex(manifest.bytes) !== binding.manifestSha256 || inputs.manifest.sha256 !== binding.manifestSha256
-            || !binary.isFile() || binary.isSymbolicLink() || binary.size > 128 * 1024 * 1024
-            || sha256Hex(fs.readFileSync(owner.hostBinaryPath)) !== owner.binarySha256) continue
-          phase = 'selected_target'
-          this.validateClaudeCandidateAuthorization({ ...owner, generationId: grant, stageRoot,
-            requestId, reservationId: binding.initialReservationId, workspacePath: binding.workspacePath, cwd: binding.cwd,
-            gitCommonDir: binding.gitCommonDir === null ? undefined : binding.gitCommonDir, manifestRelativePath: inputs.manifest.relativePath,
-            configurationBinds: inputs.configBinds ?? [], resourceMounts: inputs.resourceMounts ?? [] })
-          return undefined
+          if (!FIXTURE_GRANT.test(grant)) continue
+          if (!fs.existsSync(path.join(control, 'provider-roots', grant, '.freshell/claude-candidate-binding.json'))) continue
+          const qualified = this.qualifyPublishedClaudeCandidate(control, grant, requestId)
+          if ('authorization' in qualified) return undefined
+          phase = qualified.phase
         }
       }
     } catch { /* Unqualified publication exposes no daemon metadata. */ }
     return phase
   }
 
+  /**
+   * Qualify one controller-published Claude candidate stage against its owned
+   * soul and return its exact candidate authorization. A creation candidate is
+   * bound to its initial reservation. A recovery candidate is bound to its
+   * registry attempt and to the frozen prior generation it retains, with the
+   * same profile the closed launch fixture hands off explicitly.
+   */
+  private qualifyPublishedClaudeCandidate(control: string, grant: string, requestId: string):
+    { authorization: ClaudeCandidateAuthorization } | { phase: string } {
+    let phase = 'owner_capability'
+    try {
+      const stageRoot = path.join(control, 'provider-roots', grant)
+      const binding = this.fixtureMetadata(path.join(stageRoot, '.freshell/claude-candidate-binding.json')).value
+      const capability = this.fixtureMetadata(path.join(control, 'mcp-capabilities', `${grant}.json`), 128 * 1024).value
+      const owner = [...this.initialProviderVolumes.values()].find(owner => owner.soulId === binding.soulId && owner.soulId === capability.soulId)
+      const recovery = binding.registryAttemptId != null
+      const prior = binding.priorGenerationId
+      if (!owner || capability.incarnationId != null || binding.version !== 1 || binding.provider !== 'claude'
+        || binding.generationId !== grant || !Number.isSafeInteger(binding.intentRevision) || binding.intentRevision < 1
+        || (recovery
+          ? binding.initialReservationId != null || typeof binding.registryAttemptId !== 'string'
+            || !/^recovery-/.test(binding.registryAttemptId) || !FIXTURE_UUID.test(binding.registryAttemptId.slice('recovery-'.length))
+            || typeof prior !== 'string' || !FIXTURE_GRANT.test(prior) || prior === grant
+            // The controller issues a replacement as a pending preparation
+            // that names exactly the generation it replaces.
+            || capability.preparation?.phase !== 'pending' || !Array.isArray(capability.preparation?.priorGrantIds)
+            || !capability.preparation.priorGrantIds.includes(prior)
+          : binding.intentRevision !== 1 || prior != null || !FIXTURE_UUID.test(binding.initialReservationId ?? ''))) return { phase }
+      phase = 'prepared_metadata'
+      const context = this.fixtureMetadata(path.join(stageRoot, '.freshell/prepared-context.json'))
+      const result = this.fixtureMetadata(path.join(stageRoot, '.freshell/prepared-result.json'))
+      const inputs = context.value.preparedInputs
+      if (!context.value.preparation?.claude || !inputs || inputs.generationId !== grant
+        || context.value.mcpCapability?.grantId !== grant
+        || (inputs.initialAuthReservation ?? null) !== (binding.initialReservationId ?? null)
+        || JSON.stringify(result.value.providerLaunchContext) !== JSON.stringify(context.value)
+        || !Array.isArray(result.value.providerSecretReferences) || inputs.manifest?.relativePath !== '.freshell/input-manifest.json'
+        || sha256Hex(context.bytes) !== binding.preparedContextSha256 || sha256Hex(result.bytes) !== binding.preparedResultSha256) return { phase }
+      phase = 'authority_binding'
+      // A creation authority is its public reservation. A recovery authority is
+      // the supervisor's private callback scope, which the supervisor itself
+      // requalifies against this publication; here only its digest shape is
+      // observable.
+      if (recovery
+        ? !/^[a-f0-9]{64}$/.test(binding.authoritySha256 ?? '')
+        : sha256Hex(Buffer.from(JSON.stringify({ kind: 'initial_reservation', reservationId: binding.initialReservationId }))) !== binding.authoritySha256) return { phase }
+      let reservationId: string = binding.initialReservationId
+      let retained: ClaudeCandidateAuthorization['retained']
+      if (recovery) {
+        phase = 'retained_generation'
+        const priorRoot = path.join(control, 'provider-roots', prior)
+        const priorContext = this.fixtureMetadata(path.join(priorRoot, '.freshell/prepared-context.json')).value
+        const priorInputs = priorContext.preparedInputs
+        // Original creation metadata is carried only by the frozen prior
+        // generation; the recovery authority remains its registry attempt.
+        if (!priorContext.preparation?.claude || !priorInputs || priorInputs.generationId !== prior
+          || priorContext.mcpCapability?.grantId !== prior || !FIXTURE_UUID.test(priorInputs.initialAuthReservation ?? '')) return { phase }
+        reservationId = priorInputs.initialAuthReservation
+        retained = { attemptId: binding.registryAttemptId, generationId: prior, stageRoot: priorRoot,
+          configurationBinds: priorInputs.configBinds ?? [], resourceMounts: priorInputs.resourceMounts ?? [] }
+      }
+      phase = 'manifest_artifact'
+      const manifest = this.fixtureMetadata(path.join(stageRoot, inputs.manifest.relativePath), 64 * 1024 * 1024)
+      const binary = fs.lstatSync(owner.hostBinaryPath)
+      if (sha256Hex(manifest.bytes) !== binding.manifestSha256 || inputs.manifest.sha256 !== binding.manifestSha256
+        || !binary.isFile() || binary.isSymbolicLink() || binary.size > 128 * 1024 * 1024
+        || sha256Hex(fs.readFileSync(owner.hostBinaryPath)) !== owner.binarySha256) return { phase }
+      phase = 'selected_target'
+      const authorization: ClaudeCandidateAuthorization = { ...owner, generationId: grant, stageRoot,
+        requestId, reservationId, workspacePath: binding.workspacePath, cwd: binding.cwd,
+        gitCommonDir: binding.gitCommonDir === null ? undefined : binding.gitCommonDir, manifestRelativePath: inputs.manifest.relativePath,
+        configurationBinds: inputs.configBinds ?? [], resourceMounts: inputs.resourceMounts ?? [],
+        ...(retained ? { retained } : {}) }
+      this.validateClaudeCandidateAuthorization(authorization)
+      return { authorization }
+    } catch {
+      return { phase }
+    }
+  }
+
   private async registerPublishedFixtureInputs(body: Buffer, url: string): Promise<void> {
     let phase = 'owner'
     try {
@@ -1350,21 +1401,11 @@ export class RestrictedDockerBroker {
             || sha256Hex(fs.readFileSync(owner.hostBinaryPath)) !== owner.binarySha256) continue
           if (candidate) {
             phase = 'candidate_binding'
-            const binding = this.fixtureMetadata(path.join(stageRoot, '.freshell/claude-candidate-binding.json')).value
-            if (binding.version !== 1 || binding.provider !== 'claude' || binding.soulId !== owner.soulId
-              || binding.generationId !== grant || binding.intentRevision !== 1 || binding.registryAttemptId != null
-              || binding.initialReservationId !== inputs.initialAuthReservation
-              || binding.initialReservationId !== labels['com.freshell.source-reservation-id']
-              || binding.manifestSha256 !== inputs.manifest.sha256
-              || binding.preparedContextSha256 !== sha256Hex(contextDocument.bytes)
-              || binding.preparedResultSha256 !== sha256Hex(resultDocument.bytes)) continue
+            const qualified = this.qualifyPublishedClaudeCandidate(control, grant, labels['com.freshell.source-request-id'])
+            if (!('authorization' in qualified) || qualified.authorization.soulId !== owner.soulId
+              || !equalRecord(labels, candidateLabels(qualified.authorization, this.policy.testRunId))) continue
             phase = 'candidate_authorization'
-            this.authorizeClaudeCandidate({ ...owner, generationId: grant, stageRoot,
-              reservationId: binding.initialReservationId, requestId: labels['com.freshell.source-request-id'],
-              workspacePath: binding.workspacePath, cwd: binding.cwd, gitCommonDir: binding.gitCommonDir === null ? undefined : binding.gitCommonDir,
-              manifestRelativePath: inputs.manifest.relativePath, configurationBinds: inputs.configBinds ?? [],
-              resourceMounts: inputs.resourceMounts ?? [],
-            })
+            this.authorizeClaudeCandidate(qualified.authorization)
           } else {
             if (initial ? inputs.initialAuthReservation !== reservation!.reservationId
               : inputs.initialAuthReservation != null) continue
@@ -1668,6 +1709,9 @@ function sha256(bytes: Buffer): string {
   return `sha256:${sha256Hex(bytes)}`
 }
 
+const FIXTURE_UUID = /^[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/
+const FIXTURE_GRANT = /^grant-[a-f0-9]{8}-[a-f0-9]{4}-4[a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$/
+
 function sha256Hex(bytes: Buffer): string {
   return createHash('sha256').update(bytes).digest('hex')
 }
diff --git a/test/unit/tooling/testing/runtime-test-broker.test.ts b/test/unit/tooling/testing/runtime-test-broker.test.ts
index 74396f18f..02ef70177 100644
--- a/test/unit/tooling/testing/runtime-test-broker.test.ts
+++ b/test/unit/tooling/testing/runtime-test-broker.test.ts
@@ -1120,7 +1120,7 @@ function publisherBody(input:PreparedInputAuthorization):any {
 
 async function publishControllerPreparation(directory: string, input: Pick<PreparedInputAuthorization,
   'soulId' | 'generationId' | 'stageRoot' | 'configurationBinds' | 'resourceMounts' | 'indexedFiles'>,
-  reservationId: string, provider: 'claude' | 'codex') {
+  reservationId: string | null, provider: 'claude' | 'codex') {
   const metadata = path.join(input.stageRoot, '.freshell')
   await fs.mkdir(metadata, { recursive: true })
   const manifest = Buffer.from(JSON.stringify({ version: 1, sources: [], aliases: [], recipes: [] }))
@@ -1324,6 +1324,77 @@ describe('fixed indexed publisher broker',()=>{
       expect(forwarded.every(operation => operation.method === 'GET')).toBe(true)
     }, undefined, true)
   })
+  it('probes candidate absence for a recovery publication only against its frozen retained generation', async () => {
+    await withHttpBroker(async ({ directory, request, forwarded, broker: instance }) => {
+      const prior = candidateAuthorization()
+      const binary = Buffer.from('recovery candidate host artifact')
+      prior.hostBinaryPath = path.join(directory, 'freshell-session-host')
+      await fs.writeFile(prior.hostBinaryPath, binary)
+      prior.binarySha256 = createHash('sha256').update(binary).digest('hex')
+      prior.stageRoot = path.join(directory, 'scenarios/scenario/control/provider-roots', prior.generationId)
+      instance.authorizeInitialProviderVolume({ ...sourceAuthorization(), hostBinaryPath: prior.hostBinaryPath, binarySha256: prior.binarySha256 })
+      await publishControllerPreparation(directory, prior, prior.reservationId, 'claude')
+      const capabilities = path.join(directory, 'scenarios/scenario/control/mcp-capabilities')
+      await fs.writeFile(path.join(capabilities, `${prior.generationId}.json`), JSON.stringify({
+        endpoint: 'http://127.0.0.1:1/api/managed-mcp', scope: 'synthetic-controller-scope', soulId: prior.soulId,
+        incarnationId: 'incarnation-initial', expiresAt: Date.now() + 60_000, preparation: { phase: 'committed', priorGrantIds: [] },
+      }))
+      const attemptId = 'recovery-77777777-7777-4777-8777-777777777777'
+      const candidate: ClaudeCandidateAuthorization = { ...prior,
+        generationId: 'grant-55555555-5555-4555-8555-555555555555',
+        requestId: '66666666-6666-4666-8666-666666666666',
+        retained: { attemptId, generationId: prior.generationId, stageRoot: prior.stageRoot,
+          configurationBinds: prior.configurationBinds, resourceMounts: prior.resourceMounts } }
+      candidate.stageRoot = path.join(directory, 'scenarios/scenario/control/provider-roots', candidate.generationId)
+      const { metadata, context, contextBytes, resultBytes } = await publishControllerPreparation(directory, candidate, null, 'claude')
+      const pending = { endpoint: 'http://127.0.0.1:1/api/managed-mcp', scope: 'synthetic-controller-scope', soulId: candidate.soulId,
+        incarnationId: null, expiresAt: Date.now() + 60_000, preparation: { phase: 'pending', priorGrantIds: [prior.generationId] } }
+      const capability = path.join(capabilities, `${candidate.generationId}.json`)
+      await fs.writeFile(capability, JSON.stringify(pending))
+      const binding = { version: 1, provider: 'claude', soulId: candidate.soulId, generationId: candidate.generationId,
+        initialReservationId: null, intentRevision: 1, registryAttemptId: attemptId, priorGenerationId: prior.generationId,
+        authoritySha256: 'b'.repeat(64), workspacePath: candidate.workspacePath, cwd: candidate.cwd, gitCommonDir: null,
+        manifestSha256: context.preparedInputs.manifest.sha256,
+        preparedContextSha256: createHash('sha256').update(contextBytes).digest('hex'),
+        preparedResultSha256: createHash('sha256').update(resultBytes).digest('hex') }
+      const publication = path.join(metadata, 'claude-candidate-binding.json')
+      const url = `/v1.47/volumes/freshell-claude-candidate-${candidate.requestId}`
+      const refused = async (mutated: Record<string, unknown>, phase: string) => {
+        await fs.writeFile(publication, JSON.stringify(mutated))
+        expect((await request('GET', url)).status).toBe(403)
+        expect(instance.eventsSnapshot().at(-1)).toMatchObject({
+          reason: `candidate absence qualification unavailable at ${phase}`,
+        })
+      }
+      await refused({ ...binding, registryAttemptId: 'recovery-not-a-uuid' }, 'owner_capability')
+      await refused({ ...binding, priorGenerationId: candidate.generationId }, 'owner_capability')
+      await refused({ ...binding, initialReservationId: prior.reservationId }, 'owner_capability')
+      await fs.writeFile(capability, JSON.stringify({ ...pending, preparation: { phase: 'pending', priorGrantIds: [] } }))
+      await refused(binding, 'owner_capability')
+      await fs.writeFile(capability, JSON.stringify({ ...pending, incarnationId: 'incarnation-replacement' }))
+      await refused(binding, 'owner_capability')
+      await fs.writeFile(capability, JSON.stringify(pending))
+      await refused({ ...binding, preparedResultSha256: 'f'.repeat(64) }, 'prepared_metadata')
+      await refused({ ...binding, authoritySha256: 'not-a-digest' }, 'authority_binding')
+      const priorContext = path.join(prior.stageRoot, '.freshell/prepared-context.json')
+      const priorBytes = await fs.readFile(priorContext)
+      const unrecorded = JSON.parse(priorBytes.toString())
+      unrecorded.preparedInputs.initialAuthReservation = null
+      await fs.writeFile(priorContext, JSON.stringify(unrecorded))
+      await refused(binding, 'retained_generation')
+      await fs.writeFile(priorContext, priorBytes)
+      expect(forwarded).toHaveLength(0)
+
+      await fs.writeFile(publication, JSON.stringify(binding))
+      expect((await request('GET', url)).status).toBe(404)
+      const create = { Name: `freshell-claude-candidate-${candidate.requestId}`, Driver: 'local', DriverOpts: {},
+        Labels: candidateLabels(candidate) }
+      expect((await request('POST', '/v1.47/volumes/create', create)).status).toBe(201)
+      expect(forwarded.map(operation => `${operation.method} ${operation.url}`)).toEqual([
+        `GET ${url}`, 'POST /v1.47/volumes/create',
+      ])
+    }, undefined, true)
+  })
   it('cleans the exact registered failed-launch volume and retains changed or attached ownership', async () => {
     await withHttpBroker(async ({ directory }) => {
       const binary = Buffer.from('registered failed-launch artifact')
@@ -1379,7 +1450,8 @@ describe('fixed indexed publisher broker',()=>{
       const { metadata, context, contextBytes, resultBytes } = await publishControllerPreparation(directory, input, input.reservationId, 'claude')
       await fs.writeFile(path.join(metadata, 'claude-candidate-binding.json'), JSON.stringify({
         version: 1, provider: 'claude', soulId: input.soulId, generationId: input.generationId,
-        initialReservationId: input.reservationId, intentRevision: 1,
+        initialReservationId: input.reservationId, intentRevision: 1, registryAttemptId: null,
+        authoritySha256: createHash('sha256').update(JSON.stringify({ kind: 'initial_reservation', reservationId: input.reservationId })).digest('hex'),
         workspacePath: input.workspacePath, cwd: input.cwd, gitCommonDir: null,
         manifestSha256: context.preparedInputs.manifest.sha256,
         preparedContextSha256: createHash('sha256').update(contextBytes).digest('hex'),