diff --git a/Cargo.lock b/Cargo.lock
index 5b9d7ed73..b0c3844d7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6484,6 +6484,7 @@ dependencies = [
"criterion",
"serde",
"serde_json",
+ "sha2 0.10.9",
"temper-runtime",
"temper-spec",
"thiserror 2.0.18",
diff --git a/crates/temper-cli/src/migrate_turso_to_postgres.rs b/crates/temper-cli/src/migrate_turso_to_postgres.rs
index cf616944d..7a1cf8c09 100644
--- a/crates/temper-cli/src/migrate_turso_to_postgres.rs
+++ b/crates/temper-cli/src/migrate_turso_to_postgres.rs
@@ -432,7 +432,10 @@ async fn migrate_specs(
.filter(|hash| !hash.is_empty())
.unwrap_or_else(|| spec_content_hash(&row.ioa_source));
sqlx::query(
- "INSERT INTO specs \
+ "WITH cleared_staging AS ( \
+ DELETE FROM staged_specs WHERE tenant = $1 AND entity_type = $2 \
+ ) \
+ INSERT INTO specs \
(tenant, entity_type, ioa_source, csdl_xml, version, verified, verification_status, \
levels_passed, levels_total, verification_result, content_hash, committed, updated_at) \
VALUES ($1, $2, $3, $4, 1, $5, $6, $7, $8, $9, $10, $11, $12) \
@@ -2077,6 +2080,23 @@ mod tests {
.await
.expect("put source blob");
+ let pool = PgPool::connect(&database_url).await.expect("target pool");
+ temper_store_postgres::migration::run_migrations(&pool)
+ .await
+ .expect("target migrations");
+ sqlx::query(
+ "INSERT INTO staged_specs \
+ (tenant, entity_type, ioa_source, csdl_xml, content_hash, version, updated_at) \
+ VALUES ($1, 'SmokeEntity', 'stale staged bytes', '', 'stale-hash', 1, now()) \
+ ON CONFLICT (tenant, entity_type) DO UPDATE SET \
+ ioa_source = EXCLUDED.ioa_source, csdl_xml = EXCLUDED.csdl_xml, \
+ content_hash = EXCLUDED.content_hash, updated_at = now()",
+ )
+ .bind(&tenant)
+ .execute(&pool)
+ .await
+ .expect("seed stale target staging");
+
run(MigrationOptions {
tenant: tenant.clone(),
dry_run: false,
@@ -2090,7 +2110,15 @@ mod tests {
.await
.expect("run migration");
- let pool = PgPool::connect(&database_url).await.expect("target pool");
+ let stale_staged_count: i64 = sqlx::query_scalar(
+ "SELECT COUNT(*)::bigint FROM staged_specs \
+ WHERE tenant = $1 AND entity_type = 'SmokeEntity'",
+ )
+ .bind(&tenant)
+ .fetch_one(&pool)
+ .await
+ .expect("count stale staging");
+ assert_eq!(stale_staged_count, 0);
let event_count: i64 =
sqlx::query_scalar("SELECT COUNT(*)::bigint FROM events WHERE tenant = $1")
.bind(&tenant)
diff --git a/crates/temper-cli/src/serve/bootstrap.rs b/crates/temper-cli/src/serve/bootstrap.rs
index 2e206e453..bdc98de2e 100644
--- a/crates/temper-cli/src/serve/bootstrap.rs
+++ b/crates/temper-cli/src/serve/bootstrap.rs
@@ -212,6 +212,18 @@ pub(super) fn load_webhooks(apps: &[(String, String)]) -> Option std::collections::BTreeSet {
+ state
+ .server
+ .registry
+ .read()
+ .map(|registry| registry.tenant_ids().into_iter().cloned().collect())
+ .unwrap_or_else(|error| {
+ eprintln!(" Warning: registry lock poisoned during hydration: {error}");
+ std::collections::BTreeSet::new()
+ })
+}
+
pub(super) async fn hydrate_entities(state: &PlatformState, apps: &[(String, String)]) {
if state.server.storage_stack.is_none() {
return;
@@ -225,15 +237,9 @@ pub(super) async fn hydrate_entities(state: &PlatformState, apps: &[(String, Str
)
})
.unwrap_or(false);
- let mut all_tenants = Vec::new();
+ let mut all_tenants = registered_hydration_tenants(state);
for (tenant, _dir) in apps {
- let tenant_id = TenantId::new(tenant.as_str());
- if eager_hydrate {
- state.server.hydrate_from_store(&tenant_id).await;
- } else {
- state.server.populate_index_from_store(&tenant_id).await;
- }
- all_tenants.push(tenant_id);
+ all_tenants.insert(TenantId::new(tenant.as_str()));
}
// In TenantRouted mode, also hydrate all registered tenants.
if let Some(provider) = state
@@ -243,13 +249,14 @@ pub(super) async fn hydrate_entities(state: &PlatformState, apps: &[(String, Str
.and_then(|stack| stack.turso.clone())
{
for tenant in provider.connected_tenants().await {
- let tenant_id = TenantId::new(&tenant);
- if eager_hydrate {
- state.server.hydrate_from_store(&tenant_id).await;
- } else {
- state.server.populate_index_from_store(&tenant_id).await;
- }
- all_tenants.push(tenant_id);
+ all_tenants.insert(TenantId::new(&tenant));
+ }
+ }
+ for tenant_id in &all_tenants {
+ if eager_hydrate {
+ state.server.hydrate_from_store(tenant_id).await;
+ } else {
+ state.server.populate_index_from_store(tenant_id).await;
}
}
@@ -406,16 +413,17 @@ pub(super) async fn recover_secrets(state: &PlatformState) {
}
}
-/// Load the verification cache from Turso for a tenant (hash + verified status).
+/// Load the verification cache from the active platform store for a tenant.
///
-/// Routes to the per-tenant store in TenantRouted mode.
-/// Returns an empty map if no Turso store is available.
+/// Routes to a per-tenant Turso store in TenantRouted mode and to the shared
+/// tenant-scoped Postgres store otherwise. Returns an empty map when platform
+/// persistence is unavailable.
async fn load_verified_cache(
state: &PlatformState,
tenant: &str,
) -> std::collections::BTreeMap {
- if let Some(turso) = state.server.turso_store_for_tenant(tenant).await {
- match turso.load_verification_cache(tenant).await {
+ if let Some(store) = state.server.platform_store_for_tenant(tenant).await {
+ match store.load_verification_cache(tenant).await {
Ok(cache) => cache,
Err(e) => {
eprintln!(" Warning: failed to load verification cache for {tenant}: {e}");
@@ -435,16 +443,20 @@ async fn load_verified_cache(
pub(super) async fn bootstrap_tenants(state: &PlatformState, apps: &[(String, String)]) {
let sys_cache = load_verified_cache(state, "temper-system").await;
let sys_hashes = temper_platform::bootstrap_system_tenant(state, &sys_cache);
- if let Some(turso) = state.server.turso_store_for_tenant("temper-system").await {
- temper_platform::persist_system_verification(&turso, &sys_hashes, &sys_cache).await;
+ if let Some(store) = state
+ .server
+ .platform_store_for_tenant("temper-system")
+ .await
+ {
+ temper_platform::persist_system_verification(store.as_ref(), &sys_hashes, &sys_cache).await;
}
let default_cache = load_verified_cache(state, "default").await;
let default_hashes =
temper_platform::bootstrap_agent_specs(state, "default", false, &default_cache);
- if let Some(turso) = state.server.turso_store_for_tenant("default").await {
+ if let Some(store) = state.server.platform_store_for_tenant("default").await {
temper_platform::persist_agent_verification(
- &turso,
+ store.as_ref(),
"default",
&default_hashes,
&default_cache,
@@ -457,8 +469,9 @@ pub(super) async fn bootstrap_tenants(state: &PlatformState, apps: &[(String, St
// App tenants already have user specs loaded in Phase 2; merge the
// built-in agent OS entities so we do not replace their entity-set map.
let hashes = temper_platform::bootstrap_agent_specs(state, tenant, true, &cache);
- if let Some(turso) = state.server.turso_store_for_tenant(tenant).await {
- temper_platform::persist_agent_verification(&turso, tenant, &hashes, &cache).await;
+ if let Some(store) = state.server.platform_store_for_tenant(tenant).await {
+ temper_platform::persist_agent_verification(store.as_ref(), tenant, &hashes, &cache)
+ .await;
}
}
// In TenantRouted mode, bootstrap agent specs for all registered tenants.
@@ -474,8 +487,14 @@ pub(super) async fn bootstrap_tenants(state: &PlatformState, apps: &[(String, St
for tenant in provider.connected_tenants().await {
let cache = load_verified_cache(state, &tenant).await;
let hashes = temper_platform::bootstrap_agent_specs(state, &tenant, true, &cache);
- if let Some(turso) = state.server.turso_store_for_tenant(&tenant).await {
- temper_platform::persist_agent_verification(&turso, &tenant, &hashes, &cache).await;
+ if let Some(store) = state.server.platform_store_for_tenant(&tenant).await {
+ temper_platform::persist_agent_verification(
+ store.as_ref(),
+ &tenant,
+ &hashes,
+ &cache,
+ )
+ .await;
}
}
}
@@ -621,14 +640,180 @@ pub(super) async fn bootstrap_installed_apps(
#[cfg(test)]
mod tests {
+ use sqlx::PgPool;
use temper_platform::os_apps::get_os_app;
use temper_platform::state::PlatformState;
+ use temper_runtime::persistence::EventStore;
use temper_runtime::tenant::TenantId;
use temper_server::storage::StorageStack;
use temper_spec::csdl::parse_csdl;
+ use temper_store_postgres::{PostgresEventStore, PostgresSpecVerificationUpdate};
use temper_store_turso::TursoEventStore;
- use super::bootstrap_installed_apps;
+ use super::{bootstrap_installed_apps, load_verified_cache, registered_hydration_tenants};
+
+ #[test]
+ fn restored_registry_tenants_are_hydrated_without_cli_apps() {
+ let tenant = "restored-postgres-tenant";
+ let bundle = get_os_app("temper-fs").expect("temper-fs bundle");
+ let csdl_xml = bundle.csdl.clone().expect("temper-fs CSDL");
+ let csdl = parse_csdl(&csdl_xml).expect("parse CSDL");
+ let refs = bundle
+ .specs
+ .iter()
+ .map(|(entity_type, source)| (entity_type.as_str(), source.as_str()))
+ .collect::>();
+ let state = PlatformState::new(None);
+ state
+ .registry
+ .write()
+ .expect("registry")
+ .register_tenant(tenant, csdl, csdl_xml, &refs);
+
+ assert_eq!(
+ registered_hydration_tenants(&state),
+ [TenantId::from(tenant)].into_iter().collect(),
+ "startup must hydrate tenants restored from PostgreSQL even with no --app"
+ );
+ }
+
+ #[tokio::test]
+ async fn postgres_agent_bootstrap_republishes_replacement_tombstone() {
+ let Ok(database_url) = std::env::var("DATABASE_URL") else {
+ return;
+ };
+ let pool = PgPool::connect(&database_url)
+ .await
+ .expect("connect Postgres");
+ temper_store_postgres::migration::run_migrations(&pool)
+ .await
+ .expect("migrate Postgres");
+ let store = PostgresEventStore::new(pool.clone());
+ let tenant = format!("bootstrap-agent-postgres-{}", uuid::Uuid::new_v4());
+ let legacy_fingerprint = temper_store_turso::spec_content_hash("legacy Agent declaration");
+ let unrelated_a = "[automaton]\nname = \"Unrelated\"\n# committed-a\n";
+ let unrelated_b = "[automaton]\nname = \"Unrelated\"\n# staged-b\n";
+ let unrelated_a_fingerprint = temper_store_turso::spec_content_hash(unrelated_a);
+ let unrelated_b_fingerprint = temper_store_turso::spec_content_hash(unrelated_b);
+
+ store
+ .begin_vector_index_reconciliation(
+ &tenant,
+ "Agent",
+ "v2|legacy",
+ 1,
+ &legacy_fingerprint,
+ )
+ .await
+ .expect("bootstrap compatibility authority");
+ store
+ .persist_spec_catalog_update(&tenant, &[], "", &[], true, None)
+ .await
+ .expect("replacement tombstones compatibility authority");
+
+ store
+ .upsert_spec(
+ &tenant,
+ "Unrelated",
+ unrelated_a,
+ "",
+ &unrelated_a_fingerprint,
+ )
+ .await
+ .expect("stage unrelated A");
+ store
+ .commit_verified_spec(
+ &tenant,
+ "Unrelated",
+ &unrelated_a_fingerprint,
+ "",
+ PostgresSpecVerificationUpdate {
+ status: "completed",
+ verified: true,
+ levels_passed: None,
+ levels_total: None,
+ verification_result_json: None,
+ },
+ )
+ .await
+ .expect("commit unrelated A");
+ store
+ .upsert_spec(
+ &tenant,
+ "Unrelated",
+ unrelated_b,
+ "",
+ &unrelated_b_fingerprint,
+ )
+ .await
+ .expect("stage unrelated B during built-in bootstrap window");
+
+ let mut state = PlatformState::new(None);
+ state
+ .server
+ .set_storage_stack(StorageStack::from_postgres(store));
+ assert!(state.server.turso_store_for_tenant(&tenant).await.is_none());
+ let cache = load_verified_cache(&state, &tenant).await;
+ let hashes = temper_platform::bootstrap_agent_specs(&state, &tenant, true, &cache);
+ let platform_store = state
+ .server
+ .platform_store_for_tenant(&tenant)
+ .await
+ .expect("Postgres must provide tenant platform persistence");
+ temper_platform::persist_agent_verification(
+ platform_store.as_ref(),
+ &tenant,
+ &hashes,
+ &cache,
+ )
+ .await;
+
+ let expected_agent_fingerprint = hashes
+ .iter()
+ .find(|(entity_type, _)| entity_type == "Agent")
+ .map(|(_, fingerprint)| fingerprint)
+ .expect("Agent bootstrap fingerprint");
+ let authority: (String, bool) = sqlx::query_as(
+ "SELECT declaration_fingerprint, present \
+ FROM spec_declaration_authority \
+ WHERE tenant = $1 AND entity_type = 'Agent'",
+ )
+ .bind(&tenant)
+ .fetch_one(&pool)
+ .await
+ .expect("re-published Agent authority");
+ assert_eq!(&authority.0, expected_agent_fingerprint);
+ assert!(authority.1);
+
+ let unrelated_catalog: (String, bool) = sqlx::query_as(
+ "SELECT content_hash, committed FROM specs \
+ WHERE tenant = $1 AND entity_type = 'Unrelated'",
+ )
+ .bind(&tenant)
+ .fetch_one(&pool)
+ .await
+ .expect("read committed unrelated A after built-in bootstrap");
+ assert_eq!(unrelated_catalog, (unrelated_a_fingerprint.clone(), true));
+ let unrelated_staging: (String,) = sqlx::query_as(
+ "SELECT content_hash FROM staged_specs \
+ WHERE tenant = $1 AND entity_type = 'Unrelated'",
+ )
+ .bind(&tenant)
+ .fetch_one(&pool)
+ .await
+ .expect("read unrelated staging B after built-in bootstrap");
+ assert_eq!(unrelated_staging.0, unrelated_b_fingerprint);
+ let unrelated_authority: (String, bool) = sqlx::query_as(
+ "SELECT declaration_fingerprint, present \
+ FROM spec_declaration_authority \
+ WHERE tenant = $1 AND entity_type = 'Unrelated'",
+ )
+ .bind(&tenant)
+ .fetch_one(&pool)
+ .await
+ .expect("read unrelated authority after built-in bootstrap");
+ assert_eq!(unrelated_authority, (unrelated_a_fingerprint, true));
+ }
#[tokio::test]
async fn bootstrap_installed_apps_replays_persisted_app_when_registry_specs_are_stale() {
diff --git a/crates/temper-cli/src/serve/storage.rs b/crates/temper-cli/src/serve/storage.rs
index 3f831faa5..2fd20734a 100644
--- a/crates/temper-cli/src/serve/storage.rs
+++ b/crates/temper-cli/src/serve/storage.rs
@@ -52,51 +52,32 @@ pub(super) async fn upsert_loaded_specs_to_postgres(
tenant: &str,
loaded: &LoadedTenantSpecs,
) -> Result<()> {
- for (entity_type, ioa_source) in &loaded.ioa_sources {
- sqlx::query(
- "INSERT INTO specs \
- (tenant, entity_type, ioa_source, csdl_xml, version, verified, verification_status, updated_at) \
- VALUES ($1, $2, $3, $4, 1, false, 'pending', now()) \
- ON CONFLICT (tenant, entity_type) DO UPDATE SET \
- ioa_source = EXCLUDED.ioa_source, \
- csdl_xml = EXCLUDED.csdl_xml, \
- version = specs.version + 1, \
- verified = false, \
- verification_status = 'pending', \
- levels_passed = NULL, \
- levels_total = NULL, \
- verification_result = NULL, \
- updated_at = now()",
+ let fingerprints = loaded
+ .ioa_sources
+ .iter()
+ .map(|(entity_type, ioa_source)| {
+ (
+ entity_type.as_str(),
+ ioa_source.as_str(),
+ temper_store_turso::spec_content_hash(ioa_source),
+ )
+ })
+ .collect::>();
+ let specs = fingerprints
+ .iter()
+ .map(|(entity_type, source, fingerprint)| (*entity_type, *source, fingerprint.as_str()))
+ .collect::>();
+ PostgresEventStore::new(pool.clone())
+ .persist_spec_catalog_update(
+ tenant,
+ &specs,
+ &loaded.csdl_xml,
+ &[],
+ true,
+ loaded.cross_invariants_toml.as_deref(),
)
- .bind(tenant)
- .bind(entity_type)
- .bind(ioa_source)
- .bind(&loaded.csdl_xml)
- .execute(pool)
.await
- .with_context(|| format!("Failed to persist spec {tenant}/{entity_type}"))?;
- }
- if let Some(source) = loaded.cross_invariants_toml.as_deref() {
- sqlx::query(
- "INSERT INTO tenant_constraints (tenant, cross_invariants_toml, version, updated_at) \
- VALUES ($1, $2, 1, now()) \
- ON CONFLICT (tenant) DO UPDATE SET \
- cross_invariants_toml = EXCLUDED.cross_invariants_toml, \
- version = tenant_constraints.version + 1, \
- updated_at = now()",
- )
- .bind(tenant)
- .bind(source)
- .execute(pool)
- .await
- .with_context(|| format!("Failed to persist tenant constraints for {tenant}"))?;
- } else {
- sqlx::query("DELETE FROM tenant_constraints WHERE tenant = $1")
- .bind(tenant)
- .execute(pool)
- .await
- .with_context(|| format!("Failed to clear tenant constraints for {tenant}"))?;
- }
+ .with_context(|| format!("Failed to persist spec catalog for {tenant} in Postgres"))?;
Ok(())
}
@@ -110,26 +91,32 @@ pub(super) async fn upsert_loaded_specs_to_turso(
tenant: &str,
loaded: &LoadedTenantSpecs,
) -> Result<()> {
- for (entity_type, ioa_source) in &loaded.ioa_sources {
- let hash = temper_store_turso::spec_content_hash(ioa_source);
- turso
- .upsert_spec(tenant, entity_type, ioa_source, &loaded.csdl_xml, &hash)
- .await
- .with_context(|| format!("Failed to persist spec {tenant}/{entity_type} in Turso"))?;
- }
- if let Some(source) = loaded.cross_invariants_toml.as_deref() {
- turso
- .upsert_tenant_constraints(tenant, source)
- .await
- .with_context(|| {
- format!("Failed to persist tenant constraints for {tenant} in Turso")
- })?;
- } else {
- turso
- .delete_tenant_constraints(tenant)
- .await
- .with_context(|| format!("Failed to clear tenant constraints for {tenant} in Turso"))?;
- }
+ let fingerprints = loaded
+ .ioa_sources
+ .iter()
+ .map(|(entity_type, ioa_source)| {
+ (
+ entity_type.as_str(),
+ ioa_source.as_str(),
+ temper_store_turso::spec_content_hash(ioa_source),
+ )
+ })
+ .collect::>();
+ let specs = fingerprints
+ .iter()
+ .map(|(entity_type, source, fingerprint)| (*entity_type, *source, fingerprint.as_str()))
+ .collect::>();
+ turso
+ .persist_spec_catalog_update(
+ tenant,
+ &specs,
+ &loaded.csdl_xml,
+ &[],
+ true,
+ loaded.cross_invariants_toml.as_deref(),
+ )
+ .await
+ .with_context(|| format!("Failed to persist spec catalog for {tenant} in Turso"))?;
if let Some(policy_text) = loaded.cedar_policy_text.as_deref() {
turso
.save_policy(tenant, "primary", policy_text, "system")
diff --git a/crates/temper-jit/Cargo.toml b/crates/temper-jit/Cargo.toml
index 39ccbeb70..6a449262e 100644
--- a/crates/temper-jit/Cargo.toml
+++ b/crates/temper-jit/Cargo.toml
@@ -11,6 +11,7 @@ temper-runtime = { workspace = true }
temper-spec = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
+sha2 = { workspace = true }
thiserror = { workspace = true }
[dev-dependencies]
diff --git a/crates/temper-jit/src/shadow.rs b/crates/temper-jit/src/shadow.rs
index 8cd41174f..0d8da03ac 100644
--- a/crates/temper-jit/src/shadow.rs
+++ b/crates/temper-jit/src/shadow.rs
@@ -113,6 +113,7 @@ mod tests {
fn base_table() -> TransitionTable {
let mut table = TransitionTable {
entity_name: "Order".into(),
+ spec_declaration_fingerprint: None,
states: vec!["Draft".into(), "Submitted".into(), "Cancelled".into()],
initial_state: "Draft".into(),
keys: vec![],
diff --git a/crates/temper-jit/src/swap.rs b/crates/temper-jit/src/swap.rs
index 6b05c59f6..823422e52 100644
--- a/crates/temper-jit/src/swap.rs
+++ b/crates/temper-jit/src/swap.rs
@@ -84,6 +84,7 @@ mod tests {
fn dummy_table(name: &str) -> TransitionTable {
let mut table = TransitionTable {
entity_name: name.to_string(),
+ spec_declaration_fingerprint: None,
states: vec!["A".into(), "B".into()],
initial_state: "A".into(),
keys: vec![],
diff --git a/crates/temper-jit/src/table/builder.rs b/crates/temper-jit/src/table/builder.rs
index bf48afc67..371d47b1f 100644
--- a/crates/temper-jit/src/table/builder.rs
+++ b/crates/temper-jit/src/table/builder.rs
@@ -4,6 +4,7 @@
//! translation layer in `temper-spec`. The shared layer eliminates duplicated
//! guard/effect translation logic between JIT and verification paths.
+use sha2::{Digest, Sha256};
use temper_spec::automaton::{self, Automaton, ResolvedEffect, ResolvedGuard, translate_actions};
use super::guard::Guard;
@@ -21,7 +22,11 @@ impl TransitionTable {
pub fn try_from_ioa_source(ioa_toml: &str) -> Result {
let automaton = automaton::parse_automaton(ioa_toml)
.map_err(|e| format!("failed to parse I/O Automaton TOML: {e}"))?;
- Ok(Self::from_automaton(&automaton))
+ let mut table = Self::from_automaton(&automaton);
+ let mut hasher = Sha256::new();
+ hasher.update(ioa_toml.as_bytes());
+ table.spec_declaration_fingerprint = Some(format!("{:x}", hasher.finalize()));
+ Ok(table)
}
/// Build a TransitionTable from I/O Automaton TOML source.
@@ -118,6 +123,7 @@ impl TransitionTable {
TransitionTable {
entity_name: automaton.automaton.name.clone(),
+ spec_declaration_fingerprint: None,
states: automaton.automaton.states.clone(),
initial_state: automaton.automaton.initial.clone(),
rules,
diff --git a/crates/temper-jit/src/table/types.rs b/crates/temper-jit/src/table/types.rs
index e18fe1ffe..1edb19045 100644
--- a/crates/temper-jit/src/table/types.rs
+++ b/crates/temper-jit/src/table/types.rs
@@ -44,6 +44,13 @@ pub struct DeclaredVector {
pub struct TransitionTable {
/// The entity this table governs (e.g. "Order").
pub entity_name: String,
+ /// SHA-256 fingerprint of the IOA source compiled into this table.
+ ///
+ /// Tables built directly from an already-parsed automaton may not have a
+ /// source fingerprint. Persistent entity writes require tables built from
+ /// IOA source so stores can reject stale-replica index rows atomically.
+ #[serde(default)]
+ pub spec_declaration_fingerprint: Option,
/// All valid state values.
pub states: Vec,
/// The state an entity starts in.
@@ -152,6 +159,8 @@ impl<'de> Deserialize<'de> for TransitionTable {
#[derive(Deserialize)]
struct TransitionTableRaw {
entity_name: String,
+ #[serde(default)]
+ spec_declaration_fingerprint: Option,
states: Vec,
initial_state: String,
rules: Vec,
@@ -168,6 +177,7 @@ impl<'de> Deserialize<'de> for TransitionTable {
let raw = TransitionTableRaw::deserialize(deserializer)?;
let mut table = TransitionTable {
entity_name: raw.entity_name,
+ spec_declaration_fingerprint: raw.spec_declaration_fingerprint,
states: raw.states,
initial_state: raw.initial_state,
rules: raw.rules,
@@ -284,6 +294,7 @@ mod tests {
fn rebuild_index_groups_by_name() {
let mut table = TransitionTable {
entity_name: "TestEntity".to_string(),
+ spec_declaration_fingerprint: None,
states: vec!["Draft".to_string(), "Active".to_string()],
initial_state: "Draft".to_string(),
keys: vec![],
diff --git a/crates/temper-platform/src/bootstrap.rs b/crates/temper-platform/src/bootstrap.rs
index 8cbbdd176..75a74d31b 100644
--- a/crates/temper-platform/src/bootstrap.rs
+++ b/crates/temper-platform/src/bootstrap.rs
@@ -293,7 +293,6 @@ pub(crate) async fn persist_bootstrap_verification(
verified_cache: &BTreeMap,
) {
let hashes_to_persist = hashes_requiring_persistence(hashes, verified_cache);
- let mut wrote_specs = false;
for (entity_type, content_hash) in &hashes_to_persist {
// Find the IOA source for this entity type.
@@ -311,13 +310,16 @@ pub(crate) async fn persist_bootstrap_verification(
tracing::warn!("Failed to persist bootstrap spec {tenant}/{entity_type}: {e}");
continue;
}
- wrote_specs = true;
- // Mark as verified (bootstrap panics on failure, so all specs here passed).
+ // Atomically publish verification for exactly the bytes that passed.
+ // Another replica may stage the same tenant/type between the upsert
+ // above and this call; the expected fingerprint makes that fail closed.
if let Err(e) = store
- .persist_spec_verification(
+ .commit_verified_spec(
tenant,
entity_type,
+ content_hash,
+ csdl_source,
SpecVerificationUpdate {
status: "completed",
verified: true,
@@ -328,16 +330,9 @@ pub(crate) async fn persist_bootstrap_verification(
)
.await
{
- tracing::warn!("Failed to persist verification status for {tenant}/{entity_type}: {e}");
+ tracing::warn!("Failed to commit verified bootstrap spec {tenant}/{entity_type}: {e}");
}
}
-
- // `upsert_spec` marks rows as uncommitted while content is rewritten. Once
- // bootstrap verification succeeds, promote the tenant's spec set back to a
- // durable committed state so restart recovery can actually see the rows.
- if wrote_specs && let Err(e) = store.commit_specs(tenant).await {
- tracing::warn!("Failed to commit bootstrap specs for tenant '{tenant}': {e}");
- }
}
fn hashes_requiring_persistence(
diff --git a/crates/temper-platform/src/os_apps/mod.rs b/crates/temper-platform/src/os_apps/mod.rs
index b1e0b78ad..292541fcc 100644
--- a/crates/temper-platform/src/os_apps/mod.rs
+++ b/crates/temper-platform/src/os_apps/mod.rs
@@ -13,6 +13,7 @@ use std::time::Instant;
use chrono::{DateTime, NaiveDateTime, Utc};
use serde::Serialize;
use temper_runtime::tenant::TenantId;
+use temper_server::platform_store::SpecCommitExpectation;
use temper_server::state::WasmModuleSource;
use temper_spec::automaton;
use temper_spec::csdl::{emit_csdl_xml, merge_csdl, parse_csdl};
@@ -1249,6 +1250,7 @@ pub(super) async fn install_os_app_with_plan(
.as_ref()
.and_then(|stack| stack.platform.clone())
{
+ let mut expected_specs = Vec::with_capacity(bundle.specs.len());
if plan.specs
&& let Some(ref merged) = merged_csdl
{
@@ -1257,6 +1259,7 @@ pub(super) async fn install_os_app_with_plan(
ps.upsert_spec(tenant, entity_type, ioa_source, merged, &hash)
.await
.map_err(|e| format!("Failed to persist spec {entity_type}: {e}"))?;
+ expected_specs.push((entity_type.as_str(), hash));
}
}
if let Some(ref policy_text) = combined_policy {
@@ -1274,11 +1277,20 @@ pub(super) async fn install_os_app_with_plan(
ps.record_installed_app(tenant, app_name)
.await
.map_err(|e| format!("Failed to record os-app installation: {e}"))?;
- if plan.specs {
- // Commit only when this path used individual spec writes.
- ps.commit_specs(tenant)
+ if let Some(ref merged) = merged_csdl
+ && !expected_specs.is_empty()
+ {
+ let expected = expected_specs
+ .iter()
+ .map(|(entity_type, hash)| SpecCommitExpectation {
+ entity_type,
+ content_hash: hash,
+ csdl_xml: merged,
+ })
+ .collect::>();
+ ps.commit_spec_batch(tenant, &expected)
.await
- .map_err(|e| format!("Failed to commit specs: {e}"))?;
+ .map_err(|e| format!("Failed to commit app spec batch: {e}"))?;
}
}
diff --git a/crates/temper-runtime/src/actor/actor_ref.rs b/crates/temper-runtime/src/actor/actor_ref.rs
index 28802a23c..2eb4d49e7 100644
--- a/crates/temper-runtime/src/actor/actor_ref.rs
+++ b/crates/temper-runtime/src/actor/actor_ref.rs
@@ -1,4 +1,6 @@
use std::fmt;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tokio::sync::oneshot;
@@ -38,6 +40,9 @@ pub enum SystemSignal {
pub struct ActorRef {
pub(crate) sender: MailboxSender,
pub(crate) id: ActorId,
+ /// Packed supervised-incarnation state. The low bit is readiness and the
+ /// remaining bits are a monotonically increasing `pre_start` epoch.
+ pub(crate) lifecycle: Arc,
}
/// Unique identifier for an actor instance.
@@ -111,6 +116,21 @@ impl ActorRef {
&self.id
}
+ /// Whether this actor incarnation completed `pre_start` and is serving messages.
+ pub fn is_ready(&self) -> bool {
+ self.ready_incarnation().is_some()
+ }
+
+ /// Return the ready supervised-incarnation epoch in one atomic observation.
+ ///
+ /// An [`ActorId`] identifies the mailbox/task. Supervision can run
+ /// `pre_start` repeatedly inside that task, so callers that must distinguish
+ /// initialized state across a restart also need this epoch.
+ pub fn ready_incarnation(&self) -> Option {
+ let lifecycle = self.lifecycle.load(Ordering::Acquire);
+ (lifecycle & 1 == 1).then_some(lifecycle >> 1)
+ }
+
/// Current in-flight mailbox depth (messages queued but not yet processed).
/// Exposed for observability; see `runtime_metrics::record_actor_mailbox_depth`.
pub fn mailbox_depth(&self) -> usize {
@@ -133,6 +153,7 @@ impl Clone for ActorRef {
Self {
sender: self.sender.clone(),
id: self.id.clone(),
+ lifecycle: self.lifecycle.clone(),
}
}
}
diff --git a/crates/temper-runtime/src/actor/cell.rs b/crates/temper-runtime/src/actor/cell.rs
index 5bbca1c24..bd5466149 100644
--- a/crates/temper-runtime/src/actor/cell.rs
+++ b/crates/temper-runtime/src/actor/cell.rs
@@ -1,3 +1,6 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
use tracing::{error, info, warn};
use super::actor_ref::{ActorId, ActorRef, Envelope, SystemSignal};
@@ -16,6 +19,53 @@ pub struct ActorCell {
mailbox_capacity: usize,
}
+/// Clears publication readiness whenever the actor task future is dropped.
+///
+/// Normal shutdown, panic unwinding, and task cancellation all drop the run
+/// future, so a dead incarnation can never remain externally marked ready.
+struct ActorReadiness {
+ lifecycle: Arc,
+}
+
+impl ActorReadiness {
+ fn new(lifecycle: Arc) -> Self {
+ lifecycle.store(0, Ordering::Release);
+ Self { lifecycle }
+ }
+
+ fn begin_incarnation(&self) {
+ let current = self.lifecycle.load(Ordering::Acquire);
+ assert_eq!(
+ current & 1,
+ 0,
+ "actor cannot begin a supervised incarnation while marked ready"
+ );
+ let next = current
+ .checked_add(2)
+ .expect("actor supervised-incarnation epoch exhausted");
+ self.lifecycle.store(next, Ordering::Release);
+ }
+
+ fn mark_ready(&self) {
+ let previous = self.lifecycle.fetch_or(1, Ordering::AcqRel);
+ assert_eq!(
+ previous & 1,
+ 0,
+ "actor supervised incarnation was already marked ready"
+ );
+ }
+
+ fn mark_unready(&self) {
+ self.lifecycle.fetch_and(!1, Ordering::AcqRel);
+ }
+}
+
+impl Drop for ActorReadiness {
+ fn drop(&mut self) {
+ self.mark_unready();
+ }
+}
+
impl ActorCell {
/// Create a new actor cell with the given actor and ID.
pub fn new(actor: A, id: ActorId) -> Self {
@@ -36,13 +86,15 @@ impl ActorCell {
pub fn spawn(self) -> ActorRef {
let (tx, rx) = mailbox::mailbox(self.mailbox_capacity);
let id = self.id.clone();
+ let lifecycle = Arc::new(AtomicU64::new(0));
let actor_ref = ActorRef {
sender: tx,
id: id.clone(),
+ lifecycle: lifecycle.clone(),
};
- tokio::spawn(self.run(rx)); // determinism-ok: production actor cell, not on simulation path
+ tokio::spawn(self.run(rx, lifecycle)); // determinism-ok: production actor cell, not on simulation path
actor_ref
}
@@ -51,7 +103,8 @@ impl ActorCell {
/// 1. pre_start → initialize state
/// 2. loop: receive message → handle
/// 3. post_stop → cleanup
- async fn run(self, mut rx: MailboxReceiver) {
+ async fn run(self, mut rx: MailboxReceiver, lifecycle: Arc) {
+ let readiness = ActorReadiness::new(lifecycle);
let actor = self.actor;
let id = self.id;
let strategy = actor.supervision_strategy();
@@ -59,6 +112,8 @@ impl ActorCell {
let mut restart_count: u32 = 0;
loop {
+ readiness.mark_unready();
+ readiness.begin_incarnation();
// Phase 1: Initialize
let mut ctx = ActorContext::new(id.clone());
info!(actor = %id, "actor starting");
@@ -67,6 +122,7 @@ impl ActorCell {
Ok(s) => {
info!(actor = %id, "actor started");
restart_count = 0;
+ readiness.mark_ready();
s
}
Err(e) => {
@@ -133,6 +189,7 @@ impl ActorCell {
};
// Phase 3: Cleanup
+ readiness.mark_unready();
info!(actor = %id, "actor stopping");
actor.post_stop(state, &mut ctx).await;
@@ -159,7 +216,122 @@ fn should_restart(strategy: &SupervisionStrategy, current_restarts: u32) -> bool
#[cfg(test)]
mod tests {
use super::*;
+ use crate::actor::{ActorContext, Message};
use std::time::Duration;
+ use tokio::sync::Notify;
+
+ #[derive(Debug)]
+ enum PanickingMsg {
+ Crash,
+ }
+
+ impl Message for PanickingMsg {}
+
+ struct PanickingActor {
+ started: Arc,
+ }
+
+ impl Actor for PanickingActor {
+ type Msg = PanickingMsg;
+ type State = ();
+
+ async fn pre_start(
+ &self,
+ _ctx: &mut ActorContext,
+ ) -> Result {
+ self.started.notify_one();
+ Ok(())
+ }
+
+ async fn handle(
+ &self,
+ msg: Self::Msg,
+ _state: &mut Self::State,
+ _ctx: &mut ActorContext,
+ ) -> Result<(), ActorError> {
+ match msg {
+ PanickingMsg::Crash => panic!("intentional handler panic"),
+ }
+ }
+
+ async fn post_stop(&self, _state: Self::State, _ctx: &mut ActorContext) {}
+ }
+
+ #[tokio::test]
+ async fn handler_panic_clears_actor_readiness() {
+ let started = Arc::new(Notify::new());
+ let actor = ActorCell::new(
+ PanickingActor {
+ started: started.clone(),
+ },
+ ActorId::new("panicking", "system/panicking"),
+ )
+ .spawn();
+ started.notified().await;
+ tokio::time::timeout(Duration::from_secs(1), async {
+ while !actor.is_ready() {
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .expect("actor must publish readiness after pre_start");
+
+ actor
+ .tell(PanickingMsg::Crash)
+ .expect("enqueue crashing message");
+ tokio::time::timeout(Duration::from_secs(1), async {
+ while actor.is_ready() {
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .expect("actor panic must clear readiness through the run-future drop guard");
+ assert!(!actor.is_ready());
+ }
+
+ #[tokio::test]
+ async fn supervised_restart_advances_ready_incarnation_without_changing_actor_id() {
+ let started = Arc::new(Notify::new());
+ let actor = ActorCell::new(
+ PanickingActor {
+ started: started.clone(),
+ },
+ ActorId::new("restarting", "system/restarting"),
+ )
+ .spawn();
+ started.notified().await;
+ let first_incarnation = tokio::time::timeout(Duration::from_secs(1), async {
+ loop {
+ if let Some(incarnation) = actor.ready_incarnation() {
+ break incarnation;
+ }
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .expect("actor must publish its first ready incarnation");
+ let actor_uid = actor.id().uid;
+
+ actor
+ .signal(SystemSignal::Restart)
+ .expect("enqueue supervised restart");
+ let second_incarnation = tokio::time::timeout(Duration::from_secs(1), async {
+ loop {
+ if let Some(incarnation) = actor.ready_incarnation()
+ && incarnation != first_incarnation
+ {
+ break incarnation;
+ }
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .expect("actor must publish its restarted incarnation");
+
+ assert_eq!(actor.id().uid, actor_uid);
+ assert!(second_incarnation > first_incarnation);
+ actor.stop().expect("stop restarted actor");
+ }
#[test]
fn stop_strategy_never_restarts() {
diff --git a/crates/temper-runtime/src/persistence/indexing.rs b/crates/temper-runtime/src/persistence/indexing.rs
new file mode 100644
index 000000000..f95fb4e9f
--- /dev/null
+++ b/crates/temper-runtime/src/persistence/indexing.rs
@@ -0,0 +1,106 @@
+use serde::{Deserialize, Serialize};
+
+use super::PersistenceEnvelope;
+
+/// A declared-key row to co-commit with an append (ADR-0153). The entity claims
+/// `key_hash` for `key_name`; the store writes it into `entity_key_index` in the
+/// same transaction as the journal append, giving the read plane an `O(log n)`
+/// present/absent probe (the negative-existence access path, ARN-68).
+#[derive(Debug, Clone)]
+pub struct EntityKeyRow {
+ /// The declared key's identifier (the `[[key]]` block's `name`).
+ pub key_name: String,
+ /// The canonical, type-tagged hash of the key's values.
+ pub key_hash: String,
+}
+
+/// A derived vector-index row to co-commit with an append (ADR-0155). Parsed from
+/// the entity's post-transition state for one declared `[[vector]]` path: the
+/// float vector and the model tag that partitions its space. Stores that maintain
+/// `entity_vector_index` write one row per `(decl_name, model_tag, entity_id)`; the
+/// blob is packed little-endian f32. Unlike a key row this has no uniqueness
+/// constraint — it is derived, rebuildable ranking state.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct EntityVectorRow {
+ /// The declared vector path's identifier (the `[[vector]]` block's `name`).
+ pub decl_name: String,
+ /// The model tag that partitions this vector's space (only same-tag vectors
+ /// are ever compared).
+ pub model_tag: String,
+ /// The float vector, exactly `dims` long.
+ pub vector: Vec,
+}
+
+/// Pack an `f32` slice to little-endian bytes — the `entity_vector_index` blob
+/// encoding shared by every backend (ADR-0155). Kept here beside [`EntityVectorRow`]
+/// so the stores and the kernel ranking agree on the byte layout.
+pub fn pack_f32_le(vector: &[f32]) -> Vec {
+ let mut bytes = Vec::with_capacity(vector.len() * 4);
+ for value in vector {
+ bytes.extend_from_slice(&value.to_le_bytes());
+ }
+ bytes
+}
+
+/// Unpack little-endian bytes back to `f32`. `None` if the byte length is not a
+/// multiple of 4, or if any component is not finite (both signal a corrupt blob),
+/// so a bad row is skipped rather than panicking or feeding a `NaN`/`inf` into the
+/// kNN ranking — where a `NaN` would sort ahead of every real score.
+pub fn unpack_f32_le(bytes: &[u8]) -> Option> {
+ if !bytes.len().is_multiple_of(4) {
+ return None;
+ }
+ let mut out = Vec::with_capacity(bytes.len() / 4);
+ for chunk in bytes.chunks_exact(4) {
+ let value = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
+ if !value.is_finite() {
+ return None;
+ }
+ out.push(value);
+ }
+ Some(out)
+}
+
+/// One candidate row returned from the vector index for a kNN read (ADR-0155):
+/// an entity and its packed vector for one `(tenant, type, decl, model_tag)`
+/// partition. The kernel — not the store — computes the metric over these in the
+/// store-supplied (entity-id) order, so ranking is identical across backends.
+#[derive(Debug, Clone, PartialEq)]
+pub struct EntityVectorCandidate {
+ /// The entity holding this vector.
+ pub entity_id: String,
+ /// The float vector, exactly `dims` long.
+ pub vector: Vec,
+}
+
+/// One stream append inside an atomic multi-journal append.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PersistenceAppend {
+ /// Persistence ID in the form `{tenant}:{entity_type}:{entity_id}`.
+ pub persistence_id: String,
+ /// Optimistic-concurrency sequence expected before this append.
+ pub expected_sequence: u64,
+ /// Events to append to this journal.
+ pub events: Vec,
+ /// Complete post-transition vector rows to co-commit for this stream.
+ #[serde(default)]
+ pub vector_rows: Vec,
+ /// Whether this stream's type declares vectors. When true, an empty
+ /// `vector_rows` purges candidates while retaining the live-write fence.
+ #[serde(default)]
+ pub reconcile_vectors: bool,
+ /// SHA-256 fingerprint of the IOA source that produced this stream's
+ /// post-transition index rows. Durable stores validate it against the spec
+ /// catalog in the same transaction as the journal append.
+ #[serde(default)]
+ pub spec_declaration_fingerprint: Option,
+}
+
+/// New sequence number for one stream after an atomic multi-journal append.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct PersistenceAppendResult {
+ /// Persistence ID that was appended.
+ pub persistence_id: String,
+ /// New highest sequence number for this journal.
+ pub sequence_nr: u64,
+}
diff --git a/crates/temper-runtime/src/persistence/mod.rs b/crates/temper-runtime/src/persistence/mod.rs
index 80c236adc..e7b67a222 100644
--- a/crates/temper-runtime/src/persistence/mod.rs
+++ b/crates/temper-runtime/src/persistence/mod.rs
@@ -1,5 +1,16 @@
use serde::{Deserialize, Serialize};
+mod indexing;
+pub use indexing::{
+ EntityKeyRow, EntityVectorCandidate, EntityVectorRow, PersistenceAppend,
+ PersistenceAppendResult, pack_f32_le, unpack_f32_le,
+};
+mod types;
+pub use types::{
+ CompositeEvent, CompositeEventSubWrite, EventMetadata, PersistenceEnvelope, PersistenceError,
+ storage_error,
+};
+
/// Event type used for the parent-journal record of a Composite action.
///
/// Concrete sub-write events remain the state-changing events on their target
@@ -7,27 +18,6 @@ use serde::{Deserialize, Serialize};
/// journals/idempotency keys that were committed atomically with it.
pub const COMPOSITE_EVENT_TYPE: &str = "CompositeEvent";
-/// Replay/audit record for one Composite action application.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct CompositeEvent {
- pub tenant: String,
- pub parent_entity_type: String,
- pub parent_entity_id: String,
- pub parent_action: String,
- pub composite_idempotency_key: String,
- pub sub_writes: Vec,
-}
-
-/// One concrete sub-write recorded in a [`CompositeEvent`].
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
-pub struct CompositeEventSubWrite {
- pub index: usize,
- pub entity_type: String,
- pub entity_id: String,
- pub action: String,
- pub idempotency_key: String,
-}
-
/// Marker trait for domain events.
/// Events must be serializable (for persistence) and Send + 'static (for async).
pub trait DomainEvent:
@@ -35,21 +25,6 @@ pub trait DomainEvent:
{
}
-/// Metadata attached to every persisted event.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct EventMetadata {
- /// Unique ID of this event.
- pub event_id: uuid::Uuid,
- /// ID of the command/message that caused this event.
- pub causation_id: uuid::Uuid,
- /// Correlation ID for tracing across actor boundaries.
- pub correlation_id: uuid::Uuid,
- /// Timestamp of persistence.
- pub timestamp: chrono::DateTime,
- /// Actor that produced this event.
- pub actor_id: String,
-}
-
/// Trait for event-sourced persistent actors.
/// Extends the base Actor trait with event journal and snapshot capabilities.
///
@@ -83,77 +58,6 @@ pub trait PersistentActor: Send + 'static {
}
}
-/// A declared-key row to co-commit with an append (ADR-0153). The entity claims
-/// `key_hash` for `key_name`; the store writes it into `entity_key_index` in the
-/// same transaction as the journal append, giving the read plane an `O(log n)`
-/// present/absent probe (the negative-existence access path, ARN-68).
-#[derive(Debug, Clone)]
-pub struct EntityKeyRow {
- /// The declared key's identifier (the `[[key]]` block's `name`).
- pub key_name: String,
- /// The canonical, type-tagged hash of the key's values.
- pub key_hash: String,
-}
-
-/// A derived vector-index row to co-commit with an append (ADR-0155). Parsed from
-/// the entity's post-transition state for one declared `[[vector]]` path: the
-/// float vector and the model tag that partitions its space. Stores that maintain
-/// `entity_vector_index` write one row per `(decl_name, model_tag, entity_id)`; the
-/// blob is packed little-endian f32. Unlike a key row this has no uniqueness
-/// constraint — it is derived, rebuildable ranking state.
-#[derive(Debug, Clone, PartialEq)]
-pub struct EntityVectorRow {
- /// The declared vector path's identifier (the `[[vector]]` block's `name`).
- pub decl_name: String,
- /// The model tag that partitions this vector's space (only same-tag vectors
- /// are ever compared).
- pub model_tag: String,
- /// The float vector, exactly `dims` long.
- pub vector: Vec,
-}
-
-/// Pack an `f32` slice to little-endian bytes — the `entity_vector_index` blob
-/// encoding shared by every backend (ADR-0155). Kept here beside [`EntityVectorRow`]
-/// so the stores and the kernel ranking agree on the byte layout.
-pub fn pack_f32_le(vector: &[f32]) -> Vec {
- let mut bytes = Vec::with_capacity(vector.len() * 4);
- for value in vector {
- bytes.extend_from_slice(&value.to_le_bytes());
- }
- bytes
-}
-
-/// Unpack little-endian bytes back to `f32`. `None` if the byte length is not a
-/// multiple of 4, or if any component is not finite (both signal a corrupt blob),
-/// so a bad row is skipped rather than panicking or feeding a `NaN`/`inf` into the
-/// kNN ranking — where a `NaN` would sort ahead of every real score.
-pub fn unpack_f32_le(bytes: &[u8]) -> Option> {
- if !bytes.len().is_multiple_of(4) {
- return None;
- }
- let mut out = Vec::with_capacity(bytes.len() / 4);
- for chunk in bytes.chunks_exact(4) {
- let value = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
- if !value.is_finite() {
- return None;
- }
- out.push(value);
- }
- Some(out)
-}
-
-/// One candidate row returned from the vector index for a kNN read (ADR-0155):
-/// an entity and its packed vector for one `(tenant, type, decl, model_tag)`
-/// partition. The kernel — not the store — computes the metric over these in the
-/// store-supplied (entity-id) order, so ranking is identical across backends.
-#[derive(Debug, Clone, PartialEq)]
-pub struct EntityVectorCandidate {
- /// The entity holding this vector.
- pub entity_id: String,
- /// The float vector, exactly `dims` long.
- pub vector: Vec,
-}
-
/// Trait for the event store backend (implemented by temper-store-postgres).
/// Uses desugared async-in-trait to enforce Send bounds on futures.
pub trait EventStore: Send + Sync + 'static {
@@ -184,6 +88,7 @@ pub trait EventStore: Send + Sync + 'static {
key_rows,
&[],
false,
+ None,
)
}
@@ -192,13 +97,19 @@ pub trait EventStore: Send + Sync + 'static {
/// journal append. This is the single co-commit entry point the entity actor
/// calls. The default ignores the index kinds and delegates to
/// [`EventStore::append`] — stores with a query plane that co-commit (postgres,
- /// sim) override it; Turso also overrides it to maintain the vector index
- /// write-behind (event first, index follows). When `reconcile_vectors` is true
+ /// sim, Turso) override it. When `reconcile_vectors` is true
/// (the entity's type declares ≥1 `[[vector]]` path) the store first DELETES all
/// of the entity's vector rows, then inserts `vector_rows` — so a delete
/// transition or a cleared vector/model property purges the stale rows instead of
/// leaving them to be ranked forever. The sequence and atomicity contract is
- /// identical to `append`.
+ /// identical to `append`. `spec_declaration_fingerprint` binds the writer's
+ /// compiled table to durable spec authority; indexing stores reject a stale
+ /// fingerprint before advancing the journal. Callers that reconcile vectors
+ /// must always provide it.
+ #[expect(
+ clippy::too_many_arguments,
+ reason = "journal, key, vector, and declaration data form one atomic storage boundary"
+ )]
fn append_with_index_rows(
&self,
persistence_id: &str,
@@ -207,27 +118,97 @@ pub trait EventStore: Send + Sync + 'static {
key_rows: &[EntityKeyRow],
vector_rows: &[EntityVectorRow],
reconcile_vectors: bool,
+ spec_declaration_fingerprint: Option<&str>,
) -> impl std::future::Future