khive-runtime 0.10.0

//! Key and revision guards evaluated by the transaction owner.

use khive_storage::{SqlStatement, SqlValue, SqlWriter, StorageError};
use khive_types::{Details, KhiveError};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::{KhiveRuntime, NamespaceToken, RuntimeError, RuntimeResult};

/// Test-only pause point between a keyed singleton create's preparation
/// (annotation-target validation and embedding-model resolution, both
/// outside the writer) and its writer-transaction admission, so a race
/// between two independent callers of `create_note_with_options`, or an
/// independent writer proceeding while one caller's preparation is parked,
/// can be reproduced deterministically (ADR-172 Amendment 6, acceptance
/// items 3 and 5) instead of relying on scheduler luck or sleeps. A no-op
/// unless the calling task runs inside `AFTER_PREPARE_BARRIER.scope(...)`;
/// production code never establishes that scope, so `pause_after_prepare`
/// costs nothing outside these regression tests, and it does not exist at
/// all in non-test builds. Deliberately a separate barrier from
/// `crate::curation::race_seam`: that module's pause point is a different
/// production entry point's read/write boundary, and sharing one
/// `Barrier` between two unrelated pause points would make a test of one
/// silently depend on the party count of the other.
#[cfg(test)]
pub(crate) mod race_seam {
    use std::sync::Arc;
    use tokio::sync::Barrier;

    tokio::task_local! {
        pub(crate) static AFTER_PREPARE_BARRIER: Arc<Barrier>;
    }

    /// Meets the other party twice: the first meeting tells it this caller
    /// has finished its initial holder check and its preparation, the
    /// second is the release. Two callers sharing one barrier simply meet
    /// twice.
    pub(crate) async fn pause_after_prepare() {
        if let Ok(barrier) = AFTER_PREPARE_BARRIER.try_with(Arc::clone) {
            barrier.wait().await;
            barrier.wait().await;
        }
    }
}

/// Each fence adds a keyed read while holding the writer, like a batch observation.
pub const MAX_NOTE_FENCES: usize = 100;

#[derive(Clone, Debug, Serialize)]
pub struct NoteFence {
    pub key: String,
    pub kind: String,
    /// None asserts absence; the JSON field is still required.
    pub expected_version: Option<i64>,
    /// Dotted document path whose RFC 3339 value must exceed the writer clock,
    /// on the same semantics a batch observation's `live_until` carries. A
    /// version comparison alone cannot express live ownership: a lease that
    /// merely ran out changes no document and moves no version, so the version
    /// still matches and the write commits.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub live_until: Option<String>,
    /// The note that must hold this key, on the same semantics a batch
    /// observation's `id` carries. A recreated note starts at version 1, so a
    /// version comparison alone cannot tell the note the caller read from a
    /// different note sitting at the same number.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<Uuid>,
}

impl<'de> Deserialize<'de> for NoteFence {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        #[serde(deny_unknown_fields)]
        struct Fields {
            key: String,
            kind: String,
            #[serde(alias = "version")]
            expected_version: Option<i64>,
            live_until: Option<String>,
            id: Option<Uuid>,
        }

        let value = serde_json::Value::deserialize(deserializer)?;
        // Option alone treats omission as null. Absence must be asserted explicitly.
        if value.get("expected_version").is_none() && value.get("version").is_none() {
            return Err(serde::de::Error::custom(
                "fence requires expected_version (positive integer or null)",
            ));
        }
        let fields = Fields::deserialize(value).map_err(|error| {
            serde::de::Error::custom(format!(
                "fence requires expected_version (positive integer or null): {error}"
            ))
        })?;
        Ok(Self {
            key: fields.key,
            kind: fields.kind,
            expected_version: fields.expected_version,
            live_until: fields.live_until,
            id: fields.id,
        })
    }
}

impl NoteFence {
    pub fn validate(&self) -> RuntimeResult<()> {
        crate::keyed_memory::validate_memory_key(&self.key)?;
        if self.kind.is_empty() {
            return Err(RuntimeError::InvalidInput(
                "fence requires a note kind".into(),
            ));
        }
        if self.expected_version.is_some_and(|version| version < 1) {
            return Err(RuntimeError::InvalidInput(
                "fence requires expected_version (positive integer or null)".into(),
            ));
        }
        if self.live_until.is_some() && self.expected_version.is_none() {
            // An absence assertion has no document, so there is nothing to read
            // a deadline out of. The batch route refuses the same pairing.
            return Err(RuntimeError::InvalidInput(
                "fence live_until requires a positive expected_version".into(),
            ));
        }
        if self.live_until.as_ref().is_some_and(|path| path.is_empty()) {
            return Err(RuntimeError::InvalidInput(
                "fence live_until requires a document path".into(),
            ));
        }
        if self.id.is_some() && self.expected_version.is_none() {
            // An absence assertion has no note, so there is nothing for an
            // identity to name. The batch route refuses the same pairing.
            return Err(RuntimeError::InvalidInput(
                "fence id requires a positive expected_version".into(),
            ));
        }
        Ok(())
    }
}

/// An object retains the original refusal shape; a list identifies its failing entry.
#[derive(Clone, Debug, Serialize)]
#[serde(untagged)]
pub enum NoteFences {
    One(NoteFence),
    Many(Vec<NoteFence>),
}

impl From<NoteFence> for NoteFences {
    fn from(fence: NoteFence) -> Self {
        Self::One(fence)
    }
}

impl NoteFences {
    fn validate_count(count: usize) -> RuntimeResult<()> {
        if count > MAX_NOTE_FENCES {
            return Err(RuntimeError::InvalidInput(format!(
                "fence list admits at most {MAX_NOTE_FENCES} entries; this call sent {count}: each fence is a read taken while holding the writer"
            )));
        }
        Ok(())
    }

