khive_runtime/

atomic_prepare.rs

//! ADR-099: the per-verb async prepare pass for the KG-substrate v1
//! admissible verbs (`update`, `delete`, `link`, `merge`), plus
//! [`prepare_add_entity`]/[`prepare_add_note`] for the ADR-046 proposal
//! changeset `AddEntity`/`AddNote` arms. Each `prepare_*` function reads
//! current state (async, outside any transaction) and returns a plain-data
//! [`crate::atomic_runner::AtomicOpPlan`] ([`crate::atomic_plan`]) for the
//! synchronous commit pass ([`crate::atomic_runner::run_atomic_unit`]) to
//! apply.
//!
//! `gtd.transition`/`gtd.complete` prepare is deliberately not here (lives in
//! `kkernel` instead), and `propose`/`review`/`withdraw`/`merge` are on the
//! v1 admissible list but have no working prepare implementation in this
//! module (`prepare_governance_unimplemented` fails loudly rather than
//! silently no-opping; `prepare_merge` is unreachable through `--atomic` and
//! kept only for its own tests and as defense in depth). See
//! `docs/api/atomic_prepare.md#scope-what-is-excluded-and-why` for why each of these is excluded
//! and what would be required to admit them.

use serde_json::Value;
use uuid::Uuid;

use khive_storage::types::SqlValue;
use khive_storage::{AttachmentSubstrate, EdgeRelation, EdgeUpsertDisposition, SqlStatement};
use khive_types::{EventKind, SubstrateKind};

use crate::atomic_plan::{
    AddEntityPlan, AddNotePlan, AffectedRowGuard, DeletePlan, EdgeNaturalKey, LinkPlan, MergePlan,
    PlanStatement, PostCommitEffect, UpdatePlan,
};
use crate::atomic_runner::AtomicOpPlan;
use crate::atomic_runner::CommittedPostCommitEffects;
use crate::curation::{entity_fts_document, note_fts_document};
use crate::error::{RuntimeError, RuntimeResult};
use crate::operations::{
    canonical_edge_endpoints, merge_dependency_kind, validate_edge_metadata, validate_edge_weight,
    Resolved,
};
use crate::runtime::{KhiveRuntime, NamespaceToken};

use khive_db::stores::attachment::delete_record_attachments_statement;
use khive_db::stores::entity::{
    entity_hard_delete_statement, entity_replace_if_unchanged_statement,
    entity_soft_delete_statement, entity_upsert_statement,
};
use khive_db::stores::event::event_insert_statements;
use khive_db::stores::event::hard_delete_lineage_warning_statements;
use khive_db::stores::graph::{
    edge_hard_delete_statement, edge_insert_new_guarded_by_endpoints_statement,
    edge_link_replace_if_unchanged_and_endpoints_exist_statement,
    edge_replace_if_unchanged_statement, edge_soft_delete_statement,
    edge_symmetric_absorb_or_update_inplace_statement, edge_symmetric_delete_if_conflict_statement,
    purge_incident_edges_statement,
};
use khive_db::stores::note::{
    note_hard_delete_statement, note_soft_delete_statement, note_upsert_statement,
};
use khive_db::stores::text::{delete_document_statements, insert_document_statements};

// ---------------------------------------------------------------------------
// arg extraction helpers
// ---------------------------------------------------------------------------

fn obj(args: &Value) -> RuntimeResult<&serde_json::Map<String, Value>> {
    args.as_object()
        .ok_or_else(|| RuntimeError::InvalidInput("op args must be a JSON object".into()))
}

fn require_str<'a>(args: &'a Value, key: &str) -> RuntimeResult<&'a str> {
    obj(args)?
        .get(key)
        .and_then(|v| v.as_str())
        .ok_or_else(|| RuntimeError::InvalidInput(format!("missing required field {key:?}")))
}

/// Parse `key` as a bare UUID — never a short hex prefix.
///
/// A short prefix is a *resolution*: an unfiltered search that applies no
/// namespace predicate, so it can match nothing, match exactly one record
/// across every namespace, or match ambiguously. That search already
/// happened upstream, at the `kkernel` CLI boundary that has namespace
/// context and calls `resolve_uuid_unfiltered` before handing args down to
/// this module
/// (`crates/kkernel/src/atomic_apply.rs::resolve_kg_ids_in_args`). By the
/// time an id reaches this plan-preparation stage it must already name one
/// specific, already-identified record — which is exactly what a full UUID
/// demonstrates and a prefix does not.
fn require_uuid(args: &Value, key: &str) -> RuntimeResult<Uuid> {
    let raw = require_str(args, key)?;
    Uuid::parse_str(raw).map_err(|_| {
        RuntimeError::InvalidInput(format!(
            "{key} must be a full UUID; got {raw:?}. This atomic-plan stage consumes an \
             already-resolved record and performs no namespace-scoped search of its own, so a \
             short hex prefix cannot be resolved here — resolve it to a full UUID first (e.g. \
             via `get`) and pass that."
        ))
    })
}

fn optional_str<'a>(args: &'a Value, key: &str) -> Option<&'a str> {
    obj(args).ok()?.get(key).and_then(|v| v.as_str())
}

fn optional_create_string(args: &Value, key: &str) -> RuntimeResult<Option<String>> {
    match obj(args)?.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(value)) => Ok(Some(value.clone())),
        Some(other) => Err(RuntimeError::InvalidInput(format!(
            "{key} must be a string or null, got: {other}"
        ))),
    }
}

/// ADR-014 tri-state patch for entity `entity_type`, read from raw JSON to
/// mirror `UpdateParams.entity_type`'s `tri_string` deserializer (the
/// kkernel `--atomic` seam deserializes through that struct first, so the
/// two surfaces cannot diverge): key absent -> `None` (unchanged), key
/// present as `null` -> `Some(None)` (explicit clear), key present as a
/// string -> `Some(Some(s))` (set); any other JSON type -> a hard error.
fn optional_entity_type_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<String>>> {
    match obj(args)?.get(key) {
        None => Ok(None),
        Some(Value::Null) => Ok(Some(None)),
        Some(Value::String(value)) => Ok(Some(Some(value.clone()))),
        Some(other) => Err(RuntimeError::InvalidInput(format!(
            "{key} must be a string or null, got: {other}"
        ))),
    }
}

/// Nullable-string patch semantics shared by note updates and mirroring the
/// actually-reachable entity description behavior of
/// `khive-pack-kg::handlers::common::description_patch`. Canonical's field
/// type is `Option<Value>` (`UpdateParams.name`/`.description`); serde_json's
/// derived `Deserialize` for `Option<T>` intercepts a literal JSON `null` at
/// the outer `Option` boundary and maps it straight to Rust `None`
/// regardless of the inner type, so canonical's own "clear" arm is
/// unreachable through normal struct deserialization: `update(name=null)` /
/// `update(description=null)` are no-ops, not clears. This module reads raw,
/// un-deserialized JSON, so it must replicate that collapse explicitly: key
/// absent OR JSON `null` -> `None` (leave unchanged, no-op); key present as a
/// string -> `Some(Some(s))` (set); any other JSON type -> a hard error.
fn optional_string_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<String>>> {
    match obj(args)?.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(s)) => Ok(Some(Some(s.clone()))),
        Some(other) => Err(RuntimeError::InvalidInput(format!(
            "{key} must be a string or null, got: {other}"
        ))),
    }
}

/// Strict string-or-absent-or-null patch for entity `name`. Unlike
/// `optional_str`'s `.as_str()`, this does not silently drop a non-string,
/// non-null value like `name: 123` as absent: it rejects it instead of
/// reporting success for an invalid update. Canonical validates entity
/// `name` via `string_value` on `UpdateParams.name: Option<Value>`: null
/// collapses to absent at the struct-deserialize boundary (see
/// `optional_string_patch` doc above), so the reachable behavior is:
/// absent/null -> unchanged; non-null string -> set; any other JSON type ->
/// hard error. This mirrors that exactly, reading raw JSON instead of a
/// deserialized struct.
fn entity_name_patch(args: &Value) -> RuntimeResult<Option<String>> {
    match obj(args)?.get("name") {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(s)) => Ok(Some(s.clone())),
        Some(other) => Err(RuntimeError::InvalidInput(format!(
            "name must be a string, got: {other}"
        ))),
    }
}

/// Nullable-JSON-value patch for `properties`: canonical
/// `properties: Option<Value>` on `UpdateParams` collapses a literal JSON
/// `null` to Rust `None` at the struct-deserialize boundary (same collapse
/// as `optional_string_patch` above), so `properties=null` is canonically a
/// no-op (leave existing properties unchanged): not a stored JSON `null`.
/// This module reads raw JSON, so it must replicate that collapse: key
/// absent OR JSON `null` -> `None` (no merge); any other JSON value ->
/// `Some(value)` (merge), with no further shape validation at this layer.
fn optional_properties(args: &Value, key: &str) -> RuntimeResult<Option<Value>> {
    match obj(args)?.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(v) => Ok(Some(v.clone())),
    }
}

/// `tags` patch: canonical `tags: Option<Vec<String>>` on `UpdateParams`
/// collapses a literal JSON `null` to Rust `None` at the struct-deserialize
/// boundary (same collapse as above), so `tags=null` is canonically a no-op
/// (leave existing tags unchanged). A non-array, non-null value is still a
/// hard error (mirrors the type failure `UpdateParams` deserialization would
/// itself produce for a malformed `tags`).
fn optional_tags(args: &Value) -> RuntimeResult<Option<Vec<String>>> {
    match obj(args)?.get("tags") {
        None | Some(Value::Null) => Ok(None),
        Some(Value::Array(items)) => {
            let mut tags = Vec::with_capacity(items.len());
            for item in items {
                let s = item.as_str().ok_or_else(|| {
                    RuntimeError::InvalidInput("tags must be an array of strings".into())
                })?;
                tags.push(s.to_string());
            }
            Ok(Some(tags))
        }
        Some(_) => Err(RuntimeError::InvalidInput(
            "tags must be an array of strings".into(),
        )),
    }
}

fn optional_f64(args: &Value, key: &str) -> RuntimeResult<Option<f64>> {
    match obj(args)?.get(key) {
        None => Ok(None),
        Some(Value::Null) => Ok(None),
        Some(v) => v
            .as_f64()
            .map(Some)
            .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
    }
}

/// Tri-state patch extraction for `Option<Option<f64>>`-shaped fields
/// (`NotePatch::salience` / `NotePatch::decay_factor`): key absent -> `None`
/// (untouched), key present as JSON `null` -> `Some(None)` (clear), key
/// present as a number -> `Some(Some(v))` (set). Range validation lives in
/// curation.rs's `prepare_update_note_from_snapshot`, not here.
fn optional_f64_patch(args: &Value, key: &str) -> RuntimeResult<Option<Option<f64>>> {
    match obj(args)?.get(key) {
        None => Ok(None),
        Some(Value::Null) => Ok(Some(None)),
        Some(v) => v
            .as_f64()
            .map(|f| Some(Some(f)))
            .ok_or_else(|| RuntimeError::InvalidInput(format!("{key} must be a number"))),
    }
}

/// Every registered embedding model's vector table name, in the exact format
/// `curation::merge_entity_sql` uses (`"vec_{sanitize_key(model_name)}"`) —
/// reused here so atomic delete/merge purge the same tables the non-atomic
/// paths do.
fn vector_table_names(runtime: &KhiveRuntime) -> Vec<String> {
    runtime
        .registered_embedding_model_names()
        .iter()
        .map(|name| format!("vec_{}", crate::config::sanitize_key(name)))
        .collect()
}

/// A guarded (`guard: None` — best-effort mirror, matching the non-atomic
/// index-cleanup calls which don't assert a row existed) `DELETE` statement
/// against one vector table for a single subject, scoped by namespace.
///
/// Vector tables carry a real index on `(subject_id, namespace)`
/// ([`khive_db::stores::vectors`]) — this row-scan predicate is not the FTS
/// full-table-scan class this module's `purge_fts_document_statements`
/// exists to avoid, so it is left as a direct `namespace = ? AND subject_id =
/// ?` predicate.
fn purge_index_row_statement(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
    label: &str,
) -> PlanStatement {
    PlanStatement {
        statement: SqlStatement {
            sql: format!("DELETE FROM {table} WHERE namespace = ?1 AND subject_id = ?2"),
            params: vec![
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(subject_id.to_string()),
            ],
            label: Some(label.to_string()),
        },
        guard: None,
    }
}

/// The FTS-document half of an index purge: `fts_table`'s row for `subject_id`
/// (looked up via `khive_db::stores::text::rowid_map_table`, not a
/// `namespace`/`subject_id` scan — those columns are `UNINDEXED` in every
/// FTS5 DDL) plus that row's own entry in the sidecar map. Order-sensitive:
/// index 0 must run before index 1 — see `delete_document_statements`'s
/// adjacency contract.
fn purge_fts_document_statements(
    fts_table: &str,
    namespace: &str,
    subject_id: Uuid,
    label_prefix: &str,
) -> [PlanStatement; 2] {
    let [mut fts_stmt, mut map_stmt] = delete_document_statements(fts_table, namespace, subject_id);
    fts_stmt.label = Some(label_prefix.to_string());
    map_stmt.label = Some(format!("{label_prefix}-map"));
    [
        PlanStatement {
            statement: fts_stmt,
            guard: None,
        },
        PlanStatement {
            statement: map_stmt,
            guard: None,
        },
    ]
}

fn log_vector_row_delete_statement(
    table: &str,
    namespace: &str,
    subject_id: Uuid,
    label: &str,
) -> PlanStatement {
    PlanStatement {
        statement: SqlStatement {
            sql: format!(
                "INSERT INTO ann_write_log \
                 (namespace, embedding_model, kind, field, subject_id, op) \
                 SELECT namespace, embedding_model, kind, field, subject_id, 'delete' \
                 FROM {table} WHERE namespace = ?1 AND subject_id = ?2"
            ),
            params: vec![
                SqlValue::Text(namespace.to_string()),
                SqlValue::Text(subject_id.to_string()),
            ],
            label: Some(label.to_string()),
        },
        guard: None,
    }
}

/// `true` iff a table named `table` currently exists in the backing SQLite
/// database (`sqlite_master` probe, read-only — safe in async prepare, does
/// NOT open/create the vector store, so it cannot lazily create the table
/// itself).
async fn vector_table_exists(runtime: &KhiveRuntime, table: &str) -> RuntimeResult<bool> {
    let mut reader = runtime
        .sql()
        .reader()
        .await
        .map_err(RuntimeError::Storage)?;
    let row = reader
        .query_scalar(SqlStatement {
            sql: "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1".to_string(),
            params: vec![SqlValue::Text(table.to_string())],
            label: Some("atomic-delete-vec-table-exists".to_string()),
        })
        .await
        .map_err(RuntimeError::Storage)?;
    Ok(row.is_some())
}

/// Append the FTS + every registered model's vector-row purge for `subject_id`
/// (scoped to the RECORD's own namespace, matching `delete_entity`/
/// `delete_note`'s `record_tok`/`record_ns` convention: not the caller
/// token's namespace, per by-ID namespace-agnosticism) onto `statements`.
///
/// FTS tables (`fts_entities`/`fts_notes`) always exist (created at schema
/// migration time) so their purge is unconditional. `vec_*` tables are
/// created lazily on first vector-store open, so a default runtime can
/// register embedding models before any vector table necessarily exists:
/// a raw unconditional `DELETE FROM vec_*` can hit `no such table` on a
/// fresh DB. Only push the vec purge for tables that actually exist:
/// absence means the record definitionally has no vector row for that
/// model, so skipping is data-parity-correct (the non-atomic path would
/// lazily create the table then delete zero rows: same data outcome,
/// without this read-only prepare pass performing an init side effect).
async fn push_index_purge_statements(
    runtime: &KhiveRuntime,
    statements: &mut Vec<PlanStatement>,
    fts_table: &str,
    namespace: &str,
    subject_id: Uuid,
    label_prefix: &str,
) -> RuntimeResult<()> {
    statements.extend(purge_fts_document_statements(
        fts_table,
        namespace,
        subject_id,
        &format!("{label_prefix}-purge-fts"),
    ));
    for vec_table in vector_table_names(runtime) {
        if vector_table_exists(runtime, &vec_table).await? {
            statements.push(log_vector_row_delete_statement(
                &vec_table,
                namespace,
                subject_id,
                &format!("{label_prefix}-log-delete-vec-{vec_table}"),
            ));
            statements.push(purge_index_row_statement(
                &vec_table,
                namespace,
                subject_id,
                &format!("{label_prefix}-purge-vec-{vec_table}"),
            ));
        }
    }
    Ok(())
}

/// Event-store append parity for the canonical handlers that emit a
/// lifecycle event after their row mutation: `update_entity` ->
/// `EntityUpdated`, `delete_entity` -> `EntityDeleted`, `delete_note` ->
/// `NoteDeleted`, `update_edge` -> `EdgeUpdated`, `delete_edge` ->
/// `EdgeDeleted`, `link` -> `LinkCreated`/`EdgeUpdated`, and
/// `update_note` -> `NoteUpdated`. See
/// `docs/api/atomic_prepare.md#event_append_statements` for why
/// this is a `PlanStatement` rather than a `PostCommitEffect`.
///
/// Invariant: returned statements are unguarded — appended after the plan's
/// own guarded row statement, so [`apply_plan`]'s stop-on-first-failure
/// contract means they are only reached once that row mutation's guard has
/// already held. Committing the event row atomically with the mutation it
/// describes strengthens canonical's guarantee: the non-atomic handlers write
/// the event in a separate transaction, ordered but not atomic with the row
/// mutation.
pub(crate) fn event_append_statements(
    token: &NamespaceToken,
    namespace: &str,
    verb: &str,
    kind: EventKind,
    substrate: SubstrateKind,
    target_id: Uuid,
    payload: Value,
) -> RuntimeResult<Vec<PlanStatement>> {
    let record_token = token
        .with_namespace(crate::Namespace::parse(namespace).map_err(|error| {
            RuntimeError::Internal(format!("event namespace invalid: {error}"))
        })?);
    let event = crate::EventAttribution::from_token(&record_token).stamp(
        khive_storage::event::Event::new(namespace.to_string(), verb, kind, substrate, "")
            .with_target(target_id)
            .with_payload(payload),
    );
    let statements = event_insert_statements(&event)
        .map_err(|e| RuntimeError::Internal(format!("event_insert_statements: {e}")))?;
    Ok(statements
        .into_iter()
        .map(|statement| PlanStatement {
            statement,
            guard: None,
        })
        .collect())
}

// ---------------------------------------------------------------------------
// dispatch
// ---------------------------------------------------------------------------

/// Build the prepared [`AtomicOpPlan`] for one KG-substrate admissible op
/// (`update`, `delete`, `link`, `merge`). Returns a loud [`RuntimeError`] for
/// `propose`/`review`/`withdraw` (known scope gap, see module doc) and any
/// other verb (the CLI boundary must reject those before calling this — a
/// verb reaching here is either KG-substrate-admissible or a bug upstream).
pub async fn prepare_op(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    tool: &str,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    match tool {
        // `expected_kind: None` here — same reasoning as the `"delete"` arm
        // below: callers that need `update(kind=...)` parity must resolve
        // the kind spec themselves (it needs a `VerbRegistry`, unreachable
        // from this crate: see `AtomicUpdateKind`'s doc comment) and call
        // `prepare_update` directly with the resolved value; `kkernel`'s
        // `--atomic` seam does exactly this and bypasses this dispatch arm.
        // A caller reaching `prepare_op("update", ...)` without going
        // through that seam gets kind-unchecked behavior.
        "update" => prepare_update(runtime, token, args, None).await,
        // `expected_kind: None` here — callers that need `delete(kind=...)`
        // parity must resolve the kind spec themselves (it needs a
        // `VerbRegistry`, unreachable from this crate: see
        // `AtomicDeleteKind`'s doc comment) and call `prepare_delete`
        // directly with the resolved value; `kkernel`'s `--atomic` seam does
        // exactly this and bypasses this dispatch arm. A caller reaching
        // `prepare_op("delete", ...)` without going through that seam gets
        // kind-unchecked behavior.
        "delete" => prepare_delete(runtime, token, args, None).await,
        "link" => prepare_link(runtime, token, args).await,
        "merge" => prepare_merge(runtime, token, args).await,
        "propose" | "review" | "withdraw" => prepare_governance_unimplemented(tool),
        other => Err(RuntimeError::InvalidInput(format!(
            "{other:?} has no atomic_prepare::prepare_op implementation; the CLI \
             admissibility check should have rejected this before prepare"
        ))),
    }
}

fn prepare_governance_unimplemented(tool: &str) -> RuntimeResult<AtomicOpPlan> {
    Err(RuntimeError::InvalidInput(format!(
        "{tool:?} is on the ADR-099 v1 admissible verb list but has no --atomic \
         prepare/apply implementation yet: its lifecycle (ADR-046) is an \
         event-sourced changeset-interpreter over a dedicated `proposals_open` \
         table, not a small guarded-DML plan — a faithful non-stub atomic \
         prepare for it is tracked as ADR-099 follow-up work, not implemented \
         in slice B3. No write was attempted."
    )))
}

// ---------------------------------------------------------------------------
// create (AddEntity / AddNote)
// ---------------------------------------------------------------------------

/// Build the prepared plan for an `AddEntity` proposal change. The entity
/// row and FTS document are committed together; vector indexing is deferred
/// until after commit because embedding may suspend. `kind` must already be
/// canonicalized by the caller because pack-aware resolution requires a
/// `VerbRegistry`.
pub async fn prepare_add_entity(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    let kind = require_str(args, "kind")?;
    let name = require_str(args, "name")?;
    runtime.validate_entity_kind(kind)?;
    if name.trim().is_empty() {
        return Err(RuntimeError::InvalidInput(
            "name must not be empty".to_string(),
        ));
    }

    let description = optional_create_string(args, "description")?;
    let properties = optional_properties(args, "properties")?;
    let tags = optional_tags(args)?.unwrap_or_default();

    crate::secret_gate::check_at(name, "entity", "name")?;
    if let Some(ref d) = description {
        crate::secret_gate::check_at(d, "entity", "description")?;
    }
    if let Some(ref p) = properties {
        crate::secret_gate::check_json_at(p, "entity", "properties")?;
    }
    crate::secret_gate::check_tags_at(&tags, "entity", "tags")?;
    crate::secret_gate::reject_reserved_secret_gate_property(properties.as_ref())?;

    let ns = token.namespace().as_str();
    let mut entity = khive_storage::Entity::new(ns, kind, name);
    if let Some(d) = description {
        entity = entity.with_description(d);
    }
    if let Some(p) = properties {
        entity = entity.with_properties(p);
    }
    if !tags.is_empty() {
        entity = entity.with_tags(tags);
    }

    let mut statements = vec![PlanStatement {
        statement: entity_upsert_statement(&entity),
        guard: Some(AffectedRowGuard::exactly(1)),
    }];
    // Order-sensitive pair — see `insert_document_statements`'s adjacency
    // contract: the map upsert's `last_insert_rowid()` must read back the
    // FTS insert immediately before it.
    for statement in insert_document_statements("fts_entities", &entity_fts_document(&entity)) {
        statements.push(PlanStatement {
            statement,
            guard: None,
        });
    }

    Ok(AtomicOpPlan::AddEntity(AddEntityPlan {
        entity_id: entity.id,
        statements,
        post_commit: PostCommitEffect::ReindexEntity {
            entity_id: entity.id,
        },
    }))
}

