khive-db 0.9.0

//! SQL-backed `GraphStore`: edge CRUD, neighbor queries, and bounded BFS traversal.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;

use async_trait::async_trait;
use chrono::{DateTime, TimeZone, Utc};
use rusqlite::OptionalExtension;
use uuid::Uuid;

use khive_storage::error::StorageError;
use khive_storage::types::{
    BatchWriteSummary, DeleteMode, DirectedNeighborHit, Direction, Edge, EdgeEndpointBaseCounts,
    EdgeFilter, EdgeSeekPage, EdgeSortField, EdgeUpsertDisposition, EdgeUpsertRefusal,
    EdgeUpsertRequest, EdgeUpsertResult, GraphPath, GuardedBatchOutcome, GuardedBatchRefusal,
    GuardedEdgeBatchRefusal, GuardedEdgeBatchUpsertOutcome, GuardedEdgeUpsertOutcome,
    GuardedWriteOutcome, MissingEndpoints, NeighborCursor, NeighborHit, NeighborQuery, Page,
    PageRequest, PathNode, SeekCursor, SeekPage, SortDirection, SortOrder, SqlStatement, SqlValue,
    TraversalExecutionBudget, TraversalOptions, TraversalRequest,
};
use khive_storage::GraphStore;
use khive_storage::LinkId;
use khive_storage::StorageCapability;
use khive_types::EdgeRelation;

use crate::error::SqliteError;
use crate::pool::ConnectionPool;
use crate::sql_bridge::bind_params;
use crate::writer_task::WriterTaskHandle;

/// Map a rusqlite error to `StorageError` with `Graph` capability.
fn map_err(e: rusqlite::Error, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Graph, op, e)
}

fn map_sqlite_err(e: SqliteError, op: &'static str) -> StorageError {
    StorageError::driver(StorageCapability::Graph, op, e)
}

fn resurrection_required_error(operation: &'static str, edge: &Edge) -> StorageError {
    StorageError::Conflict {
        capability: StorageCapability::Graph,
        operation: operation.into(),
        message: format!(
            "edge {} is soft-deleted; explicit resurrection is required",
            edge.id
        ),
    }
}

const NAMESPACE_COUNT_CHUNK_SIZE: usize = 500;

// Latest eligible annotation of one target. First read at most three incident
// `annotates` edge rows, tombstones included so the probe cannot scan a long
// deleted prefix. The unique (namespace, source, target, relation) index allows
// one row per source, so two or fewer rows is the complete incident set and
// those notes are filtered and ordered directly. Three rows falls back to
// walking notes newest-first, which keeps a long receipt history from being
// enumerated or sorted. That fallback still visits newer unrelated notes when
// a target with three or more incident edges has only old eligible annotations.
const LATEST_ANNOTATING_NOTE_SQL: &str = r#"WITH incident AS MATERIALIZED (
    SELECT source_id, deleted_at
    FROM graph_edges INDEXED BY idx_graph_edges_ns_tgt_rel
    WHERE namespace = ?1 AND target_id = ?2 AND relation = 'annotates'
    LIMIT 3
)
SELECT result.id, result.created_at
FROM notes AS result
WHERE result.id = CASE WHEN (SELECT count(*) FROM incident) <= 2 THEN (
    SELECT n.id
    FROM incident AS e CROSS JOIN notes AS n
    WHERE n.id = e.source_id AND e.deleted_at IS NULL
      AND n.deleted_at IS NULL AND n.kind = ?3
      AND EXISTS (SELECT 1 FROM json_each(CASE
          WHEN json_type(n.properties, '$.tags') = 'array'
          THEN json_extract(n.properties, '$.tags') ELSE '[]' END) AS tag
          WHERE tag.type = 'text' AND tag.value = ?4 COLLATE BINARY)
    ORDER BY n.created_at DESC, n.id ASC LIMIT 1
) ELSE (
    SELECT n.id
    FROM notes AS n INDEXED BY idx_notes_created
    WHERE n.deleted_at IS NULL AND n.kind = ?3
      AND EXISTS (SELECT 1 FROM json_each(CASE
          WHEN json_type(n.properties, '$.tags') = 'array'
          THEN json_extract(n.properties, '$.tags') ELSE '[]' END) AS tag
          WHERE tag.type = 'text' AND tag.value = ?4 COLLATE BINARY)
      AND EXISTS (SELECT 1 FROM graph_edges AS e INDEXED BY idx_graph_edges_unique_triple
          WHERE e.namespace = ?1 AND e.source_id = n.id AND e.target_id = ?2
            AND e.relation = 'annotates' AND e.deleted_at IS NULL)
    ORDER BY n.created_at DESC, n.id ASC LIMIT 1
) END"#;

// ---------------------------------------------------------------------------
// Pure statement builders (ADR-099 B3 r6 structural cut) — see entity.rs's
// sibling block for the full rationale. `upsert_edge`/`delete_edge` below and
// `purge_incident_edges` (plus ADR-099's atomic prepare path in
// `khive-runtime`, and `khive-runtime::operations::update_edge`'s
// non-symmetric branch) all call these.
// ---------------------------------------------------------------------------

/// Replacement clause shared by canonical and guarded upserts. Live rows use
/// replace semantics. Tombstones are excluded unless resurrection was
/// explicitly requested; only that branch clears `deleted_at`.
fn edge_conflict_clause(resurrect: bool) -> String {
    let deleted_at = if resurrect {
        "NULL"
    } else {
        "graph_edges.deleted_at"
    };
    let predicate = if resurrect {
        ""
    } else {
        " WHERE graph_edges.deleted_at IS NULL"
    };
    format!(
        "weight = excluded.weight, \
         updated_at = excluded.updated_at, \
         deleted_at = {deleted_at}, \
         metadata = excluded.metadata, \
         target_backend = excluded.target_backend{predicate}"
    )
}

/// A `WHERE`-clause fragment asserting the id bound to `id_param` (an SQL
/// placeholder like `?3`) resolves to a live edge endpoint — an
/// undeleted entity or note, an event (append-only, no `deleted_at`), or an
/// undeleted edge (the `annotates` relation's target may be any substrate,
/// including another edge; ADR-002/ADR-055). Shared by
/// [`edge_insert_guarded_by_endpoints_statement`] and
/// [`edge_endpoints_exist`] (#769) so the two "does this endpoint still
/// exist" probes — the guarded single-row insert and the guarded batch
/// pre-check — can never drift on which substrates count as valid
/// endpoints.
fn endpoint_exists_clause(id_param: &str) -> String {
    format!(
        "EXISTS (SELECT 1 FROM entities WHERE id = {id_param} AND deleted_at IS NULL) \
         OR EXISTS (SELECT 1 FROM notes WHERE id = {id_param} AND deleted_at IS NULL) \
         OR EXISTS (SELECT 1 FROM events WHERE id = {id_param}) \
         OR EXISTS (SELECT 1 FROM graph_edges WHERE id = {id_param} AND deleted_at IS NULL)"
    )
}

/// Assert an exact live edge snapshot without updating its row or firing write
/// triggers. Retained links additionally require their endpoints to remain live;
/// deletion assertions may remove an annotation whose former target is gone.
pub fn edge_snapshot_assertion_statement(edge: &Edge, require_endpoints: bool) -> SqlStatement {
    let mut sql = "SELECT 1 FROM graph_edges WHERE id=?1 AND namespace=?2 \
        AND source_id=?3 AND target_id=?4 AND relation=?5 \
        AND updated_at=?6 AND deleted_at IS NULL"
        .to_string();
    if require_endpoints {
        sql.push_str(&format!(
            " AND ({}) AND ({})",
            endpoint_exists_clause("?3"),
            endpoint_exists_clause("?4")
        ));
    }
    SqlStatement {
        sql,
        params: vec![
            SqlValue::Text(Uuid::from(edge.id).to_string()),
            SqlValue::Text(edge.namespace.clone()),
            SqlValue::Text(edge.source_id.to_string()),
            SqlValue::Text(edge.target_id.to_string()),
            SqlValue::Text(edge.relation.to_string()),
            SqlValue::Integer(edge.updated_at.timestamp_micros()),
        ],
        label: Some("edge-snapshot-assertion".into()),
    }
}

/// The exact natural-key-upserting `INSERT ... ON CONFLICT` this store's
/// `upsert_edge` issues. Canonicalizes symmetric-relation endpoints first,
/// matching `upsert_edge`'s own call to `canonical_edge_endpoints`.
pub fn edge_upsert_statement(edge: &Edge) -> SqlStatement {
    edge_upsert_statement_with_resurrection(edge, false)
}

/// Policy-aware form of [`edge_upsert_statement`].
pub fn edge_upsert_statement_with_resurrection(edge: &Edge, resurrect: bool) -> SqlStatement {
    let (source_id, target_id) =
        canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
    let metadata_str = edge
        .metadata
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    let conflict_clause = edge_conflict_clause(resurrect);
    SqlStatement {
        sql: format!(
            "INSERT INTO graph_edges \
              (namespace, id, source_id, target_id, relation, weight, \
               created_at, updated_at, deleted_at, metadata, target_backend) \
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) \
              ON CONFLICT(namespace, id) DO UPDATE SET \
                  source_id = excluded.source_id, \
                  target_id = excluded.target_id, \
                  relation = excluded.relation, \
                  {conflict_clause} \
              ON CONFLICT(namespace, source_id, target_id, relation) DO UPDATE SET \
                  {conflict_clause}"
        ),
        params: vec![
            SqlValue::Text(edge.namespace.clone()),
            SqlValue::Text(Uuid::from(edge.id).to_string()),
            SqlValue::Text(source_id.to_string()),
            SqlValue::Text(target_id.to_string()),
            SqlValue::Text(edge.relation.to_string()),
            SqlValue::Float(edge.weight),
            SqlValue::Integer(edge.created_at.timestamp_micros()),
            SqlValue::Integer(edge.updated_at.timestamp_micros()),
            match edge.deleted_at {
                Some(t) => SqlValue::Integer(t.timestamp_micros()),
                None => SqlValue::Null,
            },
            match metadata_str {
                Some(m) => SqlValue::Text(m),
                None => SqlValue::Null,
            },
            match &edge.target_backend {
                Some(b) => SqlValue::Text(b.clone()),
                None => SqlValue::Null,
            },
        ],
        label: Some("edge-upsert".to_string()),
    }
}

/// Insert a new edge only while both endpoints still exist.
/// Competing IDs and natural keys, including tombstones, cause a constraint
/// error rather than replacing or restoring the competing row. Callers must
/// require one affected row to reject an endpoint removed after prepare.
pub fn edge_insert_only_guarded_by_endpoints_statement(edge: &Edge) -> SqlStatement {
    let mut statement = edge_upsert_statement(edge);
    let src_exists = endpoint_exists_clause("?3");
    let tgt_exists = endpoint_exists_clause("?4");
    statement.sql = format!(
        "INSERT INTO graph_edges \
          (namespace, id, source_id, target_id, relation, weight, \
           created_at, updated_at, deleted_at, metadata, target_backend) \
          SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11 \
          WHERE ({src_exists}) AND ({tgt_exists})"
    );
    statement.label = Some("edge-insert-only-where-endpoints-exist".to_string());
    statement
}

/// Conditional-insert companion to [`edge_upsert_statement`]. Conflicts on
/// either the id or natural key leave the existing edge untouched so callers
/// can read the winner and explicitly reapply their intended delta.
pub fn edge_insert_if_absent_statement(edge: &Edge) -> SqlStatement {
    let mut statement = edge_upsert_statement(edge);
    statement.sql = "INSERT INTO graph_edges \
              (namespace, id, source_id, target_id, relation, weight, \
               created_at, updated_at, deleted_at, metadata, target_backend) \
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11) \
              ON CONFLICT DO NOTHING"
        .to_string();
    statement.label = Some("edge-insert-if-absent".to_string());
    statement
}

/// Full-edge compare-and-swap update used after caller-side normalization
/// was derived from a read snapshot. Unlike [`edge_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. Mirrors `note_replace_if_unchanged_statement`
/// (`crates/khive-db/src/stores/note.rs`). `created_at` is deliberately
/// excluded from the `SET` list, matching the note/entity siblings — a
/// replacement never rewrites the row's original creation time.
pub fn edge_replace_if_unchanged_statement(
    edge: &Edge,
    expected_updated_at: DateTime<Utc>,
    expected_deleted_at: Option<DateTime<Utc>>,
) -> SqlStatement {
    let (source_id, target_id) =
        canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
    let metadata_str = edge
        .metadata
        .as_ref()
        .map(|v| serde_json::to_string(v).unwrap_or_default());
    SqlStatement {
        sql: "UPDATE graph_edges SET \
                namespace = ?1, source_id = ?2, target_id = ?3, relation = ?4, weight = ?5, \
                updated_at = ?6, deleted_at = ?7, metadata = ?8, target_backend = ?9 \
              WHERE id = ?10 AND updated_at = ?11 AND deleted_at IS ?12 \
                AND ?6 > updated_at"
            .to_string(),
        params: vec![
            SqlValue::Text(edge.namespace.clone()),
            SqlValue::Text(source_id.to_string()),
            SqlValue::Text(target_id.to_string()),
            SqlValue::Text(edge.relation.to_string()),
            SqlValue::Float(edge.weight),
            SqlValue::Integer(edge.updated_at.timestamp_micros()),
            match edge.deleted_at {
                Some(t) => SqlValue::Integer(t.timestamp_micros()),
                None => SqlValue::Null,
            },
            match metadata_str {
                Some(m) => SqlValue::Text(m),
                None => SqlValue::Null,
            },
            match &edge.target_backend {
                Some(b) => SqlValue::Text(b.clone()),
                None => SqlValue::Null,
            },
            SqlValue::Text(Uuid::from(edge.id).to_string()),
            SqlValue::Integer(expected_updated_at.timestamp_micros()),
            match expected_deleted_at {
                Some(value) => SqlValue::Integer(value.timestamp_micros()),
                None => SqlValue::Null,
            },
        ],
        label: Some("edge-replace-if-unchanged".to_string()),
    }
}

