khive_db/stores/

note.rs

//! SQL-backed `NoteStore` implementation.

pub mod recipient;
pub mod transport;

use std::collections::HashSet;
use std::sync::Arc;

use async_trait::async_trait;
use rusqlite::OptionalExtension;
use uuid::Uuid;

use khive_storage::attachment::{Attachment, AttachmentSubstrate};
use khive_storage::error::StorageError;
use khive_storage::note::{
    FilterOp, Note, NoteFilter, NoteInstantSeekAfter, NoteKeyCursor, NoteSeekAfter, NoteTagMode,
    NoteTimeOrder, NoteVisibility, SortDir,
};
use khive_storage::types::{
    BatchWriteSummary, BoundedCount, DeleteMode, Page, PageRequest, SeekCursor, SeekPage,
    SqlStatement, SqlValue,
};
use khive_storage::NoteStore;
use khive_storage::{StorageCapability, StorageResult};

use crate::error::SqliteError;
use crate::pool::ConnectionPool;
use crate::sql_bridge::bind_params;
use crate::stores::attachment::{attachment_upsert_statement, delete_record_attachments_statement};
use crate::writer_task::{execute_wrapped_transaction, WriterTaskHandle};

fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Notes, op, e)
}

fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
    e.into_storage_error(StorageCapability::Notes, op)
}

const NAMESPACE_COUNT_CHUNK_SIZE: usize = 500;

pub fn note_key_prefix_successor(prefix: &str) -> Option<String> {
    let mut chars: Vec<char> = prefix.chars().collect();
    while let Some(last) = chars.pop() {
        if last == char::MAX {
            continue;
        }
        let next = if last == '\u{d7ff}' {
            '\u{e000}'
        } else {
            char::from_u32(u32::from(last) + 1).expect("incremented non-max scalar")
        };
        chars.push(next);
        return Some(chars.into_iter().collect());
    }
    None
}

// ---------------------------------------------------------------------------
// Pure statement builders (ADR-099 B3 r6 structural cut) — see entity.rs's
// sibling block for the full rationale. `upsert_note`/`delete_note` below
// and ADR-099's atomic prepare path (`khive-runtime`) both call these.
// ---------------------------------------------------------------------------

/// The single true UPSERT every note writer (single, batch, and note-merge)
/// issues against `notes` (ADR-116 memory-ANN-generation-coherence prereq).
///
/// `INSERT OR REPLACE` is a SQLite DELETE-then-INSERT on a conflicting `id`:
/// it fires DELETE-path triggers (spuriously invalidating ANN generation
/// state once ADR-116's liveness triggers land) and discards the row's
/// original `created_at`/rowid identity. This form updates in place on a
/// primary-key conflict instead. `created_at` is deliberately absent from
/// the `DO UPDATE SET` list — it is bound for the INSERT branch only and
/// left untouched by the UPDATE branch, so an existing row keeps its
/// original `created_at` across any number of upserts. The caller-chosen
/// `key` is likewise insert-only: updates cannot replace or clear its identity.
pub const NOTE_UPSERT_SQL: &str = "INSERT INTO notes \
     (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
      properties, created_at, updated_at, deleted_at, key, strict_due_key, due_source) \
     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16) \
     ON CONFLICT(id) DO UPDATE SET \
       namespace = excluded.namespace, \
       kind = excluded.kind, \
       status = excluded.status, \
       name = excluded.name, \
       content = excluded.content, \
       salience = excluded.salience, \
       decay_factor = excluded.decay_factor, \
       expires_at = excluded.expires_at, \
       properties = excluded.properties, \
       strict_due_key = excluded.strict_due_key, \
       due_source = excluded.due_source, \
       updated_at = excluded.updated_at, \
       deleted_at = excluded.deleted_at";

/// Insert-if-absent counterpart to [`NOTE_UPSERT_SQL`], for a caller whose
/// read found no row.
///
/// Identical column list and parameter order, and deliberately
/// `DO NOTHING` rather than `DO UPDATE`: the point is that a row already
/// present must survive untouched, so the losing caller can be told it lost
/// instead of silently overwriting the winner. `changes()` is then the
/// answer to "did I insert it", which `DO UPDATE` cannot report.
pub const NOTE_INSERT_IF_ABSENT_SQL: &str = "INSERT INTO notes \
     (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
      properties, created_at, updated_at, deleted_at, key, strict_due_key, due_source) \
     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16) \
     ON CONFLICT(id) DO NOTHING";

/// The exact statement this store's `insert_note_if_absent` issues.
pub fn note_insert_if_absent_statement(note: &Note) -> SqlStatement {
    let mut statement = note_upsert_statement(note);
    statement.sql = NOTE_INSERT_IF_ABSENT_SQL.to_string();
    statement.label = Some("note-insert-if-absent".to_string());
    statement
}

/// The live key index is the arbiter; an occupied key changes no row.
pub fn note_insert_keyed_statement(note: &Note) -> SqlStatement {
    let mut statement = note_upsert_statement(note);
    statement.sql = "INSERT INTO notes \
        (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
         properties, created_at, updated_at, deleted_at, key, strict_due_key, due_source) \
        VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16) \
        ON CONFLICT(namespace,kind,key) WHERE key IS NOT NULL AND deleted_at IS NULL DO NOTHING"
        .into();
    statement.label = Some("note-keyed-create".into());
    statement
}

/// Compute a strict, order-preserving deadline key for supported note writers.
/// Missing, non-text, and malformed values remain NULL and enter the index as
/// conservative due candidates, where the strict read residual decides.
pub fn note_due_key_values(
    properties: &Option<serde_json::Value>,
) -> (Option<Vec<u8>>, Option<String>) {
    let source = properties
        .as_ref()
        .and_then(|value| value.get("next_attempt_at"))
        .and_then(|value| value.as_str());
    match source.and_then(|text| crate::pool::strict_rfc3339_key(text).map(|key| (key, text))) {
        Some((key, source)) => (Some(key), Some(source.to_string())),
        None => (None, None),
    }
}

/// The exact true UPSERT this store's `upsert_note` issues.
pub fn note_upsert_statement(note: &Note) -> SqlStatement {
    let (due_key, due_source) = note_due_key_values(&note.properties);
    let properties_str = note
        .properties
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: NOTE_UPSERT_SQL.to_string(),
        params: vec![
            SqlValue::Text(note.id.to_string()),
            SqlValue::Text(note.namespace.clone()),
            SqlValue::Text(note.kind.to_string()),
            SqlValue::Text(note.status.clone()),
            match &note.name {
                Some(n) => SqlValue::Text(n.clone()),
                None => SqlValue::Null,
            },
            SqlValue::Text(note.content.clone()),
            match note.salience {
                Some(s) => SqlValue::Float(s),
                None => SqlValue::Null,
            },
            match note.decay_factor {
                Some(d) => SqlValue::Float(d),
                None => SqlValue::Null,
            },
            match note.expires_at {
                Some(e) => SqlValue::Integer(e),
                None => SqlValue::Null,
            },
            match properties_str {
                Some(p) => SqlValue::Text(p),
                None => SqlValue::Null,
            },
            SqlValue::Integer(note.created_at),
            SqlValue::Integer(note.updated_at),
            match note.deleted_at {
                Some(d) => SqlValue::Integer(d),
                None => SqlValue::Null,
            },
            match &note.key {
                Some(key) => SqlValue::Text(key.clone()),
                None => SqlValue::Null,
            },
            due_key.map_or(SqlValue::Null, SqlValue::Blob),
            due_source.map_or(SqlValue::Null, SqlValue::Text),
        ],
        label: Some("note-upsert".to_string()),
    }
}

/// Full-note compare-and-swap update used after caller-side normalization was
/// derived from a read snapshot. Unlike [`note_upsert_statement`], this never
/// inserts and cannot overwrite a row whose revision or deletion marker moved
/// after the snapshot was read. The replacement revision must also be strictly
/// greater than the persisted snapshot revision; equality is a refused CAS,
/// never a successful write with an unchanged concurrency token.
pub fn note_replace_if_unchanged_statement(
    note: &Note,
    expected_updated_at: i64,
    expected_deleted_at: Option<i64>,
) -> SqlStatement {
    let (due_key, due_source) = note_due_key_values(&note.properties);
    let properties_str = note
        .properties
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: "UPDATE notes SET \
                namespace = ?1, kind = ?2, status = ?3, name = ?4, content = ?5, \
                salience = ?6, decay_factor = ?7, expires_at = ?8, properties = ?9, \
                updated_at = ?10, deleted_at = ?11, strict_due_key = ?15, due_source = ?16 \
              WHERE id = ?12 AND updated_at = ?13 AND deleted_at IS ?14 \
                AND ?10 > updated_at"
            .to_string(),
        params: vec![
            SqlValue::Text(note.namespace.clone()),
            SqlValue::Text(note.kind.to_string()),
            SqlValue::Text(note.status.clone()),
            match &note.name {
                Some(name) => SqlValue::Text(name.clone()),
                None => SqlValue::Null,
            },
            SqlValue::Text(note.content.clone()),
            match note.salience {
                Some(value) => SqlValue::Float(value),
                None => SqlValue::Null,
            },
            match note.decay_factor {
                Some(value) => SqlValue::Float(value),
                None => SqlValue::Null,
            },
            match note.expires_at {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
            match properties_str {
                Some(value) => SqlValue::Text(value),
                None => SqlValue::Null,
            },
            SqlValue::Integer(note.updated_at),
            match note.deleted_at {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
            SqlValue::Text(note.id.to_string()),
            SqlValue::Integer(expected_updated_at),
            match expected_deleted_at {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
            due_key.map_or(SqlValue::Null, SqlValue::Blob),
            due_source.map_or(SqlValue::Null, SqlValue::Text),
        ],
        label: Some("note-replace-if-unchanged".to_string()),
    }
}

/// Metadata-only CAS for an immutable stream entry. Protected fields are
/// predicates, never SET targets: SQLite UPDATE OF triggers fire even when a
/// value is assigned to itself. Keep the full snapshot and revision guards.
pub fn note_metadata_replace_if_unchanged_statement(
    note: &Note,
    expected_updated_at: i64,
    expected_deleted_at: Option<i64>,
) -> SqlStatement {
    let mut statement =
        note_replace_if_unchanged_statement(note, expected_updated_at, expected_deleted_at);
    statement.sql = "UPDATE notes SET status=?3, name=?4, salience=?6, decay_factor=?7, expires_at=?8, updated_at=?10 \
                     WHERE id=?12 AND updated_at=?13 AND deleted_at IS ?14 AND ?10 > updated_at \
                       AND namespace=?1 AND kind=?2 AND content=?5 AND properties IS ?9 AND deleted_at IS ?11".into();
    statement.params.truncate(14);
    statement.label = Some("stream-note-metadata-cas".into());
    statement
}

/// The exact `properties`/`updated_at` `UPDATE` this store's
/// `update_note_properties` issues. The row is patched in place without
/// rewriting any other note column or its stable row identity (#780).
/// The `comm.probe` cursor is keyed on `notes_seq.seq`, which is fixed at
/// first insert and survives a delete+reinsert of the same note id, so this
/// is defensive rather than load-bearing for cursor correctness; a metadata
/// patch should never rewrite the row regardless.
pub fn note_update_properties_statement(
    id: Uuid,
    properties: &Option<serde_json::Value>,
    updated_at: i64,
) -> SqlStatement {
    let (due_key, due_source) = note_due_key_values(properties);
    let properties_str = properties
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: "UPDATE notes SET properties = ?1, updated_at = ?2, \
              strict_due_key = ?4, due_source = ?5 \
              WHERE id = ?3 AND deleted_at IS NULL"
            .to_string(),
        params: vec![
            match properties_str {
                Some(p) => SqlValue::Text(p),
                None => SqlValue::Null,
            },
            SqlValue::Integer(updated_at),
            SqlValue::Text(id.to_string()),
            due_key.map_or(SqlValue::Null, SqlValue::Blob),
            due_source.map_or(SqlValue::Null, SqlValue::Text),
        ],
        label: Some("note-update-properties".to_string()),
    }
}

/// The atomic top-level JSON-property `UPDATE` issued by
/// [`NoteStore::set_note_property`]. The key is encoded as one quoted JSON
/// path segment, so punctuation is literal rather than interpreted as nested
/// path syntax. `json(?2)` preserves the bound value's JSON type instead of
/// storing objects, arrays, booleans, or numbers as JSON strings.
///
/// SQL-NULL documents start from `{}`. JSON arrays/scalars/null are not
/// property objects and are left untouched; the caller receives `false`.
pub fn note_set_property_statement(
    id: Uuid,
    key: &str,
    value: &serde_json::Value,
    updated_at: i64,
) -> Result<SqlStatement, StorageError> {
    // SQLite JSON-path object labels terminate at U+0000. Passing such a key
    // through `json_set` can therefore select a shorter sibling key instead
    // of the requested literal key, so reject it before constructing SQL.
    if key.contains('\0') {
        return Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "set_note_property".into(),
            message: "property key must not contain U+0000".to_string(),
        });
    }
    let path = format!("$.{}", serde_json::Value::String(key.to_string()));
    let updating_due = key == "next_attempt_at";
    let (due_key, due_source) = if updating_due {
        note_due_key_values(&Some(serde_json::json!({"next_attempt_at": value})))
    } else {
        (None, None)
    };
    Ok(SqlStatement {
        sql: "UPDATE notes \
              SET properties = json_set(COALESCE(properties, '{}'), ?1, json(?2)), \
                  updated_at = ?3, \
                  strict_due_key = CASE WHEN ?5 = 1 THEN ?6 ELSE strict_due_key END, \
                  due_source = CASE WHEN ?5 = 1 THEN ?7 ELSE due_source END \
              WHERE id = ?4 AND deleted_at IS NULL \
                AND (properties IS NULL OR json_type(properties) = 'object')"
            .to_string(),
        params: vec![
            SqlValue::Text(path),
            SqlValue::Text(value.to_string()),
            SqlValue::Integer(updated_at),
            SqlValue::Text(id.to_string()),
            SqlValue::Integer(i64::from(updating_due)),
            due_key.map_or(SqlValue::Null, SqlValue::Blob),
            due_source.map_or(SqlValue::Null, SqlValue::Text),
        ],
        label: Some("note-set-property".to_string()),
    })
}

