Skip to main content

khive_runtime/
atomic_prepare.rs

1//! ADR-099: the per-verb async prepare pass for the KG-substrate v1
2//! admissible verbs (`update`, `delete`, `link`, `merge`), plus
3//! [`prepare_add_entity`]/[`prepare_add_note`] for the ADR-046 proposal
4//! changeset `AddEntity`/`AddNote` arms. Each `prepare_*` function reads
5//! current state (async, outside any transaction) and returns a plain-data
6//! [`crate::atomic_runner::AtomicOpPlan`] ([`crate::atomic_plan`]) for the
7//! synchronous commit pass ([`crate::atomic_runner::run_atomic_unit`]) to
8//! apply.
9//!
10//! `gtd.transition`/`gtd.complete` prepare is deliberately not here (lives in
11//! `kkernel` instead), and `propose`/`review`/`withdraw`/`merge` are on the
12//! v1 admissible list but have no working prepare implementation in this
13//! module (`prepare_governance_unimplemented` fails loudly rather than
14//! silently no-opping; `prepare_merge` is unreachable through `--atomic` and
15//! kept only for its own tests and as defense in depth). See
16//! `docs/api/atomic_prepare.md#scope-what-is-excluded-and-why` for why each of these is excluded
17//! and what would be required to admit them.
18
19use 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
54// ---------------------------------------------------------------------------
55// arg extraction helpers
56// ---------------------------------------------------------------------------
57
58fn 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
90/// Nullable-string patch semantics mirroring the actually-reachable behavior
91/// of `khive-pack-kg::handlers::common::optional_string_patch`/
92/// `description_patch`, reimplemented here rather than imported (that
93/// module has no dependency edge back to `khive-runtime`). Canonical's field
94/// type is `Option<Value>` (`UpdateParams.name`/`.description`); serde_json's
95/// derived `Deserialize` for `Option<T>` intercepts a literal JSON `null` at
96/// the outer `Option` boundary and maps it straight to Rust `None`
97/// regardless of the inner type, so canonical's own "clear" arm is
98/// unreachable through normal struct deserialization: `update(name=null)` /
99/// `update(description=null)` are no-ops, not clears. This module reads raw,
100/// un-deserialized JSON, so it must replicate that collapse explicitly: key
101/// absent OR JSON `null` -> `None` (leave unchanged, no-op); key present as a
102/// string -> `Some(Some(s))` (set); any other JSON type -> a hard error.
103fn 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
113/// Strict string-or-absent-or-null patch for entity `name`. Unlike
114/// `optional_str`'s `.as_str()`, this does not silently drop a non-string,
115/// non-null value like `name: 123` as absent: it rejects it instead of
116/// reporting success for an invalid update. Canonical validates entity
117/// `name` via `string_value` on `UpdateParams.name: Option<Value>`: null
118/// collapses to absent at the struct-deserialize boundary (see
119/// `optional_string_patch` doc above), so the reachable behavior is:
120/// absent/null -> unchanged; non-null string -> set; any other JSON type ->
121/// hard error. This mirrors that exactly, reading raw JSON instead of a
122/// deserialized struct.
123fn 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
133/// Nullable-JSON-value patch for `properties`: canonical
134/// `properties: Option<Value>` on `UpdateParams` collapses a literal JSON
135/// `null` to Rust `None` at the struct-deserialize boundary (same collapse
136/// as `optional_string_patch` above), so `properties=null` is canonically a
137/// no-op (leave existing properties unchanged): not a stored JSON `null`.
138/// This module reads raw JSON, so it must replicate that collapse: key
139/// absent OR JSON `null` -> `None` (no merge); any other JSON value ->
140/// `Some(value)` (merge), with no further shape validation at this layer.
141fn 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
148/// `tags` patch: canonical `tags: Option<Vec<String>>` on `UpdateParams`
149/// collapses a literal JSON `null` to Rust `None` at the struct-deserialize
150/// boundary (same collapse as above), so `tags=null` is canonically a no-op
151/// (leave existing tags unchanged). A non-array, non-null value is still a
152/// hard error (mirrors the type failure `UpdateParams` deserialization would
153/// itself produce for a malformed `tags`).
154fn 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
184/// Tri-state patch extraction for `Option<Option<f64>>`-shaped fields
185/// (`NotePatch::salience` / `NotePatch::decay_factor`): key absent -> `None`
186/// (untouched), key present as JSON `null` -> `Some(None)` (clear), key
187/// present as a number -> `Some(Some(v))` (set). Range validation lives in
188/// curation.rs's `prepare_update_note`, not here.
189fn 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
200/// Every registered embedding model's vector table name, in the exact format
201/// `curation::merge_entity_sql` uses (`"vec_{sanitize_key(model_name)}"`) —
202/// reused here so atomic delete/merge purge the same tables the non-atomic
203/// paths do.
204fn 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
212/// A guarded (`guard: None` — best-effort mirror, matching the non-atomic
213/// index-cleanup calls which don't assert a row existed) `DELETE` statement
214/// against one FTS or vector table for a single subject, scoped by namespace.
215fn 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
234/// `true` iff a table named `table` currently exists in the backing SQLite
235/// database (`sqlite_master` probe, read-only — safe in async prepare, does
236/// NOT open/create the vector store, so it cannot lazily create the table
237/// itself).
238async 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
255/// Append the FTS + every registered model's vector-row purge for `subject_id`
256/// (scoped to the RECORD's own namespace, matching `delete_entity`/
257/// `delete_note`'s `record_tok`/`record_ns` convention: not the caller
258/// token's namespace, per by-ID namespace-agnosticism) onto `statements`.
259///
260/// FTS tables (`fts_entities`/`fts_notes`) always exist (created at schema
261/// migration time) so their purge is unconditional. `vec_*` tables are
262/// created lazily on first vector-store open, so a default runtime can
263/// register embedding models before any vector table necessarily exists:
264/// a raw unconditional `DELETE FROM vec_*` can hit `no such table` on a
265/// fresh DB. Only push the vec purge for tables that actually exist:
266/// absence means the record definitionally has no vector row for that
267/// model, so skipping is data-parity-correct (the non-atomic path would
268/// lazily create the table then delete zero rows: same data outcome,
269/// without this read-only prepare pass performing an init side effect).
270async 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
297/// Event-store append parity for the canonical handlers that emit a
298/// lifecycle event after their row mutation: `update_entity` ->
299/// `EntityUpdated`, `delete_entity` -> `EntityDeleted`, `delete_note` ->
300/// `NoteDeleted`, `update_edge` -> `EdgeUpdated`, `delete_edge` ->
301/// `EdgeDeleted`. `update_note` and `link` append no event and must never
302/// call this. See `docs/api/atomic_prepare.md#event_append_statements` for why
303/// this is a `PlanStatement` rather than a `PostCommitEffect`.
304///
305/// Invariant: returned statements are unguarded — appended after the plan's
306/// own guarded row statement, so [`apply_plan`]'s stop-on-first-failure
307/// contract means they are only reached once that row mutation's guard has
308/// already held. Committing the event row atomically with the mutation it
309/// describes strengthens canonical's guarantee: the non-atomic handlers write
310/// the event in a separate transaction, ordered but not atomic with the row
311/// mutation.
312fn 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
334// ---------------------------------------------------------------------------
335// dispatch
336// ---------------------------------------------------------------------------
337
338/// Build the prepared [`AtomicOpPlan`] for one KG-substrate admissible op
339/// (`update`, `delete`, `link`, `merge`). Returns a loud [`RuntimeError`] for
340/// `propose`/`review`/`withdraw` (known scope gap, see module doc) and any
341/// other verb (the CLI boundary must reject those before calling this — a
342/// verb reaching here is either KG-substrate-admissible or a bug upstream).
343pub async fn prepare_op(
344    runtime: &KhiveRuntime,
345    token: &NamespaceToken,
346    tool: &str,
347    args: &Value,
348) -> RuntimeResult<AtomicOpPlan> {
349    match tool {
350        // `expected_kind: None` here — same reasoning as the `"delete"` arm
351        // below: callers that need `update(kind=...)` parity must resolve
352        // the kind spec themselves (it needs a `VerbRegistry`, unreachable
353        // from this crate: see `AtomicUpdateKind`'s doc comment) and call
354        // `prepare_update` directly with the resolved value; `kkernel`'s
355        // `--atomic` seam does exactly this and bypasses this dispatch arm.
356        // A caller reaching `prepare_op("update", ...)` without going
357        // through that seam gets kind-unchecked behavior.
358        "update" => prepare_update(runtime, token, args, None).await,
359        // `expected_kind: None` here — callers that need `delete(kind=...)`
360        // parity must resolve the kind spec themselves (it needs a
361        // `VerbRegistry`, unreachable from this crate: see
362        // `AtomicDeleteKind`'s doc comment) and call `prepare_delete`
363        // directly with the resolved value; `kkernel`'s `--atomic` seam does
364        // exactly this and bypasses this dispatch arm. A caller reaching
365        // `prepare_op("delete", ...)` without going through that seam gets
366        // kind-unchecked behavior.
367        "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
389// ---------------------------------------------------------------------------
390// create (AddEntity / AddNote)
391// ---------------------------------------------------------------------------
392
393/// Build the prepared plan for an `AddEntity` proposal change. The entity
394/// row and FTS document are committed together; vector indexing is deferred
395/// until after commit because embedding may suspend. `kind` must already be
396/// canonicalized by the caller because pack-aware resolution requires a
397/// `VerbRegistry`.
398pub 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
457/// Build the prepared plan for an `AddNote` proposal change. Mirrors
458/// [`prepare_add_entity`]'s shape and the same
459/// `kind`-already-canonicalized split. `annotates` is out of scope: the
460/// proposal `NoteDraft` this backs carries no annotates targets, unlike
461/// `KhiveRuntime::create_note`'s general-purpose signature.
462pub 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(&note),
494            guard: Some(AffectedRowGuard::exactly(1)),
495        },
496        PlanStatement {
497            statement: insert_document_statement("fts_notes", &note_fts_document(&note)),
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
509// ---------------------------------------------------------------------------
510// update
511// ---------------------------------------------------------------------------
512
513/// Mirrors `khive-pack-kg::handlers::update::reject_inapplicable_fields`: a
514/// hard `InvalidInput` when a caller passes a field that does not apply to
515/// the resolved substrate (e.g. `salience` on an entity, or
516/// `description`/`tags` on a note). That function has no dependency edge
517/// back to `khive-runtime`, so its exact field-applicability check list and
518/// error message shape are reimplemented here rather than imported: same
519/// pattern as `optional_string_patch` above. Presence is checked directly on
520/// the raw args object (this module has no `UpdateParams` struct); a JSON
521/// `null` value is treated as absent, matching `Option<T>` deserialization
522/// semantics.
523fn 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        // `update` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`, so
558        // this arm must reject entity/note-only fields (e.g. `name`) on an
559        // edge update rather than silently skip the guard, mirroring
560        // `khive-pack-kg::handlers::update::reject_inapplicable_fields`'s
561        // `KindSpec::Edge` arm.
562        "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
596/// Caller-supplied update-kind expectation, resolved via the canonical
597/// `resolve_kind_spec` at the kkernel `--atomic` seam: the same pattern
598/// [`AtomicDeleteKind`] uses. Without this check, `update(kind="document",
599/// id=<concept>)` would be canonically `NotFound` but the atomic path would
600/// ignore the explicit kind and mutate the resolved entity anyway.
601/// `khive-runtime` must not depend on `khive-pack-kg`, so this is a plain
602/// substrate-level shape rather than `khive_pack_kg::handlers::KindSpec`
603/// itself: the kkernel seam does the pack-aware resolution and passes down
604/// only what `prepare_update` needs to enforce the mismatch check.
605pub enum AtomicUpdateKind {
606    Entity { specific: Option<String> },
607    Note { specific: Option<String> },
608    Edge,
609}
610
611/// `expected_kind`: `None` when the caller omitted `kind` (no check, parity
612/// with canonical's own optional discriminator); `Some(_)` enforces an
613/// exact-parity mismatch check against the resolved record's actual
614/// substrate/specific kind, mirroring `handle_update`'s
615/// `entity.kind != *k` / note kind checks (update.rs:200-201, :229-234).
616pub 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    // Mirrors update.rs's entity_kind immutability guard: entity_kind is a
625    // legacy top-level field, independent of the `kind` substrate
626    // discriminator handled elsewhere.
627    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            // Decide step lives in curation.rs's `prepare_update_entity` —
655            // the SAME function canonical `update_entity` calls. Only the
656            // arg-extraction (raw JSON -> `EntityPatch`) and the plan-shape
657            // wiring (domain object -> `PlanStatement` via the shared
658            // `entity_upsert_statement` builder) are atomic-path-specific.
659            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 &note.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            // Decide step lives in curation.rs's `prepare_update_note` — the
697            // same function canonical `update_note` calls, including the
698            // salience/decay_factor range validation. `optional_f64_patch`
699            // below preserves tri-state patch semantics (key absent =
700            // untouched, key null = clear, key present = set) when
701            // constructing the `NotePatch`.
702            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(&note),
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        // `Resolved` (khive-runtime::operations) has no `Edge` variant — an
742        // id that isn't an entity/note/pack-private/event record is checked
743        // against the graph store directly, mirroring
744        // `khive-pack-kg::handlers::KgPack::infer_kind_from_uuid`'s own
745        // entity/note-then-edge fallback order. `update` admits
746        // `kind="edge"`, so this arm must be able to build a plan for one.
747        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
762/// Build an entity update plan from a typed patch. Proposal changesets use
763/// this entry point so their explicit `description: null` clear operation is
764/// preserved instead of being collapsed by raw verb deserialization.
765pub 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
802/// Edge branch of `prepare_update`. Mirrors `KhiveRuntime::update_edge`'s
803/// patch semantics: `relation`/`weight`/`properties` are the only applicable
804/// fields, a changed `relation` is endpoint-validated first, `weight` is
805/// range-checked, and `properties` REPLACES `metadata` wholesale (no merge).
806/// See `docs/api/atomic_prepare.md#prepare_update_edge` for the DML-shape parity
807/// detail with `update_edge`.
808///
809/// Invariant (symmetric relations `competes_with`/`composed_with`): this
810/// function must never branch on a prepare-time conflict probe — a different
811/// op in the same atomic unit could change the conflict landscape between
812/// probe and commit, making any such branch stale by construction. It always
813/// emits BOTH statements from [`edge_symmetric_delete_if_conflict_statement`]
814/// and [`edge_symmetric_refresh_or_update_inplace_statement`], each carrying
815/// its own commit-time `WHERE`/`CASE WHEN` predicate that re-evaluates the
816/// conflict condition fresh inside the transaction. This function reads no
817/// state to guess a surviving id; the plan instead carries `edge_natural_key`
818/// so a post-commit caller derives the actual surviving id from the
819/// committed row, never from a value computed before the rest of this atomic
820/// unit has even run.
821async 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        // The write for a symmetric relation never branches on a
875        // prepare-time probe result: it always carries both self-guarding,
876        // commit-time-predicate statements (see their doc comment in
877        // khive-db's graph.rs for the full rationale). This avoids the
878        // staleness window a prepare-time probe would expose: an earlier op
879        // in the same atomic unit could change the conflict landscape before
880        // commit. Canonical's own probe-then-branch
881        // `update_edge_symmetric_dml` has no such exposure (single
882        // transaction, no interleaving) and is unaffected.
883        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        // No prepare-time read needed: the two statements above are
917        // self-guarding at commit time (see their doc comment). Post-commit
918        // result rendering derives the actual surviving id from THIS
919        // natural key, never from a value computed here.
920        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        // Non-symmetric: bit-for-bit the same builder `graph.upsert_edge`
928        // calls — see doc comment above.
929        edge.updated_at = now;
930        statements.push(PlanStatement {
931            statement: edge_upsert_statement(&edge),
932            guard: Some(AffectedRowGuard::exactly(1)),
933        });
934    }
935
936    // Mirrors `update_edge`'s unconditional post-mutation `EdgeUpdated`
937    // event append, keyed on the original `edge_id` the caller supplied:
938    // canonical does the same (the event target is `edge_id`, not the
939    // post-absorption surviving id).
940    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
957// ---------------------------------------------------------------------------
958// delete
959// ---------------------------------------------------------------------------
960
961/// Caller-supplied delete-kind expectation, resolved via the canonical
962/// `resolve_kind_spec` at the kkernel `--atomic` seam. `khive-runtime` must
963/// not depend on `khive-pack-kg` (packs depend on the runtime, not the other
964/// way around), so this is a plain substrate-level shape rather than
965/// `khive_pack_kg::handlers::KindSpec` itself: the kkernel seam does the
966/// pack-aware `resolve_kind_spec` resolution (which needs a `VerbRegistry`,
967/// unreachable from this crate) and passes down only what `prepare_delete`
968/// needs to enforce the mismatch check.
969///
970/// `delete` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`, hence the
971/// `Edge` variant. `Event`/`Proposal` remain rejected at the kkernel seam
972/// (not v1-admissible for atomic delete at all).
973pub enum AtomicDeleteKind {
974    Entity { specific: Option<String> },
975    Note { specific: Option<String> },
976    Edge,
977}
978
979/// `expected_kind`: `None` when the caller omitted `kind` (no check, parity
980/// with canonical's own optional discriminator); `Some(_)` enforces an
981/// exact-parity mismatch check against the resolved record's actual
982/// substrate/specific kind, mirroring `handle_delete`'s
983/// `entity.kind != *expected` / `note.kind != *expected` checks.
984pub 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    // `delete(id, hard=true)` is the public purge route after a prior soft
997    // delete, so it must resolve including already-tombstoned rows (a
998    // live-only resolve would never find one). Soft delete keeps the
999    // live-only resolve: a soft delete of an already-tombstoned row is a
1000    // no-op, matching non-atomic behavior.
1001    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            // Storage parity: `entity_soft_delete_statement`/
1028            // `entity_hard_delete_statement` are the SAME khive-db builders
1029            // khive-db's own `SqlEntityStore::delete_entity` calls — no DML
1030            // text is hand-duplicated here.
1031            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                // Same builder canonical `delete_entity`'s hard-delete
1045                // cascade calls (`graph.purge_incident_edges`).
1046                statements.push(PlanStatement {
1047                    statement: purge_incident_edges_statement(id),
1048                    guard: None,
1049                });
1050            }
1051            // FTS + vector index purge, matching operations.rs
1052            // `delete_entity`: both soft and hard delete clean indexes (a
1053            // hard delete of an already-tombstoned record must still purge
1054            // them); only hard additionally cascades edges above.
1055            push_index_purge_statements(
1056                runtime,
1057                &mut statements,
1058                "fts_entities",
1059                &namespace,
1060                id,
1061                "atomic-delete-entity",
1062            )
1063            .await?;
1064            // operations.rs's `delete_entity` appends an `EntityDeleted`
1065            // event after a successful row delete, on both soft and hard
1066            // delete. `apply_plan` never reaches this statement unless the
1067            // guarded row statement above affected a row, so no extra `if`
1068            // is needed here.
1069            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 &note.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            // Storage parity: `note_soft_delete_statement`/
1103            // `note_hard_delete_statement` are the SAME khive-db builders
1104            // khive-db's own `SqlNoteStore::delete_note` calls.
1105            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            // FTS + vector index purge, matching operations.rs
1124            // `delete_note`: both soft and hard delete clean indexes (a hard
1125            // delete of an already-tombstoned record must still purge
1126            // them); only hard additionally cascades edges above.
1127            push_index_purge_statements(
1128                runtime,
1129                &mut statements,
1130                "fts_notes",
1131                &namespace,
1132                id,
1133                "atomic-delete-note",
1134            )
1135            .await?;
1136            // operations.rs's `delete_note` appends a `NoteDeleted` event
1137            // after a successful row delete, on both soft and hard delete:
1138            // same reasoning as the entity branch above.
1139            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                // A committed atomic note delete must fire the same
1151                // pack-installed note-mutation hook `operations.rs::
1152                // delete_note` fires, so a warm ANN cache sees the deletion
1153                // even when the mutation went through the atomic-plan path.
1154                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        // `Resolved` has no `Edge` variant (same reasoning as
1164        // `prepare_update`'s fallback above) — probe the graph store
1165        // directly.
1166        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
1188/// Edge branch of `prepare_delete`. Mirrors
1189/// `khive-runtime::operations::KhiveRuntime::delete_edge` exactly: hard
1190/// delete cascades `purge_incident_edges` (any `annotates` edge — or any
1191/// other edge — pointing AT this edge as a node) BEFORE deleting the edge
1192/// row itself, then a soft or hard delete statement, then an unconditional
1193/// `EdgeDeleted` event (edges are never FTS/vector-indexed, so unlike the
1194/// entity/note branches there is no index purge here — `delete_edge` has
1195/// none either).
1196async 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        // Mirrors `delete_edge`'s `graph.purge_incident_edges(edge_id)` —
1206        // unguarded: zero incident edges is a legitimate outcome, not a
1207        // failure (same reasoning as the entity/note cascade-edges
1208        // statements above).
1209        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
1241// ---------------------------------------------------------------------------
1242// link
1243// ---------------------------------------------------------------------------
1244
1245fn 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    // Top-level `dependency_kind` param merges into `metadata`: only fills
1262    // the key when metadata doesn't already carry one. Calls the same
1263    // `khive_runtime::merge_entry_metadata` `khive-pack-kg`'s canonical
1264    // `handle_link` calls, so both sides depend on one function instead of
1265    // each maintaining their own copy.
1266    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    // Endpoint-kind `dependency_kind` inference for `depends_on` edges,
1279    // matching operations.rs `link()`: only applies when both endpoints
1280    // resolve as entities and the key is still absent after the
1281    // top-level-param merge above. Runs against the canonical endpoints,
1282    // mirroring `KhiveRuntime::link`'s own ordering (canonicalize, then
1283    // infer).
1284    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    // The guarded `INSERT ... SELECT ... WHERE EXISTS(...)` shape is
1303    // load-bearing (see `LinkPlan`'s own doc comment): it re-probes both
1304    // endpoints inside the transaction, closing the intra-batch hazard
1305    // where an earlier op in the same atomic unit, e.g. `delete(X, hard)`,
1306    // could invalidate this op's prepare-time endpoint validation before
1307    // commit. The conflict-arm SET list shares the same
1308    // `EDGE_NATURAL_KEY_CONFLICT_SET` text `edge_upsert_statement`
1309    // (canonical `link`'s builder) uses, so the two cannot silently diverge
1310    // (a prior bug: this atomic literal never set
1311    // `target_backend = excluded.target_backend`, so a re-link of an edge
1312    // carrying a cross-backend `target_backend` stamp behaved differently
1313    // under `--atomic`).
1314    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
1335// ---------------------------------------------------------------------------
1336// merge (entity-only)
1337// ---------------------------------------------------------------------------
1338
1339// Full atomic-merge parity (field folding, survivor FTS/vector reindex,
1340// loser index purge, merge provenance, same-kind rejection) is deferred:
1341// atomic `merge` is rejected entirely at the pre-runtime admissibility
1342// guard (`khive_types::pack::ATOMIC_KNOWN_UNIMPLEMENTED_VERBS`, alongside
1343// `propose`/`review`/`withdraw`). This function still produces a plan
1344// (kept for the existing direct-prepare test coverage below and as
1345// defense in depth), but the CLI's `--atomic` surface never reaches it,
1346// since `check_atomic_admissible` rejects `merge` before any runtime is
1347// built.
1348async 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
1423// ---------------------------------------------------------------------------
1424// post-commit effects
1425// ---------------------------------------------------------------------------
1426
1427/// Run every deferred [`PostCommitEffect`] after a committed atomic unit,
1428/// outside any transaction. Re-fetches each target's now-committed row and
1429/// reuses the existing `reindex_entity`/`reindex_note` (FTS + embedding,
1430/// same as the non-atomic path) for exact parity.
1431pub 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, &note).await?;
1447                    // This handler calls `reindex_note` directly, bypassing
1448                    // `update_note()` and the note-mutation hook it fires
1449                    // after its own reindex (see `curation.rs`). Fire it
1450                    // here so any in-process consumer (e.g.
1451                    // khive-pack-memory's warm ANN cache) sees a bumped
1452                    // generation after a committed atomic note update,
1453                    // matching the non-atomic path.
1454                    runtime.fire_note_mutation_hook(&note.kind, note.id).await;
1455                }
1456            }
1457            PostCommitEffect::NoteDeleted { note_id, kind } => {
1458                // Unlike `operations.rs`'s `delete_note`, which fires
1459                // `fire_note_mutation_hook` directly (with the already-known
1460                // kind, no refetch) after a successful row delete, an atomic
1461                // note delete needs this post-commit pass to reach it. The
1462                // note row is gone (hard delete) or tombstoned (soft
1463                // delete) by the time this runs, so it mirrors
1464                // `delete_note`'s direct-fire shape rather than
1465                // `ReindexNote`'s refetch-then-fire shape.
1466                runtime.fire_note_mutation_hook(&kind, note_id).await;
1467            }
1468            PostCommitEffect::GtdAudit { .. } => {
1469                // Applied by the `kkernel` caller's own post-commit pass,
1470                // not here: `khive-pack-gtd` (owner of
1471                // `ensure_audit_schema`/`write_audit_record`) depends on
1472                // `khive-runtime`, not the other way around, so this crate
1473                // cannot act on the effect itself. See
1474                // `PostCommitEffect::GtdAudit`'s doc comment.
1475            }
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    /// Atomic `update` must reject a field that does not apply to the
1550    /// resolved substrate: parity with
1551    /// `khive-pack-kg::handlers::update::reject_inapplicable_fields`.
1552    /// Without this check, atomic prepare would silently ignore `salience`
1553    /// on an entity: it would set every entity field to its current value,
1554    /// bump `updated_at`, satisfy the `exactly(1)` guard, and commit: a
1555    /// spurious no-op reported as success.
1556    #[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        // A valid entity update (name/description/tags) must still work.
1585        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    /// Symmetric note-substrate case: `description` and `tags` are
1613    /// entity-only fields; passing either for a note must be rejected the
1614    /// same way update.rs rejects them.
1615    #[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        // A valid note update (content) must still work.
1645        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    /// Updating a note's content inside an atomic unit must, after commit,
1663    /// leave the note recallable via FTS under its new content and its
1664    /// vector row refreshed: parity with the non-atomic
1665    /// `update_note` -> `reindex_note` path.
1666    #[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        // Sanity: no vector row yet for the stub model.
1685        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        // FTS: the note must be recallable under its NEW content.
1717        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        // Vector: a row must now exist for the registered stub model.
1731        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    /// The atomic-plan path must fire the pack-installed note-mutation hook
1739    /// for both an atomic note UPDATE (`PostCommitEffect::ReindexNote`'s
1740    /// handler fires it after its own reindex, mirroring `update_note()`
1741    /// on the non-atomic path) and an atomic note DELETE (`DeletePlan`
1742    /// carries a `PostCommitEffect::NoteDeleted` that
1743    /// `apply_post_commit_effects` dispatches directly, mirroring
1744    /// `operations.rs::delete_note`'s direct-fire, no-refetch shape: the
1745    /// row may already be gone by the time this runs, for a hard delete).
1746    /// A minimal counting hook proves both fire; no `khive-pack-memory`
1747    /// dependency is needed at this layer, since the hook itself is
1748    /// generic.
1749    #[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        // Update path.
1769        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        // Delete path (soft delete: the row still exists, but the hook
1799        // fires directly from the captured kind rather than refetching).
1800        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    /// Atomic delete must purge the note's FTS row and vector row for both
1850    /// soft and hard delete: parity with `KhiveRuntime::delete_note`'s
1851    /// index-cleanup contract.
1852    #[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, &note)
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    /// Atomic delete must purge the entity's FTS row and vector row for
1933    /// both soft and hard delete: parity with
1934    /// `KhiveRuntime::delete_entity`'s index-cleanup contract.
1935    #[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    /// Atomic link must persist an explicit top-level `dependency_kind`
2015    /// param into edge metadata, and must infer one for `depends_on` edges
2016    /// when absent: parity with
2017    /// `link.rs`'s `merge_entry_metadata` and `operations.rs`'s
2018    /// `infer_dependency_kind` table.
2019    #[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        // (a) explicit top-level `dependency_kind` param persists in metadata.
2039        {
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        // (b) `depends_on` with no explicit dependency_kind infers from
2066        // endpoint kinds: (service, service) -> "runtime".
2067        {
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    /// Raw natural-key probe of `graph_edges` (namespace, source_id,
2094    /// target_id, relation) — returns `(weight, metadata_json, deleted_at)`
2095    /// for exactly the ONE row a UNIQUE(namespace, source_id, target_id,
2096    /// relation) constraint permits. `None` means no row at all.
2097    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    /// Atomic `link` must be an upsert, exactly like canonical `link` ->
2141    /// `upsert_edge`'s natural-key `ON CONFLICT` arm: re-linking an
2142    /// already-linked triple must succeed and update weight/metadata, not
2143    /// hit the `UNIQUE(namespace, source_id, target_id, relation)`
2144    /// constraint and roll back the whole atomic unit.
2145    #[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        // (a) a fresh link still works.
2160        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        // (b) re-linking the SAME triple with a different weight/metadata
2189        // must SUCCEED (not a constraint-violation rollback) and UPDATE the
2190        // existing row in place — natural key stays unique.
2191        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    /// Atomic `link` of a soft-deleted triple must resurrect it
2231    /// (`deleted_at = NULL`), matching `upsert_edge`'s natural-key
2232    /// `ON CONFLICT ... DO UPDATE SET deleted_at = NULL`.
2233    #[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        // Soft-delete the edge row directly (natural-key UPDATE — mirrors
2267        // what `delete_edge(hard=false)` does to this same row).
2268        {
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        // Re-link the same triple: must resurrect (deleted_at -> NULL), not
2296        // fail on the UNIQUE constraint of the still-present soft-deleted row.
2297        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    /// Atomic delete of an entity and a note must succeed even when the
2331    /// registered embedding model's `vec_*` table has never been lazily
2332    /// created (a fresh DB registers models before any vector store is
2333    /// opened): the raw purge DML must skip tables that don't exist
2334    /// rather than hit `no such table` and roll back the whole atomic
2335    /// unit. FTS purge still fires (those tables always exist) and the
2336    /// delete itself is a clean commit.
2337    #[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        // Seed via raw upsert ONLY — never call reindex_entity/reindex_note
2346        // or vectors_for_model, so the stub model's `vec_*` table is never
2347        // lazily created (opening the vector store is what creates it).
2348        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    /// Atomic hard delete must be able to purge a record that was already
2408    /// soft-deleted: parity with `delete(id, hard=true)` being the public
2409    /// purge route after a prior soft delete (the non-atomic hard path
2410    /// resolves including deleted rows and its DML carries no `deleted_at`
2411    /// predicate).
2412    #[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, &note)
2446            .await
2447            .expect("seed index rows");
2448
2449        // First: SOFT delete both (via atomic prepare) so they're tombstoned
2450        // going into the hard-delete attempt below.
2451        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        // Now: HARD delete the already-tombstoned records.
2485        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    // ------------------------------------------------------------------
2558    // event-store append parity
2559    // ------------------------------------------------------------------
2560
2561    /// Fetch every event of `kind` targeting `target_id`, via the same
2562    /// `EventStore::query_events` surface `--atomic` callers would use to
2563    /// verify parity — not a raw SQL probe.
2564    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    /// Atomic `update(id=<entity>, name=...)` must append an
2586    /// `EntityUpdated` event, matching `curation::update_entity`: the
2587    /// event is appended unconditionally after a successful row update,
2588    /// not only on the reindex-triggering subset.
2589    #[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    /// Atomic soft and hard delete of an entity must each append an
2636    /// `EntityDeleted` event, matching `operations::delete_entity`, which
2637    /// fires on both delete modes.
2638    #[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    /// Atomic soft and hard delete of a note must each append a
2687    /// `NoteDeleted` event, matching `operations::delete_note`, which
2688    /// fires on both delete modes.
2689    #[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    /// `update` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`; this
2741    /// asserts `prepare_update` actually builds a plan for one, a
2742    /// non-symmetric relation (`extends`) exercises the
2743    /// `edge_upsert_statement` reuse branch, and that the committed row +
2744    /// `EdgeUpdated` event match canonical `update_edge`'s shape (weight
2745    /// persisted, relation unchanged, exactly one event).
2746    #[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    /// The symmetric-relation conflict-absorption branch of
2806    /// `prepare_update_edge` — mirrors `update_edge_symmetric_dml`'s case
2807    /// (b): changing a non-symmetric edge's `relation` to a symmetric one
2808    /// whose canonical natural key collides with an ALREADY-EXISTING
2809    /// symmetric edge between the same two entities must delete the
2810    /// requested (non-canonical) row and refresh the surviving canonical
2811    /// row in place, rather than raising a uniqueness error.
2812    #[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        // The non-canonical edge under test: A -> B, non-symmetric relation.
2826        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        // The pre-existing canonical row this update will collide with once
2833        // `relation` becomes `competes_with` (symmetric).
2834        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        // The plan does not compute a prepare-time advisory surviving id
2850        // (`target_id` is just the requested id): it carries
2851        // `edge_natural_key` so a post-commit caller can derive the real
2852        // surviving id itself. Assert the plan carries the right natural
2853        // key to look up; the actual surviving row's identity is verified
2854        // against the DB after commit, below.
2855        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        // The requested (non-canonical) row must be gone.
2883        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        // The surviving canonical row must carry the patch.
2893        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        // Event target is the CALLER-supplied id, not the surviving id —
2902        // mirrors `update_edge`'s event using `edge_id` (the caller's
2903        // original argument), not the post-absorption id.
2904        let events =
2905            events_for_target(&runtime, &token, requested_id, EventKind::EdgeUpdated).await;
2906        assert_eq!(events.len(), 1);
2907    }
2908
2909    /// The same-unit race: `[delete(X), update(X -> competes_with)]` where
2910    /// an already-existing canonical row sits at the post-update natural
2911    /// key. Both ops' async prepare passes run before either commits, so at
2912    /// prepare time `X` still exists and both plans build. At commit time
2913    /// `delete(X)` removes it first; `update(X -> competes_with)`'s own
2914    /// commit-time statements must then fail loud (its target no longer
2915    /// exists) rather than silently absorbing into the pre-existing
2916    /// canonical row it never causally touched. The whole atomic unit must
2917    /// roll back — parity with canonical `update_edge`'s `NotFound` for a
2918    /// missing edge, expressed here as the unit-level abort for any op
2919    /// whose commit-time guard fails.
2920    #[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        // The row op 1 will try to update — deleted by op 0 in the SAME unit.
2934        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        // The pre-existing canonical row the buggy `id = ?2 OR natural-key`
2941        // predicate used to silently absorb into.
2942        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        // Whole-unit rollback: op 0's delete must be undone too.
2984        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        // The pre-existing canonical row must be completely untouched.
2993        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    /// `update` rejects an entity/note-only field (`name`) on an edge
3005    /// target, mirroring
3006    /// `khive-pack-kg::handlers::update::reject_inapplicable_fields`'s
3007    /// `KindSpec::Edge` arm.
3008    #[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    /// `delete` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`; this
3042    /// asserts `prepare_delete` actually builds a plan for one on both soft
3043    /// and hard delete, matching `operations::delete_edge`'s row-mode DML
3044    /// and unconditional `EdgeDeleted` event.
3045    #[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    /// Parity boundary: atomic `update` of a note must append no event:
3108    /// canonical `update_note` never calls `append_event` (unlike
3109    /// `update_entity`, which always does).
3110    #[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    /// Parity boundary: atomic `link` must append no event: canonical
3161    /// `link` never calls `append_event`.
3162    #[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    // ------------------------------------------------------------------
3219    // AddEntity and AddNote plans alongside link
3220    // ------------------------------------------------------------------
3221
3222    #[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        // Prepare sees x before the transaction; the guarded link must detect
3433        // that the preceding hard delete removed it inside the transaction.
3434        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}