    pub fn entries(&self) -> &[NoteFence] {
        match self {
            Self::One(fence) => std::slice::from_ref(fence),
            Self::Many(fences) => fences,
        }
    }

    pub fn validate(&self) -> RuntimeResult<()> {
        Self::validate_count(self.entries().len())?;
        if self.entries().is_empty() {
            return Err(RuntimeError::InvalidInput(
                "fence list must not be empty".into(),
            ));
        }
        let mut seen = std::collections::HashMap::new();
        for (index, fence) in self.entries().iter().enumerate() {
            fence.validate()?;
            if let Some(first) = seen.insert((&fence.kind, &fence.key), index) {
                return Err(RuntimeError::InvalidInput(format!(
                    "duplicate fence (kind, key) at indices {first} and {index}"
                )));
            }
        }
        Ok(())
    }
}

impl<'de> Deserialize<'de> for NoteFences {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let value = serde_json::Value::deserialize(deserializer)?;
        let fences = match value {
            serde_json::Value::Object(_) => {
                Self::One(NoteFence::deserialize(value).map_err(serde::de::Error::custom)?)
            }
            serde_json::Value::Array(entries) => {
                Self::validate_count(entries.len()).map_err(serde::de::Error::custom)?;
                let mut fences = Vec::with_capacity(entries.len());
                for (index, entry) in entries.into_iter().enumerate() {
                    fences.push(NoteFence::deserialize(entry).map_err(|error| {
                        serde::de::Error::custom(format!("fence entry {index}: {error}"))
                    })?);
                }
                Self::Many(fences)
            }
            _ => {
                return Err(serde::de::Error::custom(
                    "fence requires an object or a non-empty list of objects with key, kind and expected_version (positive integer or null)",
                ));
            }
        };
        fences.validate().map_err(serde::de::Error::custom)?;
        Ok(fences)
    }
}

/// Missing optional fences default to None; explicitly supplied null is invalid.
pub fn deserialize_optional_fences<'de, D: serde::Deserializer<'de>>(
    deserializer: D,
) -> Result<Option<NoteFences>, D::Error> {
    NoteFences::deserialize(deserializer).map(Some)
}

#[derive(Clone, Debug, Default)]
pub struct NoteWriteOptions {
    pub key: Option<String>,
    pub expected_version: Option<i64>,
    pub fence: Option<NoteFences>,
    pub embed: Option<bool>,
}

impl NoteWriteOptions {
    pub fn validate(&self) -> RuntimeResult<()> {
        if let Some(key) = &self.key {
            crate::keyed_memory::validate_memory_key(key)?;
        }
        if self.expected_version.is_some_and(|version| version < 1) {
            return Err(RuntimeError::InvalidInput(
                "expected_version must be positive".into(),
            ));
        }
        if let Some(fence) = &self.fence {
            fence.validate()?;
        }
        Ok(())
    }
}

/// The candidate identity and fully-validated payload a keyed singleton
/// `create` is claiming: kind and key resolve the live holder, and content
/// plus properties let [`NoteWriteGuard::classify_refusal`] decide, inside
/// the writer transaction, whether a live holder is this same create
/// replayed (ADR-172 Amendment 6) or a genuine conflict.
#[derive(Clone, Debug)]
pub(crate) struct CreateKeyClaim {
    pub kind: String,
    pub key: String,
    pub content: String,
    pub properties: Option<serde_json::Value>,
    /// Set only by the singleton `create` route (`AtomicNoteOptions::replay_receipt`).
    /// Every other keyed-write caller leaves this `false`, so
    /// [`NoteWriteGuard::classify_refusal`] never reports a comparison
    /// outcome for their conflicts.
    pub replay_signal: bool,
}

#[derive(Clone, Debug)]
pub(crate) struct NoteWriteGuard {
    pub namespace: String,
    pub target_id: Uuid,
    pub expected_version: Option<i64>,
    pub fence: Option<NoteFences>,
    pub create_key: Option<CreateKeyClaim>,
}

#[derive(Clone, Debug)]
pub(crate) struct NoteVectors {
    namespace: String,
    subject_id: Uuid,
}

impl NoteVectors {
    pub(crate) fn new(namespace: String, subject_id: Uuid) -> Self {
        Self {
            namespace,
            subject_id,
        }
    }

    async fn tables(writer: &mut dyn SqlWriter) -> Result<Vec<String>, StorageError> {
        // vec_* is the backend's reserved vector-table namespace. Catalog
        // types exclude sqlite-vec shadow tables without guessing suffixes.
        let tables = writer
            .query_all(statement(
                "SELECT name FROM pragma_table_list \
             WHERE schema='main' AND type='virtual' AND name GLOB 'vec_*' ORDER BY name",
                vec![],
            ))
            .await?;
        let mut names = Vec::with_capacity(tables.len());
        for row in tables {
            let Some(SqlValue::Text(table)) = row.get("name") else {
                return Err(StorageError::Internal(
                    "invalid vector table catalog row".into(),
                ));
            };
            if !table.strip_prefix("vec_").is_some_and(|key| {
                !key.is_empty() && key.bytes().all(|c| c.is_ascii_alphanumeric() || c == b'_')
            }) {
                return Err(StorageError::Internal(
                    "invalid persisted vector table name".into(),
                ));
            }
            names.push(table.clone());
        }
        Ok(names)
    }

