From d2ca66301fb6af4e7d2d27fff0b772b0f0fccdf4 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 22 Aug 2026 18:28:36 +0200 Subject: [PATCH] Recover dead worker attempts from durable state --- epiphany-core/src/bin/epiphany-swarm.rs | 18 +- epiphany-core/src/current_work.rs | 88 ++++++- epiphany-core/src/resident_self.rs | 46 +++- epiphany-core/src/runtime_spine.rs | 323 ++++++++++++++++++++++-- 4 files changed, 445 insertions(+), 30 deletions(-) diff --git a/epiphany-core/src/bin/epiphany-swarm.rs b/epiphany-core/src/bin/epiphany-swarm.rs index a6603bb1..70cc3bf3 100644 --- a/epiphany-core/src/bin/epiphany-swarm.rs +++ b/epiphany-core/src/bin/epiphany-swarm.rs @@ -13,14 +13,15 @@ use epiphany_core::{ load_epiphany_cultmesh_swarm_brake, load_resident_self_state, materialize_resident_self_domain_obligations, observe_process_instance, pending_resident_self_acks, prepare_resident_self_launch, publish_resident_provider_readiness, - reap_exited_child_process, resident_cognitive_runtime_id, resident_prepared_launch_thread_id, - resident_self_child_claim, resident_self_local_provider_status, - retain_completed_runtime_sessions, retain_coordinator_run_receipts, - retain_failed_runtime_worker_attempts, retain_fulfilled_runtime_worker_attempts, - retain_resident_self_lifecycles, runtime_worker_process_claims, - settle_resident_self_exited_coordinator, 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, + reap_exited_child_process, recover_dead_runtime_worker_attempts, resident_cognitive_runtime_id, + resident_prepared_launch_thread_id, resident_self_child_claim, + resident_self_local_provider_status, retain_completed_runtime_sessions, + retain_coordinator_run_receipts, retain_failed_runtime_worker_attempts, + retain_fulfilled_runtime_worker_attempts, retain_resident_self_lifecycles, + runtime_worker_process_claims, settle_resident_self_exited_coordinator, + 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, }; use serde_json::json; use std::collections::{BTreeMap, BTreeSet}; @@ -305,6 +306,7 @@ fn cycle( ) -> Result { ports.reap_adopted_worker_children()?; let now = Utc::now().timestamp_millis().max(0) as u64; + recover_dead_runtime_worker_attempts(&args.policy.runtime_store, now)?; let brake_engaged = ports.brake_engaged()?; let cognitive_runtime_id = ports.cognitive_runtime_id().to_string(); run_feedback_import_if_released(brake_engaged || shutdown_requested, || { diff --git a/epiphany-core/src/current_work.rs b/epiphany-core/src/current_work.rs index 4ab04a41..f084d709 100644 --- a/epiphany-core/src/current_work.rs +++ b/epiphany-core/src/current_work.rs @@ -3494,12 +3494,82 @@ mod tests { assert_eq!(crate::repository_body_read_counters(), (0, 0)); let mut cache = crate::runtime_spine_cache(&store)?; cache.pull_all_backing_stores()?; - let mut failed_job = cache - .get::("body-scheduled-job")? - .expect("scheduled Body job exists"); - failed_job.status = crate::EpiphanyRuntimeJobStatus::Failed; - failed_job.updated_at = "2026-08-17T00:00:03Z".into(); - cache.put(&failed_job.job_id, &failed_job)?; + let failed_launch = cache + .get::("body-scheduled-job")? + .expect("scheduled Body launch exists"); + drop(cache); + let failed_basis = crate::worker_reasoning_basis(&store, &failed_launch)?; + crate::put_reasoning_basis(&store, &failed_basis)?; + let mut failed_native = epiphany_model_adapter::EpiphanyModelRequest::new( + "body-failed-request", + "body-failed-conversation", + "openai-codex", + "gpt-test", + "model", + ); + failed_native.reasoning_basis_id = Some(failed_basis.basis_id.clone()); + failed_native.source_worker_job_id = Some(failed_launch.job_id.clone()); + crate::open_runtime_model_execution( + &store, + crate::RuntimeSpineSessionOptions { + session_id: "body-failed-model-session".into(), + objective: "Model the exact Body projection.".into(), + created_at: "2026-08-17T00:00:02.100Z".into(), + coordinator_note: "Test dead worker audit context.".into(), + }, + crate::RuntimeSpineJobOptions { + job_id: "body-failed-model-job".into(), + session_id: "body-failed-model-session".into(), + role: "openai-model".into(), + created_at: "2026-08-17T00:00:02.100Z".into(), + summary: "Bound Body inference.".into(), + artifact_refs: Vec::new(), + }, + &failed_native, + "2026-08-17T00:00:02.100Z", + )?; + let failed_process = crate::ProcessInstanceIdentity { + process_id: u32::MAX, + creation_token: 7, + created_at_rfc3339: Some("2026-08-17T00:00:02.200Z".into()), + executable_path: "body-failed-worker".into(), + }; + let failed_activation = "body-failed-activation"; + crate::claim_runtime_worker_process( + &store, + &failed_launch.job_id, + &failed_process, + &format!("{:x}", Sha256::digest(failed_activation.as_bytes())), + "2026-08-17T00:00:02.200Z", + )?; + crate::activate_runtime_worker_process( + &store, + &failed_launch.job_id, + &failed_process, + failed_activation, + "2026-08-17T00:00:02.300Z", + )?; + assert_eq!( + crate::recover_dead_runtime_worker_attempts(&store, 1_871_270_403_000)?, + 1 + ); + let mut death_cache = crate::runtime_spine_cache(&store)?; + death_cache.pull_all_backing_stores()?; + let death = death_cache + .get::("result-worker-death-body-scheduled-job")? + .expect("worker death emits its terminal runtime result"); + assert_eq!(death.verdict, "failed"); + let death_context = death + .decision_context_id + .as_deref() + .expect("model-backed death preserves its decision context"); + assert!(crate::audit_decision_context(&store, death_context).is_ok()); + assert_eq!( + crate::runtime_worker_process_claim(&store, &failed_launch.job_id)? + .expect("dead worker claim remains auditable") + .status, + crate::WorkerProcessStatus::TerminalDeath.as_str() + ); let retry_work = project_current_work(&store)?; assert_eq!( retry_work @@ -3879,6 +3949,12 @@ mod tests { decision_context_id: Some(proposal_context.context_id.clone()), }, )?; + assert_eq!( + crate::runtime_worker_process_claim(&store, &proposal_launch.job_id)? + .expect("structured Modeling result terminalizes its process claim") + .status, + crate::WorkerProcessStatus::TerminalResult.as_str() + ); assert_eq!( project_current_work(&store)? .proposal_modeling diff --git a/epiphany-core/src/resident_self.rs b/epiphany-core/src/resident_self.rs index a46072cc..b1a32f28 100644 --- a/epiphany-core/src/resident_self.rs +++ b/epiphany-core/src/resident_self.rs @@ -1640,7 +1640,7 @@ pub fn recover_dead_resident_typed_worker( chrono::DateTime::::from_timestamp_millis(recovered_at_millis as i64) .ok_or_else(|| anyhow!("typed worker death timestamp is out of range"))? .to_rfc3339_opts(chrono::SecondsFormat::Millis, true); - crate::runtime_spine::terminalize_runtime_worker_process_death( + crate::runtime_spine::terminalize_dead_runtime_worker_attempt( runtime_store, &launch.job_id, &format!("worker-death-recovery-{}", launch.job_id), @@ -1651,6 +1651,50 @@ pub fn recover_dead_resident_typed_worker( } } +/// Reconciles worker physiology against durable attempt state without relying +/// on a still-live coordinator grant. Exact alive processes remain owners; +/// inaccessible identities fail closed; exact death either admits an already +/// sealed structured outcome or terminalizes the attempt as a typed failure. +pub fn recover_dead_runtime_worker_attempts( + runtime_store: &Path, + recovered_at_millis: u64, +) -> Result { + let recovered_at = + chrono::DateTime::::from_timestamp_millis(recovered_at_millis as i64) + .ok_or_else(|| anyhow!("worker recovery timestamp is out of range"))? + .to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let mut recovered = 0; + for claim in crate::runtime_worker_process_claims(runtime_store)? { + let status = crate::WorkerProcessStatus::parse(&claim.status)?; + if !status.is_live() { + continue; + } + let identity = crate::ProcessInstanceIdentity { + process_id: claim.process_id, + creation_token: claim.process_creation_token, + created_at_rfc3339: None, + executable_path: PathBuf::from(&claim.process_executable_path), + }; + match crate::observe_process_instance(&identity) { + crate::ProcessInstanceObservation::ExactAlive + | crate::ProcessInstanceObservation::Inaccessible + | crate::ProcessInstanceObservation::Indeterminate { .. } => {} + crate::ProcessInstanceObservation::ExactExited { .. } + | crate::ProcessInstanceObservation::Missing + | crate::ProcessInstanceObservation::Replaced { .. } => { + crate::runtime_spine::terminalize_dead_runtime_worker_attempt( + runtime_store, + &claim.job_id, + &format!("worker-death-recovery-{}", claim.job_id), + &recovered_at, + )?; + recovered += 1; + } + } + } + Ok(recovered) +} + pub fn resident_self_grant_has_typed_request( resident_store: &Path, grant_id: &str, diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index fc4400de..2a3a9f9a 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -4800,12 +4800,125 @@ pub fn abandon_unactivated_runtime_worker_process( } } -pub(crate) fn terminalize_runtime_worker_process_death( +fn terminal_worker_request_id(cache: &CultCache, worker_job_id: &str) -> Result> { + let bindings = cache + .get_all::()? + .into_iter() + .filter(|binding| binding.source_worker_job_id.as_deref() == Some(worker_job_id)) + .map(|binding| { + let validated = validate_runtime_model_execution_binding(cache, &binding.request_id)?; + let job = cache + .get::(&validated.job_id)? + .ok_or_else(|| anyhow!("worker model execution lost its runtime job"))?; + let request = cache + .get::(&validated.request_id)? + .ok_or_else(|| anyhow!("worker model execution lost its native request"))?; + Ok((validated, job, request)) + }) + .collect::>>()?; + if bindings.is_empty() { + return Ok(None); + } + + let live = bindings + .iter() + .filter(|(_, job, _)| { + matches!( + job.status, + EpiphanyRuntimeJobStatus::Queued + | EpiphanyRuntimeJobStatus::Running + | EpiphanyRuntimeJobStatus::WaitingForReview + ) + }) + .collect::>(); + if live.len() > 1 { + return Err(anyhow!( + "worker attempt has multiple live model request authorities" + )); + } + if let Some((binding, _, _)) = live.first() { + return Ok(Some(binding.request_id.clone())); + } + + // Tool continuations are self-contained requests whose ordered input is a + // strict extension of the request that preceded them. This finds the + // unique terminal request from request bytes, without granting timestamps + // or provider event order causal authority. + let terminal = bindings + .iter() + .filter(|(_, _, candidate)| { + bindings.iter().all(|(_, _, other)| { + candidate.request_id == other.request_id + || (candidate.input.len() > other.input.len() + && candidate.input.starts_with(&other.input)) + }) + }) + .collect::>(); + match terminal.as_slice() { + [(binding, _, _)] => Ok(Some(binding.request_id.clone())), + [] => Err(anyhow!( + "worker model request family has no unique terminal self-contained request" + )), + _ => Err(anyhow!( + "worker model request family has multiple terminal request authorities" + )), + } +} + +fn complete_persisted_worker_outcome( + store_path: &Path, + job_id: &str, + completed_at: &str, +) -> Result> { + let mut cache = runtime_spine_cache(store_path)?; + cache.pull_all_backing_stores()?; + let role = cache.get::(job_id)?; + let reorient = cache.get::(job_id)?; + match (role, reorient) { + (Some(_), Some(_)) => Err(anyhow!( + "worker attempt has both role and reorientation terminal outcomes" + )), + (Some(result), None) => complete_runtime_job( + store_path, + RuntimeSpineJobResultOptions { + result_id: result.result_id, + job_id: job_id.to_string(), + completed_at: completed_at.to_string(), + verdict: result.verdict, + summary: result.summary, + next_safe_move: result.next_safe_move, + evidence_refs: result.evidence_ids, + artifact_refs: result.artifact_refs, + decision_context_id: Some(result.decision_context_id), + }, + ) + .map(Some), + (None, Some(result)) => complete_runtime_job( + store_path, + RuntimeSpineJobResultOptions { + result_id: result.result_id, + job_id: job_id.to_string(), + completed_at: completed_at.to_string(), + verdict: result.mode, + summary: result.summary, + next_safe_move: result.next_safe_move, + evidence_refs: result.evidence_ids, + artifact_refs: result.artifact_refs, + decision_context_id: Some(result.decision_context_id), + }, + ) + .map(Some), + (None, None) => Ok(None), + } +} + +fn commit_runtime_worker_process_death( store_path: impl AsRef, job_id: &str, terminal_authority_id: &str, terminal_at: &str, -) -> Result { + decision_context_id: Option<&str>, +) -> Result { let store_path = store_path.as_ref(); validate_non_empty(terminal_authority_id, "worker death terminal authority")?; chrono::DateTime::parse_from_rfc3339(terminal_at) @@ -4820,7 +4933,22 @@ pub(crate) fn terminalize_runtime_worker_process_death( if status == WorkerProcessStatus::TerminalDeath && current.terminal_authority_id.as_deref() == Some(terminal_authority_id) { - return Ok(current); + let result_id = format!("result-worker-death-{job_id}"); + let result = cache + .get::(&result_id)? + .ok_or_else(|| anyhow!("worker death claim lost its terminal job result"))?; + let job = cache + .get::(job_id)? + .ok_or_else(|| anyhow!("worker death claim lost its runtime job"))?; + if job.status != EpiphanyRuntimeJobStatus::Failed + || result.job_id != job_id + || result.decision_context_id.as_deref() != decision_context_id + { + return Err(anyhow!( + "worker death replay found split terminal authority" + )); + } + return Ok(result); } if !status.is_live() { return Err(anyhow!( @@ -4830,9 +4958,36 @@ pub(crate) fn terminalize_runtime_worker_process_death( if cache .get::(job_id)? .is_some() + || cache + .get::(job_id)? + .is_some() { - return Err(anyhow!("worker death recovery races a terminal result")); + return Err(anyhow!("worker death recovery races a terminal outcome")); } + let mut job = cache + .get::(job_id)? + .ok_or_else(|| anyhow!("worker death recovery requires its runtime job"))?; + if !matches!( + job.status, + EpiphanyRuntimeJobStatus::Queued + | EpiphanyRuntimeJobStatus::Running + | EpiphanyRuntimeJobStatus::WaitingForReview + ) { + return Err(anyhow!( + "worker death recovery found a terminal runtime job" + )); + } + let existing_results = cache + .get_all::()? + .into_iter() + .filter(|result| result.job_id == job_id) + .collect::>(); + if !existing_results.is_empty() { + return Err(anyhow!("live worker death recovery found a job result")); + } + let job_envelope = cache + .get_envelope::(job_id)? + .ok_or_else(|| anyhow!("worker death recovery lost its job envelope"))?; let current_envelope = cache .get_envelope::(&claim_id)? .ok_or_else(|| anyhow!("worker death recovery lost its claim envelope"))?; @@ -4840,16 +4995,146 @@ pub(crate) fn terminalize_runtime_worker_process_death( next.status = WorkerProcessStatus::TerminalDeath.as_str().into(); next.terminal_at = Some(terminal_at.into()); next.terminal_authority_id = Some(terminal_authority_id.into()); - let next_envelope = cache.prepare_entry(&claim_id, &next)?.0; - if SingleFileMessagePackBackingStore::new(store_path) - .compare_and_swap_batch(&[current_envelope], vec![next_envelope])? + let summary = "Runtime Continuity proved the exact worker process terminal before a structured outcome was admitted.".to_string(); + job.status = EpiphanyRuntimeJobStatus::Failed; + job.updated_at = terminal_at.to_string(); + job.summary = summary.clone(); + let result = EpiphanyRuntimeJobResult { + schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(), + result_id: format!("result-worker-death-{job_id}"), + job_id: job_id.to_string(), + session_id: job.session_id.clone(), + role: job.role.clone(), + verdict: "failed".to_string(), + summary: summary.clone(), + completed_at: terminal_at.to_string(), + next_safe_move: "Derive a fresh attempt from current typed work; never rebase the abandoned model output.".to_string(), + evidence_refs: Vec::new(), + artifact_refs: Vec::new(), + metadata: BTreeMap::new(), + decision_context_id: decision_context_id.map(str::to_string), + }; + let event = EpiphanyRuntimeEvent { + schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(), + event_id: format!("event-job-completed-{job_id}"), + occurred_at: terminal_at.to_string(), + event_type: "job.completed".to_string(), + source: "runtime-continuity".to_string(), + session_id: Some(job.session_id.clone()), + job_id: Some(job_id.to_string()), + summary, + metadata: BTreeMap::from([("resultId".to_string(), result.result_id.clone())]), + }; + if cache + .get::(&event.event_id)? + .is_some() + || cache + .get::(&result.result_id)? + .is_some() { - Ok(next) - } else { - Err(anyhow!( - "worker death recovery lost its exact claim snapshot" - )) + return Err(anyhow!("worker death terminal identities already exist")); } + let mut expected = vec![current_envelope, job_envelope]; + let mut unchanged_strong_reads = Vec::new(); + if let Some(context_id) = decision_context_id { + let context = cache + .get::(context_id)? + .ok_or_else(|| anyhow!("model-backed worker death lost its decision context"))?; + if context.native_request()?.source_worker_job_id.as_deref() != Some(job_id) { + return Err(anyhow!( + "worker death decision context belongs to another worker" + )); + } + let failures = cache + .get_all::()? + .into_iter() + .filter(|failure| { + failure.decision_context_id == context_id && failure.pass_id == job_id + }) + .collect::>(); + if failures.len() != 1 { + return Err(anyhow!( + "model-backed worker death requires one exact typed pass failure" + )); + } + let context_envelope = cache + .get_envelope::(context_id)? + .ok_or_else(|| anyhow!("worker death lost its context envelope"))?; + let failure_envelope = cache + .get_envelope::(&failures[0].failure_id)? + .ok_or_else(|| anyhow!("worker death lost its pass-failure envelope"))?; + expected.extend([context_envelope.clone(), failure_envelope.clone()]); + unchanged_strong_reads.extend([context_envelope, failure_envelope]); + } else if cache + .get_all::()? + .iter() + .any(|binding| binding.source_worker_job_id.as_deref() == Some(job_id)) + { + return Err(anyhow!( + "model-backed worker death cannot terminalize without its exact context" + )); + } + let mut writes = vec![ + cache.prepare_entry(&claim_id, &next)?.0, + cache.prepare_entry(job_id, &job)?.0, + cache.prepare_entry(&result.result_id, &result)?.0, + cache.prepare_entry(&event.event_id, &event)?.0, + ]; + writes.extend(unchanged_strong_reads); + if !SingleFileMessagePackBackingStore::new(store_path) + .compare_and_swap_batch(&expected, writes)? + { + return commit_runtime_worker_process_death( + store_path, + job_id, + terminal_authority_id, + terminal_at, + decision_context_id, + ); + } + Ok(result) +} + +pub(crate) fn terminalize_dead_runtime_worker_attempt( + store_path: impl AsRef, + job_id: &str, + terminal_authority_id: &str, + terminal_at: &str, +) -> Result { + let store_path = store_path.as_ref(); + if let Some(result) = complete_persisted_worker_outcome(store_path, job_id, terminal_at)? { + return Ok(result); + } + let mut cache = runtime_spine_cache(store_path)?; + cache.pull_all_backing_stores()?; + let terminal_request_id = terminal_worker_request_id(&cache, job_id)?; + drop(cache); + let decision_context_id = if let Some(request_id) = terminal_request_id { + if let Some(failure) = model_pass_failure_for_request(store_path, &request_id)? { + Some(failure.decision_context_id) + } else { + let context = crate::seal_model_decision_context(store_path, &request_id)?; + let failure = terminalize_model_pass_failure_session( + store_path, + ModelPassFailureTerminalOptions { + decision_context_id: context.context_id, + failure_kind: "worker_process_death".to_string(), + summary: "Runtime Continuity proved the exact model worker process terminal before a structured outcome was admitted.".to_string(), + failed_at: terminal_at.to_string(), + }, + )?; + Some(failure.decision_context_id) + } + } else { + None + }; + commit_runtime_worker_process_death( + store_path, + job_id, + terminal_authority_id, + terminal_at, + decision_context_id.as_deref(), + ) } pub fn put_runtime_role_worker_result( @@ -12340,11 +12625,19 @@ pub fn complete_runtime_job( .cloned() .ok_or_else(|| anyhow!("runtime worker completion lost its claim envelope"))?; let mut terminal = claim; - if let Some(reorient) = - cache.get::(&result.job_id)? + let role = cache.get::(&result.job_id)?; + let reorient = cache.get::(&result.job_id)?; + if role.is_some() && reorient.is_some() { + return Err(anyhow!( + "runtime worker job has both role and reorientation terminal outcomes" + )); + } + if let Some(terminal_result_id) = role + .map(|result| result.result_id) + .or_else(|| reorient.map(|result| result.result_id)) { terminal.status = crate::WorkerProcessStatus::TerminalResult.as_str().into(); - terminal.terminal_authority_id = Some(reorient.result_id); + terminal.terminal_authority_id = Some(terminal_result_id); } else { terminal.status = crate::WorkerProcessStatus::TerminalFailure.as_str().into(); terminal.terminal_authority_id = Some(result.result_id.clone());