1use serde_json::Value;
38use uuid::Uuid;
39
40use khive_storage::types::SqlValue;
41use khive_storage::{EdgeRelation, SqlStatement};
42use khive_types::{EventKind, SubstrateKind};
43
44use crate::atomic_plan::{
45 AffectedRowGuard, DeletePlan, EdgeNaturalKey, LinkPlan, MergePlan, PlanStatement,
46 PostCommitEffect, UpdatePlan,
47};
48use crate::atomic_runner::AtomicOpPlan;
49use crate::error::{RuntimeError, RuntimeResult};
50use crate::operations::{
51 canonical_edge_endpoints, merge_dependency_kind, validate_edge_metadata, validate_edge_weight,
52 Resolved,
53};
54use crate::runtime::{KhiveRuntime, NamespaceToken};
55
56use khive_db::stores::entity::{
57 entity_hard_delete_statement, entity_soft_delete_statement, entity_upsert_statement,
58};
59use khive_db::stores::event::event_insert_statements;
60use khive_db::stores::graph::{
61 edge_hard_delete_statement, edge_insert_guarded_by_endpoints_statement,
62 edge_soft_delete_statement, edge_symmetric_delete_if_conflict_statement,
63 edge_symmetric_refresh_or_update_inplace_statement, edge_upsert_statement,
64 purge_incident_edges_statement,
65};
66use khive_db::stores::note::{
67 note_hard_delete_statement, note_soft_delete_statement, note_upsert_statement,
68};
69
70fn obj(args: &Value) -> RuntimeResult<&serde_json::Map<String, Value>> {
75 args.as_object()
76 .ok_or_else(|| RuntimeError::InvalidInput("op args must be a JSON object".into()))
77}
78
79fn require_str<'a>(args: &'a Value, key: &str) -> RuntimeResult<&'a str> {
80 obj(args)?
81 .get(key)
82 .and_then(|v| v.as_str())
83 .ok_or_else(|| RuntimeError::InvalidInput(format!("missing required field {key:?}")))
84}
85
86fn require_uuid(args: &Value, key: &str) -> RuntimeResult<Uuid> {
87 let raw = require_str(args, key)?;
88 Uuid::parse_str(raw)
89 .map_err(|_| RuntimeError::InvalidInput(format!("{key} must be a full UUID; got {raw:?}")))
90}
91
92fn optional_str<'a>(args: &'a Value, key: &str) -> Option<&'a str> {
93 obj(args).ok()?.get(key).and_then(|v| v.as_str())
94}
95
96fn optional_string_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<String>>> {
110 match obj(args)?.get(key) {
111 None | Some(Value::Null) => Ok(None),
112 Some(Value::String(s)) => Ok(Some(Some(s.clone()))),
113 Some(other) => Err(RuntimeError::InvalidInput(format!(
114 "{key} must be a string or null, got: {other}"
115 ))),
116 }
117}
118
119fn entity_name_patch(args: &Value) -> RuntimeResult<Option<String>> {
130 match obj(args)?.get("name") {
131 None | Some(Value::Null) => Ok(None),
132 Some(Value::String(s)) => Ok(Some(s.clone())),
133 Some(other) => Err(RuntimeError::InvalidInput(format!(
134 "name must be a string, got: {other}"
135 ))),
136 }
137}
138
139fn optional_properties(args: &Value, key: &str) -> RuntimeResult<Option<Value>> {
148 match obj(args)?.get(key) {
149 None | Some(Value::Null) => Ok(None),
150 Some(v) => Ok(Some(v.clone())),
151 }
152}
153
154fn optional_tags(args: &Value) -> RuntimeResult<Option<Vec<String>>> {
161 match obj(args)?.get("tags") {
162 None | Some(Value::Null) => Ok(None),
163 Some(Value::Array(items)) => {
164 let mut tags = Vec::with_capacity(items.len());
165 for item in items {
166 let s = item.as_str().ok_or_else(|| {
167 RuntimeError::InvalidInput("tags must be an array of strings".into())
168 })?;
169 tags.push(s.to_string());
170 }
171 Ok(Some(tags))
172 }
173 Some(_) => Err(RuntimeError::InvalidInput(
174 "tags must be an array of strings".into(),
175 )),
176 }
177}
178
179fn optional_f64(args: &Value, key: &str) -> RuntimeResult<Option<f64>> {
180 match obj(args)?.get(key) {
181 None => Ok(None),
182 Some(Value::Null) => Ok(None),
183 Some(v) => v
184 .as_f64()
185 .map(Some)
186 .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
187 }
188}
189
190fn optional_f64_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<f64>>> {
196 match obj(args)?.get(key) {
197 None => Ok(None),
198 Some(Value::Null) => Ok(Some(None)),
199 Some(v) => v
200 .as_f64()
201 .map(|f| Some(Some(f)))
202 .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
203 }
204}
205
206fn vector_table_names(runtime: &KhiveRuntime) -> Vec<String> {
211 runtime
212 .registered_embedding_model_names()
213 .iter()
214 .map(|name| format!("vec_{}", crate::config::sanitize_key(name)))
215 .collect()
216}
217
218fn purge_index_row_statement(
222 table: &str,
223 namespace: &str,
224 subject_id: Uuid,
225 label: &str,
226) -> PlanStatement {
227 PlanStatement {
228 statement: SqlStatement {
229 sql: format!("DELETE FROM {table} WHERE namespace = ?1 AND subject_id = ?2"),
230 params: vec![
231 SqlValue::Text(namespace.to_string()),
232 SqlValue::Text(subject_id.to_string()),
233 ],
234 label: Some(label.to_string()),
235 },
236 guard: None,
237 }
238}
239
240async fn vector_table_exists(runtime: &KhiveRuntime, table: &str) -> RuntimeResult<bool> {
245 let mut reader = runtime
246 .sql()
247 .reader()
248 .await
249 .map_err(RuntimeError::Storage)?;
250 let row = reader
251 .query_scalar(SqlStatement {
252 sql: "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1".to_string(),
253 params: vec![SqlValue::Text(table.to_string())],
254 label: Some("atomic-delete-vec-table-exists".to_string()),
255 })
256 .await
257 .map_err(RuntimeError::Storage)?;
258 Ok(row.is_some())
259}
260
261async fn push_index_purge_statements(
277 runtime: &KhiveRuntime,
278 statements: &mut Vec<PlanStatement>,
279 fts_table: &str,
280 namespace: &str,
281 subject_id: Uuid,
282 label_prefix: &str,
283) -> RuntimeResult<()> {
284 statements.push(purge_index_row_statement(
285 fts_table,
286 namespace,
287 subject_id,
288 &format!("{label_prefix}-purge-fts"),
289 ));
290 for vec_table in vector_table_names(runtime) {
291 if vector_table_exists(runtime, &vec_table).await? {
292 statements.push(purge_index_row_statement(
293 &vec_table,
294 namespace,
295 subject_id,
296 &format!("{label_prefix}-purge-vec-{vec_table}"),
297 ));
298 }
299 }
300 Ok(())
301}
302
303fn event_append_statements(
332 namespace: &str,
333 verb: &str,
334 kind: EventKind,
335 substrate: SubstrateKind,
336 target_id: Uuid,
337 payload: Value,
338) -> RuntimeResult<Vec<PlanStatement>> {
339 let event = khive_storage::event::Event::new(namespace.to_string(), verb, kind, substrate, "")
340 .with_target(target_id)
341 .with_payload(payload);
342 let statements = event_insert_statements(&event)
343 .map_err(|e| RuntimeError::Internal(format!("event_insert_statements: {e}")))?;
344 Ok(statements
345 .into_iter()
346 .map(|statement| PlanStatement {
347 statement,
348 guard: None,
349 })
350 .collect())
351}
352
353pub async fn prepare_op(
363 runtime: &KhiveRuntime,
364 token: &NamespaceToken,
365 tool: &str,
366 args: &Value,
367) -> RuntimeResult<AtomicOpPlan> {
368 match tool {
369 "update" => prepare_update(runtime, token, args, None).await,
378 "delete" => prepare_delete(runtime, token, args, None).await,
387 "link" => prepare_link(runtime, token, args).await,
388 "merge" => prepare_merge(runtime, token, args).await,
389 "propose" | "review" | "withdraw" => prepare_governance_unimplemented(tool),
390 other => Err(RuntimeError::InvalidInput(format!(
391 "{other:?} has no atomic_prepare::prepare_op implementation; the CLI \
392 admissibility check should have rejected this before prepare"
393 ))),
394 }
395}
396
397fn prepare_governance_unimplemented(tool: &str) -> RuntimeResult<AtomicOpPlan> {
398 Err(RuntimeError::InvalidInput(format!(
399 "{tool:?} is on the ADR-099 v1 admissible verb list but has no --atomic \
400 prepare/apply implementation yet: its lifecycle (ADR-046) is an \
401 event-sourced changeset-interpreter over a dedicated `proposals_open` \
402 table, not a small guarded-DML plan — a faithful non-stub atomic \
403 prepare for it is tracked as ADR-099 follow-up work, not implemented \
404 in slice B3. No write was attempted."
405 )))
406}
407
408fn reject_inapplicable_update_fields(args: &Value, substrate: &str) -> RuntimeResult<()> {
423 let o = obj(args)?;
424 let present = |k: &str| o.get(k).is_some_and(|v| !v.is_null());
425 let (bad_field, valid): (Option<&str>, &str) = match substrate {
426 "entity" => {
427 let bad = if present("content") {
428 Some("content")
429 } else if present("salience") {
430 Some("salience")
431 } else if present("decay_factor") {
432 Some("decay_factor")
433 } else if present("relation") {
434 Some("relation")
435 } else if present("weight") {
436 Some("weight")
437 } else {
438 None
439 };
440 (bad, "name, description, tags, properties")
441 }
442 "note" => {
443 let bad = if present("description") {
444 Some("description")
445 } else if present("tags") {
446 Some("tags")
447 } else if present("relation") {
448 Some("relation")
449 } else if present("weight") {
450 Some("weight")
451 } else {
452 None
453 };
454 (bad, "name, content, salience, decay_factor, properties")
455 }
456 "edge" => {
462 let bad = if present("name") {
463 Some("name")
464 } else if present("description") {
465 Some("description")
466 } else if present("content") {
467 Some("content")
468 } else if present("tags") {
469 Some("tags")
470 } else if present("salience") {
471 Some("salience")
472 } else if present("decay_factor") {
473 Some("decay_factor")
474 } else {
475 None
476 };
477 (bad, "relation, weight, properties")
478 }
479 _ => (None, ""),
480 };
481 if let Some(field) = bad_field {
482 let substrate_label = match substrate {
483 "entity" => "an entity",
484 "note" => "a note",
485 "edge" => "an edge",
486 other => other,
487 };
488 return Err(RuntimeError::InvalidInput(format!(
489 "field '{field}' is not valid for {substrate_label}; valid fields: {valid}"
490 )));
491 }
492 Ok(())
493}
494
495pub enum AtomicUpdateKind {
505 Entity { specific: Option<String> },
506 Note { specific: Option<String> },
507 Edge,
508}
509
510pub async fn prepare_update(
516 runtime: &KhiveRuntime,
517 token: &NamespaceToken,
518 args: &Value,
519 expected_kind: Option<AtomicUpdateKind>,
520) -> RuntimeResult<AtomicOpPlan> {
521 let id = require_uuid(args, "id")?;
522
523 if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
527 return Err(RuntimeError::InvalidInput(
528 "entity_kind is immutable; to change kind, delete then re-create the entity, \
529 or use merge() if this is a deduplication correction"
530 .into(),
531 ));
532 }
533
534 match runtime.resolve_by_id(token, id).await? {
535 Some(Resolved::Entity(entity)) => {
536 match &expected_kind {
537 None => {}
538 Some(AtomicUpdateKind::Entity {
539 specific: Some(expected),
540 }) => {
541 if &entity.kind != expected {
542 return Err(RuntimeError::NotFound(format!("entity {id}")));
543 }
544 }
545 Some(AtomicUpdateKind::Entity { specific: None }) => {}
546 Some(AtomicUpdateKind::Note { .. }) => {
547 return Err(RuntimeError::NotFound(format!("note {id}")));
548 }
549 Some(AtomicUpdateKind::Edge) => {
550 return Err(RuntimeError::NotFound(format!("edge {id}")));
551 }
552 }
553 reject_inapplicable_update_fields(args, "entity")?;
559 let name = entity_name_patch(args)?;
560 let description = optional_string_patch(args, "description")?;
561 let properties = optional_properties(args, "properties")?;
562 let tags = optional_tags(args)?;
563
564 let (entity, text_changed, changed_fields) = runtime
565 .prepare_update_entity(
566 token,
567 id,
568 crate::curation::EntityPatch {
569 name,
570 description,
571 properties,
572 tags,
573 },
574 )
575 .await?;
576
577 let mut statements = vec![PlanStatement {
578 statement: entity_upsert_statement(&entity),
579 guard: Some(AffectedRowGuard::exactly(1)),
580 }];
581 statements.extend(event_append_statements(
586 &entity.namespace,
587 "update",
588 EventKind::EntityUpdated,
589 SubstrateKind::Entity,
590 id,
591 serde_json::json!({
592 "id": id,
593 "namespace": entity.namespace,
594 "changed_fields": changed_fields,
595 }),
596 )?);
597 let post_commit = if text_changed {
598 PostCommitEffect::ReindexEntity { entity_id: id }
599 } else {
600 PostCommitEffect::None
601 };
602 Ok(AtomicOpPlan::Update(UpdatePlan {
603 target_id: id,
604 statements,
605 post_commit,
606 edge_natural_key: None,
607 }))
608 }
609 Some(Resolved::Note(note)) => {
610 match &expected_kind {
611 None => {}
612 Some(AtomicUpdateKind::Note {
613 specific: Some(expected),
614 }) => {
615 if ¬e.kind != expected {
616 return Err(RuntimeError::NotFound(format!("note {id}")));
617 }
618 }
619 Some(AtomicUpdateKind::Note { specific: None }) => {}
620 Some(AtomicUpdateKind::Entity { .. }) => {
621 return Err(RuntimeError::NotFound(format!("entity {id}")));
622 }
623 Some(AtomicUpdateKind::Edge) => {
624 return Err(RuntimeError::NotFound(format!("edge {id}")));
625 }
626 }
627 reject_inapplicable_update_fields(args, "note")?;
634 let name = optional_string_patch(args, "name")?;
635 let content = optional_str(args, "content").map(|s| s.to_string());
636 let properties = optional_properties(args, "properties")?;
637 let salience = optional_f64_patch(args, "salience")?;
638 let decay_factor = optional_f64_patch(args, "decay_factor")?;
639
640 let (note, text_changed) = runtime
641 .prepare_update_note(
642 token,
643 id,
644 crate::curation::NotePatch::new(
645 name,
646 content,
647 salience,
648 decay_factor,
649 properties,
650 ),
651 )
652 .await?;
653
654 let post_commit = if text_changed {
655 PostCommitEffect::ReindexNote { note_id: id }
656 } else {
657 PostCommitEffect::None
658 };
659 Ok(AtomicOpPlan::Update(UpdatePlan {
660 target_id: id,
661 statements: vec![PlanStatement {
662 statement: note_upsert_statement(¬e),
663 guard: Some(AffectedRowGuard::exactly(1)),
664 }],
665 post_commit,
666 edge_natural_key: None,
667 }))
668 }
669 Some(_) => Err(RuntimeError::InvalidInput(format!(
670 "update target {id} must be an entity, note, or edge"
671 ))),
672 None => match &expected_kind {
679 Some(AtomicUpdateKind::Entity { .. }) => {
680 Err(RuntimeError::NotFound(format!("entity/note {id}")))
681 }
682 Some(AtomicUpdateKind::Note { .. }) => {
683 Err(RuntimeError::NotFound(format!("entity/note {id}")))
684 }
685 Some(AtomicUpdateKind::Edge) | None => match runtime.get_edge(token, id).await? {
686 Some(edge) => prepare_update_edge(runtime, id, edge, args).await,
687 None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
688 },
689 },
690 }
691}
692
693async fn prepare_update_edge(
729 runtime: &KhiveRuntime,
730 id: Uuid,
731 mut edge: khive_storage::types::Edge,
732 args: &Value,
733) -> RuntimeResult<AtomicOpPlan> {
734 reject_inapplicable_update_fields(args, "edge")?;
735
736 let relation_raw = optional_str(args, "relation");
737 let weight = optional_f64(args, "weight")?;
738 let properties = optional_properties(args, "properties")?;
739
740 if let Some(ref p) = properties {
741 crate::secret_gate::check_json(p)?;
742 }
743
744 let namespace = edge.namespace.clone();
745 let record_tok = NamespaceToken::for_namespace(
746 khive_types::Namespace::parse(&namespace)
747 .map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
748 );
749
750 let mut changed_fields: Vec<&'static str> = Vec::new();
751 if let Some(raw) = relation_raw {
752 let relation = parse_edge_relation(raw)?;
753 runtime
754 .validate_edge_relation_endpoints(&record_tok, edge.source_id, edge.target_id, relation)
755 .await?;
756 edge.relation = relation;
757 changed_fields.push("relation");
758 }
759 if let Some(w) = weight {
760 if !w.is_finite() || !(0.0..=1.0).contains(&w) {
761 return Err(RuntimeError::InvalidInput(format!(
762 "edge weight must be a finite value in [0.0, 1.0]; got {w}"
763 )));
764 }
765 edge.weight = w;
766 changed_fields.push("weight");
767 }
768 if let Some(p) = properties {
769 edge.metadata = Some(p);
770 changed_fields.push("properties");
771 }
772
773 let (canon_src, canon_tgt) =
774 canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
775 let now = chrono::Utc::now();
776
777 let mut statements: Vec<PlanStatement> = Vec::new();
778 let mut edge_natural_key: Option<EdgeNaturalKey> = None;
779
780 if edge.relation.is_symmetric() {
781 let metadata_str = edge
791 .metadata
792 .as_ref()
793 .map(|v| serde_json::to_string(v).unwrap_or_default());
794
795 statements.push(PlanStatement {
796 statement: edge_symmetric_delete_if_conflict_statement(
797 &namespace,
798 id,
799 canon_src,
800 canon_tgt,
801 edge.relation,
802 ),
803 guard: Some(AffectedRowGuard {
804 expected_min: 0,
805 expected_max: Some(1),
806 }),
807 });
808 statements.push(PlanStatement {
809 statement: edge_symmetric_refresh_or_update_inplace_statement(
810 &namespace,
811 id,
812 canon_src,
813 canon_tgt,
814 edge.relation,
815 edge.weight,
816 now.timestamp_micros(),
817 metadata_str.as_deref(),
818 edge.target_backend.as_deref(),
819 ),
820 guard: Some(AffectedRowGuard::exactly(1)),
821 });
822
823 edge_natural_key = Some(EdgeNaturalKey {
828 namespace: namespace.clone(),
829 canon_source_id: canon_src,
830 canon_target_id: canon_tgt,
831 relation: edge.relation,
832 });
833 } else {
834 edge.updated_at = now;
837 statements.push(PlanStatement {
838 statement: edge_upsert_statement(&edge),
839 guard: Some(AffectedRowGuard::exactly(1)),
840 });
841 }
842
843 statements.extend(event_append_statements(
848 &namespace,
849 "update",
850 EventKind::EdgeUpdated,
851 SubstrateKind::Entity,
852 id,
853 serde_json::json!({"id": id, "namespace": namespace, "changed_fields": changed_fields}),
854 )?);
855
856 Ok(AtomicOpPlan::Update(UpdatePlan {
857 target_id: id,
858 statements,
859 post_commit: PostCommitEffect::None,
860 edge_natural_key,
861 }))
862}
863
864pub enum AtomicDeleteKind {
881 Entity { specific: Option<String> },
882 Note { specific: Option<String> },
883 Edge,
884}
885
886pub async fn prepare_delete(
892 runtime: &KhiveRuntime,
893 token: &NamespaceToken,
894 args: &Value,
895 expected_kind: Option<AtomicDeleteKind>,
896) -> RuntimeResult<AtomicOpPlan> {
897 let id = require_uuid(args, "id")?;
898 let hard = obj(args)?
899 .get("hard")
900 .and_then(|v| v.as_bool())
901 .unwrap_or(false);
902
903 let resolved = if hard {
909 runtime.resolve_by_id_including_deleted(token, id).await?
910 } else {
911 runtime.resolve_by_id(token, id).await?
912 };
913
914 match resolved {
915 Some(Resolved::Entity(entity)) => {
916 match &expected_kind {
917 None => {}
918 Some(AtomicDeleteKind::Entity {
919 specific: Some(expected),
920 }) => {
921 if &entity.kind != expected {
922 return Err(RuntimeError::NotFound(format!("{expected} {id}")));
923 }
924 }
925 Some(AtomicDeleteKind::Entity { specific: None }) => {}
926 Some(AtomicDeleteKind::Note { .. }) => {
927 return Err(RuntimeError::NotFound(format!("note {id}")));
928 }
929 Some(AtomicDeleteKind::Edge) => {
930 return Err(RuntimeError::NotFound(format!("edge {id}")));
931 }
932 }
933 let namespace = entity.namespace.clone();
934 let mut statements = if hard {
939 vec![PlanStatement {
940 statement: entity_hard_delete_statement(id),
941 guard: Some(AffectedRowGuard::exactly(1)),
942 }]
943 } else {
944 let deleted_at = chrono::Utc::now().timestamp_micros();
945 vec![PlanStatement {
946 statement: entity_soft_delete_statement(id, deleted_at),
947 guard: Some(AffectedRowGuard::exactly(1)),
948 }]
949 };
950 if hard {
951 statements.push(PlanStatement {
954 statement: purge_incident_edges_statement(id),
955 guard: None,
956 });
957 }
958 push_index_purge_statements(
963 runtime,
964 &mut statements,
965 "fts_entities",
966 &namespace,
967 id,
968 "atomic-delete-entity",
969 )
970 .await?;
971 statements.extend(event_append_statements(
977 &namespace,
978 "delete",
979 EventKind::EntityDeleted,
980 SubstrateKind::Entity,
981 id,
982 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
983 )?);
984 Ok(AtomicOpPlan::Delete(DeletePlan {
985 target_id: id,
986 statements,
987 post_commit: PostCommitEffect::None,
988 }))
989 }
990 Some(Resolved::Note(note)) => {
991 match &expected_kind {
992 None => {}
993 Some(AtomicDeleteKind::Note {
994 specific: Some(expected),
995 }) => {
996 if ¬e.kind != expected {
997 return Err(RuntimeError::NotFound(format!("{expected} {id}")));
998 }
999 }
1000 Some(AtomicDeleteKind::Note { specific: None }) => {}
1001 Some(AtomicDeleteKind::Entity { .. }) => {
1002 return Err(RuntimeError::NotFound(format!("entity {id}")));
1003 }
1004 Some(AtomicDeleteKind::Edge) => {
1005 return Err(RuntimeError::NotFound(format!("edge {id}")));
1006 }
1007 }
1008 let namespace = note.namespace.clone();
1009 let mut statements = if hard {
1013 vec![PlanStatement {
1014 statement: note_hard_delete_statement(id),
1015 guard: Some(AffectedRowGuard::exactly(1)),
1016 }]
1017 } else {
1018 let deleted_at = chrono::Utc::now().timestamp_micros();
1019 vec![PlanStatement {
1020 statement: note_soft_delete_statement(id, deleted_at),
1021 guard: Some(AffectedRowGuard::exactly(1)),
1022 }]
1023 };
1024 if hard {
1025 statements.push(PlanStatement {
1026 statement: purge_incident_edges_statement(id),
1027 guard: None,
1028 });
1029 }
1030 push_index_purge_statements(
1035 runtime,
1036 &mut statements,
1037 "fts_notes",
1038 &namespace,
1039 id,
1040 "atomic-delete-note",
1041 )
1042 .await?;
1043 statements.extend(event_append_statements(
1047 &namespace,
1048 "delete",
1049 EventKind::NoteDeleted,
1050 SubstrateKind::Note,
1051 id,
1052 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
1053 )?);
1054 Ok(AtomicOpPlan::Delete(DeletePlan {
1055 target_id: id,
1056 statements,
1057 post_commit: PostCommitEffect::NoteDeleted {
1062 note_id: id,
1063 kind: note.kind.clone(),
1064 },
1065 }))
1066 }
1067 Some(_) => Err(RuntimeError::InvalidInput(format!(
1068 "delete target {id} must be an entity, note, or edge"
1069 ))),
1070 None => match &expected_kind {
1074 Some(AtomicDeleteKind::Entity { .. }) => {
1075 Err(RuntimeError::NotFound(format!("entity/note {id}")))
1076 }
1077 Some(AtomicDeleteKind::Note { .. }) => {
1078 Err(RuntimeError::NotFound(format!("entity/note {id}")))
1079 }
1080 Some(AtomicDeleteKind::Edge) | None => {
1081 let edge = if hard {
1082 runtime.get_edge_including_deleted(token, id).await?
1083 } else {
1084 runtime.get_edge(token, id).await?
1085 };
1086 match edge {
1087 Some(edge) => prepare_delete_edge(id, edge, hard).await,
1088 None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
1089 }
1090 }
1091 },
1092 }
1093}
1094
1095async fn prepare_delete_edge(
1104 id: Uuid,
1105 edge: khive_storage::types::Edge,
1106 hard: bool,
1107) -> RuntimeResult<AtomicOpPlan> {
1108 let namespace = edge.namespace.clone();
1109 let mut statements: Vec<PlanStatement> = Vec::new();
1110
1111 if hard {
1112 statements.push(PlanStatement {
1117 statement: purge_incident_edges_statement(id),
1118 guard: None,
1119 });
1120 statements.push(PlanStatement {
1121 statement: edge_hard_delete_statement(id),
1122 guard: Some(AffectedRowGuard::exactly(1)),
1123 });
1124 } else {
1125 let now = chrono::Utc::now().timestamp_micros();
1126 statements.push(PlanStatement {
1127 statement: edge_soft_delete_statement(id, now),
1128 guard: Some(AffectedRowGuard::exactly(1)),
1129 });
1130 }
1131
1132 statements.extend(event_append_statements(
1133 &namespace,
1134 "delete",
1135 EventKind::EdgeDeleted,
1136 SubstrateKind::Entity,
1137 id,
1138 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
1139 )?);
1140
1141 Ok(AtomicOpPlan::Delete(DeletePlan {
1142 target_id: id,
1143 statements,
1144 post_commit: PostCommitEffect::None,
1145 }))
1146}
1147
1148fn parse_edge_relation(raw: &str) -> RuntimeResult<EdgeRelation> {
1153 raw.parse::<EdgeRelation>()
1154 .map_err(|e| RuntimeError::InvalidInput(format!("unknown edge relation {raw:?}: {e}")))
1155}
1156
1157async fn prepare_link(
1158 runtime: &KhiveRuntime,
1159 token: &NamespaceToken,
1160 args: &Value,
1161) -> RuntimeResult<AtomicOpPlan> {
1162 let source_id = require_uuid(args, "source_id")?;
1163 let target_id = require_uuid(args, "target_id")?;
1164 let relation = parse_edge_relation(require_str(args, "relation")?)?;
1165 let weight = optional_f64(args, "weight")?.unwrap_or(1.0);
1166 let metadata = obj(args)?.get("metadata").cloned();
1167
1168 let mut metadata = crate::merge_entry_metadata(
1174 metadata,
1175 optional_str(args, "dependency_kind").map(String::from),
1176 )?;
1177
1178 validate_edge_weight(weight)?;
1179 runtime
1180 .validate_edge_relation_endpoints(token, source_id, target_id, relation)
1181 .await?;
1182
1183 let (canon_source, canon_target) = canonical_edge_endpoints(relation, source_id, target_id);
1184
1185 if relation == EdgeRelation::DependsOn {
1192 metadata = match (
1193 runtime.resolve_edge_endpoint(token, canon_source).await?,
1194 runtime.resolve_edge_endpoint(token, canon_target).await?,
1195 ) {
1196 (Some(Resolved::Entity(src_e)), Some(Resolved::Entity(tgt_e))) => {
1197 merge_dependency_kind(&src_e.kind, &tgt_e.kind, metadata)
1198 }
1199 _ => metadata,
1200 };
1201 }
1202
1203 validate_edge_metadata(relation, metadata.as_ref())?;
1204 let edge_id = Uuid::new_v4();
1205 let namespace = token.namespace().as_str().to_string();
1206 let now = chrono::Utc::now().timestamp_micros();
1207 let metadata_str = metadata.map(|m| serde_json::to_string(&m).unwrap_or_default());
1208
1209 let statement = edge_insert_guarded_by_endpoints_statement(
1222 &namespace,
1223 edge_id,
1224 canon_source,
1225 canon_target,
1226 relation,
1227 weight,
1228 now,
1229 metadata_str.as_deref(),
1230 );
1231
1232 Ok(AtomicOpPlan::Link(LinkPlan {
1233 source_id: canon_source,
1234 target_id: canon_target,
1235 statement: PlanStatement {
1236 statement,
1237 guard: Some(AffectedRowGuard::exactly(1)),
1238 },
1239 }))
1240}
1241
1242async fn prepare_merge(
1256 runtime: &KhiveRuntime,
1257 token: &NamespaceToken,
1258 args: &Value,
1259) -> RuntimeResult<AtomicOpPlan> {
1260 let into_id = require_uuid(args, "into_id")?;
1261 let from_id = require_uuid(args, "from_id")?;
1262 if into_id == from_id {
1263 return Err(RuntimeError::InvalidInput(
1264 "cannot merge an entity into itself".into(),
1265 ));
1266 }
1267
1268 let entities = runtime.entities(token)?;
1269 entities
1270 .get_entity(into_id)
1271 .await?
1272 .ok_or_else(|| RuntimeError::NotFound(format!("entity {into_id}")))?;
1273 entities
1274 .get_entity(from_id)
1275 .await?
1276 .ok_or_else(|| RuntimeError::NotFound(format!("entity {from_id}")))?;
1277
1278 let now = chrono::Utc::now().timestamp_micros();
1279 let rewires = vec![
1280 crate::atomic_plan::PlanPredicate {
1281 description: "source_id = :from".to_string(),
1282 statement: SqlStatement {
1283 sql: "UPDATE graph_edges SET source_id = ?1, updated_at = ?2 WHERE source_id = ?3"
1284 .to_string(),
1285 params: vec![
1286 SqlValue::Text(into_id.to_string()),
1287 SqlValue::Integer(now),
1288 SqlValue::Text(from_id.to_string()),
1289 ],
1290 label: Some("atomic-merge-rewire-source".to_string()),
1291 },
1292 },
1293 crate::atomic_plan::PlanPredicate {
1294 description: "target_id = :from".to_string(),
1295 statement: SqlStatement {
1296 sql: "UPDATE graph_edges SET target_id = ?1, updated_at = ?2 WHERE target_id = ?3"
1297 .to_string(),
1298 params: vec![
1299 SqlValue::Text(into_id.to_string()),
1300 SqlValue::Integer(now),
1301 SqlValue::Text(from_id.to_string()),
1302 ],
1303 label: Some("atomic-merge-rewire-target".to_string()),
1304 },
1305 },
1306 ];
1307 let lifecycle = vec![PlanStatement {
1308 statement: SqlStatement {
1309 sql: "UPDATE entities SET deleted_at = ?1, merged_into = ?2 \
1310 WHERE id = ?3 AND deleted_at IS NULL"
1311 .to_string(),
1312 params: vec![
1313 SqlValue::Integer(now),
1314 SqlValue::Text(into_id.to_string()),
1315 SqlValue::Text(from_id.to_string()),
1316 ],
1317 label: Some("atomic-merge-tombstone-from-entity".to_string()),
1318 },
1319 guard: Some(AffectedRowGuard::exactly(1)),
1320 }];
1321
1322 Ok(AtomicOpPlan::Merge(MergePlan {
1323 into_id,
1324 from_id,
1325 rewires,
1326 lifecycle,
1327 }))
1328}
1329
1330pub async fn apply_post_commit_effects(
1339 runtime: &KhiveRuntime,
1340 token: &NamespaceToken,
1341 effects: Vec<PostCommitEffect>,
1342) -> RuntimeResult<()> {
1343 for effect in effects {
1344 match effect {
1345 PostCommitEffect::None => {}
1346 PostCommitEffect::ReindexEntity { entity_id } => {
1347 if let Some(entity) = runtime.entities(token)?.get_entity(entity_id).await? {
1348 runtime.reindex_entity(token, &entity).await?;
1349 }
1350 }
1351 PostCommitEffect::ReindexNote { note_id } => {
1352 if let Some(note) = runtime.notes(token)?.get_note(note_id).await? {
1353 runtime.reindex_note(token, ¬e).await?;
1354 runtime.fire_note_mutation_hook(¬e.kind, note.id).await;
1362 }
1363 }
1364 PostCommitEffect::NoteDeleted { note_id, kind } => {
1365 runtime.fire_note_mutation_hook(&kind, note_id).await;
1374 }
1375 PostCommitEffect::GtdAudit { .. } => {
1376 }
1383 }
1384 }
1385 Ok(())
1386}
1387
1388#[cfg(test)]
1389mod tests {
1390 use super::*;
1391
1392 use async_trait::async_trait;
1393 use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService};
1394 use serde_json::json;
1395
1396 use khive_types::Namespace;
1397
1398 use crate::embedder_registry::EmbedderProvider;
1399 use crate::runtime::RuntimeConfig;
1400
1401 const STUB_MODEL: &str = "stub-adr099-b3";
1402 const STUB_DIMS: usize = 4;
1403
1404 struct StubService;
1405
1406 #[async_trait]
1407 impl EmbeddingService for StubService {
1408 async fn embed(
1409 &self,
1410 texts: &[String],
1411 _model: EmbeddingModel,
1412 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1413 Ok(texts.iter().map(|_| vec![0.5_f32; STUB_DIMS]).collect())
1414 }
1415
1416 fn supports_model(&self, _model: EmbeddingModel) -> bool {
1417 true
1418 }
1419
1420 fn name(&self) -> &'static str {
1421 STUB_MODEL
1422 }
1423 }
1424
1425 struct StubProvider;
1426
1427 #[async_trait]
1428 impl EmbedderProvider for StubProvider {
1429 fn name(&self) -> &str {
1430 STUB_MODEL
1431 }
1432
1433 fn dimensions(&self) -> usize {
1434 STUB_DIMS
1435 }
1436
1437 async fn build(&self) -> RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
1438 Ok(std::sync::Arc::new(StubService))
1439 }
1440 }
1441
1442 fn scratch_runtime() -> KhiveRuntime {
1443 let dir = tempfile::tempdir().expect("tempdir");
1444 let path = dir.path().join("atomic_prepare_reindex.db");
1445 let rt = KhiveRuntime::new(RuntimeConfig {
1446 db_path: Some(path),
1447 embedding_model: None,
1448 additional_embedding_models: vec![],
1449 ..RuntimeConfig::default()
1450 })
1451 .expect("runtime");
1452 std::mem::forget(dir);
1453 rt
1454 }
1455
1456 #[tokio::test]
1464 async fn atomic_update_entity_rejects_note_only_field_salience() {
1465 let runtime = scratch_runtime();
1466 let token = runtime
1467 .authorize(Namespace::parse("local").expect("ns"))
1468 .expect("authorize");
1469 let entity = khive_storage::Entity::new("local", "concept", "GapFourEntity");
1470 let entity_id = entity.id;
1471 runtime
1472 .entities(&token)
1473 .expect("entities store")
1474 .upsert_entity(entity)
1475 .await
1476 .expect("seed entity");
1477
1478 let err = prepare_update(
1479 &runtime,
1480 &token,
1481 &json!({"id": entity_id.to_string(), "salience": 0.9}),
1482 None,
1483 )
1484 .await
1485 .expect_err("salience on an entity must be rejected, not silently accepted");
1486 assert!(
1487 matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("salience") && msg.contains("not valid for an entity")),
1488 "expected an InvalidInput naming the offending field, got: {err:?}"
1489 );
1490
1491 let plan = prepare_update(
1493 &runtime,
1494 &token,
1495 &json!({
1496 "id": entity_id.to_string(),
1497 "name": "GapFourEntity-renamed",
1498 "description": "updated description",
1499 "tags": ["a", "b"],
1500 }),
1501 None,
1502 )
1503 .await
1504 .expect("a valid entity field set must still be accepted");
1505 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1506 .await
1507 .expect("seam call ok");
1508 assert!(matches!(
1509 outcome,
1510 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1511 ));
1512 let entity = runtime
1513 .get_entity(&token, entity_id)
1514 .await
1515 .expect("get_entity");
1516 assert_eq!(entity.name, "GapFourEntity-renamed");
1517 }
1518
1519 #[tokio::test]
1523 async fn atomic_update_note_rejects_entity_only_field_description() {
1524 let runtime = scratch_runtime();
1525 let token = runtime
1526 .authorize(Namespace::parse("local").expect("ns"))
1527 .expect("authorize");
1528 let mut note = khive_storage::note::Note::new("local", "observation", "gap-4 note content");
1529 note.name = Some("gap-four-note".to_string());
1530 let note_id = note.id;
1531 runtime
1532 .notes(&token)
1533 .expect("notes store")
1534 .upsert_note(note)
1535 .await
1536 .expect("seed note");
1537
1538 let err = prepare_update(
1539 &runtime,
1540 &token,
1541 &json!({"id": note_id.to_string(), "description": "entities have descriptions, notes don't"}),
1542 None,
1543 )
1544 .await
1545 .expect_err("description on a note must be rejected, not silently accepted");
1546 assert!(
1547 matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("description") && msg.contains("not valid for a note")),
1548 "expected an InvalidInput naming the offending field, got: {err:?}"
1549 );
1550
1551 let plan = prepare_update(
1553 &runtime,
1554 &token,
1555 &json!({"id": note_id.to_string(), "content": "gap-4 note content, revised"}),
1556 None,
1557 )
1558 .await
1559 .expect("a valid note field must still be accepted");
1560 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1561 .await
1562 .expect("seam call ok");
1563 assert!(matches!(
1564 outcome,
1565 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1566 ));
1567 }
1568
1569 #[tokio::test]
1574 async fn atomic_update_note_content_is_fts_and_vector_reindexed_post_commit() {
1575 let runtime = scratch_runtime();
1576 runtime.register_embedder(StubProvider);
1577 let token = runtime
1578 .authorize(Namespace::parse("local").expect("ns"))
1579 .expect("authorize");
1580
1581 let mut note = khive_storage::note::Note::new("local", "observation", "original content");
1582 note.name = Some("reindex-target".to_string());
1583 let note_id = note.id;
1584 runtime
1585 .notes(&token)
1586 .expect("notes store")
1587 .upsert_note(note)
1588 .await
1589 .expect("seed note");
1590
1591 let vec_store = runtime
1593 .vectors_for_model(&token, STUB_MODEL)
1594 .expect("vec store");
1595 assert_eq!(vec_store.count().await.expect("count before"), 0);
1596
1597 let plan = prepare_update(
1598 &runtime,
1599 &token,
1600 &json!({"id": note_id.to_string(), "content": "freshly-updated-content-xyz"}),
1601 None,
1602 )
1603 .await
1604 .expect("prepare update");
1605
1606 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1607 .await
1608 .expect("seam call ok");
1609 let post_commit = match outcome {
1610 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1611 other => panic!("expected Committed, got {other:?}"),
1612 };
1613 assert_eq!(
1614 post_commit,
1615 vec![PostCommitEffect::ReindexNote { note_id }],
1616 "content change must schedule exactly one ReindexNote post-commit effect"
1617 );
1618
1619 apply_post_commit_effects(&runtime, &token, post_commit)
1620 .await
1621 .expect("apply post-commit effects");
1622
1623 let doc = runtime
1625 .text_for_notes(&token)
1626 .expect("text store")
1627 .get_document("local", note_id)
1628 .await
1629 .expect("get_document")
1630 .expect("document must be indexed after post-commit reindex");
1631 assert!(
1632 doc.body.contains("freshly-updated-content-xyz"),
1633 "FTS body must reflect the committed content: {:?}",
1634 doc.body
1635 );
1636
1637 assert_eq!(
1639 vec_store.count().await.expect("count after"),
1640 1,
1641 "post-commit reindex must have inserted a vector row for the stub model"
1642 );
1643 }
1644
1645 #[tokio::test]
1657 async fn atomic_note_update_and_delete_post_commit_fire_the_note_mutation_hook() {
1658 let runtime = scratch_runtime();
1659 let token = runtime
1660 .authorize(Namespace::parse("local").expect("ns"))
1661 .expect("authorize");
1662
1663 let fired: std::sync::Arc<std::sync::Mutex<Vec<(String, uuid::Uuid)>>> =
1664 std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
1665 let fired_for_hook = fired.clone();
1666 runtime.install_note_mutation_hook(std::sync::Arc::new(
1667 move |kind: String, id: uuid::Uuid| {
1668 let fired = fired_for_hook.clone();
1669 Box::pin(async move {
1670 fired.lock().expect("lock").push((kind, id));
1671 })
1672 },
1673 ));
1674
1675 let mut note = khive_storage::note::Note::new("local", "observation", "hook-update-target");
1677 note.name = Some("hook-update-target".to_string());
1678 let update_note_id = note.id;
1679 runtime
1680 .notes(&token)
1681 .expect("notes store")
1682 .upsert_note(note)
1683 .await
1684 .expect("seed update-target note");
1685
1686 let plan = prepare_update(
1687 &runtime,
1688 &token,
1689 &json!({"id": update_note_id.to_string(), "content": "hook-update-target, revised"}),
1690 None,
1691 )
1692 .await
1693 .expect("prepare update");
1694 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1695 .await
1696 .expect("seam call ok");
1697 let post_commit = match outcome {
1698 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1699 other => panic!("expected Committed, got {other:?}"),
1700 };
1701 apply_post_commit_effects(&runtime, &token, post_commit)
1702 .await
1703 .expect("apply post-commit effects (update)");
1704
1705 let mut del_note =
1708 khive_storage::note::Note::new("local", "observation", "hook-delete-target");
1709 del_note.name = Some("hook-delete-target".to_string());
1710 let delete_note_id = del_note.id;
1711 runtime
1712 .notes(&token)
1713 .expect("notes store")
1714 .upsert_note(del_note)
1715 .await
1716 .expect("seed delete-target note");
1717
1718 let plan = prepare_delete(
1719 &runtime,
1720 &token,
1721 &json!({"id": delete_note_id.to_string(), "hard": false}),
1722 None,
1723 )
1724 .await
1725 .expect("prepare delete");
1726 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1727 .await
1728 .expect("seam call ok");
1729 let post_commit = match outcome {
1730 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1731 other => panic!("expected Committed, got {other:?}"),
1732 };
1733 assert_eq!(
1734 post_commit,
1735 vec![PostCommitEffect::NoteDeleted {
1736 note_id: delete_note_id,
1737 kind: "observation".to_string(),
1738 }],
1739 "a committed note delete must schedule exactly one NoteDeleted post-commit effect"
1740 );
1741 apply_post_commit_effects(&runtime, &token, post_commit)
1742 .await
1743 .expect("apply post-commit effects (delete)");
1744
1745 let seen = fired.lock().expect("lock").clone();
1746 assert!(
1747 seen.contains(&("observation".to_string(), update_note_id)),
1748 "the note-mutation hook must fire for the atomic UPDATE path: {seen:?}"
1749 );
1750 assert!(
1751 seen.contains(&("observation".to_string(), delete_note_id)),
1752 "the note-mutation hook must fire for the atomic DELETE path: {seen:?}"
1753 );
1754 }
1755
1756 #[tokio::test]
1760 async fn atomic_delete_note_purges_fts_and_vector_indexes_soft_and_hard() {
1761 let runtime = scratch_runtime();
1762 runtime.register_embedder(StubProvider);
1763 let token = runtime
1764 .authorize(Namespace::parse("local").expect("ns"))
1765 .expect("authorize");
1766
1767 for hard in [false, true] {
1768 let mut note =
1769 khive_storage::note::Note::new("local", "observation", "purge-target content");
1770 note.name = Some(format!("purge-target-hard-{hard}"));
1771 let note_id = note.id;
1772 runtime
1773 .notes(&token)
1774 .expect("notes store")
1775 .upsert_note(note.clone())
1776 .await
1777 .expect("seed note");
1778 runtime
1779 .reindex_note(&token, ¬e)
1780 .await
1781 .expect("seed index rows");
1782
1783 let vec_store = runtime
1784 .vectors_for_model(&token, STUB_MODEL)
1785 .expect("vec store");
1786 assert_eq!(
1787 vec_store.count().await.expect("count before"),
1788 1,
1789 "seeded note must have a vector row before delete (hard={hard})"
1790 );
1791 assert!(
1792 runtime
1793 .text_for_notes(&token)
1794 .expect("text store")
1795 .get_document("local", note_id)
1796 .await
1797 .expect("get_document")
1798 .is_some(),
1799 "seeded note must have an FTS row before delete (hard={hard})"
1800 );
1801
1802 let plan = prepare_delete(
1803 &runtime,
1804 &token,
1805 &json!({"id": note_id.to_string(), "hard": hard}),
1806 None,
1807 )
1808 .await
1809 .expect("prepare delete");
1810 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1811 .await
1812 .expect("seam call ok");
1813 assert!(
1814 matches!(
1815 outcome,
1816 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1817 ),
1818 "expected commit (hard={hard}): {outcome:?}"
1819 );
1820
1821 assert!(
1822 runtime
1823 .text_for_notes(&token)
1824 .expect("text store")
1825 .get_document("local", note_id)
1826 .await
1827 .expect("get_document")
1828 .is_none(),
1829 "FTS row must be purged after atomic delete (hard={hard})"
1830 );
1831 assert_eq!(
1832 vec_store.count().await.expect("count after"),
1833 0,
1834 "vector row must be purged after atomic delete (hard={hard})"
1835 );
1836 }
1837 }
1838
1839 #[tokio::test]
1843 async fn atomic_delete_entity_purges_fts_and_vector_indexes_soft_and_hard() {
1844 let runtime = scratch_runtime();
1845 runtime.register_embedder(StubProvider);
1846 let token = runtime
1847 .authorize(Namespace::parse("local").expect("ns"))
1848 .expect("authorize");
1849
1850 for hard in [false, true] {
1851 let entity =
1852 khive_storage::Entity::new("local", "concept", format!("purge-target-hard-{hard}"));
1853 let entity_id = entity.id;
1854 runtime
1855 .entities(&token)
1856 .expect("entities store")
1857 .upsert_entity(entity.clone())
1858 .await
1859 .expect("seed entity");
1860 runtime
1861 .reindex_entity(&token, &entity)
1862 .await
1863 .expect("seed index rows");
1864
1865 let vec_store = runtime
1866 .vectors_for_model(&token, STUB_MODEL)
1867 .expect("vec store");
1868 assert_eq!(
1869 vec_store.count().await.expect("count before"),
1870 1,
1871 "seeded entity must have a vector row before delete (hard={hard})"
1872 );
1873 assert!(
1874 runtime
1875 .text(&token)
1876 .expect("text store")
1877 .get_document("local", entity_id)
1878 .await
1879 .expect("get_document")
1880 .is_some(),
1881 "seeded entity must have an FTS row before delete (hard={hard})"
1882 );
1883
1884 let plan = prepare_delete(
1885 &runtime,
1886 &token,
1887 &json!({"id": entity_id.to_string(), "hard": hard}),
1888 None,
1889 )
1890 .await
1891 .expect("prepare delete");
1892 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1893 .await
1894 .expect("seam call ok");
1895 assert!(
1896 matches!(
1897 outcome,
1898 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1899 ),
1900 "expected commit (hard={hard}): {outcome:?}"
1901 );
1902
1903 assert!(
1904 runtime
1905 .text(&token)
1906 .expect("text store")
1907 .get_document("local", entity_id)
1908 .await
1909 .expect("get_document")
1910 .is_none(),
1911 "FTS row must be purged after atomic delete (hard={hard})"
1912 );
1913 assert_eq!(
1914 vec_store.count().await.expect("count after"),
1915 0,
1916 "vector row must be purged after atomic delete (hard={hard})"
1917 );
1918 }
1919 }
1920
1921 #[tokio::test]
1927 async fn atomic_link_persists_explicit_dependency_kind_and_infers_when_absent() {
1928 let runtime = scratch_runtime();
1929 let token = runtime
1930 .authorize(Namespace::parse("local").expect("ns"))
1931 .expect("authorize");
1932 let entities = runtime.entities(&token).expect("entities store");
1933
1934 fn metadata_json(plan: &AtomicOpPlan) -> String {
1935 let link_plan = match plan {
1936 AtomicOpPlan::Link(p) => p,
1937 other => panic!("expected an AtomicOpPlan::Link, got {other:?}"),
1938 };
1939 match link_plan.statement.statement.params.last() {
1940 Some(SqlValue::Text(s)) => s.clone(),
1941 other => panic!("expected the metadata param to be SqlValue::Text, got {other:?}"),
1942 }
1943 }
1944
1945 {
1947 let svc = khive_storage::Entity::new("local", "service", "SvcA");
1948 let proj = khive_storage::Entity::new("local", "project", "ProjB");
1949 let (svc_id, proj_id) = (svc.id, proj.id);
1950 entities.upsert_entity(svc).await.expect("seed svc");
1951 entities.upsert_entity(proj).await.expect("seed proj");
1952
1953 let plan = prepare_link(
1954 &runtime,
1955 &token,
1956 &json!({
1957 "source_id": svc_id.to_string(),
1958 "target_id": proj_id.to_string(),
1959 "relation": "depends_on",
1960 "dependency_kind": "artifact",
1961 }),
1962 )
1963 .await
1964 .expect("prepare link");
1965 let json_str = metadata_json(&plan);
1966 assert!(
1967 json_str.contains(r#""dependency_kind":"artifact""#),
1968 "explicit dependency_kind param must persist: {json_str}"
1969 );
1970 }
1971
1972 {
1975 let svc_a = khive_storage::Entity::new("local", "service", "SvcC");
1976 let svc_b = khive_storage::Entity::new("local", "service", "SvcD");
1977 let (a_id, b_id) = (svc_a.id, svc_b.id);
1978 entities.upsert_entity(svc_a).await.expect("seed svc a");
1979 entities.upsert_entity(svc_b).await.expect("seed svc b");
1980
1981 let plan = prepare_link(
1982 &runtime,
1983 &token,
1984 &json!({
1985 "source_id": a_id.to_string(),
1986 "target_id": b_id.to_string(),
1987 "relation": "depends_on",
1988 }),
1989 )
1990 .await
1991 .expect("prepare link");
1992 let json_str = metadata_json(&plan);
1993 assert!(
1994 json_str.contains(r#""dependency_kind":"runtime""#),
1995 "inferred dependency_kind for (service, service) must persist: {json_str}"
1996 );
1997 }
1998 }
1999
2000 async fn probe_edge_natural_key(
2005 runtime: &KhiveRuntime,
2006 namespace: &str,
2007 source_id: Uuid,
2008 target_id: Uuid,
2009 relation: &str,
2010 ) -> (usize, Option<f64>, Option<String>, Option<i64>) {
2011 let mut reader = runtime.sql().reader().await.expect("reader");
2012 let rows = reader
2013 .query_all(SqlStatement {
2014 sql: "SELECT weight, metadata, deleted_at FROM graph_edges \
2015 WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 AND relation = ?4"
2016 .to_string(),
2017 params: vec![
2018 SqlValue::Text(namespace.to_string()),
2019 SqlValue::Text(source_id.to_string()),
2020 SqlValue::Text(target_id.to_string()),
2021 SqlValue::Text(relation.to_string()),
2022 ],
2023 label: Some("test-probe-edge-natural-key".to_string()),
2024 })
2025 .await
2026 .expect("probe edge natural key");
2027 let count = rows.len();
2028 let Some(row) = rows.into_iter().next() else {
2029 return (count, None, None, None);
2030 };
2031 let weight = match row.get("weight") {
2032 Some(SqlValue::Float(f)) => Some(*f),
2033 Some(SqlValue::Integer(i)) => Some(*i as f64),
2034 _ => None,
2035 };
2036 let metadata = match row.get("metadata") {
2037 Some(SqlValue::Text(s)) => Some(s.clone()),
2038 _ => None,
2039 };
2040 let deleted_at = match row.get("deleted_at") {
2041 Some(SqlValue::Integer(i)) => Some(*i),
2042 _ => None,
2043 };
2044 (count, weight, metadata, deleted_at)
2045 }
2046
2047 #[tokio::test]
2053 async fn atomic_link_of_already_linked_triple_upserts_weight_and_metadata() {
2054 let runtime = scratch_runtime();
2055 let token = runtime
2056 .authorize(Namespace::parse("local").expect("ns"))
2057 .expect("authorize");
2058 let entities = runtime.entities(&token).expect("entities store");
2059
2060 let a = khive_storage::Entity::new("local", "concept", "GapTwoA");
2061 let b = khive_storage::Entity::new("local", "concept", "GapTwoB");
2062 let (a_id, b_id) = (a.id, b.id);
2063 entities.upsert_entity(a).await.expect("seed a");
2064 entities.upsert_entity(b).await.expect("seed b");
2065
2066 let plan1 = prepare_link(
2068 &runtime,
2069 &token,
2070 &json!({
2071 "source_id": a_id.to_string(),
2072 "target_id": b_id.to_string(),
2073 "relation": "extends",
2074 "weight": 0.5,
2075 }),
2076 )
2077 .await
2078 .expect("prepare first link");
2079 let outcome1 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan1])
2080 .await
2081 .expect("seam call ok");
2082 assert!(
2083 matches!(
2084 outcome1,
2085 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2086 ),
2087 "fresh link must commit: {outcome1:?}"
2088 );
2089 let (count, weight, _metadata, deleted_at) =
2090 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2091 assert_eq!(count, 1, "exactly one edge row after the fresh link");
2092 assert_eq!(weight, Some(0.5));
2093 assert!(deleted_at.is_none());
2094
2095 let plan2 = prepare_link(
2099 &runtime,
2100 &token,
2101 &json!({
2102 "source_id": a_id.to_string(),
2103 "target_id": b_id.to_string(),
2104 "relation": "extends",
2105 "weight": 0.9,
2106 "metadata": {"note": "relinked"},
2107 }),
2108 )
2109 .await
2110 .expect("prepare second link");
2111 let outcome2 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan2])
2112 .await
2113 .expect("seam call ok");
2114 assert!(
2115 matches!(
2116 outcome2,
2117 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2118 ),
2119 "re-link of an already-linked triple must upsert, not roll back: {outcome2:?}"
2120 );
2121 let (count, weight, metadata, deleted_at) =
2122 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2123 assert_eq!(
2124 count, 1,
2125 "the natural-key UNIQUE constraint must still hold exactly one row (upsert, not a second insert)"
2126 );
2127 assert_eq!(weight, Some(0.9), "weight must be updated to the new value");
2128 assert!(
2129 metadata
2130 .as_deref()
2131 .is_some_and(|m| m.contains(r#""note":"relinked""#)),
2132 "metadata must be updated to the new value: {metadata:?}"
2133 );
2134 assert!(deleted_at.is_none());
2135 }
2136
2137 #[tokio::test]
2141 async fn atomic_link_of_soft_deleted_triple_resurrects_it() {
2142 let runtime = scratch_runtime();
2143 let token = runtime
2144 .authorize(Namespace::parse("local").expect("ns"))
2145 .expect("authorize");
2146 let entities = runtime.entities(&token).expect("entities store");
2147
2148 let a = khive_storage::Entity::new("local", "concept", "GapTwoResurrectA");
2149 let b = khive_storage::Entity::new("local", "concept", "GapTwoResurrectB");
2150 let (a_id, b_id) = (a.id, b.id);
2151 entities.upsert_entity(a).await.expect("seed a");
2152 entities.upsert_entity(b).await.expect("seed b");
2153
2154 let plan = prepare_link(
2155 &runtime,
2156 &token,
2157 &json!({
2158 "source_id": a_id.to_string(),
2159 "target_id": b_id.to_string(),
2160 "relation": "extends",
2161 }),
2162 )
2163 .await
2164 .expect("prepare link");
2165 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2166 .await
2167 .expect("seam call ok");
2168 assert!(matches!(
2169 outcome,
2170 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2171 ));
2172
2173 {
2176 let mut writer = runtime.sql().writer().await.expect("writer");
2177 let affected = writer
2178 .execute(SqlStatement {
2179 sql: "UPDATE graph_edges SET deleted_at = ?1 \
2180 WHERE namespace = ?2 AND source_id = ?3 AND target_id = ?4 AND relation = ?5"
2181 .to_string(),
2182 params: vec![
2183 SqlValue::Integer(chrono::Utc::now().timestamp_micros()),
2184 SqlValue::Text("local".to_string()),
2185 SqlValue::Text(a_id.to_string()),
2186 SqlValue::Text(b_id.to_string()),
2187 SqlValue::Text("extends".to_string()),
2188 ],
2189 label: Some("test-soft-delete-edge".to_string()),
2190 })
2191 .await
2192 .expect("soft delete edge");
2193 assert_eq!(affected, 1, "soft-delete must touch exactly the seeded row");
2194 }
2195 let (_, _, _, deleted_at) =
2196 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2197 assert!(
2198 deleted_at.is_some(),
2199 "row must be soft-deleted before the resurrect attempt"
2200 );
2201
2202 let plan_relink = prepare_link(
2205 &runtime,
2206 &token,
2207 &json!({
2208 "source_id": a_id.to_string(),
2209 "target_id": b_id.to_string(),
2210 "relation": "extends",
2211 "weight": 0.75,
2212 }),
2213 )
2214 .await
2215 .expect("prepare resurrecting link");
2216 let outcome_relink =
2217 crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan_relink])
2218 .await
2219 .expect("seam call ok");
2220 assert!(
2221 matches!(
2222 outcome_relink,
2223 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2224 ),
2225 "re-linking a soft-deleted triple must resurrect it, not roll back: {outcome_relink:?}"
2226 );
2227 let (count, weight, _, deleted_at) =
2228 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2229 assert_eq!(count, 1);
2230 assert_eq!(weight, Some(0.75));
2231 assert!(
2232 deleted_at.is_none(),
2233 "re-link must resurrect the soft-deleted row (deleted_at -> NULL)"
2234 );
2235 }
2236
2237 #[tokio::test]
2245 async fn atomic_delete_succeeds_when_vec_table_never_created() {
2246 let runtime = scratch_runtime();
2247 runtime.register_embedder(StubProvider);
2248 let token = runtime
2249 .authorize(Namespace::parse("local").expect("ns"))
2250 .expect("authorize");
2251
2252 let entity = khive_storage::Entity::new("local", "concept", "no-vec-table-entity");
2256 let entity_id = entity.id;
2257 runtime
2258 .entities(&token)
2259 .expect("entities store")
2260 .upsert_entity(entity)
2261 .await
2262 .expect("seed entity");
2263
2264 let mut note = khive_storage::note::Note::new("local", "observation", "no-vec-table-note");
2265 note.name = Some("no-vec-table-note".to_string());
2266 let note_id = note.id;
2267 runtime
2268 .notes(&token)
2269 .expect("notes store")
2270 .upsert_note(note)
2271 .await
2272 .expect("seed note");
2273
2274 for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
2275 let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
2276 .await
2277 .unwrap_or_else(|e| panic!("prepare delete ({kind}) must not fail: {e}"));
2278 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2279 .await
2280 .unwrap_or_else(|e| {
2281 panic!("atomic delete ({kind}) must not hit `no such table`: {e}")
2282 });
2283 assert!(
2284 matches!(
2285 outcome,
2286 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2287 ),
2288 "expected a clean commit ({kind}): {outcome:?}"
2289 );
2290 }
2291
2292 assert!(
2293 runtime
2294 .get_entity_including_deleted(&token, entity_id)
2295 .await
2296 .expect("get entity")
2297 .expect("entity row still present (soft delete)")
2298 .deleted_at
2299 .is_some(),
2300 "entity must be soft-deleted"
2301 );
2302 assert!(
2303 runtime
2304 .get_note_including_deleted(&token, note_id)
2305 .await
2306 .expect("get note")
2307 .expect("note row still present (soft delete)")
2308 .deleted_at
2309 .is_some(),
2310 "note must be soft-deleted"
2311 );
2312 }
2313
2314 #[tokio::test]
2320 async fn atomic_hard_delete_purges_already_soft_deleted_entity_and_note() {
2321 let runtime = scratch_runtime();
2322 runtime.register_embedder(StubProvider);
2323 let token = runtime
2324 .authorize(Namespace::parse("local").expect("ns"))
2325 .expect("authorize");
2326
2327 let entity =
2328 khive_storage::Entity::new("local", "concept", "tombstoned-entity-hard-delete");
2329 let entity_id = entity.id;
2330 runtime
2331 .entities(&token)
2332 .expect("entities store")
2333 .upsert_entity(entity.clone())
2334 .await
2335 .expect("seed entity");
2336 runtime
2337 .reindex_entity(&token, &entity)
2338 .await
2339 .expect("seed index rows");
2340
2341 let mut note =
2342 khive_storage::note::Note::new("local", "observation", "tombstoned-note-hard-delete");
2343 note.name = Some("tombstoned-note-hard-delete".to_string());
2344 let note_id = note.id;
2345 runtime
2346 .notes(&token)
2347 .expect("notes store")
2348 .upsert_note(note.clone())
2349 .await
2350 .expect("seed note");
2351 runtime
2352 .reindex_note(&token, ¬e)
2353 .await
2354 .expect("seed index rows");
2355
2356 for id in [entity_id, note_id] {
2359 let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
2360 .await
2361 .expect("prepare soft delete");
2362 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2363 .await
2364 .expect("soft delete commit");
2365 assert!(matches!(
2366 outcome,
2367 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2368 ));
2369 }
2370 assert!(
2371 runtime
2372 .get_entity_including_deleted(&token, entity_id)
2373 .await
2374 .expect("get entity")
2375 .expect("entity present")
2376 .deleted_at
2377 .is_some(),
2378 "entity must be soft-deleted before the hard-delete attempt"
2379 );
2380 assert!(
2381 runtime
2382 .get_note_including_deleted(&token, note_id)
2383 .await
2384 .expect("get note")
2385 .expect("note present")
2386 .deleted_at
2387 .is_some(),
2388 "note must be soft-deleted before the hard-delete attempt"
2389 );
2390
2391 for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
2393 let plan = prepare_delete(
2394 &runtime,
2395 &token,
2396 &json!({"id": id.to_string(), "hard": true}),
2397 None,
2398 )
2399 .await
2400 .unwrap_or_else(|e| {
2401 panic!(
2402 "prepare hard delete ({kind}) of an already-soft-deleted record \
2403 must resolve it: {e}"
2404 )
2405 });
2406 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2407 .await
2408 .unwrap_or_else(|e| panic!("hard delete ({kind}) commit failed: {e}"));
2409 assert!(
2410 matches!(
2411 outcome,
2412 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2413 ),
2414 "expected a clean hard-delete commit ({kind}): {outcome:?}"
2415 );
2416 }
2417
2418 assert!(
2419 runtime
2420 .get_entity_including_deleted(&token, entity_id)
2421 .await
2422 .expect("get entity")
2423 .is_none(),
2424 "entity row must be fully purged after hard delete"
2425 );
2426 assert!(
2427 runtime
2428 .get_note_including_deleted(&token, note_id)
2429 .await
2430 .expect("get note")
2431 .is_none(),
2432 "note row must be fully purged after hard delete"
2433 );
2434 assert!(
2435 runtime
2436 .text(&token)
2437 .expect("text store")
2438 .get_document("local", entity_id)
2439 .await
2440 .expect("get_document")
2441 .is_none(),
2442 "entity FTS row must be purged after hard delete"
2443 );
2444 assert!(
2445 runtime
2446 .text_for_notes(&token)
2447 .expect("text store")
2448 .get_document("local", note_id)
2449 .await
2450 .expect("get_document")
2451 .is_none(),
2452 "note FTS row must be purged after hard delete"
2453 );
2454 let vec_store = runtime
2455 .vectors_for_model(&token, STUB_MODEL)
2456 .expect("vec store");
2457 assert_eq!(
2458 vec_store.count().await.expect("count after"),
2459 0,
2460 "vector rows for both records must be purged after hard delete"
2461 );
2462 }
2463
2464 async fn events_for_target(
2472 runtime: &KhiveRuntime,
2473 token: &NamespaceToken,
2474 target_id: Uuid,
2475 kind: EventKind,
2476 ) -> Vec<khive_storage::Event> {
2477 let event_store = runtime.events(token).expect("event store");
2478 let filter = khive_storage::EventFilter {
2479 kinds: vec![kind],
2480 ..Default::default()
2481 };
2482 let page = event_store
2483 .query_events(filter, khive_storage::types::PageRequest::default())
2484 .await
2485 .expect("query_events");
2486 page.items
2487 .into_iter()
2488 .filter(|e| e.target_id == Some(target_id))
2489 .collect()
2490 }
2491
2492 #[tokio::test]
2497 async fn atomic_update_entity_appends_entity_updated_event() {
2498 let runtime = scratch_runtime();
2499 let token = runtime
2500 .authorize(Namespace::parse("local").expect("ns"))
2501 .expect("authorize");
2502 let entity = khive_storage::Entity::new("local", "concept", "gap1-entity");
2503 let entity_id = entity.id;
2504 runtime
2505 .entities(&token)
2506 .expect("entities store")
2507 .upsert_entity(entity)
2508 .await
2509 .expect("seed entity");
2510
2511 let plan = prepare_update(
2512 &runtime,
2513 &token,
2514 &json!({"id": entity_id.to_string(), "name": "gap1-entity-renamed"}),
2515 None,
2516 )
2517 .await
2518 .expect("prepare update");
2519 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2520 .await
2521 .expect("seam call ok");
2522 assert!(matches!(
2523 outcome,
2524 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2525 ));
2526
2527 let events = events_for_target(&runtime, &token, entity_id, EventKind::EntityUpdated).await;
2528 assert_eq!(
2529 events.len(),
2530 1,
2531 "expected exactly one EntityUpdated event for {entity_id}"
2532 );
2533 assert_eq!(events[0].namespace, "local");
2534 assert_eq!(events[0].payload["id"], json!(entity_id.to_string()));
2535 assert_eq!(
2536 events[0].payload["changed_fields"],
2537 json!(["name"]),
2538 "changed_fields must name exactly the patched fields"
2539 );
2540 }
2541
2542 #[tokio::test]
2546 async fn atomic_delete_entity_appends_entity_deleted_event_soft_and_hard() {
2547 let runtime = scratch_runtime();
2548 let token = runtime
2549 .authorize(Namespace::parse("local").expect("ns"))
2550 .expect("authorize");
2551
2552 for hard in [false, true] {
2553 let entity =
2554 khive_storage::Entity::new("local", "concept", format!("gap1-entity-hard-{hard}"));
2555 let entity_id = entity.id;
2556 runtime
2557 .entities(&token)
2558 .expect("entities store")
2559 .upsert_entity(entity)
2560 .await
2561 .expect("seed entity");
2562
2563 let args = if hard {
2564 json!({"id": entity_id.to_string(), "hard": true})
2565 } else {
2566 json!({"id": entity_id.to_string()})
2567 };
2568 let plan = prepare_delete(&runtime, &token, &args, None)
2569 .await
2570 .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
2571 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2572 .await
2573 .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
2574 assert!(
2575 matches!(
2576 outcome,
2577 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2578 ),
2579 "expected a clean delete commit (hard={hard}): {outcome:?}"
2580 );
2581
2582 let events =
2583 events_for_target(&runtime, &token, entity_id, EventKind::EntityDeleted).await;
2584 assert_eq!(
2585 events.len(),
2586 1,
2587 "expected exactly one EntityDeleted event for {entity_id} (hard={hard})"
2588 );
2589 assert_eq!(events[0].payload["hard"], json!(hard));
2590 }
2591 }
2592
2593 #[tokio::test]
2597 async fn atomic_delete_note_appends_note_deleted_event_soft_and_hard() {
2598 let runtime = scratch_runtime();
2599 let token = runtime
2600 .authorize(Namespace::parse("local").expect("ns"))
2601 .expect("authorize");
2602
2603 for hard in [false, true] {
2604 let mut note = khive_storage::note::Note::new(
2605 "local",
2606 "observation",
2607 format!("gap1-note-content-hard-{hard}"),
2608 );
2609 note.name = Some(format!("gap1-note-hard-{hard}"));
2610 let note_id = note.id;
2611 runtime
2612 .notes(&token)
2613 .expect("notes store")
2614 .upsert_note(note)
2615 .await
2616 .expect("seed note");
2617
2618 let args = if hard {
2619 json!({"id": note_id.to_string(), "hard": true})
2620 } else {
2621 json!({"id": note_id.to_string()})
2622 };
2623 let plan = prepare_delete(&runtime, &token, &args, None)
2624 .await
2625 .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
2626 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2627 .await
2628 .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
2629 assert!(
2630 matches!(
2631 outcome,
2632 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2633 ),
2634 "expected a clean delete commit (hard={hard}): {outcome:?}"
2635 );
2636
2637 let events = events_for_target(&runtime, &token, note_id, EventKind::NoteDeleted).await;
2638 assert_eq!(
2639 events.len(),
2640 1,
2641 "expected exactly one NoteDeleted event for {note_id} (hard={hard})"
2642 );
2643 assert_eq!(events[0].payload["hard"], json!(hard));
2644 }
2645 }
2646
2647 #[tokio::test]
2654 async fn atomic_update_edge_patches_weight_and_appends_edge_updated_event() {
2655 let runtime = scratch_runtime();
2656 let token = runtime
2657 .authorize(Namespace::parse("local").expect("ns"))
2658 .expect("authorize");
2659 let entities = runtime.entities(&token).expect("entities store");
2660 let a = khive_storage::Entity::new("local", "concept", "GapEdgeA");
2661 let b = khive_storage::Entity::new("local", "concept", "GapEdgeB");
2662 let (a_id, b_id) = (a.id, b.id);
2663 entities.upsert_entity(a).await.expect("seed a");
2664 entities.upsert_entity(b).await.expect("seed b");
2665
2666 let edge = runtime
2667 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.4, None)
2668 .await
2669 .expect("seed edge");
2670 let edge_id = Uuid::from(edge.id);
2671
2672 let plan = prepare_update(
2673 &runtime,
2674 &token,
2675 &json!({"id": edge_id.to_string(), "weight": 0.75}),
2676 None,
2677 )
2678 .await
2679 .expect("prepare update edge");
2680 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2681 .await
2682 .expect("seam call ok");
2683 assert!(
2684 matches!(
2685 outcome,
2686 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2687 ),
2688 "expected a clean edge update commit: {outcome:?}"
2689 );
2690
2691 let updated = runtime
2692 .get_edge(&token, edge_id)
2693 .await
2694 .expect("get_edge")
2695 .expect("edge still present");
2696 assert_eq!(updated.weight, 0.75, "weight patch must persist");
2697 assert_eq!(updated.relation, EdgeRelation::Extends);
2698
2699 let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeUpdated).await;
2700 assert_eq!(
2701 events.len(),
2702 1,
2703 "expected exactly one EdgeUpdated event for {edge_id}"
2704 );
2705 assert_eq!(
2706 events[0].payload["changed_fields"],
2707 json!(["weight"]),
2708 "changed_fields must name exactly the patched field"
2709 );
2710 }
2711
2712 #[tokio::test]
2720 async fn atomic_update_edge_symmetric_conflict_absorbs_into_surviving_row() {
2721 let runtime = scratch_runtime();
2722 let token = runtime
2723 .authorize(Namespace::parse("local").expect("ns"))
2724 .expect("authorize");
2725 let entities = runtime.entities(&token).expect("entities store");
2726 let a = khive_storage::Entity::new("local", "concept", "GapEdgeSymA");
2727 let b = khive_storage::Entity::new("local", "concept", "GapEdgeSymB");
2728 let (a_id, b_id) = (a.id, b.id);
2729 entities.upsert_entity(a).await.expect("seed a");
2730 entities.upsert_entity(b).await.expect("seed b");
2731
2732 let requested_edge = runtime
2734 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
2735 .await
2736 .expect("seed requested edge");
2737 let requested_id = Uuid::from(requested_edge.id);
2738
2739 let canonical_edge = runtime
2742 .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
2743 .await
2744 .expect("seed canonical edge");
2745 let canonical_id = Uuid::from(canonical_edge.id);
2746 assert_ne!(requested_id, canonical_id);
2747
2748 let plan = prepare_update(
2749 &runtime,
2750 &token,
2751 &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
2752 None,
2753 )
2754 .await
2755 .expect("prepare update edge (symmetric conflict)");
2756 let (canon_src, canon_tgt) =
2763 canonical_edge_endpoints(EdgeRelation::CompetesWith, a_id, b_id);
2764 match &plan {
2765 AtomicOpPlan::Update(p) => {
2766 assert_eq!(p.target_id, requested_id);
2767 let key = p
2768 .edge_natural_key
2769 .as_ref()
2770 .expect("symmetric edge update must carry edge_natural_key");
2771 assert_eq!(key.canon_source_id, canon_src);
2772 assert_eq!(key.canon_target_id, canon_tgt);
2773 assert_eq!(key.relation, EdgeRelation::CompetesWith);
2774 }
2775 other => panic!("expected an Update plan, got {other:?}"),
2776 }
2777
2778 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2779 .await
2780 .expect("seam call ok");
2781 assert!(
2782 matches!(
2783 outcome,
2784 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2785 ),
2786 "expected a clean symmetric-conflict-absorption commit: {outcome:?}"
2787 );
2788
2789 let requested_after = runtime
2791 .get_edge_including_deleted(&token, requested_id)
2792 .await
2793 .expect("get_edge_including_deleted");
2794 assert!(
2795 requested_after.is_none(),
2796 "the non-canonical requested row must be deleted, not just tombstoned"
2797 );
2798
2799 let surviving = runtime
2801 .get_edge(&token, canonical_id)
2802 .await
2803 .expect("get_edge")
2804 .expect("surviving canonical row must remain");
2805 assert_eq!(surviving.weight, 0.9);
2806 assert_eq!(surviving.relation, EdgeRelation::CompetesWith);
2807
2808 let events =
2812 events_for_target(&runtime, &token, requested_id, EventKind::EdgeUpdated).await;
2813 assert_eq!(events.len(), 1);
2814 }
2815
2816 #[tokio::test]
2828 async fn atomic_update_edge_symmetric_same_unit_delete_race_aborts_the_unit() {
2829 let runtime = scratch_runtime();
2830 let token = runtime
2831 .authorize(Namespace::parse("local").expect("ns"))
2832 .expect("authorize");
2833 let entities = runtime.entities(&token).expect("entities store");
2834 let a = khive_storage::Entity::new("local", "concept", "GapEdgeRaceA");
2835 let b = khive_storage::Entity::new("local", "concept", "GapEdgeRaceB");
2836 let (a_id, b_id) = (a.id, b.id);
2837 entities.upsert_entity(a).await.expect("seed a");
2838 entities.upsert_entity(b).await.expect("seed b");
2839
2840 let requested_edge = runtime
2842 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
2843 .await
2844 .expect("seed requested edge");
2845 let requested_id = Uuid::from(requested_edge.id);
2846
2847 let canonical_edge = runtime
2850 .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
2851 .await
2852 .expect("seed canonical edge");
2853 let canonical_id = Uuid::from(canonical_edge.id);
2854
2855 let delete_plan = prepare_delete(
2856 &runtime,
2857 &token,
2858 &json!({"id": requested_id.to_string(), "hard": true}),
2859 None,
2860 )
2861 .await
2862 .expect("prepare delete edge");
2863 let update_plan = prepare_update(
2864 &runtime,
2865 &token,
2866 &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
2867 None,
2868 )
2869 .await
2870 .expect("prepare update edge (both prepares run before either commits)");
2871
2872 let outcome = crate::atomic_runner::run_atomic_unit(
2873 runtime.sql().as_ref(),
2874 vec![delete_plan, update_plan],
2875 )
2876 .await
2877 .expect("the seam call itself must not error — the unit rolls back cleanly");
2878 match outcome {
2879 crate::atomic_runner::AtomicRunOutcome::RolledBack {
2880 failed_op_index, ..
2881 } => {
2882 assert_eq!(
2883 failed_op_index, 1,
2884 "op 1 (the update) must be the one whose guard fails"
2885 );
2886 }
2887 other => panic!("expected the whole unit to roll back, got {other:?}"),
2888 }
2889
2890 let requested_after = runtime
2892 .get_edge(&token, requested_id)
2893 .await
2894 .expect("get_edge");
2895 assert!(
2896 requested_after.is_some(),
2897 "delete(X) must have rolled back along with the failed update"
2898 );
2899 let canonical_after = runtime
2901 .get_edge(&token, canonical_id)
2902 .await
2903 .expect("get_edge")
2904 .expect("canonical row must still be present");
2905 assert_eq!(
2906 canonical_after.weight, 0.6,
2907 "the pre-existing canonical row must never have been touched by the aborted update"
2908 );
2909 }
2910
2911 #[tokio::test]
2916 async fn atomic_update_edge_rejects_entity_only_field_name() {
2917 let runtime = scratch_runtime();
2918 let token = runtime
2919 .authorize(Namespace::parse("local").expect("ns"))
2920 .expect("authorize");
2921 let entities = runtime.entities(&token).expect("entities store");
2922 let a = khive_storage::Entity::new("local", "concept", "GapEdgeRejectA");
2923 let b = khive_storage::Entity::new("local", "concept", "GapEdgeRejectB");
2924 let (a_id, b_id) = (a.id, b.id);
2925 entities.upsert_entity(a).await.expect("seed a");
2926 entities.upsert_entity(b).await.expect("seed b");
2927 let edge = runtime
2928 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
2929 .await
2930 .expect("seed edge");
2931 let edge_id = Uuid::from(edge.id);
2932
2933 let err = prepare_update(
2934 &runtime,
2935 &token,
2936 &json!({"id": edge_id.to_string(), "name": "not-a-valid-edge-field"}),
2937 None,
2938 )
2939 .await
2940 .expect_err("edge update with an entity-only field must be rejected");
2941 let message = err.to_string();
2942 assert!(
2943 message.contains("name") && message.contains("edge"),
2944 "error must name the offending field and the substrate: {message}"
2945 );
2946 }
2947
2948 #[tokio::test]
2953 async fn atomic_delete_edge_soft_and_hard_appends_edge_deleted_event() {
2954 let runtime = scratch_runtime();
2955 let token = runtime
2956 .authorize(Namespace::parse("local").expect("ns"))
2957 .expect("authorize");
2958
2959 for hard in [false, true] {
2960 let entities = runtime.entities(&token).expect("entities store");
2961 let a = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelA{hard}"));
2962 let b = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelB{hard}"));
2963 let (a_id, b_id) = (a.id, b.id);
2964 entities.upsert_entity(a).await.expect("seed a");
2965 entities.upsert_entity(b).await.expect("seed b");
2966 let edge = runtime
2967 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
2968 .await
2969 .expect("seed edge");
2970 let edge_id = Uuid::from(edge.id);
2971
2972 let args = if hard {
2973 json!({"id": edge_id.to_string(), "hard": true})
2974 } else {
2975 json!({"id": edge_id.to_string()})
2976 };
2977 let plan = prepare_delete(&runtime, &token, &args, None)
2978 .await
2979 .unwrap_or_else(|e| panic!("prepare delete edge (hard={hard}): {e}"));
2980 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2981 .await
2982 .unwrap_or_else(|e| panic!("edge delete commit (hard={hard}): {e}"));
2983 assert!(
2984 matches!(
2985 outcome,
2986 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2987 ),
2988 "expected a clean edge delete commit (hard={hard}): {outcome:?}"
2989 );
2990
2991 let after = runtime
2992 .get_edge_including_deleted(&token, edge_id)
2993 .await
2994 .expect("get_edge_including_deleted");
2995 if hard {
2996 assert!(after.is_none(), "hard delete must purge the row entirely");
2997 } else {
2998 assert!(
2999 after.as_ref().is_some_and(|e| e.deleted_at.is_some()),
3000 "soft delete must tombstone, not purge"
3001 );
3002 }
3003
3004 let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeDeleted).await;
3005 assert_eq!(
3006 events.len(),
3007 1,
3008 "expected exactly one EdgeDeleted event for {edge_id} (hard={hard})"
3009 );
3010 assert_eq!(events[0].payload["hard"], json!(hard));
3011 }
3012 }
3013
3014 #[tokio::test]
3018 async fn atomic_update_note_appends_no_event() {
3019 let runtime = scratch_runtime();
3020 let token = runtime
3021 .authorize(Namespace::parse("local").expect("ns"))
3022 .expect("authorize");
3023 let mut note = khive_storage::note::Note::new("local", "observation", "gap1-note-noevent");
3024 note.name = Some("gap1-note-noevent".to_string());
3025 let note_id = note.id;
3026 runtime
3027 .notes(&token)
3028 .expect("notes store")
3029 .upsert_note(note)
3030 .await
3031 .expect("seed note");
3032
3033 let plan = prepare_update(
3034 &runtime,
3035 &token,
3036 &json!({"id": note_id.to_string(), "content": "gap1-note-noevent, revised"}),
3037 None,
3038 )
3039 .await
3040 .expect("prepare update");
3041 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
3042 .await
3043 .expect("seam call ok");
3044 assert!(matches!(
3045 outcome,
3046 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
3047 ));
3048
3049 let event_store = runtime.events(&token).expect("event store");
3050 let page = event_store
3051 .query_events(
3052 khive_storage::EventFilter::default(),
3053 khive_storage::types::PageRequest::default(),
3054 )
3055 .await
3056 .expect("query_events");
3057 assert!(
3058 page.items.iter().all(|e| e.target_id != Some(note_id)),
3059 "update_note must append no event; found: {:?}",
3060 page.items
3061 .iter()
3062 .filter(|e| e.target_id == Some(note_id))
3063 .collect::<Vec<_>>()
3064 );
3065 }
3066
3067 #[tokio::test]
3070 async fn atomic_link_appends_no_event() {
3071 let runtime = scratch_runtime();
3072 let token = runtime
3073 .authorize(Namespace::parse("local").expect("ns"))
3074 .expect("authorize");
3075 let source = khive_storage::Entity::new("local", "concept", "gap1-link-source");
3076 let target = khive_storage::Entity::new("local", "concept", "gap1-link-target");
3077 let (source_id, target_id) = (source.id, target.id);
3078 runtime
3079 .entities(&token)
3080 .expect("entities store")
3081 .upsert_entity(source)
3082 .await
3083 .expect("seed source");
3084 runtime
3085 .entities(&token)
3086 .expect("entities store")
3087 .upsert_entity(target)
3088 .await
3089 .expect("seed target");
3090
3091 let plan = prepare_link(
3092 &runtime,
3093 &token,
3094 &json!({
3095 "source_id": source_id.to_string(),
3096 "target_id": target_id.to_string(),
3097 "relation": "extends",
3098 }),
3099 )
3100 .await
3101 .expect("prepare link");
3102 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
3103 .await
3104 .expect("seam call ok");
3105 assert!(matches!(
3106 outcome,
3107 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
3108 ));
3109
3110 let event_store = runtime.events(&token).expect("event store");
3111 let page = event_store
3112 .query_events(
3113 khive_storage::EventFilter::default(),
3114 khive_storage::types::PageRequest::default(),
3115 )
3116 .await
3117 .expect("query_events");
3118 assert!(
3119 page.items.is_empty(),
3120 "link must append no event; found: {:?}",
3121 page.items
3122 );
3123 }
3124}