/// The exact soft-delete `UPDATE` this store's `delete_note(Soft)` issues.
pub fn note_soft_delete_statement(id: Uuid, deleted_at: i64) -> SqlStatement {
    SqlStatement {
        sql: "UPDATE notes SET status = 'deleted', deleted_at = ?1 \
              WHERE id = ?2 AND deleted_at IS NULL"
            .to_string(),
        params: vec![
            SqlValue::Integer(deleted_at),
            SqlValue::Text(id.to_string()),
        ],
        label: Some("note-delete-soft".to_string()),
    }
}

/// The exact hard-delete `DELETE` this store's `delete_note(Hard)` issues.
pub fn note_hard_delete_statement(id: Uuid) -> SqlStatement {
    SqlStatement {
        sql: "DELETE FROM notes WHERE id = ?1".to_string(),
        params: vec![SqlValue::Text(id.to_string())],
        label: Some("note-delete-hard".to_string()),
    }
}

/// A NoteStore backed by SQLite. Namespace is the caller's responsibility.
///
/// UUID is globally unique — get/delete by ID alone. Query/count use the
/// namespace parameter as passed. Read routing is always pool-backed; the
/// constructor's legacy file-backed flag is retained for API compatibility.
pub struct SqlNoteStore {
    pool: Arc<ConnectionPool>,
    index_repair: Option<super::index_repair::IndexRepairContext>,
    writer_task: Option<WriterTaskHandle>,
}

impl SqlNoteStore {
    /// Create a new store.
    pub fn new(pool: Arc<ConnectionPool>, _is_file_backed: bool) -> Self {
        // Enabled by default for file-backed pools; explicit off/degraded
        // fallback remains possible (ADR-067 Component A, mirrors
        // entity.rs policy): a missing writer task — explicitly disabled,
        // spawn degraded, or no Tokio runtime available at this first
        // access — is cached without failing construction. Every write
        // re-resolves it and applies strict/compatibility policy then.
        let writer_task = pool.writer_task_handle().ok().flatten();

        Self {
            pool,
            writer_task,
            index_repair: None,
        }
    }

    pub(crate) fn with_index_repair(
        mut self,
        repair: super::index_repair::IndexRepairContext,
    ) -> Self {
        self.index_repair = Some(repair);
        self
    }

    async fn with_indexed_reader<F, R>(&self, op: &'static str, read: F) -> Result<R, StorageError>
    where
        F: FnMut(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        super::index_repair::run_indexed_read(
            Arc::clone(&self.pool),
            self.index_repair.clone(),
            StorageCapability::Notes,
            op,
            read,
        )
        .await
    }

    fn current_writer_task(
        &self,
        operation: &'static str,
    ) -> Result<Option<WriterTaskHandle>, StorageError> {
        self.pool
            .writer_task_for_write(self.writer_task.as_ref(), operation)
    }

    /// Route a single-row write through the pool-wide `WriterTask` when
    /// the write queue is enabled and a handle is available. Strict mode
    /// refuses a missing handle; compatibility mode falls back to the legacy
    /// pool-mutex path (ADR-067 Component A, Fork C slice 2).
    ///
    /// This is the routing point for single-statement `with_writer` callers
    /// in this store (`update_note_properties`, `set_note_property`,
    /// `delete_note`). `f` must be DML-only — on the flag-on path it runs
    /// inside the WriterTask's own transaction, so a bare `BEGIN IMMEDIATE`
    /// would violate SQLite's nested-transaction rule. `upsert_notes` (the
    /// batch method) performs the same write-time lookup first; a non-strict
    /// `None` then falls through this helper, which records the actual
    /// compatibility fallback. Strict mode returns before the direct-writer
    /// seam. Callers whose `f` issues more than one
    /// DML statement that must land atomically together (`upsert_note`,
    /// `try_insert_note`, `patch_note_property_atomic`) use
    /// [`Self::with_writer_tx`] instead — see its doc comment (khive #827,
    /// #1387).
    async fn with_writer<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = self.current_writer_task(op)? {
            return writer_task
                .send_bounded_at(move |conn| f(conn).map_err(|e| map_err(e, op)), op)
                .await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteNote);
        let pool = Arc::clone(&self.pool);
        tokio::task::spawn_blocking(move || {
            let guard = pool
                .autocommit_write_unit()
                .map_err(|e| map_sqlite_err(e, op))?;
            f(guard.conn())
                .map_err(|e| map_err(e, op))
                .map_err(crate::error::statement_failure)
                .inspect_err(|error| pool.record_direct_writer_error(error))
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
    }

    /// Like [`Self::with_writer`], but for callers whose closure issues more
    /// than one DML statement that must land atomically together (khive
    /// #827, #1387), such as a note insert plus `assign_note_seq` or a guarded
    /// property patch across several notes. On the flag-on path the
    /// WriterTask already wraps every request in its own `BEGIN
    /// IMMEDIATE`/`COMMIT`/`ROLLBACK`, so `f` is sent unwrapped, same as
    /// `with_writer`. On the flag-off (pool-mutex) path, `with_writer` runs
    /// `f` in SQLite's default autocommit mode -- each statement inside `f`
    /// is its own implicit transaction -- so a later failure could leave an
    /// earlier statement committed. This wraps that path in one explicit
    /// transaction, matching `upsert_notes`' own flag-off branch.
    async fn with_writer_tx<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        self.with_writer_tx_storage(op, move |conn| f(conn).map_err(|error| map_err(error, op)))
            .await
    }

    async fn with_writer_tx_storage<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, StorageError> + Send + 'static,
        R: Send + 'static,
    {
        if let Some(writer_task) = self.current_writer_task(op)? {
            return writer_task.send_bounded_at(f, op).await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteNote);
        let pool = Arc::clone(&self.pool);
        tokio::task::spawn_blocking(move || {
            let guard = pool
                .transaction_write_unit()
                .map_err(|e| map_sqlite_err(e, op))
                .inspect_err(|error| pool.record_direct_writer_error(error))?;
            let conn = guard.conn();
            let (result, terminal_state) = execute_wrapped_transaction(conn, op, f);
            if terminal_state.is_some() {
                pool.retire_pooled_writer(conn);
            }
            result.inspect_err(|error| pool.record_direct_writer_error(error))
        })
        .await
        .map_err(|e| StorageError::driver(StorageCapability::Notes, op, e))?
    }

    async fn get_notes_batch_inner(
        &self,
        ids: &[Uuid],
        include_deleted: bool,
    ) -> Result<Vec<Note>, StorageError> {
        if ids.is_empty() {
            return Ok(vec![]);
        }
        let operation = if include_deleted {
            "get_notes_batch_including_deleted"
        } else {
            "get_notes_batch"
        };
        let deletion_filter = if include_deleted {
            ""
        } else {
            " AND deleted_at IS NULL"
        };
        // SQLite SQLITE_MAX_VARIABLE_NUMBER defaults to 999; chunk below that
        // ceiling so callers can safely hydrate arbitrarily large ID sets.
        const CHUNK: usize = 900;
        let id_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();

        let mut result = Vec::with_capacity(ids.len());
        for chunk in id_strings.chunks(CHUNK) {
            let chunk_owned = chunk.to_vec();
            let notes = self
                .with_reader(operation, move |conn| {
                    let placeholders: String = (1..=chunk_owned.len())
                        .map(|i| format!("?{i}"))
                        .collect::<Vec<_>>()
                        .join(", ");
                    let sql = format!(
                        "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                         properties, created_at, updated_at, deleted_at, key, version \
                         FROM notes WHERE id IN ({placeholders}){deletion_filter}"
                    );
                    let mut stmt = conn.prepare(&sql)?;
                    let params: Vec<&dyn rusqlite::types::ToSql> = chunk_owned
                        .iter()
                        .map(|s| s as &dyn rusqlite::types::ToSql)
                        .collect();
                    let rows = stmt.query_map(params.as_slice(), read_note)?;
                    let mut notes = Vec::new();
                    for row in rows {
                        notes.push(row?);
                    }
                    Ok(notes)
                })
                .await?;
            result.extend(notes);
        }
        Ok(result)
    }

    async fn with_reader<F, R>(&self, op: &'static str, f: F) -> Result<R, StorageError>
    where
        F: FnOnce(&rusqlite::Connection) -> Result<R, rusqlite::Error> + Send + 'static,
        R: Send + 'static,
    {
        super::run_pooled_store_read(
            Arc::clone(&self.pool),
            StorageCapability::Notes,
            op,
            move |conn| f(conn).map_err(|error| map_err(error, op)),
        )
        .await
    }
}

// =============================================================================
// Helpers
// =============================================================================

fn read_note(row: &rusqlite::Row<'_>) -> Result<Note, rusqlite::Error> {
    let id_str: String = row.get(0)?;
    let namespace: String = row.get(1)?;
    let kind: String = row.get(2)?;
    let status: String = row.get(3)?;
    let name: Option<String> = row.get(4)?;
    let content: String = row.get(5)?;
    let salience: Option<f64> = row.get(6)?;
    let decay_factor: Option<f64> = row.get(7)?;
    let expires_at: Option<i64> = row.get(8)?;
    let properties_str: Option<String> = row.get(9)?;
    let created_at: i64 = row.get(10)?;
    let updated_at: i64 = row.get(11)?;
    let deleted_at: Option<i64> = row.get(12)?;
    let key: Option<String> = row.get(13)?;
    let version: i64 = row.get(14)?;

    let id = parse_uuid(&id_str)?;

    let properties = properties_str
        .map(|s| {
            serde_json::from_str(&s).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    9,
                    rusqlite::types::Type::Text,
                    Box::new(e),
                )
            })
        })
        .transpose()?;

    Ok(Note {
        id,
        namespace,
        kind,
        status,
        name,
        content,
        salience,
        decay_factor,
        expires_at,
        properties,
        created_at,
        updated_at,
        deleted_at,
        key,
        version,
    })
}

fn parse_uuid(s: &str) -> Result<Uuid, rusqlite::Error> {
    Uuid::parse_str(s).map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(e))
    })
}

fn query_note_page_snapshot(
    conn: &rusqlite::Connection,
    operation: &'static str,
    namespace: &str,
    count_sql: &str,
    count_params: &[Box<dyn rusqlite::types::ToSql>],
    data_sql: &str,
    data_params: &[Box<dyn rusqlite::types::ToSql>],
) -> Result<Page<Note>, rusqlite::Error> {
    let tx = rusqlite::Transaction::new_unchecked(conn, rusqlite::TransactionBehavior::Deferred)?;

    let total: i64 = {
        let mut stmt = tx.prepare(count_sql)?;
        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
            count_params.iter().map(|param| param.as_ref()).collect();
        stmt.query_row(param_refs.as_slice(), |row| row.get(0))?
    };

    #[cfg(test)]
    tests::page_snapshot_seam::hook(operation, namespace);
    #[cfg(not(test))]
    let _ = (operation, namespace);

    let items = {
        let mut stmt = tx.prepare(data_sql)?;
        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
            data_params.iter().map(|param| param.as_ref()).collect();
        let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
        rows.collect::<Result<Vec<_>, _>>()?
    };

    tx.commit()?;
    Ok(Page {
        items,
        total: Some(total as u64),
    })
}