    pub(crate) async fn has_rows(&self, writer: &mut dyn SqlWriter) -> Result<bool, StorageError> {
        for table in Self::tables(writer).await? {
            if writer
                .query_scalar(statement(
                    format!(
                        "SELECT 1 FROM main.{table} WHERE namespace=?1 AND subject_id=?2 LIMIT 1"
                    ),
                    vec![
                        SqlValue::Text(self.namespace.clone()),
                        SqlValue::Text(self.subject_id.to_string()),
                    ],
                ))
                .await?
                .is_some()
            {
                return Ok(true);
            }
        }
        Ok(false)
    }

    pub(crate) async fn apply(&self, writer: &mut dyn SqlWriter) -> Result<(), StorageError> {
        for table in Self::tables(writer).await? {
            let scope = vec![
                SqlValue::Text(self.namespace.clone()),
                SqlValue::Text(self.subject_id.to_string()),
            ];
            writer
                .execute(statement(
                    format!(
                "INSERT INTO ann_write_log (namespace,embedding_model,kind,field,subject_id,op) \
                 SELECT namespace,embedding_model,kind,field,subject_id,'delete' \
                 FROM main.{table} WHERE namespace=?1 AND subject_id=?2"),
                    scope.clone(),
                ))
                .await?;
            writer
                .execute(statement(
                    format!("DELETE FROM main.{table} WHERE namespace=?1 AND subject_id=?2"),
                    scope,
                ))
                .await?;
            let model_key = table.strip_prefix("vec_").ok_or_else(|| {
                StorageError::Internal("invalid persisted vector table name".into())
            })?;
            writer
                .execute(statement(
                    "DELETE FROM vector_provenance \
                     WHERE model_key=?1 AND namespace=?2 AND subject_id=?3",
                    vec![
                        SqlValue::Text(model_key.to_string()),
                        SqlValue::Text(self.namespace.clone()),
                        SqlValue::Text(self.subject_id.to_string()),
                    ],
                ))
                .await?;
        }
        Ok(())
    }
}

#[derive(Clone, Debug)]
pub(crate) struct NoteEmbeddingInheritance {
    pub vectors: NoteVectors,
    pub kind: String,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub enum NoteWriteConflict {
    Version {
        expected: i64,
        current: i64,
    },
    Fence {
        key: String,
        expected: Option<i64>,
        current: Option<i64>,
        index: Option<usize>,
    },
    Key {
        key: String,
        existing_id: String,
        /// Whether the live holder's validated content and properties equal
        /// the candidate's (ADR-172 Amendment 6). `None` unless the write
        /// that produced this conflict asked for the comparison
        /// (`CreateKeyClaim::replay_signal`, set only by the singleton
        /// `create` route): every other keyed route never carries this
        /// signal, so it can never reach their callers. Internal signal
        /// only even when present: it never reaches a client verbatim. The
        /// create handler turns an equal, disclosed conflict into a
        /// successful replay result, and rebuilds `Details` without this
        /// field on every other outcome.
        equal: Option<bool>,
    },
    /// A fence whose `live_until` path did not admit the write. Separate from
    /// `Fence` because the version matched: what failed is the deadline. Boxed
    /// so this variant does not set the size of every conflict result.
    FenceDeadline(Box<FenceDeadline>),
    /// A fence whose `id` did not hold the key. Separate from `Fence` because
    /// the version is not what disagreed, and it may well agree: a recreated
    /// note starts at version 1. Boxed for the same reason as `FenceDeadline`.
    FenceIdentity(Box<FenceIdentity>),
}

/// The fields an identity refusal reports: what was fenced, at which version,
/// and both identities.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FenceIdentity {
    pub key: String,
    pub kind: String,
    pub version: i64,
    pub index: Option<usize>,
    pub evidence: Vec<(&'static str, String)>,
}

/// The fields a deadline refusal reports: what was fenced, at which version and
/// path, and the predicate's own reason and evidence.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FenceDeadline {
    pub key: String,
    pub kind: String,
    pub version: i64,
    pub field: String,
    pub index: Option<usize>,
    pub reason: &'static str,
    pub evidence: Vec<(&'static str, String)>,
}

impl NoteWriteConflict {
    pub fn into_error(self) -> KhiveError {
        self.into_error_at_member(None)
    }

