diff --git a/epiphany-core/src/bin/epiphany-mvp-coordinator.rs b/epiphany-core/src/bin/epiphany-mvp-coordinator.rs index a213604a..b4750d2b 100644 --- a/epiphany-core/src/bin/epiphany-mvp-coordinator.rs +++ b/epiphany-core/src/bin/epiphany-mvp-coordinator.rs @@ -301,7 +301,21 @@ fn run_coordinator(args: &Args) -> Result { "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 { &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!({ diff --git a/epiphany-core/src/bin/epiphany-swarm.rs b/epiphany-core/src/bin/epiphany-swarm.rs index 1374954d..d6b77343 100644 --- a/epiphany-core/src/bin/epiphany-swarm.rs +++ b/epiphany-core/src/bin/epiphany-swarm.rs @@ -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> { + 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::>(); + 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 { + 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()? { diff --git a/epiphany-core/src/lib.rs b/epiphany-core/src/lib.rs index 019793f2..84689940 100644 --- a/epiphany-core/src/lib.rs +++ b/epiphany-core/src/lib.rs @@ -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; diff --git a/epiphany-core/src/resident_self.rs b/epiphany-core/src/resident_self.rs index 8fdcb312..ed78ef7a 100644 --- a/epiphany-core/src/resident_self.rs +++ b/epiphany-core/src/resident_self.rs @@ -553,9 +553,12 @@ pub fn coordinator_argv( policy: &ResidentSelfPolicy, runtime_id: &str, turn_id: &str, + artifact_incarnation: Option<&str>, wake: &ResidentSelfWake, ) -> Vec { - 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::(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>> { + 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 { + let grant = state_cache(resident_store)? + .get::(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 { + let grant = state_cache(resident_store)? + .get::(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> { + 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::(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::(&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::::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> { 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 { + 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 { + 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 { let cache = state_cache(path)?; let mut state = cache .get::(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::(&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()?; diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index c9971329..0d598d99 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -1083,6 +1083,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef) -> Result { cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; + cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; @@ -1928,6 +1929,7 @@ pub fn retain_completed_runtime_sessions( let jobs = cache.get_all::()?; let bindings = cache.get_all::()?; let receipts = cache.get_all::()?; + let death_recoveries = cache.get_all::()?; let events = cache.get_all::()?; let mut candidates = cache .get_all::()? @@ -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::()? .iter() .any(|receipt| receipt.session_id == session_id) + || cache + .get_all::()? + .iter() + .any(|recovery| recovery.session_id == session_id) || cache .get_all::()? .iter() @@ -2604,16 +2614,28 @@ where .into_iter() .filter(|receipt| receipt.session_id == session_id) .collect::>(); - if receipts.len() != 1 { + let recoveries = cache + .get_all::()? + .into_iter() + .filter(|recovery| recovery.session_id == session_id) + .collect::>(); + 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::>(); 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, + request: RuntimeTypedRequestRef<'_>, +) -> Result { + 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::()? + .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, +} + +pub fn coordinator_run_session_id( + thread_id: &str, + resident_launch_digest: Option<&str>, +) -> Result { + 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, 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( store_path: impl AsRef, 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::(session_id)?.is_some() + || cache + .get::(session_id)? + .is_some() || cache.get::(&event_id)?.is_some() || cache .get_all::()? .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::()? + .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, + recovery: &EpiphanyCoordinatorDeathRecovery, + expected_objective: &str, +) -> Result { + 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, + thread_id: &str, + resident_launch_digest: &str, +) -> Result { + 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::(&session_id)?.is_none() + && cache + .get::(&session_id)? + .is_none() + && !cache + .get_all::()? + .iter() + .any(|event| event.session_id.as_deref() == Some(session_id.as_str())) + && !cache + .get_all::()? + .iter() + .any(|job| job.session_id == session_id) + && !cache + .get_all::()? + .iter() + .any(|receipt| receipt.session_id == session_id) + && !cache + .get_all::()? + .iter() + .any(|recovery| recovery.session_id == session_id)) +} + +fn recover_coordinator_run_after_exact_process_death_with_before_commit( + store_path: impl AsRef, + recovery: &EpiphanyCoordinatorDeathRecovery, + expected_objective: &str, + before_commit: F, +) -> Result +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::(&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::(&recovery.recovery_id)?; + let existing_event = cache.get::(&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::(&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::()? + .iter() + .any(|job| job.session_id == recovery.session_id) + || cache + .get_all::()? + .iter() + .any(|receipt| receipt.session_id == recovery.session_id) + || cache + .get_all::()? + .iter() + .any(|existing| existing.session_id == recovery.session_id) + || cache.get::(&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( store_path: impl AsRef, 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::(&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::(&raced.session_id)? + .expect("raced active session") + .status, + EpiphanyRuntimeSessionStatus::Active + ); + assert!( + cache + .get::(&raced.recovery_id)? + .is_none() + ); + Ok(()) + } + #[test] fn coordinator_session_retention_repairs_legacy_and_owns_the_exact_family() -> Result<()> { let temp = tempdir()?; diff --git a/notes/epiphany-current-algorithmic-map.md b/notes/epiphany-current-algorithmic-map.md index 42cf9d95..77d36f9a 100644 --- a/notes/epiphany-current-algorithmic-map.md +++ b/notes/epiphany-current-algorithmic-map.md @@ -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 diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index 9b15de17..8b2e6ce9 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -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. diff --git a/state/ledgers.msgpack b/state/ledgers.msgpack index ea667ec9..28f8c6e7 100644 Binary files a/state/ledgers.msgpack and b/state/ledgers.msgpack differ diff --git a/state/map.yaml b/state/map.yaml index 52e892f4..d1b4ee67 100644 --- a/state/map.yaml +++ b/state/map.yaml @@ -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.