/// DML-only batch upsert loop shared by both the legacy (flag-off) and
/// WriterTask-routed (flag-on) `upsert_notes` paths (ADR-067 Component A).
///
/// Issues no `BEGIN` / `COMMIT` / `ROLLBACK` itself — the caller owns the
/// enclosing transaction. Per-row failures are captured into
/// `BatchWriteSummary::failed`/`first_error` rather than aborting the loop,
/// matching the existing partial-success contract.
fn batch_upsert_notes(
    conn: &rusqlite::Connection,
    notes: &[Note],
    attempted: u64,
) -> Result<BatchWriteSummary, rusqlite::Error> {
    let mut summary = BatchWriteSummary {
        attempted,
        ..BatchWriteSummary::default()
    };

    // Prepare the UPSERT once for the whole batch — `Connection::execute`
    // re-parses and re-plans the statement on every call, which dominates
    // wall time at conflict-heavy batch sizes (measured 536ms vs 260ms at
    // 50k conflicts; see PR #1082 review).
    let mut stmt = conn.prepare_cached(NOTE_UPSERT_SQL)?;

    for (index, note) in notes.iter().enumerate() {
        let id_str = note.id.to_string();
        let kind_str = note.kind.to_string();
        let status_str = note.status.clone();
        let properties_str = note
            .properties
            .as_ref()
            .map(|v| serde_json::to_string(v).unwrap_or_default());
        let (due_key, due_source) = note_due_key_values(&note.properties);

        match stmt.execute(rusqlite::params![
            id_str,
            &note.namespace,
            kind_str,
            status_str,
            &note.name,
            note.content,
            note.salience,
            note.decay_factor,
            note.expires_at,
            properties_str,
            note.created_at,
            note.updated_at,
            note.deleted_at,
            note.key,
            due_key,
            due_source,
        ]) {
            Ok(_) => {
                assign_note_seq(conn, &id_str)?;
                summary.affected = summary.affected.saturating_add(1);
            }
            Err(e) => {
                let (class, retryability) = super::classify_batch_sqlite_error(&e);
                summary.record_failure(index, Some(id_str), class, retryability, e.to_string());
            }
        }
    }

    Ok(summary)
}

/// Assign a note id its durable, non-reusing sequence number the first time
/// it is inserted (khive #827 — see `sql/007-notes-seq.sql`). `INSERT OR
/// IGNORE` makes this idempotent across repeated upserts of the same note
/// id: the sequence value is fixed at the note's first insert and never
/// reassigned, unlike `notes`' own implicit rowid.
fn assign_note_seq(conn: &rusqlite::Connection, note_id: &str) -> Result<(), rusqlite::Error> {
    conn.execute(
        "INSERT OR IGNORE INTO notes_seq (note_id) VALUES (?1)",
        rusqlite::params![note_id],
    )?;
    Ok(())
}

fn build_note_where(
    namespace: &str,
    kind: Option<&str>,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut conditions: Vec<String> = vec![
        "namespace = ?1".to_string(),
        "deleted_at IS NULL".to_string(),
    ];
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(namespace.to_string())];

    if let Some(k) = kind {
        params.push(Box::new(k.to_string()));
        conditions.push(format!("kind = ?{}", params.len()));
    }

    let clause = format!(" WHERE {}", conditions.join(" AND "));
    (clause, params)
}

fn build_note_where_for_namespaces(
    namespaces: &[String],
    kind: Option<&str>,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = namespaces
        .iter()
        .map(|namespace| -> Box<dyn rusqlite::types::ToSql> { Box::new(namespace.clone()) })
        .collect();
    let namespace_condition = match namespaces.len() {
        0 => "0".to_string(),
        1 => "namespace = ?1".to_string(),
        _ => {
            let placeholders: Vec<String> =
                (1..=namespaces.len()).map(|i| format!("?{i}")).collect();
            format!("namespace IN ({})", placeholders.join(", "))
        }
    };
    let mut conditions = vec![namespace_condition, "deleted_at IS NULL".to_string()];

    if let Some(kind) = kind {
        params.push(Box::new(kind.to_string()));
        conditions.push(format!("kind = ?{}", params.len()));
    }

    let clause = format!(" WHERE {}", conditions.join(" AND "));
    (clause, params)
}

/// Validate that a json_path is safe to interpolate into SQL.
/// Accepts only `$.field` or `$.field.subfield` paths with alphanumeric/underscore segments.
fn validate_json_path(path: &str) -> Result<(), StorageError> {
    let valid = path.starts_with("$.")
        && path[2..].split('.').all(|part| {
            !part.is_empty() && part.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
        });
    if valid {
        Ok(())
    } else {
        Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_filtered".into(),
            message: format!("invalid JSON path for note filter: {path:?}"),
        })
    }
}

fn json_extract_expr(path: &str) -> String {
    format!("json_extract(properties, '{path}')")
}

fn json_type_expr(path: &str) -> String {
    format!("json_type(properties, '{path}')")
}

fn validate_note_time_order(
    filter: &NoteFilter,
    operation: &'static str,
    supports_time_order: bool,
) -> StorageResult<()> {
    if filter.time_order.is_some()
        && (!supports_time_order
            || filter.order_by.is_some()
            || filter.order_by_instant
            || filter.unordered
            || filter.after.is_some()
            || filter.after_instant.is_some())
    {
        return Err(StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: operation.into(),
            message: "time_order requires a native-order page without property order, unordered mode or cursors; fixed-order reads and filtered mutations do not support it".into(),
        });
    }
    Ok(())
}

fn note_time_order_clause(order: NoteTimeOrder) -> &'static str {
    match order {
        NoteTimeOrder::ExpiresAt => " ORDER BY expires_at ASC, id ASC",
        NoteTimeOrder::ExpiresAtOrCreatedAt => {
            " ORDER BY COALESCE(expires_at, created_at) ASC, id ASC"
        }
    }
}

/// Deterministic total order shared by exact-count and count-free filtered
/// pages. Keeping the clause in one helper prevents the cheaper projection
/// from drifting into a different offset sequence.
fn note_filter_page_order_clause(filter: &NoteFilter) -> String {
    if let Some(order) = filter.time_order {
        return note_time_order_clause(order).to_string();
    }
    if filter.unordered {
        return String::new();
    }
    match &filter.order_by {
        Some((path, dir)) => {
            if filter.order_by_instant {
                let expr = json_extract_expr(path);
                return format!(" ORDER BY khive_rfc3339_key({expr}) ASC, {expr} ASC, id ASC");
            }
            let dir_str = match dir {
                SortDir::Asc => "ASC",
                SortDir::Desc => "DESC",
            };
            // #1671: append `id` as the final tiebreak in the sort field's
            // direction so equal JSON sort values still form a total order.
            format!(
                " ORDER BY {} {dir_str}, id {dir_str}",
                json_extract_expr(path)
            )
        }
        // After ANALYZE, SQLite may choose the creation-order index to fill
        // LIMIT and scan every future retry. The unary plus preserves the
        // INTEGER timestamp order while making SQLite sort only due rows
        // reached through the channel/deadline index.
        None if filter
            .property_filters
            .iter()
            .any(|property| matches!(&property.op, FilterOp::TextColonPrefixBucketIndexed))
            && filter
                .property_filters
                .iter()
                .any(|property| matches!(&property.op, FilterOp::Rfc3339LteOrInvalid)) =>
        {
            " ORDER BY +created_at DESC, id ASC".to_string()
        }
        // `id ASC` over the primary key is already the stable tiebreak for
        // notes sharing a creation timestamp.
        None => DEFAULT_NOTE_PAGE_ORDER.to_string(),
    }
}

/// Validate a value destined for inline comparison against `json_type()`.
/// The only admissible values are SQLite's own json_type result strings —
/// a closed vocabulary — so a validated value can be inlined into SQL text
/// without any injection surface. Anything else is a caller bug, rejected
/// rather than parameterized.
fn json_type_literal(value: &SqlValue) -> Result<&str, rusqlite::Error> {
    const JSON_TYPES: [&str; 8] = [
        "true", "false", "integer", "real", "text", "array", "object", "null",
    ];
    match value {
        SqlValue::Text(s) if JSON_TYPES.contains(&s.as_str()) => Ok(s.as_str()),
        other => Err(rusqlite::Error::InvalidParameterName(format!(
            "json_type comparison value must be one of SQLite's json_type strings \
             ({JSON_TYPES:?}), got {other:?}"
        ))),
    }
}

/// Exclusive upper bound of the set of strings starting with `prefix`, in
/// SQLite's BINARY (UTF-8 byte) order: the prefix with its last code point
/// incremented, after dropping any trailing `char::MAX` (which has no
/// successor). `None` when every string starting with `prefix` is also
/// `>= prefix` without an upper bound (empty prefix, or all `char::MAX`).
/// UTF-8 preserves code point order byte-wise, so every string with the
/// prefix sorts strictly below the returned bound.
fn text_prefix_upper_bound(prefix: &str) -> Option<String> {
    let mut chars: Vec<char> = prefix.chars().collect();
    while let Some(last) = chars.pop() {
        let mut next = u32::from(last) + 1;
        if (0xD800..=0xDFFF).contains(&next) {
            next = 0xE000;
        }
        if let Some(next) = char::from_u32(next) {
            chars.push(next);
            return Some(chars.into_iter().collect());
        }
    }
    None
}

fn sql_value_param(value: &SqlValue) -> Result<Box<dyn rusqlite::types::ToSql>, rusqlite::Error> {
    Ok(match value {
        SqlValue::Null => Box::new(Option::<String>::None),
        SqlValue::Bool(v) => Box::new(*v as i64),
        SqlValue::Integer(v) => Box::new(*v),
        SqlValue::Float(v) => Box::new(*v),
        SqlValue::Text(v) => Box::new(v.clone()),
        SqlValue::Blob(v) => Box::new(v.clone()),
        SqlValue::Json(v) => Box::new(
            serde_json::to_string(v)
                .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?,
        ),
        SqlValue::Uuid(v) => Box::new(v.to_string()),
        SqlValue::Timestamp(v) => Box::new(v.timestamp_micros()),
    })
}

