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

Recover exact receipt-free coordinator deaths

Browse source
This commit is contained in:
Meta Vens 2026-08-10 09:51:10 +01:00
commit e97ee1fa1b
9 changed files with 1250 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

19
epiphany-core/src/bin/epiphany-mvp-coordinator.rs
View file

@ -301,7 +301,21 @@ fn run_coordinator(args: &Args) -> Result<Value> {
"ephemeral": args.ephemeral,
"workspace": cwd,
}));
let runtime_session_id = format!("coordinator-{thread_id}");
let resident_launch_digest = args
.resident_binding
.get("launch-digest")
.map(String::as_str);
let runtime_session_id = epiphany_core::coordinator_run_session_id(
&thread_id,
resident_launch_digest,
)?;
let runtime_session_objective = args
.resident_binding
.get("objective-digest")
.map(|digest| format!("Resident coordinator launch {digest}."))
.unwrap_or_else(|| {
"Coordinate the Epiphany MVP lanes with native runtime job receipts.".to_string()
});
let runtime_identity = initialize_runtime_spine(
&runtime_store,
coordinator_runtime_identity_options(
@ -313,7 +327,8 @@ fn run_coordinator(args: &Args) -> Result<Value> {
&runtime_store,
&runtime_session_id,
&thread_id,
"Coordinate the Epiphany MVP lanes with native runtime job receipts.",
resident_launch_digest,
&runtime_session_objective,
&now(),
)?;
startup_events.push(json!({

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

@ -7,16 +7,21 @@ use epiphany_core::{
ResidentSelfState, acknowledge_resident_self_launch, acquire_resident_process_singleton,
authenticate_resident_self_policy, bind_runtime_repository_domain,
bridge_admitted_persona_feedback_to_heartbeat, cancel_resident_self_turn,
capture_process_instance, coordinator_run_receipts, derive_resident_cognition_readiness,
capture_process_instance, complete_resident_self_turn_after_death,
coordinator_run_receipts, derive_resident_cognition_readiness,
enqueue_resident_self_pressure, import_bifrost_persona_feedback_deliveries,
ingest_resident_self_domain_pressure, load_epiphany_cultmesh_swarm_brake,
load_resident_self_state, observe_process_instance, pending_resident_self_acks,
prepare_resident_self_launch, publish_resident_provider_readiness,
recover_receipt_free_dead_coordinator_session,
resident_prepared_launch_thread_id, resident_self_child_claim,
resident_self_local_provider_status, retain_completed_runtime_sessions,
resident_self_grant_has_typed_request, resident_self_local_provider_status,
resident_self_typed_attempt_exists,
retain_completed_runtime_sessions,
retain_coordinator_run_receipts, retain_resident_self_lifecycles,
settle_resident_self_exited_coordinator, terminate_process_instance,
validate_persona_feedback_store_separation, validate_resident_self_store_separation,
validate_persona_feedback_store_separation, validate_resident_self_coordinator_receipt,
validate_resident_self_store_separation, verify_resident_self_grant_fulfillment,
};
use serde_json::json;
use std::collections::{BTreeMap, BTreeSet};
@ -243,6 +248,100 @@ fn outcome_after_cancel(shutdown_requested: bool) -> ResidentSelfOutcome {
}
}
fn exact_resident_coordinator_receipt(
runtime_store: &std::path::Path,
lease: &epiphany_core::ResidentSelfTurnLease,
) -> Result<Option<epiphany_core::EpiphanyCoordinatorRunReceipt>> {
let session_id = epiphany_core::coordinator_run_session_id(
&lease.turn_id,
Some(&lease.launch_digest),
)?;
let mut receipts = coordinator_run_receipts(runtime_store)?
.into_iter()
.filter(|receipt| receipt.session_id == session_id)
.collect::<Vec<_>>();
if receipts.len() > 1 {
return Err(anyhow!(
"resident coordinator session has multiple terminal receipts"
));
}
let Some(receipt) = receipts.pop() else {
return Ok(None);
};
validate_resident_self_coordinator_receipt(lease, &receipt)?;
Ok(Some(receipt))
}
fn settle_receipt_free_dead_coordinator(
args: &Args,
lease: &epiphany_core::ResidentSelfTurnLease,
observation: ChildObservation,
shutdown_requested: bool,
brake_engaged: bool,
timed_out: bool,
now: u64,
) -> Result<ResidentSelfOutcome> {
let typed = resident_self_grant_has_typed_request(&args.state_store, &lease.grant_id)?;
if typed {
match verify_resident_self_grant_fulfillment(
&args.state_store,
&args.policy.runtime_store,
&lease.grant_id,
)? {
epiphany_core::ResidentSelfGrantFulfillment::Fulfilled => {
let recovery = recover_receipt_free_dead_coordinator_session(
&args.state_store,
&args.policy.runtime_store,
lease,
observation,
now,
)?
.ok_or_else(|| {
anyhow!(
"typed fulfillment cannot predate its coordinator runtime incarnation"
)
})?;
complete_resident_self_turn_after_death(
&args.state_store,
lease,
&recovery,
now,
args.policy.cooldown_seconds,
)?;
return Ok(ResidentSelfOutcome::Completed);
}
epiphany_core::ResidentSelfGrantFulfillment::Pending => {
if resident_self_typed_attempt_exists(
&args.state_store,
&args.policy.runtime_store,
&lease.grant_id,
)? {
return Ok(ResidentSelfOutcome::AwaitingFulfillment);
}
}
}
}
let recovery = recover_receipt_free_dead_coordinator_session(
&args.state_store,
&args.policy.runtime_store,
lease,
observation,
now,
)?;
cancel_resident_self_turn(
&args.state_store,
lease,
cancelled_turn_status(shutdown_requested, brake_engaged, timed_out),
if recovery.is_some() {
"exact coordinator process died after atomic opening and before a terminal receipt"
} else {
"exact coordinator process died before atomic runtime opening"
},
now,
)?;
Ok(outcome_after_cancel(shutdown_requested))
}
fn cycle(
args: &Args,
state: &mut ResidentSelfState,
@ -344,55 +443,52 @@ fn cycle(
}
return match ports.observe_child(&lease)? {
ChildObservation::Running => Ok(ResidentSelfOutcome::Running),
ChildObservation::Exited(0) => {
let receipt = coordinator_run_receipts(&args.policy.runtime_store)?
.into_iter()
.find(|receipt| {
receipt.thread_id == lease.turn_id
&& receipt.resident_launch_digest.as_deref()
== Some(&lease.launch_digest)
})
.ok_or_else(|| {
anyhow!(
"coordinator exited zero without its exact resident terminal receipt"
)
})?;
let outcome = settle_resident_self_exited_coordinator(
&args.state_store,
&args.policy.runtime_store,
&lease,
&receipt,
shutdown_requested,
brake_engaged,
timed_out,
now,
args.policy.cooldown_seconds,
)?;
observation => {
let receipt =
exact_resident_coordinator_receipt(&args.policy.runtime_store, &lease)?;
if let Some(receipt) = receipt.as_ref().filter(|receipt| {
matches!(
receipt.status.as_str(),
"planned" | "needsReview" | "completed"
)
}) {
let outcome = settle_resident_self_exited_coordinator(
&args.state_store,
&args.policy.runtime_store,
&lease,
receipt,
shutdown_requested,
brake_engaged,
timed_out,
now,
args.policy.cooldown_seconds,
)?;
*state = load_resident_self_state(&args.state_store)?;
return Ok(outcome);
}
let outcome = if receipt.is_some() {
cancel_resident_self_turn(
&args.state_store,
&lease,
cancelled_turn_status(shutdown_requested, brake_engaged, timed_out),
"exact coordinator terminal receipt reports failure",
now,
)?;
outcome_after_cancel(shutdown_requested)
} else {
settle_receipt_free_dead_coordinator(
args,
&lease,
observation,
shutdown_requested,
brake_engaged,
timed_out,
now,
)?
};
*state = load_resident_self_state(&args.state_store)?;
Ok(outcome)
}
ChildObservation::Exited(code) => {
cancel_resident_self_turn(
&args.state_store,
&lease,
cancelled_turn_status(shutdown_requested, brake_engaged, timed_out),
&format!("coordinator terminal observation exit={code}"),
now,
)?;
*state = load_resident_self_state(&args.state_store)?;
Ok(outcome_after_cancel(shutdown_requested))
}
ChildObservation::Missing => {
cancel_resident_self_turn(
&args.state_store,
&lease,
cancelled_turn_status(shutdown_requested, brake_engaged, timed_out),
"exact coordinator process is missing",
now,
)?;
*state = load_resident_self_state(&args.state_store)?;
Ok(outcome_after_cancel(shutdown_requested))
}
};
}
if shutdown_requested || ports.brake_engaged()? {

4
epiphany-core/src/lib.rs
View file

@ -981,8 +981,10 @@ pub use runtime_spine::ARCHIVED_RUNTIME_SESSION_SCHEMA_VERSION;
pub use runtime_spine::ARCHIVED_RUNTIME_SESSION_TYPE;
pub use runtime_spine::COORDINATOR_RUN_RECEIPT_SCHEMA_VERSION;
pub use runtime_spine::COORDINATOR_RUN_RECEIPT_TYPE;
pub use runtime_spine::COORDINATOR_DEATH_RECOVERY_SCHEMA_VERSION;
pub use runtime_spine::EpiphanyArchivedRuntimeSession;
pub use runtime_spine::EpiphanyCoordinatorRunReceipt;
pub use runtime_spine::EpiphanyCoordinatorDeathRecovery;
pub use runtime_spine::EpiphanyCoordinatorRunReceiptRetentionHead;
pub use runtime_spine::EpiphanyJobInterruptRequest;
pub use runtime_spine::EpiphanyJobInterruptResult;
@ -1056,6 +1058,7 @@ pub use runtime_spine::commit_repo_model_claim_challenge;
pub use runtime_spine::commit_repo_model_claim_repair_request;
pub use runtime_spine::complete_runtime_job;
pub use runtime_spine::coordinator_run_receipts;
pub use runtime_spine::coordinator_run_session_id;
pub use runtime_spine::create_runtime_job;
pub use runtime_spine::create_runtime_session;
pub use runtime_spine::ensure_runtime_repo_model;
@ -1142,6 +1145,7 @@ pub use runtime_spine::runtime_substrate_gate_repo_access_grant_receipt;
pub use runtime_spine::runtime_swarm_binding;
pub use runtime_spine::runtime_tool_invocation_statuses;
pub use runtime_spine::runtime_typed_request_fulfillment;
pub use runtime_spine::runtime_typed_request_attempt_exists;
pub use runtime_spine::runtime_worker_launch_body_basis;
pub use runtime_spine::runtime_worker_launch_request;
pub use runtime_spine::select_and_commit_repo_frontier_planning_request;

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

@ -553,9 +553,12 @@ pub fn coordinator_argv(
policy: &ResidentSelfPolicy,
runtime_id: &str,
turn_id: &str,
artifact_incarnation: Option<&str>,
wake: &ResidentSelfWake,
) -> Vec<String> {
let artifact_dir = policy.artifact_root.join(turn_id);
let artifact_dir = policy
.artifact_root
.join(artifact_incarnation.unwrap_or(turn_id));
let mut argv = vec![
"--model-runtime-bin".into(),
policy.model_runtime_bin.display().to_string(),
@ -870,7 +873,13 @@ fn reconcile_resident_self(
let launch = CoordinatorLaunch {
turn_id: turn_id.clone(),
wake: wake.clone(),
argv: coordinator_argv(policy, &policy.release_runtime_id, &turn_id, &wake),
argv: coordinator_argv(
policy,
&policy.release_runtime_id,
&turn_id,
None,
&wake,
),
};
let process = ports.launch_coordinator(&launch)?;
let process_id = process.process_id;
@ -1207,7 +1216,23 @@ pub fn verify_resident_self_grant_fulfillment(
let grant = state_cache(resident_store)?
.get::<ResidentSelfHeartbeatGrant>(grant_id)?
.ok_or_else(|| anyhow!("resident Self fulfillment check lost its grant"))?;
let request = match grant.pressure_kind.as_str() {
let request = resident_self_typed_request_ref(&grant)?;
let Some(request) = request else {
return Ok(ResidentSelfGrantFulfillment::Fulfilled);
};
Ok(
if crate::runtime_typed_request_fulfillment(runtime_store, request)?.is_some() {
ResidentSelfGrantFulfillment::Fulfilled
} else {
ResidentSelfGrantFulfillment::Pending
},
)
}
fn resident_self_typed_request_ref<'a>(
grant: &'a ResidentSelfHeartbeatGrant,
) -> Result<Option<crate::RuntimeTypedRequestRef<'a>>> {
Ok(match grant.pressure_kind.as_str() {
"repo-frontier-proposal-modeling" => {
let request_id = grant
.provenance_ref
@ -1239,17 +1264,31 @@ pub fn verify_resident_self_grant_fulfillment(
))
}
_ => None,
})
}
pub fn resident_self_typed_attempt_exists(
resident_store: &Path,
runtime_store: &Path,
grant_id: &str,
) -> Result<bool> {
let grant = state_cache(resident_store)?
.get::<ResidentSelfHeartbeatGrant>(grant_id)?
.ok_or_else(|| anyhow!("resident Self attempt check lost its grant"))?;
let Some(request) = resident_self_typed_request_ref(&grant)? else {
return Ok(false);
};
let Some(request) = request else {
return Ok(ResidentSelfGrantFulfillment::Fulfilled);
};
Ok(
if crate::runtime_typed_request_fulfillment(runtime_store, request)?.is_some() {
ResidentSelfGrantFulfillment::Fulfilled
} else {
ResidentSelfGrantFulfillment::Pending
},
)
crate::runtime_typed_request_attempt_exists(runtime_store, request)
}
pub fn resident_self_grant_has_typed_request(
resident_store: &Path,
grant_id: &str,
) -> Result<bool> {
let grant = state_cache(resident_store)?
.get::<ResidentSelfHeartbeatGrant>(grant_id)?
.ok_or_else(|| anyhow!("resident Self typed-request check lost its grant"))?;
Ok(resident_self_typed_request_ref(&grant)?.is_some())
}
pub fn settle_resident_self_exited_coordinator(
@ -1315,6 +1354,118 @@ pub fn settle_resident_self_exited_coordinator(
}
}
pub fn recover_receipt_free_dead_coordinator_session(
resident_store: &Path,
runtime_store: &Path,
lease: &ResidentSelfTurnLease,
observation: ChildObservation,
now_millis: u64,
) -> Result<Option<crate::EpiphanyCoordinatorDeathRecovery>> {
let (observation, exit_code) = match observation {
ChildObservation::Exited(code) => ("exited", Some(code)),
ChildObservation::Missing => ("missing", None),
ChildObservation::Running => {
return Err(anyhow!(
"Continuity refuses coordinator death recovery while the exact child is running"
));
}
};
let cache = state_cache(resident_store)?;
let state = cache
.get::<ResidentSelfState>(RESIDENT_SELF_STATE_KEY)?
.ok_or_else(|| anyhow!("coordinator death recovery lost resident state"))?;
if state.active_turn.as_ref() != Some(lease) {
return Err(anyhow!(
"coordinator death recovery lease is not the active resident authority"
));
}
let grant = cache
.get::<ResidentSelfHeartbeatGrant>(&lease.grant_id)?
.ok_or_else(|| anyhow!("coordinator death recovery lost its grant"))?;
if grant.consumed_at_millis.is_none()
|| grant.terminal_at_millis.is_some()
|| grant.terminal_status.is_some()
{
return Err(anyhow!(
"coordinator death recovery grant is not active and unterminated"
));
}
let preparation_id = format!("resident-self-prepared-{}", lease.grant_id);
let claim = resident_self_child_claim(resident_store, &preparation_id)?
.ok_or_else(|| anyhow!("coordinator death recovery lost its immutable child claim"))?;
if claim.grant_id != lease.grant_id
|| claim.launch_digest != lease.launch_digest
|| claim.process_id != lease.process_id
|| claim.process_creation_token != lease.process_creation_token
|| claim.executable_path != lease.process_executable_path
|| claim.executable_digest != lease.coordinator_executable_digest
{
return Err(anyhow!(
"coordinator death recovery child claim disagrees with the exact lease"
));
}
let expected_process = crate::ProcessInstanceIdentity {
process_id: lease.process_id,
creation_token: lease.process_creation_token,
created_at_rfc3339: None,
executable_path: lease.process_executable_path.clone(),
};
match crate::observe_process_instance(&expected_process) {
crate::ProcessInstanceObservation::ExactAlive => {
return Err(anyhow!(
"Continuity refuses coordinator death recovery while the exact process incarnation is alive"
));
}
crate::ProcessInstanceObservation::Inaccessible
| crate::ProcessInstanceObservation::Indeterminate { .. } => {
return Err(anyhow!(
"Continuity cannot prove the exact coordinator process incarnation dead"
));
}
crate::ProcessInstanceObservation::ExactExited { .. }
| crate::ProcessInstanceObservation::Missing
| crate::ProcessInstanceObservation::Replaced { .. } => {}
}
let recovered_at = chrono::DateTime::<chrono::Utc>::from_timestamp_millis(now_millis as i64)
.ok_or_else(|| anyhow!("coordinator death recovery timestamp is out of range"))?
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
let session_id = crate::coordinator_run_session_id(
&lease.turn_id,
Some(&lease.launch_digest),
)?;
if crate::runtime_spine::coordinator_run_incarnation_is_absent(
runtime_store,
&lease.turn_id,
&lease.launch_digest,
)? {
return Ok(None);
}
let recovery = crate::EpiphanyCoordinatorDeathRecovery {
schema_version: crate::runtime_spine::COORDINATOR_DEATH_RECOVERY_SCHEMA_VERSION.into(),
recovery_id: format!("coordinator-death-recovery-{session_id}"),
session_id,
thread_id: lease.turn_id.clone(),
resident_grant_id: lease.grant_id.clone(),
resident_launch_digest: lease.launch_digest.clone(),
process_id: lease.process_id,
process_creation_token: lease.process_creation_token,
process_executable_path: lease.process_executable_path.display().to_string(),
resident_started_at_millis: lease.started_at_millis,
observation: observation.into(),
recovered_at,
private_state_exposed: false,
exit_code,
};
let runtime_session_objective =
format!("Resident coordinator launch {}.", lease.objective_digest);
crate::runtime_spine::recover_coordinator_run_after_exact_process_death(
runtime_store,
&recovery,
&runtime_session_objective,
)?;
Ok(Some(recovery))
}
pub fn pending_resident_self_grant(path: &Path) -> Result<Option<ResidentSelfHeartbeatGrant>> {
let cache = state_cache(path)?;
let legacy_terminal_grant_ids = cache
@ -1390,7 +1541,13 @@ pub fn prepare_resident_self_launch(
let wake = ResidentSelfWake::Explicit {
objective: grant.objective.clone(),
};
let mut argv = coordinator_argv(policy, &binding.runtime_id, &binding.thread_id, &wake);
let mut argv = coordinator_argv(
policy,
&binding.runtime_id,
&binding.thread_id,
Some(&grant.grant_id),
&wake,
);
if let Some((request_flag, request_id)) = binding.typed_request {
let objective = argv.pop();
let flag = argv.pop();
@ -1623,7 +1780,24 @@ fn complete_resident_self_turn(
now_millis: u64,
cooldown_seconds: u64,
) -> Result<ResidentSelfTerminalAck> {
validate_resident_self_coordinator_receipt(lease, coordinator)?;
complete_resident_self_turn_with_terminal(
path,
lease,
&coordinator.receipt_id,
&coordinator.status,
now_millis,
cooldown_seconds,
)
}
pub fn validate_resident_self_coordinator_receipt(
lease: &ResidentSelfTurnLease,
coordinator: &crate::EpiphanyCoordinatorRunReceipt,
) -> Result<()> {
if coordinator.thread_id != lease.turn_id
|| coordinator.session_id
!= crate::coordinator_run_session_id(&lease.turn_id, Some(&lease.launch_digest))?
|| !matches!(
coordinator.status.as_str(),
"planned" | "needsReview" | "completed"
@ -1643,6 +1817,48 @@ fn complete_resident_self_turn(
"coordinator terminal receipt does not exactly bind the resident launch contract"
));
}
Ok(())
}
pub fn complete_resident_self_turn_after_death(
path: &Path,
lease: &ResidentSelfTurnLease,
recovery: &crate::EpiphanyCoordinatorDeathRecovery,
now_millis: u64,
cooldown_seconds: u64,
) -> Result<ResidentSelfTerminalAck> {
if recovery.session_id
!= crate::coordinator_run_session_id(&lease.turn_id, Some(&lease.launch_digest))?
|| recovery.thread_id != lease.turn_id
|| recovery.resident_grant_id != lease.grant_id
|| recovery.resident_launch_digest != lease.launch_digest
|| recovery.process_id != lease.process_id
|| recovery.process_creation_token != lease.process_creation_token
|| recovery.process_executable_path != lease.process_executable_path.display().to_string()
|| recovery.resident_started_at_millis != lease.started_at_millis
{
return Err(anyhow!(
"coordinator death recovery does not exactly bind the resident lease"
));
}
complete_resident_self_turn_with_terminal(
path,
lease,
&recovery.recovery_id,
"recovered-fulfilled",
now_millis,
cooldown_seconds,
)
}
fn complete_resident_self_turn_with_terminal(
path: &Path,
lease: &ResidentSelfTurnLease,
terminal_id: &str,
terminal_status: &str,
now_millis: u64,
cooldown_seconds: u64,
) -> Result<ResidentSelfTerminalAck> {
let cache = state_cache(path)?;
let mut state = cache
.get::<ResidentSelfState>(RESIDENT_SELF_STATE_KEY)?
@ -1662,8 +1878,8 @@ fn complete_resident_self_turn(
heartbeat_schedule_id: grant.heartbeat_schedule_id.clone(),
heartbeat_action_id: grant.heartbeat_action_id.clone(),
launch_digest: lease.launch_digest.clone(),
coordinator_receipt_id: coordinator.receipt_id.clone(),
terminal_status: coordinator.status.clone(),
coordinator_receipt_id: terminal_id.to_string(),
terminal_status: terminal_status.to_string(),
completed_at_millis: now_millis,
consumed_by_heartbeat_at_millis: None,
private_state_exposed: false,
@ -1686,13 +1902,13 @@ fn complete_resident_self_turn(
.cloned()
.ok_or_else(|| anyhow!("resident Self grant lost envelope at terminal ack"))?;
state.active_turn = None;
state.last_coordinator_receipt_id = Some(coordinator.receipt_id.clone());
state.last_coordinator_receipt_id = Some(terminal_id.to_string());
state.next_eligible_at_millis =
now_millis.saturating_add(cooldown_seconds.saturating_mul(1000));
state.revision += 1;
grant.schema_version = RESIDENT_SELF_GRANT_SCHEMA_VERSION.into();
grant.terminal_at_millis = Some(now_millis);
grant.terminal_status = Some(coordinator.status.clone());
grant.terminal_status = Some(terminal_status.to_string());
let (state_entry, _) = cache.prepare_entry(RESIDENT_SELF_STATE_KEY, &state)?;
let (grant_entry, _) = cache.prepare_entry(&grant.grant_id, &grant)?;
let (ack_entry, _) = cache.prepare_entry(&ack.ack_id, &ack)?;
@ -2610,7 +2826,11 @@ mod tests {
crate::EpiphanyCoordinatorRunReceipt {
schema_version: crate::COORDINATOR_RUN_RECEIPT_SCHEMA_VERSION.into(),
receipt_id: format!("coordinator-terminal-{}", lease.turn_id),
session_id: format!("session-{}", lease.turn_id),
session_id: crate::coordinator_run_session_id(
&lease.turn_id,
Some(&lease.launch_digest),
)
.expect("resident lease has a valid session incarnation"),
thread_id: lease.turn_id.clone(),
mode: "plan".into(),
status: "planned".into(),
@ -2766,6 +2986,7 @@ mod tests {
let mut policy = policy();
policy.coordinator_bin = coordinator.clone();
policy.runtime_store = temp.path().join("runtime.cc");
policy.artifact_root = temp.path().join("artifacts");
seed_model_direction_request(
&policy.runtime_store,
"request-model-1",
@ -2869,7 +3090,10 @@ mod tests {
let receipt = crate::EpiphanyCoordinatorRunReceipt {
schema_version: crate::COORDINATOR_RUN_RECEIPT_SCHEMA_VERSION.into(),
receipt_id: "coordinator-terminal-1".into(),
session_id: "session-1".into(),
session_id: crate::coordinator_run_session_id(
&lease.turn_id,
Some(&lease.launch_digest),
)?,
thread_id: lease.turn_id.clone(),
mode: "plan".into(),
status: "planned".into(),
@ -2897,6 +3121,9 @@ mod tests {
complete_resident_self_turn(&store, &lease, &wrong, 8, policy.cooldown_seconds)
.is_err()
);
let mut wrong_session = receipt.clone();
wrong_session.session_id = format!("{}-substituted", receipt.session_id);
assert!(validate_resident_self_coordinator_receipt(&lease, &wrong_session).is_err());
assert_eq!(
settle_resident_self_exited_coordinator(
&store,
@ -3439,6 +3666,253 @@ mod tests {
Ok(())
}
#[test]
fn receipt_free_dead_coordinator_recovery_preserves_exact_lease_then_requeues() -> Result<()> {
let temp = tempfile::tempdir()?;
let resident_store = temp.path().join("resident-self.cc");
let coordinator = temp.path().join("epiphany-mvp-coordinator");
std::fs::write(&coordinator, b"witnessed executable")?;
let mut policy = policy();
policy.coordinator_bin = coordinator.clone();
policy.runtime_store = temp.path().join("runtime.cc");
policy.artifact_root = temp.path().join("artifacts");
seed_runtime_identity(&policy.runtime_store, "test-runtime")?;
let objective = "Recover a coordinator killed after atomic opening.";
enqueue_resident_self_pressure(
&resident_store,
&ResidentSelfPressure {
schema_version: RESIDENT_SELF_PRESSURE_SCHEMA_VERSION.into(),
pressure_id: "pressure-dead-coordinator".into(),
kind: "operator-objective".into(),
provenance_ref: "operator://dead-coordinator-test".into(),
objective: objective.into(),
created_at_millis: 1,
status: "pending".into(),
consumed_by_grant_id: None,
private_state_exposed: false,
},
)?;
let grant = heartbeat_issue_resident_self_grant(
&resident_store,
"heartbeat-dead",
"action-dead",
2,
)?
.expect("dead coordinator grant");
let prepared = prepare_resident_self_launch(&resident_store, &policy, 3)?
.expect("dead coordinator preparation");
let first_artifact_index = prepared
.argv
.iter()
.position(|value| value == "--artifact-dir")
.expect("first artifact flag");
let first_artifact_dir = PathBuf::from(&prepared.argv[first_artifact_index + 1]);
std::fs::create_dir_all(&first_artifact_dir)?;
let first_artifact = first_artifact_dir.join("sealed-first-attempt.txt");
std::fs::write(&first_artifact, b"first attempt evidence")?;
let process = LaunchedCoordinator {
process_id: u32::MAX - 1,
process_creation_token: 155,
process_executable_path: coordinator,
};
claim_resident_self_preparation_as_child(
&resident_store,
&prepared.preparation_id,
&process,
4,
)?;
let lease = acknowledge_resident_self_launch(
&resident_store,
&prepared.preparation_id,
&process,
5,
)?;
let runtime_session_id =
crate::coordinator_run_session_id(&lease.turn_id, Some(&lease.launch_digest))?;
let runtime_session_objective =
format!("Resident coordinator launch {}.", lease.objective_digest);
crate::open_coordinator_run(
&policy.runtime_store,
&runtime_session_id,
&lease.turn_id,
Some(&lease.launch_digest),
&runtime_session_objective,
"1970-01-01T00:00:00.005Z",
)?;
assert!(
recover_receipt_free_dead_coordinator_session(
&resident_store,
&policy.runtime_store,
&lease,
ChildObservation::Running,
6,
)
.is_err()
);
let recovery = recover_receipt_free_dead_coordinator_session(
&resident_store,
&policy.runtime_store,
&lease,
ChildObservation::Missing,
6,
)?
.expect("opened coordinator death recovery");
assert_eq!(
load_resident_self_state(&resident_store)?.active_turn,
Some(lease.clone())
);
assert_eq!(
recover_receipt_free_dead_coordinator_session(
&resident_store,
&policy.runtime_store,
&lease,
ChildObservation::Missing,
6,
)?,
Some(recovery.clone())
);
let fulfilled_copy = temp.path().join("resident-self-fulfilled-copy.cc");
std::fs::copy(&resident_store, &fulfilled_copy)?;
let recovered_ack = complete_resident_self_turn_after_death(
&fulfilled_copy,
&lease,
&recovery,
7,
policy.cooldown_seconds,
)?;
assert_eq!(recovered_ack.terminal_status, "recovered-fulfilled");
assert!(load_resident_self_state(&fulfilled_copy)?.active_turn.is_none());
assert!(crate::coordinator_run_receipts(&policy.runtime_store)?.is_empty());
cancel_resident_self_turn(
&resident_store,
&lease,
"process-failed",
"exact coordinator process died without a run receipt",
7,
)?;
assert!(load_resident_self_state(&resident_store)?.active_turn.is_none());
assert!(pending_resident_self_pressure(&resident_store)?);
let retry_grant = heartbeat_issue_resident_self_grant(
&resident_store,
"heartbeat-retry",
"action-retry",
8,
)?
.expect("retry grant");
let retry_prepared = prepare_resident_self_launch(&resident_store, &policy, 9)?
.expect("retry preparation");
let retry_artifact_index = retry_prepared
.argv
.iter()
.position(|value| value == "--artifact-dir")
.expect("retry artifact flag");
let retry_artifact_dir = PathBuf::from(&retry_prepared.argv[retry_artifact_index + 1]);
assert_ne!(retry_artifact_dir, first_artifact_dir);
assert_eq!(std::fs::read(&first_artifact)?, b"first attempt evidence");
let retry_thread = resident_prepared_launch_thread_id(&retry_prepared)?;
assert_eq!(retry_thread, lease.turn_id);
assert_ne!(retry_prepared.launch_digest, lease.launch_digest);
let retry_session_id = crate::coordinator_run_session_id(
&retry_thread,
Some(&retry_prepared.launch_digest),
)?;
assert_ne!(retry_session_id, runtime_session_id);
crate::open_coordinator_run(
&policy.runtime_store,
&retry_session_id,
&retry_thread,
Some(&retry_prepared.launch_digest),
&format!(
"Resident coordinator launch {}.",
retry_prepared.objective_digest
),
"1970-01-01T00:00:00.009Z",
)?;
assert_eq!(retry_grant.terminal_at_millis, None);
assert_eq!(
state_cache(&resident_store)?
.get::<ResidentSelfHeartbeatGrant>(&grant.grant_id)?
.expect("terminal original grant")
.terminal_status
.as_deref(),
Some("process-failed")
);
Ok(())
}
#[test]
fn receipt_free_death_before_runtime_opening_is_proven_absent_and_requeued() -> Result<()> {
let temp = tempfile::tempdir()?;
let resident_store = temp.path().join("resident-self.cc");
let coordinator = temp.path().join("epiphany-mvp-coordinator");
std::fs::write(&coordinator, b"witnessed executable")?;
let mut policy = policy();
policy.coordinator_bin = coordinator.clone();
policy.runtime_store = temp.path().join("runtime.cc");
policy.artifact_root = temp.path().join("artifacts");
seed_runtime_identity(&policy.runtime_store, "test-runtime")?;
enqueue_resident_self_pressure(
&resident_store,
&ResidentSelfPressure {
schema_version: RESIDENT_SELF_PRESSURE_SCHEMA_VERSION.into(),
pressure_id: "pressure-pre-open-death".into(),
kind: "operator-objective".into(),
provenance_ref: "operator://pre-open-death".into(),
objective: "Recover death before runtime opening.".into(),
created_at_millis: 1,
status: "pending".into(),
consumed_by_grant_id: None,
private_state_exposed: false,
},
)?;
heartbeat_issue_resident_self_grant(
&resident_store,
"heartbeat-pre-open",
"action-pre-open",
2,
)?
.expect("pre-open grant");
let prepared = prepare_resident_self_launch(&resident_store, &policy, 3)?
.expect("pre-open preparation");
let process = LaunchedCoordinator {
process_id: u32::MAX - 2,
process_creation_token: 255,
process_executable_path: coordinator,
};
claim_resident_self_preparation_as_child(
&resident_store,
&prepared.preparation_id,
&process,
4,
)?;
let lease = acknowledge_resident_self_launch(
&resident_store,
&prepared.preparation_id,
&process,
5,
)?;
assert_eq!(
recover_receipt_free_dead_coordinator_session(
&resident_store,
&policy.runtime_store,
&lease,
ChildObservation::Missing,
6,
)?,
None
);
cancel_resident_self_turn(
&resident_store,
&lease,
"process-failed",
"exact child died before runtime opening",
7,
)?;
assert!(pending_resident_self_pressure(&resident_store)?);
assert!(load_resident_self_state(&resident_store)?.active_turn.is_none());
Ok(())
}
#[test]
fn shutdown_cancellation_requeues_pressure_without_failure_backoff() -> Result<()> {
let temp = tempfile::tempdir()?;

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

@ -1083,6 +1083,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef<Path>) -> Result<CultCache> {
cache.register_entry_type::<EpiphanyRuntimeJobResult>()?;
cache.register_entry_type::<EpiphanyRuntimeEvent>()?;
cache.register_entry_type::<EpiphanyCoordinatorRunReceipt>()?;
cache.register_entry_type::<EpiphanyCoordinatorDeathRecovery>()?;
cache.register_entry_type::<EpiphanyCoordinatorRunReceiptRetentionHead>()?;
cache.register_entry_type::<MindGatewayReview>()?;
cache.register_entry_type::<MindStateCommitReceipt>()?;
@ -1928,6 +1929,7 @@ pub fn retain_completed_runtime_sessions(
let jobs = cache.get_all::<EpiphanyRuntimeJob>()?;
let bindings = cache.get_all::<EpiphanyRuntimeModelExecutionBinding>()?;
let receipts = cache.get_all::<EpiphanyCoordinatorRunReceipt>()?;
let death_recoveries = cache.get_all::<EpiphanyCoordinatorDeathRecovery>()?;
let events = cache.get_all::<EpiphanyRuntimeEvent>()?;
let mut candidates = cache
.get_all::<EpiphanyRuntimeSession>()?
@ -1945,11 +1947,15 @@ pub fn retain_completed_runtime_sessions(
});
let coordinator_family = session_jobs.is_empty()
&& session.session_id.starts_with("coordinator-")
&& receipts
&& (receipts
.iter()
.filter(|receipt| receipt.session_id == session.session_id)
.count()
== 1
+ death_recoveries
.iter()
.filter(|recovery| recovery.session_id == session.session_id)
.count()
== 1)
&& !receipts.iter().any(|receipt| {
receipt.session_id == session.session_id
&& preserve_coordinator_receipt_ids.contains(&receipt.receipt_id)
@ -2574,6 +2580,10 @@ where
.get_all::<EpiphanyCoordinatorRunReceipt>()?
.iter()
.any(|receipt| receipt.session_id == session_id)
|| cache
.get_all::<EpiphanyCoordinatorDeathRecovery>()?
.iter()
.any(|recovery| recovery.session_id == session_id)
|| cache
.get_all::<EpiphanyRuntimeEvent>()?
.iter()
@ -2604,16 +2614,28 @@ where
.into_iter()
.filter(|receipt| receipt.session_id == session_id)
.collect::<Vec<_>>();
if receipts.len() != 1 {
let recoveries = cache
.get_all::<EpiphanyCoordinatorDeathRecovery>()?
.into_iter()
.filter(|recovery| recovery.session_id == session_id)
.collect::<Vec<_>>();
if receipts.len() + recoveries.len() != 1 {
return Err(anyhow!(
"coordinator session archive requires one exact run receipt"
"coordinator session archive requires one exact terminal authority"
));
}
let receipt = &receipts[0];
if session_id
.strip_prefix("coordinator-")
.is_none_or(|thread_id| thread_id != receipt.thread_id)
{
let expected_session_id = if let Some(receipt) = receipts.first() {
coordinator_run_session_id(
&receipt.thread_id,
receipt.resident_launch_digest.as_deref(),
)?
} else {
coordinator_run_session_id(
&recoveries[0].thread_id,
Some(&recoveries[0].resident_launch_digest),
)?
};
if session_id != expected_session_id {
return Err(anyhow!(
"coordinator session archive found a substituted thread binding"
));
@ -2624,7 +2646,11 @@ where
.filter(|event| event.session_id.as_deref() == Some(session_id))
.collect::<Vec<_>>();
events.sort_by(|left, right| left.event_id.cmp(&right.event_id));
let expected_completion = coordinator_completion_event(receipt);
let expected_completion = if let Some(receipt) = receipts.first() {
coordinator_completion_event(receipt)
} else {
coordinator_death_recovery_event(&recoveries[0])
};
if events
.iter()
.filter(|event| **event == expected_completion)
@ -2641,7 +2667,11 @@ where
|| event.source != "epiphany-mvp-coordinator"
|| event.job_id.is_some())
})
|| session.updated_at != receipt.created_at
|| session.updated_at
!= match receipts.first() {
Some(receipt) => receipt.created_at.as_str(),
None => recoveries[0].recovered_at.as_str(),
}
|| session.coordinator_note != expected_completion.summary
{
return Err(anyhow!(
@ -2650,12 +2680,17 @@ where
}
let snapshot = cache.snapshot_envelopes();
let mut deletions = Vec::new();
let terminal_envelope = if let Some(receipt) = receipts.first() {
(EpiphanyCoordinatorRunReceipt::TYPE, receipt.receipt_id.clone())
} else {
(
EpiphanyCoordinatorDeathRecovery::TYPE,
recoveries[0].recovery_id.clone(),
)
};
for (document_type, key) in
std::iter::once((EpiphanyRuntimeSession::TYPE, session_id.to_string()))
.chain(std::iter::once((
EpiphanyCoordinatorRunReceipt::TYPE,
receipt.receipt_id.clone(),
)))
.chain(std::iter::once(terminal_envelope))
.chain(
events
.iter()
@ -10108,6 +10143,37 @@ fn validate_coordinator_run_receipt(receipt: &EpiphanyCoordinatorRunReceipt) ->
Ok(())
}
pub fn runtime_typed_request_attempt_exists(
store_path: impl AsRef<Path>,
request: RuntimeTypedRequestRef<'_>,
) -> Result<bool> {
let request_id = match request {
RuntimeTypedRequestRef::ProposalModeling(id)
| RuntimeTypedRequestRef::ImaginationConsideration(id)
| RuntimeTypedRequestRef::AdmittedModelDirection(id) => id,
};
validate_non_empty(request_id, "typed attempt request id")?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
Ok(cache
.get_all::<EpiphanyRuntimeWorkerLaunchRequest>()?
.iter()
.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)
}
}))
}
fn coordinator_completion_event(receipt: &EpiphanyCoordinatorRunReceipt) -> EpiphanyRuntimeEvent {
EpiphanyRuntimeEvent {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
@ -10125,10 +10191,71 @@ fn coordinator_completion_event(receipt: &EpiphanyCoordinatorRunReceipt) -> Epip
}
}
pub const COORDINATOR_DEATH_RECOVERY_SCHEMA_VERSION: &str =
"epiphany.coordinator_death_recovery.v0";
#[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)]
#[cultcache(
type = "epiphany.coordinator_death_recovery.v0",
schema = "EpiphanyCoordinatorDeathRecovery"
)]
pub struct EpiphanyCoordinatorDeathRecovery {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub recovery_id: String,
#[cultcache(key = 2)]
pub session_id: String,
#[cultcache(key = 3)]
pub thread_id: String,
#[cultcache(key = 4)]
pub resident_grant_id: String,
#[cultcache(key = 5)]
pub resident_launch_digest: String,
#[cultcache(key = 6)]
pub process_id: u32,
#[cultcache(key = 7)]
pub process_creation_token: u64,
#[cultcache(key = 8)]
pub process_executable_path: String,
#[cultcache(key = 9)]
pub resident_started_at_millis: u64,
#[cultcache(key = 10)]
pub observation: String,
#[cultcache(key = 11)]
pub recovered_at: String,
#[cultcache(key = 12, default)]
pub private_state_exposed: bool,
#[cultcache(key = 13, default)]
pub exit_code: Option<i32>,
}
pub fn coordinator_run_session_id(
thread_id: &str,
resident_launch_digest: Option<&str>,
) -> Result<String> {
validate_non_empty(thread_id, "coordinator run thread id")?;
match resident_launch_digest {
None => Ok(format!("coordinator-{thread_id}")),
Some(digest) => {
let hex = digest.strip_prefix("sha256:").ok_or_else(|| {
anyhow!("resident coordinator session requires a SHA-256 launch digest")
})?;
if hex.len() != 64 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) {
return Err(anyhow!(
"resident coordinator session launch digest is invalid"
));
}
Ok(format!("coordinator-{thread_id}-resident-{}", hex.to_ascii_lowercase()))
}
}
}
pub fn open_coordinator_run(
store_path: impl AsRef<Path>,
session_id: &str,
thread_id: &str,
resident_launch_digest: Option<&str>,
objective: &str,
started_at: &str,
) -> Result<(EpiphanyRuntimeSession, EpiphanyRuntimeEvent)> {
@ -10136,6 +10263,7 @@ pub fn open_coordinator_run(
store_path,
session_id,
thread_id,
resident_launch_digest,
objective,
started_at,
|| Ok(()),
@ -10146,6 +10274,7 @@ fn open_coordinator_run_with_before_commit<F>(
store_path: impl AsRef<Path>,
session_id: &str,
thread_id: &str,
resident_launch_digest: Option<&str>,
objective: &str,
started_at: &str,
before_commit: F,
@ -10158,7 +10287,7 @@ where
validate_non_empty(objective, "coordinator run objective")?;
chrono::DateTime::parse_from_rfc3339(started_at)
.map_err(|error| anyhow!("coordinator run start timestamp is invalid: {error}"))?;
if session_id != format!("coordinator-{thread_id}") {
if session_id != coordinator_run_session_id(thread_id, resident_launch_digest)? {
return Err(anyhow!(
"coordinator run session id is not bound to its thread"
));
@ -10169,11 +10298,21 @@ where
require_identity(&cache)?;
let event_id = format!("event-coordinator-started-{session_id}");
if cache.get::<EpiphanyRuntimeSession>(session_id)?.is_some()
|| cache
.get::<EpiphanyArchivedRuntimeSession>(session_id)?
.is_some()
|| cache.get::<EpiphanyRuntimeEvent>(&event_id)?.is_some()
|| cache
.get_all::<EpiphanyCoordinatorRunReceipt>()?
.iter()
.any(|receipt| receipt.session_id == session_id || receipt.thread_id == thread_id)
.any(|receipt| {
receipt.session_id == session_id
|| (resident_launch_digest.is_none() && receipt.thread_id == thread_id)
})
|| cache
.get_all::<EpiphanyCoordinatorDeathRecovery>()?
.iter()
.any(|recovery| recovery.session_id == session_id)
{
return Err(anyhow!(
"coordinator run session or thread authority already exists"
@ -10224,6 +10363,189 @@ pub fn finalize_coordinator_run(
finalize_coordinator_run_with_before_commit(store_path, receipt, || Ok(()))
}
fn coordinator_death_recovery_event(
recovery: &EpiphanyCoordinatorDeathRecovery,
) -> EpiphanyRuntimeEvent {
EpiphanyRuntimeEvent {
schema_version: RUNTIME_SPINE_SCHEMA_VERSION.to_string(),
event_id: format!("event-coordinator-death-recovered-{}", recovery.session_id),
occurred_at: recovery.recovered_at.clone(),
event_type: "coordinator.death-recovered".to_string(),
source: "epiphany-continuity".to_string(),
session_id: Some(recovery.session_id.clone()),
job_id: None,
summary: format!(
"Continuity terminalized coordinator session after exact process observation {:?}{}.",
recovery.observation,
recovery
.exit_code
.map(|code| format!(" with exit code {code}"))
.unwrap_or_default()
),
metadata: BTreeMap::new(),
}
}
pub(crate) fn recover_coordinator_run_after_exact_process_death(
store_path: impl AsRef<Path>,
recovery: &EpiphanyCoordinatorDeathRecovery,
expected_objective: &str,
) -> Result<EpiphanyRuntimeSession> {
recover_coordinator_run_after_exact_process_death_with_before_commit(
store_path,
recovery,
expected_objective,
|| Ok(()),
)
}
pub(crate) fn coordinator_run_incarnation_is_absent(
store_path: impl AsRef<Path>,
thread_id: &str,
resident_launch_digest: &str,
) -> Result<bool> {
let session_id = coordinator_run_session_id(thread_id, Some(resident_launch_digest))?;
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
Ok(cache.get::<EpiphanyRuntimeSession>(&session_id)?.is_none()
&& cache
.get::<EpiphanyArchivedRuntimeSession>(&session_id)?
.is_none()
&& !cache
.get_all::<EpiphanyRuntimeEvent>()?
.iter()
.any(|event| event.session_id.as_deref() == Some(session_id.as_str()))
&& !cache
.get_all::<EpiphanyRuntimeJob>()?
.iter()
.any(|job| job.session_id == session_id)
&& !cache
.get_all::<EpiphanyCoordinatorRunReceipt>()?
.iter()
.any(|receipt| receipt.session_id == session_id)
&& !cache
.get_all::<EpiphanyCoordinatorDeathRecovery>()?
.iter()
.any(|recovery| recovery.session_id == session_id))
}
fn recover_coordinator_run_after_exact_process_death_with_before_commit<F>(
store_path: impl AsRef<Path>,
recovery: &EpiphanyCoordinatorDeathRecovery,
expected_objective: &str,
before_commit: F,
) -> Result<EpiphanyRuntimeSession>
where
F: FnOnce() -> Result<()>,
{
validate_non_empty(expected_objective, "recovered coordinator objective")?;
if recovery.schema_version != COORDINATOR_DEATH_RECOVERY_SCHEMA_VERSION
|| recovery.recovery_id != format!("coordinator-death-recovery-{}", recovery.session_id)
|| recovery.session_id
!= coordinator_run_session_id(
&recovery.thread_id,
Some(&recovery.resident_launch_digest),
)?
|| recovery.resident_grant_id.trim().is_empty()
|| !recovery.resident_launch_digest.starts_with("sha256:")
|| recovery.process_id == 0
|| recovery.process_creation_token == 0
|| recovery.process_executable_path.trim().is_empty()
|| recovery.resident_started_at_millis == 0
|| !matches!(recovery.observation.as_str(), "exited" | "missing")
|| (recovery.observation == "missing" && recovery.exit_code.is_some())
|| recovery.private_state_exposed
{
return Err(anyhow!("coordinator death recovery contract is invalid"));
}
chrono::DateTime::parse_from_rfc3339(&recovery.recovered_at)
.map_err(|error| anyhow!("coordinator death recovery timestamp is invalid: {error}"))?;
let store_path = store_path.as_ref();
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
require_identity(&cache)?;
let mut session = cache
.get::<EpiphanyRuntimeSession>(&recovery.session_id)?
.ok_or_else(|| anyhow!("coordinator death recovery session is absent"))?;
let expected_event = coordinator_death_recovery_event(recovery);
if session.status == EpiphanyRuntimeSessionStatus::Completed {
let existing_recovery = cache
.get::<EpiphanyCoordinatorDeathRecovery>(&recovery.recovery_id)?;
let existing_event = cache.get::<EpiphanyRuntimeEvent>(&expected_event.event_id)?;
if existing_recovery.as_ref() == Some(recovery)
&& existing_event.as_ref() == Some(&expected_event)
&& session.objective == expected_objective
&& session.updated_at == recovery.recovered_at
&& session.coordinator_note == expected_event.summary
{
return Ok(session);
}
return Err(anyhow!(
"completed coordinator session does not match exact death recovery"
));
}
if session.status != EpiphanyRuntimeSessionStatus::Active
|| session.objective != expected_objective
{
return Err(anyhow!(
"coordinator death recovery requires the exact active session objective"
));
}
let start_event_id = format!("event-coordinator-started-{}", recovery.session_id);
let start_event = cache
.get::<EpiphanyRuntimeEvent>(&start_event_id)?
.ok_or_else(|| anyhow!("coordinator death recovery lost its start event"))?;
let session_started_at = chrono::DateTime::parse_from_rfc3339(&session.created_at)
.map_err(|error| anyhow!("coordinator session start timestamp is invalid: {error}"))?;
let recovered_at = chrono::DateTime::parse_from_rfc3339(&recovery.recovered_at)
.map_err(|error| anyhow!("coordinator death recovery timestamp is invalid: {error}"))?;
if (session_started_at.timestamp_millis().max(0) as u64)
< recovery.resident_started_at_millis
|| recovered_at < session_started_at
|| start_event.occurred_at != session.created_at
|| start_event.event_type != "coordinator.started"
|| start_event.source != "epiphany-mvp-coordinator"
|| start_event.session_id.as_deref() != Some(recovery.session_id.as_str())
|| start_event.job_id.is_some()
|| cache
.get_all::<EpiphanyRuntimeJob>()?
.iter()
.any(|job| job.session_id == recovery.session_id)
|| cache
.get_all::<EpiphanyCoordinatorRunReceipt>()?
.iter()
.any(|receipt| receipt.session_id == recovery.session_id)
|| cache
.get_all::<EpiphanyCoordinatorDeathRecovery>()?
.iter()
.any(|existing| existing.session_id == recovery.session_id)
|| cache.get::<EpiphanyRuntimeEvent>(&expected_event.event_id)?.is_some()
{
return Err(anyhow!(
"coordinator death recovery found substituted or competing authority"
));
}
let snapshot = cache.snapshot_envelopes();
session.status = EpiphanyRuntimeSessionStatus::Completed;
session.updated_at = recovery.recovered_at.clone();
session.coordinator_note = expected_event.summary.clone();
let replacements = vec![
cache.prepare_entry(&session.session_id, &session)?.0,
cache.prepare_entry(&recovery.recovery_id, recovery)?.0,
cache.prepare_entry(&expected_event.event_id, &expected_event)?.0,
];
before_commit()?;
if !runtime_spine_backing_store(store_path)?
.replace_and_append_if_snapshot_unchanged(&snapshot, replacements)?
{
return Err(anyhow!(
"coordinator death recovery lost its full snapshot fence"
));
}
Ok(session)
}
fn finalize_coordinator_run_with_before_commit<F>(
store_path: impl AsRef<Path>,
receipt: &EpiphanyCoordinatorRunReceipt,
@ -10233,6 +10555,16 @@ where
F: FnOnce() -> Result<()>,
{
validate_coordinator_run_receipt(receipt)?;
if receipt.session_id
!= coordinator_run_session_id(
&receipt.thread_id,
receipt.resident_launch_digest.as_deref(),
)?
{
return Err(anyhow!(
"coordinator run receipt session is not bound to its run incarnation"
));
}
let store_path = store_path.as_ref();
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
@ -12677,6 +13009,14 @@ pub(crate) mod tests {
.proposal_modeling_request_id
.clone()
.expect("proposal request echo");
assert!(runtime_typed_request_attempt_exists(
&store,
RuntimeTypedRequestRef::ProposalModeling(&request_id),
)?);
assert!(!runtime_typed_request_attempt_exists(
&store,
RuntimeTypedRequestRef::ProposalModeling("absent-proposal-request"),
)?);
assert_eq!(
runtime_typed_request_fulfillment(
&store,
@ -15935,6 +16275,7 @@ pub(crate) mod tests {
&store,
"coordinator-thread-open",
"thread-open",
None,
"Open one coordinator authority family.",
"2026-08-10T10:00:01Z",
)?;
@ -15947,6 +16288,7 @@ pub(crate) mod tests {
&store,
"coordinator-thread-open",
"thread-open",
None,
"Open one coordinator authority family.",
"2026-08-10T10:00:01Z",
)
@ -15958,6 +16300,7 @@ pub(crate) mod tests {
&store,
"coordinator-swapped",
"thread-bound",
None,
"Refuse split identity.",
"2026-08-10T10:00:02Z",
)
@ -15968,6 +16311,7 @@ pub(crate) mod tests {
&store,
"coordinator-thread-raced-open",
"thread-raced-open",
None,
"Lose the opening snapshot.",
"2026-08-10T10:00:03Z",
|| {
@ -16009,6 +16353,129 @@ pub(crate) mod tests {
Ok(())
}
#[test]
fn coordinator_death_recovery_is_exact_single_use_and_archivable() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("runtime.msgpack");
initialize_runtime_spine(
&store,
RuntimeSpineInitOptions {
runtime_id: "epiphany-coordinator-death-test".to_string(),
display_name: "Epiphany Coordinator Death Test".to_string(),
created_at: "2026-08-10T11:00:00Z".to_string(),
},
)?;
let objective = "Recover one exact dead coordinator.";
let launch_digest = format!("sha256:{}", "a".repeat(64));
let session_id = coordinator_run_session_id("thread-dead", Some(&launch_digest))?;
open_coordinator_run(
&store,
&session_id,
"thread-dead",
Some(&launch_digest),
objective,
"2026-08-10T11:00:01Z",
)?;
let recovery = EpiphanyCoordinatorDeathRecovery {
schema_version: COORDINATOR_DEATH_RECOVERY_SCHEMA_VERSION.to_string(),
recovery_id: format!("coordinator-death-recovery-{session_id}"),
session_id,
thread_id: "thread-dead".to_string(),
resident_grant_id: "grant-dead".to_string(),
resident_launch_digest: launch_digest,
process_id: 41,
process_creation_token: 73,
process_executable_path: "/release/epiphany-mvp-coordinator".to_string(),
resident_started_at_millis: 1,
observation: "missing".to_string(),
recovered_at: "2026-08-10T11:00:02Z".to_string(),
private_state_exposed: false,
exit_code: None,
};
let completed = recover_coordinator_run_after_exact_process_death(
&store,
&recovery,
objective,
)?;
assert_eq!(completed.status, EpiphanyRuntimeSessionStatus::Completed);
assert_eq!(
recover_coordinator_run_after_exact_process_death(&store, &recovery, objective)?,
completed
);
let mut substituted = recovery.clone();
substituted.process_creation_token += 1;
assert!(
recover_coordinator_run_after_exact_process_death(&store, &substituted, objective)
.is_err()
);
let archived = archive_completed_coordinator_session(
&store,
&recovery.session_id,
"2026-08-10T11:00:03Z",
)?;
assert_eq!(archived.retired_envelope_count, 4);
let mut cache = runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
assert!(
cache
.get::<EpiphanyCoordinatorDeathRecovery>(&recovery.recovery_id)?
.is_none()
);
let raced_launch_digest = format!("sha256:{}", "b".repeat(64));
let raced_session_id =
coordinator_run_session_id("thread-raced-death", Some(&raced_launch_digest))?;
open_coordinator_run(
&store,
&raced_session_id,
"thread-raced-death",
Some(&raced_launch_digest),
objective,
"2026-08-10T11:01:01Z",
)?;
let mut raced = recovery.clone();
raced.recovery_id = format!("coordinator-death-recovery-{raced_session_id}");
raced.session_id = raced_session_id;
raced.thread_id = "thread-raced-death".to_string();
raced.resident_launch_digest = raced_launch_digest;
let result = recover_coordinator_run_after_exact_process_death_with_before_commit(
&store,
&raced,
objective,
|| {
append_runtime_event(
&store,
RuntimeSpineEventOptions {
event_id: "event-racing-death-recovery".to_string(),
occurred_at: "2026-08-10T11:01:02Z".to_string(),
event_type: "test.concurrent".to_string(),
source: "test".to_string(),
session_id: None,
job_id: None,
summary: "Invalidate death recovery snapshot.".to_string(),
},
)?;
Ok(())
},
);
assert!(result.is_err());
let mut cache = runtime_spine_cache(&store)?;
cache.pull_all_backing_stores()?;
assert_eq!(
cache
.get::<EpiphanyRuntimeSession>(&raced.session_id)?
.expect("raced active session")
.status,
EpiphanyRuntimeSessionStatus::Active
);
assert!(
cache
.get::<EpiphanyCoordinatorDeathRecovery>(&raced.recovery_id)?
.is_none()
);
Ok(())
}
#[test]
fn coordinator_session_retention_repairs_legacy_and_owns_the_exact_family() -> Result<()> {
let temp = tempdir()?;

57
notes/epiphany-current-algorithmic-map.md
View file

@ -2860,6 +2860,63 @@ bind the opened session to exact process or resident lease identity, prove that
identity dead, and terminalize or quarantine the session through a dedicated
receipt-grounded recovery path before retention can see it as closed.
#### Receipt-free coordinator death authority map
- Owner: resident Self owns the exact child lease and its process-incarnation
observation; runtime Continuity owns terminal recovery of the corresponding
orphaned coordinator session.
- Inputs: the persisted active resident lease, its immutable child claim and
terminal grant, an exact `Missing` or `Exited` observation for that process
incarnation, and the runtime's matching Active coordinator session plus
deterministic start event.
- Outputs: one typed coordinator-death recovery receipt and one deterministic
recovery event, atomically replacing the runtime session as Completed. Self
then uses its existing exact-grant cancellation transaction to requeue the
pressure.
- Identity split: the cognitive thread remains request-owned context. A
resident coordinator session is a distinct run incarnation derived from the
exact resident launch digest; ordinary non-resident runs retain the simple
`coordinator-{thread}` identity. Requeue therefore creates a new session
without rewriting or reusing the cognitive thread.
- Derived state: timestamps, session age, receipt absence, PID presence, and
operator projections remain observations only. None can establish death.
- Forbidden writers: retention, legacy coordinator repair, the coordinator
itself after death, generic runtime session closure, and timeout policy may
not synthesize this terminal family. A normal coordinator receipt and a
death-recovery receipt are mutually exclusive.
- Shared paths: zero exit without a receipt, nonzero exit without a receipt,
and restart observation of a missing exact child share one settlement
classifier. Death before runtime opening proves that exact incarnation family
wholly absent and requeues without inventing runtime state. Death after
opening uses the recovery transaction. Normal exits with an exact fully bound
run receipt retain the normal finalizer path.
- Typed-work boundary: authenticated terminal typed fulfillment lets the
death-recovery authority close the original grant exactly once without a
duplicate launch. No typed attempt permits canonical requeue. A typed launch
attempt without terminal fulfillment remains active and
`awaiting-fulfillment`; Continuity does not yet own orphan-worker terminality
or supersession, so timeout or brake cannot manufacture a retry.
- Cut line: replace direct receipt-free resident cancellation with runtime
recovery followed by canonical resident cancellation, and stop using the
cognitive thread as the resident run identity; do not add an age-based
scavenger or let session retention repair Active state.
- Verification: exact lease/claim/session/thread/objective/start-event binding,
hostile substitution refusal, live or indeterminate process refusal,
single-use replay, full-snapshot CAS refusal, mutual exclusion with normal
receipts, restart recovery, pressure requeue without grant resurrection, and
eventual archive of the recovered terminal family.
Retry artifacts are grant-incarnation-owned rather than thread-owned. Prepared
argv carries `artifact_root/{grant_id}`, so a new launch on the same cognitive
thread cannot erase the prior attempt's sealed directory.
Source verification for this cut is complete: the serial core suite passed 653
tests with one intentional ignore, followed by all 17 coordinator-binary tests
and all 8 swarm tests. The detached receipt is
`.epiphany-run/continuity-death-recovery-source`; stderr is empty and exit is
zero. Exact packaging and copied-live process-death falsification remain the
acceptance boundary.
### Semantic logical lifecycle implementation
`retain_memory_semantic_projection_lifecycles` now realizes the safe unattempted

45
notes/fresh-workspace-handoff.md
View file

@ -2075,3 +2075,48 @@ Normal coordinator lifecycle ownership is accepted. The next distinct fault is
a hard process death after atomic opening but before any receipt. Continuity
must ground that repair in exact process/lease authority; age, retention, and
absence of a receipt cannot infer death.
## Receipt-free coordinator death recovery source cut — 2026-08-10
The Continuity owner is now implemented but not yet packaged. Resident Self
requires its exact persisted active lease, immutable child claim, active
unterminated grant, and an independent native observation proving the exact
process incarnation exited, disappeared, or was replaced. Exact-alive,
inaccessible, and indeterminate observations refuse recovery. Runtime
Continuity then atomically replaces only the matching Active jobless
coordinator session as Completed while appending one typed death-recovery
receipt and one deterministic recovery event. A normal coordinator receipt and
a death-recovery receipt cannot coexist. Replays are exact and snapshot races
leave the Active family unchanged.
The cut exposed and removed a deeper identity conflation. The cognitive thread
is request-owned context and must survive retries; it is not a run identity.
Ordinary coordinators retain `coordinator-{thread}` sessions, while resident
coordinator sessions are now derived from the exact launch digest. A cancelled
grant therefore terminalizes and requeues through the existing resident CAS,
then the new grant produces a distinct session incarnation on the same
cognitive thread. Archived session tombstones and terminal families still
refuse exact incarnation reuse.
The receipt-free classifier distinguishes three crash boundaries. If the exact
process died before atomic runtime opening, the entire incarnation family must
be absent and resident cancellation may requeue without synthetic runtime
state. If a typed result is already authenticated, runtime death recovery can
terminalize the process and the original grant closes as
`recovered-fulfilled`. If a typed worker launch exists without terminal
fulfillment, the active lease remains `awaiting-fulfillment`; duplicate launch
is forbidden until orphan-worker terminality and explicit supersession have a
real owner. This last Continuity seam remains open. Retry artifact directories
are keyed by grant incarnation, not cognitive thread, preserving prior attempt
evidence byte-for-byte.
Focused death-recovery, resident requeue/new-incarnation, and legacy retention
tests pass. Final serial verification under
`.epiphany-run/continuity-death-recovery-source` passed 653 library tests with
one intentional ignore, all 17 coordinator-binary tests, and all 8 swarm tests;
stderr is empty and the detached suite recorded exit zero. The first parallel
full-library run had exposed one real eager-fallback archive bug plus two Git
fixture flakes; the archive bug is fixed, the serial suite removes the fixture
contention, and the complete corrected source pass is green. Commit, exact
packaging, and copied-live hard-death proof remain. Active c005 remains
untouched.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

8
state/map.yaml
View file

@ -36,9 +36,9 @@ invariants:
current_status:
summary: Epiphany is a supervised engineering alpha on Starfire, not deployment-ready. Exact b8138a6b active-child SIGTERM and restart, exact 9f369528 deterministic warm packaging, exact c930f505 cross-container ownership, exact ed15f83a resident lifecycle retention, exact 16c4b19f coordinator receipt and Heartbeat artifact retention, exact cce2b8ac narrow model-generation retention, exact eea15e82 model-turn JSON ingress, exact b1e8c89a semantic unattempted-generation retention, exact 6bc3ec3f semantic physical-epoch retirement, and exact 44230a05 Persona conversation retention are proven. Gamecult-ops 059821f owns Yggdrasil stream retention and exact 6ee08a4 owns stable Starfire Persona-permit firewall ingress. Exact 16c4b19f genuine idle and exact a987d2f7 active cognition completed 900-pulse bounded plateaus. Runtime-owned typed fulfillment is accepted live under exact 6dd9b132. Yggdrasil is the small public crossing body, never the release forge.
latest_phase: Exact 497df458 coordinator lifecycle ownership is accepted for all normal exits and legacy receipt-grounded repair. Release sha256-3563c909 with witness sha256-d02d2fe5 passed 24-binary inspection with no warnings; identical warm replay completed in 13.82s wall and Cargo 0.83s. Copied-live execution atomically terminalized a deliberate coordinator error, repaired and archived one older exact legacy family, retained the newer failed family, exposed zero active sessions/jobs/actionable frontiers, replayed with a byte-identical runtime store, and held that digest across 26 additional braked serve cycles at roughly 28 MiB before SIGTERM exit zero. Hard process death after atomic opening but before a receipt remains the next Continuity boundary.
latest_phase: Receipt-free coordinator death recovery is source-verified and ready to commit. Resident Self preserves the exact child claim/lease and independently refuses live or indeterminate process identity; the only exposed combined primitive lets runtime Continuity atomically close the matching Active session with a dedicated recovery receipt/event. Pre-open death instead proves the exact runtime family absent before canonical requeue. Cognitive thread identity is no longer misused as run identity: each launch digest derives a unique resident session and each grant owns a distinct artifact directory. Exact typed fulfillment can close once through death recovery; a typed launch without terminal fulfillment remains fail-closed awaiting orphan-worker Continuity rather than duplicating work. Normal receipts are fully lease-authenticated and mutually exclusive with recovery. Serial verification passed 653 library tests with one intentional ignore, all 17 coordinator-binary tests, and all 8 swarm tests; package and copied-live proof remain open.
operational_state:
local_worktree: coordinator_lifecycle_497df458_live_accepted_pending_receipt_commit
local_worktree: coordinator_death_recovery_source_verified_pending_commit
operator_wireguard: restored
starfire_persona_permit_firewall: persistent_exact_gamecult_ops_6ee08a4
yggdrasil: canonical_small_public_crossing_body
@ -69,8 +69,8 @@ current_status:
starfire_linux_modeling_peak: 109887488_bytes_resident_body_plus_modeling_circuit
yggdrasil_epiphany_service_contract: repaired_in_gamecult_ops_812b427_not_installed
resident_physiology: authenticated_9b8df0b4_c005_bound_to_eyes_tools_nervous_system_with_217dd_and_bd96305e_helpers_running
unresolved_authority: Coordinator receipt, Heartbeat pulse-artifact, deployed supervisor/projector log retention, genuine-idle and active resource plateaus, exact-grant terminality, restart-persistent cooldown refusal, schedule/action issuance cardinality, runtime-owned typed fulfillment, 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, and normal coordinator lifecycle opening/error closure/legacy repair/archival are accepted live. A hard process death between atomic opening and terminal receipt remains Continuity-owned recovery work rather than inferred retention. Historical migrated semantic attempts without recorded physical namespace remain durable and non-retirable. Repository Body observation still performs two authenticated full-tree Git index/manifest passes. Execute-mode workers require a persistent process body. Keyed storage has not earned live migration. One completed Discord consequence remains blocked by the absent Yggdrasil credential.
next_action: Map and implement receipt-free dead coordinator-session recovery from exact process/lease authority; do not infer death from age or let retention synthesize terminality. Then continue supervisor/projector log ownership and remaining organ/endurance audit. Preserve active c005 and exact prior receipts. Yggdrasil never builds.
unresolved_authority: Coordinator receipt, Heartbeat pulse-artifact, deployed supervisor/projector log retention, genuine-idle and active resource plateaus, exact-grant terminality, restart-persistent cooldown refusal, schedule/action issuance cardinality, runtime-owned typed fulfillment, 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, and normal coordinator lifecycle opening/error closure/legacy repair/archival are accepted live. Receipt-free hard-death recovery is source-implemented but remains uncommitted, unpackaged, and unproven on copied exact pre-open/opened process kills. A typed worker launch without valid terminal fulfillment deliberately remains active awaiting an owner for exact orphan-worker death and supersession; it is not retryable yet. Historical migrated semantic attempts without recorded physical namespace remain durable and non-retirable. Repository Body observation still performs two authenticated full-tree Git index/manifest passes. Execute-mode workers require a persistent process body. Keyed storage has not earned live migration. One completed Discord consequence remains blocked by the absent Yggdrasil credential.
next_action: Commit and push the source-verified Continuity cut, package it once with the persistent Starfire cache, then prove copied exact pre-open and post-open/pre-attempt deaths, canonical requeue under a new session/artifact incarnation, post-fulfillment recovery, archival/non-resurrection, and bounded replay. Then map orphan-worker death/supersession for typed launched attempts. Preserve active c005. Yggdrasil never builds.
active_subgoals:
- Prove Persona, Self, Eyes, Modeling, Imagination, Hands, Soul, Mind, Continuity, and nervous-system behavior individually.