From 43a017543505f0db1655498f509d999c1bd04352 Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:08:00 -0400 Subject: [PATCH 1/6] fix(kg): persist tags on note create (#747) create(kind=note, tags=[...]) whitelisted `tags` as an accepted param but never read it in the note branch, silently dropping the value. Notes have no dedicated tags column (schema: khive-db/sql/notes-ddl.sql) so, matching the existing convention already used by memory.remember and honored by this pack's own search/list note-tag filters, tags now merge into properties["tags"]. Precedence when the caller supplies both a top-level tags param and a conflicting properties.tags: the top-level tags param wins (documented in merge_note_tags and asserted by a regression test). Co-Authored-By: Claude Fable 5 --- crates/khive-pack-kg/src/handlers/common.rs | 32 ++++ crates/khive-pack-kg/src/handlers/create.rs | 3 +- crates/khive-pack-kg/tests/integration.rs | 157 ++++++++++++++++++++ 3 files changed, 191 insertions(+), 1 deletion(-) 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..9eb3d4f8 100644 --- a/crates/khive-pack-kg/tests/integration.rs +++ b/crates/khive-pack-kg/tests/integration.rs @@ -9886,3 +9886,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") + ); +} From 91f5ae1dc74f6a42e9acc8830f3403af68748ff5 Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:16:45 -0400 Subject: [PATCH 2/6] fix(runtime): screen soft-deleted notes out of neighbors() (#748) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit deleted_entity_ids only queried the entities table, so a soft-deleted note-kind neighbor (e.g. reached via an annotates edge) leaked through neighbors()/neighbors_with_query_directed()/traverse() and hydrated as a blank/missing hit. Extended the screen to UNION the notes table using an independent numbered-placeholder block per UNION half (matching the existing batch_neighbors convention in khive-db/src/stores/graph.rs — SQLite numbered params bind by index, so reusing the same numbers across UNION halves would collapse to one shared block instead of binding the id list twice). View-layer only: no edges are touched, delete() semantics are unchanged. Updated annotated_note_soft_delete_preserves_annotate_edge, which had been asserting 'edge not cascaded' by reading it back through neighbors() — that assertion happened to pass only because neighbors() was, itself, exhibiting the #748 bug (failing to filter the soft-deleted note target). It now checks edge survival directly via get_edge() and separately asserts neighbors() correctly omits the soft-deleted note. Co-Authored-By: Claude Fable 5 --- crates/khive-pack-kg/tests/integration.rs | 166 ++++++++++++++++++++++ crates/khive-runtime/src/operations.rs | 62 ++++++-- 2 files changed, 216 insertions(+), 12 deletions(-) diff --git a/crates/khive-pack-kg/tests/integration.rs b/crates/khive-pack-kg/tests/integration.rs index 9eb3d4f8..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 diff --git a/crates/khive-runtime/src/operations.rs b/crates/khive-runtime/src/operations.rs index 17bb1b23..bddee354 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, @@ -7847,11 +7871,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 +7902,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:?}" ); } From 12bdfca6e9384a6eb4f2f134712901df1d470ca2 Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:19:54 -0400 Subject: [PATCH 3/6] fix(runtime): dedup resolve_prefix_inner matches across tables (#749) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit resolve_prefix_inner scanned entities/notes/events/graph_edges in sequence, pushing every hit into matches with no cross-table dedup. A UUID legitimately present in more than one table produced duplicate entries, so matches.len() > 1 fired AmbiguousPrefix even though exactly one record was addressed — the error message named the same UUID twice. Added a seen: HashSet so matches (and every length check derived from it, including the mid-scan early-exit 'if matches.len() > 1 { break }') reflects distinct UUIDs only. Two genuinely distinct UUIDs sharing a prefix still trip AmbiguousPrefix (regression-covered by the pre-existing resolve_prefix_ambiguous_same_namespace test, reconfirmed green). Co-Authored-By: Claude Fable 5 --- crates/khive-runtime/src/operations.rs | 87 +++++++++++++++++++++++++- 1 file changed, 86 insertions(+), 1 deletion(-) diff --git a/crates/khive-runtime/src/operations.rs b/crates/khive-runtime/src/operations.rs index bddee354..2a8e122e 100644 --- a/crates/khive-runtime/src/operations.rs +++ b/crates/khive-runtime/src/operations.rs @@ -3129,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 { @@ -3155,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()); + } } } } @@ -6357,6 +6368,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 From 3b095ed7a22c546e5771e98d0fd40e5c22f52c24 Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:38:40 -0400 Subject: [PATCH 4/6] fix(memory): write-generation-checked ANN warm-cache install (#750) ensure_ann_for_model_inner installed a finished build via entry(key).or_insert(bridge), which always kept whichever build reached the install call first and was a permanent no-op once the key was occupied - so a slow build that snapshotted the corpus before a still-in-flight write committed could install after a newer write's invalidate+re-warm cycle and silently clobber freshness, with no later rebuild able to correct it (the double-fingerprint check only bounds the scan window, not the gap between computing fp_after and the eventual or_insert, e.g. during persist_snapshot's I/O). Added a per-model monotonic write-generation counter (AnnState.generations, bumped by memory.remember alongside the existing cache invalidation - no existing sequence/timestamp signal on the write path was suitable: the notes table's updated_at is wall-clock and not collision-free under concurrent writes, and the vec0 virtual vector table exposes no reliable rowid). ensure_ann_for_model now captures this counter's value BEFORE its own fast-path check and before the corpus scan, stamps it on the resulting AnnBridge, and installs via install_if_fresher: a real compare-and-replace that only overwrites an existing entry when the candidate's generation is >= the installed entry's. ensure_ann_background's own fast path is generation-aware the same way, so a caller triggered by a write that just bumped the counter always spawns a rebuild rather than trusting a stale present entry. Closed the residual recall-side gap too: search_loaded's cache-hit check was presence-only, so a stale-but-installed entry (one that lost the install race for freshness but still reached an empty slot first) would still be served directly. handlers/common.rs's recall path now gates on the new ann::is_current instead, treating "present but stale" the same as a genuine miss and falling through to the existing synchronous ensure_ann_for_model fallback. ensure_ann_background keeps its fire-and-forget shape; nothing is serialized onto the request path except this pre-existing on-demand-warm fallback. Removed the bounded fresh-runtime + poll retry workarounds from the three tests exercising this (ns733_recall_ann_overfetch_retry_loop_*, ns733b_recall_verbose_multi_model_breakdown_*, ns733b_recall_candidates_multi_model_*) - all three now assert directly in a single attempt. Stress-verified via the compiled test binary run standalone 50 consecutive times each (150 runs total, each a fresh process with a fresh in-memory KhiveRuntime): 150/150 passed, 0 failures. Added direct regression coverage for install_if_fresher's compare-and- replace invariant and is_current's stale-is-a-miss semantics (ann.rs::tests). A true end-to-end interleaving test (pause a build mid-scan, land a write, assert the newer generation wins) was not attempted: no test-only pause hook exists in ensure_ann_for_model_inner, and adding one solely to force a schedule would test the hook rather than the production path - the unit-level invariant tests plus the 150-run stress evidence stand in for it. Co-Authored-By: Claude Fable 5 --- crates/khive-pack-memory/src/ann.rs | 288 +++++++++- .../khive-pack-memory/src/handlers/common.rs | 69 ++- .../khive-pack-memory/src/handlers/recall.rs | 534 +++++++----------- .../src/handlers/remember.rs | 10 + 4 files changed, 532 insertions(+), 369 deletions(-) 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/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; } } From fd0880763230af240e9bb77552d35417e6112323 Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 02:15:22 -0400 Subject: [PATCH 5/6] fix(runtime): close #750 ANN-generation coverage gap (codex fix-round 1) The #750 freshness gate (ann::is_current) only holds if every memory-corpus mutation advances AnnState::generations. The only production bump was memory.remember's own write path; memory.prune, KG update() reaching a kind="memory" note's reindex, and KG delete() on a kind="memory" note all left the warm ANN cache trusted as current after mutating the corpus. Extends khive-runtime's existing pack-installed extension-point pattern (EntityTypeValidatorFn / install_entity_type_validator / PackRuntime:: register_entity_type_validator / call_register_entity_type_validators) with a parallel NoteMutationHookFn seam: update_note (on text_changed reindex) and delete_note (soft or hard) now fire a pack-installed hook after the mutation is durably applied, with zero cost to packs that don't install one. khive-pack-memory installs a hook that invalidates + bumps the generation for kind="memory" notes, the same two calls memory.remember already makes. khive-mcp's serve.rs/server.rs wire call_register_note_mutation_hooks at boot, alongside the existing call_register_entity_type_validators call -- load-bearing: without it the hook mechanism is entirely inert in the running daemon. memory.prune (same crate as ann, no inversion needed) invalidates and bumps inline, mirroring remember.rs's pattern directly. Added three regression tests (pack.rs's note_mutation_hook_tests): warm the ANN cache, mutate a memory-kind note via each of the three paths with no subsequent memory.remember, assert ann::is_current is false afterward. These assert the white-box is_current signal rather than a black-box recall-result check -- a throwaway experiment (temporarily disabling all three fix-round-1 call sites) proved the black-box form of the delete-path test is NOT discriminating: recall's post-search hydration filters soft-deleted notes regardless of ANN cache freshness, so "the deleted note is absent from results" passed even with the bug reintroduced. is_current does not have that blind spot; the disable/re-enable cycle confirmed all three corrected tests genuinely fail without the fix and pass with it restored. Co-Authored-By: Claude Fable 5 --- crates/khive-mcp/src/serve.rs | 5 + crates/khive-mcp/src/server.rs | 5 + .../khive-pack-memory/src/handlers/prune.rs | 19 ++ crates/khive-pack-memory/src/pack.rs | 275 ++++++++++++++++++ crates/khive-runtime/src/curation.rs | 7 + crates/khive-runtime/src/lib.rs | 2 +- crates/khive-runtime/src/operations.rs | 7 + crates/khive-runtime/src/pack.rs | 33 +++ crates/khive-runtime/src/runtime.rs | 68 ++++- 9 files changed, 419 insertions(+), 2 deletions(-) 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-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/pack.rs b/crates/khive-pack-memory/src/pack.rs index ca822a26..46f7033e 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,244 @@ 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)" + ); + } + + 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/curation.rs b/crates/khive-runtime/src/curation.rs index 484cd7b3..69f49d8d 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) 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 2a8e122e..e5b7aa83 100644 --- a/crates/khive-runtime/src/operations.rs +++ b/crates/khive-runtime/src/operations.rs @@ -3479,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) } 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` From ecfe461877700094b9648bc582138b879d56579b Mon Sep 17 00:00:00 2001 From: oceanwaves630 <33291608+oceanwaves630@users.noreply.github.com> Date: Thu, 9 Jul 2026 02:43:47 -0400 Subject: [PATCH 6/6] fix(runtime): close #750 note-mutation-hook coverage gaps (codex fix-round 2) Codex r2 REJECT, 2 Highs: High 1: merge_note's non-dry-run success path reindexed the surviving note but never fired the note-mutation hook (unlike update_note's identical text-changed branch). Fixed by firing it after reindex_note, same shape as update_note. New regression test fr2_kg_merge_invalidates_warm_ann_without_subsequent_remember, verified to discriminate via disable/re-enable. High 2 (verified AnnState::generations is process-local in-memory state, per source read of khive-pack-memory/src/ann.rs -- confirms the coordinator's hypothesis): in-process portion (fixed here): - kkernel exec --atomic's registry build never called call_register_note_mutation_hooks, so the hook slot was always None for the whole --atomic process. - atomic_prepare.rs's ReindexNote post-commit handler called reindex_note directly, bypassing update_note (and its hook-fire) entirely. - DeletePlan carried NO post_commit field at all -- a deeper gap than ReindexNote's bypass. Added a post_commit field, a new PostCommitEffect::NoteDeleted variant, and wired it through all five DeletePlan construction sites plus apply_post_commit_effects. New regression test in khive-runtime's own atomic_prepare.rs tests module: atomic_note_update_and_delete_post_commit_fire_the_note_mutation_hook, verified to discriminate for BOTH paths independently via disable/re-enable. cross-process portion (NOT fixed here, documented for a follow-up issue): kkernel exec --atomic and kkernel reindex run in separate OS processes from any daemon; a hook fired inside them can only mutate their own transient AnnState, never a running daemon's warm cache. kkernel reindex additionally has no in-process AnnState concept at all -- it invalidates only persisted Vamana snapshot rows, a different mechanism. Full evidence in impl_report_quadfix.md. Gates: cargo fmt --all -- --check; cargo clippy -p khive-pack-kg -p khive-runtime -p khive-pack-memory -p kkernel --all-targets -- -D warnings; cargo test -p khive-pack-kg -p khive-runtime -p khive-pack-memory. All green. Co-Authored-By: Claude Fable 5 --- crates/khive-pack-memory/src/pack.rs | 89 ++++++++++++ crates/khive-runtime/src/atomic_plan.rs | 15 +++ crates/khive-runtime/src/atomic_prepare.rs | 149 +++++++++++++++++++++ crates/khive-runtime/src/atomic_runner.rs | 10 +- crates/khive-runtime/src/curation.rs | 10 ++ crates/kkernel/src/atomic_apply.rs | 13 ++ 6 files changed, 284 insertions(+), 2 deletions(-) diff --git a/crates/khive-pack-memory/src/pack.rs b/crates/khive-pack-memory/src/pack.rs index 46f7033e..45c283e4 100644 --- a/crates/khive-pack-memory/src/pack.rs +++ b/crates/khive-pack-memory/src/pack.rs @@ -896,6 +896,95 @@ mod note_mutation_hook_tests { ); } + /// #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], 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 69f49d8d..9c805840 100644 --- a/crates/khive-runtime/src/curation.rs +++ b/crates/khive-runtime/src/curation.rs @@ -782,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/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());