fn build_note_filter_where(
    namespace: &str,
    filter: &NoteFilter,
) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
    // When filter.namespaces is non-empty use `namespace IN (...)` for
    // multi-namespace read visibility. Otherwise fall back to equality.
    let (ns_condition, ns_params): (String, Vec<Box<dyn rusqlite::types::ToSql>>) =
        if !filter.namespaces.is_empty() {
            let placeholders: Vec<String> = (1..=filter.namespaces.len())
                .map(|i| format!("?{i}"))
                .collect();
            let params: Vec<Box<dyn rusqlite::types::ToSql>> = filter
                .namespaces
                .iter()
                .map(|ns| -> Box<dyn rusqlite::types::ToSql> { Box::new(ns.clone()) })
                .collect();
            (
                format!("namespace IN ({})", placeholders.join(", ")),
                params,
            )
        } else {
            (
                "namespace = ?1".to_string(),
                vec![Box::new(namespace.to_string())],
            )
        };

    let mut conditions = vec![ns_condition];
    if !filter.include_deleted {
        conditions.push("deleted_at IS NULL".to_string());
    }
    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = ns_params;

    if let Some(kind) = &filter.kind {
        params.push(Box::new(kind.clone()));
        conditions.push(format!("kind = ?{}", params.len()));
    }

    if let Some(since) = filter.min_updated_at {
        params.push(Box::new(since));
        conditions.push(format!("updated_at >= ?{}", params.len()));
    }
    if !filter.tags.is_empty() {
        let mut tag_predicates = Vec::new();
        for tag in &filter.tags {
            params.push(Box::new(tag.clone()));
            tag_predicates.push(format!(
                "EXISTS (SELECT 1 FROM json_each(CASE WHEN json_type(properties,'$.tags')='array' \
                 THEN json_extract(properties,'$.tags') ELSE '[]' END) AS tag \
                 WHERE tag.type='text' AND tag.value = ?{} COLLATE NOCASE)",
                params.len()
            ));
        }
        let join = match filter.tag_mode {
            NoteTagMode::Any => " OR ",
            NoteTagMode::All => " AND ",
        };
        conditions.push(format!("({})", tag_predicates.join(join)));
    }

    for pf in &filter.property_filters {
        match &pf.op {
            FilterOp::Rfc3339Valid => {
                let expr = json_extract_expr(&pf.json_path);
                conditions.push(format!("khive_rfc3339_key({expr}) IS NOT NULL"));
            }
            FilterOp::Rfc3339Gte | FilterOp::Rfc3339Lte | FilterOp::Rfc3339LteOrInvalid => {
                let instant = match &pf.value {
                    SqlValue::Timestamp(instant) => *instant,
                    SqlValue::Text(text) => {
                        text.parse::<chrono::DateTime<chrono::Utc>>()
                            .map_err(|error| {
                                rusqlite::Error::ToSqlConversionFailure(Box::new(error))
                            })?
                    }
                    _ => {
                        return Err(rusqlite::Error::ToSqlConversionFailure(
                            "RFC 3339 filters require a timestamp or text value".into(),
                        ));
                    }
                };
                let expr = json_extract_expr(&pf.json_path);
                let op = if matches!(&pf.op, FilterOp::Rfc3339Gte) {
                    ">="
                } else {
                    "<="
                };
                params.push(Box::new(crate::pool::rfc3339_instant_key(instant)));
                if matches!(&pf.op, FilterOp::Rfc3339LteOrInvalid) {
                    // Byte-for-byte match the built-in-only V44 expression
                    // index. A raw properties edit invalidates due_source and
                    // enters the empty-key candidate range; the strict read
                    // residual below rejects its future/malformed distinction.
                    conditions.push(format!(
                        "CASE WHEN due_source = {expr} \
                         THEN ifnull(strict_due_key, x'') ELSE x'' END <= ?{n} \
                         AND ifnull(khive_rfc3339_strict_key({expr}), x'') <= ?{n}",
                        n = params.len()
                    ));
                } else {
                    conditions.push(format!("khive_rfc3339_key({expr}) {op} ?{}", params.len()));
                }
            }
            FilterOp::EqOrMissing => {
                let expr = json_extract_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!(
                    "({expr} = ?{n} OR {expr} IS NULL)",
                    n = params.len()
                ));
            }
            FilterOp::EqOrMissingIndexed => {
                let expr = json_extract_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!("ifnull({expr}, '') = ?{}", params.len()));
            }
            FilterOp::TextEqOrNonText => {
                let expr = json_extract_expr(&pf.json_path);
                let type_expr = json_type_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                let n = params.len();
                conditions.push(format!(
                    "CASE WHEN {type_expr} = 'text' THEN {expr} ELSE ?{n} END = ?{n}"
                ));
            }
            FilterOp::TextInOrNonText(values) => {
                let expr = json_extract_expr(&pf.json_path);
                let type_expr = json_type_expr(&pf.json_path);
                let mut placeholders = Vec::with_capacity(values.len());
                for value in values {
                    params.push(sql_value_param(value)?);
                    placeholders.push(format!("?{}", params.len()));
                }
                let text_match = if placeholders.is_empty() {
                    "0".to_string()
                } else {
                    format!("{expr} IN ({})", placeholders.join(", "))
                };
                conditions.push(format!(
                    "CASE WHEN {type_expr} = 'text' THEN {text_match} ELSE 1 END"
                ));
            }
            FilterOp::MissingNullOrSpaceEmptyText => {
                let expr = json_extract_expr(&pf.json_path);
                let ty = json_type_expr(&pf.json_path);
                conditions.push(format!(
                    "({ty} IS NULL OR {ty} = 'null' OR ({ty} = 'text' AND trim({expr}) = ''))"
                ));
            }
            FilterOp::TrueOrTextTrue => {
                let expr = json_extract_expr(&pf.json_path);
                let ty = json_type_expr(&pf.json_path);
                conditions.push(format!("({expr} = 'true' OR {ty} = 'true')"));
            }
            FilterOp::JsonTypeEq => {
                let type_expr = json_type_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!("{type_expr} = ?{}", params.len()));
            }
            FilterOp::JsonTypeMissing => {
                let type_expr = json_type_expr(&pf.json_path);
                conditions.push(format!("{type_expr} IS NULL"));
            }
            FilterOp::JsonTypeMissingOrNullIndexed => {
                let expr = json_extract_expr(&pf.json_path);
                let type_expr = json_type_expr(&pf.json_path);
                conditions.push(format!(
                    "ifnull({expr}, '') = '' AND ({type_expr} IS NULL OR {type_expr} = 'null')"
                ));
            }
            FilterOp::EqOrLegacyIndexed => {
                let expr = json_extract_expr(&pf.json_path);
                let type_expr = json_type_expr(&pf.json_path);
                params.push(sql_value_param(&pf.value)?);
                let n = params.len();
                conditions.push(format!(
                    "ifnull({expr}, '') IN (?{n}, '') AND \
                     ({type_expr} IS NULL OR {type_expr} = 'null' OR ifnull({expr}, '') != '')"
                ));
            }
            FilterOp::JsonTypeNeMissing => {
                let type_expr = json_type_expr(&pf.json_path);
                // Inlined as a validated literal, NOT a parameter: the
                // partial unread index (`idx_notes_unread_probe_recipient_direction`)
                // carries this exact predicate in its WHERE clause, and SQLite can
                // only prove a query implies an index predicate when the
                // compared value is known at plan time — a bound parameter
                // defeats the index and the scan degrades to
                // mailbox-proportional work. The value domain is SQLite's
                // closed json_type vocabulary, so inlining is injection-safe
                // by construction.
                let literal = json_type_literal(&pf.value)?;
                conditions.push(format!(
                    "({type_expr} IS NULL OR {type_expr} != '{literal}')"
                ));
            }
            FilterOp::In(values) => {
                let expr = json_extract_expr(&pf.json_path);
                if values.is_empty() {
                    // An empty set can never match any row.
                    conditions.push("0".to_string());
                    continue;
                }
                let mut placeholders = Vec::with_capacity(values.len());
                for v in values {
                    params.push(sql_value_param(v)?);
                    placeholders.push(format!("?{}", params.len()));
                }
                conditions.push(format!("{expr} IN ({})", placeholders.join(", ")));
            }
            FilterOp::TextStartsWithIndexed => {
                let expr = json_extract_expr(&pf.json_path);
                let SqlValue::Text(prefix) = &pf.value else {
                    return Err(rusqlite::Error::ToSqlConversionFailure(
                        "TextStartsWithIndexed takes a text prefix in PropertyFilter.value".into(),
                    ));
                };
                params.push(Box::new(prefix.clone()));
                let lower = params.len();
                match text_prefix_upper_bound(prefix) {
                    Some(upper) => {
                        params.push(Box::new(upper));
                        conditions.push(format!(
                            "({expr} >= ?{lower} AND {expr} < ?{})",
                            params.len()
                        ));
                    }
                    None => {
                        let type_expr = json_type_expr(&pf.json_path);
                        conditions.push(format!("({type_expr} = 'text' AND {expr} >= ?{lower})"));
                    }
                }
            }
            FilterOp::TextColonPrefixBucketIndexed => {
                let SqlValue::Text(prefix) = &pf.value else {
                    return Err(rusqlite::Error::ToSqlConversionFailure(
                        "TextColonPrefixBucketIndexed takes a text prefix".into(),
                    ));
                };
                if prefix
                    .strip_suffix(':')
                    .is_none_or(|head| head.is_empty() || head.contains(':'))
                {
                    return Err(rusqlite::Error::ToSqlConversionFailure(
                        "TextColonPrefixBucketIndexed requires one trailing colon".into(),
                    ));
                }
                let expr = json_extract_expr(&pf.json_path);
                params.push(Box::new(prefix.clone()));
                conditions.push(format!(
                    "substr({expr}, 1, instr({expr}, ':')) = ?{}",
                    params.len()
                ));
            }
            FilterOp::NotInOrMissing(values) => {
                let expr = json_extract_expr(&pf.json_path);
                if values.is_empty() {
                    // Nothing to exclude — every row (including missing) matches.
                    continue;
                }
                let mut placeholders = Vec::with_capacity(values.len());
                for v in values {
                    params.push(sql_value_param(v)?);
                    placeholders.push(format!("?{}", params.len()));
                }
                conditions.push(format!(
                    "({expr} IS NULL OR {expr} NOT IN ({}))",
                    placeholders.join(", ")
                ));
            }
            _ => {
                let expr = json_extract_expr(&pf.json_path);
                let op = match pf.op {
                    FilterOp::Eq => "=",
                    FilterOp::Ne => "!=",
                    FilterOp::Lt => "<",
                    FilterOp::Lte => "<=",
                    FilterOp::Gt => ">",
                    FilterOp::Gte => ">=",
                    FilterOp::EqOrMissing
                    | FilterOp::EqOrMissingIndexed
                    | FilterOp::TextEqOrNonText
                    | FilterOp::TextInOrNonText(_)
                    | FilterOp::JsonTypeEq
                    | FilterOp::JsonTypeMissing
                    | FilterOp::MissingNullOrSpaceEmptyText
                    | FilterOp::TrueOrTextTrue
                    | FilterOp::JsonTypeMissingOrNullIndexed
                    | FilterOp::EqOrLegacyIndexed
                    | FilterOp::JsonTypeNeMissing
                    | FilterOp::In(_)
                    | FilterOp::NotInOrMissing(_)
                    | FilterOp::TextStartsWithIndexed
                    | FilterOp::TextColonPrefixBucketIndexed => {
                        unreachable!()
                    }
                    FilterOp::Rfc3339Valid
                    | FilterOp::Rfc3339Gte
                    | FilterOp::Rfc3339Lte
                    | FilterOp::Rfc3339LteOrInvalid => {
                        unreachable!()
                    }
                };
                params.push(sql_value_param(&pf.value)?);
                conditions.push(format!("{expr} {op} ?{}", params.len()));
            }
        }
    }

    if let Some(min_ts) = filter.min_created_at {
        params.push(Box::new(min_ts));
        conditions.push(format!("created_at >= ?{}", params.len()));
    }

    if let Some(max_ts) = filter.max_created_at {
        params.push(Box::new(max_ts));
        conditions.push(format!("created_at <= ?{}", params.len()));
    }
    if let Some(max_ts) = filter.max_expires_at {
        params.push(Box::new(max_ts));
        conditions.push(format!("expires_at <= ?{}", params.len()));
    }
    if let Some(cutoff) = filter.expiry_fallback {
        params.push(Box::new(cutoff.expires_at_or_before));
        let expiry = params.len();
        params.push(Box::new(cutoff.created_at_or_before));
        let creation = params.len();
        conditions.push(format!(
            "((expires_at IS NOT NULL AND expires_at <= ?{expiry}) OR (expires_at IS NULL AND created_at <= ?{creation}))"
        ));
    }
    if let Some(scope) = &filter.mailbox {
        conditions.push(mailbox_condition(scope, &mut params));
    }

    Ok((format!(" WHERE {}", conditions.join(" AND ")), params))
}

/// SQL form of the row-level mailbox rule for `message` rows. Every other kind
/// passes. JSON `null`, a missing key and a missing `properties` blob all read
/// as "absent", and a non-text routing field never equals the actor id: the
/// row-level rule reads routing fields as strings, while `json_extract` renders
/// an array or object as its JSON text, which an actor id can spell.
fn mailbox_condition(
    scope: &khive_storage::note::NoteMailboxScope,
    params: &mut Vec<Box<dyn rusqlite::types::ToSql>>,
) -> String {
    params.push(Box::new(scope.actor_id.clone()));
    let actor = format!("?{}", params.len());
    let absent = |path: &str| {
        let ty = json_type_expr(path);
        format!("({ty} IS NULL OR {ty} = 'null')")
    };
    let not_text = |path: &str| format!("ifnull({}, '') != 'text'", json_type_expr(path));
    let direction = json_extract_expr("$.direction");
    let text_is_actor = |path: &str| {
        format!(
            "({} = 'text' AND {} = {actor})",
            json_type_expr(path),
            json_extract_expr(path)
        )
    };
    let to = text_is_actor("$.to_actor");
    let from = text_is_actor("$.from_actor");
    let (inbound_legacy, outbound_legacy, unrouted) = if scope.legacy_local {
        (
            format!(" OR {}", absent("$.to_actor")),
            format!(" OR {}", absent("$.from_actor")),
            format!(
                " OR ({} AND {} AND {})",
                absent("$.direction"),
                not_text("$.from_actor"),
                not_text("$.to_actor")
            ),
        )
    } else {
        Default::default()
    };
    format!(
        "(kind != 'message' OR ({direction} = 'inbound' AND ({to}{inbound_legacy})) \
         OR ({direction} = 'outbound' AND ({from}{outbound_legacy})){unrouted})"
    )
}

