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

Make semantic projection execution causal

Browse source
This commit is contained in:
Meta Vens 2026-07-15 05:48:32 +01:00
commit 327a92d78f
15 changed files with 1983 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

286
epiphany-core/src/agent_memory.rs
View file

@ -868,6 +868,89 @@ pub fn load_agent_memory_generation_witness(
cache.get(AGENT_MEMORY_GENERATION_WITNESS_LATEST_KEY)
}
pub fn agent_memory_semantic_projection_input(
store_path: impl AsRef<Path>,
) -> Result<crate::MemorySemanticProjectionInput> {
let store_path = store_path.as_ref();
let mut cache = agent_memory_cache(store_path)?;
cache.pull_all_backing_stores()?;
let identity = cache
.get::<AgentMemorySwarmIdentity>(AGENT_MEMORY_SWARM_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("Mind projection requires immutable swarm identity"))?;
let witness = cache
.get::<AgentMemoryGenerationWitness>(AGENT_MEMORY_GENERATION_WITNESS_LATEST_KEY)?
.ok_or_else(|| anyhow!("Mind projection requires admitted generation witness"))?;
let (generation, source_hash) = agent_memory_source_head(&mut cache)?;
if generation != witness.generation || source_hash != witness.source_hash {
return Err(anyhow!("Mind projection generation witness is not current"));
}
let mut entries = Vec::with_capacity(ROLE_TARGETS.len());
for (role_id, _, _) in ROLE_TARGETS {
entries.push(cache.get_required::<EpiphanyAgentMemoryEntry>(role_id)?);
}
let mut snapshot =
crate::memory_graph_from_agent_memories(format!("{}-mind", identity.swarm_id), &entries);
snapshot.model_revision = witness.generation;
let matches = cache
.get_all::<crate::MemorySemanticProjectionObligation>()?
.into_iter()
.filter(|obligation| obligation.source_commit_id == witness.witness_id)
.collect::<Vec<_>>();
if matches.len() != 1 {
return Err(anyhow!(
"Mind projection requires exactly one obligation for current generation"
));
}
let obligation = matches.into_iter().next().expect("one obligation");
let expected = crate::derive_memory_semantic_projection_obligation(
&snapshot,
&identity.swarm_id,
crate::SemanticPartition::Mind,
&format!("epiphany.agent-memory/{}/mind", identity.swarm_id),
&witness.witness_id,
&witness.committed_at,
)?;
if obligation != expected {
return Err(anyhow!(
"Mind projection obligation does not match canonical head"
));
}
let opening = cache.snapshot_envelopes();
let authority_envelopes = opening
.into_iter()
.filter(|envelope| {
(envelope.r#type == AGENT_MEMORY_SWARM_IDENTITY_TYPE
&& envelope.key == AGENT_MEMORY_SWARM_IDENTITY_KEY)
|| (envelope.r#type == AGENT_MEMORY_GENERATION_WITNESS_TYPE
&& envelope.key == AGENT_MEMORY_GENERATION_WITNESS_LATEST_KEY)
|| (envelope.r#type == AGENT_MEMORY_TYPE
&& ROLE_TARGETS
.iter()
.any(|(role_id, _, _)| envelope.key == *role_id))
})
.collect::<Vec<_>>();
if authority_envelopes.len() != ROLE_TARGETS.len() + 2 {
return Err(anyhow!("Mind projection authority snapshot is incomplete"));
}
Ok(crate::MemorySemanticProjectionInput {
snapshot,
authority: crate::memory_graph::MemorySemanticProjectionAuthoritySnapshot {
head: crate::MemorySemanticProjectionSourceHead {
swarm_id: obligation.swarm_id.clone(),
partition: obligation.partition.clone(),
canonical_source_id: obligation.canonical_source_id.clone(),
source_commit_id: obligation.source_commit_id.clone(),
graph_id: obligation.graph_id.clone(),
source_generation: obligation.source_generation,
source_model_hash: obligation.source_model_hash.clone(),
canonical_content_set_hash: obligation.canonical_content_set_hash.clone(),
},
envelopes: authority_envelopes,
},
obligation,
})
}
fn canonical_agent_memory_source_hash(
cache: &mut CultCache,
identity: &AgentMemorySwarmIdentity,
@ -1060,6 +1143,148 @@ fn commit_agent_memory_generation(
Ok(witness)
}
pub fn admit_legacy_agent_memory_generation(
store_path: impl AsRef<Path>,
) -> Result<AgentMemoryGenerationWitness> {
let store_path = store_path.as_ref();
let mut cache = agent_memory_cache(store_path)?;
cache.pull_all_backing_stores()?;
let identity = cache
.get::<AgentMemorySwarmIdentity>(AGENT_MEMORY_SWARM_IDENTITY_KEY)?
.ok_or_else(|| anyhow!("legacy Mind admission requires immutable swarm identity"))?;
let source_hash = canonical_agent_memory_source_hash(&mut cache, &identity, &BTreeMap::new())?;
if let Some(existing) =
cache.get::<AgentMemoryGenerationWitness>(AGENT_MEMORY_GENERATION_WITNESS_LATEST_KEY)?
{
if existing.schema_version != AGENT_MEMORY_GENERATION_WITNESS_SCHEMA_VERSION
|| existing.swarm_id != identity.swarm_id
|| existing.generation != 1
|| existing.previous_generation != 0
|| existing.source_hash != source_hash
|| existing.mutation_kind != "legacy-bootstrap-admission"
{
return Err(anyhow!(
"legacy Mind admission collides with live generation"
));
}
let obligations = cache
.get_all::<crate::MemorySemanticProjectionObligation>()?
.into_iter()
.filter(|obligation| obligation.source_commit_id == existing.witness_id)
.collect::<Vec<_>>();
if obligations.len() != 1 {
return Err(anyhow!(
"legacy Mind admission lost its exact semantic projection obligation"
));
}
return Ok(existing);
}
let mut entries = Vec::with_capacity(ROLE_TARGETS.len());
for (role_id, expected_agent_id, _) in ROLE_TARGETS {
let entry = cache
.get::<EpiphanyAgentMemoryEntry>(role_id)?
.ok_or_else(|| anyhow!("legacy Mind admission requires role {role_id:?}"))?;
let errors = validate_agent_entry(&entry, expected_agent_id);
if !errors.is_empty() {
return Err(anyhow!(
"legacy Mind role {role_id:?} is invalid: {}",
errors.join("; ")
));
}
entries.push(entry);
}
let committed_at = now_rfc3339();
let fingerprint = format!(
"{}|legacy-bootstrap-admission|{}",
identity.swarm_id, source_hash
);
let witness_id = format!(
"mind-generation-{}",
uuid::Uuid::new_v5(&uuid::Uuid::NAMESPACE_OID, fingerprint.as_bytes())
);
let receipt_id = format!(
"mind-admission-{}",
uuid::Uuid::new_v5(
&uuid::Uuid::NAMESPACE_OID,
format!("{fingerprint}|admit existing canonical Mind rows").as_bytes()
)
);
let receipt = AgentMemoryMindAdmissionReceipt {
schema_version: AGENT_MEMORY_MIND_ADMISSION_SCHEMA_VERSION.to_string(),
receipt_id: receipt_id.clone(),
swarm_id: identity.swarm_id.clone(),
role_id: "mind".to_string(),
mutation_kind: "legacy-bootstrap-admission".to_string(),
reason: "Admit existing canonical Mind rows without changing their content.".to_string(),
status: "admitted".to_string(),
resulting_source_hash: source_hash.clone(),
};
let witness = AgentMemoryGenerationWitness {
schema_version: AGENT_MEMORY_GENERATION_WITNESS_SCHEMA_VERSION.to_string(),
witness_id: witness_id.clone(),
swarm_id: identity.swarm_id.clone(),
generation: 1,
previous_generation: 0,
previous_source_hash: source_hash.clone(),
source_hash: source_hash.clone(),
authority_receipt_id: receipt_id.clone(),
mutation_kind: "legacy-bootstrap-admission".to_string(),
changed_role_ids: Vec::new(),
committed_at: committed_at.clone(),
};
let mut snapshot =
crate::memory_graph_from_agent_memories(format!("{}-mind", identity.swarm_id), &entries);
snapshot.model_revision = witness.generation;
let obligation = crate::derive_memory_semantic_projection_obligation(
&snapshot,
&identity.swarm_id,
crate::SemanticPartition::Mind,
&format!("epiphany.agent-memory/{}/mind", identity.swarm_id),
&witness_id,
&committed_at,
)?;
let opening = cache.snapshot_envelopes();
let mut expected = Vec::new();
let mut replacements = Vec::new();
for envelope in opening.iter().filter(|envelope| {
(envelope.r#type == AGENT_MEMORY_SWARM_IDENTITY_TYPE
&& envelope.key == AGENT_MEMORY_SWARM_IDENTITY_KEY)
|| (envelope.r#type == AGENT_MEMORY_TYPE
&& ROLE_TARGETS
.iter()
.any(|(role_id, _, _)| envelope.key == *role_id))
}) {
expected.push(envelope.clone());
replacements.push(envelope.clone());
}
if expected.len() != ROLE_TARGETS.len() + 1 {
return Err(anyhow!(
"legacy Mind admission could not authenticate every canonical source envelope"
));
}
replacements.push(
cache
.prepare_entry(AGENT_MEMORY_GENERATION_WITNESS_LATEST_KEY, &witness)?
.0,
);
replacements.push(cache.prepare_entry(&witness_id, &witness)?.0);
replacements.push(cache.prepare_entry(&receipt_id, &receipt)?.0);
replacements.push(
cache
.prepare_entry(&obligation.obligation_id, &obligation)?
.0,
);
let backing = SingleFileMessagePackBackingStore::new(store_path);
if !backing.compare_and_swap_batch(&expected, replacements)? {
return Err(anyhow!(
"legacy Mind admission lost exact atomic compare-and-swap"
));
}
Ok(witness)
}
fn agent_memory_source_head(cache: &mut CultCache) -> Result<(u64, String)> {
let identity = cache
.get::<AgentMemorySwarmIdentity>(AGENT_MEMORY_SWARM_IDENTITY_KEY)?
@ -2304,6 +2529,9 @@ fn agent_memory_cache(store_path: &Path) -> Result<CultCache> {
cache.register_entry_type::<AgentMemoryGenerationWitness>()?;
cache.register_entry_type::<AgentMemoryMindAdmissionReceipt>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionObligation>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionClaim>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionAttempt>()?;
cache.register_entry_type::<crate::MemorySemanticIndexReceipt>()?;
cache.register_entry_type::<EpiphanyAgentMemoryEntry>()?;
cache.register_entry_type::<EpiphanyAgentStateSoaEntry>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(store_path));
@ -2945,6 +3173,64 @@ mod tests {
use super::*;
use tempfile::tempdir;
#[test]
fn legacy_mind_admission_preserves_rows_and_adds_exact_generation_pressure() -> Result<()> {
let temp = tempdir()?;
let agent_dir = temp.path().join("agents-legacy-admission");
fs::create_dir_all(&agent_dir)?;
for (role_id, agent_id, filename) in ROLE_TARGETS {
fs::write(
agent_dir.join(filename),
sample_agent_json(agent_id, &format!("Agent {role_id}")),
)?;
}
let store = temp.path().join("legacy-admission.msgpack");
migrate_agent_memory_json_dir_to_cultcache(&agent_dir, &store)?;
ensure_agent_memory_swarm_identity(&store, "legacy-admission-swarm")?;
let mut before_cache = agent_memory_cache(&store)?;
before_cache.pull_all_backing_stores()?;
let before = before_cache
.snapshot_envelopes()
.into_iter()
.filter(|envelope| envelope.r#type == AGENT_MEMORY_TYPE)
.collect::<Vec<_>>();
let witness = admit_legacy_agent_memory_generation(&store)?;
assert_eq!(witness.generation, 1);
assert_eq!(witness.previous_generation, 0);
assert_eq!(witness.mutation_kind, "legacy-bootstrap-admission");
assert!(witness.changed_role_ids.is_empty());
assert_eq!(admit_legacy_agent_memory_generation(&store)?, witness);
let mut cache = agent_memory_cache(&store)?;
cache.pull_all_backing_stores()?;
let after = cache
.snapshot_envelopes()
.into_iter()
.filter(|envelope| envelope.r#type == AGENT_MEMORY_TYPE)
.collect::<Vec<_>>();
assert_eq!(after, before);
let receipt = cache
.get::<AgentMemoryMindAdmissionReceipt>(&witness.authority_receipt_id)?
.expect("legacy admission receipt");
assert_eq!(receipt.resulting_source_hash, witness.source_hash);
let obligations = cache
.get_all::<crate::MemorySemanticProjectionObligation>()?
.into_iter()
.filter(|obligation| obligation.source_commit_id == witness.witness_id)
.collect::<Vec<_>>();
assert_eq!(obligations.len(), 1);
assert_eq!(obligations[0].partition, "mind");
assert_eq!(obligations[0].source_generation, 1);
assert!(!obligations[0].source_model_hash.is_empty());
let input = agent_memory_semantic_projection_input(&store)?;
assert_eq!(input.obligation, obligations[0]);
assert_eq!(input.snapshot.model_revision, 1);
assert_eq!(input.authority.envelopes.len(), ROLE_TARGETS.len() + 2);
assert!(repair_agent_memory_store(&store).is_err());
Ok(())
}
#[test]
fn concurrent_mind_commits_have_one_generation_and_no_partial_companions() -> Result<()> {
let temp = tempdir()?;

7
epiphany-core/src/bin/epiphany-agent-memory-store.rs
View file

@ -2,6 +2,7 @@ use anyhow::Context;
use anyhow::Result;
use epiphany_core::AgentMemoryLifecycleOperation;
use epiphany_core::AgentSelfPatch;
use epiphany_core::admit_legacy_agent_memory_generation;
use epiphany_core::agent_memory_status;
use epiphany_core::apply_agent_memory_lifecycle_operation;
use epiphany_core::apply_agent_self_patch_document;
@ -53,6 +54,10 @@ fn main() -> Result<()> {
let swarm_id = require_string_arg(&mut args, "--swarm-id")?;
print_json(&ensure_agent_memory_swarm_identity(store, &swarm_id)?)?;
}
"admit-legacy-mind" => {
let store = require_path_arg(&mut args, "--store")?;
print_json(&admit_legacy_agent_memory_generation(store)?)?;
}
"refresh-soa" => {
let store = require_path_arg(&mut args, "--store")?;
print_json(&refresh_agent_state_soa(store)?)?;
@ -198,7 +203,7 @@ fn print_json<T: serde::Serialize>(value: &T) -> Result<()> {
fn print_usage() {
eprintln!(
"usage: epiphany-agent-memory-store <migrate-json-dir|project-json-dir|project-persona|set-swarm-identity|status|refresh-soa|repair|validate|review-patch|apply-patch|review-lifecycle|apply-lifecycle|smoke> ..."
"usage: epiphany-agent-memory-store <migrate-json-dir|project-json-dir|project-persona|set-swarm-identity|admit-legacy-mind|status|refresh-soa|repair|validate|review-patch|apply-patch|review-lifecycle|apply-lifecycle|smoke> ..."
);
}

158
epiphany-core/src/bin/epiphany-memory-semantic.rs
View file

@ -1,10 +1,9 @@
use anyhow::{Context, Result, anyhow};
use chrono::{SecondsFormat, Utc};
use epiphany_core::{
EpiphanyMemoryContextQuery, EpiphanyMemoryGraphSnapshot, MemorySemanticIndexConfig,
SemanticPartition, agent_memory_role_ids, index_memory_semantic_partition,
load_agent_memory_entry_for_role, load_memory_graph_snapshot, memory_graph_from_agent_memories,
persist_memory_semantic_index_receipt, runtime_current_repo_model, runtime_swarm_binding,
MemorySemanticProjectionInput, SemanticPartition, agent_memory_semantic_projection_input,
execute_memory_semantic_projection, load_memory_graph_snapshot,
load_memory_semantic_projection_readiness, runtime_modeling_semantic_projection_input,
semantic_memory_context,
};
use std::env;
@ -14,46 +13,31 @@ fn main() -> Result<()> {
let mut args = env::args().skip(1);
let command = args.next().ok_or_else(|| usage_error("missing command"))?;
let options = Options::parse(args)?;
let (snapshot, swarm_id) = options.load_source()?;
let config = MemorySemanticIndexConfig::from_env();
match command.as_str() {
"index" => {
let receipt_store = options
.receipt_store
.as_ref()
.ok_or_else(|| usage_error("index requires --receipt-store <path>"))?;
let receipt = index_memory_semantic_partition(
&snapshot,
&swarm_id,
options.partition,
&Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
&config,
)?;
persist_memory_semantic_index_receipt(receipt_store, &receipt)?;
print_json(&serde_json::json!({
"status": receipt.status,
"receiptId": receipt.receipt_id,
"receiptStore": receipt_store,
"swarmId": receipt.swarm_id,
"partition": receipt.partition,
"collectionName": receipt.collection_name,
"graphId": receipt.graph_id,
"modelRevision": receipt.model_revision,
"modelHash": receipt.model_hash,
"embeddingProviderId": receipt.embedding_provider_id,
"embeddingModel": receipt.embedding_model,
"vectorDimensions": receipt.vector_dimensions,
"indexedDocumentCount": receipt.indexed_document_count,
"deletedDocumentCount": receipt.deleted_document_count,
"canonicalContentSetHash": receipt.canonical_content_set_hash,
"indexedAt": receipt.indexed_at,
}))?;
let (input, source_store) = options.load_projection_input()?;
let executor_id = options
.executor_id
.as_deref()
.ok_or_else(|| usage_error("index requires --executor-id <id>"))?;
let receipt =
execute_memory_semantic_projection(source_store, &input, executor_id, &config)?;
print_receipt(&receipt, source_store)?;
}
"context" => {
let (snapshot, swarm_id) = options.load_source()?;
let text = options
.text
.clone()
.ok_or_else(|| usage_error("context requires --text <query>"))?;
let projection_input = options.load_projection_input().ok().map(|(input, _)| input);
let readiness = match (&projection_input, options.source_store()) {
(Some(input), Some(store)) => {
load_memory_semantic_projection_readiness(store, input)?
}
_ => None,
};
let packet = semantic_memory_context(
&snapshot,
&swarm_id,
@ -68,6 +52,7 @@ fn main() -> Result<()> {
budget: options.budget,
..Default::default()
},
readiness.as_ref(),
&config,
);
print_json(&packet)?;
@ -81,7 +66,7 @@ struct Options {
graph_store: Option<PathBuf>,
runtime_store: Option<PathBuf>,
agent_store: Option<PathBuf>,
receipt_store: Option<PathBuf>,
executor_id: Option<String>,
swarm_id: Option<String>,
partition: SemanticPartition,
text: Option<String>,
@ -96,7 +81,7 @@ impl Options {
graph_store: None,
runtime_store: None,
agent_store: None,
receipt_store: None,
executor_id: None,
swarm_id: None,
partition: SemanticPartition::Modeling,
text: None,
@ -114,7 +99,7 @@ impl Options {
"--graph-store" => options.graph_store = Some(PathBuf::from(value()?)),
"--runtime-store" => options.runtime_store = Some(PathBuf::from(value()?)),
"--agent-store" => options.agent_store = Some(PathBuf::from(value()?)),
"--receipt-store" => options.receipt_store = Some(PathBuf::from(value()?)),
"--executor-id" => options.executor_id = Some(value()?),
"--swarm-id" => options.swarm_id = Some(value()?),
"--partition" => options.partition = parse_partition(&value()?)?,
"--text" => options.text = Some(value()?),
@ -161,47 +146,11 @@ impl Options {
}
fn load_source(&self) -> Result<(EpiphanyMemoryGraphSnapshot, String)> {
if let Some(path) = &self.runtime_store {
let binding = runtime_swarm_binding(path)?
.ok_or_else(|| anyhow!("runtime store has no immutable swarm binding"))?;
if self
.swarm_id
.as_ref()
.is_some_and(|claimed| claimed != &binding.swarm_id)
{
return Err(anyhow!("--swarm-id does not match runtime swarm binding"));
}
let snapshot = runtime_current_repo_model(path)?
.ok_or_else(|| anyhow!("runtime store has no admitted RepoModel"))?;
return Ok((snapshot, binding.swarm_id));
}
if let Some(path) = &self.agent_store {
let mut entries = Vec::new();
for role in agent_memory_role_ids() {
if let Some(entry) = load_agent_memory_entry_for_role(path, role)? {
entries.push(entry);
}
}
if entries.is_empty() {
return Err(anyhow!("agent store has no admitted organ memory"));
}
let identity = epiphany_core::load_agent_memory_swarm_identity(path)?
.ok_or_else(|| anyhow!("agent store has no immutable swarm identity; run epiphany-agent-memory-store set-swarm-identity"))?;
if self
.swarm_id
.as_ref()
.is_some_and(|claimed| claimed != &identity.swarm_id)
{
return Err(anyhow!(
"--swarm-id {:?} does not match the canonical agent-store swarm identity {:?}",
self.swarm_id,
identity.swarm_id
));
}
let swarm_id = identity.swarm_id;
if self.runtime_store.is_some() || self.agent_store.is_some() {
let (input, _) = self.load_projection_input()?;
return Ok((
memory_graph_from_agent_memories(format!("{swarm_id}-mind"), &entries),
swarm_id,
input.snapshot().clone(),
input.obligation().swarm_id.clone(),
));
}
let path = self.graph_store.as_ref().expect("validated graph store");
@ -209,6 +158,32 @@ impl Options {
.ok_or_else(|| anyhow!("memory graph store {} is missing", path.display()))?;
Ok((snapshot, self.swarm_id.clone().expect("validated swarm id")))
}
fn load_projection_input(&self) -> Result<(MemorySemanticProjectionInput, &PathBuf)> {
let (input, store) = if let Some(path) = &self.runtime_store {
(runtime_modeling_semantic_projection_input(path)?, path)
} else if let Some(path) = &self.agent_store {
(agent_memory_semantic_projection_input(path)?, path)
} else {
return Err(anyhow!(
"graph-store snapshots have no canonical admission authority and are BM25-only"
));
};
if self
.swarm_id
.as_ref()
.is_some_and(|claimed| claimed != &input.obligation().swarm_id)
{
return Err(anyhow!(
"--swarm-id does not match canonical source identity"
));
}
Ok((input, store))
}
fn source_store(&self) -> Option<&PathBuf> {
self.runtime_store.as_ref().or(self.agent_store.as_ref())
}
}
fn parse_partition(value: &str) -> Result<SemanticPartition> {
@ -226,7 +201,7 @@ fn parse_profile(value: &str) -> Result<epiphany_core::EpiphanyMemoryProfile> {
fn usage_error(message: &str) -> anyhow::Error {
anyhow!(
"{message}\nusage: epiphany-memory-semantic <index|context> (--runtime-store <path>|--agent-store <path>|--graph-store <path>) [--swarm-id <id>] --partition <mind|modeling> [--receipt-store <path>] [--text <query>] [--query-id <id>] [--budget <n>] [--profile <profile>]"
"{message}\nusage: epiphany-memory-semantic <index|context> (--runtime-store <path>|--agent-store <path>|--graph-store <path>) [--swarm-id <id>] --partition <mind|modeling> [--executor-id <id>] [--text <query>] [--query-id <id>] [--budget <n>] [--profile <profile>]"
)
}
@ -234,3 +209,28 @@ fn print_json(value: &impl serde::Serialize) -> Result<()> {
println!("{}", serde_json::to_string_pretty(value)?);
Ok(())
}
fn print_receipt(
receipt: &epiphany_core::MemorySemanticIndexReceipt,
source_store: &PathBuf,
) -> Result<()> {
print_json(&serde_json::json!({
"status": receipt.status,
"receiptId": receipt.receipt_id,
"sourceStore": source_store,
"obligationId": receipt.obligation_id,
"swarmId": receipt.swarm_id,
"partition": receipt.partition,
"collectionName": receipt.collection_name,
"graphId": receipt.graph_id,
"modelRevision": receipt.model_revision,
"modelHash": receipt.model_hash,
"embeddingProviderId": receipt.embedding_provider_id,
"embeddingModel": receipt.embedding_model,
"vectorDimensions": receipt.vector_dimensions,
"indexedDocumentCount": receipt.indexed_document_count,
"deletedDocumentCount": receipt.deleted_document_count,
"canonicalContentSetHash": receipt.canonical_content_set_hash,
"indexedAt": receipt.indexed_at,
}))
}

12
epiphany-core/src/heartbeat_state.rs
View file

@ -1165,7 +1165,11 @@ fn persona_memory_recall_for_scheduled_turn(
}
}
let swarm_id = swarm_identity.swarm_id;
let graph = crate::memory_graph_from_agent_memories(&format!("{swarm_id}-mind"), &entries);
let projection_input = crate::agent_memory_semantic_projection_input(agent_store).ok();
let graph = projection_input.as_ref().map_or_else(
|| crate::memory_graph_from_agent_memories(&format!("{swarm_id}-mind"), &entries),
|input| input.snapshot.clone(),
);
let persona_domain_id =
crate::memory_graph_domain_id(crate::EpiphanyMemoryProfile::RoleSelf, "role", "Persona");
let query_document = EpiphanyMemoryContextQuery {
@ -1178,11 +1182,17 @@ fn persona_memory_recall_for_scheduled_turn(
budget: Some(8),
};
let config = memory_semantic_config_for_heartbeat();
let readiness = projection_input.as_ref().and_then(|input| {
crate::load_memory_semantic_projection_readiness(agent_store, input)
.ok()
.flatten()
});
let packet = crate::semantic_memory_context(
&graph,
&swarm_id,
crate::SemanticPartition::Mind,
&query_document,
readiness.as_ref(),
&config,
);
let fallback = packet

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

@ -105,7 +105,9 @@ pub use agent_memory::GhostlightTraitVector;
pub use agent_memory::GhostlightValue;
pub use agent_memory::GhostlightWorld;
pub use agent_memory::PERSONA_STATE_SCHEMA_VERSION;
pub use agent_memory::admit_legacy_agent_memory_generation;
pub use agent_memory::agent_memory_role_ids;
pub use agent_memory::agent_memory_semantic_projection_input;
pub use agent_memory::agent_memory_status;
pub use agent_memory::apply_agent_canonical_trait_seeds;
pub use agent_memory::apply_agent_memory_lifecycle_operation;
@ -630,13 +632,17 @@ pub use memory_graph::MEMORY_GRAPH_SCHEMA_VERSION;
pub use memory_graph::MEMORY_GRAPH_TYPE;
pub use memory_graph::MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION;
pub use memory_graph::MEMORY_SEMANTIC_PROJECTION_ATTEMPT_SCHEMA_VERSION;
pub use memory_graph::MEMORY_SEMANTIC_PROJECTION_CLAIM_SCHEMA_VERSION;
pub use memory_graph::MEMORY_SEMANTIC_PROJECTION_OBLIGATION_SCHEMA_VERSION;
pub use memory_graph::MemorySemanticIndexConfig;
pub use memory_graph::MemorySemanticIndexReceipt;
pub use memory_graph::MemorySemanticProjectionAttempt;
pub use memory_graph::MemorySemanticProjectionClaim;
pub use memory_graph::MemorySemanticProjectionHealth;
pub use memory_graph::MemorySemanticProjectionHealthStatus;
pub use memory_graph::MemorySemanticProjectionInput;
pub use memory_graph::MemorySemanticProjectionObligation;
pub use memory_graph::MemorySemanticProjectionReadiness;
pub use memory_graph::MemorySemanticProjectionSourceHead;
pub use memory_graph::RepoFrontierAdoptedPlan;
pub use memory_graph::RepoFrontierItem;
@ -662,10 +668,12 @@ pub use memory_graph::derive_memory_semantic_projection_health;
pub use memory_graph::derive_memory_semantic_projection_obligation;
pub use memory_graph::derive_repo_model_patch;
pub use memory_graph::derive_semantic_projection;
pub use memory_graph::index_memory_semantic_partition;
pub use memory_graph::execute_memory_semantic_projection;
pub use memory_graph::lifecycle_allowed_for_profile;
pub use memory_graph::load_memory_graph_entry;
pub use memory_graph::load_memory_graph_snapshot;
pub use memory_graph::load_memory_semantic_projection_readiness;
pub use memory_graph::load_memory_semantic_projection_success;
pub use memory_graph::memory_graph_cache;
pub use memory_graph::memory_graph_domain_id;
pub use memory_graph::memory_graph_edge_id;
@ -675,7 +683,6 @@ pub use memory_graph::memory_graph_from_heartbeat_cognition;
pub use memory_graph::memory_graph_model_hash;
pub use memory_graph::memory_graph_node_id;
pub use memory_graph::memory_semantic_projection_query_eligible;
pub use memory_graph::persist_memory_semantic_index_receipt;
pub use memory_graph::plan_memory_graph_context_cut;
pub use memory_graph::plan_memory_graph_context_cut_with_ranked_ids;
pub use memory_graph::refresh_or_validate_repo_memory_graph;
@ -873,6 +880,7 @@ pub use runtime_spine::runtime_job_snapshot;
pub use runtime_spine::runtime_latest_hands_receipt_chain_after;
pub use runtime_spine::runtime_mind_gateway_review;
pub use runtime_spine::runtime_mind_state_commit_receipt;
pub use runtime_spine::runtime_modeling_semantic_projection_input;
pub use runtime_spine::runtime_registered_document_types;
pub use runtime_spine::runtime_reorient_worker_result;
pub use runtime_spine::runtime_repo_frontier_proposal_modeling_request;

2
epiphany-core/src/memory_graph/mod.rs
View file

@ -5,6 +5,7 @@ pub mod ids;
pub mod profiles;
pub mod semantic_index;
pub mod semantic_projection;
pub mod semantic_projector;
pub mod store;
pub mod validation;
@ -48,6 +49,7 @@ pub use profiles::memory_graph_from_heartbeat_cognition;
pub use profiles::refresh_or_validate_repo_memory_graph;
pub use semantic_index::*;
pub use semantic_projection::*;
pub use semantic_projector::*;
pub use store::EpiphanyMemoryGraphEntry;
pub use store::MEMORY_GRAPH_KEY;
pub use store::MEMORY_GRAPH_SCHEMA_VERSION;

368
epiphany-core/src/memory_graph/semantic_index.rs
View file

@ -9,12 +9,11 @@ use crate::semantic_backend::{
SemanticPoint,
};
use anyhow::{Result, anyhow};
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use cultcache_rs::DatabaseEntry;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashSet;
use std::collections::{BTreeMap, HashSet};
use std::env;
use std::path::Path;
pub const MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION: &str =
"gamecult.epiphany.memory_semantic_index_receipt.v0";
@ -191,6 +190,13 @@ pub struct MemorySemanticProjectionHealth {
pub query_eligible: bool,
}
#[derive(Clone, Debug)]
pub struct MemorySemanticProjectionReadiness {
pub(crate) obligation: MemorySemanticProjectionObligation,
pub(crate) current: MemorySemanticProjectionSourceHead,
pub(crate) receipt: MemorySemanticIndexReceipt,
}
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "gamecult.epiphany.memory_semantic_index_receipt",
@ -336,6 +342,11 @@ pub fn validate_memory_semantic_projection_attempt(
if attempt.attempt_id.trim().is_empty()
|| attempt.obligation_id.trim().is_empty()
|| attempt.started_at.trim().is_empty()
|| chrono::DateTime::parse_from_rfc3339(&attempt.started_at).is_err()
|| attempt
.completed_at
.as_deref()
.is_some_and(|completed_at| chrono::DateTime::parse_from_rfc3339(completed_at).is_err())
{
return Err(anyhow!(
"semantic projection attempt is missing identity or time"
@ -344,6 +355,7 @@ pub fn validate_memory_semantic_projection_attempt(
match attempt.status.as_str() {
"running" if attempt.completed_at.is_none() && attempt.error.is_none() => Ok(()),
"failed" if attempt.completed_at.is_some() && attempt.error.is_some() => Ok(()),
"succeeded" if attempt.completed_at.is_some() && attempt.error.is_none() => Ok(()),
_ => Err(anyhow!(
"semantic projection attempt status does not match its terminal fields"
)),
@ -368,26 +380,40 @@ pub fn derive_memory_semantic_projection_health(
.iter()
.filter(|receipt| receipt_matches_obligation(receipt, obligation))
.collect::<Vec<_>>();
matching_receipts.sort_by(|left, right| left.indexed_at.cmp(&right.indexed_at));
matching_receipts.sort_by_key(|receipt| {
chrono::DateTime::parse_from_rfc3339(&receipt.indexed_at)
.expect("matching receipt has valid RFC3339 time")
});
let receipt = matching_receipts.last().copied();
let mut matching_attempts = attempts
.iter()
.filter(|attempt| attempt.obligation_id == obligation.obligation_id)
.collect::<Vec<_>>();
matching_attempts.sort_by(|left, right| {
left.completed_at
.as_deref()
.unwrap_or(left.started_at.as_str())
.cmp(
right
.completed_at
.as_deref()
.unwrap_or(right.started_at.as_str()),
)
matching_attempts.sort_by_key(|attempt| {
chrono::DateTime::parse_from_rfc3339(
attempt
.completed_at
.as_deref()
.unwrap_or(attempt.started_at.as_str()),
)
.expect("validated attempt has RFC3339 ordering time")
});
let latest_attempt = matching_attempts.last().copied();
let repair_after_receipt = latest_attempt
.zip(receipt)
.is_some_and(|(attempt, receipt)| {
chrono::DateTime::parse_from_rfc3339(&attempt.started_at).ok()
> chrono::DateTime::parse_from_rfc3339(&receipt.indexed_at).ok()
&& attempt.status != "succeeded"
});
let status = if stale {
MemorySemanticProjectionHealthStatus::Stale
} else if repair_after_receipt
&& latest_attempt.is_some_and(|attempt| attempt.status == "failed")
{
MemorySemanticProjectionHealthStatus::Failed
} else if repair_after_receipt {
MemorySemanticProjectionHealthStatus::Pending
} else if receipt.is_some() {
MemorySemanticProjectionHealthStatus::Ready
} else if latest_attempt.is_some_and(|attempt| attempt.status == "failed") {
@ -471,9 +497,10 @@ fn receipt_matches_obligation(
&& receipt.model_hash == obligation.source_model_hash
&& receipt.canonical_content_set_hash == obligation.canonical_content_set_hash
&& receipt.projection_schema_version == obligation.projection_schema_version
&& chrono::DateTime::parse_from_rfc3339(&receipt.indexed_at).is_ok()
}
pub fn index_memory_semantic_partition(
pub(super) fn index_memory_semantic_partition(
snapshot: &EpiphanyMemoryGraphSnapshot,
swarm_id: &str,
partition: SemanticPartition,
@ -484,11 +511,53 @@ pub fn index_memory_semantic_partition(
.into_iter()
.filter(|document| document.partition == partition)
.collect::<Vec<_>>();
let backend = qdrant(config)?;
let collection = config.collection(partition);
let scope = [
("swarmId", swarm_id),
("partition", partition_name(partition)),
];
let content_set_hash = canonical_content_set_hash(&documents);
let model_hash = crate::memory_graph_model_hash(snapshot)?;
if documents.is_empty() {
return Err(anyhow!(
"semantic projection partition has no live documents"
let (vector_size, deleted_document_count) = if backend.collection_exists(collection)? {
let actual = backend.collection_compatibility(collection)?;
let expected = compatibility(config, partition, actual.vector_size);
if actual != expected {
return Err(anyhow!(
"Qdrant collection {collection} is incompatible: actual {actual:?}, expected {expected:?}"
));
}
let existing_ids = backend.point_ids_for_scope(collection, &scope)?;
backend.delete_points(collection, &existing_ids)?;
let observed_ids = backend.point_ids_for_scope(collection, &scope)?;
if !observed_ids.is_empty() {
return Err(anyhow!(
"Qdrant scope synchronization failed: expected no points, observed {observed_ids:?}"
));
}
(actual.vector_size as u32, existing_ids.len() as u32)
} else {
// An empty projection has nothing to embed and does not justify creating
// physical projection state. Absence already represents the empty set.
(0, 0)
};
return Ok(memory_semantic_index_receipt(
snapshot,
swarm_id,
partition,
indexed_at,
config,
collection,
&model_hash,
&content_set_hash,
vector_size,
0,
deleted_document_count,
));
}
let embedder = embedder(config, partition)?;
let texts = documents
.iter()
@ -499,8 +568,6 @@ pub fn index_memory_semantic_partition(
.first()
.map(Vec::len)
.ok_or_else(|| anyhow!("semantic projection produced no embeddings"))?;
let backend = qdrant(config)?;
let collection = config.collection(partition);
let compatibility = compatibility(config, partition, vector_size);
if backend.collection_exists(collection)? {
let actual = backend.collection_compatibility(collection)?;
@ -512,13 +579,7 @@ pub fn index_memory_semantic_partition(
} else {
backend.create_collection(collection, &compatibility)?;
}
let existing_ids = backend.point_ids_for_scope(
collection,
&[
("swarmId", swarm_id),
("partition", partition_name(partition)),
],
)?;
let existing_ids = backend.point_ids_for_scope(collection, &scope)?;
let points = documents
.iter()
.zip(embeddings)
@ -538,7 +599,65 @@ pub fn index_memory_semantic_partition(
.filter(|id| !live_ids.contains(id.as_str()))
.collect::<Vec<_>>();
backend.delete_points(collection, &deleted_ids)?;
let content_set_hash = canonical_content_set_hash(&documents);
let observed_ids = backend.point_ids_for_scope(collection, &scope)?;
let desired_ids = documents
.iter()
.map(|document| document.point_id.clone())
.collect::<HashSet<_>>();
let observed_ids = observed_ids.into_iter().collect::<HashSet<_>>();
if observed_ids != desired_ids {
return Err(anyhow!(
"Qdrant scope synchronization failed: observed IDs do not equal the desired projection"
));
}
let observed_payloads = backend
.points_for_scope::<MemorySemanticPointPayload>(collection, &scope)?
.into_iter()
.map(|point| {
point
.payload
.map(|payload| (point.id, payload))
.ok_or_else(|| anyhow!("Qdrant scoped point omitted typed projection payload"))
})
.collect::<Result<BTreeMap<_, _>>>()?;
let desired_payloads = documents
.iter()
.map(|document| (document.point_id.clone(), point_payload(document, config)))
.collect::<BTreeMap<_, _>>();
if observed_payloads != desired_payloads {
return Err(anyhow!(
"Qdrant scope synchronization failed: observed payload identities do not equal the desired projection"
));
}
Ok(memory_semantic_index_receipt(
snapshot,
swarm_id,
partition,
indexed_at,
config,
collection,
&model_hash,
&content_set_hash,
vector_size as u32,
documents.len() as u32,
deleted_ids.len() as u32,
))
}
#[allow(clippy::too_many_arguments)]
fn memory_semantic_index_receipt(
snapshot: &EpiphanyMemoryGraphSnapshot,
swarm_id: &str,
partition: SemanticPartition,
indexed_at: &str,
config: &MemorySemanticIndexConfig,
collection: &str,
model_hash: &str,
content_set_hash: &str,
vector_dimensions: u32,
indexed_document_count: u32,
deleted_document_count: u32,
) -> MemorySemanticIndexReceipt {
let receipt_id = format!(
"memory-semantic-index-{}-{}-{}-{}",
partition_name(partition),
@ -546,7 +665,7 @@ pub fn index_memory_semantic_partition(
&content_set_hash[..16],
&format!("{:x}", Sha256::digest(indexed_at.as_bytes()))[..12]
);
Ok(MemorySemanticIndexReceipt {
MemorySemanticIndexReceipt {
schema_version: MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION.to_string(),
receipt_id,
swarm_id: swarm_id.to_string(),
@ -554,13 +673,13 @@ pub fn index_memory_semantic_partition(
collection_name: collection.to_string(),
graph_id: snapshot.graph_id.clone(),
model_revision: snapshot.model_revision,
model_hash: documents[0].model_hash.clone(),
model_hash: model_hash.to_string(),
embedding_provider_id: config.embedding_provider_id.clone(),
embedding_model: config.ollama_model.clone(),
vector_dimensions: vector_size as u32,
indexed_document_count: documents.len() as u32,
deleted_document_count: deleted_ids.len() as u32,
canonical_content_set_hash: content_set_hash,
vector_dimensions,
indexed_document_count,
deleted_document_count,
canonical_content_set_hash: content_set_hash.to_string(),
indexed_at: indexed_at.to_string(),
status: "ready".to_string(),
obligation_id: String::new(),
@ -568,19 +687,7 @@ pub fn index_memory_semantic_partition(
source_commit_id: String::new(),
source_generation: snapshot.model_revision,
projection_schema_version: SEMANTIC_PROJECTION_SCHEMA_VERSION.to_string(),
})
}
pub fn persist_memory_semantic_index_receipt(
store_path: impl AsRef<Path>,
receipt: &MemorySemanticIndexReceipt,
) -> Result<()> {
let mut cache = CultCache::new();
cache.register_entry_type::<MemorySemanticIndexReceipt>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(store_path.as_ref()));
cache.pull_all_backing_stores()?;
cache.put(&receipt.receipt_id, receipt)?;
Ok(())
}
}
pub fn semantic_memory_context(
@ -588,8 +695,41 @@ pub fn semantic_memory_context(
swarm_id: &str,
partition: SemanticPartition,
query: &EpiphanyMemoryContextQuery,
readiness: Option<&MemorySemanticProjectionReadiness>,
config: &MemorySemanticIndexConfig,
) -> EpiphanyMemoryContextPacket {
let eligible = readiness.is_some_and(|readiness| {
let source_matches_query = (|| -> Result<bool> {
let documents = derive_semantic_projection(swarm_id, snapshot)?
.into_iter()
.filter(|document| document.partition == partition)
.collect::<Vec<_>>();
Ok(readiness.obligation.swarm_id == swarm_id
&& readiness.obligation.partition == partition_name(partition)
&& readiness.obligation.graph_id == snapshot.graph_id
&& readiness.obligation.source_generation == snapshot.model_revision
&& readiness.obligation.source_model_hash
== crate::memory_graph_model_hash(snapshot)?
&& readiness.obligation.canonical_content_set_hash
== canonical_content_set_hash(&documents))
})()
.unwrap_or(false);
source_matches_query
&& memory_semantic_projection_query_eligible(
&readiness.obligation,
&readiness.current,
&readiness.receipt,
)
});
if !eligible {
let mut packet =
plan_memory_graph_context_cut_for_partition(snapshot, query, &[], partition);
packet.warnings.push(
"semantic projection unavailable; used canonical BM25 fallback: newest canonical obligation has no exact success receipt"
.to_string(),
);
return packet;
}
match try_semantic_memory_context(snapshot, swarm_id, partition, query, config) {
Ok(packet) => packet,
Err(error) => {
@ -775,6 +915,9 @@ mod tests {
EpiphanyMemoryDomain, EpiphanyMemoryLifecycle, EpiphanyMemoryNode, EpiphanyMemoryNodeKind,
EpiphanyMemoryProfile, MEMORY_GRAPH_SCHEMA_VERSION, memory_graph_model_hash,
};
use tokio::runtime::Runtime;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
fn snapshot() -> EpiphanyMemoryGraphSnapshot {
let mut snapshot = EpiphanyMemoryGraphSnapshot {
@ -809,6 +952,50 @@ mod tests {
snapshot
}
#[test]
fn empty_partition_bypasses_ollama_and_does_not_create_collection() -> Result<()> {
let runtime = Runtime::new()?;
let qdrant = runtime.block_on(MockServer::start());
runtime.block_on(
Mock::given(method("GET"))
.and(path("/collections/empty_modeling/exists"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"result": { "exists": false },
"status": "ok",
"time": 0.0
})))
.expect(1)
.mount(&qdrant),
);
let mut empty = EpiphanyMemoryGraphSnapshot {
schema_version: Some(MEMORY_GRAPH_SCHEMA_VERSION.to_string()),
graph_id: "empty-semantic-source".to_string(),
model_revision: 9,
..Default::default()
};
empty.model_hash = memory_graph_model_hash(&empty)?;
let mut config = MemorySemanticIndexConfig::from_env();
config.qdrant_url = qdrant.uri();
config.modeling_collection = "empty_modeling".to_string();
config.ollama_base_url = "http://127.0.0.1:1".to_string();
config.ollama_timeout_ms = 5;
let receipt = index_memory_semantic_partition(
&empty,
"swarm-empty",
SemanticPartition::Modeling,
"2026-07-15T12:00:00Z",
&config,
)?;
assert_eq!(receipt.indexed_document_count, 0);
assert_eq!(receipt.deleted_document_count, 0);
assert_eq!(receipt.vector_dimensions, 0);
assert_eq!(receipt.model_hash, empty.model_hash);
assert_eq!(receipt.status, "ready");
Ok(())
}
#[test]
fn projection_obligation_identity_is_commit_owned_and_deterministic() -> Result<()> {
let snapshot = snapshot();
@ -866,6 +1053,7 @@ mod tests {
"swarm",
SemanticPartition::Modeling,
&query,
None,
&config,
);
assert_eq!(actual.nodes, expected.nodes);
@ -879,6 +1067,77 @@ mod tests {
);
}
#[test]
fn exact_readiness_for_one_source_cannot_open_another_query_scope() -> Result<()> {
let snapshot = snapshot();
let obligation = derive_memory_semantic_projection_obligation(
&snapshot,
"swarm-a",
SemanticPartition::Modeling,
"runtime/repo-model",
"admission-current",
"2026-07-15T10:00:00Z",
)?;
let current = MemorySemanticProjectionSourceHead {
swarm_id: obligation.swarm_id.clone(),
partition: obligation.partition.clone(),
canonical_source_id: obligation.canonical_source_id.clone(),
source_commit_id: obligation.source_commit_id.clone(),
graph_id: obligation.graph_id.clone(),
source_generation: obligation.source_generation,
source_model_hash: obligation.source_model_hash.clone(),
canonical_content_set_hash: obligation.canonical_content_set_hash.clone(),
};
let receipt = MemorySemanticIndexReceipt {
schema_version: MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION.to_string(),
receipt_id: "ready-current".to_string(),
swarm_id: obligation.swarm_id.clone(),
partition: obligation.partition.clone(),
collection_name: "modeling".to_string(),
graph_id: obligation.graph_id.clone(),
model_revision: obligation.source_generation,
model_hash: obligation.source_model_hash.clone(),
embedding_provider_id: "provider".to_string(),
embedding_model: "model".to_string(),
vector_dimensions: 3,
indexed_document_count: 3,
deleted_document_count: 0,
canonical_content_set_hash: obligation.canonical_content_set_hash.clone(),
indexed_at: "2026-07-15T10:01:00Z".to_string(),
status: "ready".to_string(),
obligation_id: obligation.obligation_id.clone(),
canonical_source_id: obligation.canonical_source_id.clone(),
source_commit_id: obligation.source_commit_id.clone(),
source_generation: obligation.source_generation,
projection_schema_version: obligation.projection_schema_version.clone(),
};
let query = EpiphanyMemoryContextQuery {
id: "swapped-source".to_string(),
text: Some("authority".to_string()),
..Default::default()
};
let mut config = MemorySemanticIndexConfig::from_env();
config.qdrant_url = "http://127.0.0.1:1".to_string();
config.ollama_base_url = "http://127.0.0.1:1".to_string();
let readiness = MemorySemanticProjectionReadiness {
obligation: obligation.clone(),
current: current.clone(),
receipt: receipt.clone(),
};
let packet = semantic_memory_context(
&snapshot,
"swarm-b",
SemanticPartition::Modeling,
&query,
Some(&readiness),
&config,
);
assert!(packet.warnings.iter().any(|warning| {
warning.contains("newest canonical obligation has no exact success receipt")
}));
Ok(())
}
#[test]
fn mind_and_modeling_have_distinct_physical_collections_and_query_profiles() {
let config = MemorySemanticIndexConfig::from_env();
@ -994,6 +1253,23 @@ mod tests {
assert_eq!(ready.status, MemorySemanticProjectionHealthStatus::Ready);
assert!(ready.query_eligible);
let mut repair = failed_attempt();
repair.attempt_id = "attempt-repair".to_string();
repair.started_at = "2026-07-15T10:02:00Z".to_string();
repair.completed_at = Some("2026-07-15T10:02:01Z".to_string());
let repairing = derive_memory_semantic_projection_health(
&obligation,
&current,
&[repair],
&[receipt()],
)
.unwrap();
assert_eq!(
repairing.status,
MemorySemanticProjectionHealthStatus::Failed
);
assert!(!repairing.query_eligible);
let mut newer = current.clone();
newer.source_generation += 1;
newer.source_commit_id = "commit-8".to_string();

968
epiphany-core/src/memory_graph/semantic_projector.rs Normal file
View file

@ -0,0 +1,968 @@
use super::{
MEMORY_SEMANTIC_PROJECTION_ATTEMPT_SCHEMA_VERSION, MemorySemanticIndexReceipt,
MemorySemanticProjectionAttempt, MemorySemanticProjectionObligation,
MemorySemanticProjectionSourceHead, bind_memory_semantic_index_receipt,
memory_semantic_projection_query_eligible, validate_memory_semantic_projection_attempt,
validate_memory_semantic_projection_obligation,
};
use anyhow::{Result, anyhow};
use cultcache_rs::{
CacheBackingStore, CultCache, CultCacheEnvelope, DatabaseEntry,
SingleFileMessagePackBackingStore,
};
use serde::de::DeserializeOwned;
use sha2::{Digest, Sha256};
use std::path::Path;
pub const MEMORY_SEMANTIC_PROJECTION_CLAIM_SCHEMA_VERSION: &str =
"gamecult.epiphany.memory_semantic_projection_claim.v0";
#[derive(Clone, Debug, PartialEq, DatabaseEntry)]
#[cultcache(
type = "gamecult.epiphany.memory_semantic_projection_claim",
schema = "MemorySemanticProjectionClaim"
)]
pub struct MemorySemanticProjectionClaim {
#[cultcache(key = 0)]
pub schema_version: String,
#[cultcache(key = 1)]
pub scope_id: String,
#[cultcache(key = 2)]
pub claim_id: String,
#[cultcache(key = 3)]
pub obligation_id: String,
#[cultcache(key = 4)]
pub attempt_id: String,
#[cultcache(key = 5)]
pub executor_id: String,
#[cultcache(key = 6)]
pub epoch: u64,
#[cultcache(key = 7)]
pub status: String,
#[cultcache(key = 8)]
pub claimed_at: String,
#[cultcache(key = 9)]
pub completed_at: Option<String>,
}
#[derive(Clone, Debug)]
pub(crate) struct MemorySemanticProjectionAuthoritySnapshot {
pub(crate) head: MemorySemanticProjectionSourceHead,
pub(crate) envelopes: Vec<CultCacheEnvelope>,
}
#[derive(Clone, Debug)]
pub struct MemorySemanticProjectionInput {
pub(crate) snapshot: super::EpiphanyMemoryGraphSnapshot,
pub(crate) obligation: MemorySemanticProjectionObligation,
pub(crate) authority: MemorySemanticProjectionAuthoritySnapshot,
}
impl MemorySemanticProjectionInput {
pub fn snapshot(&self) -> &super::EpiphanyMemoryGraphSnapshot {
&self.snapshot
}
pub fn obligation(&self) -> &MemorySemanticProjectionObligation {
&self.obligation
}
pub fn source_head(&self) -> &MemorySemanticProjectionSourceHead {
&self.authority.head
}
}
pub(crate) fn projection_scope_id(swarm_id: &str, partition: &str) -> Result<String> {
if swarm_id.trim().is_empty() || !matches!(partition, "mind" | "modeling") {
return Err(anyhow!("semantic projection scope identity is invalid"));
}
Ok(format!(
"memory-semantic-scope-{partition}-{:x}",
Sha256::digest(format!("{swarm_id}|{partition}").as_bytes())
))
}
pub fn load_memory_semantic_projection_success(
store_path: impl AsRef<Path>,
obligation: &MemorySemanticProjectionObligation,
head: &MemorySemanticProjectionSourceHead,
) -> Result<Option<MemorySemanticIndexReceipt>> {
validate_memory_semantic_projection_obligation(obligation)?;
let envelopes = SingleFileMessagePackBackingStore::new(store_path.as_ref()).pull_all()?;
let attempts = decode_all::<MemorySemanticProjectionAttempt>(&envelopes)?;
let receipts = decode_all::<MemorySemanticIndexReceipt>(&envelopes)?;
let health =
super::derive_memory_semantic_projection_health(obligation, head, &attempts, &receipts)?;
if health.status != super::MemorySemanticProjectionHealthStatus::Ready {
return Ok(None);
}
let receipt_id = health
.receipt_id
.ok_or_else(|| anyhow!("ready semantic projection health lost receipt identity"))?;
receipts
.into_iter()
.find(|receipt| receipt.receipt_id == receipt_id)
.map(Some)
.ok_or_else(|| anyhow!("ready semantic projection receipt disappeared"))
}
pub fn load_memory_semantic_projection_readiness(
store_path: impl AsRef<Path>,
input: &MemorySemanticProjectionInput,
) -> Result<Option<super::MemorySemanticProjectionReadiness>> {
let receipt = load_memory_semantic_projection_success(
store_path,
&input.obligation,
&input.authority.head,
)?;
Ok(
receipt.map(|receipt| super::MemorySemanticProjectionReadiness {
obligation: input.obligation.clone(),
current: input.authority.head.clone(),
receipt,
}),
)
}
pub fn execute_memory_semantic_projection(
store_path: impl AsRef<Path>,
input: &MemorySemanticProjectionInput,
executor_id: &str,
config: &super::MemorySemanticIndexConfig,
) -> Result<MemorySemanticIndexReceipt> {
let partition = match input.obligation.partition.as_str() {
"mind" => super::SemanticPartition::Mind,
"modeling" => super::SemanticPartition::Modeling,
_ => {
return Err(anyhow!(
"semantic projection obligation partition is invalid"
));
}
};
let store_path = store_path.as_ref();
let started_at = now_rfc3339();
let claim =
claim_memory_semantic_projection(store_path, &input.obligation, executor_id, &started_at)?;
let claim = if claim.status == "succeeded" {
if let Some(receipt) = load_memory_semantic_projection_success(
store_path,
&input.obligation,
&input.authority.head,
)? {
return Ok(receipt);
}
reopen_succeeded_projection_claim(
store_path,
&claim,
&input.obligation,
executor_id,
&now_rfc3339(),
)?
} else {
claim
};
let raw_receipt = match super::semantic_index::index_memory_semantic_partition(
&input.snapshot,
&input.obligation.swarm_id,
partition,
&started_at,
config,
) {
Ok(receipt) => receipt,
Err(error) => {
let _ = fail_memory_semantic_projection_claim(
store_path,
&claim.claim_id,
&now_rfc3339(),
&format!("{error:#}"),
);
return Err(error);
}
};
match succeed_memory_semantic_projection_claim(
store_path,
&claim.claim_id,
&input.authority,
raw_receipt,
&now_rfc3339(),
) {
Ok(receipt) => Ok(receipt),
Err(error) => {
let _ = fail_memory_semantic_projection_claim(
store_path,
&claim.claim_id,
&now_rfc3339(),
&format!("terminal success refused: {error:#}"),
);
Err(error)
}
}
}
fn reopen_succeeded_projection_claim(
store_path: &Path,
current: &MemorySemanticProjectionClaim,
obligation: &MemorySemanticProjectionObligation,
executor_id: &str,
reopened_at: &str,
) -> Result<MemorySemanticProjectionClaim> {
validate_memory_semantic_projection_claim(current)?;
validate_identity_and_time(executor_id, reopened_at)?;
if current.status != "succeeded" || current.obligation_id != obligation.obligation_id {
return Err(anyhow!(
"semantic projection repair requires exact succeeded claim"
));
}
let cache = semantic_projector_cache(store_path)?;
let opening = SingleFileMessagePackBackingStore::new(store_path).pull_all()?;
let stored_claim = decode_one::<MemorySemanticProjectionClaim>(&opening, &current.scope_id)?
.ok_or_else(|| anyhow!("semantic projection repair claim disappeared"))?;
if stored_claim != *current {
return Err(anyhow!("semantic projection repair claim advanced"));
}
let old_attempt = decode_one::<MemorySemanticProjectionAttempt>(&opening, &current.attempt_id)?
.ok_or_else(|| anyhow!("semantic projection repair lost succeeded attempt"))?;
if old_attempt.status != "succeeded" {
return Err(anyhow!(
"semantic projection repair attempt is not succeeded"
));
}
let next = running_claim(
&current.scope_id,
obligation,
executor_id,
current.epoch + 1,
reopened_at,
);
let next_attempt = running_attempt(&next);
terminal_or_recovery_cas(
store_path,
&opening,
current,
&old_attempt,
vec![
cache.prepare_entry(&current.scope_id, &next)?.0,
cache
.prepare_entry(&old_attempt.attempt_id, &old_attempt)?
.0,
cache
.prepare_entry(&next_attempt.attempt_id, &next_attempt)?
.0,
],
)?;
Ok(next)
}
pub(crate) fn claim_memory_semantic_projection(
store_path: impl AsRef<Path>,
obligation: &MemorySemanticProjectionObligation,
executor_id: &str,
claimed_at: &str,
) -> Result<MemorySemanticProjectionClaim> {
validate_memory_semantic_projection_obligation(obligation)?;
validate_identity_and_time(executor_id, claimed_at)?;
let store_path = store_path.as_ref();
let cache = semantic_projector_cache(store_path)?;
let opening = SingleFileMessagePackBackingStore::new(store_path).pull_all()?;
let stored =
decode_one::<MemorySemanticProjectionObligation>(&opening, &obligation.obligation_id)?
.ok_or_else(|| anyhow!("semantic projection claim requires persisted obligation"))?;
if stored != *obligation {
return Err(anyhow!("semantic projection obligation identity collision"));
}
let scope_id = projection_scope_id(&obligation.swarm_id, &obligation.partition)?;
let current = decode_one::<MemorySemanticProjectionClaim>(&opening, &scope_id)?;
if let Some(current) = &current {
validate_memory_semantic_projection_claim(current)?;
if current.status == "running" {
return if current.obligation_id == obligation.obligation_id
&& current.executor_id == executor_id
{
Ok(current.clone())
} else {
Err(anyhow!(
"semantic projection scope is already claimed by executor {:?}",
current.executor_id
))
};
}
if current.status == "succeeded" && current.obligation_id == obligation.obligation_id {
return Ok(current.clone());
}
}
let epoch = current.as_ref().map_or(1, |claim| claim.epoch + 1);
let claim = running_claim(&scope_id, obligation, executor_id, epoch, claimed_at);
let attempt = running_attempt(&claim);
let obligation_envelope = exact_envelope(
&opening,
"gamecult.epiphany.memory_semantic_projection_obligation",
&obligation.obligation_id,
)?;
let mut expected = vec![obligation_envelope.clone()];
let mut replacements = vec![obligation_envelope];
if current.is_some() {
let current_envelope = exact_envelope(
&opening,
"gamecult.epiphany.memory_semantic_projection_claim",
&scope_id,
)?;
expected.push(current_envelope);
}
replacements.push(cache.prepare_entry(&scope_id, &claim)?.0);
replacements.push(cache.prepare_entry(&attempt.attempt_id, &attempt)?.0);
if !SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&expected, replacements)?
{
return Err(anyhow!("semantic projection claim lost exact CAS"));
}
Ok(claim)
}
#[allow(dead_code)] // Deliberately withheld from callers until Idunn supplies typed recovery authority.
pub(crate) fn recover_memory_semantic_projection_claim(
store_path: impl AsRef<Path>,
expected_claim_id: &str,
executor_id: &str,
recovered_at: &str,
reason: &str,
) -> Result<MemorySemanticProjectionClaim> {
validate_identity_and_time(executor_id, recovered_at)?;
if expected_claim_id.trim().is_empty() || reason.trim().is_empty() {
return Err(anyhow!(
"semantic projection recovery requires claim and reason"
));
}
let store_path = store_path.as_ref();
let cache = semantic_projector_cache(store_path)?;
let opening = SingleFileMessagePackBackingStore::new(store_path).pull_all()?;
let current = decode_all::<MemorySemanticProjectionClaim>(&opening)?
.into_iter()
.find(|claim| claim.claim_id == expected_claim_id)
.ok_or_else(|| anyhow!("semantic projection recovery claim is missing"))?;
validate_memory_semantic_projection_claim(&current)?;
if current.status != "running" {
return Err(anyhow!(
"only a running semantic projection claim can recover"
));
}
let obligation =
decode_one::<MemorySemanticProjectionObligation>(&opening, &current.obligation_id)?
.ok_or_else(|| anyhow!("semantic projection recovery lost obligation"))?;
let old_attempt = decode_one::<MemorySemanticProjectionAttempt>(&opening, &current.attempt_id)?
.ok_or_else(|| anyhow!("semantic projection recovery lost running attempt"))?;
validate_memory_semantic_projection_attempt(&old_attempt)?;
if old_attempt.status != "running" {
return Err(anyhow!(
"semantic projection recovery attempt is not running"
));
}
let next = running_claim(
&current.scope_id,
&obligation,
executor_id,
current.epoch + 1,
recovered_at,
);
let next_attempt = running_attempt(&next);
let failed_attempt = MemorySemanticProjectionAttempt {
completed_at: Some(recovered_at.to_string()),
status: "failed".to_string(),
error: Some(format!("superseded by fenced recovery: {reason}")),
..old_attempt.clone()
};
terminal_or_recovery_cas(
store_path,
&opening,
&current,
&old_attempt,
vec![
cache.prepare_entry(&current.scope_id, &next)?.0,
cache
.prepare_entry(&failed_attempt.attempt_id, &failed_attempt)?
.0,
cache
.prepare_entry(&next_attempt.attempt_id, &next_attempt)?
.0,
],
)?;
Ok(next)
}
pub(crate) fn fail_memory_semantic_projection_claim(
store_path: impl AsRef<Path>,
claim_id: &str,
completed_at: &str,
error: &str,
) -> Result<MemorySemanticProjectionClaim> {
if error.trim().is_empty() {
return Err(anyhow!("semantic projection failure requires an error"));
}
chrono::DateTime::parse_from_rfc3339(completed_at)
.map_err(|_| anyhow!("semantic projection completion time must be RFC3339"))?;
let store_path = store_path.as_ref();
let (cache, opening, claim, attempt) = load_running_claim(store_path, claim_id)?;
let terminal_claim = MemorySemanticProjectionClaim {
status: "failed".to_string(),
completed_at: Some(completed_at.to_string()),
..claim.clone()
};
let terminal_attempt = MemorySemanticProjectionAttempt {
status: "failed".to_string(),
completed_at: Some(completed_at.to_string()),
error: Some(error.to_string()),
..attempt.clone()
};
terminal_or_recovery_cas(
store_path,
&opening,
&claim,
&attempt,
vec![
cache.prepare_entry(&claim.scope_id, &terminal_claim)?.0,
cache
.prepare_entry(&attempt.attempt_id, &terminal_attempt)?
.0,
],
)?;
Ok(terminal_claim)
}
pub(crate) fn succeed_memory_semantic_projection_claim(
store_path: impl AsRef<Path>,
claim_id: &str,
authority: &MemorySemanticProjectionAuthoritySnapshot,
receipt: MemorySemanticIndexReceipt,
completed_at: &str,
) -> Result<MemorySemanticIndexReceipt> {
chrono::DateTime::parse_from_rfc3339(completed_at)
.map_err(|_| anyhow!("semantic projection completion time must be RFC3339"))?;
if authority.envelopes.is_empty() {
return Err(anyhow!(
"semantic projection success requires canonical authority envelopes"
));
}
let store_path = store_path.as_ref();
let (cache, opening, claim, attempt) = load_running_claim(store_path, claim_id)?;
let obligation =
decode_one::<MemorySemanticProjectionObligation>(&opening, &claim.obligation_id)?
.ok_or_else(|| anyhow!("semantic projection success lost obligation"))?;
let receipt = bind_memory_semantic_index_receipt(receipt, &obligation)?;
if !memory_semantic_projection_query_eligible(&obligation, &authority.head, &receipt) {
return Err(anyhow!(
"semantic projection success no longer matches canonical source head"
));
}
let terminal_claim = MemorySemanticProjectionClaim {
status: "succeeded".to_string(),
completed_at: Some(completed_at.to_string()),
..claim.clone()
};
let terminal_attempt = MemorySemanticProjectionAttempt {
status: "succeeded".to_string(),
completed_at: Some(completed_at.to_string()),
error: None,
..attempt.clone()
};
let obligation_envelope = exact_envelope(
&opening,
"gamecult.epiphany.memory_semantic_projection_obligation",
&obligation.obligation_id,
)?;
let claim_envelope = exact_envelope(
&opening,
"gamecult.epiphany.memory_semantic_projection_claim",
&claim.scope_id,
)?;
let attempt_envelope = exact_envelope(
&opening,
"gamecult.epiphany.memory_semantic_projection_attempt",
&attempt.attempt_id,
)?;
let mut expected = vec![
obligation_envelope.clone(),
claim_envelope,
attempt_envelope,
];
let mut replacements = vec![
obligation_envelope,
cache.prepare_entry(&claim.scope_id, &terminal_claim)?.0,
cache
.prepare_entry(&attempt.attempt_id, &terminal_attempt)?
.0,
cache.prepare_entry(&receipt.receipt_id, &receipt)?.0,
];
for authority_envelope in &authority.envelopes {
let current = opening
.iter()
.find(|candidate| {
candidate.r#type == authority_envelope.r#type
&& candidate.key == authority_envelope.key
})
.ok_or_else(|| anyhow!("canonical projection authority envelope disappeared"))?;
if current != authority_envelope {
return Err(anyhow!("canonical projection authority head advanced"));
}
if !expected.iter().any(|candidate| {
candidate.r#type == authority_envelope.r#type && candidate.key == authority_envelope.key
}) {
expected.push(authority_envelope.clone());
replacements.push(authority_envelope.clone());
}
}
if !SingleFileMessagePackBackingStore::new(store_path)
.compare_and_swap_batch(&expected, replacements)?
{
return Err(anyhow!(
"semantic projection success lost exact terminal CAS"
));
}
Ok(receipt)
}
pub(crate) fn validate_memory_semantic_projection_claim(
claim: &MemorySemanticProjectionClaim,
) -> Result<()> {
if claim.schema_version != MEMORY_SEMANTIC_PROJECTION_CLAIM_SCHEMA_VERSION
|| claim.scope_id.trim().is_empty()
|| claim.claim_id.trim().is_empty()
|| claim.obligation_id.trim().is_empty()
|| claim.attempt_id.trim().is_empty()
|| claim.executor_id.trim().is_empty()
|| claim.epoch == 0
|| chrono::DateTime::parse_from_rfc3339(&claim.claimed_at).is_err()
{
return Err(anyhow!("semantic projection claim is invalid"));
}
match claim.status.as_str() {
"running" if claim.completed_at.is_none() => Ok(()),
"failed" | "succeeded"
if claim
.completed_at
.as_deref()
.is_some_and(|time| chrono::DateTime::parse_from_rfc3339(time).is_ok()) =>
{
Ok(())
}
_ => Err(anyhow!(
"semantic projection claim terminal fields disagree"
)),
}
}
fn semantic_projector_cache(store_path: &Path) -> Result<CultCache> {
let mut cache = CultCache::new();
cache.register_entry_type::<MemorySemanticProjectionObligation>()?;
cache.register_entry_type::<MemorySemanticProjectionClaim>()?;
cache.register_entry_type::<MemorySemanticProjectionAttempt>()?;
cache.register_entry_type::<MemorySemanticIndexReceipt>()?;
cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(store_path));
Ok(cache)
}
fn running_claim(
scope_id: &str,
obligation: &MemorySemanticProjectionObligation,
executor_id: &str,
epoch: u64,
claimed_at: &str,
) -> MemorySemanticProjectionClaim {
let fingerprint = format!(
"{}|{}|{}|{}",
scope_id, obligation.obligation_id, executor_id, epoch
);
let claim_id = format!(
"memory-semantic-claim-{:x}",
Sha256::digest(fingerprint.as_bytes())
);
MemorySemanticProjectionClaim {
schema_version: MEMORY_SEMANTIC_PROJECTION_CLAIM_SCHEMA_VERSION.to_string(),
scope_id: scope_id.to_string(),
claim_id: claim_id.clone(),
obligation_id: obligation.obligation_id.clone(),
attempt_id: format!("memory-semantic-attempt-{claim_id}"),
executor_id: executor_id.to_string(),
epoch,
status: "running".to_string(),
claimed_at: claimed_at.to_string(),
completed_at: None,
}
}
fn running_attempt(claim: &MemorySemanticProjectionClaim) -> MemorySemanticProjectionAttempt {
MemorySemanticProjectionAttempt {
schema_version: MEMORY_SEMANTIC_PROJECTION_ATTEMPT_SCHEMA_VERSION.to_string(),
attempt_id: claim.attempt_id.clone(),
obligation_id: claim.obligation_id.clone(),
started_at: claim.claimed_at.clone(),
completed_at: None,
status: "running".to_string(),
error: None,
}
}
fn load_running_claim(
store_path: &Path,
claim_id: &str,
) -> Result<(
CultCache,
Vec<CultCacheEnvelope>,
MemorySemanticProjectionClaim,
MemorySemanticProjectionAttempt,
)> {
let cache = semantic_projector_cache(store_path)?;
let opening = SingleFileMessagePackBackingStore::new(store_path).pull_all()?;
let claim = decode_all::<MemorySemanticProjectionClaim>(&opening)?
.into_iter()
.find(|claim| claim.claim_id == claim_id)
.ok_or_else(|| anyhow!("semantic projection claim is missing"))?;
validate_memory_semantic_projection_claim(&claim)?;
if claim.status != "running" {
return Err(anyhow!("semantic projection claim is not running"));
}
let attempt = decode_one::<MemorySemanticProjectionAttempt>(&opening, &claim.attempt_id)?
.ok_or_else(|| anyhow!("semantic projection running attempt is missing"))?;
validate_memory_semantic_projection_attempt(&attempt)?;
if attempt.status != "running" || attempt.obligation_id != claim.obligation_id {
return Err(anyhow!("semantic projection attempt does not own claim"));
}
Ok((cache, opening, claim, attempt))
}
fn terminal_or_recovery_cas(
store_path: &Path,
opening: &[CultCacheEnvelope],
claim: &MemorySemanticProjectionClaim,
attempt: &MemorySemanticProjectionAttempt,
mut replacements: Vec<CultCacheEnvelope>,
) -> Result<()> {
let obligation = exact_envelope(
opening,
"gamecult.epiphany.memory_semantic_projection_obligation",
&claim.obligation_id,
)?;
let claim_envelope = exact_envelope(
opening,
"gamecult.epiphany.memory_semantic_projection_claim",
&claim.scope_id,
)?;
let attempt_envelope = exact_envelope(
opening,
"gamecult.epiphany.memory_semantic_projection_attempt",
&attempt.attempt_id,
)?;
replacements.push(obligation.clone());
if !SingleFileMessagePackBackingStore::new(store_path).compare_and_swap_batch(
&[obligation, claim_envelope, attempt_envelope],
replacements,
)? {
return Err(anyhow!(
"semantic projection terminal transition lost exact CAS"
));
}
Ok(())
}
fn exact_envelope(
envelopes: &[CultCacheEnvelope],
type_name: &str,
key: &str,
) -> Result<CultCacheEnvelope> {
envelopes
.iter()
.find(|envelope| envelope.r#type == type_name && envelope.key == key)
.cloned()
.ok_or_else(|| anyhow!("missing {type_name} envelope {key:?}"))
}
fn decode_one<T: DatabaseEntry + DeserializeOwned>(
envelopes: &[CultCacheEnvelope],
key: &str,
) -> Result<Option<T>> {
envelopes
.iter()
.find(|envelope| envelope.r#type == T::TYPE && envelope.key == key)
.map(|envelope| {
rmp_serde::from_slice(&envelope.payload)
.map_err(|error| anyhow!("failed to decode {} {key:?}: {error}", T::TYPE))
})
.transpose()
}
fn decode_all<T: DatabaseEntry + DeserializeOwned>(
envelopes: &[CultCacheEnvelope],
) -> Result<Vec<T>> {
envelopes
.iter()
.filter(|envelope| envelope.r#type == T::TYPE)
.map(|envelope| {
rmp_serde::from_slice(&envelope.payload).map_err(|error| {
anyhow!("failed to decode {} {:?}: {error}", T::TYPE, envelope.key)
})
})
.collect()
}
fn validate_identity_and_time(identity: &str, time: &str) -> Result<()> {
if identity.trim().is_empty() {
return Err(anyhow!("semantic projection executor identity is empty"));
}
chrono::DateTime::parse_from_rfc3339(time)
.map_err(|_| anyhow!("semantic projection time must be RFC3339"))?;
Ok(())
}
fn now_rfc3339() -> String {
chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION,
MEMORY_SEMANTIC_PROJECTION_OBLIGATION_SCHEMA_VERSION, SEMANTIC_PROJECTION_SCHEMA_VERSION,
};
use tempfile::tempdir;
fn obligation() -> MemorySemanticProjectionObligation {
MemorySemanticProjectionObligation {
schema_version: MEMORY_SEMANTIC_PROJECTION_OBLIGATION_SCHEMA_VERSION.to_string(),
obligation_id: "obligation-modeling-7".to_string(),
swarm_id: "swarm-a".to_string(),
partition: "modeling".to_string(),
canonical_source_id: "runtime/repo-model".to_string(),
source_commit_id: "admission-7".to_string(),
graph_id: "repo-model".to_string(),
source_generation: 7,
source_model_hash: "model-hash-7".to_string(),
canonical_content_set_hash: "content-set-7".to_string(),
projection_schema_version: SEMANTIC_PROJECTION_SCHEMA_VERSION.to_string(),
created_at: "2026-07-15T04:00:00Z".to_string(),
}
}
fn source_head() -> MemorySemanticProjectionSourceHead {
let obligation = obligation();
MemorySemanticProjectionSourceHead {
swarm_id: obligation.swarm_id,
partition: obligation.partition,
canonical_source_id: obligation.canonical_source_id,
source_commit_id: obligation.source_commit_id,
graph_id: obligation.graph_id,
source_generation: obligation.source_generation,
source_model_hash: obligation.source_model_hash,
canonical_content_set_hash: obligation.canonical_content_set_hash,
}
}
fn receipt() -> MemorySemanticIndexReceipt {
let obligation = obligation();
MemorySemanticIndexReceipt {
schema_version: MEMORY_SEMANTIC_INDEX_RECEIPT_SCHEMA_VERSION.to_string(),
receipt_id: "receipt-modeling-7".to_string(),
swarm_id: obligation.swarm_id,
partition: obligation.partition,
collection_name: "modeling".to_string(),
graph_id: obligation.graph_id,
model_revision: obligation.source_generation,
model_hash: obligation.source_model_hash,
embedding_provider_id: "embedder".to_string(),
embedding_model: "model".to_string(),
vector_dimensions: 3,
indexed_document_count: 1,
deleted_document_count: 0,
canonical_content_set_hash: obligation.canonical_content_set_hash,
indexed_at: "2026-07-15T04:03:00Z".to_string(),
status: "ready".to_string(),
obligation_id: String::new(),
canonical_source_id: String::new(),
source_commit_id: String::new(),
source_generation: obligation.source_generation,
projection_schema_version: SEMANTIC_PROJECTION_SCHEMA_VERSION.to_string(),
}
}
#[test]
fn scope_claim_recovery_fences_old_executor_and_terminalizes_exact_success() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("projector.msgpack");
let obligation = obligation();
let mut cache = semantic_projector_cache(&store)?;
cache.put(&obligation.obligation_id, &obligation)?;
let first = claim_memory_semantic_projection(
&store,
&obligation,
"executor-a",
"2026-07-15T04:01:00Z",
)?;
assert!(
claim_memory_semantic_projection(
&store,
&obligation,
"executor-b",
"2026-07-15T04:01:01Z"
)
.is_err()
);
let recovered = recover_memory_semantic_projection_claim(
&store,
&first.claim_id,
"executor-b",
"2026-07-15T04:02:00Z",
"daemon restarted under operator recovery",
)?;
assert_eq!(recovered.epoch, first.epoch + 1);
assert!(
fail_memory_semantic_projection_claim(
&store,
&first.claim_id,
"2026-07-15T04:02:30Z",
"late old executor"
)
.is_err()
);
let mut cache = semantic_projector_cache(&store)?;
cache.pull_all_backing_stores()?;
let authority_envelope = cache
.snapshot_envelopes()
.into_iter()
.find(|envelope| {
envelope.r#type == "gamecult.epiphany.memory_semantic_projection_obligation"
&& envelope.key == obligation.obligation_id
})
.expect("obligation envelope");
let bound = succeed_memory_semantic_projection_claim(
&store,
&recovered.claim_id,
&MemorySemanticProjectionAuthoritySnapshot {
head: source_head(),
envelopes: vec![authority_envelope],
},
receipt(),
"2026-07-15T04:03:00Z",
)?;
assert_eq!(bound.obligation_id, obligation.obligation_id);
let mut cache = semantic_projector_cache(&store)?;
cache.pull_all_backing_stores()?;
let terminal = cache
.get::<MemorySemanticProjectionClaim>(&recovered.scope_id)?
.expect("terminal claim");
assert_eq!(terminal.status, "succeeded");
let old_attempt = cache
.get::<MemorySemanticProjectionAttempt>(&first.attempt_id)?
.expect("fenced old attempt");
assert_eq!(old_attempt.status, "failed");
let new_attempt = cache
.get::<MemorySemanticProjectionAttempt>(&recovered.attempt_id)?
.expect("successful recovered attempt");
assert_eq!(new_attempt.status, "succeeded");
let repair_attempt = MemorySemanticProjectionAttempt {
schema_version: MEMORY_SEMANTIC_PROJECTION_ATTEMPT_SCHEMA_VERSION.to_string(),
attempt_id: "repair-after-success".to_string(),
obligation_id: obligation.obligation_id.clone(),
started_at: "2026-07-15T04:04:00Z".to_string(),
completed_at: Some("2026-07-15T04:04:01Z".to_string()),
status: "failed".to_string(),
error: Some("observed cache corruption".to_string()),
};
cache.put(&repair_attempt.attempt_id, &repair_attempt)?;
assert!(
load_memory_semantic_projection_success(&store, &obligation, &source_head())?.is_none()
);
Ok(())
}
#[test]
fn terminal_success_refuses_a_preexisting_receipt_identity_collision() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("receipt-collision.msgpack");
let obligation = obligation();
let mut cache = semantic_projector_cache(&store)?;
cache.put(&obligation.obligation_id, &obligation)?;
let claim = claim_memory_semantic_projection(
&store,
&obligation,
"executor-a",
"2026-07-15T04:01:00Z",
)?;
let mut collision = receipt();
collision.status = "failed".to_string();
cache.put(&collision.receipt_id, &collision)?;
let authority_envelope = SingleFileMessagePackBackingStore::new(&store)
.pull_all()?
.into_iter()
.find(|envelope| {
envelope.r#type == "gamecult.epiphany.memory_semantic_projection_obligation"
&& envelope.key == obligation.obligation_id
})
.expect("obligation envelope");
assert!(
succeed_memory_semantic_projection_claim(
&store,
&claim.claim_id,
&MemorySemanticProjectionAuthoritySnapshot {
head: source_head(),
envelopes: vec![authority_envelope],
},
receipt(),
"2026-07-15T04:03:00Z",
)
.is_err()
);
Ok(())
}
#[test]
fn succeeded_claim_with_missing_receipt_reopens_under_a_new_fence() -> Result<()> {
let temp = tempdir()?;
let store = temp.path().join("missing-receipt.msgpack");
let obligation = obligation();
let mut cache = semantic_projector_cache(&store)?;
cache.put(&obligation.obligation_id, &obligation)?;
let claim = claim_memory_semantic_projection(
&store,
&obligation,
"executor-a",
"2026-07-15T04:01:00Z",
)?;
let authority_envelope = SingleFileMessagePackBackingStore::new(&store)
.pull_all()?
.into_iter()
.find(|envelope| {
envelope.r#type == "gamecult.epiphany.memory_semantic_projection_obligation"
&& envelope.key == obligation.obligation_id
})
.expect("obligation envelope");
let bound = succeed_memory_semantic_projection_claim(
&store,
&claim.claim_id,
&MemorySemanticProjectionAuthoritySnapshot {
head: source_head(),
envelopes: vec![authority_envelope],
},
receipt(),
"2026-07-15T04:03:00Z",
)?;
let mut cache = semantic_projector_cache(&store)?;
cache.pull_all_backing_stores()?;
cache.delete::<MemorySemanticIndexReceipt>(&bound.receipt_id)?;
let terminal = cache
.get::<MemorySemanticProjectionClaim>(&claim.scope_id)?
.expect("succeeded claim");
let reopened = reopen_succeeded_projection_claim(
&store,
&terminal,
&obligation,
"executor-b",
"2026-07-15T04:04:00Z",
)?;
assert_eq!(reopened.status, "running");
assert_eq!(reopened.epoch, terminal.epoch + 1);
assert_ne!(reopened.claim_id, terminal.claim_id);
Ok(())
}
}

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

@ -843,6 +843,9 @@ pub fn runtime_spine_cache(store_path: impl AsRef<Path>) -> Result<CultCache> {
cache.register_entry_type::<EpiphanyRuntimeIdentity>()?;
cache.register_entry_type::<EpiphanyRuntimeSwarmBinding>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionObligation>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionClaim>()?;
cache.register_entry_type::<crate::MemorySemanticProjectionAttempt>()?;
cache.register_entry_type::<crate::MemorySemanticIndexReceipt>()?;
cache.register_entry_type::<EpiphanyRuntimeSession>()?;
cache.register_entry_type::<EpiphanyRuntimeJob>()?;
cache.register_entry_type::<EpiphanyRuntimeWorkerLaunchRequest>()?;
@ -3235,6 +3238,83 @@ pub fn runtime_current_repo_model(
.transpose()
}
pub fn runtime_modeling_semantic_projection_input(
store_path: impl AsRef<Path>,
) -> Result<crate::MemorySemanticProjectionInput> {
let mut cache = runtime_spine_cache(store_path)?;
cache.pull_all_backing_stores()?;
let binding = require_runtime_swarm_binding(&cache)?;
let entry = cache
.get::<crate::EpiphanyMemoryGraphEntry>(crate::MEMORY_GRAPH_KEY)?
.ok_or_else(|| anyhow!("Modeling projection requires admitted RepoModel"))?;
let snapshot = entry.snapshot()?;
let model_hash = crate::memory_graph_model_hash(&snapshot)?;
let canonical_source_id = format!("epiphany.runtime/{}/repo-model", binding.runtime_id);
let matches = cache
.get_all::<crate::MemorySemanticProjectionObligation>()?
.into_iter()
.filter(|obligation| {
obligation.swarm_id == binding.swarm_id
&& obligation.partition == "modeling"
&& obligation.canonical_source_id == canonical_source_id
&& obligation.graph_id == snapshot.graph_id
&& obligation.source_generation == snapshot.model_revision
&& obligation.source_model_hash == model_hash
})
.collect::<Vec<_>>();
if matches.len() != 1 {
return Err(anyhow!(
"Modeling projection requires exactly one obligation for current RepoModel"
));
}
let obligation = matches.into_iter().next().expect("one obligation");
let expected = crate::derive_memory_semantic_projection_obligation(
&snapshot,
&binding.swarm_id,
crate::SemanticPartition::Modeling,
&canonical_source_id,
&obligation.source_commit_id,
&obligation.created_at,
)?;
if obligation != expected {
return Err(anyhow!(
"Modeling projection obligation does not match canonical RepoModel"
));
}
let opening = cache.snapshot_envelopes();
let authority_envelopes = opening
.into_iter()
.filter(|envelope| {
(envelope.r#type == RUNTIME_SWARM_BINDING_TYPE
&& envelope.key == RUNTIME_SWARM_BINDING_KEY)
|| (envelope.r#type == crate::MEMORY_GRAPH_TYPE
&& envelope.key == crate::MEMORY_GRAPH_KEY)
})
.collect::<Vec<_>>();
if authority_envelopes.len() != 2 {
return Err(anyhow!(
"Modeling projection authority snapshot is incomplete"
));
}
Ok(crate::MemorySemanticProjectionInput {
snapshot,
authority: crate::memory_graph::MemorySemanticProjectionAuthoritySnapshot {
head: crate::MemorySemanticProjectionSourceHead {
swarm_id: obligation.swarm_id.clone(),
partition: obligation.partition.clone(),
canonical_source_id: obligation.canonical_source_id.clone(),
source_commit_id: obligation.source_commit_id.clone(),
graph_id: obligation.graph_id.clone(),
source_generation: obligation.source_generation,
source_model_hash: obligation.source_model_hash.clone(),
canonical_content_set_hash: obligation.canonical_content_set_hash.clone(),
},
envelopes: authority_envelopes,
},
obligation,
})
}
pub fn select_repo_frontier_work_proposal_for_modeling(
store_path: impl AsRef<Path>,
proposal_id: &str,
@ -7865,7 +7945,11 @@ pub(crate) mod tests {
migrate_legacy_repo_model_projection_obligation(&store)?,
Some(obligation.clone())
);
assert_eq!(runtime_current_repo_model(&store)?, Some(snapshot));
assert_eq!(runtime_current_repo_model(&store)?, Some(snapshot.clone()));
let input = runtime_modeling_semantic_projection_input(&store)?;
assert_eq!(input.snapshot, snapshot);
assert_eq!(input.obligation, obligation);
assert_eq!(input.authority.envelopes.len(), 2);
Ok(())
}

64
epiphany-core/src/semantic_backend.rs
View file

@ -62,6 +62,12 @@ pub(crate) struct SemanticCandidate<P> {
pub(crate) payload: Option<P>,
}
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct SemanticStoredPoint<P> {
pub(crate) id: String,
pub(crate) payload: Option<P>,
}
pub(crate) struct QdrantBackend {
base_url: String,
timeout_seconds: u64,
@ -286,6 +292,54 @@ impl QdrantBackend {
}
Ok(ids)
}
pub(crate) fn points_for_scope<P: DeserializeOwned>(
&self,
name: &str,
scope: &[(&str, &str)],
) -> Result<Vec<SemanticStoredPoint<P>>> {
let must = scope
.iter()
.map(|(key, value)| json!({ "key": key, "match": { "value": value } }))
.collect::<Vec<_>>();
let mut offset: Option<Value> = None;
let mut points = Vec::new();
loop {
let response = self
.client
.post(format!(
"{}/collections/{name}/points/scroll",
self.base_url
))
.query(&[("timeout", self.timeout_seconds)])
.json(&json!({
"filter": { "must": must },
"limit": POINT_BATCH_SIZE,
"offset": offset,
"with_payload": true,
"with_vector": false,
}))
.send()
.with_context(|| format!("failed to observe Qdrant scope in {name}"))?;
let envelope: Envelope<PayloadScrollResult<P>> = parse_response(response)
.with_context(|| format!("failed to decode Qdrant scope payloads in {name}"))?;
points.extend(
envelope
.result
.points
.into_iter()
.map(|point| SemanticStoredPoint {
id: point.id,
payload: point.payload,
}),
);
offset = envelope.result.next_page_offset;
if offset.is_none() {
break;
}
}
Ok(points)
}
}
pub(crate) struct OllamaEmbedder {
@ -433,6 +487,16 @@ struct ScrollPoint {
id: String,
}
#[derive(Deserialize)]
struct PayloadScrollResult<P> {
points: Vec<PayloadScrollPoint<P>>,
next_page_offset: Option<Value>,
}
#[derive(Deserialize)]
struct PayloadScrollPoint<P> {
id: String,
payload: Option<P>,
}
#[derive(Deserialize)]
struct EmbedResponse {
embeddings: Option<Vec<Vec<f32>>>,
}

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

@ -1102,10 +1102,40 @@ admission receipt, and Mind obligation together. Every extension re-hashes the
canonical rows and authenticates the prior witness. Bootstrap/import/repair/raw
replacement/trait seeding cannot write after a generation exists.
Core proof after fixture repair: 342 tests passed, 0 failed, 1 intentional
ignore; focused legacy migration and hostile concurrent Mind CAS tests pass;
all binaries compile. The current `state/agents.msgpack` predates the generation
witness and still needs one explicit legacy Mind-generation migration before
query gating can regard its Qdrant projection as ready. Next implement
restart-safe projector claim/attempt/terminal receipt execution and exact query
eligibility, then CultMesh/Idunn physiology.
Core proof for the landed atomic admission pass: 343 tests passed, 0 failed, 1
intentional ignore; all binaries compile; the OpenAI runtime suites pass 22
tests. The current `state/agents.msgpack` has now been admitted without row
mutation as generation 1 under swarm `gamecult.epiphany.main`, with immutable
witness `mind-generation-eb394e34-1c73-5d0a-8691-6ff1ba6145a4`, its Mind
admission receipt, and its exact projection obligation. Repeating the migration
is idempotent and validation remains clean. Next replace the direct projector
CLI with restart-safe scope claims, attempt/terminal receipt CAS, observed
zero-document synchronization, and exact query eligibility; then add
CultMesh/Idunn physiology.
## Restart-safe semantic projector and gated query (2026-07-15)
The projection executor now serializes each `(swarm, partition)` mutation scope
with a durable executor-bound claim and fencing epoch. An internal recovery
transition proves that an abandoned attempt can be failed and fenced, but it is
not exposed until Idunn supplies typed stale/recovery authority; arbitrary peers
cannot steal a live scope. Terminal success requires the same live claim, exact
bound receipt, unchanged canonical source envelopes, and observed Qdrant scope.
The raw indexer and ordinary receipt writer are no longer public authority
surfaces. Empty projections bypass Ollama and delete only their exact scope.
Non-empty projections observe exact IDs and typed payload identities after
synchronization.
Query gating is live in the CLI and Persona heartbeat. Without the newest exact
obligation/success pair they use canonical BM25 without contacting Ollama or
Qdrant. With the pair they use Qdrant only for ranking, reload canonical typed
documents, and ignore payload prose. The live `gamecult.epiphany.main` Mind and
Modeling obligations were discharged through the protocol: 43 Mind documents
and 3 Modeling documents, both 1024-dimensional. Repeated execution returned
the same immutable receipts. Core proof: 349 passed, 0 failed, 1 intentional
ignore; all binaries type-check.
Next: derive and publish provider-owned health through CultMesh, make Idunn own
projector process survival/restart and explicit fenced recovery routing, then
run the deployability audit against daemon packaging, configuration, and live
Yggdrasil assumptions. Neither CultMesh nor Idunn may mint readiness.

112
notes/memory-graph-semantic-projection-authority-map.md
View file

@ -279,3 +279,115 @@ add projector claims/attempt terminal CAS and bind index receipts to exact
obligations; make queries require the newest exact success before Qdrant
ranking; then publish derived health through provider-owned CultMesh state and
attach Idunn process survival.
## Projector Execution Rebuild Map (2026-07-15)
### Owner
One projector executor owns mutation of a semantic scope identified by
`(swarm_id, partition)` for the duration of an exact canonical obligation.
Canonical admission owns the obligation. The executor owns only its durable
claim, Qdrant synchronization, post-write observation, and terminal evidence.
Qdrant, health projections, schedulers, Idunn, CultMesh, Eve, and Gjallar own
none of those decisions.
### Inputs And Outputs
- Inputs: the newest exact obligation for the authenticated canonical source
head; the canonical snapshot reconstructed from that source; immutable swarm
identity; embedding and collection compatibility; projector incarnation.
- Success output: an immutable receipt bound to every obligation/source field,
emitted only after the observed scoped point set and typed payload identities
equal the derived canonical projection.
- Failure output: a terminal failed attempt for the claimed obligation. It
never erases the obligation or permits semantic query.
- Derived output: `pending`, `failed`, `stale`, or `ready` health. Health is
recomputed sight, not writable state.
### Derived State And Demotions
- `MemorySemanticProjectionHealth` is no longer a prospective daemon status;
it is derived from canonical head, newest matching obligation, exact attempt,
and exact receipt.
- Collection existence, compatibility, point counts, HTTP success, embedding
success, and process liveness are observations only.
- An older success is no longer readiness once a newer canonical obligation
exists. A repair attempt begun after success suppresses that success until a
newer successful terminal receipt proves the scope again.
### Forbidden Writers And Shared Paths
- The CLI may not call the raw indexer and `put` an unbound ready receipt.
- Query and Persona heartbeat recall may not touch Qdrant until the same gate
authenticates the current head, selects its newest exact obligation, and
finds its immutable exact success receipt.
- A caller-supplied swarm id, arbitrary graph-store snapshot, timer, collection
metadata, Qdrant payload, or renderer status cannot open the gate.
- Initial execution, crash replay, daemon restart, operator retry, and repair
all use the same scope claim and terminalization primitives.
### Cut Line
1. Remove the CLI's direct `index -> ordinary receipt put` authority path.
2. Replace it with a scope-serialized claim, idempotent synchronization, fresh
canonical-head reauthentication, post-write verification, and terminal CAS.
3. Delete the empty-partition refusal. An empty desired set bypasses Ollama,
deletes every point in its exact scope, preserves other scopes, and earns a
zero-document receipt; an absent collection is already synchronized.
4. Require the exact readiness gate before embedding or Qdrant query; otherwise
use canonical BM25 without touching either external service.
5. Only after execution and query authority are sealed, lower derived health to
CultMesh and give Idunn survival responsibility for the executor daemon.
### Restart And Concurrency Law
- Claims serialize the shared mutation scope, not merely an obligation id;
generations of one swarm partition overwrite the same point population.
- A crash before terminal evidence leaves pressure open. Replay is idempotent
and repairs partial upsert/delete work; partial state is never queryable.
- Recovery requires an explicit fenced incarnation/epoch. A wall-clock lease
alone may identify a candidate for recovery but cannot let the previous
executor publish afterward.
- Terminal success CAS expects the exact live claim and reauthenticated source
head. If the source advanced mid-run, the old mutation may exist in Qdrant
but cannot become ready; the newer obligation repairs it while queries use
canonical BM25.
## Projector Execution Landed (2026-07-15)
- The direct CLI `raw index -> ordinary receipt put` authority path is gone.
The public executor claims the exact `(swarm, partition)` scope, records a
running attempt, synchronizes Qdrant, observes the resulting scope, and can
publish success only through an exact terminal CAS against the unchanged
canonical authority envelopes.
- Claims carry executor identity and a monotonically fenced epoch. The internal
recovery transition terminalizes the abandoned attempt, advances the epoch,
and makes the old executor structurally unable to publish. That transition is
deliberately withheld from production callers until Idunn supplies typed
stale/recovery authority; a random peer cannot fence a live executor. Failed
execution leaves an exact failed terminal attempt and never grants readiness.
- Mind source authentication includes immutable swarm identity, all seven
canonical role rows, and the latest generation witness. Modeling source
authentication includes immutable runtime swarm binding and the canonical
RepoModel envelope. Both reconstruct and compare the obligation before work.
- Empty canonical partitions bypass Ollama. A missing collection already
represents the empty set; an existing compatible collection deletes every
point in the exact swarm/partition scope and confirms that none remain.
Non-empty synchronization observes the exact ID set and typed locator payload
set after upsert/delete before returning a candidate receipt.
- Semantic query now requires the newest authenticated source input plus its
exact immutable success receipt before either Ollama or Qdrant is touched.
CLI context and Persona heartbeat use this gate. Missing, stale, foreign, or
unbound evidence selects canonical BM25 and reports fallback.
- The live Mind generation-1 and Modeling revision-0 obligations for
`gamecult.epiphany.main` were discharged through the new protocol. Mind
indexed 43 documents and Modeling indexed 3 at 1024 dimensions. Repeated
execution returned the same receipts without another projection mutation;
live context then used Qdrant ranking while resolving every hit back to
canonical documents and ignoring payload prose.
Remaining physiology: publish provider-owned derived projection health through
CultMesh, attach Idunn-owned daemon survival and authorize the fenced recovery
transition with typed stale/recovery evidence, and add the daemon pulse that
discovers open obligations without letting time own their creation or
completion.

BIN
state/agents.msgpack
View file

Binary file not shown.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

2
state/map.yaml
View file

@ -714,4 +714,4 @@ current_status:
- Service lifecycle receipt history is now event-only: `load_epiphany_cultmesh_daemon_service_lifecycle_receipts` filters out the `epiphany-local/daemon-service-lifecycle-receipt/latest` mirror, while `load_latest_epiphany_cultmesh_daemon_service_lifecycle_receipt` remains the latest-state reader. Focused unit test `service_lifecycle_receipt_history_excludes_latest_mirror` proves two per-receipt writes return exactly two history records and latest points at the second. Fresh smoke `verse-query-lifecycle-history-filter-smoke` and live overview/triage `local-20260618-145632-497-efac6f0d` / `local-20260618-145632-321-867947af` prove compact swarm sights still expose service lifecycle attention plus the present elevated runbook artifact without daemon pokes or service mutation. Do not use a broad unscoped cargo test filter for this check on Windows; it fans out across bin harnesses and hit a broken pipe during this pass.
- `epiphany-verse-query seed-compact` is the compact seed receipt for wrappers and read-only rites. The older `seed` command still prints the full local Verse context for explicit debugging, but compact wrapper modes now avoid full `seed`/final `query` readback after the combined overview proof exposed Windows debug stack overflow from full-context serialization pressure.
last_completed_repo_swarm_cut: Huginn live-fire has crossed the first real repo Body from Persona pressure to branch-local Hands commit and Soul/Mind closure, and now has repeatable closure and redacted-public-proof mouths. On `E:\Projects\Huginn` branch `epiphany/huginn/first-awakening`, `tools/epiphany_local_run.ps1 -Mode repo-persona-intake` accepted item `huginn-inspection-readback-livefire`, Weksa/Persona intake remained sealed, and `epiphany-work derive-plan` lowered it to model-authored safe family `repo.status_section` for `README.md` only. `tools/epiphany_local_run.ps1 -Mode repo-swarm-run` advanced four bounded pulses: run-from-plan, adopt-from-plan, execute-from-plan, and close-from-execute. Huginn commit `e9b07c38fceabbdc89ea35c2156326c80c0393a2` updated only `README.md`; closure receipt `.epiphany\work\work-close-huginn-inspection-readback-livefire.json` records status `closed`, Soul verdict `passed`, Mind state commit `repo-work-close-huginn-inspection-readback-livefire-mind-commit`, safe family `repo.status_section`, path scope `README.md`, Bifrost publication gate, and `privateStateExposed=false`. Verification also ran Huginn's TypeScript compiler through concrete Node path: `C:\Program Files\nodejs\node.exe E:\Projects\Huginn\node_modules\typescript\bin\tsc -p tsconfig.json` passed. Native verifier `epiphany-repo-livefire-closure-smoke --workspace E:\Projects\Huginn --item huginn-inspection-readback-livefire --expected-path README.md --expected-family repo.status_section --expected-gate awaiting-publication --expected-blocker bifrost-publication-missing` checks closure receipt, Mind map entry, overview gate/blocker, authority seals, and `git show` changed paths, returning `status=ok`, `soulVerdict=passed`, `publicationGate=Bifrost`, and `privateStateExposed=false`. Wrapper `tools/epiphany_local_run.ps1 -Mode repo-livefire-closure -Workspace E:\Projects\Huginn -RepoWorkItem huginn-inspection-readback-livefire -SkipBuild` exposes the same read-only proof through the local operator mouth; artifact `local-20260621-190156-803-2f85ba15` printed commit, path, gate, blocker, Soul verdict, Mind commit receipt, and `privateStateExposed=False` without push, merge, publication, or target mutation. New wrapper `tools/epiphany_local_run.ps1 -Mode repo-work-public-proof -Workspace E:\Projects\Huginn -RepoWorkItem huginn-inspection-readback-livefire -SkipBuild` exports the redacted proof bundle and repo-local Verse index without publishing; artifact `local-20260621-190645-345-4555b1dc` wrote `.epiphany\public\proof-bundles\repo-work-public-proof-huginn-inspection-readback-livefire.json`, proof id `repo-work-public-proof-huginn-inspection-readback-livefire`, SHA-256 `857545b199f3e37eb06c2e757e5cfc201958e46667bb1f8403fd333a22620f63`, nine artifact rows, zero publication rows, and `privateStateExposed=False`. swarm overview readback `local-20260621-190701-140-b9519ec6` sees the `REPO-WORK-PUBLIC-PROOF` row and action row from Huginn's repo-local Verse with the same SHA-256 while Bifrost publication remains missing. Launcher compact modes are now named compact operator-context modes instead of read-only modes, and `bifrost-public-proof` summaries print item, target public Verse, room, proof SHA-256, review receipts, and credit receipts so publication receipts are auditable without claiming upstream sync.
next_action: Build restart-safe projector execution on the landed atomic admission nerve. Runtime-spine is immutably bound to canonical Mind swarm identity; RepoModel bootstrap/generic admission/frontier Adopt atomically emit Modeling obligations; reviewed Mind self-patch/lifecycle commits atomically emit fixed-role generation witness, persisted admission receipt, and Mind obligation; bootstrap-only writers refuse after generation and witness hashes are authenticated against current canonical rows. The live legacy runtime has its exact Modeling obligation. Next add an explicit legacy Mind generation-zero migration for current `state/agents.msgpack`, then claim/attempt/terminal receipt CAS, zero-document scoped deletion, and query gating that permits Qdrant only for the newest exact obligation/success receipt. Only after that publish derived provider-owned CultMesh health and attach Idunn survival. Canonical partitioned BM25 remains correct throughout. Do not let timers, Qdrant compatibility, point counts, runtime ids, paths, CLI labels, Idunn liveness, CultMesh mirrors, Eve/Gjallar, or payload prose impersonate swarm identity or readiness.
next_action: Complete semantic projection physiology on the landed atomic admission, fenced projector, and exact query gate. Mind and Modeling now emit canonical obligations; the live `gamecult.epiphany.main` obligations are discharged through scope-serialized executor claims, post-write observation, terminal CAS, and immutable receipts; CLI and Persona heartbeat touch Qdrant only for the newest exact success and otherwise use canonical BM25. Next publish provider-owned derived pending/failed/stale/ready health through CultMesh without writable health state, package the projector as an Idunn-owned daemon with explicit fenced recovery routing and open-obligation discovery, and run the deployability audit for local/Yggdrasil Qdrant+Ollama configuration, daemon identity, restart behavior, and operator-safe Eve lowering. Timers, collection metadata, point counts, runtime ids, paths, CLI labels, Idunn liveness, CultMesh mirrors, Eve/Gjallar, and payload prose own no readiness.