From 465af24d03adebee7a875f2c5f301debf58b4eb3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 11 Aug 2026 19:57:01 +0200 Subject: [PATCH] Reap adopted model worker children --- epiphany-core/src/bin/epiphany-swarm.rs | 16 ++++++- epiphany-core/src/lib.rs | 2 + epiphany-core/src/process_observation.rs | 53 ++++++++++++++++++++++++ epiphany-core/src/runtime_spine.rs | 10 +++++ 4 files changed, 79 insertions(+), 2 deletions(-) diff --git a/epiphany-core/src/bin/epiphany-swarm.rs b/epiphany-core/src/bin/epiphany-swarm.rs index decd4522..298f3224 100644 --- a/epiphany-core/src/bin/epiphany-swarm.rs +++ b/epiphany-core/src/bin/epiphany-swarm.rs @@ -12,7 +12,7 @@ use epiphany_core::{ ingest_resident_self_coordinator_continuation_pressure, ingest_resident_self_domain_pressure, load_epiphany_cultmesh_swarm_brake, live_resident_self_typed_request_ids, load_resident_self_state, observe_process_instance, - pending_resident_self_acks, + pending_resident_self_acks, reap_exited_child_process, prepare_resident_self_launch, publish_resident_provider_readiness, resident_cognitive_runtime_id, resident_prepared_launch_thread_id, resident_self_child_claim, @@ -21,7 +21,8 @@ use epiphany_core::{ retain_coordinator_run_receipts, retain_failed_runtime_worker_attempts, retain_fulfilled_runtime_worker_attempts, retain_resident_self_lifecycles, settle_resident_self_exited_coordinator, - settle_resident_self_receipt_free_dead_coordinator, terminate_process_instance, + runtime_worker_process_claims, settle_resident_self_receipt_free_dead_coordinator, + terminate_process_instance, validate_persona_feedback_store_separation, validate_resident_self_coordinator_receipt_binding, validate_resident_self_store_separation, @@ -310,6 +311,7 @@ fn cycle( ports: &mut NativePorts, shutdown_requested: bool, ) -> Result { + ports.reap_adopted_worker_children()?; let now = Utc::now().timestamp_millis().max(0) as u64; let brake_engaged = ports.brake_engaged()?; let cognitive_runtime_id = ports.cognitive_runtime_id().to_string(); @@ -670,6 +672,16 @@ impl<'a> NativePorts<'a> { fn cognitive_runtime_id(&self) -> &str { &self.cognitive_runtime_id } + + fn reap_adopted_worker_children(&mut self) -> Result { + let mut reaped = 0; + for claim in runtime_worker_process_claims(&self.policy.runtime_store)? { + if reap_exited_child_process(claim.process_id)? { + reaped += 1; + } + } + Ok(reaped) + } } impl ResidentSelfPorts for NativePorts<'_> { diff --git a/epiphany-core/src/lib.rs b/epiphany-core/src/lib.rs index c3e87811..a789c044 100644 --- a/epiphany-core/src/lib.rs +++ b/epiphany-core/src/lib.rs @@ -964,6 +964,7 @@ pub use process_observation::native_boot_identity; pub use process_observation::native_process_executable_path; pub use process_observation::observe_native_process; pub use process_observation::observe_process_instance; +pub use process_observation::reap_exited_child_process; pub use process_observation::terminate_process_instance; pub use promotion::EpiphanyPromotionDecision; pub use promotion::EpiphanyPromotionInput; @@ -1009,6 +1010,7 @@ pub use runtime_spine::EpiphanyRuntimeSwarmBinding; pub use runtime_spine::EpiphanyRuntimeToolExecutionBinding; pub use runtime_spine::EpiphanyRuntimeWorkerLaunchRequest; pub use runtime_spine::EpiphanyRuntimeWorkerProcessClaim; +pub use runtime_spine::runtime_worker_process_claims; pub use runtime_spine::EpiphanyArchivedRuntimeWorkerAttempt; pub use runtime_spine::EpiphanyToolInvocationStatus; pub use runtime_spine::PreparedRuntimeSpineHeartbeatJob; diff --git a/epiphany-core/src/process_observation.rs b/epiphany-core/src/process_observation.rs index cacd1fa1..32f6d70a 100644 --- a/epiphany-core/src/process_observation.rs +++ b/epiphany-core/src/process_observation.rs @@ -76,6 +76,38 @@ pub fn terminate_process_instance(expected: &ProcessInstanceIdentity) -> Result< platform::terminate(expected) } +/// Reap one exact exited child of the current process without waiting. +/// +/// This is OS lifecycle cleanup only. Callers must supply identity from their +/// own typed authority; this function does not discover or classify processes. +#[cfg(unix)] +pub fn reap_exited_child_process(process_id: u32) -> Result { + let pid = i32::try_from(process_id).context("child process id exceeds Unix pid range")?; + loop { + let mut status = 0; + let observed = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) }; + if observed == pid { + return Ok(true); + } + if observed == 0 { + return Ok(false); + } + let error = std::io::Error::last_os_error(); + if error.kind() == std::io::ErrorKind::Interrupted { + continue; + } + if error.raw_os_error() == Some(libc::ECHILD) { + return Ok(false); + } + return Err(error).context("failed to reap exact exited child process"); + } +} + +#[cfg(windows)] +pub fn reap_exited_child_process(_process_id: u32) -> Result { + Ok(false) +} + /// Returns a boot-incarnation token only when the platform can prove it. pub fn native_boot_identity() -> Option { platform::boot_identity() @@ -589,4 +621,25 @@ mod tests { } | ProcessInstanceObservation::Missing )); } + + #[cfg(unix)] + #[test] + fn exact_exited_child_is_reaped_once_without_wildcard_waiting() { + let child = std::process::Command::new("sh") + .args(["-c", "exit 0"]) + .spawn() + .unwrap(); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + if reap_exited_child_process(child.id()).unwrap() { + break; + } + assert!( + std::time::Instant::now() < deadline, + "exact exited child was not reaped" + ); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + assert!(!reap_exited_child_process(child.id()).unwrap()); + } } diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index d60b9530..425cb396 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -3527,6 +3527,16 @@ pub fn runtime_worker_process_claim( cache.get::(&worker_process_claim_id(job_id)) } +pub fn runtime_worker_process_claims( + store_path: impl AsRef, +) -> Result> { + let mut cache = runtime_spine_cache(store_path)?; + cache.pull_all_backing_stores()?; + let mut claims = cache.get_all::()?; + claims.sort_by(|left, right| left.job_id.cmp(&right.job_id)); + Ok(claims) +} + pub fn claim_runtime_worker_process( store_path: impl AsRef, job_id: &str,