/// The atomic `link` op's variant of [`edge_upsert_statement`] (ADR-099
/// §B3). Shares the SAME
/// `EDGE_NATURAL_KEY_CONFLICT_SET` conflict-arm text — the two builders
/// cannot diverge on write behavior — but wraps the `INSERT` in a guarded
/// `SELECT ... WHERE EXISTS(...)` that re-probes both endpoints for
/// existence INSIDE the transaction, at commit time, rather than trusting
/// prepare-time validation alone.
///
/// This guard is atomic-`link`-specific, not a `edge_upsert_statement`
/// concern: `LinkPlan`'s own doc comment (`khive-runtime::atomic_plan`)
/// records why it must be commit-time, not prepare-time — a `link` op's
/// async prepare pass (`validate_edge_relation_endpoints`) can run and pass
/// BEFORE an earlier op in the SAME atomic unit (e.g. `delete(X, hard)`)
/// removes that very endpoint; only a commit-time, in-transaction guard
/// closes that intra-batch ordering hazard (ADR-099 acceptance criteria:
/// `[delete(X, hard), link(A, X)]` must fail, not silently create a
/// dangling edge). Canonical `link` has no equivalent need — it executes
/// and commits standalone, with no other op's write interleaved between its
/// own validation and its own write.
#[allow(clippy::too_many_arguments)]
pub fn edge_insert_guarded_by_endpoints_statement(
    namespace: &str,
    edge_id: Uuid,
    source_id: Uuid,
    target_id: Uuid,
    relation: EdgeRelation,
    weight: f64,
    now: i64,
    metadata: Option<&str>,
) -> SqlStatement {
    edge_insert_guarded_by_endpoints_with_resurrection_statement(
        namespace, source_id, target_id, edge_id, relation, weight, now, metadata, false,
    )
}

/// Policy-aware form of [`edge_insert_guarded_by_endpoints_statement`].
#[allow(clippy::too_many_arguments)]
pub fn edge_insert_guarded_by_endpoints_with_resurrection_statement(
    namespace: &str,
    source_id: Uuid,
    target_id: Uuid,
    edge_id: Uuid,
    relation: EdgeRelation,
    weight: f64,
    now: i64,
    metadata: Option<&str>,
    resurrect: bool,
) -> SqlStatement {
    let src_exists = endpoint_exists_clause("?3");
    let tgt_exists = endpoint_exists_clause("?4");
    let conflict_clause = edge_conflict_clause(resurrect);
    SqlStatement {
        sql: format!(
            "INSERT INTO graph_edges \
              (namespace, id, source_id, target_id, relation, weight, \
               created_at, updated_at, metadata) \
              SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?7, ?8 \
              WHERE ({src_exists}) AND ({tgt_exists}) \
              ON CONFLICT(namespace, source_id, target_id, relation) DO UPDATE SET \
                  {conflict_clause}"
        ),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(edge_id.to_string()),
            SqlValue::Text(source_id.to_string()),
            SqlValue::Text(target_id.to_string()),
            SqlValue::Text(relation.as_str().to_string()),
            SqlValue::Float(weight),
            SqlValue::Integer(now),
            match metadata {
                Some(m) => SqlValue::Text(m.to_string()),
                None => SqlValue::Null,
            },
        ],
        label: Some("atomic-link-insert-edge-where-exists".to_string()),
    }
}

/// Atomic `link` create arm. Unlike the compatibility upsert builder above,
/// this statement never changes a row that appeared after prepare: a natural-
/// key conflict affects zero rows and therefore fails the caller's guard.
#[allow(clippy::too_many_arguments)]
pub fn edge_insert_new_guarded_by_endpoints_statement(
    namespace: &str,
    edge_id: Uuid,
    source_id: Uuid,
    target_id: Uuid,
    relation: EdgeRelation,
    weight: f64,
    now: i64,
    metadata: Option<&str>,
) -> SqlStatement {
    let src_exists = endpoint_exists_clause("?3");
    let tgt_exists = endpoint_exists_clause("?4");
    SqlStatement {
        sql: format!(
            "INSERT INTO graph_edges \
              (namespace, id, source_id, target_id, relation, weight, \
               created_at, updated_at, metadata) \
              SELECT ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?7, ?8 \
              WHERE ({src_exists}) AND ({tgt_exists}) \
              ON CONFLICT DO NOTHING"
        ),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(edge_id.to_string()),
            SqlValue::Text(source_id.to_string()),
            SqlValue::Text(target_id.to_string()),
            SqlValue::Text(relation.to_string()),
            SqlValue::Float(weight),
            SqlValue::Integer(now),
            metadata
                .map(|value| SqlValue::Text(value.to_string()))
                .unwrap_or(SqlValue::Null),
        ],
        label: Some("edge-link-create-if-absent-and-endpoints-exist".to_string()),
    }
}

/// Atomic `link` replace/resurrection arm. The natural-key row and both
/// concurrency markers must still match the prepare snapshot, and both
/// endpoints must still exist. Clearing `deleted_at` is deliberate here:
/// callers only build this statement for a live replacement or an explicitly
/// authorized resurrection.
pub fn edge_link_replace_if_unchanged_and_endpoints_exist_statement(
    previous: &Edge,
    weight: f64,
    now: i64,
    metadata: Option<&str>,
) -> SqlStatement {
    let src_exists = endpoint_exists_clause("?6");
    let tgt_exists = endpoint_exists_clause("?7");
    SqlStatement {
        sql: format!(
            "UPDATE graph_edges SET \
               weight = ?1, updated_at = ?2, deleted_at = NULL, \
               metadata = ?3, target_backend = NULL \
             WHERE namespace = ?4 AND id = ?5 \
               AND source_id = ?6 AND target_id = ?7 AND relation = ?8 \
               AND updated_at = ?9 AND deleted_at IS ?10 AND ?2 > updated_at \
               AND ({src_exists}) AND ({tgt_exists})"
        ),
        params: vec![
            SqlValue::Float(weight),
            SqlValue::Integer(now),
            metadata
                .map(|value| SqlValue::Text(value.to_string()))
                .unwrap_or(SqlValue::Null),
            SqlValue::Text(previous.namespace.clone()),
            SqlValue::Text(Uuid::from(previous.id).to_string()),
            SqlValue::Text(previous.source_id.to_string()),
            SqlValue::Text(previous.target_id.to_string()),
            SqlValue::Text(previous.relation.to_string()),
            SqlValue::Integer(previous.updated_at.timestamp_micros()),
            previous
                .deleted_at
                .map(|value| SqlValue::Integer(value.timestamp_micros()))
                .unwrap_or(SqlValue::Null),
        ],
        label: Some("edge-link-replace-if-unchanged-and-endpoints-exist".to_string()),
    }
}

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

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

/// The exact cascade `DELETE` this store's `purge_incident_edges` issues.
pub fn purge_incident_edges_statement(node_id: Uuid) -> SqlStatement {
    SqlStatement {
        sql: "DELETE FROM graph_edges WHERE source_id = ?1 OR target_id = ?1".to_string(),
        params: vec![SqlValue::Text(node_id.to_string())],
        label: Some("edge-purge-incident".to_string()),
    }
}

// ---------------------------------------------------------------------------
// Symmetric-relation update DML (ADR-099 B3 r6 second pass) — the SQL text
// `khive-runtime::operations::KhiveRuntime::update_edge_symmetric_dml` (the
// synchronous raw-connection commit-time path, run inside the writer-task/
// pool-mutex transaction) and ADR-099's atomic `prepare_update_edge` symmetric
// branch (the async plan-time path) both bind. `upsert_edge` cannot be used
// here: it resolves `ON CONFLICT(namespace, id)` first and cannot detect a
// natural-key collision at (namespace, source_id, target_id, relation) with a
// *different* id, which is exactly the case a symmetric-relation endpoint
// canonicalization can produce.
//
// The two call sites bind these against different parameter-passing
// mechanisms — `conn.execute`/`conn.query_row` with `rusqlite::params!` in the
// synchronous path (it must run inside an existing transaction on a borrowed
// `&rusqlite::Connection`, so it cannot go through the `SqlStatement`/
// `SqlValue` plan-shape khive-storage abstracts elsewhere) vs. `SqlValue`
// plan params for the async `PlanStatement` path — but the SQL TEXT itself
// (the `EDGE_SYMMETRIC_*_SQL` constants below) is the single source of truth
// for both, closing the class of drift that produced a hand-copied SQL
// literal silently diverging from canonical (ADR-099 §B3).
pub const EDGE_SYMMETRIC_CONFLICT_PROBE_SQL: &str = "SELECT id FROM graph_edges \
     WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 \
     AND relation = ?4 AND id != ?5";

pub const EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL: &str =
    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2";

/// Canonical `update_edge`'s guarded variant of
/// [`EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL`]: `?3`/`?4` pin the fetched
/// snapshot's `updated_at`/`deleted_at` so a writer whose edge changed
/// concurrently after it read that snapshot cannot delete the row out from
/// under the concurrent write, even though a canonical survivor genuinely
/// exists at the natural key. Zero affected rows means stale, not
/// "no conflict" — the caller has already confirmed a conflicting canonical
/// row exists before running this statement. Deliberately a DIFFERENT
/// constant from the unguarded one above: merge's predicate-based rewrites
/// (`khive-runtime::curation`) intentionally keep running the unguarded form
/// inside their own single writer transaction and must not be changed to
/// bind this one.
pub const EDGE_SYMMETRIC_DELETE_NONCANONICAL_GUARDED_SQL: &str =
    "DELETE FROM graph_edges WHERE namespace = ?1 AND id = ?2 \
     AND updated_at = ?3 AND deleted_at IS ?4";

/// Case (a) update, guarded on the fetched snapshot's revision and deletion
/// marker: `?9`/`?10` pin `updated_at`/`deleted_at` as read, and `?5 >
/// updated_at` requires the replacement revision to strictly advance —
/// mirroring `edge_replace_if_unchanged_statement`'s guard so the symmetric
/// path cannot silently overwrite a concurrent writer's change between the
/// snapshot read and this write.
pub const EDGE_SYMMETRIC_UPDATE_INPLACE_SQL: &str = "UPDATE graph_edges SET \
     source_id = ?1, target_id = ?2, relation = ?3, \
     weight = ?4, updated_at = ?5, metadata = ?6 \
     WHERE namespace = ?7 AND id = ?8 \
       AND updated_at = ?9 AND deleted_at IS ?10 \
       AND ?5 > updated_at";

/// Plan-shape builder for [`EDGE_SYMMETRIC_CONFLICT_PROBE_SQL`] — the
/// async prepare-time conflict probe.
pub fn edge_symmetric_conflict_probe_statement(
    namespace: &str,
    canon_src: Uuid,
    canon_tgt: Uuid,
    relation: EdgeRelation,
    exclude_id: Uuid,
) -> SqlStatement {
    SqlStatement {
        sql: EDGE_SYMMETRIC_CONFLICT_PROBE_SQL.to_string(),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(canon_src.to_string()),
            SqlValue::Text(canon_tgt.to_string()),
            SqlValue::Text(relation.to_string()),
            SqlValue::Text(exclude_id.to_string()),
        ],
        label: Some("edge-symmetric-conflict-probe".to_string()),
    }
}

/// Plan-shape builder for [`EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL`] —
/// case (b): a canonical row already exists, delete the requested row.
pub fn edge_symmetric_delete_noncanonical_statement(namespace: &str, id: Uuid) -> SqlStatement {
    SqlStatement {
        sql: EDGE_SYMMETRIC_DELETE_NONCANONICAL_SQL.to_string(),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(id.to_string()),
        ],
        label: Some("edge-symmetric-delete-noncanonical".to_string()),
    }
}