// ADR-187: match only compiler-emitted equality/IN terms that constrain every
// comm seek key. The unread pin additionally requires its exact partial predicate.
// Inspect whole AND terms, never substrings that might occur inside an OR branch.
fn comm_filter_index_clause(filter: &NoteFilter, where_sql: &str) -> &'static str {
    if filter.kind.as_deref() != Some("message") {
        return "";
    }
    let Some(predicate) = where_sql.strip_prefix(" WHERE ") else {
        return "";
    };
    let terms: Vec<_> = predicate.split(" AND ").collect();
    let numbered_param = |value: &str| {
        value.strip_prefix('?').is_some_and(|number| {
            !number.is_empty() && number.bytes().all(|byte| byte.is_ascii_digit())
        })
    };
    let equality = |prefix: &str| {
        terms
            .iter()
            .any(|term| term.strip_prefix(prefix).is_some_and(numbered_param))
    };
    let namespace = equality("namespace = ")
        || terms.iter().any(|term| {
            term.strip_prefix("namespace IN (")
                .and_then(|term| term.strip_suffix(')'))
                .is_some_and(|values| values.split(", ").all(numbered_param))
        });
    let recipient = equality("ifnull(json_extract(properties, '$.to_actor'), '') = ")
        || terms.contains(&"ifnull(json_extract(properties, '$.to_actor'), '') = ''")
        || terms.iter().any(|term| {
            term.strip_prefix("ifnull(json_extract(properties, '$.to_actor'), '') IN (")
                .and_then(|term| term.strip_suffix(", '')"))
                .is_some_and(numbered_param)
        });
    if !namespace || !terms.contains(&"deleted_at IS NULL") || !equality("kind = ") {
        return "";
    }
    if !equality("json_extract(properties, '$.direction') = ") || !recipient {
        // A thread-keyed read selects the thread's own rows. Left to the
        // planner, `LIMIT` under a `created_at` order picks the creation-order
        // index and evaluates `$.thread_id` on every message in the namespace,
        // and a direction term ties the thread index with
        // `idx_comm_message_outbound_ref`, which walks every outbound message.
        let thread = equality("json_extract(properties, '$.thread_id') = ")
            || terms.iter().any(|term| {
                term.strip_prefix("json_extract(properties, '$.thread_id') IN (")
                    .and_then(|term| term.strip_suffix(')'))
                    .is_some_and(|values| values.split(", ").all(numbered_param))
            });
        return if thread {
            " INDEXED BY idx_comm_message_thread"
        } else {
            ""
        };
    }
    if terms.contains(
        &"(json_type(properties, '$.read') IS NULL OR json_type(properties, '$.read') != 'true')",
    ) {
        // Delegated reads require an exact, string-typed recipient. Match both
        // whole compiler-emitted terms and the bound type value; legacy/own
        // shapes retain the existing ADR-187 pin.
        let typed_exact_recipient =
            equality("ifnull(json_extract(properties, '$.to_actor'), '') = ")
                && equality("json_type(properties, '$.to_actor') = ")
                && filter.property_filters.iter().any(|property| {
                    property.json_path == "$.to_actor"
                        && matches!(property.op, FilterOp::JsonTypeEq)
                        && matches!(&property.value, SqlValue::Text(value) if value == "text")
                });
        if typed_exact_recipient {
            " INDEXED BY idx_notes_unread_probe_recipient_type_direction"
        } else {
            " INDEXED BY idx_notes_unread_probe_recipient_direction"
        }
    } else {
        " INDEXED BY idx_notes_message_recipient_direction"
    }
}

fn build_note_filter_read_clause(
    namespace: &str,
    filter: &NoteFilter,
) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
    let (where_sql, params) = build_note_filter_where(namespace, filter)?;
    let index_clause = comm_filter_index_clause(filter, &where_sql);
    Ok((format!("{index_clause}{where_sql}"), params))
}

/// `SELECT` column list for a plain note-row projection. Used by
/// [`fetch_notes_after`] and `query_notes_filtered_count_free`; the other
/// note-row projection queries in this file (`query_notes`,
/// `query_notes_count_free`, `query_notes_filtered`,
/// `query_notes_filtered_after`, `query_notes_filtered_bounded`) still spell
/// the same column list out inline.
const NOTE_COLUMNS: &str = "id, namespace, kind, status, name, content, salience, decay_factor, \
     expires_at, properties, created_at, updated_at, deleted_at, key, version";

/// The default note page order, served for one namespace by
/// `idx_notes_namespace_created`.
const DEFAULT_NOTE_PAGE_ORDER: &str = " ORDER BY created_at DESC, id ASC";

/// Most namespaces a page read splits into per-namespace arms; a wider visible
/// set keeps the single `namespace IN (...)` statement.
const NAMESPACE_ARMS_MAX: usize = 32;

/// The statement and parameters for one `query_notes_filtered_count_free` page
/// read by `LIMIT`/`OFFSET`.
///
/// Several visible namespaces filter as `namespace IN (...)`, and SQLite cannot
/// take `created_at DESC, id ASC` from `idx_notes_namespace_created` across an
/// `IN` list: it reads every visible live row and sorts them all before taking
/// the page. In that case the page is built from one arm per namespace. Each arm
/// is the same filter with `namespace = ?i`, so it can walk the order index and
/// stop after `offset + limit` rows, and the outer order takes the page from at
/// most that many rows per namespace. A row lives in exactly one namespace and
/// `id` makes the order total, so the page is the one the `IN` form returns.
///
/// Every other read keeps the single `IN` statement: an order other than the
/// default, an index pin chosen for a message read, deleted rows included (the
/// order index is partial on `deleted_at IS NULL`), and a visible set wider than
/// [`NAMESPACE_ARMS_MAX`].
fn note_filter_page_sql(
    namespace: &str,
    filter: &NoteFilter,
    limit: i64,
    offset: i64,
) -> Result<(String, Vec<Box<dyn rusqlite::types::ToSql>>), rusqlite::Error> {
    let mut filter = filter.clone();
    let mut seen = HashSet::new();
    filter.namespaces.retain(|ns| seen.insert(ns.clone()));
    let (read_clause, mut params) = build_note_filter_read_clause(namespace, &filter)?;
    let order_clause = note_filter_page_order_clause(&filter);
    let arms = filter.namespaces.len();
    let in_term = format!(
        " WHERE namespace IN ({})",
        (1..=arms)
            .map(|i| format!("?{i}"))
            .collect::<Vec<_>>()
            .join(", ")
    );
    let arm_rest = read_clause.strip_prefix(in_term.as_str()).filter(|_| {
        (2..=NAMESPACE_ARMS_MAX).contains(&arms)
            && !filter.include_deleted
            && order_clause == DEFAULT_NOTE_PAGE_ORDER
    });
    let Some(rest) = arm_rest else {
        params.push(Box::new(limit));
        params.push(Box::new(offset));
        let limit_idx = params.len() - 1;
        let offset_idx = params.len();
        return Ok((
            format!(
                "SELECT {NOTE_COLUMNS} FROM notes{read_clause}{order_clause} \
                 LIMIT ?{limit_idx} OFFSET ?{offset_idx}"
            ),
            params,
        ));
    };
    params.push(Box::new(offset.saturating_add(limit)));
    let arm_limit_idx = params.len();
    let arm_sql: Vec<String> = (1..=arms)
        .map(|i| {
            format!(
                "SELECT * FROM (SELECT {NOTE_COLUMNS} FROM notes WHERE namespace = ?{i}{rest}\
                 {order_clause} LIMIT ?{arm_limit_idx})"
            )
        })
        .collect();
    params.push(Box::new(limit));
    params.push(Box::new(offset));
    let limit_idx = params.len() - 1;
    let offset_idx = params.len();
    Ok((
        format!(
            "{}{order_clause} LIMIT ?{limit_idx} OFFSET ?{offset_idx}",
            arm_sql.join(" UNION ALL ")
        ),
        params,
    ))
}

fn fetch_notes_after_instant(
    conn: &rusqlite::Connection,
    namespace: &str,
    base_filter: &NoteFilter,
    after: &NoteInstantSeekAfter,
    limit: i64,
) -> Result<Vec<Note>, rusqlite::Error> {
    let (where_sql, mut params) = build_note_filter_read_clause(namespace, base_filter)?;
    params.push(Box::new(after.value.clone()));
    let value_idx = params.len();
    params.push(Box::new(after.id.to_string()));
    let id_idx = params.len();
    params.push(Box::new(limit));
    let limit_idx = params.len();
    let (path, _) = base_filter
        .order_by
        .as_ref()
        .expect("instant cursor requires order_by");
    let expr = json_extract_expr(path);
    let order_clause = note_filter_page_order_clause(base_filter);
    let sql = format!(
        "SELECT {NOTE_COLUMNS} FROM notes{where_sql} \
         AND (khive_rfc3339_key({expr}), {expr}, id) > \
             (khive_rfc3339_key(?{value_idx}), ?{value_idx}, ?{id_idx}) \
         {order_clause} LIMIT ?{limit_idx}"
    );
    let mut stmt = conn.prepare_cached(&sql)?;
    let param_refs: Vec<&dyn rusqlite::types::ToSql> =
        params.iter().map(|param| param.as_ref()).collect();
    let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
    rows.collect()
}

/// Fetch up to `limit` rows strictly after `after` in the notes store's
/// default `created_at DESC, id ASC` total order, for `NoteFilter.after`
/// keyset pagination.
///
/// Deliberately NOT a single `WHERE ... (created_at, id) < (?, ?)` (row
/// value) or `WHERE ... (created_at < ?1 OR (created_at = ?1 AND id > ?2))`
/// predicate: `created_at` sorts DESC while `id` sorts ASC, and neither form
/// gets index-seek treatment from SQLite for a mixed-direction boundary on
/// this build (`EXPLAIN QUERY PLAN` showed the same full ordered index scan
/// as no boundary at all, i.e. exactly the `OFFSET` cost this exists to
/// avoid — confirmed empirically, not assumed). Splitting into two
/// single-direction queries keeps each one a plain equality/range AND-chain,
/// which SQLite reliably turns into an index seek: the first grabs any tied
/// rows at the exact boundary timestamp (`id ASC` order matches the index
/// order within that tie group), the second grabs the (strictly smaller)
/// timestamps that follow. Concatenating the two batches in that order
/// reproduces `created_at DESC, id ASC` exactly with no merge step, since a
/// tie-group's rows all sort before every following (smaller) timestamp.
fn fetch_notes_after(
    conn: &rusqlite::Connection,
    namespace: &str,
    base_filter: &NoteFilter,
    after: &NoteSeekAfter,
    limit: i64,
) -> Result<Vec<Note>, rusqlite::Error> {
    let mut items = Vec::new();
    if limit <= 0 {
        return Ok(items);
    }

    {
        let (where_sql, mut params) = build_note_filter_read_clause(namespace, base_filter)?;
        params.push(Box::new(after.created_at));
        let ts_idx = params.len();
        params.push(Box::new(after.id.to_string()));
        let id_idx = params.len();
        params.push(Box::new(limit));
        let limit_idx = params.len();
        let sql = format!(
            "SELECT {NOTE_COLUMNS} FROM notes{where_sql} AND created_at = ?{ts_idx} \
             AND id > ?{id_idx} ORDER BY id ASC LIMIT ?{limit_idx}"
        );
        let mut stmt = conn.prepare_cached(&sql)?;
        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
            params.iter().map(|p| p.as_ref()).collect();
        let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
        for row in rows {
            items.push(row?);
        }
    }

    let remaining = limit - items.len() as i64;
    if remaining > 0 {
        let (where_sql, mut params) = build_note_filter_read_clause(namespace, base_filter)?;
        params.push(Box::new(after.created_at));
        let ts_idx = params.len();
        params.push(Box::new(remaining));
        let limit_idx = params.len();
        let sql = format!(
            "SELECT {NOTE_COLUMNS} FROM notes{where_sql} AND created_at < ?{ts_idx} \
             ORDER BY created_at DESC, id ASC LIMIT ?{limit_idx}"
        );
        let mut stmt = conn.prepare_cached(&sql)?;
        let param_refs: Vec<&dyn rusqlite::types::ToSql> =
            params.iter().map(|p| p.as_ref()).collect();
        let rows = stmt.query_map(param_refs.as_slice(), read_note)?;
        for row in rows {
            items.push(row?);
        }
    }

    Ok(items)
}

