GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Recover dead worker attempts from durable state

Browse source
This commit is contained in:
Meta Vens 2026-08-22 18:28:36 +02:00
commit d2ca66301f
4 changed files with 445 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

18
epiphany-core/src/bin/epiphany-swarm.rs
View file

@ -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<ResidentSelfOutcome> {
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, || {

88
epiphany-core/src/current_work.rs
View file

@ -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::<crate::EpiphanyRuntimeJob>("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::<crate::EpiphanyRuntimeWorkerLaunchRequest>("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::<crate::EpiphanyRuntimeJobResult>("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

46
epiphany-core/src/resident_self.rs
View file

@ -1640,7 +1640,7 @@ pub fn recover_dead_resident_typed_worker(
chrono::DateTime::<chrono::Utc>::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<usize> {
let recovered_at =
chrono::DateTime::<chrono::Utc>::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,

323
epiphany-core/src/runtime_spine.rs
View file

@ -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<Option<String>> {
let bindings = cache
.get_all::<EpiphanyRuntimeModelExecutionBinding>()?
.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::<EpiphanyRuntimeJob>(&validated.job_id)?
.ok_or_else(|| anyhow!("worker model execution lost its runtime job"))?;
let request = cache
.get::<EpiphanyModelRequest>(&validated.request_id)?
.ok_or_else(|| anyhow!("worker model execution lost its native request"))?;
Ok((validated, job, request))
})
.collect::<Result<Vec<_>>>()?;
if bindings.is_empty() {
return Ok(None);
}
let live = bindings
.iter()
.filter(|(_, job, _)| {
matches!(
job.status,
EpiphanyRuntimeJobStatus::Queued
| EpiphanyRuntimeJobStatus::Running
| EpiphanyRuntimeJobStatus::WaitingForReview
)
})
.collect::<Vec<_>>();
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::<Vec<_>>();
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<Option<EpiphanyRuntimeJobResult>> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let role = cache.get::<EpiphanyRuntimeRoleWorkerResult>(job_id)?;
let reorient = cache.get::<EpiphanyRuntimeReorientWorkerResult>(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<Path>,
job_id: &str,
terminal_authority_id: &str,
terminal_at: &str,
) -> Result<EpiphanyRuntimeWorkerProcessClaim> {
decision_context_id: Option<&str>,
) -> Result<EpiphanyRuntimeJobResult> {
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::<EpiphanyRuntimeJobResult>(&result_id)?
.ok_or_else(|| anyhow!("worker death claim lost its terminal job result"))?;
let job = cache
.get::<EpiphanyRuntimeJob>(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::<EpiphanyRuntimeRoleWorkerResult>(job_id)?
.is_some()
|| cache
.get::<EpiphanyRuntimeReorientWorkerResult>(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::<EpiphanyRuntimeJob>(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::<EpiphanyRuntimeJobResult>()?
.into_iter()
.filter(|result| result.job_id == job_id)
.collect::<Vec<_>>();
if !existing_results.is_empty() {
return Err(anyhow!("live worker death recovery found a job result"));
}
let job_envelope = cache
.get_envelope::<EpiphanyRuntimeJob>(job_id)?
.ok_or_else(|| anyhow!("worker death recovery lost its job envelope"))?;
let current_envelope = cache
.get_envelope::<EpiphanyRuntimeWorkerProcessClaim>(&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::<EpiphanyRuntimeEvent>(&event.event_id)?
.is_some()
|| cache
.get::<EpiphanyRuntimeJobResult>(&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::<crate::EpiphanyDecisionContext>(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::<crate::EpiphanyModelPassFailure>()?
.into_iter()
.filter(|failure| {
failure.decision_context_id == context_id && failure.pass_id == job_id
})
.collect::<Vec<_>>();
if failures.len() != 1 {
return Err(anyhow!(
"model-backed worker death requires one exact typed pass failure"
));
}
let context_envelope = cache
.get_envelope::<crate::EpiphanyDecisionContext>(context_id)?
.ok_or_else(|| anyhow!("worker death lost its context envelope"))?;
let failure_envelope = cache
.get_envelope::<crate::EpiphanyModelPassFailure>(&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::<EpiphanyRuntimeModelExecutionBinding>()?
.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<Path>,
job_id: &str,
terminal_authority_id: &str,
terminal_at: &str,
) -> Result<EpiphanyRuntimeJobResult> {
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::<EpiphanyRuntimeReorientWorkerResult>(&result.job_id)?
let role = cache.get::<EpiphanyRuntimeRoleWorkerResult>(&result.job_id)?;
let reorient = cache.get::<EpiphanyRuntimeReorientWorkerResult>(&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());