1use serde_json::Value;
20use uuid::Uuid;
21
22use khive_storage::types::SqlValue;
23use khive_storage::{EdgeRelation, SqlStatement};
24use khive_types::{EventKind, SubstrateKind};
25
26use crate::atomic_plan::{
27 AddEntityPlan, AddNotePlan, AffectedRowGuard, DeletePlan, EdgeNaturalKey, LinkPlan, MergePlan,
28 PlanStatement, PostCommitEffect, UpdatePlan,
29};
30use crate::atomic_runner::AtomicOpPlan;
31use crate::curation::{entity_fts_document, note_fts_document};
32use crate::error::{RuntimeError, RuntimeResult};
33use crate::operations::{
34 canonical_edge_endpoints, merge_dependency_kind, validate_edge_metadata, validate_edge_weight,
35 Resolved,
36};
37use crate::runtime::{KhiveRuntime, NamespaceToken};
38
39use khive_db::stores::entity::{
40 entity_hard_delete_statement, entity_soft_delete_statement, entity_upsert_statement,
41};
42use khive_db::stores::event::event_insert_statements;
43use khive_db::stores::graph::{
44 edge_hard_delete_statement, edge_insert_guarded_by_endpoints_statement,
45 edge_soft_delete_statement, edge_symmetric_delete_if_conflict_statement,
46 edge_symmetric_refresh_or_update_inplace_statement, edge_upsert_statement,
47 purge_incident_edges_statement,
48};
49use khive_db::stores::note::{
50 note_hard_delete_statement, note_soft_delete_statement, note_upsert_statement,
51};
52use khive_db::stores::text::insert_document_statement;
53
54fn obj(args: &Value) -> RuntimeResult<&serde_json::Map<String, Value>> {
59 args.as_object()
60 .ok_or_else(|| RuntimeError::InvalidInput("op args must be a JSON object".into()))
61}
62
63fn require_str<'a>(args: &'a Value, key: &str) -> RuntimeResult<&'a str> {
64 obj(args)?
65 .get(key)
66 .and_then(|v| v.as_str())
67 .ok_or_else(|| RuntimeError::InvalidInput(format!("missing required field {key:?}")))
68}
69
70fn require_uuid(args: &Value, key: &str) -> RuntimeResult<Uuid> {
71 let raw = require_str(args, key)?;
72 Uuid::parse_str(raw)
73 .map_err(|_| RuntimeError::InvalidInput(format!("{key} must be a full UUID; got {raw:?}")))
74}
75
76fn optional_str<'a>(args: &'a Value, key: &str) -> Option<&'a str> {
77 obj(args).ok()?.get(key).and_then(|v| v.as_str())
78}
79
80fn optional_create_string(args: &Value, key: &str) -> RuntimeResult<Option<String>> {
81 match obj(args)?.get(key) {
82 None | Some(Value::Null) => Ok(None),
83 Some(Value::String(value)) => Ok(Some(value.clone())),
84 Some(other) => Err(RuntimeError::InvalidInput(format!(
85 "{key} must be a string or null, got: {other}"
86 ))),
87 }
88}
89
90fn optional_string_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<String>>> {
104 match obj(args)?.get(key) {
105 None | Some(Value::Null) => Ok(None),
106 Some(Value::String(s)) => Ok(Some(Some(s.clone()))),
107 Some(other) => Err(RuntimeError::InvalidInput(format!(
108 "{key} must be a string or null, got: {other}"
109 ))),
110 }
111}
112
113fn entity_name_patch(args: &Value) -> RuntimeResult<Option<String>> {
124 match obj(args)?.get("name") {
125 None | Some(Value::Null) => Ok(None),
126 Some(Value::String(s)) => Ok(Some(s.clone())),
127 Some(other) => Err(RuntimeError::InvalidInput(format!(
128 "name must be a string, got: {other}"
129 ))),
130 }
131}
132
133fn optional_properties(args: &Value, key: &str) -> RuntimeResult<Option<Value>> {
142 match obj(args)?.get(key) {
143 None | Some(Value::Null) => Ok(None),
144 Some(v) => Ok(Some(v.clone())),
145 }
146}
147
148fn optional_tags(args: &Value) -> RuntimeResult<Option<Vec<String>>> {
155 match obj(args)?.get("tags") {
156 None | Some(Value::Null) => Ok(None),
157 Some(Value::Array(items)) => {
158 let mut tags = Vec::with_capacity(items.len());
159 for item in items {
160 let s = item.as_str().ok_or_else(|| {
161 RuntimeError::InvalidInput("tags must be an array of strings".into())
162 })?;
163 tags.push(s.to_string());
164 }
165 Ok(Some(tags))
166 }
167 Some(_) => Err(RuntimeError::InvalidInput(
168 "tags must be an array of strings".into(),
169 )),
170 }
171}
172
173fn optional_f64(args: &Value, key: &str) -> RuntimeResult<Option<f64>> {
174 match obj(args)?.get(key) {
175 None => Ok(None),
176 Some(Value::Null) => Ok(None),
177 Some(v) => v
178 .as_f64()
179 .map(Some)
180 .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
181 }
182}
183
184fn optional_f64_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<f64>>> {
190 match obj(args)?.get(key) {
191 None => Ok(None),
192 Some(Value::Null) => Ok(Some(None)),
193 Some(v) => v
194 .as_f64()
195 .map(|f| Some(Some(f)))
196 .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
197 }
198}
199
200fn vector_table_names(runtime: &KhiveRuntime) -> Vec<String> {
205 runtime
206 .registered_embedding_model_names()
207 .iter()
208 .map(|name| format!("vec_{}", crate::config::sanitize_key(name)))
209 .collect()
210}
211
212fn purge_index_row_statement(
216 table: &str,
217 namespace: &str,
218 subject_id: Uuid,
219 label: &str,
220) -> PlanStatement {
221 PlanStatement {
222 statement: SqlStatement {
223 sql: format!("DELETE FROM {table} WHERE namespace = ?1 AND subject_id = ?2"),
224 params: vec![
225 SqlValue::Text(namespace.to_string()),
226 SqlValue::Text(subject_id.to_string()),
227 ],
228 label: Some(label.to_string()),
229 },
230 guard: None,
231 }
232}
233
234async fn vector_table_exists(runtime: &KhiveRuntime, table: &str) -> RuntimeResult<bool> {
239 let mut reader = runtime
240 .sql()
241 .reader()
242 .await
243 .map_err(RuntimeError::Storage)?;
244 let row = reader
245 .query_scalar(SqlStatement {
246 sql: "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1".to_string(),
247 params: vec![SqlValue::Text(table.to_string())],
248 label: Some("atomic-delete-vec-table-exists".to_string()),
249 })
250 .await
251 .map_err(RuntimeError::Storage)?;
252 Ok(row.is_some())
253}
254
255async fn push_index_purge_statements(
271 runtime: &KhiveRuntime,
272 statements: &mut Vec<PlanStatement>,
273 fts_table: &str,
274 namespace: &str,
275 subject_id: Uuid,
276 label_prefix: &str,
277) -> RuntimeResult<()> {
278 statements.push(purge_index_row_statement(
279 fts_table,
280 namespace,
281 subject_id,
282 &format!("{label_prefix}-purge-fts"),
283 ));
284 for vec_table in vector_table_names(runtime) {
285 if vector_table_exists(runtime, &vec_table).await? {
286 statements.push(purge_index_row_statement(
287 &vec_table,
288 namespace,
289 subject_id,
290 &format!("{label_prefix}-purge-vec-{vec_table}"),
291 ));
292 }
293 }
294 Ok(())
295}
296
297fn event_append_statements(
313 namespace: &str,
314 verb: &str,
315 kind: EventKind,
316 substrate: SubstrateKind,
317 target_id: Uuid,
318 payload: Value,
319) -> RuntimeResult<Vec<PlanStatement>> {
320 let event = khive_storage::event::Event::new(namespace.to_string(), verb, kind, substrate, "")
321 .with_target(target_id)
322 .with_payload(payload);
323 let statements = event_insert_statements(&event)
324 .map_err(|e| RuntimeError::Internal(format!("event_insert_statements: {e}")))?;
325 Ok(statements
326 .into_iter()
327 .map(|statement| PlanStatement {
328 statement,
329 guard: None,
330 })
331 .collect())
332}
333
334pub async fn prepare_op(
344 runtime: &KhiveRuntime,
345 token: &NamespaceToken,
346 tool: &str,
347 args: &Value,
348) -> RuntimeResult<AtomicOpPlan> {
349 match tool {
350 "update" => prepare_update(runtime, token, args, None).await,
359 "delete" => prepare_delete(runtime, token, args, None).await,
368 "link" => prepare_link(runtime, token, args).await,
369 "merge" => prepare_merge(runtime, token, args).await,
370 "propose" | "review" | "withdraw" => prepare_governance_unimplemented(tool),
371 other => Err(RuntimeError::InvalidInput(format!(
372 "{other:?} has no atomic_prepare::prepare_op implementation; the CLI \
373 admissibility check should have rejected this before prepare"
374 ))),
375 }
376}
377
378fn prepare_governance_unimplemented(tool: &str) -> RuntimeResult<AtomicOpPlan> {
379 Err(RuntimeError::InvalidInput(format!(
380 "{tool:?} is on the ADR-099 v1 admissible verb list but has no --atomic \
381 prepare/apply implementation yet: its lifecycle (ADR-046) is an \
382 event-sourced changeset-interpreter over a dedicated `proposals_open` \
383 table, not a small guarded-DML plan — a faithful non-stub atomic \
384 prepare for it is tracked as ADR-099 follow-up work, not implemented \
385 in slice B3. No write was attempted."
386 )))
387}
388
389pub async fn prepare_add_entity(
399 runtime: &KhiveRuntime,
400 token: &NamespaceToken,
401 args: &Value,
402) -> RuntimeResult<AtomicOpPlan> {
403 let kind = require_str(args, "kind")?;
404 let name = require_str(args, "name")?;
405 runtime.validate_entity_kind(kind)?;
406 if name.trim().is_empty() {
407 return Err(RuntimeError::InvalidInput(
408 "name must not be empty".to_string(),
409 ));
410 }
411
412 let description = optional_create_string(args, "description")?;
413 let properties = optional_properties(args, "properties")?;
414 let tags = optional_tags(args)?.unwrap_or_default();
415
416 crate::secret_gate::check(name)?;
417 if let Some(ref d) = description {
418 crate::secret_gate::check(d)?;
419 }
420 if let Some(ref p) = properties {
421 crate::secret_gate::check_json(p)?;
422 }
423 crate::secret_gate::check_tags(&tags)?;
424
425 let ns = token.namespace().as_str();
426 let mut entity = khive_storage::Entity::new(ns, kind, name);
427 if let Some(d) = description {
428 entity = entity.with_description(d);
429 }
430 if let Some(p) = properties {
431 entity = entity.with_properties(p);
432 }
433 if !tags.is_empty() {
434 entity = entity.with_tags(tags);
435 }
436
437 let statements = vec![
438 PlanStatement {
439 statement: entity_upsert_statement(&entity),
440 guard: Some(AffectedRowGuard::exactly(1)),
441 },
442 PlanStatement {
443 statement: insert_document_statement("fts_entities", &entity_fts_document(&entity)),
444 guard: None,
445 },
446 ];
447
448 Ok(AtomicOpPlan::AddEntity(AddEntityPlan {
449 entity_id: entity.id,
450 statements,
451 post_commit: PostCommitEffect::ReindexEntity {
452 entity_id: entity.id,
453 },
454 }))
455}
456
457pub async fn prepare_add_note(
463 runtime: &KhiveRuntime,
464 token: &NamespaceToken,
465 args: &Value,
466) -> RuntimeResult<AtomicOpPlan> {
467 let kind = require_str(args, "kind")?;
468 let content = require_str(args, "content")?;
469 runtime.validate_note_kind(kind)?;
470
471 let name = optional_create_string(args, "name")?;
472 let properties = optional_properties(args, "properties")?;
473
474 crate::secret_gate::check(content)?;
475 if let Some(ref n) = name {
476 crate::secret_gate::check(n)?;
477 }
478 if let Some(ref p) = properties {
479 crate::secret_gate::check_json(p)?;
480 }
481
482 let ns = token.namespace().as_str();
483 let mut note = khive_storage::note::Note::new(ns, kind, content);
484 if let Some(n) = name {
485 note = note.with_name(n);
486 }
487 if let Some(p) = properties {
488 note = note.with_properties(p);
489 }
490
491 let statements = vec![
492 PlanStatement {
493 statement: note_upsert_statement(¬e),
494 guard: Some(AffectedRowGuard::exactly(1)),
495 },
496 PlanStatement {
497 statement: insert_document_statement("fts_notes", ¬e_fts_document(¬e)),
498 guard: None,
499 },
500 ];
501
502 Ok(AtomicOpPlan::AddNote(AddNotePlan {
503 note_id: note.id,
504 statements,
505 post_commit: PostCommitEffect::ReindexNote { note_id: note.id },
506 }))
507}
508
509fn reject_inapplicable_update_fields(args: &Value, substrate: &str) -> RuntimeResult<()> {
524 let o = obj(args)?;
525 let present = |k: &str| o.get(k).is_some_and(|v| !v.is_null());
526 let (bad_field, valid): (Option<&str>, &str) = match substrate {
527 "entity" => {
528 let bad = if present("content") {
529 Some("content")
530 } else if present("salience") {
531 Some("salience")
532 } else if present("decay_factor") {
533 Some("decay_factor")
534 } else if present("relation") {
535 Some("relation")
536 } else if present("weight") {
537 Some("weight")
538 } else {
539 None
540 };
541 (bad, "name, description, tags, properties")
542 }
543 "note" => {
544 let bad = if present("description") {
545 Some("description")
546 } else if present("tags") {
547 Some("tags")
548 } else if present("relation") {
549 Some("relation")
550 } else if present("weight") {
551 Some("weight")
552 } else {
553 None
554 };
555 (bad, "name, content, salience, decay_factor, properties")
556 }
557 "edge" => {
563 let bad = if present("name") {
564 Some("name")
565 } else if present("description") {
566 Some("description")
567 } else if present("content") {
568 Some("content")
569 } else if present("tags") {
570 Some("tags")
571 } else if present("salience") {
572 Some("salience")
573 } else if present("decay_factor") {
574 Some("decay_factor")
575 } else {
576 None
577 };
578 (bad, "relation, weight, properties")
579 }
580 _ => (None, ""),
581 };
582 if let Some(field) = bad_field {
583 let substrate_label = match substrate {
584 "entity" => "an entity",
585 "note" => "a note",
586 "edge" => "an edge",
587 other => other,
588 };
589 return Err(RuntimeError::InvalidInput(format!(
590 "field '{field}' is not valid for {substrate_label}; valid fields: {valid}"
591 )));
592 }
593 Ok(())
594}
595
596pub enum AtomicUpdateKind {
606 Entity { specific: Option<String> },
607 Note { specific: Option<String> },
608 Edge,
609}
610
611pub async fn prepare_update(
617 runtime: &KhiveRuntime,
618 token: &NamespaceToken,
619 args: &Value,
620 expected_kind: Option<AtomicUpdateKind>,
621) -> RuntimeResult<AtomicOpPlan> {
622 let id = require_uuid(args, "id")?;
623
624 if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
628 return Err(RuntimeError::InvalidInput(
629 "entity_kind is immutable; to change kind, delete then re-create the entity, \
630 or use merge() if this is a deduplication correction"
631 .into(),
632 ));
633 }
634
635 match runtime.resolve_by_id(token, id).await? {
636 Some(Resolved::Entity(entity)) => {
637 match &expected_kind {
638 None => {}
639 Some(AtomicUpdateKind::Entity {
640 specific: Some(expected),
641 }) => {
642 if &entity.kind != expected {
643 return Err(RuntimeError::NotFound(format!("entity {id}")));
644 }
645 }
646 Some(AtomicUpdateKind::Entity { specific: None }) => {}
647 Some(AtomicUpdateKind::Note { .. }) => {
648 return Err(RuntimeError::NotFound(format!("note {id}")));
649 }
650 Some(AtomicUpdateKind::Edge) => {
651 return Err(RuntimeError::NotFound(format!("edge {id}")));
652 }
653 }
654 reject_inapplicable_update_fields(args, "entity")?;
660 let name = entity_name_patch(args)?;
661 let description = optional_string_patch(args, "description")?;
662 let properties = optional_properties(args, "properties")?;
663 let tags = optional_tags(args)?;
664
665 prepare_update_entity_plan(
666 runtime,
667 token,
668 id,
669 crate::curation::EntityPatch {
670 name,
671 description,
672 properties,
673 tags,
674 },
675 )
676 .await
677 }
678 Some(Resolved::Note(note)) => {
679 match &expected_kind {
680 None => {}
681 Some(AtomicUpdateKind::Note {
682 specific: Some(expected),
683 }) => {
684 if ¬e.kind != expected {
685 return Err(RuntimeError::NotFound(format!("note {id}")));
686 }
687 }
688 Some(AtomicUpdateKind::Note { specific: None }) => {}
689 Some(AtomicUpdateKind::Entity { .. }) => {
690 return Err(RuntimeError::NotFound(format!("entity {id}")));
691 }
692 Some(AtomicUpdateKind::Edge) => {
693 return Err(RuntimeError::NotFound(format!("edge {id}")));
694 }
695 }
696 reject_inapplicable_update_fields(args, "note")?;
703 let name = optional_string_patch(args, "name")?;
704 let content = optional_str(args, "content").map(|s| s.to_string());
705 let properties = optional_properties(args, "properties")?;
706 let salience = optional_f64_patch(args, "salience")?;
707 let decay_factor = optional_f64_patch(args, "decay_factor")?;
708
709 let (note, text_changed) = runtime
710 .prepare_update_note(
711 token,
712 id,
713 crate::curation::NotePatch::new(
714 name,
715 content,
716 salience,
717 decay_factor,
718 properties,
719 ),
720 )
721 .await?;
722
723 let post_commit = if text_changed {
724 PostCommitEffect::ReindexNote { note_id: id }
725 } else {
726 PostCommitEffect::None
727 };
728 Ok(AtomicOpPlan::Update(UpdatePlan {
729 target_id: id,
730 statements: vec![PlanStatement {
731 statement: note_upsert_statement(¬e),
732 guard: Some(AffectedRowGuard::exactly(1)),
733 }],
734 post_commit,
735 edge_natural_key: None,
736 }))
737 }
738 Some(_) => Err(RuntimeError::InvalidInput(format!(
739 "update target {id} must be an entity, note, or edge"
740 ))),
741 None => match &expected_kind {
748 Some(AtomicUpdateKind::Entity { .. }) => {
749 Err(RuntimeError::NotFound(format!("entity/note {id}")))
750 }
751 Some(AtomicUpdateKind::Note { .. }) => {
752 Err(RuntimeError::NotFound(format!("entity/note {id}")))
753 }
754 Some(AtomicUpdateKind::Edge) | None => match runtime.get_edge(token, id).await? {
755 Some(edge) => prepare_update_edge(runtime, id, edge, args).await,
756 None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
757 },
758 },
759 }
760}
761
762pub async fn prepare_update_entity_plan(
766 runtime: &KhiveRuntime,
767 token: &NamespaceToken,
768 id: Uuid,
769 patch: crate::curation::EntityPatch,
770) -> RuntimeResult<AtomicOpPlan> {
771 let (entity, text_changed, changed_fields) =
772 runtime.prepare_update_entity(token, id, patch).await?;
773 let mut statements = vec![PlanStatement {
774 statement: entity_upsert_statement(&entity),
775 guard: Some(AffectedRowGuard::exactly(1)),
776 }];
777 statements.extend(event_append_statements(
778 &entity.namespace,
779 "update",
780 EventKind::EntityUpdated,
781 SubstrateKind::Entity,
782 id,
783 serde_json::json!({
784 "id": id,
785 "namespace": entity.namespace,
786 "changed_fields": changed_fields,
787 }),
788 )?);
789 let post_commit = if text_changed {
790 PostCommitEffect::ReindexEntity { entity_id: id }
791 } else {
792 PostCommitEffect::None
793 };
794 Ok(AtomicOpPlan::Update(UpdatePlan {
795 target_id: id,
796 statements,
797 post_commit,
798 edge_natural_key: None,
799 }))
800}
801
802async fn prepare_update_edge(
822 runtime: &KhiveRuntime,
823 id: Uuid,
824 mut edge: khive_storage::types::Edge,
825 args: &Value,
826) -> RuntimeResult<AtomicOpPlan> {
827 reject_inapplicable_update_fields(args, "edge")?;
828
829 let relation_raw = optional_str(args, "relation");
830 let weight = optional_f64(args, "weight")?;
831 let properties = optional_properties(args, "properties")?;
832
833 if let Some(ref p) = properties {
834 crate::secret_gate::check_json(p)?;
835 }
836
837 let namespace = edge.namespace.clone();
838 let record_tok = NamespaceToken::for_namespace(
839 khive_types::Namespace::parse(&namespace)
840 .map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
841 );
842
843 let mut changed_fields: Vec<&'static str> = Vec::new();
844 if let Some(raw) = relation_raw {
845 let relation = parse_edge_relation(raw)?;
846 runtime
847 .validate_edge_relation_endpoints(&record_tok, edge.source_id, edge.target_id, relation)
848 .await?;
849 edge.relation = relation;
850 changed_fields.push("relation");
851 }
852 if let Some(w) = weight {
853 if !w.is_finite() || !(0.0..=1.0).contains(&w) {
854 return Err(RuntimeError::InvalidInput(format!(
855 "edge weight must be a finite value in [0.0, 1.0]; got {w}"
856 )));
857 }
858 edge.weight = w;
859 changed_fields.push("weight");
860 }
861 if let Some(p) = properties {
862 edge.metadata = Some(p);
863 changed_fields.push("properties");
864 }
865
866 let (canon_src, canon_tgt) =
867 canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
868 let now = chrono::Utc::now();
869
870 let mut statements: Vec<PlanStatement> = Vec::new();
871 let mut edge_natural_key: Option<EdgeNaturalKey> = None;
872
873 if edge.relation.is_symmetric() {
874 let metadata_str = edge
884 .metadata
885 .as_ref()
886 .map(|v| serde_json::to_string(v).unwrap_or_default());
887
888 statements.push(PlanStatement {
889 statement: edge_symmetric_delete_if_conflict_statement(
890 &namespace,
891 id,
892 canon_src,
893 canon_tgt,
894 edge.relation,
895 ),
896 guard: Some(AffectedRowGuard {
897 expected_min: 0,
898 expected_max: Some(1),
899 }),
900 });
901 statements.push(PlanStatement {
902 statement: edge_symmetric_refresh_or_update_inplace_statement(
903 &namespace,
904 id,
905 canon_src,
906 canon_tgt,
907 edge.relation,
908 edge.weight,
909 now.timestamp_micros(),
910 metadata_str.as_deref(),
911 edge.target_backend.as_deref(),
912 ),
913 guard: Some(AffectedRowGuard::exactly(1)),
914 });
915
916 edge_natural_key = Some(EdgeNaturalKey {
921 namespace: namespace.clone(),
922 canon_source_id: canon_src,
923 canon_target_id: canon_tgt,
924 relation: edge.relation,
925 });
926 } else {
927 edge.updated_at = now;
930 statements.push(PlanStatement {
931 statement: edge_upsert_statement(&edge),
932 guard: Some(AffectedRowGuard::exactly(1)),
933 });
934 }
935
936 statements.extend(event_append_statements(
941 &namespace,
942 "update",
943 EventKind::EdgeUpdated,
944 SubstrateKind::Entity,
945 id,
946 serde_json::json!({"id": id, "namespace": namespace, "changed_fields": changed_fields}),
947 )?);
948
949 Ok(AtomicOpPlan::Update(UpdatePlan {
950 target_id: id,
951 statements,
952 post_commit: PostCommitEffect::None,
953 edge_natural_key,
954 }))
955}
956
957pub enum AtomicDeleteKind {
974 Entity { specific: Option<String> },
975 Note { specific: Option<String> },
976 Edge,
977}
978
979pub async fn prepare_delete(
985 runtime: &KhiveRuntime,
986 token: &NamespaceToken,
987 args: &Value,
988 expected_kind: Option<AtomicDeleteKind>,
989) -> RuntimeResult<AtomicOpPlan> {
990 let id = require_uuid(args, "id")?;
991 let hard = obj(args)?
992 .get("hard")
993 .and_then(|v| v.as_bool())
994 .unwrap_or(false);
995
996 let resolved = if hard {
1002 runtime.resolve_by_id_including_deleted(token, id).await?
1003 } else {
1004 runtime.resolve_by_id(token, id).await?
1005 };
1006
1007 match resolved {
1008 Some(Resolved::Entity(entity)) => {
1009 match &expected_kind {
1010 None => {}
1011 Some(AtomicDeleteKind::Entity {
1012 specific: Some(expected),
1013 }) => {
1014 if &entity.kind != expected {
1015 return Err(RuntimeError::NotFound(format!("{expected} {id}")));
1016 }
1017 }
1018 Some(AtomicDeleteKind::Entity { specific: None }) => {}
1019 Some(AtomicDeleteKind::Note { .. }) => {
1020 return Err(RuntimeError::NotFound(format!("note {id}")));
1021 }
1022 Some(AtomicDeleteKind::Edge) => {
1023 return Err(RuntimeError::NotFound(format!("edge {id}")));
1024 }
1025 }
1026 let namespace = entity.namespace.clone();
1027 let mut statements = if hard {
1032 vec![PlanStatement {
1033 statement: entity_hard_delete_statement(id),
1034 guard: Some(AffectedRowGuard::exactly(1)),
1035 }]
1036 } else {
1037 let deleted_at = chrono::Utc::now().timestamp_micros();
1038 vec![PlanStatement {
1039 statement: entity_soft_delete_statement(id, deleted_at),
1040 guard: Some(AffectedRowGuard::exactly(1)),
1041 }]
1042 };
1043 if hard {
1044 statements.push(PlanStatement {
1047 statement: purge_incident_edges_statement(id),
1048 guard: None,
1049 });
1050 }
1051 push_index_purge_statements(
1056 runtime,
1057 &mut statements,
1058 "fts_entities",
1059 &namespace,
1060 id,
1061 "atomic-delete-entity",
1062 )
1063 .await?;
1064 statements.extend(event_append_statements(
1070 &namespace,
1071 "delete",
1072 EventKind::EntityDeleted,
1073 SubstrateKind::Entity,
1074 id,
1075 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
1076 )?);
1077 Ok(AtomicOpPlan::Delete(DeletePlan {
1078 target_id: id,
1079 statements,
1080 post_commit: PostCommitEffect::None,
1081 }))
1082 }
1083 Some(Resolved::Note(note)) => {
1084 match &expected_kind {
1085 None => {}
1086 Some(AtomicDeleteKind::Note {
1087 specific: Some(expected),
1088 }) => {
1089 if ¬e.kind != expected {
1090 return Err(RuntimeError::NotFound(format!("{expected} {id}")));
1091 }
1092 }
1093 Some(AtomicDeleteKind::Note { specific: None }) => {}
1094 Some(AtomicDeleteKind::Entity { .. }) => {
1095 return Err(RuntimeError::NotFound(format!("entity {id}")));
1096 }
1097 Some(AtomicDeleteKind::Edge) => {
1098 return Err(RuntimeError::NotFound(format!("edge {id}")));
1099 }
1100 }
1101 let namespace = note.namespace.clone();
1102 let mut statements = if hard {
1106 vec![PlanStatement {
1107 statement: note_hard_delete_statement(id),
1108 guard: Some(AffectedRowGuard::exactly(1)),
1109 }]
1110 } else {
1111 let deleted_at = chrono::Utc::now().timestamp_micros();
1112 vec![PlanStatement {
1113 statement: note_soft_delete_statement(id, deleted_at),
1114 guard: Some(AffectedRowGuard::exactly(1)),
1115 }]
1116 };
1117 if hard {
1118 statements.push(PlanStatement {
1119 statement: purge_incident_edges_statement(id),
1120 guard: None,
1121 });
1122 }
1123 push_index_purge_statements(
1128 runtime,
1129 &mut statements,
1130 "fts_notes",
1131 &namespace,
1132 id,
1133 "atomic-delete-note",
1134 )
1135 .await?;
1136 statements.extend(event_append_statements(
1140 &namespace,
1141 "delete",
1142 EventKind::NoteDeleted,
1143 SubstrateKind::Note,
1144 id,
1145 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
1146 )?);
1147 Ok(AtomicOpPlan::Delete(DeletePlan {
1148 target_id: id,
1149 statements,
1150 post_commit: PostCommitEffect::NoteDeleted {
1155 note_id: id,
1156 kind: note.kind.clone(),
1157 },
1158 }))
1159 }
1160 Some(_) => Err(RuntimeError::InvalidInput(format!(
1161 "delete target {id} must be an entity, note, or edge"
1162 ))),
1163 None => match &expected_kind {
1167 Some(AtomicDeleteKind::Entity { .. }) => {
1168 Err(RuntimeError::NotFound(format!("entity/note {id}")))
1169 }
1170 Some(AtomicDeleteKind::Note { .. }) => {
1171 Err(RuntimeError::NotFound(format!("entity/note {id}")))
1172 }
1173 Some(AtomicDeleteKind::Edge) | None => {
1174 let edge = if hard {
1175 runtime.get_edge_including_deleted(token, id).await?
1176 } else {
1177 runtime.get_edge(token, id).await?
1178 };
1179 match edge {
1180 Some(edge) => prepare_delete_edge(id, edge, hard).await,
1181 None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
1182 }
1183 }
1184 },
1185 }
1186}
1187
1188async fn prepare_delete_edge(
1197 id: Uuid,
1198 edge: khive_storage::types::Edge,
1199 hard: bool,
1200) -> RuntimeResult<AtomicOpPlan> {
1201 let namespace = edge.namespace.clone();
1202 let mut statements: Vec<PlanStatement> = Vec::new();
1203
1204 if hard {
1205 statements.push(PlanStatement {
1210 statement: purge_incident_edges_statement(id),
1211 guard: None,
1212 });
1213 statements.push(PlanStatement {
1214 statement: edge_hard_delete_statement(id),
1215 guard: Some(AffectedRowGuard::exactly(1)),
1216 });
1217 } else {
1218 let now = chrono::Utc::now().timestamp_micros();
1219 statements.push(PlanStatement {
1220 statement: edge_soft_delete_statement(id, now),
1221 guard: Some(AffectedRowGuard::exactly(1)),
1222 });
1223 }
1224
1225 statements.extend(event_append_statements(
1226 &namespace,
1227 "delete",
1228 EventKind::EdgeDeleted,
1229 SubstrateKind::Entity,
1230 id,
1231 serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
1232 )?);
1233
1234 Ok(AtomicOpPlan::Delete(DeletePlan {
1235 target_id: id,
1236 statements,
1237 post_commit: PostCommitEffect::None,
1238 }))
1239}
1240
1241fn parse_edge_relation(raw: &str) -> RuntimeResult<EdgeRelation> {
1246 raw.parse::<EdgeRelation>()
1247 .map_err(|e| RuntimeError::InvalidInput(format!("unknown edge relation {raw:?}: {e}")))
1248}
1249
1250async fn prepare_link(
1251 runtime: &KhiveRuntime,
1252 token: &NamespaceToken,
1253 args: &Value,
1254) -> RuntimeResult<AtomicOpPlan> {
1255 let source_id = require_uuid(args, "source_id")?;
1256 let target_id = require_uuid(args, "target_id")?;
1257 let relation = parse_edge_relation(require_str(args, "relation")?)?;
1258 let weight = optional_f64(args, "weight")?.unwrap_or(1.0);
1259 let metadata = obj(args)?.get("metadata").cloned();
1260
1261 let mut metadata = crate::merge_entry_metadata(
1267 metadata,
1268 optional_str(args, "dependency_kind").map(String::from),
1269 )?;
1270
1271 validate_edge_weight(weight)?;
1272 runtime
1273 .validate_edge_relation_endpoints(token, source_id, target_id, relation)
1274 .await?;
1275
1276 let (canon_source, canon_target) = canonical_edge_endpoints(relation, source_id, target_id);
1277
1278 if relation == EdgeRelation::DependsOn {
1285 metadata = match (
1286 runtime.resolve_edge_endpoint(token, canon_source).await?,
1287 runtime.resolve_edge_endpoint(token, canon_target).await?,
1288 ) {
1289 (Some(Resolved::Entity(src_e)), Some(Resolved::Entity(tgt_e))) => {
1290 merge_dependency_kind(&src_e.kind, &tgt_e.kind, metadata)
1291 }
1292 _ => metadata,
1293 };
1294 }
1295
1296 validate_edge_metadata(relation, metadata.as_ref())?;
1297 let edge_id = Uuid::new_v4();
1298 let namespace = token.namespace().as_str().to_string();
1299 let now = chrono::Utc::now().timestamp_micros();
1300 let metadata_str = metadata.map(|m| serde_json::to_string(&m).unwrap_or_default());
1301
1302 let statement = edge_insert_guarded_by_endpoints_statement(
1315 &namespace,
1316 edge_id,
1317 canon_source,
1318 canon_target,
1319 relation,
1320 weight,
1321 now,
1322 metadata_str.as_deref(),
1323 );
1324
1325 Ok(AtomicOpPlan::Link(LinkPlan {
1326 source_id: canon_source,
1327 target_id: canon_target,
1328 statement: PlanStatement {
1329 statement,
1330 guard: Some(AffectedRowGuard::exactly(1)),
1331 },
1332 }))
1333}
1334
1335async fn prepare_merge(
1349 runtime: &KhiveRuntime,
1350 token: &NamespaceToken,
1351 args: &Value,
1352) -> RuntimeResult<AtomicOpPlan> {
1353 let into_id = require_uuid(args, "into_id")?;
1354 let from_id = require_uuid(args, "from_id")?;
1355 if into_id == from_id {
1356 return Err(RuntimeError::InvalidInput(
1357 "cannot merge an entity into itself".into(),
1358 ));
1359 }
1360
1361 let entities = runtime.entities(token)?;
1362 entities
1363 .get_entity(into_id)
1364 .await?
1365 .ok_or_else(|| RuntimeError::NotFound(format!("entity {into_id}")))?;
1366 entities
1367 .get_entity(from_id)
1368 .await?
1369 .ok_or_else(|| RuntimeError::NotFound(format!("entity {from_id}")))?;
1370
1371 let now = chrono::Utc::now().timestamp_micros();
1372 let rewires = vec![
1373 crate::atomic_plan::PlanPredicate {
1374 description: "source_id = :from".to_string(),
1375 statement: SqlStatement {
1376 sql: "UPDATE graph_edges SET source_id = ?1, updated_at = ?2 WHERE source_id = ?3"
1377 .to_string(),
1378 params: vec![
1379 SqlValue::Text(into_id.to_string()),
1380 SqlValue::Integer(now),
1381 SqlValue::Text(from_id.to_string()),
1382 ],
1383 label: Some("atomic-merge-rewire-source".to_string()),
1384 },
1385 },
1386 crate::atomic_plan::PlanPredicate {
1387 description: "target_id = :from".to_string(),
1388 statement: SqlStatement {
1389 sql: "UPDATE graph_edges SET target_id = ?1, updated_at = ?2 WHERE target_id = ?3"
1390 .to_string(),
1391 params: vec![
1392 SqlValue::Text(into_id.to_string()),
1393 SqlValue::Integer(now),
1394 SqlValue::Text(from_id.to_string()),
1395 ],
1396 label: Some("atomic-merge-rewire-target".to_string()),
1397 },
1398 },
1399 ];
1400 let lifecycle = vec![PlanStatement {
1401 statement: SqlStatement {
1402 sql: "UPDATE entities SET deleted_at = ?1, merged_into = ?2 \
1403 WHERE id = ?3 AND deleted_at IS NULL"
1404 .to_string(),
1405 params: vec![
1406 SqlValue::Integer(now),
1407 SqlValue::Text(into_id.to_string()),
1408 SqlValue::Text(from_id.to_string()),
1409 ],
1410 label: Some("atomic-merge-tombstone-from-entity".to_string()),
1411 },
1412 guard: Some(AffectedRowGuard::exactly(1)),
1413 }];
1414
1415 Ok(AtomicOpPlan::Merge(MergePlan {
1416 into_id,
1417 from_id,
1418 rewires,
1419 lifecycle,
1420 }))
1421}
1422
1423pub async fn apply_post_commit_effects(
1432 runtime: &KhiveRuntime,
1433 token: &NamespaceToken,
1434 effects: Vec<PostCommitEffect>,
1435) -> RuntimeResult<()> {
1436 for effect in effects {
1437 match effect {
1438 PostCommitEffect::None => {}
1439 PostCommitEffect::ReindexEntity { entity_id } => {
1440 if let Some(entity) = runtime.entities(token)?.get_entity(entity_id).await? {
1441 runtime.reindex_entity(token, &entity).await?;
1442 }
1443 }
1444 PostCommitEffect::ReindexNote { note_id } => {
1445 if let Some(note) = runtime.notes(token)?.get_note(note_id).await? {
1446 runtime.reindex_note(token, ¬e).await?;
1447 runtime.fire_note_mutation_hook(¬e.kind, note.id).await;
1455 }
1456 }
1457 PostCommitEffect::NoteDeleted { note_id, kind } => {
1458 runtime.fire_note_mutation_hook(&kind, note_id).await;
1467 }
1468 PostCommitEffect::GtdAudit { .. } => {
1469 }
1476 }
1477 }
1478 Ok(())
1479}
1480
1481#[cfg(test)]
1482mod tests {
1483 use super::*;
1484
1485 use async_trait::async_trait;
1486 use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService};
1487 use serde_json::json;
1488
1489 use khive_types::Namespace;
1490
1491 use crate::embedder_registry::EmbedderProvider;
1492 use crate::runtime::RuntimeConfig;
1493
1494 const STUB_MODEL: &str = "stub-adr099-b3";
1495 const STUB_DIMS: usize = 4;
1496
1497 struct StubService;
1498
1499 #[async_trait]
1500 impl EmbeddingService for StubService {
1501 async fn embed(
1502 &self,
1503 texts: &[String],
1504 _model: EmbeddingModel,
1505 ) -> Result<Vec<Vec<f32>>, EmbedError> {
1506 Ok(texts.iter().map(|_| vec![0.5_f32; STUB_DIMS]).collect())
1507 }
1508
1509 fn supports_model(&self, _model: EmbeddingModel) -> bool {
1510 true
1511 }
1512
1513 fn name(&self) -> &'static str {
1514 STUB_MODEL
1515 }
1516 }
1517
1518 struct StubProvider;
1519
1520 #[async_trait]
1521 impl EmbedderProvider for StubProvider {
1522 fn name(&self) -> &str {
1523 STUB_MODEL
1524 }
1525
1526 fn dimensions(&self) -> usize {
1527 STUB_DIMS
1528 }
1529
1530 async fn build(&self) -> RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
1531 Ok(std::sync::Arc::new(StubService))
1532 }
1533 }
1534
1535 fn scratch_runtime() -> KhiveRuntime {
1536 let dir = tempfile::tempdir().expect("tempdir");
1537 let path = dir.path().join("atomic_prepare_reindex.db");
1538 let rt = KhiveRuntime::new(RuntimeConfig {
1539 db_path: Some(path),
1540 embedding_model: None,
1541 additional_embedding_models: vec![],
1542 ..RuntimeConfig::default()
1543 })
1544 .expect("runtime");
1545 std::mem::forget(dir);
1546 rt
1547 }
1548
1549 #[tokio::test]
1557 async fn atomic_update_entity_rejects_note_only_field_salience() {
1558 let runtime = scratch_runtime();
1559 let token = runtime
1560 .authorize(Namespace::parse("local").expect("ns"))
1561 .expect("authorize");
1562 let entity = khive_storage::Entity::new("local", "concept", "GapFourEntity");
1563 let entity_id = entity.id;
1564 runtime
1565 .entities(&token)
1566 .expect("entities store")
1567 .upsert_entity(entity)
1568 .await
1569 .expect("seed entity");
1570
1571 let err = prepare_update(
1572 &runtime,
1573 &token,
1574 &json!({"id": entity_id.to_string(), "salience": 0.9}),
1575 None,
1576 )
1577 .await
1578 .expect_err("salience on an entity must be rejected, not silently accepted");
1579 assert!(
1580 matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("salience") && msg.contains("not valid for an entity")),
1581 "expected an InvalidInput naming the offending field, got: {err:?}"
1582 );
1583
1584 let plan = prepare_update(
1586 &runtime,
1587 &token,
1588 &json!({
1589 "id": entity_id.to_string(),
1590 "name": "GapFourEntity-renamed",
1591 "description": "updated description",
1592 "tags": ["a", "b"],
1593 }),
1594 None,
1595 )
1596 .await
1597 .expect("a valid entity field set must still be accepted");
1598 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1599 .await
1600 .expect("seam call ok");
1601 assert!(matches!(
1602 outcome,
1603 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1604 ));
1605 let entity = runtime
1606 .get_entity(&token, entity_id)
1607 .await
1608 .expect("get_entity");
1609 assert_eq!(entity.name, "GapFourEntity-renamed");
1610 }
1611
1612 #[tokio::test]
1616 async fn atomic_update_note_rejects_entity_only_field_description() {
1617 let runtime = scratch_runtime();
1618 let token = runtime
1619 .authorize(Namespace::parse("local").expect("ns"))
1620 .expect("authorize");
1621 let mut note = khive_storage::note::Note::new("local", "observation", "gap-4 note content");
1622 note.name = Some("gap-four-note".to_string());
1623 let note_id = note.id;
1624 runtime
1625 .notes(&token)
1626 .expect("notes store")
1627 .upsert_note(note)
1628 .await
1629 .expect("seed note");
1630
1631 let err = prepare_update(
1632 &runtime,
1633 &token,
1634 &json!({"id": note_id.to_string(), "description": "entities have descriptions, notes don't"}),
1635 None,
1636 )
1637 .await
1638 .expect_err("description on a note must be rejected, not silently accepted");
1639 assert!(
1640 matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("description") && msg.contains("not valid for a note")),
1641 "expected an InvalidInput naming the offending field, got: {err:?}"
1642 );
1643
1644 let plan = prepare_update(
1646 &runtime,
1647 &token,
1648 &json!({"id": note_id.to_string(), "content": "gap-4 note content, revised"}),
1649 None,
1650 )
1651 .await
1652 .expect("a valid note field must still be accepted");
1653 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1654 .await
1655 .expect("seam call ok");
1656 assert!(matches!(
1657 outcome,
1658 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1659 ));
1660 }
1661
1662 #[tokio::test]
1667 async fn atomic_update_note_content_is_fts_and_vector_reindexed_post_commit() {
1668 let runtime = scratch_runtime();
1669 runtime.register_embedder(StubProvider);
1670 let token = runtime
1671 .authorize(Namespace::parse("local").expect("ns"))
1672 .expect("authorize");
1673
1674 let mut note = khive_storage::note::Note::new("local", "observation", "original content");
1675 note.name = Some("reindex-target".to_string());
1676 let note_id = note.id;
1677 runtime
1678 .notes(&token)
1679 .expect("notes store")
1680 .upsert_note(note)
1681 .await
1682 .expect("seed note");
1683
1684 let vec_store = runtime
1686 .vectors_for_model(&token, STUB_MODEL)
1687 .expect("vec store");
1688 assert_eq!(vec_store.count().await.expect("count before"), 0);
1689
1690 let plan = prepare_update(
1691 &runtime,
1692 &token,
1693 &json!({"id": note_id.to_string(), "content": "freshly-updated-content-xyz"}),
1694 None,
1695 )
1696 .await
1697 .expect("prepare update");
1698
1699 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1700 .await
1701 .expect("seam call ok");
1702 let post_commit = match outcome {
1703 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1704 other => panic!("expected Committed, got {other:?}"),
1705 };
1706 assert_eq!(
1707 post_commit,
1708 vec![PostCommitEffect::ReindexNote { note_id }],
1709 "content change must schedule exactly one ReindexNote post-commit effect"
1710 );
1711
1712 apply_post_commit_effects(&runtime, &token, post_commit)
1713 .await
1714 .expect("apply post-commit effects");
1715
1716 let doc = runtime
1718 .text_for_notes(&token)
1719 .expect("text store")
1720 .get_document("local", note_id)
1721 .await
1722 .expect("get_document")
1723 .expect("document must be indexed after post-commit reindex");
1724 assert!(
1725 doc.body.contains("freshly-updated-content-xyz"),
1726 "FTS body must reflect the committed content: {:?}",
1727 doc.body
1728 );
1729
1730 assert_eq!(
1732 vec_store.count().await.expect("count after"),
1733 1,
1734 "post-commit reindex must have inserted a vector row for the stub model"
1735 );
1736 }
1737
1738 #[tokio::test]
1750 async fn atomic_note_update_and_delete_post_commit_fire_the_note_mutation_hook() {
1751 let runtime = scratch_runtime();
1752 let token = runtime
1753 .authorize(Namespace::parse("local").expect("ns"))
1754 .expect("authorize");
1755
1756 let fired: std::sync::Arc<std::sync::Mutex<Vec<(String, uuid::Uuid)>>> =
1757 std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
1758 let fired_for_hook = fired.clone();
1759 runtime.install_note_mutation_hook(std::sync::Arc::new(
1760 move |kind: String, id: uuid::Uuid| {
1761 let fired = fired_for_hook.clone();
1762 Box::pin(async move {
1763 fired.lock().expect("lock").push((kind, id));
1764 })
1765 },
1766 ));
1767
1768 let mut note = khive_storage::note::Note::new("local", "observation", "hook-update-target");
1770 note.name = Some("hook-update-target".to_string());
1771 let update_note_id = note.id;
1772 runtime
1773 .notes(&token)
1774 .expect("notes store")
1775 .upsert_note(note)
1776 .await
1777 .expect("seed update-target note");
1778
1779 let plan = prepare_update(
1780 &runtime,
1781 &token,
1782 &json!({"id": update_note_id.to_string(), "content": "hook-update-target, revised"}),
1783 None,
1784 )
1785 .await
1786 .expect("prepare update");
1787 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1788 .await
1789 .expect("seam call ok");
1790 let post_commit = match outcome {
1791 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1792 other => panic!("expected Committed, got {other:?}"),
1793 };
1794 apply_post_commit_effects(&runtime, &token, post_commit)
1795 .await
1796 .expect("apply post-commit effects (update)");
1797
1798 let mut del_note =
1801 khive_storage::note::Note::new("local", "observation", "hook-delete-target");
1802 del_note.name = Some("hook-delete-target".to_string());
1803 let delete_note_id = del_note.id;
1804 runtime
1805 .notes(&token)
1806 .expect("notes store")
1807 .upsert_note(del_note)
1808 .await
1809 .expect("seed delete-target note");
1810
1811 let plan = prepare_delete(
1812 &runtime,
1813 &token,
1814 &json!({"id": delete_note_id.to_string(), "hard": false}),
1815 None,
1816 )
1817 .await
1818 .expect("prepare delete");
1819 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1820 .await
1821 .expect("seam call ok");
1822 let post_commit = match outcome {
1823 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
1824 other => panic!("expected Committed, got {other:?}"),
1825 };
1826 assert_eq!(
1827 post_commit,
1828 vec![PostCommitEffect::NoteDeleted {
1829 note_id: delete_note_id,
1830 kind: "observation".to_string(),
1831 }],
1832 "a committed note delete must schedule exactly one NoteDeleted post-commit effect"
1833 );
1834 apply_post_commit_effects(&runtime, &token, post_commit)
1835 .await
1836 .expect("apply post-commit effects (delete)");
1837
1838 let seen = fired.lock().expect("lock").clone();
1839 assert!(
1840 seen.contains(&("observation".to_string(), update_note_id)),
1841 "the note-mutation hook must fire for the atomic UPDATE path: {seen:?}"
1842 );
1843 assert!(
1844 seen.contains(&("observation".to_string(), delete_note_id)),
1845 "the note-mutation hook must fire for the atomic DELETE path: {seen:?}"
1846 );
1847 }
1848
1849 #[tokio::test]
1853 async fn atomic_delete_note_purges_fts_and_vector_indexes_soft_and_hard() {
1854 let runtime = scratch_runtime();
1855 runtime.register_embedder(StubProvider);
1856 let token = runtime
1857 .authorize(Namespace::parse("local").expect("ns"))
1858 .expect("authorize");
1859
1860 for hard in [false, true] {
1861 let mut note =
1862 khive_storage::note::Note::new("local", "observation", "purge-target content");
1863 note.name = Some(format!("purge-target-hard-{hard}"));
1864 let note_id = note.id;
1865 runtime
1866 .notes(&token)
1867 .expect("notes store")
1868 .upsert_note(note.clone())
1869 .await
1870 .expect("seed note");
1871 runtime
1872 .reindex_note(&token, ¬e)
1873 .await
1874 .expect("seed index rows");
1875
1876 let vec_store = runtime
1877 .vectors_for_model(&token, STUB_MODEL)
1878 .expect("vec store");
1879 assert_eq!(
1880 vec_store.count().await.expect("count before"),
1881 1,
1882 "seeded note must have a vector row before delete (hard={hard})"
1883 );
1884 assert!(
1885 runtime
1886 .text_for_notes(&token)
1887 .expect("text store")
1888 .get_document("local", note_id)
1889 .await
1890 .expect("get_document")
1891 .is_some(),
1892 "seeded note must have an FTS row before delete (hard={hard})"
1893 );
1894
1895 let plan = prepare_delete(
1896 &runtime,
1897 &token,
1898 &json!({"id": note_id.to_string(), "hard": hard}),
1899 None,
1900 )
1901 .await
1902 .expect("prepare delete");
1903 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1904 .await
1905 .expect("seam call ok");
1906 assert!(
1907 matches!(
1908 outcome,
1909 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1910 ),
1911 "expected commit (hard={hard}): {outcome:?}"
1912 );
1913
1914 assert!(
1915 runtime
1916 .text_for_notes(&token)
1917 .expect("text store")
1918 .get_document("local", note_id)
1919 .await
1920 .expect("get_document")
1921 .is_none(),
1922 "FTS row must be purged after atomic delete (hard={hard})"
1923 );
1924 assert_eq!(
1925 vec_store.count().await.expect("count after"),
1926 0,
1927 "vector row must be purged after atomic delete (hard={hard})"
1928 );
1929 }
1930 }
1931
1932 #[tokio::test]
1936 async fn atomic_delete_entity_purges_fts_and_vector_indexes_soft_and_hard() {
1937 let runtime = scratch_runtime();
1938 runtime.register_embedder(StubProvider);
1939 let token = runtime
1940 .authorize(Namespace::parse("local").expect("ns"))
1941 .expect("authorize");
1942
1943 for hard in [false, true] {
1944 let entity =
1945 khive_storage::Entity::new("local", "concept", format!("purge-target-hard-{hard}"));
1946 let entity_id = entity.id;
1947 runtime
1948 .entities(&token)
1949 .expect("entities store")
1950 .upsert_entity(entity.clone())
1951 .await
1952 .expect("seed entity");
1953 runtime
1954 .reindex_entity(&token, &entity)
1955 .await
1956 .expect("seed index rows");
1957
1958 let vec_store = runtime
1959 .vectors_for_model(&token, STUB_MODEL)
1960 .expect("vec store");
1961 assert_eq!(
1962 vec_store.count().await.expect("count before"),
1963 1,
1964 "seeded entity must have a vector row before delete (hard={hard})"
1965 );
1966 assert!(
1967 runtime
1968 .text(&token)
1969 .expect("text store")
1970 .get_document("local", entity_id)
1971 .await
1972 .expect("get_document")
1973 .is_some(),
1974 "seeded entity must have an FTS row before delete (hard={hard})"
1975 );
1976
1977 let plan = prepare_delete(
1978 &runtime,
1979 &token,
1980 &json!({"id": entity_id.to_string(), "hard": hard}),
1981 None,
1982 )
1983 .await
1984 .expect("prepare delete");
1985 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
1986 .await
1987 .expect("seam call ok");
1988 assert!(
1989 matches!(
1990 outcome,
1991 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
1992 ),
1993 "expected commit (hard={hard}): {outcome:?}"
1994 );
1995
1996 assert!(
1997 runtime
1998 .text(&token)
1999 .expect("text store")
2000 .get_document("local", entity_id)
2001 .await
2002 .expect("get_document")
2003 .is_none(),
2004 "FTS row must be purged after atomic delete (hard={hard})"
2005 );
2006 assert_eq!(
2007 vec_store.count().await.expect("count after"),
2008 0,
2009 "vector row must be purged after atomic delete (hard={hard})"
2010 );
2011 }
2012 }
2013
2014 #[tokio::test]
2020 async fn atomic_link_persists_explicit_dependency_kind_and_infers_when_absent() {
2021 let runtime = scratch_runtime();
2022 let token = runtime
2023 .authorize(Namespace::parse("local").expect("ns"))
2024 .expect("authorize");
2025 let entities = runtime.entities(&token).expect("entities store");
2026
2027 fn metadata_json(plan: &AtomicOpPlan) -> String {
2028 let link_plan = match plan {
2029 AtomicOpPlan::Link(p) => p,
2030 other => panic!("expected an AtomicOpPlan::Link, got {other:?}"),
2031 };
2032 match link_plan.statement.statement.params.last() {
2033 Some(SqlValue::Text(s)) => s.clone(),
2034 other => panic!("expected the metadata param to be SqlValue::Text, got {other:?}"),
2035 }
2036 }
2037
2038 {
2040 let svc = khive_storage::Entity::new("local", "service", "SvcA");
2041 let proj = khive_storage::Entity::new("local", "project", "ProjB");
2042 let (svc_id, proj_id) = (svc.id, proj.id);
2043 entities.upsert_entity(svc).await.expect("seed svc");
2044 entities.upsert_entity(proj).await.expect("seed proj");
2045
2046 let plan = prepare_link(
2047 &runtime,
2048 &token,
2049 &json!({
2050 "source_id": svc_id.to_string(),
2051 "target_id": proj_id.to_string(),
2052 "relation": "depends_on",
2053 "dependency_kind": "artifact",
2054 }),
2055 )
2056 .await
2057 .expect("prepare link");
2058 let json_str = metadata_json(&plan);
2059 assert!(
2060 json_str.contains(r#""dependency_kind":"artifact""#),
2061 "explicit dependency_kind param must persist: {json_str}"
2062 );
2063 }
2064
2065 {
2068 let svc_a = khive_storage::Entity::new("local", "service", "SvcC");
2069 let svc_b = khive_storage::Entity::new("local", "service", "SvcD");
2070 let (a_id, b_id) = (svc_a.id, svc_b.id);
2071 entities.upsert_entity(svc_a).await.expect("seed svc a");
2072 entities.upsert_entity(svc_b).await.expect("seed svc b");
2073
2074 let plan = prepare_link(
2075 &runtime,
2076 &token,
2077 &json!({
2078 "source_id": a_id.to_string(),
2079 "target_id": b_id.to_string(),
2080 "relation": "depends_on",
2081 }),
2082 )
2083 .await
2084 .expect("prepare link");
2085 let json_str = metadata_json(&plan);
2086 assert!(
2087 json_str.contains(r#""dependency_kind":"runtime""#),
2088 "inferred dependency_kind for (service, service) must persist: {json_str}"
2089 );
2090 }
2091 }
2092
2093 async fn probe_edge_natural_key(
2098 runtime: &KhiveRuntime,
2099 namespace: &str,
2100 source_id: Uuid,
2101 target_id: Uuid,
2102 relation: &str,
2103 ) -> (usize, Option<f64>, Option<String>, Option<i64>) {
2104 let mut reader = runtime.sql().reader().await.expect("reader");
2105 let rows = reader
2106 .query_all(SqlStatement {
2107 sql: "SELECT weight, metadata, deleted_at FROM graph_edges \
2108 WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 AND relation = ?4"
2109 .to_string(),
2110 params: vec![
2111 SqlValue::Text(namespace.to_string()),
2112 SqlValue::Text(source_id.to_string()),
2113 SqlValue::Text(target_id.to_string()),
2114 SqlValue::Text(relation.to_string()),
2115 ],
2116 label: Some("test-probe-edge-natural-key".to_string()),
2117 })
2118 .await
2119 .expect("probe edge natural key");
2120 let count = rows.len();
2121 let Some(row) = rows.into_iter().next() else {
2122 return (count, None, None, None);
2123 };
2124 let weight = match row.get("weight") {
2125 Some(SqlValue::Float(f)) => Some(*f),
2126 Some(SqlValue::Integer(i)) => Some(*i as f64),
2127 _ => None,
2128 };
2129 let metadata = match row.get("metadata") {
2130 Some(SqlValue::Text(s)) => Some(s.clone()),
2131 _ => None,
2132 };
2133 let deleted_at = match row.get("deleted_at") {
2134 Some(SqlValue::Integer(i)) => Some(*i),
2135 _ => None,
2136 };
2137 (count, weight, metadata, deleted_at)
2138 }
2139
2140 #[tokio::test]
2146 async fn atomic_link_of_already_linked_triple_upserts_weight_and_metadata() {
2147 let runtime = scratch_runtime();
2148 let token = runtime
2149 .authorize(Namespace::parse("local").expect("ns"))
2150 .expect("authorize");
2151 let entities = runtime.entities(&token).expect("entities store");
2152
2153 let a = khive_storage::Entity::new("local", "concept", "GapTwoA");
2154 let b = khive_storage::Entity::new("local", "concept", "GapTwoB");
2155 let (a_id, b_id) = (a.id, b.id);
2156 entities.upsert_entity(a).await.expect("seed a");
2157 entities.upsert_entity(b).await.expect("seed b");
2158
2159 let plan1 = prepare_link(
2161 &runtime,
2162 &token,
2163 &json!({
2164 "source_id": a_id.to_string(),
2165 "target_id": b_id.to_string(),
2166 "relation": "extends",
2167 "weight": 0.5,
2168 }),
2169 )
2170 .await
2171 .expect("prepare first link");
2172 let outcome1 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan1])
2173 .await
2174 .expect("seam call ok");
2175 assert!(
2176 matches!(
2177 outcome1,
2178 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2179 ),
2180 "fresh link must commit: {outcome1:?}"
2181 );
2182 let (count, weight, _metadata, deleted_at) =
2183 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2184 assert_eq!(count, 1, "exactly one edge row after the fresh link");
2185 assert_eq!(weight, Some(0.5));
2186 assert!(deleted_at.is_none());
2187
2188 let plan2 = prepare_link(
2192 &runtime,
2193 &token,
2194 &json!({
2195 "source_id": a_id.to_string(),
2196 "target_id": b_id.to_string(),
2197 "relation": "extends",
2198 "weight": 0.9,
2199 "metadata": {"note": "relinked"},
2200 }),
2201 )
2202 .await
2203 .expect("prepare second link");
2204 let outcome2 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan2])
2205 .await
2206 .expect("seam call ok");
2207 assert!(
2208 matches!(
2209 outcome2,
2210 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2211 ),
2212 "re-link of an already-linked triple must upsert, not roll back: {outcome2:?}"
2213 );
2214 let (count, weight, metadata, deleted_at) =
2215 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2216 assert_eq!(
2217 count, 1,
2218 "the natural-key UNIQUE constraint must still hold exactly one row (upsert, not a second insert)"
2219 );
2220 assert_eq!(weight, Some(0.9), "weight must be updated to the new value");
2221 assert!(
2222 metadata
2223 .as_deref()
2224 .is_some_and(|m| m.contains(r#""note":"relinked""#)),
2225 "metadata must be updated to the new value: {metadata:?}"
2226 );
2227 assert!(deleted_at.is_none());
2228 }
2229
2230 #[tokio::test]
2234 async fn atomic_link_of_soft_deleted_triple_resurrects_it() {
2235 let runtime = scratch_runtime();
2236 let token = runtime
2237 .authorize(Namespace::parse("local").expect("ns"))
2238 .expect("authorize");
2239 let entities = runtime.entities(&token).expect("entities store");
2240
2241 let a = khive_storage::Entity::new("local", "concept", "GapTwoResurrectA");
2242 let b = khive_storage::Entity::new("local", "concept", "GapTwoResurrectB");
2243 let (a_id, b_id) = (a.id, b.id);
2244 entities.upsert_entity(a).await.expect("seed a");
2245 entities.upsert_entity(b).await.expect("seed b");
2246
2247 let plan = prepare_link(
2248 &runtime,
2249 &token,
2250 &json!({
2251 "source_id": a_id.to_string(),
2252 "target_id": b_id.to_string(),
2253 "relation": "extends",
2254 }),
2255 )
2256 .await
2257 .expect("prepare link");
2258 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2259 .await
2260 .expect("seam call ok");
2261 assert!(matches!(
2262 outcome,
2263 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2264 ));
2265
2266 {
2269 let mut writer = runtime.sql().writer().await.expect("writer");
2270 let affected = writer
2271 .execute(SqlStatement {
2272 sql: "UPDATE graph_edges SET deleted_at = ?1 \
2273 WHERE namespace = ?2 AND source_id = ?3 AND target_id = ?4 AND relation = ?5"
2274 .to_string(),
2275 params: vec![
2276 SqlValue::Integer(chrono::Utc::now().timestamp_micros()),
2277 SqlValue::Text("local".to_string()),
2278 SqlValue::Text(a_id.to_string()),
2279 SqlValue::Text(b_id.to_string()),
2280 SqlValue::Text("extends".to_string()),
2281 ],
2282 label: Some("test-soft-delete-edge".to_string()),
2283 })
2284 .await
2285 .expect("soft delete edge");
2286 assert_eq!(affected, 1, "soft-delete must touch exactly the seeded row");
2287 }
2288 let (_, _, _, deleted_at) =
2289 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2290 assert!(
2291 deleted_at.is_some(),
2292 "row must be soft-deleted before the resurrect attempt"
2293 );
2294
2295 let plan_relink = prepare_link(
2298 &runtime,
2299 &token,
2300 &json!({
2301 "source_id": a_id.to_string(),
2302 "target_id": b_id.to_string(),
2303 "relation": "extends",
2304 "weight": 0.75,
2305 }),
2306 )
2307 .await
2308 .expect("prepare resurrecting link");
2309 let outcome_relink =
2310 crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan_relink])
2311 .await
2312 .expect("seam call ok");
2313 assert!(
2314 matches!(
2315 outcome_relink,
2316 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2317 ),
2318 "re-linking a soft-deleted triple must resurrect it, not roll back: {outcome_relink:?}"
2319 );
2320 let (count, weight, _, deleted_at) =
2321 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
2322 assert_eq!(count, 1);
2323 assert_eq!(weight, Some(0.75));
2324 assert!(
2325 deleted_at.is_none(),
2326 "re-link must resurrect the soft-deleted row (deleted_at -> NULL)"
2327 );
2328 }
2329
2330 #[tokio::test]
2338 async fn atomic_delete_succeeds_when_vec_table_never_created() {
2339 let runtime = scratch_runtime();
2340 runtime.register_embedder(StubProvider);
2341 let token = runtime
2342 .authorize(Namespace::parse("local").expect("ns"))
2343 .expect("authorize");
2344
2345 let entity = khive_storage::Entity::new("local", "concept", "no-vec-table-entity");
2349 let entity_id = entity.id;
2350 runtime
2351 .entities(&token)
2352 .expect("entities store")
2353 .upsert_entity(entity)
2354 .await
2355 .expect("seed entity");
2356
2357 let mut note = khive_storage::note::Note::new("local", "observation", "no-vec-table-note");
2358 note.name = Some("no-vec-table-note".to_string());
2359 let note_id = note.id;
2360 runtime
2361 .notes(&token)
2362 .expect("notes store")
2363 .upsert_note(note)
2364 .await
2365 .expect("seed note");
2366
2367 for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
2368 let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
2369 .await
2370 .unwrap_or_else(|e| panic!("prepare delete ({kind}) must not fail: {e}"));
2371 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2372 .await
2373 .unwrap_or_else(|e| {
2374 panic!("atomic delete ({kind}) must not hit `no such table`: {e}")
2375 });
2376 assert!(
2377 matches!(
2378 outcome,
2379 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2380 ),
2381 "expected a clean commit ({kind}): {outcome:?}"
2382 );
2383 }
2384
2385 assert!(
2386 runtime
2387 .get_entity_including_deleted(&token, entity_id)
2388 .await
2389 .expect("get entity")
2390 .expect("entity row still present (soft delete)")
2391 .deleted_at
2392 .is_some(),
2393 "entity must be soft-deleted"
2394 );
2395 assert!(
2396 runtime
2397 .get_note_including_deleted(&token, note_id)
2398 .await
2399 .expect("get note")
2400 .expect("note row still present (soft delete)")
2401 .deleted_at
2402 .is_some(),
2403 "note must be soft-deleted"
2404 );
2405 }
2406
2407 #[tokio::test]
2413 async fn atomic_hard_delete_purges_already_soft_deleted_entity_and_note() {
2414 let runtime = scratch_runtime();
2415 runtime.register_embedder(StubProvider);
2416 let token = runtime
2417 .authorize(Namespace::parse("local").expect("ns"))
2418 .expect("authorize");
2419
2420 let entity =
2421 khive_storage::Entity::new("local", "concept", "tombstoned-entity-hard-delete");
2422 let entity_id = entity.id;
2423 runtime
2424 .entities(&token)
2425 .expect("entities store")
2426 .upsert_entity(entity.clone())
2427 .await
2428 .expect("seed entity");
2429 runtime
2430 .reindex_entity(&token, &entity)
2431 .await
2432 .expect("seed index rows");
2433
2434 let mut note =
2435 khive_storage::note::Note::new("local", "observation", "tombstoned-note-hard-delete");
2436 note.name = Some("tombstoned-note-hard-delete".to_string());
2437 let note_id = note.id;
2438 runtime
2439 .notes(&token)
2440 .expect("notes store")
2441 .upsert_note(note.clone())
2442 .await
2443 .expect("seed note");
2444 runtime
2445 .reindex_note(&token, ¬e)
2446 .await
2447 .expect("seed index rows");
2448
2449 for id in [entity_id, note_id] {
2452 let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
2453 .await
2454 .expect("prepare soft delete");
2455 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2456 .await
2457 .expect("soft delete commit");
2458 assert!(matches!(
2459 outcome,
2460 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2461 ));
2462 }
2463 assert!(
2464 runtime
2465 .get_entity_including_deleted(&token, entity_id)
2466 .await
2467 .expect("get entity")
2468 .expect("entity present")
2469 .deleted_at
2470 .is_some(),
2471 "entity must be soft-deleted before the hard-delete attempt"
2472 );
2473 assert!(
2474 runtime
2475 .get_note_including_deleted(&token, note_id)
2476 .await
2477 .expect("get note")
2478 .expect("note present")
2479 .deleted_at
2480 .is_some(),
2481 "note must be soft-deleted before the hard-delete attempt"
2482 );
2483
2484 for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
2486 let plan = prepare_delete(
2487 &runtime,
2488 &token,
2489 &json!({"id": id.to_string(), "hard": true}),
2490 None,
2491 )
2492 .await
2493 .unwrap_or_else(|e| {
2494 panic!(
2495 "prepare hard delete ({kind}) of an already-soft-deleted record \
2496 must resolve it: {e}"
2497 )
2498 });
2499 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2500 .await
2501 .unwrap_or_else(|e| panic!("hard delete ({kind}) commit failed: {e}"));
2502 assert!(
2503 matches!(
2504 outcome,
2505 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2506 ),
2507 "expected a clean hard-delete commit ({kind}): {outcome:?}"
2508 );
2509 }
2510
2511 assert!(
2512 runtime
2513 .get_entity_including_deleted(&token, entity_id)
2514 .await
2515 .expect("get entity")
2516 .is_none(),
2517 "entity row must be fully purged after hard delete"
2518 );
2519 assert!(
2520 runtime
2521 .get_note_including_deleted(&token, note_id)
2522 .await
2523 .expect("get note")
2524 .is_none(),
2525 "note row must be fully purged after hard delete"
2526 );
2527 assert!(
2528 runtime
2529 .text(&token)
2530 .expect("text store")
2531 .get_document("local", entity_id)
2532 .await
2533 .expect("get_document")
2534 .is_none(),
2535 "entity FTS row must be purged after hard delete"
2536 );
2537 assert!(
2538 runtime
2539 .text_for_notes(&token)
2540 .expect("text store")
2541 .get_document("local", note_id)
2542 .await
2543 .expect("get_document")
2544 .is_none(),
2545 "note FTS row must be purged after hard delete"
2546 );
2547 let vec_store = runtime
2548 .vectors_for_model(&token, STUB_MODEL)
2549 .expect("vec store");
2550 assert_eq!(
2551 vec_store.count().await.expect("count after"),
2552 0,
2553 "vector rows for both records must be purged after hard delete"
2554 );
2555 }
2556
2557 async fn events_for_target(
2565 runtime: &KhiveRuntime,
2566 token: &NamespaceToken,
2567 target_id: Uuid,
2568 kind: EventKind,
2569 ) -> Vec<khive_storage::Event> {
2570 let event_store = runtime.events(token).expect("event store");
2571 let filter = khive_storage::EventFilter {
2572 kinds: vec![kind],
2573 ..Default::default()
2574 };
2575 let page = event_store
2576 .query_events(filter, khive_storage::types::PageRequest::default())
2577 .await
2578 .expect("query_events");
2579 page.items
2580 .into_iter()
2581 .filter(|e| e.target_id == Some(target_id))
2582 .collect()
2583 }
2584
2585 #[tokio::test]
2590 async fn atomic_update_entity_appends_entity_updated_event() {
2591 let runtime = scratch_runtime();
2592 let token = runtime
2593 .authorize(Namespace::parse("local").expect("ns"))
2594 .expect("authorize");
2595 let entity = khive_storage::Entity::new("local", "concept", "gap1-entity");
2596 let entity_id = entity.id;
2597 runtime
2598 .entities(&token)
2599 .expect("entities store")
2600 .upsert_entity(entity)
2601 .await
2602 .expect("seed entity");
2603
2604 let plan = prepare_update(
2605 &runtime,
2606 &token,
2607 &json!({"id": entity_id.to_string(), "name": "gap1-entity-renamed"}),
2608 None,
2609 )
2610 .await
2611 .expect("prepare update");
2612 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2613 .await
2614 .expect("seam call ok");
2615 assert!(matches!(
2616 outcome,
2617 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2618 ));
2619
2620 let events = events_for_target(&runtime, &token, entity_id, EventKind::EntityUpdated).await;
2621 assert_eq!(
2622 events.len(),
2623 1,
2624 "expected exactly one EntityUpdated event for {entity_id}"
2625 );
2626 assert_eq!(events[0].namespace, "local");
2627 assert_eq!(events[0].payload["id"], json!(entity_id.to_string()));
2628 assert_eq!(
2629 events[0].payload["changed_fields"],
2630 json!(["name"]),
2631 "changed_fields must name exactly the patched fields"
2632 );
2633 }
2634
2635 #[tokio::test]
2639 async fn atomic_delete_entity_appends_entity_deleted_event_soft_and_hard() {
2640 let runtime = scratch_runtime();
2641 let token = runtime
2642 .authorize(Namespace::parse("local").expect("ns"))
2643 .expect("authorize");
2644
2645 for hard in [false, true] {
2646 let entity =
2647 khive_storage::Entity::new("local", "concept", format!("gap1-entity-hard-{hard}"));
2648 let entity_id = entity.id;
2649 runtime
2650 .entities(&token)
2651 .expect("entities store")
2652 .upsert_entity(entity)
2653 .await
2654 .expect("seed entity");
2655
2656 let args = if hard {
2657 json!({"id": entity_id.to_string(), "hard": true})
2658 } else {
2659 json!({"id": entity_id.to_string()})
2660 };
2661 let plan = prepare_delete(&runtime, &token, &args, None)
2662 .await
2663 .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
2664 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2665 .await
2666 .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
2667 assert!(
2668 matches!(
2669 outcome,
2670 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2671 ),
2672 "expected a clean delete commit (hard={hard}): {outcome:?}"
2673 );
2674
2675 let events =
2676 events_for_target(&runtime, &token, entity_id, EventKind::EntityDeleted).await;
2677 assert_eq!(
2678 events.len(),
2679 1,
2680 "expected exactly one EntityDeleted event for {entity_id} (hard={hard})"
2681 );
2682 assert_eq!(events[0].payload["hard"], json!(hard));
2683 }
2684 }
2685
2686 #[tokio::test]
2690 async fn atomic_delete_note_appends_note_deleted_event_soft_and_hard() {
2691 let runtime = scratch_runtime();
2692 let token = runtime
2693 .authorize(Namespace::parse("local").expect("ns"))
2694 .expect("authorize");
2695
2696 for hard in [false, true] {
2697 let mut note = khive_storage::note::Note::new(
2698 "local",
2699 "observation",
2700 format!("gap1-note-content-hard-{hard}"),
2701 );
2702 note.name = Some(format!("gap1-note-hard-{hard}"));
2703 let note_id = note.id;
2704 runtime
2705 .notes(&token)
2706 .expect("notes store")
2707 .upsert_note(note)
2708 .await
2709 .expect("seed note");
2710
2711 let args = if hard {
2712 json!({"id": note_id.to_string(), "hard": true})
2713 } else {
2714 json!({"id": note_id.to_string()})
2715 };
2716 let plan = prepare_delete(&runtime, &token, &args, None)
2717 .await
2718 .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
2719 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2720 .await
2721 .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
2722 assert!(
2723 matches!(
2724 outcome,
2725 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2726 ),
2727 "expected a clean delete commit (hard={hard}): {outcome:?}"
2728 );
2729
2730 let events = events_for_target(&runtime, &token, note_id, EventKind::NoteDeleted).await;
2731 assert_eq!(
2732 events.len(),
2733 1,
2734 "expected exactly one NoteDeleted event for {note_id} (hard={hard})"
2735 );
2736 assert_eq!(events[0].payload["hard"], json!(hard));
2737 }
2738 }
2739
2740 #[tokio::test]
2747 async fn atomic_update_edge_patches_weight_and_appends_edge_updated_event() {
2748 let runtime = scratch_runtime();
2749 let token = runtime
2750 .authorize(Namespace::parse("local").expect("ns"))
2751 .expect("authorize");
2752 let entities = runtime.entities(&token).expect("entities store");
2753 let a = khive_storage::Entity::new("local", "concept", "GapEdgeA");
2754 let b = khive_storage::Entity::new("local", "concept", "GapEdgeB");
2755 let (a_id, b_id) = (a.id, b.id);
2756 entities.upsert_entity(a).await.expect("seed a");
2757 entities.upsert_entity(b).await.expect("seed b");
2758
2759 let edge = runtime
2760 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.4, None)
2761 .await
2762 .expect("seed edge");
2763 let edge_id = Uuid::from(edge.id);
2764
2765 let plan = prepare_update(
2766 &runtime,
2767 &token,
2768 &json!({"id": edge_id.to_string(), "weight": 0.75}),
2769 None,
2770 )
2771 .await
2772 .expect("prepare update edge");
2773 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2774 .await
2775 .expect("seam call ok");
2776 assert!(
2777 matches!(
2778 outcome,
2779 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2780 ),
2781 "expected a clean edge update commit: {outcome:?}"
2782 );
2783
2784 let updated = runtime
2785 .get_edge(&token, edge_id)
2786 .await
2787 .expect("get_edge")
2788 .expect("edge still present");
2789 assert_eq!(updated.weight, 0.75, "weight patch must persist");
2790 assert_eq!(updated.relation, EdgeRelation::Extends);
2791
2792 let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeUpdated).await;
2793 assert_eq!(
2794 events.len(),
2795 1,
2796 "expected exactly one EdgeUpdated event for {edge_id}"
2797 );
2798 assert_eq!(
2799 events[0].payload["changed_fields"],
2800 json!(["weight"]),
2801 "changed_fields must name exactly the patched field"
2802 );
2803 }
2804
2805 #[tokio::test]
2813 async fn atomic_update_edge_symmetric_conflict_absorbs_into_surviving_row() {
2814 let runtime = scratch_runtime();
2815 let token = runtime
2816 .authorize(Namespace::parse("local").expect("ns"))
2817 .expect("authorize");
2818 let entities = runtime.entities(&token).expect("entities store");
2819 let a = khive_storage::Entity::new("local", "concept", "GapEdgeSymA");
2820 let b = khive_storage::Entity::new("local", "concept", "GapEdgeSymB");
2821 let (a_id, b_id) = (a.id, b.id);
2822 entities.upsert_entity(a).await.expect("seed a");
2823 entities.upsert_entity(b).await.expect("seed b");
2824
2825 let requested_edge = runtime
2827 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
2828 .await
2829 .expect("seed requested edge");
2830 let requested_id = Uuid::from(requested_edge.id);
2831
2832 let canonical_edge = runtime
2835 .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
2836 .await
2837 .expect("seed canonical edge");
2838 let canonical_id = Uuid::from(canonical_edge.id);
2839 assert_ne!(requested_id, canonical_id);
2840
2841 let plan = prepare_update(
2842 &runtime,
2843 &token,
2844 &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
2845 None,
2846 )
2847 .await
2848 .expect("prepare update edge (symmetric conflict)");
2849 let (canon_src, canon_tgt) =
2856 canonical_edge_endpoints(EdgeRelation::CompetesWith, a_id, b_id);
2857 match &plan {
2858 AtomicOpPlan::Update(p) => {
2859 assert_eq!(p.target_id, requested_id);
2860 let key = p
2861 .edge_natural_key
2862 .as_ref()
2863 .expect("symmetric edge update must carry edge_natural_key");
2864 assert_eq!(key.canon_source_id, canon_src);
2865 assert_eq!(key.canon_target_id, canon_tgt);
2866 assert_eq!(key.relation, EdgeRelation::CompetesWith);
2867 }
2868 other => panic!("expected an Update plan, got {other:?}"),
2869 }
2870
2871 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
2872 .await
2873 .expect("seam call ok");
2874 assert!(
2875 matches!(
2876 outcome,
2877 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
2878 ),
2879 "expected a clean symmetric-conflict-absorption commit: {outcome:?}"
2880 );
2881
2882 let requested_after = runtime
2884 .get_edge_including_deleted(&token, requested_id)
2885 .await
2886 .expect("get_edge_including_deleted");
2887 assert!(
2888 requested_after.is_none(),
2889 "the non-canonical requested row must be deleted, not just tombstoned"
2890 );
2891
2892 let surviving = runtime
2894 .get_edge(&token, canonical_id)
2895 .await
2896 .expect("get_edge")
2897 .expect("surviving canonical row must remain");
2898 assert_eq!(surviving.weight, 0.9);
2899 assert_eq!(surviving.relation, EdgeRelation::CompetesWith);
2900
2901 let events =
2905 events_for_target(&runtime, &token, requested_id, EventKind::EdgeUpdated).await;
2906 assert_eq!(events.len(), 1);
2907 }
2908
2909 #[tokio::test]
2921 async fn atomic_update_edge_symmetric_same_unit_delete_race_aborts_the_unit() {
2922 let runtime = scratch_runtime();
2923 let token = runtime
2924 .authorize(Namespace::parse("local").expect("ns"))
2925 .expect("authorize");
2926 let entities = runtime.entities(&token).expect("entities store");
2927 let a = khive_storage::Entity::new("local", "concept", "GapEdgeRaceA");
2928 let b = khive_storage::Entity::new("local", "concept", "GapEdgeRaceB");
2929 let (a_id, b_id) = (a.id, b.id);
2930 entities.upsert_entity(a).await.expect("seed a");
2931 entities.upsert_entity(b).await.expect("seed b");
2932
2933 let requested_edge = runtime
2935 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
2936 .await
2937 .expect("seed requested edge");
2938 let requested_id = Uuid::from(requested_edge.id);
2939
2940 let canonical_edge = runtime
2943 .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
2944 .await
2945 .expect("seed canonical edge");
2946 let canonical_id = Uuid::from(canonical_edge.id);
2947
2948 let delete_plan = prepare_delete(
2949 &runtime,
2950 &token,
2951 &json!({"id": requested_id.to_string(), "hard": true}),
2952 None,
2953 )
2954 .await
2955 .expect("prepare delete edge");
2956 let update_plan = prepare_update(
2957 &runtime,
2958 &token,
2959 &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
2960 None,
2961 )
2962 .await
2963 .expect("prepare update edge (both prepares run before either commits)");
2964
2965 let outcome = crate::atomic_runner::run_atomic_unit(
2966 runtime.sql().as_ref(),
2967 vec![delete_plan, update_plan],
2968 )
2969 .await
2970 .expect("the seam call itself must not error — the unit rolls back cleanly");
2971 match outcome {
2972 crate::atomic_runner::AtomicRunOutcome::RolledBack {
2973 failed_op_index, ..
2974 } => {
2975 assert_eq!(
2976 failed_op_index, 1,
2977 "op 1 (the update) must be the one whose guard fails"
2978 );
2979 }
2980 other => panic!("expected the whole unit to roll back, got {other:?}"),
2981 }
2982
2983 let requested_after = runtime
2985 .get_edge(&token, requested_id)
2986 .await
2987 .expect("get_edge");
2988 assert!(
2989 requested_after.is_some(),
2990 "delete(X) must have rolled back along with the failed update"
2991 );
2992 let canonical_after = runtime
2994 .get_edge(&token, canonical_id)
2995 .await
2996 .expect("get_edge")
2997 .expect("canonical row must still be present");
2998 assert_eq!(
2999 canonical_after.weight, 0.6,
3000 "the pre-existing canonical row must never have been touched by the aborted update"
3001 );
3002 }
3003
3004 #[tokio::test]
3009 async fn atomic_update_edge_rejects_entity_only_field_name() {
3010 let runtime = scratch_runtime();
3011 let token = runtime
3012 .authorize(Namespace::parse("local").expect("ns"))
3013 .expect("authorize");
3014 let entities = runtime.entities(&token).expect("entities store");
3015 let a = khive_storage::Entity::new("local", "concept", "GapEdgeRejectA");
3016 let b = khive_storage::Entity::new("local", "concept", "GapEdgeRejectB");
3017 let (a_id, b_id) = (a.id, b.id);
3018 entities.upsert_entity(a).await.expect("seed a");
3019 entities.upsert_entity(b).await.expect("seed b");
3020 let edge = runtime
3021 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
3022 .await
3023 .expect("seed edge");
3024 let edge_id = Uuid::from(edge.id);
3025
3026 let err = prepare_update(
3027 &runtime,
3028 &token,
3029 &json!({"id": edge_id.to_string(), "name": "not-a-valid-edge-field"}),
3030 None,
3031 )
3032 .await
3033 .expect_err("edge update with an entity-only field must be rejected");
3034 let message = err.to_string();
3035 assert!(
3036 message.contains("name") && message.contains("edge"),
3037 "error must name the offending field and the substrate: {message}"
3038 );
3039 }
3040
3041 #[tokio::test]
3046 async fn atomic_delete_edge_soft_and_hard_appends_edge_deleted_event() {
3047 let runtime = scratch_runtime();
3048 let token = runtime
3049 .authorize(Namespace::parse("local").expect("ns"))
3050 .expect("authorize");
3051
3052 for hard in [false, true] {
3053 let entities = runtime.entities(&token).expect("entities store");
3054 let a = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelA{hard}"));
3055 let b = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelB{hard}"));
3056 let (a_id, b_id) = (a.id, b.id);
3057 entities.upsert_entity(a).await.expect("seed a");
3058 entities.upsert_entity(b).await.expect("seed b");
3059 let edge = runtime
3060 .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
3061 .await
3062 .expect("seed edge");
3063 let edge_id = Uuid::from(edge.id);
3064
3065 let args = if hard {
3066 json!({"id": edge_id.to_string(), "hard": true})
3067 } else {
3068 json!({"id": edge_id.to_string()})
3069 };
3070 let plan = prepare_delete(&runtime, &token, &args, None)
3071 .await
3072 .unwrap_or_else(|e| panic!("prepare delete edge (hard={hard}): {e}"));
3073 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
3074 .await
3075 .unwrap_or_else(|e| panic!("edge delete commit (hard={hard}): {e}"));
3076 assert!(
3077 matches!(
3078 outcome,
3079 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
3080 ),
3081 "expected a clean edge delete commit (hard={hard}): {outcome:?}"
3082 );
3083
3084 let after = runtime
3085 .get_edge_including_deleted(&token, edge_id)
3086 .await
3087 .expect("get_edge_including_deleted");
3088 if hard {
3089 assert!(after.is_none(), "hard delete must purge the row entirely");
3090 } else {
3091 assert!(
3092 after.as_ref().is_some_and(|e| e.deleted_at.is_some()),
3093 "soft delete must tombstone, not purge"
3094 );
3095 }
3096
3097 let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeDeleted).await;
3098 assert_eq!(
3099 events.len(),
3100 1,
3101 "expected exactly one EdgeDeleted event for {edge_id} (hard={hard})"
3102 );
3103 assert_eq!(events[0].payload["hard"], json!(hard));
3104 }
3105 }
3106
3107 #[tokio::test]
3111 async fn atomic_update_note_appends_no_event() {
3112 let runtime = scratch_runtime();
3113 let token = runtime
3114 .authorize(Namespace::parse("local").expect("ns"))
3115 .expect("authorize");
3116 let mut note = khive_storage::note::Note::new("local", "observation", "gap1-note-noevent");
3117 note.name = Some("gap1-note-noevent".to_string());
3118 let note_id = note.id;
3119 runtime
3120 .notes(&token)
3121 .expect("notes store")
3122 .upsert_note(note)
3123 .await
3124 .expect("seed note");
3125
3126 let plan = prepare_update(
3127 &runtime,
3128 &token,
3129 &json!({"id": note_id.to_string(), "content": "gap1-note-noevent, revised"}),
3130 None,
3131 )
3132 .await
3133 .expect("prepare update");
3134 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
3135 .await
3136 .expect("seam call ok");
3137 assert!(matches!(
3138 outcome,
3139 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
3140 ));
3141
3142 let event_store = runtime.events(&token).expect("event store");
3143 let page = event_store
3144 .query_events(
3145 khive_storage::EventFilter::default(),
3146 khive_storage::types::PageRequest::default(),
3147 )
3148 .await
3149 .expect("query_events");
3150 assert!(
3151 page.items.iter().all(|e| e.target_id != Some(note_id)),
3152 "update_note must append no event; found: {:?}",
3153 page.items
3154 .iter()
3155 .filter(|e| e.target_id == Some(note_id))
3156 .collect::<Vec<_>>()
3157 );
3158 }
3159
3160 #[tokio::test]
3163 async fn atomic_link_appends_no_event() {
3164 let runtime = scratch_runtime();
3165 let token = runtime
3166 .authorize(Namespace::parse("local").expect("ns"))
3167 .expect("authorize");
3168 let source = khive_storage::Entity::new("local", "concept", "gap1-link-source");
3169 let target = khive_storage::Entity::new("local", "concept", "gap1-link-target");
3170 let (source_id, target_id) = (source.id, target.id);
3171 runtime
3172 .entities(&token)
3173 .expect("entities store")
3174 .upsert_entity(source)
3175 .await
3176 .expect("seed source");
3177 runtime
3178 .entities(&token)
3179 .expect("entities store")
3180 .upsert_entity(target)
3181 .await
3182 .expect("seed target");
3183
3184 let plan = prepare_link(
3185 &runtime,
3186 &token,
3187 &json!({
3188 "source_id": source_id.to_string(),
3189 "target_id": target_id.to_string(),
3190 "relation": "extends",
3191 }),
3192 )
3193 .await
3194 .expect("prepare link");
3195 let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
3196 .await
3197 .expect("seam call ok");
3198 assert!(matches!(
3199 outcome,
3200 crate::atomic_runner::AtomicRunOutcome::Committed { .. }
3201 ));
3202
3203 let event_store = runtime.events(&token).expect("event store");
3204 let page = event_store
3205 .query_events(
3206 khive_storage::EventFilter::default(),
3207 khive_storage::types::PageRequest::default(),
3208 )
3209 .await
3210 .expect("query_events");
3211 assert!(
3212 page.items.is_empty(),
3213 "link must append no event; found: {:?}",
3214 page.items
3215 );
3216 }
3217
3218 #[tokio::test]
3223 async fn prepare_add_entity_rejects_whitespace_only_name() {
3224 let runtime = scratch_runtime();
3225 let token = runtime
3226 .authorize(Namespace::parse("local").expect("ns"))
3227 .expect("authorize");
3228
3229 let err = prepare_add_entity(&runtime, &token, &json!({"kind": "concept", "name": " "}))
3230 .await
3231 .expect_err("whitespace-only entity name must fail prepare");
3232
3233 assert!(matches!(
3234 err,
3235 RuntimeError::InvalidInput(message) if message.contains("name must not be empty")
3236 ));
3237 }
3238
3239 #[tokio::test]
3240 async fn prepare_add_entity_rejects_non_string_description() {
3241 let runtime = scratch_runtime();
3242 let token = runtime
3243 .authorize(Namespace::parse("local").expect("ns"))
3244 .expect("authorize");
3245
3246 let err = prepare_add_entity(
3247 &runtime,
3248 &token,
3249 &json!({"kind": "concept", "name": "Valid", "description": 42}),
3250 )
3251 .await
3252 .expect_err("non-string entity description must fail prepare");
3253
3254 assert!(matches!(
3255 err,
3256 RuntimeError::InvalidInput(message)
3257 if message.contains("description must be a string or null")
3258 ));
3259 }
3260
3261 #[tokio::test]
3262 async fn prepare_add_note_rejects_non_string_name() {
3263 let runtime = scratch_runtime();
3264 let token = runtime
3265 .authorize(Namespace::parse("local").expect("ns"))
3266 .expect("authorize");
3267
3268 let err = prepare_add_note(
3269 &runtime,
3270 &token,
3271 &json!({"kind": "observation", "content": "Valid", "name": 42}),
3272 )
3273 .await
3274 .expect_err("non-string note name must fail prepare");
3275
3276 assert!(matches!(
3277 err,
3278 RuntimeError::InvalidInput(message) if message.contains("name must be a string or null")
3279 ));
3280 }
3281
3282 #[tokio::test]
3283 async fn atomic_add_entity_link_add_note_plan_commits_entity_edge_note_and_fts_together() {
3284 let runtime = scratch_runtime();
3285 runtime.register_embedder(StubProvider);
3286 let token = runtime
3287 .authorize(Namespace::parse("local").expect("ns"))
3288 .expect("authorize");
3289 let entities = runtime.entities(&token).expect("entities store");
3290 let a = khive_storage::Entity::new("local", "concept", "ProposalPlanLinkA");
3291 let b = khive_storage::Entity::new("local", "concept", "ProposalPlanLinkB");
3292 let (a_id, b_id) = (a.id, b.id);
3293 entities.upsert_entity(a).await.expect("seed a");
3294 entities.upsert_entity(b).await.expect("seed b");
3295
3296 let add_entity_plan = prepare_add_entity(
3297 &runtime,
3298 &token,
3299 &json!({"kind": "concept", "name": "ProposalPlanNewEntity", "description": "created atomically"}),
3300 )
3301 .await
3302 .expect("prepare add_entity");
3303 let link_plan = prepare_link(
3304 &runtime,
3305 &token,
3306 &json!({"source_id": a_id.to_string(), "target_id": b_id.to_string(), "relation": "extends"}),
3307 )
3308 .await
3309 .expect("prepare link");
3310 let add_note_plan = prepare_add_note(
3311 &runtime,
3312 &token,
3313 &json!({"kind": "observation", "content": "created atomically alongside the entity"}),
3314 )
3315 .await
3316 .expect("prepare add_note");
3317
3318 let entity_id = match &add_entity_plan {
3319 AtomicOpPlan::AddEntity(p) => p.entity_id,
3320 other => panic!("expected an AddEntity plan, got {other:?}"),
3321 };
3322 let note_id = match &add_note_plan {
3323 AtomicOpPlan::AddNote(p) => p.note_id,
3324 other => panic!("expected an AddNote plan, got {other:?}"),
3325 };
3326
3327 let outcome = crate::atomic_runner::run_atomic_unit(
3328 runtime.sql().as_ref(),
3329 vec![add_entity_plan, link_plan, add_note_plan],
3330 )
3331 .await
3332 .expect("seam call ok");
3333 let post_commit = match outcome {
3334 crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
3335 other => panic!("expected the whole unit to commit: {other:?}"),
3336 };
3337 let entity = runtime
3338 .entities(&token)
3339 .expect("entities store")
3340 .get_entity(entity_id)
3341 .await
3342 .expect("get_entity")
3343 .expect("entity must exist after commit");
3344 assert_eq!(entity.name, "ProposalPlanNewEntity");
3345 assert!(
3346 runtime
3347 .text(&token)
3348 .expect("text store")
3349 .get_document("local", entity_id)
3350 .await
3351 .expect("get_document")
3352 .is_some(),
3353 "entity's FTS document must exist after commit"
3354 );
3355
3356 let (edge_count, _, _, edge_deleted_at) =
3357 probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
3358 assert_eq!(
3359 edge_count, 1,
3360 "the edge must be committed alongside the entity/note"
3361 );
3362 assert!(edge_deleted_at.is_none());
3363
3364 let note = runtime
3365 .notes(&token)
3366 .expect("notes store")
3367 .get_note(note_id)
3368 .await
3369 .expect("get_note")
3370 .expect("note must exist after commit");
3371 assert_eq!(note.content, "created atomically alongside the entity");
3372 assert!(
3373 runtime
3374 .text_for_notes(&token)
3375 .expect("text store")
3376 .get_document("local", note_id)
3377 .await
3378 .expect("get_document")
3379 .is_some(),
3380 "note's FTS document must exist after commit"
3381 );
3382
3383 apply_post_commit_effects(&runtime, &token, post_commit)
3384 .await
3385 .expect("apply post-commit effects");
3386
3387 let vec_store = runtime
3388 .vectors_for_model(&token, STUB_MODEL)
3389 .expect("vec store");
3390 assert_eq!(
3391 vec_store.count().await.expect("count after"),
3392 2,
3393 "post-commit reindex must have embedded both the new entity and the new note"
3394 );
3395 }
3396
3397 #[tokio::test]
3398 async fn atomic_add_entity_and_add_note_roll_back_on_later_link_failure_leaving_zero_trace() {
3399 let runtime = scratch_runtime();
3400 let token = runtime
3401 .authorize(Namespace::parse("local").expect("ns"))
3402 .expect("authorize");
3403 let entities = runtime.entities(&token).expect("entities store");
3404 let a = khive_storage::Entity::new("local", "concept", "ProposalPlanRollbackA");
3405 let x = khive_storage::Entity::new("local", "concept", "ProposalPlanRollbackX");
3406 let (a_id, x_id) = (a.id, x.id);
3407 entities.upsert_entity(a).await.expect("seed a");
3408 entities.upsert_entity(x.clone()).await.expect("seed x");
3409
3410 let add_entity_plan = prepare_add_entity(
3411 &runtime,
3412 &token,
3413 &json!({"kind": "concept", "name": "ProposalPlanRollbackNewEntity"}),
3414 )
3415 .await
3416 .expect("prepare add_entity");
3417 let add_note_plan = prepare_add_note(
3418 &runtime,
3419 &token,
3420 &json!({"kind": "observation", "content": "must not survive the rollback"}),
3421 )
3422 .await
3423 .expect("prepare add_note");
3424 let delete_plan = prepare_delete(
3425 &runtime,
3426 &token,
3427 &json!({"id": x_id.to_string(), "hard": true}),
3428 None,
3429 )
3430 .await
3431 .expect("prepare delete x");
3432 let link_plan = prepare_link(
3435 &runtime,
3436 &token,
3437 &json!({"source_id": a_id.to_string(), "target_id": x_id.to_string(), "relation": "extends"}),
3438 )
3439 .await
3440 .expect("prepare link (endpoint still exists at prepare time)");
3441
3442 let entity_id = match &add_entity_plan {
3443 AtomicOpPlan::AddEntity(p) => p.entity_id,
3444 other => panic!("expected an AddEntity plan, got {other:?}"),
3445 };
3446 let note_id = match &add_note_plan {
3447 AtomicOpPlan::AddNote(p) => p.note_id,
3448 other => panic!("expected an AddNote plan, got {other:?}"),
3449 };
3450
3451 let outcome = crate::atomic_runner::run_atomic_unit(
3452 runtime.sql().as_ref(),
3453 vec![add_entity_plan, add_note_plan, delete_plan, link_plan],
3454 )
3455 .await
3456 .expect("the seam call itself must not error; the unit rolls back cleanly");
3457 match outcome {
3458 crate::atomic_runner::AtomicRunOutcome::RolledBack {
3459 failed_op_index, ..
3460 } => {
3461 assert_eq!(
3462 failed_op_index, 3,
3463 "the trailing link (index 3) must be the op whose guard fails"
3464 );
3465 }
3466 other => panic!("expected the whole unit to roll back, got {other:?}"),
3467 }
3468
3469 assert!(
3470 runtime
3471 .get_entity_including_deleted(&token, entity_id)
3472 .await
3473 .expect("get_entity_including_deleted")
3474 .is_none(),
3475 "the new entity must leave zero trace after rollback"
3476 );
3477 assert!(
3478 runtime
3479 .text(&token)
3480 .expect("text store")
3481 .get_document("local", entity_id)
3482 .await
3483 .expect("get_document")
3484 .is_none(),
3485 "the new entity's FTS document must leave zero trace after rollback"
3486 );
3487 assert!(
3488 runtime
3489 .get_note_including_deleted(&token, note_id)
3490 .await
3491 .expect("get_note_including_deleted")
3492 .is_none(),
3493 "the new note must leave zero trace after rollback"
3494 );
3495 assert!(
3496 runtime
3497 .text_for_notes(&token)
3498 .expect("text store")
3499 .get_document("local", note_id)
3500 .await
3501 .expect("get_document")
3502 .is_none(),
3503 "the new note's FTS document must leave zero trace after rollback"
3504 );
3505
3506 let x_after = runtime
3507 .get_entity_including_deleted(&token, x_id)
3508 .await
3509 .expect("get_entity_including_deleted")
3510 .expect("x must still be present because its delete rolled back too");
3511 assert!(
3512 x_after.deleted_at.is_none(),
3513 "x's delete must have rolled back along with the failed link"
3514 );
3515
3516 let (edge_count, _, _, _) =
3517 probe_edge_natural_key(&runtime, "local", a_id, x_id, "extends").await;
3518 assert_eq!(edge_count, 0, "no edge may have been committed");
3519 }
3520}