/// Build the prepared plan for an `AddNote` proposal change. Mirrors
/// [`prepare_add_entity`]'s shape and the same
/// `kind`-already-canonicalized split. `annotates` is out of scope: the
/// proposal `NoteDraft` this backs carries no annotates targets, unlike
/// `KhiveRuntime::create_note`'s general-purpose signature.
pub async fn prepare_add_note(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    let kind = require_str(args, "kind")?;
    let content = require_str(args, "content")?;
    runtime.validate_note_kind(kind)?;

    let name = optional_create_string(args, "name")?;
    let properties = optional_properties(args, "properties")?;
    // Same note-write validator `create_note_inner` runs: this path builds its
    // args itself and dispatches no pack hook, so without this call a proposal
    // changeset would be the one note-write that stores caller-supplied owned
    // identity properties verbatim. The token here is the applying caller's
    // (threaded in by the apply worker), not the proposer's.
    let properties = runtime.derive_note_write_properties(kind, token, properties)?;

    crate::secret_gate::check_at(content, "note", "content")?;
    if let Some(ref n) = name {
        crate::secret_gate::check_at(n, "note", "name")?;
    }
    if let Some(ref p) = properties {
        crate::secret_gate::check_json_at(p, "note", "properties")?;
    }
    crate::secret_gate::reject_reserved_secret_gate_property(properties.as_ref())?;

    let ns = token.namespace().as_str();
    let mut note = khive_storage::note::Note::new(ns, kind, content);
    if let Some(n) = name {
        note = note.with_name(n);
    }
    if let Some(p) = properties {
        note = note.with_properties(p);
    }

    let mut statements = vec![PlanStatement {
        statement: note_upsert_statement(&note),
        guard: Some(AffectedRowGuard::exactly(1)),
    }];
    // Order-sensitive pair — see `insert_document_statements`'s adjacency
    // contract.
    for statement in insert_document_statements("fts_notes", &note_fts_document(&note)) {
        statements.push(PlanStatement {
            statement,
            guard: None,
        });
    }

    Ok(AtomicOpPlan::AddNote(Box::new(AddNotePlan {
        note_guard: None,
        note_id: note.id,
        statements,
        post_commit: PostCommitEffect::ReindexNote {
            note_id: note.id,
            version: note.version,
        },
    })))
}

// ---------------------------------------------------------------------------
// update
// ---------------------------------------------------------------------------

/// Mirrors `khive-pack-kg::handlers::update::reject_inapplicable_fields`: a
/// hard `InvalidInput` when a caller passes a field that does not apply to
/// the resolved substrate (e.g. `salience` on an entity, or
/// `description` on a note). That function has no dependency edge
/// back to `khive-runtime`, so its exact field-applicability check list and
/// error message shape are reimplemented here rather than imported: same
/// pattern as `optional_string_patch` above. Presence is checked directly on
/// the raw args object (this module has no `UpdateParams` struct); a JSON
/// `null` value is treated as absent, matching `Option<T>` deserialization
/// semantics.
fn reject_inapplicable_update_fields(args: &Value, substrate: &str) -> RuntimeResult<()> {
    let o = obj(args)?;
    if substrate == "edge" && o.get("expected_version").is_some_and(|v| !v.is_null()) {
        return Err(RuntimeError::InvalidInput(
            "expected_version applies only to entities and notes".into(),
        ));
    }
    if substrate != "note"
        && ["embed", "fence"]
            .iter()
            .any(|field| o.contains_key(*field))
    {
        return Err(RuntimeError::InvalidInput(
            "embed and fence apply only to notes".into(),
        ));
    }
    let present = |k: &str| o.get(k).is_some_and(|v| !v.is_null());
    let (bad_field, valid): (Option<&str>, &str) = match substrate {
        "entity" => {
            let bad = if present("content") {
                Some("content")
            } else if present("salience") {
                Some("salience")
            } else if present("decay_factor") {
                Some("decay_factor")
            } else if present("relation") {
                Some("relation")
            } else if present("weight") {
                Some("weight")
            } else {
                None
            };
            (bad, "name, description, tags, properties, entity_type")
        }
        "note" => {
            let bad = if present("description") {
                Some("description")
            } else if present("relation") {
                Some("relation")
            } else if present("weight") {
                Some("weight")
            } else if o.contains_key("entity_type") {
                // ADR-014 tri-state: a PRESENT key (including JSON `null`,
                // the explicit clear) is inapplicable to notes.
                Some("entity_type")
            } else {
                None
            };
            (
                bad,
                "name, content, salience, decay_factor, properties, tags",
            )
        }
        // `update` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`, so
        // this arm must reject entity/note-only fields (e.g. `name`) on an
        // edge update rather than silently skip the guard, mirroring
        // `khive-pack-kg::handlers::update::reject_inapplicable_fields`'s
        // `KindSpec::Edge` arm.
        "edge" => {
            let bad = if present("name") {
                Some("name")
            } else if present("description") {
                Some("description")
            } else if present("content") {
                Some("content")
            } else if present("tags") {
                Some("tags")
            } else if present("salience") {
                Some("salience")
            } else if present("decay_factor") {
                Some("decay_factor")
            } else if o.contains_key("entity_type") {
                // ADR-014 tri-state: a PRESENT key (including JSON `null`,
                // the explicit clear) is inapplicable to edges.
                Some("entity_type")
            } else {
                None
            };
            (bad, "relation, weight, properties")
        }
        _ => (None, ""),
    };
    if let Some(field) = bad_field {
        let substrate_label = match substrate {
            "entity" => "an entity",
            "note" => "a note",
            "edge" => "an edge",
            other => other,
        };
        return Err(RuntimeError::InvalidInput(format!(
            "field '{field}' is not valid for {substrate_label}; valid fields: {valid}"
        )));
    }
    Ok(())
}

/// Caller-supplied update-kind expectation, resolved via the canonical
/// `resolve_kind_spec` at the kkernel `--atomic` seam: the same pattern
/// [`AtomicDeleteKind`] uses. Without this check, `update(kind="document",
/// id=<concept>)` would be canonically `NotFound` but the atomic path would
/// ignore the explicit kind and mutate the resolved entity anyway.
/// `khive-runtime` must not depend on `khive-pack-kg`, so this is a plain
/// substrate-level shape rather than `khive_pack_kg::handlers::KindSpec`
/// itself: the kkernel seam does the pack-aware resolution and passes down
/// only what `prepare_update` needs to enforce the mismatch check.
pub enum AtomicUpdateKind {
    Entity { specific: Option<String> },
    Note { specific: Option<String> },
    Edge,
}

/// Enforce a caller's explicit update-kind discriminator against a resolved
/// note before any pack hook can inspect or normalize the request. Canonical
/// KG dispatch performs this mismatch check before its hook; the atomic
/// adapter calls this same helper to preserve that error ordering.
pub fn validate_note_update_expected_kind(
    note: &khive_storage::note::Note,
    expected_kind: &Option<AtomicUpdateKind>,
) -> RuntimeResult<()> {
    let id = note.id;
    match expected_kind {
        None => Ok(()),
        Some(AtomicUpdateKind::Note {
            specific: Some(expected),
        }) if &note.kind != expected => Err(RuntimeError::NotFound(format!("note {id}"))),
        Some(AtomicUpdateKind::Note { .. }) => Ok(()),
        Some(AtomicUpdateKind::Entity { .. }) => {
            Err(RuntimeError::NotFound(format!("entity {id}")))
        }
        Some(AtomicUpdateKind::Edge) => Err(RuntimeError::NotFound(format!("edge {id}"))),
    }
}

async fn prepare_note_update_plan_from_snapshot(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
    expected_kind: &Option<AtomicUpdateKind>,
    note: khive_storage::note::Note,
    policy: crate::NoteUpdatePolicy,
    registry: Option<&crate::VerbRegistry>,
) -> RuntimeResult<(khive_storage::Note, UpdatePlan)> {
    let id = require_uuid(args, "id")?;
    if note.id != id {
        return Err(RuntimeError::NotFound(format!("note {id}")));
    }
    validate_note_update_expected_kind(&note, expected_kind)?;

    reject_inapplicable_update_fields(args, "note")?;
    let mut normalized_args = args.clone();
    crate::curation::normalize_note_update_tags(&mut normalized_args)?;
    let args = &normalized_args;
    let name = optional_string_patch(args, "name")?;
    let content = optional_str(args, "content").map(str::to_string);
    let properties = optional_properties(args, "properties")?;
    let salience = optional_f64_patch(args, "salience")?;
    let decay_factor = optional_f64_patch(args, "decay_factor")?;
    let options = crate::note_write::NoteWriteOptions {
        expected_version: obj(args)?
            .get("expected_version")
            .filter(|v| !v.is_null())
            .map(|v| {
                v.as_i64().ok_or_else(|| {
                    RuntimeError::InvalidInput("expected_version must be an integer".into())
                })
            })
            .transpose()?,
        fence: obj(args)?
            .get("fence")
            .map(|v| {
                serde_json::from_value(v.clone())
                    .map_err(|error| RuntimeError::InvalidInput(format!("invalid fence: {error}")))
            })
            .transpose()?,
        embed: obj(args)?
            .get("embed")
            .filter(|v| !v.is_null())
            .map(|v| {
                v.as_bool()
                    .ok_or_else(|| RuntimeError::InvalidInput("embed must be boolean".into()))
            })
            .transpose()?,
        key: None,
    };
    let patch = crate::curation::NotePatch::new(name, content, salience, decay_factor, properties)
        .with_update_policy(policy)
        .with_write_options(options);
    let (updated, mut plan) = runtime
        .prepare_versioned_note_update(token, note.clone(), patch.clone())
        .await?;
    if let Some(registry) = registry {
        attach_note_update_effects(runtime, token, registry, &note, &patch, &mut plan).await?;
    }
    Ok((updated, plan))
}

/// Build an atomic update plan from the exact note snapshot already supplied
/// to a pack update hook. Persistence is guarded by that snapshot's revision
/// and deletion marker, so hook normalization cannot race a second read. The
/// caller must first run the registry normalizer/validator against this snapshot.
/// This shared canonical/atomic seam prepares all patch fields with the owning
/// kind's property policy (the value `VerbRegistry::prepare_note_update_policy`
/// returned for this snapshot), then derives and attaches the owner's typed
/// graph effects. It returns the projected note and one atomic plan, preserving
/// the caller's operation index.
pub async fn prepare_update_from_note_snapshot(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
    expected_kind: Option<AtomicUpdateKind>,
    note: khive_storage::note::Note,
    policy: crate::NoteUpdatePolicy,
    registry: &crate::VerbRegistry,
) -> RuntimeResult<(khive_storage::Note, AtomicOpPlan)> {
    if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
        return Err(RuntimeError::InvalidInput(
            "entity_kind is immutable; to change kind, delete then re-create the entity, \
             or use merge() if this is a deduplication correction"
                .into(),
        ));
    }
    let (note, plan) = prepare_note_update_plan_from_snapshot(
        runtime,
        token,
        args,
        &expected_kind,
        note,
        policy,
        Some(registry),
    )
    .await?;
    Ok((note, AtomicOpPlan::Update(Box::new(plan))))
}

/// The only companion attachment site. Canonical, CLI atomic, and stream note
/// updates all enter through `prepare_update_from_note_snapshot` after running
/// the registry's normalizer/validator on the same snapshot.
async fn attach_note_update_effects(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    registry: &crate::VerbRegistry,
    snapshot: &khive_storage::Note,
    patch: &crate::curation::NotePatch,
    plan: &mut UpdatePlan,
) -> RuntimeResult<()> {
    use crate::atomic_plan::NoteUpdateStatement;
    use crate::NoteUpdateEffect;

    let Some(hook) = registry.find_kind_hook(&snapshot.kind) else {
        return Ok(());
    };
    // Refuse already-known stale input before asking the owner to derive effects.
    // A change after this read remains protected by the note's first CAS statement.
    if let Some(expected) = patch.write_options.expected_version {
        if expected != snapshot.version {
            return Err(crate::note_write::NoteWriteConflict::Version {
                expected,
                current: snapshot.version,
            }
            .into_error()
            .into());
        }
    }
    let current = runtime.notes(token)?.get_note(snapshot.id).await?;
    if !current.is_some_and(|current| {
        current.updated_at == snapshot.updated_at
            && current.deleted_at == snapshot.deleted_at
            && current.version == snapshot.version
    }) {
        return Err(crate::curation::stale_note_snapshot_error(snapshot.id));
    }
    let effects = hook
        .note_update_effects(runtime, token, snapshot, patch)
        .await?;
    if plan.idempotent_noop && !effects.is_empty() {
        return Err(RuntimeError::InvalidInput(
            "an unchanged note update cannot carry graph effects".into(),
        ));
    }
    let edge_token = token.with_namespace(
        crate::Namespace::parse(&snapshot.namespace)
            .map_err(|error| RuntimeError::Internal(format!("invalid note namespace: {error}")))?,
    );
    for effect in effects {
        match effect {
            NoteUpdateEffect::Link(spec) => {
                if spec.source_id != snapshot.id
                    || spec
                        .namespace
                        .as_deref()
                        .is_some_and(|ns| ns != snapshot.namespace)
                {
                    return Err(RuntimeError::InvalidInput(
                        "note update links must originate from the note in its namespace".into(),
                    ));
                }
                let mut args = serde_json::json!({
                    "source_id": spec.source_id, "target_id": spec.target_id,
                    "relation": spec.relation, "weight": spec.weight,
                    "resurrect": spec.resurrect,
                });
                if let Some(metadata) = spec.metadata {
                    args["metadata"] = metadata;
                }
                let AtomicOpPlan::Link(link) = prepare_link(runtime, &edge_token, &args).await?
                else {
                    return Err(RuntimeError::Internal("expected a link plan".into()));
                };
                // A live annotation inserted between the owner's read and this
                // lookup belongs to that writer. Refuse instead of replacing its
                // weight/metadata; a fresh preparation can preserve it explicitly.
                if link.disposition == EdgeUpsertDisposition::Updated {
                    return Err(khive_types::KhiveError::conflict(
                        "a live edge appeared while preparing the note update; retry with fresh state",
                    ).into());
                }
                plan.graph_effects
                    .extend(link.statements.into_iter().map(NoteUpdateStatement::Write));
            }
            NoteUpdateEffect::DeleteEdge(edge) => {
                let id = Uuid::from(edge.id);
                if edge.source_id != snapshot.id
                    || edge.namespace != snapshot.namespace
                    || edge.deleted_at.is_some()
                {
                    return Err(RuntimeError::InvalidInput(
                        "note update deletes must select an outgoing edge in the note namespace"
                            .into(),
                    ));
                }
                plan.graph_effects
                    .push(NoteUpdateStatement::Assert(PlanStatement {
                        statement: khive_db::stores::graph::edge_snapshot_assertion_statement(
                            &edge, false,
                        ),
                        guard: Some(AffectedRowGuard::exactly(1)),
                    }));
                let actor = format!("{}:{}", token.actor().kind, token.actor().id);
                let AtomicOpPlan::Delete(delete) =
                    prepare_delete_edge(&edge_token, id, edge, false, &actor).await?
                else {
                    return Err(RuntimeError::Internal(
                        "expected an edge delete plan".into(),
                    ));
                };
                if delete.post_commit != PostCommitEffect::None {
                    return Err(RuntimeError::Internal(
                        "edge delete has a deferred effect".into(),
                    ));
                }
                plan.graph_effects.extend(
                    delete
                        .statements
                        .into_iter()
                        .map(NoteUpdateStatement::Write),
                );
            }
            NoteUpdateEffect::AssertLink(edge) => {
                if edge.source_id != snapshot.id
                    || edge.namespace != snapshot.namespace
                    || edge.deleted_at.is_some()
                {
                    return Err(RuntimeError::InvalidInput(
                        "note update assertions must select a live outgoing edge in the note namespace".into(),
                    ));
                }
                plan.graph_effects
                    .push(NoteUpdateStatement::Assert(PlanStatement {
                        statement: khive_db::stores::graph::edge_snapshot_assertion_statement(
                            &edge, true,
                        ),
                        guard: Some(AffectedRowGuard::exactly(1)),
                    }));
            }
        }
    }
    Ok(())
}

/// `expected_kind`: `None` when the caller omitted `kind` (no check, parity
/// with canonical's own optional discriminator); `Some(_)` enforces an
/// exact-parity mismatch check against the resolved record's actual
/// substrate/specific kind, mirroring `handle_update`'s
/// `entity.kind != *k` / note kind checks (update.rs:200-201, :229-234).
/// Task notes must pass the GTD pack hook before this or
/// `prepare_op("update", ..)`; neither runs it.
pub async fn prepare_update(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
    expected_kind: Option<AtomicUpdateKind>,
) -> RuntimeResult<AtomicOpPlan> {
    let id = require_uuid(args, "id")?;

    // Mirrors update.rs's entity_kind immutability guard: entity_kind is a
    // legacy top-level field, independent of the `kind` substrate
    // discriminator handled elsewhere.
    if obj(args)?.get("entity_kind").is_some_and(|v| !v.is_null()) {
        return Err(RuntimeError::InvalidInput(
            "entity_kind is immutable; to change kind, delete then re-create the entity, \
             or use merge() if this is a deduplication correction"
                .into(),
        ));
    }

    match runtime.resolve_by_id(token, id).await? {
        Some(Resolved::Entity(entity)) => {
            match &expected_kind {
                None => {}
                Some(AtomicUpdateKind::Entity {
                    specific: Some(expected),
                }) if &entity.kind != expected => {
                    return Err(RuntimeError::NotFound(format!("entity {id}")));
                }
                Some(AtomicUpdateKind::Entity { .. }) => {}
                Some(AtomicUpdateKind::Note { .. }) => {
                    return Err(RuntimeError::NotFound(format!("note {id}")));
                }
                Some(AtomicUpdateKind::Edge) => {
                    return Err(RuntimeError::NotFound(format!("edge {id}")));
                }
            }
            // Decide step lives in curation.rs's `prepare_update_entity` —
            // the SAME function canonical `update_entity` calls. Only the
            // arg-extraction (raw JSON -> `EntityPatch`) and the plan-shape
            // wiring are atomic-path-specific: the domain object becomes a
            // `PlanStatement` via `entity_replace_if_unchanged_statement`,
            // whose CAS predicate binds the snapshot's revision and deletion
            // marker, under an exactly-one affected-row guard.
            reject_inapplicable_update_fields(args, "entity")?;
            let name = entity_name_patch(args)?;
            let description = optional_string_patch(args, "description")?;
            let properties = optional_properties(args, "properties")?;
            let tags = optional_tags(args)?;
            let entity_type = optional_entity_type_patch(args, "entity_type")?;

            let expected_version = obj(args)?
                .get("expected_version")
                .filter(|v| !v.is_null())
                .map(|value| {
                    value.as_i64().ok_or_else(|| {
                        RuntimeError::InvalidInput("expected_version must be an integer".into())
                    })
                })
                .transpose()?;
            prepare_update_entity_plan_with_version(
                runtime,
                token,
                id,
                crate::curation::EntityPatch {
                    name,
                    description,
                    properties,
                    tags,
                    entity_type,
                },
                expected_version,
            )
            .await
        }
        Some(Resolved::Note(note)) => {
            // Patch application lives in curation.rs's
            // `prepare_update_note_from_snapshot` — the same implementation
            // canonical guarded update calls, including salience/decay range
            // validation. The plan retains this exact snapshot's revision.
            let (_, plan) = prepare_note_update_plan_from_snapshot(
                runtime,
                token,
                args,
                &expected_kind,
                note,
                crate::NoteUpdatePolicy::default(),
                None,
            )
            .await?;
            Ok(AtomicOpPlan::Update(Box::new(plan)))
        }
        Some(_) => Err(RuntimeError::InvalidInput(format!(
            "update target {id} must be an entity, note, or edge"
        ))),
        // `Resolved` (khive-runtime::operations) has no `Edge` variant — an
        // id that isn't an entity/note/pack-private/event record is checked
        // against the graph store directly, mirroring
        // `khive-pack-kg::handlers::KgPack::infer_kind_from_uuid`'s own
        // entity/note-then-edge fallback order. `update` admits
        // `kind="edge"`, so this arm must be able to build a plan for one.
        None => match &expected_kind {
            Some(AtomicUpdateKind::Entity { .. }) => {
                Err(RuntimeError::NotFound(format!("entity/note {id}")))
            }
            Some(AtomicUpdateKind::Note { .. }) => {
                Err(RuntimeError::NotFound(format!("entity/note {id}")))
            }
            Some(AtomicUpdateKind::Edge) | None => match runtime.get_edge(token, id).await? {
                Some(edge) => prepare_update_edge(runtime, token, id, edge, args).await,
                None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
            },
        },
    }
}

/// Build an entity update plan from a typed patch. Proposal changesets use
/// this entry point so their explicit `description: null` clear operation is
/// preserved instead of being collapsed by raw verb deserialization.
pub async fn prepare_update_entity_plan(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    id: Uuid,
    patch: crate::curation::EntityPatch,
) -> RuntimeResult<AtomicOpPlan> {
    prepare_update_entity_plan_with_version(runtime, token, id, patch, None).await
}

pub(crate) async fn prepare_update_entity_plan_with_version(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    id: Uuid,
    patch: crate::curation::EntityPatch,
    expected_version: Option<i64>,
) -> RuntimeResult<AtomicOpPlan> {
    crate::entity_write::validate_expected_version(expected_version)?;
    let (entity, reindex_required, changed_fields, expected_updated_at, expected_deleted_at) =
        runtime.prepare_update_entity(token, id, patch).await?;
    let mut statements = vec![PlanStatement {
        statement: entity_replace_if_unchanged_statement(
            &entity,
            expected_updated_at,
            expected_deleted_at,
        ),
        guard: Some(AffectedRowGuard::exactly(1)),
    }];
    statements.extend(event_append_statements(
        token,
        &entity.namespace,
        "update",
        EventKind::EntityUpdated,
        SubstrateKind::Entity,
        id,
        serde_json::json!({
            "id": id,
            "namespace": entity.namespace,
            "changed_fields": changed_fields,
        }),
    )?);
    let post_commit = if reindex_required {
        PostCommitEffect::ReindexEntity { entity_id: id }
    } else {
        PostCommitEffect::None
    };
    Ok(AtomicOpPlan::Update(Box::new(UpdatePlan {
        graph_effects: Vec::new(),
        note_vector_purge: None,
        note_embedding_inheritance: None,
        entity_guard: expected_version.map(|expected_version| {
            crate::entity_write::EntityWriteGuard {
                id,
                expected_version,
            }
        }),
        note_guard: None,
        target_id: id,
        statements,
        post_commit,
        edge_natural_key: None,
        idempotent_noop: false,
    })))
}

