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 e706ee9b..6dc83225 100644 Binary files a/state/ledgers.msgpack and b/state/ledgers.msgpack differ diff --git a/state/map.yaml b/state/map.yaml index bb48c9a6..ff7bbbd8 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -70,7 +70,7 @@ current_status: yggdrasil_epiphany_service_contract: repaired_in_gamecult_ops_812b427_not_installed resident_physiology: c011_exact_465af24d_alive_braked_revision384_after_modeling_and_imagination_terminal_settlement_and_zero_zombie_reaping unresolved_authority: Coordinator receipt, Heartbeat pulse-artifact, genuine-idle and active resource plateaus, exact-grant terminality, restart-persistent cooldown refusal, schedule/action issuance cardinality, runtime-owned typed fulfillment, all three packaged fulfilled-before-receipt lanes, genuine packaged Imagination option generation, narrow model-generation retention, published model-turn JSON ingress, semantic logical/physical retention, Persona conversation retention, stable Starfire firewall ownership, completed model-adapter session archival, normal coordinator lifecycle opening/error closure/legacy repair/archival, co-resident resident lifecycle retention, authenticated outer-worker attempt retention, deployed supervisor/projector journald ownership, Docker named-volume atomic snapshot visibility, deterministic incremental release construction, and canonical-brake refusal of direct daemon lifecycle actuation are accepted. Starfire's Docker Desktop F: bind is rejected for mutable authority after an exact packaged lockless reader observed an empty published snapshot during writer interruption. Historical migrated semantic attempts without recorded physical namespace remain durable and non-retirable. One completed Discord consequence remains blocked by the absent Yggdrasil credential. Exact aa841873 also makes legacy coordinator repair one full-snapshot transaction so a later hostile family cannot leave an earlier partial repair. - next_action: Keep exact c011 alive and cognitively braked. Commit and push the generalized causal-work and immutable-public-source identity owners after final source review. Then extract the narrow runtime_worker_attempt aggregate: typed process-status predicates and request-family association only, without merging Heartbeat, resident, coordinator, runtime-job, OS-process, or Mind authority. Replay accepted source tests before resuming the packaged public Eyes gates. Do not build on Yggdrasil. + next_action: Keep exact c011 alive and cognitively braked. Commit and push the extracted runtime_worker_attempt owner: one typed process-status vocabulary and one exhaustive request-family association, without merging Heartbeat, resident, coordinator, runtime-job, OS-process, or Mind authority. Resume with the copied packaged public Eyes success and no-grant denial pair. Do not build on Yggdrasil. active_subgoals: - Prove Persona, Self, Eyes, Modeling, Imagination, Hands, Soul, Mind, Continuity, and nervous-system behavior individually.