1use 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#[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#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
59#[serde(rename_all = "snake_case")]
60pub enum EntityDedupMergePolicy {
61 #[default]
64 PreferInto,
65 PreferFrom,
67 Union,
69}
70
71#[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#[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#[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#[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 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#[derive(Clone, Debug, Default)]
143pub struct EdgeListFilter {
144 pub source_id: Option<Uuid>,
145 pub target_id: Option<Uuid>,
146 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#[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
186impl KhiveRuntime {
191 pub async fn update_entity(
199 &self,
200 token: &NamespaceToken,
201 id: Uuid,
202 patch: EntityPatch,
203 ) -> RuntimeResult<Entity> {
204 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 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 {
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 let _ = self.entities(token)?;
328 let _ = self.graph(token)?;
329 let _ = self.text(token)?;
330 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 !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 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 pub(crate) async fn reindex_entity(
401 &self,
402 token: &NamespaceToken,
403 entity: &Entity,
404 ) -> RuntimeResult<()> {
405 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 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 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, ¬e.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 pub async fn update_note(
533 &self,
534 token: &NamespaceToken,
535 id: Uuid,
536 patch: NotePatch,
537 ) -> RuntimeResult<khive_storage::note::Note> {
538 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 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 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 ¬e.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, ¬e).await?;
604 }
605
606 Ok(note)
607 }
608
609 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 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
686pub 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
721pub fn note_fts_document(note: &Note) -> TextDocument {
734 let body = match ¬e.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
752pub(crate) struct NoteFtsScalars {
758 pub title: String,
761 pub body: String,
762 pub tags: String,
764 pub metadata: Option<String>,
766 pub updated_at_micros: i64,
768}
769
770pub(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
788fn 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#[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 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 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 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 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 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 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 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 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 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 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 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 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 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
1196fn 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#[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 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 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 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 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 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 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 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 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 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 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 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 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
1633fn 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 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
1675fn 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
1695fn 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 (_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#[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 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 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 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 let hits_before = fts_hit(&rt, &tok, "StableIndexed").await;
1958 assert!(hits_before.contains(&entity.id));
1959
1960 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 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 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 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 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 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 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 #[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 #[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 assert!(
2317 rt.get_entity(&tok, from_id).await.is_err(),
2318 "tombstoned source should not be returned by get_entity"
2319 );
2320
2321 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 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 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 #[tokio::test]
2495 async fn merge_nameless_notes_fts_document_is_parity_correct() {
2496 use khive_storage::types::TextSearchRequest;
2497
2498 let rt = rt(); 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 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 let expected = note_fts_document(&merged_note);
2550
2551 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 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 assert!(
2571 stored.title.is_none(),
2572 "nameless merged note must have title=None in FTS (was NULL before fix)"
2573 );
2574
2575 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 #[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 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 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 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 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 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 #[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 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 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 #[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 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 #[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 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 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 #[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 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 #[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 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 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 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 #[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 #[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 #[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 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 #[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 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 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 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 #[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 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 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 #[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}