diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..5448eda1 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +/schemas/cultnet/epiphany.pipeline.*.schema.json text eol=lf diff --git a/AGENTS.md b/AGENTS.md index 6db3f50a..ac980edc 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,7 +2,10 @@ ## Project Purpose -This repo is the forge-temple of Epiphany: an opinionated Codex fork where +This repo is the forge-temple of Epiphany: an opinionated agent harness, born +as a Codex fork and now backend-agnostic. Codex, Claude, and OpenAI-compatible +APIs speak to it through connectors that share one standardized +Epiphany-facing CultNet API, the same one Ghostlight uses. It is the harness where external typed state, explicit mental maps, bounded scratch work, verifier evidence, and anti-churn discipline are wired into the blessed harness instead of taped onto the chat transcript like a purity seal over a cracked reactor. diff --git a/Cargo.lock b/Cargo.lock index 975f7a2a..7d75c250 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -363,12 +363,30 @@ dependencies = [ [[package]] name = "cultcache-rs" -version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" +version = "0.2.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=8fc74c709444b884a2bd339e0c306bf9484d99d2#8fc74c709444b884a2bd339e0c306bf9484d99d2" dependencies = [ "anyhow", "chrono", - "cultcache-rs-derive", + "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=8fc74c709444b884a2bd339e0c306bf9484d99d2)", + "fs2", + "redb", + "rmp-serde", + "serde", + "serde_bytes", + "serde_json", + "uuid", + "windows-sys 0.59.0", +] + +[[package]] +name = "cultcache-rs" +version = "0.2.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6#a0813c6eed24d30bf88073ef615b633c77ebfcd6" +dependencies = [ + "anyhow", + "chrono", + "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "fs2", "redb", "rmp-serde", @@ -382,7 +400,17 @@ dependencies = [ [[package]] name = "cultcache-rs-derive" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" +source = "git+https://github.com/GameCult/CultLib.git?rev=8fc74c709444b884a2bd339e0c306bf9484d99d2#8fc74c709444b884a2bd339e0c306bf9484d99d2" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "cultcache-rs-derive" +version = "0.1.0" +source = "git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6#a0813c6eed24d30bf88073ef615b633c77ebfcd6" dependencies = [ "proc-macro2", "quote", @@ -392,10 +420,10 @@ dependencies = [ [[package]] name = "cultmesh-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" +source = "git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6#a0813c6eed24d30bf88073ef615b633c77ebfcd6" dependencies = [ "anyhow", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "cultnet-rs", "serde", ] @@ -403,13 +431,13 @@ dependencies = [ [[package]] name = "cultnet-rs" version = "0.1.0" -source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" +source = "git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6#a0813c6eed24d30bf88073ef615b633c77ebfcd6" dependencies = [ "aes-gcm", "anyhow", "base64 0.22.1", "chrono", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "ed25519-dalek", "hmac", "rand 0.9.5", @@ -502,6 +530,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ed25519" version = "2.2.3" @@ -533,7 +567,7 @@ dependencies = [ "anyhow", "chrono", "ctrlc", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "cultmesh-rs", "cultnet-rs", "ed25519-dalek", @@ -559,7 +593,7 @@ version = "0.1.0" dependencies = [ "anyhow", "codex-connector", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "serde", "serde_json", "sha2", @@ -571,7 +605,7 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "epiphany-core", "epiphany-model-adapter", "epiphany-tool-adapter", @@ -584,6 +618,19 @@ dependencies = [ "uuid", ] +[[package]] +name = "epiphany-pipeline" +version = "0.1.0" +dependencies = [ + "anyhow", + "chrono", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=8fc74c709444b884a2bd339e0c306bf9484d99d2)", + "rmp-serde", + "schemars", + "serde", + "serde_json", +] + [[package]] name = "epiphany-release-bundle" version = "0.1.0" @@ -592,7 +639,7 @@ dependencies = [ "chrono", "codex-connector", "ctrlc", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "cultnet-rs", "epiphany-core", "epiphany-model-adapter", @@ -615,7 +662,7 @@ dependencies = [ name = "epiphany-tool-adapter" version = "0.1.0" dependencies = [ - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "serde", "serde_json", ] @@ -626,7 +673,7 @@ version = "0.1.0" dependencies = [ "anyhow", "chrono", - "cultcache-rs", + "cultcache-rs 0.2.0 (git+https://github.com/GameCult/CultLib.git?rev=a0813c6eed24d30bf88073ef615b633c77ebfcd6)", "epiphany-core", "epiphany-tool-adapter", "http", @@ -849,7 +896,7 @@ dependencies = [ [[package]] name = "ghostlight-persona-projection" version = "0.1.0" -source = "git+https://github.com/GameCult/Ghostlight.git?rev=22281891cbf9258db80c15a9fd465e6fa251e7dd#22281891cbf9258db80c15a9fd465e6fa251e7dd" +source = "git+https://github.com/GameCult/Ghostlight.git?rev=22937b71f056d2000dd7ea1f6af0c45825de495f#22937b71f056d2000dd7ea1f6af0c45825de495f" dependencies = [ "serde", "sha2", @@ -1557,6 +1604,26 @@ dependencies = [ "libc", ] +[[package]] +name = "ref-cast" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e440fb4e4b4147295338efb76001ab9e4efc0e5839df2c47fc5ac2381d365c3" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "regex-automata" version = "0.4.18" @@ -1810,6 +1877,31 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "schemars" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a" +dependencies = [ + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.3", +] + [[package]] name = "security-framework" version = "3.7.0" @@ -1883,6 +1975,17 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_json" version = "1.0.151" diff --git a/Cargo.toml b/Cargo.toml index dfb47187..2d59e523 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,6 +3,7 @@ members = [ "epiphany-core", "epiphany-model-adapter", "epiphany-openai-runtime", + "epiphany-pipeline", "epiphany-tool-adapter", "epiphany-tool-mcp-runtime", ] @@ -20,8 +21,8 @@ license-file = "LICENSE" anyhow = "1" chrono = "0.4.44" ctrlc = { version = "3.5.1", features = ["termination"] } -cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs" } -cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultnet-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultcache-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultnet-rs" } epiphany-core = { path = "epiphany-core" } epiphany-model-adapter = { path = "epiphany-model-adapter" } epiphany-openai-runtime = { path = "epiphany-openai-runtime", optional = true } diff --git a/epiphany-core/Cargo.toml b/epiphany-core/Cargo.toml index 35a994c3..e0986e4e 100644 --- a/epiphany-core/Cargo.toml +++ b/epiphany-core/Cargo.toml @@ -14,9 +14,9 @@ path = "src/lib.rs" anyhow = "1" chrono = "0.4.44" ctrlc = { version = "3.5.1", features = ["termination"] } -cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs" } -cultmesh-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultmesh-rs" } -cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultnet-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultcache-rs" } +cultmesh-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultmesh-rs" } +cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultnet-rs" } epiphany-model-adapter = { path = "../epiphany-model-adapter" } epiphany-tool-adapter = { path = "../epiphany-tool-adapter" } fs2 = "0.4" @@ -24,7 +24,7 @@ ghostlight-persona-projection = { git = "https://github.com/GameCult/Ghostlight. ed25519-dalek = "2" ignore = "0.4.23" rmp-serde = "1" -serde = { version = "1", features = ["derive"] } +serde ={ version = "1", features = ["derive"] } serde_json = "1" semver = { version = "1", features = ["serde"] } sha2 = "0.10" diff --git a/epiphany-core/src/atlas/store.rs b/epiphany-core/src/atlas/store.rs index e82ccf75..3cee306c 100644 --- a/epiphany-core/src/atlas/store.rs +++ b/epiphany-core/src/atlas/store.rs @@ -355,7 +355,7 @@ impl AtlasCultCacheStore { cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&self.path)); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&self.path))?; cache.pull_all_backing_stores()?; Ok(cache) } diff --git a/epiphany-core/src/persona_conversation.rs b/epiphany-core/src/persona_conversation.rs index 105b2873..d27f9e59 100644 --- a/epiphany-core/src/persona_conversation.rs +++ b/epiphany-core/src/persona_conversation.rs @@ -958,7 +958,7 @@ fn require_store_retirement_receipt( })?; let mut cache = CultCache::new(); cache.register_entry_type::()?; - cache.load_envelope::(envelope.clone())?; + cache.put_envelope::(envelope.clone())?; if cache .get::(&expected.receipt_id)? .as_ref() diff --git a/epiphany-core/src/persona_feedback_admission.rs b/epiphany-core/src/persona_feedback_admission.rs index dd477ed0..8a5ae614 100644 --- a/epiphany-core/src/persona_feedback_admission.rs +++ b/epiphany-core/src/persona_feedback_admission.rs @@ -1042,7 +1042,7 @@ mod tests { let delivery = signed_delivery(&signer, "epiphany")?; let mut source = CultCache::new(); source.register_entry_type::()?; - source.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&source_store)); + source.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&source_store))?; source.put(&delivery.admission_id, &delivery)?; let source_lock = source_store.with_file_name(format!( "{}.lock", diff --git a/epiphany-core/src/persona_social_state.rs b/epiphany-core/src/persona_social_state.rs index 29758733..a755d7ff 100644 --- a/epiphany-core/src/persona_social_state.rs +++ b/epiphany-core/src/persona_social_state.rs @@ -846,19 +846,19 @@ fn is_persona_social_type(document_type: &str) -> bool { fn load_persona_social_envelope(cache: &mut CultCache, envelope: CultCacheEnvelope) -> Result<()> { match envelope.r#type.as_str() { PERSONA_SOCIAL_MENTION_TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } PERSONA_SOCIAL_TURN_REQUEST_TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } PERSONA_SOCIAL_TURN_TERMINAL_TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } PERSONA_SOCIAL_RETENTION_HEAD_TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } PERSONA_SOCIAL_RETENTION_PLAN_TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } _ => unreachable!("caller filtered Persona social document types"), } @@ -891,3 +891,50 @@ fn stable_pending_mention_id( fn now_iso() -> String { chrono::Utc::now().to_rfc3339() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn persona_social_cache_reads_without_writing_the_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = temp.path().join("social.cc"); + let cache = persona_social_cache(&store)?; + let head = PersonaSocialRetentionHeadDocument { + head: PersonaConversationRetentionHead::default(), + }; + SingleFileMessagePackBackingStore::new(&store).push( + &cache + .prepare_entry(PERSONA_SOCIAL_RETENTION_HEAD_KEY, &head)? + .0, + )?; + let past = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000_000); + std::fs::File::options() + .write(true) + .open(&store)? + .set_modified(past)?; + let listing = || -> Result> { + let mut entries = std::fs::read_dir(temp.path())? + .map(|entry| { + let entry = entry?; + let metadata = entry.metadata()?; + Ok((entry.file_name(), metadata.len(), metadata.modified()?)) + }) + .collect::>>()?; + entries.sort(); + Ok(entries) + }; + let before = listing()?; + let names = before.iter().map(|entry| entry.0.clone()).collect::>(); + assert_eq!(names, ["social.cc", "social.cc.lock"]); + let reopened = persona_social_cache(&store)?; + assert_eq!( + reopened + .get::(PERSONA_SOCIAL_RETENTION_HEAD_KEY)?, + Some(head) + ); + assert_eq!(listing()?, before, "a read attached or wrote a store"); + Ok(()) + } +} diff --git a/epiphany-core/src/reasoning_context.rs b/epiphany-core/src/reasoning_context.rs index 0414319a..11f5a8a4 100644 --- a/epiphany-core/src/reasoning_context.rs +++ b/epiphany-core/src/reasoning_context.rs @@ -1572,7 +1572,16 @@ pub(crate) fn commit_external_typed_observation_mind_mutation( ) } -fn commit_authorized_mind_mutation( +/// The one owner of a Mind receipt commit. It resolves the Mind backing store +/// once per commit from `store_path` and uses that one store for the cache +/// load, the batch CAS, and the conflict re-read, so no path in this function +/// can target a second store. +/// +/// Order is load-bearing and fail-closed: identities must be unique, the store +/// must open (which refuses a foreign epoch), and every write must validate, +/// all before a stored receipt may answer a replay. A batch the current +/// validator refuses is refused, not answered from the store. +pub(crate) fn commit_authorized_mind_mutation( store_path: &Path, authority: EpiphanyMindCommitAuthority, invariant_owner: &str, @@ -1588,9 +1597,6 @@ fn commit_authorized_mind_mutation( if writes.is_empty() { return Err(anyhow!("Mind mutation requires at least one write")); } - for write in &writes { - crate::mind_documents::validate_mind_write_envelope(write)?; - } validate_unique_envelope_identities(&strong_reads, "strong read")?; validate_unique_envelope_identities(&writes, "write")?; validate_unique_envelope_identities(&companions, "companion")?; @@ -1602,8 +1608,12 @@ fn commit_authorized_mind_mutation( .iter() .map(|entry| (entry.r#type.clone(), entry.key.clone())) .collect::>(); - let mut cache = runtime_spine_cache(store_path)?; + let backing_store = runtime_spine_backing_store(store_path)?; + let mut cache = crate::runtime_spine::open_runtime_spine_cache(backing_store.clone())?; cache.pull_all_backing_stores()?; + for write in &writes { + crate::mind_documents::validate_mind_write_envelope(write)?; + } let mut companion_expected = Vec::new(); let mut companion_replacements = Vec::new(); for companion in companions { @@ -1666,10 +1676,10 @@ fn commit_authorized_mind_mutation( replacements.push(cache.prepare_entry(&receipt_id, &receipt)?.0); let mut expected = strong_reads.clone(); expected.extend(companion_expected); - if runtime_spine_backing_store(store_path)?.compare_and_swap_batch(&expected, replacements)? { + if backing_store.compare_and_swap_batch(&expected, replacements)? { return Ok(EpiphanyMindCommitOutcome::Committed(receipt)); } - let current = runtime_spine_backing_store(store_path)?.pull_all()?; + let current = backing_store.pull_all()?; let mut conflicts = strong_reads .iter() .filter(|expected| { @@ -2478,6 +2488,386 @@ mod tests { Ok(()) } + /// An envelope the Mind validator refuses: its key is not the value's + /// semantic identity. + fn refused_mind_write(store: &Path) -> Result { + let value = crate::EpiphanyObservation { + id: "right-id".into(), + summary: "one".into(), + source_kind: "test".into(), + status: "accepted".into(), + code_refs: Vec::new(), + evidence_ids: Vec::new(), + }; + Ok(runtime_spine_cache(store)? + .prepare_entry("wrong-key", &crate::EpiphanyMindObservationDocument { value })? + .0) + } + + /// Ruling 12, and Soul's finding S6: every write is validated before a + /// stored receipt may answer a replay, so a replay of a batch the current + /// validator refuses is refused rather than answered from the store. + #[test] + fn mind_commit_validates_before_answering_a_replay() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-replay-test".into(), + display_name: "Mind replay test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let refused = refused_mind_write(&store)?; + let provenance = EpiphanyMindDocumentVersion::from_envelope("epiphany-organ", &refused)?; + let authority = EpiphanyMindCommitAuthority::TypedOrganProvenance { + organ: "test-organ".into(), + provenance: provenance.clone(), + }; + let writes = vec![EpiphanyMindDocumentVersion::from_envelope( + "epiphany-mind", + &refused, + )?]; + // Plant exactly the receipt this batch would produce, so a replay check + // that ran first would have a stored answer to hand back. + let receipt_id = mind_commit_receipt_id(&authority, "test-owner", &[], &writes)?; + let planted = EpiphanyMindCommitReceipt { + schema_version: MIND_COMMIT_RECEIPT_SCHEMA_VERSION.to_string(), + receipt_id: receipt_id.clone(), + authority, + invariant_owner: "test-owner".into(), + strong_reads: Vec::new(), + writes, + committed_at: "2026-09-15T00:00:01Z".into(), + }; + planted.validate()?; + let mut planting = runtime_spine_cache(&store)?; + planting.pull_all_backing_stores()?; + planting.put(&receipt_id, &planted)?; + let before = std::fs::read(&store)?; + let error = commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance, + "test-owner", + Vec::new(), + vec![refused], + "2026-09-15T00:00:02Z", + ) + .unwrap_err(); + assert!( + error.to_string().contains("semantic identity"), + "a refused batch must be refused, not answered with the stored receipt: {error}" + ); + assert_eq!(std::fs::read(&store)?, before); + Ok(()) + } + + /// Every write in the batch is validated, not just the first. A batch of + /// two distinct writes whose second is refused is refused whole, with + /// nothing written. + #[test] + fn mind_commit_validates_every_write_in_the_batch() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-batch-validation-test".into(), + display_name: "Mind batch validation test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let value = crate::EpiphanyObservation { + id: "first".into(), + summary: "one".into(), + source_kind: "test".into(), + status: "accepted".into(), + code_refs: Vec::new(), + evidence_ids: Vec::new(), + }; + let accepted = runtime_spine_cache(&store)? + .prepare_entry("first", &crate::EpiphanyMindObservationDocument { value })? + .0; + let refused = refused_mind_write(&store)?; + assert_ne!( + (&accepted.r#type, &accepted.key), + (&refused.r#type, &refused.key), + "the two writes carry distinct identities, so uniqueness cannot answer first" + ); + let provenance = EpiphanyMindDocumentVersion::from_envelope("epiphany-organ", &accepted)?; + let before = std::fs::read(&store)?; + let error = commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance, + "test-owner", + Vec::new(), + vec![accepted, refused], + "2026-09-15T00:00:01Z", + ) + .unwrap_err(); + assert!( + error.to_string().contains("semantic identity"), + "the second write is validated too: {error}" + ); + assert_eq!( + std::fs::read(&store)?, + before, + "a refused batch writes nothing, including its valid writes" + ); + Ok(()) + } + + /// Ruling 12 again: identity uniqueness is checked before write validation, + /// so a batch that repeats an identity reports the repeat, not the + /// validator's refusal. The order is fail-closed either way; this pins + /// which refusal the caller is told about. + #[test] + fn mind_commit_refuses_repeated_write_identities_before_validating() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-uniqueness-test".into(), + display_name: "Mind uniqueness test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let refused = refused_mind_write(&store)?; + let provenance = EpiphanyMindDocumentVersion::from_envelope("epiphany-organ", &refused)?; + let before = std::fs::read(&store)?; + let error = commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance, + "test-owner", + Vec::new(), + vec![refused.clone(), refused], + "2026-09-15T00:00:01Z", + ) + .unwrap_err(); + assert!( + error.to_string().contains("repeats write identity"), + "uniqueness is checked before the writes are validated: {error}" + ); + assert_eq!(std::fs::read(&store)?, before); + Ok(()) + } + + /// The owner resolves one backing store from the path and uses that one for + /// the cache load, the batch CAS and the conflict re-read. A commit that + /// wrote anywhere else would leave a second store beside this one, and a + /// replay read from anywhere else would not find the receipt. + /// + /// The file assertions below cannot see a handle re-derived from the *same* + /// path, because a handle carries nothing but its path. The resolution + /// count can, and that is the shape the map's redb plan makes load-bearing: + /// redb permits one writable handle per path, so a second derivation is a + /// second handle there. + #[test] + fn mind_commit_reads_and_writes_one_store() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-one-store-test".into(), + display_name: "Mind one store test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let value = crate::EpiphanyObservation { + id: "first".into(), + summary: "one".into(), + source_kind: "test".into(), + status: "accepted".into(), + code_refs: Vec::new(), + evidence_ids: Vec::new(), + }; + let write = runtime_spine_cache(&store)? + .prepare_entry("first", &crate::EpiphanyMindObservationDocument { value })? + .0; + let provenance = EpiphanyMindDocumentVersion::from_envelope("epiphany-organ", &write)?; + let commit = |at: &str| { + commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance.clone(), + "test-owner", + Vec::new(), + vec![write.clone()], + at, + ) + }; + let resolutions_before = + crate::runtime_store_backend::backing_store_resolutions(&store); + let EpiphanyMindCommitOutcome::Committed(receipt) = commit("2026-09-15T00:00:01Z")? else { + panic!("the first Mind write must commit"); + }; + assert_eq!( + crate::runtime_store_backend::backing_store_resolutions(&store) - resolutions_before, + 1, + "the commit resolves the backing store once and reuses that handle" + ); + let mut files = std::fs::read_dir(temp.path())? + .map(|entry| Ok(entry?.file_name().to_string_lossy().into_owned())) + .collect::>>()?; + files.sort(); + assert_eq!( + files, + vec!["mind.cc".to_string(), "mind.cc.lock".to_string()], + "the commit writes exactly the store it was given, and locks that one" + ); + let stored = SingleFileMessagePackBackingStore::new(&store).pull_all()?; + assert!( + stored.iter().any(|entry| entry.key == receipt.receipt_id), + "the receipt lands in the store the owner read" + ); + assert_eq!( + commit("2026-09-15T00:00:09Z")?, + EpiphanyMindCommitOutcome::Committed(receipt), + "the replay is answered from that same store" + ); + Ok(()) + } + + #[test] + fn mind_commit_refuses_a_foreign_epoch_store() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-epoch-test".into(), + display_name: "Mind epoch test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let cache = runtime_spine_cache(&store)?; + let backing = runtime_spine_backing_store(&store)?; + let current = backing + .pull_all()? + .into_iter() + .find(|entry| entry.r#type == crate::EpiphanyMindIdentity::TYPE) + .expect("an initialized store carries its Mind identity"); + let foreign = crate::EpiphanyMindIdentity { + schema_epoch: "epiphany.mind.epoch.foreign".into(), + runtime_id: "mind-epoch-test".into(), + }; + let foreign = cache.prepare_entry(¤t.key, &foreign)?.0; + assert!(backing.compare_and_swap_batch(&[current], vec![foreign])?); + let value = crate::EpiphanyObservation { + id: "foreign".into(), + summary: "foreign".into(), + source_kind: "test".into(), + status: "accepted".into(), + code_refs: Vec::new(), + evidence_ids: Vec::new(), + }; + let write = cache + .prepare_entry("foreign", &crate::EpiphanyMindObservationDocument { value })? + .0; + let provenance = EpiphanyMindDocumentVersion::from_envelope("epiphany-organ", &write)?; + let before = std::fs::read(&store)?; + let error = commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance, + "test-owner", + Vec::new(), + vec![write], + "2026-09-15T00:00:01Z", + ) + .unwrap_err(); + assert!(error.to_string().contains("unsupported writable schema epoch")); + assert_eq!(std::fs::read(&store)?, before); + Ok(()) + } + + #[test] + fn mind_commit_keeps_validation_receipts_replay_and_conflicts() -> Result<()> { + let temp = tempdir()?; + let store = temp.path().join("mind.cc"); + initialize_runtime_spine( + &store, + RuntimeSpineInitOptions { + runtime_id: "mind-commit-keeps-test".into(), + display_name: "Mind commit keeps test".into(), + created_at: "2026-09-15T00:00:00Z".into(), + }, + )?; + let cache = runtime_spine_cache(&store)?; + let observation = |key: &str, id: &str, summary: &str| -> Result { + let value = crate::EpiphanyObservation { + id: id.into(), + summary: summary.into(), + source_kind: "test".into(), + status: "accepted".into(), + code_refs: Vec::new(), + evidence_ids: Vec::new(), + }; + Ok(cache + .prepare_entry(key, &crate::EpiphanyMindObservationDocument { value })? + .0) + }; + let provenance = EpiphanyMindDocumentVersion::from_envelope( + "epiphany-organ", + &observation("provenance", "provenance", "organ")?, + )?; + let commit = |reads: Vec, + writes: Vec, + at: &str| { + commit_external_typed_observation_mind_mutation( + &store, + "test-organ", + provenance.clone(), + "test-owner", + reads, + writes, + at, + ) + }; + let before = std::fs::read(&store)?; + let error = commit( + Vec::new(), + vec![observation("wrong-key", "right-id", "x")?], + "2026-09-15T00:00:01Z", + ) + .unwrap_err(); + assert!(error.to_string().contains("semantic identity")); + assert_eq!(std::fs::read(&store)?, before); + let first = observation("first", "first", "one")?; + let EpiphanyMindCommitOutcome::Committed(receipt) = + commit(Vec::new(), vec![first.clone()], "2026-09-15T00:00:02Z")? + else { + panic!("the first Mind write must commit"); + }; + assert!(receipt.writes.iter().all(|version| version.store_id == "epiphany-mind")); + assert_eq!( + receipt.receipt_id, + "mind-commit-sha256:415283192ba888e99ce011b5e7a2d2ff82e93d9cbf9b5c8af96eebc57acc281e" + ); + assert_eq!( + commit(Vec::new(), vec![first.clone()], "2026-09-15T00:00:30Z")?, + EpiphanyMindCommitOutcome::Committed(receipt), + "exact replay returns the stored receipt" + ); + assert_eq!( + commit( + vec![observation("first", "first", "stale")?], + vec![observation("second", "second", "two")?], + "2026-09-15T00:00:03Z", + )?, + EpiphanyMindCommitOutcome::Conflict { + document_identities: vec![(first.r#type.clone(), "first".into())], + } + ); + Ok(()) + } + #[test] fn disjoint_mind_mutations_merge_and_same_identity_conflicts() -> Result<()> { let temp = tempdir()?; diff --git a/epiphany-core/src/repository_body_observer.rs b/epiphany-core/src/repository_body_observer.rs index 9f406fba..d7625251 100644 --- a/epiphany-core/src/repository_body_observer.rs +++ b/epiphany-core/src/repository_body_observer.rs @@ -979,7 +979,7 @@ fn sanitized_git_command() -> Command { sanitize_git_command(&mut command); command } -fn repository_git_command(repo: &Path) -> Result { +pub(crate) fn repository_git_command(repo: &Path) -> Result { let canonical = std::fs::canonicalize(repo) .with_context(|| format!("failed to canonicalize Git repository {}", repo.display()))?; let mut command = sanitized_git_command(); diff --git a/epiphany-core/src/resident_readiness.rs b/epiphany-core/src/resident_readiness.rs index 4e48704c..a8a947c9 100644 --- a/epiphany-core/src/resident_readiness.rs +++ b/epiphany-core/src/resident_readiness.rs @@ -788,7 +788,7 @@ mod tests { let store = temp.path().join("mixed.cc"); let mut mixed = CultCache::new(); mixed.register_entry_type::()?; - mixed.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store)); + mixed.add_generic_backing_store(SingleFileMessagePackBackingStore::new(&store))?; let foreign = ForeignOwnerEntry { value: "must survive readiness CAS".into(), }; diff --git a/epiphany-core/src/resident_self.rs b/epiphany-core/src/resident_self.rs index 85cfe508..9871ca45 100644 --- a/epiphany-core/src/resident_self.rs +++ b/epiphany-core/src/resident_self.rs @@ -797,22 +797,22 @@ fn state_cache(path: &Path) -> Result { } match envelope.r#type.as_str() { ResidentSelfState::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } ResidentSelfPressure::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } ResidentSelfGrant::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } ResidentSelfTerminalReceipt::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } ResidentSelfChildClaim::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } ResidentSelfRetentionHead::TYPE => { - cache.load_envelope::(envelope)?; + cache.put_envelope::(envelope)?; } _ => unreachable!("owned resident Self type was matched above"), }; @@ -2730,6 +2730,45 @@ mod pressure_replay_tests { Ok(()) } + #[test] + fn state_cache_reads_without_writing_the_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = temp.path().join("resident.cc"); + let cache = state_cache(&store)?; + SingleFileMessagePackBackingStore::new(&store).push( + &cache + .prepare_entry(RESIDENT_SELF_STATE_KEY, &ResidentSelfState::default())? + .0, + )?; + let past = std::time::UNIX_EPOCH + std::time::Duration::from_secs(1_000_000_000); + std::fs::File::options() + .write(true) + .open(&store)? + .set_modified(past)?; + let listing = || -> Result> { + let mut entries = std::fs::read_dir(temp.path())? + .map(|entry| { + let entry = entry?; + let metadata = entry.metadata()?; + Ok((entry.file_name(), metadata.len(), metadata.modified()?)) + }) + .collect::>>()?; + entries.sort(); + Ok(entries) + }; + let before = listing()?; + let names = before.iter().map(|entry| entry.0.clone()).collect::>(); + assert_eq!(names, ["resident.cc", "resident.cc.lock"]); + let reopened = state_cache(&store)?; + assert!( + reopened + .get::(RESIDENT_SELF_STATE_KEY)? + .is_some() + ); + assert_eq!(listing()?, before, "a read attached or wrote a store"); + Ok(()) + } + #[test] fn configured_operator_pressure_replay_preserves_the_first_document() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index 5e6804c5..af9e626a 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -640,11 +640,15 @@ pub struct RuntimeSpineHeartbeatJobOptions { } pub fn runtime_spine_cache(store_path: impl AsRef) -> Result { - let store_path = store_path.as_ref(); - let backing_store = runtime_spine_backing_store(store_path)?; + open_runtime_spine_cache(runtime_spine_backing_store(store_path.as_ref())?) +} + +pub(crate) fn open_runtime_spine_cache( + backing_store: SingleFileMessagePackBackingStore, +) -> Result { validate_runtime_store_epoch(&backing_store.pull_all()?)?; let mut cache = runtime_spine_schema_cache()?; - cache.add_generic_backing_store(backing_store); + cache.add_generic_backing_store(backing_store)?; Ok(cache) } @@ -8655,7 +8659,7 @@ pub(crate) mod tests { let mut historical = CultCache::new(); historical.register_entry_type::()?; historical.register_entry_type::()?; - historical.add_generic_backing_store(runtime_spine_backing_store(&store)?); + historical.add_generic_backing_store(runtime_spine_backing_store(&store)?)?; historical.put( "epiphany.mind.epoch.v1", &crate::EpiphanyMindIdentity { @@ -8691,7 +8695,7 @@ pub(crate) mod tests { let mut archive_v0 = CultCache::new(); archive_v0.register_entry_type::()?; archive_v0.register_entry_type::()?; - archive_v0.add_generic_backing_store(runtime_spine_backing_store(&archive_v0_store)?); + archive_v0.add_generic_backing_store(runtime_spine_backing_store(&archive_v0_store)?)?; archive_v0.put( crate::MIND_SCHEMA_EPOCH, &crate::EpiphanyMindIdentity { @@ -8720,7 +8724,7 @@ pub(crate) mod tests { responses_only.register_entry_type::()?; responses_only.register_entry_type::()?; responses_only - .add_generic_backing_store(runtime_spine_backing_store(&responses_only_store)?); + .add_generic_backing_store(runtime_spine_backing_store(&responses_only_store)?)?; responses_only.put( crate::MIND_SCHEMA_EPOCH, &crate::EpiphanyMindIdentity { @@ -8747,6 +8751,58 @@ pub(crate) mod tests { Ok(()) } + /// A stand-in for a document of the Eureka pipeline, which + /// `epiphany-pipeline` owns and this package deliberately does not depend + /// on. The rule under test is the spine registry's, not the document's: + /// what the spine cache sees is a type id it never registered, and that is + /// all this needs to carry. + #[derive(Clone, Debug, PartialEq, Eq, DatabaseEntry)] + #[cultcache( + type = "epiphany.pipeline.campaign.v1", + schema = "EpiphanyPipelineCampaignDocument" + )] + struct ForeignPipelineDocument { + #[cultcache(key = 0)] + slug: String, + } + + /// Soul F4: the spine store and a pipeline store are different stores, and + /// the spine cache refuses the latter *by type* — it registers its own + /// types and nothing else. The epoch checks beside this one cannot stand in + /// for it: a pipeline store carries no runtime or Mind identity at all, so + /// `validate_runtime_store_epoch` returns `Ok` and the registry is the only + /// thing left refusing. + #[test] + fn runtime_spine_cache_refuses_a_pipeline_store() -> Result<()> { + let temp = tempfile::tempdir()?; + let store = temp.path().join("pipeline.cc"); + let mut pipeline = CultCache::new(); + pipeline.register_entry_type::()?; + let envelope = pipeline + .prepare_entry_named( + "eureka-state", + &ForeignPipelineDocument { slug: "eureka-state".into() }, + )? + .0; + assert!( + envelope.r#type.starts_with("epiphany.pipeline."), + "the sample is a pipeline document: {}", + envelope.r#type + ); + let mut backing = runtime_spine_backing_store(&store)?; + backing.push(&envelope)?; + + assert!( + validate_runtime_store_epoch(&runtime_spine_backing_store(&store)?.pull_all()?).is_ok(), + "the epoch check passes a pipeline store, so only the registry refuses it" + ); + let error = runtime_spine_cache(&store) + .and_then(|mut runtime| runtime.pull_all_backing_stores()) + .unwrap_err(); + assert!(error.to_string().contains("epiphany.pipeline."), "{error:#}"); + Ok(()) + } + #[test] fn coordinator_terminal_receipts_preserve_exact_immutable_run_basis() -> Result<()> { let temp = tempfile::tempdir()?; diff --git a/epiphany-core/src/runtime_store_backend.rs b/epiphany-core/src/runtime_store_backend.rs index 4f0add24..42310a2a 100644 --- a/epiphany-core/src/runtime_store_backend.rs +++ b/epiphany-core/src/runtime_store_backend.rs @@ -108,7 +108,11 @@ impl CacheBackingStore for RuntimeSpineBackingStore { pub(crate) fn runtime_spine_backing_store(path: &Path) -> Result { match path.extension().and_then(|extension| extension.to_str()) { - Some("cc" | "msgpack" | "redb") => Ok(RuntimeSpineBackingStore::new(path)), + Some("cc" | "msgpack" | "redb") => { + #[cfg(test)] + record_backing_store_resolution(path); + Ok(RuntimeSpineBackingStore::new(path)) + } extension => Err(anyhow!( "runtime spine store must use an explicit .cc, .msgpack, or .redb extension; got {:?}", extension @@ -116,6 +120,39 @@ pub(crate) fn runtime_spine_backing_store(path: &Path) -> Result> = + std::sync::Mutex::new(Vec::new()); + +#[cfg(test)] +fn record_backing_store_resolution(path: &Path) { + RESOLVED_BACKING_STORES + .lock() + .expect("the backing-store resolution log is never poisoned") + .push(path.to_path_buf()); +} + +/// How many times this path has been resolved into a handle. Tests run in +/// parallel, so a caller compares this around its own call against its own +/// temporary path rather than reading a global total. +#[cfg(test)] +pub(crate) fn backing_store_resolutions(path: &Path) -> usize { + RESOLVED_BACKING_STORES + .lock() + .expect("the backing-store resolution log is never poisoned") + .iter() + .filter(|resolved| resolved.as_path() == path) + .count() +} + #[cfg(test)] mod tests { use super::*; diff --git a/epiphany-core/src/state_ledger.rs b/epiphany-core/src/state_ledger.rs index ac9f88a9..7ae104bd 100644 --- a/epiphany-core/src/state_ledger.rs +++ b/epiphany-core/src/state_ledger.rs @@ -133,7 +133,7 @@ pub fn close_state_branch( fn state_ledger_cache(store_path: &Path) -> Result { let mut cache = CultCache::new(); cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(store_path)); + cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new(store_path))?; Ok(cache) } diff --git a/epiphany-model-adapter/Cargo.toml b/epiphany-model-adapter/Cargo.toml index 304cdad2..693650ea 100644 --- a/epiphany-model-adapter/Cargo.toml +++ b/epiphany-model-adapter/Cargo.toml @@ -12,7 +12,7 @@ path = "src/lib.rs" [dependencies] anyhow = "1" codex-connector = { git = "https://github.com/GameCult/CodexConnector", rev = "be1776ab455fd79bd74b393518ab07211dc9c1f0", default-features = false } -cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultcache-rs" } serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/epiphany-openai-runtime/Cargo.toml b/epiphany-openai-runtime/Cargo.toml index 089f23ca..b3aba99c 100644 --- a/epiphany-openai-runtime/Cargo.toml +++ b/epiphany-openai-runtime/Cargo.toml @@ -13,7 +13,7 @@ path = "src/lib.rs" [dependencies] anyhow = "1" chrono = "0.4.44" -cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs" } +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "a0813c6eed24d30bf88073ef615b633c77ebfcd6", package = "cultcache-rs" } epiphany-core = { path = "../epiphany-core" } epiphany-model-adapter = { path = "../epiphany-model-adapter" } epiphany-tool-adapter = { path = "../epiphany-tool-adapter" } diff --git a/epiphany-pipeline/Cargo.toml b/epiphany-pipeline/Cargo.toml new file mode 100644 index 00000000..48d2fdd8 --- /dev/null +++ b/epiphany-pipeline/Cargo.toml @@ -0,0 +1,27 @@ +[package] +name = "epiphany-pipeline" +version = "0.1.0" +edition = "2024" +publish = false +autobins = false +license-file = "../LICENSE" + +[lib] +name = "epiphany_pipeline" +path = "src/lib.rs" + +# A leaf type library. Every dependency here is one the document types or the +# write path need in a non-test build: the organ prepares, decodes and +# validates envelopes through this crate, so `anyhow` (the cache's error type) +# and `rmp-serde` (the payload decode) are normal dependencies. Only the +# schema-publication check needs JSON. +[dependencies] +anyhow = "1" +chrono = "0.4.44" +cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "8fc74c709444b884a2bd339e0c306bf9484d99d2", package = "cultcache-rs" } +rmp-serde = "1" +schemars = "1" +serde = { version = "1", features = ["derive"] } + +[dev-dependencies] +serde_json = "1" diff --git a/epiphany-pipeline/src/lib.rs b/epiphany-pipeline/src/lib.rs new file mode 100644 index 00000000..e8697753 --- /dev/null +++ b/epiphany-pipeline/src/lib.rs @@ -0,0 +1,3018 @@ +//! Eureka pipeline documents: the typed shape of a Eureka campaign's state. +//! +//! This is a leaf type library, and deliberately nothing else. It owns document +//! shape, field bounds, formats, key derivation and the JSON schemas published +//! under `schemas/cultnet`; it owns no storage, no admission, no process and no +//! network. That is the whole reason it is a package: the organ that will admit +//! these documents depends on these types without depending on the harness +//! that used to hold them. +//! +//! Every kind is a plain `serde` + `JsonSchema` value inside a one-slot +//! `DatabaseEntry` wrapper, always prepared with `prepare_entry_named`, so the +//! stored payload is `[value]` and the value is the named map the published +//! schema describes. Keys are semantic: `pipeline_key` derives them from the +//! value, and the write validator refuses any envelope whose key differs. +//! +//! Every key is `::`, three segments for every kind with +//! none excused: the root is a `Slug`, the kind is the literal kind name, and +//! the local is `Label`s joined by `.`, so no local part carries a dot and a +//! reader recovers the parts by splitting. A root's local is the constant +//! `self`; a resolution's local is its subject's kind and local. A +//! resolution's local ends in its per-subject sequence, `n`, and a +//! stewardship's in its per-repo sequence, so a subject's resolutions and a +//! repo's assignments each share a prefix. +//! +//! **Validation follows the type.** `value_types!` is the single field list: it +//! emits the struct and its `Bounded` impl from the same tokens, so a field +//! cannot be declared without being validated. Format rules live in the field's +//! type (`Label`, `Slug`, `OrgRepo`, `Sha`, `Sha256Hex`, `Date`), never in a +//! hand-written impl that a later field can slip past. A `Vec` field must carry +//! a maximum, because `Vec` has no `Bounded` impl of its own. +//! +//! The wrappers are crate-private: outside code registers, prepares and +//! decodes them through `register_pipeline_document_types`, +//! `PipelineDocument::prepare` and `PipelineDocument::decode`, and validates a +//! write through `validate_pipeline_write_envelope`. + +// `anyhow` is the cache's own error type, which `prepare` and the registrar +// pass through; every refusal this crate decides is a `PipelineRefusal`. +use anyhow::Result; +use cultcache_rs::{CultCache, CultCacheEnvelope, DatabaseEntry}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; + +/// Typed refusals of the pipeline documents: bounds, formats and key identity +/// (D2's document half), and `ForeignStore`, raised by `decode` for an +/// envelope of any other type. This is every refusal the leaf decides; a +/// store's identity and its schema epoch are not decided here, since this +/// crate owns no store, and belong to the organ that will admit these +/// documents. +/// +/// Serialised externally tagged, serde's default for an enum: each variant is +/// a one-key map from the variant's name to its named fields, so `field` and +/// `value` are on the wire under their own names and a reader of a refusal +/// carried over a wire reads the same parts a local caller matches on. No +/// attribute buys that, which is why there is none: an internal or adjacent +/// tag would move the fields under a content key without making any of them +/// more visible. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +pub enum PipelineRefusal { + FieldBound { field: String, limit: u32, actual: u32 }, + InvalidFormat { field: String, value: String }, + InvalidIdentity { kind: PipelineKind, key: String, expected: String }, + DuplicateRepo { field: String, repo: String }, + ForeignStore { r#type: String }, +} + +impl std::fmt::Display for PipelineRefusal { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "pipeline refusal: {self:?}") + } +} + +impl std::error::Error for PipelineRefusal {} + +pub(crate) trait Bounded { + fn validate(&self, field: &str) -> Result<(), PipelineRefusal>; +} + +impl Bounded for Option { + fn validate(&self, field: &str) -> Result<(), PipelineRefusal> { + self.as_ref().map_or(Ok(()), |value| value.validate(field)) + } +} + +macro_rules! unbounded_scalars { + ($($ty:ty),* $(,)?) => {$( + impl Bounded for $ty { + fn validate(&self, _field: &str) -> Result<(), PipelineRefusal> { + Ok(()) + } + } + )*}; +} + +unbounded_scalars!(u32, u64, bool); + +fn bound(field: &str, limit: usize, actual: usize) -> Result<(), PipelineRefusal> { + if actual > limit { + return Err(PipelineRefusal::FieldBound { + field: field.into(), + limit: limit as u32, + actual: actual as u32, + }); + } + Ok(()) +} + +/// List maximums only. Minimum counts are admission rules with their own +/// refusals (Cut 3b), so a bounds pass never pre-empts them. +fn list(field: &str, items: &[T], max: usize) -> Result<(), PipelineRefusal> { + bound(field, max, items.len())?; + for (index, item) in items.iter().enumerate() { + item.validate(&format!("{field}[{index}]"))?; + } + Ok(()) +} + +fn format_error(field: &str, value: &str) -> PipelineRefusal { + PipelineRefusal::InvalidFormat { + field: field.into(), + value: value.into(), + } +} + +fn hex(field: &str, value: &str, lengths: std::ops::RangeInclusive) -> Result<(), PipelineRefusal> { + let lower_hex = value.bytes().all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f')); + if !lengths.contains(&value.len()) || !lower_hex { + return Err(format_error(field, value)); + } + Ok(()) +} + +/// A key label: `[A-Za-z0-9_-]{1,64}`. Dots are excluded so that a composed key +/// segments unambiguously; see `local`. +fn label_text(field: &str, value: &str) -> Result<(), PipelineRefusal> { + let valid = value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')); + if !valid || value.is_empty() || value.len() > 64 { + return Err(format_error(field, value)); + } + Ok(()) +} + +/// A dotted key segment: label parts joined by `.`, at most 64 bytes. Every +/// part must be a label, so `..`, a leading or trailing dot, and an empty part +/// are all refused. +fn dotted_text(field: &str, value: &str) -> Result<(), PipelineRefusal> { + if value.is_empty() || value.len() > 64 { + return Err(format_error(field, value)); + } + for part in value.split('.') { + label_text(field, part)?; + } + Ok(()) +} + +/// GitHub's own grammar for `owner/repo`, exactly one `/`: the owner is +/// `[A-Za-z0-9-]`, 1 to 39 bytes, with no leading or trailing hyphen; the repo +/// is `[A-Za-z0-9._-]`, 1 to 100 bytes, is never `.` or `..`, and does not end +/// in `.git`, ASCII-case-insensitively (`a/b.GIT` is refused exactly as +/// `a/b.git` is), so the suffix rule cannot be dodged by a case variant whose +/// `identity()` is refused. The repo's own byte alphabet excludes `/`, so a +/// second `/` inside it is already refused by the byte check; no separate +/// `contains('/')` check is needed. +fn org_repo_text(field: &str, value: &str) -> Result<(), PipelineRefusal> { + let valid = value.split_once('/').is_some_and(|(owner, repo)| { + (1..=39).contains(&owner.len()) + && owner.bytes().all(|byte| byte.is_ascii_alphanumeric() || byte == b'-') + && !owner.starts_with('-') + && !owner.ends_with('-') + && (1..=100).contains(&repo.len()) + && repo.bytes().all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-')) + && repo != "." + && repo != ".." + && !(repo.len() >= 4 && repo[repo.len() - 4..].eq_ignore_ascii_case(".git")) + }); + if valid { + Ok(()) + } else { + Err(format_error(field, value)) + } +} + +macro_rules! bounded_text { + ($($(#[$doc:meta])* $name:ident = $limit:literal, $check:expr;)*) => {$( + #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema)] + #[serde(transparent)] + #[schemars(extend("maxLength" = $limit))] + $(#[$doc])* + pub struct $name(pub String); + impl Bounded for $name { + fn validate(&self, field: &str) -> Result<(), PipelineRefusal> { + let check: fn(&str, &str) -> Result<(), PipelineRefusal> = $check; + check(field, &self.0) + } + } + impl From<&str> for $name { + fn from(value: &str) -> Self { + Self(value.into()) + } + } + )*}; +} + +fn within(limit: usize) -> impl Fn(&str, &str) -> Result<(), PipelineRefusal> { + move |field: &str, value: &str| bound(field, limit, value.len()) +} + +/// A title holds at least one `char::is_alphanumeric()` character, no +/// `char::is_control()` character, no U+2028 LINE SEPARATOR or U+2029 +/// PARAGRAPH SEPARATOR (both `Zl`/`Zp`, so `is_control()` does not reach +/// them, and a one-line field should carry no hard line break), and no bidi +/// control character: `U+061C`, `U+200E`, `U+200F`, `U+202A`-`U+202E`, +/// `U+2066`-`U+2069`. No denylist of individual zero-width or format +/// characters is maintained here; a character not covered by one of these +/// tests is allowed, so ZWJ and ZWNJ still work. +fn title_text(field: &str, value: &str) -> Result<(), PipelineRefusal> { + let alphanumeric = value.chars().any(|character| character.is_alphanumeric()); + let control = value + .chars() + .any(|character| character.is_control() || matches!(character, '\u{2028}' | '\u{2029}')); + let bidi = value.chars().any(|character| { + matches!( + character, + '\u{061C}' | '\u{200E}' | '\u{200F}' | '\u{202A}'..='\u{202E}' | '\u{2066}'..='\u{2069}' + ) + }); + if !alphanumeric || control || bidi { + return Err(format_error(field, value)); + } + within(200)(field, value) +} + +bounded_text! { + /// Text of at most 200 UTF-8 bytes. + Short = 200, |field, value| within(200)(field, value); + /// A title: 1 to 200 UTF-8 bytes, enforced by admission; JSON Schema + /// length counts characters. Holds at least one alphanumeric character, + /// no control character, and no bidi control character (`U+061C`, + /// `U+200E`, `U+200F`, `U+202A`-`U+202E`, `U+2066`-`U+2069`). + #[schemars(extend("minLength" = 1))] + Title = 200, title_text; + /// Text of at most 1,000 UTF-8 bytes. + Line = 1000, |field, value| within(1000)(field, value); + /// Text of at most 4,000 UTF-8 bytes; longer narrative is cited by `DocRef`. + Para = 4000, |field, value| within(4000)(field, value); + /// A key label: `[A-Za-z0-9_-]{1,64}`. + Label = 64, label_text; + /// A campaign slug or key segment: label parts joined by `.`. + Slug = 64, dotted_text; + /// A git commit id: 7-40 lowercase hex characters. + Sha = 40, |field, value| hex(field, value, 7..=40); + /// A full git commit id: exactly 40 lowercase hex characters. + FullSha = 40, |field, value| hex(field, value, 40..=40); + /// A SHA-256 digest: exactly 64 lowercase hex characters. + Sha256Hex = 64, |field, value| hex(field, value, 64..=64); +} + +impl Slug { + /// The public door onto the slug grammar, for a caller outside this crate + /// that holds a bare `Slug` and no way to reach `Bounded`, which is + /// crate-private. On the pattern of `PipelineRef::validate_ref`: it + /// delegates so the grammar keeps one owner, and it applies exactly the + /// `dotted_text` check that every `Slug` field in this crate is held to, + /// so a caller that validates a declared name this way and a document + /// field of the same type refuse the same inputs. Named `validate_slug` + /// rather than a bare `validate`, because an inherent `validate` would + /// shadow `Bounded`'s for every in-crate caller holding a `Slug`. + pub fn validate_slug(&self) -> Result<(), PipelineRefusal> { + Bounded::validate(self, "slug") + } +} + +/// A GitHub repository as `Org/Repo`, at most 140 UTF-8 bytes (39 + 1 + +/// 100). The owner is 1 to 39 `[A-Za-z0-9-]` bytes with no leading or +/// trailing hyphen; the repo is 1 to 100 `[A-Za-z0-9._-]` bytes, is never +/// `.` or `..`, and does not end in `.git`, ASCII-case-insensitively. +/// Identity is case-insensitive: `GameCult/Epiphany` and `gamecult/epiphany` +/// are the same repository. +// +// The raw string is display-only. `OrgRepo` is not generated by +// `bounded_text!`: that macro derives `PartialEq`, `Eq`, `Hash`, `PartialOrd` +// and `Ord` on the raw string, with no per-member override, so a type built +// from it compares and keys on spelling. `OrgRepo` instead carries +// hand-written impls of those five traits, every one of them over +// `identity()` (the ASCII-lowercased value), so `==`, a `BTreeMap` key and a +// `HashMap` key are all case-insensitive by construction: `GameCult/Epiphany` +// and `gamecult/epiphany` compare equal, hash equal, order together and +// collapse to one map entry, and a caller cannot reach the case-sensitive +// behaviour by accident. `Hash` and `Eq` agree because both are defined over +// the same `identity()` string. Every other bounded type keeps the macro's +// derives untouched. This reasoning is a source comment, not a doc comment, +// so it does not enter the published schema description (RS-L closing fix 2): +// a consumer needs the grammar, the bound and the case-insensitivity, not the +// Rust-macro justification for how that got built. +#[derive(Clone, Debug, Default, Serialize, Deserialize, JsonSchema)] +#[serde(transparent)] +#[schemars(extend("maxLength" = 140))] +pub struct OrgRepo(pub String); + +impl Bounded for OrgRepo { + fn validate(&self, field: &str) -> Result<(), PipelineRefusal> { + org_repo_text(field, &self.0) + } +} + +impl From<&str> for OrgRepo { + fn from(value: &str) -> Self { + Self(value.into()) + } +} + +impl PartialEq for OrgRepo { + fn eq(&self, other: &Self) -> bool { + self.identity() == other.identity() + } +} + +impl Eq for OrgRepo {} + +impl std::hash::Hash for OrgRepo { + fn hash(&self, state: &mut H) { + self.identity().hash(state); + } +} + +impl PartialOrd for OrgRepo { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for OrgRepo { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.identity().cmp(&other.identity()) + } +} + +impl OrgRepo { + /// The public door onto the `org_repo` grammar, for a caller outside this + /// crate that holds a bare `OrgRepo` (the organ's `repo` alias, RS-3) and + /// no way to reach `Bounded`. On the pattern of `Slug::validate_slug`: it + /// delegates to `org_repo_text` through `Bounded::validate`, so the + /// grammar keeps one owner and a declared name checked this way refuses + /// exactly what a document's own `OrgRepo` field refuses. + pub fn validate_org_repo(&self) -> Result<(), PipelineRefusal> { + Bounded::validate(self, "org_repo") + } + + /// The one canonical key for an `OrgRepo`: the value, ASCII-lowercased. + /// Every comparison, hash, ordering and lookup keyed by repository + /// identity goes through this — `PartialEq`, `Hash` and `Ord` above are + /// all defined in terms of it — so `GameCult/Epiphany` and + /// `gamecult/epiphany` are the same repository everywhere, not only where + /// a caller remembers to ask. + pub fn identity(&self) -> String { + self.0.to_ascii_lowercase() + } +} + +impl Label { + /// The public door onto the label grammar, for a caller outside this + /// crate that holds a bare `Label` (the organ's `cut` alias, RS-3) and no + /// way to reach `Bounded`. On the pattern of `Slug::validate_slug`: it + /// delegates to `label_text` through `Bounded::validate`, so the grammar + /// keeps one owner and a declared name checked this way refuses exactly + /// what a document's own `Label` field refuses. + pub fn validate_label(&self) -> Result<(), PipelineRefusal> { + Bounded::validate(self, "label") + } +} + +/// A calendar date, `YYYY-MM-DD`. +#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema)] +#[serde(transparent)] +#[schemars(extend("maxLength" = 10))] +pub struct Date(pub String); + +impl Bounded for Date { + fn validate(&self, field: &str) -> Result<(), PipelineRefusal> { + match chrono::NaiveDate::parse_from_str(&self.0, "%Y-%m-%d") { + Ok(_) if self.0.len() == 10 => Ok(()), + _ => Err(format_error(field, &self.0)), + } + } +} + +macro_rules! bounded_field { + ($value:expr, $field:expr) => { + Bounded::validate(&$value, $field)? + }; + ($value:expr, $field:expr, $max:literal) => { + list($field, &$value, $max)? + }; +} + +/// The single field list: one invocation emits the value struct and its +/// `Bounded` impl, so a field can never be left out of validation. +macro_rules! value_types { + ($($(#[$attr:meta])* pub struct $name:ident { $($field:ident: $ty:ty $([$max:literal])?),* $(,)? } $(=> $extra:path)?)*) => {$( + $(#[$attr])* + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] + pub struct $name { + $( + $(#[schemars(extend("maxItems" = $max))])? + pub $field: $ty + ),* + } + + impl Bounded for $name { + fn validate(&self, at: &str) -> Result<(), PipelineRefusal> { + $(bounded_field!(self.$field, &format!("{at}.{}", stringify!($field)) $(, $max)?);)* + $($extra(self, at)?;)? + Ok(()) + } + } + )*}; +} + +macro_rules! unit_enums { + ($($name:ident { $($variant:ident),* $(,)? })*) => {$( + #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize, JsonSchema)] + pub enum $name { $($variant),* } + impl Bounded for $name { + fn validate(&self, _field: &str) -> Result<(), PipelineRefusal> { + Ok(()) + } + } + )*}; +} + +/// RS-L closing fix 4: `campaign.repos` refuses two entries naming the same +/// repository by identity, not by raw spelling, so `GameCult/Epiphany` and +/// `gamecult/epiphany` in one list refuse exactly as one entry twice would. +/// This is a cross-element invariant on `PipelineCampaign` alone, not a +/// per-type format rule any other `value_types!` struct shares, so it is +/// wired in as that struct's one `=> extra` hook rather than folded into +/// `Bounded for OrgRepo` or the list bound in `bounded_field!`. +fn campaign_repos_are_distinct(campaign: &PipelineCampaign, at: &str) -> Result<(), PipelineRefusal> { + let mut seen = std::collections::HashSet::new(); + for repo in &campaign.repos { + if !seen.insert(repo.identity()) { + return Err(PipelineRefusal::DuplicateRepo { + field: format!("{at}.repos"), + repo: repo.0.clone(), + }); + } + } + Ok(()) +} + +unit_enums! { + EvidenceKind { Command, Test, Mutation, Probe, SourceRead, Capture } + ClaimOutcome { Holds, Falsified, Unproven } + FindingConfidence { Confirmed, Plausible } + FindingSeverity { Blocker, High, Medium, Low } + RulingAuthority { Operator, Standing, Defaulted } + FindingOrigin { Introduced, PreExisting } +} + +value_types! { + pub struct CodeLocation { path: Short, line: u32, end_line: Option } + pub struct CommitRange { base: Sha, head: Sha } + pub struct Evidence { kind: EvidenceKind, locator: Line, result: Line } + /// A citation of a document in another repo's pipeline store (D6). + pub struct ForeignRef { repo: OrgRepo, commit: FullSha, kind: PipelineKind, id: Short, payload_sha256: Sha256Hex } + pub struct DocRef { path: Short, start_line: u32, end_line: u32, commit: Sha } + pub struct TargetInvariant { label: Label, statement: Line } + pub struct QuestionOption { label: Label, text: Line } + pub struct CutDelete { path: Short, lines: u32, note: Line } + pub struct FileChange { location: CodeLocation, change: Line } + pub struct AuthorityMap { + owner: Line, inputs: Vec[16], outputs: Vec[16], derived_state: Vec[16], + forbidden_writers: Vec[16], shared_paths: Vec[16], deletion_line: Line, + } + pub struct VerificationTest { name: Short, pins: Line } + pub struct NegativeCheck { pattern: Short, scope: Line } + pub struct CutVerification { + builds: Vec[64], tests: Vec[64], + negative: Vec[64], operator: Vec[64], + } + pub struct ReportCommit { sha: Sha, subject: Line, builds: bool } + /// One mutation a report ran: a key-safe label a verdict claim can name, + /// the exact edit, and the tree it was applied to. + pub struct MutationRecord { label: Label, rule: Line, location: CodeLocation, before: Line, after: Line, commit: Sha, failed_as_expected: bool } + pub struct Deviation { what: Line, why: Line } + // D1 tables no maximum for these lists, so they take the shared list + // default of 64 rather than a number invented for this field alone. + pub struct StructuralDelta { + lines_added: u32, lines_removed: u32, + dependencies_added: Vec[64], dependencies_removed: Vec[64], + formats_added: Vec[64], formats_removed: Vec[64], + targets_added: Vec[64], targets_removed: Vec[64], + } + /// A name the cut landed, and where it lives. + pub struct LandedName { name: Short, path: Short } + /// A promise a report makes about what it landed. Not a document and not + /// resolvable: it is identified by its report's id and its label, as a + /// `TargetInvariant` is by its target's. Soul measures every one (ruling + /// A), and the rule that every one is measured is admission's. + pub struct Promise { label: Label, text: Line } + /// The promise this claim measured, and the report's mutations it ran. An + /// `Unproven` claim with a promise is one Soul could not reach. + pub struct VerdictClaim { + claim: Line, outcome: ClaimOutcome, evidence: Vec[8], findings: Vec[16], + promise: Option