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

Prove decision-state reentry

Browse source
This commit is contained in:
Meta Vens 2026-08-18 07:27:31 +02:00
commit d3300bba95
5 changed files with 472 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

22
epiphany-core/src/current_work.rs
View file

@ -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, &current_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(())
}
}

92
epiphany-core/src/persona_conversation.rs
View file

@ -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::<crate::EpiphanyMindPersonaPassInputDocument>(turn_id)? else {
return Ok(None);
};
document.validate()?;
let envelope = cache
.get_envelope::<crate::EpiphanyMindPersonaPassInputDocument>(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::<crate::EpiphanyMindCommitReceipt>()?
.into_iter()
.filter(|receipt| {
receipt.invariant_owner == "Persona.pass_input"
&& receipt.writes.iter().any(|write| write == &source)
})
.collect::<Vec<_>>();
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::<Vec<_>>();
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(())
}

78
epiphany-core/src/reasoning_context.rs
View file

@ -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::<EpiphanyMindCommitReceipt>(&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::<epiphany_model_adapter::EpiphanyModelStreamEvent>(
"request-2:00000000"
)?);
assert!(cache.delete::<epiphany_openai_adapter::EpiphanyOpenAiStreamEvent>(
"request-2:00000000"
)?);
assert!(
cache
.delete::<epiphany_model_adapter::EpiphanyModelStreamEvent>("request-2:00000000")?
);
assert!(
cache.delete::<epiphany_openai_adapter::EpiphanyOpenAiStreamEvent>(
"request-2:00000000"
)?
);
cache.pull_all_backing_stores()?;
let retained = cache
.get::<EpiphanyDecisionContext>(&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(())
}

373
epiphany-openai-runtime/src/bin/epiphany-persona-service.rs
View file

@ -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<bool> {
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::<Vec<_>>();
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::<EpiphanyMindPersonaPassInputDocument>(&request.request_id)?
{
existing.validate()?;
let envelope = runtime_cache
.get_envelope::<EpiphanyMindPersonaPassInputDocument>(&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<bool> {
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::<Vec<_>>();
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<String, String>, 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::EpiphanyRuntimeIdentity>(
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(())
}
}

149
epiphany-openai-runtime/src/persona_executor.rs
View file

@ -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<PersonaTranscriptMessage>,
pub allowed_channel_ids: Vec<String>,
pub dynamic_semantic_memory_recall: String,
pub source_documents: Vec<epiphany_core::EpiphanyMindDocumentVersion>,
pub cultmesh_store: PathBuf,
pub runtime_id: String,
turn_id: String,
provider: String,
model: String,
projector_input: PersonaProjectorInput,
transcript: Vec<PersonaTranscriptMessage>,
allowed_channel_ids: Vec<String>,
dynamic_semantic_memory_recall: String,
source_documents: Vec<epiphany_core::EpiphanyMindDocumentVersion>,
cultmesh_store: PathBuf,
runtime_id: String,
}
impl PersonaModelExecutionPlan {
pub fn from_admitted_input(
store_path: &Path,
turn_id: &str,
provider: impl Into<String>,
model: impl Into<String>,
cultmesh_store: PathBuf,
runtime_id: impl Into<String>,
) -> Result<Self> {
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<PersonaModelExecutionPlan> {
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::EpiphanyRuntimeIdentity>(
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::<epiphany_core::EpiphanyMindPersonaPassInputDocument>(
@ -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::<epiphany_core::EpiphanyMindPersonaPassInputDocument>(&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> {
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::<PersonaModelStageReceipt>(&store, receipt_id)?.unwrap();
let basis = load_document::<epiphany_core::EpiphanyReasoningBasis>(
&store,
&receipt.reasoning_basis_id,
)?
.unwrap();
assert_eq!(basis.source_documents, vec![admitted_source.clone()]);
}
let effects =
load_document::<PersonaInterpreterEffectDocument>(&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(())