From 5d6a7316079cad39e2df3c74effa5fa93cf546ac Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 8 Aug 2026 19:07:43 +0100 Subject: [PATCH] runtime: add keyed spine migration and narrow release builds --- epiphany-core/Cargo.toml | 4 + .../src/bin/epiphany-runtime-store-migrate.rs | 25 +++ .../src/coordinator_state_transaction.rs | 7 +- epiphany-core/src/lib.rs | 6 + epiphany-core/src/packaged_release.rs | 12 +- epiphany-core/src/runtime_spine.rs | 10 +- epiphany-core/src/runtime_store_backend.rs | 129 +++++++++++++ epiphany-core/src/runtime_store_migration.rs | 178 ++++++++++++++++++ notes/fresh-workspace-handoff.md | 36 ++++ state/ledgers.msgpack | Bin 576425 -> 576858 bytes state/map.yaml | 8 +- 11 files changed, 402 insertions(+), 13 deletions(-) create mode 100644 epiphany-core/src/bin/epiphany-runtime-store-migrate.rs create mode 100644 epiphany-core/src/runtime_store_backend.rs create mode 100644 epiphany-core/src/runtime_store_migration.rs diff --git a/epiphany-core/Cargo.toml b/epiphany-core/Cargo.toml index 8edd5d0c..2b85327d 100644 --- a/epiphany-core/Cargo.toml +++ b/epiphany-core/Cargo.toml @@ -272,6 +272,10 @@ path = "src/bin/epiphany-rider-bridge-smoke.rs" name = "epiphany-runtime-spine" path = "src/bin/epiphany-runtime-spine.rs" +[[bin]] +name = "epiphany-runtime-store-migrate" +path = "src/bin/epiphany-runtime-store-migrate.rs" + [[bin]] name = "epiphany-state" path = "src/bin/epiphany-state.rs" diff --git a/epiphany-core/src/bin/epiphany-runtime-store-migrate.rs b/epiphany-core/src/bin/epiphany-runtime-store-migrate.rs new file mode 100644 index 00000000..b7cd385c --- /dev/null +++ b/epiphany-core/src/bin/epiphany-runtime-store-migrate.rs @@ -0,0 +1,25 @@ +use anyhow::{Result, anyhow}; +use epiphany_core::migrate_runtime_store_to_redb; + +fn main() -> Result<()> { + let mut args = std::env::args().skip(1); + let source = args.next().ok_or_else(|| anyhow!("usage: epiphany-runtime-store-migrate "))?; + let destination = args + .next() + .ok_or_else(|| anyhow!("missing destination.redb"))?; + let migrated_at = args + .next() + .ok_or_else(|| anyhow!("missing migrated-at-rfc3339"))?; + if args.next().is_some() { + return Err(anyhow!("unexpected extra migration arguments")); + } + let receipt = migrate_runtime_store_to_redb(source, destination, &migrated_at)?; + println!( + "{}", + rmp_serde::to_vec_named(&receipt)? + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::() + ); + Ok(()) +} diff --git a/epiphany-core/src/coordinator_state_transaction.rs b/epiphany-core/src/coordinator_state_transaction.rs index 9affb4e1..2bc4de17 100644 --- a/epiphany-core/src/coordinator_state_transaction.rs +++ b/epiphany-core/src/coordinator_state_transaction.rs @@ -1,10 +1,13 @@ use crate::{ EpiphanyThreadStateEntry, THREAD_STATE_KEY, THREAD_STATE_TYPE, coordinator_acceptance_cache, }; +use crate::runtime_store_backend::runtime_spine_backing_store; +#[cfg(test)] +use crate::runtime_store_backend::RuntimeSpineBackingStore as SingleFileMessagePackBackingStore; use anyhow::{Result, anyhow}; #[cfg(test)] use cultcache_rs::CacheBackingStore; -use cultcache_rs::{CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore}; +use cultcache_rs::{CultCache, CultCacheEnvelope}; use epiphany_state_model::EpiphanyThreadState; use std::ops::{Deref, DerefMut}; use std::path::{Path, PathBuf}; @@ -126,7 +129,7 @@ pub(crate) fn commit_coordinator_state_transaction( replacements.push(replacement); } - let backing = SingleFileMessagePackBackingStore::new(&transaction.store); + let backing = runtime_spine_backing_store(&transaction.store)?; if !backing.compare_and_swap_batch(&expected, replacements)? { return Err(anyhow!( "coordinator state transaction lost its exact atomic compare-and-swap" diff --git a/epiphany-core/src/lib.rs b/epiphany-core/src/lib.rs index cf197feb..681d6208 100644 --- a/epiphany-core/src/lib.rs +++ b/epiphany-core/src/lib.rs @@ -41,6 +41,8 @@ mod repository_readiness; mod resident_readiness; mod resident_self; mod runtime_spine; +mod runtime_store_backend; +mod runtime_store_migration; mod semantic_backend; mod semantic_projector_service; mod soul_gateway; @@ -1105,6 +1107,10 @@ pub use runtime_spine::select_repo_frontier_work_proposal_for_modeling; pub use runtime_spine::validate_hands_action_authority; pub use runtime_spine::write_runtime_hello_frame; pub use runtime_spine::write_runtime_schema_catalog_json; +pub use runtime_store_migration::{ + RuntimeStoreMigrationReceipt, RUNTIME_STORE_MIGRATION_RECEIPT_SCHEMA_VERSION, + RUNTIME_STORE_MIGRATION_RECEIPT_TYPE, migrate_runtime_store_to_redb, +}; pub use semantic_projector_service::SemanticProjectorServiceBody; pub use semantic_projector_service::SemanticProjectorServicePulse; pub use semantic_projector_service::validate_semantic_projector_source_pair; diff --git a/epiphany-core/src/packaged_release.rs b/epiphany-core/src/packaged_release.rs index f119671c..d5bc6cfe 100644 --- a/epiphany-core/src/packaged_release.rs +++ b/epiphany-core/src/packaged_release.rs @@ -456,8 +456,12 @@ fn build_required_release_siblings( })?; let mut outputs = BTreeMap::new(); let required = required_packaged_release_binaries(target); + let mut first_pass_bins = BTreeSet::new(); for (role, file_name) in &required { - required_release_build_target(role)?; + let (_, binary) = required_release_build_target(role)?; + if *role != "coordinator" { + first_pass_bins.insert(binary); + } outputs.insert( *role, target_dir.join(target).join("release").join(file_name), @@ -473,8 +477,10 @@ fn build_required_release_siblings( .arg(&target_dir) .arg("--target") .arg(target) - .arg("--locked") - .arg("--bins"); + .arg("--locked"); + for binary in first_pass_bins { + command.arg("--bin").arg(binary); + } let status = command .status() .context("failed to start Epiphany release bundle build")?; diff --git a/epiphany-core/src/runtime_spine.rs b/epiphany-core/src/runtime_spine.rs index 2dae6a2d..c9287b7d 100644 --- a/epiphany-core/src/runtime_spine.rs +++ b/epiphany-core/src/runtime_spine.rs @@ -1,5 +1,9 @@ use crate::EpiphanyWorkerLaunchDocument; use crate::RepoFrontierPlanMindContextProjection; +use crate::runtime_store_backend::{ + RuntimeSpineBackingStore as SingleFileMessagePackBackingStore, + runtime_spine_backing_store, +}; use crate::agent_launch::{ EPIPHANY_IMAGINATION_OWNER_ROLE, EPIPHANY_IMAGINATION_ROLE_BINDING_ID, EPIPHANY_MIND_OWNER_ROLE, EPIPHANY_MIND_ROLE_BINDING_ID, EPIPHANY_MODELING_OWNER_ROLE, @@ -116,7 +120,6 @@ use cultcache_rs::CacheBackingStore; use cultcache_rs::CultCache; use cultcache_rs::CultCacheEnvelope; use cultcache_rs::DatabaseEntry; -use cultcache_rs::SingleFileMessagePackBackingStore; use cultnet_rs::CultNetDocumentMutationContract; use cultnet_rs::CultNetDocumentOperation; use cultnet_rs::CultNetMessage; @@ -905,6 +908,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef) -> Result { cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; + cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; @@ -983,9 +987,7 @@ pub fn runtime_spine_cache(store_path: impl AsRef) -> Result { cache.register_entry_type::()?; cache.register_entry_type::()?; cache.register_entry_type::()?; - cache.add_generic_backing_store(SingleFileMessagePackBackingStore::new( - store_path.to_path_buf(), - )); + cache.add_generic_backing_store(runtime_spine_backing_store(store_path)?); Ok(cache) } diff --git a/epiphany-core/src/runtime_store_backend.rs b/epiphany-core/src/runtime_store_backend.rs new file mode 100644 index 00000000..8c110a2e --- /dev/null +++ b/epiphany-core/src/runtime_store_backend.rs @@ -0,0 +1,129 @@ +use anyhow::{Result, anyhow}; +use cultcache_rs::{ + CacheBackingStore, CultCacheEnvelope, PushAllOptions, RedbMessagePackBackingStore, + SingleFileMessagePackBackingStore, +}; +use std::path::{Path, PathBuf}; + +#[derive(Clone, Debug)] +pub(crate) struct RuntimeSpineBackingStore { + path: PathBuf, +} + +enum SelectedBackingStore { + Snapshot(SingleFileMessagePackBackingStore), + Keyed(RedbMessagePackBackingStore), +} + +impl RuntimeSpineBackingStore { + pub(crate) fn new(path: impl Into) -> Self { + Self { path: path.into() } + } + + fn selected(&self) -> Result { + match self + .path + .extension() + .and_then(|extension| extension.to_str()) + { + Some("cc" | "msgpack") => Ok(SelectedBackingStore::Snapshot( + SingleFileMessagePackBackingStore::new(&self.path), + )), + Some("redb") => Ok(SelectedBackingStore::Keyed( + RedbMessagePackBackingStore::new(&self.path)?, + )), + extension => Err(anyhow!( + "runtime spine store must use an explicit .cc, .msgpack, or .redb extension; got {:?}", + extension + )), + } + } + + pub(crate) fn compare_and_swap_batch( + &self, + expected: &[CultCacheEnvelope], + replacements: Vec, + ) -> Result { + match self.selected()? { + SelectedBackingStore::Snapshot(store) => { + store.compare_and_swap_batch(expected, replacements) + } + SelectedBackingStore::Keyed(store) => { + store.compare_and_swap_batch(expected, replacements) + } + } + } +} + +impl CacheBackingStore for RuntimeSpineBackingStore { + fn pull_all(&self) -> Result> { + match self.selected()? { + SelectedBackingStore::Snapshot(store) => store.pull_all(), + SelectedBackingStore::Keyed(store) => store.pull_all(), + } + } + + fn push(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + match self.selected()? { + SelectedBackingStore::Snapshot(mut store) => store.push(entry), + SelectedBackingStore::Keyed(mut store) => store.push(entry), + } + } + + fn delete(&mut self, entry: &CultCacheEnvelope) -> Result<()> { + match self.selected()? { + SelectedBackingStore::Snapshot(mut store) => store.delete(entry), + SelectedBackingStore::Keyed(mut store) => store.delete(entry), + } + } + + fn push_all(&mut self, entries: &[CultCacheEnvelope], options: PushAllOptions) -> Result<()> { + match self.selected()? { + SelectedBackingStore::Snapshot(mut store) => store.push_all(entries, options), + SelectedBackingStore::Keyed(mut store) => store.push_all(entries, options), + } + } +} + +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)), + extension => Err(anyhow!( + "runtime spine store must use an explicit .cc, .msgpack, or .redb extension; got {:?}", + extension + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn envelope(payload: u8) -> CultCacheEnvelope { + CultCacheEnvelope { + key: "state".to_string(), + r#type: "epiphany.test".to_string(), + payload: vec![payload], + stored_at: "2026-08-08T00:00:00Z".to_string(), + schema_id: Some("epiphany.test".to_string()), + } + } + + #[test] + fn explicit_extensions_select_one_backend_and_preserve_cas() -> Result<()> { + let temp = tempfile::tempdir()?; + for file_name in ["runtime.cc", "runtime.redb"] { + let path = temp.path().join(file_name); + let store = RuntimeSpineBackingStore::new(path); + let first = envelope(1); + let second = envelope(2); + assert!(store.compare_and_swap_batch(&[], vec![first.clone()])?); + assert_eq!(store.pull_all()?, vec![first.clone()], "{file_name}"); + assert!(store.compare_and_swap_batch(&[first.clone()], vec![second.clone()])?); + assert!(!store.compare_and_swap_batch(&[first], vec![envelope(3)])?); + assert_eq!(store.pull_all()?, vec![second]); + } + assert!(runtime_spine_backing_store(&temp.path().join("runtime.bin")).is_err()); + Ok(()) + } +} diff --git a/epiphany-core/src/runtime_store_migration.rs b/epiphany-core/src/runtime_store_migration.rs new file mode 100644 index 00000000..7c2442a0 --- /dev/null +++ b/epiphany-core/src/runtime_store_migration.rs @@ -0,0 +1,178 @@ +use anyhow::{Result, anyhow}; +use cultcache_rs::{ + CacheBackingStore, CultCacheEnvelope, DatabaseEntry, RedbMessagePackBackingStore, + SingleFileMessagePackBackingStore, +}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::collections::BTreeSet; +use std::fs; +use std::path::{Path, PathBuf}; + +pub const RUNTIME_STORE_MIGRATION_RECEIPT_TYPE: &str = "epiphany.runtime_store_migration_receipt"; +pub const RUNTIME_STORE_MIGRATION_RECEIPT_SCHEMA_VERSION: &str = + "epiphany.runtime_store_migration_receipt.v1"; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct RuntimeStoreMigrationReceipt { + pub schema_version: String, + pub migration_id: String, + pub source_path: String, + pub source_sha256: String, + pub source_entry_count: usize, + pub source_envelope_set_sha256: String, + pub destination_path: String, + pub migrated_at: String, +} + +impl DatabaseEntry for RuntimeStoreMigrationReceipt { + const TYPE: &'static str = RUNTIME_STORE_MIGRATION_RECEIPT_TYPE; + const SCHEMA_NAME: &'static str = "RuntimeStoreMigrationReceipt"; +} + +fn sha256(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + +fn sorted_envelopes(entries: &[CultCacheEnvelope]) -> Vec { + let mut sorted = entries.to_vec(); + sorted.sort_by(|left, right| (&left.r#type, &left.key).cmp(&(&right.r#type, &right.key))); + sorted +} + +fn envelope_set_sha256(entries: &[CultCacheEnvelope]) -> Result { + Ok(sha256(&rmp_serde::to_vec_named(&sorted_envelopes( + entries, + ))?)) +} + +pub fn migrate_runtime_store_to_redb( + source: impl AsRef, + destination: impl AsRef, + migrated_at: &str, +) -> Result { + chrono::DateTime::parse_from_rfc3339(migrated_at) + .map_err(|_| anyhow!("migration timestamp must be RFC3339"))?; + let source = source.as_ref(); + let destination = destination.as_ref(); + if source.extension().and_then(|value| value.to_str()) != Some("cc") { + return Err(anyhow!("migration source must be a legacy .cc snapshot")); + } + if destination.extension().and_then(|value| value.to_str()) != Some("redb") { + return Err(anyhow!("migration destination must use .redb")); + } + if destination.exists() { + return Err(anyhow!("migration destination already exists")); + } + + let source_bytes = fs::read(source)?; + let source_sha256 = sha256(&source_bytes); + let source_store = SingleFileMessagePackBackingStore::new(source); + let entries = source_store.pull_all()?; + let identities = entries + .iter() + .map(|entry| (&entry.r#type, &entry.key)) + .collect::>(); + if identities.len() != entries.len() { + return Err(anyhow!( + "legacy runtime snapshot contains duplicate typed identities" + )); + } + if entries + .iter() + .any(|entry| entry.r#type == RUNTIME_STORE_MIGRATION_RECEIPT_TYPE) + { + return Err(anyhow!( + "legacy runtime snapshot already contains a migration receipt" + )); + } + + let set_sha256 = envelope_set_sha256(&entries)?; + let migration_id = format!("runtime-store-migration-{source_sha256}"); + let receipt = RuntimeStoreMigrationReceipt { + schema_version: RUNTIME_STORE_MIGRATION_RECEIPT_SCHEMA_VERSION.to_string(), + migration_id: migration_id.clone(), + source_path: source.canonicalize()?.display().to_string(), + source_sha256: source_sha256.clone(), + source_entry_count: entries.len(), + source_envelope_set_sha256: set_sha256.clone(), + destination_path: PathBuf::from(destination).display().to_string(), + migrated_at: migrated_at.to_string(), + }; + let receipt_envelope = CultCacheEnvelope { + key: migration_id, + r#type: RUNTIME_STORE_MIGRATION_RECEIPT_TYPE.to_string(), + payload: rmp_serde::to_vec_named(&receipt)?, + stored_at: migrated_at.to_string(), + schema_id: Some(RUNTIME_STORE_MIGRATION_RECEIPT_SCHEMA_VERSION.to_string()), + }; + let destination_store = RedbMessagePackBackingStore::new(destination)?; + let mut writes = entries.clone(); + writes.push(receipt_envelope.clone()); + if !destination_store.compare_and_swap_batch(&[], writes)? { + return Err(anyhow!( + "keyed runtime migration lost its empty-destination CAS" + )); + } + let mut migrated = destination_store.pull_all()?; + let persisted_receipt = migrated + .iter() + .find(|entry| entry.r#type == RUNTIME_STORE_MIGRATION_RECEIPT_TYPE) + .cloned() + .ok_or_else(|| anyhow!("keyed runtime migration lost its receipt"))?; + if persisted_receipt != receipt_envelope { + return Err(anyhow!( + "keyed runtime migration receipt changed during persistence" + )); + } + migrated.retain(|entry| entry.r#type != RUNTIME_STORE_MIGRATION_RECEIPT_TYPE); + if sorted_envelopes(&migrated) != sorted_envelopes(&entries) + || envelope_set_sha256(&migrated)? != set_sha256 + { + return Err(anyhow!( + "keyed runtime migration failed exact envelope equivalence" + )); + } + if sha256(&fs::read(source)?) != source_sha256 { + return Err(anyhow!("legacy runtime snapshot changed during migration")); + } + Ok(receipt) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn migration_is_exact_seals_source_and_refuses_destination_reuse() -> Result<()> { + let temp = tempfile::tempdir()?; + let source = temp.path().join("runtime.cc"); + let destination = temp.path().join("runtime.redb"); + let entry = CultCacheEnvelope { + key: "state".to_string(), + r#type: "epiphany.test".to_string(), + payload: vec![1, 2, 3], + stored_at: "2026-08-08T00:00:00Z".to_string(), + schema_id: Some("epiphany.test".to_string()), + }; + let source_store = SingleFileMessagePackBackingStore::new(&source); + assert!(source_store.compare_and_swap_batch(&[], vec![entry.clone()])?); + let before = fs::read(&source)?; + let receipt = migrate_runtime_store_to_redb(&source, &destination, "2026-08-08T01:00:00Z")?; + assert_eq!(receipt.source_entry_count, 1); + assert_eq!(fs::read(&source)?, before); + assert!( + migrate_runtime_store_to_redb(&source, &destination, "2026-08-08T02:00:00Z").is_err() + ); + let rows = RedbMessagePackBackingStore::new(&destination)?.pull_all()?; + assert!(rows.contains(&entry)); + assert_eq!( + rows.iter() + .filter(|row| row.r#type == RUNTIME_STORE_MIGRATION_RECEIPT_TYPE) + .count(), + 1 + ); + Ok(()) + } +} diff --git a/notes/fresh-workspace-handoff.md b/notes/fresh-workspace-handoff.md index 9b9cf214..8f53aa3b 100644 --- a/notes/fresh-workspace-handoff.md +++ b/notes/fresh-workspace-handoff.md @@ -150,6 +150,42 @@ probe reported `Removed 0 files` and is rejected, not evidence. - Fresh Modeling job commit: 51.637s; coordinator total 51.865s. - The launch wound is the whole-store transaction, not context assembly. +## Keyed runtime and packaging cut in progress + +The current uncommitted pass after `b070e121` introduces one runtime-spine +backend owner selected by explicit extension: `.cc`/`.msgpack` retain the +sealed legacy snapshot implementation and `.redb` selects CultCache's keyed +redb implementation. All 29 runtime-spine CAS construction sites and the +coordinator state transaction now resolve through that owner; unrelated stores +retain their existing authorities. Focused backend parity proves successful and +stale CAS on both formats. All eight coordinator transaction tests pass. + +The one-time migration refuses an existing destination, verifies unique typed +identities, writes source envelopes plus a typed in-store migration receipt in +one empty-destination CAS, reads back exact envelope equivalence, and rehashes +the source bytes to prove the legacy evidence was not changed. A disposable +migration of live v53 copied 9,376 envelopes from source SHA-256 +`78eb340a028c88d42770345518d810841029411510a2abab233d9a89761dd5e4`; the +sorted envelope-set SHA-256 is +`bbe288e348ab3a6c8af6d64d205fe818a59218988a493f9751ed65588f2b338e`. +Artifacts and the receipt are under +`.epiphany-run/storage-benchmark-20260808-2/`; authoritative v53 was not +mutated. + +Release-mode live-sized measurements are mixed but promising: legacy full read +563.70ms versus keyed 718.18ms; legacy event mutation 896.91ms versus keyed +363.61ms. Debug redb timings were misleadingly slow and are rejected for +production judgment. Do not migrate live from this proxy alone. Measure the +exact coordinator launch replacement set against the keyed v53 copy next. + +Packaging no longer requests Cargo `--bins`, which linked every diagnostic and +smoke executable after core edits. It now requests only the authenticated +release sibling binaries, leaving the feature-gated coordinator in its existing +second phase. The focused 17 packaging tests pass. A deliberately detached +release benchmark that missed the shared target directory took 4m41s, proving +that target-cache location must remain tool-owned rather than operator memory; +the authenticated packager already owns a stable graph cache explicitly. + ## Open readiness work - cut the measured 51.482s whole-store Modeling job-commit seam; diff --git a/state/ledgers.msgpack b/state/ledgers.msgpack index 53e7394b71f2c877bb466301740ec7d80d8654ed..0bec8ad022f67aca78eb77dbd42df5239e591c87 100644 GIT binary patch delta 480 zcmXX?OHRWu5GCciVS^Amq>;L%QQL_0!zG}wV&^zRV%4#Y?WAN?Ar>3}DkN^e4v7Wg z0!T=lfKzY;CZ&>}J(_ub^JZS2+CR_j*MnepIPMO{(ctoPZRdON)()E)D(MEjtVCrMTMQ}I0CLs4TZ=oTmCRsSD3=+C>q4XXWRp-3Z*va zU?zM{>P!;D>2hv40;7h$Hk7f*Q5YY>g!(gwKw3AXO>t?dKE;~uqV)frqBsiUUg4lZ z>q-X_GsbHej^c1c=_5!=zqARC`trOPrZJGCOu^8cVy?24MXt3?g<@XYj>}o&CPBsX zg4t;Zr?eNA31TZfTW!aIR