    pub(crate) fn into_error_at_member(self, member: Option<usize>) -> KhiveError {
        let (message, mut details) = match self {
            Self::Version { expected, current } => (
                "note version precondition failed",
                vec![
                    ("reason", "version_conflict".into()),
                    ("expected_version", expected.to_string()),
                    ("current_version", current.to_string()),
                ],
            ),
            Self::Fence {
                key,
                expected,
                current,
                index,
            } => {
                let mut fields = vec![
                    ("reason", "fence_conflict".into()),
                    ("key", key),
                    (
                        "expected_version",
                        expected.map_or_else(|| "absent".into(), |version| version.to_string()),
                    ),
                ];
                if let Some(current) = current {
                    fields.push(("current_version", current.to_string()));
                }
                if let Some(index) = index {
                    fields.push(("index", index.to_string()));
                }
                ("note fence precondition failed", fields)
            }
            Self::Key {
                key,
                existing_id,
                equal,
            } => {
                let mut fields = vec![
                    ("reason", "key_conflict".into()),
                    ("key", key),
                    ("existing_id", existing_id),
                ];
                if let Some(equal) = equal {
                    fields.push(("equal", equal.to_string()));
                }
                ("a live note already holds this key", fields)
            }
            Self::FenceDeadline(deadline) => {
                let FenceDeadline {
                    key,
                    kind,
                    version,
                    field,
                    index,
                    reason,
                    evidence,
                } = *deadline;
                let mut fields = vec![
                    ("reason", reason.into()),
                    ("key", key),
                    ("kind", kind),
                    ("version", version.to_string()),
                    ("field", field),
                ];
                if let Some(index) = index {
                    fields.push(("index", index.to_string()));
                }
                fields.extend(evidence);
                ("note fence time precondition failed", fields)
            }
            Self::FenceIdentity(identity) => {
                let FenceIdentity {
                    key,
                    kind,
                    version,
                    index,
                    evidence,
                } = *identity;
                let mut fields = vec![
                    ("reason", "identity_conflict".into()),
                    ("key", key),
                    ("kind", kind),
                    ("version", version.to_string()),
                ];
                fields.extend(evidence);
                if let Some(index) = index {
                    fields.push(("index", index.to_string()));
                }
                ("note fence identity precondition failed", fields)
            }
        };
        if let Some(member) = member {
            details.push(("member", member.to_string()));
        }
        KhiveError::conflict(message).with_details(Details::new_owned(details))
    }
}

pub(crate) fn statement(sql: impl Into<String>, params: Vec<SqlValue>) -> SqlStatement {
    SqlStatement {
        sql: sql.into(),
        params,
        label: Some("note-write-guard".into()),
    }
}

impl NoteWriteGuard {
    pub(crate) async fn check_fence(
        &self,
        writer: &mut dyn SqlWriter,
    ) -> Result<Option<NoteWriteConflict>, StorageError> {
        let Some(fences) = &self.fence else {
            return Ok(None);
        };
        // One clock reading for this transaction, taken only when some entry
        // asks for one, and shared by all of them: two entries in one write
        // must not be judged against two instants.
        let mut now: Option<i64> = None;
        for (index, fence) in fences.entries().iter().enumerate() {
            let holder = crate::fence_identity::read_holder(
                writer,
                &self.namespace,
                &fence.kind,
                &fence.key,
                "note-write-guard",
            )
            .await?;
            // Identity first, then version, then the deadline: the batch route
            // orders them this way, and a caller must not learn a different
            // failure for the same state depending on which route it used. An
            // absent holder has no identity to name, so it falls through to the
            // version comparison that already reports it.
            if let (Some(asserted), Some(holder)) = (fence.id, holder.as_ref()) {
                if asserted != holder.id {
                    return Ok(Some(NoteWriteConflict::FenceIdentity(Box::new(
                        FenceIdentity {
                            key: fence.key.clone(),
                            kind: fence.kind.clone(),
                            // validate() refuses id without a positive
                            // expected_version, so the version is present here.
                            version: fence.expected_version.unwrap_or_default(),
                            index: matches!(fences, NoteFences::Many(_)).then_some(index),
                            evidence: crate::fence_identity::identity_evidence(asserted, holder.id),
                        },
                    ))));
                }
            }
            let current = holder.map(|holder| holder.version);
            if current != fence.expected_version {
                return Ok(Some(NoteWriteConflict::Fence {
                    key: fence.key.clone(),
                    expected: fence.expected_version,
                    current,
                    index: matches!(fences, NoteFences::Many(_)).then_some(index),
                }));
            }
            if let Some(field) = &fence.live_until {
                let clock = match now {
                    Some(clock) => clock,
                    None => {
                        let clock =
                            crate::live_until::writer_clock(writer, "note-write-guard-clock")
                                .await?;
                        now = Some(clock);
                        clock
                    }
                };
                if let Some(refusal) = crate::live_until::evaluate(
                    writer,
                    &self.namespace,
                    &fence.kind,
                    &fence.key,
                    field,
                    clock,
                    "note-write-guard-live-until",
                )
                .await?
                {
                    return Ok(Some(NoteWriteConflict::FenceDeadline(Box::new(
                        FenceDeadline {
                            key: fence.key.clone(),
                            kind: fence.kind.clone(),
                            // validate() refuses live_until without a positive
                            // expected_version, so the version is present here.
                            version: fence.expected_version.unwrap_or_default(),
                            field: field.clone(),
                            index: matches!(fences, NoteFences::Many(_)).then_some(index),
                            reason: refusal.reason(),
                            evidence: refusal.details(),
                        },
                    ))));
                }
            }
        }
        Ok(None)
    }