/// Edge branch of `prepare_update`. Mirrors `KhiveRuntime::update_edge`'s
/// patch semantics: `relation`/`weight`/`properties` are the only applicable
/// fields, a changed `relation` is endpoint-validated first, `weight` is
/// range-checked, and `properties` REPLACES `metadata` wholesale (no merge).
/// See `docs/api/atomic_prepare.md#prepare_update_edge` for the DML-shape parity
/// detail with `update_edge`.
///
/// Invariant (symmetric relations `competes_with`/`composed_with`): this
/// function must never branch on a prepare-time conflict probe — a different
/// op in the same atomic unit could change the conflict landscape between
/// probe and commit, making any such branch stale by construction. It always
/// emits BOTH statements from [`edge_symmetric_delete_if_conflict_statement`]
/// and [`edge_symmetric_absorb_or_update_inplace_statement`], each carrying
/// its own commit-time `WHERE`/`CASE WHEN` predicate that re-evaluates the
/// conflict condition fresh inside the transaction. This function reads no
/// state to guess a surviving id; the plan instead carries `edge_natural_key`
/// so a post-commit caller derives the actual surviving id from the
/// committed row, never from a value computed before the rest of this atomic
/// unit has even run.
async fn prepare_update_edge(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    id: Uuid,
    mut edge: khive_storage::types::Edge,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    reject_inapplicable_update_fields(args, "edge")?;

    let expected_updated_at = edge.updated_at;
    let expected_deleted_at = edge.deleted_at;

    let relation_raw = optional_str(args, "relation");
    let weight = optional_f64(args, "weight")?;
    let properties = optional_properties(args, "properties")?;

    if let Some(ref p) = properties {
        crate::secret_gate::check_json_at(p, "edge", "properties")?;
    }
    crate::secret_gate::reject_reserved_secret_gate_property(properties.as_ref())?;

    let namespace = edge.namespace.clone();
    let record_tok = token.with_namespace(
        khive_types::Namespace::parse(&namespace)
            .map_err(|e| RuntimeError::Internal(format!("edge namespace invalid: {e}")))?,
    );

    let mut changed_fields: Vec<&'static str> = Vec::new();
    if let Some(raw) = relation_raw {
        let relation = parse_edge_relation(raw)?;
        runtime
            .validate_edge_relation_endpoints(&record_tok, edge.source_id, edge.target_id, relation)
            .await?;
        edge.relation = relation;
        changed_fields.push("relation");
    }
    if let Some(w) = weight {
        if !w.is_finite() || !(0.0..=1.0).contains(&w) {
            return Err(RuntimeError::InvalidInput(format!(
                "edge weight must be a finite value in [0.0, 1.0]; got {w}"
            )));
        }
        edge.weight = w;
        changed_fields.push("weight");
    }
    if let Some(p) = properties {
        edge.metadata = Some(p);
        changed_fields.push("properties");
    }

    let (canon_src, canon_tgt) =
        canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
    let now = chrono::Utc::now();

    let mut statements: Vec<PlanStatement> = Vec::new();
    let mut edge_natural_key: Option<EdgeNaturalKey> = None;

    if edge.relation.is_symmetric() {
        // The write for a symmetric relation never branches on a
        // prepare-time probe result: it always carries both self-guarding,
        // commit-time-predicate statements (see their doc comment in
        // khive-db's graph.rs for the full rationale). This avoids the
        // staleness window a prepare-time probe would expose: an earlier op
        // in the same atomic unit could change the conflict landscape before
        // commit. Canonical's own probe-then-branch
        // `update_edge_symmetric_dml` has no such exposure (single
        // transaction, no interleaving) and is unaffected.
        let metadata_str = edge
            .metadata
            .as_ref()
            .map(|v| serde_json::to_string(v).unwrap_or_default());

        // `updated_at` must strictly advance past the snapshot even when two
        // operations land inside one clock microsecond; saturating to
        // i64::MAX would let the CAS accept a write without advancing its
        // revision, so that is not a valid fallback (mirrors the note path).
        let minimum_updated_at_micros = expected_updated_at
            .timestamp_micros()
            .checked_add(1)
            .ok_or_else(|| {
                RuntimeError::Internal(format!(
                    "edge {id} updated_at is already at i64::MAX and cannot advance"
                ))
            })?;
        let symmetric_updated_at_micros = now.timestamp_micros().max(minimum_updated_at_micros);
        let expected_deleted_at_micros = expected_deleted_at.map(|v| v.timestamp_micros());

        statements.push(PlanStatement {
            statement: edge_symmetric_delete_if_conflict_statement(
                &namespace,
                id,
                canon_src,
                canon_tgt,
                edge.relation,
                expected_updated_at.timestamp_micros(),
                expected_deleted_at_micros,
            ),
            guard: Some(AffectedRowGuard {
                expected_min: 0,
                expected_max: Some(1),
            }),
        });
        statements.push(PlanStatement {
            statement: edge_symmetric_absorb_or_update_inplace_statement(
                &namespace,
                id,
                canon_src,
                canon_tgt,
                edge.relation,
                edge.weight,
                symmetric_updated_at_micros,
                metadata_str.as_deref(),
                edge.target_backend.as_deref(),
                expected_updated_at.timestamp_micros(),
                expected_deleted_at_micros,
            ),
            guard: Some(AffectedRowGuard::exactly(1)),
        });

        // No prepare-time read needed: the two statements above are
        // self-guarding at commit time (see their doc comment). Post-commit
        // result rendering derives the actual surviving id from THIS
        // natural key, never from a value computed here.
        edge_natural_key = Some(EdgeNaturalKey {
            namespace: namespace.clone(),
            canon_source_id: canon_src,
            canon_target_id: canon_tgt,
            relation: edge.relation,
        });
    } else {
        // Non-symmetric: guarded replace of the read snapshot rather than
        // `graph.upsert_edge`'s unconditional natural-key upsert — a zero
        // affected-row result now means a concurrent writer moved this edge
        // between PREPARE and commit, and the atomic unit must roll back
        // instead of silently overwriting it.
        //
        // `updated_at` must strictly advance past the snapshot even when two
        // operations land inside one clock microsecond; saturating to
        // i64::MAX would let the CAS accept a write without advancing its
        // revision, so that is not a valid fallback (mirrors the note path).
        let minimum_updated_at_micros = expected_updated_at
            .timestamp_micros()
            .checked_add(1)
            .ok_or_else(|| {
                RuntimeError::Internal(format!(
                    "edge {id} updated_at is already at i64::MAX and cannot advance"
                ))
            })?;
        let now_micros = now.timestamp_micros().max(minimum_updated_at_micros);
        edge.updated_at = chrono::DateTime::from_timestamp_micros(now_micros).ok_or_else(|| {
            RuntimeError::Internal(format!(
                "edge {id}: computed updated_at {now_micros} is not a valid timestamp"
            ))
        })?;
        statements.push(PlanStatement {
            statement: edge_replace_if_unchanged_statement(
                &edge,
                expected_updated_at,
                expected_deleted_at,
            ),
            guard: Some(AffectedRowGuard::exactly(1)),
        });
    }

    // Mirrors `update_edge`'s unconditional post-mutation `EdgeUpdated`
    // event append, keyed on the original `edge_id` the caller supplied:
    // canonical does the same (the event target is `edge_id`, not the
    // post-absorption surviving id).
    statements.extend(event_append_statements(
        token,
        &namespace,
        "update",
        EventKind::EdgeUpdated,
        SubstrateKind::Entity,
        id,
        serde_json::json!({"id": id, "namespace": namespace, "changed_fields": changed_fields}),
    )?);

    Ok(AtomicOpPlan::Update(Box::new(UpdatePlan {
        graph_effects: Vec::new(),
        note_vector_purge: None,
        note_embedding_inheritance: None,
        entity_guard: None,
        note_guard: None,
        target_id: id,
        statements,
        post_commit: PostCommitEffect::None,
        edge_natural_key,
        idempotent_noop: false,
    })))
}

// ---------------------------------------------------------------------------
// delete
// ---------------------------------------------------------------------------

/// Caller-supplied delete-kind expectation, resolved via the canonical
/// `resolve_kind_spec` at the kkernel `--atomic` seam. `khive-runtime` must
/// not depend on `khive-pack-kg` (packs depend on the runtime, not the other
/// way around), so this is a plain substrate-level shape rather than
/// `khive_pack_kg::handlers::KindSpec` itself: the kkernel seam does the
/// pack-aware `resolve_kind_spec` resolution (which needs a `VerbRegistry`,
/// unreachable from this crate) and passes down only what `prepare_delete`
/// needs to enforce the mismatch check.
///
/// `delete` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`, hence the
/// `Edge` variant. `Event`/`Proposal` remain rejected at the kkernel seam
/// (not v1-admissible for atomic delete at all).
pub enum AtomicDeleteKind {
    Entity { specific: Option<String> },
    Note { specific: Option<String> },
    Edge,
}

/// `expected_kind`: `None` when the caller omitted `kind` (no check, parity
/// with canonical's own optional discriminator); `Some(_)` enforces an
/// exact-parity mismatch check against the resolved record's actual
/// substrate/specific kind, mirroring `handle_delete`'s
/// `entity.kind != *expected` / `note.kind != *expected` checks.
pub async fn prepare_delete(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
    expected_kind: Option<AtomicDeleteKind>,
) -> RuntimeResult<AtomicOpPlan> {
    let id = require_uuid(args, "id")?;
    let actor = format!("{}:{}", token.actor().kind, token.actor().id);
    let hard = obj(args)?
        .get("hard")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);

    // `delete(id, hard=true)` is the public purge route after a prior soft
    // delete, so it must resolve including already-tombstoned rows (a
    // live-only resolve would never find one). Soft delete keeps the
    // live-only resolve: a soft delete of an already-tombstoned row is a
    // no-op, matching non-atomic behavior.
    let resolved = if hard {
        runtime.resolve_by_id_including_deleted(token, id).await?
    } else {
        runtime.resolve_by_id(token, id).await?
    };

    match resolved {
        Some(Resolved::Entity(entity)) => {
            match &expected_kind {
                None => {}
                Some(AtomicDeleteKind::Entity {
                    specific: Some(expected),
                }) if &entity.kind != expected => {
                    return Err(RuntimeError::NotFound(format!("{expected} {id}")));
                }
                Some(AtomicDeleteKind::Entity { .. }) => {}
                Some(AtomicDeleteKind::Note { .. }) => {
                    return Err(RuntimeError::NotFound(format!("note {id}")));
                }
                Some(AtomicDeleteKind::Edge) => {
                    return Err(RuntimeError::NotFound(format!("edge {id}")));
                }
            }
            let namespace = entity.namespace.clone();
            // Storage parity: `entity_soft_delete_statement`/
            // `entity_hard_delete_statement` are the SAME khive-db builders
            // khive-db's own `SqlEntityStore::delete_entity` calls — no DML
            // text is hand-duplicated here.
            let mut statements = if hard {
                vec![
                    PlanStatement {
                        statement: delete_record_attachments_statement(
                            id,
                            AttachmentSubstrate::Entity,
                        ),
                        guard: None,
                    },
                    PlanStatement {
                        statement: entity_hard_delete_statement(id),
                        guard: Some(AffectedRowGuard::exactly(1)),
                    },
                ]
            } else {
                let deleted_at = chrono::Utc::now().timestamp_micros();
                vec![PlanStatement {
                    statement: entity_soft_delete_statement(id, deleted_at),
                    guard: Some(AffectedRowGuard::exactly(1)),
                }]
            };
            if hard {
                statements.extend(
                    hard_delete_lineage_warning_statements(
                        &namespace,
                        &actor,
                        id,
                        SubstrateKind::Entity,
                    )
                    .into_iter()
                    .map(|statement| PlanStatement {
                        statement,
                        guard: None,
                    }),
                );
                // Same builder canonical `delete_entity`'s hard-delete
                // cascade calls (`graph.purge_incident_edges`).
                statements.push(PlanStatement {
                    statement: purge_incident_edges_statement(id),
                    guard: None,
                });
            }
            // FTS + vector index purge, matching operations.rs
            // `delete_entity`: both soft and hard delete clean indexes (a
            // hard delete of an already-tombstoned record must still purge
            // them); only hard additionally cascades edges above.
            push_index_purge_statements(
                runtime,
                &mut statements,
                "fts_entities",
                &namespace,
                id,
                "atomic-delete-entity",
            )
            .await?;
            // operations.rs's `delete_entity` appends an `EntityDeleted`
            // event after a successful row delete, on both soft and hard
            // delete. `apply_plan` never reaches this statement unless the
            // guarded row statement above affected a row, so no extra `if`
            // is needed here.
            statements.extend(event_append_statements(
                token,
                &namespace,
                "delete",
                EventKind::EntityDeleted,
                SubstrateKind::Entity,
                id,
                serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
            )?);
            Ok(AtomicOpPlan::Delete(DeletePlan {
                target_id: id,
                statements,
                post_commit: PostCommitEffect::None,
            }))
        }
        Some(Resolved::Note(note)) => {
            match &expected_kind {
                None => {}
                Some(AtomicDeleteKind::Note {
                    specific: Some(expected),
                }) if &note.kind != expected => {
                    return Err(RuntimeError::NotFound(format!("{expected} {id}")));
                }
                Some(AtomicDeleteKind::Note { .. }) => {}
                Some(AtomicDeleteKind::Entity { .. }) => {
                    return Err(RuntimeError::NotFound(format!("entity {id}")));
                }
                Some(AtomicDeleteKind::Edge) => {
                    return Err(RuntimeError::NotFound(format!("edge {id}")));
                }
            }
            if let Some(error) = runtime.stream_member_error(&note).await? {
                return Err(error);
            }
            let namespace = note.namespace.clone();
            // Storage parity: `note_soft_delete_statement`/
            // `note_hard_delete_statement` are the SAME khive-db builders
            // khive-db's own `SqlNoteStore::delete_note` calls.
            let mut statements = if hard {
                vec![
                    PlanStatement {
                        statement: delete_record_attachments_statement(
                            id,
                            AttachmentSubstrate::Note,
                        ),
                        guard: None,
                    },
                    PlanStatement {
                        statement: note_hard_delete_statement(id),
                        guard: Some(AffectedRowGuard::exactly(1)),
                    },
                ]
            } else {
                let deleted_at = chrono::Utc::now().timestamp_micros();
                vec![PlanStatement {
                    statement: note_soft_delete_statement(id, deleted_at),
                    guard: Some(AffectedRowGuard::exactly(1)),
                }]
            };
            if hard {
                statements.extend(
                    hard_delete_lineage_warning_statements(
                        &namespace,
                        &actor,
                        id,
                        SubstrateKind::Note,
                    )
                    .into_iter()
                    .map(|statement| PlanStatement {
                        statement,
                        guard: None,
                    }),
                );
                statements.push(PlanStatement {
                    statement: purge_incident_edges_statement(id),
                    guard: None,
                });
            }
            // FTS + vector index purge, matching operations.rs
            // `delete_note`: both soft and hard delete clean indexes (a hard
            // delete of an already-tombstoned record must still purge
            // them); only hard additionally cascades edges above.
            push_index_purge_statements(
                runtime,
                &mut statements,
                "fts_notes",
                &namespace,
                id,
                "atomic-delete-note",
            )
            .await?;
            // operations.rs's `delete_note` appends a `NoteDeleted` event
            // after a successful row delete, on both soft and hard delete:
            // same reasoning as the entity branch above.
            statements.extend(event_append_statements(
                token,
                &namespace,
                "delete",
                EventKind::NoteDeleted,
                SubstrateKind::Note,
                id,
                serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
            )?);
            Ok(AtomicOpPlan::Delete(DeletePlan {
                target_id: id,
                statements,
                // A committed atomic note delete must fire the same
                // pack-installed note-mutation hook `operations.rs::
                // delete_note` fires, so a warm ANN cache sees the deletion
                // even when the mutation went through the atomic-plan path.
                post_commit: PostCommitEffect::NoteDeleted {
                    note_id: id,
                    kind: note.kind.clone(),
                },
            }))
        }
        Some(_) => Err(RuntimeError::InvalidInput(format!(
            "delete target {id} must be an entity, note, or edge"
        ))),
        // `Resolved` has no `Edge` variant (same reasoning as
        // `prepare_update`'s fallback above) — probe the graph store
        // directly.
        None => match &expected_kind {
            Some(AtomicDeleteKind::Entity { .. }) => {
                Err(RuntimeError::NotFound(format!("entity/note {id}")))
            }
            Some(AtomicDeleteKind::Note { .. }) => {
                Err(RuntimeError::NotFound(format!("entity/note {id}")))
            }
            Some(AtomicDeleteKind::Edge) | None => {
                let edge = if hard {
                    runtime.get_edge_including_deleted(token, id).await?
                } else {
                    runtime.get_edge(token, id).await?
                };
                match edge {
                    Some(edge) => prepare_delete_edge(token, id, edge, hard, &actor).await,
                    None => Err(RuntimeError::NotFound(format!("entity/note/edge {id}"))),
                }
            }
        },
    }
}

/// Edge branch of `prepare_delete`. Mirrors
/// `khive-runtime::operations::KhiveRuntime::delete_edge` exactly: hard
/// delete cascades `purge_incident_edges` (any `annotates` edge — or any
/// other edge — pointing AT this edge as a node) BEFORE deleting the edge
/// row itself, then a soft or hard delete statement, then an unconditional
/// `EdgeDeleted` event (edges are never FTS/vector-indexed, so unlike the
/// entity/note branches there is no index purge here — `delete_edge` has
/// none either).
async fn prepare_delete_edge(
    token: &NamespaceToken,
    id: Uuid,
    edge: khive_storage::types::Edge,
    hard: bool,
    actor: &str,
) -> RuntimeResult<AtomicOpPlan> {
    let namespace = edge.namespace.clone();
    let mut statements: Vec<PlanStatement> = Vec::new();

    if hard {
        statements.extend(
            hard_delete_lineage_warning_statements(&namespace, actor, id, SubstrateKind::Entity)
                .into_iter()
                .map(|statement| PlanStatement {
                    statement,
                    guard: None,
                }),
        );
        // Mirrors `delete_edge`'s `graph.purge_incident_edges(edge_id)` —
        // unguarded: zero incident edges is a legitimate outcome, not a
        // failure (same reasoning as the entity/note cascade-edges
        // statements above).
        statements.push(PlanStatement {
            statement: purge_incident_edges_statement(id),
            guard: None,
        });
        statements.push(PlanStatement {
            statement: edge_hard_delete_statement(id),
            guard: Some(AffectedRowGuard::exactly(1)),
        });
    } else {
        let now = chrono::Utc::now().timestamp_micros();
        statements.push(PlanStatement {
            statement: edge_soft_delete_statement(id, now),
            guard: Some(AffectedRowGuard::exactly(1)),
        });
    }

    statements.extend(event_append_statements(
        token,
        &namespace,
        "delete",
        EventKind::EdgeDeleted,
        SubstrateKind::Entity,
        id,
        serde_json::json!({"id": id, "namespace": namespace, "hard": hard}),
    )?);

    Ok(AtomicOpPlan::Delete(DeletePlan {
        target_id: id,
        statements,
        post_commit: PostCommitEffect::None,
    }))
}

// ---------------------------------------------------------------------------
// link
// ---------------------------------------------------------------------------

fn parse_edge_relation(raw: &str) -> RuntimeResult<EdgeRelation> {
    raw.parse::<EdgeRelation>()
        .map_err(|e| RuntimeError::InvalidInput(format!("unknown edge relation {raw:?}: {e}")))
}

async fn prepare_link(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    let source_id = require_uuid(args, "source_id")?;
    let target_id = require_uuid(args, "target_id")?;
    let relation = parse_edge_relation(require_str(args, "relation")?)?;
    let weight = optional_f64(args, "weight")?.unwrap_or(1.0);
    let metadata = obj(args)?.get("metadata").cloned();
    let resurrect = match obj(args)?.get("resurrect") {
        None => false,
        Some(Value::Bool(value)) => *value,
        Some(other) => {
            return Err(RuntimeError::InvalidInput(format!(
                "resurrect must be a boolean, got: {other}"
            )))
        }
    };

    // Top-level `dependency_kind` param merges into `metadata`: only fills
    // the key when metadata doesn't already carry one. Calls the same
    // `khive_runtime::merge_entry_metadata` `khive-pack-kg`'s canonical
    // `handle_link` calls, so both sides depend on one function instead of
    // each maintaining their own copy.
    let mut metadata = crate::merge_entry_metadata(
        metadata,
        optional_str(args, "dependency_kind").map(String::from),
    )?;

    validate_edge_weight(weight)?;
    runtime
        .validate_edge_relation_endpoints(token, source_id, target_id, relation)
        .await?;

    let (canon_source, canon_target) = canonical_edge_endpoints(relation, source_id, target_id);

    // Endpoint-kind `dependency_kind` inference for `depends_on` edges,
    // matching operations.rs `link()`: only applies when both endpoints
    // resolve as entities and the key is still absent after the
    // top-level-param merge above. Runs against the canonical endpoints,
    // mirroring `KhiveRuntime::link`'s own ordering (canonicalize, then
    // infer).
    if relation == EdgeRelation::DependsOn {
        metadata = match (
            runtime.resolve_edge_endpoint(token, canon_source).await?,
            runtime.resolve_edge_endpoint(token, canon_target).await?,
        ) {
            (Some(Resolved::Entity(src_e)), Some(Resolved::Entity(tgt_e))) => {
                merge_dependency_kind(&src_e.kind, &tgt_e.kind, metadata)
            }
            _ => metadata,
        };
    }

    validate_edge_metadata(relation, metadata.as_ref())?;
    let namespace = token.namespace().as_str().to_string();
    let previous = runtime
        .get_edge_by_natural_key_including_deleted(
            token,
            &namespace,
            canon_source,
            canon_target,
            relation,
        )
        .await?;
    if let Some(edge) = previous.as_ref() {
        if edge.deleted_at.is_some() && !resurrect {
            return Err(RuntimeError::InvalidInput(format!(
                "edge natural key is soft-deleted; pass resurrect=true to link explicitly: {}",
                Uuid::from(edge.id)
            )));
        }
    }

    let disposition = match previous.as_ref() {
        None => EdgeUpsertDisposition::Created,
        Some(edge) if edge.deleted_at.is_some() => EdgeUpsertDisposition::Resurrected,
        Some(_) => EdgeUpsertDisposition::Updated,
    };
    let edge_id = previous
        .as_ref()
        .map(|edge| Uuid::from(edge.id))
        .unwrap_or_else(Uuid::new_v4);
    let now = previous.as_ref().map_or_else(
        || chrono::Utc::now().timestamp_micros(),
        |edge| {
            chrono::Utc::now()
                .timestamp_micros()
                .max(edge.updated_at.timestamp_micros().saturating_add(1))
        },
    );
    let metadata_str = metadata
        .as_ref()
        .map(|value| serde_json::to_string(value).unwrap_or_default());

    // The guarded mutation closes both atomic seams: endpoints are re-probed
    // inside the transaction, and the natural-key row must still match the
    // prepare snapshot. That makes the disposition used below truthful.
    let statement = match previous.as_ref() {
        None => edge_insert_new_guarded_by_endpoints_statement(
            &namespace,
            edge_id,
            canon_source,
            canon_target,
            relation,
            weight,
            now,
            metadata_str.as_deref(),
        ),
        Some(edge) => edge_link_replace_if_unchanged_and_endpoints_exist_statement(
            edge,
            weight,
            now,
            metadata_str.as_deref(),
        ),
    };
    let mut statements = vec![PlanStatement {
        statement,
        guard: Some(AffectedRowGuard::exactly(1)),
    }];
    let kind = match disposition {
        EdgeUpsertDisposition::Created => EventKind::LinkCreated,
        EdgeUpsertDisposition::Updated | EdgeUpsertDisposition::Resurrected => {
            EventKind::EdgeUpdated
        }
    };
    statements.extend(event_append_statements(
        token,
        &namespace,
        "link",
        kind,
        SubstrateKind::Entity,
        edge_id,
        serde_json::json!({
            "id": edge_id,
            "namespace": namespace,
            "mutation": disposition.name(),
            "source_id": canon_source,
            "target_id": canon_target,
            "relation": relation,
            "weight": weight,
            "metadata": metadata,
            "previous": previous,
        }),
    )?);

    Ok(AtomicOpPlan::Link(LinkPlan {
        source_id: canon_source,
        target_id: canon_target,
        statements,
        disposition,
    }))
}