fn execute_filtered_note_property_patch(
    conn: &rusqlite::Connection,
    id: Uuid,
    namespace: &str,
    filter: &NoteFilter,
    json_path: &str,
    value_json: &str,
    updated_at: i64,
) -> Result<usize, rusqlite::Error> {
    let (mut where_clause, mut params) = build_note_filter_where(namespace, filter)?;
    // A read selector cannot relax either property-patch method's live-row guard.
    if filter.include_deleted {
        where_clause.push_str(" AND deleted_at IS NULL");
    }
    let updating_due = json_path == "$.next_attempt_at";
    let due_source = if updating_due {
        serde_json::from_str::<serde_json::Value>(value_json)
            .ok()
            .and_then(|value| value.as_str().map(str::to_owned))
    } else {
        None
    };
    let due_key = due_source
        .as_deref()
        .and_then(crate::pool::strict_rfc3339_key);

    let base = params.len();
    let sql = format!(
        "UPDATE notes SET properties = json_set(COALESCE(properties, '{{}}'), ?{p1}, json(?{p2})), \
         updated_at = ?{p3}, \
         strict_due_key = CASE WHEN ?{p5} = 1 THEN ?{p6} ELSE strict_due_key END, \
         due_source = CASE WHEN ?{p5} = 1 THEN ?{p7} ELSE due_source END \
         {where_clause} \
         AND (properties IS NULL OR json_type(properties) = 'object') AND id = ?{p4}",
        p1 = base + 1,
        p2 = base + 2,
        p3 = base + 3,
        p4 = base + 4,
        p5 = base + 5,
        p6 = base + 6,
        p7 = base + 7,
    );
    params.push(Box::new(json_path.to_string()));
    params.push(Box::new(value_json.to_string()));
    params.push(Box::new(updated_at));
    params.push(Box::new(id.to_string()));
    params.push(Box::new(i64::from(updating_due)));
    params.push(Box::new(due_key));
    params.push(Box::new(due_source));

    let mut stmt = conn.prepare_cached(&sql)?;
    let param_refs: Vec<&dyn rusqlite::types::ToSql> =
        params.iter().map(|param| param.as_ref()).collect();
    stmt.execute(param_refs.as_slice())
}

// =============================================================================
// NoteStore implementation
// =============================================================================

#[async_trait]
impl NoteStore for SqlNoteStore {
    async fn get_live_notes_by_key(
        &self,
        namespace: &str,
        key: &str,
        kind: Option<&str>,
    ) -> StorageResult<Vec<Note>> {
        let namespace = namespace.to_owned();
        let key = key.to_owned();
        let kind = kind.map(str::to_owned);
        self.with_reader("get_live_notes_by_key", move |conn| {
            let (mut clause, mut params) = build_note_where(&namespace, kind.as_deref());
            params.push(Box::new(key));
            clause.push_str(&format!(" AND key = ?{}", params.len()));
            let sql = format!("SELECT {NOTE_COLUMNS} FROM notes{clause} ORDER BY kind ASC");
            let mut stmt = conn.prepare(&sql)?;
            let params: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(params.as_slice(), read_note)?.collect();
            rows
        })
        .await
    }