/// Plan-shape builder for [`EDGE_SYMMETRIC_UPDATE_INPLACE_SQL`] —
/// case (a): no conflict, update the requested row in place, guarded on the
/// fetched snapshot's revision and deletion marker.
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_update_inplace_statement(
    namespace: &str,
    id: Uuid,
    canon_src: Uuid,
    canon_tgt: Uuid,
    relation: EdgeRelation,
    weight: f64,
    updated_at_micros: i64,
    metadata: Option<&str>,
    expected_updated_at_micros: i64,
    expected_deleted_at_micros: Option<i64>,
) -> SqlStatement {
    SqlStatement {
        sql: EDGE_SYMMETRIC_UPDATE_INPLACE_SQL.to_string(),
        params: vec![
            SqlValue::Text(canon_src.to_string()),
            SqlValue::Text(canon_tgt.to_string()),
            SqlValue::Text(relation.to_string()),
            SqlValue::Float(weight),
            SqlValue::Integer(updated_at_micros),
            match metadata {
                Some(m) => SqlValue::Text(m.to_string()),
                None => SqlValue::Null,
            },
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(id.to_string()),
            SqlValue::Integer(expected_updated_at_micros),
            match expected_deleted_at_micros {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
        ],
        label: Some("edge-symmetric-update-inplace".to_string()),
    }
}

// ---------------------------------------------------------------------------
// Symmetric-relation update DML — atomic-only, commit-time self-guarding
// variant (ADR-099 §B3).
//
// Canonical `update_edge_symmetric_dml` binds the shared SQL constants above
// directly, not the plan-shape builders — the builders are used by the atomic
// prepare path. Canonical probes and branches synchronously INSIDE its own
// writer-task transaction, with no other op interleaved between its probe and
// its write, so the interleaving exposure the atomic path has does not arise
// there. Canonical's absorption delete is nonetheless bound to the GUARDED
// constant, because its snapshot is read before the transaction and can be
// stale by the time the delete runs.
//
// The atomic path is structurally different: its conflict probe runs in the
// async PREPARE phase, which for a multi-op `--atomic` unit completes for
// EVERY op before the synchronous COMMIT phase begins for ANY of them. An
// earlier op in the SAME atomic unit (e.g. a `delete` or another symmetric
// `update` touching the same natural key) can change the conflict landscape
// between this op's prepare-time probe and its own statements finally
// executing at commit time — this is a real staleness
// window, not just an SQL-text duplication concern.
//
// The two builders below close it: instead of a Rust-level `if let
// Some(conflict) { ... } else { ... }` that hand-picks ONE of three
// statements at prepare time (the second hand-assembled branch this
// closes), the atomic plan ALWAYS carries both statements, in order, and
// each is a self-guarding, commit-time predicate that re-evaluates conflict
// state fresh against whatever the transaction's state actually is when it
// runs — not what prepare's probe said:
//
// 1. [`edge_symmetric_delete_if_conflict_statement`]: deletes the requested
//    (non-canonical) row IF AND ONLY IF a differently-id'd canonical row
//    exists at the target natural key at THIS moment AND the row's own
//    `updated_at`/`deleted_at` still match the snapshot this plan was built
//    from (guard: 0 or 1 rows) — a plan built from a since-changed snapshot
//    must not delete the row just because some other, unrelated conflict
//    happens to exist; the second statement's own commit-time predicate
//    (below) then sees `changes() = 0` and fails its `exactly(1)` guard,
//    aborting the whole atomic unit rather than silently absorbing a
//    concurrent writer's change.
// 2. [`edge_symmetric_absorb_or_update_inplace_statement`]: a single
//    `UPDATE` that no longer trusts an `id = ?2 OR natural-key` predicate
//    (ADR-099 §B3 — that predicate could
//    match the WRONG row: if a different op earlier in the SAME atomic unit
//    had already deleted the requested edge, statement 1 above no-ops (its
//    "0 rows" result is indistinguishable at the Rust level from "no
//    conflict existed"), yet the natural-key arm could still hit a
//    pre-existing canonical row that this update never causally touched).
//    The fix ties the natural-key arm to `changes()` — SQLite's per-
//    connection scalar reporting the row count of the most recently
//    COMPLETED statement, i.e. statement 1's own result, evaluated fresh at
//    THIS statement's execution, not at prepare time:
//    - `id = ?2 AND changes() = 0`: statement 1 deleted nothing, so the
//      requested row is still live under its own id — update it in place.
//    - `source_id = ?3 AND target_id = ?4 AND relation = ?5 AND id != ?2
//      AND changes() = 1`: statement 1 just deleted the requested row
//      BECAUSE a conflict existed — match the surviving canonical row and
//      leave its attributes unchanged (ADR-039 DO NOTHING; see below).
//    These two arms are mutually exclusive and, together with statement 1's
//    own guard, jointly exhaustive: if the requested row no longer existed
//    when statement 1 ran (the same-unit race above), statement 1 affects 0
//    rows for a reason unrelated to conflict absorption, `id = ?2` no
//    longer matches anything (the row is gone), and the natural-key arm's
//    `changes() = 1` guard is false — so this statement affects ZERO rows
//    and the plan's `AffectedRowGuard::exactly(1)` on it fails the op,
//    aborting the whole atomic unit rather than silently mutating an
//    unrelated row.
//
//    ADR-039's edge-conflict contract is ON CONFLICT DO NOTHING: the
//    natural-key (absorbed-conflict) arm must leave the surviving canonical
//    row's attributes exactly as they were — refreshing them from the
//    discarded edge (and forcing `deleted_at = NULL`, resurrecting a
//    tombstone) is the same defect already fixed on the merge-rewire path
//    (khive#1213). Every SET expression is therefore keyed on `id = ?2`:
//    that condition is true only for the WHERE clause's first (in-place)
//    arm — the second (absorbed) arm only ever matches a row whose id is
//    NOT ?2 — so each column either takes its new value (in-place arm) or
//    self-assigns its current value (absorbed arm, a true no-op). This
//    still affects exactly one row either way (SQLite's `changes()` counts
//    matched rows, not changed bytes), so `AffectedRowGuard::exactly(1)`
//    and the race-abort behavior above are unaffected by this being a
//    no-op write in the absorbed case.
//
// No probe, no branch, no read at all is needed to APPLY this pair. Which
// row this plan actually touched is derived post-commit by the caller via a
// fresh natural-key lookup (`khive-runtime::KhiveRuntime::get_edge_by_natural_key_including_deleted`,
// filtered on the canonicalized endpoints/relation and including soft-deleted rows — unlike
// `list_edges`, which unconditionally filters `deleted_at IS NULL` and would report "not
// found" for a surviving row this absorption arm left tombstoned) — ADR-099 §B3
// removed the prior prepare-time advisory `target_id` probe entirely:
// a value computed before the
// SAME atomic unit's other ops have run is not a fact this plan can stand
// behind, so result rendering no longer trusts it.
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_delete_if_conflict_statement(
    namespace: &str,
    id: Uuid,
    canon_src: Uuid,
    canon_tgt: Uuid,
    relation: EdgeRelation,
    expected_updated_at_micros: i64,
    expected_deleted_at_micros: Option<i64>,
) -> SqlStatement {
    SqlStatement {
        sql: "DELETE FROM graph_edges \
              WHERE namespace = ?1 AND id = ?2 \
                AND updated_at = ?6 AND deleted_at IS ?7 \
                AND EXISTS ( \
                  SELECT 1 FROM graph_edges \
                  WHERE namespace = ?1 AND source_id = ?3 AND target_id = ?4 \
                    AND relation = ?5 AND id != ?2 \
                )"
        .to_string(),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(id.to_string()),
            SqlValue::Text(canon_src.to_string()),
            SqlValue::Text(canon_tgt.to_string()),
            SqlValue::Text(relation.to_string()),
            SqlValue::Integer(expected_updated_at_micros),
            match expected_deleted_at_micros {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
        ],
        label: Some("edge-symmetric-delete-if-conflict".to_string()),
    }
}

/// `?10`/`?11` pin the fetched snapshot's `updated_at`/`deleted_at` and
/// `?7 > updated_at` requires the replacement revision to strictly advance —
/// guarding the in-place arm (`id = ?2`) against a concurrent writer that
/// changed the row between the snapshot read and this statement. The
/// absorbed arm (a differently-id'd canonical row) is intentionally left
/// unguarded on revision: per ADR-039 DO NOTHING it self-assigns every
/// column, so it is a no-op write regardless of the survivor's current
/// state.
#[allow(clippy::too_many_arguments)]
pub fn edge_symmetric_absorb_or_update_inplace_statement(
    namespace: &str,
    id: Uuid,
    canon_src: Uuid,
    canon_tgt: Uuid,
    relation: EdgeRelation,
    weight: f64,
    updated_at_micros: i64,
    metadata: Option<&str>,
    target_backend: Option<&str>,
    expected_updated_at_micros: i64,
    expected_deleted_at_micros: Option<i64>,
) -> SqlStatement {
    SqlStatement {
        sql: "UPDATE graph_edges SET \
              source_id = CASE WHEN id = ?2 THEN ?3 ELSE source_id END, \
              target_id = CASE WHEN id = ?2 THEN ?4 ELSE target_id END, \
              relation = CASE WHEN id = ?2 THEN ?5 ELSE relation END, \
              weight = CASE WHEN id = ?2 THEN ?6 ELSE weight END, \
              updated_at = CASE WHEN id = ?2 THEN ?7 ELSE updated_at END, \
              deleted_at = CASE WHEN id = ?2 THEN NULL ELSE deleted_at END, \
              metadata = CASE WHEN id = ?2 THEN ?8 ELSE metadata END, \
              target_backend = CASE WHEN id = ?2 THEN ?9 ELSE target_backend END \
              WHERE namespace = ?1 \
                AND ( \
                  (id = ?2 AND changes() = 0 AND updated_at = ?10 AND deleted_at IS ?11 \
                      AND ?7 > updated_at) \
                  OR (source_id = ?3 AND target_id = ?4 AND relation = ?5 \
                      AND id != ?2 AND changes() = 1) \
                )"
        .to_string(),
        params: vec![
            SqlValue::Text(namespace.to_string()),
            SqlValue::Text(id.to_string()),
            SqlValue::Text(canon_src.to_string()),
            SqlValue::Text(canon_tgt.to_string()),
            SqlValue::Text(relation.to_string()),
            SqlValue::Float(weight),
            SqlValue::Integer(updated_at_micros),
            match metadata {
                Some(m) => SqlValue::Text(m.to_string()),
                None => SqlValue::Null,
            },
            match target_backend {
                Some(b) => SqlValue::Text(b.to_string()),
                None => SqlValue::Null,
            },
            SqlValue::Integer(expected_updated_at_micros),
            match expected_deleted_at_micros {
                Some(value) => SqlValue::Integer(value),
                None => SqlValue::Null,
            },
        ],
        label: Some("edge-symmetric-absorb-or-update-inplace".to_string()),
    }
}

/// A GraphStore backed by SQLite tables.
pub struct SqlGraphStore {
    pool: Arc<ConnectionPool>,
    is_file_backed: bool,
    /// Default namespace for multi-record queries (ADR-007 PARAM-ONLY: used as a
    /// WHERE filter on `query_edges`/`neighbors`/`traverse`, never as an
    /// enforcement gate on by-ID operations).
    namespace: String,
    writer_task: Option<WriterTaskHandle>,
}

impl SqlGraphStore {
    /// Create a new store with a default namespace for multi-record query filtering.
    ///
    /// The namespace is a PARAM-ONLY hint (ADR-007 rule 4) — it is used as a
    /// WHERE filter in multi-record queries and as the write namespace stamped on
    /// upserted edges, but it does NOT enforce isolation: `upsert_edge` accepts
    /// edges from any namespace, and by-ID ops (`get_edge`, `delete_edge`) ignore
    /// the namespace entirely.
    pub fn new_scoped(
        pool: Arc<ConnectionPool>,
        is_file_backed: bool,
        namespace: impl Into<String>,
    ) -> Self {
        // Enabled by default for file-backed pools; explicit off/degraded
        // construction remains synchronous (ADR-067 Component A, mirrors
        // entity.rs policy): a missing writer task 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,
            is_file_backed,
            namespace: namespace.into(),
            writer_task,
        }
    }

    fn open_standalone_writer(&self) -> Result<rusqlite::Connection, StorageError> {
        self.pool
            .open_standalone_writer()
            .map_err(|e| map_sqlite_err(e, "open_graph_writer"))
    }

    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
    /// standalone-connection / pool-mutex path (ADR-067 Component A, Fork C
    /// slice 2). `f` must be DML-only. See
    /// `crates/khive-db/docs/api/graph.md` for the per-caller routing rules.
    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(move |conn| f(conn).map_err(|e| map_err(e, op)))
                .await;
        }

        self.pool
            .record_direct_route(crate::timeout_sink::Site::DirectRouteGraphGeneralWrite);
        if self.is_file_backed {
            let conn = self.open_standalone_writer()?;
            let db = crate::timeout_sink::db_label(&self.pool);
            tokio::task::spawn_blocking(move || {
                f(&conn).map_err(|e| {
                    crate::timeout_sink::maybe_emit_busy(
                        &db,
                        crate::timeout_sink::Site::StandaloneGraph,
                        &e,
                    );
                    map_err(e, op)
                })
            })
            .await
            .map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
        } else {
            let pool = Arc::clone(&self.pool);
            tokio::task::spawn_blocking(move || {
                let guard = pool.try_writer().map_err(|e| map_sqlite_err(e, op))?;
                f(guard.conn()).map_err(|e| map_err(e, op))
            })
            .await
            .map_err(|e| StorageError::driver(StorageCapability::Graph, op, e))?
        }
    }

    async fn observed_edge_write(
        &self,
        operation: &'static str,
        request: EdgeUpsertRequest,
        guard_endpoints: bool,
    ) -> Result<GuardedEdgeUpsertOutcome, StorageError> {
        if let Some(writer_task) = self.current_writer_task(operation)? {
            return writer_task
                .send_bounded(move |conn| {
                    observed_edge_upsert(conn, &request, guard_endpoints)
                        .map_err(|error| map_err(error, operation))
                })
                .await;
        }

        let origin = self.pool.origin();
        self.with_writer(operation, move |conn| {
            conn.execute_batch("BEGIN IMMEDIATE")?;
            let _tx_handle =
                khive_storage::tx_registry::register_scoped(Some(operation.to_string()), origin);
            let outcome = match observed_edge_upsert(conn, &request, guard_endpoints) {
                Ok(outcome) => outcome,
                Err(error) => {
                    let _ = conn.execute_batch("ROLLBACK");
                    return Err(error);
                }
            };
            if let Err(error) = conn.execute_batch("COMMIT") {
                let _ = conn.execute_batch("ROLLBACK");
                return Err(error);
            }
            Ok(outcome)
        })
        .await
    }

    async fn observed_edge_batch_write(
        &self,
        operation: &'static str,
        requests: Vec<EdgeUpsertRequest>,
        guard_endpoints: bool,
    ) -> Result<GuardedEdgeBatchUpsertOutcome, StorageError> {
        if let Some(writer_task) = self.current_writer_task(operation)? {
            return writer_task
                .send_bounded(move |conn| {
                    observed_edge_batch_upsert(conn, &requests, guard_endpoints)
                        .map_err(|error| map_err(error, operation))
                })
                .await;
        }

        let origin = self.pool.origin();
        self.with_writer(operation, move |conn| {
            conn.execute_batch("BEGIN IMMEDIATE")?;
            let _tx_handle =
                khive_storage::tx_registry::register_scoped(Some(operation.to_string()), origin);
            let outcome = match observed_edge_batch_upsert(conn, &requests, guard_endpoints) {
                Ok(outcome) => outcome,
                Err(error) => {
                    let _ = conn.execute_batch("ROLLBACK");
                    return Err(error);
                }
            };
            if let Err(error) = conn.execute_batch("COMMIT") {
                let _ = conn.execute_batch("ROLLBACK");
                return Err(error);
            }
            Ok(outcome)
        })
        .await
    }

    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::Graph,
            op,
            move |conn| f(conn).map_err(|error| map_err(error, op)),
        )
        .await
    }
}

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

/// Publish a graph read's accounting after the read resolves, on either
/// outcome.
///
/// Both counters are collected inside the storage closure because that closure
/// runs on a blocking thread where the task-local usage context is invisible,
/// and because a value returned only on success reports nothing at all when
/// the read fails partway. `queries` is incremented immediately before a
/// statement executes, so a read that never got that far — a connection the
/// pool could not hand out, a blocking task that died — counts no round trip.
/// `rows` is incremented as each adjacency entry materialises, so entries
/// storage already returned still count when a later row fails to convert.
fn report_graph_usage(queries: &std::sync::atomic::AtomicU64, rows: &std::sync::atomic::AtomicU64) {
    khive_storage::usage::count(
        khive_storage::usage::UsageUnit::DbRoundTrips,
        queries.load(std::sync::atomic::Ordering::Relaxed),
    );
    khive_storage::usage::count(
        khive_storage::usage::UsageUnit::GraphHops,
        rows.load(std::sync::atomic::Ordering::Relaxed),
    );
}

