From e6d8ea2e0aacb73f4a4775f55a5620ecb0d8b8e9 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 5 Aug 2026 13:10:27 -0600 Subject: [PATCH] feat(entity): add facet-link identity groups and ambiguity rescope read_identity_group_map exposes full collision groups so facets can refuse\nambiguous journal matches. rescope_facet_ambiguities rewrites facet\nscopes, origins, origin keys, prior choices, and ambiguity ids when a\nfacet directory is renamed, preserving recorded choices. --- core/crates/solstone-core-entity/src/lib.rs | 23 +- .../src/store/ambiguity.rs | 224 +++++++++++++++++- .../solstone-core-entity/src/store/map.rs | 55 ++++- .../solstone-core-entity/src/store/mod.rs | 10 +- .../solstone-core-entity/src/store/write.rs | 4 +- .../solstone-core-entity/src/store_tests.rs | 134 ++++++++++- 6 files changed, 420 insertions(+), 30 deletions(-) diff --git a/core/crates/solstone-core-entity/src/lib.rs b/core/crates/solstone-core-entity/src/lib.rs index 4b32067b0..e0dfaa4c7 100644 --- a/core/crates/solstone-core-entity/src/lib.rs +++ b/core/crates/solstone-core-entity/src/lib.rs @@ -22,17 +22,18 @@ pub use resolution::{ }; pub use slug::{MAX_ENTITY_SLUG_LENGTH, entity_slug}; pub use store::{ - AmbiguityChoiceEntity, AmbiguityChoiceRequest, AmbiguityObservation, EntityIdentityMap, - EntityIdentityRepairError, EntityIdentityRepairGuard, EntityIdentityRepairRefusal, - EntityIdentityRepairReport, EntityIdentityRepairSkip, EntityIdentityRepairSkipReason, - EntityOperationContext, EntityOperationKind, EntitySaveResult, EntityStoreError, - EntityWriteError, HistoryEvent, IdentityMapCacheLoad, IdentityMapLoser, IdentityMapLoserReason, - IdentitySnapshot, PreparedHistoryEvent, PreparedHistoryOutcome, classify_prepared_history, - guard_restore_does_not_cross_merge, guard_visible_event_collision, - load_resolved_ambiguity_choice, read_ambiguities, read_entity_identity, read_identity_map, - read_prepared_history, read_visible_history, record_ambiguity_choice, - record_ambiguity_observation, refresh_identity_map_cache, repair_entity_identities, - save_entity_identity, + AmbiguityChoiceEntity, AmbiguityChoiceRequest, AmbiguityObservation, + EntityAmbiguityRescopeError, EntityAmbiguityRescopeReport, EntityIdentityGroupMap, + EntityIdentityMap, EntityIdentityRepairError, EntityIdentityRepairGuard, + EntityIdentityRepairRefusal, EntityIdentityRepairReport, EntityIdentityRepairSkip, + EntityIdentityRepairSkipReason, EntityOperationContext, EntityOperationKind, EntitySaveResult, + EntityStoreError, EntityWriteError, HistoryEvent, IdentityMapCacheLoad, IdentityMapLoser, + IdentityMapLoserReason, IdentitySnapshot, PreparedHistoryEvent, PreparedHistoryOutcome, + classify_prepared_history, guard_restore_does_not_cross_merge, guard_visible_event_collision, + load_resolved_ambiguity_choice, read_ambiguities, read_entity_identity, + read_identity_group_map, read_identity_map, read_prepared_history, read_visible_history, + record_ambiguity_choice, record_ambiguity_observation, refresh_identity_map_cache, + repair_entity_identities, rescope_facet_ambiguities, save_entity_identity, }; pub use trust_lock::{EntityTrustLock, EntityTrustLockError, hold_entity_trust_lock}; diff --git a/core/crates/solstone-core-entity/src/store/ambiguity.rs b/core/crates/solstone-core-entity/src/store/ambiguity.rs index 178fb847a..012e6f848 100644 --- a/core/crates/solstone-core-entity/src/store/ambiguity.rs +++ b/core/crates/solstone-core-entity/src/store/ambiguity.rs @@ -1,16 +1,236 @@ // SPDX-License-Identifier: AGPL-3.0-only // Copyright (c) 2026 sol pbc +use std::error::Error; +use std::fmt; use std::path::Path; use serde_json::{Map, Value}; -use solstone_core_journal_io::{MalformedPolicy, read_text}; +use solstone_core_journal_io::AtomicWriteError; +use solstone_core_journal_io::AtomicWriteOptions; +use solstone_core_journal_io::LockError; +use solstone_core_journal_io::LockOptions; +use solstone_core_journal_io::MalformedPolicy; +use solstone_core_journal_io::hold_lock; +use solstone_core_journal_io::read_text; +use solstone_core_journal_io::write_text; + +use crate::{EntityTrustLockError, hold_entity_trust_lock}; use super::error::EntityStoreError; use super::paths::ambiguities_path; +use super::write::{origin_key, serialize_ambiguity_rows}; const AMBIGUITY_SCHEMA_VERSION: u64 = 1; +/// The ambiguity identifiers rewritten by a facet-directory rename. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EntityAmbiguityRescopeReport { + pub rewritten_ambiguity_ids: Vec, +} + +/// Failure while rescoping facet references in durable ambiguity rows. +#[derive(Debug)] +pub enum EntityAmbiguityRescopeError { + TrustLock(EntityTrustLockError), + Read(EntityStoreError), + Lock(LockError), + Write(AtomicWriteError), + InvalidRow { + ambiguity_id: Option, + detail: String, + }, +} + +impl fmt::Display for EntityAmbiguityRescopeError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::TrustLock(error) => error.fmt(formatter), + Self::Read(error) => error.fmt(formatter), + Self::Lock(error) => error.fmt(formatter), + Self::Write(error) => error.fmt(formatter), + Self::InvalidRow { + ambiguity_id, + detail, + } => write!( + formatter, + "invalid ambiguity row {}: {detail}", + ambiguity_id.as_deref().unwrap_or("") + ), + } + } +} + +impl Error for EntityAmbiguityRescopeError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + match self { + Self::TrustLock(error) => Some(error), + Self::Read(error) => Some(error), + Self::Lock(error) => Some(error), + Self::Write(error) => Some(error), + Self::InvalidRow { .. } => None, + } + } +} + +/// Rescope every strict ambiguity row that references `old_facet`. +pub fn rescope_facet_ambiguities( + journal_root: &Path, + old_facet: &str, + new_facet: &str, +) -> Result { + let _trust = + hold_entity_trust_lock(journal_root).map_err(EntityAmbiguityRescopeError::TrustLock)?; + let path = ambiguities_path(journal_root).map_err(EntityAmbiguityRescopeError::Read)?; + let _lock = hold_lock( + &path, + LockOptions { + mode: Some(0o600), + ..LockOptions::default() + }, + ) + .map_err(EntityAmbiguityRescopeError::Lock)?; + let mut rows = read_ambiguities(journal_root, MalformedPolicy::Raise) + .map_err(EntityAmbiguityRescopeError::Read)?; + let mut rewritten_ambiguity_ids = Vec::new(); + + for row in &mut rows { + let ambiguity_id = row + .get("ambiguity_id") + .and_then(Value::as_str) + .map(str::to_owned); + let object = + row.as_object_mut() + .ok_or_else(|| EntityAmbiguityRescopeError::InvalidRow { + ambiguity_id: ambiguity_id.clone(), + detail: "row is not an object".to_owned(), + })?; + if !row_references_facet(object, old_facet) { + continue; + } + + if let Some(scope) = object.get_mut("scope").and_then(Value::as_object_mut) + && scope.get("facet").and_then(Value::as_str) == Some(old_facet) + { + scope.insert("facet".to_owned(), Value::String(new_facet.to_owned())); + } + if let Some(origins) = object.get_mut("origins").and_then(Value::as_array_mut) { + for origin in origins.iter_mut() { + rescope_origin(origin, old_facet, new_facet); + } + let keys = origins + .iter() + .map(|origin| { + origin_key(origin).map(Value::String).map_err(|error| { + EntityAmbiguityRescopeError::InvalidRow { + ambiguity_id: ambiguity_id.clone(), + detail: error.to_string(), + } + }) + }) + .collect::, _>>()?; + object.insert("origin_keys".to_owned(), Value::Array(keys)); + } + if let Some(priors) = object + .get_mut("audit") + .and_then(Value::as_object_mut) + .and_then(|audit| audit.get_mut("prior_choices")) + .and_then(Value::as_array_mut) + { + for prior in priors { + if let Some(origin) = prior + .as_object_mut() + .and_then(|prior| prior.get_mut("replaced_by_origin")) + { + rescope_origin(origin, old_facet, new_facet); + } + } + } + + let scope_key = object.get("scope").and_then(scope_key).ok_or_else(|| { + EntityAmbiguityRescopeError::InvalidRow { + ambiguity_id: ambiguity_id.clone(), + detail: "invalid scope".to_owned(), + } + })?; + let normalized_query = + non_empty_string(object.get("normalized_query")).ok_or_else(|| { + EntityAmbiguityRescopeError::InvalidRow { + ambiguity_id: ambiguity_id.clone(), + detail: "missing normalized_query".to_owned(), + } + })?; + let rewritten_id = crate::ambiguity_id(&format!("{scope_key}|{normalized_query}")); + object.insert( + "ambiguity_id".to_owned(), + Value::String(rewritten_id.clone()), + ); + validate_row(object).map_err(|detail| EntityAmbiguityRescopeError::InvalidRow { + ambiguity_id, + detail: detail.to_owned(), + })?; + rewritten_ambiguity_ids.push(rewritten_id); + } + + let contents = serialize_ambiguity_rows(&rows).map_err(EntityAmbiguityRescopeError::Write)?; + write_text(&path, &contents, AtomicWriteOptions { mode: Some(0o600) }) + .map_err(EntityAmbiguityRescopeError::Write)?; + Ok(EntityAmbiguityRescopeReport { + rewritten_ambiguity_ids, + }) +} + +fn row_references_facet(row: &Map, facet: &str) -> bool { + row.get("scope") + .and_then(Value::as_object) + .and_then(|scope| scope.get("facet")) + .and_then(Value::as_str) + == Some(facet) + || row + .get("origins") + .and_then(Value::as_array) + .is_some_and(|origins| origins.iter().any(|origin| origin_facet_is(origin, facet))) + || row + .get("audit") + .and_then(Value::as_object) + .and_then(|audit| audit.get("prior_choices")) + .and_then(Value::as_array) + .is_some_and(|priors| { + priors.iter().any(|prior| { + prior + .as_object() + .and_then(|prior| prior.get("replaced_by_origin")) + .is_some_and(|origin| origin_facet_is(origin, facet)) + }) + }) +} + +fn origin_facet_is(origin: &Value, facet: &str) -> bool { + origin + .as_object() + .and_then(|origin| origin.get("facet")) + .and_then(Value::as_str) + == Some(facet) +} + +fn rescope_origin(origin: &mut Value, old_facet: &str, new_facet: &str) { + let Some(origin) = origin.as_object_mut() else { + return; + }; + if origin.get("facet").and_then(Value::as_str) == Some(old_facet) { + origin.insert("facet".to_owned(), Value::String(new_facet.to_owned())); + } + if let Some(path) = origin.get("path").and_then(Value::as_str) { + let old_segment = format!("facets/{old_facet}/"); + if path.contains(&old_segment) { + origin.insert( + "path".to_owned(), + Value::String(path.replace(&old_segment, &format!("facets/{new_facet}/"))), + ); + } + } +} + /// Read durable ambiguity rows with Python-compatible strictness behavior. pub fn read_ambiguities( journal_root: &Path, @@ -285,7 +505,7 @@ pub(super) fn validate_row(row: &Map) -> Result<(), &'static str> Ok(()) } -fn scope_key(scope: &Value) -> Option { +pub(super) fn scope_key(scope: &Value) -> Option { let scope = scope.as_object()?; match scope.get("kind").and_then(Value::as_str) { Some("journal") if scope.get("facet").is_none_or(Value::is_null) => { diff --git a/core/crates/solstone-core-entity/src/store/map.rs b/core/crates/solstone-core-entity/src/store/map.rs index ac8376fa2..1e7d257e2 100644 --- a/core/crates/solstone-core-entity/src/store/map.rs +++ b/core/crates/solstone-core-entity/src/store/map.rs @@ -16,6 +16,16 @@ pub struct EntityIdentityMap { pub losers: Vec, } +/// In-memory grouping from effective identity id to every matching directory. +/// +/// Each group is ordered deterministically: explicit written identities first, +/// then directory fallbacks, with lexical directory-name tie breaking. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EntityIdentityGroupMap { + pub groups: HashMap>, + pub losers: Vec, +} + /// An entity omitted from the identity map with its visible reason. #[derive(Debug, Clone, PartialEq, Eq)] pub struct IdentityMapLoser { @@ -45,6 +55,32 @@ struct Candidate { /// Build a deterministic, non-persisted durable identity lookup. pub fn read_identity_map(journal_root: &Path) -> Result { + let group_map = read_identity_group_map(journal_root)?; + let mut resolved = HashMap::new(); + let mut losers = group_map.losers; + for (identity_id, candidates) in group_map.groups { + let winner = candidates + .first() + .expect("non-empty identity candidate group"); + resolved.insert(identity_id, winner.clone()); + losers.extend( + candidates + .into_iter() + .skip(1) + .map(|entity_dir| IdentityMapLoser { + entity_dir, + reason: IdentityMapLoserReason::CollisionLost, + }), + ); + } + losers.sort_by(|left, right| left.entity_dir.cmp(&right.entity_dir)); + Ok(EntityIdentityMap { resolved, losers }) +} + +/// Build deterministic effective-identity groups without discarding collisions. +pub fn read_identity_group_map( + journal_root: &Path, +) -> Result { let entities_dir = contained_path(journal_root, "entities")?; let mut candidates = Vec::new(); let mut losers = Vec::new(); @@ -84,22 +120,21 @@ pub fn read_identity_map(journal_root: &Path) -> Result Option { } } -fn origin_key(origin: &Value) -> Result { +pub(super) fn origin_key(origin: &Value) -> Result { let mut origin = origin.clone(); sort_json_keys(&mut origin); serialize_value(&origin, AsciiCompactFormatter).map_err(|error| { @@ -1005,7 +1005,7 @@ fn origin_key(origin: &Value) -> Result { }) } -fn serialize_ambiguity_rows(rows: &[Value]) -> Result { +pub(super) fn serialize_ambiguity_rows(rows: &[Value]) -> Result { let mut contents = String::new(); for row in rows { let line = diff --git a/core/crates/solstone-core-entity/src/store_tests.rs b/core/crates/solstone-core-entity/src/store_tests.rs index 975c75a8c..c31522547 100644 --- a/core/crates/solstone-core-entity/src/store_tests.rs +++ b/core/crates/solstone-core-entity/src/store_tests.rs @@ -19,13 +19,14 @@ use solstone_core_journal_io::{ use crate::{ AmbiguityChoiceEntity, AmbiguityChoiceRequest, AmbiguityObservation, EntityIdentityRepairError, EntityIdentityRepairGuard, EntityIdentityRepairSkipReason, EntityWriteError, - IdentityMapLoserReason, PreparedHistoryOutcome, classify_prepared_history, + IdentityMapLoserReason, PreparedHistoryOutcome, ambiguity_id, classify_prepared_history, guard_restore_does_not_cross_merge, guard_visible_event_collision, load_resolved_ambiguity_choice, read_ambiguities, read_entity_identity, read_identity_map, read_prepared_history, read_visible_history, record_ambiguity_choice, record_ambiguity_observation, refresh_identity_map_cache, repair_entity_identities, - save_entity_identity, save_entity_identity_with_timeout, set_forced_identity_write_failure, - set_repair_identity_write_failure_on_attempt, write_history_event_json_for_test, + rescope_facet_ambiguities, save_entity_identity, save_entity_identity_with_timeout, + set_forced_identity_write_failure, set_repair_identity_write_failure_on_attempt, + write_history_event_json_for_test, }; const ENTITY_STORE_FIXTURE: &str = include_str!(concat!( @@ -140,6 +141,48 @@ fn identity_map_distinguishes_written_self_id_from_directory_fallback() { })); } +#[test] +fn identity_group_map_retains_every_collision_candidate_in_precedence_order() { + let temporary = TempDir::new(); + write_json( + temporary.path(), + "entities/written_z/entity.json", + &json!({"id": "shared"}), + ); + write_json( + temporary.path(), + "entities/written_a/entity.json", + &json!({"id": "shared"}), + ); + write_json(temporary.path(), "entities/shared/entity.json", &json!({})); + + let groups = crate::read_identity_group_map(temporary.path()).unwrap(); + + assert_eq!( + groups.groups.get("shared"), + Some(&vec![ + "written_a".to_owned(), + "written_z".to_owned(), + "shared".to_owned(), + ]) + ); + assert!(groups.losers.is_empty()); + let resolved = read_identity_map(temporary.path()).unwrap(); + assert_eq!( + resolved.resolved.get("shared"), + Some(&"written_a".to_owned()) + ); + assert_eq!( + resolved + .losers + .iter() + .filter(|loser| loser.reason == IdentityMapLoserReason::CollisionLost) + .map(|loser| loser.entity_dir.as_str()) + .collect::>(), + vec!["shared", "written_z"] + ); +} + #[test] fn prepared_history_is_discovered_from_staging_directory() { let fixture = fixture(); @@ -570,6 +613,91 @@ fn ambiguity_fixture_rows_obey_strict_validation_in_order() { } } +#[test] +fn ambiguity_rescope_updates_a_facet_scope_and_its_identifier() { + let temporary = TempDir::new(); + let mut row = valid_ambiguity_row(); + row["scope"] = json!({"kind": "facet", "facet": "old"}); + let normalized_query = row["normalized_query"].as_str().unwrap().to_owned(); + row["ambiguity_id"] = json!(ambiguity_id(&format!("facet:old|{normalized_query}"))); + write_json(temporary.path(), "entities/ambiguities.jsonl", &row); + + let report = rescope_facet_ambiguities(temporary.path(), "old", "new").unwrap(); + let rows = read_ambiguities(temporary.path(), MalformedPolicy::Raise).unwrap(); + + assert_eq!(report.rewritten_ambiguity_ids.len(), 1); + assert_eq!(rows[0]["scope"]["facet"], "new"); + assert_eq!( + rows[0]["ambiguity_id"], + ambiguity_id(&format!("facet:new|{normalized_query}")) + ); +} + +#[test] +fn ambiguity_rescope_updates_facet_origins_without_changing_journal_scope() { + let temporary = TempDir::new(); + let mut row = valid_ambiguity_row(); + row["origins"] = json!([{ + "lane": "facet", + "facet": "old", + "path": "facets/old/entities/alice/entity.json" + }]); + row["origin_keys"] = json!(["stale"]); + write_json(temporary.path(), "entities/ambiguities.jsonl", &row); + + let report = rescope_facet_ambiguities(temporary.path(), "old", "new").unwrap(); + let rows = read_ambiguities(temporary.path(), MalformedPolicy::Raise).unwrap(); + + assert_eq!( + report.rewritten_ambiguity_ids, + vec![row["ambiguity_id"].as_str().unwrap().to_owned()] + ); + assert_eq!(rows[0]["scope"]["kind"], "journal"); + assert!(rows[0]["scope"].get("facet").is_none()); + assert_eq!(rows[0]["origins"][0]["facet"], "new"); + assert_eq!( + rows[0]["origins"][0]["path"], + "facets/new/entities/alice/entity.json" + ); + assert_eq!( + rows[0]["origin_keys"][0], + "{\"facet\":\"new\",\"lane\":\"facet\",\"path\":\"facets/new/entities/alice/entity.json\"}" + ); +} + +#[test] +fn ambiguity_rescope_updates_prior_choice_replacement_origins() { + let temporary = TempDir::new(); + let mut row = valid_ambiguity_row(); + row["audit"]["prior_choices"] = json!([{ + "resolved_entity_id": "alice_chen", + "resolved_at": "2026-08-04T00:00:00Z", + "replaced_at": "2026-08-05T00:00:00Z", + "replaced_by_origin": { + "lane": "facet", + "facet": "old", + "path": "facets/old/entities/alice/entity.json" + } + }]); + write_json(temporary.path(), "entities/ambiguities.jsonl", &row); + + let report = rescope_facet_ambiguities(temporary.path(), "old", "new").unwrap(); + let rows = read_ambiguities(temporary.path(), MalformedPolicy::Raise).unwrap(); + + assert_eq!( + report.rewritten_ambiguity_ids, + vec![row["ambiguity_id"].as_str().unwrap().to_owned()] + ); + assert_eq!( + rows[0]["audit"]["prior_choices"][0]["replaced_by_origin"]["facet"], + "new" + ); + assert_eq!( + rows[0]["audit"]["prior_choices"][0]["replaced_by_origin"]["path"], + "facets/new/entities/alice/entity.json" + ); +} + #[test] fn ambiguity_reader_distinguishes_malformed_non_object_and_lenient_rows() { let fixture = fixture(); -- 2.51.2