// ---------------------------------------------------------------------------
// merge (entity-only)
// ---------------------------------------------------------------------------

// Full atomic-merge parity (field folding, survivor FTS/vector reindex,
// loser index purge, merge provenance, same-kind rejection) is deferred:
// atomic `merge` is rejected entirely at the pre-runtime admissibility
// guard (`khive_types::pack::ATOMIC_KNOWN_UNIMPLEMENTED_VERBS`, alongside
// `propose`/`review`/`withdraw`). This function still produces a plan
// (kept for the existing direct-prepare test coverage below and as
// defense in depth), but the CLI's `--atomic` surface never reaches it,
// since `check_atomic_admissible` rejects `merge` before any runtime is
// built.
async fn prepare_merge(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    args: &Value,
) -> RuntimeResult<AtomicOpPlan> {
    let into_id = require_uuid(args, "into_id")?;
    let from_id = require_uuid(args, "from_id")?;
    if into_id == from_id {
        return Err(RuntimeError::InvalidInput(
            "cannot merge an entity into itself".into(),
        ));
    }

    let entities = runtime.entities(token)?;
    entities
        .get_entity(into_id)
        .await?
        .ok_or_else(|| RuntimeError::NotFound(format!("entity {into_id}")))?;
    entities
        .get_entity(from_id)
        .await?
        .ok_or_else(|| RuntimeError::NotFound(format!("entity {from_id}")))?;

    let now = chrono::Utc::now().timestamp_micros();
    let rewires = vec![
        crate::atomic_plan::PlanPredicate {
            description: "source_id = :from".to_string(),
            statement: SqlStatement {
                sql: "UPDATE graph_edges SET source_id = ?1, updated_at = ?2 WHERE source_id = ?3"
                    .to_string(),
                params: vec![
                    SqlValue::Text(into_id.to_string()),
                    SqlValue::Integer(now),
                    SqlValue::Text(from_id.to_string()),
                ],
                label: Some("atomic-merge-rewire-source".to_string()),
            },
        },
        crate::atomic_plan::PlanPredicate {
            description: "target_id = :from".to_string(),
            statement: SqlStatement {
                sql: "UPDATE graph_edges SET target_id = ?1, updated_at = ?2 WHERE target_id = ?3"
                    .to_string(),
                params: vec![
                    SqlValue::Text(into_id.to_string()),
                    SqlValue::Integer(now),
                    SqlValue::Text(from_id.to_string()),
                ],
                label: Some("atomic-merge-rewire-target".to_string()),
            },
        },
    ];
    let lifecycle = vec![PlanStatement {
        statement: SqlStatement {
            sql: "UPDATE entities SET deleted_at = ?1, merged_into = ?2, version = version + 1 \
                  WHERE id = ?3 AND deleted_at IS NULL"
                .to_string(),
            params: vec![
                SqlValue::Integer(now),
                SqlValue::Text(into_id.to_string()),
                SqlValue::Text(from_id.to_string()),
            ],
            label: Some("atomic-merge-tombstone-from-entity".to_string()),
        },
        guard: Some(AffectedRowGuard::exactly(1)),
    }];

    Ok(AtomicOpPlan::Merge(MergePlan {
        into_id,
        from_id,
        rewires,
        lifecycle,
    }))
}

// ---------------------------------------------------------------------------
// post-commit effects
// ---------------------------------------------------------------------------

/// Embedding metadata produced by one successfully applied reindex effect.
///
/// The effect identity is retained so response builders can attach an advisory
/// to the exact atomic op that scheduled the reindex. Effects whose target is
/// no longer present are omitted from the returned outcome list.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct PostCommitEmbeddingOutcome {
    /// The committed reindex effect that produced this outcome.
    pub effect: PostCommitEffect,
    /// Actual input bounding observed while executing the effect.
    pub truncation: crate::retrieval::EmbeddingTruncationReport,
}

/// Run every deferred [`PostCommitEffect`] after a committed atomic unit.
pub async fn apply_post_commit_effects(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    effects: CommittedPostCommitEffects,
) -> RuntimeResult<()> {
    apply_post_commit_effects_with_report(runtime, token, effects)
        .await
        .map(|_| ())
}

/// Truncation-reporting form of [`apply_post_commit_effects`]. Re-fetches each
/// target's now-committed row outside any transaction and reuses the existing
/// `reindex_entity`/`reindex_note` (FTS + embedding, same as the non-atomic
/// path) for exact parity. Returns the typed embedding outcome for each reindex
/// effect so callers can preserve write-response advisories instead of
/// discarding them after commit.
pub async fn apply_post_commit_effects_with_report(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    effects: CommittedPostCommitEffects,
) -> RuntimeResult<Vec<PostCommitEmbeddingOutcome>> {
    let mut embedding_outcomes = Vec::new();
    let mut failures = Vec::new();
    for (index, effect) in effects.into_effects().into_iter().enumerate() {
        let identity = format!("{effect:?}");
        match apply_one_post_commit_effect(runtime, token, effect).await {
            Ok(Some(outcome)) => embedding_outcomes.push(outcome),
            Ok(None) => {}
            Err(error) => failures.push(format!("effect[{index}] {identity}: {error}")),
        }
    }
    if failures.is_empty() {
        Ok(embedding_outcomes)
    } else {
        Err(RuntimeError::Internal(format!(
            "post-commit effects failed after commit: {}",
            failures.join("; ")
        )))
    }
}