fn read_edge(row: &rusqlite::Row<'_>) -> Result<Edge, rusqlite::Error> {
    let namespace: String = row.get(0)?;
    let id_str: String = row.get(1)?;
    let source_str: String = row.get(2)?;
    let target_str: String = row.get(3)?;
    let relation_str: String = row.get(4)?;
    let weight: f64 = row.get(5)?;
    let created_micros: i64 = row.get(6)?;
    let updated_micros: i64 = row.get(7)?;
    let deleted_micros: Option<i64> = row.get(8)?;
    let metadata_str: Option<String> = row.get(9)?;
    let target_backend: Option<String> = row.get(10)?;

    let id = parse_uuid(&id_str)?;
    let source_id = parse_uuid(&source_str)?;
    let target_id = parse_uuid(&target_str)?;
    let created_at = micros_to_datetime(created_micros);
    let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
        rusqlite::Error::FromSqlConversionFailure(4, rusqlite::types::Type::Text, Box::new(e))
    })?;
    let metadata = match metadata_str {
        Some(s) => {
            let v = serde_json::from_str(&s).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    9,
                    rusqlite::types::Type::Text,
                    Box::new(e),
                )
            })?;
            Some(v)
        }
        None => None,
    };

    Ok(Edge {
        id: id.into(),
        namespace,
        source_id,
        target_id,
        relation,
        weight,
        created_at,
        updated_at: micros_to_datetime(updated_micros),
        deleted_at: deleted_micros.map(micros_to_datetime),
        metadata,
        target_backend,
    })
}

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))
    })
}

/// Build the `relation IN (...)` / `weight >= ?` `WHERE`-extra clause and the
/// `LIMIT` clause shared by `neighbors` and `neighbors_both_directions` —
/// both filter and cap identically, differing only in which direction(s) the
/// base `SELECT`s cover. `start_param_idx` is the next free `?N` placeholder
/// (both callers bind `namespace` and `node_id` as `?1`/`?2` first).
fn neighbor_extra_clause(
    query: &NeighborQuery,
    start_param_idx: usize,
    after: Option<&NeighborCursor>,
    neighbor_kinds: Option<&[String]>,
) -> (String, String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut conditions: Vec<String> = Vec::new();
    let mut extra_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
    let mut param_idx = start_param_idx;

    if let Some(ref rels) = query.relations {
        if !rels.is_empty() {
            let placeholders: Vec<String> = rels
                .iter()
                .map(|r| {
                    extra_params.push(Box::new(r.to_string()));
                    let p = format!("?{}", param_idx);
                    param_idx += 1;
                    p
                })
                .collect();
            conditions.push(format!("relation IN ({})", placeholders.join(",")));
        }
    }

    if let Some(min_w) = query.min_weight {
        extra_params.push(Box::new(min_w));
        conditions.push(format!("weight >= ?{}", param_idx));
        param_idx += 1;
    }

    if let Some(cursor) = after {
        extra_params.push(Box::new(cursor.weight));
        let weight_idx = param_idx;
        param_idx += 1;
        extra_params.push(Box::new(cursor.node_id.to_string()));
        let node_idx = param_idx;
        param_idx += 1;
        extra_params.push(Box::new(cursor.edge_id.to_string()));
        let edge_idx = param_idx;
        param_idx += 1;
        conditions.push(format!(
            "(weight < ?{weight_idx} OR (weight = ?{weight_idx} AND node_id > ?{node_idx}) OR (weight = ?{weight_idx} AND node_id = ?{node_idx} AND edge_id > ?{edge_idx}))"
        ));
    }

    if let Some(kinds) = neighbor_kinds.filter(|kinds| !kinds.is_empty()) {
        let placeholders: Vec<String> = kinds
            .iter()
            .map(|kind| {
                extra_params.push(Box::new(kind.clone()));
                let p = format!("?{param_idx}");
                param_idx += 1;
                p
            })
            .collect();
        let entity_placeholders = placeholders.join(",");
        let note_placeholders: Vec<String> = kinds
            .iter()
            .map(|kind| {
                extra_params.push(Box::new(kind.clone()));
                let p = format!("?{param_idx}");
                param_idx += 1;
                p
            })
            .collect();
        conditions.push(format!(
            "(EXISTS (SELECT 1 FROM entities AS neighbor_entities WHERE neighbor_entities.id = node_id AND neighbor_entities.namespace = ?1 AND neighbor_entities.deleted_at IS NULL AND neighbor_entities.kind IN ({entity_placeholders})) OR EXISTS (SELECT 1 FROM notes AS neighbor_notes WHERE neighbor_notes.id = node_id AND neighbor_notes.namespace = ?1 AND neighbor_notes.deleted_at IS NULL AND neighbor_notes.kind IN ({})))",
            note_placeholders.join(",")
        ));
    }

    let where_extra = if conditions.is_empty() {
        String::new()
    } else {
        format!(" WHERE {}", conditions.join(" AND "))
    };

    let limit_clause = if let Some(lim) = query.limit {
        extra_params.push(Box::new(lim as i64));
        format!(" LIMIT ?{}", param_idx)
    } else {
        String::new()
    };

    (where_extra, limit_clause, extra_params)
}

// Test-only counter of storage-level neighbor SELECT executions (`neighbors`
// and `neighbors_both_directions` each issue exactly one `graph_edges`
// query per call). Lets tests assert the query-count halving a
// `Direction::Both` caller gets from `neighbors_both_directions` vs the old
// pattern of two separate `neighbors` calls (ADR-089 context-verb
// optimization). Gated out of release builds — no counter overhead on the
// hot path in production.
#[cfg(test)]
static NEIGHBOR_SELECT_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

#[cfg(test)]
fn count_neighbor_select() {
    NEIGHBOR_SELECT_COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}

#[cfg(not(test))]
fn count_neighbor_select() {}

#[cfg(test)]
pub(crate) fn reset_neighbor_select_count() {
    NEIGHBOR_SELECT_COUNT.store(0, std::sync::atomic::Ordering::Relaxed);
}

#[cfg(test)]
pub(crate) fn neighbor_select_count() -> usize {
    NEIGHBOR_SELECT_COUNT.load(std::sync::atomic::Ordering::Relaxed)
}

fn micros_to_datetime(micros: i64) -> DateTime<Utc> {
    Utc.timestamp_micros(micros)
        .single()
        .unwrap_or_else(Utc::now)
}

/// Deterministic ORDER BY for edge queries. #1671: an `id` tiebreak is
/// always appended (following the last sort field's direction, DESC for the
/// empty default) so pages over equal sort values keep a total order. That
/// removes tie-order instability only — offset paging can still duplicate or
/// skip rows under concurrent inserts/deletes or sort-key updates (that
/// would need snapshot isolation or keyset pagination).
fn edge_order_clause(sort: &[SortOrder<EdgeSortField>]) -> String {
    if sort.is_empty() {
        return " ORDER BY created_at DESC, id DESC".to_string();
    }
    let mut parts: Vec<String> = sort
        .iter()
        .map(|s| {
            let dir = match s.direction {
                SortDirection::Asc => "ASC",
                SortDirection::Desc => "DESC",
            };
            format!("{} {}", edge_sort_col(&s.field), dir)
        })
        .collect();
    let dir = match sort.last().map(|s| &s.direction) {
        Some(SortDirection::Asc) => "ASC",
        _ => "DESC",
    };
    parts.push(format!("id {dir}"));
    format!(" ORDER BY {}", parts.join(", "))
}

/// `CASE` expression classifying one edge endpoint by the base it resolves
/// against. Soft-deleted endpoints are already excluded by
/// [`LIVE_ENDPOINTS_CONDITION`] in the surrounding `WHERE`, so a row reaching
/// this expression has live endpoints or none at all.
///
/// The probe reads the same local tables as [`LIVE_ENDPOINTS_CONDITION`] and
/// [`endpoint_exists_clause`], which is what makes the three agree on what an
/// endpoint is. Two consequences are deliberate. An `annotates` endpoint that
/// is an event or another edge is a live endpoint that is not in either base,
/// so it classifies as `none` and lands in `unresolved`; that is the bucket's
/// meaning, not a miscount, and `annotates` is excluded from structural
/// density anyway. And when ADR-009 edge routing is wired, an endpoint held in
/// another backend will not be in these tables either: `target_backend` is
/// written by nothing today, so no such row exists yet, but whoever wires the
/// routing has to decide what the breakdown should say about it and change
/// this expression deliberately rather than discover it as a silent
/// `unresolved`.
fn endpoint_base_case(column: &str) -> String {
    format!(
        "CASE WHEN EXISTS (SELECT 1 FROM entities be WHERE be.id = graph_edges.{column}) \
         THEN 'entity' \
         WHEN EXISTS (SELECT 1 FROM notes bn WHERE bn.id = graph_edges.{column}) \
         THEN 'note' ELSE 'none' END"
    )
}

/// Folds one `(source_base, target_base, count)` group into the tally. An
/// unrecognized pair lands in `unresolved` rather than being dropped, so the
/// buckets keep summing to the live edge total.
fn fold_endpoint_base_row(counts: &mut EdgeEndpointBaseCounts, source: &str, target: &str, n: u64) {
    let slot = match (source, target) {
        ("entity", "entity") => &mut counts.entity_entity,
        ("entity", "note") => &mut counts.entity_note,
        ("note", "entity") => &mut counts.note_entity,
        ("note", "note") => &mut counts.note_note,
        _ => &mut counts.unresolved,
    };
    *slot = slot.saturating_add(n);
}

/// Restricts an edge count to edges whose endpoints are still live.
///
/// A soft delete leaves incident edges in place on purpose (`KhiveRuntime::delete_entity`
/// documents it; only a hard delete purges them). Every reader that walks the graph
/// hydrates the endpoint records, so it cannot reach an edge whose endpoint is
/// tombstoned. A count that includes those edges therefore reports a density no reader
/// can walk, under a `count_scope` that says `live_only`.
///
/// The predicate excludes an endpoint only when it is PRESENT AND tombstoned. An id
/// found in neither table belongs to a substrate this database does not hold, and
/// treating its absence as a tombstone would silently under-count; erring toward
/// counting keeps the failure in the direction the old behaviour already had.
const LIVE_ENDPOINTS_CONDITION: &str = "NOT EXISTS (SELECT 1 FROM entities le \
     WHERE le.id = graph_edges.source_id AND le.deleted_at IS NOT NULL) \
     AND NOT EXISTS (SELECT 1 FROM entities le \
     WHERE le.id = graph_edges.target_id AND le.deleted_at IS NOT NULL) \
     AND NOT EXISTS (SELECT 1 FROM notes ln \
     WHERE ln.id = graph_edges.source_id AND ln.deleted_at IS NOT NULL) \
     AND NOT EXISTS (SELECT 1 FROM notes ln \
     WHERE ln.id = graph_edges.target_id AND ln.deleted_at IS NOT NULL)";

/// Append [`LIVE_ENDPOINTS_CONDITION`] to a `WHERE` clause produced by the edge filter
/// builders, which always emit a non-empty ` WHERE ...`.
fn with_live_endpoints(where_clause: &str) -> String {
    format!("{where_clause} AND {LIVE_ENDPOINTS_CONDITION}")
}

fn build_edge_filter_sql(
    namespace: &str,
    filter: &EdgeFilter,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    build_edge_filter_sql_for_namespaces(&[namespace.to_string()], filter)
}

fn build_edge_filter_sql_for_namespaces(
    namespaces: &[String],
    filter: &EdgeFilter,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let 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(", "))
        }
    };
    build_edge_filter_conditions(namespace_condition, params, filter)
}

/// Same filter conditions as [`build_edge_filter_sql_for_namespaces`], but
/// the namespace set is bound as a single pre-serialized JSON array
/// parameter (`?1`) matched via `json_each` rather than one `?N` per
/// namespace.
///
/// `namespace IN (?1, ?2, ..., ?N)` costs one bound SQLite variable per
/// namespace; a caller with a large visible-namespace set (hundreds to
/// thousands, e.g. a broad `[actor]` visibility grant) can exceed
/// `SQLITE_LIMIT_VARIABLE_NUMBER` (999 by default) before any of the
/// filter's own parameters are even added. Binding one JSON string instead
/// keeps this at O(1) parameters regardless of namespace count, so paged
/// enumeration (which needs one exact-order SQL statement per page — see
/// [`SqlGraphStore::query_edges_in_namespaces`]) never has to chunk the
/// namespace set and cannot silently corrupt paging by doing so (#2088).
fn build_edge_filter_sql_for_namespaces_json(
    namespaces_json: &str,
    filter: &EdgeFilter,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![Box::new(namespaces_json.to_string())];
    let namespace_condition = "namespace IN (SELECT value FROM json_each(?1))".to_string();
    build_edge_filter_conditions(namespace_condition, params, filter)
}

fn build_edge_filter_conditions(
    namespace_condition: String,
    mut params: Vec<Box<dyn rusqlite::types::ToSql>>,
    filter: &EdgeFilter,
) -> (String, Vec<Box<dyn rusqlite::types::ToSql>>) {
    let mut conditions = vec![namespace_condition, "deleted_at IS NULL".to_string()];

    if !filter.ids.is_empty() {
        let placeholders: Vec<String> = filter
            .ids
            .iter()
            .map(|id| {
                params.push(Box::new(id.to_string()));
                format!("?{}", params.len())
            })
            .collect();
        conditions.push(format!("id IN ({})", placeholders.join(",")));
    }

    if !filter.source_ids.is_empty() {
        let placeholders: Vec<String> = filter
            .source_ids
            .iter()
            .map(|id| {
                params.push(Box::new(id.to_string()));
                format!("?{}", params.len())
            })
            .collect();
        conditions.push(format!("source_id IN ({})", placeholders.join(",")));
    }

    if !filter.target_ids.is_empty() {
        let placeholders: Vec<String> = filter
            .target_ids
            .iter()
            .map(|id| {
                params.push(Box::new(id.to_string()));
                format!("?{}", params.len())
            })
            .collect();
        conditions.push(format!("target_id IN ({})", placeholders.join(",")));
    }

    if !filter.relations.is_empty() {
        let placeholders: Vec<String> = filter
            .relations
            .iter()
            .map(|r| {
                params.push(Box::new(r.to_string()));
                format!("?{}", params.len())
            })
            .collect();
        conditions.push(format!("relation IN ({})", placeholders.join(",")));
    }

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

    if let Some(max_w) = filter.max_weight {
        params.push(Box::new(max_w));
        conditions.push(format!("weight <= ?{}", params.len()));
    }

    if let Some(ref time_range) = filter.created_at {
        if let Some(start) = time_range.start {
            params.push(Box::new(start.timestamp_micros()));
            conditions.push(format!("created_at >= ?{}", params.len()));
        }
        if let Some(end) = time_range.end {
            params.push(Box::new(end.timestamp_micros()));
            conditions.push(format!("created_at < ?{}", params.len()));
        }
    }

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

fn edge_sort_col(field: &EdgeSortField) -> &'static str {
    match field {
        EdgeSortField::CreatedAt => "created_at",
        EdgeSortField::Weight => "weight",
        EdgeSortField::Relation => "relation",
    }
}