    async fn query_keyed_notes(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        prefix: &str,
        after: Option<&NoteKeyCursor>,
        page: PageRequest,
    ) -> StorageResult<(Vec<Note>, Option<NoteKeyCursor>)> {
        validate_note_time_order(filter, "query_keyed_notes", false)?;
        if !filter.namespaces.is_empty()
            || filter.order_by.is_some()
            || filter.after.is_some()
            || (after.is_some() && page.offset != 0)
            || page.limit == 0
        {
            return Err(StorageError::InvalidInput { capability: StorageCapability::Notes,
                operation: "query_keyed_notes".into(), message: "keyed paging requires primary namespace, keyed order and a positive limit; cursor excludes offset".into() });
        }
        for property in &filter.property_filters {
            validate_json_path(&property.json_path)?;
        }
        let offset = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_keyed_notes".into(),
            message: "offset exceeds the supported integer range".into(),
        })?;
        let namespace = namespace.to_owned();
        let filter = filter.clone();
        let prefix = prefix.to_owned();
        let after = after.cloned();
        self.with_reader("query_keyed_notes", move |conn| {
            let (mut clause, mut params) = build_note_filter_where(&namespace, &filter)?;
            params.push(Box::new(prefix.clone()));
            clause.push_str(&format!(" AND key IS NOT NULL AND key >= ?{}", params.len()));
            if let Some(upper) = note_key_prefix_successor(&prefix) {
                params.push(Box::new(upper));
                clause.push_str(&format!(" AND key < ?{}", params.len()));
            }
            if let Some(after) = after {
                params.push(Box::new(after.updated_at)); let u = params.len();
                params.push(Box::new(after.key)); let k = params.len();
                params.push(Box::new(after.id.to_string())); let id = params.len();
                clause.push_str(&format!(" AND (updated_at < ?{u} OR (updated_at = ?{u} AND key < ?{k}) \
                    OR (updated_at = ?{u} AND key = ?{k} AND id > ?{id}))"));
            }
            params.push(Box::new(i64::from(page.limit) + 1)); let limit = params.len();
            params.push(Box::new(offset)); let offset = params.len();
            let sql = format!("SELECT {NOTE_COLUMNS} FROM notes{clause} ORDER BY updated_at DESC, key DESC, id ASC LIMIT ?{limit} OFFSET ?{offset}");
            let mut stmt = conn.prepare(&sql)?;
            let params: Vec<&dyn rusqlite::types::ToSql> = params.iter().map(|p| p.as_ref()).collect();
            let mut notes = stmt.query_map(params.as_slice(), read_note)?.collect::<Result<Vec<_>, _>>()?;
            let has_more = notes.len() > page.limit as usize;
            notes.truncate(page.limit as usize);
            let next = if has_more { notes.last().map(NoteKeyCursor::from) } else { None };
            Ok((notes, next))
        }).await
    }

    async fn upsert_note(&self, note: Note) -> Result<(), StorageError> {
        let id_str = note.id.to_string();
        let statement = note_upsert_statement(&note);
        self.with_writer_tx("upsert_note", move |conn| {
            let mut stmt = conn.prepare_cached(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            stmt.raw_execute()?;
            assign_note_seq(conn, &id_str)?;
            Ok(())
        })
        .await
    }

    async fn insert_note_if_absent(&self, note: Note) -> Result<bool, StorageError> {
        let id_str = note.id.to_string();
        let statement = note_insert_if_absent_statement(&note);
        self.with_writer_tx("insert_note_if_absent", move |conn| {
            let mut stmt = conn.prepare_cached(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            let inserted = stmt.raw_execute()? > 0;
            // Only a real insert needs a sequence number. Assigning one on the
            // no-op path would write to a row this call did not create, which
            // is the overwrite this primitive exists to avoid.
            if inserted {
                assign_note_seq(conn, &id_str)?;
            }
            Ok(inserted)
        })
        .await
    }

    async fn replace_note_if_unchanged(
        &self,
        note: Note,
        expected_updated_at: i64,
        expected_deleted_at: Option<i64>,
    ) -> Result<bool, StorageError> {
        let statement =
            note_replace_if_unchanged_statement(&note, expected_updated_at, expected_deleted_at);
        self.with_writer("replace_note_if_unchanged", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn update_note_properties(
        &self,
        id: Uuid,
        properties: Option<serde_json::Value>,
        updated_at: i64,
    ) -> Result<bool, StorageError> {
        let statement = note_update_properties_statement(id, &properties, updated_at);
        self.with_writer("update_note_properties", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn set_note_property(
        &self,
        id: Uuid,
        key: &str,
        value: serde_json::Value,
        updated_at: i64,
    ) -> Result<bool, StorageError> {
        let statement = note_set_property_statement(id, key, &value, updated_at)?;
        self.with_writer("set_note_property", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn try_patch_note_property(
        &self,
        id: Uuid,
        namespace: &str,
        filter: &NoteFilter,
        json_path: &str,
        value: serde_json::Value,
        updated_at: i64,
    ) -> Result<bool, StorageError> {
        validate_note_time_order(filter, "try_patch_note_property", false)?;
        for property in &filter.property_filters {
            validate_json_path(&property.json_path)?;
        }
        let namespace = namespace.to_string();
        let filter = filter.clone();
        let value_json = serde_json::to_string(&value).map_err(|e| {
            StorageError::driver(StorageCapability::Notes, "try_patch_note_property", e)
        })?;
        let json_path = json_path.to_string();

        self.with_writer("try_patch_note_property", move |conn| {
            execute_filtered_note_property_patch(
                conn,
                id,
                &namespace,
                &filter,
                &json_path,
                &value_json,
                updated_at,
            )
            .map(|rows| rows > 0)
        })
        .await
    }

    async fn patch_note_property_atomic(
        &self,
        mut ids: Vec<Uuid>,
        namespace: &str,
        filter: &NoteFilter,
        json_path: &str,
        value: serde_json::Value,
        updated_at: i64,
    ) -> Result<(), StorageError> {
        validate_note_time_order(filter, "patch_note_property_atomic", false)?;
        for property in &filter.property_filters {
            validate_json_path(&property.json_path)?;
        }
        let mut seen = HashSet::with_capacity(ids.len());
        ids.retain(|id| seen.insert(*id));
        if ids.is_empty() {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "patch_note_property_atomic".into(),
                message: "at least one note id is required".to_string(),
            });
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let value_json = serde_json::to_string(&value).map_err(|e| {
            StorageError::driver(StorageCapability::Notes, "patch_note_property_atomic", e)
        })?;
        let json_path = json_path.to_string();

        self.with_writer_tx_storage("patch_note_property_atomic", move |conn| {
            for id in ids {
                let rows = execute_filtered_note_property_patch(
                    conn,
                    id,
                    &namespace,
                    &filter,
                    &json_path,
                    &value_json,
                    updated_at,
                )
                .map_err(|error| map_err(error, "patch_note_property_atomic"))?;
                if rows != 1 {
                    return Err(StorageError::Conflict {
                        capability: StorageCapability::Notes,
                        operation: "patch_note_property_atomic".into(),
                        message: format!(
                            "precondition failed for note {id}: guarded update changed {rows} rows; expected 1"
                        ),
                    });
                }
            }
            Ok(())
        })
        .await
    }

    async fn try_insert_note(&self, note: Note) -> Result<bool, StorageError> {
        self.try_insert_note_with_attachments(note, Vec::new())
            .await
    }

    async fn try_insert_note_with_attachments(
        &self,
        note: Note,
        attachments: Vec<Attachment>,
    ) -> Result<bool, StorageError> {
        let mut attachment_statements = Vec::with_capacity(attachments.len());
        for attachment in attachments {
            attachment.validate()?;
            if attachment.record_uuid != note.id
                || attachment.substrate != AttachmentSubstrate::Note
            {
                return Err(StorageError::InvalidInput {
                    capability: StorageCapability::Attachments,
                    operation: "try_insert_note_with_attachments".into(),
                    message: format!(
                        "attachment {} must target note {}",
                        attachment.role, note.id
                    ),
                });
            }
            attachment_statements.push(attachment_upsert_statement(&attachment)?);
        }
        let namespace = note.namespace.clone();
        let id_str = note.id.to_string();
        let kind_str = note.kind.to_string();
        let status_str = note.status.clone();
        let properties_str = note
            .properties
            .as_ref()
            .map(|v| serde_json::to_string(v).unwrap_or_default());
        let (due_key, due_source) = note_due_key_values(&note.properties);

        // Capture the durable dedup key for verification after a zero-row insert.
        let ext_id_opt: Option<String> = note
            .properties
            .as_ref()
            .and_then(|v| v.get("external_id"))
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string());
        let channel_kind = note
            .properties
            .as_ref()
            .and_then(|v| v.get("channel_kind"))
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();
        let channel_slug = note
            .properties
            .as_ref()
            .and_then(|v| v.get("channel_slug"))
            .and_then(|v| v.as_str())
            .unwrap_or_default()
            .to_string();

        self.with_writer_tx("try_insert_note", move |conn| {
            let rows = conn.execute(
                "INSERT OR IGNORE INTO notes \
                 (id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                  properties, created_at, updated_at, deleted_at, key, strict_due_key, due_source) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)",
                rusqlite::params![
                    id_str,
                    namespace,
                    kind_str,
                    status_str,
                    note.name,
                    note.content,
                    note.salience,
                    note.decay_factor,
                    note.expires_at,
                    properties_str,
                    note.created_at,
                    note.updated_at,
                    note.deleted_at,
                    note.key,
                    due_key,
                    due_source,
                ],
            )?;

            if rows > 0 {
                assign_note_seq(conn, &id_str)?;
                for statement in attachment_statements {
                    let mut stmt = conn.prepare(&statement.sql)?;
                    bind_params(&mut stmt, &statement.params)?;
                    stmt.raw_execute()?;
                }
                return Ok(true);
            }

            // Zero rows: the INSERT was silently skipped by OR IGNORE.
            // Only treat this as a dedup hit when a live note with the same
            // non-empty external_id and exact channel provenance already exists
            // in this namespace and kind.
            // Any other ignored constraint (e.g. a PRIMARY KEY collision) must
            // surface as an error rather than being misreported as a duplicate.
            if let Some(ref ext_id) = ext_id_opt {
                let is_dedup: bool = conn.query_row(
                    "SELECT COUNT(*) > 0 FROM notes \
                     WHERE namespace = ?1 \
                       AND kind = ?2 \
                       AND json_extract(properties, '$.external_id') = ?3 \
                       AND ifnull(json_extract(properties, '$.channel_kind'), '') = ?4 \
                       AND ifnull(json_extract(properties, '$.channel_slug'), '') = ?5 \
                       AND deleted_at IS NULL",
                    rusqlite::params![namespace, kind_str, ext_id, channel_kind, channel_slug],
                    |row| row.get(0),
                )?;
                if is_dedup {
                    return Ok(false);
                }
            }

            // The INSERT was dropped for a reason other than an external_id
            // collision.  Surface it as a constraint error.
            Err(rusqlite::Error::SqliteFailure(
                rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT),
                Some(
                    "try_insert_note: INSERT ignored for a constraint other than \
                     external_id dedup; not masking as deduplication"
                        .to_string(),
                ),
            ))
        })
        .await
    }

    async fn upsert_notes(&self, notes: Vec<Note>) -> Result<BatchWriteSummary, StorageError> {
        let attempted = notes.len() as u64;

        // khive #827: route through `with_writer_tx`
        // instead of hand-rolling BEGIN IMMEDIATE/COMMIT/ROLLBACK here. The
        // old flag-off path only rolled back when the final COMMIT failed —
        // an earlier error from `batch_upsert_notes` (e.g. a failed
        // `assign_note_seq`) propagated via `?` straight out of the closure,
        // skipping ROLLBACK entirely and leaving BEGIN IMMEDIATE open on the
        // shared pool-mutex connection, poisoning every later write on that
        // connection. `with_writer_tx` rolls back on ANY error from `f`, on
        // both the flag-on (WriterTask, which wraps its own transaction) and
        // flag-off (pool-mutex) paths.
        let origin = self.pool.origin();
        self.with_writer_tx("upsert_notes", move |conn| {
            let _tx_handle = khive_storage::tx_registry::register_scoped(
                Some("note_upsert_batch".to_string()),
                origin,
            );
            batch_upsert_notes(conn, &notes, attempted)
        })
        .await
    }

    async fn get_note(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
        let id_str = id.to_string();

        self.with_reader("get_note", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes WHERE id = ?1 AND deleted_at IS NULL",
            )?;
            let mut rows = stmt.query(rusqlite::params![id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_note(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn get_note_including_deleted(&self, id: Uuid) -> Result<Option<Note>, StorageError> {
        let id_str = id.to_string();

        self.with_reader("get_note_including_deleted", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes WHERE id = ?1",
            )?;
            let mut rows = stmt.query(rusqlite::params![id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_note(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn note_sequence(&self, id: Uuid) -> Result<Option<i64>, StorageError> {
        let id = id.to_string();
        self.with_reader("note_sequence", move |conn| {
            conn.query_row(
                "SELECT seq FROM notes_seq WHERE note_id = ?1",
                rusqlite::params![id],
                |row| row.get(0),
            )
            .optional()
        })
        .await
    }

    async fn get_notes_batch(&self, ids: &[Uuid]) -> Result<Vec<Note>, StorageError> {
        self.get_notes_batch_inner(ids, false).await
    }

    async fn get_notes_batch_including_deleted(
        &self,
        ids: &[Uuid],
    ) -> Result<Vec<Note>, StorageError> {
        self.get_notes_batch_inner(ids, true).await
    }

    async fn get_note_visibility_batch(
        &self,
        ids: &[Uuid],
    ) -> Result<Vec<NoteVisibility>, StorageError> {
        if ids.is_empty() {
            return Ok(vec![]);
        }
        // Stay below SQLite's baseline 999-parameter limit for large ANN candidate sets.
        const CHUNK: usize = 900;
        let id_strings: Vec<String> = ids.iter().map(Uuid::to_string).collect();
        let mut result = Vec::with_capacity(ids.len());
        for chunk in id_strings.chunks(CHUNK) {
            let chunk_owned = chunk.to_vec();
            let rows = self
                .with_reader("get_note_visibility_batch", move |conn| {
                    let placeholders = (1..=chunk_owned.len())
                        .map(|i| format!("?{i}"))
                        .collect::<Vec<_>>()
                        .join(", ");
                    let sql = format!(
                        "SELECT id, namespace, deleted_at FROM notes WHERE id IN ({placeholders})"
                    );
                    let mut stmt = conn.prepare(&sql)?;
                    let params: Vec<&dyn rusqlite::types::ToSql> = chunk_owned
                        .iter()
                        .map(|s| s as &dyn rusqlite::types::ToSql)
                        .collect();
                    let rows = stmt.query_map(params.as_slice(), |row| {
                        let id: String = row.get(0)?;
                        Ok(NoteVisibility {
                            id: parse_uuid(&id)?,
                            namespace: row.get(1)?,
                            deleted_at: row.get(2)?,
                        })
                    })?;
                    rows.collect::<rusqlite::Result<Vec<_>>>()
                })
                .await?;
            result.extend(rows);
        }
        Ok(result)
    }

    async fn delete_note(&self, id: Uuid, mode: DeleteMode) -> Result<bool, StorageError> {
        match mode {
            DeleteMode::Soft => {
                let now = chrono::Utc::now().timestamp_micros();
                let statement = note_soft_delete_statement(id, now);
                self.with_writer("delete_note_soft", move |conn| {
                    let mut stmt = conn.prepare(&statement.sql)?;
                    bind_params(&mut stmt, &statement.params)?;
                    Ok(stmt.raw_execute()? > 0)
                })
                .await
            }
            DeleteMode::Hard => {
                let note_statement = note_hard_delete_statement(id);
                let attachment_statement =
                    delete_record_attachments_statement(id, AttachmentSubstrate::Note);
                self.with_writer_tx("delete_note_hard", move |conn| {
                    let mut note_stmt = conn.prepare(&note_statement.sql)?;
                    bind_params(&mut note_stmt, &note_statement.params)?;
                    let deleted = note_stmt.raw_execute()? > 0;
                    drop(note_stmt);
                    if deleted {
                        let mut attachment_stmt = conn.prepare(&attachment_statement.sql)?;
                        bind_params(&mut attachment_stmt, &attachment_statement.params)?;
                        attachment_stmt.raw_execute()?;
                    }
                    Ok(deleted)
                })
                .await
            }
        }
    }

    async fn query_notes(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        let namespace = namespace.to_string();
        let kind = kind.map(|k| k.to_string());
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_reader("query_notes", move |conn| {
            let (count_sql, count_params) = build_note_where(&namespace, kind.as_deref());
            let count_sql = format!("SELECT COUNT(*) FROM notes{count_sql}");

            let (where_sql, mut data_params) = build_note_where(&namespace, kind.as_deref());
            data_params.push(Box::new(limit_i64));
            data_params.push(Box::new(offset_i64));

            let limit_idx = data_params.len() - 1;
            let offset_idx = data_params.len();

            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, expires_at, \
                 properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes{} ORDER BY created_at DESC, id ASC LIMIT ?{} OFFSET ?{}",
                where_sql, limit_idx, offset_idx,
            );

            query_note_page_snapshot(
                conn,
                "query_notes",
                &namespace,
                &count_sql,
                &count_params,
                &data_sql,
                &data_params,
            )
        })
        .await
    }

    async fn query_notes_count_free(
        &self,
        namespace: &str,
        kind: Option<&str>,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        let namespace = namespace.to_string();
        let kind = kind.map(str::to_string);
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_count_free".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_reader("query_notes_count_free", move |conn| {
            let (where_sql, mut params) = build_note_where(&namespace, kind.as_deref());
            params.push(Box::new(limit_i64));
            params.push(Box::new(offset_i64));
            let limit_idx = params.len() - 1;
            let offset_idx = params.len();
            let sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes{where_sql} ORDER BY created_at DESC, id ASC \
                 LIMIT ?{limit_idx} OFFSET ?{offset_idx}"
            );

            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|param| param.as_ref()).collect();
            let mut rows = stmt.query(param_refs.as_slice())?;
            let mut items = Vec::new();
            while let Some(row) = rows.next()? {
                items.push(read_note(row)?);
            }

            Ok(Page { items, total: None })
        })
        .await
    }

    async fn query_notes_filtered(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        validate_note_time_order(filter, "query_notes_filtered", true)?;
        if filter.unordered {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered".into(),
                message: "NoteFilter.unordered is supported only by \
                          query_notes_filtered_count_free"
                    .into(),
            });
        }
        // Validate paths before entering spawn_blocking (closures return rusqlite::Error).
        for pf in &filter.property_filters {
            validate_json_path(&pf.json_path)?;
        }
        if let Some((path, _)) = &filter.order_by {
            validate_json_path(path)?;
        }
        if filter.order_by_instant && !matches!(filter.order_by.as_ref(), Some((_, SortDir::Asc))) {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered".into(),
                message: "order_by_instant requires an ascending property order".into(),
            });
        }
        if filter.after.is_some() || filter.after_instant.is_some() {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered".into(),
                message: "NoteFilter.after or after_instant (keyset pagination) is not supported by this \
                          method: it computes an exact COUNT(*) total over the whole \
                          matching set, which has no defined meaning paired with a seek \
                          boundary; use query_notes_filtered_count_free instead, which \
                          seeks and returns total: None"
                    .into(),
            });
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_filtered".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_indexed_reader("query_notes_filtered", move |conn| {
            let (count_sql, count_params) = build_note_filter_read_clause(&namespace, &filter)?;
            let count_sql = format!("SELECT COUNT(*) FROM notes{count_sql}");

            let (where_sql, mut data_params) = build_note_filter_read_clause(&namespace, &filter)?;
            data_params.push(Box::new(limit_i64));
            data_params.push(Box::new(offset_i64));

            // The total order removes tie-order instability only — offset
            // paging can still duplicate or skip rows across separate requests
            // under concurrent inserts/deletes or sort-key updates.
            let order_clause = note_filter_page_order_clause(&filter);

            let limit_idx = data_params.len() - 1;
            let offset_idx = data_params.len();
            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes{}{order_clause} LIMIT ?{} OFFSET ?{}",
                where_sql, limit_idx, offset_idx,
            );

            query_note_page_snapshot(
                conn,
                "query_notes_filtered",
                &namespace,
                &count_sql,
                &count_params,
                &data_sql,
                &data_params,
            )
        })
        .await
    }

    async fn query_notes_filtered_count_free(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        page: PageRequest,
    ) -> Result<Page<Note>, StorageError> {
        validate_note_time_order(filter, "query_notes_filtered_count_free", true)?;
        for property_filter in &filter.property_filters {
            validate_json_path(&property_filter.json_path)?;
        }
        if let Some((path, _)) = &filter.order_by {
            validate_json_path(path)?;
        }
        if filter.order_by_instant && !matches!(filter.order_by.as_ref(), Some((_, SortDir::Asc))) {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "order_by_instant requires an ascending property order".into(),
            });
        }
        if filter.order_by_instant && filter.unordered {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "order_by_instant is incompatible with unordered pages".into(),
            });
        }
        if filter.after_instant.is_some() && !filter.order_by_instant {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "after_instant requires order_by_instant".into(),
            });
        }
        if filter.after.is_some() && filter.after_instant.is_some() {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "after and after_instant are mutually exclusive".into(),
            });
        }
        if filter.after.is_some() && filter.order_by.is_some() {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "NoteFilter.after is incompatible with a custom order_by; it is \
                          defined only over the default created_at DESC, id ASC order"
                    .into(),
            });
        }
        if (filter.after.is_some() || filter.after_instant.is_some()) && page.offset != 0 {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_count_free".into(),
                message: "NoteFilter.after or after_instant and a non-zero PageRequest.offset \
                          are mutually exclusive pagination strategies; pass offset: 0"
                    .into(),
            });
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Notes,
            operation: "query_notes_filtered_count_free".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;

        self.with_indexed_reader("query_notes_filtered_count_free", move |conn| {
            if let Some(after) = &filter.after_instant {
                let mut base_filter = filter.clone();
                base_filter.after_instant = None;
                let items =
                    fetch_notes_after_instant(conn, &namespace, &base_filter, after, limit_i64)?;
                return Ok(Page { items, total: None });
            }
            if let Some(after) = &filter.after {
                let mut base_filter = filter.clone();
                base_filter.after = None;
                let items = fetch_notes_after(conn, &namespace, &base_filter, after, limit_i64)?;
                return Ok(Page { items, total: None });
            }

            let (sql, params) = note_filter_page_sql(&namespace, &filter, limit_i64, offset_i64)?;

            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|param| param.as_ref()).collect();
            let mut rows = stmt.query(param_refs.as_slice())?;
            let mut items = Vec::new();
            while let Some(row) = rows.next()? {
                items.push(read_note(row)?);
                // A single SQLite statement pins its read snapshot on the
                // first step. The test seam pauses there so a concurrent WAL
                // commit can prove the remainder stays on that snapshot.
                #[cfg(test)]
                if items.len() == 1 {
                    tests::page_snapshot_seam::hook("query_notes_filtered_count_free", &namespace);
                }
            }

            Ok(Page { items, total: None })
        })
        .await
    }

    async fn count_notes_filtered_in_snapshot(
        &self,
        namespace: &str,
        filters: &[NoteFilter],
    ) -> Result<Vec<u64>, StorageError> {
        for filter in filters {
            for pf in &filter.property_filters {
                validate_json_path(&pf.json_path)?;
            }
        }

        let namespace = namespace.to_string();
        let filters = filters.to_vec();
        self.with_indexed_reader("count_notes_filtered_in_snapshot", move |conn| {
            let tx = rusqlite::Transaction::new_unchecked(
                conn,
                rusqlite::TransactionBehavior::Deferred,
            )?;
            let mut counts = Vec::with_capacity(filters.len());
            for filter in filters.iter() {
                #[cfg(test)]
                if !counts.is_empty() {
                    tests::page_snapshot_seam::hook("count_notes_filtered_in_snapshot", &namespace);
                }

                let (where_sql, params) = build_note_filter_read_clause(&namespace, filter)?;
                let sql = format!("SELECT COUNT(*) FROM notes{where_sql}");
                let mut stmt = tx.prepare(&sql)?;
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    params.iter().map(|param| param.as_ref()).collect();
                let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
                counts.push(count as u64);
            }
            tx.commit()?;
            Ok(counts)
        })
        .await
    }

    async fn count_notes_filtered_bounded_in_snapshot(
        &self,
        namespace: &str,
        filters: &[NoteFilter],
        cap: u32,
    ) -> Result<Vec<BoundedCount>, StorageError> {
        for filter in filters {
            for property_filter in &filter.property_filters {
                validate_json_path(&property_filter.json_path)?;
            }
        }

        let namespace = namespace.to_string();
        let filters = filters.to_vec();
        let cap_u64 = u64::from(cap);
        let probe_limit_i64 = i64::from(cap) + 1;
        self.with_indexed_reader("count_notes_filtered_bounded_in_snapshot", move |conn| {
            let tx = rusqlite::Transaction::new_unchecked(
                conn,
                rusqlite::TransactionBehavior::Deferred,
            )?;
            let mut counts = Vec::with_capacity(filters.len());
            for filter in &filters {
                #[cfg(test)]
                if !counts.is_empty() {
                    tests::page_snapshot_seam::hook(
                        "count_notes_filtered_bounded_in_snapshot",
                        &namespace,
                    );
                }

                let (where_sql, mut params) = build_note_filter_read_clause(&namespace, filter)?;
                params.push(Box::new(probe_limit_i64));
                let limit_idx = params.len();
                // The inner LIMIT is the work bound. Selecting a constant
                // and omitting ORDER BY lets SQLite stop after cap + 1
                // matching index entries instead of hydrating or sorting
                // the complete population.
                let sql = format!(
                    "SELECT COUNT(*) FROM (SELECT 1 FROM notes{where_sql} LIMIT ?{limit_idx})"
                );
                let mut stmt = tx.prepare(&sql)?;
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    params.iter().map(|param| param.as_ref()).collect();
                let observed: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
                let observed = observed as u64;
                counts.push(BoundedCount {
                    count: observed.min(cap_u64),
                    cap: cap_u64,
                    saturated: observed > cap_u64,
                });
            }
            tx.commit()?;
            Ok(counts)
        })
        .await
    }

    async fn query_notes_filtered_after(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        after: Option<SeekCursor>,
        limit: u32,
    ) -> Result<SeekPage<Note>, StorageError> {
        validate_note_time_order(filter, "query_notes_filtered_after", false)?;
        if limit == 0 {
            return Ok(SeekPage::default());
        }
        if filter.order_by.is_some() {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_after".into(),
                message: "custom order_by is not compatible with insertion-sequence pagination"
                    .into(),
            });
        }
        for property_filter in &filter.property_filters {
            validate_json_path(&property_filter.json_path)?;
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_usize = limit as usize;
        let probe_limit_i64 = i64::from(limit) + 1;
        self.with_reader("query_notes_filtered_after", move |conn| {
            let (mut where_sql, mut params) = build_note_filter_where(&namespace, &filter)?;
            if let Some(cursor) = after {
                params.push(Box::new(cursor.sequence));
                where_sql.push_str(&format!(" AND notes_seq.seq > ?{}", params.len()));
            }
            params.push(Box::new(probe_limit_i64));
            let limit_idx = params.len();
            // CROSS JOIN fixes the ledger as the outer loop, preserving an
            // indexed `seq > boundary` scan with no full-match sort.
            let sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at, key, version, notes_seq.seq \
                 FROM notes_seq CROSS JOIN notes ON notes.id = notes_seq.note_id{where_sql} \
                 ORDER BY notes_seq.seq ASC LIMIT ?{limit_idx}"
            );
            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|param| param.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), |row| {
                Ok((read_note(row)?, row.get::<_, i64>(15)?))
            })?;
            let mut entries = rows.collect::<Result<Vec<_>, _>>()?;
            let has_more = entries.len() > limit_usize;
            if has_more {
                entries.truncate(limit_usize);
            }
            let next_after = if has_more {
                entries.last().map(|(note, sequence)| SeekCursor {
                    sequence: *sequence,
                    id: note.id,
                })
            } else {
                None
            };
            let items = entries.into_iter().map(|(note, _)| note).collect();
            Ok(SeekPage { items, next_after })
        })
        .await
    }

    async fn query_notes_filtered_bounded(
        &self,
        namespace: &str,
        filter: &NoteFilter,
        max_rows: u32,
    ) -> Result<Vec<Note>, StorageError> {
        validate_note_time_order(filter, "query_notes_filtered_bounded", true)?;
        if filter.unordered {
            return Err(StorageError::InvalidInput {
                capability: StorageCapability::Notes,
                operation: "query_notes_filtered_bounded".into(),
                message: "NoteFilter.unordered is supported only by \
                          query_notes_filtered_count_free"
                    .into(),
            });
        }
        for pf in &filter.property_filters {
            validate_json_path(&pf.json_path)?;
        }
        if let Some((path, _)) = &filter.order_by {
            validate_json_path(path)?;
        }

        let namespace = namespace.to_string();
        let filter = filter.clone();
        let limit_i64 = i64::from(max_rows) + 1;

        self.with_indexed_reader("query_notes_filtered_bounded", move |conn| {
            let (where_sql, mut data_params) = build_note_filter_read_clause(&namespace, &filter)?;
            data_params.push(Box::new(limit_i64));
            let limit_idx = data_params.len();

            // Tie-break on `id` in addition to the primary sort key so the
            // snapshot ordering is fully deterministic even when many rows
            // share the same `created_at` (or the same custom sort value).
            let order_clause = match (filter.time_order, &filter.order_by) {
                (Some(order), _) => note_time_order_clause(order).to_string(),
                (None, Some((path, dir))) => {
                    let dir_str = match dir {
                        SortDir::Asc => "ASC",
                        SortDir::Desc => "DESC",
                    };
                    format!(" ORDER BY {} {dir_str}, id ASC", json_extract_expr(path))
                }
                (None, None) => " ORDER BY created_at DESC, id ASC".to_string(),
            };

            let data_sql = format!(
                "SELECT id, namespace, kind, status, name, content, salience, decay_factor, \
                 expires_at, properties, created_at, updated_at, deleted_at, key, version \
                 FROM notes{where_sql}{order_clause} LIMIT ?{limit_idx}",
            );

            let mut stmt = conn.prepare(&data_sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                data_params.iter().map(|p| p.as_ref()).collect();
            let rows = stmt.query_map(param_refs.as_slice(), read_note)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }
            Ok(items)
        })
        .await
    }

    async fn count_notes(&self, namespace: &str, kind: Option<&str>) -> Result<u64, StorageError> {
        let namespace = namespace.to_string();
        let kind = kind.map(|k| k.to_string());

        self.with_reader("count_notes", move |conn| {
            let (where_sql, params) = build_note_where(&namespace, kind.as_deref());
            let sql = format!("SELECT COUNT(*) FROM notes{}", where_sql);
            let mut stmt = conn.prepare(&sql)?;
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|p| p.as_ref()).collect();
            let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
            Ok(count as u64)
        })
        .await
    }

    async fn count_notes_in_namespaces(
        &self,
        namespaces: &[String],
        kind: Option<&str>,
    ) -> Result<u64, StorageError> {
        let namespaces: Vec<String> = namespaces
            .iter()
            .cloned()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        let kind = kind.map(str::to_string);

        self.with_reader("count_notes_in_namespaces", move |conn| {
            let mut total = 0;
            for chunk in namespaces.chunks(NAMESPACE_COUNT_CHUNK_SIZE) {
                let (where_sql, params) = build_note_where_for_namespaces(chunk, kind.as_deref());
                let sql = format!("SELECT COUNT(*) FROM notes{where_sql}");
                let mut stmt = conn.prepare(&sql)?;
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    params.iter().map(|p| p.as_ref()).collect();
                let count: i64 = stmt.query_row(param_refs.as_slice(), |row| row.get(0))?;
                total += count as u64;
            }
            Ok(total)
        })
        .await
    }
}