    pub(crate) async fn classify_refusal(
        &self,
        writer: &mut dyn SqlWriter,
    ) -> Result<Option<NoteWriteConflict>, StorageError> {
        if let Some(claim) = &self.create_key {
            // ADR-172 Amendment 6: the final holder comparison happens here,
            // inside the writer transaction, reading the same row the failed
            // guarded INSERT just contended on. Never a later, separate
            // lookup after the transaction has settled.
            let holder = writer
                .query_row(statement(
                    "SELECT id, content, properties FROM notes \
                     WHERE namespace=?1 AND kind=?2 AND key=?3 AND deleted_at IS NULL",
                    vec![
                        SqlValue::Text(self.namespace.clone()),
                        SqlValue::Text(claim.kind.clone()),
                        SqlValue::Text(claim.key.clone()),
                    ],
                ))
                .await?;
            if let Some(row) = holder {
                let existing_id = match row.get("id") {
                    Some(SqlValue::Text(id)) => id.clone(),
                    _ => {
                        return Err(StorageError::Internal(
                            "invalid keyed note holder identity".into(),
                        ))
                    }
                };
                let existing_content = match row.get("content") {
                    Some(SqlValue::Text(content)) => content.clone(),
                    _ => {
                        return Err(StorageError::Internal(
                            "invalid keyed note holder content".into(),
                        ))
                    }
                };
                let existing_properties: Option<serde_json::Value> = match row.get("properties") {
                    Some(SqlValue::Text(text)) => {
                        Some(serde_json::from_str(text).map_err(|_| {
                            StorageError::Internal("invalid keyed note holder properties".into())
                        })?)
                    }
                    Some(SqlValue::Null) | None => None,
                    _ => {
                        return Err(StorageError::Internal(
                            "invalid keyed note holder properties".into(),
                        ))
                    }
                };
                // Decoded content is byte-exact text comparison; properties
                // use typed JSON equality (object member order irrelevant,
                // member presence/array order/value types significant,
                // `None` distinct from `Some(Value::Object(empty))`).
                // `serde_json::Value`'s own `PartialEq` already has exactly
                // this shape (an `IndexMap`/`Map` compares order-insensitively,
                // a `Vec` compares order-sensitively, `Number` compares its
                // typed representation without coercion), so no extra
                // normalization runs here. Only reported back when the
                // caller asked for it: every route besides singleton
                // `create` leaves `replay_signal` false and gets `None`,
                // which `into_error_at_member` renders as no `equal` key at
                // all in the returned Details.
                let equal = claim.replay_signal.then(|| {
                    existing_content == claim.content && existing_properties == claim.properties
                });
                return Ok(Some(NoteWriteConflict::Key {
                    key: claim.key.clone(),
                    existing_id,
                    equal,
                }));
            }
        }
        if let Some(expected) = self.expected_version {
            let current = writer
                .query_scalar(statement(
                    "SELECT version FROM notes WHERE id=?1 AND deleted_at IS NULL",
                    vec![SqlValue::Text(self.target_id.to_string())],
                ))
                .await?;
            if let Some(SqlValue::Integer(current)) = current {
                if current != expected {
                    return Ok(Some(NoteWriteConflict::Version { expected, current }));
                }
            }
        }
        Ok(None)
    }
}

/// The ADR-172 Amendment 6 authoritative holder check, run with no insert
/// statement of its own: [`NoteWriteGuard::check_fence`] then
/// [`NoteWriteGuard::classify_refusal`], inside their own single writer
/// checkout ([`crate::atomic_runner::run_prepared_atomic_unit`], the same
/// "one `atomic_unit`" seam [`crate::atomic_runner::apply_plan`] uses for the
/// real guarded insert). `create_note_with_options` calls this twice for a
/// keyed create: once before preparation (creation-only annotation
/// resolution and embedding work must not run ahead of this decision), and
/// once more, only when preparation itself failed, to revalidate before
/// falling back to that failure. `Ok(None)` means fences passed and no live
/// holder exists yet; `Ok(Some(conflict))` is the caller's answer (already
/// in the same shape a failed guarded insert reports): replay, key
/// conflict, or a stale fence.
async fn check_keyed_create_holder(
    access: &dyn khive_storage::SqlAccess,
    guard: NoteWriteGuard,
) -> RuntimeResult<Option<NoteWriteConflict>> {
    use crate::atomic_runner::{
        run_prepared_atomic_unit, PreparedAtomicError, PreparedAtomicOp, PreparedAtomicOutcome,
    };
    let op: PreparedAtomicOp<(), NoteWriteConflict> = Box::new(move |writer| {
        Box::pin(async move {
            if let Some(conflict) = guard.check_fence(writer).await? {
                return Err(PreparedAtomicError::Refused {
                    failure: conflict,
                    message: "keyed create: fence stale at holder check".into(),
                });
            }
            if let Some(conflict) = guard.classify_refusal(writer).await? {
                return Err(PreparedAtomicError::Refused {
                    failure: conflict,
                    message: "keyed create: live holder found at holder check".into(),
                });
            }
            Ok(((), Vec::new()))
        })
    });
    match run_prepared_atomic_unit(access, op).await? {
        PreparedAtomicOutcome::Committed { .. } => Ok(None),
        PreparedAtomicOutcome::RolledBack(conflict) => Ok(Some(conflict)),
    }
}

pub(crate) fn validate_head(note: &khive_storage::note::Note) -> RuntimeResult<()> {
    if note.kind != "head" {
        return Ok(());
    }
    if note.name.is_some() {
        return Err(RuntimeError::InvalidInput("head notes have no name".into()));
    }
    serde_json::from_str::<serde_json::Value>(&note.content).map_err(|error| {
        RuntimeError::InvalidInput(format!("head content must be JSON text: {error}"))
    })?;
    if let Some(tags) = note
        .properties
        .as_ref()
        .and_then(|p| p.get("tags"))
        .and_then(|t| t.as_array())
    {
        for tag in tags.iter().filter_map(|tag| tag.as_str()) {
            if let Some(kind) = tag.strip_prefix("kind:") {
                if kind.len() > 64 || kind.contains('\0') {
                    return Err(RuntimeError::InvalidInput(
                        "head document kind must be at most 64 bytes without U+0000".into(),
                    ));
                }
            }
        }
    }
    Ok(())
}

impl KhiveRuntime {
    pub async fn get_note_by_key(
        &self,
        token: &NamespaceToken,
        key: &str,
        kind: Option<&str>,
        after_key: bool,
    ) -> RuntimeResult<khive_storage::note::Note> {
        self.get_note_by_key_in_scope(token, key, kind, after_key, None)
            .await
    }

