1#[cfg(doc)]
2use super::AtomicDeleteKind;
3use super::{
4 entity_fts_document, entity_name_patch, entity_replace_if_unchanged_statement,
5 entity_upsert_statement, event_append_statements, insert_document_statements,
6 note_fts_document, note_upsert_statement, obj, optional_create_string,
7 optional_entity_type_patch, optional_f64_patch, optional_properties, optional_str,
8 optional_string_patch, optional_tags, prepare_delete, prepare_delete_edge, prepare_link,
9 prepare_merge, prepare_update_edge, prepare_update_entity_plan_with_version,
10 refuse_pack_registry_tags, require_str, require_uuid, AddEntityPlan, AddNotePlan,
11 AffectedRowGuard, AtomicOpPlan, EdgeUpsertDisposition, EventKind, KhiveRuntime, NamespaceToken,
12 PlanStatement, PostCommitEffect, Resolved, RuntimeError, RuntimeResult, SubstrateKind,
13 UpdatePlan, Uuid, Value,
14};
15
16pub async fn prepare_op(
26 runtime: &KhiveRuntime,
27 token: &NamespaceToken,
28 tool: &str,
29 args: &Value,
30) -> RuntimeResult<AtomicOpPlan> {
31 match tool {
32 "update" => prepare_update(runtime, token, args, None).await,
41 "delete" => prepare_delete(runtime, token, args, None).await,
50 "link" => prepare_link(runtime, token, args).await,
51 "merge" => prepare_merge(runtime, token, args).await,
52 "propose" | "review" | "withdraw" => prepare_governance_unimplemented(tool),
53 other => Err(RuntimeError::InvalidInput(format!(
54 "{other:?} has no atomic_prepare::prepare_op implementation; the CLI \
55 admissibility check should have rejected this before prepare"
56 ))),
57 }
58}
59
60fn prepare_governance_unimplemented(tool: &str) -> RuntimeResult<AtomicOpPlan> {
61 Err(RuntimeError::InvalidInput(format!(
62 "{tool:?} is on the ADR-099 v1 admissible verb list but has no --atomic \
63 prepare/apply implementation yet: its lifecycle (ADR-046) is an \
64 event-sourced changeset-interpreter over a dedicated `proposals_open` \
65 table, not a small guarded-DML plan — a faithful non-stub atomic \
66 prepare for it is tracked as ADR-099 follow-up work, not implemented \
67 in slice B3. No write was attempted."
68 )))
69}
70
71pub async fn prepare_add_entity(
81 runtime: &KhiveRuntime,
82 token: &NamespaceToken,
83 args: &Value,
84) -> RuntimeResult<AtomicOpPlan> {
85 let kind = require_str(args, "kind")?;
86 let name = require_str(args, "name")?;
87 runtime.validate_entity_kind(kind)?;
88 if name.trim().is_empty() {
89 return Err(RuntimeError::InvalidInput(
90 "name must not be empty".to_string(),
91 ));
92 }
93
94 let description = optional_create_string(args, "description")?;
95 let properties = optional_properties(args, "properties")?;
96 let tags = optional_tags(args)?.unwrap_or_default();
97
98 crate::secret_gate::check_at(name, "entity", "name")?;
99 if let Some(ref d) = description {
100 crate::secret_gate::check_at(d, "entity", "description")?;
101 }
102 if let Some(ref p) = properties {
103 crate::secret_gate::check_json_at(p, "entity", "properties")?;
104 }
105 crate::secret_gate::check_tags_at(&tags, "entity", "tags")?;
106 crate::secret_gate::reject_reserved_secret_gate_property(properties.as_ref())?;
107
108 let ns = token.namespace().as_str();
109 let mut entity = khive_storage::Entity::new(ns, kind, name);
110 if let Some(d) = description {
111 entity = entity.with_description(d);
112 }
113 if let Some(p) = properties {
114 entity = entity.with_properties(p);
115 }
116 if !tags.is_empty() {
117 entity = entity.with_tags(tags);
118 }
119
120 let mut statements = vec![PlanStatement {
121 statement: entity_upsert_statement(&entity),
122 guard: Some(AffectedRowGuard::exactly(1)),
123 }];
124 for statement in insert_document_statements("fts_entities", &entity_fts_document(&entity)) {
128 statements.push(PlanStatement {
129 statement,
130 guard: None,
131 });
132 }
133
134 Ok(AtomicOpPlan::AddEntity(AddEntityPlan {
135 entity_id: entity.id,
136 statements,
137 post_commit: PostCommitEffect::ReindexEntity {
138 entity_id: entity.id,
139 },
140 }))
141}
142
143pub async fn prepare_add_note(
149 runtime: &KhiveRuntime,
150 token: &NamespaceToken,
151 args: &Value,
152) -> RuntimeResult<AtomicOpPlan> {
153 let kind = require_str(args, "kind")?;
154 let content = require_str(args, "content")?;
155 runtime.validate_note_kind(kind)?;
156
157 let name = optional_create_string(args, "name")?;
158 let properties = optional_properties(args, "properties")?;
159 let properties = runtime.derive_note_write_properties(kind, token, properties)?;
165
166 crate::secret_gate::check_note_fields_at(
167 content,
168 name.as_deref(),
169 properties.as_ref(),
170 "note",
171 )?;
172 crate::secret_gate::reject_reserved_secret_gate_property(properties.as_ref())?;
173
174 let ns = token.namespace().as_str();
175 let mut note = khive_storage::note::Note::new(ns, kind, content);
176 if let Some(n) = name {
177 note = note.with_name(n);
178 }
179 if let Some(p) = properties {
180 note = note.with_properties(p);
181 }
182
183 let mut statements = vec![PlanStatement {
184 statement: note_upsert_statement(¬e),
185 guard: Some(AffectedRowGuard::exactly(1)),
186 }];
187 for statement in insert_document_statements("fts_notes", ¬e_fts_document(¬e)) {
190 statements.push(PlanStatement {
191 statement,
192 guard: None,
193 });
194 }
195
196 Ok(AtomicOpPlan::AddNote(Box::new(AddNotePlan {
197 note_guard: None,
198 note_id: note.id,
199 statements,
200 post_commit: PostCommitEffect::ReindexNote {
201 note_id: note.id,
202 version: note.version,
203 },
204 })))
205}
206
207pub(super) fn reject_inapplicable_update_fields(
222 args: &Value,
223 substrate: &str,
224) -> RuntimeResult<()> {
225 let o = obj(args)?;
226 if substrate == "edge" && o.get("expected_version").is_some_and(|v| !v.is_null()) {
227 return Err(RuntimeError::InvalidInput(
228 "expected_version applies only to entities and notes".into(),
229 ));
230 }
231 if substrate != "note"
232 && ["embed", "fence"]
233 .iter()
234 .any(|field| o.contains_key(*field))
235 {
236 return Err(RuntimeError::InvalidInput(
237 "embed and fence apply only to notes".into(),
238 ));
239 }
240 let present = |k: &str| o.get(k).is_some_and(|v| !v.is_null());
241 let (bad_field, valid): (Option<&str>, &str) = match substrate {
242 "entity" => {
243 let bad = if present("content") {
244 Some("content")
245 } else if o.contains_key("salience") {
246 Some("salience")
247 } else if o.contains_key("decay_factor") {
248 Some("decay_factor")
249 } else if present("relation") {
250 Some("relation")
251 } else if present("weight") {
252 Some("weight")
253 } else {
254 None
255 };
256 (bad, "name, description, tags, properties, entity_type")
257 }
258 "note" => {
259 let bad = if o.contains_key("description") {
260 Some("description")
261 } else if present("relation") {
262 Some("relation")
263 } else if present("weight") {
264 Some("weight")
265 } else if o.contains_key("entity_type") {
266 Some("entity_type")
269 } else {
270 None
271 };
272 (
273 bad,
274 "name, content, salience, decay_factor, properties, tags",
275 )
276 }
277 "edge" => {
283 let bad = if present("name") {
284 Some("name")
285 } else if o.contains_key("description") {
286 Some("description")
287 } else if present("content") {
288 Some("content")
289 } else if present("tags") {
290 Some("tags")
291 } else if o.contains_key("salience") {
292 Some("salience")
293 } else if o.contains_key("decay_factor") {
294 Some("decay_factor")
295 } else if o.contains_key("entity_type") {
296 Some("entity_type")
299 } else {
300 None
301 };
302 (bad, "relation, weight, properties")
303 }
304 _ => (None, ""),
305 };
306 if let Some(field) = bad_field {
307 let substrate_label = match substrate {
308 "entity" => "an entity",
309 "note" => "a note",
310 "edge" => "an edge",
311 other => other,
312 };
313 return Err(RuntimeError::InvalidInput(format!(
314 "field '{field}' is not valid for {substrate_label}; valid fields: {valid}"
315 )));
316 }
317 Ok(())
318}
319
320pub enum AtomicUpdateKind {
330 Entity {
331 specific: Option<String>,
332 entity_type: Option<String>,
333 },
334 Note {
335 specific: Option<String>,
336 },
337 Edge,
338}
339
340pub fn validate_note_update_expected_kind(
345 note: &khive_storage::note::Note,
346 expected_kind: &Option<AtomicUpdateKind>,
347) -> RuntimeResult<()> {
348 let id = note.id;
349 match expected_kind {
350 None => Ok(()),
351 Some(AtomicUpdateKind::Note {
352 specific: Some(expected),
353 }) if ¬e.kind != expected => Err(RuntimeError::NotFound(format!("note {id}"))),
354 Some(AtomicUpdateKind::Note { .. }) => Ok(()),
355 Some(AtomicUpdateKind::Entity { .. }) => {
356 Err(RuntimeError::NotFound(format!("entity {id}")))
357 }
358 Some(AtomicUpdateKind::Edge) => Err(RuntimeError::NotFound(format!("edge {id}"))),
359 }
360}
361
362async fn prepare_note_update_plan_from_snapshot(
363 runtime: &KhiveRuntime,
364 token: &NamespaceToken,
365 args: &Value,
366 expected_kind: &Option<AtomicUpdateKind>,
367 note: khive_storage::note::Note,
368 policy: crate::NoteUpdatePolicy,
369 registry: Option<&crate::VerbRegistry>,
370) -> RuntimeResult<(khive_storage::Note, UpdatePlan)> {
371 let id = require_uuid(args, "id")?;
372 if note.id != id {
373 return Err(RuntimeError::NotFound(format!("note {id}")));
374 }
375 validate_note_update_expected_kind(¬e, expected_kind)?;
376
377 reject_inapplicable_update_fields(args, "note")?;
378 let mut normalized_args = args.clone();
379 crate::curation::normalize_note_update_tags(&mut normalized_args)?;
380 let args = &normalized_args;
381 let name = optional_string_patch(args, "name")?;
382 let content = optional_str(args, "content").map(str::to_string);
383 let properties = optional_properties(args, "properties")?;
384 let salience = optional_f64_patch(args, "salience")?;
385 let decay_factor = optional_f64_patch(args, "decay_factor")?;
386 let options = crate::note_write::NoteWriteOptions {
387 expected_version: obj(args)?
388 .get("expected_version")
389 .filter(|v| !v.is_null())
390 .map(|v| {
391 v.as_i64().ok_or_else(|| {
392 RuntimeError::InvalidInput("expected_version must be an integer".into())
393 })
394 })
395 .transpose()?,
396 fence: obj(args)?
397 .get("fence")
398 .map(|v| {
399 serde_json::from_value(v.clone())
400 .map_err(|error| RuntimeError::InvalidInput(format!("invalid fence: {error}")))
401 })
402 .transpose()?,
403 embed: obj(args)?
404 .get("embed")
405 .filter(|v| !v.is_null())
406 .map(|v| {
407 v.as_bool()
408 .ok_or_else(|| RuntimeError::InvalidInput("embed must be boolean".into()))
409 })
410 .transpose()?,
411 key: None,
412 };
413 let patch = crate::curation::NotePatch::new(name, content, salience, decay_factor, properties)
414 .with_update_policy(policy)
415 .with_write_options(options);
416 let (updated, mut plan) = runtime
417 .prepare_versioned_note_update(token, note.clone(), patch.clone())
418 .await?;
419 if let Some(registry) = registry {
420 attach_note_update_effects(runtime, token, registry, ¬e, &patch, &mut plan).await?;
421 }
422 Ok((updated, plan))
423}
424
425pub async fn prepare_update_from_note_snapshot(
435 runtime: &KhiveRuntime,
436 token: &NamespaceToken,
437 args: &Value,
438 expected_kind: Option<AtomicUpdateKind>,
439 note: khive_storage::note::Note,
440 policy: crate::NoteUpdatePolicy,
441 registry: &crate::VerbRegistry,
442) -> RuntimeResult<(khive_storage::Note, AtomicOpPlan)> {
443 if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
444 return Err(RuntimeError::InvalidInput(
445 "entity_kind is immutable; to change kind, delete then re-create the entity, \
446 or use merge() if this is a deduplication correction"
447 .into(),
448 ));
449 }
450 let (note, plan) = prepare_note_update_plan_from_snapshot(
451 runtime,
452 token,
453 args,
454 &expected_kind,
455 note,
456 policy,
457 Some(registry),
458 )
459 .await?;
460 Ok((note, AtomicOpPlan::Update(Box::new(plan))))
461}
462
463async fn attach_note_update_effects(
467 runtime: &KhiveRuntime,
468 token: &NamespaceToken,
469 registry: &crate::VerbRegistry,
470 snapshot: &khive_storage::Note,
471 patch: &crate::curation::NotePatch,
472 plan: &mut UpdatePlan,
473) -> RuntimeResult<()> {
474 use crate::atomic_plan::NoteUpdateStatement;
475 use crate::NoteUpdateEffect;
476
477 let Some(hook) = registry.find_kind_hook(&snapshot.kind) else {
478 return Ok(());
479 };
480 if let Some(expected) = patch.write_options.expected_version {
483 if expected != snapshot.version {
484 return Err(crate::note_write::NoteWriteConflict::Version {
485 expected,
486 current: snapshot.version,
487 }
488 .into_error()
489 .into());
490 }
491 }
492 let current = runtime.notes(token)?.get_note(snapshot.id).await?;
493 if !current.is_some_and(|current| {
494 current.updated_at == snapshot.updated_at
495 && current.deleted_at == snapshot.deleted_at
496 && current.version == snapshot.version
497 }) {
498 return Err(crate::curation::stale_note_snapshot_error(snapshot.id));
499 }
500 let effects = hook
501 .note_update_effects(runtime, token, snapshot, patch)
502 .await?;
503 if plan.idempotent_noop && !effects.is_empty() {
504 return Err(RuntimeError::InvalidInput(
505 "an unchanged note update cannot carry graph effects".into(),
506 ));
507 }
508 let edge_token = token.with_namespace(
509 crate::Namespace::parse(&snapshot.namespace)
510 .map_err(|error| RuntimeError::Internal(format!("invalid note namespace: {error}")))?,
511 );
512 for effect in effects {
513 match effect {
514 NoteUpdateEffect::Link(spec) => {
515 if spec.source_id != snapshot.id
516 || spec
517 .namespace
518 .as_deref()
519 .is_some_and(|ns| ns != snapshot.namespace)
520 {
521 return Err(RuntimeError::InvalidInput(
522 "note update links must originate from the note in its namespace".into(),
523 ));
524 }
525 let mut args = serde_json::json!({
526 "source_id": spec.source_id, "target_id": spec.target_id,
527 "relation": spec.relation, "weight": spec.weight,
528 "resurrect": spec.resurrect,
529 });
530 if let Some(metadata) = spec.metadata {
531 args["metadata"] = metadata;
532 }
533 let AtomicOpPlan::Link(link) = prepare_link(runtime, &edge_token, &args).await?
534 else {
535 return Err(RuntimeError::Internal("expected a link plan".into()));
536 };
537 if link.disposition == EdgeUpsertDisposition::Updated {
541 return Err(khive_types::KhiveError::conflict(
542 "a live edge appeared while preparing the note update; retry with fresh state",
543 ).into());
544 }
545 plan.graph_effects
546 .extend(link.statements.into_iter().map(NoteUpdateStatement::Write));
547 }
548 NoteUpdateEffect::DeleteEdge(edge) => {
549 let id = Uuid::from(edge.id);
550 if edge.source_id != snapshot.id
551 || edge.namespace != snapshot.namespace
552 || edge.deleted_at.is_some()
553 {
554 return Err(RuntimeError::InvalidInput(
555 "note update deletes must select an outgoing edge in the note namespace"
556 .into(),
557 ));
558 }
559 plan.graph_effects
560 .push(NoteUpdateStatement::Assert(PlanStatement {
561 statement: khive_db::stores::graph::edge_snapshot_assertion_statement(
562 &edge, false,
563 ),
564 guard: Some(AffectedRowGuard::exactly(1)),
565 }));
566 let actor = format!("{}:{}", token.actor().kind, token.actor().id);
567 let AtomicOpPlan::Delete(delete) =
568 prepare_delete_edge(&edge_token, id, edge, false, &actor).await?
569 else {
570 return Err(RuntimeError::Internal(
571 "expected an edge delete plan".into(),
572 ));
573 };
574 if delete.post_commit != PostCommitEffect::None {
575 return Err(RuntimeError::Internal(
576 "edge delete has a deferred effect".into(),
577 ));
578 }
579 plan.graph_effects.extend(
580 delete
581 .statements
582 .into_iter()
583 .map(NoteUpdateStatement::Write),
584 );
585 }
586 NoteUpdateEffect::AssertLink(edge) => {
587 if edge.source_id != snapshot.id
588 || edge.namespace != snapshot.namespace
589 || edge.deleted_at.is_some()
590 {
591 return Err(RuntimeError::InvalidInput(
592 "note update assertions must select a live outgoing edge in the note namespace".into(),
593 ));
594 }
595 plan.graph_effects
596 .push(NoteUpdateStatement::Assert(PlanStatement {
597 statement: khive_db::stores::graph::edge_snapshot_assertion_statement(
598 &edge, true,
599 ),
600 guard: Some(AffectedRowGuard::exactly(1)),
601 }));
602 }
603 }
604 }
605 Ok(())
606}
607
608pub async fn prepare_update(
616 runtime: &KhiveRuntime,
617 token: &NamespaceToken,
618 args: &Value,
619 expected_kind: Option<AtomicUpdateKind>,
620) -> RuntimeResult<AtomicOpPlan> {
621 let id = require_uuid(args, "id")?;
622
623 if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
627 return Err(RuntimeError::InvalidInput(
628 "entity_kind is immutable; to change kind, delete then re-create the entity, \
629 or use merge() if this is a deduplication correction"
630 .into(),
631 ));
632 }
633
634 match runtime.resolve_by_id(token, id).await? {
635 Some(Resolved::Entity(entity)) => {
636 match &expected_kind {
637 None => {}
638 Some(AtomicUpdateKind::Entity {
639 specific: Some(expected),
640 ..
641 }) if &entity.kind != expected => {
642 return Err(RuntimeError::NotFound(format!("entity {id}")));
643 }
644 Some(AtomicUpdateKind::Entity { .. }) => {}
645 Some(AtomicUpdateKind::Note { .. }) => {
646 return Err(RuntimeError::NotFound(format!("note {id}")));
647 }
648 Some(AtomicUpdateKind::Edge) => {
649 return Err(RuntimeError::NotFound(format!("edge {id}")));
650 }
651 }
652 if let Some(AtomicUpdateKind::Entity {
653 entity_type: Some(expected),
654 ..
655 }) = &expected_kind
656 {
657 if entity
658 .entity_type
659 .as_deref()
660 .is_some_and(|actual| actual != expected.as_str())
661 {
662 return Err(RuntimeError::NotFound(format!("entity {id}")));
663 }
664 }
665 refuse_pack_registry_tags(&entity.tags, "update")?;
666 reject_inapplicable_update_fields(args, "entity")?;
674 let name = entity_name_patch(args)?;
675 let description = optional_entity_type_patch(args, "description")?;
676 let properties = optional_properties(args, "properties")?;
677 let tags = optional_tags(args)?;
678 if let Some(ref tags) = tags {
679 refuse_pack_registry_tags(tags, "update")?;
680 }
681 let entity_type = optional_entity_type_patch(args, "entity_type")?;
682
683 let expected_version = obj(args)?
684 .get("expected_version")
685 .filter(|v| !v.is_null())
686 .map(|value| {
687 value.as_i64().ok_or_else(|| {
688 RuntimeError::InvalidInput("expected_version must be an integer".into())
689 })
690 })
691 .transpose()?;
692 let required_entity_type = match &expected_kind {
693 Some(AtomicUpdateKind::Entity { entity_type, .. }) => entity_type.as_deref(),
694 _ => None,
695 };
696 prepare_update_entity_plan_with_version_and_type(
697 runtime,
698 token,
699 id,
700 crate::curation::EntityPatch {
701 name,
702 description,
703 properties,
704 tags,
705 entity_type,
706 },
707 expected_version,
708 required_entity_type,
709 )
710 .await
711 }
712 Some(Resolved::Note(note)) => {
713 let (_, plan) = prepare_note_update_plan_from_snapshot(
718 runtime,
719 token,
720 args,
721 &expected_kind,
722 note,
723 crate::NoteUpdatePolicy::default(),
724 None,
725 )
726 .await?;
727 Ok(AtomicOpPlan::Update(Box::new(plan)))
728 }
729 Some(_) => Err(RuntimeError::InvalidInput(format!(
730 "update target {id} must be an entity, note, or edge"
731 ))),
732 None => match &expected_kind {
739 Some(AtomicUpdateKind::Entity { .. }) => {
740 Err(RuntimeError::NotFound(format!("entity/note {id}")))
741 }
742 Some(AtomicUpdateKind::Note { .. }) => {
743 Err(RuntimeError::NotFound(format!("entity/note {id}")))
744 }
745 Some(AtomicUpdateKind::Edge) | None => match runtime.get_edge(token, id).await? {
746 Some(edge) => prepare_update_edge(runtime, token, id, edge, args).await,
747 None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
748 },
749 },
750 }
751}
752
753pub async fn prepare_update_entity_plan(
757 runtime: &KhiveRuntime,
758 token: &NamespaceToken,
759 id: Uuid,
760 patch: crate::curation::EntityPatch,
761) -> RuntimeResult<AtomicOpPlan> {
762 prepare_update_entity_plan_with_version(runtime, token, id, patch, None).await
763}
764
765pub(super) async fn prepare_update_entity_plan_with_version_and_type(
766 runtime: &KhiveRuntime,
767 token: &NamespaceToken,
768 id: Uuid,
769 patch: crate::curation::EntityPatch,
770 expected_version: Option<i64>,
771 required_entity_type: Option<&str>,
772) -> RuntimeResult<AtomicOpPlan> {
773 crate::entity_write::validate_expected_version(expected_version)?;
774 let explicit_entity_type_patch = patch.entity_type.is_some();
775 let (entity, reindex_required, changed_fields, expected_updated_at, expected_deleted_at) =
776 runtime.prepare_update_entity(token, id, patch).await?;
777 if required_entity_type.is_some_and(|expected| {
778 (explicit_entity_type_patch || entity.entity_type.is_some())
779 && entity.entity_type.as_deref() != Some(expected)
780 }) {
781 return Err(RuntimeError::InvalidInput(
782 "kind subtype contradicts the requested entity_type update".into(),
783 ));
784 }
785 let mut statements = vec![PlanStatement {
786 statement: entity_replace_if_unchanged_statement(
787 &entity,
788 expected_updated_at,
789 expected_deleted_at,
790 ),
791 guard: Some(AffectedRowGuard::exactly(1)),
792 }];
793 statements.extend(event_append_statements(
794 token,
795 &entity.namespace,
796 "update",
797 EventKind::EntityUpdated,
798 SubstrateKind::Entity,
799 id,
800 serde_json::json!({
801 "id": id,
802 "namespace": entity.namespace,
803 "changed_fields": changed_fields,
804 }),
805 )?);
806 let post_commit = if reindex_required {
807 PostCommitEffect::ReindexEntity { entity_id: id }
808 } else {
809 PostCommitEffect::None
810 };
811 Ok(AtomicOpPlan::Update(Box::new(UpdatePlan {
812 graph_effects: Vec::new(),
813 note_vector_purge: None,
814 note_embedding_inheritance: None,
815 entity_guard: expected_version.map(|expected_version| {
816 crate::entity_write::EntityWriteGuard {
817 id,
818 expected_version,
819 }
820 }),
821 note_guard: None,
822 target_id: id,
823 statements,
824 post_commit,
825 edge_natural_key: None,
826 idempotent_noop: false,
827 })))
828}