// =============================================================================
// GraphStore implementation
// =============================================================================

/// Canonical endpoint order for symmetric relations (F012).
///
/// For `competes_with` and `composed_with`, ensures `source_uuid < target_uuid`
/// so A→B and B→A collapse to a single canonical row in storage.
fn canonical_edge_endpoints(
    relation: EdgeRelation,
    source_id: Uuid,
    target_id: Uuid,
) -> (Uuid, Uuid) {
    if relation.is_symmetric() && target_id < source_id {
        (target_id, source_id)
    } else {
        (source_id, target_id)
    }
}

/// Standalone existence probe for both endpoints of a would-be edge (#769),
/// matching the `WHERE EXISTS(...)` shape
/// [`edge_insert_guarded_by_endpoints_statement`] embeds in its own guarded
/// `INSERT`. Returns per-endpoint existence (not a single AND'd bool) so
/// callers can report exactly which side was missing. See
/// `crates/khive-db/docs/api/graph.md` for its two call sites and why both
/// need in-transaction, not reconstructed-after-the-fact, results.
fn edge_endpoints_exist(
    conn: &rusqlite::Connection,
    source_id: Uuid,
    target_id: Uuid,
) -> Result<MissingEndpoints, rusqlite::Error> {
    let src_exists = endpoint_exists_clause("?1");
    let tgt_exists = endpoint_exists_clause("?2");
    let sql = format!("SELECT ({src_exists}), ({tgt_exists})");
    conn.query_row(
        &sql,
        rusqlite::params![source_id.to_string(), target_id.to_string()],
        |row| {
            let src_exists: bool = row.get(0)?;
            let tgt_exists: bool = row.get(1)?;
            Ok(MissingEndpoints {
                source: !src_exists,
                target: !tgt_exists,
            })
        },
    )
}

fn edge_by_natural_key_including_deleted(
    conn: &rusqlite::Connection,
    edge: &Edge,
) -> Result<Option<Edge>, rusqlite::Error> {
    let (source_id, target_id) =
        canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
    conn.query_row(
        "SELECT namespace, id, source_id, target_id, relation, weight, \
                created_at, updated_at, deleted_at, metadata, target_backend \
         FROM graph_edges \
         WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 AND relation = ?4",
        rusqlite::params![
            &edge.namespace,
            source_id.to_string(),
            target_id.to_string(),
            edge.relation.as_str(),
        ],
        read_edge,
    )
    .optional()
}

/// Apply one replacement-style upsert and derive its disposition/preimage on
/// the same write connection. Callers keep this inside one write transaction.
fn observed_edge_upsert(
    conn: &rusqlite::Connection,
    request: &EdgeUpsertRequest,
    guard_endpoints: bool,
) -> Result<GuardedEdgeUpsertOutcome, rusqlite::Error> {
    let (source_id, target_id) = canonical_edge_endpoints(
        request.edge.relation,
        request.edge.source_id,
        request.edge.target_id,
    );
    if guard_endpoints {
        // Test-only rendezvous proving the endpoint decision is made while
        // this write transaction still excludes racing writers.
        #[cfg(test)]
        tests::insert_probe_seam::hook((source_id, target_id));
        let missing = edge_endpoints_exist(conn, source_id, target_id)?;
        if missing.any() {
            return Ok(GuardedEdgeUpsertOutcome::Refused(
                EdgeUpsertRefusal::MissingEndpoints(missing),
            ));
        }
    }

    let previous = edge_by_natural_key_including_deleted(conn, &request.edge)?;
    if let Some(edge) = previous.as_ref() {
        if edge.deleted_at.is_some() && !request.resurrect {
            return Ok(GuardedEdgeUpsertOutcome::Refused(
                EdgeUpsertRefusal::ResurrectionRequired { edge: edge.clone() },
            ));
        }
    }

    let statement = edge_upsert_statement_with_resurrection(&request.edge, request.resurrect);
    let mut stmt = conn.prepare(&statement.sql)?;
    bind_params(&mut stmt, &statement.params)?;
    let affected = stmt.raw_execute()?;
    if affected == 0 {
        let edge = edge_by_natural_key_including_deleted(conn, &request.edge)?
            .ok_or(rusqlite::Error::QueryReturnedNoRows)?;
        return Ok(GuardedEdgeUpsertOutcome::Refused(
            EdgeUpsertRefusal::ResurrectionRequired { edge },
        ));
    }

    let edge = edge_by_natural_key_including_deleted(conn, &request.edge)?
        .ok_or(rusqlite::Error::QueryReturnedNoRows)?;
    let disposition = match previous.as_ref().and_then(|edge| edge.deleted_at) {
        None if previous.is_none() => EdgeUpsertDisposition::Created,
        None => EdgeUpsertDisposition::Updated,
        Some(_) => EdgeUpsertDisposition::Resurrected,
    };
    Ok(GuardedEdgeUpsertOutcome::Written(EdgeUpsertResult {
        edge,
        disposition,
        previous,
    }))
}

/// Shared DML-only batch engine for observed and legacy upserts on both writer
/// routes. The caller owns the transaction and rolls back SQL failures.
fn observed_edge_batch_upsert(
    conn: &rusqlite::Connection,
    requests: &[EdgeUpsertRequest],
    guard_endpoints: bool,
) -> Result<GuardedEdgeBatchUpsertOutcome, rusqlite::Error> {
    // Preflight every refusal before the first mutation so the successful
    // return path is genuinely all-or-nothing even under WriterTask, whose
    // transaction commits any `Ok(...)` closure result.
    for (entry_index, request) in requests.iter().enumerate() {
        let (source_id, target_id) = canonical_edge_endpoints(
            request.edge.relation,
            request.edge.source_id,
            request.edge.target_id,
        );
        if guard_endpoints {
            let missing = edge_endpoints_exist(conn, source_id, target_id)?;
            if missing.any() {
                return Ok(GuardedEdgeBatchUpsertOutcome {
                    rows: Vec::new(),
                    refusal: Some(GuardedEdgeBatchRefusal {
                        entry_index,
                        reason: EdgeUpsertRefusal::MissingEndpoints(missing),
                    }),
                });
            }
        }
        if let Some(edge) = edge_by_natural_key_including_deleted(conn, &request.edge)? {
            if edge.deleted_at.is_some() && !request.resurrect {
                return Ok(GuardedEdgeBatchUpsertOutcome {
                    rows: Vec::new(),
                    refusal: Some(GuardedEdgeBatchRefusal {
                        entry_index,
                        reason: EdgeUpsertRefusal::ResurrectionRequired { edge },
                    }),
                });
            }
        }
    }

    let mut rows = Vec::with_capacity(requests.len());
    for request in requests {
        match observed_edge_upsert(conn, request, false)? {
            GuardedEdgeUpsertOutcome::Written(row) => rows.push(row),
            GuardedEdgeUpsertOutcome::Refused(_) => {
                return Err(rusqlite::Error::ExecuteReturnedResults)
            }
        }
    }
    Ok(GuardedEdgeBatchUpsertOutcome {
        rows,
        refusal: None,
    })
}

/// One index-seek adjacency statement used by bounded level-synchronous BFS.
/// There is deliberately no `ORDER BY`: the public contract promises minimum
/// depth, not a same-depth tie order, and sorting a high-degree node before a
/// low `LIMIT` would reintroduce the unbounded SQL work fixed by #1444.
fn traversal_neighbor_sql(
    direction: Direction,
    relation_count: usize,
    has_min_weight: bool,
) -> String {
    let (node_column, endpoint_column, index) = match direction {
        Direction::Out => ("target_id", "source_id", "idx_graph_edges_ns_src_rel"),
        Direction::In => ("source_id", "target_id", "idx_graph_edges_ns_tgt_rel"),
        Direction::Both => unreachable!("Direction::Both is split into indexed Out/In seeks"),
    };
    let mut sql = format!(
        "SELECT {node_column}, id, weight \
         FROM graph_edges INDEXED BY {index} \
         WHERE namespace = ?1 AND {endpoint_column} = ?2 AND deleted_at IS NULL"
    );
    if relation_count > 0 {
        let placeholders = (0..relation_count)
            .map(|offset| format!("?{}", 4 + offset))
            .collect::<Vec<_>>()
            .join(",");
        sql.push_str(&format!(" AND relation IN ({placeholders})"));
    }
    if has_min_weight {
        sql.push_str(&format!(" AND weight >= ?{}", 4 + relation_count));
    }
    sql.push_str(" LIMIT ?3");
    sql
}

fn traversal_timeout_error(budget: &TraversalExecutionBudget) -> StorageError {
    StorageError::Timeout {
        operation: format!(
            "traverse ({}ms execution budget)",
            budget.max_duration().as_millis()
        )
        .into(),
    }
}

fn traversal_work_error(budget: &TraversalExecutionBudget) -> StorageError {
    StorageError::InvalidInput {
        capability: StorageCapability::Graph,
        operation: "traverse".into(),
        message: format!(
            "traversal work budget exceeded after {} adjacency rows; \
             narrow roots, depth, relations, or result limit",
            budget.work_limit()
        ),
    }
}

#[derive(Clone, Copy)]
struct TraversalFrontierNode {
    node_id: Uuid,
    depth: usize,
    total_weight: f64,
}

#[allow(clippy::too_many_arguments)]
fn run_bounded_traversal(
    conn: &rusqlite::Connection,
    roots: Vec<Uuid>,
    opts: TraversalOptions,
    include_roots: bool,
    namespace: String,
    origin: khive_storage::tx_registry::TxOrigin,
    budget: TraversalExecutionBudget,
    counted_rows: &std::sync::atomic::AtomicU64,
    counted_queries: &std::sync::atomic::AtomicU64,
) -> Result<Vec<GraphPath>, StorageError> {
    let progress_timed_out = Arc::new(std::sync::atomic::AtomicBool::new(false));
    let callback_timed_out = Arc::clone(&progress_timed_out);
    let callback_budget = budget.clone();
    #[cfg(test)]
    let progress_seam_root = roots.first().copied();
    conn.progress_handler(
        1_000,
        Some(move || {
            if crate::read_cancellation::current_read_should_interrupt() {
                return true;
            }
            #[cfg(test)]
            if tests::traverse_progress_seam::hook(progress_seam_root) {
                callback_timed_out.store(true, std::sync::atomic::Ordering::Relaxed);
                return true;
            }
            let expired = callback_budget.is_expired();
            if expired {
                callback_timed_out.store(true, std::sync::atomic::Ordering::Relaxed);
            }
            expired
        }),
    )
    .map_err(|e| map_err(e, "traverse_progress_handler"))?;

    let result = (|| {
        let result_limit = opts.effective_limit() as usize;
        let relation_count = opts.relations.as_ref().map_or(0, Vec::len);
        let directions = match opts.direction {
            Direction::Out => vec![Direction::Out],
            Direction::In => vec![Direction::In],
            Direction::Both => vec![Direction::Out, Direction::In],
        };
        let statements = directions
            .into_iter()
            .map(|direction| {
                traversal_neighbor_sql(direction, relation_count, opts.min_weight.is_some())
            })
            .collect::<Vec<_>>();
        let map_sql_error = |error| {
            if progress_timed_out.load(std::sync::atomic::Ordering::Relaxed) {
                traversal_timeout_error(&budget)
            } else {
                map_err(error, "traverse")
            }
        };

        let mut all_paths = Vec::with_capacity(roots.len());
        for root_id in roots {
            let mut seen = HashSet::new();
            seen.insert(root_id);
            let mut frontier = VecDeque::from([TraversalFrontierNode {
                node_id: root_id,
                depth: 0,
                total_weight: 0.0,
            }]);
            let mut nodes = Vec::with_capacity(result_limit + usize::from(include_roots));
            if include_roots {
                nodes.push(PathNode {
                    node_id: root_id,
                    via_edge: None,
                    depth: 0,
                    name: None,
                    kind: None,
                    properties: None,
                    weight: 0.0,
                });
            }
            let mut non_root_count = 0usize;

            'root_walk: while non_root_count < result_limit {
                let Some(current) = frontier.pop_front() else {
                    break;
                };
                if current.depth >= opts.max_depth {
                    continue;
                }
                if budget.is_expired() {
                    return Err(traversal_timeout_error(&budget));
                }

                for sql in &statements {
                    let row_cap = budget.remaining_work().saturating_add(1);
                    let mut params: Vec<Box<dyn rusqlite::types::ToSql>> = vec![
                        Box::new(namespace.clone()),
                        Box::new(current.node_id.to_string()),
                        Box::new(row_cap as i64),
                    ];
                    if let Some(relations) = &opts.relations {
                        params.extend(relations.iter().map(|relation| {
                            Box::new(relation.to_string()) as Box<dyn rusqlite::types::ToSql>
                        }));
                    }
                    if let Some(min_weight) = opts.min_weight {
                        params.push(Box::new(min_weight));
                    }
                    let param_refs = params
                        .iter()
                        .map(|param| param.as_ref())
                        .collect::<Vec<&dyn rusqlite::types::ToSql>>();

                    let _snapshot = khive_storage::tx_registry::register_scoped(
                        Some("graph_traverse_read".to_string()),
                        origin.clone(),
                    );
                    let mut stmt = conn.prepare(sql).map_err(&map_sql_error)?;
                    counted_queries.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    let mut rows = stmt.query(param_refs.as_slice()).map_err(&map_sql_error)?;
                    while let Some(row) = rows.next().map_err(&map_sql_error)? {
                        // Test seam after `sqlite3_step` has produced a row:
                        // the cursor's read snapshot is demonstrably live.
                        #[cfg(test)]
                        tests::traverse_snapshot_seam::hook(current.node_id);
                        counted_rows.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        if budget.is_expired() {
                            return Err(traversal_timeout_error(&budget));
                        }
                        if !budget.try_consume_row() {
                            return Err(traversal_work_error(&budget));
                        }
                        let node_str: String = row.get(0).map_err(&map_sql_error)?;
                        let edge_str: String = row.get(1).map_err(&map_sql_error)?;
                        let edge_weight: f64 = row.get(2).map_err(&map_sql_error)?;
                        let node_id = parse_uuid(&node_str).map_err(&map_sql_error)?;
                        if !seen.insert(node_id) {
                            continue;
                        }
                        let via_edge = parse_uuid(&edge_str).map_err(&map_sql_error)?;
                        let depth = current.depth + 1;
                        let total_weight = current.total_weight + edge_weight;
                        nodes.push(PathNode {
                            node_id,
                            via_edge: Some(via_edge),
                            depth,
                            name: None,
                            kind: None,
                            properties: None,
                            weight: total_weight,
                        });
                        non_root_count += 1;
                        if depth < opts.max_depth {
                            frontier.push_back(TraversalFrontierNode {
                                node_id,
                                depth,
                                total_weight,
                            });
                        }
                        if non_root_count == result_limit {
                            break 'root_walk;
                        }
                    }
                }
            }

            if !nodes.is_empty() {
                let total_weight = nodes.iter().map(|node| node.weight).fold(0.0_f64, f64::max);
                all_paths.push(GraphPath {
                    root_id,
                    nodes,
                    total_weight,
                });
            }
        }
        Ok(all_paths)
    })();

    conn.progress_handler(0, None::<fn() -> bool>)
        .map_err(|e| map_err(e, "traverse_progress_handler_clear"))?;
    result
}