    /// [`Self::get_note_by_key`] counting only the rows `mailbox` admits, so
    /// a message the caller's mailbox hides answers as a missing key and
    /// never enters the ambiguity count.
    pub async fn get_note_by_key_in_scope(
        &self,
        token: &NamespaceToken,
        key: &str,
        kind: Option<&str>,
        after_key: bool,
        mailbox: Option<&khive_storage::note::NoteMailboxScope>,
    ) -> RuntimeResult<khive_storage::note::Note> {
        crate::keyed_memory::validate_memory_key(key)?;
        let mut matches = self
            .notes(token)?
            .get_live_notes_by_key(token.namespace().as_str(), key, kind)
            .await?;
        if let Some(scope) = mailbox {
            matches.retain(|note| crate::MailboxView::scope_permits_message_note(scope, note));
        }
        match matches.len() {
            0 => {
                let error = KhiveError::not_found("note key", key);
                Err(if after_key {
                    error.with_details(Details::new_owned([
                        ("reason", "after_key_missing".into()),
                        ("key", key.into()),
                    ]))
                } else {
                    error
                }
                .into())
            }
            1 => Ok(matches.remove(0)),
            _ => {
                let kinds = matches
                    .iter()
                    .map(|note| note.kind.as_str())
                    .collect::<Vec<_>>()
                    .join(",");
                Err(KhiveError::conflict("note key is ambiguous")
                    .with_details(Details::new_owned([
                        ("reason", "key_ambiguous".into()),
                        ("key", key.into()),
                        ("kinds", kinds),
                    ]))
                    .into())
            }
        }
    }

