diff --git a/crates/freshell-server/src/managed_provider_launch_tests.rs b/crates/freshell-server/src/managed_provider_launch_tests.rs index 175ba25ea..b5cc7deee 100644 --- a/crates/freshell-server/src/managed_provider_launch_tests.rs +++ b/crates/freshell-server/src/managed_provider_launch_tests.rs @@ -504,11 +504,12 @@ impl RuntimeBackend for BrokerAuthorizedBackend { owner: &freshell_supervisor::backend::ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, test_run_id: &str, + until: tokio::time::Instant, ) -> Result { self.authorize_claude_candidate(owner, request).await?; let reply = match self .real() - .observe_claude_sources(owner, request, test_run_id) + .observe_claude_sources(owner, request, test_run_id, until) .await { Ok(reply) => reply, diff --git a/crates/freshell-server/src/managed_provider_references/claude.rs b/crates/freshell-server/src/managed_provider_references/claude.rs index d93a58366..4f650a11d 100644 --- a/crates/freshell-server/src/managed_provider_references/claude.rs +++ b/crates/freshell-server/src/managed_provider_references/claude.rs @@ -632,6 +632,7 @@ mod tests { owner: &ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, _: &str, + _: tokio::time::Instant, ) -> Result { assert!(matches!(owner, ClaudeSourceOwner::Initial { .. })); diff --git a/crates/freshell-server/src/managed_runtime_opencode_replan_tests.rs b/crates/freshell-server/src/managed_runtime_opencode_replan_tests.rs index be79d1218..e168dd342 100644 --- a/crates/freshell-server/src/managed_runtime_opencode_replan_tests.rs +++ b/crates/freshell-server/src/managed_runtime_opencode_replan_tests.rs @@ -43,6 +43,7 @@ mod uncertain_launch { _: &ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, _: &str, + _: tokio::time::Instant, ) -> Result { let r = &request.observation_request; diff --git a/crates/freshell-supervisor/src/backend.rs b/crates/freshell-supervisor/src/backend.rs index d73981b16..654b89ba6 100644 --- a/crates/freshell-supervisor/src/backend.rs +++ b/crates/freshell-supervisor/src/backend.rs @@ -1,5 +1,7 @@ mod claude_source_observation; pub mod docker; +mod helper_lifecycle; +pub(crate) use helper_lifecycle::CLAUDE_SOURCE_READER_RETRY; mod input_publisher; mod native_auth; mod native_history; @@ -138,11 +140,14 @@ pub enum ClaudeSourceOwner { #[async_trait] pub trait RuntimeBackend: Send + Sync { + /// Observe Claude sources through the fixed reader. The read, its owned + /// cleanup included, ends by `until`. async fn observe_claude_sources( &self, _owner: &ClaudeSourceOwner, _request: &freshell_runtime_protocol::ObserveClaudeSourcesRequest, _test_run_id: &str, + _until: tokio::time::Instant, ) -> Result { Err(BackendError::Unavailable( "Claude native source reader unavailable; retain inputs and reprepare".into(), @@ -230,6 +235,8 @@ pub trait RuntimeBackend: Send + Sync { #[derive(Clone)] pub struct DockerEngineBackend { socket_path: PathBuf, + /// The supervisor's lifecycle JSONL, for helper retry and cleanup events. + lifecycle_log: Option>, input_publishers: std::sync::Arc>>, input_publication_claims: std::sync::Arc>>, @@ -239,11 +246,19 @@ impl DockerEngineBackend { pub fn new(socket_path: impl Into) -> Self { Self { socket_path: socket_path.into(), + lifecycle_log: None, input_publishers: Default::default(), input_publication_claims: Default::default(), } } + /// Record helper retry and late-cleanup events in the supervisor's + /// lifecycle JSONL. + pub fn with_lifecycle_log(mut self, path: impl Into) -> Self { + self.lifecycle_log = Some(std::sync::Arc::new(path.into())); + self + } + pub async fn ping(&self) -> Result<(), BackendError> { let response = self.request("GET", "/_ping", None).await?; if response.status == 200 && response.body.starts_with(b"OK") { @@ -709,12 +724,14 @@ impl RuntimeBackend for DockerEngineBackend { owner: &ClaudeSourceOwner, request: &freshell_runtime_protocol::ObserveClaudeSourcesRequest, test_run_id: &str, + until: tokio::time::Instant, ) -> Result { - self.read_claude_source_helper( + self.observe_claude_sources_with( + &helper_lifecycle::CLAUDE_SOURCE_READER_RETRY, owner, request, test_run_id, - std::time::Duration::from_secs(25), + until, ) .await } diff --git a/crates/freshell-supervisor/src/backend/claude_source_observation.rs b/crates/freshell-supervisor/src/backend/claude_source_observation.rs index 3672cac20..e9b502db3 100644 --- a/crates/freshell-supervisor/src/backend/claude_source_observation.rs +++ b/crates/freshell-supervisor/src/backend/claude_source_observation.rs @@ -5,7 +5,7 @@ use freshell_platform::claude_inputs::observation::{ ObservationReply, MAX_REPLY_BYTES, MAX_REQUEST_BYTES, }; use freshell_runtime_protocol::ObserveClaudeSourcesRequest; -use std::time::Duration; +use helper_lifecycle::{AttemptStep, HelperContainer, HelperDeadlines, HelperFailure, HelperKind}; use tokio::net::UnixListener; fn fail(reason: &'static str) -> BackendError { @@ -52,9 +52,17 @@ struct CandidateScratch { labels: Value, daemon: DockerDaemonId, attempted: bool, + create: helper_lifecycle::InFlightCreate, } impl CandidateScratch { + /// Wait until the volume create answers; false if it still has not. + async fn settle(&mut self, deadline: tokio::time::Instant) -> bool { + self.create + .settle_ownership(deadline, &mut self.attempted) + .await + } + fn owns(&self, actual: &Value) -> bool { actual["Name"] == self.name && actual["Driver"] == "local" @@ -164,6 +172,9 @@ impl CandidateInputView { } fn cleanup(&mut self) -> Result<(), BackendError> { + if self.1 { + return Ok(()); + } std::fs::remove_dir_all(&self.0) .map_err(|_| fail("Claude candidate ordinary input view cleanup failed"))?; if !matches!(std::fs::symlink_metadata(&self.0), Err(error) if error.kind()==std::io::ErrorKind::NotFound) @@ -188,6 +199,63 @@ impl Drop for CandidateInputView { } } +/// Everything one candidate read may leave in Docker or on disk: the +/// mountpoint helper, the reader, the ordinary scratch volume both mount, and +/// the mount skeletons. Released together, helpers first, so an uncertain +/// helper removal never deletes a still-mounted volume. The skeletons back +/// only the helpers' bind mounts, so they go as soon as the helpers are gone, +/// before the scratch volume, whose removal may still need retrying. +struct CandidateObjects { + mountpoints: Option, + reader: Option, + scratch: Option, + views: Vec, + directory: Option, +} + +#[async_trait] +impl helper_lifecycle::HelperObjects for CandidateObjects { + fn names(&self) -> Vec { + self.mountpoints + .iter() + .chain(self.reader.iter()) + .map(|helper| helper.name.clone()) + .chain( + self.scratch + .iter() + .filter(|scratch| scratch.attempted) + .map(|scratch| scratch.name.clone()), + ) + .collect() + } + + async fn settle(&mut self, deadline: tokio::time::Instant) -> bool { + let mut settled = true; + for helper in self.mountpoints.iter_mut().chain(self.reader.iter_mut()) { + settled &= helper.settle(deadline).await; + } + if let Some(scratch) = self.scratch.as_mut() { + settled &= scratch.settle(deadline).await; + } + settled + } + + async fn remove(&mut self, backend: &DockerEngineBackend) -> Result<(), BackendError> { + for helper in self.mountpoints.iter_mut().chain(self.reader.iter_mut()) { + helper.remove(backend).await?; + } + for view in &mut self.views { + view.cleanup()?; + } + // The skeletons lived inside the private exchange; it goes after them. + self.directory.take(); + if let Some(scratch) = self.scratch.as_mut() { + backend.cleanup_claude_candidate_scratch(scratch).await?; + } + Ok(()) + } +} + impl DockerEngineBackend { async fn retained_claude_volume( &self, @@ -242,28 +310,28 @@ impl DockerEngineBackend { request: &ObserveClaudeSourcesRequest, root: &Path, test_run_id: &str, - budget: Duration, + deadlines: HelperDeadlines, prior_stage: Option<&Path>, - ) -> Result { + ) -> Result { use freshell_runtime_protocol::{bounded_json_bytes, MAX_CONTROL_FRAME_BYTES}; - let deadline = tokio::time::Instant::now() + budget; + let deadline = deadlines.operation; let name = format!("freshell-claude-candidate-reader-{}", uuid::Uuid::new_v4()); let mut scratch = None::; - let mut mountpoint_helper = None::<(String, Value)>; - let mut expected = None::; - let mut attempted = false; + let mut mountpoint_helper = None::; + let mut reader = None::; let mut directory = None::; let mut input_view = None::; let mut retained_view = None::; + let step = AttemptStep::new(); let operation = async { let projection = qualify_candidate_stage(root, owner, request)?; let raw = bounded_json_bytes(&projection, MAX_CONTROL_FRAME_BYTES) .map_err(|_| fail("Claude candidate projection exceeds metadata profile"))?; let (daemon, rootless) = self.daemon_info().await?; if !rootless { - return Err(fail( - "Claude candidate reader requires the qualified rootless daemon", - )); + return Err( + fail("Claude candidate reader requires the qualified rootless daemon").into(), + ); } let mut retained_volume = None; let (installation, image, binary_path, reservation_id) = match owner { @@ -276,7 +344,7 @@ impl DockerEngineBackend { || self.observe_initial_volume(reservation).await? != *observation || !matches!(observation, freshell_runtime_protocol::InitialAuthVolumeObservation::Absent { daemon_id } if daemon_id == &daemon) { - return Err(fail("Claude candidate creation absence changed")); + return Err(fail("Claude candidate creation absence changed").into()); } ( reservation.installation_id.as_str(), @@ -304,7 +372,7 @@ impl DockerEngineBackend { .map_err(|_| fail("Claude candidate reader artifact unavailable"))?; if let ClaudeSourceOwner::Initial { reservation, .. } = owner { if digest != reservation.binary_sha256 { - return Err(fail("Claude candidate reserved reader artifact changed")); + return Err(fail("Claude candidate reserved reader artifact changed").into()); } } let workspace = canonical_dir(Path::new(&request.workspace_path)) @@ -315,7 +383,7 @@ impl DockerEngineBackend { || cwd != request.observation_request.cwd || !cwd.starts_with(&workspace) { - return Err(fail("Claude candidate workload topology changed")); + return Err(fail("Claude candidate workload topology changed").into()); } let mounts = docker::ordinary_claude_candidate_mounts( &workspace, @@ -343,8 +411,10 @@ impl DockerEngineBackend { labels: labels.clone(), daemon: daemon.clone(), attempted: false, + create: Default::default(), }); let backing = scratch.as_mut().unwrap(); + step.enter("scratch_create"); self.create_claude_candidate_scratch(backing).await?; self.run_provider_mountpoint_helper( provider_mountpoints::MountpointHelperTarget { @@ -355,13 +425,12 @@ impl DockerEngineBackend { labels: labels.clone(), }, provider_mountpoints::selections(&mounts)?, - ( - deadline.saturating_duration_since(tokio::time::Instant::now()), - Duration::from_secs(5), - ), + deadlines, + &step, &mut mountpoint_helper, ) .await?; + step.enter("prepare"); directory = Some(native_auth::ExchangeDirectory::create_input()?); let exchange_directory = directory.as_ref().unwrap(); let socket = exchange_directory.0.join("reply.sock"); @@ -407,22 +476,26 @@ impl DockerEngineBackend { "HostConfig":{"NetworkMode":"none","ReadonlyRootfs":true,"CapDrop":["ALL"],"CapAdd":["CHOWN","SETGID","SETUID"], "SecurityOpt":["no-new-privileges"],"Memory":256*1024*1024,"MemorySwap":256*1024*1024,"NanoCpus":500_000_000,"PidsLimit":32, "LogConfig":{"Type":"none","Config":{}},"Tmpfs":{},"Mounts":descriptors}}); - expected = Some(body.clone()); - attempted = true; - let created = self - .request_bounded( - "POST", - &format!("{DOCKER_API}/containers/create?name={name}"), - Some(&body), - 64 * 1024, - ) - .await?; - if created.status == 409 { - attempted = false; - } + step.enter("create"); + let created = reader + .insert(HelperContainer::new( + HelperKind::ClaudeCandidateReader, + name.clone(), + body.clone(), + )) + .create(self) + .await + .map_err(|_| { + HelperFailure::unanswered( + "create", + fail("Claude candidate reader creation acknowledgement unavailable"), + ) + })?; if created.status != 201 { - return Err(fail( - "Claude candidate reader creation acknowledgement unavailable", + return Err(HelperFailure::answered( + "create", + created.status, + fail("Claude candidate reader creation acknowledgement unavailable"), )); } let created: Value = serde_json::from_slice(&created.body) @@ -446,9 +519,7 @@ impl DockerEngineBackend { || actual["State"]["Running"] != false || !owned_candidate_helper(&actual, &body) { - return Err(fail( - "Claude candidate reader namespace changed before start", - )); + return Err(fail("Claude candidate reader namespace changed before start").into()); } let volume_reply = self .request_bounded( @@ -461,7 +532,7 @@ impl DockerEngineBackend { let volume: Value = serde_json::from_slice(&volume_reply.body) .map_err(|_| fail("Claude candidate ordinary volume metadata invalid"))?; if volume_reply.status != 200 { - return Err(fail("Claude candidate ordinary backing disappeared")); + return Err(fail("Claude candidate ordinary backing disappeared").into()); } verify_candidate_mount_reports( &actual, @@ -472,6 +543,7 @@ impl DockerEngineBackend { retained_volume.as_ref(), )?; let supervision = async { + step.enter("start"); let started = self .request_bounded( "POST", @@ -479,10 +551,21 @@ impl DockerEngineBackend { None, 64 * 1024, ) - .await?; + .await + .map_err(|_| { + HelperFailure::unanswered( + "start", + fail("Claude candidate reader start unanswered"), + ) + })?; if started.status != 204 { - return Err(fail("Claude candidate reader start failed")); + return Err(HelperFailure::answered( + "start", + started.status, + fail("Claude candidate reader start failed"), + )); } + step.enter("read"); let waited = self .request_bounded( "POST", @@ -494,13 +577,21 @@ impl DockerEngineBackend { let value: Value = serde_json::from_slice(&waited.body) .map_err(|_| fail("Claude candidate reader completion invalid"))?; if waited.status != 200 || value["StatusCode"] != 0 { - return Err(fail( - "Claude candidate reader did not return complete evidence", - )); + return Err( + fail("Claude candidate reader did not return complete evidence").into(), + ); } - Ok::<_, BackendError>(()) + Ok::<_, HelperFailure>(()) }; - let (reply, ()) = tokio::try_join!(exchange(&listener, &raw), supervision)?; + let (reply, ()) = tokio::try_join!( + async { + let reply = exchange(&listener, &raw).await?; + step.enter("exit"); + Ok::<_, HelperFailure>(reply) + }, + supervision + )?; + step.enter("verify"); let source_raw = bounded_json_bytes(&request.observation_request, MAX_REQUEST_BYTES) .map_err(|_| fail("Claude candidate source metadata exceeds profile"))?; reply @@ -520,7 +611,7 @@ impl DockerEngineBackend { .map_err(|_| fail("Claude candidate reader artifact unavailable"))? != digest { - return Err(fail("Claude candidate actual namespace changed")); + return Err(fail("Claude candidate actual namespace changed").into()); } match owner { ClaudeSourceOwner::Initial { @@ -528,7 +619,7 @@ impl DockerEngineBackend { observation, } => { if self.observe_initial_volume(reservation).await? != *observation { - return Err(fail("Claude candidate creation absence changed")); + return Err(fail("Claude candidate creation absence changed").into()); } } ClaudeSourceOwner::Retained { handle, .. } => { @@ -539,7 +630,8 @@ impl DockerEngineBackend { { return Err(fail( "Claude retained HOME metadata changed during candidate projection", - )); + ) + .into()); } qualify_retained_claude_stage( prior_stage @@ -556,42 +648,28 @@ impl DockerEngineBackend { let result = tokio::time::timeout_at(deadline, operation) .await .unwrap_or_else(|_| { - Err(fail( - "Claude candidate reader timed out; retain inputs and reprepare", + Err(HelperFailure::timed_out( + fail("Claude candidate reader timed out; retain inputs and reprepare"), + &step, )) }); - // One cleanup reserve covers both helpers and the ordinary backing. - // Any uncertain helper removal prevents deleting its still-mounted HOME. - let cleanup = async { - if let Some((helper, body)) = mountpoint_helper.as_ref() { - self.cleanup_provider_mountpoint_helper(helper, body) - .await?; - mountpoint_helper = None; - } - if attempted { - self.cleanup_claude_candidate_reader(&name, expected.as_ref().unwrap()) - .await?; - attempted = false; - } - if let Some(backing) = scratch.as_mut() { - self.cleanup_claude_candidate_scratch(backing).await?; - } - if let Some(view) = input_view.as_mut() { - view.cleanup()?; - } - if let Some(view) = retained_view.as_mut() { - view.cleanup()?; - } - Ok::<_, BackendError>(()) + // One release covers both helpers and the ordinary backing. Any + // uncertain helper removal prevents deleting its still-mounted HOME. + let objects = CandidateObjects { + mountpoints: mountpoint_helper, + reader, + scratch, + views: input_view.into_iter().chain(retained_view).collect(), + directory, }; - tokio::time::timeout(Duration::from_secs(5), cleanup) - .await - .map_err(|_| fail("Claude candidate owned cleanup timed out"))??; + let result = self + .release_helper_objects("claude_candidate_reader", objects, deadlines, result) + .await; tracing::debug!(event="provider.inputs.claude_candidate_helper.settled",soul_id=%request.soul_id,succeeded=result.is_ok(),"Claude candidate reader settled after owned cleanup"); result } - async fn cleanup_claude_candidate_reader( + pub(super) async fn cleanup_claude_candidate_reader( &self, name: &str, expected: &Value, @@ -644,9 +722,9 @@ impl DockerEngineBackend { async fn create_claude_candidate_scratch( &self, scratch: &mut CandidateScratch, - ) -> Result<(), BackendError> { + ) -> Result<(), HelperFailure> { if scratch.attempted || self.daemon_id().await? != scratch.daemon { - return Err(fail("Claude candidate scratch namespace changed")); + return Err(fail("Claude candidate scratch namespace changed").into()); } let absent = self .request_bounded( @@ -663,17 +741,36 @@ impl DockerEngineBackend { "Claude candidate scratch name is already present" } else { "Claude candidate scratch absence could not be confirmed" - })); + }) + .into()); } // Retain ownership on a lost acknowledgement. A conflict is foreign. scratch.attempted = true; - let created = self.request_bounded("POST", &format!("{DOCKER_API}/volumes/create"), Some(&json!({"Name":scratch.name,"Driver":"local","Labels":scratch.labels,"DriverOpts":{}})), 64 * 1024).await?; + let body = + json!({"Name":scratch.name,"Driver":"local","Labels":scratch.labels,"DriverOpts":{}}); + let created = scratch + .create + .send( + self, + &scratch.name, + format!("{DOCKER_API}/volumes/create"), + body, + ) + .await + .map_err(|_| { + HelperFailure::unanswered( + "create", + fail("Claude candidate scratch creation acknowledgement unavailable"), + ) + })?; if created.status == 409 { scratch.attempted = false; } if !matches!(created.status, 200 | 201) { - return Err(fail( - "Claude candidate scratch creation acknowledgement unavailable", + return Err(HelperFailure::answered( + "create", + created.status, + fail("Claude candidate scratch creation acknowledgement unavailable"), )); } let inspected = self @@ -690,7 +787,7 @@ impl DockerEngineBackend { || !scratch.owns(&actual) || self.daemon_id().await? != scratch.daemon { - return Err(fail("Claude candidate ordinary scratch ownership changed")); + return Err(fail("Claude candidate ordinary scratch ownership changed").into()); } Ok(()) } @@ -1178,13 +1275,38 @@ async fn exchange(listener: &UnixListener, raw: &[u8]) -> Result Result { + let helper = if request.candidate.is_some() { + "claude_candidate_reader" + } else { + "claude_source_reader" + }; + let context = json!({"soulId":request.soul_id,"provider":request.provider, + "requestId":request.observation_request.request_id}); + self.retry_helper(policy, helper, context, until, |deadlines| { + self.read_claude_source_helper(owner, request, test_run_id, deadlines) + }) + .await + } + + pub(super) async fn read_claude_source_helper( + &self, + owner: &ClaudeSourceOwner, + request: &ObserveClaudeSourcesRequest, + test_run_id: &str, + deadlines: HelperDeadlines, + ) -> Result { request .validate() .map_err(|_| fail("Claude source request invalid"))?; @@ -1215,7 +1337,7 @@ impl DockerEngineBackend { request, &root, test_run_id, - budget, + deadlines, prior_stage.as_deref(), ) .await; @@ -1224,7 +1346,7 @@ impl DockerEngineBackend { owner, request, test_run_id, - budget, + deadlines, prior_stage.as_deref(), ) .await @@ -1235,13 +1357,13 @@ impl DockerEngineBackend { owner: &ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, test_run_id: &str, - budget: Duration, + deadlines: HelperDeadlines, prior_stage: Option<&Path>, - ) -> Result { + ) -> Result { let name = format!("freshell-claude-source-{}", uuid::Uuid::new_v4()); - let mut attempted = false; - let mut expected = None::; + let mut helper = None::; let mut directory = None::; + let step = AttemptStep::new(); let operation = async { let mut retained_volume = None::; let (daemon, rootless) = self @@ -1249,9 +1371,9 @@ impl DockerEngineBackend { .await .map_err(|_| fail("Claude source daemon unavailable"))?; if !rootless { - return Err(fail( - "Claude source reader requires the qualified rootless daemon", - )); + return Err( + fail("Claude source reader requires the qualified rootless daemon").into(), + ); } let (installation, soul, image, binary, retained) = match owner { ClaudeSourceOwner::Initial { @@ -1261,7 +1383,7 @@ impl DockerEngineBackend { if reservation.provider != request.provider || self.observe_initial_volume(reservation).await? != *observation { - return Err(fail("Claude source creation absence changed")); + return Err(fail("Claude source creation absence changed").into()); } ( &reservation.installation_id, @@ -1297,7 +1419,8 @@ impl DockerEngineBackend { { return Err(fail( "Claude source requires its exact quiescent retained owner", - )); + ) + .into()); } let response = self .request_bounded( @@ -1313,7 +1436,7 @@ impl DockerEngineBackend { || volume["Name"] != handle.provider_volume_name() || volume["Driver"] != "local" { - return Err(fail("Claude retained HOME is unavailable; preserve native stores and repair the source")); + return Err(fail("Claude retained HOME is unavailable; preserve native stores and repair the source").into()); } retained_volume = Some(volume); ( @@ -1326,7 +1449,7 @@ impl DockerEngineBackend { } }; if soul != &request.soul_id { - return Err(fail("Claude source soul ownership changed")); + return Err(fail("Claude source soul ownership changed").into()); } let binary = canonical_regular_file(binary) .map_err(|_| fail("Claude source reader artifact unavailable"))?; @@ -1334,7 +1457,7 @@ impl DockerEngineBackend { .map_err(|_| fail("Claude source reader artifact changed"))?; if let ClaudeSourceOwner::Initial { reservation, .. } = owner { if digest != reservation.binary_sha256 { - return Err(fail("Claude reserved reader artifact changed")); + return Err(fail("Claude reserved reader artifact changed").into()); } } let workspace = canonical_dir(Path::new(&request.workspace_path)) @@ -1345,7 +1468,7 @@ impl DockerEngineBackend { || cwd != request.observation_request.cwd || !cwd.starts_with(&workspace) { - return Err(fail("Claude source workload topology changed")); + return Err(fail("Claude source workload topology changed").into()); } let common = request .git_common_dir @@ -1379,7 +1502,7 @@ impl DockerEngineBackend { let tmpfs = if let Some(volume) = retained { mounts.push(json!({"Type":"volume","Source":volume,"Target":"/home/freshell/provider","ReadOnly":true,"VolumeOptions":{"NoCopy":true}})); let ClaudeSourceOwner::Retained { handle, .. } = owner else { - return Err(fail("Claude retained ordinary owner unavailable")); + return Err(fail("Claude retained ordinary owner unavailable").into()); }; let ordinary = docker::ordinary_claude_candidate_mounts( &workspace, @@ -1415,22 +1538,27 @@ impl DockerEngineBackend { MAX_REQUEST_BYTES, ) .map_err(|_| fail("Claude source request exceeds metadata profile"))?; - expected = Some(body.clone()); - attempted = true; - let created = self - .request_bounded( - "POST", - &format!("{DOCKER_API}/containers/create?name={name}"), - Some(&body), - 64 * 1024, - ) + step.enter("create"); + let created = helper + .insert(HelperContainer::new( + HelperKind::ClaudeSource, + name.clone(), + body.clone(), + )) + .create(self) .await - .map_err(|_| fail("Claude source helper creation acknowledgement unavailable"))?; - if created.status == 409 { - attempted = false; - } + .map_err(|_| { + HelperFailure::unanswered( + "create", + fail("Claude source helper creation acknowledgement unavailable"), + ) + })?; if created.status != 201 { - return Err(fail("Claude source helper creation failed")); + return Err(HelperFailure::answered( + "create", + created.status, + fail("Claude source helper creation failed"), + )); } let created: Value = serde_json::from_slice(&created.body) .map_err(|_| fail("Claude source helper creation invalid"))?; @@ -1453,14 +1581,15 @@ impl DockerEngineBackend { || !owned_helper(&actual, &body) || actual["State"]["Running"] != false { - return Err(fail( - "Claude source helper namespace/topology ownership changed", - )); + return Err( + fail("Claude source helper namespace/topology ownership changed").into(), + ); } if let Some(volume) = &retained_volume { verify_retained_claude_mount_reports(&actual, &body, volume)?; } let supervision = async { + step.enter("start"); let started = self .request_bounded( "POST", @@ -1468,10 +1597,21 @@ impl DockerEngineBackend { None, 64 * 1024, ) - .await?; + .await + .map_err(|_| { + HelperFailure::unanswered( + "start", + fail("Claude source helper start unanswered"), + ) + })?; if started.status != 204 { - return Err(fail("Claude source helper start failed")); + return Err(HelperFailure::answered( + "start", + started.status, + fail("Claude source helper start failed"), + )); } + step.enter("read"); let waited = self .request_bounded( "POST", @@ -1483,13 +1623,21 @@ impl DockerEngineBackend { let value: Value = serde_json::from_slice(&waited.body) .map_err(|_| fail("Claude source helper completion invalid"))?; if waited.status != 200 || value["StatusCode"] != 0 { - return Err(fail( - "Claude source helper did not return complete evidence", - )); + return Err( + fail("Claude source helper did not return complete evidence").into(), + ); } - Ok::<_, BackendError>(()) + Ok::<_, HelperFailure>(()) }; - let (reply, ()) = tokio::try_join!(exchange(&listener, &raw), supervision)?; + let (reply, ()) = tokio::try_join!( + async { + let reply = exchange(&listener, &raw).await?; + step.enter("exit"); + Ok::<_, HelperFailure>(reply) + }, + supervision + )?; + step.enter("verify"); reply .validate_for_request( &request.observation_request, @@ -1503,14 +1651,14 @@ impl DockerEngineBackend { reply.observation.egid, ) != (65534, 65534, 0, 0) { - return Err(fail("Claude source native reader identity changed")); + return Err(fail("Claude source native reader identity changed").into()); } if self.daemon_id().await? != daemon || crate::registry::binary_digest(&binary) .map_err(|_| fail("Claude source reader artifact unavailable"))? != digest { - return Err(fail("Claude source daemon or reader artifact changed")); + return Err(fail("Claude source daemon or reader artifact changed").into()); } if let ClaudeSourceOwner::Initial { reservation, @@ -1518,76 +1666,86 @@ impl DockerEngineBackend { } = owner { if self.observe_initial_volume(reservation).await? != *observation { - return Err(fail("Claude source initial HOME presence changed")); + return Err(fail("Claude source initial HOME presence changed").into()); } } Ok(reply) }; - let result = tokio::time::timeout(budget, operation) + let result = tokio::time::timeout_at(deadlines.operation, operation) .await .unwrap_or_else(|_| { - Err(fail( - "Claude source helper timed out; retain inputs and reprepare", + Err(HelperFailure::timed_out( + fail("Claude source helper timed out; retain inputs and reprepare"), + &step, )) }); - if attempted { - let cleanup = async { - let inspection = self - .request_bounded( - "GET", - &format!("{DOCKER_API}/containers/{name}/json"), - None, - 1024 * 1024, - ) - .await?; - if inspection.status == 404 { - return Ok(()); - } - let actual: Value = serde_json::from_slice(&inspection.body) - .map_err(|_| fail("Claude source cleanup ownership invalid"))?; - if inspection.status != 200 - || !expected - .as_ref() - .is_some_and(|expected| owned_helper(&actual, expected)) - { - return Err(fail( - "Claude source cleanup ownership changed; no provider workload was stopped", - )); - } - let removed = self - .request_bounded( - "DELETE", - &format!("{DOCKER_API}/containers/{name}?force=1"), - None, - 64 * 1024, - ) - .await?; - if !matches!(removed.status, 204 | 404) { - return Err(fail("Claude source helper cleanup failed")); - } - let absent = self - .request_bounded( - "GET", - &format!("{DOCKER_API}/containers/{name}/json"), - None, - 64 * 1024, - ) - .await?; - if absent.status != 404 { - return Err(fail("Claude source helper cleanup not confirmed")); - } - Ok(()) - }; - tokio::time::timeout(Duration::from_secs(5), cleanup) - .await - .map_err(|_| fail("Claude source helper cleanup timed out"))??; - } + let result = match helper { + Some(helper) => { + self.release_helper_objects("claude_source_reader", helper, deadlines, result) + .await + } + None => result, + }; tracing::debug!(event="provider.inputs.native_source_helper.settled",soul_id=%request.soul_id,helper=%name,succeeded=result.is_ok(), "Claude source helper settled after owned cleanup"); result } + + /// Remove an owned Claude source helper by exact name and confirm its + /// absence; a changed helper is never removed. + pub(super) async fn cleanup_claude_source_helper( + &self, + name: &str, + expected: &Value, + ) -> Result<(), BackendError> { + let inspection = self + .request_bounded( + "GET", + &format!("{DOCKER_API}/containers/{name}/json"), + None, + 1024 * 1024, + ) + .await?; + if inspection.status == 404 { + return Ok(()); + } + let actual: Value = serde_json::from_slice(&inspection.body) + .map_err(|_| fail("Claude source cleanup ownership invalid"))?; + if inspection.status != 200 || !owned_helper(&actual, expected) { + return Err(fail( + "Claude source cleanup ownership changed; no provider workload was stopped", + )); + } + let removed = self + .request_bounded( + "DELETE", + &format!("{DOCKER_API}/containers/{name}?force=1"), + None, + 64 * 1024, + ) + .await?; + if !matches!(removed.status, 204 | 404) { + return Err(fail("Claude source helper cleanup failed")); + } + let absent = self + .request_bounded( + "GET", + &format!("{DOCKER_API}/containers/{name}/json"), + None, + 64 * 1024, + ) + .await?; + if absent.status != 404 { + return Err(fail("Claude source helper cleanup not confirmed")); + } + Ok(()) + } } #[cfg(test)] #[path = "claude_source_observation_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "claude_source_reader_retry_tests.rs"] +mod reader_retry_tests; diff --git a/crates/freshell-supervisor/src/backend/claude_source_observation_tests.rs b/crates/freshell-supervisor/src/backend/claude_source_observation_tests.rs index cdf6ffc86..bd12d2725 100644 --- a/crates/freshell-supervisor/src/backend/claude_source_observation_tests.rs +++ b/crates/freshell-supervisor/src/backend/claude_source_observation_tests.rs @@ -4,6 +4,7 @@ use freshell_runtime_protocol::{ ClaudeSourceAuthority, InitialAuthVolumeObservation, ReserveProviderInputsRequest, }; use std::sync::{Arc, Mutex}; +use std::time::Duration; use tokio::{net::UnixListener, task::JoinSet}; #[tokio::test] @@ -26,6 +27,7 @@ async fn provider_claude_candidate_scratch_owns_disk_backing_and_confirmed_clean labels: labels.clone(), daemon: daemon.clone(), attempted: false, + create: Default::default(), }; let present = Arc::new(Mutex::new(mode == "existing")); let calls = Arc::new(Mutex::new(Vec::<(String, Value)>::new())); @@ -857,7 +859,7 @@ async fn qualify_transport(mode: &'static str) { &owner, &request, "synthetic-source-run", - Duration::from_secs(3), + HelperDeadlines::after(Duration::from_secs(3), Duration::from_secs(5)), Some(&stage), ) .await @@ -867,7 +869,7 @@ async fn qualify_transport(mode: &'static str) { &owner, &request, "synthetic-source-run", - Duration::from_millis(250), + HelperDeadlines::after(Duration::from_millis(250), Duration::from_secs(5)), ) .await }; @@ -990,6 +992,26 @@ async fn qualify_transport(mode: &'static str) { } } +/// The candidate reader's ordinary input roots (its mount skeletons), from the +/// reader create the synthetic daemon saw. +fn candidate_input_roots(seen: &[(String, Value)]) -> Vec { + seen.iter() + .filter(|(_, body)| body["Cmd"] == json!(["project-claude-candidate-private"])) + .flat_map(|(_, body)| { + body["HostConfig"]["Mounts"] + .as_array() + .unwrap() + .iter() + .filter(|mount| { + mount["Target"] == "/run/freshell-candidate-inputs" + || mount["Target"] == "/run/freshell-retained-inputs" + }) + .map(|mount| PathBuf::from(mount["Source"].as_str().unwrap())) + .collect::>() + }) + .collect() +} + async fn qualify_candidate_helper_transport( owner: &ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, @@ -1005,6 +1027,9 @@ async fn qualify_candidate_helper_transport( "volume-still-present", "wrong-mount-report", "wrong-resource-subpath", + "late-mountpoints-create", + "slow-reader-delete", + "late-scratch-create", ]; if matches!(owner, ClaudeSourceOwner::Retained { .. }) { modes.extend([ @@ -1064,6 +1089,9 @@ async fn qualify_candidate_helper_transport( let listener = UnixListener::bind(owned.path().join("docker.sock")).unwrap(); let (stop, mut stopped) = tokio::sync::oneshot::channel::<()>(); let (seen, bodies, backing) = (calls.clone(), containers.clone(), volume.clone()); + // Docker has made the scratch volume whose create it answers late. + let scratch_made = Arc::new(tokio::sync::Notify::new()); + let made = scratch_made.clone(); let expected_scratch = format!( "freshell-claude-candidate-{}", request.observation_request.request_id @@ -1078,10 +1106,11 @@ async fn qualify_candidate_helper_transport( _=&mut stopped=>break, accepted=listener.accept()=>{ let (mut stream,_)=accepted.unwrap(); - let (seen,bodies,backing,daemon,private_jobs,expected_scratch,retained_name)=(seen.clone(),bodies.clone(),backing.clone(),daemon.clone(),private_jobs.clone(),expected_scratch.clone(),retained_name.clone()); + let (seen,bodies,backing,daemon,private_jobs,expected_scratch,retained_name,made)=(seen.clone(),bodies.clone(),backing.clone(),daemon.clone(),private_jobs.clone(),expected_scratch.clone(),retained_name.clone(),made.clone()); jobs.spawn(async move { let mut header=Vec::new();let mut byte=[0]; - while !header.ends_with(b"\r\n\r\n") {stream.read_exact(&mut byte).await.unwrap();header.push(byte[0]);assert!(header.len()<=32*1024);} + // A caller may abandon a request it no longer waits for. + while !header.ends_with(b"\r\n\r\n") {if stream.read_exact(&mut byte).await.is_err() {return;}header.push(byte[0]);assert!(header.len()<=32*1024);} let header=String::from_utf8(header).unwrap();let line=header.lines().next().unwrap().to_string(); let length=header.lines().find_map(|line|line.strip_prefix("Content-Length: ")).unwrap().trim().parse::().unwrap(); assert!(length<=1024*1024);let mut raw=vec![0;length];stream.read_exact(&mut raw).await.unwrap(); @@ -1092,13 +1121,23 @@ async fn qualify_candidate_helper_transport( else if line.starts_with("POST") && line.contains("/volumes/create "){ assert_eq!(body["Name"],expected_scratch,"private candidate request must identify its exact scratch backing before creation"); if serde_json::from_value::>(body["Labels"].clone()).is_err() { (400,json!({"message":"volume labels require string metadata"})) } - else { let actual=json!({"Name":body["Name"],"Driver":"local","Options":{},"Labels":body["Labels"],"Mountpoint":"/synthetic/ordinary-volume"});*backing.lock().unwrap()=Some(actual.clone());(201,actual) } + else { + // A stalled daemon makes the volume only after the reader's deadline. + if mode=="late-scratch-create" {tokio::time::sleep(Duration::from_millis(900)).await;} + let actual=json!({"Name":body["Name"],"Driver":"local","Options":{},"Labels":body["Labels"],"Mountpoint":"/synthetic/ordinary-volume"});*backing.lock().unwrap()=Some(actual.clone());made.notify_one();(201,actual) } } else if line.starts_with("GET") && line.contains("/volumes/"){if retained_name.as_ref().is_some_and(|name|line.contains(name)) {(200,json!({"Name":retained_name,"Driver":"local","Options":{},"Mountpoint":"/synthetic/retained-home"}))} else {match backing.lock().unwrap().clone(){Some(v) if line.contains(v["Name"].as_str().unwrap())=>(200,v),_=>(404,json!({}))}}} else if line.starts_with("DELETE") && line.contains("/volumes/"){assert!(bodies.lock().unwrap().is_empty(),"candidate helpers must be gone before volume deletion");if mode!="volume-still-present" {*backing.lock().unwrap()=None;}(204,json!({}))} else if line.contains("/containers/create?"){ if body["Cmd"]==json!(["project-claude-candidate-private"]) {let root=body["HostConfig"]["Mounts"].as_array().unwrap().iter().find(|m|m["Target"]=="/run/freshell-candidate-inputs").unwrap()["Source"].as_str().unwrap();assert!(!Path::new(root).join(".freshell/mcp-capability.json").exists(),"ordinary candidate input root must not expose capability credentials");} - let name=line.split("name=").nth(1).unwrap().split_whitespace().next().unwrap().to_string();let id=if body["Cmd"]==json!(["prepare-provider-mountpoints"]) {"a".repeat(64)} else {"b".repeat(64)};bodies.lock().unwrap().insert(name,json!({"Id":id,"Config":body,"HostConfig":body["HostConfig"],"State":{"Running":false}}));(if mode=="lost-projection-ack" && body["Cmd"]==json!(["project-claude-candidate-private"]) {503}else{201},json!({"Id":id}))} + let name=line.split("name=").nth(1).unwrap().split_whitespace().next().unwrap().to_string();let id=if body["Cmd"]==json!(["prepare-provider-mountpoints"]) {"a".repeat(64)} else {"b".repeat(64)};let record=json!({"Id":id,"Config":body,"HostConfig":body["HostConfig"],"State":{"Running":false}}); + if mode=="late-mountpoints-create" && body["Cmd"]==json!(["prepare-provider-mountpoints"]) { + // Docker makes this helper after the reader's deadline, + // whether or not the reader still waits for the answer. + let (late,late_name)=(bodies.clone(),name.clone()); + private_jobs.lock().unwrap().push(tokio::spawn(async move {tokio::time::sleep(Duration::from_millis(900)).await;late.lock().unwrap().insert(late_name,record);})); + while !bodies.lock().unwrap().contains_key(&name) {tokio::time::sleep(Duration::from_millis(10)).await;} + } else {bodies.lock().unwrap().insert(name,record);}(if mode=="lost-projection-ack" && body["Cmd"]==json!(["project-claude-candidate-private"]) {503}else{201},json!({"Id":id}))} else if line.starts_with("GET") && line.contains("/containers/"){let key=line.split("/containers/").nth(1).unwrap().split('/').next().unwrap();let found=bodies.lock().unwrap().iter().find(|(name,v)|*name==key || v["Id"]==key).map(|(_,v)|v.clone());match found {Some(mut v)=>{ let reports=v["HostConfig"]["Mounts"].as_array().unwrap().iter().map(|m|json!({"Type":m["Type"],"Destination":m["Target"],"RW":!m["ReadOnly"].as_bool().unwrap(),"Source":if m["Type"]=="volume" {json!(if m["Target"]=="/run/freshell-retained-home" {"/synthetic/retained-home"} else {"/synthetic/ordinary-volume"})} else {m["Source"].clone()},"Name":if m["Type"]=="volume" {m["Source"].clone()} else {Value::Null}})).collect::>(); v["Mounts"]=json!(reports); @@ -1111,7 +1150,12 @@ async fn qualify_candidate_helper_transport( if mode=="writable-retained-home" {v["Mounts"].as_array_mut().unwrap().iter_mut().find(|m|m["Destination"]=="/run/freshell-retained-home").unwrap()["RW"]=json!(true);} if mode=="wrong-retained-volume" {v["Mounts"].as_array_mut().unwrap().iter_mut().find(|m|m["Destination"]=="/run/freshell-retained-home").unwrap()["Name"]=json!("foreign-provider-home");} } (200,v)},None=>(404,json!({}))}} - else if line.starts_with("DELETE") && line.contains("/containers/"){let key=line.split("/containers/").nth(1).unwrap().split('?').next().unwrap();let name=bodies.lock().unwrap().iter().find(|(name,v)|*name==key || v["Id"]==key).map(|(name,_)|name.clone());if let Some(name)=name{bodies.lock().unwrap().remove(&name);}(204,json!({}))} + else if line.starts_with("DELETE") && line.contains("/containers/"){let key=line.split("/containers/").nth(1).unwrap().split('?').next().unwrap();let name=bodies.lock().unwrap().iter().find(|(name,v)|*name==key || v["Id"]==key).map(|(name,_)|name.clone());if let Some(name)=name{bodies.lock().unwrap().remove(&name);} + // Docker removes the reader but answers only after the + // caller's cleanup window, as a force-remove that waits + // on a host-wide writeback does. + if mode=="slow-reader-delete" && key=="b".repeat(64) {tokio::time::sleep(Duration::from_millis(5_500)).await;} + (204,json!({}))} else if line.contains("/start "){let id=line.split("/containers/").nth(1).unwrap().split('/').next().unwrap();let v=bodies.lock().unwrap().values().find(|v|v["Id"]==id).unwrap().clone();private_jobs.lock().unwrap().push(tokio::spawn(async move { let is_mountpoint=v["Config"]["Cmd"]==json!(["prepare-provider-mountpoints"]); let target=if is_mountpoint {"/run/freshell-input-exchange"} else {"/run/freshell-source-exchange"}; @@ -1169,7 +1213,10 @@ async fn qualify_candidate_helper_transport( } }); let backend = DockerEngineBackend::new(owned.path().join("docker.sock")); - let budget = if mode == "projection-timeout" { + let budget = if matches!( + mode, + "projection-timeout" | "late-mountpoints-create" | "late-scratch-create" + ) { Duration::from_millis(250) } else { Duration::from_secs(3) @@ -1180,12 +1227,36 @@ async fn qualify_candidate_helper_transport( request, stage, "synthetic-source-run", - budget, + HelperDeadlines::after(budget, Duration::from_secs(5)), retained_name.as_ref().map(|_| prior_stage.as_path()), ) .await; + if mode == "late-scratch-create" { + // Let Docker finish the volume the read gave up on, then give a + // cleanup that settled the create time to remove it. + tokio::time::timeout(Duration::from_secs(10), scratch_made.notified()) + .await + .expect("the synthetic daemon makes the late scratch volume"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while volume.lock().unwrap().is_some() && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(20)).await; + } + } + if mode == "slow-reader-delete" { + // The reader's removal outlived the cleanup window, so the + // attempt failed; what it still owned is removed in the + // background, the scratch volume after the helpers. + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + while volume.lock().unwrap().is_some() && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(50)).await; + } + let roots = candidate_input_roots(&calls.lock().unwrap()); + while roots.iter().any(|root| root.exists()) && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(50)).await; + } + } let _ = stop.send(()); - tokio::time::timeout(Duration::from_secs(3), server) + tokio::time::timeout(Duration::from_secs(10), server) .await .unwrap() .unwrap(); diff --git a/crates/freshell-supervisor/src/backend/claude_source_reader_retry_tests.rs b/crates/freshell-supervisor/src/backend/claude_source_reader_retry_tests.rs new file mode 100644 index 000000000..11fefa0ce --- /dev/null +++ b/crates/freshell-supervisor/src/backend/claude_source_reader_retry_tests.rs @@ -0,0 +1,857 @@ +//! The Claude source reader against a synthetic Docker daemon whose helper +//! creates can be refused or answered late, as a stalled daemon answers them. +use super::*; +use freshell_platform::claude_inputs::observation::{LocalDecision, NativeObservation}; +use freshell_runtime_protocol::{InitialAuthVolumeObservation, ReserveProviderInputsRequest}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::{net::UnixListener, task::JoinSet}; + +/// How the synthetic daemon answers one helper create, in request order. +#[derive(Clone, Copy, Debug)] +enum Create { + /// Create the helper and answer at once. + Now, + /// Answer with this status and create nothing. + Refuse(u16), + /// Create the helper and answer only after this long, as a create that + /// waits on a host-wide writeback answers. + Late(Duration), +} + +/// The daemon's view: what was asked for and what exists. +#[derive(Default)] +struct Inventory { + creates: Vec, + containers: BTreeMap, + removed: Vec, +} + +/// What the synthetic reader does once Docker starts it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Reader { + /// Reply over the private exchange and exit with this status. + Exits(i64), + /// Accept the exchange, never reply and never exit. + Hangs, +} + +struct ReaderDaemon { + inventory: Arc>, + late: Arc>>>, + stop: Option>, + server: Option>, +} + +impl ReaderDaemon { + fn start(socket: &Path, daemon: DockerDaemonId, script: Vec, exit_code: i64) -> Self { + Self::with_reader( + socket, + daemon, + script, + Reader::Exits(exit_code), + Duration::ZERO, + ) + } + + /// `delete_delay`: how long after removing a container Docker answers + /// the DELETE, as a force-remove that waits on a flush does. + fn with_reader( + socket: &Path, + daemon: DockerDaemonId, + script: Vec, + reader: Reader, + delete_delay: Duration, + ) -> Self { + let inventory = Arc::new(Mutex::new(Inventory::default())); + let late = Arc::new(Mutex::new(Vec::new())); + let listener = UnixListener::bind(socket).unwrap(); + let (stop, mut stopped) = tokio::sync::oneshot::channel::<()>(); + let (state, late_jobs) = (inventory.clone(), late.clone()); + let server = tokio::spawn(async move { + let mut jobs = JoinSet::new(); + loop { + tokio::select! { + _ = &mut stopped => break, + accepted = listener.accept() => { + let (stream, _) = accepted.unwrap(); + let (state, late_jobs, daemon, script) = + (state.clone(), late_jobs.clone(), daemon.clone(), script.clone()); + jobs.spawn(answer(stream, state, late_jobs, daemon, script, reader, delete_delay)); + } + result = jobs.join_next(), if !jobs.is_empty() => { result.unwrap().unwrap(); } + } + } + jobs.shutdown().await; + }); + Self { + inventory, + late, + stop: Some(stop), + server: Some(server), + } + } + + /// Stop answering, after every late create has created its helper. + async fn finish(&mut self) { + let late = std::mem::take(&mut *self.late.lock().unwrap()); + for job in late { + match tokio::time::timeout(Duration::from_secs(30), job).await { + Ok(joined) => joined.unwrap(), + // A hung reader holds its exchange until the daemon stops. + Err(_) => panic!("a late create never finished"), + } + } + let _ = self.stop.take().unwrap().send(()); + tokio::time::timeout(Duration::from_secs(5), self.server.take().unwrap()) + .await + .unwrap() + .unwrap(); + } + + /// Stop every hung reader the daemon is holding. + fn abort_readers(&self) { + for job in std::mem::take(&mut *self.late.lock().unwrap()) { + job.abort(); + } + } + + fn creates(&self) -> Vec { + self.inventory.lock().unwrap().creates.clone() + } + + fn live(&self) -> Vec { + self.inventory + .lock() + .unwrap() + .containers + .keys() + .cloned() + .collect() + } + + fn removed(&self) -> Vec { + self.inventory.lock().unwrap().removed.clone() + } +} + +async fn answer( + mut stream: UnixStream, + state: Arc>, + late: Arc>>>, + daemon: DockerDaemonId, + script: Vec, + reader: Reader, + delete_delay: Duration, +) { + let mut header = Vec::new(); + let mut byte = [0]; + while !header.ends_with(b"\r\n\r\n") { + // A caller may abandon a request it no longer waits for. + if stream.read_exact(&mut byte).await.is_err() { + return; + } + header.push(byte[0]); + } + let header = String::from_utf8(header).unwrap(); + let length = header + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + .unwrap() + .trim() + .parse::() + .unwrap(); + let mut raw = vec![0; length]; + if stream.read_exact(&mut raw).await.is_err() { + return; + } + let line = header.lines().next().unwrap().to_string(); + let key = |line: &str| { + line.split("/containers/") + .nth(1) + .unwrap() + .split(['/', '?', ' ']) + .next() + .unwrap() + .to_string() + }; + let (status, response) = if line.contains("/info ") { + ( + 200, + json!({"ID":daemon,"SecurityOptions":["name=rootless"]}), + ) + } else if line.starts_with("GET") && line.contains("/volumes/") { + (404, json!({})) + } else if line.starts_with("GET") && line.contains("/images/") { + (200, json!({"Config":{"Env":["PATH=/usr/bin"]}})) + } else if line.starts_with("POST") && line.contains("/containers/create?") { + let name = line + .split("name=") + .nth(1) + .unwrap() + .split_whitespace() + .next() + .unwrap() + .to_string(); + let body: Value = serde_json::from_slice(&raw).unwrap(); + let (index, id) = { + let mut inventory = state.lock().unwrap(); + inventory.creates.push(name.clone()); + let index = inventory.creates.len() - 1; + (index, format!("{:064x}", index + 1)) + }; + let container = json!({"Id":id,"Name":format!("/{name}"),"Config":body, + "HostConfig":body["HostConfig"],"Mounts":[],"State":{"Running":false}}); + match script.get(index).copied().unwrap_or(Create::Now) { + Create::Refuse(status) => (status, json!({"message":"synthetic refusal"})), + Create::Now => { + state.lock().unwrap().containers.insert(name, container); + (201, json!({"Id":id})) + } + Create::Late(delay) => { + // The daemon finishes this create even when its caller has + // stopped waiting for the answer. + let created = state.clone(); + let job = tokio::spawn(async move { + tokio::time::sleep(delay).await; + created.lock().unwrap().containers.insert(name, container); + }); + let settled = Arc::new(tokio::sync::Notify::new()); + let notify = settled.clone(); + let watcher = tokio::spawn(async move { + let _ = job.await; + notify.notify_one(); + }); + late.lock().unwrap().push(watcher); + settled.notified().await; + (201, json!({"Id":id})) + } + } + } else if line.starts_with("GET") && line.contains("/containers/") { + let key = key(&line); + let found = state + .lock() + .unwrap() + .containers + .iter() + .find(|(name, value)| **name == key || value["Id"] == key) + .map(|(_, value)| value.clone()); + match found { + Some(value) => (200, value), + None => (404, json!({})), + } + } else if line.starts_with("POST") && line.contains("/start ") { + let key = key(&line); + let body = state + .lock() + .unwrap() + .containers + .values() + .find(|value| value["Id"] == key) + .map(|value| value["Config"].clone()) + .unwrap(); + let exchange = body["HostConfig"]["Mounts"] + .as_array() + .unwrap() + .iter() + .find(|mount| mount["Target"] == "/run/freshell-source-exchange") + .unwrap()["Source"] + .as_str() + .unwrap() + .to_string(); + let mut channel = UnixStream::connect(Path::new(&exchange).join("reply.sock")) + .await + .unwrap(); + if reader == Reader::Hangs { + // Started promptly; the reader itself never answers. + late.lock().unwrap().push(tokio::spawn(async move { + let _held = channel; + tokio::time::sleep(Duration::from_secs(3600)).await; + })); + let response = "{}"; + let _ = stream + .write_all( + format!( + "HTTP/1.1 204 fixture\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}", + response.len() + ) + .as_bytes(), + ) + .await; + return; + } + let length = channel.read_u32().await.unwrap() as usize; + let mut raw = vec![0; length]; + channel.read_exact(&mut raw).await.unwrap(); + let request: freshell_platform::claude_inputs::observation::ObservationRequest = + serde_json::from_slice(&raw).unwrap(); + let reply = ObservationReply { + request_sha256: format!("{:x}", Sha256::digest(&raw)), + observation: NativeObservation { + version: 1, + native_source_runtime: request.native_source_runtime, + setting_sources: request.setting_sources, + uid: 65534, + euid: 65534, + gid: 0, + egid: 0, + supplementary_groups: vec![], + local_settings: vec![request.cwd.join(".claude/settings.local.json")], + cwd: request.cwd, + home: Some(request.home), + roots: None, + local_decision: LocalDecision::NoRoot, + root_uid: None, + git_uid: None, + claude_uid: None, + }, + }; + let raw = serde_json::to_vec(&reply).unwrap(); + channel.write_u32(raw.len() as u32).await.unwrap(); + channel.write_all(&raw).await.unwrap(); + channel.shutdown().await.unwrap(); + (204, json!({})) + } else if line.starts_with("POST") && line.contains("/wait?") { + match reader { + Reader::Exits(code) => (200, json!({"StatusCode":code})), + Reader::Hangs => { + // A running reader never reports an exit. + tokio::time::sleep(Duration::from_secs(3600)).await; + return; + } + } + } else if line.starts_with("DELETE") && line.contains("/containers/") { + let key = key(&line); + { + let mut inventory = state.lock().unwrap(); + let name = inventory + .containers + .iter() + .find(|(name, value)| **name == key || value["Id"] == key) + .map(|(name, _)| name.clone()); + if let Some(name) = name { + inventory.containers.remove(&name); + inventory.removed.push(name); + } + } + tokio::time::sleep(delete_delay).await; + (204, json!({})) + } else { + (500, json!({})) + }; + let response = response.to_string(); + let _ = stream + .write_all( + format!( + "HTTP/1.1 {status} fixture\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{response}", + response.len() + ) + .as_bytes(), + ) + .await; +} + +/// A first-creation Claude source request and the owner that authorizes it. +async fn initial_request( + root: &Path, + daemon: &DockerDaemonId, +) -> (ClaudeSourceOwner, ObserveClaudeSourcesRequest) { + let workspace = root.join("workspace"); + std::fs::create_dir(&workspace).unwrap(); + let binary = root.join("synthetic-host"); + std::fs::write(&binary, b"synthetic helper artifact").unwrap(); + let registry = crate::registry::Registry::open(root.join("registry"), None).unwrap(); + let soul = SoulId::new(); + let reservation = registry + .reserve_initial_inputs( + ReserveProviderInputsRequest { + expected_control_epoch: Some(registry.control_epoch()), + soul_id: soul.clone(), + provider: "claude".into(), + creation_seed_ref: "synthetic-seed".into(), + }, + format!("sha256:{}", "a".repeat(64)), + binary, + ) + .await + .unwrap(); + let request: ObserveClaudeSourcesRequest = serde_json::from_value(json!({ + "expectedControlEpoch":registry.control_epoch(), "soulId":soul, "provider":"claude", + "authority":{"kind":"initial_reservation","reservationId":reservation.id}, + "workspacePath":workspace,"gitCommonDir":null,"candidate":null, + "observationRequest":{"version":1,"requestId":uuid::Uuid::new_v4().to_string(), + "nativeSourceRuntime":"terminal","cwd":workspace,"home":"/home/freshell/provider", + "settingSources":["local"],"selectorManifestSha256":"b".repeat(64),"planRevision":0} + })) + .unwrap(); + let owner = ClaudeSourceOwner::Initial { + reservation, + observation: InitialAuthVolumeObservation::Absent { + daemon_id: daemon.clone(), + }, + }; + (owner, request) +} + +fn lifecycle_events(path: &Path) -> Vec { + std::fs::read_to_string(path) + .unwrap_or_default() + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect() +} + +fn named<'a>(events: &'a [Value], name: &str) -> Vec<&'a Value> { + events + .iter() + .filter(|event| event["event"] == name) + .collect() +} + +/// The end the supervisor gives a read with no recovery scope around it. +fn production_until() -> tokio::time::Instant { + tokio::time::Instant::now() + helper_lifecycle::CLAUDE_SOURCE_READER_RETRY.total +} + +/// A caller's end far beyond any test policy's total. +fn far() -> tokio::time::Instant { + tokio::time::Instant::now() + Duration::from_secs(3600) +} + +/// Wait until the background reaper reports how its removal ended. +async fn reaper_outcome(path: &Path) -> Vec { + let deadline = tokio::time::Instant::now() + Duration::from_secs(60); + loop { + let events = lifecycle_events(path); + let settled = events.iter().any(|event| { + event["event"] == "supervisor.helper.cleanup.removed" + || event["event"] == "supervisor.helper.cleanup.unconfirmed" + }); + if settled || tokio::time::Instant::now() >= deadline { + return events; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +#[tokio::test] +async fn claude_source_reader_retries_a_create_the_daemon_refused_transiently() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Refuse(500), Create::Now], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = tokio::time::timeout( + Duration::from_secs(20), + backend.observe_claude_sources(&owner, &request, "synthetic-run", production_until()), + ) + .await + .unwrap(); + daemon.finish().await; + assert!(result.is_ok(), "a refused create is retried: {result:?}"); + assert_eq!(daemon.creates().len(), 2, "one retry after the refusal"); + assert!(daemon.live().is_empty(), "the successful helper is removed"); + let events = lifecycle_events(&log); + let retries = named(&events, "supervisor.helper.retry"); + assert_eq!(retries.len(), 1, "{events:?}"); + let retry = &retries[0]["data"]; + assert_eq!(retry["severity"], "warn"); + assert_eq!(retry["helper"], "claude_source_reader"); + assert_eq!(retry["cause"], "refused"); + assert_eq!(retry["step"], "create"); + assert_eq!(retry["attempt"], 1); + assert_eq!( + retry["requestId"], + json!(request.observation_request.request_id) + ); + assert_eq!(named(&events, "supervisor.helper.recovered").len(), 1); + assert!( + named(&events, "supervisor.helper.create_slow").is_empty(), + "prompt creates are not reported as slow" + ); +} + +#[tokio::test] +async fn claude_source_reader_keeps_final_failures_final() { + for (script, exit_code) in [ + (vec![Create::Refuse(409)], 0), + (vec![Create::Refuse(400)], 0), + (vec![Create::Now], 1), + ] { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + script.clone(), + exit_code, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = tokio::time::timeout( + Duration::from_secs(20), + backend.observe_claude_sources(&owner, &request, "synthetic-run", production_until()), + ) + .await + .unwrap(); + daemon.finish().await; + assert!(result.is_err(), "{script:?}/{exit_code}: {result:?}"); + assert_eq!( + daemon.creates().len(), + 1, + "{script:?}/{exit_code}: a final failure is never retried" + ); + assert!(daemon.live().is_empty(), "{script:?}/{exit_code}"); + assert!(named(&lifecycle_events(&log), "supervisor.helper.retry").is_empty()); + } +} + +#[tokio::test] +async fn claude_source_reader_removes_a_helper_whose_create_answered_after_its_deadline() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Late(Duration::from_secs(15))], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + // Seconds-long windows: the create must be sent inside the first, and + // answered well after both. + let result = backend + .read_claude_source_helper( + &owner, + &request, + "synthetic-run", + HelperDeadlines::after(Duration::from_secs(5), Duration::from_secs(2)), + ) + .await; + assert!(result.is_err(), "the helper's deadline passed: {result:?}"); + let creates = daemon.creates(); + assert_eq!(creates.len(), 1); + // Docker creates the helper after the reader stopped waiting. Nothing + // else will ever ask for it, so the reader's own cleanup must remove it. + let events = reaper_outcome(&log).await; + daemon.finish().await; + assert_eq!(daemon.removed(), creates, "the late helper is removed"); + assert!(daemon.live().is_empty(), "no helper is left behind"); + let deferred = named(&events, "supervisor.helper.cleanup.deferred"); + assert_eq!(deferred.len(), 1, "{events:?}"); + assert_eq!(deferred[0]["data"]["severity"], "warn"); + assert_eq!(deferred[0]["data"]["reason"], "create_unanswered"); + assert_eq!(deferred[0]["data"]["names"], json!(creates)); + let removed = named(&events, "supervisor.helper.cleanup.removed"); + assert_eq!(removed.len(), 1, "{events:?}"); + assert_eq!(removed[0]["data"]["severity"], "info"); +} + +/// A small policy with the production shape, so stalls resolve in tests. +/// Its windows are seconds long: on this 96-core host at a load of 800 a +/// healthy synthetic read can take more than a second, and scheduling delay +/// must never decide which attempt times out. +const QUICK: helper_lifecycle::HelperRetryPolicy = helper_lifecycle::HelperRetryPolicy { + attempt: Duration::from_secs(5), + total: Duration::from_secs(60), + cleanup: Duration::from_secs(3), + backoff: [Duration::from_millis(50), Duration::from_millis(100)], + min_attempt: Duration::from_secs(3), +}; + +#[tokio::test] +async fn claude_source_reader_retries_after_a_stalled_create_answers_and_removes_it_first() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Late(Duration::from_secs(9)), Create::Now], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = backend + .observe_claude_sources_with(&QUICK, &owner, &request, "synthetic-run", far()) + .await; + let creates = daemon.creates(); + let removed = daemon.removed(); + daemon.finish().await; + assert!(result.is_ok(), "the retry reads the sources: {result:?}"); + assert_eq!(creates.len(), 2); + // The stalled helper was settled and removed before the retry began, so + // the daemon never held two of this read's helpers at once. + assert_eq!(removed, creates, "both helpers are removed, in order"); + assert!(daemon.live().is_empty()); + let events = lifecycle_events(&log); + let retries = named(&events, "supervisor.helper.retry"); + assert_eq!(retries.len(), 1, "{events:?}"); + assert_eq!(retries[0]["data"]["cause"], "timed_out"); + assert_eq!( + retries[0]["data"]["step"], "create", + "the retry names the step the deadline hit" + ); + assert!(named(&events, "supervisor.helper.cleanup.deferred").is_empty()); + assert_eq!(named(&events, "supervisor.helper.recovered").len(), 1); +} + +#[tokio::test] +async fn claude_source_reader_reports_exhaustion_after_its_bounded_attempts() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Refuse(503); 5], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = backend + .observe_claude_sources_with(&QUICK, &owner, &request, "synthetic-run", far()) + .await; + daemon.finish().await; + assert!(result.is_err()); + assert_eq!(daemon.creates().len(), 3, "three attempts at most"); + let events = lifecycle_events(&log); + assert_eq!( + named(&events, "supervisor.helper.retry").len(), + 2, + "{events:?}" + ); + let exhausted = named(&events, "supervisor.helper.retry_exhausted"); + assert_eq!(exhausted.len(), 1, "{events:?}"); + let exhausted = &exhausted[0]["data"]; + assert_eq!(exhausted["severity"], "error"); + assert_eq!(exhausted["attempt"], 3); + assert_eq!(exhausted["maxAttempts"], 3); + assert_eq!(exhausted["cause"], "refused"); + assert_eq!(exhausted["soulId"], json!(request.soul_id)); +} + +#[tokio::test] +async fn claude_source_reader_stays_inside_its_total_budget_and_reaps_what_outlives_it() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Late(Duration::from_secs(25))], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + // One attempt, then a settle that ends before Docker answers. + let policy = helper_lifecycle::HelperRetryPolicy { + total: Duration::from_secs(10), + ..QUICK + }; + let started = tokio::time::Instant::now(); + let result = backend + .observe_claude_sources_with(&policy, &owner, &request, "synthetic-run", far()) + .await; + let elapsed = started.elapsed(); + assert!(result.is_err()); + assert!( + elapsed < Duration::from_secs(18), + "the read returns near its total budget, not when Docker answers: {elapsed:?}" + ); + let creates = daemon.creates(); + assert_eq!(creates.len(), 1, "no retry once the budget has no room"); + let events = reaper_outcome(&log).await; + daemon.finish().await; + assert_eq!( + daemon.removed(), + creates, + "the reaper removes the late helper" + ); + assert_eq!( + named(&events, "supervisor.helper.cleanup.deferred").len(), + 1 + ); + assert_eq!(named(&events, "supervisor.helper.cleanup.removed").len(), 1); + let exhausted = named(&events, "supervisor.helper.retry_exhausted"); + assert_eq!(exhausted.len(), 1, "{events:?}"); + assert_eq!(exhausted[0]["data"]["cause"], "timed_out"); +} + +#[tokio::test] +async fn a_helper_create_docker_answers_slowly_is_logged_with_its_latency() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Late(Duration::from_millis(5_300))], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = backend + .read_claude_source_helper( + &owner, + &request, + "synthetic-run", + HelperDeadlines::after(Duration::from_secs(10), Duration::from_secs(1)), + ) + .await; + let creates = daemon.creates(); + daemon.finish().await; + assert!( + result.is_ok(), + "a slow create inside the budget still reads: {result:?}" + ); + let events = lifecycle_events(&log); + let slow = named(&events, "supervisor.helper.create_slow"); + assert_eq!(slow.len(), 1, "{events:?}"); + let slow = &slow[0]["data"]; + assert_eq!(slow["severity"], "warn"); + assert_eq!(slow["name"], json!(creates[0])); + assert_eq!(slow["status"], 201); + assert!(slow["elapsedMs"].as_u64().unwrap() >= 5_000, "{slow}"); +} + +#[tokio::test] +async fn claude_source_reader_ends_by_its_callers_deadline_inside_its_own_budget() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::start( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Late(Duration::from_secs(25))], + 0, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + // The policy alone would wait out the stalled create and retry; the + // caller's end (a recovery scope) leaves room for one attempt only. + let started = tokio::time::Instant::now(); + let result = backend + .observe_claude_sources_with( + &QUICK, + &owner, + &request, + "synthetic-run", + started + Duration::from_secs(10), + ) + .await; + let elapsed = started.elapsed(); + assert!(result.is_err(), "{result:?}"); + assert!( + elapsed < Duration::from_secs(18), + "the read ends near its caller's deadline, not when Docker answers: {elapsed:?}" + ); + let creates = daemon.creates(); + assert_eq!(creates.len(), 1, "no retry past the caller's deadline"); + let events = reaper_outcome(&log).await; + daemon.finish().await; + assert_eq!( + daemon.removed(), + creates, + "the reaper removes the late helper" + ); + let exhausted = named(&events, "supervisor.helper.retry_exhausted"); + assert_eq!(exhausted.len(), 1, "{events:?}"); +} + +#[tokio::test] +async fn claude_source_reader_names_a_hung_read_apart_from_a_docker_stall() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + let mut daemon = ReaderDaemon::with_reader( + &temp.path().join("docker.sock"), + daemon_id, + vec![], + Reader::Hangs, + Duration::ZERO, + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = backend + .observe_claude_sources_with(&QUICK, &owner, &request, "synthetic-run", far()) + .await; + let creates = daemon.creates(); + let removed = daemon.removed(); + daemon.abort_readers(); + daemon.finish().await; + assert!(result.is_err(), "{result:?}"); + // A timeout is retried wherever it hit: the same disk flush that stalls a + // create also stalls a reader's writes and Docker's report of its exit. + // The log tells the reader's own step apart from Docker's. + assert_eq!( + creates.len(), + 3, + "a timed-out read is retried within its bound" + ); + assert_eq!(removed, creates, "every hung reader is removed"); + let events = lifecycle_events(&log); + let mut steps = named(&events, "supervisor.helper.retry") + .iter() + .map(|event| event["data"]["step"].clone()) + .collect::>(); + steps.extend( + named(&events, "supervisor.helper.retry_exhausted") + .iter() + .map(|event| event["data"]["step"].clone()), + ); + assert_eq!(steps, vec![json!("read"); 3], "{events:?}"); +} + +#[tokio::test] +async fn claude_source_reader_lets_a_slow_removal_use_the_time_its_retry_reserved() { + let temp = tempfile::tempdir().unwrap(); + let daemon_id = DockerDaemonId::new(); + let (owner, request) = initial_request(temp.path(), &daemon_id).await; + // Docker answers the reader's removal well after one cleanup window, + // and well inside the time the read reserved for cleanup and retries. + let mut daemon = ReaderDaemon::with_reader( + &temp.path().join("docker.sock"), + daemon_id, + vec![Create::Now], + Reader::Exits(0), + QUICK.cleanup + Duration::from_secs(4), + ); + let log = temp.path().join("lifecycle.jsonl"); + let backend = + DockerEngineBackend::new(temp.path().join("docker.sock")).with_lifecycle_log(&log); + let result = backend + .observe_claude_sources_with(&QUICK, &owner, &request, "synthetic-run", far()) + .await; + let creates = daemon.creates(); + let removed = daemon.removed(); + daemon.finish().await; + assert!( + result.is_ok(), + "a removal confirmed inside the reserve keeps the evidence: {result:?}" + ); + assert_eq!(creates.len(), 1); + assert_eq!(removed, creates); + let events = lifecycle_events(&log); + assert!( + named(&events, "supervisor.helper.cleanup.deferred").is_empty(), + "{events:?}" + ); +} diff --git a/crates/freshell-supervisor/src/backend/helper_lifecycle.rs b/crates/freshell-supervisor/src/backend/helper_lifecycle.rs new file mode 100644 index 000000000..c24b21ec9 --- /dev/null +++ b/crates/freshell-supervisor/src/backend/helper_lifecycle.rs @@ -0,0 +1,662 @@ +//! Lifecycle of the short-lived helper containers the supervisor runs for one +//! bounded read: the Claude source readers and the provider mountpoint helper. +//! +//! Docker can take tens of seconds to answer a container create. It mounts +//! and unmounts the new container's overlay root filesystem, and the kernel +//! syncs the whole filesystem under Docker's data root when that overlay +//! mount goes away, so the create waits for every dirty page on that disk. On +//! the development host a create took about 0.1s idle, 4.5s with 1 GB of dirty +//! data and 16.7s with 3 GB, and up to 31s during busy test runs; the next +//! create, with little left to flush, is fast again. Three rules follow: +//! +//! 1. A create keeps running when its caller stops waiting +//! ([`InFlightCreate`]). Dropping the request does not stop Docker; it only +//! hides whether the helper now exists. +//! 2. An attempt's objects are removed only once every create they came from +//! has answered, by exact name with confirmed absence +//! ([`DockerEngineBackend::release_helper_objects`]). Objects stay owned +//! until their absence is confirmed: a create still unanswered when its +//! caller's wait ends, or a removal that runs out of time or fails, goes to +//! a background reaper that keeps removing them +//! ([`DockerEngineBackend::reap_helper_objects`]). +//! 3. A helper run that timed out, or whose create or start Docker refused or +//! dropped transiently, is retried within a bounded budget +//! ([`DockerEngineBackend::retry_helper`]). Every other failure is final. +//! A timeout is retried whichever step it hit: the same flush that stalls +//! a create stalls a start, the helper's own writes and Docker's report of +//! its exit. On the development host a candidate read whose create and +//! start answered at once timed out while 2 GB of unwritten data drained, +//! and its retry succeeded. The retry log names the step +//! ([`AttemptStep`]), so a Docker stall can be told from a hung helper. +use super::*; +use std::future::Future; +use std::time::Duration; +use tokio::task::JoinHandle; +use tokio::time::Instant; + +/// Exact-name removal of one attempt's helper objects, once every create they +/// came from has answered. +pub(super) const HELPER_CLEANUP_BUDGET: Duration = Duration::from_secs(5); + +/// How long the background reaper waits for Docker to answer a create whose +/// caller gave up, before it reports the helper's absence as unconfirmed. +const LATE_CREATE_LIMIT: Duration = Duration::from_secs(600); + +/// How long the background reaper keeps removing what one attempt left, once +/// every create it came from has answered. A force-remove can wait on the +/// same disk flush a create does, so one removal can take tens of seconds. +const LATE_REMOVAL_LIMIT: Duration = Duration::from_secs(120); + +/// The longest wait between two of the background reaper's removal tries. +const LATE_REMOVAL_BACKOFF_CAP: Duration = Duration::from_secs(15); + +/// A helper create Docker answers after this long is logged with its +/// latency: a healthy create answers in well under a second. +const SLOW_CREATE: Duration = Duration::from_secs(5); + +/// When one helper attempt's work and its cleanup wait end. +#[derive(Debug, Clone, Copy)] +pub(super) struct HelperDeadlines { + /// The helper's own work (create, start, exchange, wait) ends here. + pub(super) operation: Instant, + /// Cleanup waits until here for a create Docker has not answered yet; + /// after that the background reaper owns its removal. + pub(super) settle: Instant, + /// The exact-name removal after every create answered gets at least + /// this long... + pub(super) cleanup: Duration, + /// ...and may run until here when that is later: the time the caller + /// set aside for this attempt's cleanup and any retry after it. A + /// removal that waits on the same stall as the attempt then costs + /// reserved time instead of the evidence. + pub(super) reserved: Instant, +} + +impl HelperDeadlines { + /// A stand-alone attempt: `budget` of work, then up to `settle_after` for + /// a create Docker has not answered yet. + pub(super) fn after(budget: Duration, settle_after: Duration) -> Self { + let started = Instant::now(); + let operation = started + budget; + Self { + operation, + settle: operation + settle_after, + cleanup: HELPER_CLEANUP_BUDGET, + // Nothing beyond the cleanup window itself. + reserved: started, + } + } + + /// When a removal that starts now must have confirmed absence: one + /// cleanup window from now, or the reserved instant if that is later. + pub(super) fn removal_deadline(&self) -> Instant { + (Instant::now() + self.cleanup).max(self.reserved) + } +} + +/// Why a failed helper attempt may be retried. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum TransientCause { + /// The attempt's work outlasted its deadline. + TimedOut, + /// Docker answered a create or start with a server error. + Refused, + /// A create or start got no answer at all. + Unanswered, +} + +impl TransientCause { + fn as_str(self) -> &'static str { + match self { + Self::TimedOut => "timed_out", + Self::Refused => "refused", + Self::Unanswered => "unanswered", + } + } +} + +/// Where one helper attempt stands, so a timed-out attempt's retry log names +/// the step its deadline hit: `prepare` (checks and inspections before the +/// create), `create`, `start`, `read` (the helper's own work, until it +/// replies over its private exchange), `exit` (Docker reporting the helper's +/// exit) or `verify` (checks after the reply). The candidate read adds +/// `scratch_create` and its mountpoint helper's `mountpoints_*` steps. +pub(super) struct AttemptStep(std::sync::Mutex<&'static str>); + +impl AttemptStep { + pub(super) fn new() -> Self { + Self(std::sync::Mutex::new("prepare")) + } + + pub(super) fn enter(&self, step: &'static str) { + *self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = step; + } + + fn current(&self) -> &'static str { + *self + .0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +/// A failed helper attempt. It is transient only where the attempt timed +/// out, or Docker refused or dropped a create or start without judging the +/// request. Every other failure stays final: changed ownership or topology, a +/// helper that exited without evidence, a client error, an unconfirmed +/// cleanup. +#[derive(Debug)] +pub(super) struct HelperFailure { + pub(super) error: BackendError, + transient: Option<(TransientCause, &'static str)>, +} + +impl From for HelperFailure { + fn from(error: BackendError) -> Self { + Self { + error, + transient: None, + } + } +} + +impl HelperFailure { + /// The attempt's work outlasted its deadline during `step`. + pub(super) fn timed_out(error: BackendError, step: &AttemptStep) -> Self { + Self { + error, + transient: Some((TransientCause::TimedOut, step.current())), + } + } + + /// Docker answered `step` ("create" or "start") with `status`. Only a + /// server error is transient; a client error judged the request. + pub(super) fn answered(step: &'static str, status: u16, error: BackendError) -> Self { + Self { + error, + transient: (500..600) + .contains(&status) + .then_some((TransientCause::Refused, step)), + } + } + + /// `step` ("create" or "start") got no answer. + pub(super) fn unanswered(step: &'static str, error: BackendError) -> Self { + Self { + error, + transient: Some((TransientCause::Unanswered, step)), + } + } +} + +type CreateAnswer = Result; + +/// Where an [`InFlightCreate`] stands after a bounded wait. +enum Settled { + /// Nothing was outstanding. + Idle, + /// The outstanding create answered while we waited. + Answered(CreateAnswer), + /// The create is still unanswered. + Pending, +} + +/// One create request for a helper object. It runs as its own task, so a +/// caller that stops waiting at its deadline does not abandon it: cleanup can +/// still learn whether Docker made the object, and remove it. +#[derive(Default)] +pub(super) struct InFlightCreate(Option>); + +impl InFlightCreate { + /// Send the create of the object called `name`. A slow answer is logged + /// with its latency when it arrives, whether or not anyone still waits. + pub(super) async fn send( + &mut self, + backend: &DockerEngineBackend, + name: &str, + path: String, + body: Value, + ) -> CreateAnswer { + let backend = backend.clone(); + let name = name.to_owned(); + let request = self.0.insert(tokio::spawn(async move { + let sent = Instant::now(); + let answer = backend + .request_bounded("POST", &path, Some(&body), 64 * 1024) + .await; + let elapsed = sent.elapsed(); + if elapsed >= SLOW_CREATE { + backend.helper_event( + "supervisor.helper.create_slow", + json!({"severity":"warn","name":name,"elapsedMs":millis(elapsed), + "status":answer.as_ref().ok().map(|response| response.status)}), + ); + } + answer + })); + let answer = joined(request.await); + self.0 = None; + answer + } + + /// Wait until the outstanding create answers, at most until `deadline`. + async fn settle(&mut self, deadline: Instant) -> Settled { + let Some(request) = self.0.as_mut() else { + return Settled::Idle; + }; + match tokio::time::timeout_at(deadline, request).await { + Ok(answer) => { + self.0 = None; + Settled::Answered(joined(answer)) + } + Err(_) => Settled::Pending, + } + } + + /// Wait until the outstanding create answers, at most until `deadline`; + /// false while it is still unanswered. `attempted` records whether Docker + /// may hold an object under the create's name: a name conflict proves the + /// name was never ours. + pub(super) async fn settle_ownership( + &mut self, + deadline: Instant, + attempted: &mut bool, + ) -> bool { + match self.settle(deadline).await { + Settled::Pending => false, + Settled::Answered(Ok(response)) if response.status == 409 => { + *attempted = false; + true + } + Settled::Idle | Settled::Answered(_) => true, + } + } +} + +fn joined(answer: Result) -> CreateAnswer { + answer.unwrap_or_else(|_| { + Err(BackendError::Unavailable( + "helper create request ended without an answer".into(), + )) + }) +} + +/// The Docker objects one helper attempt created or may have created. They +/// are removed only once every create they came from has answered. +#[async_trait] +pub(super) trait HelperObjects: Send + 'static { + /// The objects' names, for logs. + fn names(&self) -> Vec; + /// Wait until every create has answered; false if one is still + /// unanswered at `deadline`. + async fn settle(&mut self, deadline: Instant) -> bool; + /// Remove every owned object by exact name and confirm its absence. + async fn remove(&mut self, backend: &DockerEngineBackend) -> Result<(), BackendError>; +} + +/// Why an attempt's objects went to the background reaper. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum Deferred { + /// A create was still unanswered when the caller's wait ended. + CreateUnanswered, + /// Every create answered, but removing the objects ran out of time or + /// failed before their absence was confirmed. + RemovalUnconfirmed, +} + +impl Deferred { + fn as_str(self) -> &'static str { + match self { + Self::CreateUnanswered => "create_unanswered", + Self::RemovalUnconfirmed => "removal_unconfirmed", + } + } +} + +/// Which fixed helper a [`HelperContainer`] is, for its ownership check. +#[derive(Debug, Clone, Copy)] +pub(super) enum HelperKind { + ClaudeSource, + ClaudeCandidateReader, + ProviderMountpoints, +} + +/// A helper container the supervisor asked Docker to create, owned until its +/// exact-name absence is confirmed. +pub(super) struct HelperContainer { + kind: HelperKind, + pub(super) name: String, + pub(super) expected: Value, + create: InFlightCreate, + /// Docker may hold an object under `name` because of this create. + attempted: bool, +} + +impl HelperContainer { + pub(super) fn new(kind: HelperKind, name: String, expected: Value) -> Self { + Self { + kind, + name, + expected, + create: InFlightCreate::default(), + attempted: false, + } + } + + /// Docker has not answered this helper's create yet. + pub(super) fn is_pending(&self) -> bool { + self.create.0.is_some() + } + + /// Ask Docker to create the helper. A name conflict proves the name was + /// never ours. + pub(super) async fn create(&mut self, backend: &DockerEngineBackend) -> CreateAnswer { + self.attempted = true; + let path = format!("{DOCKER_API}/containers/create?name={}", self.name); + let answer = self + .create + .send(backend, &self.name, path, self.expected.clone()) + .await; + if matches!(&answer, Ok(response) if response.status == 409) { + self.attempted = false; + } + answer + } +} + +#[async_trait] +impl HelperObjects for HelperContainer { + fn names(&self) -> Vec { + vec![self.name.clone()] + } + + async fn settle(&mut self, deadline: Instant) -> bool { + self.create + .settle_ownership(deadline, &mut self.attempted) + .await + } + + async fn remove(&mut self, backend: &DockerEngineBackend) -> Result<(), BackendError> { + if !self.attempted { + return Ok(()); + } + match self.kind { + HelperKind::ClaudeSource => { + backend + .cleanup_claude_source_helper(&self.name, &self.expected) + .await + } + HelperKind::ClaudeCandidateReader => { + backend + .cleanup_claude_candidate_reader(&self.name, &self.expected) + .await + } + HelperKind::ProviderMountpoints => { + backend + .cleanup_provider_mountpoint_helper(&self.name, &self.expected) + .await + } + }?; + self.attempted = false; + Ok(()) + } +} + +/// Bounded retry of one helper read. Each attempt keeps the helper's own +/// work bound; the whole retry, its waits and its cleanup stay inside +/// `total`. +#[derive(Debug, Clone, Copy)] +pub(crate) struct HelperRetryPolicy { + /// One attempt's helper work. + pub(crate) attempt: Duration, + /// Every attempt, the waits between them and the cleanup after them. + pub(crate) total: Duration, + /// The exact-name removal after each attempt. + pub(crate) cleanup: Duration, + /// The waits before the second and third attempt. + pub(crate) backoff: [Duration; 2], + /// A retry starts only with at least this much helper time left. + pub(crate) min_attempt: Duration, +} + +impl HelperRetryPolicy { + fn max_attempts(&self) -> usize { + self.backoff.len() + 1 + } +} + +/// The Claude source reader: 25 seconds of helper work per attempt (a +/// healthy read takes about two). A stalled create answers only once the host +/// has flushed what was unwritten on Docker's disk: up to 8 GB and a 43-second +/// stall during the provider parity runs on the development host. The whole +/// read therefore gets three attempt lengths, each with its five-second +/// cleanup: the attempt the stall times out, up to one more attempt length +/// for Docker to answer that create, and one full retry after it. A third +/// attempt fits only when an earlier one was refused quickly. +pub(crate) const CLAUDE_SOURCE_READER_RETRY: HelperRetryPolicy = HelperRetryPolicy { + attempt: Duration::from_secs(25), + total: Duration::from_secs(3 * (25 + HELPER_CLEANUP_BUDGET.as_secs())), + cleanup: HELPER_CLEANUP_BUDGET, + backoff: [Duration::from_secs(1), Duration::from_secs(2)], + min_attempt: Duration::from_secs(10), +}; + +fn with_fields(context: &Value, fields: Value) -> Value { + let mut merged = context.clone(); + if let (Some(merged), Value::Object(fields)) = (merged.as_object_mut(), fields) { + merged.extend(fields); + } + merged +} + +fn millis(duration: Duration) -> u64 { + duration.as_millis().min(u128::from(u64::MAX)) as u64 +} + +impl DockerEngineBackend { + /// A helper lifecycle event: always a tracing event, and a line in the + /// supervisor's lifecycle JSONL when the backend has one. + pub(super) fn helper_event(&self, event: &'static str, data: Value) { + match data["severity"].as_str() { + Some("error") => tracing::error!(event, data = %data, "helper lifecycle"), + Some("warn") => tracing::warn!(event, data = %data, "helper lifecycle"), + _ => tracing::info!(event, data = %data, "helper lifecycle"), + } + if let Some(path) = &self.lifecycle_log { + crate::service::append_event(path, event, data); + } + } + + /// Release one attempt's helper objects, then hand back the attempt's + /// `result`. Waits until `settle` for every create the objects came from + /// to answer, then removes them by exact name within one cleanup window + /// or until `reserved`, whichever is later; an unconfirmed removal + /// replaces the result with a final failure. Objects whose create is + /// still unanswered, or whose removal is unconfirmed, go to the + /// background reaper. [`Self::retry_helper`] settles until its last + /// cleanup reserve and treats an unconfirmed removal as final, so no + /// retry follows a hand-off: it could not reuse what may still appear + /// under those names. + pub(super) async fn release_helper_objects( + &self, + helper: &'static str, + mut objects: O, + deadlines: HelperDeadlines, + result: Result, + ) -> Result { + if !objects.settle(deadlines.settle).await { + self.reap_helper_objects(helper, objects, Deferred::CreateUnanswered); + // Every create answered before an attempt could succeed. + return result.and_then(|_| { + Err(BackendError::Unavailable(format!( + "{helper} create is still unanswered; its removal continues in the background" + )) + .into()) + }); + } + let removed = tokio::time::timeout_at(deadlines.removal_deadline(), objects.remove(self)) + .await + .unwrap_or_else(|_| { + Err(BackendError::Unavailable(format!( + "{helper} cleanup timed out" + ))) + }); + if let Err(error) = removed { + // The objects stay ours until their absence is confirmed. + self.reap_helper_objects(helper, objects, Deferred::RemovalUnconfirmed); + return Err(error.into()); + } + result + } + + /// Keep removing `objects` in the background: wait for every create they + /// came from to answer, then remove them by exact name, retrying with a + /// growing wait until their absence is confirmed or the reaper's limit + /// passes. Each removal re-checks ownership, so a retry never removes an + /// object that is no longer ours. + pub(super) fn reap_helper_objects( + &self, + helper: &'static str, + mut objects: O, + why: Deferred, + ) { + let names = objects.names(); + self.helper_event( + "supervisor.helper.cleanup.deferred", + json!({"severity":"warn","helper":helper,"names":names,"reason":why.as_str()}), + ); + let backend = self.clone(); + tokio::spawn(async move { + let started = Instant::now(); + let removed = if objects.settle(started + LATE_CREATE_LIMIT).await { + backend + .keep_removing(&mut objects, Instant::now() + LATE_REMOVAL_LIMIT) + .await + } else { + Err(BackendError::Unavailable(format!( + "{helper} create still unanswered after {}s", + LATE_CREATE_LIMIT.as_secs() + ))) + }; + match removed { + Ok(()) => backend.helper_event( + "supervisor.helper.cleanup.removed", + json!({"severity":"info","helper":helper,"names":names, + "reason":why.as_str(),"elapsedMs":millis(started.elapsed())}), + ), + Err(error) => backend.helper_event( + "supervisor.helper.cleanup.unconfirmed", + json!({"severity":"error","helper":helper,"names":names, + "reason":why.as_str(),"error":error.to_string()}), + ), + } + }); + } + + /// Remove `objects` until their absence is confirmed or `deadline` + /// passes, waiting longer between tries. + async fn keep_removing( + &self, + objects: &mut O, + deadline: Instant, + ) -> Result<(), BackendError> { + let mut wait = Duration::from_secs(1); + loop { + let error = match tokio::time::timeout_at(deadline, objects.remove(self)).await { + Ok(Ok(())) => return Ok(()), + Ok(Err(error)) => error, + Err(_) => { + return Err(BackendError::Unavailable(format!( + "helper removal still unconfirmed after {}s", + LATE_REMOVAL_LIMIT.as_secs() + ))) + } + }; + if Instant::now() + wait >= deadline { + return Err(error); + } + tokio::time::sleep(wait).await; + wait = (wait * 2).min(LATE_REMOVAL_BACKOFF_CAP); + } + } + + /// Run `attempt` until it succeeds, fails finally, or `policy` leaves no + /// room for another attempt. An attempt runs only after the previous + /// one's helpers were removed, so retries never pile creates onto a + /// stalled daemon. Every attempt and its cleanup end by `until` (and + /// inside `policy.total`). `context` identifies the read in every event. + pub(super) async fn retry_helper( + &self, + policy: &HelperRetryPolicy, + helper: &'static str, + context: Value, + until: Instant, + mut attempt: F, + ) -> Result + where + F: FnMut(HelperDeadlines) -> Fut, + Fut: Future>, + { + let started = Instant::now(); + // Every attempt's work and cleanup wait end here, which leaves the + // last cleanup inside the total and before the caller's end. + let end = (started + policy.total).min(until); + let reserve = end + .checked_sub(policy.cleanup) + .unwrap_or(started) + .max(started); + let mut number = 1; + loop { + let deadlines = HelperDeadlines { + operation: (Instant::now() + policy.attempt).min(reserve), + settle: reserve, + cleanup: policy.cleanup, + reserved: end, + }; + let failure = match attempt(deadlines).await { + Ok(value) => { + if number > 1 { + self.helper_event( + "supervisor.helper.recovered", + with_fields( + &context, + json!({"severity":"info","helper":helper,"attempts":number, + "elapsedMs":millis(started.elapsed())}), + ), + ); + } + return Ok(value); + } + Err(failure) => failure, + }; + let Some((cause, step)) = failure.transient else { + return Err(failure.error); + }; + let detail = json!({"helper":helper,"attempt":number, + "maxAttempts":policy.max_attempts(),"cause":cause.as_str(),"step":step, + "reason":failure.error.to_string(),"elapsedMs":millis(started.elapsed())}); + let wait = policy.backoff.get(number - 1).copied().filter(|wait| { + reserve.saturating_duration_since(Instant::now() + *wait) >= policy.min_attempt + }); + let Some(wait) = wait else { + self.helper_event( + "supervisor.helper.retry_exhausted", + with_fields(&with_fields(&context, detail), json!({"severity":"error"})), + ); + return Err(failure.error); + }; + self.helper_event( + "supervisor.helper.retry", + with_fields( + &with_fields(&context, detail), + json!({"severity":"warn","backoffMs":millis(wait)}), + ), + ); + tokio::time::sleep(wait).await; + number += 1; + } + } +} diff --git a/crates/freshell-supervisor/src/backend/provider_mountpoints.rs b/crates/freshell-supervisor/src/backend/provider_mountpoints.rs index fe9c86241..aed40dfa3 100644 --- a/crates/freshell-supervisor/src/backend/provider_mountpoints.rs +++ b/crates/freshell-supervisor/src/backend/provider_mountpoints.rs @@ -1,4 +1,8 @@ //! Fixed helper for descriptor-owned placeholders in the exact soul volume. +use super::helper_lifecycle::{ + AttemptStep, Deferred, HelperContainer, HelperDeadlines, HelperFailure, HelperKind, + HelperObjects, HELPER_CLEANUP_BUDGET, +}; use super::*; use freshell_platform::provider_mountpoints::Mountpoint; use std::time::Duration; @@ -206,7 +210,6 @@ impl DockerEngineBackend { if self.daemon_id().await? != *daemon { return Err(fail("provider mountpoint daemon changed")); } - let mut owned = None; let target = MountpointHelperTarget { image: &spec.image_ref, volume: &spec.provider_volume_name, @@ -214,28 +217,67 @@ impl DockerEngineBackend { daemon, labels: json!({"com.freshell.installation-id":spec.installation_id,"com.freshell.soul-id":spec.soul_id,"com.freshell.incarnation-id":spec.incarnation_id}), }; - self.run_provider_mountpoint_helper( + self.run_launch_mountpoint_helper( target, selected, - (Duration::from_secs(30), Duration::from_secs(5)), - &mut owned, + HelperDeadlines::after(Duration::from_secs(30), HELPER_CLEANUP_BUDGET), ) .await } - pub(super) async fn run_provider_mountpoint_helper( + /// Run the fixed mountpoint helper for a launch within `deadlines`. A + /// launch has no later cleanup of its own, so whatever the run still owns + /// when it returns goes to the background reaper. + pub(super) async fn run_launch_mountpoint_helper( &self, target: MountpointHelperTarget<'_>, selected: Vec, - budgets: (Duration, Duration), - owned: &mut Option<(String, Value)>, + deadlines: HelperDeadlines, ) -> Result<(), BackendError> { + let mut owned = None; + let result = self + .run_provider_mountpoint_helper( + target, + selected, + deadlines, + &AttemptStep::new(), + &mut owned, + ) + .await; + // The run still owns a helper Docker has not created yet, or one whose + // removal it could not confirm: either is still ours to remove. + if let Some(helper) = owned.take() { + let why = if helper.is_pending() { + Deferred::CreateUnanswered + } else { + Deferred::RemovalUnconfirmed + }; + self.reap_helper_objects("provider_mountpoints", helper, why); + } + result.map_err(|failure| failure.error) + } + + /// Run the fixed mountpoint helper against `target` within `deadlines`. + /// `owned` holds the helper from its create until its removal is + /// confirmed; it is still set on return when the removal is unconfirmed or + /// Docker has not answered the create yet, and the caller then owns the + /// helper's removal. A run that timed out, or whose create or start was + /// refused transiently, is reported as a transient failure; `step` records + /// the `mountpoints_*` step the run is in. + pub(super) async fn run_provider_mountpoint_helper( + &self, + target: MountpointHelperTarget<'_>, + selected: Vec, + deadlines: HelperDeadlines, + step: &AttemptStep, + owned: &mut Option, + ) -> Result<(), HelperFailure> { if owned.is_some() { - return Err(fail( - "provider mountpoint previous helper cleanup remains unconfirmed", - )); + return Err( + fail("provider mountpoint previous helper cleanup remains unconfirmed").into(), + ); } - let deadline = tokio::time::Instant::now() + budgets.0; + let deadline = deadlines.operation; let request = freshell_runtime_protocol::bounded_json_bytes(&selected, 1024 * 1024) .map_err(|_| fail("provider mountpoint metadata profile exceeded"))?; let directory = native_auth::ExchangeDirectory::create_input()?; @@ -249,9 +291,7 @@ impl DockerEngineBackend { .await .map_err(|_| fail("provider mountpoint image lookup timed out"))??; if tokio::time::Instant::now() >= deadline { - return Err(fail( - "provider mountpoint preparation timed out before create", - )); + return Err(fail("provider mountpoint preparation timed out before create").into()); } let body = json!({ "Image":target.image,"User":"0:0","Tty":false, @@ -265,24 +305,28 @@ impl DockerEngineBackend { {"Type":"volume","Source":target.volume,"Target":"/home/freshell/provider","ReadOnly":false,"VolumeOptions":{"NoCopy":true}}, {"Type":"bind","Source":directory.0,"Target":"/run/freshell-input-exchange","ReadOnly":true}]} }); - let mut attempted = false; let operation = async { - attempted = true; - *owned = Some((name.clone(), body.clone())); - let created = self - .request_bounded( - "POST", - &format!("{DOCKER_API}/containers/create?name={name}"), - Some(&body), - 64 * 1024, - ) - .await?; - if created.status == 409 { - attempted = false; - *owned = None; - } + step.enter("mountpoints_create"); + let created = owned + .insert(HelperContainer::new( + HelperKind::ProviderMountpoints, + name.clone(), + body.clone(), + )) + .create(self) + .await + .map_err(|_| { + HelperFailure::unanswered( + "mountpoints_create", + fail("provider mountpoint helper creation unanswered"), + ) + })?; if created.status != 201 { - return Err(fail("provider mountpoint helper creation failed")); + return Err(HelperFailure::answered( + "mountpoints_create", + created.status, + fail("provider mountpoint helper creation failed"), + )); } let created: Value = serde_json::from_slice(&created.body) .map_err(|_| fail("provider mountpoint helper identity unavailable"))?; @@ -305,16 +349,28 @@ impl DockerEngineBackend { || actual["State"]["Running"] != false || !owned_mountpoint_helper(&actual, &body) { - return Err(fail( - "provider mountpoint helper namespace changed before start", - )); + return Err( + fail("provider mountpoint helper namespace changed before start").into(), + ); } + step.enter("mountpoints_start"); let started = self .request("POST", &format!("{DOCKER_API}/containers/{id}/start"), None) - .await?; + .await + .map_err(|_| { + HelperFailure::unanswered( + "mountpoints_start", + fail("provider mountpoint helper start unanswered"), + ) + })?; if started.status != 204 { - return Err(fail("provider mountpoint helper start failed")); + return Err(HelperFailure::answered( + "mountpoints_start", + started.status, + fail("provider mountpoint helper start failed"), + )); } + step.enter("mountpoints_read"); let exchange = async { let (mut stream, _) = listener .accept() @@ -333,6 +389,7 @@ impl DockerEngineBackend { .read_exact(&mut acknowledgement) .await .map_err(|_| fail("provider mountpoint acknowledgement unavailable"))?; + step.enter("mountpoints_exit"); match &acknowledgement { b"ok" => Ok(()), b"er" => { @@ -378,24 +435,29 @@ impl DockerEngineBackend { let (exchange_result, wait_result) = tokio::join!(exchange, wait); exchange_result?; wait_result?; + step.enter("mountpoints_verify"); if self.daemon_id().await? != *target.daemon { - return Err(fail( - "provider mountpoint daemon changed before main create", - )); + return Err(fail("provider mountpoint daemon changed before main create").into()); } Ok(()) }; let result = tokio::time::timeout_at(deadline, operation) .await - .unwrap_or_else(|_| Err(fail("provider mountpoint preparation timed out"))); - if attempted { - tokio::time::timeout( - budgets.1, - self.cleanup_provider_mountpoint_helper(&name, &body), - ) - .await - .map_err(|_| fail("provider mountpoint cleanup timed out"))??; - *owned = None; + .unwrap_or_else(|_| { + Err(HelperFailure::timed_out( + fail("provider mountpoint preparation timed out"), + step, + )) + }); + // A create Docker has not answered by the settle deadline stays in + // `owned`: the caller owns removing the helper it may still make. + if let Some(helper) = owned.as_mut() { + if helper.settle(deadlines.settle).await { + tokio::time::timeout_at(deadlines.removal_deadline(), helper.remove(self)) + .await + .map_err(|_| fail("provider mountpoint cleanup timed out"))??; + *owned = None; + } } tracing::debug!( event = "runtime.provider_mountpoints.settled", @@ -558,6 +620,7 @@ mod tests { }); let backend = DockerEngineBackend::new(owned.path().join("docker.sock")); let mut helper = None; + let step = AttemptStep::new(); let budget = Duration::from_millis(250); let caller_bound = if mode == "cancel-private-exchange" { Duration::from_millis(50) @@ -587,18 +650,18 @@ mod tests { }; let outcome = tokio::select! { outcome = tokio::time::timeout(caller_bound,backend.run_provider_mountpoint_helper( - MountpointHelperTarget {image:"synthetic-image",volume:"owned-candidate-volume",binary:&binary,daemon:&daemon,labels:json!({"com.freshell.soul-id":"synthetic-owner","com.freshell.source-request-id":"synthetic-request"})},selected,(budget,budget),&mut helper + MountpointHelperTarget {image:"synthetic-image",volume:"owned-candidate-volume",binary:&binary,daemon:&daemon,labels:json!({"com.freshell.soul-id":"synthetic-owner","com.freshell.source-request-id":"synthetic-request"})},selected,HelperDeadlines::after(budget,budget),&step,&mut helper )) => outcome, () = expire_at_stall => unreachable!("the stall clock never completes"), }; if mode == "cancel-private-exchange" { assert!(outcome.is_err()); - let (name, body) = helper + let owned = helper .as_ref() .expect("canceled private exchange must retain exact helper ownership"); tokio::time::timeout( Duration::from_millis(250), - backend.cleanup_provider_mountpoint_helper(name, body), + backend.cleanup_provider_mountpoint_helper(&owned.name, &owned.expected), ) .await .unwrap() @@ -680,6 +743,130 @@ mod tests { } } + /// A launch has no later cleanup of its own: whatever its helper run + /// still owns when it returns (a create Docker had not answered, or a + /// removal it could not confirm) goes to the background reaper. Real time, + /// because the reaper's removal is real socket I/O after the run returns. + #[tokio::test] + async fn launch_mountpoint_helper_hands_what_it_still_owns_to_the_reaper() { + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, Mutex}; + use tokio::task::JoinSet; + for mode in ["late-create", "slow-cleanup"] { + let owned = tempfile::tempdir().unwrap(); + let daemon = DockerDaemonId::new(); + let binary = owned.path().join("host"); + std::fs::write(&binary, b"synthetic helper artifact").unwrap(); + let listener = UnixListener::bind(owned.path().join("docker.sock")).unwrap(); + let created = Arc::new(Mutex::new(None::)); + let deleted = Arc::new(AtomicBool::new(false)); + // The cleanup inspection that answers late has happened. + let slowed = Arc::new(AtomicBool::new(false)); + let (stop, mut stopped) = tokio::sync::oneshot::channel::<()>(); + let (fixture_created, fixture_deleted, fixture_daemon) = + (created.clone(), deleted.clone(), daemon.clone()); + let server = tokio::spawn(async move { + let mut jobs = JoinSet::new(); + loop { + tokio::select! { + _=&mut stopped=>break, + accepted=listener.accept()=>{ + let (mut stream,_)=accepted.unwrap(); + let (created,deleted,slowed,daemon)=(fixture_created.clone(),fixture_deleted.clone(),slowed.clone(),fixture_daemon.clone()); + jobs.spawn(async move { + let mut header=Vec::new(); let mut byte=[0]; + // A caller may abandon a request it no longer waits for. + while !header.ends_with(b"\r\n\r\n") {if stream.read_exact(&mut byte).await.is_err() {return;}header.push(byte[0]);} + let header=String::from_utf8(header).unwrap(); + let line=header.lines().next().unwrap().to_string(); + let length=header.lines().find_map(|line|line.strip_prefix("Content-Length: ")).unwrap().trim().parse::().unwrap(); + let mut raw=vec![0;length];if stream.read_exact(&mut raw).await.is_err() {return;} + let (status,value)=if line.contains("/images/") {(200,json!({"Config":{"Env":["PATH=/usr/bin"]}}))} + else if line.contains("/info ") {(200,json!({"ID":daemon,"SecurityOptions":["name=rootless"]}))} + else if line.contains("/containers/create?") { + // A stalled daemon makes the helper well after the run gave up on it. + if mode=="late-create" {tokio::time::sleep(Duration::from_secs(9)).await;} + *created.lock().unwrap()=Some(serde_json::from_slice(&raw).unwrap());(201,json!({"Id":"c".repeat(64)}))} + else if line.starts_with("GET") && line.contains("/containers/") { + // Cleanup's first inspection, by name, answers after the + // cleanup window: the helper is still there and still ours. + if mode=="slow-cleanup" && line.contains("/containers/freshell-mountpoints-") && !slowed.swap(true,Ordering::SeqCst) {tokio::time::sleep(Duration::from_secs(8)).await;} + let body=created.lock().unwrap().clone(); + match body {Some(body) if !deleted.load(Ordering::SeqCst)=>(200,json!({"Id":"c".repeat(64),"Config":body,"HostConfig":body["HostConfig"],"State":{"Running":false}})),_=>(404,json!({}))} + } + else if line.contains("/start ") { + let body=created.lock().unwrap().clone().unwrap(); + let exchange=body["HostConfig"]["Mounts"].as_array().unwrap().iter().find(|mount|mount["Target"]=="/run/freshell-input-exchange").unwrap()["Source"].as_str().unwrap().to_owned(); + // The helper answers over its exchange once it is running. + tokio::spawn(async move { + let Ok(mut channel)=UnixStream::connect(Path::new(&exchange).join("reply.sock")).await else {return;}; + let Ok(length)=channel.read_u32().await else {return;}; + let mut request=vec![0;length as usize]; + if channel.read_exact(&mut request).await.is_ok() {let _=channel.write_all(b"ok").await;} + }); + (204,json!({})) + } + else if line.contains("/wait?") {(200,json!({"StatusCode":0}))} + else if line.starts_with("DELETE") {deleted.store(true,Ordering::SeqCst);(204,json!({}))} + else {(500,json!({}))}; + let raw=value.to_string(); + let _=stream.write_all(format!("HTTP/1.1 {status} fixture\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{raw}",raw.len()).as_bytes()).await; + }); + }, + result=jobs.join_next(),if !jobs.is_empty()=>{result.unwrap().unwrap();} + } + } + jobs.shutdown().await; + }); + let backend = DockerEngineBackend::new(owned.path().join("docker.sock")); + let target = MountpointHelperTarget { + image: "synthetic-image", + volume: "owned-provider-volume", + binary: &binary, + daemon: &daemon, + labels: json!({"com.freshell.soul-id":"synthetic-owner"}), + }; + let selected = vec![Mountpoint { + relative_path: ".claude/settings.json".into(), + directory: false, + }]; + // Seconds-long windows: the run reaches its create inside the + // first, and Docker answers the late create well after both. + let deadlines = if mode == "late-create" { + HelperDeadlines::after(Duration::from_secs(3), Duration::from_secs(2)) + } else { + HelperDeadlines::after(Duration::from_secs(10), Duration::from_secs(2)) + }; + let result = tokio::time::timeout( + Duration::from_secs(30), + backend.run_launch_mountpoint_helper(target, selected, deadlines), + ) + .await + .expect("the launch's helper window is bounded"); + if mode == "late-create" { + assert!(result.is_err(), "{result:?}"); + } else { + assert!( + format!("{result:?}").contains("cleanup timed out"), + "an unconfirmed removal stays a failure: {result:?}" + ); + } + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + while !deleted.load(Ordering::SeqCst) && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(50)).await; + } + let _ = stop.send(()); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); + assert!( + deleted.load(Ordering::SeqCst), + "{mode}: the helper the launch still owned is removed" + ); + } + } + #[tokio::test] async fn provider_mountpoint_cleanup_requires_confirmed_absence_after_delete() { for mode in [ diff --git a/crates/freshell-supervisor/src/claude_source_observation.rs b/crates/freshell-supervisor/src/claude_source_observation.rs index 716955273..6514db52b 100644 --- a/crates/freshell-supervisor/src/claude_source_observation.rs +++ b/crates/freshell-supervisor/src/claude_source_observation.rs @@ -79,6 +79,9 @@ impl Supervisor { } }; let mut recovery = None; + // A recovery's observation serves its callback; it must end inside + // the recovery scope that callback carries. + let mut scope_expires = None; let owner = match &request.authority { ClaudeSourceAuthority::InitialReservation { reservation_id } => { let (reservation, expected) = self @@ -126,6 +129,7 @@ impl Supervisor { .cloned() .ok_or_else(|| denied("recovery attempt scope is absent or expired"))? }; + scope_expires = Some(authority.expires); if authority.start.context.soul_id != request.soul_id || authority.start.context.provider != request.provider || authority.start.context.prior_handle.image_ref() != self.config.image_ref @@ -171,12 +175,23 @@ impl Supervisor { ) .map_err(|_| denied("observation request exceeds its metadata profile"))?; let digest = format!("{:x}", Sha256::digest(&raw)); - // The backend owns a 25-second operation and five-second finite cleanup. - // This outer deadline permits that cleanup to finish before returning. - let reply = tokio::time::timeout( - CLAUDE_SOURCE_OBSERVATION_BUDGET, + // The backend owns a bounded retry of the fixed reader, its owned + // cleanup included, and ends it by `until`: the reader's whole retry, + // or a reply slack before a recovery's scope ends, whichever is first. + // A retry inside a recovery therefore uses only what is left of the + // callback it serves. The outer wait adds the reply slack. + let mut until = Instant::now() + crate::backend::CLAUDE_SOURCE_READER_RETRY.total; + if let Some(expires) = scope_expires { + until = until.min( + expires + .checked_sub(CLAUDE_SOURCE_REPLY_SLACK) + .unwrap_or_else(Instant::now), + ); + } + let reply = tokio::time::timeout_at( + until + CLAUDE_SOURCE_REPLY_SLACK, self.backend - .observe_claude_sources(&owner, &request, &self.config.test_run_id), + .observe_claude_sources(&owner, &request, &self.config.test_run_id, until), ) .await .map_err(|_| unavailable("source reader timed out".into()))? @@ -261,6 +276,8 @@ mod tests { StdMutex>>, StdMutex>, std::sync::atomic::AtomicUsize, + // When the supervisor asked the read to end. + StdMutex>, ); #[async_trait] impl RuntimeBackend for ReplyBackend { @@ -269,8 +286,10 @@ mod tests { owner: &ClaudeSourceOwner, request: &ObserveClaudeSourcesRequest, _: &str, + until: Instant, ) -> Result { self.2.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + *self.3.lock().unwrap() = Some(until); let ClaudeSourceOwner::Retained { handle, recovery_source, @@ -367,6 +386,7 @@ mod tests { async fn provider_claude_nested_source_reply_requires_live_scope_after_private_ipc() { for mode in [ "live", + "short-scope", "dropped", "expired", "candidate", @@ -443,6 +463,7 @@ mod tests { StdMutex::new(None), StdMutex::new(None), std::sync::atomic::AtomicUsize::new(0), + StdMutex::new(None), )); let supervisor = Supervisor::new( registry, @@ -461,7 +482,15 @@ mod tests { .unwrap(); let lock = supervisor.lifecycle_lock(&soul).await; let _held = lock.lock().await; - let scope = supervisor.auth_lookup_scope(&start, CLAUDE_RECOVERY_CALLBACK_BUDGET); + // A scope shorter than the reader's whole retry bounds the read. + let lifetime = if mode == "short-scope" { + Duration::from_secs(40) + } else { + CLAUDE_RECOVERY_CALLBACK_BUDGET + }; + let scope = supervisor.auth_lookup_scope(&start, lifetime); + let scope_expires = + supervisor.auth_lookup_authorities.lock().unwrap()[&scope.key].expires; if matches!( mode, "dropped" | "expired" | "candidate-dropped" | "candidate-expired" @@ -533,6 +562,7 @@ mod tests { .await .expect("nested Claude source lookup must not reacquire the held soul lifecycle lock") .result; + let returned = Instant::now(); assert_eq!( backend.2.load(std::sync::atomic::Ordering::SeqCst), usize::from(mode != "candidate-unrecorded-prior"), @@ -540,9 +570,22 @@ mod tests { ); assert_eq!( result.is_ok(), - matches!(mode, "live" | "candidate"), + matches!(mode, "live" | "short-scope" | "candidate"), "{mode}: changed scoped authority after IPC cannot qualify source evidence: {result:?}" ); + if let Some(until) = *backend.3.lock().unwrap() { + // The read serves its recovery callback: it ends, its owned + // cleanup included, a reply slack before the scope does, and + // never later than the reader's whole retry. + assert!( + until + CLAUDE_SOURCE_REPLY_SLACK <= scope_expires, + "{mode}: the read outlives its recovery scope" + ); + assert!( + until <= returned + crate::backend::CLAUDE_SOURCE_READER_RETRY.total, + "{mode}: the read outlives the reader's own retry" + ); + } let failures = std::fs::read_to_string(owned.path().join("lifecycle.jsonl")) .unwrap_or_default() .lines() diff --git a/crates/freshell-supervisor/src/main.rs b/crates/freshell-supervisor/src/main.rs index d09e795c5..aa4b802b4 100644 --- a/crates/freshell-supervisor/src/main.rs +++ b/crates/freshell-supervisor/src/main.rs @@ -77,7 +77,7 @@ async fn run() -> Result<(), String> { .join("lifecycle.jsonl"); let supervisor = Supervisor::new( registry, - default_backend(docker_socket), + default_backend(docker_socket, lifecycle_log.clone()), SupervisorConfig { runtime_root, control_socket_path: control_socket.clone(), diff --git a/crates/freshell-supervisor/src/native_auth_lookup_tests.rs b/crates/freshell-supervisor/src/native_auth_lookup_tests.rs index 851522421..3d8196648 100644 --- a/crates/freshell-supervisor/src/native_auth_lookup_tests.rs +++ b/crates/freshell-supervisor/src/native_auth_lookup_tests.rs @@ -404,6 +404,7 @@ impl RuntimeBackend for AuthBackend { owner: &ClaudeSourceOwner, _: &ObserveClaudeSourcesRequest, _: &str, + _: tokio::time::Instant, ) -> Result { assert!(matches!(owner, ClaudeSourceOwner::Initial { .. })); self.0.fetch_add(1, Ordering::SeqCst); diff --git a/crates/freshell-supervisor/src/provider_callback_budget_tests.rs b/crates/freshell-supervisor/src/provider_callback_budget_tests.rs index 8179ab2cf..79ebda62d 100644 --- a/crates/freshell-supervisor/src/provider_callback_budget_tests.rs +++ b/crates/freshell-supervisor/src/provider_callback_budget_tests.rs @@ -85,7 +85,10 @@ async fn claude_recovery(budgets: ProviderCallbackBudgets) -> RecoveryFixture { .unwrap(); let mut supervisor = Supervisor::new( registry, - default_backend(root.path().join("absent-docker.sock")), + default_backend( + root.path().join("absent-docker.sock"), + root.path().join("lifecycle.jsonl"), + ), SupervisorConfig { runtime_root: root.path().join("runtime"), control_socket_path: root.path().join("control.sock"), diff --git a/crates/freshell-supervisor/src/service.rs b/crates/freshell-supervisor/src/service.rs index ac85cad96..be7d566a3 100644 --- a/crates/freshell-supervisor/src/service.rs +++ b/crates/freshell-supervisor/src/service.rs @@ -3186,8 +3186,13 @@ where let _ = freshell_runtime_protocol::write_frame_with_limit(&mut stream, &reply, limit).await; } -pub fn default_backend(socket: impl Into) -> Arc { - Arc::new(DockerEngineBackend::new(socket)) +/// The Docker backend. Its helper retry and late-cleanup events go to the +/// supervisor's lifecycle JSONL. +pub fn default_backend( + socket: impl Into, + lifecycle_log: impl Into, +) -> Arc { + Arc::new(DockerEngineBackend::new(socket).with_lifecycle_log(lifecycle_log)) } /// Default bound on one supervisor→session-host control round trip. @@ -3327,22 +3332,35 @@ 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); +/// Room between the end of a Claude source read, its owned cleanup included, +/// and the end of the supervisor's wait for it: the reply's slack. Outside a +/// recovery an observation therefore waits the fixed reader's whole bounded +/// retry ([`crate::backend::CLAUDE_SOURCE_READER_RETRY`], 90s) plus this. +pub(crate) const CLAUDE_SOURCE_REPLY_SLACK: Duration = Duration::from_secs(5); + +/// One Claude source observation whose fixed reader succeeds on its first +/// attempt: that attempt, its owned cleanup and the reply slack (35s). +pub(crate) const CLAUDE_SOURCE_FIRST_ATTEMPT_BUDGET: Duration = Duration::from_secs( + crate::backend::CLAUDE_SOURCE_READER_RETRY.attempt.as_secs() + + crate::backend::CLAUDE_SOURCE_READER_RETRY.cleanup.as_secs() + + CLAUDE_SOURCE_REPLY_SLACK.as_secs(), +); /// 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. +/// Wait for a Claude recovery's `issue` callback (105s). Before it answers, +/// the web server observes the current Claude sources and then the candidate +/// projection, back to back, each through this supervisor. The wait covers +/// the callback's own work plus two observations whose readers succeed on +/// their first attempt ([`CLAUDE_SOURCE_FIRST_ATTEMPT_BUDGET`]). A reader +/// that needs a retry runs inside what is left of the recovery scope, which +/// lives exactly as long as this wait: the supervisor ends every recovery +/// observation, its owned cleanup included, a reply slack before the scope +/// does, so a retry never outlives the callback it serves. pub(crate) const CLAUDE_RECOVERY_CALLBACK_BUDGET: Duration = Duration::from_secs( - PROVIDER_CALLBACK_BUDGET.as_secs() + 2 * CLAUDE_SOURCE_OBSERVATION_BUDGET.as_secs(), + PROVIDER_CALLBACK_BUDGET.as_secs() + 2 * CLAUDE_SOURCE_FIRST_ATTEMPT_BUDGET.as_secs(), ); /// How long the supervisor waits for a provider-preparation callback. The @@ -3778,7 +3796,10 @@ mod host_command_budget_tests { std::fs::write(&host_binary_path, "synthetic-host-artifact").unwrap(); let mut supervisor = Supervisor::new( registry, - default_backend(root.join("absent-docker.sock")), + default_backend( + root.join("absent-docker.sock"), + root.join("lifecycle.jsonl"), + ), SupervisorConfig { runtime_root: root.join("runtime"), control_socket_path: root.join("control.sock"), @@ -4516,7 +4537,10 @@ mod control_connection_tests { std::fs::write(&host_binary_path, "synthetic-host-artifact").unwrap(); Supervisor::new( Registry::open(registry_root, None).unwrap(), - default_backend(root.join("absent-docker.sock")), + default_backend( + root.join("absent-docker.sock"), + root.join("lifecycle.jsonl"), + ), SupervisorConfig { runtime_root: root.join("runtime"), control_socket_path: root.join("control.sock"), diff --git a/docs/development/managed-runtime.md b/docs/development/managed-runtime.md index 0815e5656..17845031d 100644 --- a/docs/development/managed-runtime.md +++ b/docs/development/managed-runtime.md @@ -160,10 +160,14 @@ 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. +back to back, through the supervisor. The wait covers two observations whose +reader succeeds on its first attempt (35s each: the reader's 25s attempt, its +5s cleanup and 5s of reply slack). The recovery scope those nested lookups +present lives exactly as long as the callback wait, and the supervisor ends +every recovery observation, its cleanup included, 5s before that scope does: +an observation never finds its scope expired, and a reader that needs a retry +uses only what is left of the scope. Outside a recovery an observation waits +up to 95s (the reader's bounded retry, 90s, plus the reply slack). A launch or recovery holds the soul's lifecycle lock for its whole activation, callback and grant included, and `stop` takes the same lock @@ -188,6 +192,60 @@ The session host itself keeps a worker-thread floor makes the control plane share one thread with every workload task — a long synchronous step then looks indistinguishable from a dead host. +### Short-lived helper containers + +The Claude source readers (`freshell-claude-source-*`, +`freshell-claude-candidate-reader-*` with its `freshell-claude-candidate-*` +scratch volume) and the provider mountpoint helper (`freshell-mountpoints-*`) +are one-shot containers the supervisor creates, runs and removes. + +A Docker container create can take tens of seconds on a busy host. Docker +mounts and unmounts the new container's overlay root filesystem, and the +kernel flushes the whole filesystem under Docker's data root when that mount +goes away, so the create waits for every unwritten change on that disk (on +the development host: about 0.1s idle, 4.5s with 1 GB unwritten, 16.7s with +3 GB; up to 48s during the provider parity runs, which build Rust release +binaries while up to 8 GB sits unwritten). The create right after it is fast +again, which is why managed-session creates stay quick while the helper +created just before them is slow. A host whose load far exceeds its cores +slows every Docker step the same way, exits and removals included. + +- A helper create keeps running when its caller stops waiting, and cleanup + waits for Docker's answer before it removes the helper by exact name. The + supervisor owns a helper until its absence is confirmed. A removal gets 5s, + or, inside a retried read, all the time the read still has reserved. A + create Docker answers only after the caller's window has passed, or a + removal that runs past its window or fails, goes to a background reaper. The reaper removes + the helper (and, for a candidate read, the scratch volume after it), retrying + with a growing wait for up to 120s; every try re-checks ownership. A create + Docker answers after 5s or more is logged as `supervisor.helper.create_slow` + (warn) with its latency. Lifecycle events: `supervisor.helper.cleanup.deferred` + (warn, with `reason` `create_unanswered` or `removal_unconfirmed`), then + `supervisor.helper.cleanup.removed` (info) or + `supervisor.helper.cleanup.unconfirmed` (error). +- The Claude source reader retries a run that timed out, or whose create or + start Docker refused with a server error or left unanswered. A timeout is + retried whichever step it hit: the same flush that stalls a create also + stalls a start, the reader's own writes and Docker's report of its exit (a + candidate read whose create and start answered at once timed out while 2 GB + of unwritten data drained, and its retry succeeded). Each attempt + keeps the reader's 25s bound; the whole read gets up to 90s, three attempt + lengths with their 5s cleanup: the attempt a stall times out, up to one + more attempt length for Docker to answer that create, and one full retry; + a third attempt fits only after a fast refusal. Inside a recovery the read + also ends 5s before the recovery scope does (see above). Waits are 1s, + then 2s, and a retry starts only after the previous attempt's helpers were + removed, so retries never pile creates onto a stalled daemon. Every other failure (a client + error, changed ownership or topology, a reader that exited without + evidence, an unconfirmed cleanup) is final, as before. Events: + `supervisor.helper.retry` (warn, per retry), `supervisor.helper.recovered` + (info) and `supervisor.helper.retry_exhausted` (error). Their `step` names + where a timed-out attempt's deadline hit: `prepare`, `create`, `start`, + `read` (the reader's own work, until it replies), `exit` (Docker reporting + its exit) or `verify`; a candidate read adds `scratch_create` and its + mountpoint helper's `mountpoints_*` steps. A reader that hangs in `read` + while Docker answers promptly is therefore visible as such. + ## Launch transaction Managed launch is intentionally staged: diff --git a/scripts/testing/runtime-sandbox.ts b/scripts/testing/runtime-sandbox.ts index 262ea3e47..fdd2a286f 100644 --- a/scripts/testing/runtime-sandbox.ts +++ b/scripts/testing/runtime-sandbox.ts @@ -7,7 +7,7 @@ import os from 'node:os' import path from 'node:path' import { configuredAmplifierOnecliGrantFiles } from './runtime-amplifier-onecli.js' -import { RestrictedDockerBroker, type BrokerEvent, type BrokerReceipt } from './runtime-test-broker.js' +import { parseHelperCreateDelay, RestrictedDockerBroker, type BrokerEvent, type BrokerReceipt } from './runtime-test-broker.js' export const PHASE1_RUNTIME_IMAGE_TAG = 'ubuntu:24.04' export const PHASE2_RUNTIME_IMAGE_TAG = 'freshell-managed-runtime:phase2' @@ -69,6 +69,10 @@ export type FreshAgentQualificationBuildRecord = { sessionHostBinary: string } +/** How long cleanup waits for Docker to answer requests the broker forwarded: + * a create can wait tens of seconds for the host to flush its disk. */ +const BROKER_SETTLE_MS = 90_000 + export function phase2BootstrapFiles( env: NodeJS.ProcessEnv = process.env, ): string[] { @@ -250,6 +254,11 @@ export class RuntimeHarness { testRunId: this.runId, logPath: path.join(this.evidenceDir, 'broker.jsonl'), }) + const helperCreateDelay = parseHelperCreateDelay(process.env.FRESHELL_TEST_BROKER_DELAY_HELPER_CREATE) + if (helperCreateDelay) { + this.broker.delayHelperCreates(helperCreateDelay) + this.recordLifecycle('gate.broker.helper_create_delay', helperCreateDelay) + } await this.broker.start() this.brokerStarted = true @@ -264,6 +273,13 @@ export class RuntimeHarness { for (const id of [...this.trackedSupervisorContainers]) { try { this.removeContainerExact(id) } catch (error) { errors.push(`supervisor ${id}: ${String(error)}`) } } + // A helper create Docker answers after the supervisor stopped waiting, or + // after the supervisor container above was removed with its background + // cleanup, still makes the helper. Let every forwarded request finish so + // the receipt removals below include that helper. + if (this.brokerStarted && !(await this.broker.settleForwarded(BROKER_SETTLE_MS))) { + errors.push(`docker requests forwarded by the broker were still unanswered after ${BROKER_SETTLE_MS / 1000}s`) + } for (const id of [...this.broker?.receiptIds?.() ?? []]) { try { this.removeContainerExact(id) } catch (error) { errors.push(`managed runtime ${id}: ${String(error)}`) } } @@ -297,6 +313,21 @@ export class RuntimeHarness { } } catch (error) { errors.push(`indexed volume ${volume}: ${String(error)}`) } } + // A candidate read whose supervisor was removed mid-read, or before its + // background cleanup ran, leaves its scratch volume. The broker knows + // exactly which ones this run created; every helper that mounted one is + // gone by now, so delete each through the broker's ownership check. + const candidateVolumes = this.broker?.candidateVolumeNames?.() ?? [] + for (const volume of candidateVolumes) { + try { + const url = `/v1.47/volumes/${volume}` + const deleted = await brokerRequestStatus(this.broker.proxySocketPath, 'DELETE', url) + if (deleted !== 204 && deleted !== 404) throw new Error('owned candidate volume deletion unavailable') + if (await brokerRequestStatus(this.broker.proxySocketPath, 'GET', url) !== 404) { + throw new Error('owned candidate volume absence unconfirmed') + } + } catch (error) { errors.push(`candidate volume ${volume}: ${String(error)}`) } + } for (const volume of [...this.trackedVolumes]) { try { docker(['volume', 'rm', volume]) } catch (error) { errors.push(`volume ${volume}: ${String(error)}`) } } @@ -317,7 +348,7 @@ export class RuntimeHarness { ...this.trackedContainers, ...this.broker?.receiptIds?.() ?? [], ])], - volumes: [...new Set([...this.trackedVolumes, ...providerVolumes, ...indexedVolumes])], + volumes: [...new Set([...this.trackedVolumes, ...providerVolumes, ...indexedVolumes, ...candidateVolumes])], unsafeBrokerAttempts: this.broker?.unsafeAttempts?.() ?? [], completedAt: new Date().toISOString(), } diff --git a/scripts/testing/runtime-test-broker.ts b/scripts/testing/runtime-test-broker.ts index 0ae038c99..35f227b91 100644 --- a/scripts/testing/runtime-test-broker.ts +++ b/scripts/testing/runtime-test-broker.ts @@ -209,6 +209,29 @@ type DockerResponse = { body: Buffer } +/** The short-lived helpers whose create can be held by a test. */ +const DELAYABLE_HELPER_PREFIXES = ['freshell-claude-source-', 'freshell-claude-candidate-reader-', 'freshell-mountpoints-'] as const + +export type HelperCreateDelay = { prefix: typeof DELAYABLE_HELPER_PREFIXES[number]; delayMs: number; count: number } + +/** + * Parse `FRESHELL_TEST_BROKER_DELAY_HELPER_CREATE` (`:[:]`, + * e.g. `freshell-claude-source-:30000:1`): hold that many helper creates for + * that long before forwarding them, as a Docker create stalled behind a + * host-wide writeback is answered late. + */ +export function parseHelperCreateDelay(value: string | undefined): HelperCreateDelay | undefined { + if (value === undefined || value.trim() === '') return undefined + const [prefix, delay, count = '1', ...rest] = value.trim().split(':') + const delayMs = Number(delay), times = Number(count) + if (rest.length > 0 || !DELAYABLE_HELPER_PREFIXES.includes(prefix as HelperCreateDelay['prefix']) + || !Number.isInteger(delayMs) || delayMs < 1 || delayMs > 600_000 + || !Number.isInteger(times) || times < 1 || times > 100) { + throw new Error(`invalid helper create delay ${JSON.stringify(value)}: expected <${DELAYABLE_HELPER_PREFIXES.join('|')}>:[:]`) + } + return { prefix: prefix as HelperCreateDelay['prefix'], delayMs, count: times } +} + export class RestrictedDockerBroker { private readonly claudeCandidates = new Map() private readonly candidateVolumes = new Set() @@ -229,6 +252,8 @@ export class RestrictedDockerBroker { private server?: http.Server private stopFailuresRemaining = 0 private inputPublicationBeforeCreateFailure = false + private readonly helperCreateDelays: HelperCreateDelay[] = [] + private readonly inFlight = new Set>() constructor(private readonly policy: RestrictedDockerBrokerPolicy) {} @@ -375,6 +400,13 @@ export class RestrictedDockerBroker { return new Set([...this.knownContainerIds, ...[...this.historyHelpers.values(),...this.inputPublishers.values(),...this.claudeSourceHelpers.values()].flatMap((helper) => helper.containerId ? [helper.containerId] : [])]) } + /** Every Claude candidate scratch volume this proxy forwarded a create + * for (a lost acknowledgement included). Its proxied DELETE keeps the + * exact ownership check, so cleanup can remove what a supervisor left. */ + candidateVolumeNames(): string[] { + return [...this.candidateVolumes] + } + eventsSnapshot(): BrokerEvent[] { return [...this.events] } @@ -393,13 +425,31 @@ export class RestrictedDockerBroker { this.stopFailuresRemaining = Math.max(0, count) } + /** Hold the next `count` creates of helpers named `prefix*` for `delayMs` + * before forwarding them, so Docker makes each helper after its caller's + * deadline, as a create stalled behind a host-wide writeback does. */ + delayHelperCreates(delay: HelperCreateDelay): void { + this.helperCreateDelays.push({ ...delay }) + } + + private async holdHelperCreate(method: string, url: string, name: string): Promise { + const delay = this.helperCreateDelays.find(candidate => candidate.count > 0 && name.startsWith(candidate.prefix)) + if (!delay) return + delay.count -= 1 + this.record({ method, url, decision: 'inject_failure', reason: `configured slow helper create (${delay.delayMs}ms)`, + destructive: false, unsafeAttempt: false, helperName: name }) + await new Promise(resolve => setTimeout(resolve, delay.delayMs)) + } + async start(): Promise { fs.mkdirSync(path.dirname(this.policy.proxySocketPath), { recursive: true, mode: 0o700 }) fs.mkdirSync(path.dirname(this.policy.logPath), { recursive: true, mode: 0o700 }) fs.chmodSync(path.dirname(this.policy.logPath), 0o700) fs.rmSync(this.policy.proxySocketPath, { force: true }) this.server = http.createServer((request, response) => { - void this.handle(request, response) + const handled = this.handle(request, response) + this.inFlight.add(handled) + void handled.finally(() => this.inFlight.delete(handled)) }) await new Promise((resolve, reject) => { this.server!.once('error', reject) @@ -408,6 +458,26 @@ export class RestrictedDockerBroker { fs.chmodSync(this.policy.proxySocketPath, 0o600) } + /** Wait until every request this proxy accepted has its Docker answer + * recorded, at most `timeoutMs`. A helper create Docker answers after the + * supervisor stopped waiting, or after the supervisor container was + * removed, still makes the helper; once settled, receiptIds() names it. + * False when requests are still unanswered at the deadline. */ + async settleForwarded(timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs + while (this.inFlight.size > 0) { + const remaining = deadline - Date.now() + if (remaining <= 0) return false + let timer: NodeJS.Timeout | undefined + await Promise.race([ + Promise.allSettled([...this.inFlight]), + new Promise(resolve => { timer = setTimeout(resolve, remaining) }), + ]) + clearTimeout(timer) + } + return true + } + async close(): Promise { if (!this.server) return await new Promise((resolve) => this.server!.close(() => resolve())) @@ -704,6 +774,7 @@ export class RestrictedDockerBroker { } const helper: { expected: Record; containerId?: string } = { expected: JSON.parse(body.toString()) } this.claudeSourceHelpers.set(name, helper) + await this.holdHelperCreate(method, url, name) try { const forwarded = await this.forward(request, body) if (forwarded.statusCode === 201) { @@ -735,6 +806,7 @@ export class RestrictedDockerBroker { } const publisher:{input:PreparedInputAuthorization;expected:Record;containerId?:string}={input,expected:JSON.parse(body.toString())} this.inputPublishers.set(name,publisher) + await this.holdHelperCreate(method, url, name) try { const forwarded=await this.forward(request,body) if (forwarded.statusCode===201) { diff --git a/test/helpers/provider-parity-environment.ts b/test/helpers/provider-parity-environment.ts index 995e65073..29bff9d74 100644 --- a/test/helpers/provider-parity-environment.ts +++ b/test/helpers/provider-parity-environment.ts @@ -150,3 +150,18 @@ export function providerParityCommands(workspace: string): Recordx.startsWith('runtime.sqlite3'))){const value=fs.readFileSync(p+'/'+f);if(markers.some(marker=>value.includes(marker)))process.exit(4)}` + + "break}catch(error){if(error.code!=='ENOENT'||attempt>=20)throw error}}"] +} diff --git a/test/integration/server/fresh-agent-parity.test.ts b/test/integration/server/fresh-agent-parity.test.ts index 3a37d4f88..568312f4b 100644 --- a/test/integration/server/fresh-agent-parity.test.ts +++ b/test/integration/server/fresh-agent-parity.test.ts @@ -10,7 +10,7 @@ import { afterAll, beforeAll, describe, expect, it } from 'vitest' import { WS_PROTOCOL_VERSION } from '../../../shared/ws-protocol.js' import { ManagedRuntimeBrowserRig } from '../../e2e-browser/helpers/managed-runtime.js' import { authorizeFreshAgentParityRequest, captureProviderParityVolumes, cleanupProviderParityVolumes, readProviderParitySource, type ProviderParitySource } from '../../helpers/provider-parity-broker.js' -import { makeProviderParityResourcesReadable, prepareProviderParityUserRoots } from '../../helpers/provider-parity-environment.js' +import { makeProviderParityResourcesReadable, prepareProviderParityUserRoots, supervisorRegistrySecretScan } from '../../helpers/provider-parity-environment.js' import { exposeGrantsToHarness, fakeOnecliGrants, withMissingGrant } from '../../helpers/provider-parity-onecli.js' import type { InitialProviderVolumeAuthorization } from '../../../scripts/testing/runtime-test-broker.js' @@ -318,9 +318,7 @@ describe('hosted fresh-agent provider inputs', () => { const evidenceDir = process.env.FRESHELL_PROVIDER_PARITY_ROWS_DIR if (evidenceDir) { const sentinel = 'fixture-secret-byte' - rig.runtime.execOwnedContainerExact(rig.supervisor.containerId, [ - 'node', '-e', `const fs=require('fs');const p='/var/lib/freshell-supervisor';for(const f of fs.readdirSync(p).filter(x=>x.startsWith('runtime.sqlite3'))){if(fs.readFileSync(p+'/'+f).includes('${sentinel}'))process.exit(4)}`, - ]) + rig.runtime.execOwnedContainerExact(rig.supervisor.containerId, supervisorRegistrySecretScan([sentinel])) for (const material of [ JSON.stringify(rig.runtime.inspectContainer(view.containerId!)), rig.runtime.containerLogs(view.containerId!), diff --git a/test/integration/server/managed-provider-parity.test.ts b/test/integration/server/managed-provider-parity.test.ts index 6501e8138..91fc540e9 100644 --- a/test/integration/server/managed-provider-parity.test.ts +++ b/test/integration/server/managed-provider-parity.test.ts @@ -10,7 +10,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest' import { WS_PROTOCOL_VERSION } from '../../../shared/ws-protocol.js' import { ManagedRuntimeBrowserRig } from '../../e2e-browser/helpers/managed-runtime.js' import { exposeGrantsToHarness, fakeOnecliGrants, withMissingGrant } from '../../helpers/provider-parity-onecli.js' -import { makeProviderParityResourcesReadable, prepareProviderParityUserRoots, providerParityCommands, providerParityFailureDiagnostics, writeProviderParityEnvironment } from '../../helpers/provider-parity-environment.js' +import { makeProviderParityResourcesReadable, prepareProviderParityUserRoots, providerParityCommands, providerParityFailureDiagnostics, supervisorRegistrySecretScan, writeProviderParityEnvironment } from '../../helpers/provider-parity-environment.js' import { authorizeProviderParityRequest, captureProviderParityVolumes, cleanupProviderParityVolumes, readProviderParitySource, type ProviderParitySource } from '../../helpers/provider-parity-broker.js' import type { InitialProviderVolumeAuthorization } from '../../../scripts/testing/runtime-test-broker.js' @@ -745,9 +745,8 @@ process.stdout.write(JSON.stringify({uid:process.getuid(),gid:process.getgid(),c expect(observed[1].onecliControlPresent).toBe(false) const supervisorLogs = rig.runtime.containerLogs(rig.supervisor.containerId) const providerLogs = rig.runtime.containerLogs(managedView) - rig.runtime.execOwnedContainerExact(rig.supervisor.containerId, [ - 'node', '-e', `const fs=require('fs');const p='/var/lib/freshell-supervisor';for(const f of fs.readdirSync(p).filter(x=>x.startsWith('runtime.sqlite3'))){const value=fs.readFileSync(p+'/'+f);if(['${nestedSecretMarker}','fixture-secret-byte'].some(marker=>value.includes(marker)))process.exit(4)}`, - ]) + rig.runtime.execOwnedContainerExact(rig.supervisor.containerId, + supervisorRegistrySecretScan([nestedSecretMarker, 'fixture-secret-byte'])) for (const material of [inspect, supervisorLogs, providerLogs, fs.readFileSync(rig.runtime.evidenceDir + '/lifecycle.jsonl', 'utf8')]) { expect(material).not.toContain(nestedSecretMarker) expect(material).not.toContain('fixture-secret-byte') diff --git a/test/unit/tooling/testing/provider-parity-registry-scan.test.ts b/test/unit/tooling/testing/provider-parity-registry-scan.test.ts new file mode 100644 index 000000000..35351c737 --- /dev/null +++ b/test/unit/tooling/testing/provider-parity-registry-scan.test.ts @@ -0,0 +1,53 @@ +import { spawnSync } from 'node:child_process' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' + +import { supervisorRegistrySecretScan } from '../../../helpers/provider-parity-environment.js' + +const owned: string[] = [] +afterEach(() => { + for (const directory of owned.splice(0)) fs.rmSync(directory, { recursive: true, force: true }) +}) + +function registry(files: Record): string { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'parity-registry-scan-')) + owned.push(directory) + for (const [name, contents] of Object.entries(files)) fs.writeFileSync(path.join(directory, name), contents) + return directory +} + +/** Run the scan as the parity tests run it in the supervisor container. */ +function scan(directory: string, markers: string[], preload?: string): number | null { + const [program, ...args] = supervisorRegistrySecretScan(markers, directory) + return spawnSync(program, preload ? ['--require', preload, ...args] : args, { encoding: 'utf8' }).status +} + +/** Make SQLite's checkpoint race happen: the first read of `name` finds it + * deleted after the directory listing named it. */ +function vanishOnFirstRead(directory: string, name: string): string { + const preload = path.join(directory, '..', `${path.basename(directory)}-vanish.cjs`) + owned.push(preload) + fs.writeFileSync(preload, `const fs=require('fs');const read=fs.readFileSync;let vanished=false; +fs.readFileSync=function(file,...rest){if(!vanished&&String(file).endsWith(${JSON.stringify(`/${name}`)})){vanished=true;fs.rmSync(file);}return read.call(this,file,...rest)}`) + return preload +} + +describe('supervisor registry secret scan', () => { + it('reports a marker in any registry file and passes a clean registry', () => { + expect(scan(registry({ 'runtime.sqlite3': 'clean', 'runtime.sqlite3-wal': 'x fixture-secret-byte' }), ['fixture-secret-byte'])).toBe(4) + expect(scan(registry({ 'runtime.sqlite3': 'nested-secret-byte' }), ['fixture-secret-byte', 'nested-secret-byte'])).toBe(4) + expect(scan(registry({ 'runtime.sqlite3': 'clean', 'other-file': 'fixture-secret-byte' }), ['fixture-secret-byte'])).toBe(0) + }) + + it('rescans when SQLite removes a -wal or -shm file between listing and reading it', () => { + for (const name of ['runtime.sqlite3-wal', 'runtime.sqlite3-shm']) { + const clean = registry({ 'runtime.sqlite3': 'clean', [name]: 'clean' }) + expect(scan(clean, ['fixture-secret-byte'], vanishOnFirstRead(clean, name)), name).toBe(0) + // The checkpoint moved the marker into the main file: the rescan reads it. + const moved = registry({ 'runtime.sqlite3': 'x fixture-secret-byte', [name]: 'clean' }) + expect(scan(moved, ['fixture-secret-byte'], vanishOnFirstRead(moved, name)), name).toBe(4) + } + }) +}) diff --git a/test/unit/tooling/testing/runtime-sandbox.test.ts b/test/unit/tooling/testing/runtime-sandbox.test.ts index 440c35416..b74b10f84 100644 --- a/test/unit/tooling/testing/runtime-sandbox.test.ts +++ b/test/unit/tooling/testing/runtime-sandbox.test.ts @@ -897,3 +897,89 @@ describe('runtime harness indexed input cleanup', () => { }, ) }) + +describe('runtime harness candidate scratch cleanup', () => { + it.each(['removed', 'already-gone', 'foreign-label'] as const)( + 'removes the Claude candidate scratch volumes the broker saw created: %s', async outcome => { + const directory = await fs.mkdtemp(path.join(os.tmpdir(), 'frs-candidate-cleanup-')) + const calls: { method: string; url: string }[] = [] + let volume: any + fakeDocker.settled = false + fakeDocker.commands = [] + const daemon = http.createServer(async (request, response) => { + const chunks: Buffer[] = [] + for await (const chunk of request) chunks.push(Buffer.from(chunk)) + const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString()) : undefined + calls.push({ method: request.method!, url: request.url! }) + if (request.url === '/v1.47/volumes/create') { + volume = { Name: body.Name, Driver: body.Driver, Options: body.DriverOpts, Labels: body.Labels } + response.statusCode = 201 + response.end(JSON.stringify(volume)) + } else if (request.method === 'GET') { + response.statusCode = volume ? 200 : 404 + response.end(JSON.stringify(volume ?? {})) + } else if (request.method === 'DELETE') { + response.statusCode = volume ? 204 : 404 + volume = undefined + response.end() + } else { response.statusCode = 500; response.end() } + }) + const harness = new RuntimeHarness(directory) + const binary = path.join(directory, 'host') + const broker = new RestrictedDockerBroker({ realSocketPath: path.join(directory, 'daemon.sock'), + proxySocketPath: path.join(directory, 'broker.sock'), runtimeRootPrefix: path.join(directory, 'r'), + allowedHostBinaryPaths: new Set([binary]), allowedImageRefs: new Set(['sha256:fixture']), + allowedWorkspaceRoots: new Set(['/workspace']), testRunId: harness.runId, logPath: path.join(directory, 'broker.jsonl') }) + harness.broker = broker + const request = (method: string, url: string, body?: unknown) => new Promise((resolve, reject) => { + const outgoing = http.request({ socketPath: broker.proxySocketPath, path: url, method }, response => { + response.resume() + response.once('end', () => resolve(response.statusCode!)) + }) + outgoing.once('error', reject) + outgoing.end(body === undefined ? undefined : JSON.stringify(body)) + }) + try { + await new Promise(resolve => daemon.listen(path.join(directory, 'daemon.sock'), resolve)) + await broker.start() + // Exercise cleanup's actual live-broker lifecycle without image/build preparation. + ;(harness as unknown as { brokerStarted: boolean }).brokerStarted = true + const installationId = 'installation-cleanup', soulId = 'soul-cleanup' + const owner = { installationId, soulId, + providerVolumeName: `freshell-provider-${createHash('sha256').update(`${installationId}\0${soulId}`).digest('hex').slice(0, 24)}`, + imageRef: 'sha256:fixture', hostBinaryPath: binary, binarySha256: 'a'.repeat(64), daemonId: 'fixture-daemon' } + broker.authorizeInitialProviderVolume(owner) + const generationId = 'grant-44444444-4444-4444-8444-444444444444' + const candidate = { ...owner, generationId, reservationId: '11111111-1111-4111-8111-111111111111', + requestId: '22222222-2222-4222-8222-222222222222', + stageRoot: path.join(directory, 'scenarios/claude/control/provider-roots', generationId), + workspacePath: '/workspace', cwd: '/workspace', manifestRelativePath: '.freshell/input-manifest.json', + configurationBinds: [], resourceMounts: [] } + broker.authorizeClaudeCandidate(candidate) + const name = `freshell-claude-candidate-${candidate.requestId}` + const labels = { 'com.freshell.installation-id': installationId, 'com.freshell.soul-id': soulId, + 'com.freshell.runtime-test-run-id': harness.runId, 'com.freshell.source-request-id': candidate.requestId, + 'com.freshell.source-binary-sha256': owner.binarySha256, 'com.freshell.input-generation': generationId, + 'com.freshell.source-reservation-id': candidate.reservationId, 'com.freshell.claude-candidate': 'ordinary' } + // The supervisor created the scratch volume; it was removed mid-read + // (or before its background cleanup ran) and left the volume behind. + expect(await request('POST', '/v1.47/volumes/create', { Name: name, Driver: 'local', DriverOpts: {}, Labels: labels })).toBe(201) + if (outcome === 'already-gone') volume = undefined + if (outcome === 'foreign-label') volume.Labels['com.freshell.soul-id'] = 'foreign' + const result = await harness.cleanup() + expect(result.ok).toBe(outcome !== 'foreign-label') + expect(volume === undefined).toBe(outcome !== 'foreign-label') + const deletes = calls.filter(call => call.method === 'DELETE') + expect(deletes).toEqual(outcome === 'foreign-label' ? [] : [{ method: 'DELETE', url: `/v1.47/volumes/${name}` }]) + if (outcome === 'foreign-label') expect(result.errors.join('\n')).toContain(name) + else expect(calls.at(-1)).toEqual({ method: 'GET', url: `/v1.47/volumes/${name}` }) + const cleanup = JSON.parse(await fs.readFile(path.join(harness.evidenceDir, 'cleanup.json'), 'utf8')) + expect(cleanup.volumes).toContain(name) + } finally { + await broker.close() + await new Promise(resolve => daemon.close(() => resolve())) + await fs.rm(directory, { recursive: true, force: true }) + } + }, + ) +}) diff --git a/test/unit/tooling/testing/runtime-test-broker.test.ts b/test/unit/tooling/testing/runtime-test-broker.test.ts index 02ef70177..950e77d53 100644 --- a/test/unit/tooling/testing/runtime-test-broker.test.ts +++ b/test/unit/tooling/testing/runtime-test-broker.test.ts @@ -7,7 +7,7 @@ import path from 'node:path' import { pathToFileURL } from 'node:url' import { describe, expect, it, vi } from 'vitest' -import { RestrictedDockerBroker, type ClaudeCandidateAuthorization, type PreparedInputAuthorization } from '../../../../scripts/testing/runtime-test-broker.js' +import { parseHelperCreateDelay, RestrictedDockerBroker, type ClaudeCandidateAuthorization, type PreparedInputAuthorization } from '../../../../scripts/testing/runtime-test-broker.js' import { authorizeFreshAgentParityRequest, authorizeProviderParityRequest, captureProviderParityVolumes, cleanupProviderParityVolumes } from '../../../helpers/provider-parity-broker.js' import { makeProviderParityResourcesReadable, providerParityCommands, providerParityFailureDiagnostics, writeProviderParityEnvironment } from '../../../helpers/provider-parity-environment.js' @@ -899,6 +899,61 @@ describe('scoped Claude source observer broker', () => { expect((await request('POST', `/v1.47/containers/create?name=${name}`, body)).status).toBe(201) }, undefined, true) }) + it('holds a configured slow helper create before Docker sees it, for the configured count only', async () => { + await withHttpBroker(async ({ request, forwarded, broker: instance, returnHelperId }) => { + instance.authorizeInitialProviderVolume(sourceAuthorization()) + instance.delayHelperCreates({ prefix: 'freshell-claude-source-', delayMs: 300, count: 1 }) + const started = Date.now() + const held = request('POST', `/v1.47/containers/create?name=${sourceName}`, sourceHelperBody()) + await new Promise(resolve => setTimeout(resolve, 100)) + expect(forwarded, 'Docker has not been asked to make the helper yet').toHaveLength(0) + expect((await held).status).toBe(201) + expect(Date.now() - started).toBeGreaterThanOrEqual(300) + expect(forwarded.map(operation => operation.url)).toEqual([`/v1.47/containers/create?name=${sourceName}`]) + expect(instance.eventsSnapshot()).toContainEqual(expect.objectContaining({ + decision: 'inject_failure', reason: 'configured slow helper create (300ms)', helperName: sourceName, + })) + const next = 'freshell-claude-source-22222222-2222-4222-8222-222222222222' + returnHelperId('c'.repeat(64)) + const quick = Date.now() + expect((await request('POST', `/v1.47/containers/create?name=${next}`, sourceHelperBody())).status).toBe(201) + expect(Date.now() - quick, 'only the configured count is held').toBeLessThan(250) + }) + }) + it('lets cleanup wait until a helper create Docker answers late is recorded', async () => { + await withHttpBroker(async ({ request, broker: instance }) => { + instance.authorizeInitialProviderVolume(sourceAuthorization()) + instance.delayHelperCreates({ prefix: 'freshell-claude-source-', delayMs: 300, count: 1 }) + // The supervisor has stopped waiting; the proxy still owes Docker's answer. + const owed = request('POST', `/v1.47/containers/create?name=${sourceName}`, sourceHelperBody()) + await new Promise(resolve => setTimeout(resolve, 50)) + expect(instance.receiptIds().has(helperId)).toBe(false) + expect(await instance.settleForwarded(5_000)).toBe(true) + expect(instance.receiptIds().has(helperId), 'cleanup now knows the late helper').toBe(true) + expect((await owed).status).toBe(201) + }) + }) + it('reports requests that are still unanswered at the settle deadline', async () => { + await withHttpBroker(async ({ request, broker: instance }) => { + instance.authorizeInitialProviderVolume(sourceAuthorization()) + instance.delayHelperCreates({ prefix: 'freshell-claude-source-', delayMs: 1_000, count: 1 }) + const owed = request('POST', `/v1.47/containers/create?name=${sourceName}`, sourceHelperBody()) + await new Promise(resolve => setTimeout(resolve, 50)) + expect(await instance.settleForwarded(100)).toBe(false) + expect((await owed).status).toBe(201) + expect(await instance.settleForwarded(100)).toBe(true) + }) + }) + it('parses the slow helper create setting and refuses anything but the fixed helpers', () => { + expect(parseHelperCreateDelay(undefined)).toBeUndefined() + expect(parseHelperCreateDelay('')).toBeUndefined() + expect(parseHelperCreateDelay('freshell-claude-source-:30000')).toEqual({ prefix: 'freshell-claude-source-', delayMs: 30_000, count: 1 }) + expect(parseHelperCreateDelay('freshell-mountpoints-:1500:2')).toEqual({ prefix: 'freshell-mountpoints-', delayMs: 1_500, count: 2 }) + for (const invalid of ['freshell-incarnation-:1000', 'freshell-claude-source-:soon', 'freshell-claude-source-:0', + 'freshell-claude-source-:1000:0', 'freshell-claude-source-:1000:1:extra', 'freshell-mountpoints-:700000']) { + expect(() => parseHelperCreateDelay(invalid), invalid).toThrow(/invalid helper create delay/) + } + }) it('records bounded upstream helper rejection and host exchange absence without daemon payloads', async () => { await withHttpBroker(async ({ request, broker: instance, rejectHelperCreate }) => { instance.authorizeInitialProviderVolume(sourceAuthorization())