Reap adopted model worker children

This commit is contained in:
Meta Vens 2026-08-11 19:57:01 +02:00
commit 465af24d03
4 changed files with 79 additions and 2 deletions

View file

@ -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<ResidentSelfOutcome> {
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<usize> {
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<'_> {

View file

@ -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;

View file

@ -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<bool> {
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<bool> {
Ok(false)
}
/// Returns a boot-incarnation token only when the platform can prove it.
pub fn native_boot_identity() -> Option<String> {
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());
}
}

View file

@ -3527,6 +3527,16 @@ pub fn runtime_worker_process_claim(
cache.get::<EpiphanyRuntimeWorkerProcessClaim>(&worker_process_claim_id(job_id))
}
pub fn runtime_worker_process_claims(
store_path: impl AsRef<Path>,
) -> Result<Vec<EpiphanyRuntimeWorkerProcessClaim>> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let mut claims = cache.get_all::<EpiphanyRuntimeWorkerProcessClaim>()?;
claims.sort_by(|left, right| left.job_id.cmp(&right.job_id));
Ok(claims)
}
pub fn claim_runtime_worker_process(
store_path: impl AsRef<Path>,
job_id: &str,