From 7b404d130d6b37c8aa22e9a54e0010de2900a75d Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 12 Aug 2026 02:05:24 +0200 Subject: [PATCH] Extract worker attempt authority --- epiphany-core/src/coordinator_launch.rs | 27 ++-- epiphany-core/src/lib.rs | 3 +- epiphany-core/src/resident_self.rs | 46 ++---- epiphany-core/src/runtime_spine.rs | 162 +++++++------------ epiphany-core/src/runtime_worker_attempt.rs | 164 ++++++++++++++++++++ notes/epiphany-current-algorithmic-map.md | 11 +- notes/fresh-workspace-handoff.md | 40 +++-- notes/perfect-machine-audit-roadmap.md | 13 +- state/ledgers.msgpack | Bin 618834 -> 619284 bytes state/map.yaml | 2 +- 10 files changed, 280 insertions(+), 188 deletions(-) create mode 100644 epiphany-core/src/runtime_worker_attempt.rs diff --git a/epiphany-core/src/coordinator_launch.rs b/epiphany-core/src/coordinator_launch.rs index 9fcfb3a5..f50cccc1 100644 --- a/epiphany-core/src/coordinator_launch.rs +++ b/epiphany-core/src/coordinator_launch.rs @@ -971,7 +971,7 @@ fn validate_imagination_consideration_launch( .cloned() .collect::>(); for binding in prior_bindings { - if !worker_process_attempt_is_terminal(cache, &binding.job_id)? { + if !worker_attempt_is_retryable_terminal_failure(cache, &binding.job_id)? { return Err(anyhow!("consideration request already bound")); } } @@ -1041,7 +1041,7 @@ fn validate_admitted_model_direction_consideration_launch( .cloned() .collect::>(); for worker in prior_launches { - if !worker_process_attempt_is_terminal(cache, &worker.job_id)? { + if !worker_attempt_is_retryable_terminal_failure(cache, &worker.job_id)? { return Err(anyhow!("model direction consideration already bound")); } } @@ -1216,20 +1216,16 @@ fn proposal_modeling_launch_was_superseded( cache: &CultCache, binding: &RepoFrontierProposalModelingLaunchBinding, ) -> Result { - worker_process_attempt_is_terminal(cache, &binding.job_id) + worker_attempt_is_retryable_terminal_failure(cache, &binding.job_id) } -fn worker_process_attempt_is_terminal(cache: &CultCache, job_id: &str) -> Result { - Ok(cache +fn worker_attempt_is_retryable_terminal_failure(cache: &CultCache, job_id: &str) -> Result { + cache .get::(&format!( - "runtime-worker-process-{job_id}" - ))? - .is_some_and(|claim| { - matches!( - claim.status.as_str(), - "terminal-death" | "terminal-unactivated" | "terminal-failure" - ) - })) + "runtime-worker-process-{job_id}" + ))? + .map(|claim| Ok(crate::WorkerProcessStatus::parse(&claim.status)?.allows_retry())) + .unwrap_or(Ok(false)) } fn terminal_runtime_link_for_binding( @@ -1259,10 +1255,7 @@ fn terminal_runtime_link_for_binding( let Some(claim) = crate::runtime_worker_process_claim(runtime_store, &link.runtime_job_id)? else { return Ok(None); }; - if !matches!( - claim.status.as_str(), - "terminal-death" | "terminal-unactivated" | "terminal-failure" - ) { + if !crate::WorkerProcessStatus::parse(&claim.status)?.is_failed_terminal() { return Ok(None); } let terminal_authority_id = claim diff --git a/epiphany-core/src/lib.rs b/epiphany-core/src/lib.rs index 41ab28ee..5b16b7a1 100644 --- a/epiphany-core/src/lib.rs +++ b/epiphany-core/src/lib.rs @@ -45,6 +45,7 @@ mod resident_self; mod runtime_spine; mod runtime_store_backend; mod runtime_store_migration; +mod runtime_worker_attempt; mod semantic_backend; mod semantic_projector_service; mod soul_gateway; @@ -1055,7 +1056,7 @@ pub use runtime_spine::RuntimeSpineJobResultOptions; pub use runtime_spine::RuntimeSpineSessionClosureOptions; pub use runtime_spine::RuntimeSpineSessionOptions; pub use runtime_spine::RuntimeTypedFulfillmentEvidence; -pub use runtime_spine::RuntimeTypedRequestRef; +pub use runtime_worker_attempt::{RuntimeTypedRequestRef, WorkerProcessStatus}; pub use runtime_spine::amend_repo_frontier_execution; pub use runtime_spine::append_runtime_event; pub use runtime_spine::abandon_unactivated_runtime_worker_process; diff --git a/epiphany-core/src/resident_self.rs b/epiphany-core/src/resident_self.rs index 1936f5b1..e73cec0c 100644 --- a/epiphany-core/src/resident_self.rs +++ b/epiphany-core/src/resident_self.rs @@ -1645,12 +1645,7 @@ pub fn live_resident_self_typed_request_ids(path: &Path) -> Result()? { if let Some(request) = resident_self_typed_request_ref(&grant)? { - let request_id = match request { - crate::RuntimeTypedRequestRef::ProposalModeling(id) - | crate::RuntimeTypedRequestRef::ImaginationConsideration(id) - | crate::RuntimeTypedRequestRef::AdmittedModelDirection(id) => id, - }; - request_ids.insert(request_id.to_string()); + request_ids.insert(request.request_id().to_string()); } } Ok(request_ids) @@ -1762,28 +1757,12 @@ pub fn recover_dead_resident_typed_worker( let Some(request) = resident_self_typed_request_ref(&grant)? else { return Ok(false); }; - let request_id = match request { - crate::RuntimeTypedRequestRef::ProposalModeling(id) - | crate::RuntimeTypedRequestRef::ImaginationConsideration(id) - | crate::RuntimeTypedRequestRef::AdmittedModelDirection(id) => id, - }; let mut cache = crate::runtime_spine_cache(runtime_store)?; cache.pull_all_backing_stores()?; let mut launches = cache .get_all::()? .into_iter() - .filter(|launch| match request { - crate::RuntimeTypedRequestRef::ProposalModeling(_) => { - launch.proposal_modeling_request_id.as_deref() == Some(request_id) - } - crate::RuntimeTypedRequestRef::ImaginationConsideration(_) => { - launch.imagination_consideration_request_id.as_deref() == Some(request_id) - } - crate::RuntimeTypedRequestRef::AdmittedModelDirection(_) => launch - .admitted_model_direction_consideration_request_id - .as_deref() - == Some(request_id), - }) + .filter(|launch| request.matches_launch(launch)) .collect::>(); launches.sort_by(|a, b| a.job_id.cmp(&b.job_id)); let mut retryable = Vec::new(); @@ -1792,14 +1771,15 @@ pub fn recover_dead_resident_typed_worker( "runtime-worker-process-{}", launch.job_id ))?; + if let Some(claim) = &claim { + crate::WorkerProcessStatus::parse(&claim.status)?; + } retryable.push((launch, claim)); } let live = retryable .iter() .filter(|(_, claim)| { - claim.as_ref().is_some_and(|claim| { - matches!(claim.status.as_str(), "claimed" | "active" | "terminal-result") - }) + claim.as_ref().is_some_and(|claim| crate::WorkerProcessStatus::parse(&claim.status).is_ok_and(|status| status.is_live() || status.is_fulfilled_terminal())) }) .count(); if live > 1 { @@ -1808,19 +1788,17 @@ pub fn recover_dead_resident_typed_worker( let selected = retryable .iter() .find(|(_, claim)| { - claim.as_ref().is_some_and(|claim| { - matches!(claim.status.as_str(), "claimed" | "active" | "terminal-result") - }) + claim.as_ref().is_some_and(|claim| crate::WorkerProcessStatus::parse(&claim.status).is_ok_and(|status| status.is_live() || status.is_fulfilled_terminal())) }) .or_else(|| retryable.last()); let Some((launch, Some(claim))) = selected else { return Ok(false); }; - match claim.status.as_str() { - "terminal-death" | "terminal-unactivated" | "terminal-failure" => return Ok(true), - "terminal-result" => return Ok(false), - "claimed" | "active" => {} - _ => return Err(anyhow!("typed worker process claim has invalid status")), + match crate::WorkerProcessStatus::parse(&claim.status)? { + status if status.allows_retry() => return Ok(true), + status if status.is_fulfilled_terminal() => return Ok(false), + status if status.is_live() => {} + _ => unreachable!("worker process status classes are exhaustive"), } let identity = crate::ProcessInstanceIdentity { process_id: claim.process_id, diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index 5ef068e7..c194c225 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -1,5 +1,6 @@ use crate::EpiphanyWorkerLaunchDocument; use crate::RepoFrontierPlanMindContextProjection; +use crate::{RuntimeTypedRequestRef, WorkerProcessStatus}; use crate::agent_launch::{ EPIPHANY_IMAGINATION_OWNER_ROLE, EPIPHANY_IMAGINATION_ROLE_BINDING_ID, EPIPHANY_MIND_OWNER_ROLE, EPIPHANY_MIND_ROLE_BINDING_ID, EPIPHANY_MODELING_OWNER_ROLE, @@ -3962,7 +3963,7 @@ pub fn claim_runtime_worker_process( process_creation_token: process.creation_token, process_executable_path: process.executable_path.display().to_string(), activation_token_sha256: activation_token_sha256.into(), - status: "claimed".into(), + status: WorkerProcessStatus::Claimed.as_str().into(), claimed_at: claimed_at.into(), activated_at: None, terminal_at: None, @@ -4011,17 +4012,18 @@ pub fn activate_runtime_worker_process( { return Err(anyhow!("worker activation does not bind the exact process claim")); } - if current.status == "active" { + let status = WorkerProcessStatus::parse(¤t.status)?; + if status == WorkerProcessStatus::Active { return Ok(current); } - if current.status != "claimed" { + if status != WorkerProcessStatus::Claimed { return Err(anyhow!("worker activation found terminal process authority")); } let current_envelope = cache .get_envelope::(&claim_id)? .ok_or_else(|| anyhow!("worker activation lost its claim envelope"))?; let mut next = current; - next.status = "active".into(); + next.status = WorkerProcessStatus::Active.as_str().into(); next.activated_at = Some(activated_at.into()); let next_envelope = cache.prepare_entry(&claim_id, &next)?.0; if SingleFileMessagePackBackingStore::new(store_path) @@ -4051,7 +4053,7 @@ pub fn abandon_unactivated_runtime_worker_process( if current.process_id != process.process_id || current.process_creation_token != process.creation_token || current.process_executable_path != process.executable_path.display().to_string() - || current.status != "claimed" + || WorkerProcessStatus::parse(¤t.status)? != WorkerProcessStatus::Claimed { return Err(anyhow!("worker abandonment does not own an unactivated exact claim")); } @@ -4059,7 +4061,7 @@ pub fn abandon_unactivated_runtime_worker_process( .get_envelope::(&claim_id)? .ok_or_else(|| anyhow!("worker abandonment lost its claim envelope"))?; let mut next = current; - next.status = "terminal-unactivated".into(); + next.status = WorkerProcessStatus::TerminalUnactivated.as_str().into(); next.terminal_at = Some(terminal_at.into()); next.terminal_authority_id = Some(format!("worker-unactivated-{job_id}")); let next_envelope = cache.prepare_entry(&claim_id, &next)?.0; @@ -4088,12 +4090,13 @@ pub(crate) fn terminalize_runtime_worker_process_death( let current = cache .get::(&claim_id)? .ok_or_else(|| anyhow!("worker death recovery requires its process claim"))?; - if current.status == "terminal-death" + let status = WorkerProcessStatus::parse(¤t.status)?; + if status == WorkerProcessStatus::TerminalDeath && current.terminal_authority_id.as_deref() == Some(terminal_authority_id) { return Ok(current); } - if !matches!(current.status.as_str(), "claimed" | "active") { + if !status.is_live() { return Err(anyhow!("worker death recovery found terminal process authority")); } if cache @@ -4106,7 +4109,7 @@ pub(crate) fn terminalize_runtime_worker_process_death( .get_envelope::(&claim_id)? .ok_or_else(|| anyhow!("worker death recovery lost its claim envelope"))?; let mut next = current; - next.status = "terminal-death".into(); + 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; @@ -4617,7 +4620,7 @@ pub fn put_runtime_role_worker_result( )); } if let Some(claim) = process_claim.as_ref() - && (claim.status != "terminal-result" + && (!crate::WorkerProcessStatus::parse(&claim.status)?.is_fulfilled_terminal() || claim.terminal_authority_id.as_deref() != Some(result.result_id.as_str())) { return Err(anyhow!( @@ -4650,7 +4653,7 @@ pub fn put_runtime_role_worker_result( let mut writes = vec![envelope]; let mut expected = Vec::new(); if let Some(claim) = process_claim.as_ref() { - if claim.status != "active" { + if crate::WorkerProcessStatus::parse(&claim.status)? != crate::WorkerProcessStatus::Active { return Err(anyhow!( "role worker result requires its active process claim" )); @@ -4659,7 +4662,7 @@ pub fn put_runtime_role_worker_result( .get_envelope::(&process_claim_id)? .ok_or_else(|| anyhow!("role worker result lost its process claim envelope"))?; let mut terminal_claim = claim.clone(); - terminal_claim.status = "terminal-result".into(); + terminal_claim.status = crate::WorkerProcessStatus::TerminalResult.as_str().into(); terminal_claim.terminal_at = Some(chrono::Utc::now().to_rfc3339()); terminal_claim.terminal_authority_id = Some(result.result_id.clone()); expected.push(current_envelope); @@ -4715,7 +4718,7 @@ pub fn put_runtime_role_worker_result( Some(_) => reloaded .get::(&process_claim_id)? .is_some_and(|claim| { - claim.status == "terminal-result" + crate::WorkerProcessStatus::parse(&claim.status).is_ok_and(|status| status.is_fulfilled_terminal()) && claim.terminal_authority_id.as_deref() == Some(result.result_id.as_str()) }), None => true, @@ -4743,13 +4746,6 @@ pub fn runtime_role_worker_result( cache.get::(job_id) } -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum RuntimeTypedRequestRef<'a> { - ProposalModeling(&'a str), - ImaginationConsideration(&'a str), - AdmittedModelDirection(&'a str), -} - #[derive(Clone, Debug, PartialEq, Eq)] pub struct RuntimeTypedFulfillmentEvidence { pub job_id: String, @@ -4980,11 +4976,7 @@ pub fn runtime_typed_request_fulfillment( request: RuntimeTypedRequestRef<'_>, ) -> Result> { let store_path = store_path.as_ref(); - let request_id = match request { - RuntimeTypedRequestRef::ProposalModeling(id) - | RuntimeTypedRequestRef::ImaginationConsideration(id) - | RuntimeTypedRequestRef::AdmittedModelDirection(id) => id, - }; + let request_id = request.request_id(); validate_non_empty(request_id, "typed fulfillment request id")?; let mut cache = runtime_spine_cache(store_path)?; cache.pull_all_backing_stores()?; @@ -4999,15 +4991,10 @@ pub fn runtime_typed_request_fulfillment( )); } if let Some(attempt) = archived_matches.first() { - let expected_kind = match request { - RuntimeTypedRequestRef::ProposalModeling(_) => "proposal-modeling", - RuntimeTypedRequestRef::ImaginationConsideration(_) => "imagination-consideration", - RuntimeTypedRequestRef::AdmittedModelDirection(_) => "admitted-model-direction", - }; if attempt.schema_version != ARCHIVED_RUNTIME_WORKER_ATTEMPT_SCHEMA_VERSION || attempt.archive_id != attempt.job_id - || attempt.request_kind != expected_kind - || attempt.terminal_process_status != "terminal-result" + || attempt.request_kind != request.kind() + || !crate::WorkerProcessStatus::parse(&attempt.terminal_process_status)?.is_fulfilled_terminal() || !attempt.retired_chain_digest.starts_with("sha256:") { return Err(anyhow!("archived typed fulfillment tombstone is invalid")); @@ -5027,20 +5014,7 @@ pub fn runtime_typed_request_fulfillment( let matches = cache .get_all::()? .into_iter() - .filter(|result| match request { - RuntimeTypedRequestRef::ProposalModeling(_) => { - result.proposal_modeling_request_id.as_deref() == Some(request_id) - } - RuntimeTypedRequestRef::ImaginationConsideration(_) => { - result.imagination_consideration_request_id.as_deref() == Some(request_id) - } - RuntimeTypedRequestRef::AdmittedModelDirection(_) => { - result - .admitted_model_direction_consideration_request_id - .as_deref() - == Some(request_id) - } - }) + .filter(|result| request.matches_result(result)) .collect::>(); if matches.is_empty() { return Ok(None); @@ -11205,44 +11179,26 @@ pub fn runtime_typed_request_attempt_exists( store_path: impl AsRef, request: RuntimeTypedRequestRef<'_>, ) -> Result { - let request_id = match request { - RuntimeTypedRequestRef::ProposalModeling(id) - | RuntimeTypedRequestRef::ImaginationConsideration(id) - | RuntimeTypedRequestRef::AdmittedModelDirection(id) => id, - }; + let request_id = request.request_id(); validate_non_empty(request_id, "typed attempt request id")?; let mut cache = runtime_spine_cache(store_path)?; cache.pull_all_backing_stores()?; let claims = cache .get_all::()? .into_iter() - .map(|claim| (claim.job_id.clone(), claim)) - .collect::>(); + .map(|claim| { + Ok((claim.job_id.clone(), WorkerProcessStatus::parse(&claim.status)?)) + }) + .collect::>>()?; Ok(cache .get_all::()? .iter() .filter(|launch| { - claims.get(&launch.job_id).is_none_or(|claim| { - !matches!( - claim.status.as_str(), - "terminal-death" | "terminal-unactivated" | "terminal-failure" - ) - }) + claims + .get(&launch.job_id) + .is_none_or(|status| !status.allows_retry()) }) - .any(|launch| match request { - RuntimeTypedRequestRef::ProposalModeling(_) => { - launch.proposal_modeling_request_id.as_deref() == Some(request_id) - } - RuntimeTypedRequestRef::ImaginationConsideration(_) => { - launch.imagination_consideration_request_id.as_deref() == Some(request_id) - } - RuntimeTypedRequestRef::AdmittedModelDirection(_) => { - launch - .admitted_model_direction_consideration_request_id - .as_deref() - == Some(request_id) - } - })) + .any(|launch| request.matches_launch(launch))) } pub fn archive_failed_runtime_worker_attempt( @@ -11448,15 +11404,9 @@ where } let launch = cache.get::(job_id)? .ok_or_else(|| anyhow!("worker attempt archive requires its immutable launch"))?; - let typed = [ - ("proposal-modeling", launch.proposal_modeling_request_id.as_deref()), - ("imagination-consideration", launch.imagination_consideration_request_id.as_deref()), - ("admitted-model-direction", launch.admitted_model_direction_consideration_request_id.as_deref()), - ].into_iter().filter_map(|(kind, id)| id.map(|id| (kind, id))).collect::>(); - if typed.len() != 1 { - return Err(anyhow!("worker attempt archive requires exactly one supported typed request")); - } - let (request_kind, request_id) = typed[0]; + let request = launch.typed_request_ref()?.ok_or_else(|| anyhow!("worker attempt archive requires exactly one supported typed request"))?; + let request_kind = request.kind(); + let request_id = request.request_id(); validate_archivable_typed_worker_launch(&cache, &launch, request_kind, request_id)?; if live_resident_request_ids.contains(request_id) { return Err(anyhow!("worker attempt archive refuses resident-live typed request")); @@ -11468,11 +11418,8 @@ where let claim_id = worker_process_claim_id(job_id); let claim = cache.get::(&claim_id)? .ok_or_else(|| anyhow!("worker attempt archive requires its process claim"))?; - let valid_claim = if fulfilled { - claim.status == "terminal-result" - } else { - matches!(claim.status.as_str(), "terminal-death" | "terminal-unactivated" | "terminal-failure") - }; + let status = crate::WorkerProcessStatus::parse(&claim.status)?; + let valid_claim = if fulfilled { status.is_fulfilled_terminal() } else { status.is_failed_terminal() }; if !valid_claim || claim.terminal_at.is_none() || claim.terminal_authority_id.is_none() { @@ -11576,18 +11523,20 @@ pub fn retain_failed_runtime_worker_attempts( require_identity(&cache)?; let launches = cache.get_all::()?.into_iter() .map(|launch| (launch.job_id.clone(), launch)).collect::>(); + let typed_request_ids = launches.values().map(|launch| Ok((launch.job_id.clone(), launch.typed_request_ref()?.map(|request| request.request_id().to_string())))).collect::>>()?; let jobs = cache.get_all::()?.into_iter() .map(|job| (job.job_id.clone(), job)).collect::>(); - let mut candidates = cache.get_all::()?.into_iter() - .filter(|claim| matches!(claim.status.as_str(), "terminal-death" | "terminal-unactivated" | "terminal-failure")) - .filter_map(|claim| { - let launch = launches.get(&claim.job_id)?; + let claims = cache.get_all::()?.into_iter() + .map(|claim| Ok((crate::WorkerProcessStatus::parse(&claim.status)?, claim))) + .collect::>>()?; + let mut candidates = claims.into_iter() + .filter(|(status, _)| status.is_failed_terminal()) + .filter_map(|(_, claim)| { let job = jobs.get(&claim.job_id)?; if !matches!(job.status, EpiphanyRuntimeJobStatus::Failed | EpiphanyRuntimeJobStatus::Cancelled) { return None; } - let request_id = [launch.proposal_modeling_request_id.as_deref(), launch.imagination_consideration_request_id.as_deref(), launch.admitted_model_direction_consideration_request_id.as_deref()] - .into_iter().flatten().next()?; + let request_id = typed_request_ids.get(&claim.job_id)?.as_deref()?; (!live_resident_request_ids.contains(request_id)).then_some(claim) }).collect::>(); candidates.sort_by(|a, b| b.terminal_at.cmp(&a.terminal_at).then(b.job_id.cmp(&a.job_id))); @@ -11606,15 +11555,19 @@ pub fn retain_fulfilled_runtime_worker_attempts( require_identity(&cache)?; let launches = cache.get_all::()?.into_iter() .map(|launch| (launch.job_id.clone(), launch)).collect::>(); + let typed_request_ids = launches.values().map(|launch| Ok((launch.job_id.clone(), launch.typed_request_ref()?.map(|request| request.request_id().to_string())))).collect::>>()?; let jobs = cache.get_all::()?.into_iter() .map(|job| (job.job_id.clone(), job)).collect::>(); let role_results = cache.get_all::()?.into_iter() .map(|result| (result.job_id.clone(), result)).collect::>(); let admitted_result_ids = cache.get_all::()?.into_iter() .filter_map(|receipt| receipt.result_id).collect::>(); - let mut candidates = cache.get_all::()?.into_iter() - .filter(|claim| claim.status == "terminal-result") - .filter_map(|claim| { + let claims = cache.get_all::()?.into_iter() + .map(|claim| Ok((crate::WorkerProcessStatus::parse(&claim.status)?, claim))) + .collect::>>()?; + let mut candidates = claims.into_iter() + .filter(|(status, _)| status.is_fulfilled_terminal()) + .filter_map(|(_, claim)| { let launch = launches.get(&claim.job_id)?; let job = jobs.get(&claim.job_id)?; let result = role_results.get(&claim.job_id)?; @@ -11624,8 +11577,7 @@ pub fn retain_fulfilled_runtime_worker_attempts( { return None; } - let request_id = [launch.proposal_modeling_request_id.as_deref(), launch.imagination_consideration_request_id.as_deref(), launch.admitted_model_direction_consideration_request_id.as_deref()] - .into_iter().flatten().next()?; + let request_id = typed_request_ids.get(&claim.job_id)?.as_deref()?; (!live_resident_request_ids.contains(request_id)).then_some(claim) }).collect::>(); candidates.sort_by(|a, b| b.terminal_at.cmp(&a.terminal_at).then(b.job_id.cmp(&a.job_id))); @@ -12341,8 +12293,8 @@ pub fn complete_runtime_job( ]; let claim_id = worker_process_claim_id(&result.job_id); if let Some(claim) = cache.get::(&claim_id)? { - match claim.status.as_str() { - "active" => { + match crate::WorkerProcessStatus::parse(&claim.status)? { + crate::WorkerProcessStatus::Active => { let claim_envelope = snapshot .iter() .find(|entry| { @@ -12355,25 +12307,25 @@ pub fn complete_runtime_job( if let Some(reorient) = cache.get::(&result.job_id)? { - terminal.status = "terminal-result".into(); + terminal.status = crate::WorkerProcessStatus::TerminalResult.as_str().into(); terminal.terminal_authority_id = Some(reorient.result_id); } else { - terminal.status = "terminal-failure".into(); + terminal.status = crate::WorkerProcessStatus::TerminalFailure.as_str().into(); terminal.terminal_authority_id = Some(result.result_id.clone()); } terminal.terminal_at = Some(result.completed_at.clone()); expected.push(claim_envelope); writes.push(cache.prepare_entry(&claim_id, &terminal)?.0); } - "terminal-result" => {} - "claimed" => { + crate::WorkerProcessStatus::TerminalResult => {} + crate::WorkerProcessStatus::Claimed => { return Err(anyhow!( "runtime worker job cannot complete before activation" )); } status => { return Err(anyhow!( - "runtime worker job completion found terminal process status {status:?}" + "runtime worker job completion found process status {:?}", status.as_str() )); } } diff --git a/epiphany-core/src/runtime_worker_attempt.rs b/epiphany-core/src/runtime_worker_attempt.rs new file mode 100644 index 00000000..055231c0 --- /dev/null +++ b/epiphany-core/src/runtime_worker_attempt.rs @@ -0,0 +1,164 @@ +use anyhow::{Result, anyhow}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum WorkerProcessStatus { + Claimed, + Active, + TerminalResult, + TerminalDeath, + TerminalUnactivated, + TerminalFailure, +} + +impl WorkerProcessStatus { + pub fn parse(value: &str) -> Result { + match value { + "claimed" => Ok(Self::Claimed), + "active" => Ok(Self::Active), + "terminal-result" => Ok(Self::TerminalResult), + "terminal-death" => Ok(Self::TerminalDeath), + "terminal-unactivated" => Ok(Self::TerminalUnactivated), + "terminal-failure" => Ok(Self::TerminalFailure), + other => Err(anyhow!("unknown runtime worker process status {other:?}")), + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::Claimed => "claimed", + Self::Active => "active", + Self::TerminalResult => "terminal-result", + Self::TerminalDeath => "terminal-death", + Self::TerminalUnactivated => "terminal-unactivated", + Self::TerminalFailure => "terminal-failure", + } + } + + pub fn is_live(self) -> bool { + matches!(self, Self::Claimed | Self::Active) + } + pub fn is_fulfilled_terminal(self) -> bool { + self == Self::TerminalResult + } + pub fn is_failed_terminal(self) -> bool { + matches!( + self, + Self::TerminalDeath | Self::TerminalUnactivated | Self::TerminalFailure + ) + } + pub fn is_terminal(self) -> bool { + self.is_fulfilled_terminal() || self.is_failed_terminal() + } + pub fn allows_retry(self) -> bool { + self.is_failed_terminal() + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum RuntimeTypedRequestRef<'a> { + ProposalModeling(&'a str), + ImaginationConsideration(&'a str), + AdmittedModelDirection(&'a str), +} + +impl<'a> RuntimeTypedRequestRef<'a> { + pub fn request_id(self) -> &'a str { + match self { + Self::ProposalModeling(id) + | Self::ImaginationConsideration(id) + | Self::AdmittedModelDirection(id) => id, + } + } + + pub fn kind(self) -> &'static str { + match self { + Self::ProposalModeling(_) => "proposal-modeling", + Self::ImaginationConsideration(_) => "imagination-consideration", + Self::AdmittedModelDirection(_) => "admitted-model-direction", + } + } + + pub fn matches_launch(self, launch: &crate::EpiphanyRuntimeWorkerLaunchRequest) -> bool { + match self { + Self::ProposalModeling(id) => { + launch.proposal_modeling_request_id.as_deref() == Some(id) + } + Self::ImaginationConsideration(id) => { + launch.imagination_consideration_request_id.as_deref() == Some(id) + } + Self::AdmittedModelDirection(id) => { + launch + .admitted_model_direction_consideration_request_id + .as_deref() + == Some(id) + } + } + } + + pub fn matches_result(self, result: &crate::EpiphanyRuntimeRoleWorkerResult) -> bool { + match self { + Self::ProposalModeling(id) => { + result.proposal_modeling_request_id.as_deref() == Some(id) + } + Self::ImaginationConsideration(id) => { + result.imagination_consideration_request_id.as_deref() == Some(id) + } + Self::AdmittedModelDirection(id) => { + result + .admitted_model_direction_consideration_request_id + .as_deref() + == Some(id) + } + } + } +} + +impl crate::EpiphanyRuntimeWorkerLaunchRequest { + pub fn typed_request_ref(&self) -> Result>> { + let requests = [ + self.proposal_modeling_request_id + .as_deref() + .map(RuntimeTypedRequestRef::ProposalModeling), + self.imagination_consideration_request_id + .as_deref() + .map(RuntimeTypedRequestRef::ImaginationConsideration), + self.admitted_model_direction_consideration_request_id + .as_deref() + .map(RuntimeTypedRequestRef::AdmittedModelDirection), + ] + .into_iter() + .flatten() + .collect::>(); + match requests.as_slice() { + [] => Ok(None), + [request] => Ok(Some(*request)), + _ => Err(anyhow!( + "worker launch carries multiple typed request authorities" + )), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn status_classes_are_exhaustive_and_disjoint() -> Result<()> { + for value in [ + "claimed", + "active", + "terminal-result", + "terminal-death", + "terminal-unactivated", + "terminal-failure", + ] { + let status = WorkerProcessStatus::parse(value)?; + assert_eq!(status.as_str(), value); + assert_ne!(status.is_live(), status.is_terminal()); + assert_eq!(status.allows_retry(), status.is_failed_terminal()); + } + assert!(WorkerProcessStatus::parse("completed").is_err()); + Ok(()) + } +} diff --git a/notes/epiphany-current-algorithmic-map.md b/notes/epiphany-current-algorithmic-map.md index e73d45e6..6c069e68 100644 --- a/notes/epiphany-current-algorithmic-map.md +++ b/notes/epiphany-current-algorithmic-map.md @@ -3522,7 +3522,9 @@ model direction binds the request/model projection directly. Failed terminal status is not permission to launder malformed launch state into a tombstone. Model-direction result admission now checks the same projected model hash. -- Owner: runtime spine owns archival of one immutable outer worker attempt. +- Owner: `runtime_worker_attempt` owns one immutable outer worker attempt's + typed request association and process-status semantics. Runtime spine retains + the CultCache transaction primitives for claim, result, and archival commits. Resident Self supplies cross-store liveness but cannot delete runtime rows. - Inputs: exact typed request identity; worker launch; lane-specific launch binding; process claim; terminal role result and typed companion; runtime job @@ -3539,13 +3541,18 @@ Model-direction result admission now checks the same projected model hash. and artifact cleanup cannot retire outer worker authority. No generic sweep may delete a claim or result independently. - Shared paths: proposal Modeling, Imagination consideration, and admitted model - direction use the same archive commit primitive and common launch/claim/job + direction use one exhaustive typed-request projection, the same status + predicates, the same archive commit primitive, and common launch/claim/job invariants, with lane-specific binding/companion validators. - Cut line: archive only exact launch, terminal claim, terminal result, lane-specific attempt binding/companion, and terminal job/event rows. Preserve request, proposal, model, admission, and unrelated runtime families until a separately mapped lifecycle owns them. The tombstone must be written in the same full-snapshot CAS as deletion. +- Demotion: serialized process-status strings are wire representation, not local + policy owners. Resident recovery, coordinator retry, runtime fulfillment, and + retention derive their decisions from `WorkerProcessStatus`. Per-consumer + request-family arrays no longer decide which typed work exists. - Verification layer: terminal-result/failure/death attempts; prepared/active/ unconsumed-ack refusal; downstream-reference refusal; stale snapshot race; unknown-row byte identity; retry after archived failure; fulfillment after diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index de55e7b6..07c9da31 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -5,10 +5,9 @@ Updated: 2026-08-12 ## Current state The five-day live shakedown campaign is paused for architectural consolidation. -The source branch is `codex/epiphany-shakedown-live`; the last pushed source -before surgery is `80541843`. The working tree contains the consolidation cut -described below and must be committed only after final source checks and map -updates. +The source branch is `codex/epiphany-shakedown-live`. Causal-work and immutable +public-source ownership landed at `868a5be0`. The working tree contains the +second consolidation cut described below. The accepted live body remains c011 at exact `465af24d` on Starfire, cognitively braked at resident revision 384 with no active lease, admitted work, or defunct @@ -35,10 +34,11 @@ repair. public-source receipt. Provider validation is no longer trusted as a substitute for admission validation. -The full `epiphany-core` library passes 683 tests with one intentional ignored -cross-process helper. `epiphany-tool-mcp-runtime` passes 14 tests with one -intentional ignored live-network proof. Focused stale-thread refusal preserves -the exact store bytes. +The first cut's full `epiphany-core` library passed 683 tests with one intentional +ignored cross-process helper. After worker-attempt extraction, core passes 684 +active tests with one ignored helper, `epiphany-tool-mcp-runtime` passes 14 +active tests with one ignored live-network proof, and OpenAI runtime passes all +39 tests. Focused stale-thread refusal preserves the exact store bytes. ## Authority map @@ -76,25 +76,23 @@ Do not collapse these into one enum or service: - Runtime worker attempt owns launch, exact process claim, result, and archive. - Mind decides adoption; fulfillment is evidence, not admission. -The next physical extraction is a narrow `runtime_worker_attempt` module. It -should centralize typed process-status predicates and typed-request association -without merging the authorities above or introducing a registry. +The narrow `runtime_worker_attempt` module now owns the serialized process-status +vocabulary, live/fulfilled/failed/retry classification, and the complete set of +typed request families carried by one launch. Runtime, resident, and coordinator +consumers call those predicates and accessors instead of maintaining local +string sets and parallel family arrays. Persisted documents and transactions are +unchanged; the lifecycle authorities above remain separate. ## Acceptance gates before shakedown resumes -1. Extract the worker-attempt aggregate from `runtime_spine` with no schema or - state migration. -2. Replay full core, resident, coordinator, tool-runtime, and OpenAI-runtime - suites after extraction. -3. Prove immutable public Eyes success and no-grant denial on a copied packaged +1. Prove immutable public Eyes success and no-grant denial on a copied packaged body. This is the paused shakedown's first remaining live gate, not part of the consolidation commit. -4. Run one fresh-repository source -> Hands -> Soul -> Modeling -> Mind -> Self +2. Run one fresh-repository source -> Hands -> Soul -> Modeling -> Mind -> Self capstone before claiming organism-level readiness. ## Immediate next action -Review the current diff for any surviving family-local causal hash or public -GitHub parser. Commit and push this first consolidation cut. Then map and extract -`runtime_worker_attempt`; do not add new capability surfaces and do not replace -c011. +Commit and push the worker-attempt extraction, then resume with the copied +packaged public Eyes success/denial pair. Do not add new capability surfaces and +do not replace c011 before that gate closes. diff --git a/notes/perfect-machine-audit-roadmap.md b/notes/perfect-machine-audit-roadmap.md index dbf36fcd..832f7efc 100644 --- a/notes/perfect-machine-audit-roadmap.md +++ b/notes/perfect-machine-audit-roadmap.md @@ -18,7 +18,7 @@ consolidation gates below close. |---|---|---|---| | Durable request identity excludes mutable transport provenance | `causal_work_identity` | Proposal Modeling, Research, Planning, PlanMind, and Admitted Model Direction use one pure derivation owner; stale thread restoration refuses byte-identically | replay packaged coordinated circuits after worker-attempt extraction | | Immutable GitHub identity has one grammar | `ImmutableGithubSource` | Modeling selection, Eyes execution, and Mind authentication share one parser/canonicalizer; malformed identities fail at Mind | copied packaged public success and no-grant denial | -| One worker attempt has one process/result/archive authority | planned `runtime_worker_attempt` | current lifecycle and archive proofs remain accepted | extract typed status predicates and request association from `runtime_spine` without schema migration | +| One worker attempt has one process/result/archive authority | `runtime_worker_attempt` | typed status classes and request association are shared by runtime, resident, and coordinator; core 684/684 (+1 ignored), tool runtime 14/14 (+1 ignored), OpenAI runtime 39/39 | replay packaged settlement/archive circuits | | Canonical state admission has one writer | coordinator state transaction / Mind | CAS companions and negative source guards accepted | continue legacy-writer audit only when source evidence finds a live seam | ## Worker-attempt extraction boundary @@ -49,14 +49,13 @@ consolidation gates below close. | Proposal/Planning causal identity | deterministic shared owner | stale carrier cannot restore thread | cross-incarnation replay | source complete | | Hands -> Soul -> Modeling -> Mind -> Self | bounded adjacent circuits | bypass/substitution refusals | fresh-repository full loop | capstone open | | Persona -> Bifrost -> public consequence | signed failure/unknown/no-repost | private-state and permit refusal | successful receipted consequence | blocked by external credential | -| Worker process -> result/death -> archive | genuine Linux workers and zero-zombie reaping | resurrection and competing retry refused | both receipt orderings and restart | accepted; physical extraction pending | +| Worker process -> result/death -> archive | genuine Linux workers and zero-zombie reaping | resurrection and competing retry refused | both receipt orderings and restart | source extraction complete; packaged replay pending | | Eve/CultUI operator interface | typed state surfaces | renderer cannot own truth | load/transition/settled/re-entry probes | open | ## Immediate order -1. Commit and push the causal/public-source identity consolidation. -2. Extract `runtime_worker_attempt` without changing persisted contracts. -3. Replay source tests and the accepted worker settlement/archive circuits. -4. Resume shakedown with the packaged public Eyes success/denial pair. -5. Run the fresh-repository full-organ capstone before any deployment-ready +1. Commit and push `runtime_worker_attempt`. +2. Replay the accepted worker settlement/archive circuits. +3. Resume shakedown with the packaged public Eyes success/denial pair. +4. Run the fresh-repository full-organ capstone before any deployment-ready claim. diff --git a/state/ledgers.msgpack b/state/ledgers.msgpack index e706ee9b68927c1cd9ce54f11fb00ee69172e9d6..6dc832259e2d42dff63404473b2845ac98c8c8e8 100644 GIT binary patch delta 487 zcmZ{fzfJ-{5XQapFP)Lb!eSCbq7ldehA31PnpoM`XhYUzfDL!M$L<_RMPXsBf$$2J zCKkrR)(7wnw6OOL96%Zqi(fXI`Mz(ydH>7|zcP<$Z@qXP1eKsv3Cfpm3#rcc)3(>~ zyoZ~Se>?I$X7VmFOUyE}!mKiD&m%wesyBu&#h_Tu1qZppZ`SzZ!rvNr_s7>{g(R$l zO*E+pYdvGzXlsH9qnJQApcaB_3CkkW(`dmoV5WlD8i~$@6DnZKVYI?pj%PObps3|+XEBc!Q>p+{{Y%3Uz&^1zAC84l==qYNM zgdjrORJu8-r!0i3fwC_{1d;IEDpKgcSdcBDr@9!eQ*LsqG!T