impl SqlGraphStore {
    async fn query_neighbors_page(
        &self,
        operation: &'static str,
        node_id: Uuid,
        query: NeighborQuery,
        after: Option<NeighborCursor>,
        neighbor_kinds: Option<Vec<String>>,
    ) -> Result<Vec<NeighborHit>, StorageError> {
        count_neighbor_select();

        let namespace = self.namespace.clone();
        let node_str = node_id.to_string();
        let counted_queries = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let counted_rows = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let closure_queries = Arc::clone(&counted_queries);
        let closure_rows = Arc::clone(&counted_rows);
        let result = self
            .with_reader(operation, move |conn| {
                let base_out = "SELECT target_id AS node_id, id AS edge_id, relation, weight \
                            FROM graph_edges \
                            WHERE namespace = ?1 AND source_id = ?2 AND deleted_at IS NULL";
                let base_in = "SELECT source_id AS node_id, id AS edge_id, relation, weight \
                           FROM graph_edges \
                           WHERE namespace = ?1 AND target_id = ?2 AND deleted_at IS NULL";
                let sql = match query.direction {
                    Direction::Out => base_out.to_string(),
                    Direction::In => base_in.to_string(),
                    Direction::Both => format!("{} UNION ALL {}", base_out, base_in),
                };
                let (where_extra, limit_clause, extra_params) =
                    neighbor_extra_clause(&query, 3, after.as_ref(), neighbor_kinds.as_deref());
                let full_sql = format!(
                    "SELECT node_id, edge_id, relation, weight FROM ({}){} \
                 ORDER BY weight DESC, node_id ASC, edge_id ASC{}",
                    sql, where_extra, limit_clause
                );
                let mut stmt = conn.prepare(&full_sql)?;
                let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
                all_params.push(Box::new(namespace.clone()));
                all_params.push(Box::new(node_str.clone()));
                all_params.extend(extra_params);
                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    all_params.iter().map(|p| p.as_ref()).collect();

                closure_queries.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let rows = stmt.query_map(param_refs.as_slice(), |row| {
                    let nid_str: String = row.get(0)?;
                    let eid_str: String = row.get(1)?;
                    let relation_str: String = row.get(2)?;
                    let weight: f64 = row.get(3)?;
                    Ok((nid_str, eid_str, relation_str, weight))
                })?;
                let mut hits = Vec::new();
                for row in rows {
                    let (nid_str, eid_str, relation_str, weight) = row?;
                    closure_rows.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            2,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;
                    hits.push(NeighborHit {
                        node_id: parse_uuid(&nid_str)?,
                        edge_id: parse_uuid(&eid_str)?,
                        relation,
                        weight,
                        name: None,
                        kind: None,
                        entity_type: None,
                    });
                }
                Ok(hits)
            })
            .await;

        report_graph_usage(&counted_queries, &counted_rows);
        result
    }
}

#[async_trait]
impl GraphStore for SqlGraphStore {
    async fn latest_annotating_note(
        &self,
        node_id: Uuid,
        kind: &str,
        tag: &str,
    ) -> Result<Option<(Uuid, i64)>, StorageError> {
        let namespace = self.namespace.clone();
        let node_id = node_id.to_string();
        let kind = kind.to_owned();
        let tag = tag.to_owned();
        self.with_reader("latest_annotating_note", move |conn| {
            conn.query_row(
                LATEST_ANNOTATING_NOTE_SQL,
                rusqlite::params![namespace, node_id, kind, tag],
                |row| {
                    let id: String = row.get(0)?;
                    Ok((parse_uuid(&id)?, row.get(1)?))
                },
            )
            .optional()
        })
        .await
    }

    async fn upsert_edge(&self, edge: Edge) -> Result<(), StorageError> {
        self.upsert_edge_observed(EdgeUpsertRequest {
            edge,
            resurrect: false,
        })
        .await
        .map(|_| ())
    }

    async fn upsert_edge_observed(
        &self,
        request: EdgeUpsertRequest,
    ) -> Result<EdgeUpsertResult, StorageError> {
        match self
            .observed_edge_write("upsert_edge_observed", request, false)
            .await?
        {
            GuardedEdgeUpsertOutcome::Written(result) => Ok(result),
            GuardedEdgeUpsertOutcome::Refused(EdgeUpsertRefusal::ResurrectionRequired { edge }) => {
                Err(resurrection_required_error("upsert_edge_observed", &edge))
            }
            GuardedEdgeUpsertOutcome::Refused(EdgeUpsertRefusal::MissingEndpoints(_)) => {
                Err(StorageError::Conflict {
                    capability: StorageCapability::Graph,
                    operation: "upsert_edge_observed".into(),
                    message: "unguarded edge upsert reported a missing-endpoint refusal".into(),
                })
            }
        }
    }

