From d3300bba95871add262bd6cfe67e62473b4e7c3c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 18 Aug 2026 07:27:31 +0200 Subject: [PATCH] Prove decision-state reentry --- epiphany-core/src/current_work.rs | 22 ++ epiphany-core/src/persona_conversation.rs | 92 ++++- epiphany-core/src/reasoning_context.rs | 78 +++- .../src/bin/epiphany-persona-service.rs | 373 +++++++++++------- .../src/persona_executor.rs | 149 ++++--- 5 files changed, 496 insertions(+), 218 deletions(-) diff --git a/epiphany-core/src/current_work.rs b/epiphany-core/src/current_work.rs index 1d8a7e98..5cbd6e72 100644 --- a/epiphany-core/src/current_work.rs +++ b/epiphany-core/src/current_work.rs @@ -3191,6 +3191,24 @@ mod tests { model_basis_with_payload(vec![1]) } + fn assert_current_work_reentry_is_read_only( + store: &Path, + expected: &EpiphanyCurrentWorkProjection, + ) -> Result<()> { + let mut before = crate::runtime_spine_cache(store)?; + before.pull_all_backing_stores()?; + let before = before.snapshot_envelopes(); + assert_eq!(&project_current_work(store)?, expected); + let mut after = crate::runtime_spine_cache(store)?; + after.pull_all_backing_stores()?; + let after = after.snapshot_envelopes(); + assert_eq!(after, before); + assert!(after.iter().all(|envelope| { + envelope.r#type != "epiphany.thread_state" && envelope.r#type != "epiphany.memory_graph" + })); + Ok(()) + } + #[test] fn body_generation_owns_work_identity_while_repo_model_is_projection_cargo() -> Result<()> { let first = EpiphanyBodyModelingWorkProjection::derive( @@ -3314,6 +3332,7 @@ mod tests { crate::assemble_mind_view(&store)?.projection_digest ); assert_eq!(crate::repository_body_read_counters(), (0, 0)); + assert_current_work_reentry_is_read_only(&store, ¤t_work)?; let resident_pressure = crate::resident_self_body_modeling_pressure(&store, 1)? .expect("unresolved Body work must create Resident Self pressure"); assert_eq!( @@ -3339,6 +3358,7 @@ mod tests { scheduled_work.body_modeling_action, Some(EpiphanyAgentPassContinuationAction::Wait) ); + assert_current_work_reentry_is_read_only(&store, &scheduled_work)?; assert!(crate::resident_self_body_modeling_pressure(&store, 2)?.is_none()); let mut scheduled_cache = crate::runtime_spine_cache(&store)?; scheduled_cache.pull_all_backing_stores()?; @@ -3485,6 +3505,7 @@ mod tests { .is_none() ); assert_eq!(crate::repository_body_read_counters(), (0, 0)); + assert_current_work_reentry_is_read_only(&store, &completed_work)?; let proposal = crate::intake_user_repo_frontier_proposal( &store, @@ -5414,6 +5435,7 @@ mod tests { assert!(after_research.body_modeling_action.is_none()); assert!(after_research.proposal_modeling.is_none()); assert!(after_research.frontier_verdict_modeling.is_none()); + assert_current_work_reentry_is_read_only(&store, &after_research)?; Ok(()) } } diff --git a/epiphany-core/src/persona_conversation.rs b/epiphany-core/src/persona_conversation.rs index d2ed9b29..e212094e 100644 --- a/epiphany-core/src/persona_conversation.rs +++ b/epiphany-core/src/persona_conversation.rs @@ -45,11 +45,7 @@ pub fn admit_persona_pass_input( )); } let cache = runtime_spine_cache(runtime_store)?; - let write = crate::mind_documents::prepare_mind_document( - &cache, - &document.turn_id, - document, - )?; + let write = crate::mind_documents::prepare_mind_document(&cache, &document.turn_id, document)?; match crate::reasoning_context::commit_external_typed_observation_mind_mutation( runtime_store, "Persona", @@ -76,6 +72,59 @@ pub fn admit_persona_pass_input( crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &envelope) } +pub fn load_admitted_persona_pass_input( + runtime_store: &Path, + turn_id: &str, +) -> Result< + Option<( + crate::EpiphanyMindPersonaPassInputDocument, + crate::EpiphanyMindDocumentVersion, + )>, +> { + let mut cache = runtime_spine_cache(runtime_store)?; + cache.pull_all_backing_stores()?; + let Some(document) = cache.get::(turn_id)? else { + return Ok(None); + }; + document.validate()?; + let envelope = cache + .get_envelope::(turn_id)? + .ok_or_else(|| anyhow!("Persona pass input lost its exact envelope"))?; + let source = crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &envelope)?; + let candidates = cache + .get_all::()? + .into_iter() + .filter(|receipt| { + receipt.invariant_owner == "Persona.pass_input" + && receipt.writes.iter().any(|write| write == &source) + }) + .collect::>(); + for receipt in &candidates { + receipt.validate()?; + } + let receipts = candidates + .into_iter() + .filter(|receipt| { + receipt.committed_at == document.admitted_at + && receipt.strong_reads.is_empty() + && receipt.writes == vec![source.clone()] + && matches!( + &receipt.authority, + crate::EpiphanyMindCommitAuthority::TypedOrganProvenance { + organ, + provenance, + } if organ == "Persona" && document.observed_sources.contains(provenance) + ) + }) + .collect::>(); + if receipts.len() != 1 { + return Err(anyhow!( + "Persona pass input does not have one exact causal Mind commit receipt" + )); + } + Ok(Some((document, source))) +} + #[derive(Clone, Debug, PartialEq, DatabaseEntry)] #[cultcache( type = "epiphany.persona_effect_execution_intent.v0", @@ -1090,9 +1139,7 @@ pub(crate) fn admit_persona_state_notes( writes, &document.created_at, )? { - crate::EpiphanyMindCommitOutcome::Committed(_) => { - Ok(("admitted".into(), Vec::new())) - } + crate::EpiphanyMindCommitOutcome::Committed(_) => Ok(("admitted".into(), Vec::new())), crate::EpiphanyMindCommitOutcome::Conflict { document_identities, } => Err(anyhow!( @@ -1620,7 +1667,8 @@ mod tests { } #[test] - fn persona_pass_input_is_admitted_once_and_substitution_refuses_byte_identically() -> Result<()> { + fn persona_pass_input_is_admitted_once_and_substitution_refuses_byte_identically() -> Result<()> + { let temp = tempfile::tempdir()?; let runtime = temp.path().join("runtime.cc"); crate::initialize_runtime_spine( @@ -1636,10 +1684,8 @@ mod tests { .into_iter() .find(|entry| entry.r#type == crate::RUNTIME_IDENTITY_TYPE) .expect("runtime identity provenance"); - let provenance = crate::EpiphanyMindDocumentVersion::from_envelope( - "epiphany-heartbeat", - &identity, - )?; + let provenance = + crate::EpiphanyMindDocumentVersion::from_envelope("epiphany-heartbeat", &identity)?; let document = crate::EpiphanyMindPersonaPassInputDocument { turn_id: "persona-pass-1".into(), projector_input: crate::PersonaProjectorInput { @@ -1661,11 +1707,25 @@ mod tests { admit_persona_pass_input(&runtime, provenance.clone(), &document)?, first ); + assert_eq!( + load_admitted_persona_pass_input(&runtime, &document.turn_id)?, + Some((document.clone(), first.clone())) + ); let before = SingleFileMessagePackBackingStore::new(&runtime).pull_all()?; - let mut substituted = document; + let mut substituted = document.clone(); substituted.allowed_channel_ids = vec!["elsewhere".into()]; - assert!(admit_persona_pass_input(&runtime, provenance, &substituted).is_err()); - assert_eq!(SingleFileMessagePackBackingStore::new(&runtime).pull_all()?, before); + assert!(admit_persona_pass_input(&runtime, provenance.clone(), &substituted).is_err()); + assert_eq!( + SingleFileMessagePackBackingStore::new(&runtime).pull_all()?, + before + ); + + let mut naked = document; + naked.turn_id = "persona-pass-without-commit".into(); + let mut cache = crate::runtime_spine_cache(&runtime)?; + cache.pull_all_backing_stores()?; + cache.put(&naked.turn_id, &naked)?; + assert!(load_admitted_persona_pass_input(&runtime, &naked.turn_id).is_err()); Ok(()) } diff --git a/epiphany-core/src/reasoning_context.rs b/epiphany-core/src/reasoning_context.rs index b0823eb5..c020148e 100644 --- a/epiphany-core/src/reasoning_context.rs +++ b/epiphany-core/src/reasoning_context.rs @@ -447,6 +447,45 @@ pub struct EpiphanyMindCommitReceipt { pub committed_at: String, } +impl EpiphanyMindCommitReceipt { + pub fn validate(&self) -> Result<()> { + if self.schema_version != MIND_COMMIT_RECEIPT_SCHEMA_VERSION { + return Err(anyhow!("Mind commit receipt schema is unsupported")); + } + require_non_empty(&self.invariant_owner, "Mind commit invariant owner")?; + chrono::DateTime::parse_from_rfc3339(&self.committed_at) + .map_err(|error| anyhow!("Mind commit time is invalid: {error}"))?; + if self.writes.is_empty() { + return Err(anyhow!("Mind commit receipt has no writes")); + } + match &self.authority { + EpiphanyMindCommitAuthority::ModelDecisionContext { + decision_context_id, + } => require_non_empty(decision_context_id, "Mind decision context")?, + EpiphanyMindCommitAuthority::OperatorProvenance { provenance } => { + provenance.validate()? + } + EpiphanyMindCommitAuthority::TypedOrganProvenance { organ, provenance } => { + require_non_empty(organ, "Mind provenance organ")?; + provenance.validate()?; + } + } + validate_mind_document_versions(&self.strong_reads, "strong read")?; + validate_mind_document_versions(&self.writes, "write")?; + if self.receipt_id + != mind_commit_receipt_id( + &self.authority, + &self.invariant_owner, + &self.strong_reads, + &self.writes, + )? + { + return Err(anyhow!("Mind commit receipt identity digest mismatch")); + } + Ok(()) + } +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum EpiphanyMindCommitAuthority { ModelDecisionContext { @@ -1057,6 +1096,7 @@ fn commit_authorized_mind_mutation( writes: write_versions, committed_at: committed_at.to_string(), }; + receipt.validate()?; if let Some(existing) = cache.get::(&receipt_id)? { if existing != receipt { return Err(anyhow!("Mind commit receipt identity collision")); @@ -1258,6 +1298,20 @@ fn validate_unique_envelope_identities(entries: &[CultCacheEnvelope], label: &st Ok(()) } +fn validate_mind_document_versions( + entries: &[EpiphanyMindDocumentVersion], + label: &str, +) -> Result<()> { + let mut identities = BTreeSet::new(); + for entry in entries { + entry.validate()?; + if !identities.insert(entry.identity()) { + return Err(anyhow!("Mind commit receipt repeats {label} identity")); + } + } + Ok(()) +} + fn current_write_collisions( current: &[CultCacheEnvelope], writes: &[EpiphanyMindDocumentVersion], @@ -1553,8 +1607,8 @@ mod tests { } #[test] - fn decision_context_reuses_runtime_binding_owner_and_survives_transcript_deletion() - -> Result<()> { + fn decision_context_reuses_runtime_binding_owner_and_survives_transcript_deletion() -> Result<()> + { let temp = tempdir()?; let store = temp.path().join("mind.cc"); initialize_runtime_spine( @@ -1677,12 +1731,15 @@ mod tests { cache = runtime_spine_cache(&store)?; cache.put("request-2:00000000", &native_delta)?; cache.put("request-2:00000000", &provider_delta)?; - assert!(cache.delete::( - "request-2:00000000" - )?); - assert!(cache.delete::( - "request-2:00000000" - )?); + assert!( + cache + .delete::("request-2:00000000")? + ); + assert!( + cache.delete::( + "request-2:00000000" + )? + ); cache.pull_all_backing_stores()?; let retained = cache .get::(&context.context_id)? @@ -1699,7 +1756,10 @@ mod tests { cache.put(&intent.intent_id, &hostile_binding)?; let before = SingleFileMessagePackBackingStore::new(&store).pull_all()?; assert!(seal_model_decision_context(&store, &terminal_native.request_id).is_err()); - assert_eq!(SingleFileMessagePackBackingStore::new(&store).pull_all()?, before); + assert_eq!( + SingleFileMessagePackBackingStore::new(&store).pull_all()?, + before + ); Ok(()) } diff --git a/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs b/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs index 97e96157..46d5a93a 100644 --- a/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs +++ b/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs @@ -5,12 +5,12 @@ use cultcache_rs::{CacheBackingStore, DatabaseEntry, SingleFileMessagePackBackin use epiphany_core::{ EpiphanyAgentMemoryEntry, EpiphanyHeartbeatStateEntry, EpiphanyMindDocumentVersion, EpiphanyMindPersonaPassInputDocument, PersonaIdentity, PersonaProjectorInput, - PersonaRepoActivity, PersonaSocialAffordance, PersonaTranscriptMessage, + PersonaRepoActivity, PersonaSocialAffordance, PersonaTranscriptMessage, PersonaTurnRequest, PersonaTurnTerminalOptions, admit_persona_pass_input, assemble_mind_view, complete_persona_turn_request_store, default_organ_dependencies_for, - exchange_persona_discord_delivery_rudp, load_agent_memory_entry_for_role, - load_epiphany_cultmesh_swarm_brake, load_heartbeat_state_entry, - load_persona_discord_receipt_anchor, + exchange_persona_discord_delivery_rudp, load_admitted_persona_pass_input, + load_agent_memory_entry_for_role, load_epiphany_cultmesh_swarm_brake, + load_heartbeat_state_entry, load_persona_discord_receipt_anchor, load_persona_discord_service_anchor, open_persona_discord_request_identity, pending_persona_discord_delivery_request_for_turn, persona_delivery_receipt_exists_for_turn, persona_model_terminal_exists, poll_persona_discord_crossing, @@ -90,149 +90,15 @@ async fn poll_once(options: &Options) -> Result { let Some(request) = candidates.into_iter().next() else { return Ok(false); }; - let memory = load_agent_memory_entry_for_role(&options.agent_store, &request.role_id) - .ok() - .flatten(); - let identity = PersonaIdentity { - identity_id: request.agent_id.clone(), - display_name: options.persona_name.clone(), - repo_name: options.repo_name.clone(), - public_description: options.persona_description.clone(), - jurisdiction: vec![options.repo_name.clone()], - }; - let transcript = request - .mentions - .iter() - .map(|mention| PersonaTranscriptMessage { - channel_id: mention.channel_id.clone(), - message_id: mention.message_id.clone(), - author_id: mention.author_id.clone(), - author_name: mention - .author_name - .clone() - .unwrap_or_else(|| mention.author_id.clone()), - is_agent: false, - content: mention.content.clone(), - timestamp: mention.queued_at.clone(), - }) - .collect::>(); - let semantic_recall = semantic_memory_recall_from_heartbeat_action( - &serde_json::json!({"persona_memory_recall": request.semantic_memory_recall}), - ); - let social_affordances = request - .mentions - .iter() - .map(|mention| PersonaSocialAffordance { - person_id: mention.author_id.clone(), - summary: format!( - "{} directly addressed the Persona in this reserved turn.", - mention.author_name.as_deref().unwrap_or(&mention.author_id) - ), - recent_message_ids: vec![mention.message_id.clone()], - }) - .collect(); - let mind = assemble_mind_view(&options.runtime_store)?; - let repo_activity = mind - .repository_body_observation - .as_ref() - .map(|body| PersonaRepoActivity { - repo_name: options.repo_name.clone(), - summary: format!( - "Typed repository Body generation {} at manifest {}.", - body.generation, body.manifest_root_sha256 - ), - refs: vec![body.observation_id.clone()], - }) - .into_iter() - .collect(); - let has_memory = memory.is_some(); - let projector_input = PersonaProjectorInput { - identity, - memory, - semantic_memory_recall: semantic_recall.clone(), - pending_mentions: request.mentions.clone(), - repo_activity, - social_affordances, - organ_dependencies: vec![default_organ_dependencies_for("Persona")], - }; - let mut observed_sources = Vec::new(); - let heartbeat_envelope = SingleFileMessagePackBackingStore::new(&options.heartbeat_store) - .pull_all()? - .into_iter() - .find(|entry| entry.r#type == EpiphanyHeartbeatStateEntry::TYPE) - .ok_or_else(|| anyhow!("Persona pass input lost its heartbeat source"))?; - observed_sources.push(EpiphanyMindDocumentVersion::from_envelope( - "epiphany-heartbeat", - &heartbeat_envelope, - )?); - if has_memory { - let agent_envelope = SingleFileMessagePackBackingStore::new(&options.agent_store) - .pull_all()? - .into_iter() - .find(|entry| { - entry.r#type == EpiphanyAgentMemoryEntry::TYPE && entry.key == request.role_id - }) - .ok_or_else(|| anyhow!("Persona pass input lost its agent-memory source"))?; - observed_sources.push(EpiphanyMindDocumentVersion::from_envelope( - "epiphany-agent-memory", - &agent_envelope, - )?); - } - if let Some(source) = mind.source_documents.iter().find(|source| { - source.document_type - == epiphany_core::EpiphanyMindRepositoryBodyObservationDocument::TYPE - && mind.repository_body_observation.as_ref().is_some_and(|body| { - source.document_key == body.observation_id - }) - }) { - observed_sources.push(source.clone()); - } - let observed_pass_input = EpiphanyMindPersonaPassInputDocument { - turn_id: request.request_id.clone(), - projector_input, - transcript, - allowed_channel_ids: request - .mentions - .iter() - .map(|mention| mention.channel_id.clone()) - .collect(), - dynamic_semantic_memory_recall: semantic_recall, - observed_sources, - admitted_at: request.reserved_at.clone(), - }; - let mut runtime_cache = epiphany_core::runtime_spine_cache(&options.runtime_store)?; - runtime_cache.pull_all_backing_stores()?; - let (pass_input, source) = if let Some(existing) = runtime_cache - .get::(&request.request_id)? - { - existing.validate()?; - let envelope = runtime_cache - .get_envelope::(&request.request_id)? - .ok_or_else(|| anyhow!("Persona pass input lost its exact envelope"))?; - ( - existing, - EpiphanyMindDocumentVersion::from_envelope("epiphany-mind", &envelope)?, - ) - } else { - let source = admit_persona_pass_input( - &options.runtime_store, - observed_pass_input.observed_sources[0].clone(), - &observed_pass_input, - )?; - (observed_pass_input, source) - }; - let plan = PersonaModelExecutionPlan { - turn_id: request.request_id.clone(), - provider: options.provider.clone(), - model: options.model.clone(), - projector_input: pass_input.projector_input, - transcript: pass_input.transcript, - allowed_channel_ids: pass_input.allowed_channel_ids, - dynamic_semantic_memory_recall: pass_input.dynamic_semantic_memory_recall, - source_documents: vec![source], - cultmesh_store: options.cultmesh_store.clone(), - runtime_id: options.runtime_id.clone(), - }; + ensure_persona_pass_input_admitted(options, &request)?; + let plan = PersonaModelExecutionPlan::from_admitted_input( + &options.runtime_store, + &request.request_id, + options.provider.clone(), + options.model.clone(), + options.cultmesh_store.clone(), + options.runtime_id.clone(), + )?; let mut runner = NativePersonaModelRunner { store_path: options.runtime_store.clone(), codex_home: options.codex_home.clone(), @@ -310,6 +176,128 @@ async fn poll_once(options: &Options) -> Result { Ok(result.is_some()) } +fn ensure_persona_pass_input_admitted( + options: &Options, + request: &PersonaTurnRequest, +) -> Result<()> { + if load_admitted_persona_pass_input(&options.runtime_store, &request.request_id)?.is_some() { + return Ok(()); + } + + let memory = load_agent_memory_entry_for_role(&options.agent_store, &request.role_id) + .ok() + .flatten(); + let identity = PersonaIdentity { + identity_id: request.agent_id.clone(), + display_name: options.persona_name.clone(), + repo_name: options.repo_name.clone(), + public_description: options.persona_description.clone(), + jurisdiction: vec![options.repo_name.clone()], + }; + let transcript = request + .mentions + .iter() + .map(|mention| PersonaTranscriptMessage { + channel_id: mention.channel_id.clone(), + message_id: mention.message_id.clone(), + author_id: mention.author_id.clone(), + author_name: mention + .author_name + .clone() + .unwrap_or_else(|| mention.author_id.clone()), + is_agent: false, + content: mention.content.clone(), + timestamp: mention.queued_at.clone(), + }) + .collect::>(); + let semantic_recall = semantic_memory_recall_from_heartbeat_action( + &serde_json::json!({"persona_memory_recall": request.semantic_memory_recall}), + ); + let social_affordances = request + .mentions + .iter() + .map(|mention| PersonaSocialAffordance { + person_id: mention.author_id.clone(), + summary: format!( + "{} directly addressed the Persona in this reserved turn.", + mention.author_name.as_deref().unwrap_or(&mention.author_id) + ), + recent_message_ids: vec![mention.message_id.clone()], + }) + .collect(); + let mind = assemble_mind_view(&options.runtime_store)?; + let repo_activity = mind + .repository_body_observation + .as_ref() + .map(|body| PersonaRepoActivity { + repo_name: options.repo_name.clone(), + summary: format!( + "Typed repository Body generation {} at manifest {}.", + body.generation, body.manifest_root_sha256 + ), + refs: vec![body.observation_id.clone()], + }) + .into_iter() + .collect(); + let has_memory = memory.is_some(); + let projector_input = PersonaProjectorInput { + identity, + memory, + semantic_memory_recall: semantic_recall.clone(), + pending_mentions: request.mentions.clone(), + repo_activity, + social_affordances, + organ_dependencies: vec![default_organ_dependencies_for("Persona")], + }; + let heartbeat_envelope = SingleFileMessagePackBackingStore::new(&options.heartbeat_store) + .pull_all()? + .into_iter() + .find(|entry| entry.r#type == EpiphanyHeartbeatStateEntry::TYPE) + .ok_or_else(|| anyhow!("Persona pass input lost its heartbeat source"))?; + let heartbeat_source = + EpiphanyMindDocumentVersion::from_envelope("epiphany-heartbeat", &heartbeat_envelope)?; + let mut observed_sources = vec![heartbeat_source.clone()]; + if has_memory { + let agent_envelope = SingleFileMessagePackBackingStore::new(&options.agent_store) + .pull_all()? + .into_iter() + .find(|entry| { + entry.r#type == EpiphanyAgentMemoryEntry::TYPE && entry.key == request.role_id + }) + .ok_or_else(|| anyhow!("Persona pass input lost its agent-memory source"))?; + observed_sources.push(EpiphanyMindDocumentVersion::from_envelope( + "epiphany-agent-memory", + &agent_envelope, + )?); + } + if let Some(source) = mind.source_documents.iter().find(|source| { + source.document_type == epiphany_core::EpiphanyMindRepositoryBodyObservationDocument::TYPE + && mind + .repository_body_observation + .as_ref() + .is_some_and(|body| source.document_key == body.observation_id) + }) { + observed_sources.push(source.clone()); + } + let document = EpiphanyMindPersonaPassInputDocument { + turn_id: request.request_id.clone(), + projector_input, + transcript, + allowed_channel_ids: request + .mentions + .iter() + .map(|mention| mention.channel_id.clone()) + .collect(), + dynamic_semantic_memory_recall: semantic_recall, + observed_sources, + admitted_at: request.reserved_at.clone(), + }; + admit_persona_pass_input(&options.runtime_store, heartbeat_source, &document)?; + load_admitted_persona_pass_input(&options.runtime_store, &request.request_id)? + .ok_or_else(|| anyhow!("Persona pass input admission was not durable"))?; + Ok(()) +} + struct Options { runtime_store: PathBuf, heartbeat_store: PathBuf, @@ -396,3 +384,94 @@ impl Options { fn value(values: &std::collections::BTreeMap, key: &str, default: &str) -> String { values.get(key).cloned().unwrap_or_else(|| default.into()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn admitted_reentry_does_not_reobserve_external_stores() -> Result<()> { + let temp = tempfile::tempdir()?; + let runtime_store = temp.path().join("runtime.cc"); + epiphany_core::initialize_runtime_spine( + &runtime_store, + epiphany_core::RuntimeSpineInitOptions { + runtime_id: "persona-reentry".into(), + display_name: "Persona reentry".into(), + created_at: "2026-08-18T03:00:00Z".into(), + }, + )?; + let mut cache = epiphany_core::runtime_spine_cache(&runtime_store)?; + cache.pull_all_backing_stores()?; + let provenance = EpiphanyMindDocumentVersion::from_envelope( + "epiphany-heartbeat", + &cache + .get_envelope::( + epiphany_core::RUNTIME_IDENTITY_KEY, + )? + .expect("runtime identity"), + )?; + let document = EpiphanyMindPersonaPassInputDocument { + turn_id: "turn-reentry".into(), + projector_input: PersonaProjectorInput { + identity: PersonaIdentity { + identity_id: "epiphany.Persona".into(), + display_name: "Epiphany".into(), + ..Default::default() + }, + ..Default::default() + }, + transcript: Vec::new(), + allowed_channel_ids: vec!["aquarium".into()], + dynamic_semantic_memory_recall: String::new(), + observed_sources: vec![provenance.clone()], + admitted_at: "2026-08-18T03:00:01Z".into(), + }; + admit_persona_pass_input(&runtime_store, provenance, &document)?; + let request = PersonaTurnRequest { + request_id: document.turn_id.clone(), + role_id: "Persona".into(), + agent_id: "epiphany.Persona".into(), + reserved_at: document.admitted_at.clone(), + status: "reserved".into(), + ..Default::default() + }; + let absent = temp.path().join("intentionally-absent"); + let options = Options { + runtime_store: runtime_store.clone(), + heartbeat_store: absent.join("heartbeat.cc"), + agent_store: absent.join("agent.cc"), + cultmesh_store: absent.join("cultmesh.cc"), + codex_home: absent.join("codex-home"), + runtime_id: "persona-reentry".into(), + provider: "test".into(), + model: "test-model".into(), + repo_name: "Epiphany".into(), + persona_name: "Epiphany".into(), + persona_description: String::new(), + _repo_root: absent.join("repo"), + mouth_request_store: absent.join("mouth-requests.cc"), + mouth_receipt_store: absent.join("mouth-receipts.cc"), + mouth_identity_store: absent.join("mouth-identity.cc"), + mouth_request_anchor: absent.join("request-anchor.cc"), + mouth_receipt_anchor: absent.join("receipt-anchor.cc"), + mouth_rudp: "127.0.0.1:1".parse()?, + mouth_timeout_ms: 1, + retained_terminal_conversations: 1, + poll_ms: 1, + once: true, + }; + + ensure_persona_pass_input_admitted(&options, &request)?; + PersonaModelExecutionPlan::from_admitted_input( + &runtime_store, + &request.request_id, + "test", + "test-model", + options.cultmesh_store, + "persona-reentry", + )?; + assert!(!absent.exists()); + Ok(()) + } +} diff --git a/epiphany-openai-runtime/src/persona_executor.rs b/epiphany-openai-runtime/src/persona_executor.rs index 12bc92e4..1e6a3c22 100644 --- a/epiphany-openai-runtime/src/persona_executor.rs +++ b/epiphany-openai-runtime/src/persona_executor.rs @@ -1,5 +1,5 @@ use std::future::Future; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::pin::Pin; use anyhow::{Context, Result, anyhow}; @@ -22,16 +22,43 @@ use crate::{EpiphanyOpenAiRuntimeOptions, assistant_text_from_model_events, run_ #[derive(Clone, Debug)] pub struct PersonaModelExecutionPlan { - pub turn_id: String, - pub provider: String, - pub model: String, - pub projector_input: PersonaProjectorInput, - pub transcript: Vec, - pub allowed_channel_ids: Vec, - pub dynamic_semantic_memory_recall: String, - pub source_documents: Vec, - pub cultmesh_store: PathBuf, - pub runtime_id: String, + turn_id: String, + provider: String, + model: String, + projector_input: PersonaProjectorInput, + transcript: Vec, + allowed_channel_ids: Vec, + dynamic_semantic_memory_recall: String, + source_documents: Vec, + cultmesh_store: PathBuf, + runtime_id: String, +} + +impl PersonaModelExecutionPlan { + pub fn from_admitted_input( + store_path: &Path, + turn_id: &str, + provider: impl Into, + model: impl Into, + cultmesh_store: PathBuf, + runtime_id: impl Into, + ) -> Result { + let (document, source) = + epiphany_core::load_admitted_persona_pass_input(store_path, turn_id)? + .ok_or_else(|| anyhow!("Persona execution plan has no admitted Mind input"))?; + Ok(Self { + turn_id: document.turn_id, + provider: provider.into(), + model: model.into(), + projector_input: document.projector_input, + transcript: document.transcript, + allowed_channel_ids: document.allowed_channel_ids, + dynamic_semantic_memory_recall: document.dynamic_semantic_memory_recall, + source_documents: vec![source], + cultmesh_store, + runtime_id: runtime_id.into(), + }) + } } #[derive(Clone, Debug)] @@ -569,10 +596,28 @@ mod tests { } fn plan(store: &PathBuf, cultmesh_store: PathBuf) -> Result { - let mut plan = PersonaModelExecutionPlan { + if !epiphany_core::runtime_spine_status(store)?.present { + epiphany_core::initialize_runtime_spine( + store, + epiphany_core::RuntimeSpineInitOptions { + runtime_id: "persona-test".into(), + display_name: "Persona test".into(), + created_at: "2026-08-14T00:00:00Z".into(), + }, + )?; + } + let mut cache = epiphany_core::runtime_spine_cache(store)?; + cache.pull_all_backing_stores()?; + let provenance = epiphany_core::EpiphanyMindDocumentVersion::from_envelope( + "epiphany-runtime-bootstrap", + &cache + .get_envelope::( + epiphany_core::RUNTIME_IDENTITY_KEY, + )? + .ok_or_else(|| anyhow!("test runtime identity is missing"))?, + )?; + let document = epiphany_core::EpiphanyMindPersonaPassInputDocument { turn_id: "turn-1".into(), - provider: "test".into(), - model: "test-model".into(), projector_input: PersonaProjectorInput { identity: PersonaIdentity { identity_id: "epiphany.Persona".into(), @@ -586,17 +631,7 @@ mod tests { transcript: vec![], allowed_channel_ids: vec!["aquarium".into()], dynamic_semantic_memory_recall: String::new(), - source_documents: Vec::new(), - cultmesh_store, - runtime_id: "epiphany-test".into(), - }; - let document = epiphany_core::EpiphanyMindPersonaPassInputDocument { - turn_id: plan.turn_id.clone(), - projector_input: plan.projector_input.clone(), - transcript: plan.transcript.clone(), - allowed_channel_ids: plan.allowed_channel_ids.clone(), - dynamic_semantic_memory_recall: plan.dynamic_semantic_memory_recall.clone(), - observed_sources: Vec::new(), + observed_sources: vec![provenance.clone()], admitted_at: "2026-08-14T00:00:00Z".into(), }; if load_document::( @@ -605,20 +640,23 @@ mod tests { )? .is_none() { - put_new_document(store, &document.turn_id, &document)?; + epiphany_core::admit_persona_pass_input(store, provenance, &document)?; } - let mut cache = epiphany_core::runtime_spine_cache(store)?; - cache.pull_all_backing_stores()?; - let envelope = cache - .get_envelope::(&document.turn_id)? - .expect("test Persona pass input"); - plan.source_documents = vec![ - epiphany_core::EpiphanyMindDocumentVersion::from_envelope( - "epiphany-mind", - &envelope, - )?, - ]; - Ok(plan) + admitted_plan(store, cultmesh_store) + } + + fn admitted_plan( + store: &PathBuf, + cultmesh_store: PathBuf, + ) -> Result { + PersonaModelExecutionPlan::from_admitted_input( + store, + "turn-1", + "test", + "test-model", + cultmesh_store, + "epiphany-test", + ) } fn release_brake(path: &PathBuf) -> Result<()> { @@ -637,14 +675,32 @@ mod tests { let cultmesh = dir.path().join("cultmesh.cc"); release_brake(&cultmesh)?; let mut runner = FakeRunner { calls: vec![] }; - let first = - execute_persona_model_turn_with_runner(&store, &plan(&store, cultmesh.clone())?, &mut runner) - .await?; + let first = execute_persona_model_turn_with_runner( + &store, + &plan(&store, cultmesh.clone())?, + &mut runner, + ) + .await?; assert_eq!(runner.calls, ["projector", "persona", "interpreter"]); + drop(runner); + + let restarted_plan = admitted_plan(&store, cultmesh)?; + let admitted_source = restarted_plan.source_documents[0].clone(); + let mut restarted_runner = FakeRunner { calls: vec![] }; let second = - execute_persona_model_turn_with_runner(&store, &plan(&store, cultmesh)?, &mut runner).await?; + execute_persona_model_turn_with_runner(&store, &restarted_plan, &mut restarted_runner) + .await?; assert_eq!(first, second); - assert_eq!(runner.calls.len(), 3); + assert!(restarted_runner.calls.is_empty()); + for receipt_id in &second.stage_receipt_ids { + let receipt = load_document::(&store, receipt_id)?.unwrap(); + let basis = load_document::( + &store, + &receipt.reasoning_basis_id, + )? + .unwrap(); + assert_eq!(basis.source_documents, vec![admitted_source.clone()]); + } let effects = load_document::(&store, &first.effect_document_id)? .unwrap(); @@ -699,9 +755,10 @@ mod tests { put_new_document(&store, &poisoned.receipt_id, &poisoned)?; let mut runner = FakeRunner { calls: vec![] }; - let error = execute_persona_model_turn_with_runner(&store, &plan(&store, cultmesh)?, &mut runner) - .await - .unwrap_err(); + let error = + execute_persona_model_turn_with_runner(&store, &plan(&store, cultmesh)?, &mut runner) + .await + .unwrap_err(); assert!(error.to_string().contains("replay binding is invalid")); assert!(runner.calls.is_empty()); Ok(())