Skip to main content

khive_runtime/
curation.rs

1// Licensed under the Apache License, Version 2.0.
2
3// FILE SIZE JUSTIFICATION: curation.rs holds entity/note/edge patch types alongside
4// their update and merge implementations. The implementations share private helpers
5// (merge_properties, namespace checks, dedup policy) that need pub(crate) access to
6// runtime internals. Inline tests cover merge semantics that require direct access to
7// those helpers. Split plan: extract patch types into `curation/patch.rs` and merge
8// logic into `curation/merge.rs` once the dedup policy API stabilises.
9//! Curation operations: entity update/merge and edge-list filter type.
10
11use std::collections::HashSet;
12
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15use uuid::Uuid;
16
17use khive_db::SqliteError;
18use khive_storage::note::Note;
19use khive_storage::types::{EdgeFilter, TextDocument};
20use khive_storage::{EdgeRelation, Entity, SubstrateKind};
21use khive_types::EventKind;
22use rusqlite::OptionalExtension;
23
24use crate::error::{RuntimeError, RuntimeResult};
25use crate::operations::canonical_edge_endpoints;
26use crate::runtime::{KhiveRuntime, NamespaceToken};
27
28// ---------------------------------------------------------------------------
29// Public types
30// ---------------------------------------------------------------------------
31
32/// Patch for `update_entity`. Only `Some(_)` fields are applied; `None` means "leave unchanged".
33///
34/// For `description`:
35/// - `None` (outer) — leave the current description as-is
36/// - `Some(None)` — clear the description (set to NULL)
37/// - `Some(Some(s))` — set the description to `s`
38///
39/// For `properties` (deep-merge semantics):
40/// - `None` — leave properties as-is
41/// - `Some(value)` — deep-merge `value` into existing properties. Keys present in
42///   the patch overwrite existing keys; keys absent from the patch are preserved.
43///   Removing a key requires explicit replacement of the parent object (or a future
44///   `unset`/`null-marker` extension).
45///
46/// For `tags` — replace semantics: `Some(vec)` sets tags to exactly `vec`. To add
47/// a tag without losing existing tags, read the entity first, push the new tag,
48/// and pass the full list back.
49#[derive(Clone, Debug, Default)]
50pub struct EntityPatch {
51    pub name: Option<String>,
52    pub description: Option<Option<String>>,
53    pub properties: Option<Value>,
54    pub tags: Option<Vec<String>>,
55}
56
57/// Policy used when deduplicating two entities.
58#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
59#[serde(rename_all = "snake_case")]
60pub enum EntityDedupMergePolicy {
61    /// `into` values win on conflict. Tags are unioned. Properties from `from` fill in
62    /// keys that `into` doesn't have. This is the default.
63    #[default]
64    PreferInto,
65    /// `from` values win on conflict.
66    PreferFrom,
67    /// Deep-merge: object properties merge recursively. Scalar conflicts go to `into`.
68    Union,
69}
70
71/// Strategy for merging note content when two notes are combined.
72#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
73#[serde(rename_all = "snake_case")]
74pub enum ContentMergeStrategy {
75    #[default]
76    Append,
77    PreferInto,
78    PreferFrom,
79}
80
81/// Result returned by `merge_entity` / `merge_note`.
82#[derive(Clone, Debug, Serialize, Deserialize)]
83pub struct MergeSummary {
84    pub kept_id: Uuid,
85    pub removed_id: Uuid,
86    pub edges_rewired: usize,
87    pub properties_merged: usize,
88    pub tags_unioned: usize,
89    pub content_appended: bool,
90    pub dry_run: bool,
91}
92
93/// Patch for `update_edge`. Only `Some(_)` fields are applied; `None` means "leave unchanged".
94///
95/// For `properties` — replacement semantics (not deep merge): `Some(value)` replaces
96/// the entire metadata object. `None` leaves metadata unchanged.
97#[derive(Clone, Debug, Default)]
98pub struct EdgePatch {
99    pub relation: Option<EdgeRelation>,
100    pub weight: Option<f64>,
101    pub properties: Option<Value>,
102}
103
104/// Patch for `update_note`. Only `Some(_)` fields are applied; `None` means "leave unchanged".
105///
106/// For `salience`/`decay_factor`:
107/// - `None` (outer) — leave unchanged
108/// - `Some(None)` — clear the value
109/// - `Some(Some(v))` — set to v
110#[derive(Clone, Debug, Default)]
111pub struct NotePatch {
112    pub name: Option<Option<String>>,
113    pub content: Option<String>,
114    pub salience: Option<Option<f64>>,
115    pub decay_factor: Option<Option<f64>>,
116    pub properties: Option<Value>,
117    pub(crate) kind_status: Option<String>,
118}
119
120impl NotePatch {
121    /// Construct a `NotePatch` from the public fields only.
122    /// Use this from external crates; `kind_status` is set to `None`.
123    pub fn new(
124        name: Option<Option<String>>,
125        content: Option<String>,
126        salience: Option<Option<f64>>,
127        decay_factor: Option<Option<f64>>,
128        properties: Option<Value>,
129    ) -> Self {
130        Self {
131            name,
132            content,
133            salience,
134            decay_factor,
135            properties,
136            kind_status: None,
137        }
138    }
139}
140
141/// Filter for `list_edges` / `count_edges`.
142#[derive(Clone, Debug, Default)]
143pub struct EdgeListFilter {
144    pub source_id: Option<Uuid>,
145    pub target_id: Option<Uuid>,
146    /// Empty = any relation.
147    pub relations: Vec<EdgeRelation>,
148    pub min_weight: Option<f64>,
149    pub max_weight: Option<f64>,
150}
151
152impl From<EdgeListFilter> for EdgeFilter {
153    fn from(f: EdgeListFilter) -> Self {
154        EdgeFilter {
155            source_ids: f.source_id.into_iter().collect(),
156            target_ids: f.target_id.into_iter().collect(),
157            relations: f.relations,
158            min_weight: f.min_weight,
159            max_weight: f.max_weight,
160            ..Default::default()
161        }
162    }
163}
164
165// ---------------------------------------------------------------------------
166// Private types
167// ---------------------------------------------------------------------------
168
169// REASON: EdgeRow fields are populated via rusqlite row mapping. The struct is fully
170// constructed even when not all fields are read back after construction. The complete
171// field mapping guards against column-order bugs when the schema changes.
172#[allow(dead_code)]
173struct EdgeRow {
174    id: Uuid,
175    source_id: Uuid,
176    target_id: Uuid,
177    relation: String,
178    weight: f64,
179    created_at: i64,
180    updated_at: i64,
181    deleted_at: Option<i64>,
182    target_backend: Option<String>,
183    metadata: Option<String>,
184}
185
186// ---------------------------------------------------------------------------
187// Implementation
188// ---------------------------------------------------------------------------
189
190impl KhiveRuntime {
191    /// Patch-style entity update.
192    ///
193    /// Only fields set to `Some(_)` are changed. Re-indexes FTS5 (and vectors if configured)
194    /// when `name` or `description` changes; skips re-indexing for property/tag-only patches.
195    ///
196    /// Returns `RuntimeError::NotFound` if the entity does not exist or belongs to a different
197    /// namespace. Namespace isolation is enforced at the runtime layer.
198    pub async fn update_entity(
199        &self,
200        token: &NamespaceToken,
201        id: Uuid,
202        patch: EntityPatch,
203    ) -> RuntimeResult<Entity> {
204        // Secret gate: scan incoming text fields, properties, and tags.
205        if let Some(ref name) = patch.name {
206            crate::secret_gate::check(name)?;
207        }
208        if let Some(Some(ref desc)) = patch.description {
209            crate::secret_gate::check(desc)?;
210        }
211        if let Some(ref props) = patch.properties {
212            crate::secret_gate::check_json(props)?;
213        }
214        if let Some(ref tags) = patch.tags {
215            crate::secret_gate::check_tags(tags)?;
216        }
217        let store = self.entities(token)?;
218        let mut entity = store
219            .get_entity(id)
220            .await?
221            .ok_or_else(|| RuntimeError::NotFound(format!("entity {id}")))?;
222
223        let mut text_changed = false;
224        let mut changed_fields: Vec<&'static str> = Vec::new();
225
226        if let Some(name) = patch.name {
227            text_changed |= entity.name != name;
228            entity.name = name;
229            changed_fields.push("name");
230        }
231        if let Some(desc_patch) = patch.description {
232            text_changed |= entity.description != desc_patch;
233            entity.description = desc_patch;
234            changed_fields.push("description");
235        }
236        if let Some(props) = patch.properties {
237            let (merged, _) = merge_properties(
238                &entity.properties,
239                &Some(props),
240                EntityDedupMergePolicy::PreferFrom,
241            );
242            entity.properties = merged;
243            changed_fields.push("properties");
244        }
245        if let Some(tags) = patch.tags {
246            entity.tags = tags;
247            changed_fields.push("tags");
248        }
249
250        entity.updated_at = chrono::Utc::now().timestamp_micros();
251        store.upsert_entity(entity.clone()).await?;
252
253        if text_changed {
254            self.reindex_entity(token, &entity).await?;
255        }
256
257        let event_store = self.events(token)?;
258        let event = khive_storage::event::Event::new(
259            entity.namespace.clone(),
260            "update",
261            EventKind::EntityUpdated,
262            SubstrateKind::Entity,
263            "",
264        )
265        .with_target(entity.id)
266        .with_payload(serde_json::json!({
267            "id": entity.id,
268            "namespace": entity.namespace,
269            "changed_fields": changed_fields,
270        }));
271        event_store.append_event(event).await.map_err(|e| {
272            RuntimeError::Internal(format!("update_entity: event store write failed: {e}"))
273        })?;
274
275        Ok(entity)
276    }
277
278    /// Merge `from_id` into `into_id`.
279    ///
280    /// All edges incident to `from_id` are rewired to `into_id`. Self-loops that would
281    /// result from the rewire are dropped. Properties and tags are merged per `strategy`.
282    /// `from_id` is tombstoned with merge provenance and removed from indexes. Returns a summary.
283    ///
284    /// If `dry_run` is true, computes and returns the planned summary without mutating any rows.
285    ///
286    /// Atomic: all SQL (entity reads/writes, edge rewires, FTS updates, vec-index delete)
287    /// runs on a single pool connection inside one `BEGIN IMMEDIATE` transaction via
288    /// `merge_entity_sql`. If embedding vectors are configured, the vector re-insert for
289    /// `into_id` is performed after the transaction (requires async embedding computation).
290    pub async fn merge_entity(
291        &self,
292        token: &NamespaceToken,
293        into_id: Uuid,
294        from_id: Uuid,
295        strategy: EntityDedupMergePolicy,
296        dry_run: bool,
297    ) -> RuntimeResult<MergeSummary> {
298        if into_id == from_id {
299            return Err(RuntimeError::InvalidInput(
300                "cannot merge an entity into itself".into(),
301            ));
302        }
303        // H2 fix: enforce same-kind constraint at the runtime layer.
304        // The handler also checks this, but any direct runtime caller (CLI, tests,
305        // future SDK) would bypass the handler guard without this check here.
306        {
307            let into_entity = self.get_entity(token, into_id).await?;
308            let from_entity = self.get_entity(token, from_id).await?;
309            if into_entity.kind != from_entity.kind {
310                return Err(RuntimeError::InvalidInput(format!(
311                    "cannot merge entities of different kinds: into={} ({}), from={} ({}); \
312                     merge requires both entities to share the same kind",
313                    into_id, into_entity.kind, from_id, from_entity.kind
314                )));
315            }
316        }
317        let ns = token.namespace().as_str().to_owned();
318        let fts_table = "fts_entities".to_string();
319        let vec_tables: Vec<String> = self
320            .registered_embedding_model_names()
321            .iter()
322            .map(|name| format!("vec_{}", crate::config::sanitize_key(name)))
323            .collect();
324
325        // Ensure all required tables exist before entering the transaction.
326        // Each accessor applies its DDL idempotently via `CREATE TABLE IF NOT EXISTS`.
327        let _ = self.entities(token)?;
328        let _ = self.graph(token)?;
329        let _ = self.text(token)?;
330        // Prime DDL for every registered model's vec table.  Use vectors_for_model
331        // (not the default-model-only self.vectors()) so that custom-only runtimes
332        // (embedding_model: None but custom providers registered) work correctly.
333        for model_name in &self.registered_embedding_model_names() {
334            let _ = self.vectors_for_model(token, model_name)?;
335        }
336
337        let pool = self.backend().pool_arc();
338
339        let (summary, updated_entity) = tokio::task::spawn_blocking(move || {
340            let guard = pool.writer()?;
341            guard.transaction(|conn| {
342                merge_entity_sql(
343                    conn, ns, fts_table, vec_tables, into_id, from_id, strategy, dry_run,
344                )
345            })
346        })
347        .await
348        .map_err(|e| RuntimeError::Internal(e.to_string()))??;
349
350        // If vectors are configured, reindex into_entity (requires async embedding).
351        // FTS and vec-deletes for all registered models were already committed inside
352        // the transaction above.
353        if !dry_run && !self.registered_embedding_model_names().is_empty() {
354            self.reindex_entity(token, &updated_entity).await?;
355        }
356
357        let event_store = self.events(token)?;
358        // Mirror the wire-level strategy spelling from MergeParams so consumers
359        // can round-trip the policy string back into a request.
360        let policy_str = match strategy {
361            EntityDedupMergePolicy::PreferInto => "prefer_into",
362            EntityDedupMergePolicy::PreferFrom => "prefer_from",
363            EntityDedupMergePolicy::Union => "union",
364        };
365        let event = khive_storage::event::Event::new(
366            updated_entity.namespace.clone(),
367            "merge",
368            EventKind::EntityMerged,
369            SubstrateKind::Entity,
370            "",
371        )
372        .with_target(summary.kept_id)
373        .with_payload(serde_json::json!({
374            "into_id": summary.kept_id,
375            "from_id": summary.removed_id,
376            "policy": policy_str,
377            "edges_rewired": summary.edges_rewired,
378        }));
379        event_store.append_event(event).await.map_err(|e| {
380            RuntimeError::Internal(format!("merge_entity: event store write failed: {e}"))
381        })?;
382
383        Ok(summary)
384    }
385
386    // ---- Internal helpers ----
387
388    /// Re-upsert FTS5 document and vector(s) for the entity across all registered models.
389    ///
390    /// Uses `entity.namespace` — the authoritative namespace stored on the record — rather
391    /// than the caller-supplied `namespace` parameter. This prevents a cross-namespace
392    /// reindex from writing the search document into the wrong namespace's FTS index.
393    ///
394    /// Best-effort for vectors: if embedding or inserting for a particular model fails,
395    /// logs a warning and continues to the next model. The FTS step is fail-closed
396    /// (propagates error). Callers (update_entity, merge_entity) have already committed
397    /// the entity row, so a partial embed miss leaves a stale vector rather than
398    /// rolling back the update — acceptable because SqliteVecStore::insert is an upsert
399    /// (the prior vector stays intact on failure, keeping the record searchable).
400    pub(crate) async fn reindex_entity(
401        &self,
402        token: &NamespaceToken,
403        entity: &Entity,
404    ) -> RuntimeResult<()> {
405        // Use entity.namespace (authoritative) rather than token.namespace().as_str() (caller claim).
406        let ns = entity.namespace.clone();
407        let doc = entity_fts_document(entity);
408        let embed_body = doc.body.clone();
409        self.text(token)?.upsert_document(doc).await?;
410
411        let embed_model_names = self.registered_embedding_model_names();
412        for model_name in &embed_model_names {
413            match self
414                .embed_document_with_model(model_name, &embed_body)
415                .await
416            {
417                Ok(vector) => match self.vectors_for_model(token, model_name) {
418                    Ok(vs) => {
419                        if let Err(e) = vs
420                            .insert(
421                                entity.id,
422                                SubstrateKind::Entity,
423                                &ns,
424                                "entity.body",
425                                vec![vector],
426                            )
427                            .await
428                        {
429                            tracing::warn!(
430                                model = model_name,
431                                id = %entity.id,
432                                "reindex_entity: vector insert failed, skipping model: {e}"
433                            );
434                        }
435                    }
436                    Err(e) => {
437                        tracing::warn!(
438                            model = model_name,
439                            id = %entity.id,
440                            "reindex_entity: could not access vector store for model, skipping: {e}"
441                        );
442                    }
443                },
444                Err(e) => {
445                    tracing::warn!(
446                        model = model_name,
447                        id = %entity.id,
448                        "reindex_entity: embed failed for model, skipping: {e}"
449                    );
450                }
451            }
452        }
453
454        Ok(())
455    }
456
457    /// Remove an entity from FTS5 and vector indexes across all registered models.
458    pub(crate) async fn remove_from_indexes(
459        &self,
460        token: &NamespaceToken,
461        id: Uuid,
462    ) -> RuntimeResult<()> {
463        let ns = token.namespace().as_str().to_owned();
464        self.text(token)?.delete_document(&ns, id).await?;
465        for model_name in self.registered_embedding_model_names() {
466            self.vectors_for_model(token, &model_name)?
467                .delete(id)
468                .await?;
469        }
470        Ok(())
471    }
472
473    /// Re-upsert FTS5 document and vector(s) for the note across all registered models.
474    ///
475    /// Best-effort for vectors: mirrors reindex_entity's warn-and-continue policy.
476    pub(crate) async fn reindex_note(
477        &self,
478        token: &NamespaceToken,
479        note: &khive_storage::note::Note,
480    ) -> RuntimeResult<()> {
481        self.text_for_notes(token)?
482            .upsert_document(note_fts_document(note))
483            .await?;
484
485        let ns = note.namespace.clone();
486        let embed_model_names = self.registered_embedding_model_names();
487        for model_name in &embed_model_names {
488            match self
489                .embed_document_with_model(model_name, &note.content)
490                .await
491            {
492                Ok(vector) => match self.vectors_for_model(token, model_name) {
493                    Ok(vs) => {
494                        if let Err(e) = vs
495                            .insert(
496                                note.id,
497                                SubstrateKind::Note,
498                                &ns,
499                                "note.content",
500                                vec![vector],
501                            )
502                            .await
503                        {
504                            tracing::warn!(
505                                model = model_name,
506                                id = %note.id,
507                                "reindex_note: vector insert failed, skipping model: {e}"
508                            );
509                        }
510                    }
511                    Err(e) => {
512                        tracing::warn!(
513                            model = model_name,
514                            id = %note.id,
515                            "reindex_note: could not access vector store for model, skipping: {e}"
516                        );
517                    }
518                },
519                Err(e) => {
520                    tracing::warn!(
521                        model = model_name,
522                        id = %note.id,
523                        "reindex_note: embed failed for model, skipping: {e}"
524                    );
525                }
526            }
527        }
528        Ok(())
529    }
530
531    /// Patch-style note update.
532    pub async fn update_note(
533        &self,
534        token: &NamespaceToken,
535        id: Uuid,
536        patch: NotePatch,
537    ) -> RuntimeResult<khive_storage::note::Note> {
538        // Secret gate: scan incoming text fields and structured properties.
539        if let Some(ref content) = patch.content {
540            crate::secret_gate::check(content)?;
541        }
542        if let Some(Some(ref name)) = patch.name {
543            crate::secret_gate::check(name)?;
544        }
545        if let Some(ref props) = patch.properties {
546            crate::secret_gate::check_json(props)?;
547        }
548        let store = self.notes(token)?;
549        let mut note = store
550            .get_note(id)
551            .await?
552            .ok_or_else(|| RuntimeError::NotFound(format!("note {id}")))?;
553
554        let mut text_changed = false;
555
556        if let Some(name_patch) = patch.name {
557            text_changed |= note.name != name_patch;
558            note.name = name_patch;
559        }
560        if let Some(content) = patch.content {
561            text_changed |= note.content != content;
562            note.content = content;
563        }
564        if let Some(salience_patch) = patch.salience {
565            // Reject non-finite or out-of-range salience at the runtime boundary
566            // rather than silently clamping invalid caller input (coding-standards §608-622).
567            if let Some(s) = salience_patch {
568                if !s.is_finite() || !(0.0..=1.0).contains(&s) {
569                    return Err(crate::RuntimeError::InvalidInput(format!(
570                        "salience must be a finite value in [0.0, 1.0]; got {s}"
571                    )));
572                }
573            }
574            note.salience = salience_patch;
575        }
576        if let Some(decay_patch) = patch.decay_factor {
577            // Reject non-finite or negative decay_factor at the runtime boundary.
578            if let Some(d) = decay_patch {
579                if !d.is_finite() || d < 0.0 {
580                    return Err(crate::RuntimeError::InvalidInput(format!(
581                        "decay_factor must be a finite value >= 0.0; got {d}"
582                    )));
583                }
584            }
585            note.decay_factor = decay_patch;
586        }
587        if let Some(props) = patch.properties {
588            let (merged, _) = merge_properties(
589                &note.properties,
590                &Some(props),
591                EntityDedupMergePolicy::PreferFrom,
592            );
593            note.properties = merged;
594        }
595        if let Some(status) = patch.kind_status {
596            note.status = status;
597        }
598
599        note.updated_at = chrono::Utc::now().timestamp_micros();
600        store.upsert_note(note.clone()).await?;
601
602        if text_changed {
603            self.reindex_note(token, &note).await?;
604        }
605
606        Ok(note)
607    }
608
609    /// Merge `from_id` note into `into_id` note.
610    ///
611    /// Both notes must exist in the namespace and have the same `kind`. Content is merged
612    /// per `content_strategy`. Properties are merged per `strategy`. `from_id` is
613    /// tombstoned (status='deleted', deleted_at set). Returns a summary.
614    ///
615    /// If `dry_run` is true, computes and returns the planned summary without mutating
616    /// any rows, edges, or indexes.
617    pub async fn merge_note(
618        &self,
619        token: &NamespaceToken,
620        into_id: Uuid,
621        from_id: Uuid,
622        strategy: EntityDedupMergePolicy,
623        content_strategy: ContentMergeStrategy,
624        dry_run: bool,
625    ) -> RuntimeResult<MergeSummary> {
626        if into_id == from_id {
627            return Err(RuntimeError::InvalidInput(
628                "cannot merge a note into itself".into(),
629            ));
630        }
631        let ns = token.namespace().as_str().to_string();
632        let fts_table = "fts_notes".to_string();
633        let vec_tables: Vec<String> = self
634            .registered_embedding_model_names()
635            .iter()
636            .map(|name| format!("vec_{}", crate::config::sanitize_key(name)))
637            .collect();
638
639        let note_store = self.notes(token)?;
640        let into_note = note_store
641            .get_note(into_id)
642            .await?
643            .ok_or_else(|| RuntimeError::NotFound("not found in this namespace".into()))?;
644        Self::ensure_namespace(&into_note.namespace, &ns)?;
645
646        let from_note = note_store
647            .get_note(from_id)
648            .await?
649            .ok_or_else(|| RuntimeError::NotFound("not found in this namespace".into()))?;
650        Self::ensure_namespace(&from_note.namespace, &ns)?;
651
652        let _ = self.graph(token)?;
653        let _ = self.text_for_notes(token)?;
654        // Prime DDL for every registered model's vec table (mirrors merge_entity fix).
655        for model_name in &self.registered_embedding_model_names() {
656            let _ = self.vectors_for_model(token, model_name)?;
657        }
658
659        let pool = self.backend().pool_arc();
660        let (summary, updated_note) = tokio::task::spawn_blocking(move || {
661            let guard = pool.writer()?;
662            guard.transaction(|conn| {
663                merge_note_sql(
664                    conn,
665                    ns,
666                    fts_table,
667                    vec_tables,
668                    into_id,
669                    from_id,
670                    strategy,
671                    content_strategy,
672                    dry_run,
673                )
674            })
675        })
676        .await
677        .map_err(|e| RuntimeError::Internal(e.to_string()))??;
678
679        if !dry_run && !self.registered_embedding_model_names().is_empty() {
680            self.reindex_note(token, &updated_note).await?;
681        }
682        Ok(summary)
683    }
684}
685
686// ---------------------------------------------------------------------------
687// FTS document construction
688// ---------------------------------------------------------------------------
689
690/// Build the `TextDocument` for an entity. This is the single source of truth for
691/// entity FTS document shape; all write paths (create, update, merge, reindex, backfill)
692/// must go through this function so search parity is guaranteed.
693///
694/// Body rule: when the entity has a non-empty description, prepend the name
695/// (`"<name> <description>"`). Otherwise the body is just the name. This
696/// matches the FTS index contract: `title` and `body` are the ranked columns;
697/// `tags`, `metadata`, and `namespace` are UNINDEXED.
698///
699/// `updated_at` is taken from the entity's own timestamp so that backfill and
700/// reindex runs record the entity's actual mutation time rather than the
701/// reindex execution time.
702pub fn entity_fts_document(entity: &Entity) -> TextDocument {
703    let body = match &entity.description {
704        Some(d) if !d.is_empty() => format!("{} {}", entity.name, d),
705        _ => entity.name.clone(),
706    };
707    let updated_at =
708        chrono::DateTime::from_timestamp_micros(entity.updated_at).unwrap_or_else(chrono::Utc::now);
709    TextDocument {
710        subject_id: entity.id,
711        kind: SubstrateKind::Entity,
712        title: Some(entity.name.clone()),
713        body,
714        tags: entity.tags.clone(),
715        namespace: entity.namespace.clone(),
716        metadata: entity.properties.clone(),
717        updated_at,
718    }
719}
720
721/// Build the `TextDocument` for a note. This is the single source of truth for
722/// note FTS document shape; all write paths (create, update, reindex) must go
723/// through this function so recall parity is guaranteed. Changes here apply to
724/// every caller automatically.
725///
726/// Body rule: when the note has a `name`, prepend it to the content
727/// (`"<name> <content>"`). This matches the FTS index contract: title and body
728/// both contribute to ranking, and the name is the most salient signal.
729///
730/// `updated_at` is taken from the note's own timestamp (not `Utc::now()`) so
731/// that backfill and reindex runs record the note's actual mutation time rather
732/// than the reindex execution time.
733pub fn note_fts_document(note: &Note) -> TextDocument {
734    let body = match &note.name {
735        Some(n) => format!("{n} {}", note.content),
736        None => note.content.clone(),
737    };
738    let updated_at =
739        chrono::DateTime::from_timestamp_micros(note.updated_at).unwrap_or_else(chrono::Utc::now);
740    TextDocument {
741        subject_id: note.id,
742        kind: SubstrateKind::Note,
743        title: note.name.clone(),
744        body,
745        tags: vec![],
746        namespace: note.namespace.clone(),
747        metadata: note.properties.clone(),
748        updated_at,
749    }
750}
751
752/// SQL-bind–ready scalars derived from [`note_fts_document`].
753///
754/// Used by `merge_note_sql` to guarantee that the raw SQL FTS INSERT stores
755/// exactly what [`Fts5TextSearch::upsert_document`] would write, preventing
756/// null/empty-string divergence on the `title` column for nameless notes.
757pub(crate) struct NoteFtsScalars {
758    /// Empty string when `note.name` is `None` — matches the `unwrap_or("")` in
759    /// `Fts5TextSearch::upsert_document`.
760    pub title: String,
761    pub body: String,
762    /// Always the JSON array `"[]"`.
763    pub tags: String,
764    /// Serialised `note.properties`, or `None` when properties are absent.
765    pub metadata: Option<String>,
766    /// `note.updated_at` converted to `DateTime<Utc>` timestamp_micros.
767    pub updated_at_micros: i64,
768}
769
770/// Derive [`NoteFtsScalars`] from a [`Note`].
771///
772/// All values match the encoding that [`Fts5TextSearch::upsert_document`]
773/// applies when given the output of [`note_fts_document`].
774pub(crate) fn note_fts_scalars(note: &Note) -> NoteFtsScalars {
775    let doc = note_fts_document(note);
776    NoteFtsScalars {
777        title: doc.title.unwrap_or_default(),
778        body: doc.body,
779        tags: "[]".to_string(),
780        metadata: doc
781            .metadata
782            .as_ref()
783            .map(|v| serde_json::to_string(v).unwrap_or_default()),
784        updated_at_micros: doc.updated_at.timestamp_micros(),
785    }
786}
787
788// ---------------------------------------------------------------------------
789// Transactional merge SQL helpers
790// ---------------------------------------------------------------------------
791
792/// Read one entity row by ID within a namespace, returning `SqliteError` on missing/wrong-ns.
793fn read_merge_entity(
794    conn: &rusqlite::Connection,
795    id: Uuid,
796    namespace: &str,
797) -> Result<Entity, SqliteError> {
798    let id_str = id.to_string();
799    let mut stmt = conn.prepare(
800        "SELECT id, namespace, kind, entity_type, name, description, properties, tags, \
801         created_at, updated_at, deleted_at, merged_into, merge_event_id \
802         FROM entities WHERE id = ?1 AND deleted_at IS NULL",
803    )?;
804    let mut rows = stmt.query(rusqlite::params![id_str])?;
805    let row = rows
806        .next()?
807        .ok_or_else(|| SqliteError::InvalidData(format!("entity {id} not found")))?;
808
809    let id_s: String = row.get(0)?;
810    let ns: String = row.get(1)?;
811    let kind: String = row.get(2)?;
812    let entity_type: Option<String> = row.get(3)?;
813    let name: String = row.get(4)?;
814    let description: Option<String> = row.get(5)?;
815    let properties_str: Option<String> = row.get(6)?;
816    let tags_str: String = row.get(7)?;
817    let created_at: i64 = row.get(8)?;
818    let updated_at: i64 = row.get(9)?;
819    let deleted_at: Option<i64> = row.get(10)?;
820    let merged_into_str: Option<String> = row.get(11)?;
821    let merge_event_id_str: Option<String> = row.get(12)?;
822
823    if ns != namespace {
824        return Err(SqliteError::InvalidData(format!(
825            "entity {id} belongs to namespace '{ns}', not '{namespace}'"
826        )));
827    }
828
829    let entity_id = Uuid::parse_str(&id_s).map_err(|e| SqliteError::InvalidData(e.to_string()))?;
830    let properties: Option<Value> = properties_str
831        .map(|s| {
832            serde_json::from_str::<Value>(&s).map_err(|e| SqliteError::InvalidData(e.to_string()))
833        })
834        .transpose()?;
835    let tags: Vec<String> =
836        serde_json::from_str(&tags_str).map_err(|e| SqliteError::InvalidData(e.to_string()))?;
837    let merged_into = merged_into_str
838        .as_deref()
839        .map(Uuid::parse_str)
840        .transpose()
841        .map_err(|e| SqliteError::InvalidData(e.to_string()))?;
842    let merge_event_id = merge_event_id_str
843        .as_deref()
844        .map(Uuid::parse_str)
845        .transpose()
846        .map_err(|e| SqliteError::InvalidData(e.to_string()))?;
847
848    Ok(Entity {
849        id: entity_id,
850        namespace: ns,
851        kind,
852        entity_type,
853        name,
854        description,
855        properties,
856        tags,
857        created_at,
858        updated_at,
859        deleted_at,
860        merged_into,
861        merge_event_id,
862    })
863}
864
865/// All merge SQL on one connection inside an already-open `BEGIN IMMEDIATE` transaction.
866///
867/// Reads both entities, rewires/drops incident edges, merges entity fields, updates FTS,
868/// deletes the `from` vec entry (if `vec_table` is Some), and tombstones `from` with merge
869/// provenance.  Returns the updated `into` entity so the caller can do the async vec re-insert.
870///
871/// When `dry_run` is true, all reads and computations are performed but no writes are issued.
872// REASON: merge requires both entity IDs, the namespace, FTS and vec table names, merge
873// policy, and dry-run flag — all are load-bearing; reducing to a struct would obscure
874// the sync/async boundary split that keeps this function off the async runtime.
875#[allow(clippy::too_many_arguments)]
876fn merge_entity_sql(
877    conn: &rusqlite::Connection,
878    namespace: String,
879    fts_table: String,
880    vec_tables: Vec<String>,
881    into_id: Uuid,
882    from_id: Uuid,
883    strategy: EntityDedupMergePolicy,
884    dry_run: bool,
885) -> Result<(MergeSummary, Entity), SqliteError> {
886    let into_entity = read_merge_entity(conn, into_id, &namespace)?;
887    let from_entity = read_merge_entity(conn, from_id, &namespace)?;
888
889    // --- Collect edges incident to from_id ---
890    let parse_id =
891        |s: String| Uuid::parse_str(&s).map_err(|e| SqliteError::InvalidData(e.to_string()));
892
893    let from_str = from_id.to_string();
894
895    let mut outbound: Vec<EdgeRow> = Vec::new();
896    {
897        let mut stmt = conn.prepare(
898            "SELECT id, source_id, target_id, relation, weight, created_at, \
899                    updated_at, deleted_at, target_backend, metadata \
900             FROM graph_edges WHERE namespace = ?1 AND source_id = ?2",
901        )?;
902        let mut rows = stmt.query(rusqlite::params![&namespace, &from_str])?;
903        while let Some(row) = rows.next()? {
904            outbound.push(EdgeRow {
905                id: parse_id(row.get(0)?)?,
906                source_id: parse_id(row.get(1)?)?,
907                target_id: parse_id(row.get(2)?)?,
908                relation: row.get(3)?,
909                weight: row.get(4)?,
910                created_at: row.get(5)?,
911                updated_at: row.get(6)?,
912                deleted_at: row.get(7)?,
913                target_backend: row.get(8)?,
914                metadata: row.get(9)?,
915            });
916        }
917    }
918
919    let mut inbound: Vec<EdgeRow> = Vec::new();
920    {
921        let mut stmt = conn.prepare(
922            "SELECT id, source_id, target_id, relation, weight, created_at, \
923                    updated_at, deleted_at, target_backend, metadata \
924             FROM graph_edges WHERE namespace = ?1 AND target_id = ?2",
925        )?;
926        let mut rows = stmt.query(rusqlite::params![&namespace, &from_str])?;
927        while let Some(row) = rows.next()? {
928            inbound.push(EdgeRow {
929                id: parse_id(row.get(0)?)?,
930                source_id: parse_id(row.get(1)?)?,
931                target_id: parse_id(row.get(2)?)?,
932                relation: row.get(3)?,
933                weight: row.get(4)?,
934                created_at: row.get(5)?,
935                updated_at: row.get(6)?,
936                deleted_at: row.get(7)?,
937                target_backend: row.get(8)?,
938                metadata: row.get(9)?,
939            });
940        }
941    }
942
943    // Deduplicate by edge ID (a self-edge from_id→from_id appears in both lists).
944    let mut seen: HashSet<Uuid> = HashSet::new();
945    let mut all_edges: Vec<EdgeRow> = Vec::new();
946    for edge in outbound.into_iter().chain(inbound) {
947        if seen.insert(edge.id) {
948            all_edges.push(edge);
949        }
950    }
951
952    // --- Merge entity fields ---
953    let (merged_props, properties_merged) =
954        merge_properties(&into_entity.properties, &from_entity.properties, strategy);
955    let merged_name = merge_string_field(&into_entity.name, &from_entity.name, strategy);
956    let merged_description =
957        merge_option_string_field(&into_entity.description, &from_entity.description, strategy);
958    let (merged_tags, tags_unioned) = union_tags(&into_entity.tags, &from_entity.tags);
959
960    let now = chrono::Utc::now().timestamp_micros();
961    let into_str = into_id.to_string();
962    let props_str = merged_props
963        .as_ref()
964        .map(|v| serde_json::to_string(v).unwrap_or_default());
965    let tags_json = serde_json::to_string(&merged_tags).unwrap_or_else(|_| "[]".to_string());
966
967    // --- Rewire edges ---
968    let mut edges_rewired = 0usize;
969    if !dry_run {
970        for edge in all_edges {
971            let raw_src = if edge.source_id == from_id {
972                into_id
973            } else {
974                edge.source_id
975            };
976            let raw_tgt = if edge.target_id == from_id {
977                into_id
978            } else {
979                edge.target_id
980            };
981            // Symmetric relations must be stored with source_uuid < target_uuid.
982            // Apply canonicalization so the conflict check and UPDATE both use the canonical form.
983            let (new_src, new_tgt) = match edge.relation.parse::<EdgeRelation>() {
984                Ok(rel) => canonical_edge_endpoints(rel, raw_src, raw_tgt),
985                Err(_) => (raw_src, raw_tgt),
986            };
987
988            if new_src == new_tgt {
989                conn.execute(
990                    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2",
991                    rusqlite::params![&namespace, edge.id.to_string()],
992                )?;
993                continue;
994            }
995
996            let now_ts = chrono::Utc::now().timestamp();
997            // H3 fix: preserve the original edge ID by updating
998            // source_id/target_id in-place when no conflict exists.
999            //
1000            // Two-step approach to handle all cases while keeping the original ID:
1001            //   (a) No conflict (new triple): UPDATE source_id/target_id in-place.
1002            //       The edge retains its original UUID — callers can still get() it
1003            //       by the ID they received from link().
1004            //   (b) Conflict: into_id already has an edge with this (source,target,
1005            //       relation). Delete the from-edge (it is superseded) and UPDATE
1006            //       the existing into-edge to refresh weight/metadata/deleted_at.
1007            //       The surviving edge is the into-entity's original edge (correct).
1008            //
1009            // Check for a conflict: does into_id already have this natural key?
1010            let conflict_id: Option<String> = {
1011                let conflict_src = new_src.to_string();
1012                let conflict_tgt = new_tgt.to_string();
1013                conn.query_row(
1014                    "SELECT id FROM graph_edges \
1015                     WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 \
1016                     AND relation = ?4 AND id != ?5",
1017                    rusqlite::params![
1018                        &namespace,
1019                        &conflict_src,
1020                        &conflict_tgt,
1021                        &edge.relation,
1022                        edge.id.to_string(),
1023                    ],
1024                    |row| row.get(0),
1025                )
1026                .optional()
1027                .map_err(SqliteError::Rusqlite)?
1028            };
1029
1030            let changed = if let Some(existing_id) = conflict_id {
1031                // Case (b): a live or soft-deleted row already owns this natural key.
1032                // Delete the from-edge and refresh the existing row.
1033                conn.execute(
1034                    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2",
1035                    rusqlite::params![&namespace, edge.id.to_string()],
1036                )?;
1037                conn.execute(
1038                    "UPDATE graph_edges SET \
1039                     weight = ?1, updated_at = ?2, deleted_at = NULL, \
1040                     target_backend = ?3, metadata = ?4 \
1041                     WHERE namespace = ?5 AND id = ?6",
1042                    rusqlite::params![
1043                        edge.weight,
1044                        now_ts,
1045                        edge.target_backend,
1046                        edge.metadata,
1047                        &namespace,
1048                        &existing_id,
1049                    ],
1050                )?
1051            } else {
1052                // Case (a): no conflict — update source_id/target_id in-place,
1053                // preserving the original edge ID for callers.
1054                conn.execute(
1055                    "UPDATE graph_edges SET \
1056                     source_id = ?1, target_id = ?2, updated_at = ?3 \
1057                     WHERE namespace = ?4 AND id = ?5",
1058                    rusqlite::params![
1059                        new_src.to_string(),
1060                        new_tgt.to_string(),
1061                        now_ts,
1062                        &namespace,
1063                        edge.id.to_string(),
1064                    ],
1065                )?
1066            };
1067            if changed > 0 {
1068                edges_rewired += 1;
1069            }
1070        }
1071
1072        // --- Upsert merged entity ---
1073        conn.execute(
1074            "INSERT OR REPLACE INTO entities \
1075             (id, namespace, kind, name, description, properties, tags, \
1076              created_at, updated_at, deleted_at, merged_into, merge_event_id) \
1077             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
1078            rusqlite::params![
1079                &into_str,
1080                &namespace,
1081                &into_entity.kind,
1082                &merged_name,
1083                &merged_description,
1084                &props_str,
1085                &tags_json,
1086                into_entity.created_at,
1087                now,
1088                into_entity.deleted_at,
1089                Option::<String>::None,
1090                Option::<String>::None,
1091            ],
1092        )?;
1093
1094        // --- Reindex into_id in FTS (delete existing, insert updated) ---
1095        // Body formula mirrors entity_fts_document (the canonical constructor) —
1096        // this path is sync/spawn_blocking so it cannot call entity_fts_document
1097        // directly, but must stay field-identical.
1098        let fts_body = match &merged_description {
1099            Some(d) if !d.is_empty() => format!("{} {}", merged_name, d),
1100            _ => merged_name.clone(),
1101        };
1102        let kind_str = SubstrateKind::Entity.to_string();
1103
1104        conn.execute(
1105            &format!(
1106                "DELETE FROM {} WHERE namespace = ?1 AND subject_id = ?2",
1107                fts_table
1108            ),
1109            rusqlite::params![&namespace, &into_str],
1110        )?;
1111        conn.execute(
1112            &format!(
1113                "INSERT INTO {} \
1114                 (subject_id, kind, title, body, tags, namespace, metadata, updated_at) \
1115                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
1116                fts_table
1117            ),
1118            rusqlite::params![
1119                &into_str,
1120                &kind_str,
1121                &merged_name,
1122                &fts_body,
1123                &tags_json,
1124                &namespace,
1125                &props_str,
1126                now,
1127            ],
1128        )?;
1129
1130        // --- Delete from_id from FTS ---
1131        conn.execute(
1132            &format!(
1133                "DELETE FROM {} WHERE namespace = ?1 AND subject_id = ?2",
1134                fts_table
1135            ),
1136            rusqlite::params![&namespace, &from_str],
1137        )?;
1138
1139        // --- Delete from_id from all registered model vector tables ---
1140        for vec_tbl in &vec_tables {
1141            conn.execute(
1142                &format!(
1143                    "DELETE FROM {} WHERE subject_id = ?1 AND namespace = ?2",
1144                    vec_tbl
1145                ),
1146                rusqlite::params![&from_str, &namespace],
1147            )?;
1148        }
1149
1150        // --- Tombstone from entity (soft-delete with provenance) ---
1151        let merge_event_id = Uuid::new_v4();
1152        conn.execute(
1153            "UPDATE entities \
1154             SET deleted_at = ?1, merged_into = ?2, merge_event_id = ?3, updated_at = ?1 \
1155             WHERE namespace = ?4 AND id = ?5 AND deleted_at IS NULL",
1156            rusqlite::params![
1157                now,
1158                into_str,
1159                merge_event_id.to_string(),
1160                &namespace,
1161                &from_str,
1162            ],
1163        )?;
1164    }
1165
1166    let updated_entity = Entity {
1167        id: into_id,
1168        namespace,
1169        kind: into_entity.kind,
1170        entity_type: into_entity.entity_type,
1171        name: merged_name,
1172        description: merged_description,
1173        properties: merged_props,
1174        tags: merged_tags,
1175        created_at: into_entity.created_at,
1176        updated_at: now,
1177        deleted_at: into_entity.deleted_at,
1178        merged_into: None,
1179        merge_event_id: None,
1180    };
1181
1182    Ok((
1183        MergeSummary {
1184            kept_id: into_id,
1185            removed_id: from_id,
1186            edges_rewired,
1187            properties_merged,
1188            tags_unioned,
1189            content_appended: false,
1190            dry_run,
1191        },
1192        updated_entity,
1193    ))
1194}
1195
1196// ---------------------------------------------------------------------------
1197// Note merge SQL helpers
1198// ---------------------------------------------------------------------------
1199
1200/// Read one note row by ID within a namespace, returning `SqliteError` on missing/wrong-ns.
1201fn read_merge_note(
1202    conn: &rusqlite::Connection,
1203    id: Uuid,
1204    namespace: &str,
1205) -> Result<khive_storage::note::Note, SqliteError> {
1206    use khive_storage::note::Note;
1207    let id_str = id.to_string();
1208    let mut stmt = conn.prepare(
1209        "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
1210         expires_at, properties, created_at, updated_at, deleted_at \
1211         FROM notes WHERE id = ?1 AND deleted_at IS NULL",
1212    )?;
1213    let mut rows = stmt.query(rusqlite::params![id_str])?;
1214    let row = rows
1215        .next()?
1216        .ok_or_else(|| SqliteError::InvalidData(format!("note {id} not found")))?;
1217
1218    let id_s: String = row.get(0)?;
1219    let ns: String = row.get(1)?;
1220    let kind: String = row.get(2)?;
1221    let status: String = row.get(3)?;
1222    let name: Option<String> = row.get(4)?;
1223    let content: String = row.get(5)?;
1224    let salience: Option<f64> = row.get(6)?;
1225    let decay_factor: Option<f64> = row.get(7)?;
1226    let expires_at: Option<i64> = row.get(8)?;
1227    let properties_str: Option<String> = row.get(9)?;
1228    let created_at: i64 = row.get(10)?;
1229    let updated_at: i64 = row.get(11)?;
1230    let deleted_at: Option<i64> = row.get(12)?;
1231
1232    if ns != namespace {
1233        return Err(SqliteError::InvalidData(format!(
1234            "note {id} belongs to namespace '{ns}', not '{namespace}'"
1235        )));
1236    }
1237
1238    let note_id = Uuid::parse_str(&id_s).map_err(|e| SqliteError::InvalidData(e.to_string()))?;
1239    let properties: Option<serde_json::Value> = properties_str
1240        .map(|s| serde_json::from_str(&s).map_err(|e| SqliteError::InvalidData(e.to_string())))
1241        .transpose()?;
1242
1243    Ok(Note {
1244        id: note_id,
1245        namespace: ns,
1246        kind,
1247        status,
1248        name,
1249        content,
1250        salience,
1251        decay_factor,
1252        expires_at,
1253        properties,
1254        created_at,
1255        updated_at,
1256        deleted_at,
1257    })
1258}
1259
1260fn max_option_f64(a: Option<f64>, b: Option<f64>) -> Option<f64> {
1261    match (a, b) {
1262        (Some(x), Some(y)) => Some(x.max(y)),
1263        (Some(x), None) => Some(x),
1264        (None, Some(y)) => Some(y),
1265        (None, None) => None,
1266    }
1267}
1268
1269fn append_merge_history(props: Option<Value>, entry: Value) -> Result<Option<Value>, SqliteError> {
1270    use serde_json::{json, Map};
1271    let mut obj: Map<String, Value> = match props {
1272        Some(Value::Object(m)) => m,
1273        Some(other) => {
1274            let mut m = Map::new();
1275            m.insert("_value".into(), other);
1276            m
1277        }
1278        None => Map::new(),
1279    };
1280    let history = obj
1281        .entry("_merge_history".to_string())
1282        .or_insert_with(|| json!([]));
1283    if let Value::Array(arr) = history {
1284        arr.push(entry);
1285    }
1286    Ok(Some(Value::Object(obj)))
1287}
1288
1289/// All note merge SQL on one connection inside a `BEGIN IMMEDIATE` transaction.
1290///
1291/// Reads both notes (must have same `kind`), rewires/drops incident edges, merges content
1292/// per `content_strategy`, tombstones `from`. Returns the updated `into` note for async
1293/// re-embedding.
1294///
1295/// When `dry_run` is true, all reads and computations are performed but no writes are issued.
1296// REASON: note merge additionally requires a content_strategy parameter versus entity merge;
1297// same sync/async boundary rationale as merge_entity_sql applies here.
1298#[allow(clippy::too_many_arguments)]
1299fn merge_note_sql(
1300    conn: &rusqlite::Connection,
1301    namespace: String,
1302    fts_table: String,
1303    vec_tables: Vec<String>,
1304    into_id: Uuid,
1305    from_id: Uuid,
1306    strategy: EntityDedupMergePolicy,
1307    content_strategy: ContentMergeStrategy,
1308    dry_run: bool,
1309) -> Result<(MergeSummary, khive_storage::note::Note), SqliteError> {
1310    let into_note = read_merge_note(conn, into_id, &namespace)?;
1311    let from_note = read_merge_note(conn, from_id, &namespace)?;
1312
1313    if into_note.kind != from_note.kind {
1314        return Err(SqliteError::InvalidData(format!(
1315            "cannot merge notes of different kinds: {} vs {}",
1316            into_note.kind, from_note.kind
1317        )));
1318    }
1319
1320    let now = chrono::Utc::now().timestamp_micros();
1321    let into_str = into_id.to_string();
1322    let from_str = from_id.to_string();
1323
1324    // Collect edges incident to from_id.
1325    let parse_id =
1326        |s: String| Uuid::parse_str(&s).map_err(|e| SqliteError::InvalidData(e.to_string()));
1327
1328    let mut outbound: Vec<EdgeRow> = Vec::new();
1329    {
1330        let mut stmt = conn.prepare(
1331            "SELECT id, source_id, target_id, relation, weight, created_at, updated_at, deleted_at, target_backend, metadata \
1332             FROM graph_edges WHERE namespace = ?1 AND source_id = ?2",
1333        )?;
1334        let mut rows = stmt.query(rusqlite::params![&namespace, &from_str])?;
1335        while let Some(row) = rows.next()? {
1336            outbound.push(EdgeRow {
1337                id: parse_id(row.get(0)?)?,
1338                source_id: parse_id(row.get(1)?)?,
1339                target_id: parse_id(row.get(2)?)?,
1340                relation: row.get(3)?,
1341                weight: row.get(4)?,
1342                created_at: row.get(5)?,
1343                updated_at: row.get(6)?,
1344                deleted_at: row.get(7)?,
1345                target_backend: row.get(8)?,
1346                metadata: row.get(9)?,
1347            });
1348        }
1349    }
1350    let mut inbound: Vec<EdgeRow> = Vec::new();
1351    {
1352        let mut stmt = conn.prepare(
1353            "SELECT id, source_id, target_id, relation, weight, created_at, updated_at, deleted_at, target_backend, metadata \
1354             FROM graph_edges WHERE namespace = ?1 AND target_id = ?2",
1355        )?;
1356        let mut rows = stmt.query(rusqlite::params![&namespace, &from_str])?;
1357        while let Some(row) = rows.next()? {
1358            inbound.push(EdgeRow {
1359                id: parse_id(row.get(0)?)?,
1360                source_id: parse_id(row.get(1)?)?,
1361                target_id: parse_id(row.get(2)?)?,
1362                relation: row.get(3)?,
1363                weight: row.get(4)?,
1364                created_at: row.get(5)?,
1365                updated_at: row.get(6)?,
1366                deleted_at: row.get(7)?,
1367                target_backend: row.get(8)?,
1368                metadata: row.get(9)?,
1369            });
1370        }
1371    }
1372    let mut seen: HashSet<Uuid> = HashSet::new();
1373    let mut all_edges: Vec<EdgeRow> = Vec::new();
1374    for edge in outbound.into_iter().chain(inbound) {
1375        if seen.insert(edge.id) {
1376            all_edges.push(edge);
1377        }
1378    }
1379
1380    // Merge note fields.
1381    let (merged_content, content_appended) = match content_strategy {
1382        ContentMergeStrategy::Append => {
1383            if from_note.content.is_empty() {
1384                (into_note.content.clone(), false)
1385            } else {
1386                (
1387                    format!("{}\n\n---\n\n{}", into_note.content, from_note.content),
1388                    true,
1389                )
1390            }
1391        }
1392        ContentMergeStrategy::PreferInto => (into_note.content.clone(), false),
1393        ContentMergeStrategy::PreferFrom => (from_note.content.clone(), false),
1394    };
1395
1396    let merged_name = match strategy {
1397        EntityDedupMergePolicy::PreferFrom => from_note.name.clone().or(into_note.name.clone()),
1398        _ => into_note.name.clone().or(from_note.name.clone()),
1399    };
1400
1401    let (merged_props, properties_merged) =
1402        merge_properties(&into_note.properties, &from_note.properties, strategy);
1403
1404    // Append merge history to properties.
1405    let merge_history_entry = serde_json::json!({
1406        "merged_from": from_id.to_string(),
1407        "merged_at": now,
1408        "strategy": format!("{:?}", strategy),
1409        "content_strategy": format!("{:?}", content_strategy),
1410    });
1411    let merged_props = append_merge_history(merged_props, merge_history_entry)?;
1412
1413    let merged_salience = max_option_f64(into_note.salience, from_note.salience);
1414    let merged_expires_at = match (into_note.expires_at, from_note.expires_at) {
1415        (Some(a), Some(b)) => Some(a.max(b)),
1416        (Some(a), None) => Some(a),
1417        (None, Some(b)) => Some(b),
1418        (None, None) => None,
1419    };
1420
1421    let props_str = merged_props
1422        .as_ref()
1423        .map(|v| serde_json::to_string(v).unwrap_or_default());
1424
1425    let mut edges_rewired = 0usize;
1426    if !dry_run {
1427        // Rewire and upsert.
1428        for edge in all_edges {
1429            let raw_src = if edge.source_id == from_id {
1430                into_id
1431            } else {
1432                edge.source_id
1433            };
1434            let raw_tgt = if edge.target_id == from_id {
1435                into_id
1436            } else {
1437                edge.target_id
1438            };
1439            // Canonicalize symmetric relations before conflict check + UPDATE.
1440            let (new_src, new_tgt) = match edge.relation.parse::<EdgeRelation>() {
1441                Ok(rel) => canonical_edge_endpoints(rel, raw_src, raw_tgt),
1442                Err(_) => (raw_src, raw_tgt),
1443            };
1444            if new_src == new_tgt {
1445                conn.execute(
1446                    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2",
1447                    rusqlite::params![&namespace, edge.id.to_string()],
1448                )?;
1449                continue;
1450            }
1451            let now_ts = chrono::Utc::now().timestamp();
1452            // Same two-step approach as entity merge rewire: preserve original edge ID
1453            // when no conflict, merge into existing row when conflict exists.
1454            let conflict_id: Option<String> = {
1455                let conflict_src = new_src.to_string();
1456                let conflict_tgt = new_tgt.to_string();
1457                conn.query_row(
1458                    "SELECT id FROM graph_edges \
1459                     WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 \
1460                     AND relation = ?4 AND id != ?5",
1461                    rusqlite::params![
1462                        &namespace,
1463                        &conflict_src,
1464                        &conflict_tgt,
1465                        &edge.relation,
1466                        edge.id.to_string(),
1467                    ],
1468                    |row| row.get(0),
1469                )
1470                .optional()
1471                .map_err(SqliteError::Rusqlite)?
1472            };
1473
1474            let changed = if let Some(existing_id) = conflict_id {
1475                conn.execute(
1476                    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2",
1477                    rusqlite::params![&namespace, edge.id.to_string()],
1478                )?;
1479                conn.execute(
1480                    "UPDATE graph_edges SET \
1481                     weight = ?1, updated_at = ?2, deleted_at = NULL, \
1482                     target_backend = ?3, metadata = ?4 \
1483                     WHERE namespace = ?5 AND id = ?6",
1484                    rusqlite::params![
1485                        edge.weight,
1486                        now_ts,
1487                        edge.target_backend,
1488                        edge.metadata,
1489                        &namespace,
1490                        &existing_id,
1491                    ],
1492                )?
1493            } else {
1494                conn.execute(
1495                    "UPDATE graph_edges SET \
1496                     source_id = ?1, target_id = ?2, updated_at = ?3 \
1497                     WHERE namespace = ?4 AND id = ?5",
1498                    rusqlite::params![
1499                        new_src.to_string(),
1500                        new_tgt.to_string(),
1501                        now_ts,
1502                        &namespace,
1503                        edge.id.to_string(),
1504                    ],
1505                )?
1506            };
1507            if changed > 0 {
1508                edges_rewired += 1;
1509            }
1510        }
1511
1512        // Upsert merged into-note.
1513        conn.execute(
1514            "INSERT OR REPLACE INTO notes \
1515             (id, namespace, kind, status, name, content, salience, decay_factor, \
1516              expires_at, properties, created_at, updated_at, deleted_at) \
1517             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)",
1518            rusqlite::params![
1519                &into_str,
1520                &namespace,
1521                &into_note.kind,
1522                &into_note.status,
1523                &merged_name,
1524                &merged_content,
1525                merged_salience,
1526                into_note.decay_factor,
1527                merged_expires_at,
1528                &props_str,
1529                into_note.created_at,
1530                now,
1531                into_note.deleted_at,
1532            ],
1533        )?;
1534
1535        // Update FTS for into-note.
1536        conn.execute(
1537            &format!(
1538                "DELETE FROM {} WHERE namespace = ?1 AND subject_id = ?2",
1539                fts_table
1540            ),
1541            rusqlite::params![&namespace, &into_str],
1542        )?;
1543        // Derive FTS scalars through the shared constructor so this raw SQL
1544        // path is field-identical to TextSearch::upsert_document.  Critically,
1545        // `title` is an empty string (not SQL NULL) for nameless notes —
1546        // matching the unwrap_or("") in Fts5TextSearch::upsert_document and
1547        // allowing get_document to round-trip None ↔ "" correctly.
1548        let fts_merged = {
1549            let mut merged_note = Note::new(&namespace, &*into_note.kind, &*merged_content);
1550            merged_note.id = into_id;
1551            merged_note.name = merged_name.clone();
1552            merged_note.properties = merged_props.clone();
1553            merged_note.updated_at = now;
1554            note_fts_scalars(&merged_note)
1555        };
1556        conn.execute(
1557            &format!(
1558                "INSERT INTO {} \
1559                 (subject_id, kind, title, body, tags, namespace, metadata, updated_at) \
1560                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
1561                fts_table
1562            ),
1563            rusqlite::params![
1564                &into_str,
1565                SubstrateKind::Note.to_string(),
1566                &fts_merged.title,
1567                &fts_merged.body,
1568                &fts_merged.tags,
1569                &namespace,
1570                &fts_merged.metadata,
1571                fts_merged.updated_at_micros,
1572            ],
1573        )?;
1574
1575        // Delete from-note from FTS.
1576        conn.execute(
1577            &format!(
1578                "DELETE FROM {} WHERE namespace = ?1 AND subject_id = ?2",
1579                fts_table
1580            ),
1581            rusqlite::params![&namespace, &from_str],
1582        )?;
1583
1584        // Delete from-note from all registered model vector tables.
1585        for vec_tbl in &vec_tables {
1586            conn.execute(
1587                &format!(
1588                    "DELETE FROM {} WHERE subject_id = ?1 AND namespace = ?2",
1589                    vec_tbl
1590                ),
1591                rusqlite::params![&from_str, &namespace],
1592            )?;
1593        }
1594
1595        // Tombstone the from-note.
1596        conn.execute(
1597            "UPDATE notes SET status = 'deleted', deleted_at = ?1, updated_at = ?1 \
1598             WHERE namespace = ?2 AND id = ?3 AND deleted_at IS NULL",
1599            rusqlite::params![now, &namespace, &from_str],
1600        )?;
1601    }
1602
1603    let updated_note = khive_storage::note::Note {
1604        id: into_id,
1605        namespace: namespace.clone(),
1606        kind: into_note.kind.clone(),
1607        status: into_note.status.clone(),
1608        name: merged_name,
1609        content: merged_content,
1610        salience: merged_salience,
1611        decay_factor: into_note.decay_factor,
1612        expires_at: merged_expires_at,
1613        properties: merged_props,
1614        created_at: into_note.created_at,
1615        updated_at: now,
1616        deleted_at: into_note.deleted_at,
1617    };
1618
1619    Ok((
1620        MergeSummary {
1621            kept_id: into_id,
1622            removed_id: from_id,
1623            edges_rewired,
1624            properties_merged,
1625            tags_unioned: 0,
1626            content_appended,
1627            dry_run,
1628        },
1629        updated_note,
1630    ))
1631}
1632
1633// ---------------------------------------------------------------------------
1634// Merge helpers (pure functions — easier to unit test)
1635// ---------------------------------------------------------------------------
1636
1637fn merge_string_field(into: &str, from: &str, strategy: EntityDedupMergePolicy) -> String {
1638    match strategy {
1639        EntityDedupMergePolicy::PreferInto | EntityDedupMergePolicy::Union => into.to_string(),
1640        EntityDedupMergePolicy::PreferFrom => from.to_string(),
1641    }
1642}
1643
1644fn merge_option_string_field(
1645    into: &Option<String>,
1646    from: &Option<String>,
1647    strategy: EntityDedupMergePolicy,
1648) -> Option<String> {
1649    match strategy {
1650        EntityDedupMergePolicy::PreferInto => {
1651            if into.is_some() {
1652                into.clone()
1653            } else {
1654                from.clone()
1655            }
1656        }
1657        EntityDedupMergePolicy::PreferFrom => {
1658            if from.is_some() {
1659                from.clone()
1660            } else {
1661                into.clone()
1662            }
1663        }
1664        EntityDedupMergePolicy::Union => {
1665            // Keep into's description; if empty, append from's.
1666            match (into, from) {
1667                (Some(a), _) if !a.is_empty() => Some(a.clone()),
1668                (_, Some(b)) => Some(b.clone()),
1669                _ => None,
1670            }
1671        }
1672    }
1673}
1674
1675/// Merge two property objects. Returns (merged, count_of_fields_from_from_that_were_added).
1676fn merge_properties(
1677    into: &Option<Value>,
1678    from: &Option<Value>,
1679    strategy: EntityDedupMergePolicy,
1680) -> (Option<Value>, usize) {
1681    match (into, from) {
1682        (None, None) => (None, 0),
1683        (Some(a), None) => (Some(a.clone()), 0),
1684        (None, Some(b)) => {
1685            let count = if let Value::Object(m) = b { m.len() } else { 1 };
1686            (Some(b.clone()), count)
1687        }
1688        (Some(into_val), Some(from_val)) => {
1689            let (merged, added) = merge_json(into_val, from_val, strategy);
1690            (Some(merged), added)
1691        }
1692    }
1693}
1694
1695/// Deep-merge two JSON values per strategy. Returns (merged, keys_contributed_by_from).
1696fn merge_json(into: &Value, from: &Value, strategy: EntityDedupMergePolicy) -> (Value, usize) {
1697    match (into, from, strategy) {
1698        (Value::Object(a), Value::Object(b), EntityDedupMergePolicy::Union) => {
1699            let mut result = a.clone();
1700            let mut added = 0usize;
1701            for (k, v_from) in b {
1702                if let Some(v_into) = a.get(k) {
1703                    let (merged, sub_added) =
1704                        merge_json(v_into, v_from, EntityDedupMergePolicy::Union);
1705                    result.insert(k.clone(), merged);
1706                    added += sub_added;
1707                } else {
1708                    result.insert(k.clone(), v_from.clone());
1709                    added += 1;
1710                }
1711            }
1712            (Value::Object(result), added)
1713        }
1714        (Value::Object(a), Value::Object(b), EntityDedupMergePolicy::PreferInto) => {
1715            let mut result = a.clone();
1716            let mut added = 0usize;
1717            for (k, v) in b {
1718                if !a.contains_key(k) {
1719                    result.insert(k.clone(), v.clone());
1720                    added += 1;
1721                }
1722            }
1723            (Value::Object(result), added)
1724        }
1725        (Value::Object(a), Value::Object(b), EntityDedupMergePolicy::PreferFrom) => {
1726            let mut result = a.clone();
1727            let mut added = 0usize;
1728            for (k, v) in b {
1729                result.insert(k.clone(), v.clone());
1730                if !a.contains_key(k) {
1731                    added += 1;
1732                }
1733            }
1734            (Value::Object(result), added)
1735        }
1736        // Non-object scalars: apply strategy directly.
1737        (_into_val, from_val, EntityDedupMergePolicy::PreferFrom) => (from_val.clone(), 1),
1738        _ => (into.clone(), 0),
1739    }
1740}
1741
1742fn union_tags(into: &[String], from: &[String]) -> (Vec<String>, usize) {
1743    let mut seen: HashSet<&str> = into.iter().map(|s| s.as_str()).collect();
1744    let mut result: Vec<String> = into.to_vec();
1745    let mut added = 0usize;
1746    for tag in from {
1747        if seen.insert(tag.as_str()) {
1748            result.push(tag.clone());
1749            added += 1;
1750        }
1751    }
1752    (result, added)
1753}
1754
1755// ---------------------------------------------------------------------------
1756// INLINE TEST JUSTIFICATION: tests here exercise patch/merge helpers and the
1757// update_note/update_entity paths that share private merge_properties logic.
1758// Moving them to tests/ would require pub-exporting merge_properties, which is
1759// an internal invariant not suitable for the public API surface. Broad
1760// behavioral curation tests live in tests/integration.rs.
1761// ---------------------------------------------------------------------------
1762
1763#[cfg(test)]
1764mod tests {
1765    use super::*;
1766    use crate::runtime::{KhiveRuntime, NamespaceToken};
1767    use khive_storage::types::{Direction, TextFilter, TextQueryMode, TextSearchRequest};
1768
1769    fn rt() -> KhiveRuntime {
1770        KhiveRuntime::memory().unwrap()
1771    }
1772
1773    // Helper: search FTS5 for `query` in a runtime namespace.
1774    async fn fts_hit(rt: &KhiveRuntime, token: &NamespaceToken, query: &str) -> Vec<Uuid> {
1775        let ns = token.namespace().as_str().to_string();
1776        rt.text(token)
1777            .unwrap()
1778            .search(TextSearchRequest {
1779                query: query.to_string(),
1780                mode: TextQueryMode::Plain,
1781                filter: Some(TextFilter {
1782                    namespaces: vec![ns],
1783                    ..Default::default()
1784                }),
1785                top_k: 50,
1786                snippet_chars: 100,
1787            })
1788            .await
1789            .unwrap()
1790            .into_iter()
1791            .map(|h| h.subject_id)
1792            .collect()
1793    }
1794
1795    #[tokio::test]
1796    async fn update_entity_patch_changes_only_specified_fields() {
1797        let rt = rt();
1798        let tok = NamespaceToken::local();
1799        let entity = rt
1800            .create_entity(
1801                &tok,
1802                "concept",
1803                None,
1804                "OriginalName",
1805                Some("orig desc"),
1806                Some(serde_json::json!({"k":"v"})),
1807                vec![],
1808            )
1809            .await
1810            .unwrap();
1811
1812        let updated = rt
1813            .update_entity(
1814                &tok,
1815                entity.id,
1816                EntityPatch {
1817                    description: Some(Some("new desc".to_string())),
1818                    ..Default::default()
1819                },
1820            )
1821            .await
1822            .unwrap();
1823
1824        assert_eq!(updated.name, "OriginalName");
1825        assert_eq!(updated.description.as_deref(), Some("new desc"));
1826        assert_eq!(updated.properties, Some(serde_json::json!({"k":"v"})));
1827    }
1828
1829    #[tokio::test]
1830    async fn update_entity_clear_description_with_some_none() {
1831        let rt = rt();
1832        let tok = NamespaceToken::local();
1833        let entity = rt
1834            .create_entity(
1835                &tok,
1836                "concept",
1837                None,
1838                "ClearDesc",
1839                Some("has description"),
1840                None,
1841                vec![],
1842            )
1843            .await
1844            .unwrap();
1845
1846        let updated = rt
1847            .update_entity(
1848                &tok,
1849                entity.id,
1850                EntityPatch {
1851                    description: Some(None),
1852                    ..Default::default()
1853                },
1854            )
1855            .await
1856            .unwrap();
1857
1858        assert!(
1859            updated.description.is_none(),
1860            "description should be cleared"
1861        );
1862    }
1863
1864    #[tokio::test]
1865    async fn update_entity_reindexes_when_name_changes() {
1866        let rt = rt();
1867        let tok = NamespaceToken::local();
1868        let entity = rt
1869            .create_entity(&tok, "concept", None, "OldName", None, None, vec![])
1870            .await
1871            .unwrap();
1872
1873        // Old name is findable.
1874        let hits_before = fts_hit(&rt, &tok, "OldName").await;
1875        assert!(
1876            hits_before.contains(&entity.id),
1877            "entity should be findable by old name"
1878        );
1879
1880        rt.update_entity(
1881            &tok,
1882            entity.id,
1883            EntityPatch {
1884                name: Some("NewName".to_string()),
1885                ..Default::default()
1886            },
1887        )
1888        .await
1889        .unwrap();
1890
1891        let hits_old = fts_hit(&rt, &tok, "OldName").await;
1892        let hits_new = fts_hit(&rt, &tok, "NewName").await;
1893
1894        // After rename, old name no longer matches this entity (FTS index updated).
1895        assert!(
1896            !hits_old.contains(&entity.id),
1897            "old name should no longer match after rename"
1898        );
1899        assert!(
1900            hits_new.contains(&entity.id),
1901            "new name should be findable after rename"
1902        );
1903    }
1904
1905    #[tokio::test]
1906    async fn update_entity_properties_merges_preserving_existing_keys() {
1907        let rt = rt();
1908        let tok = NamespaceToken::local();
1909        let entity = rt
1910            .create_entity(
1911                &tok,
1912                "concept",
1913                None,
1914                "MergeProps",
1915                None,
1916                Some(serde_json::json!({
1917                    "domain": "inference",
1918                    "repo": "lattice",
1919                    "status": "researched",
1920                })),
1921                vec![],
1922            )
1923            .await
1924            .unwrap();
1925
1926        let updated = rt
1927            .update_entity(
1928                &tok,
1929                entity.id,
1930                EntityPatch {
1931                    properties: Some(serde_json::json!({"status": "implemented"})),
1932                    ..Default::default()
1933                },
1934            )
1935            .await
1936            .unwrap();
1937
1938        let props = updated.properties.expect("properties should remain set");
1939        assert_eq!(props["domain"], "inference", "domain key must be preserved");
1940        assert_eq!(props["repo"], "lattice", "repo key must be preserved");
1941        assert_eq!(
1942            props["status"], "implemented",
1943            "status key must be updated by patch"
1944        );
1945    }
1946
1947    #[tokio::test]
1948    async fn update_entity_skips_reindex_when_only_properties_change() {
1949        let rt = rt();
1950        let tok = NamespaceToken::local();
1951        let entity = rt
1952            .create_entity(&tok, "concept", None, "StableIndexed", None, None, vec![])
1953            .await
1954            .unwrap();
1955
1956        // Verify it's in the index before.
1957        let hits_before = fts_hit(&rt, &tok, "StableIndexed").await;
1958        assert!(hits_before.contains(&entity.id));
1959
1960        // Only patch properties — text index should be untouched (still findable).
1961        rt.update_entity(
1962            &tok,
1963            entity.id,
1964            EntityPatch {
1965                properties: Some(serde_json::json!({"new": "prop"})),
1966                ..Default::default()
1967            },
1968        )
1969        .await
1970        .unwrap();
1971
1972        let hits_after = fts_hit(&rt, &tok, "StableIndexed").await;
1973        assert!(
1974            hits_after.contains(&entity.id),
1975            "still findable after props-only patch"
1976        );
1977    }
1978
1979    #[tokio::test]
1980    async fn merge_entity_rewires_edges() {
1981        let rt = rt();
1982        let tok = NamespaceToken::local();
1983        let a = rt
1984            .create_entity(&tok, "concept", None, "A", None, None, vec![])
1985            .await
1986            .unwrap();
1987        let b = rt
1988            .create_entity(&tok, "concept", None, "B", None, None, vec![])
1989            .await
1990            .unwrap();
1991        let c = rt
1992            .create_entity(&tok, "concept", None, "C", None, None, vec![])
1993            .await
1994            .unwrap();
1995        let d = rt
1996            .create_entity(&tok, "concept", None, "D", None, None, vec![])
1997            .await
1998            .unwrap();
1999
2000        // A→B and C→B; merge B into D → should become A→D and C→D.
2001        rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
2002            .await
2003            .unwrap();
2004        rt.link(&tok, c.id, b.id, EdgeRelation::Extends, 1.0, None)
2005            .await
2006            .unwrap();
2007
2008        let summary = rt
2009            .merge_entity(&tok, d.id, b.id, EntityDedupMergePolicy::PreferInto, false)
2010            .await
2011            .unwrap();
2012
2013        assert_eq!(summary.kept_id, d.id);
2014        assert_eq!(summary.removed_id, b.id);
2015        assert_eq!(summary.edges_rewired, 2);
2016
2017        // Verify edges now point to D.
2018        let a_neighbors = rt
2019            .neighbors(&tok, a.id, Direction::Out, None, None)
2020            .await
2021            .unwrap();
2022        assert_eq!(a_neighbors.len(), 1);
2023        assert_eq!(a_neighbors[0].node_id, d.id);
2024
2025        let c_neighbors = rt
2026            .neighbors(&tok, c.id, Direction::Out, None, None)
2027            .await
2028            .unwrap();
2029        assert_eq!(c_neighbors.len(), 1);
2030        assert_eq!(c_neighbors[0].node_id, d.id);
2031    }
2032
2033    #[tokio::test]
2034    async fn merge_entity_self_merge_rejected() {
2035        let rt = rt();
2036        let tok = NamespaceToken::local();
2037        let a = rt
2038            .create_entity(&tok, "concept", None, "A", None, None, vec![])
2039            .await
2040            .unwrap();
2041        let err = rt
2042            .merge_entity(&tok, a.id, a.id, EntityDedupMergePolicy::PreferInto, false)
2043            .await
2044            .unwrap_err();
2045        assert!(
2046            format!("{err:?}").contains("cannot merge an entity into itself"),
2047            "expected self-merge rejection, got: {err:?}"
2048        );
2049    }
2050
2051    #[tokio::test]
2052    async fn merge_entity_prefer_into_strategy() {
2053        let rt = rt();
2054        let tok = NamespaceToken::local();
2055        let into = rt
2056            .create_entity(
2057                &tok,
2058                "concept",
2059                None,
2060                "Into",
2061                None,
2062                Some(serde_json::json!({"a": 1})),
2063                vec![],
2064            )
2065            .await
2066            .unwrap();
2067        let from = rt
2068            .create_entity(
2069                &tok,
2070                "concept",
2071                None,
2072                "From",
2073                None,
2074                Some(serde_json::json!({"a": 2, "b": 3})),
2075                vec![],
2076            )
2077            .await
2078            .unwrap();
2079
2080        rt.merge_entity(
2081            &tok,
2082            into.id,
2083            from.id,
2084            EntityDedupMergePolicy::PreferInto,
2085            false,
2086        )
2087        .await
2088        .unwrap();
2089
2090        let kept = rt.get_entity(&tok, into.id).await.unwrap();
2091        let props = kept.properties.unwrap();
2092        // a stays as 1 (into wins), b is added from from.
2093        assert_eq!(props["a"], 1);
2094        assert_eq!(props["b"], 3);
2095    }
2096
2097    #[tokio::test]
2098    async fn merge_entity_prefer_from_strategy() {
2099        let rt = rt();
2100        let tok = NamespaceToken::local();
2101        let into = rt
2102            .create_entity(
2103                &tok,
2104                "concept",
2105                None,
2106                "Into",
2107                None,
2108                Some(serde_json::json!({"a": 1})),
2109                vec![],
2110            )
2111            .await
2112            .unwrap();
2113        let from = rt
2114            .create_entity(
2115                &tok,
2116                "concept",
2117                None,
2118                "From",
2119                None,
2120                Some(serde_json::json!({"a": 2, "b": 3})),
2121                vec![],
2122            )
2123            .await
2124            .unwrap();
2125
2126        rt.merge_entity(
2127            &tok,
2128            into.id,
2129            from.id,
2130            EntityDedupMergePolicy::PreferFrom,
2131            false,
2132        )
2133        .await
2134        .unwrap();
2135
2136        let kept = rt.get_entity(&tok, into.id).await.unwrap();
2137        let props = kept.properties.unwrap();
2138        // from wins on a, b also from from.
2139        assert_eq!(props["a"], 2);
2140        assert_eq!(props["b"], 3);
2141    }
2142
2143    #[tokio::test]
2144    async fn merge_entity_union_strategy() {
2145        let rt = rt();
2146        let tok = NamespaceToken::local();
2147        let into = rt
2148            .create_entity(
2149                &tok,
2150                "concept",
2151                None,
2152                "Into",
2153                None,
2154                Some(serde_json::json!({"a": 1})),
2155                vec![],
2156            )
2157            .await
2158            .unwrap();
2159        let from = rt
2160            .create_entity(
2161                &tok,
2162                "concept",
2163                None,
2164                "From",
2165                None,
2166                Some(serde_json::json!({"a": 2, "b": 3})),
2167                vec![],
2168            )
2169            .await
2170            .unwrap();
2171
2172        rt.merge_entity(&tok, into.id, from.id, EntityDedupMergePolicy::Union, false)
2173            .await
2174            .unwrap();
2175
2176        let kept = rt.get_entity(&tok, into.id).await.unwrap();
2177        let props = kept.properties.unwrap();
2178        // Scalar conflict: into wins → a=1. b added from from.
2179        assert_eq!(props["a"], 1);
2180        assert_eq!(props["b"], 3);
2181    }
2182
2183    #[tokio::test]
2184    async fn merge_entity_unions_tags() {
2185        let rt = rt();
2186        let tok = NamespaceToken::local();
2187        let into = rt
2188            .create_entity(
2189                &tok,
2190                "concept",
2191                None,
2192                "Into",
2193                None,
2194                None,
2195                vec!["x".to_string(), "y".to_string()],
2196            )
2197            .await
2198            .unwrap();
2199        let from = rt
2200            .create_entity(
2201                &tok,
2202                "concept",
2203                None,
2204                "From",
2205                None,
2206                None,
2207                vec!["y".to_string(), "z".to_string()],
2208            )
2209            .await
2210            .unwrap();
2211
2212        rt.merge_entity(
2213            &tok,
2214            into.id,
2215            from.id,
2216            EntityDedupMergePolicy::PreferInto,
2217            false,
2218        )
2219        .await
2220        .unwrap();
2221
2222        let kept = rt.get_entity(&tok, into.id).await.unwrap();
2223        let mut tags = kept.tags.clone();
2224        tags.sort();
2225        assert_eq!(tags, vec!["x", "y", "z"]);
2226    }
2227
2228    #[tokio::test]
2229    async fn merge_entity_drops_self_loops() {
2230        let rt = rt();
2231        let tok = NamespaceToken::local();
2232        let a = rt
2233            .create_entity(&tok, "concept", None, "A", None, None, vec![])
2234            .await
2235            .unwrap();
2236        let b = rt
2237            .create_entity(&tok, "concept", None, "B", None, None, vec![])
2238            .await
2239            .unwrap();
2240
2241        // A `extends` B — merging B into A would produce A `extends` A → drop it.
2242        rt.link(&tok, a.id, b.id, EdgeRelation::Extends, 1.0, None)
2243            .await
2244            .unwrap();
2245
2246        let summary = rt
2247            .merge_entity(&tok, a.id, b.id, EntityDedupMergePolicy::PreferInto, false)
2248            .await
2249            .unwrap();
2250
2251        assert_eq!(
2252            summary.edges_rewired, 0,
2253            "self-loop should be dropped, not rewired"
2254        );
2255
2256        let a_out = rt
2257            .neighbors(&tok, a.id, Direction::Out, None, None)
2258            .await
2259            .unwrap();
2260        assert!(a_out.is_empty(), "no self-loop should remain");
2261    }
2262
2263    // ---- merge helper unit tests ----
2264
2265    #[test]
2266    fn union_tags_deduplicates() {
2267        let (tags, added) = union_tags(
2268            &["x".to_string(), "y".to_string()],
2269            &["y".to_string(), "z".to_string()],
2270        );
2271        let mut sorted = tags.clone();
2272        sorted.sort();
2273        assert_eq!(sorted, vec!["x", "y", "z"]);
2274        assert_eq!(added, 1);
2275    }
2276
2277    #[test]
2278    fn merge_properties_prefer_into_fills_missing_keys() {
2279        let a = serde_json::json!({"a": 1});
2280        let b = serde_json::json!({"a": 99, "b": 2});
2281        let (merged, added) =
2282            merge_properties(&Some(a), &Some(b), EntityDedupMergePolicy::PreferInto);
2283        let m = merged.unwrap();
2284        assert_eq!(m["a"], 1);
2285        assert_eq!(m["b"], 2);
2286        assert_eq!(added, 1);
2287    }
2288
2289    // ---- tombstone and note merge tests ----
2290
2291    #[tokio::test]
2292    async fn merge_entity_tombstones_source_with_provenance() {
2293        let rt = rt();
2294        let tok = NamespaceToken::local();
2295        let into = rt
2296            .create_entity(&tok, "concept", None, "Into", None, None, vec![])
2297            .await
2298            .unwrap();
2299        let from = rt
2300            .create_entity(&tok, "concept", None, "From", None, None, vec![])
2301            .await
2302            .unwrap();
2303        let from_id = from.id;
2304
2305        rt.merge_entity(
2306            &tok,
2307            into.id,
2308            from_id,
2309            EntityDedupMergePolicy::PreferInto,
2310            false,
2311        )
2312        .await
2313        .unwrap();
2314
2315        // After merge, get_entity returns an error (soft-deleted rows are excluded).
2316        assert!(
2317            rt.get_entity(&tok, from_id).await.is_err(),
2318            "tombstoned source should not be returned by get_entity"
2319        );
2320
2321        // Verify the source row still exists in SQL with provenance.
2322        let pool = rt.backend().pool_arc();
2323        let (deleted_at, merged_into): (Option<i64>, Option<String>) =
2324            tokio::task::spawn_blocking(move || {
2325                let guard = pool.writer().unwrap();
2326                guard
2327                    .conn()
2328                    .query_row(
2329                        "SELECT deleted_at, merged_into FROM entities WHERE id = ?1",
2330                        [from_id.to_string()],
2331                        |row| Ok((row.get(0)?, row.get(1)?)),
2332                    )
2333                    .unwrap()
2334            })
2335            .await
2336            .unwrap();
2337        assert!(
2338            deleted_at.is_some(),
2339            "tombstoned entity must have deleted_at set"
2340        );
2341        assert_eq!(
2342            merged_into.as_deref(),
2343            Some(into.id.to_string().as_str()),
2344            "merged_into must point to into_id"
2345        );
2346    }
2347
2348    #[tokio::test]
2349    async fn merge_note_same_kind_appends_content() {
2350        let rt = rt();
2351        let tok = NamespaceToken::local();
2352        let into = rt
2353            .create_note(
2354                &tok,
2355                "observation",
2356                None,
2357                "Into content",
2358                None,
2359                None,
2360                vec![],
2361            )
2362            .await
2363            .unwrap();
2364        let from = rt
2365            .create_note(
2366                &tok,
2367                "observation",
2368                None,
2369                "From content",
2370                None,
2371                None,
2372                vec![],
2373            )
2374            .await
2375            .unwrap();
2376        let from_id = from.id;
2377
2378        let summary = rt
2379            .merge_note(
2380                &tok,
2381                into.id,
2382                from_id,
2383                EntityDedupMergePolicy::PreferInto,
2384                ContentMergeStrategy::Append,
2385                false,
2386            )
2387            .await
2388            .unwrap();
2389
2390        assert_eq!(summary.kept_id, into.id);
2391        assert_eq!(summary.removed_id, from_id);
2392        assert!(summary.content_appended);
2393        assert!(!summary.dry_run);
2394
2395        // Source is no longer findable.
2396        let from_store = rt.notes(&tok).unwrap();
2397        assert!(
2398            from_store.get_note(from_id).await.unwrap().is_none(),
2399            "merged-from note should be soft-deleted"
2400        );
2401    }
2402
2403    #[tokio::test]
2404    async fn merge_note_different_kinds_rejected() {
2405        let rt = rt();
2406        let tok = NamespaceToken::local();
2407        let into = rt
2408            .create_note(&tok, "observation", None, "Into", None, None, vec![])
2409            .await
2410            .unwrap();
2411        let from = rt
2412            .create_note(&tok, "decision", None, "From", None, None, vec![])
2413            .await
2414            .unwrap();
2415
2416        let result = rt
2417            .merge_note(
2418                &tok,
2419                into.id,
2420                from.id,
2421                EntityDedupMergePolicy::PreferInto,
2422                ContentMergeStrategy::Append,
2423                false,
2424            )
2425            .await;
2426        assert!(result.is_err(), "merging different note kinds must fail");
2427    }
2428
2429    #[tokio::test]
2430    async fn merge_note_dry_run_leaves_notes_unchanged() {
2431        let rt = rt();
2432        let tok = NamespaceToken::local();
2433        let into = rt
2434            .create_note(
2435                &tok,
2436                "observation",
2437                None,
2438                "Into content",
2439                None,
2440                None,
2441                vec![],
2442            )
2443            .await
2444            .unwrap();
2445        let from = rt
2446            .create_note(
2447                &tok,
2448                "observation",
2449                None,
2450                "From content",
2451                None,
2452                None,
2453                vec![],
2454            )
2455            .await
2456            .unwrap();
2457        let into_id = into.id;
2458        let from_id = from.id;
2459
2460        let summary = rt
2461            .merge_note(
2462                &tok,
2463                into_id,
2464                from_id,
2465                EntityDedupMergePolicy::PreferInto,
2466                ContentMergeStrategy::Append,
2467                true,
2468            )
2469            .await
2470            .unwrap();
2471
2472        assert!(summary.dry_run);
2473
2474        // Both notes still exist unchanged.
2475        let store = rt.notes(&tok).unwrap();
2476        let into_after = store.get_note(into_id).await.unwrap().unwrap();
2477        let from_after = store.get_note(from_id).await.unwrap().unwrap();
2478        assert_eq!(
2479            into_after.content, "Into content",
2480            "dry_run must not mutate into-note"
2481        );
2482        assert_eq!(
2483            from_after.content, "From content",
2484            "dry_run must not mutate from-note"
2485        );
2486    }
2487
2488    // Regression: merge two NAMELESS notes with no embedding model configured.
2489    // Before this fix, the raw SQL FTS INSERT bound &merged_name directly — for a
2490    // nameless note that is SQL NULL, while Fts5TextSearch::upsert_document stores
2491    // an empty string.  The mismatch caused get_document to diverge (or fail) for
2492    // nameless merged notes.  After the fix, note_fts_scalars drives every scalar
2493    // and the round-trip is field-identical.
2494    #[tokio::test]
2495    async fn merge_nameless_notes_fts_document_is_parity_correct() {
2496        use khive_storage::types::TextSearchRequest;
2497
2498        let rt = rt(); // in-memory runtime — no embedding model configured
2499        let tok = NamespaceToken::local();
2500
2501        let into = rt
2502            .create_note(
2503                &tok,
2504                "observation",
2505                None,
2506                "intosentinelzxq body",
2507                None,
2508                Some(serde_json::json!({"src": "into"})),
2509                vec![],
2510            )
2511            .await
2512            .expect("create into-note");
2513        let from = rt
2514            .create_note(
2515                &tok,
2516                "observation",
2517                None,
2518                "fromsentinelzxq body",
2519                None,
2520                None,
2521                vec![],
2522            )
2523            .await
2524            .expect("create from-note");
2525
2526        let into_id = into.id;
2527        let from_id = from.id;
2528
2529        rt.merge_note(
2530            &tok,
2531            into_id,
2532            from_id,
2533            EntityDedupMergePolicy::PreferInto,
2534            ContentMergeStrategy::Append,
2535            false,
2536        )
2537        .await
2538        .expect("merge_note must succeed");
2539
2540        // Fetch the merged note from the note store to get its current state.
2541        let note_store = rt.notes(&tok).expect("note store");
2542        let merged_note = note_store
2543            .get_note(into_id)
2544            .await
2545            .expect("get_note")
2546            .expect("merged note must exist");
2547
2548        // Compute the expected FTS document via the shared constructor.
2549        let expected = note_fts_document(&merged_note);
2550
2551        // Fetch the stored FTS document and verify field parity.
2552        let fts = rt.text_for_notes(&tok).expect("FTS store");
2553        let stored = fts
2554            .get_document("local", into_id)
2555            .await
2556            .expect("get_document must not error")
2557            .expect("FTS document must exist after merge");
2558
2559        // Core parity: subject_id, title (must be None, not Some("")), body.
2560        assert_eq!(stored.subject_id, expected.subject_id, "subject_id");
2561        assert_eq!(
2562            stored.title, expected.title,
2563            "title (None for nameless note)"
2564        );
2565        assert_eq!(stored.body, expected.body, "body");
2566        assert_eq!(stored.namespace, expected.namespace, "namespace");
2567        assert_eq!(stored.kind, expected.kind, "kind");
2568
2569        // The merged note is nameless — title must be None, matching the shared path.
2570        assert!(
2571            stored.title.is_none(),
2572            "nameless merged note must have title=None in FTS (was NULL before fix)"
2573        );
2574
2575        // The merged note must be searchable by a unique token from the into-note body.
2576        let hits = fts
2577            .search(TextSearchRequest {
2578                query: "intosentinelzxq".to_string(),
2579                mode: khive_storage::types::TextQueryMode::Plain,
2580                filter: None,
2581                top_k: 10,
2582                snippet_chars: 0,
2583            })
2584            .await
2585            .expect("search");
2586        assert!(
2587            hits.iter().any(|h| h.subject_id == into_id),
2588            "merged note must be searchable by into-note content"
2589        );
2590    }
2591
2592    #[tokio::test]
2593    async fn update_edge_updates_properties() {
2594        use khive_storage::EdgeRelation;
2595        let rt = rt();
2596        let tok = NamespaceToken::local();
2597        let a = rt
2598            .create_entity(&tok, "concept", None, "A", None, None, vec![])
2599            .await
2600            .unwrap();
2601        let b = rt
2602            .create_entity(&tok, "concept", None, "B", None, None, vec![])
2603            .await
2604            .unwrap();
2605        let edge = rt
2606            .link(&tok, a.id, b.id, EdgeRelation::Extends, 0.5, None)
2607            .await
2608            .unwrap();
2609        let edge_id: Uuid = edge.id.into();
2610
2611        let updated = rt
2612            .update_edge(
2613                &tok,
2614                edge_id,
2615                EdgePatch {
2616                    properties: Some(serde_json::json!({"source": "manual"})),
2617                    ..Default::default()
2618                },
2619            )
2620            .await
2621            .unwrap();
2622
2623        assert_eq!(updated.metadata.as_ref().unwrap()["source"], "manual");
2624        assert!((updated.weight - 0.5).abs() < 0.001, "weight unchanged");
2625    }
2626
2627    // scenario-kg-maintenance C1 regression: merge must not crash when both
2628    // entities share a common third-party edge (duplicate triple after rewire).
2629    // Before the fix, the double-ON-CONFLICT INSERT raised a UNIQUE constraint
2630    // error at the SQLite layer and the merge aborted mid-transaction.
2631    #[tokio::test]
2632    async fn merge_entity_survives_shared_edge_to_third_party() {
2633        use khive_storage::EdgeRelation;
2634        let rt = rt();
2635        let tok = NamespaceToken::local();
2636
2637        // Create three entities: A and B will be merged; shared is the common target.
2638        // Use `extends` (concept→concept) which is a valid endpoint combination.
2639        let a = rt
2640            .create_entity(&tok, "concept", None, "A", None, None, vec![])
2641            .await
2642            .unwrap();
2643        let b = rt
2644            .create_entity(&tok, "concept", None, "B", None, None, vec![])
2645            .await
2646            .unwrap();
2647        let shared = rt
2648            .create_entity(&tok, "concept", None, "Shared", None, None, vec![])
2649            .await
2650            .unwrap();
2651
2652        // Both A and B extend the same shared concept — this creates a duplicate
2653        // triple (A/B → shared, extends) that triggers the crash on rewire.
2654        rt.link(&tok, a.id, shared.id, EdgeRelation::Extends, 1.0, None)
2655            .await
2656            .unwrap();
2657        rt.link(&tok, b.id, shared.id, EdgeRelation::Extends, 1.0, None)
2658            .await
2659            .unwrap();
2660
2661        // Before the fix this would return Err with "UNIQUE constraint failed".
2662        let summary = rt
2663            .merge_entity(
2664                &tok,
2665                a.id,
2666                b.id,
2667                crate::EntityDedupMergePolicy::PreferInto,
2668                false,
2669            )
2670            .await
2671            .expect(
2672                "C1: merge must succeed even when both entities share an edge to a third party",
2673            );
2674
2675        assert_eq!(summary.kept_id, a.id);
2676        assert_eq!(summary.removed_id, b.id);
2677        // A already had the Extends edge to shared; when B→shared is rewired to
2678        // A→shared, the ON CONFLICT DO UPDATE refreshes the existing row (clears
2679        // deleted_at, updates weight). rusqlite reports this as 1 change, so
2680        // edges_rewired will be >= 0. The important invariant is that the merge
2681        // did NOT crash and exactly one live edge A→shared remains.
2682
2683        // One live edge A→shared must exist after merge.
2684        let a_edges = rt
2685            .list_edges(
2686                &tok,
2687                crate::EdgeListFilter {
2688                    source_id: Some(a.id),
2689                    target_id: Some(shared.id),
2690                    relations: vec![EdgeRelation::Extends],
2691                    ..Default::default()
2692                },
2693                10,
2694            )
2695            .await
2696            .unwrap();
2697        assert_eq!(
2698            a_edges.len(),
2699            1,
2700            "C1: exactly one live A→shared Extends edge must exist after merge; got: {a_edges:?}"
2701        );
2702
2703        // Tombstone check: B must be soft-deleted after successful merge (C3).
2704        // get_entity filters deleted_at IS NULL, so a tombstoned entity returns None.
2705        let b_after = rt.entities(&tok).unwrap().get_entity(b.id).await.unwrap();
2706        assert!(
2707            b_after.is_none(),
2708            "C3: from_entity must be tombstoned (get_entity returns None for deleted) after merge; got: {b_after:?}"
2709        );
2710    }
2711
2712    // H2 regression: merge_entity at the runtime level must reject cross-kind merges.
2713    // Before the H2 fix, only the pack handler had this guard; a direct runtime caller
2714    // could still merge concept+project, silently tombstoning the source entity.
2715    #[tokio::test]
2716    async fn merge_entity_cross_kind_rejected_at_runtime() {
2717        let rt = rt();
2718        let tok = NamespaceToken::local();
2719
2720        let concept = rt
2721            .create_entity(&tok, "concept", None, "H2Concept", None, None, vec![])
2722            .await
2723            .unwrap();
2724        let project = rt
2725            .create_entity(&tok, "project", None, "H2Project", None, None, vec![])
2726            .await
2727            .unwrap();
2728
2729        // Cross-kind merge must return InvalidInput at the runtime level.
2730        let err = rt
2731            .merge_entity(
2732                &tok,
2733                concept.id,
2734                project.id,
2735                crate::EntityDedupMergePolicy::PreferInto,
2736                false,
2737            )
2738            .await
2739            .expect_err("H2: cross-kind merge must be rejected by runtime");
2740        assert!(
2741            matches!(err, crate::RuntimeError::InvalidInput(_)),
2742            "H2: expected InvalidInput, got: {err:?}"
2743        );
2744
2745        // Both entities must survive the failed merge attempt with no tombstone.
2746        let concept_after = rt.get_entity(&tok, concept.id).await;
2747        let project_after = rt.get_entity(&tok, project.id).await;
2748        assert!(
2749            concept_after.is_ok(),
2750            "H2: concept must remain live after rejected merge; got: {concept_after:?}"
2751        );
2752        assert!(
2753            project_after.is_ok(),
2754            "H2: project must remain live after rejected merge; got: {project_after:?}"
2755        );
2756    }
2757
2758    // scenario-kg-maintenance C2 regression: same-kind merge must succeed.
2759    #[tokio::test]
2760    async fn merge_entity_same_kind_succeeds() {
2761        let rt = rt();
2762        let tok = NamespaceToken::local();
2763
2764        let c1 = rt
2765            .create_entity(&tok, "concept", None, "Concept1", None, None, vec![])
2766            .await
2767            .unwrap();
2768        let c2 = rt
2769            .create_entity(&tok, "concept", None, "Concept2", None, None, vec![])
2770            .await
2771            .unwrap();
2772
2773        let summary = rt
2774            .merge_entity(
2775                &tok,
2776                c1.id,
2777                c2.id,
2778                crate::EntityDedupMergePolicy::PreferInto,
2779                false,
2780            )
2781            .await
2782            .expect("same-kind merge must succeed");
2783        assert_eq!(summary.kept_id, c1.id);
2784        assert_eq!(summary.removed_id, c2.id);
2785
2786        // c2 must be tombstoned.
2787        let c2_after = rt.entities(&tok).unwrap().get_entity(c2.id).await.unwrap();
2788        assert!(c2_after.is_none(), "from_entity must be tombstoned");
2789    }
2790
2791    // ── #567 regression: cross-namespace merge_note must be denied on either ID ──
2792
2793    #[tokio::test]
2794    async fn merge_note_cross_namespace_either_id_returns_not_found() {
2795        use crate::error::RuntimeError;
2796        use crate::Namespace;
2797
2798        let rt = rt();
2799        let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
2800        let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
2801
2802        let into_a = rt
2803            .create_note(&ns_a, "observation", None, "Into A", None, None, vec![])
2804            .await
2805            .unwrap();
2806        let from_a = rt
2807            .create_note(&ns_a, "observation", None, "From A", None, None, vec![])
2808            .await
2809            .unwrap();
2810        let note_b = rt
2811            .create_note(&ns_b, "observation", None, "Note B", None, None, vec![])
2812            .await
2813            .unwrap();
2814
2815        // foreign into_id: note_b belongs to ns_b, caller token is ns_a
2816        let foreign_into = rt
2817            .merge_note(
2818                &ns_a,
2819                note_b.id,
2820                from_a.id,
2821                EntityDedupMergePolicy::PreferInto,
2822                ContentMergeStrategy::Append,
2823                false,
2824            )
2825            .await;
2826        assert!(
2827            matches!(foreign_into, Err(RuntimeError::NotFound(_))),
2828            "foreign into_id must be denied before merge, got {foreign_into:?}"
2829        );
2830
2831        // foreign from_id: note_b belongs to ns_b, caller token is ns_a
2832        let foreign_from = rt
2833            .merge_note(
2834                &ns_a,
2835                into_a.id,
2836                note_b.id,
2837                EntityDedupMergePolicy::PreferInto,
2838                ContentMergeStrategy::Append,
2839                false,
2840            )
2841            .await;
2842        assert!(
2843            matches!(foreign_from, Err(RuntimeError::NotFound(_))),
2844            "foreign from_id must be denied before merge, got {foreign_from:?}"
2845        );
2846    }
2847
2848    // ADR-007 PR-A1: cross-namespace update now succeeds (shared-brain model).
2849
2850    #[tokio::test]
2851    async fn update_entity_cross_namespace_succeeds() {
2852        use crate::Namespace;
2853
2854        let rt = rt();
2855        let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
2856        let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
2857
2858        let entity = rt
2859            .create_entity(
2860                &ns_a,
2861                "concept",
2862                None,
2863                "Alpha",
2864                Some("original"),
2865                None,
2866                vec![],
2867            )
2868            .await
2869            .unwrap();
2870
2871        // ADR-007 PR-A1: update from ns_b token must now succeed on an ns_a entity.
2872        let result = rt
2873            .update_entity(
2874                &ns_b,
2875                entity.id,
2876                EntityPatch {
2877                    name: Some("Updated".into()),
2878                    ..Default::default()
2879                },
2880            )
2881            .await;
2882
2883        assert!(
2884            result.is_ok(),
2885            "cross-namespace update must succeed in shared-brain OSS; got {result:?}"
2886        );
2887        assert_eq!(result.unwrap().name, "Updated");
2888    }
2889
2890    // merge_entity still requires both entities to be in the same namespace as
2891    // the token's write namespace (enforced at the SQL transaction layer, not the
2892    // runtime layer).  This is a merge-semantic constraint, not tenant isolation.
2893    #[tokio::test]
2894    async fn merge_entity_cross_namespace_ids_fail_at_sql_layer() {
2895        use crate::Namespace;
2896
2897        let rt = rt();
2898        let ns_a = NamespaceToken::for_namespace(Namespace::parse("ns-a").unwrap());
2899        let ns_b = NamespaceToken::for_namespace(Namespace::parse("ns-b").unwrap());
2900
2901        let into_a = rt
2902            .create_entity(&ns_a, "concept", None, "Into A", None, None, vec![])
2903            .await
2904            .unwrap();
2905        let from_a = rt
2906            .create_entity(&ns_a, "concept", None, "From A", None, None, vec![])
2907            .await
2908            .unwrap();
2909        let foreign_b = rt
2910            .create_entity(&ns_b, "concept", None, "Foreign B", None, None, vec![])
2911            .await
2912            .unwrap();
2913
2914        // foreign into_id: SQL read_merge_entity checks ns matches token namespace.
2915        let foreign_into = rt
2916            .merge_entity(
2917                &ns_a,
2918                foreign_b.id,
2919                from_a.id,
2920                EntityDedupMergePolicy::PreferInto,
2921                false,
2922            )
2923            .await;
2924        assert!(
2925            foreign_into.is_err(),
2926            "cross-namespace into_id must still fail at SQL layer; got {foreign_into:?}"
2927        );
2928
2929        // foreign from_id: same SQL constraint.
2930        let foreign_from = rt
2931            .merge_entity(
2932                &ns_a,
2933                into_a.id,
2934                foreign_b.id,
2935                EntityDedupMergePolicy::PreferInto,
2936                false,
2937            )
2938            .await;
2939        assert!(
2940            foreign_from.is_err(),
2941            "cross-namespace from_id must still fail at SQL layer; got {foreign_from:?}"
2942        );
2943
2944        // All three entities survive the failed merges.
2945        assert!(rt.get_entity(&ns_a, into_a.id).await.is_ok());
2946        assert!(rt.get_entity(&ns_a, from_a.id).await.is_ok());
2947        assert!(rt.get_entity(&ns_b, foreign_b.id).await.is_ok());
2948    }
2949
2950    // Parity: entity_fts_document must produce the same body/title as the
2951    // create_entity and update_entity FTS write paths.
2952    #[test]
2953    fn entity_fts_document_with_description() {
2954        let mut entity = Entity::new("local", "concept", "MyEntity");
2955        entity = entity.with_description("some description text");
2956        let doc = entity_fts_document(&entity);
2957        assert_eq!(doc.subject_id, entity.id);
2958        assert_eq!(doc.namespace, "local");
2959        assert_eq!(doc.title.as_deref(), Some("MyEntity"));
2960        assert_eq!(doc.body, "MyEntity some description text");
2961        assert_eq!(doc.kind, khive_types::SubstrateKind::Entity);
2962    }
2963
2964    #[test]
2965    fn entity_fts_document_without_description() {
2966        let entity = Entity::new("local", "concept", "NameOnly");
2967        let doc = entity_fts_document(&entity);
2968        assert_eq!(doc.title.as_deref(), Some("NameOnly"));
2969        assert_eq!(doc.body, "NameOnly");
2970    }
2971
2972    #[test]
2973    fn entity_fts_document_empty_description_uses_name_only() {
2974        let mut entity = Entity::new("local", "concept", "TitleOnly");
2975        entity = entity.with_description("");
2976        let doc = entity_fts_document(&entity);
2977        assert_eq!(
2978            doc.body, "TitleOnly",
2979            "empty description must not be appended"
2980        );
2981    }
2982
2983    // Cross-path equality: an entity created through the runtime (operations.rs
2984    // create_entity path) must produce a stored FTS document field-identical to
2985    // entity_fts_document() called on the same Entity.
2986    #[tokio::test]
2987    async fn entity_fts_document_matches_runtime_create_path() {
2988        let rt = rt();
2989        let tok = NamespaceToken::local();
2990
2991        let entity = rt
2992            .create_entity(
2993                &tok,
2994                "concept",
2995                None,
2996                "CrossPathTitle",
2997                Some("cross path description body"),
2998                Some(serde_json::json!({"key": "val"})),
2999                vec!["tag1".to_string()],
3000            )
3001            .await
3002            .expect("create_entity");
3003
3004        let fts = rt.text(&tok).expect("FTS store");
3005        let stored = fts
3006            .get_document("local", entity.id)
3007            .await
3008            .expect("get_document")
3009            .expect("document must exist after create_entity");
3010
3011        let expected = entity_fts_document(&entity);
3012
3013        assert_eq!(stored.subject_id, expected.subject_id, "subject_id");
3014        assert_eq!(stored.kind, expected.kind, "kind");
3015        assert_eq!(stored.title, expected.title, "title");
3016        assert_eq!(stored.body, expected.body, "body");
3017        assert_eq!(stored.namespace, expected.namespace, "namespace");
3018    }
3019
3020    // Cross-path equality: update_entity must produce a stored FTS document
3021    // field-identical to entity_fts_document() on the updated Entity.
3022    #[tokio::test]
3023    async fn entity_fts_document_matches_runtime_update_path() {
3024        let rt = rt();
3025        let tok = NamespaceToken::local();
3026
3027        let entity = rt
3028            .create_entity(
3029                &tok,
3030                "concept",
3031                None,
3032                "OldName",
3033                Some("old desc"),
3034                None,
3035                vec![],
3036            )
3037            .await
3038            .expect("create_entity");
3039
3040        let updated = rt
3041            .update_entity(
3042                &tok,
3043                entity.id,
3044                EntityPatch {
3045                    name: Some("NewName".to_string()),
3046                    description: Some(Some("new desc".to_string())),
3047                    ..Default::default()
3048                },
3049            )
3050            .await
3051            .expect("update_entity");
3052
3053        let fts = rt.text(&tok).expect("FTS store");
3054        let stored = fts
3055            .get_document("local", updated.id)
3056            .await
3057            .expect("get_document")
3058            .expect("document must exist after update_entity");
3059
3060        let expected = entity_fts_document(&updated);
3061
3062        assert_eq!(stored.title, expected.title, "title after update");
3063        assert_eq!(stored.body, expected.body, "body after update");
3064    }
3065
3066    // ── Fix #135 regression tests ────────────────────────────────────────────
3067    //
3068    // Verify that merge_entity / merge_note delete from_id vectors from ALL
3069    // registered model vec tables, not just the default-model table.
3070    //
3071    // Uses the same ConstVecProvider/ConstVecService pattern as operations.rs
3072    // so no real model files are required.
3073
3074    struct MergeTestVecService {
3075        dims: usize,
3076    }
3077
3078    #[async_trait::async_trait]
3079    impl lattice_embed::EmbeddingService for MergeTestVecService {
3080        async fn embed(
3081            &self,
3082            texts: &[String],
3083            _model: lattice_embed::EmbeddingModel,
3084        ) -> std::result::Result<Vec<Vec<f32>>, lattice_embed::EmbedError> {
3085            Ok(texts.iter().map(|_| vec![1.0_f32; self.dims]).collect())
3086        }
3087
3088        fn supports_model(&self, _model: lattice_embed::EmbeddingModel) -> bool {
3089            true
3090        }
3091
3092        fn name(&self) -> &'static str {
3093            "merge-test-const-vec"
3094        }
3095    }
3096
3097    struct MergeTestVecProvider {
3098        provider_name: String,
3099        dims: usize,
3100    }
3101
3102    impl MergeTestVecProvider {
3103        fn new(name: &str, dims: usize) -> Self {
3104            Self {
3105                provider_name: name.to_owned(),
3106                dims,
3107            }
3108        }
3109    }
3110
3111    #[async_trait::async_trait]
3112    impl crate::embedder_registry::EmbedderProvider for MergeTestVecProvider {
3113        fn name(&self) -> &str {
3114            &self.provider_name
3115        }
3116
3117        fn dimensions(&self) -> usize {
3118            self.dims
3119        }
3120
3121        async fn build(
3122            &self,
3123        ) -> crate::error::RuntimeResult<std::sync::Arc<dyn lattice_embed::EmbeddingService>>
3124        {
3125            Ok(std::sync::Arc::new(MergeTestVecService { dims: self.dims }))
3126        }
3127    }
3128
3129    /// merge_entity must delete from_id vectors from ALL registered model tables.
3130    ///
3131    /// Two custom embedders ("merge-vec-a", "merge-vec-b") are registered.  Both
3132    /// entities are embedded so each has a row in both model tables.  After merge,
3133    /// from_id must have zero surviving rows in either table.
3134    #[tokio::test]
3135    async fn merge_entity_clears_vectors_from_all_registered_models() {
3136        const DIMS: usize = 4;
3137        let rt = KhiveRuntime::memory().unwrap();
3138        rt.register_embedder(MergeTestVecProvider::new("merge-vec-a", DIMS));
3139        rt.register_embedder(MergeTestVecProvider::new("merge-vec-b", DIMS));
3140
3141        let ns_str = "merge-entity-vec-cleanup";
3142        let ns = crate::Namespace::parse(ns_str).unwrap();
3143        let tok = NamespaceToken::for_namespace(ns);
3144
3145        let into_e = rt
3146            .create_entity(
3147                &tok,
3148                "concept",
3149                None,
3150                "IntoVecEntity",
3151                Some("desc a"),
3152                None,
3153                vec![],
3154            )
3155            .await
3156            .expect("create into");
3157        let from_e = rt
3158            .create_entity(
3159                &tok,
3160                "concept",
3161                None,
3162                "FromVecEntity",
3163                Some("desc b"),
3164                None,
3165                vec![],
3166            )
3167            .await
3168            .expect("create from");
3169
3170        // Confirm both entities have vectors in both model tables before merge.
3171        let vs_a = rt.vectors_for_model(&tok, "merge-vec-a").unwrap();
3172        let vs_b = rt.vectors_for_model(&tok, "merge-vec-b").unwrap();
3173        use khive_storage::types::VectorSearchRequest;
3174        let query = vec![1.0_f32; DIMS];
3175        let pre_a = vs_a
3176            .search(VectorSearchRequest {
3177                query_vectors: vec![query.clone()],
3178                top_k: 100,
3179                namespace: Some(ns_str.to_string()),
3180                kind: Some(khive_types::SubstrateKind::Entity),
3181                embedding_model: Some("merge-vec-a".to_string()),
3182                filter: None,
3183                backend_hints: None,
3184            })
3185            .await
3186            .unwrap();
3187        assert!(
3188            pre_a.iter().any(|h| h.subject_id == into_e.id)
3189                && pre_a.iter().any(|h| h.subject_id == from_e.id),
3190            "both entities must be in model-a before merge; got {pre_a:?}"
3191        );
3192
3193        // model-b must ALSO hold both entities pre-merge, else the post-merge
3194        // model-b emptiness check below is vacuous (nothing to delete).
3195        let pre_b = vs_b
3196            .search(VectorSearchRequest {
3197                query_vectors: vec![query.clone()],
3198                top_k: 100,
3199                namespace: Some(ns_str.to_string()),
3200                kind: Some(khive_types::SubstrateKind::Entity),
3201                embedding_model: Some("merge-vec-b".to_string()),
3202                filter: None,
3203                backend_hints: None,
3204            })
3205            .await
3206            .unwrap();
3207        assert!(
3208            pre_b.iter().any(|h| h.subject_id == into_e.id)
3209                && pre_b.iter().any(|h| h.subject_id == from_e.id),
3210            "both entities must be in model-b before merge; got {pre_b:?}"
3211        );
3212
3213        rt.merge_entity(
3214            &tok,
3215            into_e.id,
3216            from_e.id,
3217            EntityDedupMergePolicy::PreferInto,
3218            false,
3219        )
3220        .await
3221        .expect("merge_entity");
3222
3223        // from_id must have ZERO rows in EITHER model table after merge.
3224        let post_a = vs_a
3225            .search(VectorSearchRequest {
3226                query_vectors: vec![query.clone()],
3227                top_k: 100,
3228                namespace: Some(ns_str.to_string()),
3229                kind: Some(khive_types::SubstrateKind::Entity),
3230                embedding_model: Some("merge-vec-a".to_string()),
3231                filter: None,
3232                backend_hints: None,
3233            })
3234            .await
3235            .unwrap();
3236        let from_ids_a: Vec<_> = post_a
3237            .iter()
3238            .filter(|h| h.subject_id == from_e.id)
3239            .collect();
3240        assert!(
3241            from_ids_a.is_empty(),
3242            "from_id must have no vectors in model-a after merge; got {from_ids_a:?}"
3243        );
3244
3245        let post_b = vs_b
3246            .search(VectorSearchRequest {
3247                query_vectors: vec![query],
3248                top_k: 100,
3249                namespace: Some(ns_str.to_string()),
3250                kind: Some(khive_types::SubstrateKind::Entity),
3251                embedding_model: Some("merge-vec-b".to_string()),
3252                filter: None,
3253                backend_hints: None,
3254            })
3255            .await
3256            .unwrap();
3257        let from_ids_b: Vec<_> = post_b
3258            .iter()
3259            .filter(|h| h.subject_id == from_e.id)
3260            .collect();
3261        assert!(
3262            from_ids_b.is_empty(),
3263            "from_id must have no vectors in model-b after merge; got {from_ids_b:?}"
3264        );
3265    }
3266
3267    /// merge_note must delete from_id vectors from ALL registered model tables.
3268    ///
3269    /// Two custom embedders ("merge-note-vec-a", "merge-note-vec-b") are registered.
3270    /// Both notes are embedded so each has a row in both model tables.  After merge,
3271    /// from_id must have zero surviving rows in either table.
3272    #[tokio::test]
3273    async fn merge_note_clears_vectors_from_all_registered_models() {
3274        const DIMS: usize = 4;
3275        let rt = KhiveRuntime::memory().unwrap();
3276        rt.register_embedder(MergeTestVecProvider::new("merge-note-vec-a", DIMS));
3277        rt.register_embedder(MergeTestVecProvider::new("merge-note-vec-b", DIMS));
3278
3279        let ns_str = "merge-note-vec-cleanup";
3280        let ns = crate::Namespace::parse(ns_str).unwrap();
3281        let tok = NamespaceToken::for_namespace(ns);
3282
3283        let into_n = rt
3284            .create_note(
3285                &tok,
3286                "observation",
3287                None,
3288                "IntoVecNote content",
3289                None,
3290                None,
3291                vec![],
3292            )
3293            .await
3294            .expect("create into note");
3295        let from_n = rt
3296            .create_note(
3297                &tok,
3298                "observation",
3299                None,
3300                "FromVecNote content",
3301                None,
3302                None,
3303                vec![],
3304            )
3305            .await
3306            .expect("create from note");
3307
3308        let vs_a = rt.vectors_for_model(&tok, "merge-note-vec-a").unwrap();
3309        let vs_b = rt.vectors_for_model(&tok, "merge-note-vec-b").unwrap();
3310        use khive_storage::types::VectorSearchRequest;
3311        let query = vec![1.0_f32; DIMS];
3312
3313        let pre_a = vs_a
3314            .search(VectorSearchRequest {
3315                query_vectors: vec![query.clone()],
3316                top_k: 100,
3317                namespace: Some(ns_str.to_string()),
3318                kind: Some(khive_types::SubstrateKind::Note),
3319                embedding_model: Some("merge-note-vec-a".to_string()),
3320                filter: None,
3321                backend_hints: None,
3322            })
3323            .await
3324            .unwrap();
3325        assert!(
3326            pre_a.iter().any(|h| h.subject_id == into_n.id)
3327                && pre_a.iter().any(|h| h.subject_id == from_n.id),
3328            "both notes must be in model-a before merge; got {pre_a:?}"
3329        );
3330
3331        // model-b must ALSO hold both notes pre-merge, else the post-merge
3332        // model-b emptiness check below is vacuous (nothing to delete).
3333        let pre_b = vs_b
3334            .search(VectorSearchRequest {
3335                query_vectors: vec![query.clone()],
3336                top_k: 100,
3337                namespace: Some(ns_str.to_string()),
3338                kind: Some(khive_types::SubstrateKind::Note),
3339                embedding_model: Some("merge-note-vec-b".to_string()),
3340                filter: None,
3341                backend_hints: None,
3342            })
3343            .await
3344            .unwrap();
3345        assert!(
3346            pre_b.iter().any(|h| h.subject_id == into_n.id)
3347                && pre_b.iter().any(|h| h.subject_id == from_n.id),
3348            "both notes must be in model-b before merge; got {pre_b:?}"
3349        );
3350
3351        rt.merge_note(
3352            &tok,
3353            into_n.id,
3354            from_n.id,
3355            EntityDedupMergePolicy::PreferInto,
3356            ContentMergeStrategy::PreferInto,
3357            false,
3358        )
3359        .await
3360        .expect("merge_note");
3361
3362        // from_id must have ZERO rows in EITHER model table after merge.
3363        let post_a = vs_a
3364            .search(VectorSearchRequest {
3365                query_vectors: vec![query.clone()],
3366                top_k: 100,
3367                namespace: Some(ns_str.to_string()),
3368                kind: Some(khive_types::SubstrateKind::Note),
3369                embedding_model: Some("merge-note-vec-a".to_string()),
3370                filter: None,
3371                backend_hints: None,
3372            })
3373            .await
3374            .unwrap();
3375        let from_ids_a: Vec<_> = post_a
3376            .iter()
3377            .filter(|h| h.subject_id == from_n.id)
3378            .collect();
3379        assert!(
3380            from_ids_a.is_empty(),
3381            "from_id must have no vectors in model-a after merge; got {from_ids_a:?}"
3382        );
3383
3384        let post_b = vs_b
3385            .search(VectorSearchRequest {
3386                query_vectors: vec![query],
3387                top_k: 100,
3388                namespace: Some(ns_str.to_string()),
3389                kind: Some(khive_types::SubstrateKind::Note),
3390                embedding_model: Some("merge-note-vec-b".to_string()),
3391                filter: None,
3392                backend_hints: None,
3393            })
3394            .await
3395            .unwrap();
3396        let from_ids_b: Vec<_> = post_b
3397            .iter()
3398            .filter(|h| h.subject_id == from_n.id)
3399            .collect();
3400        assert!(
3401            from_ids_b.is_empty(),
3402            "from_id must have no vectors in model-b after merge; got {from_ids_b:?}"
3403        );
3404    }
3405
3406    // Cross-path equality: merge_entity must produce a stored FTS document for
3407    // the kept entity that is field-identical to entity_fts_document().
3408    #[tokio::test]
3409    async fn entity_fts_document_matches_runtime_merge_path() {
3410        let rt = rt();
3411        let tok = NamespaceToken::local();
3412
3413        let into_e = rt
3414            .create_entity(
3415                &tok,
3416                "concept",
3417                None,
3418                "IntoEntity",
3419                Some("into desc"),
3420                None,
3421                vec![],
3422            )
3423            .await
3424            .expect("create into");
3425        let from_e = rt
3426            .create_entity(
3427                &tok,
3428                "concept",
3429                None,
3430                "FromEntity",
3431                Some("from desc"),
3432                None,
3433                vec![],
3434            )
3435            .await
3436            .expect("create from");
3437
3438        let summary = rt
3439            .merge_entity(
3440                &tok,
3441                into_e.id,
3442                from_e.id,
3443                EntityDedupMergePolicy::PreferInto,
3444                false,
3445            )
3446            .await
3447            .expect("merge_entity");
3448
3449        let kept = rt
3450            .get_entity(&tok, summary.kept_id)
3451            .await
3452            .expect("get kept");
3453
3454        let fts = rt.text(&tok).expect("FTS store");
3455        let stored = fts
3456            .get_document("local", kept.id)
3457            .await
3458            .expect("get_document")
3459            .expect("FTS document must exist for kept entity after merge");
3460
3461        let expected = entity_fts_document(&kept);
3462
3463        assert_eq!(stored.title, expected.title, "title after merge");
3464        assert_eq!(stored.body, expected.body, "body after merge");
3465    }
3466}