    async fn insert_edge_if_absent(&self, edge: Edge) -> Result<bool, StorageError> {
        let statement = edge_insert_if_absent_statement(&edge);
        self.with_writer("insert_edge_if_absent", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn replace_edge_if_unchanged(
        &self,
        edge: Edge,
        expected_updated_at: DateTime<Utc>,
        expected_deleted_at: Option<DateTime<Utc>>,
    ) -> Result<bool, StorageError> {
        let statement =
            edge_replace_if_unchanged_statement(&edge, expected_updated_at, expected_deleted_at);
        self.with_writer("replace_edge_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 upsert_edges(&self, edges: Vec<Edge>) -> Result<BatchWriteSummary, StorageError> {
        let attempted = edges.len() as u64;
        let requests = edges
            .into_iter()
            .map(|edge| EdgeUpsertRequest {
                edge,
                resurrect: false,
            })
            .collect();
        let outcome = self
            .observed_edge_batch_write("upsert_edges", requests, false)
            .await?;
        if let Some(refusal) = outcome.refusal {
            return match refusal.reason {
                EdgeUpsertRefusal::ResurrectionRequired { edge } => {
                    Err(resurrection_required_error("upsert_edges", &edge))
                }
                EdgeUpsertRefusal::MissingEndpoints(_) => Err(StorageError::Conflict {
                    capability: StorageCapability::Graph,
                    operation: "upsert_edges".into(),
                    message: "unguarded edge batch reported a missing-endpoint refusal".into(),
                }),
            };
        }
        Ok(BatchWriteSummary {
            attempted,
            affected: outcome.rows.len() as u64,
            ..BatchWriteSummary::default()
        })
    }

    async fn upsert_edge_guarded(&self, edge: Edge) -> Result<GuardedWriteOutcome, StorageError> {
        match self
            .upsert_edge_guarded_observed(EdgeUpsertRequest {
                edge,
                resurrect: false,
            })
            .await?
        {
            GuardedEdgeUpsertOutcome::Written(_) => Ok(GuardedWriteOutcome::Written),
            GuardedEdgeUpsertOutcome::Refused(EdgeUpsertRefusal::MissingEndpoints(missing)) => {
                Ok(GuardedWriteOutcome::Refused(missing))
            }
            GuardedEdgeUpsertOutcome::Refused(EdgeUpsertRefusal::ResurrectionRequired { edge }) => {
                Err(resurrection_required_error("upsert_edge_guarded", &edge))
            }
        }
    }

    async fn upsert_edge_guarded_observed(
        &self,
        request: EdgeUpsertRequest,
    ) -> Result<GuardedEdgeUpsertOutcome, StorageError> {
        self.observed_edge_write("upsert_edge_guarded_observed", request, true)
            .await
    }

    async fn upsert_edges_guarded(
        &self,
        edges: Vec<Edge>,
    ) -> Result<GuardedBatchOutcome, StorageError> {
        let attempted = edges.len() as u64;
        let requests = edges
            .iter()
            .cloned()
            .map(|edge| EdgeUpsertRequest {
                edge,
                resurrect: false,
            })
            .collect();
        let outcome = self.upsert_edges_guarded_observed(requests).await?;
        match outcome.refusal {
            None => Ok(GuardedBatchOutcome {
                summary: BatchWriteSummary {
                    attempted,
                    affected: outcome.rows.len() as u64,
                    ..BatchWriteSummary::default()
                },
                refused: None,
            }),
            Some(refusal) => match refusal.reason {
                EdgeUpsertRefusal::MissingEndpoints(missing) => {
                    let index = refusal.entry_index;
                    let edge = &edges[index];
                    let (source_id, target_id) =
                        canonical_edge_endpoints(edge.relation, edge.source_id, edge.target_id);
                    let message = format!(
                        "batch entry {index}: edge endpoint no longer exists at write time: source \
                         {source_id} or target {target_id}"
                    );
                    let mut summary = BatchWriteSummary {
                        attempted,
                        ..BatchWriteSummary::default()
                    };
                    // Keep the culprit's legacy first_error while recording
                    // per-input error classes and retryability in input order.
                    summary.first_error = message.clone();
                    let refusal = GuardedBatchRefusal {
                        entry_index: index,
                        missing,
                    };
                    for (failed_index, failed_edge) in edges.iter().enumerate() {
                        refusal.record_failure(&mut summary, failed_index, failed_edge, &message);
                    }
                    Ok(GuardedBatchOutcome {
                        summary,
                        refused: Some(refusal),
                    })
                }
                EdgeUpsertRefusal::ResurrectionRequired { edge } => {
                    Err(resurrection_required_error("upsert_edges_guarded", &edge))
                }
            },
        }
    }

    async fn upsert_edges_guarded_observed(
        &self,
        requests: Vec<EdgeUpsertRequest>,
    ) -> Result<GuardedEdgeBatchUpsertOutcome, StorageError> {
        self.observed_edge_batch_write("upsert_edges_guarded_observed", requests, true)
            .await
    }

    async fn get_edge(&self, id: LinkId) -> Result<Option<Edge>, StorageError> {
        let id_str = Uuid::from(id).to_string();

        self.with_reader("get_edge", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges 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_edge(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn get_edge_including_deleted(&self, id: LinkId) -> Result<Option<Edge>, StorageError> {
        let id_str = Uuid::from(id).to_string();

        self.with_reader("get_edge_including_deleted", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges WHERE id = ?1",
            )?;
            let mut rows = stmt.query(rusqlite::params![id_str])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_edge(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

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

    async fn edge_sequences(&self, ids: &[Uuid]) -> Result<Vec<(Uuid, i64)>, StorageError> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        let ids = ids.to_vec();
        self.with_reader("edge_sequences", move |conn| {
            const CHUNK: usize = 900;
            let mut resolved = Vec::with_capacity(ids.len());
            for chunk in ids.chunks(CHUNK) {
                let placeholders = (1..=chunk.len())
                    .map(|index| format!("?{index}"))
                    .collect::<Vec<_>>()
                    .join(", ");
                let sql = format!(
                    "SELECT edge_id, seq FROM graph_edges_seq WHERE edge_id IN ({placeholders})"
                );
                let strings = chunk.iter().map(Uuid::to_string).collect::<Vec<_>>();
                let params = strings
                    .iter()
                    .map(|id| id as &dyn rusqlite::types::ToSql)
                    .collect::<Vec<_>>();
                let mut stmt = conn.prepare(&sql)?;
                let rows = stmt.query_map(params.as_slice(), |row| {
                    let id: String = row.get(0)?;
                    Ok((parse_uuid(&id)?, row.get::<_, i64>(1)?))
                })?;
                resolved.extend(rows.collect::<Result<Vec<_>, _>>()?);
            }
            Ok(resolved)
        })
        .await
    }

    async fn get_edge_by_natural_key_including_deleted(
        &self,
        namespace: &str,
        source_id: Uuid,
        target_id: Uuid,
        relation: EdgeRelation,
    ) -> Result<Option<Edge>, StorageError> {
        let namespace = namespace.to_string();
        let source_str = source_id.to_string();
        let target_str = target_id.to_string();
        let relation_str = relation.to_string();

        self.with_reader("get_edge_by_natural_key_including_deleted", move |conn| {
            let mut stmt = conn.prepare(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges \
                 WHERE namespace = ?1 AND source_id = ?2 AND target_id = ?3 AND relation = ?4",
            )?;
            let mut rows = stmt.query(rusqlite::params![
                namespace,
                source_str,
                target_str,
                relation_str
            ])?;
            match rows.next()? {
                Some(row) => Ok(Some(read_edge(row)?)),
                None => Ok(None),
            }
        })
        .await
    }

    async fn get_edges(&self, ids: &[LinkId]) -> Result<Vec<Edge>, StorageError> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        // SQLite SQLITE_MAX_VARIABLE_NUMBER defaults to 999; chunk at 900 to stay safe.
        const CHUNK: usize = 900;
        let id_strs: Vec<String> = ids.iter().map(|id| Uuid::from(*id).to_string()).collect();

        let mut result: Vec<Edge> = Vec::with_capacity(ids.len());
        for chunk in id_strs.chunks(CHUNK) {
            let chunk_owned: Vec<String> = chunk.to_vec();
            let edges = self
                .with_reader("get_edges", move |conn| {
                    let placeholders: Vec<String> =
                        (1..=chunk_owned.len()).map(|i| format!("?{}", i)).collect();
                    let sql = format!(
                        "SELECT namespace, id, source_id, target_id, relation, weight, \
                                created_at, updated_at, deleted_at, metadata, target_backend \
                         FROM graph_edges WHERE id IN ({}) AND deleted_at IS NULL",
                        placeholders.join(",")
                    );
                    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_edge)?;
                    let mut edges = Vec::new();
                    for row in rows {
                        edges.push(row?);
                    }
                    Ok(edges)
                })
                .await?;
            result.extend(edges);
        }
        Ok(result)
    }

    async fn batch_neighbors(
        &self,
        sources: &[Uuid],
        query: NeighborQuery,
    ) -> Result<Vec<(Uuid, NeighborHit)>, StorageError> {
        use khive_storage::types::Direction;

        if sources.is_empty() {
            return Ok(Vec::new());
        }
        let mut seen_sources = HashSet::with_capacity(sources.len());
        let unique_sources: Vec<Uuid> = sources
            .iter()
            .copied()
            .filter(|source| seen_sources.insert(*source))
            .collect();
        const CHUNK_SIZE: usize = 880;

        let namespace = self.namespace.clone();
        let mut result: Vec<(Uuid, NeighborHit)> = Vec::new();
        // Declared across the whole chunk loop, not per chunk: a failure in a
        // later chunk must not erase what the earlier chunks already did.
        let counted_queries = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let counted_rows = Arc::new(std::sync::atomic::AtomicU64::new(0));

        for chunk in unique_sources.chunks(CHUNK_SIZE) {
            let chunk_owned: Vec<Uuid> = chunk.to_vec();
            let query_clone = query.clone();
            let ns = namespace.clone();
            let closure_queries = Arc::clone(&counted_queries);
            let closure_rows = Arc::clone(&counted_rows);

            let chunk_result = self
                .with_reader("batch_neighbors", move |conn| {
                    let src_strs: Vec<String> = chunk_owned.iter().map(|u| u.to_string()).collect();

                    let sources_json = serde_json::to_string(&src_strs).map_err(|error| {
                        rusqlite::Error::ToSqlConversionFailure(Box::new(error))
                    })?;

                    let build_inner_sql =
                        |direction_out: bool,
                         q: &NeighborQuery|
                         -> (String, Vec<String>, Option<f64>) {
                            let (filter_col, node_col) = if direction_out {
                                ("source_id", "target_id")
                            } else {
                                ("target_id", "source_id")
                            };

                            let mut rel_params: Vec<String> = Vec::new();
                            let mut conditions: Vec<String> = Vec::new();
                            let mut param_idx = 3;

                            if let Some(ref rels) = q.relations {
                                if !rels.is_empty() {
                                    let ps: Vec<String> = rels
                                        .iter()
                                        .map(|r| {
                                            rel_params.push(r.to_string());
                                            let p = format!("?{param_idx}");
                                            param_idx += 1;
                                            p
                                        })
                                        .collect();
                                    conditions
                                        .push(format!("edges.relation IN ({})", ps.join(",")));
                                }
                            }

                            // min_weight is returned separately so it can be added to
                            // all_params AFTER the rel_params block, at the right index.
                            let min_weight_val = if let Some(min_w) = q.min_weight {
                                conditions.push(format!("edges.weight >= ?{param_idx}"));
                                Some(min_w)
                            } else {
                                None
                            };

                            let where_extra = if conditions.is_empty() {
                                String::new()
                            } else {
                                format!(" AND {}", conditions.join(" AND "))
                            };

                            let sql = format!(
                                "SELECT requested.origin_id, edges.{node_col} AS node_id, \
                                 edges.id AS edge_id, edges.relation, edges.weight \
                                 FROM requested CROSS JOIN graph_edges AS edges \
                                   ON edges.{filter_col} = requested.origin_id \
                                 WHERE edges.namespace = ?1 \
                                   AND edges.deleted_at IS NULL{where_extra}",
                            );
                            (sql, rel_params, min_weight_val)
                        };

                    let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
                    all_params.push(Box::new(ns.to_string()));
                    all_params.push(Box::new(sources_json));

                    let (combined_inner, rel_params, min_weight_val) = match query_clone.direction {
                        Direction::Out => build_inner_sql(true, &query_clone),
                        Direction::In => build_inner_sql(false, &query_clone),
                        Direction::Both => {
                            let (out_sql, rel_params, min_weight_val) =
                                build_inner_sql(true, &query_clone);
                            let (in_sql, _, _) = build_inner_sql(false, &query_clone);
                            (
                                format!("{out_sql} UNION ALL {in_sql}"),
                                rel_params,
                                min_weight_val,
                            )
                        }
                    };

                    for relation in rel_params {
                        all_params.push(Box::new(relation));
                    }
                    if let Some(min_weight) = min_weight_val {
                        all_params.push(Box::new(min_weight));
                    }
                    let limit_param_idx = all_params.len() + 1;

                    // Wrap combined inner with per-source ROW_NUMBER limit if needed.
                    //
                    // Deterministic weight-descending order, tie-broken by node_id
                    // ascending, applied INSIDE the window's ORDER BY — otherwise a
                    // per-origin cap can silently drop high-weight neighbors in favor
                    // of arbitrary SQLite row order (mirrors neighbors(), ADR-089
                    // context-verb review; issue #589).
                    let full_sql = if let Some(lim) = query_clone.limit {
                        all_params.push(Box::new(lim as i64));
                        format!(
                            "WITH requested(origin_id) AS (\
                               SELECT value FROM json_each(?2)\
                             ) SELECT origin_id, node_id, edge_id, relation, weight \
                             FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY origin_id \
                                   ORDER BY weight DESC, node_id ASC) AS rn \
                                   FROM ({combined_inner})) WHERE rn <= ?{limit_param_idx}",
                        )
                    } else {
                        format!(
                            "WITH requested(origin_id) AS (\
                               SELECT value FROM json_each(?2)\
                             ) SELECT origin_id, node_id, edge_id, relation, weight \
                             FROM ({combined_inner})",
                        )
                    };

                    let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                        all_params.iter().map(|p| p.as_ref()).collect();

                    let mut stmt = conn.prepare(&full_sql)?;
                    closure_queries.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    let rows = stmt.query_map(param_refs.as_slice(), |row| {
                        let origin_str: String = row.get(0)?;
                        let nid_str: String = row.get(1)?;
                        let eid_str: String = row.get(2)?;
                        let relation_str: String = row.get(3)?;
                        let weight: f64 = row.get(4)?;
                        Ok((origin_str, nid_str, eid_str, relation_str, weight))
                    })?;

                    let mut pairs = Vec::new();
                    for row in rows {
                        let (origin_str, nid_str, eid_str, relation_str, weight) = row?;
                        closure_rows.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                        let origin = parse_uuid(&origin_str)?;
                        let node_id = parse_uuid(&nid_str)?;
                        let edge_id = parse_uuid(&eid_str)?;
                        let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
                            rusqlite::Error::FromSqlConversionFailure(
                                3,
                                rusqlite::types::Type::Text,
                                Box::new(e),
                            )
                        })?;
                        pairs.push((
                            origin,
                            NeighborHit {
                                node_id,
                                edge_id,
                                relation,
                                weight,
                                name: None,
                                kind: None,
                                entity_type: None,
                            },
                        ));
                    }
                    Ok(pairs)
                })
                .await;
            let pairs = match chunk_result {
                Ok(pairs) => pairs,
                Err(e) => {
                    report_graph_usage(&counted_queries, &counted_rows);
                    return Err(e);
                }
            };
            result.extend(pairs);
        }
        report_graph_usage(&counted_queries, &counted_rows);

        let requested: HashSet<Uuid> = unique_sources.iter().copied().collect();
        let mut grouped: HashMap<Uuid, Vec<NeighborHit>> =
            HashMap::with_capacity(unique_sources.len());
        for (origin, hit) in result {
            if !requested.contains(&origin) {
                return Err(StorageError::Internal(format!(
                    "batch_neighbors returned unrequested origin {origin}"
                )));
            }
            grouped.entry(origin).or_default().push(hit);
        }

        for hits in grouped.values_mut() {
            hits.sort_by(|a, b| {
                b.weight
                    .partial_cmp(&a.weight)
                    .unwrap_or(std::cmp::Ordering::Equal)
                    .then(a.node_id.cmp(&b.node_id))
                    .then(a.edge_id.cmp(&b.edge_id))
            });
        }

        let mut ordered = Vec::new();
        for &source in sources {
            if let Some(hits) = grouped.get(&source) {
                ordered.extend(hits.iter().cloned().map(|hit| (source, hit)));
            }
        }
        Ok(ordered)
    }

    async fn delete_edge(&self, id: LinkId, mode: DeleteMode) -> Result<bool, StorageError> {
        let id = Uuid::from(id);
        let statement = match mode {
            DeleteMode::Soft => {
                edge_soft_delete_statement(id, chrono::Utc::now().timestamp_micros())
            }
            DeleteMode::Hard => edge_hard_delete_statement(id),
        };
        self.with_writer("delete_edge", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? > 0)
        })
        .await
    }

    async fn query_edges(
        &self,
        filter: EdgeFilter,
        sort: Vec<SortOrder<EdgeSortField>>,
        page: PageRequest,
    ) -> Result<Page<Edge>, StorageError> {
        let namespace = self.namespace.clone();
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Graph,
            operation: "query_edges".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;
        self.with_reader("query_edges", move |conn| {
            let (where_clause, mut all_params) = build_edge_filter_sql(&namespace, &filter);
            let order_clause = edge_order_clause(&sort);
            all_params.push(Box::new(limit_i64));
            all_params.push(Box::new(offset_i64));

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

            let data_sql = format!(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges{}{} LIMIT ?{} OFFSET ?{}",
                where_clause, order_clause, limit_idx, offset_idx,
            );

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

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

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

    async fn count_edges(&self, filter: EdgeFilter) -> Result<u64, StorageError> {
        let namespace = self.namespace.clone();
        self.with_reader("count_edges", move |conn| {
            let (where_clause, params) = build_edge_filter_sql(&namespace, &filter);
            let sql = format!(
                "SELECT COUNT(*) FROM graph_edges{}",
                with_live_endpoints(&where_clause)
            );
            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_edges_in_namespaces(
        &self,
        namespaces: &[String],
        filter: EdgeFilter,
    ) -> Result<u64, StorageError> {
        let namespaces: Vec<String> = namespaces
            .iter()
            .cloned()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        self.with_reader("count_edges_in_namespaces", move |conn| {
            let mut total = 0;
            for chunk in namespaces.chunks(NAMESPACE_COUNT_CHUNK_SIZE) {
                let (where_clause, params) = build_edge_filter_sql_for_namespaces(chunk, &filter);
                let sql = format!(
                    "SELECT COUNT(*) FROM graph_edges{}",
                    with_live_endpoints(&where_clause)
                );
                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
    }

    async fn query_edges_in_namespaces(
        &self,
        namespaces: &[String],
        filter: EdgeFilter,
        sort: Vec<SortOrder<EdgeSortField>>,
        page: PageRequest,
    ) -> Result<Page<Edge>, StorageError> {
        // One statement with `namespace IN (...)` and real SQL paging: a
        // per-namespace prefix fetch merged and re-sliced client-side floats
        // the offset window between calls, silently duplicating and skipping
        // rows during enumeration (#2088).
        //
        // The namespace set is bound as a single JSON-array parameter (see
        // `build_edge_filter_sql_for_namespaces_json`), not one `?N` per
        // namespace: a caller visible in hundreds-to-thousands of namespaces
        // would otherwise blow past `SQLITE_LIMIT_VARIABLE_NUMBER` before any
        // filter parameter is even added, and chunking the namespace set (as
        // the count-only aggregate methods below do) is not an option here —
        // it would reintroduce exactly the floating-offset bug this method
        // exists to close, since a per-chunk LIMIT/OFFSET cannot be
        // re-sliced into one globally exact page.
        let namespaces: Vec<String> = {
            let mut seen = HashSet::new();
            namespaces
                .iter()
                .filter(|ns| seen.insert((*ns).clone()))
                .cloned()
                .collect()
        };
        let limit_i64 = i64::from(page.limit);
        let offset_i64 = i64::try_from(page.offset).map_err(|_| StorageError::InvalidInput {
            capability: StorageCapability::Graph,
            operation: "query_edges_in_namespaces".into(),
            message: format!(
                "PageRequest: offset must be <= i64::MAX, got {}",
                page.offset
            ),
        })?;
        self.with_reader("query_edges_in_namespaces", move |conn| {
            let namespaces_json = serde_json::to_string(&namespaces)
                .map_err(|error| rusqlite::Error::ToSqlConversionFailure(Box::new(error)))?;

            let (where_clause, mut all_params) =
                build_edge_filter_sql_for_namespaces_json(&namespaces_json, &filter);
            let order_clause = edge_order_clause(&sort);
            all_params.push(Box::new(limit_i64));
            all_params.push(Box::new(offset_i64));

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

            let data_sql = format!(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges{}{} LIMIT ?{} OFFSET ?{}",
                where_clause, order_clause, limit_idx, offset_idx,
            );

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

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

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

    async fn count_edges_by_relation(&self) -> Result<Vec<(EdgeRelation, u64)>, StorageError> {
        let namespace = self.namespace.clone();
        self.with_reader("count_edges_by_relation", move |conn| {
            let sql = format!(
                "SELECT relation, COUNT(*) FROM graph_edges \
                 WHERE namespace = ?1 AND deleted_at IS NULL AND {LIVE_ENDPOINTS_CONDITION} \
                 GROUP BY relation"
            );
            let mut stmt = conn.prepare(&sql)?;
            let rows = stmt.query_map([&namespace], |row| {
                let relation_str: String = row.get(0)?;
                let count: i64 = row.get(1)?;
                Ok((relation_str, count))
            })?;
            let mut out = Vec::new();
            for row in rows {
                let (relation_str, count) = row?;
                let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
                    rusqlite::Error::FromSqlConversionFailure(
                        0,
                        rusqlite::types::Type::Text,
                        Box::new(e),
                    )
                })?;
                out.push((relation, count as u64));
            }
            Ok(out)
        })
        .await
    }

    async fn count_edges_by_relation_in_namespaces(
        &self,
        namespaces: &[String],
    ) -> Result<Vec<(EdgeRelation, u64)>, StorageError> {
        let namespaces: Vec<String> = namespaces
            .iter()
            .cloned()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        self.with_reader("count_edges_by_relation_in_namespaces", move |conn| {
            let mut totals = HashMap::new();
            for chunk in namespaces.chunks(NAMESPACE_COUNT_CHUNK_SIZE) {
                let (where_clause, params) =
                    build_edge_filter_sql_for_namespaces(chunk, &EdgeFilter::default());
                let sql = format!(
                    "SELECT relation, COUNT(*) FROM graph_edges{} GROUP BY relation",
                    with_live_endpoints(&where_clause)
                );
                let mut stmt = conn.prepare(&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(), |row| {
                    let relation_str: String = row.get(0)?;
                    let count: i64 = row.get(1)?;
                    Ok((relation_str, count))
                })?;
                for row in rows {
                    let (relation_str, count) = row?;
                    let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            0,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;
                    *totals.entry(relation).or_insert(0) += count as u64;
                }
            }
            Ok(totals.into_iter().collect())
        })
        .await
    }

    async fn count_edges_by_endpoint_base(&self) -> Result<EdgeEndpointBaseCounts, StorageError> {
        let namespace = self.namespace.clone();
        self.with_reader("count_edges_by_endpoint_base", move |conn| {
            let source_case = endpoint_base_case("source_id");
            let target_case = endpoint_base_case("target_id");
            let sql = format!(
                "SELECT {source_case} AS source_base, {target_case} AS target_base, COUNT(*) \
                 FROM graph_edges \
                 WHERE namespace = ?1 AND deleted_at IS NULL AND {LIVE_ENDPOINTS_CONDITION} \
                 GROUP BY source_base, target_base"
            );
            let mut stmt = conn.prepare(&sql)?;
            let rows = stmt.query_map([&namespace], |row| {
                let source: String = row.get(0)?;
                let target: String = row.get(1)?;
                let count: i64 = row.get(2)?;
                Ok((source, target, count))
            })?;
            let mut counts = EdgeEndpointBaseCounts::default();
            for row in rows {
                let (source, target, count) = row?;
                fold_endpoint_base_row(&mut counts, &source, &target, count as u64);
            }
            Ok(counts)
        })
        .await
    }

    async fn count_edges_by_endpoint_base_in_namespaces(
        &self,
        namespaces: &[String],
    ) -> Result<EdgeEndpointBaseCounts, StorageError> {
        let namespaces: Vec<String> = namespaces
            .iter()
            .cloned()
            .collect::<HashSet<_>>()
            .into_iter()
            .collect();
        self.with_reader("count_edges_by_endpoint_base_in_namespaces", move |conn| {
            let source_case = endpoint_base_case("source_id");
            let target_case = endpoint_base_case("target_id");
            let mut counts = EdgeEndpointBaseCounts::default();
            for chunk in namespaces.chunks(NAMESPACE_COUNT_CHUNK_SIZE) {
                let (where_clause, params) =
                    build_edge_filter_sql_for_namespaces(chunk, &EdgeFilter::default());
                let sql = format!(
                    "SELECT {source_case} AS source_base, {target_case} AS target_base, COUNT(*) \
                     FROM graph_edges{} GROUP BY source_base, target_base",
                    with_live_endpoints(&where_clause)
                );
                let mut stmt = conn.prepare(&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(), |row| {
                    let source: String = row.get(0)?;
                    let target: String = row.get(1)?;
                    let count: i64 = row.get(2)?;
                    Ok((source, target, count))
                })?;
                for row in rows {
                    let (source, target, count) = row?;
                    fold_endpoint_base_row(&mut counts, &source, &target, count as u64);
                }
            }
            Ok(counts)
        })
        .await
    }

    async fn query_edges_after(
        &self,
        filter: EdgeFilter,
        after: Option<Uuid>,
        limit: u32,
    ) -> Result<EdgeSeekPage, StorageError> {
        let namespace = self.namespace.clone();
        let limit_usize = limit as usize;
        let probe_limit_i64 = i64::from(limit) + 1;
        self.with_reader("query_edges_after", move |conn| {
            let (mut where_clause, mut params) = build_edge_filter_sql(&namespace, &filter);
            if let Some(cursor) = after {
                params.push(Box::new(cursor.to_string()));
                where_clause.push_str(&format!(" AND id > ?{}", params.len()));
            }
            params.push(Box::new(probe_limit_i64));
            let limit_idx = params.len();

            // `where_clause` always pins `namespace = ?1`; adding `id > ?N` here
            // keeps the predicate a range scan against the implicit unique index
            // backing `PRIMARY KEY (namespace, id)` — equality on the leading
            // column plus a range on the trailing one, with `ORDER BY id ASC`
            // matching the index order, so SQLite seeks instead of scanning.
            let data_sql = format!(
                "SELECT namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend \
                 FROM graph_edges{} ORDER BY id ASC LIMIT ?{}",
                where_clause, limit_idx,
            );

            let mut stmt = conn.prepare(&data_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_edge)?;

            let mut items = Vec::new();
            for row in rows {
                items.push(row?);
            }
            let has_more = items.len() > limit_usize;
            if has_more {
                items.truncate(limit_usize);
            }
            let next_after = if has_more {
                items.last().map(|e| Uuid::from(e.id))
            } else {
                None
            };

            Ok(EdgeSeekPage { items, next_after })
        })
        .await
    }

    async fn query_edges_sequence_after(
        &self,
        filter: EdgeFilter,
        after: Option<SeekCursor>,
        limit: u32,
    ) -> Result<SeekPage<Edge>, StorageError> {
        if limit == 0 {
            return Ok(SeekPage::default());
        }
        let namespace = self.namespace.clone();
        let limit_usize = limit as usize;
        let probe_limit_i64 = i64::from(limit) + 1;
        self.with_reader("query_edges_sequence_after", move |conn| {
            let (mut where_clause, mut params) = build_edge_filter_sql(&namespace, &filter);
            if let Some(cursor) = after {
                params.push(Box::new(cursor.sequence));
                where_clause.push_str(&format!(" AND graph_edges_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 namespace, id, source_id, target_id, relation, weight, \
                        created_at, updated_at, deleted_at, metadata, target_backend, \
                        graph_edges_seq.seq \
                 FROM graph_edges_seq CROSS JOIN graph_edges \
                    ON graph_edges.id = graph_edges_seq.edge_id{where_clause} \
                 ORDER BY graph_edges_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_edge(row)?, row.get::<_, i64>(11)?))
            })?;
            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(|(edge, sequence)| SeekCursor {
                    sequence: *sequence,
                    id: Uuid::from(edge.id),
                })
            } else {
                None
            };
            let items = entries.into_iter().map(|(edge, _)| edge).collect();
            Ok(SeekPage { items, next_after })
        })
        .await
    }

    async fn neighbors(
        &self,
        node_id: Uuid,
        query: NeighborQuery,
    ) -> Result<Vec<NeighborHit>, StorageError> {
        self.query_neighbors_page("neighbors", node_id, query, None, None)
            .await
    }

    async fn neighbors_page(
        &self,
        node_id: Uuid,
        query: NeighborQuery,
        after: Option<NeighborCursor>,
        neighbor_kinds: Option<Vec<String>>,
    ) -> Result<Vec<NeighborHit>, StorageError> {
        self.query_neighbors_page("neighbors_page", node_id, query, after, neighbor_kinds)
            .await
    }

    /// Single-query both-direction neighbor fetch (ADR-089 context-verb
    /// optimization): projects a `'out'`/`'in'` literal from each `UNION ALL`
    /// arm so the caller gets direction labels without a second direction-
    /// scoped round trip. `query.direction` is ignored — always both.
    async fn neighbors_both_directions(
        &self,
        node_id: Uuid,
        query: NeighborQuery,
    ) -> Result<Vec<DirectedNeighborHit>, StorageError> {
        count_neighbor_select();

        let namespace = self.namespace.clone();
        let node_str = node_id.to_string();

        let counted_queries = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let counted_rows = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let closure_queries = Arc::clone(&counted_queries);
        let closure_rows = Arc::clone(&counted_rows);
        let result = self
            .with_reader("neighbors_both_directions", move |conn| {
                let base_out = "SELECT target_id AS node_id, id AS edge_id, relation, weight, \
                            'out' AS dir \
                            FROM graph_edges \
                            WHERE namespace = ?1 AND source_id = ?2 AND deleted_at IS NULL";
                let base_in = "SELECT source_id AS node_id, id AS edge_id, relation, weight, \
                           'in' AS dir \
                           FROM graph_edges \
                           WHERE namespace = ?1 AND target_id = ?2 AND deleted_at IS NULL";
                let sql = format!("{} UNION ALL {}", base_out, base_in);

                let (where_extra, limit_clause, extra_params) =
                    neighbor_extra_clause(&query, 3, None, None);

                // Same global weight-descending/node_id-ascending order as `neighbors`
                // (ADR-089 context-verb review),
                // applied across BOTH directions before `LIMIT` truncates. A
                // reciprocal pair (an Out edge and an In edge to/from the same
                // neighbor at the same weight) ties on `(weight, node_id)`, so the
                // order is extended with a direction rank (`out` before `in`) and
                // finally `edge_id` to make the pre-`LIMIT` order fully
                // deterministic).
                let full_sql = format!(
                    "SELECT node_id, edge_id, relation, weight, dir FROM ({}){} \
                 ORDER BY weight DESC, node_id ASC, \
                 CASE dir WHEN 'out' THEN 0 ELSE 1 END ASC, edge_id ASC{}",
                    sql, where_extra, limit_clause
                );

                let mut stmt = conn.prepare(&full_sql)?;

                let mut all_params: Vec<Box<dyn rusqlite::types::ToSql>> = Vec::new();
                all_params.push(Box::new(namespace.clone()));
                all_params.push(Box::new(node_str.clone()));
                all_params.extend(extra_params);

                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                    all_params.iter().map(|p| p.as_ref()).collect();

                closure_queries.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let rows = stmt.query_map(param_refs.as_slice(), |row| {
                    let nid_str: String = row.get(0)?;
                    let eid_str: String = row.get(1)?;
                    let relation_str: String = row.get(2)?;
                    let weight: f64 = row.get(3)?;
                    let dir_str: String = row.get(4)?;
                    Ok((nid_str, eid_str, relation_str, weight, dir_str))
                })?;

                let mut hits = Vec::new();
                for row in rows {
                    let (nid_str, eid_str, relation_str, weight, dir_str) = row?;
                    closure_rows.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                    let relation = relation_str.parse::<EdgeRelation>().map_err(|e| {
                        rusqlite::Error::FromSqlConversionFailure(
                            2,
                            rusqlite::types::Type::Text,
                            Box::new(e),
                        )
                    })?;
                    let direction = if dir_str == "out" {
                        Direction::Out
                    } else {
                        Direction::In
                    };
                    hits.push(DirectedNeighborHit {
                        hit: NeighborHit {
                            node_id: parse_uuid(&nid_str)?,
                            edge_id: parse_uuid(&eid_str)?,
                            relation,
                            weight,
                            name: None,
                            kind: None,
                            entity_type: None,
                        },
                        direction,
                    });
                }

                Ok(hits)
            })
            .await;

        report_graph_usage(&counted_queries, &counted_rows);
        result
    }

    async fn traverse(&self, request: TraversalRequest) -> Result<Vec<GraphPath>, StorageError> {
        request
            .validate()
            .map_err(|message| StorageError::InvalidInput {
                capability: StorageCapability::Graph,
                operation: "traverse".into(),
                message,
            })?;
        if request.roots.is_empty() {
            return Ok(Vec::new());
        }

        let mut distinct_roots = HashSet::with_capacity(request.roots.len());
        let roots = request
            .roots
            .iter()
            .copied()
            .filter(|root| distinct_roots.insert(*root))
            .collect::<Vec<_>>();
        let opts = request.options;
        let include_roots = request.include_roots;
        let namespace = self.namespace.clone();
        let origin = self.pool.origin();
        let budget = request.execution_budget;
        // Shared with the blocking closure so the counts survive an error.
        // `with_reader` runs the closure on a blocking thread where the
        // task-local usage context is invisible, so it cannot call
        // `usage::count` itself; returning the totals in the Ok value would
        // lose every round trip already issued when a later statement fails.
        let counted_rows = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let counted_queries = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let closure_rows = Arc::clone(&counted_rows);
        let closure_queries = Arc::clone(&counted_queries);
        let result = self
            .with_reader("traverse", move |conn| {
                Ok(run_bounded_traversal(
                    conn,
                    roots,
                    opts,
                    include_roots,
                    namespace,
                    origin,
                    budget,
                    closure_rows.as_ref(),
                    closure_queries.as_ref(),
                ))
            })
            .await
            .and_then(|inner| inner);

        // Accounted on BOTH outcomes. `db_round_trips` counts round trips
        // *issued* and `graph_hops` counts adjacency rows storage *returned*,
        // so work already done before a later statement errors is real work and
        // must appear. Reading the shared counters here rather than off the
        // Ok value is what makes that possible: the closure runs on a blocking
        // thread where the task-local usage context is not visible, so it
        // cannot count for itself, and a value returned only on success
        // reports nothing at all when the traversal fails partway.
        khive_storage::usage::count(
            khive_storage::usage::UsageUnit::DbRoundTrips,
            counted_queries.load(std::sync::atomic::Ordering::Relaxed),
        );
        khive_storage::usage::count(
            khive_storage::usage::UsageUnit::GraphHops,
            counted_rows.load(std::sync::atomic::Ordering::Relaxed),
        );

        result
    }

    async fn purge_incident_edges(&self, node_id: Uuid) -> Result<u64, StorageError> {
        // No namespace filter: UUID v4 is globally unique. Hard-delete cascade must
        // remove ALL incident edges regardless of which namespace they were written in
        // (ADR-002 no-dangling-references, ADR-007 by-ID contract).
        let statement = purge_incident_edges_statement(node_id);
        self.with_writer("purge_incident_edges", move |conn| {
            let mut stmt = conn.prepare(&statement.sql)?;
            bind_params(&mut stmt, &statement.params)?;
            Ok(stmt.raw_execute()? as u64)
        })
        .await
    }
}

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

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

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

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

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