async fn apply_one_post_commit_effect(
    runtime: &KhiveRuntime,
    token: &NamespaceToken,
    effect: PostCommitEffect,
) -> RuntimeResult<Option<PostCommitEmbeddingOutcome>> {
    match effect {
        PostCommitEffect::None => Ok(None),
        PostCommitEffect::NoteChanged { note_id, kind } => {
            runtime.fire_note_mutation_hook(&kind, note_id).await;
            Ok(None)
        }
        PostCommitEffect::ReindexEntity { entity_id } => {
            let Some(entity) = runtime.entities(token)?.get_entity(entity_id).await? else {
                return Ok(None);
            };
            let truncation = runtime.reindex_entity(token, &entity).await?;
            Ok(Some(PostCommitEmbeddingOutcome {
                effect: PostCommitEffect::ReindexEntity { entity_id },
                truncation,
            }))
        }
        PostCommitEffect::ReindexNote { note_id, version } => {
            let Some(note) = runtime.notes(token)?.get_note(note_id).await? else {
                return Ok(None);
            };
            if note.version != version {
                return Ok(None);
            }
            let truncation = runtime.reindex_note(token, &note).await?;
            if runtime
                .notes(token)?
                .get_note(note_id)
                .await?
                .is_none_or(|current| current.version != version)
            {
                return Ok(None);
            }
            // Atomic note updates bypass the regular update_note hook. Notify
            // in-process consumers only after the current version was indexed.
            runtime.fire_note_mutation_hook(&note.kind, note.id).await;
            Ok(Some(PostCommitEmbeddingOutcome {
                effect: PostCommitEffect::ReindexNote { note_id, version },
                truncation,
            }))
        }
        PostCommitEffect::NoteDeleted { note_id, kind } => {
            // The committed row may already be gone; use the captured kind.
            runtime.fire_note_mutation_hook(&kind, note_id).await;
            Ok(None)
        }
        PostCommitEffect::GtdAudit { .. } => {
            // The kkernel caller owns the GTD pack's separate audit side write.
            Ok(None)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use async_trait::async_trait;
    use lattice_embed::{EmbedError, EmbeddingModel, EmbeddingService, MAX_TEXT_BYTES};
    use serde_json::json;

    use khive_types::Namespace;

    use crate::embedder_registry::EmbedderProvider;
    use crate::runtime::RuntimeConfig;

    /// Owns a file-backed runtime and removes its database directory after shutdown.
    struct TestRuntime {
        runtime: KhiveRuntime,
        _temp_dir: tempfile::TempDir,
    }

    impl std::ops::Deref for TestRuntime {
        type Target = KhiveRuntime;

        fn deref(&self) -> &Self::Target {
            &self.runtime
        }
    }

    const STUB_MODEL: &str = "stub-adr099-b3";
    const STUB_DIMS: usize = 4;

    struct StubService;

    #[async_trait]
    impl EmbeddingService for StubService {
        async fn embed(
            &self,
            texts: &[String],
            _model: EmbeddingModel,
        ) -> Result<Vec<Vec<f32>>, EmbedError> {
            Ok(texts.iter().map(|_| vec![0.5_f32; STUB_DIMS]).collect())
        }

        fn supports_model(&self, _model: EmbeddingModel) -> bool {
            true
        }

        fn name(&self) -> &'static str {
            STUB_MODEL
        }
    }

    struct StubProvider;

    #[async_trait]
    impl EmbedderProvider for StubProvider {
        fn name(&self) -> &str {
            STUB_MODEL
        }

        fn dimensions(&self) -> usize {
            STUB_DIMS
        }

        async fn build(&self) -> RuntimeResult<std::sync::Arc<dyn EmbeddingService>> {
            Ok(std::sync::Arc::new(StubService))
        }
    }

    fn scratch_runtime() -> TestRuntime {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("atomic_prepare_reindex.db");
        let runtime = KhiveRuntime::new_for_test(RuntimeConfig {
            db_path: Some(path),
            embedding_model: None,
            additional_embedding_models: vec![],
            ..RuntimeConfig::default()
        })
        .expect("runtime");
        TestRuntime {
            runtime,
            _temp_dir: dir,
        }
    }

    /// Atomic `update` must reject a field that does not apply to the
    /// resolved substrate: parity with
    /// `khive-pack-kg::handlers::update::reject_inapplicable_fields`.
    /// Without this check, atomic prepare would silently ignore `salience`
    /// on an entity: it would set every entity field to its current value,
    /// bump `updated_at`, satisfy the `exactly(1)` guard, and commit: a
    /// spurious no-op reported as success.
    #[tokio::test]
    async fn atomic_update_entity_rejects_note_only_field_salience() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entity = khive_storage::Entity::new("local", "concept", "GapFourEntity");
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");

        let err = prepare_update(
            &runtime,
            &token,
            &json!({"id": entity_id.to_string(), "salience": 0.9}),
            None,
        )
        .await
        .expect_err("salience on an entity must be rejected, not silently accepted");
        assert!(
            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("salience") && msg.contains("not valid for an entity")),
            "expected an InvalidInput naming the offending field, got: {err:?}"
        );

        // A valid entity update (name/description/tags) must still work.
        let plan = prepare_update(
            &runtime,
            &token,
            &json!({
                "id": entity_id.to_string(),
                "name": "GapFourEntity-renamed",
                "description": "updated description",
                "tags": ["a", "b"],
            }),
            None,
        )
        .await
        .expect("a valid entity field set must still be accepted");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));
        let entity = runtime
            .get_entity(&token, entity_id)
            .await
            .expect("get_entity");
        assert_eq!(entity.name, "GapFourEntity-renamed");
    }

    /// The `--atomic` seam intentionally requires a full UUID (never a short
    /// hex prefix) for its `id` fields — prefix resolution already happened
    /// upstream, at the kkernel CLI boundary that has namespace context
    /// (`resolve_kg_ids_in_args`). The rejection message must say *why* a
    /// prefix cannot be resolved at this stage, not just restate that a full
    /// UUID is required.
    #[tokio::test]
    async fn atomic_update_short_prefix_id_rejected_with_namespace_explanation() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_update(
            &runtime,
            &token,
            &json!({"id": "deadbeef", "name": "whatever"}),
            None,
        )
        .await
        .expect_err("a short hex prefix must be rejected by the atomic-plan seam");
        let msg = match err {
            RuntimeError::InvalidInput(ref msg) => msg.clone(),
            other => panic!("expected InvalidInput, got: {other:?}"),
        };
        assert!(
            msg.contains("full UUID"),
            "message must still state the rule; got: {msg}"
        );
        assert!(
            msg.to_ascii_lowercase().contains("namespace"),
            "message must explain the namespace-scoping consequence, not just restate the \
             rule; got: {msg}"
        );
    }

    #[tokio::test]
    async fn atomic_update_entity_type_persists_patch_and_schedules_reindex() {
        let runtime = scratch_runtime();
        runtime.install_entity_type_validator(std::sync::Arc::new(|kind, entity_type| {
            let Some(raw) = entity_type else {
                return Ok(None);
            };
            let normalized = raw.trim().to_ascii_lowercase();
            if kind == "concept" && normalized == "algorithm" {
                Ok(Some(normalized))
            } else {
                Err(RuntimeError::InvalidInput(format!(
                    "unknown entity_type {raw:?} for {kind:?}; valid: algorithm"
                )))
            }
        }));
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut entity = khive_storage::Entity::new("local", "concept", "AtomicHistorical");
        entity.description = Some("keep description".to_string());
        entity.properties = Some(json!({"type": "algorithm", "keep": true}));
        entity.tags = vec!["keep-tag".to_string()];
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": entity_id.to_string(), "entity_type": " Algorithm "}),
            None,
        )
        .await
        .expect("atomic prepare must accept a registered entity_type");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("atomic update must run");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[PostCommitEffect::ReindexEntity { entity_id }],
            "entity_type patch must schedule the normal entity reindex path"
        );

        let updated = runtime
            .get_entity(&token, entity_id)
            .await
            .expect("read updated entity");
        assert_eq!(updated.entity_type.as_deref(), Some("algorithm"));
        assert_eq!(updated.name, "AtomicHistorical");
        assert_eq!(updated.description.as_deref(), Some("keep description"));
        assert_eq!(
            updated.properties,
            Some(json!({"type": "algorithm", "keep": true}))
        );
        assert_eq!(updated.tags, vec!["keep-tag"]);
    }

    /// Symmetric note-substrate case: `description` is entity-only and
    /// must be rejected the same way update.rs rejects it.
    #[tokio::test]
    async fn atomic_update_note_rejects_entity_only_field_description() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut note = khive_storage::note::Note::new("local", "observation", "gap-4 note content");
        note.name = Some("gap-four-note".to_string());
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");

        let err = prepare_update(
            &runtime,
            &token,
            &json!({"id": note_id.to_string(), "description": "entities have descriptions, notes don't"}),
                None,
            )
        .await
        .expect_err("description on a note must be rejected, not silently accepted");
        assert!(
            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("description") && msg.contains("not valid for a note")),
            "expected an InvalidInput naming the offending field, got: {err:?}"
        );

        // A valid note update (content) must still work.
        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": note_id.to_string(), "content": "gap-4 note content, revised"}),
            None,
        )
        .await
        .expect("a valid note field must still be accepted");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));
    }

    #[tokio::test]
    async fn atomic_update_note_tags_replace_preserve_clear_and_override_nested_tags() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut note = khive_storage::note::Note::new("local", "observation", "tagged note");
        note.properties = Some(json!({"tags": ["old"], "keep": {"value": 1}}));
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");

        for (mut args, expected_tags) in [
            (
                json!({"tags": ["new", "shared"], "properties": {"tags": ["nested"], "added": true}}),
                json!(["new", "shared"]),
            ),
            (
                json!({"name": "renamed note", "properties": {"omitted": true}}),
                json!(["new", "shared"]),
            ),
            (
                json!({"tags": null, "properties": null}),
                json!(["new", "shared"]),
            ),
            (
                json!({"tags": [], "properties": {"tags": ["nested-after-clear"]}}),
                json!([]),
            ),
            (
                json!({"tags": ["after-null-properties"], "properties": null}),
                json!(["after-null-properties"]),
            ),
        ] {
            args["id"] = json!(note_id.to_string());
            let original_args = args.clone();
            let plan = prepare_update(&runtime, &token, &args, None)
                .await
                .expect("valid atomic note tags patch");
            assert_eq!(
                args, original_args,
                "preparation must not mutate caller args"
            );
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("atomic note update");
            assert!(matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ));
            let updated = runtime
                .notes(&token)
                .expect("notes store")
                .get_note(note_id)
                .await
                .expect("read note")
                .expect("note exists");
            let properties = updated.properties.expect("note properties");
            assert_eq!(properties["tags"], expected_tags);
            assert_eq!(properties["keep"], json!({"value": 1}));
            assert_eq!(properties["added"], json!(true));
            assert_eq!(updated.content, "tagged note");
            if original_args.get("name").is_some() {
                assert_eq!(updated.name.as_deref(), Some("renamed note"));
                assert_eq!(properties["omitted"], json!(true));
            }
        }
    }

    #[tokio::test]
    async fn atomic_update_entity_tags_keep_replace_preserve_and_clear_semantics() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut entity = khive_storage::Entity::new("local", "concept", "tagged entity");
        entity.tags = vec!["old".to_string()];
        entity.properties = Some(json!({"keep": true}));
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");
        for (mut args, expected_tags) in [
            (json!({"tags": ["new", "shared"]}), json!(["new", "shared"])),
            (json!({"name": "renamed entity"}), json!(["new", "shared"])),
            (json!({"tags": null}), json!(["new", "shared"])),
            (json!({"tags": []}), json!([])),
        ] {
            args["id"] = json!(entity_id.to_string());
            let plan = prepare_update(&runtime, &token, &args, None)
                .await
                .expect("valid atomic entity tags patch");
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("atomic entity update");
            assert!(matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ));
            let updated = runtime
                .get_entity(&token, entity_id)
                .await
                .expect("read entity");
            assert_eq!(json!(updated.tags), expected_tags);
            assert_eq!(updated.properties, Some(json!({"keep": true})));
        }
    }

    #[tokio::test]
    async fn atomic_update_note_invalid_tags_leave_snapshot_unchanged() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut note = khive_storage::note::Note::new("local", "observation", "unchanged note");
        note.properties = Some(json!({"tags": ["keep"], "other": true}));
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");
        let before = runtime
            .notes(&token)
            .expect("notes store")
            .get_note(note_id)
            .await
            .expect("read note")
            .expect("note exists");

        for (mut args, expected_error) in [
            (
                json!({"tags": "invalid"}),
                "tags must be an array of strings",
            ),
            (
                json!({"tags": ["valid", 1]}),
                "tags must be an array of strings",
            ),
            (
                json!({"tags": {"nested": true}}),
                "tags must be an array of strings",
            ),
            (
                json!({"tags": ["valid"], "properties": []}),
                "properties must be an object",
            ),
            (
                json!({"tags": [], "properties": "invalid"}),
                "properties must be an object",
            ),
            (
                json!({"tags": "invalid", "description": "entity field"}),
                "field 'description' is not valid for a note",
            ),
        ] {
            args["id"] = json!(note_id.to_string());
            args["content"] = json!("must not persist");
            let error = prepare_update(&runtime, &token, &args, None)
                .await
                .expect_err("invalid tags patch must not produce a plan");
            assert!(
                matches!(error, RuntimeError::InvalidInput(ref message) if message.contains(expected_error)),
                "unexpected error: {error:?}"
            );
            let after = runtime
                .notes(&token)
                .expect("notes store")
                .get_note(note_id)
                .await
                .expect("read note")
                .expect("note exists");
            assert_eq!(after, before);
        }
    }

    /// Updating a note's content inside an atomic unit must, after commit,
    /// leave the note recallable via FTS under its new content and its
    /// vector row refreshed: parity with the non-atomic
    /// `update_note` -> `reindex_note` path.
    #[tokio::test]
    async fn atomic_update_note_content_is_fts_and_vector_reindexed_post_commit() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let mut note = khive_storage::note::Note::new("local", "observation", "original content");
        note.name = Some("reindex-target".to_string());
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");

        // Sanity: no vector row yet for the stub model.
        let vec_store = runtime
            .vectors_for_model(&token, STUB_MODEL)
            .expect("vec store");
        assert_eq!(vec_store.count().await.expect("count before"), 0);

        let updated_content = format!("freshly-updated-content-xyz{}", "x".repeat(MAX_TEXT_BYTES));
        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": note_id.to_string(), "content": updated_content, "embed": true}),
            None,
        )
        .await
        .expect("prepare update");

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[PostCommitEffect::ReindexNote {
                note_id,
                version: 2
            }],
            "content change must schedule exactly one ReindexNote post-commit effect"
        );

        let embedding_outcomes =
            apply_post_commit_effects_with_report(&runtime, &token, post_commit)
                .await
                .expect("apply post-commit effects");
        assert_eq!(embedding_outcomes.len(), 1);
        assert_eq!(
            embedding_outcomes[0].effect,
            PostCommitEffect::ReindexNote {
                note_id,
                version: 2
            }
        );
        assert_eq!(embedding_outcomes[0].truncation.truncated, 1);
        assert!(embedding_outcomes[0].truncation.discarded_bytes > 0);

        // FTS: the note must be recallable under its NEW content.
        let doc = runtime
            .text_for_notes(&token)
            .expect("text store")
            .get_document("local", note_id)
            .await
            .expect("get_document")
            .expect("document must be indexed after post-commit reindex");
        assert!(
            doc.body.contains("freshly-updated-content-xyz"),
            "FTS body must reflect the committed content: {:?}",
            doc.body
        );

        // Vector: a row must now exist for the registered stub model.
        assert_eq!(
            vec_store.count().await.expect("count after"),
            1,
            "post-commit reindex must have inserted a vector row for the stub model"
        );
    }

    /// The atomic-plan path must fire the pack-installed note-mutation hook
    /// for both an atomic note UPDATE (`PostCommitEffect::ReindexNote`'s
    /// handler fires it after its own reindex, mirroring `update_note()`
    /// on the non-atomic path) and an atomic note DELETE (`DeletePlan`
    /// carries a `PostCommitEffect::NoteDeleted` that
    /// `apply_post_commit_effects` dispatches directly, mirroring
    /// `operations.rs::delete_note`'s direct-fire, no-refetch shape: the
    /// row may already be gone by the time this runs, for a hard delete).
    /// A minimal counting hook proves both fire; no `khive-pack-memory`
    /// dependency is needed at this layer, since the hook itself is
    /// generic.
    #[tokio::test]
    async fn atomic_note_update_and_delete_post_commit_effects_execute_exactly_once() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let fired: std::sync::Arc<std::sync::Mutex<Vec<(String, uuid::Uuid)>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let fired_for_hook = fired.clone();
        runtime.install_note_mutation_hook(std::sync::Arc::new(
            move |kind: String, id: uuid::Uuid| {
                let fired = fired_for_hook.clone();
                Box::pin(async move {
                    fired.lock().expect("lock").push((kind, id));
                })
            },
        ));

        // Update path.
        let mut note = khive_storage::note::Note::new("local", "observation", "hook-update-target");
        note.name = Some("hook-update-target".to_string());
        let update_note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed update-target note");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": update_note_id.to_string(), "content": "hook-update-target, revised"}),
            None,
        )
        .await
        .expect("prepare update");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        apply_post_commit_effects(&runtime, &token, post_commit)
            .await
            .expect("apply post-commit effects (update)");

        // Delete path (soft delete: the row still exists, but the hook
        // fires directly from the captured kind rather than refetching).
        let mut del_note =
            khive_storage::note::Note::new("local", "observation", "hook-delete-target");
        del_note.name = Some("hook-delete-target".to_string());
        let delete_note_id = del_note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(del_note)
            .await
            .expect("seed delete-target note");

        let plan = prepare_delete(
            &runtime,
            &token,
            &json!({"id": delete_note_id.to_string(), "hard": false}),
            None,
        )
        .await
        .expect("prepare delete");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[PostCommitEffect::NoteDeleted {
                note_id: delete_note_id,
                kind: "observation".to_string(),
            }],
            "a committed note delete must schedule exactly one NoteDeleted post-commit effect"
        );
        apply_post_commit_effects(&runtime, &token, post_commit)
            .await
            .expect("apply post-commit effects (delete)");

        assert_eq!(
            *fired.lock().expect("lock"),
            vec![
                ("observation".to_string(), update_note_id),
                ("observation".to_string(), delete_note_id),
            ],
            "each committed token must execute its note-mutation effect exactly once"
        );
    }

    #[tokio::test]
    async fn failed_post_commit_reindexes_do_not_skip_later_note_mutation_hook() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let fired: std::sync::Arc<std::sync::Mutex<Vec<uuid::Uuid>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let fired_for_hook = fired.clone();
        runtime.install_note_mutation_hook(std::sync::Arc::new(
            move |_kind: String, id: uuid::Uuid| {
                let fired = fired_for_hook.clone();
                Box::pin(async move {
                    fired.lock().expect("lock").push(id);
                })
            },
        ));

        let mut plans = Vec::new();
        let mut failed_ids = Vec::new();
        for name in ["first", "second"] {
            let note = khive_storage::note::Note::new("local", "observation", name);
            let id = note.id;
            runtime
                .notes(&token)
                .expect("notes store")
                .upsert_note(note)
                .await
                .expect("seed reindex target");
            plans.push(
                prepare_update(
                    &runtime,
                    &token,
                    // Explicit embedding is required here: with embed=None
                    // and no existing vector row, atomic_runner legitimately
                    // replaces ReindexNote with NoteChanged (inheritance
                    // policy), leaving no reindex for the fault to fail.
                    &json!({"id": id.to_string(), "content": format!("{name} revised"), "embed": true}),
                    None,
                )
                .await
                .expect("prepare note update"),
            );
            failed_ids.push(id);
        }
        let deleted = khive_storage::note::Note::new("local", "observation", "deleted");
        let deleted_id = deleted.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(deleted)
            .await
            .expect("seed delete target");
        plans.push(
            prepare_delete(
                &runtime,
                &token,
                &json!({"id": deleted_id.to_string(), "hard": false}),
                None,
            )
            .await
            .expect("prepare note delete"),
        );

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), plans)
            .await
            .expect("commit atomic unit");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[
                PostCommitEffect::ReindexNote {
                    note_id: failed_ids[0],
                    version: 2,
                },
                PostCommitEffect::ReindexNote {
                    note_id: failed_ids[1],
                    version: 2,
                },
                PostCommitEffect::NoteDeleted {
                    note_id: deleted_id,
                    kind: "observation".into(),
                },
            ],
            "the fixture must commit two reindexes before the deletion hook"
        );

        // The already-committed note rows remain intact, but both deferred
        // reindexes now fail. The following deletion hook must still fire.
        let mut writer = runtime.sql().writer().await.expect("writer");
        writer
            .execute(SqlStatement {
                sql: "DROP TABLE fts_notes".into(),
                params: Vec::new(),
                label: Some("test_remove_fts_notes_after_commit".into()),
            })
            .await
            .expect("remove FTS table");
        drop(writer);

        let error = apply_post_commit_effects_with_report(&runtime, &token, post_commit)
            .await
            .expect_err("both reindexes must fail")
            .to_string();
        assert!(error.contains("effect[0]"), "{error}");
        assert!(error.contains("effect[1]"), "{error}");
        for id in failed_ids {
            assert!(error.contains(&id.to_string()), "{error}");
        }
        assert_eq!(*fired.lock().expect("lock"), vec![deleted_id]);
    }

    /// Atomic delete must purge the note's FTS row and vector row for both
    /// soft and hard delete: parity with `KhiveRuntime::delete_note`'s
    /// index-cleanup contract.
    #[tokio::test]
    async fn atomic_delete_note_purges_fts_and_vector_indexes_soft_and_hard() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        for hard in [false, true] {
            let mut note =
                khive_storage::note::Note::new("local", "observation", "purge-target content");
            note.name = Some(format!("purge-target-hard-{hard}"));
            let note_id = note.id;
            runtime
                .notes(&token)
                .expect("notes store")
                .upsert_note(note.clone())
                .await
                .expect("seed note");
            runtime
                .reindex_note(&token, &note)
                .await
                .expect("seed index rows");

            let vec_store = runtime
                .vectors_for_model(&token, STUB_MODEL)
                .expect("vec store");
            assert_eq!(
                vec_store.count().await.expect("count before"),
                1,
                "seeded note must have a vector row before delete (hard={hard})"
            );
            assert!(
                runtime
                    .text_for_notes(&token)
                    .expect("text store")
                    .get_document("local", note_id)
                    .await
                    .expect("get_document")
                    .is_some(),
                "seeded note must have an FTS row before delete (hard={hard})"
            );

            let plan = prepare_delete(
                &runtime,
                &token,
                &json!({"id": note_id.to_string(), "hard": hard}),
                None,
            )
            .await
            .expect("prepare delete");
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("seam call ok");
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected commit (hard={hard}): {outcome:?}"
            );

            assert!(
                runtime
                    .text_for_notes(&token)
                    .expect("text store")
                    .get_document("local", note_id)
                    .await
                    .expect("get_document")
                    .is_none(),
                "FTS row must be purged after atomic delete (hard={hard})"
            );
            assert_eq!(
                vec_store.count().await.expect("count after"),
                0,
                "vector row must be purged after atomic delete (hard={hard})"
            );
        }
    }

    /// Atomic delete must purge the entity's FTS row and vector row for
    /// both soft and hard delete: parity with
    /// `KhiveRuntime::delete_entity`'s index-cleanup contract.
    #[tokio::test]
    async fn atomic_delete_entity_purges_fts_and_vector_indexes_soft_and_hard() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        for hard in [false, true] {
            let entity =
                khive_storage::Entity::new("local", "concept", format!("purge-target-hard-{hard}"));
            let entity_id = entity.id;
            runtime
                .entities(&token)
                .expect("entities store")
                .upsert_entity(entity.clone())
                .await
                .expect("seed entity");
            runtime
                .reindex_entity(&token, &entity)
                .await
                .expect("seed index rows");

            let vec_store = runtime
                .vectors_for_model(&token, STUB_MODEL)
                .expect("vec store");
            assert_eq!(
                vec_store.count().await.expect("count before"),
                1,
                "seeded entity must have a vector row before delete (hard={hard})"
            );
            assert!(
                runtime
                    .text(&token)
                    .expect("text store")
                    .get_document("local", entity_id)
                    .await
                    .expect("get_document")
                    .is_some(),
                "seeded entity must have an FTS row before delete (hard={hard})"
            );

            let plan = prepare_delete(
                &runtime,
                &token,
                &json!({"id": entity_id.to_string(), "hard": hard}),
                None,
            )
            .await
            .expect("prepare delete");
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("seam call ok");
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected commit (hard={hard}): {outcome:?}"
            );

            assert!(
                runtime
                    .text(&token)
                    .expect("text store")
                    .get_document("local", entity_id)
                    .await
                    .expect("get_document")
                    .is_none(),
                "FTS row must be purged after atomic delete (hard={hard})"
            );
            assert_eq!(
                vec_store.count().await.expect("count after"),
                0,
                "vector row must be purged after atomic delete (hard={hard})"
            );
        }
    }

    #[tokio::test]
    async fn atomic_delete_entity_and_note_logs_vector_delete_rows() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let entity = khive_storage::Entity::new("local", "concept", "ann-delete-entity");
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity.clone())
            .await
            .expect("seed entity");
        runtime
            .reindex_entity(&token, &entity)
            .await
            .expect("seed entity vector row");
        let note = khive_storage::note::Note::new("local", "observation", "ann-delete-note");
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note.clone())
            .await
            .expect("seed note");
        runtime
            .reindex_note(&token, &note)
            .await
            .expect("seed note vector row");

        for id in [entity_id, note_id] {
            let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
                .await
                .expect("prepare delete");
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("seam call ok");
            assert!(matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ));
        }

        let mut reader = runtime.sql().reader().await.expect("sql reader");
        let count = reader
            .query_scalar(SqlStatement {
                sql: "SELECT COUNT(*) FROM ann_write_log \
                      WHERE namespace = 'local' AND embedding_model = ?1 AND op = 'delete' \
                      AND ((kind = 'entity' AND field = 'entity.body' AND subject_id = ?2) \
                        OR (kind = 'note' AND field = 'note.content' AND subject_id = ?3))"
                    .to_string(),
                params: vec![
                    SqlValue::Text(STUB_MODEL.to_string()),
                    SqlValue::Text(entity_id.to_string()),
                    SqlValue::Text(note_id.to_string()),
                ],
                label: Some("test-atomic-delete-ann-write-log".to_string()),
            })
            .await
            .expect("query ANN write log");
        assert!(matches!(count, Some(SqlValue::Integer(2))));
    }

    /// Atomic prepare validates the requested orientation before canonicalizing
    /// a symmetric edge for persistence. Fixed UUIDs force target < source so a
    /// regression would report the reverse ordered pair.
    #[tokio::test]
    async fn atomic_link_symmetric_rejection_preserves_requested_pair() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let concept_id =
            Uuid::parse_str("ffffffff-ffff-ffff-ffff-ffffffffffff").expect("high UUID");
        let project_id = Uuid::nil();
        assert!(project_id < concept_id, "test must exercise UUID reversal");

        let mut concept = khive_storage::Entity::new("local", "concept", "Concept source");
        concept.id = concept_id;
        let mut project = khive_storage::Entity::new("local", "project", "Project target");
        project.id = project_id;
        entities.upsert_entity(concept).await.expect("seed concept");
        entities.upsert_entity(project).await.expect("seed project");

        let error = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": concept_id.to_string(),
                "target_id": project_id.to_string(),
                "relation": "competes_with",
            }),
        )
        .await
        .expect_err("atomic link must reject concept competes_with project");
        let message = error.to_string();
        assert!(
            message.contains(
                "currently legal relations for concept -> project under the loaded endpoint rules: none"
            ),
            "atomic validation must diagnose caller order before persistence canonicalization; got: {message}"
        );
        assert!(
            !message.contains("currently legal relations for project -> concept"),
            "atomic validation must not diagnose the UUID-canonical reverse pair; got: {message}"
        );
    }

    /// Atomic link must persist an explicit top-level `dependency_kind`
    /// param into edge metadata, and must infer one for `depends_on` edges
    /// when absent: parity with
    /// `link.rs`'s `merge_entry_metadata` and `operations.rs`'s
    /// `infer_dependency_kind` table.
    #[tokio::test]
    async fn atomic_link_persists_explicit_dependency_kind_and_infers_when_absent() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");

        fn metadata_json(plan: &AtomicOpPlan) -> String {
            let link_plan = match plan {
                AtomicOpPlan::Link(p) => p,
                other => panic!("expected an AtomicOpPlan::Link, got {other:?}"),
            };
            match link_plan.statements[0].statement.params.last() {
                Some(SqlValue::Text(s)) => s.clone(),
                other => panic!("expected the metadata param to be SqlValue::Text, got {other:?}"),
            }
        }

        // (a) explicit top-level `dependency_kind` param persists in metadata.
        {
            let svc = khive_storage::Entity::new("local", "service", "SvcA");
            let proj = khive_storage::Entity::new("local", "project", "ProjB");
            let (svc_id, proj_id) = (svc.id, proj.id);
            entities.upsert_entity(svc).await.expect("seed svc");
            entities.upsert_entity(proj).await.expect("seed proj");

            let plan = prepare_link(
                &runtime,
                &token,
                &json!({
                    "source_id": svc_id.to_string(),
                    "target_id": proj_id.to_string(),
                    "relation": "depends_on",
                    "dependency_kind": "artifact",
                }),
            )
            .await
            .expect("prepare link");
            let json_str = metadata_json(&plan);
            assert!(
                json_str.contains(r#""dependency_kind":"artifact""#),
                "explicit dependency_kind param must persist: {json_str}"
            );
        }

        // (b) `depends_on` with no explicit dependency_kind infers from
        // endpoint kinds: (service, service) -> "runtime".
        {
            let svc_a = khive_storage::Entity::new("local", "service", "SvcC");
            let svc_b = khive_storage::Entity::new("local", "service", "SvcD");
            let (a_id, b_id) = (svc_a.id, svc_b.id);
            entities.upsert_entity(svc_a).await.expect("seed svc a");
            entities.upsert_entity(svc_b).await.expect("seed svc b");

            let plan = prepare_link(
                &runtime,
                &token,
                &json!({
                    "source_id": a_id.to_string(),
                    "target_id": b_id.to_string(),
                    "relation": "depends_on",
                }),
            )
            .await
            .expect("prepare link");
            let json_str = metadata_json(&plan);
            assert!(
                json_str.contains(r#""dependency_kind":"runtime""#),
                "inferred dependency_kind for (service, service) must persist: {json_str}"
            );
        }
    }

    /// Raw natural-key probe of `graph_edges` (namespace, source_id,
    /// target_id, relation) — returns `(weight, metadata_json, deleted_at)`
    /// for exactly the ONE row a UNIQUE(namespace, source_id, target_id,
    /// relation) constraint permits. `None` means no row at all.
    async fn probe_edge_natural_key(
        runtime: &KhiveRuntime,
        namespace: &str,
        source_id: Uuid,
        target_id: Uuid,
        relation: &str,
    ) -> (usize, Option<f64>, Option<String>, Option<i64>) {
        let mut reader = runtime.sql().reader().await.expect("reader");
        let rows = reader
            .query_all(SqlStatement {
                sql: "SELECT weight, metadata, deleted_at FROM graph_edges \
                      WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 AND relation = ?4"
                    .to_string(),
                params: vec![
                    SqlValue::Text(namespace.to_string()),
                    SqlValue::Text(source_id.to_string()),
                    SqlValue::Text(target_id.to_string()),
                    SqlValue::Text(relation.to_string()),
                ],
                label: Some("test-probe-edge-natural-key".to_string()),
            })
            .await
            .expect("probe edge natural key");
        let count = rows.len();
        let Some(row) = rows.into_iter().next() else {
            return (count, None, None, None);
        };
        let weight = match row.get("weight") {
            Some(SqlValue::Float(f)) => Some(*f),
            Some(SqlValue::Integer(i)) => Some(*i as f64),
            _ => None,
        };
        let metadata = match row.get("metadata") {
            Some(SqlValue::Text(s)) => Some(s.clone()),
            _ => None,
        };
        let deleted_at = match row.get("deleted_at") {
            Some(SqlValue::Integer(i)) => Some(*i),
            _ => None,
        };
        (count, weight, metadata, deleted_at)
    }

    /// Atomic `link` must be an upsert, exactly like canonical `link` ->
    /// `upsert_edge`'s natural-key `ON CONFLICT` arm: re-linking an
    /// already-linked triple must succeed and update weight/metadata, not
    /// hit the `UNIQUE(namespace, source_id, target_id, relation)`
    /// constraint and roll back the whole atomic unit.
    #[tokio::test]
    async fn atomic_link_of_already_linked_triple_upserts_weight_and_metadata() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");

        let a = khive_storage::Entity::new("local", "concept", "GapTwoA");
        let b = khive_storage::Entity::new("local", "concept", "GapTwoB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        // (a) a fresh link still works.
        let plan1 = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": a_id.to_string(),
                "target_id": b_id.to_string(),
                "relation": "extends",
                "weight": 0.5,
            }),
        )
        .await
        .expect("prepare first link");
        let outcome1 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan1])
            .await
            .expect("seam call ok");
        assert!(
            matches!(
                outcome1,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "fresh link must commit: {outcome1:?}"
        );
        let (count, weight, _metadata, deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert_eq!(count, 1, "exactly one edge row after the fresh link");
        assert_eq!(weight, Some(0.5));
        assert!(deleted_at.is_none());

        // (b) re-linking the SAME triple with a different weight/metadata
        // must SUCCEED (not a constraint-violation rollback) and UPDATE the
        // existing row in place — natural key stays unique.
        let plan2 = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": a_id.to_string(),
                "target_id": b_id.to_string(),
                "relation": "extends",
                "weight": 0.9,
                "metadata": {"note": "relinked"},
            }),
        )
        .await
        .expect("prepare second link");
        let outcome2 = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan2])
            .await
            .expect("seam call ok");
        assert!(
            matches!(
                outcome2,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "re-link of an already-linked triple must upsert, not roll back: {outcome2:?}"
        );
        let (count, weight, metadata, deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert_eq!(
            count, 1,
            "the natural-key UNIQUE constraint must still hold exactly one row (upsert, not a second insert)"
        );
        assert_eq!(weight, Some(0.9), "weight must be updated to the new value");
        assert!(
            metadata
                .as_deref()
                .is_some_and(|m| m.contains(r#""note":"relinked""#)),
            "metadata must be updated to the new value: {metadata:?}"
        );
        assert!(deleted_at.is_none());
    }

    /// Atomic `link` refuses a soft-deleted triple by default and only
    /// resurrects it when the caller opts in explicitly.
    #[tokio::test]
    async fn atomic_link_of_soft_deleted_triple_resurrects_it() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");

        let a = khive_storage::Entity::new("local", "concept", "GapTwoResurrectA");
        let b = khive_storage::Entity::new("local", "concept", "GapTwoResurrectB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let plan = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": a_id.to_string(),
                "target_id": b_id.to_string(),
                "relation": "extends",
            }),
        )
        .await
        .expect("prepare link");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));

        // Soft-delete the edge row directly (natural-key UPDATE — mirrors
        // what `delete_edge(hard=false)` does to this same row).
        {
            let mut writer = runtime.sql().writer().await.expect("writer");
            let affected = writer
                .execute(SqlStatement {
                    sql: "UPDATE graph_edges SET deleted_at = ?1 \
                          WHERE namespace = ?2 AND source_id = ?3 AND target_id = ?4 AND relation = ?5"
                        .to_string(),
                    params: vec![
                        SqlValue::Integer(chrono::Utc::now().timestamp_micros()),
                        SqlValue::Text("local".to_string()),
                        SqlValue::Text(a_id.to_string()),
                        SqlValue::Text(b_id.to_string()),
                        SqlValue::Text("extends".to_string()),
                    ],
                    label: Some("test-soft-delete-edge".to_string()),
                })
                .await
                .expect("soft delete edge");
            assert_eq!(affected, 1, "soft-delete must touch exactly the seeded row");
        }
        let (_, _, _, deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert!(
            deleted_at.is_some(),
            "row must be soft-deleted before the resurrect attempt"
        );

        let refusal = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": a_id.to_string(),
                "target_id": b_id.to_string(),
                "relation": "extends",
                "weight": 0.75,
            }),
        )
        .await
        .expect_err("implicit resurrection must be refused at prepare time");
        assert!(matches!(
            refusal,
            RuntimeError::InvalidInput(message) if message.contains("resurrect=true")
        ));
        let (_, weight, _, deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert_eq!(weight, Some(1.0), "refusal must preserve the tombstone");
        assert!(deleted_at.is_some());

        let plan_relink = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": a_id.to_string(),
                "target_id": b_id.to_string(),
                "relation": "extends",
                "weight": 0.75,
                "resurrect": true,
            }),
        )
        .await
        .expect("prepare explicitly resurrecting link");
        let outcome_relink =
            crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan_relink])
                .await
                .expect("seam call ok");
        assert!(
            matches!(
                outcome_relink,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "re-linking a soft-deleted triple must resurrect it, not roll back: {outcome_relink:?}"
        );
        let (count, weight, _, deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert_eq!(count, 1);
        assert_eq!(weight, Some(0.75));
        assert!(
            deleted_at.is_none(),
            "re-link must resurrect the soft-deleted row (deleted_at -> NULL)"
        );
    }

    /// Atomic delete of an entity and a note must succeed even when the
    /// registered embedding model's `vec_*` table has never been lazily
    /// created (a fresh DB registers models before any vector store is
    /// opened): the raw purge DML must skip tables that don't exist
    /// rather than hit `no such table` and roll back the whole atomic
    /// unit. FTS purge still fires (those tables always exist) and the
    /// delete itself is a clean commit.
    #[tokio::test]
    async fn atomic_delete_succeeds_when_vec_table_never_created() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        // Seed via raw upsert ONLY — never call reindex_entity/reindex_note
        // or vectors_for_model, so the stub model's `vec_*` table is never
        // lazily created (opening the vector store is what creates it).
        let entity = khive_storage::Entity::new("local", "concept", "no-vec-table-entity");
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");

        let mut note = khive_storage::note::Note::new("local", "observation", "no-vec-table-note");
        note.name = Some("no-vec-table-note".to_string());
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");

        for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
            let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
                .await
                .unwrap_or_else(|e| panic!("prepare delete ({kind}) must not fail: {e}"));
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .unwrap_or_else(|e| {
                    panic!("atomic delete ({kind}) must not hit `no such table`: {e}")
                });
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected a clean commit ({kind}): {outcome:?}"
            );
        }

        assert!(
            runtime
                .get_entity_including_deleted(&token, entity_id)
                .await
                .expect("get entity")
                .expect("entity row still present (soft delete)")
                .deleted_at
                .is_some(),
            "entity must be soft-deleted"
        );
        assert!(
            runtime
                .get_note_including_deleted(&token, note_id)
                .await
                .expect("get note")
                .expect("note row still present (soft delete)")
                .deleted_at
                .is_some(),
            "note must be soft-deleted"
        );
    }

    /// Atomic hard delete must be able to purge a record that was already
    /// soft-deleted: parity with `delete(id, hard=true)` being the public
    /// purge route after a prior soft delete (the non-atomic hard path
    /// resolves including deleted rows and its DML carries no `deleted_at`
    /// predicate).
    #[tokio::test]
    async fn atomic_hard_delete_purges_already_soft_deleted_entity_and_note() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let entity =
            khive_storage::Entity::new("local", "concept", "tombstoned-entity-hard-delete");
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity.clone())
            .await
            .expect("seed entity");
        runtime
            .reindex_entity(&token, &entity)
            .await
            .expect("seed index rows");

        let mut note =
            khive_storage::note::Note::new("local", "observation", "tombstoned-note-hard-delete");
        note.name = Some("tombstoned-note-hard-delete".to_string());
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note.clone())
            .await
            .expect("seed note");
        runtime
            .reindex_note(&token, &note)
            .await
            .expect("seed index rows");

        // First: SOFT delete both (via atomic prepare) so they're tombstoned
        // going into the hard-delete attempt below.
        for id in [entity_id, note_id] {
            let plan = prepare_delete(&runtime, &token, &json!({"id": id.to_string()}), None)
                .await
                .expect("prepare soft delete");
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .expect("soft delete commit");
            assert!(matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ));
        }
        assert!(
            runtime
                .get_entity_including_deleted(&token, entity_id)
                .await
                .expect("get entity")
                .expect("entity present")
                .deleted_at
                .is_some(),
            "entity must be soft-deleted before the hard-delete attempt"
        );
        assert!(
            runtime
                .get_note_including_deleted(&token, note_id)
                .await
                .expect("get note")
                .expect("note present")
                .deleted_at
                .is_some(),
            "note must be soft-deleted before the hard-delete attempt"
        );

        // Now: HARD delete the already-tombstoned records.
        for (id, kind) in [(entity_id, "entity"), (note_id, "note")] {
            let plan = prepare_delete(
                &runtime,
                &token,
                &json!({"id": id.to_string(), "hard": true}),
                None,
            )
            .await
            .unwrap_or_else(|e| {
                panic!(
                    "prepare hard delete ({kind}) of an already-soft-deleted record \
                         must resolve it: {e}"
                )
            });
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .unwrap_or_else(|e| panic!("hard delete ({kind}) commit failed: {e}"));
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected a clean hard-delete commit ({kind}): {outcome:?}"
            );
        }

        assert!(
            runtime
                .get_entity_including_deleted(&token, entity_id)
                .await
                .expect("get entity")
                .is_none(),
            "entity row must be fully purged after hard delete"
        );
        assert!(
            runtime
                .get_note_including_deleted(&token, note_id)
                .await
                .expect("get note")
                .is_none(),
            "note row must be fully purged after hard delete"
        );
        assert!(
            runtime
                .text(&token)
                .expect("text store")
                .get_document("local", entity_id)
                .await
                .expect("get_document")
                .is_none(),
            "entity FTS row must be purged after hard delete"
        );
        assert!(
            runtime
                .text_for_notes(&token)
                .expect("text store")
                .get_document("local", note_id)
                .await
                .expect("get_document")
                .is_none(),
            "note FTS row must be purged after hard delete"
        );
        let vec_store = runtime
            .vectors_for_model(&token, STUB_MODEL)
            .expect("vec store");
        assert_eq!(
            vec_store.count().await.expect("count after"),
            0,
            "vector rows for both records must be purged after hard delete"
        );
    }

    // ------------------------------------------------------------------
    // event-store append parity
    // ------------------------------------------------------------------

    /// Fetch every event of `kind` targeting `target_id`, via the same
    /// `EventStore::query_events` surface `--atomic` callers would use to
    /// verify parity — not a raw SQL probe.
    async fn events_for_target(
        runtime: &KhiveRuntime,
        token: &NamespaceToken,
        target_id: Uuid,
        kind: EventKind,
    ) -> Vec<khive_storage::Event> {
        let event_store = runtime.events(token).expect("event store");
        let filter = khive_storage::EventFilter {
            kinds: vec![kind],
            ..Default::default()
        };
        let page = event_store
            .query_events(filter, khive_storage::types::PageRequest::default())
            .await
            .expect("query_events");
        page.items
            .into_iter()
            .filter(|e| e.target_id == Some(target_id))
            .collect()
    }

    /// Atomic `update(id=<entity>, name=...)` must append an
    /// `EntityUpdated` event, matching `curation::update_entity`: the
    /// event is appended unconditionally after a successful row update,
    /// not only on the reindex-triggering subset.
    #[tokio::test]
    async fn atomic_update_entity_appends_entity_updated_event() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entity = khive_storage::Entity::new("local", "concept", "gap1-entity");
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": entity_id.to_string(), "name": "gap1-entity-renamed"}),
            None,
        )
        .await
        .expect("prepare update");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));

        let events = events_for_target(&runtime, &token, entity_id, EventKind::EntityUpdated).await;
        assert_eq!(
            events.len(),
            1,
            "expected exactly one EntityUpdated event for {entity_id}"
        );
        assert_eq!(events[0].namespace, "local");
        assert_eq!(
            events[0].actor, "anonymous:local",
            "atomic event attribution must come from the authorized token"
        );
        assert_eq!(events[0].payload["id"], json!(entity_id.to_string()));
        assert_eq!(
            events[0].payload["changed_fields"],
            json!(["name"]),
            "changed_fields must name exactly the patched fields"
        );
    }

    /// Atomic soft and hard delete of an entity must each append an
    /// `EntityDeleted` event, matching `operations::delete_entity`, which
    /// fires on both delete modes.
    #[tokio::test]
    async fn atomic_delete_entity_appends_entity_deleted_event_soft_and_hard() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        for hard in [false, true] {
            let entity =
                khive_storage::Entity::new("local", "concept", format!("gap1-entity-hard-{hard}"));
            let entity_id = entity.id;
            runtime
                .entities(&token)
                .expect("entities store")
                .upsert_entity(entity)
                .await
                .expect("seed entity");

            let args = if hard {
                json!({"id": entity_id.to_string(), "hard": true})
            } else {
                json!({"id": entity_id.to_string()})
            };
            let plan = prepare_delete(&runtime, &token, &args, None)
                .await
                .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected a clean delete commit (hard={hard}): {outcome:?}"
            );

            let events =
                events_for_target(&runtime, &token, entity_id, EventKind::EntityDeleted).await;
            assert_eq!(
                events.len(),
                1,
                "expected exactly one EntityDeleted event for {entity_id} (hard={hard})"
            );
            assert_eq!(events[0].payload["hard"], json!(hard));
        }
    }

    #[tokio::test]
    async fn atomic_hard_delete_emits_lineage_warning_in_commit_unit() {
        let runtime = scratch_runtime();
        let token = NamespaceToken::mint_authorized(
            Namespace::local(),
            crate::ActorRef::new("agent", "atomic-lineage-deleter"),
        );
        let doomed = khive_storage::Entity::new("local", "document", "atomic-doomed");
        let source = khive_storage::Entity::new("local", "artifact", "atomic-source");
        let doomed_id = doomed.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(doomed)
            .await
            .expect("seed doomed entity");
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(source.clone())
            .await
            .expect("seed source entity");
        runtime
            .link(
                &token,
                source.id,
                doomed_id,
                EdgeRelation::DerivedFrom,
                1.0,
                None,
            )
            .await
            .expect("seed protected edge");

        let plan = prepare_delete(
            &runtime,
            &token,
            &json!({"id": doomed_id.to_string(), "hard": true}),
            None,
        )
        .await
        .expect("prepare hard delete");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("delete commit");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));

        let warnings = events_for_target(&runtime, &token, doomed_id, EventKind::Audit).await;
        assert_eq!(warnings.len(), 1);
        assert_eq!(warnings[0].actor, "agent:atomic-lineage-deleter");
        assert_eq!(warnings[0].payload["relation"], "derived_from");
        assert_eq!(warnings[0].payload["warning"], "provenance_loss");
    }

    /// Atomic soft and hard delete of a note must each append a
    /// `NoteDeleted` event, matching `operations::delete_note`, which
    /// fires on both delete modes.
    #[tokio::test]
    async fn atomic_delete_note_appends_note_deleted_event_soft_and_hard() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        for hard in [false, true] {
            let mut note = khive_storage::note::Note::new(
                "local",
                "observation",
                format!("gap1-note-content-hard-{hard}"),
            );
            note.name = Some(format!("gap1-note-hard-{hard}"));
            let note_id = note.id;
            runtime
                .notes(&token)
                .expect("notes store")
                .upsert_note(note)
                .await
                .expect("seed note");

            let args = if hard {
                json!({"id": note_id.to_string(), "hard": true})
            } else {
                json!({"id": note_id.to_string()})
            };
            let plan = prepare_delete(&runtime, &token, &args, None)
                .await
                .unwrap_or_else(|e| panic!("prepare delete (hard={hard}): {e}"));
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .unwrap_or_else(|e| panic!("delete commit (hard={hard}): {e}"));
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected a clean delete commit (hard={hard}): {outcome:?}"
            );

            let events = events_for_target(&runtime, &token, note_id, EventKind::NoteDeleted).await;
            assert_eq!(
                events.len(),
                1,
                "expected exactly one NoteDeleted event for {note_id} (hard={hard})"
            );
            assert_eq!(events[0].payload["hard"], json!(hard));
        }
    }

    /// `update` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`; this
    /// asserts `prepare_update` actually builds a plan for one, a
    /// non-symmetric relation (`extends`) exercises the
    /// `edge_upsert_statement` reuse branch, and that the committed row +
    /// `EdgeUpdated` event match canonical `update_edge`'s shape (weight
    /// persisted, relation unchanged, exactly one event).
    #[tokio::test]
    async fn atomic_update_edge_patches_weight_and_appends_edge_updated_event() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "GapEdgeA");
        let b = khive_storage::Entity::new("local", "concept", "GapEdgeB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.4, None)
            .await
            .expect("seed edge");
        let edge_id = Uuid::from(edge.id);

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": edge_id.to_string(), "weight": 0.75}),
            None,
        )
        .await
        .expect("prepare update edge");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(
            matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "expected a clean edge update commit: {outcome:?}"
        );

        let updated = runtime
            .get_edge(&token, edge_id)
            .await
            .expect("get_edge")
            .expect("edge still present");
        assert_eq!(updated.weight, 0.75, "weight patch must persist");
        assert_eq!(updated.relation, EdgeRelation::Extends);

        let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeUpdated).await;
        assert_eq!(
            events.len(),
            1,
            "expected exactly one EdgeUpdated event for {edge_id}"
        );
        assert_eq!(
            events[0].payload["changed_fields"],
            json!(["weight"]),
            "changed_fields must name exactly the patched field"
        );
    }

    /// ADR-115 Amendment 1 §3: the reserved `khive:secret_gate` property key
    /// must be rejected on atomic edge-metadata updates the same way it is
    /// on canonical `update_edge`.
    #[tokio::test]
    async fn atomic_update_edge_rejects_reserved_secret_gate_property() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "ReservedEdgeA");
        let b = khive_storage::Entity::new("local", "concept", "ReservedEdgeB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.4, None)
            .await
            .expect("seed edge");
        let edge_id = Uuid::from(edge.id);

        let err = prepare_update(
            &runtime,
            &token,
            &json!({
                "id": edge_id.to_string(),
                "properties": {"khive:secret_gate": "exempted:content-sha256-manifest-v1"},
            }),
            None,
        )
        .await
        .expect_err("a caller-supplied reserved key on edge metadata must be rejected");
        assert!(
            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("khive:secret_gate") && msg.contains("runtime-owned")),
            "expected a reservation rejection, got: {err:?}"
        );

        let unchanged = runtime
            .get_edge(&token, edge_id)
            .await
            .expect("get_edge")
            .expect("edge still present");
        assert!(
            unchanged.metadata.is_none(),
            "rejected edge update must leave metadata untouched"
        );
    }

    /// The symmetric-relation conflict-absorption branch of
    /// `prepare_update_edge` — mirrors `update_edge_symmetric_dml`'s case
    /// (b): changing a non-symmetric edge's `relation` to a symmetric one
    /// whose canonical natural key collides with an ALREADY-EXISTING
    /// symmetric edge between the same two entities must delete the
    /// requested (non-canonical) row and leave the surviving canonical row
    /// untouched (ADR-039 ON CONFLICT DO NOTHING), rather than raising a
    /// uniqueness error OR overwriting the survivor with the discarded
    /// edge's attributes (khive#1213).
    #[tokio::test]
    async fn atomic_update_edge_symmetric_conflict_absorbs_into_surviving_row() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "GapEdgeSymA");
        let b = khive_storage::Entity::new("local", "concept", "GapEdgeSymB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        // The non-canonical edge under test: A -> B, non-symmetric relation.
        let requested_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
            .await
            .expect("seed requested edge");
        let requested_id = Uuid::from(requested_edge.id);

        // The pre-existing canonical row this update will collide with once
        // `relation` becomes `competes_with` (symmetric).
        let canonical_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
            .await
            .expect("seed canonical edge");
        let canonical_id = Uuid::from(canonical_edge.id);
        assert_ne!(requested_id, canonical_id);

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
            None,
        )
        .await
        .expect("prepare update edge (symmetric conflict)");
        // The plan does not compute a prepare-time advisory surviving id
        // (`target_id` is just the requested id): it carries
        // `edge_natural_key` so a post-commit caller can derive the real
        // surviving id itself. Assert the plan carries the right natural
        // key to look up; the actual surviving row's identity is verified
        // against the DB after commit, below.
        let (canon_src, canon_tgt) =
            canonical_edge_endpoints(EdgeRelation::CompetesWith, a_id, b_id);
        match &plan {
            AtomicOpPlan::Update(p) => {
                assert_eq!(p.target_id, requested_id);
                let key = p
                    .edge_natural_key
                    .as_ref()
                    .expect("symmetric edge update must carry edge_natural_key");
                assert_eq!(key.canon_source_id, canon_src);
                assert_eq!(key.canon_target_id, canon_tgt);
                assert_eq!(key.relation, EdgeRelation::CompetesWith);
            }
            other => panic!("expected an Update plan, got {other:?}"),
        }

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(
            matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "expected a clean symmetric-conflict-absorption commit: {outcome:?}"
        );

        // The requested (non-canonical) row must be gone.
        let requested_after = runtime
            .get_edge_including_deleted(&token, requested_id)
            .await
            .expect("get_edge_including_deleted");
        assert!(
            requested_after.is_none(),
            "the non-canonical requested row must be deleted, not just tombstoned"
        );

        // ADR-039 DO NOTHING: the surviving canonical row keeps its OWN
        // pre-existing attributes — the discarded edge's patched weight
        // (0.9) must never overwrite it.
        let surviving = runtime
            .get_edge(&token, canonical_id)
            .await
            .expect("get_edge")
            .expect("surviving canonical row must remain");
        assert_eq!(
            surviving.weight, 0.6,
            "survivor weight must not be overwritten by the discarded edge's patch"
        );
        assert_eq!(surviving.relation, EdgeRelation::CompetesWith);

        // Event target is the CALLER-supplied id, not the surviving id —
        // mirrors `update_edge`'s event using `edge_id` (the caller's
        // original argument), not the post-absorption id.
        let events =
            events_for_target(&runtime, &token, requested_id, EventKind::EdgeUpdated).await;
        assert_eq!(events.len(), 1);
    }

    /// Regression for khive #1753: an entity atomic update plan is prepared
    /// from one revision, a concurrent (out-of-plan) writer advances the row
    /// past that revision before the plan commits, and the plan's guarded
    /// `entity_replace_if_unchanged_statement` must then affect zero rows —
    /// which `AffectedRowGuard::exactly(1)` must turn into a whole-unit
    /// rollback, not a silent overwrite of the concurrent writer's change.
    /// Before threading the expected revision into the plan's `WHERE`
    /// predicate, this plan used the unconditional `entity_upsert_statement`,
    /// which always affects exactly 1 row regardless of staleness — the
    /// guard could never fire and this test would redden (the outcome would
    /// be `Committed`, and the concurrent writer's `name` change would be
    /// lost under the stale plan's `description`-only patch).
    #[tokio::test]
    async fn atomic_entity_update_plan_stale_revision_rolls_back_unit() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entity = runtime
            .create_entity(
                &token,
                "concept",
                None,
                "StaleEntityPlanTarget",
                None,
                Some(json!({"a": 0})),
                vec![],
            )
            .await
            .expect("seed entity");
        let id = entity.id;

        // PREPARE time: build the plan from the current (soon-to-be-stale)
        // revision.
        let mut plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": id.to_string(), "description": "from the stale plan"}),
            None,
        )
        .await
        .expect("prepare update entity plan");

        // Read the plan's OWN bound values, so the isolation below is
        // derived from the plan rather than assumed about it. Per
        // `entity_replace_if_unchanged_statement`, `?8` (index 7) is the
        // replacement revision, `?12` (index 11) the target id, `?13`
        // (index 12) the expected revision, and `?14` (index 13) the
        // expected deletion marker.
        let (planned_replacement, plan_target_id, plan_expected, plan_expected_deleted) = {
            let statements = match &plan {
                AtomicOpPlan::Update(p) => p.statements.clone(),
                other => panic!("expected an Update plan, got {other:?}"),
            };
            // Locate by LABEL, never by the guard text. A locator keyed on
            // `?8 > updated_at` would stop finding the statement in exactly the
            // mutation run that deletes that conjunct, so the arm would report
            // on this fixture's locator instead of on the guard.
            let cas = statements
                .iter()
                .find(|s| s.statement.label.as_deref() == Some("entity-replace-if-unchanged"))
                .expect("the plan must carry the guarded entity replacement");
            let read = |i: usize| match &cas.statement.params[i] {
                SqlValue::Integer(v) => *v,
                other => panic!("param {i} must be an integer revision, got {other:?}"),
            };
            let read_marker = |i: usize| match &cas.statement.params[i] {
                SqlValue::Null => None,
                SqlValue::Integer(v) => Some(*v),
                other => panic!("param {i} must be a deletion marker, got {other:?}"),
            };
            let read_text = |i: usize| match &cas.statement.params[i] {
                SqlValue::Text(v) => v.clone(),
                other => panic!("param {i} must be a text id, got {other:?}"),
            };
            (read(7), read_text(11), read(12), read_marker(13))
        };
        // The identity conjunct, on the same footing as the revision and the
        // deletion marker. `id = ?12` is live in the same UPDATE, so a plan
        // that bound any other row's id would affect zero rows and roll the
        // unit back for a reason this test does not name — an outcome
        // indistinguishable from the one it does name.
        assert_eq!(
            plan_target_id,
            id.to_string(),
            "fixture premise: the plan's `?12` must be the row under test, otherwise \
             `id = ?12` refuses on identity and the rollback stops being attributable to \
             the expected-revision guard"
        );

        // A concurrent writer commits BEFORE the plan runs, advancing the
        // row's revision past what the plan's guard expects.
        runtime
            .update_entity(
                &token,
                id,
                crate::curation::EntityPatch {
                    name: Some("ConcurrentWriterWon".to_string()),
                    ..Default::default()
                },
            )
            .await
            .expect("concurrent writer update");

        // Pin the stored revision one microsecond BELOW the plan's replacement.
        // This is what makes the test specific to the guard it names. Both the
        // plan and the concurrent writer derive their revision from
        // `max(now, expected + 1)`, so left alone the concurrent write lands at
        // or past the plan's own replacement and BOTH conjuncts refuse — the
        // test would then stay green if either guard were deleted. Pinning
        // leaves `?8 > updated_at` satisfied, so it cannot be the refuser,
        // while `updated_at = ?13` is violated, which is the single condition
        // this test names. The shape is reachable in production whenever the
        // concurrent writer's clock trails the preparing writer's.
        let stored_pinned = planned_replacement - 1;
        assert_ne!(
            stored_pinned, plan_expected,
            "fixture premise: the pinned revision must differ from the plan's expected \
             revision, otherwise `updated_at = ?13` would MATCH and nothing would refuse \
             the plan"
        );
        {
            let mut writer = runtime.sql().writer().await.expect("writer");
            let affected = writer
                .execute(SqlStatement {
                    sql: "UPDATE entities SET version = version + 1, updated_at = ?1 WHERE id = ?2"
                        .to_string(),
                    params: vec![
                        SqlValue::Integer(stored_pinned),
                        SqlValue::Text(id.to_string()),
                    ],
                    label: Some("test-pin-stored-revision".to_string()),
                })
                .await
                .expect("pin the stored revision");
            assert_eq!(affected, 1, "the pin must touch exactly the seeded row");
        }
        assert!(
            planned_replacement > stored_pinned,
            "fixture premise: the plan's replacement must still strictly advance past the \
             stored revision, otherwise `?8 > updated_at` would refuse and this stops being \
             a test of the expected-revision guard"
        );
        // The remaining non-target conjunct. `deleted_at IS ?14` is live in the
        // same UPDATE, so without this read the fixture would stay green if the
        // deletion marker were the actual refuser — the refusal would look
        // identical. Read the row as it stands at DML time, after both the
        // concurrent writer and the pin.
        {
            let stored = runtime
                .get_entity_including_deleted(&token, id)
                .await
                .expect("read the stored row")
                .expect("the seeded row is present before the plan runs");
            assert_eq!(
                stored.updated_at, stored_pinned,
                "fixture premise: the pin must be what the guard reads, so the stored \
                 revision is the pinned value and nothing re-advanced it"
            );
            assert_eq!(
                stored.deleted_at, plan_expected_deleted,
                "fixture premise: the stored deletion marker must MATCH the plan's `?14`, \
                 otherwise `deleted_at IS ?14` refuses too and this stops being a test of \
                 the expected-revision guard alone"
            );
            // Keep this legacy timestamp-conjunct oracle independent of the
            // newly added persisted-version predicate. The production plan
            // would also refuse on its old version; this fixture pins only
            // that additional predicate to the observed current value.
            let AtomicOpPlan::Update(update) = &mut plan else {
                unreachable!()
            };
            let cas = update
                .statements
                .iter_mut()
                .find(|s| s.statement.label.as_deref() == Some("entity-replace-if-unchanged"))
                .unwrap();
            cas.statement.params[14] = SqlValue::Integer(stored.version);
        }

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("the seam call itself must not error; the unit rolls back cleanly");
        match outcome {
            crate::atomic_runner::AtomicRunOutcome::RolledBack {
                failed_op_index,
                failure,
            } => {
                assert_eq!(
                    failed_op_index, 0,
                    "the sole op's guard must be the one that fails"
                );
                // Attribution rather than inference. The premises above
                // establish that every non-target conjunct holds; this names
                // the statement that actually refused. Without it a unit that
                // rolled back at some other statement reads identically from
                // the outside, which is the whole failure mode those premises
                // were approximating.
                assert_eq!(
                    failure,
                    crate::atomic_runner::AtomicOpFailure::GuardFailed {
                        statement_label: Some("entity-replace-if-unchanged".to_string()),
                        expected: crate::atomic_plan::AffectedRowGuard::exactly(1),
                        observed: 0,
                    },
                    "the guarded entity replacement must be the statement whose guard refused"
                );
            }
            other => panic!(
                "a stale entity plan must roll back, not silently overwrite the concurrent \
                 writer's change: {other:?}"
            ),
        }

        let after = runtime.get_entity(&token, id).await.expect("get_entity");
        assert_eq!(
            after.name, "ConcurrentWriterWon",
            "the concurrent writer's committed name must survive the rolled-back stale plan"
        );
        assert_eq!(
            after.description, None,
            "the stale plan's description patch must NOT have landed"
        );
    }

    /// Edge counterpart of `atomic_entity_update_plan_stale_revision_rolls_back_unit`:
    /// a non-symmetric edge atomic update plan prepared from one revision
    /// must roll back the whole unit when a concurrent writer advances the
    /// row first, rather than silently overwriting that writer's change.
    #[tokio::test]
    async fn atomic_edge_update_plan_stale_revision_rolls_back_unit() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "StaleEdgePlanA");
        let b = khive_storage::Entity::new("local", "concept", "StaleEdgePlanB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
            .await
            .expect("seed edge");
        let edge_id = Uuid::from(edge.id);

        // PREPARE time: build the plan from the current (soon-to-be-stale)
        // revision.
        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": edge_id.to_string(), "properties": {"note": "from the stale plan"}}),
            None,
        )
        .await
        .expect("prepare update edge plan");

        // Read the plan's OWN bound values, so the isolation below is
        // derived from the plan rather than assumed about it. Per
        // `edge_replace_if_unchanged_statement`, `?6` (index 5) is the
        // replacement revision, `?10` (index 9) the target id, `?11`
        // (index 10) the expected revision, and `?12` (index 11) the expected
        // deletion marker.
        let (planned_replacement, plan_target_id, plan_expected, plan_expected_deleted) = {
            let statements = match &plan {
                AtomicOpPlan::Update(p) => p.statements.clone(),
                other => panic!("expected an Update plan, got {other:?}"),
            };
            // Locate by LABEL, never by the guard text — see the entity sibling
            // above: a locator keyed on the conjunct disappears in exactly the
            // mutation run that deletes it.
            let cas = statements
                .iter()
                .find(|s| s.statement.label.as_deref() == Some("edge-replace-if-unchanged"))
                .expect("the plan must carry the guarded edge replacement");
            let read = |i: usize| match &cas.statement.params[i] {
                SqlValue::Integer(v) => *v,
                other => panic!("param {i} must be an integer revision, got {other:?}"),
            };
            let read_marker = |i: usize| match &cas.statement.params[i] {
                SqlValue::Null => None,
                SqlValue::Integer(v) => Some(*v),
                other => panic!("param {i} must be a deletion marker, got {other:?}"),
            };
            let read_text = |i: usize| match &cas.statement.params[i] {
                SqlValue::Text(v) => v.clone(),
                other => panic!("param {i} must be a text id, got {other:?}"),
            };
            (read(5), read_text(9), read(10), read_marker(11))
        };
        // The identity conjunct, for the same reason as the entity sibling:
        // `id = ?10` is live in the same UPDATE and a misbound target refuses
        // indistinguishably.
        assert_eq!(
            plan_target_id,
            edge_id.to_string(),
            "fixture premise: the plan's `?10` must be the edge under test, otherwise \
             `id = ?10` refuses on identity and the rollback stops being attributable to \
             the expected-revision guard"
        );

        // A concurrent writer commits BEFORE the plan runs, advancing the
        // row's revision past what the plan's guard expects.
        runtime
            .update_edge(
                &token,
                edge_id,
                crate::curation::EdgePatch {
                    weight: Some(0.75),
                    ..Default::default()
                },
            )
            .await
            .expect("concurrent writer update");

        // Pin the stored revision one microsecond BELOW the plan's replacement,
        // for the same reason as the entity sibling above: both the plan and
        // the concurrent writer derive their revision from
        // `max(now, expected + 1)`, so left alone BOTH conjuncts refuse and the
        // test would stay green if either guard were deleted. Pinning leaves
        // `?6 > updated_at` satisfied, so it cannot be the refuser, while
        // `updated_at = ?11` is violated — the single condition this test names.
        let stored_pinned = planned_replacement - 1;
        assert_ne!(
            stored_pinned, plan_expected,
            "fixture premise: the pinned revision must differ from the plan's expected \
             revision, otherwise `updated_at = ?11` would MATCH and nothing would refuse \
             the plan"
        );
        {
            let mut writer = runtime.sql().writer().await.expect("writer");
            let affected = writer
                .execute(SqlStatement {
                    sql: "UPDATE graph_edges SET updated_at = ?1 WHERE id = ?2".to_string(),
                    params: vec![
                        SqlValue::Integer(stored_pinned),
                        SqlValue::Text(edge_id.to_string()),
                    ],
                    label: Some("test-pin-stored-revision".to_string()),
                })
                .await
                .expect("pin the stored revision");
            assert_eq!(affected, 1, "the pin must touch exactly the seeded edge");
        }
        assert!(
            planned_replacement > stored_pinned,
            "fixture premise: the plan's replacement must still strictly advance past the \
             stored revision, otherwise `?6 > updated_at` would refuse and this stops being \
             a test of the expected-revision guard"
        );
        // The remaining non-target conjunct, for the same reason as the entity
        // sibling: `deleted_at IS ?12` is live in the same UPDATE and would
        // produce an indistinguishable refusal.
        {
            let stored = runtime
                .get_edge_including_deleted(&token, edge_id)
                .await
                .expect("read the stored edge")
                .expect("the seeded edge is present before the plan runs");
            assert_eq!(
                stored.updated_at.timestamp_micros(),
                stored_pinned,
                "fixture premise: the pin must be what the guard reads, so the stored \
                 revision is the pinned value and nothing re-advanced it"
            );
            assert_eq!(
                stored.deleted_at.map(|d| d.timestamp_micros()),
                plan_expected_deleted,
                "fixture premise: the stored deletion marker must MATCH the plan's `?12`, \
                 otherwise `deleted_at IS ?12` refuses too and this stops being a test of \
                 the expected-revision guard alone"
            );
        }

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("the seam call itself must not error; the unit rolls back cleanly");
        match outcome {
            crate::atomic_runner::AtomicRunOutcome::RolledBack {
                failed_op_index,
                failure,
            } => {
                assert_eq!(
                    failed_op_index, 0,
                    "the sole op's guard must be the one that fails"
                );
                // Attribution rather than inference — see the entity sibling.
                assert_eq!(
                    failure,
                    crate::atomic_runner::AtomicOpFailure::GuardFailed {
                        statement_label: Some("edge-replace-if-unchanged".to_string()),
                        expected: crate::atomic_plan::AffectedRowGuard::exactly(1),
                        observed: 0,
                    },
                    "the guarded edge replacement must be the statement whose guard refused"
                );
            }
            other => panic!(
                "a stale edge plan must roll back, not silently overwrite the concurrent \
                 writer's change: {other:?}"
            ),
        }

        let after = runtime
            .get_edge(&token, edge_id)
            .await
            .expect("get_edge")
            .expect("edge still exists");
        assert!(
            (after.weight - 0.75).abs() < 0.001,
            "the concurrent writer's committed weight must survive the rolled-back stale plan: {after:?}"
        );
        assert!(
            after.metadata.is_none(),
            "the stale plan's properties patch must NOT have landed: {after:?}"
        );
    }

    /// A soft-deleted surviving canonical row must not be resurrected by a
    /// conflicting symmetric-relation update (ADR-039 DO NOTHING; khive#1213):
    /// the requested edge is still deleted (conflict absorbed), but the
    /// tombstoned survivor must stay tombstoned.
    #[tokio::test]
    async fn atomic_update_edge_symmetric_conflict_does_not_resurrect_tombstoned_survivor() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "GapEdgeSymTombA");
        let b = khive_storage::Entity::new("local", "concept", "GapEdgeSymTombB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let requested_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
            .await
            .expect("seed requested edge");
        let requested_id = Uuid::from(requested_edge.id);

        let canonical_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
            .await
            .expect("seed canonical edge");
        let canonical_id = Uuid::from(canonical_edge.id);
        runtime
            .delete_edge(&token, canonical_id, false)
            .await
            .expect("soft-delete canonical edge");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
            None,
        )
        .await
        .expect("prepare update edge (symmetric conflict over tombstone)");

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(
            matches!(
                outcome,
                crate::atomic_runner::AtomicRunOutcome::Committed { .. }
            ),
            "expected a clean symmetric-conflict-absorption commit: {outcome:?}"
        );

        let requested_after = runtime
            .get_edge_including_deleted(&token, requested_id)
            .await
            .expect("get_edge_including_deleted");
        assert!(
            requested_after.is_none(),
            "the non-canonical requested row must be deleted, not just tombstoned"
        );

        let canonical_after = runtime
            .get_edge(&token, canonical_id)
            .await
            .expect("get_edge");
        assert!(
            canonical_after.is_none(),
            "a tombstoned survivor must not be resurrected by a conflicting update"
        );
    }

    /// The same-unit race: `[delete(X), update(X -> competes_with)]` where
    /// an already-existing canonical row sits at the post-update natural
    /// key. Both ops' async prepare passes run before either commits, so at
    /// prepare time `X` still exists and both plans build. At commit time
    /// `delete(X)` removes it first; `update(X -> competes_with)`'s own
    /// commit-time statements must then fail loud (its target no longer
    /// exists) rather than silently absorbing into the pre-existing
    /// canonical row it never causally touched. The whole atomic unit must
    /// roll back — parity with canonical `update_edge`'s `NotFound` for a
    /// missing edge, expressed here as the unit-level abort for any op
    /// whose commit-time guard fails.
    #[tokio::test]
    async fn atomic_update_edge_symmetric_same_unit_delete_race_aborts_the_unit() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "GapEdgeRaceA");
        let b = khive_storage::Entity::new("local", "concept", "GapEdgeRaceB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        // The row op 1 will try to update — deleted by op 0 in the SAME unit.
        let requested_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
            .await
            .expect("seed requested edge");
        let requested_id = Uuid::from(requested_edge.id);

        // The pre-existing canonical row the buggy `id = ?2 OR natural-key`
        // predicate used to silently absorb into.
        let canonical_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
            .await
            .expect("seed canonical edge");
        let canonical_id = Uuid::from(canonical_edge.id);

        let delete_plan = prepare_delete(
            &runtime,
            &token,
            &json!({"id": requested_id.to_string(), "hard": true}),
            None,
        )
        .await
        .expect("prepare delete edge");
        let update_plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
            None,
        )
        .await
        .expect("prepare update edge (both prepares run before either commits)");

        let outcome = crate::atomic_runner::run_atomic_unit(
            runtime.sql().as_ref(),
            vec![delete_plan, update_plan],
        )
        .await
        .expect("the seam call itself must not error — the unit rolls back cleanly");
        match outcome {
            crate::atomic_runner::AtomicRunOutcome::RolledBack {
                failed_op_index, ..
            } => {
                assert_eq!(
                    failed_op_index, 1,
                    "op 1 (the update) must be the one whose guard fails"
                );
            }
            other => panic!("expected the whole unit to roll back, got {other:?}"),
        }

        // Whole-unit rollback: op 0's delete must be undone too.
        let requested_after = runtime
            .get_edge(&token, requested_id)
            .await
            .expect("get_edge");
        assert!(
            requested_after.is_some(),
            "delete(X) must have rolled back along with the failed update"
        );
        // The pre-existing canonical row must be completely untouched.
        let canonical_after = runtime
            .get_edge(&token, canonical_id)
            .await
            .expect("get_edge")
            .expect("canonical row must still be present");
        assert_eq!(
            canonical_after.weight, 0.6,
            "the pre-existing canonical row must never have been touched by the aborted update"
        );
    }

    /// The symmetric absorption delete's atomic commit-time statement
    /// (`edge_symmetric_delete_if_conflict_statement`) must refuse a plan
    /// built from a since-changed snapshot, exactly like the non-symmetric
    /// `atomic_edge_update_plan_stale_revision_rolls_back_unit` case above —
    /// even though a genuine canonical survivor exists at the target natural
    /// key. `E`'s own revision changes (via an unrelated production
    /// `update_edge` call) AFTER `prepare_update` captured its snapshot but
    /// BEFORE the plan runs; the whole atomic unit must roll back rather than
    /// silently absorbing `E` into the survivor and discarding the
    /// concurrent write.
    #[tokio::test]
    async fn atomic_update_edge_symmetric_absorption_plan_stale_revision_rolls_back_unit() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "StaleAbsorbA");
        let b = khive_storage::Entity::new("local", "concept", "StaleAbsorbB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        // The pre-existing canonical row this plan will try to absorb into.
        let canonical_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::CompetesWith, 0.6, None)
            .await
            .expect("seed canonical edge");
        let canonical_id = Uuid::from(canonical_edge.id);

        // E: the requested edge under test.
        let requested_edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.2, None)
            .await
            .expect("seed requested edge");
        let requested_id = Uuid::from(requested_edge.id);

        // PREPARE time: build the plan from the current (soon-to-be-stale)
        // revision.
        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": requested_id.to_string(), "relation": "competes_with", "weight": 0.9}),
            None,
        )
        .await
        .expect("prepare update edge (symmetric absorption)");

        // A concurrent writer commits BEFORE the plan runs, advancing E's
        // revision past what the plan's guard expects. Relation stays
        // `extends` (non-symmetric), so this lands via the already-guarded
        // replace path — independent of the absorption bug under test.
        let concurrent = runtime
            .update_edge(
                &token,
                requested_id,
                crate::curation::EdgePatch {
                    weight: Some(0.77),
                    ..Default::default()
                },
            )
            .await
            .expect("concurrent writer update");
        assert!((concurrent.weight - 0.77).abs() < 1e-9);

        // Statement 2's in-place arm carries a strict-advance conjunct
        // (`?7 > updated_at`) beside the expected-revision conjunct
        // (`updated_at = ?10`) this test names. Both the plan and the
        // concurrent writer derive their revision from `max(now, expected + 1)`
        // and the concurrent writer ran LATER, so left alone the stored
        // revision sits at or past the plan's replacement and BOTH conjuncts
        // refuse — the test would stay green with either one deleted. Pin the
        // stored revision one microsecond below the plan's replacement, the
        // same treatment the entity and edge siblings above already carry.
        let (absorb_replacement, absorb_expected, absorb_expected_deleted, absorb_ns, absorb_id) = {
            let statements = match &plan {
                AtomicOpPlan::Update(p) => p.statements.clone(),
                other => panic!("expected an Update plan, got {other:?}"),
            };
            let s = statements
                .iter()
                .find(|st| {
                    st.statement.label.as_deref() == Some("edge-symmetric-absorb-or-update-inplace")
                })
                .expect("the plan must carry the guarded symmetric in-place absorb");
            let int = |i: usize| match &s.statement.params[i] {
                SqlValue::Integer(v) => *v,
                other => panic!("param {i} must be an integer revision, got {other:?}"),
            };
            let marker = |i: usize| match &s.statement.params[i] {
                SqlValue::Null => None,
                SqlValue::Integer(v) => Some(*v),
                other => panic!("param {i} must be a deletion marker, got {other:?}"),
            };
            let txt = |i: usize| match &s.statement.params[i] {
                SqlValue::Text(v) => v.clone(),
                other => panic!("param {i} must be text, got {other:?}"),
            };
            (int(6), int(9), marker(10), txt(0), txt(1))
        };
        let stored_pinned = absorb_replacement - 1;
        assert_ne!(
            stored_pinned, absorb_expected,
            "fixture premise: the pinned revision must differ from the plan's `?10`, otherwise \
             `updated_at = ?10` would MATCH and nothing would refuse the in-place arm (and the \
             guarded delete's `updated_at = ?6`, bound to the same value, would MATCH too and \
             fire, leaving `changes() = 1` and selecting the absorbed arm instead)"
        );
        {
            let mut writer = runtime.sql().writer().await.expect("writer");
            let affected = writer
                .execute(SqlStatement {
                    sql: "UPDATE graph_edges SET updated_at = ?1 WHERE id = ?2".to_string(),
                    params: vec![
                        SqlValue::Integer(stored_pinned),
                        SqlValue::Text(requested_id.to_string()),
                    ],
                    label: Some("test-pin-stored-revision".to_string()),
                })
                .await
                .expect("pin the stored revision");
            assert_eq!(affected, 1, "the pin must touch exactly the requested edge");
        }
        assert!(
            absorb_replacement > stored_pinned,
            "fixture premise: the plan's `?7` must still strictly advance past the stored \
             revision, otherwise `?7 > updated_at` refuses too and the refusal stops being \
             attributable to `updated_at = ?10`"
        );

        // A symmetric plan carries TWO guarded statements, and it matters
        // which one refuses.
        //
        //   1. `edge-symmetric-delete-if-conflict`, guarded
        //      `{expected_min: 0, expected_max: Some(1)}` — a zero-row delete
        //      SATISFIES this guard. It does not roll the unit back. Its
        //      effect here is to leave SQLite's `changes()` at 0.
        //   2. `edge-symmetric-absorb-or-update-inplace`, guarded
        //      `exactly(1)` — this is the statement that refuses and rolls the
        //      unit back, through its in-place arm
        //      `(id = ?2 AND changes() = 0 AND updated_at = ?10 AND
        //        deleted_at IS ?11 AND ?7 > updated_at)`,
        //      whose `updated_at = ?10` is the stale-revision guard under
        //      test. The absorbed arm needs `changes() = 1`, so the delete
        //      refusing is what selects the in-place arm.
        //
        // So the premises come in two layers. The delete's own predicates
        // (`?6`, `?7`, and the EXISTS survivor at `?3`/`?4`/`?5`) are premised
        // because they decide WHETHER the delete fires, and therefore which
        // arm of statement 2 is live. Statement 2's remaining conjuncts are
        // premised because each of them refuses identically to the revision
        // guard this test names. Read the plans' own bound values and check
        // the world against them, immediately before the DML.
        {
            let statements = match &plan {
                AtomicOpPlan::Update(p) => p.statements.clone(),
                other => panic!("expected an Update plan, got {other:?}"),
            };
            // By LABEL, for the same reason as the siblings above.
            let del = statements
                .iter()
                .find(|s| s.statement.label.as_deref() == Some("edge-symmetric-delete-if-conflict"))
                .expect("the plan must carry the guarded symmetric delete");
            let text = |i: usize| match &del.statement.params[i] {
                SqlValue::Text(v) => v.clone(),
                other => panic!("param {i} must be text, got {other:?}"),
            };
            let plan_expected_updated = match &del.statement.params[5] {
                SqlValue::Integer(v) => *v,
                other => panic!("param 5 must be the expected revision, got {other:?}"),
            };
            let plan_expected_deleted = match &del.statement.params[6] {
                SqlValue::Null => None,
                SqlValue::Integer(v) => Some(*v),
                other => panic!("param 6 must be a deletion marker, got {other:?}"),
            };

            // The delete's OWN identity conjuncts. Reading them into the
            // survivor-count panic message below is not asserting them: point
            // `?1` or `?2` at a row that does not exist and the delete still
            // affects zero rows, its `0..=1` guard still accepts that, and
            // statement 2 still refuses on the pinned stale revision — so the
            // test stays green while establishing nothing about which row the
            // named delete attempted.
            assert_eq!(
                text(0),
                "local",
                "fixture premise: the delete's `?1` must be the namespace the row lives in, \
                 otherwise `namespace = ?1` refuses on its own"
            );
            assert_eq!(
                text(1),
                requested_id.to_string(),
                "fixture premise: the delete's `?2` must be the edge under test, otherwise \
                 `id = ?2` refuses on identity and the delete never attempted the requested row"
            );

            let requested_now = runtime
                .get_edge_including_deleted(&token, requested_id)
                .await
                .expect("read the requested edge")
                .expect("the requested edge is present before the plan runs");
            assert_ne!(
                requested_now.updated_at.timestamp_micros(),
                plan_expected_updated,
                "fixture premise: the concurrent writer must actually have moved the \
                 revision past the plan's `?6`, otherwise nothing refuses the delete"
            );
            assert_eq!(
                requested_now.deleted_at.map(|d| d.timestamp_micros()),
                plan_expected_deleted,
                "fixture premise: the stored deletion marker must MATCH the plan's `?7`, \
                 otherwise `deleted_at IS ?7` refuses too and the refusal is not \
                 attributable to the revision guard"
            );

            // The EXISTS arm. A missing survivor refuses the delete on its own
            // and looks identical from outside, so the premise has to be that
            // arm's OWN question. Do not reconstruct it from the seeded row's
            // endpoints: the natural key the plan binds is the canonical
            // ordering, which need not equal the order the survivor was stored
            // in, and a hand-built comparison would be asserting my model of
            // canonicalisation rather than the predicate. Run the subquery
            // verbatim against the plan's own bound parameters instead.
            assert_ne!(
                canonical_id, requested_id,
                "fixture premise: the survivor must be a DIFFERENT row, since the EXISTS \
                 arm excludes the requested id"
            );
            let survivors = {
                let mut reader = runtime.sql().reader().await.expect("sql reader");
                reader
                    .query_scalar(SqlStatement {
                        sql: "SELECT count(*) FROM graph_edges \
                              WHERE namespace = ?1 AND source_id = ?3 AND target_id = ?4 \
                                AND relation = ?5 AND id != ?2"
                            .to_string(),
                        params: del.statement.params[0..5].to_vec(),
                        label: Some("test-survivor-exists-premise".to_string()),
                    })
                    .await
                    .expect("run the plan's own EXISTS predicate")
            };
            let survivors = match survivors {
                Some(SqlValue::Integer(n)) => n,
                other => panic!("count(*) must come back as an integer, got {other:?}"),
            };
            assert_eq!(
                survivors,
                1,
                "fixture premise: exactly one survivor must satisfy the plan's own EXISTS \
                 predicate (namespace {}, source {}, target {}, relation {}, id != {}), \
                 otherwise the delete refuses on the survivor arm and the refusal is not \
                 attributable to the revision guard",
                text(0),
                text(2),
                text(3),
                text(4),
                text(1),
            );
        }

        // Statement 2's remaining conjuncts. Each of these refuses the in-place
        // arm identically to the revision guard this test names, so each has to
        // be shown satisfied for the attribution to hold.
        assert_eq!(
            absorb_ns, "local",
            "fixture premise: the plan's `?1` must be the namespace the row lives in, \
             otherwise the outer `namespace = ?1` refuses on its own"
        );
        assert_eq!(
            absorb_id,
            requested_id.to_string(),
            "fixture premise: the plan's `?2` must be the edge under test, otherwise the \
             in-place arm's `id = ?2` refuses on identity"
        );
        {
            let stored = runtime
                .get_edge_including_deleted(&token, requested_id)
                .await
                .expect("read the requested edge")
                .expect("the requested edge is present before the plan runs");
            assert_eq!(
                stored.updated_at.timestamp_micros(),
                stored_pinned,
                "fixture premise: the pin must be what the guard reads at DML time, so the \
                 stored revision is the pinned value and nothing re-advanced it"
            );
            assert_eq!(
                stored.deleted_at.map(|d| d.timestamp_micros()),
                absorb_expected_deleted,
                "fixture premise: the stored deletion marker must MATCH the plan's `?11`, \
                 otherwise `deleted_at IS ?11` refuses too and the refusal stops being \
                 attributable to `updated_at = ?10`"
            );
        }

        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("the seam call itself must not error; the unit rolls back cleanly");
        match outcome {
            crate::atomic_runner::AtomicRunOutcome::RolledBack {
                failed_op_index,
                failure,
            } => {
                assert_eq!(
                    failed_op_index, 0,
                    "the sole op's guard must be the one that fails"
                );
                // The decisive attribution, and the reason the two-layer
                // premise stack above exists: `failed_op_index` cannot
                // distinguish the two statements, because both live in op 0.
                // Naming the label is what says the in-place absorb refused
                // rather than the delete — and the delete's `0..=1` guard
                // means it CANNOT be the refuser, so a run that reported it
                // here would be evidence the guards had been rewired.
                assert_eq!(
                    failure,
                    crate::atomic_runner::AtomicOpFailure::GuardFailed {
                        statement_label: Some(
                            "edge-symmetric-absorb-or-update-inplace".to_string()
                        ),
                        expected: crate::atomic_plan::AffectedRowGuard::exactly(1),
                        observed: 0,
                    },
                    "the guarded in-place absorb must be the statement whose guard refused"
                );
            }
            other => panic!(
                "a stale absorption plan must roll back, not silently absorb E into the \
                 survivor and discard the concurrent writer's change: {other:?}"
            ),
        }

        let requested_after = runtime
            .get_edge(&token, requested_id)
            .await
            .expect("get_edge")
            .expect("E must still exist after the rolled-back absorption");
        assert_eq!(
            requested_after.relation,
            EdgeRelation::Extends,
            "E must be untouched by the rolled-back absorption: {requested_after:?}"
        );
        assert!(
            (requested_after.weight - 0.77).abs() < 1e-9,
            "the concurrent writer's committed weight must survive the rolled-back plan: \
             {requested_after:?}"
        );

        let canonical_after = runtime
            .get_edge(&token, canonical_id)
            .await
            .expect("get_edge")
            .expect("S must still exist after the rolled-back absorption");
        assert_eq!(
            canonical_after.weight, 0.6,
            "the survivor must never have been touched by the aborted absorption"
        );
    }

    /// `update` rejects an entity/note-only field (`name`) on an edge
    /// target, mirroring
    /// `khive-pack-kg::handlers::update::reject_inapplicable_fields`'s
    /// `KindSpec::Edge` arm.
    #[tokio::test]
    async fn atomic_update_edge_rejects_entity_only_field_name() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "GapEdgeRejectA");
        let b = khive_storage::Entity::new("local", "concept", "GapEdgeRejectB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");
        let edge = runtime
            .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
            .await
            .expect("seed edge");
        let edge_id = Uuid::from(edge.id);

        let err = prepare_update(
            &runtime,
            &token,
            &json!({"id": edge_id.to_string(), "name": "not-a-valid-edge-field"}),
            None,
        )
        .await
        .expect_err("edge update with an entity-only field must be rejected");
        let message = err.to_string();
        assert!(
            message.contains("name") && message.contains("edge"),
            "error must name the offending field and the substrate: {message}"
        );
    }

    /// `delete` admits `kind="edge"` per `ATOMIC_ADMISSIBLE_VERBS`; this
    /// asserts `prepare_delete` actually builds a plan for one on both soft
    /// and hard delete, matching `operations::delete_edge`'s row-mode DML
    /// and unconditional `EdgeDeleted` event.
    #[tokio::test]
    async fn atomic_delete_edge_soft_and_hard_appends_edge_deleted_event() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        for hard in [false, true] {
            let entities = runtime.entities(&token).expect("entities store");
            let a = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelA{hard}"));
            let b = khive_storage::Entity::new("local", "concept", format!("GapEdgeDelB{hard}"));
            let (a_id, b_id) = (a.id, b.id);
            entities.upsert_entity(a).await.expect("seed a");
            entities.upsert_entity(b).await.expect("seed b");
            let edge = runtime
                .link(&token, a_id, b_id, EdgeRelation::Extends, 0.5, None)
                .await
                .expect("seed edge");
            let edge_id = Uuid::from(edge.id);

            let args = if hard {
                json!({"id": edge_id.to_string(), "hard": true})
            } else {
                json!({"id": edge_id.to_string()})
            };
            let plan = prepare_delete(&runtime, &token, &args, None)
                .await
                .unwrap_or_else(|e| panic!("prepare delete edge (hard={hard}): {e}"));
            let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
                .await
                .unwrap_or_else(|e| panic!("edge delete commit (hard={hard}): {e}"));
            assert!(
                matches!(
                    outcome,
                    crate::atomic_runner::AtomicRunOutcome::Committed { .. }
                ),
                "expected a clean edge delete commit (hard={hard}): {outcome:?}"
            );

            let after = runtime
                .get_edge_including_deleted(&token, edge_id)
                .await
                .expect("get_edge_including_deleted");
            if hard {
                assert!(after.is_none(), "hard delete must purge the row entirely");
            } else {
                assert!(
                    after.as_ref().is_some_and(|e| e.deleted_at.is_some()),
                    "soft delete must tombstone, not purge"
                );
            }

            let events = events_for_target(&runtime, &token, edge_id, EventKind::EdgeDeleted).await;
            assert_eq!(
                events.len(),
                1,
                "expected exactly one EdgeDeleted event for {edge_id} (hard={hard})"
            );
            assert_eq!(events[0].payload["hard"], json!(hard));
        }
    }

    /// Parity boundary: an atomic `update` of a note appends exactly one
    /// `NoteUpdated` event, because this path and canonical `update_note` build
    /// their plan through the same `prepare_versioned_note_update`, which is
    /// where the event statements are added.
    ///
    /// This test used to assert the opposite. That was a faithful record of a
    /// gap rather than a contract: notes were the substrate that recorded no
    /// update at all, so the parity it certified was parity with nothing.
    #[tokio::test]
    async fn atomic_update_note_appends_its_domain_event() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut note = khive_storage::note::Note::new("local", "observation", "gap1-note-noevent");
        note.name = Some("gap1-note-noevent".to_string());
        let note_id = note.id;
        runtime
            .notes(&token)
            .expect("notes store")
            .upsert_note(note)
            .await
            .expect("seed note");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": note_id.to_string(), "content": "gap1-note-noevent, revised"}),
            None,
        )
        .await
        .expect("prepare update");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));

        let event_store = runtime.events(&token).expect("event store");
        let page = event_store
            .query_events(
                khive_storage::EventFilter::default(),
                khive_storage::types::PageRequest::default(),
            )
            .await
            .expect("query_events");
        let for_note: Vec<_> = page
            .items
            .iter()
            .filter(|e| e.target_id == Some(note_id))
            .collect();
        assert_eq!(
            for_note.len(),
            1,
            "an atomic note update must append exactly one event; found: {for_note:?}"
        );
        assert_eq!(for_note[0].kind, EventKind::NoteUpdated);
        assert_eq!(for_note[0].substrate, SubstrateKind::Note);
        assert_eq!(for_note[0].verb, "update");
        assert_eq!(for_note[0].payload["id"], json!(note_id));
        assert_eq!(for_note[0].payload["text_changed"], json!(true));
    }

    /// Atomic `link` commits its mutation and event-plane observation in the
    /// same unit.
    #[tokio::test]
    async fn atomic_link_appends_created_event_with_edge_observation() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let source = khive_storage::Entity::new("local", "concept", "gap1-link-source");
        let target = khive_storage::Entity::new("local", "concept", "gap1-link-target");
        let (source_id, target_id) = (source.id, target.id);
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(source)
            .await
            .expect("seed source");
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(target)
            .await
            .expect("seed target");

        let plan = prepare_link(
            &runtime,
            &token,
            &json!({
                "source_id": source_id.to_string(),
                "target_id": target_id.to_string(),
                "relation": "extends",
            }),
        )
        .await
        .expect("prepare link");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("seam call ok");
        assert!(matches!(
            outcome,
            crate::atomic_runner::AtomicRunOutcome::Committed { .. }
        ));

        let event_store = runtime.events(&token).expect("event store");
        let page = event_store
            .query_events(
                khive_storage::EventFilter {
                    kinds: vec![EventKind::LinkCreated],
                    ..khive_storage::EventFilter::default()
                },
                khive_storage::types::PageRequest::default(),
            )
            .await
            .expect("query_events");
        assert_eq!(page.items.len(), 1, "link must append one created event");
        let event = &page.items[0];
        assert_eq!(event.payload["mutation"], "created");
        let edge_id = event.target_id.expect("link event targets its edge");
        let observed = event_store
            .query_events(
                khive_storage::EventFilter {
                    observed: vec![edge_id],
                    ..khive_storage::EventFilter::default()
                },
                khive_storage::types::PageRequest::default(),
            )
            .await
            .expect("query observed edge");
        assert_eq!(observed.items.len(), 1);
        assert_eq!(observed.items[0].id, event.id);
    }

    // ------------------------------------------------------------------
    // AddEntity and AddNote plans alongside link
    // ------------------------------------------------------------------

    #[tokio::test]
    async fn prepare_add_entity_rejects_whitespace_only_name() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_add_entity(&runtime, &token, &json!({"kind": "concept", "name": "   "}))
            .await
            .expect_err("whitespace-only entity name must fail prepare");

        assert!(matches!(
            err,
            RuntimeError::InvalidInput(message) if message.contains("name must not be empty")
        ));
    }

    #[tokio::test]
    async fn prepare_add_entity_rejects_non_string_description() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_add_entity(
            &runtime,
            &token,
            &json!({"kind": "concept", "name": "Valid", "description": 42}),
        )
        .await
        .expect_err("non-string entity description must fail prepare");

        assert!(matches!(
            err,
            RuntimeError::InvalidInput(message)
                if message.contains("description must be a string or null")
        ));
    }

    #[tokio::test]
    async fn prepare_add_note_rejects_non_string_name() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_add_note(
            &runtime,
            &token,
            &json!({"kind": "observation", "content": "Valid", "name": 42}),
        )
        .await
        .expect_err("non-string note name must fail prepare");

        assert!(matches!(
            err,
            RuntimeError::InvalidInput(message) if message.contains("name must be a string or null")
        ));
    }

    /// ADR-115 Amendment 1 §3: proposal materialization (`prepare_add_entity`)
    /// must reject the reserved `khive:secret_gate` property key the same way
    /// canonical `create` does — proposal apply is reservation-only, never an
    /// exemption-consuming path.
    #[tokio::test]
    async fn prepare_add_entity_rejects_reserved_secret_gate_property() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_add_entity(
            &runtime,
            &token,
            &json!({
                "kind": "concept",
                "name": "ReservedKeyEntity",
                "properties": {"khive:secret_gate": "exempted:content-sha256-manifest-v1"},
            }),
        )
        .await
        .expect_err("a caller-supplied reserved key on a new entity must be rejected");
        assert!(
            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("khive:secret_gate") && msg.contains("runtime-owned")),
            "expected a reservation rejection, got: {err:?}"
        );
    }

    /// Note-substrate counterpart of
    /// `prepare_add_entity_rejects_reserved_secret_gate_property`.
    #[tokio::test]
    async fn prepare_add_note_rejects_reserved_secret_gate_property() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");

        let err = prepare_add_note(
            &runtime,
            &token,
            &json!({
                "kind": "observation",
                "content": "a note carrying a reserved property key",
                "properties": {"khive:secret_gate": "exempted:content-sha256-manifest-v1"},
            }),
        )
        .await
        .expect_err("a caller-supplied reserved key on a new note must be rejected");
        assert!(
            matches!(err, RuntimeError::InvalidInput(ref msg) if msg.contains("khive:secret_gate") && msg.contains("runtime-owned")),
            "expected a reservation rejection, got: {err:?}"
        );
    }

    #[tokio::test]
    async fn atomic_proposal_vectors_materialize_only_after_successful_commit() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let vec_store = runtime
            .vectors_for_model(&token, STUB_MODEL)
            .expect("vec store");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "ProposalPlanLinkA");
        let b = khive_storage::Entity::new("local", "concept", "ProposalPlanLinkB");
        let (a_id, b_id) = (a.id, b.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(b).await.expect("seed b");

        let add_entity_plan = prepare_add_entity(
            &runtime,
            &token,
            &json!({"kind": "concept", "name": "ProposalPlanNewEntity", "description": "created atomically"}),
        )
        .await
        .expect("prepare add_entity");
        let link_plan = prepare_link(
            &runtime,
            &token,
            &json!({"source_id": a_id.to_string(), "target_id": b_id.to_string(), "relation": "extends"}),
        )
        .await
        .expect("prepare link");
        let add_note_plan = prepare_add_note(
            &runtime,
            &token,
            &json!({"kind": "observation", "content": "created atomically alongside the entity"}),
        )
        .await
        .expect("prepare add_note");

        let entity_id = match &add_entity_plan {
            AtomicOpPlan::AddEntity(p) => p.entity_id,
            other => panic!("expected an AddEntity plan, got {other:?}"),
        };
        let note_id = match &add_note_plan {
            AtomicOpPlan::AddNote(p) => p.note_id,
            other => panic!("expected an AddNote plan, got {other:?}"),
        };

        let outcome = crate::atomic_runner::run_atomic_unit(
            runtime.sql().as_ref(),
            vec![add_entity_plan, link_plan, add_note_plan],
        )
        .await
        .expect("seam call ok");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected the whole unit to commit: {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[
                PostCommitEffect::ReindexEntity { entity_id },
                PostCommitEffect::ReindexNote {
                    note_id,
                    version: 1
                },
            ],
            "prepare-derived effects must reach the committed token unchanged"
        );
        assert_eq!(
            vec_store.count().await.expect("count before effects"),
            0,
            "commit returns deferred effects without materializing vectors"
        );
        let entity = runtime
            .entities(&token)
            .expect("entities store")
            .get_entity(entity_id)
            .await
            .expect("get_entity")
            .expect("entity must exist after commit");
        assert_eq!(entity.name, "ProposalPlanNewEntity");
        assert!(
            runtime
                .text(&token)
                .expect("text store")
                .get_document("local", entity_id)
                .await
                .expect("get_document")
                .is_some(),
            "entity's FTS document must exist after commit"
        );

        let (edge_count, _, _, edge_deleted_at) =
            probe_edge_natural_key(&runtime, "local", a_id, b_id, "extends").await;
        assert_eq!(
            edge_count, 1,
            "the edge must be committed alongside the entity/note"
        );
        assert!(edge_deleted_at.is_none());

        let note = runtime
            .notes(&token)
            .expect("notes store")
            .get_note(note_id)
            .await
            .expect("get_note")
            .expect("note must exist after commit");
        assert_eq!(note.content, "created atomically alongside the entity");
        assert!(
            runtime
                .text_for_notes(&token)
                .expect("text store")
                .get_document("local", note_id)
                .await
                .expect("get_document")
                .is_some(),
            "note's FTS document must exist after commit"
        );

        apply_post_commit_effects(&runtime, &token, post_commit)
            .await
            .expect("apply post-commit effects");

        assert_eq!(
            vec_store.count().await.expect("count after"),
            2,
            "post-commit reindex must have embedded both the new entity and the new note"
        );
    }

    #[tokio::test]
    async fn atomic_proposal_abort_leaves_zero_vector_rows() {
        let runtime = scratch_runtime();
        runtime.register_embedder(StubProvider);
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let vec_store = runtime
            .vectors_for_model(&token, STUB_MODEL)
            .expect("vec store");
        let entities = runtime.entities(&token).expect("entities store");
        let a = khive_storage::Entity::new("local", "concept", "ProposalPlanRollbackA");
        let x = khive_storage::Entity::new("local", "concept", "ProposalPlanRollbackX");
        let (a_id, x_id) = (a.id, x.id);
        entities.upsert_entity(a).await.expect("seed a");
        entities.upsert_entity(x.clone()).await.expect("seed x");

        let add_entity_plan = prepare_add_entity(
            &runtime,
            &token,
            &json!({"kind": "concept", "name": "ProposalPlanRollbackNewEntity"}),
        )
        .await
        .expect("prepare add_entity");
        let add_note_plan = prepare_add_note(
            &runtime,
            &token,
            &json!({"kind": "observation", "content": "must not survive the rollback"}),
        )
        .await
        .expect("prepare add_note");
        let delete_plan = prepare_delete(
            &runtime,
            &token,
            &json!({"id": x_id.to_string(), "hard": true}),
            None,
        )
        .await
        .expect("prepare delete x");
        // Prepare sees x before the transaction; the guarded link must detect
        // that the preceding hard delete removed it inside the transaction.
        let link_plan = prepare_link(
            &runtime,
            &token,
            &json!({"source_id": a_id.to_string(), "target_id": x_id.to_string(), "relation": "extends"}),
        )
        .await
        .expect("prepare link (endpoint still exists at prepare time)");

        let entity_id = match &add_entity_plan {
            AtomicOpPlan::AddEntity(p) => p.entity_id,
            other => panic!("expected an AddEntity plan, got {other:?}"),
        };
        let note_id = match &add_note_plan {
            AtomicOpPlan::AddNote(p) => p.note_id,
            other => panic!("expected an AddNote plan, got {other:?}"),
        };

        let outcome = crate::atomic_runner::run_atomic_unit(
            runtime.sql().as_ref(),
            vec![add_entity_plan, add_note_plan, delete_plan, link_plan],
        )
        .await
        .expect("the seam call itself must not error; the unit rolls back cleanly");
        match outcome {
            crate::atomic_runner::AtomicRunOutcome::RolledBack {
                failed_op_index, ..
            } => {
                assert_eq!(
                    failed_op_index, 3,
                    "the trailing link (index 3) must be the op whose guard fails"
                );
            }
            other => panic!("expected the whole unit to roll back, got {other:?}"),
        }

        assert_eq!(
            vec_store
                .count()
                .await
                .expect("vector count after rollback"),
            0,
            "a rolled-back atomic apply must not materialize vectors"
        );

        assert!(
            runtime
                .get_entity_including_deleted(&token, entity_id)
                .await
                .expect("get_entity_including_deleted")
                .is_none(),
            "the new entity must leave zero trace after rollback"
        );
        assert!(
            runtime
                .text(&token)
                .expect("text store")
                .get_document("local", entity_id)
                .await
                .expect("get_document")
                .is_none(),
            "the new entity's FTS document must leave zero trace after rollback"
        );
        assert!(
            runtime
                .get_note_including_deleted(&token, note_id)
                .await
                .expect("get_note_including_deleted")
                .is_none(),
            "the new note must leave zero trace after rollback"
        );
        assert!(
            runtime
                .text_for_notes(&token)
                .expect("text store")
                .get_document("local", note_id)
                .await
                .expect("get_document")
                .is_none(),
            "the new note's FTS document must leave zero trace after rollback"
        );

        let x_after = runtime
            .get_entity_including_deleted(&token, x_id)
            .await
            .expect("get_entity_including_deleted")
            .expect("x must still be present because its delete rolled back too");
        assert!(
            x_after.deleted_at.is_none(),
            "x's delete must have rolled back along with the failed link"
        );

        let (edge_count, _, _, _) =
            probe_edge_natural_key(&runtime, "local", a_id, x_id, "extends").await;
        assert_eq!(edge_count, 0, "no edge may have been committed");
    }

    /// ADR-014 tri-state on the atomic path: `entity_type: null` must
    /// explicitly CLEAR a stored entity type (and reindex), not collapse to
    /// "unchanged" like the old `optional_create_string` did.
    #[tokio::test]
    async fn atomic_update_entity_type_null_clears_stored_type() {
        let runtime = scratch_runtime();
        runtime.install_entity_type_validator(std::sync::Arc::new(|kind, entity_type| {
            let Some(raw) = entity_type else {
                return Ok(None);
            };
            let normalized = raw.trim().to_ascii_lowercase();
            if kind == "concept" && normalized == "algorithm" {
                Ok(Some(normalized))
            } else {
                Err(RuntimeError::InvalidInput(format!(
                    "unknown entity_type {raw:?} for {kind:?}; valid: algorithm"
                )))
            }
        }));
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let mut entity = khive_storage::Entity::new("local", "concept", "AtomicNullClear");
        entity.entity_type = Some("algorithm".to_string());
        let entity_id = entity.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(entity)
            .await
            .expect("seed entity");

        let plan = prepare_update(
            &runtime,
            &token,
            &json!({"id": entity_id.to_string(), "entity_type": null}),
            None,
        )
        .await
        .expect("atomic prepare must accept entity_type: null");
        let outcome = crate::atomic_runner::run_atomic_unit(runtime.sql().as_ref(), vec![plan])
            .await
            .expect("atomic update must run");
        let post_commit = match outcome {
            crate::atomic_runner::AtomicRunOutcome::Committed { post_commit } => post_commit,
            other => panic!("expected Committed, got {other:?}"),
        };
        assert_eq!(
            post_commit.as_slice(),
            &[PostCommitEffect::ReindexEntity { entity_id }],
            "a type clear that differs from the prior value must reindex"
        );

        let updated = runtime
            .get_entity(&token, entity_id)
            .await
            .expect("read updated entity");
        assert_eq!(
            updated.entity_type, None,
            "entity_type: null must clear the stored type"
        );
        assert_eq!(updated.name, "AtomicNullClear");
    }

    /// ADR-014 tri-state on the atomic path: a PRESENT `entity_type` key,
    /// including JSON `null`, must be rejected on note and edge targets
    /// (parity with `khive-pack-kg`'s `reject_inapplicable_fields`).
    #[tokio::test]
    async fn atomic_update_null_entity_type_rejected_for_note_and_edge() {
        let runtime = scratch_runtime();
        let token = runtime
            .authorize(Namespace::parse("local").expect("ns"))
            .expect("authorize");
        let note = runtime
            .create_note(
                &token,
                "observation",
                None,
                "note body for entity_type guard",
                Some(0.5),
                None,
                vec![],
            )
            .await
            .expect("create note");

        let note_err = prepare_update(
            &runtime,
            &token,
            &json!({"id": note.id.to_string(), "entity_type": null}),
            Some(crate::atomic_prepare::AtomicUpdateKind::Note { specific: None }),
        )
        .await
        .expect_err("entity_type: null on a note must be rejected");
        assert!(
            matches!(note_err, RuntimeError::InvalidInput(ref msg) if msg.contains("entity_type") && msg.contains("not valid for a note")),
            "expected an InvalidInput naming entity_type for a note, got: {note_err:?}"
        );

        let source = khive_storage::Entity::new("local", "concept", "AtomicNullTypeEdgeSource");
        let target = khive_storage::Entity::new("local", "concept", "AtomicNullTypeEdgeTarget");
        let source_id = source.id;
        let target_id = target.id;
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(source)
            .await
            .expect("seed source");
        runtime
            .entities(&token)
            .expect("entities store")
            .upsert_entity(target)
            .await
            .expect("seed target");
        let edge = runtime
            .link(
                &token,
                source_id,
                target_id,
                "supports".parse().expect("relation"),
                0.5,
                None,
            )
            .await
            .expect("create edge");

        let edge_err = prepare_update(
            &runtime,
            &token,
            &json!({"id": edge.id.to_string(), "entity_type": null}),
            Some(crate::atomic_prepare::AtomicUpdateKind::Edge),
        )
        .await
        .expect_err("entity_type: null on an edge must be rejected");
        assert!(
            matches!(edge_err, RuntimeError::InvalidInput(ref msg) if msg.contains("entity_type") && msg.contains("not valid for an edge")),
            "expected an InvalidInput naming entity_type for an edge, got: {edge_err:?}"
        );
    }
}