    pub(crate) async fn prepare_versioned_note_update(
        &self,
        token: &NamespaceToken,
        snapshot: khive_storage::note::Note,
        patch: crate::curation::NotePatch,
    ) -> RuntimeResult<(khive_storage::note::Note, crate::atomic_plan::UpdatePlan)> {
        use crate::atomic_plan::{AffectedRowGuard, PlanStatement, PostCommitEffect, UpdatePlan};
        let options = patch.write_options.clone();
        options.validate()?;
        if options.key.is_some() {
            return Err(RuntimeError::InvalidInput("key is immutable".into()));
        }
        if let Some(fences) = &options.fence {
            for fence in fences.entries() {
                self.validate_note_kind(&fence.kind)?;
            }
        }
        let expected_updated_at = snapshot.updated_at;
        let expected_deleted_at = snapshot.deleted_at;
        let expected_snapshot_version = snapshot.version;
        let (mut note, text_changed, changed) = self
            .prepare_update_note_from_snapshot(token, snapshot, patch)
            .await?;
        validate_head(&note)?;
        // ADR-172 Amendment 5: the no-op answer is for unfenced updates only.
        // A write that names `expected_version` is asking for exactly one
        // accepted write at that version, and the version it mints is the only
        // thing a rival can fail against, so an identical fenced patch is still
        // a write. This branch serves `update` and the `stream.batch` write
        // member alike, and a batch write member is always fenced.
        if !changed && options.embed.is_none() && options.expected_version.is_none() {
            let assertion = SqlStatement {
                sql: "SELECT 1 FROM notes WHERE id=?1 AND updated_at=?2 AND deleted_at IS ?3 AND version=?4"
                    .into(),
                params: vec![
                    SqlValue::Text(note.id.to_string()),
                    SqlValue::Integer(expected_updated_at),
                    expected_deleted_at
                        .map(SqlValue::Integer)
                        .unwrap_or(SqlValue::Null),
                    SqlValue::Integer(expected_snapshot_version),
                ],
                label: Some("note-noop-assertion".into()),
            };
            let plan = UpdatePlan {
                graph_effects: Vec::new(),
                target_id: note.id,
                statements: vec![PlanStatement {
                    statement: assertion,
                    guard: Some(AffectedRowGuard::exactly(1)),
                }],
                post_commit: PostCommitEffect::None,
                edge_natural_key: None,
                idempotent_noop: true,
                entity_guard: None,
                note_guard: Some(NoteWriteGuard {
                    namespace: token.namespace().as_str().into(),
                    target_id: note.id,
                    expected_version: options.expected_version,
                    fence: options.fence,
                    create_key: None,
                }),
                note_vector_purge: None,
                note_embedding_inheritance: None,
            };
            return Ok((note, plan));
        }
        if !changed {
            // A fenced write of an identical patch rewrites the row as it is,
            // and the replace statement admits only a strictly newer
            // `updated_at`, so advance the revision here the way a changed
            // patch does (see `prepare_update_note_from_snapshot`).
            let minimum_updated_at = note.updated_at.checked_add(1).ok_or_else(|| {
                RuntimeError::Internal(format!(
                    "note {} updated_at is already at i64::MAX and cannot advance",
                    note.id
                ))
            })?;
            note.updated_at = chrono::Utc::now()
                .timestamp_micros()
                .max(minimum_updated_at);
        }
        let next_version = note
            .version
            .checked_add(1)
            .ok_or_else(|| RuntimeError::InvalidInput("note version exhausted".into()))?;
        note.version = next_version;
        let mut update = if self.stream_member_error(&note).await?.is_some() {
            khive_db::stores::note::note_metadata_replace_if_unchanged_statement(
                &note,
                expected_updated_at,
                expected_deleted_at,
            )
        } else {
            khive_db::stores::note::note_replace_if_unchanged_statement(
                &note,
                expected_updated_at,
                expected_deleted_at,
            )
        };
        // A writer such as gtd.repair may change the row while retaining its
        // updated_at. The snapshot version is therefore part of every CAS,
        // including an unfenced update.
        update
            .params
            .push(SqlValue::Integer(expected_snapshot_version));
        update
            .sql
            .push_str(&format!(" AND version = ?{}", update.params.len()));
        if let Some(version) = options.expected_version {
            update.params.push(SqlValue::Integer(version));
            update
                .sql
                .push_str(&format!(" AND version = ?{}", update.params.len()));
        }
        let mut statements = vec![PlanStatement {
            statement: update,
            guard: Some(AffectedRowGuard::exactly(1)),
        }];
        if text_changed {
            for sql in khive_db::stores::text::delete_document_statements(
                "fts_notes",
                &note.namespace,
                note.id,
            )
            .into_iter()
            .chain(khive_db::stores::text::insert_document_statements(
                "fts_notes",
                &crate::curation::note_fts_document(&note),
            )) {
                statements.push(PlanStatement {
                    statement: sql,
                    guard: None,
                });
            }
        }
        // The note update event, in the same atomic unit as the row it
        // describes, exactly as the entity update plan does. This is the third
        // of the three domain kinds that had no emitter anywhere in the tree.
        statements.extend(crate::atomic_prepare::event_append_statements(
            token,
            &note.namespace,
            "update",
            khive_types::EventKind::NoteUpdated,
            khive_types::SubstrateKind::Note,
            note.id,
            serde_json::json!({
                "id": note.id,
                "namespace": note.namespace,
                "version": note.version,
                "text_changed": text_changed,
            }),
        )?);
        // This is a potential reindex. Inherited membership is resolved by the
        // writer because vector publication can occur without a note revision.
        let post_commit =
            if options.embed != Some(false) && (text_changed || options.embed == Some(true)) {
                PostCommitEffect::ReindexNote {
                    note_id: note.id,
                    version: note.version,
                }
            } else if text_changed || options.embed == Some(false) {
                PostCommitEffect::NoteChanged {
                    note_id: note.id,
                    kind: note.kind.clone(),
                }
            } else {
                PostCommitEffect::None
            };
        let plan = UpdatePlan {
            graph_effects: Vec::new(),
            target_id: note.id,
            statements,
            post_commit,
            edge_natural_key: None,
            idempotent_noop: false,
            entity_guard: None,
            note_guard: Some(NoteWriteGuard {
                namespace: token.namespace().as_str().into(),
                target_id: note.id,
                expected_version: options.expected_version,
                fence: options.fence,
                create_key: None,
            }),
            note_vector_purge: (options.embed == Some(false))
                .then(|| NoteVectors::new(note.namespace.clone(), note.id)),
            note_embedding_inheritance: (text_changed && options.embed.is_none()).then(|| {
                NoteEmbeddingInheritance {
                    vectors: NoteVectors::new(note.namespace.clone(), note.id),
                    kind: note.kind.clone(),
                }
            }),
        };
        Ok((note, plan))
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn create_note_with_options(
        &self,
        token: &NamespaceToken,
        kind: &str,
        name: Option<&str>,
        content: &str,
        embedding_content: Option<&str>,
        salience: Option<f64>,
        decay_factor: Option<f64>,
        properties: Option<serde_json::Value>,
        annotates: Vec<Uuid>,
        embedding_model: Option<&str>,
        options: NoteWriteOptions,
    ) -> RuntimeResult<(
        khive_storage::note::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )> {
        self.create_note_with_options_resolving_annotations(
            token,
            kind,
            name,
            content,
            embedding_content,
            salience,
            decay_factor,
            properties,
            std::future::ready(Ok(annotates)),
            embedding_model,
            options,
        )
        .await
    }

    /// Defer annotation target resolution until a keyed-create holder check
    /// has decided whether this request is an exact replay. Resolution errors
    /// still take part in the final holder recheck before they are returned.
    #[allow(clippy::too_many_arguments)]
    pub async fn create_note_with_options_resolving_annotations<F>(
        &self,
        token: &NamespaceToken,
        kind: &str,
        name: Option<&str>,
        content: &str,
        embedding_content: Option<&str>,
        salience: Option<f64>,
        decay_factor: Option<f64>,
        properties: Option<serde_json::Value>,
        annotations: F,
        embedding_model: Option<&str>,
        options: NoteWriteOptions,
    ) -> RuntimeResult<(
        khive_storage::note::Note,
        crate::retrieval::EmbeddingTruncationReport,
    )>
    where
        F: std::future::Future<Output = RuntimeResult<Vec<Uuid>>> + Send,
    {
        use crate::atomic_message::{AtomicNoteOptions, AtomicNoteSpec};
        use crate::atomic_runner::{run_atomic_unit, AtomicOpFailure, AtomicRunOutcome};
        use crate::note_create::{prepare_note_create, KeyPublication};
        options.validate()?;
        if options.expected_version.is_some() {
            return Err(RuntimeError::InvalidInput(
                "expected_version applies only to update".into(),
            ));
        }
        if let Some(fences) = &options.fence {
            for fence in fences.entries() {
                self.validate_note_kind(&fence.kind)?;
            }
        }
        if let Some(prefix) = embedding_content {
            if prefix.is_empty() || prefix.len() >= content.len() || !content.starts_with(prefix) {
                return Err(RuntimeError::InvalidInput(
                    "embedding_content must be a non-empty proper prefix of content".into(),
                ));
            }
            crate::secret_gate::check_at(prefix, "note", "embedding_content")?;
        }
        let mut candidate =
            khive_storage::note::Note::new(token.namespace().as_str(), kind, content);
        candidate.name = name.map(str::to_owned);
        candidate.properties = properties.clone();
        validate_head(&candidate)?;

        // Unkeyed create: unchanged. There is no key to hold, so there is
        // nothing for an initial holder check to decide, so preparation runs
        // the same way it always has, and the single writer transaction at
        // the end is the only admission decision.
        let Some(key) = options.key.as_deref() else {
            let annotates = annotations.await?;
            let (mut prepared, _) = prepare_note_create(
                self,
                AtomicNoteSpec {
                    token,
                    id: None,
                    kind,
                    name,
                    content,
                    properties,
                },
                AtomicNoteOptions {
                    salience,
                    decay_factor,
                    embedding_model,
                    embedding_content,
                    embed: Some(options.embed.unwrap_or(kind != "head")),
                    key: None,
                    memory_visibility_receipt: false,
                    replay_receipt: true,
                    fence: options.fence.as_ref(),
                    properties_already_derived: false,
                },
                &annotates,
                KeyPublication::AtInsert,
            )
            .await?;
            let note = prepared.notes.remove(0);
            return match run_atomic_unit(self.sql().as_ref(), prepared.plans).await {
                Ok(AtomicRunOutcome::Committed { .. }) => {
                    self.fire_note_mutation_hook(&note.kind, note.id).await;
                    Ok((note, prepared.embedding_truncation))
                }
                Ok(AtomicRunOutcome::RolledBack {
                    failure: AtomicOpFailure::NoteConflict(conflict),
                    ..
                }) => Err(conflict.into_error().into()),
                Ok(AtomicRunOutcome::RolledBack { failure, .. }) => Err(RuntimeError::Internal(
                    format!("note creation rolled back: {failure:?}"),
                )),
                Err(error) => Err(RuntimeError::Storage(error.0)),
            };
        };

        // Keyed singleton create (ADR-172 Amendment 6): staged.
        //
        // Properties are derived exactly once, here, before either holder
        // check, because the installed note-write validator is not guaranteed
        // idempotent (a kg regression test counts its calls), and equality
        // compares the DERIVED value, so the initial check, the note this
        // preparation would insert, and any final-check fallback must all
        // compare/store the SAME derived value rather than three
        // independently hook-derived ones.
        let derived_properties =
            self.derive_note_write_properties(kind, token, properties.clone())?;
        let namespace: String = token.namespace().as_str().into();

        // Initial writer transaction: supplied fences and the live holder
        // for (namespace, kind, key), before any creation-only work runs.
        // Equal + disclosable => replay, no preparation. Different => the
        // existing key_conflict. A stale fence => fence_conflict, also
        // before preparation. Absent (and fences pass) => continue.
        if let Some(conflict) = check_keyed_create_holder(
            self.sql().as_ref(),
            NoteWriteGuard {
                namespace: namespace.clone(),
                target_id: candidate.id,
                expected_version: None,
                fence: options.fence.clone(),
                create_key: Some(CreateKeyClaim {
                    kind: kind.to_owned(),
                    key: key.to_owned(),
                    content: content.to_owned(),
                    properties: derived_properties.clone(),
                    replay_signal: true,
                }),
            },
        )
        .await?
        {
            return Err(conflict.into_error().into());
        }

        // Creation-only preparation (annotation-target resolution,
        // embedding-model resolution, embedding computation), outside the
        // writer, now that the key was observed absent. A failure here is
        // captured rather than returned immediately: the final writer
        // transaction below still gets to revalidate the holder before
        // either this plan or this failure is consumed.
        let prep_result = async {
            let annotates = annotations.await?;
            prepare_note_create(
                self,
                AtomicNoteSpec {
                    token,
                    id: None,
                    kind,
                    name,
                    content,
                    properties: derived_properties.clone(),
                },
                AtomicNoteOptions {
                    salience,
                    decay_factor,
                    embedding_model,
                    embedding_content,
                    embed: Some(options.embed.unwrap_or(kind != "head")),
                    key: Some(key),
                    memory_visibility_receipt: false,
                    replay_receipt: true,
                    fence: options.fence.as_ref(),
                    properties_already_derived: true,
                },
                &annotates,
                KeyPublication::AtInsert,
            )
            .await
        }
        .await;

        // Test-only pause, reached whether preparation succeeded or failed:
        // both the ordinary insert path and the prep-failure fallback below
        // re-derive the current holder fresh, so both need to be
        // interruptible by an independent writer for the race regression
        // tests (ADR-172 Amendment 6, acceptance items 3 and 5).
        #[cfg(test)]
        race_seam::pause_after_prepare().await;

        match prep_result {
            Ok((mut prepared, _)) => {
                let note = prepared.notes.remove(0);
                match run_atomic_unit(self.sql().as_ref(), prepared.plans).await {
                    Ok(AtomicRunOutcome::Committed { .. }) => {
                        self.fire_note_mutation_hook(&note.kind, note.id).await;
                        Ok((note, prepared.embedding_truncation))
                    }
                    Ok(AtomicRunOutcome::RolledBack {
                        failure: AtomicOpFailure::NoteConflict(conflict),
                        ..
                    }) => Err(conflict.into_error().into()),
                    Ok(AtomicRunOutcome::RolledBack { failure, .. }) => Err(
                        RuntimeError::Internal(format!("note creation rolled back: {failure:?}")),
                    ),
                    Err(error) => Err(RuntimeError::Storage(error.0)),
                }
            }
            Err(prep_error) => {
                // Preparation failed. Revalidate fences and the current
                // holder one more time before consuming that failure: an
                // equal holder now yields replay; a different payload
                // conflicts; continued absence preserves the preparation
                // failure (ADR-172 Amendment 6, acceptance item 5).
                match check_keyed_create_holder(
                    self.sql().as_ref(),
                    NoteWriteGuard {
                        namespace,
                        target_id: candidate.id,
                        expected_version: None,
                        fence: options.fence.clone(),
                        create_key: Some(CreateKeyClaim {
                            kind: kind.to_owned(),
                            key: key.to_owned(),
                            content: content.to_owned(),
                            properties: derived_properties,
                            replay_signal: true,
                        }),
                    },
                )
                .await?
                {
                    Some(conflict) => Err(conflict.into_error().into()),
                    None => Err(prep_error),
                }
            }
        }
    }
}