diff --git a/crates/khive-mcp/src/serve.rs b/crates/khive-mcp/src/serve.rs index 2e83ea2b..9db77d22 100644 --- a/crates/khive-mcp/src/serve.rs +++ b/crates/khive-mcp/src/serve.rs @@ -1133,6 +1133,11 @@ pub fn build_registry_for_multi_backend( } registry.call_register_embedders(&default_runtime); registry.call_register_entity_type_validators(&default_runtime); + // #750 fix-round 1: install pack-owned note-mutation hooks (currently + // only khive-pack-memory's warm-ANN-cache invalidation) so KG's + // update/delete verbs notify caching packs even though there is no + // crate-level dependency between them. + registry.call_register_note_mutation_hooks(&default_runtime); let backend_for_pack: HashMap<&str, &StorageBackend> = per_pack_runtimes_local .iter() diff --git a/crates/khive-mcp/src/server.rs b/crates/khive-mcp/src/server.rs index 1ea5fadf..dbe2382d 100644 --- a/crates/khive-mcp/src/server.rs +++ b/crates/khive-mcp/src/server.rs @@ -336,6 +336,11 @@ impl KhiveMcpServer { // entity-type validation is active at the runtime layer for all write // paths, including direct `create_many` callers that bypass the handler. registry.call_register_entity_type_validators(&runtime); + // #750 fix-round 1: install pack-owned note-mutation hooks (currently + // only khive-pack-memory's warm-ANN-cache invalidation) so KG's + // update/delete verbs notify caching packs even though there is no + // crate-level dependency between them. + registry.call_register_note_mutation_hooks(&runtime); // Apply pack-auxiliary schema plans at startup so pack tables are // present before any handler runs. Errors are logged but not propagated // so a single pack's schema failure cannot abort startup. diff --git a/crates/khive-pack-kg/src/handlers/common.rs b/crates/khive-pack-kg/src/handlers/common.rs index ffd86af2..9402d788 100644 --- a/crates/khive-pack-kg/src/handlers/common.rs +++ b/crates/khive-pack-kg/src/handlers/common.rs @@ -731,6 +731,38 @@ pub(crate) fn tags_match_any(entity_tags: &[String], wanted: &[String]) -> bool .any(|tag| wanted.iter().any(|w| tag.eq_ignore_ascii_case(w))) } +/// Merge the top-level `tags` create-param into `properties["tags"]` for a +/// note. Notes have no dedicated tags column (see search.rs's `tag_filter` +/// handling) — `properties["tags"]` is the storage convention already used +/// by `memory.remember` (khive-pack-memory/src/handlers/remember.rs) and by +/// this pack's own `search`/`list` note-tag filters. Without this merge, +/// `create(kind=note, tags=[...])` silently dropped the tags (#747). +/// +/// Precedence: an empty/absent `tags` param leaves `properties` untouched. +/// A non-empty `tags` param always WINS over any `properties["tags"]` the +/// caller also supplied — the top-level, typed param is the more explicit +/// signal, so it overwrites rather than merges with a same-named nested key. +pub(crate) fn merge_note_tags( + properties: Option, + tags: Option>, +) -> Result, RuntimeError> { + let tags = match tags { + Some(t) if !t.is_empty() => t, + _ => return Ok(properties), + }; + let mut obj = match properties { + None => serde_json::Map::new(), + Some(Value::Object(m)) => m, + Some(other) => { + return Err(RuntimeError::InvalidInput(format!( + "create: note `tags` cannot be merged into non-object `properties` (got {other})" + ))); + } + }; + obj.insert("tags".to_string(), json!(tags)); + Ok(Some(Value::Object(obj))) +} + // ---- Handler helpers ---- pub(crate) fn parse_entity_policy(s: &str) -> Result { diff --git a/crates/khive-pack-kg/src/handlers/create.rs b/crates/khive-pack-kg/src/handlers/create.rs index 6816a5a6..a2d99292 100644 --- a/crates/khive-pack-kg/src/handlers/create.rs +++ b/crates/khive-pack-kg/src/handlers/create.rs @@ -335,6 +335,7 @@ impl KgPack { for s in p.annotates.unwrap_or_default() { annotates.push(resolve_uuid_unfiltered(&s, &self.runtime, token).await?); } + let properties = super::common::merge_note_tags(p.properties, p.tags)?; let note = self .runtime .create_note( @@ -343,7 +344,7 @@ impl KgPack { p.name.as_deref(), &content, p.salience, - p.properties, + properties, annotates, ) .await?; diff --git a/crates/khive-pack-kg/tests/integration.rs b/crates/khive-pack-kg/tests/integration.rs index 8bc533d7..ce772201 100644 --- a/crates/khive-pack-kg/tests/integration.rs +++ b/crates/khive-pack-kg/tests/integration.rs @@ -3447,6 +3447,172 @@ async fn traverse_excludes_soft_deleted_entity() { ); } +// ---- #748: neighbors() must also screen soft-deleted NOTE targets ---- + +/// #748: a soft-deleted note reached via an `annotates` edge must not appear +/// in `neighbors()`. The original Fix-2 screen (`deleted_entity_ids`) only +/// queried the `entities` table, so a soft-deleted note-kind neighbor leaked +/// through. View-layer only — no edges are touched, no data is mutated. +#[tokio::test] +async fn neighbors_excludes_soft_deleted_note() { + let pack = pack(); + + let entity = pack + .dispatch( + "create", + json!({"kind": "entity", "name": "Nvk748Target", "entity_kind": "concept"}), + ) + .await + .expect("create entity must succeed"); + let entity_id = entity + .get("id") + .and_then(Value::as_str) + .unwrap() + .to_string(); + + // Annotating note: creates a note --annotates--> entity edge. + let note = pack + .dispatch( + "create", + json!({ + "kind": "note", + "content": "nvk748 note that will be soft-deleted", + "annotates": [entity_id.clone()], + }), + ) + .await + .expect("create annotating note must succeed"); + let note_id = note.get("id").and_then(Value::as_str).unwrap().to_string(); + + // A second, non-deleted annotating note — must remain visible after the + // first note is soft-deleted (regression: the fix must not over-filter). + let live_note = pack + .dispatch( + "create", + json!({ + "kind": "note", + "content": "nvk748 note that stays alive", + "annotates": [entity_id.clone()], + }), + ) + .await + .expect("create second annotating note must succeed"); + let live_note_id = live_note + .get("id") + .and_then(Value::as_str) + .unwrap() + .to_string(); + + // Before delete: both notes are visible as incoming neighbors of the entity. + let before = pack + .dispatch( + "neighbors", + json!({"node_id": entity_id, "direction": "in"}), + ) + .await + .expect("neighbors (pre-delete) must succeed"); + let before_ids: Vec<&str> = before + .as_array() + .expect("neighbors must be array") + .iter() + .filter_map(|v| v.get("id").and_then(Value::as_str)) + .collect(); + assert!( + before_ids.iter().any(|&id| id == note_id), + "#748 setup: note must appear as a neighbor before delete; ids={before_ids:?}" + ); + assert!( + before_ids.iter().any(|&id| id == live_note_id), + "#748 setup: live_note must appear as a neighbor before delete; ids={before_ids:?}" + ); + + // Soft-delete the first note. + pack.dispatch("delete", json!({"id": note_id, "kind": "note"})) + .await + .expect("delete note must succeed"); + + let after = pack + .dispatch( + "neighbors", + json!({"node_id": entity_id, "direction": "in"}), + ) + .await + .expect("neighbors (post-delete) must succeed"); + let after_ids: Vec<&str> = after + .as_array() + .expect("neighbors must be array") + .iter() + .filter_map(|v| v.get("id").and_then(Value::as_str)) + .collect(); + assert!( + !after_ids.iter().any(|&id| id == note_id), + "#748: soft-deleted note must not appear in neighbors(); ids={after_ids:?}" + ); + assert!( + after_ids.iter().any(|&id| id == live_note_id), + "#748: non-deleted note must still appear in neighbors(); ids={after_ids:?}" + ); +} + +/// #748 regression guard: the pre-existing entity-only soft-delete screen +/// (Fix 2) must keep working after extending `deleted_entity_ids` to also +/// cover notes. +#[tokio::test] +async fn neighbors_still_excludes_soft_deleted_entity_after_note_fix() { + let pack = pack(); + + let alive = pack + .dispatch( + "create", + json!({"kind": "entity", "name": "Nvk748AliveEntity", "entity_kind": "concept"}), + ) + .await + .expect("create alive must succeed"); + let alive_id = alive.get("id").and_then(Value::as_str).unwrap().to_string(); + + let deleted = pack + .dispatch( + "create", + json!({"kind": "entity", "name": "Nvk748DeletedEntity", "entity_kind": "concept"}), + ) + .await + .expect("create deleted must succeed"); + let deleted_id = deleted + .get("id") + .and_then(Value::as_str) + .unwrap() + .to_string(); + + pack.dispatch( + "link", + json!({"source_id": alive_id, "target_id": deleted_id, "relation": "contains"}), + ) + .await + .expect("link must succeed"); + + pack.dispatch("delete", json!({"id": deleted_id, "kind": "entity"})) + .await + .expect("delete must succeed"); + + let neighbors = pack + .dispatch( + "neighbors", + json!({"node_id": alive_id, "direction": "out"}), + ) + .await + .expect("neighbors must succeed"); + let ids: Vec<&str> = neighbors + .as_array() + .expect("neighbors must be array") + .iter() + .filter_map(|v| v.get("id").and_then(Value::as_str)) + .collect(); + assert!( + !ids.iter().any(|&id| id == deleted_id), + "#748 regression: soft-deleted entity must still be excluded; ids={ids:?}" + ); +} + // ---- K-C1: link response preserves caller source/target for symmetric relations ---- /// K-C1 regression: for `competes_with` (symmetric), the runtime canonicalises @@ -9886,3 +10052,160 @@ async fn stats_reports_edges_by_relation() { assert_eq!(by_relation.get("extends").and_then(Value::as_u64), Some(2)); assert_eq!(by_relation.get("enables").and_then(Value::as_u64), Some(1)); } + +// ── #747 regression: create(kind=note, tags=[...]) must persist tags ────── + +/// #747: top-level `tags` on `create(kind="note", ...)` must round-trip +/// through `properties["tags"]` (notes have no dedicated tags column — same +/// convention already used by `memory.remember` and honored by this pack's +/// own `search`/`list` note-tag filters). Before the fix, `tags` was +/// whitelisted as an accepted param but never read in the note branch, so it +/// was silently dropped. +#[tokio::test] +async fn create_note_persists_top_level_tags_into_properties() { + let f = pack(); + + let created = f + .dispatch( + "create", + json!({ + "kind": "note", + "note_kind": "observation", + "content": "note nvk747 top-level tags must persist", + "tags": ["alpha", "beta"], + }), + ) + .await + .expect("create note with tags must succeed"); + let id = created.get("id").and_then(Value::as_str).expect("id"); + + let fetched = f + .dispatch("get", json!({"id": id})) + .await + .expect("get must succeed"); + let stored_tags: Vec<&str> = fetched + .get("properties") + .and_then(|p| p.get("tags")) + .and_then(Value::as_array) + .expect("properties.tags must be present") + .iter() + .filter_map(Value::as_str) + .collect(); + assert_eq!( + stored_tags, + vec!["alpha", "beta"], + "#747: note tags must persist into properties[\"tags\"]" + ); +} + +/// #747: absent/empty `tags` must leave `properties` unchanged — no spurious +/// `properties["tags"]` key when the caller never supplied tags. +#[tokio::test] +async fn create_note_without_tags_leaves_properties_unchanged() { + let f = pack(); + + let created = f + .dispatch( + "create", + json!({ + "kind": "note", + "note_kind": "observation", + "content": "note nvk747b no tags supplied", + "properties": {"domain": "test747"}, + }), + ) + .await + .expect("create note without tags must succeed"); + let id = created.get("id").and_then(Value::as_str).expect("id"); + + let fetched = f + .dispatch("get", json!({"id": id})) + .await + .expect("get must succeed"); + let props = fetched + .get("properties") + .and_then(Value::as_object) + .expect("properties must be present"); + assert_eq!(props.get("domain").and_then(Value::as_str), Some("test747")); + assert!( + !props.contains_key("tags"), + "#747: absent tags param must not inject a properties[\"tags\"] key; got {props:?}" + ); + + // Empty tags array is equivalent to absent — must not inject the key either. + let created2 = f + .dispatch( + "create", + json!({ + "kind": "note", + "note_kind": "observation", + "content": "note nvk747c empty tags array", + "tags": [], + }), + ) + .await + .expect("create note with empty tags must succeed"); + let id2 = created2.get("id").and_then(Value::as_str).expect("id"); + let fetched2 = f + .dispatch("get", json!({"id": id2})) + .await + .expect("get must succeed"); + let has_tags_key = fetched2 + .get("properties") + .and_then(Value::as_object) + .is_some_and(|p| p.contains_key("tags")); + assert!( + !has_tags_key, + "#747: empty tags array must not inject a properties[\"tags\"] key" + ); +} + +/// #747: when the caller supplies BOTH a top-level `tags` param AND a +/// conflicting `properties.tags`, the top-level `tags` param wins (documented +/// precedence rule — the explicit, typed param overrides the nested-object +/// value rather than silently merging or erroring). +#[tokio::test] +async fn create_note_top_level_tags_wins_over_properties_tags_conflict() { + let f = pack(); + + let created = f + .dispatch( + "create", + json!({ + "kind": "note", + "note_kind": "observation", + "content": "note nvk747d conflicting tags sources", + "properties": {"tags": ["from-properties"], "domain": "test747d"}, + "tags": ["from-top-level"], + }), + ) + .await + .expect("create note with conflicting tags must succeed"); + let id = created.get("id").and_then(Value::as_str).expect("id"); + + let fetched = f + .dispatch("get", json!({"id": id})) + .await + .expect("get must succeed"); + let props = fetched + .get("properties") + .and_then(Value::as_object) + .expect("properties must be present"); + let stored_tags: Vec<&str> = props + .get("tags") + .and_then(Value::as_array) + .expect("tags must be present") + .iter() + .filter_map(Value::as_str) + .collect(); + assert_eq!( + stored_tags, + vec!["from-top-level"], + "#747: top-level `tags` param must win over properties.tags on conflict" + ); + // The rest of `properties` (unrelated keys) must survive the merge untouched. + assert_eq!( + props.get("domain").and_then(Value::as_str), + Some("test747d") + ); +} diff --git a/crates/khive-pack-memory/src/ann.rs b/crates/khive-pack-memory/src/ann.rs index 0e411477..92d643a5 100644 --- a/crates/khive-pack-memory/src/ann.rs +++ b/crates/khive-pack-memory/src/ann.rs @@ -42,6 +42,14 @@ pub(crate) struct AnnBridge { /// Used by the recall retry gate to short-circuit when the global index /// contains no vectors outside the caller's visible namespace set. pub(crate) namespace_set: HashSet, + /// Write-generation this build's corpus snapshot was taken at or after + /// (#750). Captured from `AnnState::generations` BEFORE the corpus scan + /// begins — see `ensure_ann_for_model_inner`. Cache installation compares + /// this against the currently-installed entry's generation instead of + /// blindly `or_insert`-ing, so a build that snapshotted a stale (older) + /// corpus can never clobber an already-installed fresher build, and a + /// build that snapshotted a fresher corpus always wins. + pub(crate) generation: u64, } /// Shared ANN state: per-`(namespace, model)` indexes with at-most-one-background-build guard. @@ -57,6 +65,19 @@ pub(crate) struct AnnState { /// in-flight attempt to finish, so only one caller ever emits the /// `PhaseStarted`/`PhaseCompleted` pair for a given model. model_locks: Mutex>>>, + /// Monotonic per-model write-generation counter (#750). Bumped by + /// `bump_generation` whenever a write may have changed a model's corpus + /// (`memory.remember`'s affected-models loop, right where the old + /// `invalidate_namespace` clear already ran). `ensure_ann_for_model` + /// snapshots the current value for its model BEFORE doing anything else + /// — including before its own "already loaded" fast path and before the + /// corpus scan — and stamps it on the resulting `AnnBridge`. Cache + /// install then only replaces an existing entry when the candidate's + /// generation is >= the installed entry's, instead of the old + /// `entry(key).or_insert(...)`, which always kept whichever build + /// happened to acquire the per-model lock first even if it had + /// snapshotted a now-stale corpus. + generations: Mutex>, /// Counts how many times `search_loaded` returned a warm hit. Test-only; /// call `reset_warm_route_count()` between operations to isolate counts. #[cfg(test)] @@ -70,11 +91,78 @@ pub(crate) fn new_shared() -> SharedAnn { indexes: RwLock::new(HashMap::new()), warming: Mutex::new(HashSet::new()), model_locks: Mutex::new(HashMap::new()), + generations: Mutex::new(HashMap::new()), #[cfg(test)] warm_route_count: AtomicUsize::new(0), }) } +/// Bump `key`'s write-generation counter and return the NEW value (#750). +/// Called by `memory.remember` for every model whose corpus the write may +/// have affected, right alongside the existing cache invalidation. +pub(crate) async fn bump_generation(ann: &SharedAnn, key: &AnnKey) -> u64 { + let mut gens = ann.generations.lock().await; + let slot = gens.entry(key.clone()).or_insert(0); + *slot += 1; + *slot +} + +/// Read `key`'s current write-generation counter (0 if never bumped). +async fn current_generation(ann: &SharedAnn, key: &AnnKey) -> u64 { + ann.generations.lock().await.get(key).copied().unwrap_or(0) +} + +/// True when the currently-installed entry for `key` (if any) is fresh +/// enough to satisfy a caller whose write-generation floor is +/// `min_generation` — i.e. the installed build's own generation is >= what +/// the caller needs. `Ok(None)`-style "cache miss" callers should treat +/// `false` as "must (re)build", not merely "absent". +async fn installed_is_fresh(ann: &SharedAnn, key: &AnnKey, min_generation: u64) -> bool { + ann.indexes + .read() + .await + .get(key) + .is_some_and(|b| b.generation >= min_generation) +} + +/// True when the currently-cached entry for `key` (if any) reflects at +/// least `key`'s latest recorded write generation (#750). Recall's +/// cache-hit gate uses this instead of a bare presence check, so a stale +/// entry left behind by a slow, superseded background build (one that lost +/// the freshness race but still reached an empty cache slot first) is +/// treated the same as a genuine cache miss — forcing the same +/// `ensure_ann_for_model` fallback a true miss would take, instead of +/// silently serving results that predate a write the caller can already +/// see committed. +pub(crate) async fn is_current(ann: &SharedAnn, key: &AnnKey) -> bool { + let target_generation = current_generation(ann, key).await; + installed_is_fresh(ann, key, target_generation).await +} + +/// Install `candidate` into the cache for `key` UNLESS an entry is already +/// present with a generation >= `candidate.generation` (#750). Replaces the +/// old `entry(key).or_insert(candidate)`, which always kept whichever build +/// happened to acquire the per-model lock and reach this call first — even +/// one that snapshotted a now-stale corpus — and silently discarded a +/// later, fresher build's result because `or_insert` is a no-op once the +/// key is occupied. +async fn install_if_fresher(ann: &SharedAnn, key: &AnnKey, candidate: AnnBridge) { + let mut idxs = ann.indexes.write().await; + match idxs.get(key) { + Some(existing) if existing.generation >= candidate.generation => { + tracing::debug!( + model = %key.model, + existing_generation = existing.generation, + candidate_generation = candidate.generation, + "memory ANN install skipped: cached entry is already >= this build's generation" + ); + } + _ => { + idxs.insert(key.clone(), candidate); + } + } +} + /// Fetch (creating if absent) the per-model warm single-flight lock. /// /// The outer `model_locks` mutex is only held long enough to look up or @@ -134,9 +222,21 @@ impl AnnBridge { index, id_map, namespace_set, + generation: 0, }) } + /// Stamp this build with the write-generation its corpus snapshot was + /// taken at or after (#750). Set unconditionally by every install path + /// in `ensure_ann_for_model_inner` before the compare-and-maybe-replace + /// install; defaults to `0` (oldest possible) so any construction path + /// that forgets to call this loses every generation comparison rather + /// than winning one it shouldn't. + pub(crate) fn with_generation(mut self, generation: u64) -> Self { + self.generation = generation; + self + } + pub(crate) fn search(&self, query: &[f32], k: usize) -> Result, RuntimeError> { let mut q = query.to_vec(); l2_normalize(&mut q); @@ -185,6 +285,7 @@ impl AnnBridge { // Until then it is left empty, which causes the retry gate to be // conservative (assume the index may contain non-visible namespaces). namespace_set: HashSet::new(), + generation: 0, }) } @@ -311,8 +412,17 @@ pub(crate) async fn ensure_ann_background( } let key = AnnKey::from_token(token, model); - // Fast path: already loaded. - if ann.indexes.read().await.contains_key(&key) { + // #750: snapshot the write-generation floor BEFORE the fast path so a + // caller triggered by a write that just bumped the counter always sees + // its own write reflected here — a caller reading this after a write + // must never observe a lower floor than that write set. + let target_generation = current_generation(ann, &key).await; + + // Fast path: already loaded AND fresh enough for this caller's write. + // A merely-present entry is not sufficient — a concurrent build that + // snapshotted an older corpus can still be sitting in the cache from a + // race that finished installing after this caller's write landed (#750). + if installed_is_fresh(ann, &key, target_generation).await { return false; } @@ -415,8 +525,18 @@ pub(crate) async fn ensure_ann_for_model( } let key = AnnKey::from_token(token, model); - // Fast path: no lock needed if already warm. - if ann.indexes.read().await.contains_key(&key) { + // #750: capture the write-generation floor BEFORE anything else — before + // even the fast "already loaded" check — so a caller invoked right after + // a write (memory.remember bumps the counter, then calls this) always + // requires at least that write's generation from whatever it accepts as + // "fresh enough". Read-generation-first, snapshot-corpus-second is the + // ordering that closes the race: reversing it would let a write land + // between the corpus snapshot and the generation read, understating the + // floor and letting a build miss that write without detecting it. + let target_generation = current_generation(ann, &key).await; + + // Fast path: no lock needed if already warm AND fresh enough. + if installed_is_fresh(ann, &key, target_generation).await { return Ok(AnnEnsureStatus::AlreadyLoaded); } @@ -424,8 +544,9 @@ pub(crate) async fn ensure_ann_for_model( let _single_flight_guard = lock.lock().await; // Re-check after acquiring the lock: a concurrent caller may have - // finished warming this model while we were waiting for the guard. - if ann.indexes.read().await.contains_key(&key) { + // finished warming this model (with a generation >= ours) while we were + // waiting for the guard. + if installed_is_fresh(ann, &key, target_generation).await { return Ok(AnnEnsureStatus::AlreadyLoaded); } @@ -462,7 +583,7 @@ pub(crate) async fn ensure_ann_for_model( ) .await; - let result = ensure_ann_for_model_inner(rt, token, ann, model).await; + let result = ensure_ann_for_model_inner(rt, token, ann, model, target_generation).await; let wall_us = phase_start.elapsed().as_micros() as i64; let cpu_us = khive_runtime::cpu_delta_us(cpu_start, khive_runtime::process_resource_usage()); @@ -556,11 +677,12 @@ async fn ensure_ann_for_model_inner( token: &NamespaceToken, ann: &SharedAnn, model: &str, + target_generation: u64, ) -> Result { let ns = "global"; let key = AnnKey::new(ns, model); - if ann.indexes.read().await.contains_key(&key) { + if installed_is_fresh(ann, &key, target_generation).await { return Ok(AnnEnsureStatus::AlreadyLoaded); } @@ -577,7 +699,8 @@ async fn ensure_ann_for_model_inner( .await .unwrap_or_default(); bridge.set_namespace_set(ns_set); - ann.indexes.write().await.entry(key).or_insert(bridge); + let bridge = bridge.with_generation(target_generation); + install_if_fresher(ann, &key, bridge).await; tracing::debug!(namespace = %ns, model = %model, "memory ANN loaded from snapshot"); return Ok(AnnEnsureStatus::LoadedSnapshot); } @@ -596,6 +719,16 @@ async fn ensure_ann_for_model_inner( } // Rebuild from vector store with double-fingerprint concurrency check. + // The fingerprint sandwich alone (compute before, scan, compute after) + // only bounds the SCAN window — it cannot see a write that lands after + // `fp_after` is read but before this build's `install_if_fresher` call + // below (e.g. during `persist_snapshot`'s I/O). The write-generation + // check closes that residual window: `target_generation` was captured + // in the caller BEFORE this whole attempt started, so a write landing + // after that point bumps the counter to a strictly higher value that a + // later, correctly-scoped rebuild will carry on its own bridge — + // `install_if_fresher` then refuses to let THIS (now-stale) build + // overwrite that later result, regardless of which one finishes first. let fp_before = compute_memory_fingerprint(rt, token, model).await; match load_and_build_from_vector_store(rt, token, model).await { Ok(Some(bridge)) => { @@ -610,12 +743,13 @@ async fn ensure_ann_for_model_inner( return Ok(AnnEnsureStatus::DiscardedStaleBuild); } let vector_count = bridge.id_map.len(); + let bridge = bridge.with_generation(target_generation); if let Some(fingerprint) = fp_after { if let Err(e) = persist_snapshot(rt, ns, model, &bridge, fingerprint).await { tracing::warn!(error = %e, "failed to persist memory Vamana snapshot"); } } - ann.indexes.write().await.entry(key).or_insert(bridge); + install_if_fresher(ann, &key, bridge).await; tracing::debug!(namespace = %ns, model = %model, vectors = vector_count, "memory ANN index built"); Ok(AnnEnsureStatus::Built { vectors: vector_count, @@ -1041,6 +1175,140 @@ mod tests { ); } + // ── #750: write-generation-checked install ────────────────────────────── + // + // The end-to-end race (a slow build snapshotting a stale corpus, + // finishing after a newer write's invalidate+re-warm cycle, and + // clobbering the cache) requires precise control over async task + // interleaving that no test-only pause hook currently exists for in + // `ensure_ann_for_model_inner` — adding one solely to force a specific + // schedule would test the hook, not the production code path. These + // tests instead pin down the exact invariant the fix depends on + // directly: `install_if_fresher`'s compare-and-replace semantics, and + // `is_current`'s stale-is-a-miss semantics. Both are unconditional, + // deterministic properties of the fix — no timing required to observe + // them — and the ns733/ns733b recall tests in `handlers/recall.rs` + // additionally exercise the real end-to-end `memory.remember` → + // `memory.recall` path stress-verified over 50 consecutive fresh-process + // runs each (see the #750 implementation report). + + fn tiny_bridge(id: Uuid, generation: u64) -> AnnBridge { + AnnBridge::build(vec![1.0f32, 0.0, 0.0, 0.0], 4, vec![id], HashSet::new()) + .expect("build tiny bridge") + .with_generation(generation) + } + + /// A candidate with a STRICTLY OLDER generation than the currently + /// installed entry must never replace it. This is the exact shape of + /// the pre-#750 bug: a slow build (older generation) finishing after a + /// faster, newer-generation build already installed. + #[tokio::test] + async fn install_if_fresher_rejects_older_generation_candidate() { + let ann = new_shared(); + let key = AnnKey::new("any-ns", "model-x"); + let newer_id = Uuid::new_v4(); + let older_id = Uuid::new_v4(); + + install_if_fresher(&ann, &key, tiny_bridge(newer_id, 5)).await; + install_if_fresher(&ann, &key, tiny_bridge(older_id, 2)).await; + + let installed = ann.indexes.read().await; + let bridge = installed.get(&key).expect("an entry must be installed"); + assert_eq!(bridge.generation, 5, "the newer generation must survive"); + assert_eq!( + bridge.id_map, + vec![newer_id], + "the older-generation candidate must not have replaced it" + ); + } + + /// A candidate with a STRICTLY NEWER generation must replace an + /// existing older entry — the compare-and-replace half of the fix + /// (`entry(key).or_insert(...)` never replaced anything once a key was + /// occupied, which is the direct cause of "later, more-complete + /// rebuild attempts are discarded"). + #[tokio::test] + async fn install_if_fresher_replaces_older_installed_entry() { + let ann = new_shared(); + let key = AnnKey::new("any-ns", "model-x"); + let older_id = Uuid::new_v4(); + let newer_id = Uuid::new_v4(); + + install_if_fresher(&ann, &key, tiny_bridge(older_id, 1)).await; + install_if_fresher(&ann, &key, tiny_bridge(newer_id, 9)).await; + + let installed = ann.indexes.read().await; + let bridge = installed.get(&key).expect("an entry must be installed"); + assert_eq!(bridge.generation, 9); + assert_eq!(bridge.id_map, vec![newer_id]); + } + + /// Equal generations: the existing entry is kept (no-op), matching the + /// `>=` comparison in `install_if_fresher` — ties do not thrash the + /// cache with an equivalent rebuild. + #[tokio::test] + async fn install_if_fresher_keeps_existing_entry_on_equal_generation() { + let ann = new_shared(); + let key = AnnKey::new("any-ns", "model-x"); + let first_id = Uuid::new_v4(); + let second_id = Uuid::new_v4(); + + install_if_fresher(&ann, &key, tiny_bridge(first_id, 3)).await; + install_if_fresher(&ann, &key, tiny_bridge(second_id, 3)).await; + + let installed = ann.indexes.read().await; + let bridge = installed.get(&key).expect("an entry must be installed"); + assert_eq!( + bridge.id_map, + vec![first_id], + "on an equal generation, the first-installed entry must be kept" + ); + } + + /// `is_current` (the recall-path freshness gate) must treat a cached + /// entry whose generation is behind the model's current write-generation + /// counter as NOT current — the other half of the fix, since a + /// presence-only check would still serve a stale-but-installed entry to + /// a recall issued after a later write. + #[tokio::test] + async fn is_current_false_when_installed_generation_behind_counter() { + let ann = new_shared(); + let key = AnnKey::new("any-ns", "model-x"); + + // Install a bridge stamped with generation 1 (as if built before any + // write bumped the counter further). + install_if_fresher(&ann, &key, tiny_bridge(Uuid::new_v4(), 1)).await; + assert!( + is_current(&ann, &key).await, + "with no bumps yet, generation-1 must be considered current (counter starts at 0)" + ); + + // A write lands and bumps the counter past the installed generation. + bump_generation(&ann, &key).await; // -> 1 + bump_generation(&ann, &key).await; // -> 2 + assert!( + !is_current(&ann, &key).await, + "installed generation (1) is now behind the write-generation counter (2)" + ); + + // Once a fresher build (generation >= 2) installs, it is current again. + install_if_fresher(&ann, &key, tiny_bridge(Uuid::new_v4(), 2)).await; + assert!( + is_current(&ann, &key).await, + "installed generation (2) now matches the write-generation counter (2)" + ); + } + + /// `is_current` on an absent key is false (a genuine cache miss), so + /// callers correctly fall through to the ensure/build path rather than + /// treating "no entry" as "no problem." + #[tokio::test] + async fn is_current_false_when_absent() { + let ann = new_shared(); + let key = AnnKey::new("any-ns", "model-x"); + assert!(!is_current(&ann, &key).await); + } + // ADR-103 Stage 1 / issue #723 ask 1: `ensure_ann_for_model` must bracket // its whole attempt with a `PhaseStarted`/`PhaseCompleted` event pair // whenever an `EventStore` is configured, regardless of which path diff --git a/crates/khive-pack-memory/src/handlers/common.rs b/crates/khive-pack-memory/src/handlers/common.rs index 11eb6785..52d488a0 100644 --- a/crates/khive-pack-memory/src/handlers/common.rs +++ b/crates/khive-pack-memory/src/handlers/common.rs @@ -938,36 +938,45 @@ impl MemoryPack { // retry — up to ANN_OVERFETCH_MAX_ROUNDS total rounds, or until the // index is exhausted (returned hits < requested k). Single-namespace // stores fill on round 1 at zero extra cost. - let initial_raw_hits: Option> = - match ann::search_loaded(&self.ann, &key, &vec, ann_fetch_limit).await { - Ok(Some(hits)) => Some(hits), - Ok(None) => { - let status = ann::ensure_ann_for_model( - &self.runtime, - token, - &self.ann, - &model_name, - ) - .await?; - tracing::debug!( - ?status, - model = %model_name, - namespace = %ns, - "memory ANN ensured on recall miss" - ); - ann::search_loaded(&self.ann, &key, &vec, ann_fetch_limit).await? - } - Err(e) => { - tracing::warn!( - error = %e, - namespace = %ns, - model = %model_name, - "memory ANN search failed; falling back to exact sqlite-vec" - ); - ann::clear_key(&self.ann, &key).await; - None - } - }; + // #750: a merely-present cache entry is not sufficient — a slow + // background build that snapshotted the corpus before the most + // recent write can still be sitting in the cache (it lost the + // install race for freshness but was still the first to reach + // an empty slot). `is_current` treats "present but stale + // relative to this model's write-generation counter" the same + // as a genuine cache miss, so a just-written memory is never + // silently served from a snapshot that predates it. + let cache_fresh = ann::is_current(&self.ann, &key).await; + let search_result = if cache_fresh { + ann::search_loaded(&self.ann, &key, &vec, ann_fetch_limit).await + } else { + Ok(None) + }; + let initial_raw_hits: Option> = match search_result { + Ok(Some(hits)) => Some(hits), + Ok(None) => { + let status = + ann::ensure_ann_for_model(&self.runtime, token, &self.ann, &model_name) + .await?; + tracing::debug!( + ?status, + model = %model_name, + namespace = %ns, + "memory ANN ensured on recall miss" + ); + ann::search_loaded(&self.ann, &key, &vec, ann_fetch_limit).await? + } + Err(e) => { + tracing::warn!( + error = %e, + namespace = %ns, + model = %model_name, + "memory ANN search failed; falling back to exact sqlite-vec" + ); + ann::clear_key(&self.ann, &key).await; + None + } + }; if let Some(first_raw) = initial_raw_hits { // Bounded retry: widen fetch window if visible-namespace survivors diff --git a/crates/khive-pack-memory/src/handlers/prune.rs b/crates/khive-pack-memory/src/handlers/prune.rs index d7db5c5e..aa9184c6 100644 --- a/crates/khive-pack-memory/src/handlers/prune.rs +++ b/crates/khive-pack-memory/src/handlers/prune.rs @@ -6,6 +6,7 @@ use serde_json::{json, Value}; use khive_runtime::{NamespaceToken, RuntimeError}; use khive_storage::types::{DeleteMode, SqlStatement, SqlValue}; +use crate::ann; use crate::MemoryPack; // ── Params ──────────────────────────────────────────────────────────────────── @@ -135,6 +136,24 @@ impl MemoryPack { } } + // #750 fix-round 1: `note_store.delete_note` above is the raw + // `NoteStore` capability — it bypasses `KhiveRuntime::delete_note`'s + // orchestration entirely (no FTS/vector cleanup, no note-mutation + // hook fire), so it does NOT go through the generic hook wired into + // `update_note`/`delete_note` for KG's delete path. Prune is inside + // the same crate as `ann`, so it invalidates/bumps directly, mirroring + // `memory.remember`'s own write-path pattern in `remember.rs`. The + // resulting cache miss forces a rebuild whose corpus scan already + // filters `deleted_at IS NULL` (`ann.rs`), so the just-pruned rows + // are correctly excluded from the rebuilt index. + if pruned > 0 { + ann::invalidate_namespace(&self.runtime, &self.ann, &namespace).await; + for model in self.runtime.registered_embedding_model_names() { + let key = ann::AnnKey::new(namespace.as_str(), model.as_str()); + ann::bump_generation(&self.ann, &key).await; + } + } + Ok(json!({ "pruned": pruned, "dry_run": false, diff --git a/crates/khive-pack-memory/src/handlers/recall.rs b/crates/khive-pack-memory/src/handlers/recall.rs index 3bdb9078..76f09b75 100644 --- a/crates/khive-pack-memory/src/handlers/recall.rs +++ b/crates/khive-pack-memory/src/handlers/recall.rs @@ -3555,142 +3555,110 @@ mod tests { #[tokio::test] #[serial(background_tasks)] async fn ns733_recall_ann_overfetch_retry_loop_respects_effective_namespace() { - // See the identical race documented on - // `ns733b_recall_verbose_multi_model_breakdown_excludes_off_namespace_candidates` - // below: `ensure_ann_background`'s fire-and-forget per-model warm - // (ann.rs) installs whichever queued build's scan acquires the - // per-model lock first, permanently, even if it raced ahead of a - // still-in-flight `remember` and snapshotted an incomplete corpus. - // 35 rapid-fire `remember` calls before the one `bench-a` target is - // enough to occasionally trigger it here too — a pre-existing - // production characteristic, not a consequence of the namespace - // filter under test. Retry the whole seed-and-recall flow against a - // fresh `KhiveRuntime` (fresh scheduling) up to a few times. - const MAX_ATTEMPTS: usize = 5; - let mut last_failure: Option = None; - - 'attempt: for attempt in 1..=MAX_ATTEMPTS { - let rt = KhiveRuntime::memory().expect("in-memory runtime"); - rt.register_embedder(FixedVecProvider { - model_name: NS733_ANN_MODEL.to_string(), - map: ns733_ann_fixed_vectors(), - }); + // #750: this test used to retry the whole seed-and-recall flow + // against a fresh `KhiveRuntime` up to 5 times, and separately poll + // `memory.recall` for up to 500ms per attempt, to paper over a + // pre-existing ANN warm-cache race: `ensure_ann_for_model` installed + // whichever queued build acquired the per-model lock first via + // `entry(key).or_insert(bridge)`, permanently, even when it had + // snapshotted the corpus before a still-in-flight `remember` + // committed. With the write-generation-checked install + // (`install_if_fresher` in ann.rs) and the recall-path freshness + // gate (`ann::is_current`, `handlers/common.rs`) replacing that + // blind `or_insert`, a stale build can no longer win permanently and + // a stale-but-present cache entry is no longer treated as a hit — + // so this now asserts directly, in one attempt, with no retry or + // poll. + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + rt.register_embedder(FixedVecProvider { + model_name: NS733_ANN_MODEL.to_string(), + map: ns733_ann_fixed_vectors(), + }); - let mut builder = VerbRegistryBuilder::new(); - builder.register(KgPack::new(rt.clone())); - builder.register(MemoryPack::new(rt.clone())); - let registry = builder.build().expect("registry"); + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + builder.register(MemoryPack::new(rt.clone())); + let registry = builder.build().expect("registry"); - // No `embedding_model` on remember: `create_note_inner`'s auto-detect - // path fans out to every *registered* model when the field is - // omitted (`resolve_embedding_model` only accepts lattice aliases — - // an explicit custom provider name here would hit `UnknownModel`, - // same gotcha documented on `recall_with_residual_fts5_char_fails_loud` - // above). `NS733_ANN_MODEL` is the only model registered on this - // runtime, so auto-detect resolves to exactly it. - for i in 0..NS733_FILLER_COUNT { - registry - .dispatch( - "memory.remember", - serde_json::json!({ - "content": format!("ns733 ann overfetch local filler {i}"), - "memory_type": "semantic", - "namespace": "local", - }), - ) - .await - .expect("remember filler"); - } - let target_id = registry + // No `embedding_model` on remember: `create_note_inner`'s auto-detect + // path fans out to every *registered* model when the field is + // omitted (`resolve_embedding_model` only accepts lattice aliases — + // an explicit custom provider name here would hit `UnknownModel`, + // same gotcha documented on `recall_with_residual_fts5_char_fails_loud` + // above). `NS733_ANN_MODEL` is the only model registered on this + // runtime, so auto-detect resolves to exactly it. + for i in 0..NS733_FILLER_COUNT { + registry .dispatch( "memory.remember", serde_json::json!({ - "content": NS733_TARGET_CONTENT, + "content": format!("ns733 ann overfetch local filler {i}"), "memory_type": "semantic", - "namespace": "bench-a", + "namespace": "local", }), ) .await - .expect("remember target")["id"] - .as_str() - .expect("id") - .parse::() - .expect("valid uuid"); - - let base_params = serde_json::json!({ - "query": NS733_QUERY, - "namespace": "bench-a", - "fusion_strategy": "vector_only", - "embedding_model": NS733_ANN_MODEL, - "config": { "candidate_limit": 1 }, - "limit": 1, - }); - - // Case 1: default widening — the target must be found, and only the - // target. Poll (bounded) for the ANN warm to settle on the full - // corpus before treating a mismatch as a real assertion failure. - let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); - let widened_hits: Vec; - loop { - let widened_result = registry - .dispatch("memory.recall", base_params.clone()) - .await - .expect("memory.recall with default widening"); - let hits = widened_result - .as_array() - .cloned() - .expect("bare array result"); - let settled = hits.len() == 1 - && hits[0]["id"].as_str().and_then(|s| s.parse::().ok()) - == Some(target_id); - if settled { - widened_hits = hits; - break; - } - if std::time::Instant::now() >= deadline { - last_failure = Some(format!( - "attempt {attempt}/{MAX_ATTEMPTS}: ANN warm for the bench-a target \ - did not settle within 500ms; last response: {hits:?}" - )); - continue 'attempt; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } - assert_eq!( - widened_hits.len(), - 1, - "default widening must surface exactly the bench-a target, got: {widened_hits:?}" - ); - assert_eq!( - widened_hits[0]["id"] - .as_str() - .and_then(|s| s.parse::().ok()), - Some(target_id), - "the single hit must be the bench-a target, not a local filler" - ); - - // Case 2: widening disabled (`ann_overfetch_max_rounds: 1`) — round 1's - // narrow window is exhausted entirely by `local` fillers ranked ahead - // of the target, so the namespace-scoped post-filter finds nothing. - let mut disabled_params = base_params; - disabled_params["config"]["ann_overfetch_max_rounds"] = serde_json::json!(1); - let disabled_result = registry - .dispatch("memory.recall", disabled_params) - .await - .expect("memory.recall with widening disabled"); - let disabled_hits = disabled_result.as_array().expect("bare array result"); - assert!( - disabled_hits.is_empty(), - "with widening disabled, round 1's over-fetch window is saturated by \ - closer local fillers and must not reach the bench-a target: {disabled_hits:?}" - ); - return; + .expect("remember filler"); } + let target_id = registry + .dispatch( + "memory.remember", + serde_json::json!({ + "content": NS733_TARGET_CONTENT, + "memory_type": "semantic", + "namespace": "bench-a", + }), + ) + .await + .expect("remember target")["id"] + .as_str() + .expect("id") + .parse::() + .expect("valid uuid"); + + let base_params = serde_json::json!({ + "query": NS733_QUERY, + "namespace": "bench-a", + "fusion_strategy": "vector_only", + "embedding_model": NS733_ANN_MODEL, + "config": { "candidate_limit": 1 }, + "limit": 1, + }); - panic!( - "ns733: corpus/ANN-warm race did not settle in {MAX_ATTEMPTS} fresh attempts; \ - last failure: {}", - last_failure.unwrap_or_default() + // Case 1: default widening — the target must be found, and only the + // target. + let widened_result = registry + .dispatch("memory.recall", base_params.clone()) + .await + .expect("memory.recall with default widening"); + let widened_hits = widened_result.as_array().expect("bare array result"); + assert_eq!( + widened_hits.len(), + 1, + "default widening must surface exactly the bench-a target, got: {widened_hits:?}" + ); + assert_eq!( + widened_hits[0]["id"] + .as_str() + .and_then(|s| s.parse::().ok()), + Some(target_id), + "the single hit must be the bench-a target, not a local filler" + ); + + // Case 2: widening disabled (`ann_overfetch_max_rounds: 1`) — round 1's + // narrow window is exhausted entirely by `local` fillers ranked ahead + // of the target, so the namespace-scoped post-filter finds nothing. + let mut disabled_params = base_params; + disabled_params["config"]["ann_overfetch_max_rounds"] = serde_json::json!(1); + let disabled_result = registry + .dispatch("memory.recall", disabled_params) + .await + .expect("memory.recall with widening disabled"); + let disabled_hits = disabled_result.as_array().expect("bare array result"); + assert!( + disabled_hits.is_empty(), + "with widening disabled, round 1's over-fetch window is saturated by \ + closer local fillers and must not reach the bench-a target: {disabled_hits:?}" ); } @@ -3794,102 +3762,50 @@ mod tests { #[tokio::test] #[serial(background_tasks)] async fn ns733b_recall_verbose_multi_model_breakdown_excludes_off_namespace_candidates() { - // Every `memory.remember` fires `ann::invalidate_namespace` (the ANN - // index is global per model, so any write clears *all* loaded slots) - // and re-queues `ensure_ann_background`'s fire-and-forget rebuild for - // every registered model (ann.rs). `ensure_ann_for_model` installs a - // finished build with `indexes.entry(key).or_insert(bridge)` and - // treats an already-present key as `AlreadyLoaded` — so whichever - // queued build's `spawn_blocking` scan happens to acquire the - // per-model lock *first* wins permanently, even if it raced ahead of - // a still-in-flight sibling write and snapshotted an incomplete - // corpus (the double-fingerprint check only discards a build that - // observes the corpus mutate *during its own* scan, which does not - // catch "raced ahead of a write that hadn't started yet"). Six rapid - // `remember` calls across two brand-new custom models is enough to - // occasionally trigger this pre-existing scheduling race, and once a - // stale entry wins, nothing here can invalidate it again (no further - // writes occur) — `memory.recall` polling never self-corrects. - // - // This is a genuine pre-existing production characteristic, not a - // consequence of the #733 namespace filter under test (confirmed via - // targeted debug capture: the losing runs consistently showed a - // 5-filler, target-absent vector index that never changed across a - // bounded poll window). Rather than touch the production ANN warm - // path, retry the whole corpus-seed-and-recall flow against a fresh - // `KhiveRuntime` (fresh scheduling, fresh race) up to a few times — - // deterministic in the success case (which is the overwhelming - // majority), bounded and loud on failure otherwise. - const MAX_ATTEMPTS: usize = 5; - let mut last_failure: Option = None; - - 'attempt: for attempt in 1..=MAX_ATTEMPTS { - let rt = KhiveRuntime::memory().expect("in-memory runtime"); - rt.register_embedder(FixedVecProvider { - model_name: NS733B_MODEL_A.to_string(), - map: ns733b_fixed_vectors(), - }); - rt.register_embedder(FixedVecProvider { - model_name: NS733B_MODEL_B.to_string(), - map: ns733b_fixed_vectors(), - }); - - let mut builder = VerbRegistryBuilder::new(); - builder.register(KgPack::new(rt.clone())); - builder.register(MemoryPack::new(rt.clone())); - let registry = builder.build().expect("registry"); + // #750: this test used to retry the whole corpus-seed-and-recall flow + // against a fresh `KhiveRuntime` up to 5 times, and separately poll + // `memory.recall` for up to 500ms per attempt, to paper over the + // same pre-existing ANN warm-cache race documented in detail on + // `ns733_recall_ann_overfetch_retry_loop_respects_effective_namespace` + // above — `ensure_ann_for_model`'s old `entry(key).or_insert(bridge)` + // let whichever queued build acquired the per-model lock first win + // permanently, even one that had snapshotted the corpus before a + // still-in-flight sibling `remember` committed. With the + // write-generation-checked install (`install_if_fresher`, ann.rs) + // and the recall-path freshness gate (`ann::is_current`, + // `handlers/common.rs`), a stale build can no longer win permanently + // and a stale-but-present cache entry is no longer served as a hit — + // so this now asserts directly, in one attempt, with no retry or poll. + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + rt.register_embedder(FixedVecProvider { + model_name: NS733B_MODEL_A.to_string(), + map: ns733b_fixed_vectors(), + }); + rt.register_embedder(FixedVecProvider { + model_name: NS733B_MODEL_B.to_string(), + map: ns733b_fixed_vectors(), + }); - let (local_filler_ids, target_id) = ns733b_seed_two_model_corpus(®istry).await; + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + builder.register(MemoryPack::new(rt.clone())); + let registry = builder.build().expect("registry"); - let recall_args = serde_json::json!({ - "query": NS733B_QUERY, - "namespace": "bench-a", - "fusion_strategy": "vector_only", - "include_breakdown": true, - "limit": 10, - }); - let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); - let settled_result: Value; - loop { - let result = registry - .dispatch("memory.recall", recall_args.clone()) - .await - .expect("memory.recall with namespace + multi-model + include_breakdown"); - let settled = result["candidates"]["vector_candidates_per_model"] - .as_array() - .is_some_and(|per_model| { - per_model.len() == 2 - && per_model.iter().any(|entry| { - entry["hits"].as_array().is_some_and(|hits| { - hits.iter().any(|hit| { - hit["id"].as_str().and_then(|s| s.parse::().ok()) - == Some(target_id) - }) - }) - }) - }); - if settled { - settled_result = result; - break; - } - if std::time::Instant::now() >= deadline { - last_failure = Some(format!( - "attempt {attempt}/{MAX_ATTEMPTS}: ANN warm for the bench-a target \ - did not settle within 500ms; last response: {result}" - )); - continue 'attempt; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } + let (local_filler_ids, target_id) = ns733b_seed_two_model_corpus(®istry).await; - return ns733b_assert_breakdown(&settled_result, &local_filler_ids, target_id); - } + let recall_args = serde_json::json!({ + "query": NS733B_QUERY, + "namespace": "bench-a", + "fusion_strategy": "vector_only", + "include_breakdown": true, + "limit": 10, + }); + let result = registry + .dispatch("memory.recall", recall_args) + .await + .expect("memory.recall with namespace + multi-model + include_breakdown"); - panic!( - "ns733b: corpus/ANN-warm race did not settle in {MAX_ATTEMPTS} fresh attempts; \ - last failure: {}", - last_failure.unwrap_or_default() - ); + ns733b_assert_breakdown(&result, &local_filler_ids, target_id); } /// Assertion body for @@ -3961,127 +3877,87 @@ mod tests { #[tokio::test] #[serial(background_tasks)] async fn ns733b_recall_candidates_multi_model_excludes_off_namespace_candidates() { - // Same pre-existing ANN-warm race documented in detail on + // #750: same pre-existing ANN warm-cache race documented in detail on // `ns733b_recall_verbose_multi_model_breakdown_excludes_off_namespace_candidates` - // above applies identically here — `handle_recall_candidates` reads + // above applied identically here — `handle_recall_candidates` reads // the same shared per-model ANN index via the same - // `collect_recall_candidates` path `handle_recall` uses. Same bounded - // fresh-runtime retry. - const MAX_ATTEMPTS: usize = 5; - let mut last_failure: Option = None; - - 'attempt: for attempt in 1..=MAX_ATTEMPTS { - let rt = KhiveRuntime::memory().expect("in-memory runtime"); - rt.register_embedder(FixedVecProvider { - model_name: NS733B_MODEL_A.to_string(), - map: ns733b_fixed_vectors(), - }); - rt.register_embedder(FixedVecProvider { - model_name: NS733B_MODEL_B.to_string(), - map: ns733b_fixed_vectors(), - }); + // `collect_recall_candidates` path `handle_recall` uses. Fixed the + // same way (write-generation-checked install + recall-path freshness + // gate); asserts directly, in one attempt, with no retry or poll. + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + rt.register_embedder(FixedVecProvider { + model_name: NS733B_MODEL_A.to_string(), + map: ns733b_fixed_vectors(), + }); + rt.register_embedder(FixedVecProvider { + model_name: NS733B_MODEL_B.to_string(), + map: ns733b_fixed_vectors(), + }); - let mut builder = VerbRegistryBuilder::new(); - builder.register(KgPack::new(rt.clone())); - builder.register(MemoryPack::new(rt.clone())); - let registry = builder.build().expect("registry"); + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + builder.register(MemoryPack::new(rt.clone())); + let registry = builder.build().expect("registry"); - let (local_filler_ids, target_id) = ns733b_seed_two_model_corpus(®istry).await; - - // `memory.recall_candidates`'s `HandlerDef` declares no - // `namespace` `ParamDef` (`pack.rs`: `params: &[]`), so - // `VerbRegistry::dispatch`'s generic Rule-3 explicit-namespace - // escape (ADR-007 Rev 4/6) is the *only* thing scoping this call - // — it mints the token with `visible=["bench-a"]` before - // `handle_recall_candidates` ever runs, then strips the - // `namespace` key from params (the handler never sees it). No - // `fusion_strategy`/`include_breakdown` params exist on this - // sub-verb — it always returns the per-model breakdown whenever - // more than one model is registered (`sub_handlers.rs:168`). - let recall_candidates_args = serde_json::json!({ - "query": NS733B_QUERY, - "namespace": "bench-a", - "limit": 10, - }); - let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500); - let settled_result: Value; - loop { - let result = registry - .dispatch("memory.recall_candidates", recall_candidates_args.clone()) - .await - .expect("memory.recall_candidates with namespace + multi-model"); - let settled = result["vector_candidates_per_model"] - .as_object() - .is_some_and(|per_model| { - per_model.len() == 2 - && per_model.values().any(|hits| { - hits.as_array().is_some_and(|hits| { - hits.iter().any(|hit| { - hit["id"].as_str().and_then(|s| s.parse::().ok()) - == Some(target_id) - }) - }) - }) - }); - if settled { - settled_result = result; - break; - } - if std::time::Instant::now() >= deadline { - last_failure = Some(format!( - "attempt {attempt}/{MAX_ATTEMPTS}: ANN warm for the bench-a target \ - did not settle within 500ms; last response: {result}" - )); - continue 'attempt; - } - tokio::time::sleep(std::time::Duration::from_millis(20)).await; - } + let (local_filler_ids, target_id) = ns733b_seed_two_model_corpus(®istry).await; + + // `memory.recall_candidates`'s `HandlerDef` declares no + // `namespace` `ParamDef` (`pack.rs`: `params: &[]`), so + // `VerbRegistry::dispatch`'s generic Rule-3 explicit-namespace + // escape (ADR-007 Rev 4/6) is the *only* thing scoping this call + // — it mints the token with `visible=["bench-a"]` before + // `handle_recall_candidates` ever runs, then strips the + // `namespace` key from params (the handler never sees it). No + // `fusion_strategy`/`include_breakdown` params exist on this + // sub-verb — it always returns the per-model breakdown whenever + // more than one model is registered (`sub_handlers.rs:168`). + let recall_candidates_args = serde_json::json!({ + "query": NS733B_QUERY, + "namespace": "bench-a", + "limit": 10, + }); + let result = registry + .dispatch("memory.recall_candidates", recall_candidates_args) + .await + .expect("memory.recall_candidates with namespace + multi-model"); - let per_model = settled_result["vector_candidates_per_model"] - .as_object() - .expect("multi-model breakdown present (two models registered)"); - assert_eq!( - per_model.len(), - 2, - "both registered models must appear in the breakdown: {per_model:?}" - ); + let per_model = result["vector_candidates_per_model"] + .as_object() + .expect("multi-model breakdown present (two models registered)"); + assert_eq!( + per_model.len(), + 2, + "both registered models must appear in the breakdown: {per_model:?}" + ); - for (model_name, hits) in per_model { - let hits = hits.as_array().expect("hits array"); - for hit in hits { - let id = hit["id"] - .as_str() - .expect("id") - .parse::() - .expect("valid uuid"); - assert!( - !local_filler_ids.contains(&id), - "namespace=\"bench-a\" recall_candidates breakdown must not leak a \ - local filler UUID ({id}) for model {model_name:?}: {hits:?}" - ); - } + for (model_name, hits) in per_model { + let hits = hits.as_array().expect("hits array"); + for hit in hits { + let id = hit["id"] + .as_str() + .expect("id") + .parse::() + .expect("valid uuid"); + assert!( + !local_filler_ids.contains(&id), + "namespace=\"bench-a\" recall_candidates breakdown must not leak a \ + local filler UUID ({id}) for model {model_name:?}: {hits:?}" + ); } - - // Sanity: the fix must not have filtered away everything — the - // bench-a target itself is entitled to appear (proves this is a - // real filter, not a filter-everything regression). - let any_model_has_target = per_model.values().any(|hits| { - hits.as_array().unwrap().iter().any(|hit| { - hit["id"].as_str().and_then(|s| s.parse::().ok()) == Some(target_id) - }) - }); - assert!( - any_model_has_target, - "the bench-a target must still appear in at least one model's \ - recall_candidates breakdown after the namespace filter: {per_model:?}" - ); - return; } - panic!( - "ns733b recall_candidates: corpus/ANN-warm race did not settle in {MAX_ATTEMPTS} \ - fresh attempts; last failure: {}", - last_failure.unwrap_or_default() + // Sanity: the fix must not have filtered away everything — the + // bench-a target itself is entitled to appear (proves this is a + // real filter, not a filter-everything regression). + let any_model_has_target = per_model.values().any(|hits| { + hits.as_array().unwrap().iter().any(|hit| { + hit["id"].as_str().and_then(|s| s.parse::().ok()) == Some(target_id) + }) + }); + assert!( + any_model_has_target, + "the bench-a target must still appear in at least one model's \ + recall_candidates breakdown after the namespace filter: {per_model:?}" ); } } diff --git a/crates/khive-pack-memory/src/handlers/remember.rs b/crates/khive-pack-memory/src/handlers/remember.rs index f69e57e6..faebd936 100644 --- a/crates/khive-pack-memory/src/handlers/remember.rs +++ b/crates/khive-pack-memory/src/handlers/remember.rs @@ -143,6 +143,16 @@ impl MemoryPack { None => self.runtime.registered_embedding_model_names(), }; for model in affected_models { + // #750: bump this model's write-generation counter BEFORE + // triggering the background warm, so `ensure_ann_background`'s + // (and, downstream, `ensure_ann_for_model`'s) write-generation + // floor for this call always reflects the note just written — + // any build that started capturing its own floor before this + // point is provably looking at a stale corpus and will lose + // the install race to whichever rebuild captures at or after + // this bump. + let key = ann::AnnKey::from_token(write_token, &model); + ann::bump_generation(&self.ann, &key).await; ann::ensure_ann_background(&self.runtime, write_token, &self.ann, &model).await; } } diff --git a/crates/khive-pack-memory/src/pack.rs b/crates/khive-pack-memory/src/pack.rs index ca822a26..45c283e4 100644 --- a/crates/khive-pack-memory/src/pack.rs +++ b/crates/khive-pack-memory/src/pack.rs @@ -392,6 +392,40 @@ impl PackRuntime for MemoryPack { fts_population_guard(&self.runtime).await; } + /// #750 fix-round 1: install a note-mutation hook on the runtime THIS + /// pack owns (`self.runtime`, not the `_runtime` param — mirrors + /// `KgPack::register_entity_type_validator`'s multi-backend rationale: + /// in a multi-backend deployment each pack is constructed with its own + /// per-pack runtime, and `self.runtime` is always that one). + /// + /// KG's `update`/`delete` verbs reach `KhiveRuntime::update_note`/ + /// `delete_note` with no dependency on `khive-pack-memory` at all. When + /// those touch a `kind="memory"` note (content update, soft delete, or + /// hard delete), this hook invalidates the warm ANN cache and bumps the + /// affected models' write-generation counter — the same two calls + /// `memory.remember` already makes on its own write path (`ann::invalidate_namespace` + /// + `ann::bump_generation`) — so `ann::is_current()` correctly treats + /// the stale-but-still-installed index as a miss on the next recall. + fn register_note_mutation_hook(&self, _runtime: &KhiveRuntime) { + let runtime = self.runtime.clone(); + let ann = self.ann.clone(); + let hook: khive_runtime::NoteMutationHookFn = std::sync::Arc::new(move |kind, _id| { + let runtime = runtime.clone(); + let ann = ann.clone(); + Box::pin(async move { + if kind != "memory" { + return; + } + crate::ann::invalidate_namespace(&runtime, &ann, "local").await; + for model in runtime.registered_embedding_model_names() { + let key = crate::ann::AnnKey::new("local", model.as_str()); + crate::ann::bump_generation(&ann, &key).await; + } + }) + }); + self.runtime.install_note_mutation_hook(hook); + } + async fn dispatch( &self, verb: &str, @@ -672,3 +706,333 @@ mod ann_route_tests { ); } } + +/// #750 fix-round 1 regressions: `memory.prune`, KG `update`, and KG `delete` +/// all mutate the same memory-note corpus the warm ANN cache is built from, +/// but only `memory.remember` bumped `AnnState::generations` before this +/// round. Each test here warms the cache (proving `is_current()` is `true` +/// for the registered model), performs the mutation WITHOUT any intervening +/// `memory.remember`, and asserts `is_current()` is `false` afterward — the +/// exact mechanism `handlers/common.rs`'s recall-path cache-hit gate checks +/// before trusting an installed index. +/// +/// This is a white-box, crate-internal assertion (`ann::is_current`/ +/// `ann::AnnKey` are `pub(crate)`) rather than a black-box recall-result +/// check. Confirmed via a throwaway experiment (temporarily disabling the +/// note-mutation hook call sites in `curation.rs`/`operations.rs`/ +/// `prune.rs`): for the delete path specifically, "the deleted note is +/// absent from recall results" passes EVEN WITH THE BUG REINTRODUCED, +/// because recall's post-search hydration filters soft-deleted notes +/// regardless of ANN cache freshness — it is not a discriminating +/// assertion. `is_current` is. The same experiment confirmed all three +/// tests below fail (as expected) with the fix removed, and pass with it +/// restored. +#[cfg(test)] +mod note_mutation_hook_tests { + use super::*; + use crate::ann; + use khive_pack_kg::KgPack; + use khive_runtime::VerbRegistryBuilder; + use serde_json::json; + use serial_test::serial; + use uuid::Uuid; + + const FR1_MODEL: &str = "fr1-mutation-hook-model"; + + /// Build a kg+memory registry AND wire the note-mutation hook. + /// Production boot does this via `khive-mcp`'s `serve.rs`/`server.rs` + /// calling `VerbRegistry::call_register_note_mutation_hooks`; a + /// hand-built test registry must call it explicitly, same as this + /// codebase's existing `call_register_entity_type_validators` test + /// pattern. Also returns the `MemoryPack`'s `SharedAnn` handle (cloned + /// before the pack is moved into the builder) so tests can assert + /// `ann::is_current` directly. + fn fr1_build_registry(rt: &KhiveRuntime) -> (khive_runtime::VerbRegistry, SharedAnn) { + let mut builder = VerbRegistryBuilder::new(); + builder.register(KgPack::new(rt.clone())); + let memory_pack = MemoryPack::new(rt.clone()); + let ann = memory_pack.ann_for_test(); + builder.register(memory_pack); + let registry = builder.build().expect("registry builds"); + registry.call_register_note_mutation_hooks(rt); + (registry, ann) + } + + fn fr1_key() -> ann::AnnKey { + ann::AnnKey::new("local", FR1_MODEL) + } + + /// Seed one memory note, then run a `memory.recall` to synchronously + /// warm the ANN cache — the cache-miss fallback in `handlers/common.rs` + /// calls `ensure_ann_for_model` and awaits it before returning, so a + /// single recall is enough to guarantee the index is installed. Returns + /// the seeded note's id and asserts the cache is current before the + /// caller's mutation under test. + async fn fr1_seed_and_warm( + rt: &KhiveRuntime, + registry: &khive_runtime::VerbRegistry, + ann: &SharedAnn, + content: &str, + salience: f64, + ) -> Uuid { + rt.register_embedder(Fr1FixedVecProvider { + model_name: FR1_MODEL.to_string(), + vector: [1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], + }); + let id = registry + .dispatch( + "memory.remember", + json!({"content": content, "salience": salience}), + ) + .await + .expect("seed remember")["id"] + .as_str() + .expect("id") + .parse::() + .expect("valid uuid"); + + registry + .dispatch( + "memory.recall", + json!({ + "query": content, + "namespace": "local", + "fusion_strategy": "vector_only", + "embedding_model": FR1_MODEL, + }), + ) + .await + .expect("warm recall"); + + assert!( + ann::is_current(ann, &fr1_key()).await, + "sanity: warm-up recall must leave the ANN cache current before \ + the mutation under test" + ); + id + } + + #[tokio::test] + #[serial(background_tasks)] + async fn fr1_prune_invalidates_warm_ann_without_subsequent_remember() { + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + let (registry, ann) = fr1_build_registry(&rt); + + fr1_seed_and_warm(&rt, ®istry, &ann, "fr1 prune target content", 0.9).await; + + // `min_salience: 1.0` is strictly above the seeded note's 0.9 + // salience, so it is the one candidate. No `memory.remember` call + // follows. + let pruned = registry + .dispatch( + "memory.prune", + json!({ "min_salience": 1.0, "namespace": "local" }), + ) + .await + .expect("prune"); + assert_eq!( + pruned["pruned"], 1, + "the seeded note must be the one candidate pruned: {pruned:?}" + ); + + assert!( + !ann::is_current(&ann, &fr1_key()).await, + "memory.prune deleting a candidate must invalidate the warm ANN \ + generation for affected models (#750 fix-round 1)" + ); + } + + #[tokio::test] + #[serial(background_tasks)] + async fn fr1_kg_update_reindex_invalidates_warm_ann_without_subsequent_remember() { + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + let (registry, ann) = fr1_build_registry(&rt); + + let id = fr1_seed_and_warm(&rt, ®istry, &ann, "fr1 update target content", 0.7).await; + + // KG's generic `update` verb — NOT `memory.remember` — changes the + // note's content. Same call shape `khive-pack-kg/src/handlers/ + // update.rs` dispatches through `KhiveRuntime::update_note`; no + // `kind` param needed, the UUID resolves the substrate. + registry + .dispatch( + "update", + json!({ "id": id.to_string(), "content": "entirely different rewritten content" }), + ) + .await + .expect("kg update on memory-kind note"); + + assert!( + !ann::is_current(&ann, &fr1_key()).await, + "a KG `update` that changes a memory-kind note's content must \ + invalidate the warm ANN generation (#750 fix-round 1)" + ); + } + + #[tokio::test] + #[serial(background_tasks)] + async fn fr1_kg_delete_invalidates_warm_ann_without_subsequent_remember() { + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + let (registry, ann) = fr1_build_registry(&rt); + + let id = fr1_seed_and_warm(&rt, ®istry, &ann, "fr1 delete target content", 0.7).await; + + // KG's generic `delete` verb (soft delete by default) — NOT any + // memory-pack verb. + let deleted = registry + .dispatch("delete", json!({ "id": id.to_string() })) + .await + .expect("kg delete on memory-kind note"); + assert_eq!( + deleted["deleted"].as_bool(), + Some(true), + "delete must report success: {deleted:?}" + ); + + assert!( + !ann::is_current(&ann, &fr1_key()).await, + "a KG `delete` on a memory-kind note must invalidate the warm \ + ANN generation (#750 fix-round 1)" + ); + } + + /// #750 fix-round 2 (codex r2 High 1): `merge_note`'s non-dry-run + /// success path reindexes the surviving note (a corpus change exactly + /// like `update_note`'s `text_changed` branch) but never fired the + /// note-mutation hook. Same white-box shape as the three fr1 tests + /// above. + /// + /// Both notes are seeded and the cache is warmed only ONCE, AFTER both + /// exist — NOT via `fr1_seed_and_warm` for the first note followed by a + /// second `memory.remember` for the second, because `memory.remember`'s + /// own handler already calls `ann::bump_generation` on every create + /// (`handlers/remember.rs`). Warming before the second note existed + /// would make the merge assertion pass trivially off that unrelated + /// bump — this exact non-discriminating-test trap is what caught out + /// an earlier draft of this test (confirmed by a disable/re-enable + /// run: the earlier draft still passed with the merge fix's hook-fire + /// call commented out). See the fix-round-2 report for the full + /// before/after evidence. + #[tokio::test] + #[serial(background_tasks)] + async fn fr2_kg_merge_invalidates_warm_ann_without_subsequent_remember() { + let rt = KhiveRuntime::memory().expect("in-memory runtime"); + let (registry, ann) = fr1_build_registry(&rt); + + rt.register_embedder(Fr1FixedVecProvider { + model_name: FR1_MODEL.to_string(), + vector: [1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0], + }); + + let into_id = registry + .dispatch( + "memory.remember", + json!({"content": "fr2 merge into content", "salience": 0.7}), + ) + .await + .expect("seed into-note")["id"] + .as_str() + .expect("id") + .parse::() + .expect("valid uuid"); + let from_id = registry + .dispatch( + "memory.remember", + json!({"content": "fr2 merge from content", "salience": 0.7}), + ) + .await + .expect("seed from-note")["id"] + .as_str() + .expect("id") + .parse::() + .expect("valid uuid"); + + // ONE warm-up recall, after BOTH notes exist. + registry + .dispatch( + "memory.recall", + json!({ + "query": "fr2 merge into content", + "namespace": "local", + "fusion_strategy": "vector_only", + "embedding_model": FR1_MODEL, + }), + ) + .await + .expect("warm recall"); + assert!( + ann::is_current(&ann, &fr1_key()).await, + "sanity: warm-up recall must leave the ANN cache current before \ + the merge under test" + ); + + registry + .dispatch( + "merge", + json!({ + "into_id": into_id.to_string(), + "from_id": from_id.to_string(), + "kind": "memory", + }), + ) + .await + .expect("kg merge on memory-kind notes"); + + assert!( + !ann::is_current(&ann, &fr1_key()).await, + "a KG `merge` of two memory-kind notes must invalidate the warm \ + ANN generation (#750 fix-round 2, codex r2 High 1)" + ); + } + + struct Fr1FixedVecProvider { + model_name: String, + vector: [f32; 8], + } + + #[async_trait::async_trait] + impl khive_runtime::EmbedderProvider for Fr1FixedVecProvider { + fn name(&self) -> &str { + &self.model_name + } + + fn dimensions(&self) -> usize { + 8 + } + + async fn build( + &self, + ) -> Result, RuntimeError> { + Ok(std::sync::Arc::new(Fr1FixedVecService { + vector: self.vector, + })) + } + } + + struct Fr1FixedVecService { + vector: [f32; 8], + } + + #[async_trait::async_trait] + impl lattice_embed::EmbeddingService for Fr1FixedVecService { + async fn embed( + &self, + texts: &[String], + _model: lattice_embed::EmbeddingModel, + ) -> Result>, lattice_embed::EmbedError> { + // Every text maps to the SAME fixed vector — deterministic + // cosine=1.0 between any query and any seeded content under this + // provider, so ANN warming/hit behavior is fully controlled by + // this test module, not by real embedding semantics. + Ok(texts.iter().map(|_| self.vector.to_vec()).collect()) + } + + fn supports_model(&self, _model: lattice_embed::EmbeddingModel) -> bool { + true + } + + fn name(&self) -> &'static str { + "fr1-fixed-vec" + } + } +} diff --git a/crates/khive-runtime/src/atomic_plan.rs b/crates/khive-runtime/src/atomic_plan.rs index 2310b218..814b9cb2 100644 --- a/crates/khive-runtime/src/atomic_plan.rs +++ b/crates/khive-runtime/src/atomic_plan.rs @@ -149,6 +149,16 @@ pub enum PostCommitEffect { note: Option, namespace: String, }, + /// A committed note delete (soft or hard) — fire the pack-installed + /// note-mutation hook with the deleted note's kind (#750 fix-round 2, + /// codex r2 High 2: `DeletePlan` previously carried no `post_commit` + /// slot at all, so an atomic note delete never reached + /// `KhiveRuntime::fire_note_mutation_hook`, unlike `operations.rs`'s + /// `delete_note`, which fires it directly after a successful row + /// delete). Entity deletes have no equivalent — the hook system is + /// note-only (`khive-pack-memory`'s warm ANN cache is the only + /// installed consumer today). + NoteDeleted { note_id: Uuid, kind: String }, } /// The natural key a committed symmetric edge update's surviving row must @@ -204,6 +214,10 @@ pub struct DeletePlan { /// delete only) is unguarded — it may legitimately affect zero rows if /// the target had no incident edges. pub statements: Vec, + /// Deferred note-mutation-hook fire, for a note delete (#750 fix-round + /// 2). `PostCommitEffect::None` for entity and edge deletes — the hook + /// system is note-only. + pub post_commit: PostCommitEffect, } /// Write plan for a `link` op (create a typed directed edge). Endpoint @@ -368,6 +382,7 @@ mod tests { fn delete_plan_guard_is_anchored_to_the_target_row_not_the_cascade() { let plan = DeletePlan { target_id: Uuid::new_v4(), + post_commit: PostCommitEffect::None, statements: vec![ guarded("delete-row", AffectedRowGuard::exactly(1)), unguarded("cascade-edges"), diff --git a/crates/khive-runtime/src/atomic_prepare.rs b/crates/khive-runtime/src/atomic_prepare.rs index 6a5f864e..6f944964 100644 --- a/crates/khive-runtime/src/atomic_prepare.rs +++ b/crates/khive-runtime/src/atomic_prepare.rs @@ -1035,6 +1035,7 @@ pub async fn prepare_delete( Ok(AtomicOpPlan::Delete(DeletePlan { target_id: id, statements, + post_commit: PostCommitEffect::None, })) } Some(Resolved::Note(note)) => { @@ -1106,6 +1107,15 @@ pub async fn prepare_delete( Ok(AtomicOpPlan::Delete(DeletePlan { target_id: id, statements, + // #750 fix-round 2 (codex r2 High 2): a committed atomic + // note delete must fire the same pack-installed + // note-mutation hook `operations.rs::delete_note` fires, + // so a warm ANN cache sees the deletion even when the + // mutation went through the atomic-plan path instead. + post_commit: PostCommitEffect::NoteDeleted { + note_id: id, + kind: note.kind.clone(), + }, })) } Some(_) => Err(RuntimeError::InvalidInput(format!( @@ -1185,6 +1195,7 @@ async fn prepare_delete_edge( Ok(AtomicOpPlan::Delete(DeletePlan { target_id: id, statements, + post_commit: PostCommitEffect::None, })) } @@ -1399,8 +1410,31 @@ pub async fn apply_post_commit_effects( PostCommitEffect::ReindexNote { note_id } => { if let Some(note) = runtime.notes(token)?.get_note(note_id).await? { runtime.reindex_note(token, ¬e).await?; + // #750 fix-round 2 (codex r2 High 2): this handler + // called `reindex_note` directly, bypassing + // `update_note()` entirely — the note-mutation hook + // `update_note` fires after its own reindex (see + // `curation.rs`) was never reached for an atomic + // update. Fire it here so any in-process consumer + // (e.g. khive-pack-memory's warm ANN cache) sees a + // bumped generation after a committed atomic note + // update, matching the non-atomic path. + runtime.fire_note_mutation_hook(¬e.kind, note.id).await; } } + PostCommitEffect::NoteDeleted { note_id, kind } => { + // #750 fix-round 2 (codex r2 High 2): `DeletePlan` had no + // `post_commit` slot at all prior to this round, so an + // atomic note delete never reached + // `fire_note_mutation_hook` — unlike `operations.rs`'s + // `delete_note`, which fires it directly (with the + // already-known kind, no refetch) after a successful row + // delete. The note row is gone (hard delete) or tombstoned + // (soft delete) by the time this post-commit pass runs, so + // this mirrors `delete_note`'s direct-fire shape rather + // than `ReindexNote`'s refetch-then-fire shape. + runtime.fire_note_mutation_hook(&kind, note_id).await; + } PostCommitEffect::GtdAudit { .. } => { // GAP-5 (B3 fix round 4): applied by the `kkernel` caller's // own post-commit pass, not here — `khive-pack-gtd` (owner @@ -1671,6 +1705,121 @@ mod tests { ); } + /// #750 fix-round 2 (codex r2 High 2, in-process portion): the + /// atomic-plan path never fired the pack-installed note-mutation hook + /// for either an atomic note UPDATE (`PostCommitEffect::ReindexNote`'s + /// handler called `reindex_note` directly, bypassing `update_note()`, + /// which is where the hook fires on the non-atomic path) or an atomic + /// note DELETE (`DeletePlan` carried no `post_commit` slot at all + /// before this round, so nothing could fire there regardless). Both + /// are fixed in this round: `ReindexNote`'s handler now fires the hook + /// after its own reindex, and `DeletePlan` now carries a + /// `PostCommitEffect::NoteDeleted` that `apply_post_commit_effects` + /// dispatches directly (mirroring `operations.rs::delete_note`'s + /// direct-fire, no-refetch shape — the row may already be gone by the + /// time this runs, for a hard delete). A minimal counting hook proves + /// both fire; no `khive-pack-memory` dependency is needed at this + /// layer, since the hook itself is generic. + #[tokio::test] + async fn atomic_note_update_and_delete_post_commit_fire_the_note_mutation_hook() { + let runtime = scratch_runtime(); + let token = runtime + .authorize(Namespace::parse("local").expect("ns")) + .expect("authorize"); + + let fired: std::sync::Arc>> = + std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let fired_for_hook = fired.clone(); + runtime.install_note_mutation_hook(std::sync::Arc::new( + move |kind: String, id: uuid::Uuid| { + let fired = fired_for_hook.clone(); + Box::pin(async move { + fired.lock().expect("lock").push((kind, id)); + }) + }, + )); + + // Update path. + let mut note = khive_storage::note::Note::new("local", "observation", "hook-update-target"); + note.name = Some("hook-update-target".to_string()); + let update_note_id = note.id; + runtime + .notes(&token) + .expect("notes store") + .upsert_note(note) + .await + .expect("seed update-target note"); + + let plan = prepare_update( + &runtime, + &token, + &json!({"id": update_note_id.to_string(), "content": "hook-update-target, revised"}), + None, + ) + .await + .expect("prepare update"); + let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan]) + .await + .expect("seam call ok"); + let post_commit = match outcome { + crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit, + other => panic!("expected Committed, got {other:?}"), + }; + apply_post_commit_effects(&runtime, &token, post_commit) + .await + .expect("apply post-commit effects (update)"); + + // Delete path (soft delete — the row still exists, but the fix + // fires directly from the captured kind rather than refetching). + let mut del_note = + khive_storage::note::Note::new("local", "observation", "hook-delete-target"); + del_note.name = Some("hook-delete-target".to_string()); + let delete_note_id = del_note.id; + runtime + .notes(&token) + .expect("notes store") + .upsert_note(del_note) + .await + .expect("seed delete-target note"); + + let plan = prepare_delete( + &runtime, + &token, + &json!({"id": delete_note_id.to_string(), "hard": false}), + None, + ) + .await + .expect("prepare delete"); + let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan]) + .await + .expect("seam call ok"); + let post_commit = match outcome { + crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit, + other => panic!("expected Committed, got {other:?}"), + }; + assert_eq!( + post_commit, + vec![PostCommitEffect::NoteDeleted { + note_id: delete_note_id, + kind: "observation".to_string(), + }], + "a committed note delete must schedule exactly one NoteDeleted post-commit effect" + ); + apply_post_commit_effects(&runtime, &token, post_commit) + .await + .expect("apply post-commit effects (delete)"); + + let seen = fired.lock().expect("lock").clone(); + assert!( + seen.contains(&("observation".to_string(), update_note_id)), + "the note-mutation hook must fire for the atomic UPDATE path: {seen:?}" + ); + assert!( + seen.contains(&("observation".to_string(), delete_note_id)), + "the note-mutation hook must fire for the atomic DELETE path: {seen:?}" + ); + } + /// B3 fix round (codex REJECT Blocker 1): atomic delete must purge the /// note's FTS row and vector row for BOTH soft and hard delete — parity /// with `KhiveRuntime::delete_note`'s index-cleanup contract. diff --git a/crates/khive-runtime/src/atomic_runner.rs b/crates/khive-runtime/src/atomic_runner.rs index af67b644..7bb5736b 100644 --- a/crates/khive-runtime/src/atomic_runner.rs +++ b/crates/khive-runtime/src/atomic_runner.rs @@ -110,8 +110,10 @@ impl AtomicOpPlan { /// [`UpdatePlan`] carries a [`PostCommitEffect`] field for the `update` /// reindex caveat (ADR-099 D3). Since the B3 fix round (GAP-5), /// [`GtdTransitionPlan`]/[`GtdCompletePlan`] also carry one, for the - /// best-effort lifecycle audit row. Every other admissible verb's apply - /// is pure DML with no deferred side effect. `merge`'s existing + /// best-effort lifecycle audit row. Since the #750 fix-round 2 (codex + /// r2 High 2), [`DeletePlan`] also carries one, for the note-mutation + /// hook fire on a committed note delete. Every other admissible verb's + /// apply is pure DML with no deferred side effect. `merge`'s existing /// post-transaction vector re-insert (D3: "merge already performs its /// vector re-insert after its transaction") is the handler's own /// pre-existing behavior outside this plan shape — [`MergePlan`] itself @@ -121,6 +123,9 @@ impl AtomicOpPlan { AtomicOpPlan::Update(p) if p.post_commit != PostCommitEffect::None => { Some(p.post_commit.clone()) } + AtomicOpPlan::Delete(p) if p.post_commit != PostCommitEffect::None => { + Some(p.post_commit.clone()) + } AtomicOpPlan::GtdTransition(p) if p.post_commit != PostCommitEffect::None => { Some(p.post_commit.clone()) } @@ -498,6 +503,7 @@ mod tests { }, guard: Some(AffectedRowGuard::exactly(1)), }], + post_commit: PostCommitEffect::None, }) } diff --git a/crates/khive-runtime/src/curation.rs b/crates/khive-runtime/src/curation.rs index 484cd7b3..9c805840 100644 --- a/crates/khive-runtime/src/curation.rs +++ b/crates/khive-runtime/src/curation.rs @@ -669,6 +669,13 @@ impl KhiveRuntime { if text_changed { self.reindex_note(token, ¬e).await?; + // #750 fix-round 1: text_changed means the note's embedding was + // just reindexed, which any pack-owned vector-derived cache + // (e.g. khive-pack-memory's warm ANN index) needs to know about — + // reached via this generic hook so khive-runtime/khive-pack-kg + // never take a dependency on khive-pack-memory. No-op when no + // pack has installed a hook. + self.fire_note_mutation_hook(¬e.kind, note.id).await; } Ok(note) @@ -775,6 +782,16 @@ impl KhiveRuntime { if !dry_run && !self.registered_embedding_model_names().is_empty() { self.reindex_note(token, &updated_note).await?; + // #750 fix-round 2 (codex r2 High 1): a merge changes the same + // ANN corpus as update_note's text_changed branch — the + // surviving note gets a new vector via reindex_note above, and + // the source note is tombstoned by merge_note_sql. Fire the + // same hook update_note fires after its own reindex (see + // ~line 670 above) so a pack-owned vector cache (e.g. + // khive-pack-memory's warm ANN index) is notified regardless of + // which public write path reached the corpus change. + self.fire_note_mutation_hook(&updated_note.kind, updated_note.id) + .await; } Ok(summary) } diff --git a/crates/khive-runtime/src/lib.rs b/crates/khive-runtime/src/lib.rs index b3544fec..d3540bfd 100644 --- a/crates/khive-runtime/src/lib.rs +++ b/crates/khive-runtime/src/lib.rs @@ -96,7 +96,7 @@ pub use retrieval::{SearchHit, SearchSource}; pub use runtime::{ assert_db_anchor_consistent, parse_pack_list, resolve_db_anchor, resolve_project_actor_id, runtime_config_from_khive_config, BackendId, EntityTypeValidatorFn, KhiveRuntime, - NamespaceToken, RuntimeConfig, + NamespaceToken, NoteMutationHookFn, RuntimeConfig, }; pub use secret_gate::SecretMatch; pub use validation::{ diff --git a/crates/khive-runtime/src/operations.rs b/crates/khive-runtime/src/operations.rs index 17bb1b23..e5b7aa83 100644 --- a/crates/khive-runtime/src/operations.rs +++ b/crates/khive-runtime/src/operations.rs @@ -2020,26 +2020,50 @@ impl KhiveRuntime { Ok(paths) } - /// Batch-query for soft-deleted entity UUIDs in `ids`. + /// Batch-query for soft-deleted UUIDs in `ids`, across BOTH the entities + /// and notes tables. /// - /// Returns the subset of `ids` that have `deleted_at IS NOT NULL` in the - /// entities table. Takes `Vec` (not an iterator) so the async - /// state machine holds only owned data — no iterator borrow across yields. + /// Neighbor/traverse candidates can be note-kind nodes (e.g. reached via + /// `annotates` edges) as well as entities; the original Fix-2 screen only + /// consulted `entities`, so soft-deleted note targets leaked through and + /// hydrated as blank/missing hits (#748). This is a view-layer read-only + /// screen — it does not touch edges or mutate any data. + /// + /// Returns the subset of `ids` that have `deleted_at IS NOT NULL` in + /// either table. Takes `Vec` (not an iterator) so the async state + /// machine holds only owned data — no iterator borrow across yields. async fn deleted_entity_ids(&self, ids: Vec) -> std::collections::HashSet { if ids.is_empty() { return std::collections::HashSet::new(); } let id_strs: Vec = ids.iter().map(|u| u.to_string()).collect(); - let placeholders = id_strs - .iter() - .enumerate() - .map(|(i, _)| format!("?{}", i + 1)) + let n = id_strs.len(); + // Each UNION half gets its OWN numbered-placeholder block (?1..?n for + // entities, ?(n+1)..?(2n) for notes) — numbered SQLite params bind by + // index, so reusing the same numbers across halves would silently + // collapse to a single shared block instead of binding the full list + // twice (see khive-db/src/stores/graph.rs batch_neighbors: "each half + // is a fully independent positional-parameter block"). + let entities_placeholders = (0..n) + .map(|i| format!("?{}", i + 1)) + .collect::>() + .join(","); + let notes_placeholders = (0..n) + .map(|i| format!("?{}", n + i + 1)) .collect::>() .join(","); let sql_str = format!( - "SELECT id FROM entities WHERE id IN ({placeholders}) AND deleted_at IS NOT NULL" + "SELECT id FROM entities WHERE id IN ({entities_placeholders}) AND deleted_at IS NOT NULL \ + UNION \ + SELECT id FROM notes WHERE id IN ({notes_placeholders}) AND deleted_at IS NOT NULL" ); - let params: Vec = id_strs.into_iter().map(SqlValue::Text).collect(); + // Same id list bound twice — once per UNION arm's independent placeholder block. + let params: Vec = id_strs + .iter() + .chain(id_strs.iter()) + .cloned() + .map(SqlValue::Text) + .collect(); let stmt = SqlStatement { sql: sql_str, params, @@ -3105,7 +3129,16 @@ impl KhiveRuntime { format!(" AND namespace IN ({})", placeholders.join(", ")) }); + // #749: a UUID can legitimately exist in more than one scanned table + // (e.g. an entity id string that also happens to be an edge id — the + // scan is purely a text-prefix LIKE across independent tables, not a + // substrate-exclusive lookup). Without dedup, a single record hit + // twice across tables inflated `matches.len()` past 1 and produced a + // false `AmbiguousPrefix` naming the SAME UUID twice. `seen` tracks + // UUIDs already pushed so `matches` (and thus every length check, + // including the early-exit below) reflects DISTINCT UUIDs only. let mut matches: Vec = Vec::new(); + let mut seen: std::collections::HashSet = std::collections::HashSet::new(); let mut reader = self.sql().reader().await.map_err(RuntimeError::Storage)?; for (table, has_deleted_at) in tables { @@ -3131,7 +3164,9 @@ impl KhiveRuntime { for row in rows { if let Some(col) = row.columns.first() { if let SqlValue::Text(s) = &col.value { - matches.push(s.clone()); + if seen.insert(s.clone()) { + matches.push(s.clone()); + } } } } @@ -3444,6 +3479,13 @@ impl KhiveRuntime { event_store.append_event(event).await.map_err(|e| { RuntimeError::Internal(format!("delete_note: event store write failed: {e}")) })?; + // #750 fix-round 1: a soft OR hard delete removes the note's + // vectors/FTS document above — any pack-owned vector-derived + // cache (e.g. khive-pack-memory's warm ANN index) needs to know + // the corpus changed, reached via this generic hook so + // khive-runtime never takes a dependency on khive-pack-memory. + // No-op when no pack has installed a hook. + self.fire_note_mutation_hook(¬e.kind, id).await; } Ok(deleted) } @@ -6333,6 +6375,80 @@ mod tests { ); } + /// #749: a single UUID legitimately present in TWO scanned tables + /// (entities and notes here) must resolve cleanly to that one UUID, not + /// a false `AmbiguousPrefix` naming the same UUID twice. Before the fix, + /// `resolve_prefix_inner` pushed every table hit into `matches` with no + /// cross-table dedup, so `matches.len()` became 2 for a single record. + #[tokio::test] + async fn resolve_prefix_cross_table_duplicate_uuid_resolves_cleanly() { + use khive_storage::entity::Entity; + + let rt = rt(); + let tok = NamespaceToken::local(); + let shared_id = Uuid::parse_str("ccddeeff-1111-4000-8000-000000000001").unwrap(); + + let mut entity = Entity::new("local", "concept", "Nvk749Entity"); + entity.id = shared_id; + rt.entities(&tok) + .unwrap() + .upsert_entity(entity) + .await + .unwrap(); + + let mut note = Note::new("local", "observation", "nvk749 note with the same id"); + note.id = shared_id; + rt.notes(&tok).unwrap().upsert_note(note).await.unwrap(); + + let resolved = rt + .resolve_prefix(&tok, "ccddeeff") + .await + .expect("#749: a UUID present in two tables must not be reported as ambiguous"); + assert_eq!( + resolved, + Some(shared_id), + "#749: cross-table duplicate must resolve to the single shared UUID" + ); + } + + /// #749: the early-exit inside the per-table scan loop (`if matches.len() + /// > 1 { break }`) must also operate on DEDUPED state — otherwise a + /// cross-table duplicate could still short-circuit the scan before a + /// later table contributes the SAME UUID again, which would have masked + /// the bug rather than exercising it. This drives the duplicate through + /// the earliest two tables scanned (entities, notes) so the early-exit + /// path is the one under test, not a post-loop dedup applied too late. + #[tokio::test] + async fn resolve_prefix_early_exit_uses_deduped_match_count() { + use khive_storage::entity::Entity; + + let rt = rt(); + let tok = NamespaceToken::local(); + let shared_id = Uuid::parse_str("ddeeff11-2222-4000-8000-000000000002").unwrap(); + + // entities and notes are the first two tables scanned inside + // resolve_prefix_inner — the same UUID in both must not trip the + // mid-scan `matches.len() > 1` break as if two distinct UUIDs had + // been found. + let mut entity = Entity::new("local", "concept", "Nvk749bEntity"); + entity.id = shared_id; + rt.entities(&tok) + .unwrap() + .upsert_entity(entity) + .await + .unwrap(); + + let mut note = Note::new("local", "observation", "nvk749b note with the same id"); + note.id = shared_id; + rt.notes(&tok).unwrap().upsert_note(note).await.unwrap(); + + let resolved = rt + .resolve_prefix(&tok, "ddeeff11") + .await + .expect("#749: deduped early-exit must not falsely report ambiguity"); + assert_eq!(resolved, Some(shared_id)); + } + // ---- Event resolution tests (issue #30) ---- // // resolve_prefix and handle_get already include events; these tests are @@ -7847,11 +7963,25 @@ mod tests { .await .unwrap(); assert_eq!(before.len(), 1); + let edge_id = before[0].edge_id; // Soft delete must NOT cascade edges (data-vs-view principle). let deleted = rt.delete_note(&tok, note_target.id, false).await.unwrap(); assert!(deleted, "soft delete must return true"); + // The edge itself must survive the soft delete — checked at the + // storage/edge layer directly (`get_edge`), not through `neighbors()`. + // `neighbors()` is a VIEW query and, per #748, now correctly screens + // out soft-deleted note targets — so it no longer surfaces this edge + // once note_target is soft-deleted, even though the edge row itself + // is untouched (data-vs-view principle: the edge is data, what + // `neighbors()` shows is a view decision). + let edge_after = rt.get_edge(&tok, edge_id).await.unwrap(); + assert!( + edge_after.is_some(), + "soft delete must NOT cascade edges; get_edge returned None" + ); + let after = rt .neighbors( &tok, @@ -7864,8 +7994,8 @@ mod tests { .unwrap(); assert_eq!( after.len(), - 1, - "soft delete must NOT cascade edges; got {after:?}" + 0, + "#748: neighbors() must screen out the soft-deleted note target; got {after:?}" ); } diff --git a/crates/khive-runtime/src/pack.rs b/crates/khive-runtime/src/pack.rs index b8888641..93217655 100644 --- a/crates/khive-runtime/src/pack.rs +++ b/crates/khive-runtime/src/pack.rs @@ -203,6 +203,22 @@ pub trait PackRuntime: Send + Sync { /// is the correct behaviour for bare runtimes without packs. fn register_entity_type_validator(&self, _runtime: &KhiveRuntime) {} + /// Install a pack-owned note-mutation hook on the runtime (#750 fix-round 1). + /// + /// Called by the transport during pack initialisation, after the registry + /// is built and before the first verb dispatch — same timing as + /// `register_entity_type_validator`. Packs that cache derived state keyed + /// by note content (e.g. `khive-pack-memory`'s warm ANN index) should + /// override this to install a hook via `KhiveRuntime::install_note_mutation_hook`, + /// so `update_note`/`delete_note` notify them even when the mutation + /// arrived through a different pack's verb that has no dependency on the + /// reacting pack (e.g. KG's `update`/`delete` on a `kind="memory"` note). + /// + /// The default no-op leaves the runtime hook absent (skip-when-None), + /// which is the correct behaviour for packs that don't cache note-derived + /// state and for bare runtimes without packs. + fn register_note_mutation_hook(&self, _runtime: &KhiveRuntime) {} + /// Warm up any in-memory state from persisted snapshots (optional). /// /// Called by the transport after all packs are registered but before @@ -1534,6 +1550,23 @@ impl VerbRegistry { } } + /// Invoke `PackRuntime::register_note_mutation_hook` on every registered pack + /// (#750 fix-round 1). + /// + /// Called by the transport during startup, after the registry is built and + /// before the first verb dispatch, so that note-mutation notifications at + /// the runtime layer are active for all write paths — including KG's + /// `update`/`delete` verbs reaching a `kind="memory"` note, which have no + /// crate-level dependency on `khive-pack-memory`. + /// + /// Packs whose `register_note_mutation_hook` is the default no-op pay no + /// overhead. + pub fn call_register_note_mutation_hooks(&self, runtime: &KhiveRuntime) { + for pack in self.packs.iter() { + pack.register_note_mutation_hook(runtime); + } + } + /// Invoke `PackRuntime::warm` on every registered pack. /// Called by the daemon at boot (in a background task) so expensive in-memory /// state (ANN indexes) is pre-loaded without blocking request serving. diff --git a/crates/khive-runtime/src/runtime.rs b/crates/khive-runtime/src/runtime.rs index 8903b607..e2b4809c 100644 --- a/crates/khive-runtime/src/runtime.rs +++ b/crates/khive-runtime/src/runtime.rs @@ -28,6 +28,24 @@ use crate::error::{RuntimeError, RuntimeResult}; pub type EntityTypeValidatorFn = Arc) -> Result, RuntimeError> + Send + Sync>; +/// Callback type for a pack-installed note-mutation hook (#750 fix-round 1). +/// +/// Invoked by `update_note` (when the note's text/embedding actually +/// changed) and `delete_note` (soft or hard) with `(note_kind, note_id)`, +/// AFTER the mutation has been durably applied. Returns a boxed future so +/// the hook can await async cache-invalidation work (e.g. +/// `khive-pack-memory`'s ANN warm-cache generation bump) without the +/// `KhiveRuntime`/`khive-runtime` crate depending on any pack crate — the +/// dependency points the other way (packs depend on `khive-runtime`, not +/// vice versa), so this is the same "runtime exposes an extension point, +/// pack installs into it" shape as `EntityTypeValidatorFn` / +/// `install_edge_rules`, just async. +pub type NoteMutationHookFn = Arc< + dyn Fn(String, uuid::Uuid) -> std::pin::Pin + Send>> + + Send + + Sync, +>; + pub use crate::config::{ assert_db_anchor_consistent, parse_pack_list, resolve_db_anchor, resolve_project_actor_id, runtime_config_from_khive_config, BackendId, NamespaceToken, RuntimeConfig, @@ -75,6 +93,18 @@ pub struct KhiveRuntime { /// without packs), entity-type validation is skipped — the pack handler layer /// is the primary enforcement point, same as for `valid_entity_kinds`. entity_type_validator: Arc>>, + /// Pack-installed note-mutation hook (#750 fix-round 1). + /// + /// When `Some`, `update_note` (on text change) and `delete_note` (soft + /// or hard) call this after the mutation is durably applied, so a pack + /// that caches derived state keyed by note content (e.g. `khive-pack-memory`'s + /// warm ANN index) can invalidate/advance its own generation counter even + /// when the mutation arrived through a different pack's verb (e.g. KG's + /// `update`/`delete` on a `kind="memory"` note) that has no dependency on + /// the reacting pack. `None` when no pack installs one (bare runtime, or + /// no pack cares about note-mutation notifications) — the call becomes a + /// no-op check of an `Option`. + note_mutation_hook: Arc>>, } impl KhiveRuntime { @@ -115,6 +145,7 @@ impl KhiveRuntime { valid_entity_kinds: Arc::new(RwLock::new(Vec::new())), valid_note_kinds: Arc::new(RwLock::new(Vec::new())), entity_type_validator: Arc::new(RwLock::new(None)), + note_mutation_hook: Arc::new(RwLock::new(None)), }) } @@ -143,6 +174,7 @@ impl KhiveRuntime { valid_entity_kinds: Arc::new(RwLock::new(Vec::new())), valid_note_kinds: Arc::new(RwLock::new(Vec::new())), entity_type_validator: Arc::new(RwLock::new(None)), + note_mutation_hook: Arc::new(RwLock::new(None)), }) } @@ -170,6 +202,7 @@ impl KhiveRuntime { valid_entity_kinds: Arc::new(RwLock::new(Vec::new())), valid_note_kinds: Arc::new(RwLock::new(Vec::new())), entity_type_validator: Arc::new(RwLock::new(None)), + note_mutation_hook: Arc::new(RwLock::new(None)), } } @@ -200,7 +233,7 @@ impl KhiveRuntime { /// this returns a new `KhiveRuntime` backed by the main /// `Arc` and sharing all registry state (`embedder_registry`, /// `edge_rules`, `valid_entity_kinds`, `valid_note_kinds`, - /// `entity_type_validator`) with `self`. + /// `entity_type_validator`, `note_mutation_hook`) with `self`. /// No database I/O occurs; no embedding models are reloaded. /// /// Use `core()` for notes and entities that must reside in the shared graph @@ -226,6 +259,7 @@ impl KhiveRuntime { valid_entity_kinds: self.valid_entity_kinds.clone(), valid_note_kinds: self.valid_note_kinds.clone(), entity_type_validator: self.entity_type_validator.clone(), + note_mutation_hook: self.note_mutation_hook.clone(), } } } @@ -616,6 +650,38 @@ impl KhiveRuntime { } } + /// Install a pack-owned note-mutation hook (#750 fix-round 1). + /// + /// Overwrites any previously-installed hook, same single-slot semantics + /// as [`install_entity_type_validator`](Self::install_entity_type_validator). + /// In practice only one pack (`khive-pack-memory`) installs one today; + /// if a second pack ever needs this, the slot should be widened to a + /// `Vec` at that point rather than silently overwritten. + pub fn install_note_mutation_hook(&self, f: NoteMutationHookFn) { + if let Ok(mut guard) = self.note_mutation_hook.write() { + *guard = Some(f); + } + } + + /// Invoke the pack-installed note-mutation hook, if any. + /// + /// `kind` is the note's `kind` string (e.g. `"memory"`); `id` is the + /// note's UUID. No-op when no hook is installed (bare runtime, or no + /// pack cares). Errors inside the hook are the hook's own concern to + /// handle/log — this call site cannot propagate a failure without + /// changing `update_note`/`delete_note`'s already-committed success + /// return value. + pub(crate) async fn fire_note_mutation_hook(&self, kind: &str, id: uuid::Uuid) { + let hook = self + .note_mutation_hook + .read() + .ok() + .and_then(|guard| guard.clone()); + if let Some(hook) = hook { + hook(kind.to_string(), id).await; + } + } + /// Snapshot of currently-installed pack edge rules. /// /// This is the SAME composed rule set `validate_edge_relation_endpoints` diff --git a/crates/kkernel/src/atomic_apply.rs b/crates/kkernel/src/atomic_apply.rs index bd96f8ed..3e1adac3 100644 --- a/crates/kkernel/src/atomic_apply.rs +++ b/crates/kkernel/src/atomic_apply.rs @@ -149,6 +149,19 @@ pub(crate) async fn execute_atomic_ops_file( let verb_registry = verb_registry_builder .build() .context("building VerbRegistry for --atomic kind resolution")?; + // #750 fix-round 2 (codex r2 High 2): every other entry point that + // builds a `VerbRegistry` from a freshly constructed `KhiveRuntime` + // (`khive-mcp`'s `serve.rs`/`server.rs`) calls this so a pack-installed + // note-mutation hook (e.g. khive-pack-memory's warm ANN invalidation) + // is actually wired into the runtime handle used for the rest of this + // process's lifetime. `--atomic` built its own registry without this + // call, so `fire_note_mutation_hook` was a guaranteed no-op for the + // whole `--atomic` process regardless of whether a call site invoked + // it. This closes the in-process half of the gap; the note-mutation + // hook's effect (a bumped `AnnState` generation) is itself process- + // local, so it still cannot reach a separately-running daemon's warm + // cache — see the cross-process analysis in the fix-round-2 report. + verb_registry.call_register_note_mutation_hooks(&runtime); // ── async prepare pass (reads only, no writes) ─────────────────────────── let mut plans: Vec = Vec::with_capacity(ops.len());