From 807feef3500527cd966213fdb7bcb0bed4251675 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:53:27 -0700 Subject: [PATCH 01/10] feat(session-host): record per-stage timing for every execution grant A managed terminal or fresh-agent dispatch that outran the supervisor's reply bound left no trace of where the time went once its container was gone. Every grant now records a host.grant_timing event (severity, grant id, outcome, refusal code, total and per-stage milliseconds) in the incarnation's host-lifecycle.jsonl and on the host's stderr, for accepted and refused grants alike. This is what showed that a 4,200-file selected provider environment spent 10.5-12.8 s of a grant inside prepare_provider_state while every other stage took milliseconds. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-session-host/src/main.rs | 141 +++++++++++++++++++++++ 1 file changed, 141 insertions(+) diff --git a/crates/freshell-session-host/src/main.rs b/crates/freshell-session-host/src/main.rs index 6fd13bb1f..405dc29c7 100644 --- a/crates/freshell-session-host/src/main.rs +++ b/crates/freshell-session-host/src/main.rs @@ -891,9 +891,66 @@ struct GrantExecutionRequest { resume_spec: Option, } +/// Wall-clock breakdown of one execution grant. The supervisor bounds the +/// whole grant reply; this 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)>, +} + +impl GrantStageTimeline { + fn start() -> Self { + let now = std::time::Instant::now(); + Self { + started: now, + stage_started: now, + stages: Vec::new(), + } + } + + 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(), + "stages": self + .stages + .iter() + .map(|(stage, ms)| serde_json::json!({"stage": stage, "ms": ms})) + .collect::>(), + }) + } +} + async fn grant_execution( state: &Arc, request: GrantExecutionRequest, +) -> Result { + let grant_id = request.grant_id.clone(); + let mut timeline = GrantStageTimeline::start(); + 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, + request: GrantExecutionRequest, + timeline: &mut GrantStageTimeline, ) -> Result { let GrantExecutionRequest { soul_id, @@ -959,6 +1016,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. @@ -994,6 +1052,7 @@ 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, @@ -1018,6 +1077,7 @@ async fn grant_execution( ) })?; secrets.environment.extend(generated); + timeline.completed("prepare_provider_inputs"); } else if !provider_path_inputs.is_empty() { return Err(RuntimeError::new( RuntimeErrorCode::InvalidRequest, @@ -1094,6 +1154,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 { @@ -1152,6 +1213,7 @@ async fn grant_execution( ) .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?; } + timeline.completed("prepare_provider_state"); let prepared = providers::prepare_terminal( terminal, &resolved_secrets.environment, @@ -1165,6 +1227,7 @@ async fn grant_execution( format!("prepare managed provider: {error}"), ) })?; + timeline.completed("prepare_terminal"); claude_source_observation::verify_final( prepared.terminal.provider_launch_context.as_ref(), &state @@ -1178,10 +1241,13 @@ async fn grant_execution( freshell_platform::claude_inputs::NativeSourceRuntime::Terminal, ) .await?; + timeline.completed("verify_claude_sources"); consume_grant(&mut persisted)?; + timeline.completed("consume_grant"); let prepared = providers::start_terminal_sidecar(prepared) .await .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?; + timeline.completed("start_sidecar"); let terminal = prepared.terminal; let hosted = HostedPty::spawn( &state.state_dir, @@ -1198,6 +1264,7 @@ async fn grant_execution( format!("spawn managed terminal: {e}"), ) })?; + timeline.completed("spawn_worker"); let pid = hosted.pid(); let evidence = serde_json::json!({ "workload":"terminal", @@ -1310,6 +1377,7 @@ async fn grant_execution( ) .await .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?; + timeline.completed("prepare_provider_state"); claude_source_observation::verify_final( launch.provider_launch_context.as_ref(), &state @@ -1323,7 +1391,9 @@ async fn grant_execution( freshell_platform::claude_inputs::NativeSourceRuntime::Sdk, ) .await?; + timeline.completed("verify_claude_sources"); 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()) @@ -1334,6 +1404,7 @@ async fn grant_execution( format!("hosted fresh-agent provider failed to start: {error}"), ) })?; + timeline.completed("start_fresh_agent"); let profile = actor.profile().await; let native_session_id = profile.native_session_id.clone(); let evidence = serde_json::json!({ @@ -10374,6 +10445,51 @@ mod tests { assert!(state.child.lock().await.is_none()); } + fn grant_timing_events(state_dir: &Path) -> Vec { + 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, + }, + ); + 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!([])); + } + #[tokio::test] async fn grant_from_another_host_boot_is_rejected_before_consumption() { let state = test_host_state(); @@ -10588,6 +10704,31 @@ 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::>(); + 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()); let mut observed = None; for _ in 0..100 { if let Ok(value) = std::fs::read(&child_output) { From f39575f6b2c68e47e9046c93a0cd8fb19f6a3c21 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:53:34 -0700 Subject: [PATCH 02/10] fix(session-host): overlap provider projection flushes so large environments dispatch The provider projection makes every staged node durable before it is published, and it flushed them one at a time inside the execution grant. Each flush is its own journal commit, so a selected plugin with a 4,200-file node_modules spent 8-13 s of the grant flushing on the qualification host (about 2 ms per file, far more under I/O pressure), past the supervisor's grant reply bound. The host then consumed the grant after the supervisor had already declared it unreachable. Flushes of independent nodes now overlap on a bounded worker set so the filesystem journal commits them together (4,200 files: 13.4 s serial, 1.4 s at 16 workers). Every node is still flushed exactly once with a full fsync, symlinks are still not followed, the barrier still completes before the transaction advances, and the calling thread always takes part so a refused thread spawn cannot stall it. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-session-host/src/main.rs | 165 +++++++++++++++++++++-- 1 file changed, 153 insertions(+), 12 deletions(-) diff --git a/crates/freshell-session-host/src/main.rs b/crates/freshell-session-host/src/main.rs index 405dc29c7..e246e69ee 100644 --- a/crates/freshell-session-host/src/main.rs +++ b/crates/freshell-session-host/src/main.rs @@ -3250,12 +3250,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, ) -> Result<(), String> { let metadata = std::fs::symlink_metadata(root).map_err(|error| { format!( @@ -3273,15 +3294,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::); + 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> { @@ -6588,12 +6657,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)); @@ -6604,6 +6674,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::>(); + 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, From 335d0d364945b9cee3d699d768f90df27d99d7bd Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:53:42 -0700 Subject: [PATCH 03/10] fix(supervisor): bound terminal execution grants by their dispatch budget The supervisor bounded a terminal GrantExecution reply by the 10 s control round-trip budget, the same bound as hello, status and stop. A grant is not a control round trip: the host answers only after it has prepared the selected provider inputs and started the provider, and those steps carry their own inner bounds inside the host (Codex app-server start 45 s, Claude final source reader 20 + 5 s, OpenCode native-auth reader 20 + 5 s). An outer bound shorter than the host's own inner bounds abandoned healthy dispatches half-way: the host consumed the grant and started the provider at 10.6-12.9 s while the supervisor had already recorded deliver_execution_grant HOST_UNREACHABLE, torn the incarnation down and blocked recovery as PROVIDER_UNAVAILABLE. Fresh-agent grants already used a 60 s dispatch bound for the same reason. Reply budgets are now resolved once into HostCommandBudgets and chosen by what the command makes the host do: control round trips keep FRESHELL_RUNTIME_HOST_COMMAND_TIMEOUT_MS (10 s), terminal grants use FRESHELL_RUNTIME_TERMINAL_DISPATCH_TIMEOUT_MS (60 s, documented against the inner bounds above) and fresh-agent grants and turn operations keep FRESHELL_RUNTIME_FRESH_AGENT_COMMAND_TIMEOUT_MS. A silent host is still released at the dispatch bound. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/src/service.rs | 319 ++++++++++++++++++++-- docs/development/managed-runtime.md | 18 ++ 2 files changed, 316 insertions(+), 21 deletions(-) diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs index 39efaf62c..9b7d05b29 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -117,6 +117,8 @@ pub struct Supervisor { pub(crate) live_resume_evidence_attempts: Arc>>, auth_lookup_authorities: Arc>>, provider_path_inputs: Arc>, + /// Reply budgets for supervisor→session-host commands, read once. + pub(crate) host_command_budgets: HostCommandBudgets, } impl Supervisor { @@ -166,6 +168,7 @@ 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(), }) } @@ -2730,7 +2733,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) => { @@ -2917,16 +2920,7 @@ impl Supervisor { host: &AuthenticatedHost, command: HostCommand, ) -> Result { - let is_slow_fresh_agent_command = matches!( - &command, - HostCommand::FreshAgentSend { .. } - | HostCommand::FreshAgentCompact { .. } - | HostCommand::FreshAgentRollback { .. } - | HostCommand::GrantExecution { - fresh_agent: Some(_), - .. - } - ); + let budget = self.host_command_budgets.for_command(&command); let mut envelope = Envelope::new(RequestId::new(), ControlRole::Supervisor, command); envelope.auth = Some(host_proof( &host.secret, @@ -2934,15 +2928,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 } @@ -3122,6 +3107,24 @@ 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 bound on 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 resolves and projects the selected +/// provider inputs, then starts the provider, and several of those steps carry +/// their own bounds inside the host: the Codex app-server start budget (45 s, +/// `freshell_codex::launch_lifecycle::SIDECAR_START_BUDGET`), the Claude final +/// source reader (20 s exchange + 5 s cleanup) and the OpenCode native-auth +/// reader (20 s + 5 s). An outer bound shorter than those inner bounds abandons +/// a healthy dispatch half-way: the host still consumes the grant and starts +/// the provider, while the supervisor has already declared the host +/// unreachable and torn the incarnation down. The bound matches the +/// fresh-agent dispatch bound, which covers its 50-second create for the same +/// reason, and stays finite so a silent host still releases the activation. +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. @@ -3143,6 +3146,61 @@ 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::().ok()) + .filter(|ms| *ms > 0) + .map(Duration::from_millis) + .unwrap_or(Duration::from_millis(DEFAULT_TERMINAL_DISPATCH_TIMEOUT_MS)) +} + +/// 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) -> 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 { + match command { + HostCommand::FreshAgentSend { .. } + | HostCommand::FreshAgentCompact { .. } + | HostCommand::FreshAgentRollback { .. } + | HostCommand::GrantExecution { + fresh_agent: Some(_), + .. + } => self.fresh_agent, + HostCommand::GrantExecution { + terminal: Some(_), .. + } => self.terminal_dispatch, + _ => self.control, + } + } +} + /// 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. @@ -3337,7 +3395,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, @@ -3461,6 +3520,224 @@ 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, AuthenticatedHost, HostCommandBudgets, Supervisor, SupervisorConfig, + HOST_COMMAND_TIMEOUT_ENV, + }; + use crate::registry::Registry; + use freshell_runtime_protocol::{ + read_frame, write_frame, Envelope, GrantId, HostBootId, HostCommand, HostReply, HostResult, + IncarnationId, RuntimeErrorCode, RuntimeLimits, SoulId, TerminalLaunchSpec, + }; + use std::path::Path; + 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, + } + } + + /// Every non-control budget is generous; a control round trip is short. + fn budgets(control_ms: &'static str, dispatch_ms: &'static str) -> HostCommandBudgets { + HostCommandBudgets::from_lookup(move |name| { + Some(if name == HOST_COMMAND_TIMEOUT_ENV { + control_ms + } else { + dispatch_ms + }) + .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, + } + } + + /// A synthetic host that answers every command after `delay` (or never). + fn slow_host(runtime_dir: &Path, delay: Option) -> tokio::task::JoinHandle<()> { + let listener = UnixListener::bind(runtime_dir.join("host.sock")).unwrap(); + tokio::spawn(async move { + loop { + let (mut stream, _) = listener.accept().await.unwrap(); + tokio::spawn(async move { + let envelope: Envelope = read_frame(&mut stream).await.unwrap(); + 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; + }); + } + }) + } + + 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")); + 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")); + 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(); } } diff --git a/docs/development/managed-runtime.md b/docs/development/managed-runtime.md index eb8692fc8..4502497f3 100644 --- a/docs/development/managed-runtime.md +++ b/docs/development/managed-runtime.md @@ -127,6 +127,24 @@ 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, and several of those steps carry their own inner +bounds (Codex's app-server start alone may take up to 45s). A terminal grant is +bounded by `FRESHELL_RUNTIME_TERMINAL_DISPATCH_TIMEOUT_MS` and a fresh-agent +grant or turn operation by `FRESHELL_RUNTIME_FRESH_AGENT_COMMAND_TIMEOUT_MS` +(both default 60s). An outer bound shorter than the host's inner bounds would +abandon a healthy dispatch half-way: the host still consumes the grant and +starts the provider, while the supervisor has already torn the incarnation +down. 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, for accepted and refused grants alike. + +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 From 64596272719d5511f9973e5fb5a19c5622608f40 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:53:48 -0700 Subject: [PATCH 04/10] feat(supervisor): record why a Claude source observation was refused A Claude creation or recovery whose native source observation failed surfaced only the fixed sanitized refusal ("evidence unavailable") to the web server, and the supervisor has no tracing subscriber, so the exact check that stopped it was lost. A blocked Claude recovery showed nothing but CAPABILITY_PENDING. The observation now records supervisor.claude_source_observation.failed in the supervisor lifecycle log (severity, soul, provider, authority kind, candidate flag, error code and the specific refused check or reader failure). The web caller still receives only the fixed refusal. The candidate scratch absence probe also stops calling every non-404 a name that is "already present": only an answered lookup proves presence, and any other refusal is reported as absence that could not be confirmed. Nothing is created in either case. Co-Authored-By: Claude Opus 5.5 --- .../src/backend/claude_source_observation.rs | 8 +- .../src/claude_source_observation.rs | 138 +++++++++++++++--- 2 files changed, 121 insertions(+), 25 deletions(-) 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..e8584e886 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 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 { + 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 + { 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,7 +169,7 @@ 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. @@ -119,14 +179,16 @@ impl Supervisor { .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 || ¤t.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")); } } } @@ -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::(line).unwrap()) + .filter(|event| event["event"] == "supervisor.claude_source_observation.failed") + .map(|event| event["data"].clone()) + .collect::>(); 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"); } From 60cc170386a1707da57bf5142303c82e40f6049d Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 11:53:54 -0700 Subject: [PATCH 05/10] test(runtime-broker): qualify recovery Claude candidate publications in browser fixtures The restricted Docker broker that fronts browser runtime fixtures could qualify a controller-published Claude candidate only for initial creation. A recovery candidate (registry attempt, prior generation, no initial reservation) was refused at owner_capability, so its scratch absence probe got a 403 and the supervisor blocked every managed Claude terminal recovery as CAPABILITY_PENDING. The provider parity suite timed out waiting for the Claude replacement incarnation. Initial and recovery publications now share one qualifier for both the absence probe and the later volume registration. A recovery candidate must name a well-formed registry attempt and a different prior generation, its capability must be the pending replacement of exactly that generation, and its retained profile (reservation, binds and resource units) comes from the frozen prior stage, the same profile the closed launch fixture hands off explicitly. The registration also now requires the exact candidate labels and, for a creation candidate, its reservation authority digest, which the absence probe already required. Co-Authored-By: Claude Opus 5.5 --- scripts/testing/runtime-test-broker.ts | 144 ++++++++++++------ .../testing/runtime-test-broker.test.ts | 76 ++++++++- 2 files changed, 168 insertions(+), 52 deletions(-) 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 { 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, - 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, 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'), From f701c033d84fc490b4d346876476dfeacafc237f Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:39:32 -0700 Subject: [PATCH 06/10] test(supervisor): keep the dispatch budget tests clippy-clean Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/src/service.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs index 9b7d05b29..bb7c951d2 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -3562,17 +3562,19 @@ mod host_command_budget_tests { /// Every non-control budget is generous; a control round trip is short. fn budgets(control_ms: &'static str, dispatch_ms: &'static str) -> HostCommandBudgets { HostCommandBudgets::from_lookup(move |name| { - Some(if name == HOST_COMMAND_TIMEOUT_ENV { - control_ms - } else { - dispatch_ms - }) - .map(str::to_string) + Some( + if name == HOST_COMMAND_TIMEOUT_ENV { + control_ms + } else { + dispatch_ms + } + .to_string(), + ) }) } fn supervisor(root: &Path, budgets: HostCommandBudgets) -> Supervisor { - let registry = Registry::open(&root.join("registry"), None).unwrap(); + 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( From 1b37e250c2de4773f1e73ed189b4218c3ff82451 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:25:47 -0700 Subject: [PATCH 07/10] test(supervisor): hold each host command class to its own budget The budget tests gave terminal grants and fresh-agent operations the same configured value, so routing a terminal grant to the fresh-agent budget (or reading the wrong variable for it) still passed. Give control, terminal dispatch, and fresh-agent budgets distinct values and assert that each command class times out at its own. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-supervisor/src/service.rs | 83 +++++++++++++++++++---- 1 file changed, 69 insertions(+), 14 deletions(-) diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs index 49e768a38..4280a75a7 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -3625,12 +3625,12 @@ mod host_ipc_timeout_tests { mod host_command_budget_tests { use super::{ default_backend, AuthenticatedHost, HostCommandBudgets, Supervisor, SupervisorConfig, - HOST_COMMAND_TIMEOUT_ENV, + 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, RuntimeErrorCode, RuntimeLimits, SoulId, TerminalLaunchSpec, + IncarnationId, RequestId, RuntimeErrorCode, RuntimeLimits, SoulId, TerminalLaunchSpec, }; use std::path::Path; use std::time::{Duration, Instant}; @@ -3645,17 +3645,21 @@ mod host_command_budget_tests { } } - /// Every non-control budget is generous; a control round trip is short. - fn budgets(control_ms: &'static str, dispatch_ms: &'static str) -> HostCommandBudgets { + /// 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| { - Some( - if name == HOST_COMMAND_TIMEOUT_ENV { - control_ms - } else { - dispatch_ms - } - .to_string(), - ) + 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) }) } @@ -3761,7 +3765,7 @@ mod host_command_budget_tests { 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")); + 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(); @@ -3805,7 +3809,7 @@ mod host_command_budget_tests { 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")); + 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(); @@ -3827,6 +3831,57 @@ mod host_command_budget_tests { 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(); + } } #[cfg(test)] From fe7ce56b9c235d757d12e50aab50324611b27d70 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:29:35 -0700 Subject: [PATCH 08/10] fix(runtime): end a long grant dispatch inside the host, within the supervisor's wait The 60s terminal grant wait was documented as longer than the host's inner limits, but it is not: a terminal Codex start retries its 45s app-server start up to five times (about 251s), and an exact resume runs the 25s provider-owned refresh (plus up to 32s of recovery) first. A slow but healthy dispatch could still be abandoned by the supervisor while the host kept working on it. No fixed wait covers every such sum, so the grant now carries its own dispatch budget (`dispatch_budget_ms`): the supervisor's wait less a 5s reply allowance, never less than half the wait. The host runs every awaited dispatch stage inside it and cuts the stage at the deadline, never crosses the native dispatch boundary once it is spent (the grant stays unconsumed), and refuses with a typed HostUnreachable naming the stage the budget ran out before. `host.grant_timing` records `budgetMs` and `budgetExhaustedBefore`. The wait and the budget now have accurate docs, including that a stop issued during activation waits for it to settle. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-runtime-protocol/src/lib.rs | 7 + crates/freshell-session-host/src/main.rs | 413 +++++++++++++++----- crates/freshell-supervisor/src/service.rs | 145 ++++++- docs/development/managed-runtime.md | 40 +- 4 files changed, 467 insertions(+), 138 deletions(-) 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>, #[serde(default, skip_serializing_if = "Option::is_none")] resume_spec: Option>, + /// 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, }, Stop { incarnation_id: IncarnationId, diff --git a/crates/freshell-session-host/src/main.rs b/crates/freshell-session-host/src/main.rs index 131bb3fcc..616b9a61b 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,25 +940,89 @@ struct GrantExecutionRequest { terminal: Option, fresh_agent: Option, resume_spec: Option, -} - -/// Wall-clock breakdown of one execution grant. The supervisor bounds the -/// whole grant reply; this names the stage that spent the time, for every -/// outcome, so a slow dispatch is diagnosable after the container is gone. + /// The supervisor's bound on this dispatch; see [`GrantStageTimeline`]. + dispatch_budget: Option, +} + +/// 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, + exhausted_before: Option<&'static str>, } impl GrantStageTimeline { - fn start() -> Self { + fn start(budget: Option) -> Self { let now = std::time::Instant::now(); Self { started: now, stage_started: now, stages: Vec::new(), + budget, + exhausted_before: None, + } + } + + fn deadline(&self) -> Option { + 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( + &mut self, + stage: &'static str, + work: impl std::future::Future>, + ) -> Result { + 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) { @@ -974,6 +1040,8 @@ impl GrantStageTimeline { "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() @@ -988,7 +1056,7 @@ async fn grant_execution( request: GrantExecutionRequest, ) -> Result { let grant_id = request.grant_id.clone(); - let mut timeline = GrantStageTimeline::start(); + 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()); @@ -1011,6 +1079,7 @@ async fn grant_execution_stages( terminal, fresh_agent, resume_spec, + dispatch_budget: _, } = request; let provider = terminal .as_ref() @@ -1103,30 +1172,33 @@ async fn grant_execution_stages( })?; 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); - timeline.completed("prepare_provider_inputs"); } else if !provider_path_inputs.is_empty() { return Err(RuntimeError::new( RuntimeErrorCode::InvalidRequest, @@ -1189,6 +1261,7 @@ async fn grant_execution_stages( .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( @@ -1263,57 +1336,72 @@ async fn grant_execution_stages( .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?; } timeline.completed("prepare_provider_state"); - 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}"), + 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, + ), ) - })?; - timeline.completed("prepare_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?; - timeline.completed("verify_claude_sources"); + .await?; + timeline.ensure_budget("consume_grant")?; consume_grant(&mut persisted)?; timeline.completed("consume_grant"); - let prepared = providers::start_terminal_sidecar(prepared) - .await - .map_err(|error| RuntimeError::new(RuntimeErrorCode::HostUnreachable, error))?; - timeline.completed("start_sidecar"); + 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}"), - ) - })?; - timeline.completed("spawn_worker"); + 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", @@ -1419,41 +1507,51 @@ async fn grant_execution_stages( ) .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))?; - timeline.completed("prepare_provider_state"); - 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.completed("verify_claude_sources"); + 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}"), - ) - })?; - timeline.completed("start_fresh_agent"); + 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!({ @@ -10578,6 +10676,7 @@ mod tests { terminal: Some(terminal), fresh_agent: None, resume_spec: None, + dispatch_budget: None, }, ) .await; @@ -10661,6 +10760,7 @@ mod tests { terminal: terminal.clone(), fresh_agent: fresh.clone(), resume_spec: Some(resume.clone()), + dispatch_budget: None, }, ) .await @@ -10733,6 +10833,7 @@ mod tests { terminal: Some(terminal.clone()), fresh_agent: None, resume_spec: Some(resume.clone()), + dispatch_budget: None, }, ) .await @@ -10825,6 +10926,7 @@ mod tests { terminal: None, fresh_agent: None, resume_spec: None, + dispatch_budget_ms: None, }, ); let error = dispatch(envelope, &state).await.unwrap_err(); @@ -10838,6 +10940,97 @@ mod tests { 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::>(); + 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(); @@ -10855,6 +11048,7 @@ mod tests { terminal: None, fresh_agent: None, resume_spec: None, + dispatch_budget_ms: None, }, ); let error = dispatch(envelope, &state).await.unwrap_err(); @@ -11037,6 +11231,7 @@ mod tests { terminal: Some(Box::new(terminal)), fresh_agent: None, resume_spec: Some(Box::new(resume_spec)), + dispatch_budget_ms: Some(60_000), }, ); @@ -11077,6 +11272,8 @@ mod tests { "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/service.rs b/crates/freshell-supervisor/src/service.rs index 4280a75a7..2a7ecb322 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -2887,6 +2887,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) @@ -2986,6 +2987,7 @@ impl Supervisor { command: HostCommand, ) -> Result { 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, @@ -3172,20 +3174,22 @@ 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 bound on a terminal execution grant's reply. +/// 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 resolves and projects the selected -/// provider inputs, then starts the provider, and several of those steps carry -/// their own bounds inside the host: the Codex app-server start budget (45 s, -/// `freshell_codex::launch_lifecycle::SIDECAR_START_BUDGET`), the Claude final -/// source reader (20 s exchange + 5 s cleanup) and the OpenCode native-auth -/// reader (20 s + 5 s). An outer bound shorter than those inner bounds abandons -/// a healthy dispatch half-way: the host still consumes the grant and starts -/// the provider, while the supervisor has already declared the host -/// unreachable and torn the incarnation down. The bound matches the -/// fresh-agent dispatch bound, which covers its 50-second create for the same -/// reason, and stays finite so a silent host still releases the activation. +/// 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"; @@ -3219,6 +3223,32 @@ pub(crate) fn terminal_dispatch_budget(configured: Option<&str>) -> Duration { .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. @@ -3624,8 +3654,9 @@ mod host_ipc_timeout_tests { #[cfg(test)] mod host_command_budget_tests { use super::{ - default_backend, AuthenticatedHost, HostCommandBudgets, Supervisor, SupervisorConfig, - FRESH_AGENT_COMMAND_TIMEOUT_ENV, HOST_COMMAND_TIMEOUT_ENV, TERMINAL_DISPATCH_TIMEOUT_ENV, + 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::{ @@ -3633,6 +3664,7 @@ mod host_command_budget_tests { 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; @@ -3706,17 +3738,28 @@ mod host_command_budget_tests { terminal: Some(Box::new(terminal)), fresh_agent: None, resume_spec: None, + dispatch_budget_ms: None, } } - /// A synthetic host that answers every command after `delay` (or never). - fn slow_host(runtime_dir: &Path, delay: Option) -> tokio::task::JoinHandle<()> { + type Received = Arc>>; + + /// A synthetic host that answers every command after `delay` (or never), + /// recording each command it received. + fn slow_host( + runtime_dir: &Path, + delay: Option, + ) -> (tokio::task::JoinHandle<()>, Received) { let listener = UnixListener::bind(runtime_dir.join("host.sock")).unwrap(); - tokio::spawn(async move { + 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 = 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; @@ -3744,7 +3787,8 @@ mod host_command_budget_tests { .await; }); } - }) + }); + (task, received) } fn authenticated() -> AuthenticatedHost { @@ -3766,7 +3810,7 @@ mod host_command_budget_tests { 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 (host, _) = slow_host(&runtime_dir, Some(Duration::from_millis(700))); let incarnation_id = IncarnationId::new(); let granted = supervisor @@ -3810,7 +3854,7 @@ mod host_command_budget_tests { 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 (host, _) = slow_host(&runtime_dir, None); let incarnation_id = IncarnationId::new(); let started = Instant::now(); let error = supervisor @@ -3840,7 +3884,7 @@ mod host_command_budget_tests { 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 (host, _) = slow_host(&runtime_dir, None); let incarnation_id = IncarnationId::new(); let host_auth = authenticated(); let send = |command| { @@ -3882,6 +3926,65 @@ mod host_command_budget_tests { } 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) + ); + } } #[cfg(test)] diff --git a/docs/development/managed-runtime.md b/docs/development/managed-runtime.md index 4502497f3..e91172441 100644 --- a/docs/development/managed-runtime.md +++ b/docs/development/managed-runtime.md @@ -129,16 +129,38 @@ 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, and several of those steps carry their own inner -bounds (Codex's app-server start alone may take up to 45s). A terminal grant is -bounded by `FRESHELL_RUNTIME_TERMINAL_DISPATCH_TIMEOUT_MS` and a fresh-agent -grant or turn operation by `FRESHELL_RUNTIME_FRESH_AGENT_COMMAND_TIMEOUT_MS` -(both default 60s). An outer bound shorter than the host's inner bounds would -abandon a healthy dispatch half-way: the host still consumes the grant and -starts the provider, while the supervisor has already torn the incarnation -down. Every grant records a `host.grant_timing` event (in the incarnation's +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, for accepted and refused grants alike. +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. + +A launch or recovery holds the soul's lifecycle lock for its whole +activation, 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. The provider projection inside a grant makes every staged file durable before it is published. Those flushes overlap (`PROVIDER_PROJECTION_SYNC_WORKERS`) so From dd309cb40a42eee46a5a9f9dc00904a833f6c4dc Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:33:30 -0700 Subject: [PATCH 09/10] fix(supervisor): let a Claude recovery callback cover its two source observations Recovery's provider-preparation callback to the web server waited 35s and its recovery scope expired after 30s, yet a Claude recovery's callback makes two source observations back to back through the supervisor before it answers, each bounded at 35s on its own. An observation that finished inside its own bound could still be cut off by the callback around it, or find its scope expired, and the recovery blocked as CAPABILITY_PENDING. In the failing parity run on the previous head the candidate reader hit its own 25s limit 32.3s after the recovery was scheduled, inside a callback allowed 35s, under a load average near 600. A Claude recovery callback now waits 105s (the 35s callback wait plus two 35s observation bounds); every other callback keeps 35s. The recovery scope lives exactly as long as the callback wait it serves. The timeout message names the wait it hit. Co-Authored-By: Claude Opus 5.5 --- .../src/claude_source_observation.rs | 4 +- .../src/native_auth_lookup_tests.rs | 4 +- .../src/provider_callback_budget_tests.rs | 235 ++++++++++++++++++ crates/freshell-supervisor/src/service.rs | 83 ++++++- docs/development/managed-runtime.md | 17 +- 5 files changed, 327 insertions(+), 16 deletions(-) create mode 100644 crates/freshell-supervisor/src/provider_callback_budget_tests.rs diff --git a/crates/freshell-supervisor/src/claude_source_observation.rs b/crates/freshell-supervisor/src/claude_source_observation.rs index e8584e886..716955273 100644 --- a/crates/freshell-supervisor/src/claude_source_observation.rs +++ b/crates/freshell-supervisor/src/claude_source_observation.rs @@ -174,7 +174,7 @@ impl Supervisor { // 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), ) @@ -461,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" 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 2a7ecb322..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 { @@ -119,6 +122,7 @@ pub struct Supervisor { provider_path_inputs: Arc>, /// Reply budgets for supervisor→session-host commands, read once. pub(crate) host_command_budgets: HostCommandBudgets, + pub(crate) provider_callback_budgets: ProviderCallbackBudgets, } impl Supervisor { @@ -169,6 +173,7 @@ impl Supervisor { 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(), }) } @@ -682,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 @@ -693,7 +705,7 @@ impl Supervisor { key.clone(), AuthLookupAuthority { start: start.clone(), - expires: Instant::now() + Duration::from_secs(30), + expires: Instant::now() + lifetime, }, ); AuthLookupScope { @@ -2515,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 @@ -2605,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| { @@ -3296,6 +3317,52 @@ impl HostCommandBudgets { } } +/// 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. diff --git a/docs/development/managed-runtime.md b/docs/development/managed-runtime.md index e91172441..0815e5656 100644 --- a/docs/development/managed-runtime.md +++ b/docs/development/managed-runtime.md @@ -156,11 +156,20 @@ 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, 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. +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 From 6af75b1995d68ab3c1ec5862906db5b2425b1793 Mon Sep 17 00:00:00 2001 From: Dan Shapiro <3732858+danshapiro@users.noreply.github.com> Date: Thu, 22 Oct 2026 15:53:34 -0700 Subject: [PATCH 10/10] test(session-host): run freshly written fixtures through a transient ETXTBSY The pre-push workspace run failed provider_config_two_stage_script_is_runnable with "Text file busy": another test thread forked while the fixture was still open for writing, its child held a copy of that write descriptor until it exec'd, and exec of the fixture was refused. Run freshly written fixtures through a helper that retries only that transient refusal for up to 500ms (the platform crate's fixtures already do), with a test that holds a write descriptor open to prove it. Co-Authored-By: Claude Opus 5.5 --- crates/freshell-session-host/src/main.rs | 60 +++++++++++++++++++++--- 1 file changed, 54 insertions(+), 6 deletions(-) diff --git a/crates/freshell-session-host/src/main.rs b/crates/freshell-session-host/src/main.rs index 616b9a61b..08b359d7c 100644 --- a/crates/freshell-session-host/src/main.rs +++ b/crates/freshell-session-host/src/main.rs @@ -7490,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 { + 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::*; @@ -8226,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(); @@ -8237,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"); @@ -9681,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"); }