// =============================================================================
// DDL
// =============================================================================

const NOTES_DDL: &str = include_str!("../../sql/notes-ddl.sql");

/// Same anti-join repair as `sql/008-notes-seq-repair.sql` (the V8 forward
/// migration) -- shared via `include_str!` from that single source file
/// rather than duplicated as SQL text. `INSERT OR IGNORE` targets notes
/// still missing a `notes_seq` row specifically, so it is correct to run
/// against a fresh ledger, a partially populated one, or an already fully
/// repaired one.
const NOTES_SEQ_REPAIR_DDL: &str = include_str!("../../sql/008-notes-seq-repair.sql");

pub(crate) fn ensure_notes_schema(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(NOTES_DDL)
}

/// Anti-join backfill of `notes_seq` for any note still missing a row
/// (khive #827). Scans `notes` in full, so callers MUST gate this to
/// run at most once per backend/pool rather than on every store acquisition
/// (khive #827) -- see
/// `StorageBackend::notes_for_namespace`.
pub(crate) fn repair_notes_seq(conn: &rusqlite::Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(NOTES_SEQ_REPAIR_DDL)
}

#[cfg(test)]
#[path = "note_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "comm_filter_plan_tests.rs"]
mod comm_filter_plan_tests;

#[cfg(test)]
#[path = "note_list_plan_tests.rs"]
mod note_list_plan_tests;

#[cfg(test)]
#[path = "note_busy_tests.rs"]
mod direct_busy_tests;