khive-db 0.9.0

use super::*;
use crate::pool::PoolConfig;
use khive_storage::types::{
    Direction, TraversalExecutionBudget, TraversalOptions, MAX_TRAVERSAL_DEPTH, MAX_TRAVERSAL_ROOTS,
};
use khive_storage::{BatchWriteErrorClass, BatchWriteRetryability};
use rusqlite::hooks::{AuthAction, AuthContext, Authorization};
use serial_test::serial;
use std::collections::{HashMap, HashSet};

fn deny_count_function(ctx: AuthContext<'_>) -> Authorization {
    match ctx.action {
        AuthAction::Function { function_name } if function_name.eq_ignore_ascii_case("count") => {
            Authorization::Deny
        }
        _ => Authorization::Allow,
    }
}

/// Deterministic barrier immediately before the in-transaction endpoint
/// probe used by an observed guarded edge write.
///
/// A pure wall-clock race at this seam is not observable: a refused,
/// zero-row `INSERT` autocommits (and so releases SQLite's write lock)
/// before returning, and the probe that follows it is a plain `SELECT`,
/// which never blocks on a writer in WAL mode. Two back-to-back Rust calls
/// on the same thread with no `.await` between them leave no real window
/// for another OS thread to interleave — a racer would have to win a
/// same-thread, no-I/O footrace against its own scheduling latency, which
/// it structurally cannot do. This module replaces that footrace with a
/// real rendezvous: the guarded call blocks at the seam until the test
/// releases it, so the racer's write is forced to land in the seam's
/// window on unwrapped (pre-fix) code, and forced to block behind the
/// still-open `BEGIN IMMEDIATE` on wrapped (fixed) code.
///
/// Scoped by `(source_id, target_id)` rather than global so unrelated
/// `upsert_edge_guarded` calls from other tests running concurrently in the
/// same process are unaffected (`hook` is a no-op unless its key matches an
/// installed barrier).
pub(super) mod insert_probe_seam {
    use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
    use std::sync::Mutex;
    use uuid::Uuid;

    struct Barrier {
        key: (Uuid, Uuid),
        reached_tx: SyncSender<()>,
        proceed_rx: Receiver<()>,
    }

    static BARRIER: Mutex<Option<Barrier>> = Mutex::new(None);

    /// Installs a barrier for the given `(source_id, target_id)` key.
    /// Returns the `reached` receiver (fires once the guarded call has
    /// executed its `INSERT` and is parked at the seam) and the `proceed`
    /// sender the caller uses to release it.
    pub(crate) fn install(key: (Uuid, Uuid)) -> (Receiver<()>, SyncSender<()>) {
        let (reached_tx, reached_rx) = sync_channel(0);
        let (proceed_tx, proceed_rx) = sync_channel(0);
        *BARRIER.lock().unwrap() = Some(Barrier {
            key,
            reached_tx,
            proceed_rx,
        });
        (reached_rx, proceed_tx)
    }

    /// Clears any still-installed barrier. Safe to call unconditionally
    /// after a test finishes, whether or not `hook` ever consumed it.
    pub(crate) fn uninstall() {
        *BARRIER.lock().unwrap() = None;
    }

    /// Called from the production insert-then-probe seam. A no-op unless a
    /// barrier for this exact key is currently installed. Single-use: takes
    /// the barrier out of the static the moment it matches, so a second
    /// call with the same key (there is none in this test) would fall
    /// through as a no-op rather than re-blocking.
    pub(crate) fn hook(key: (Uuid, Uuid)) {
        let barrier = {
            let mut guard = BARRIER.lock().unwrap();
            match guard.as_ref() {
                Some(barrier) if barrier.key == key => guard.take(),
                _ => None,
            }
        };
        let Some(barrier) = barrier else {
            return;
        };
        let _ = barrier.reached_tx.send(());
        let _ = barrier.proceed_rx.recv();
    }
}

/// Deterministic observation point inside one statement-scoped traversal
/// snapshot. It is keyed by frontier node so unrelated tests and later BFS
/// statements remain unaffected.
pub(super) mod traverse_snapshot_seam {
    use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
    use std::sync::Mutex;
    use uuid::Uuid;

    struct Barrier {
        node_id: Uuid,
        reached_tx: SyncSender<()>,
        proceed_rx: Receiver<()>,
    }

    static BARRIER: Mutex<Option<Barrier>> = Mutex::new(None);

    pub(crate) fn install(node_id: Uuid) -> (Receiver<()>, SyncSender<()>) {
        let (reached_tx, reached_rx) = sync_channel(0);
        let (proceed_tx, proceed_rx) = sync_channel(0);
        *BARRIER.lock().unwrap() = Some(Barrier {
            node_id,
            reached_tx,
            proceed_rx,
        });
        (reached_rx, proceed_tx)
    }

    pub(crate) fn hook(node_id: Uuid) {
        let barrier = {
            let mut slot = BARRIER.lock().unwrap();
            if slot
                .as_ref()
                .is_none_or(|barrier| barrier.node_id != node_id)
            {
                return;
            }
            slot.take().unwrap()
        };
        barrier.reached_tx.send(()).unwrap();
        barrier.proceed_rx.recv().unwrap();
    }
}

/// Deterministically interrupts SQLite from inside traversal's VM progress
/// callback. The seam is keyed by root UUID so concurrently-running traversal
/// tests remain unaffected. It stays installed after its one interrupt so a
/// same-connection follow-up query can prove the callback was actually removed:
/// a leaked handler would increment `calls` again even though it no longer
/// requests interruption.
pub(super) mod traverse_progress_seam {
    use std::sync::Mutex;
    use uuid::Uuid;

    struct State {
        root_id: Uuid,
        interrupt_next: bool,
        calls: usize,
    }

    static STATE: Mutex<Option<State>> = Mutex::new(None);

    pub(crate) fn install(root_id: Uuid) {
        *STATE.lock().unwrap() = Some(State {
            root_id,
            interrupt_next: true,
            calls: 0,
        });
    }

    pub(crate) fn hook(root_id: Option<Uuid>) -> bool {
        let mut slot = STATE.lock().unwrap();
        let Some(state) = slot.as_mut() else {
            return false;
        };
        if root_id != Some(state.root_id) {
            return false;
        }
        state.calls += 1;
        std::mem::take(&mut state.interrupt_next)
    }

    pub(crate) fn calls(root_id: Uuid) -> usize {
        STATE
            .lock()
            .unwrap()
            .as_ref()
            .filter(|state| state.root_id == root_id)
            .map_or(0, |state| state.calls)
    }

    pub(crate) fn uninstall() {
        *STATE.lock().unwrap() = None;
    }
}

/// Memory-backed store with the substrate tables seeded.
///
/// The edge counts probe `entities`/`notes` for endpoint tombstones, so a database
/// holding the graph DDL alone can no longer answer them. That shape is test-only:
/// production applies one schema to one connection (`Backend::open`), and the guarded
/// edge insert already reads those two tables.
/// The graph DDL plus the minimal `entities`/`notes` tables the edge counts probe for
/// endpoint tombstones, applied together because a database holding one without the other
/// cannot answer those counts. That split is test-only: production applies one schema to
/// one connection, and the guarded edge insert already reads both tables.
fn apply_test_schema(conn: &rusqlite::Connection) {
    conn.execute_batch(GRAPH_DDL).unwrap();
    conn.execute_batch(
        "CREATE TABLE IF NOT EXISTS entities (id TEXT PRIMARY KEY, deleted_at INTEGER);
         CREATE TABLE IF NOT EXISTS notes (id TEXT PRIMARY KEY, deleted_at INTEGER);
         CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY);",
    )
    .unwrap();
}

fn setup_memory_store() -> SqlGraphStore {
    setup_memory_store_with_substrates().1
}

/// File-backed store plus an attribution view unique to this test. Cleanup
/// assertions must not inspect the process-global registry: unrelated tests
/// legitimately hold spans while this test binary runs in parallel.
fn setup_file_store_with_origin_view() -> (
    tempfile::TempDir,
    SqlGraphStore,
    khive_storage::tx_registry::TxOriginFilter,
) {
    let dir = tempfile::tempdir().unwrap();
    let config = PoolConfig {
        path: Some(dir.path().join("graph-test.db")),
        ..PoolConfig::for_test()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }
    let identity = match pool.origin() {
        khive_storage::tx_registry::TxOrigin::Database(identity) => identity,
        other => panic!("expected file-backed test pool, got {other:?}"),
    };
    let view = khive_storage::tx_registry::TxOriginFilter::Secondary(identity);
    (dir, SqlGraphStore::new_scoped(pool, false, "default"), view)
}

/// Like [`setup_memory_store`] but also seeds minimal `entities`/`notes`
/// tables (id + deleted_at only) so the `#769` guarded-write tests can
/// exercise the real `WHERE EXISTS(entities...) OR EXISTS(notes...)` probes
/// `edge_insert_guarded_by_endpoints_statement` and `edge_endpoints_exist`
/// issue against those tables.
fn setup_memory_store_with_substrates() -> (Arc<ConnectionPool>, SqlGraphStore) {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());

    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }

    let store = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "default");
    (pool, store)
}

fn insert_live_entity(pool: &ConnectionPool, id: Uuid) {
    let writer = pool.writer().unwrap();
    writer
        .conn()
        .execute(
            "INSERT INTO entities (id, deleted_at) VALUES (?1, NULL)",
            rusqlite::params![id.to_string()],
        )
        .unwrap();
}

fn soft_delete_entity(pool: &ConnectionPool, id: Uuid) {
    let writer = pool.writer().unwrap();
    let changed = writer
        .conn()
        .execute(
            "UPDATE entities SET deleted_at = ?2 WHERE id = ?1",
            rusqlite::params![id.to_string(), Utc::now().timestamp_micros()],
        )
        .unwrap();
    assert_eq!(changed, 1, "soft delete must have tombstoned a row");
}

fn insert_note(pool: &ConnectionPool, id: Uuid, deleted: bool) {
    let writer = pool.writer().unwrap();
    writer
        .conn()
        .execute(
            "INSERT INTO notes (id, deleted_at) VALUES (?1, ?2)",
            rusqlite::params![
                id.to_string(),
                deleted.then(|| Utc::now().timestamp_micros())
            ],
        )
        .unwrap();
}

fn soft_delete_note(pool: &ConnectionPool, id: Uuid) {
    let writer = pool.writer().unwrap();
    let changed = writer
        .conn()
        .execute(
            "UPDATE notes SET deleted_at = ?2 WHERE id = ?1",
            rusqlite::params![id.to_string(), Utc::now().timestamp_micros()],
        )
        .unwrap();
    assert_eq!(changed, 1, "soft delete must have tombstoned a note row");
}

fn hard_delete_entity(pool: &ConnectionPool, id: Uuid) {
    let writer = pool.writer().unwrap();
    writer
        .conn()
        .execute(
            "DELETE FROM entities WHERE id = ?1",
            rusqlite::params![id.to_string()],
        )
        .unwrap();
}

fn make_edge(source: Uuid, target: Uuid, relation: EdgeRelation, weight: f64) -> Edge {
    let now = Utc::now();
    Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: source,
        target_id: target,
        relation,
        weight,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    }
}

#[tokio::test]
async fn edge_pages_run_when_sqlite_count_is_denied() {
    let dir = tempfile::tempdir().unwrap();
    let pool = Arc::new(
        ConnectionPool::new(PoolConfig {
            path: Some(dir.path().join("graph-count-free-pages.db")),
            max_readers: 1,
            write_queue_enabled: Some(false),
            ..PoolConfig::default()
        })
        .unwrap(),
    );
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }
    // Pooled-reader mode lets the authorizer below observe the exact
    // connection used by both the control count and the page queries.
    let store = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "count-free-a");
    for namespace in ["count-free-a", "count-free-b"] {
        let mut edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
        edge.namespace = namespace.to_string();
        store.upsert_edge(edge).await.unwrap();
    }
    assert_eq!(
        store
            .count_edges_in_namespaces(
                &["count-free-a".to_string(), "count-free-b".to_string()],
                EdgeFilter::default(),
            )
            .await
            .unwrap(),
        2,
        "the explicit count API remains exact"
    );

    {
        let reader = pool.reader().unwrap();
        reader.conn().authorizer(Some(deny_count_function)).unwrap();
    }

    assert!(
        store.count_edges(EdgeFilter::default()).await.is_err(),
        "the exact-count control must be rejected by the authorizer"
    );
    let page = store
        .query_edges(
            EdgeFilter::default(),
            vec![],
            PageRequest {
                offset: 0,
                limit: 10,
            },
        )
        .await
        .expect("single-namespace page must not invoke SQLite count");
    assert_eq!(page.total, None);
    assert_eq!(page.items.len(), 1);

    let namespaces = vec!["count-free-a".to_string(), "count-free-b".to_string()];
    assert!(
        store
            .count_edges_in_namespaces(&namespaces, EdgeFilter::default())
            .await
            .is_err(),
        "the multi-namespace exact-count control must be rejected by the authorizer"
    );
    let page = store
        .query_edges_in_namespaces(
            &namespaces,
            EdgeFilter::default(),
            vec![],
            PageRequest {
                offset: 0,
                limit: 10,
            },
        )
        .await
        .expect("multi-namespace page must not invoke SQLite count");
    assert_eq!(page.total, None);
    assert_eq!(page.items.len(), 2);

    let reader = pool.reader().unwrap();
    reader
        .conn()
        .authorizer(None::<fn(AuthContext<'_>) -> Authorization>)
        .unwrap();
}

#[tokio::test]
async fn query_edges_in_namespaces_offset_paging_enumerates_exactly() {
    // #2088 regression: two namespaces, every edge sharing one created_at
    // (the worst tie population), paged by offset to exhaustion. The single
    // `namespace IN (...)` statement with the #1671 id tiebreak must
    // enumerate every row exactly once, in a deterministic order.
    let store = setup_memory_store();
    let tied = Utc::now();
    let mut expected: Vec<Uuid> = Vec::new();
    for ns in ["ns-a", "ns-b"] {
        for _ in 0..20 {
            let mut edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 0.5);
            edge.namespace = ns.to_string();
            edge.created_at = tied;
            edge.updated_at = tied;
            expected.push(edge.id.into());
            store.upsert_edge(edge).await.unwrap();
        }
    }

    let namespaces = vec!["ns-a".to_string(), "ns-b".to_string()];
    let sort = || {
        vec![SortOrder {
            field: EdgeSortField::CreatedAt,
            direction: SortDirection::Asc,
        }]
    };

    let mut seen: Vec<Uuid> = Vec::new();
    let mut offset: u64 = 0;
    loop {
        let page = store
            .query_edges_in_namespaces(
                &namespaces,
                EdgeFilter::default(),
                sort(),
                PageRequest { offset, limit: 7 },
            )
            .await
            .unwrap();
        assert_eq!(page.total, None);
        if page.items.is_empty() {
            break;
        }
        for e in &page.items {
            seen.push(e.id.into());
        }
        offset += 7;
    }

    assert_eq!(seen.len(), 40, "offset paging must enumerate every row");
    let distinct: std::collections::HashSet<Uuid> = seen.iter().copied().collect();
    assert_eq!(distinct.len(), 40, "no row may appear on two pages");
    let mut want: Vec<Uuid> = expected.clone();
    want.sort();
    let mut got: Vec<Uuid> = seen.clone();
    got.sort();
    assert_eq!(got, want, "enumerated set must equal the seeded set");

    // Determinism: the same page re-read returns the same rows in order.
    let page1 = store
        .query_edges_in_namespaces(
            &namespaces,
            EdgeFilter::default(),
            sort(),
            PageRequest {
                offset: 14,
                limit: 7,
            },
        )
        .await
        .unwrap();
    let page2 = store
        .query_edges_in_namespaces(
            &namespaces,
            EdgeFilter::default(),
            sort(),
            PageRequest {
                offset: 14,
                limit: 7,
            },
        )
        .await
        .unwrap();
    let ids1: Vec<Uuid> = page1.items.iter().map(|e| e.id.into()).collect();
    let ids2: Vec<Uuid> = page2.items.iter().map(|e| e.id.into()).collect();
    assert_eq!(ids1, ids2, "a page must be stable across identical reads");
}

#[tokio::test]
async fn test_upsert_and_get_edge() {
    let store = setup_memory_store();

    let src = Uuid::new_v4();
    let tgt = Uuid::new_v4();
    let now = Utc::now();
    let edge = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 0.8,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };
    let edge_id = edge.id;

    store.upsert_edge(edge).await.unwrap();

    let fetched = store.get_edge(edge_id).await.unwrap();
    assert!(fetched.is_some());
    let fetched = fetched.unwrap();
    assert_eq!(fetched.id, edge_id);
    assert_eq!(fetched.namespace, "default");
    assert_eq!(fetched.source_id, src);
    assert_eq!(fetched.target_id, tgt);
    assert_eq!(fetched.relation, EdgeRelation::Extends);
    assert!((fetched.weight - 0.8).abs() < 1e-9);
}

#[tokio::test]
async fn observed_upsert_distinguishes_replace_refusal_and_resurrection() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    insert_live_entity(&pool, target);
    let mut original = make_edge(source, target, EdgeRelation::Extends, 0.4);
    original.metadata = Some(serde_json::json!({"revision": 1}));
    let original_id = original.id;

    let created = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: original.clone(),
            resurrect: false,
        })
        .await
        .unwrap();
    let GuardedEdgeUpsertOutcome::Written(created) = created else {
        panic!("first natural-key write must succeed");
    };
    assert_eq!(created.disposition, EdgeUpsertDisposition::Created);
    assert!(created.previous.is_none());

    let mut replacement = make_edge(source, target, EdgeRelation::Extends, 0.9);
    replacement.metadata = Some(serde_json::json!({"revision": 2}));
    let updated = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: replacement.clone(),
            resurrect: false,
        })
        .await
        .unwrap();
    let GuardedEdgeUpsertOutcome::Written(updated) = updated else {
        panic!("live natural-key replacement must succeed");
    };
    assert_eq!(updated.disposition, EdgeUpsertDisposition::Updated);
    assert_eq!(updated.edge.id, original_id);
    assert_eq!(updated.edge.metadata, replacement.metadata);
    assert_eq!(updated.previous.unwrap().metadata, original.metadata);

    store
        .delete_edge(original_id, DeleteMode::Soft)
        .await
        .unwrap();
    let refused = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: replacement.clone(),
            resurrect: false,
        })
        .await
        .unwrap();
    let GuardedEdgeUpsertOutcome::Refused(EdgeUpsertRefusal::ResurrectionRequired {
        edge: tombstone,
    }) = refused
    else {
        panic!("implicit resurrection must be refused");
    };
    assert_eq!(tombstone.id, original_id);
    assert!(tombstone.deleted_at.is_some());

    replacement.weight = 0.7;
    replacement.metadata = Some(serde_json::json!({"revision": 3}));
    let resurrected = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: replacement.clone(),
            resurrect: true,
        })
        .await
        .unwrap();
    let GuardedEdgeUpsertOutcome::Written(resurrected) = resurrected else {
        panic!("explicit resurrection must succeed");
    };
    assert_eq!(resurrected.disposition, EdgeUpsertDisposition::Resurrected);
    assert_eq!(resurrected.edge.id, original_id);
    assert!(resurrected.edge.deleted_at.is_none());
    assert_eq!(resurrected.edge.metadata, replacement.metadata);
    assert!(resurrected.previous.unwrap().deleted_at.is_some());
}

#[tokio::test]
async fn observed_upsert_uses_canonical_symmetric_natural_key() {
    let (pool, store) = setup_memory_store_with_substrates();
    let left = Uuid::new_v4();
    let right = Uuid::new_v4();
    insert_live_entity(&pool, left);
    insert_live_entity(&pool, right);
    let first = make_edge(left, right, EdgeRelation::CompetesWith, 0.4);
    let first_id = first.id;
    let created = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: first,
            resurrect: false,
        })
        .await
        .unwrap();
    assert!(matches!(
        created,
        GuardedEdgeUpsertOutcome::Written(EdgeUpsertResult {
            disposition: EdgeUpsertDisposition::Created,
            ..
        })
    ));

    let reverse = make_edge(right, left, EdgeRelation::CompetesWith, 0.8);
    let updated = store
        .upsert_edge_guarded_observed(EdgeUpsertRequest {
            edge: reverse,
            resurrect: false,
        })
        .await
        .unwrap();
    let GuardedEdgeUpsertOutcome::Written(updated) = updated else {
        panic!("reverse symmetric edge must target the canonical row");
    };
    assert_eq!(updated.disposition, EdgeUpsertDisposition::Updated);
    assert_eq!(updated.edge.id, first_id);
    assert_eq!(updated.edge.weight, 0.8);
}

#[tokio::test]
async fn insert_edge_if_absent_preserves_the_natural_key_winner() {
    let store = setup_memory_store();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    let winner = make_edge(source, target, EdgeRelation::DependsOn, 1.0);
    let winner_id = winner.id;
    let mut loser = make_edge(source, target, EdgeRelation::DependsOn, 0.25);
    loser.metadata = Some(serde_json::json!({"kind": "loser"}));
    let loser_id = loser.id;

    assert!(store.insert_edge_if_absent(winner).await.unwrap());
    assert!(!store.insert_edge_if_absent(loser).await.unwrap());

    let persisted = store.get_edge(winner_id).await.unwrap().unwrap();
    assert!((persisted.weight - 1.0).abs() < f64::EPSILON);
    assert_eq!(persisted.metadata, None);
    assert!(store.get_edge(loser_id).await.unwrap().is_none());
}

#[tokio::test]
async fn observed_batch_upsert_reports_updated_preimage_and_created_rows() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    let other_target = Uuid::new_v4();
    for id in [source, target, other_target] {
        insert_live_entity(&pool, id);
    }
    let mut original = make_edge(source, target, EdgeRelation::Extends, 0.25);
    original.metadata = Some(serde_json::json!({"revision": 1, "old_only": true}));
    original.target_backend = Some("old-backend".into());
    let original_id = original.id;
    store.upsert_edge(original).await.unwrap();
    let original = store.get_edge(original_id).await.unwrap().unwrap();

    let mut replacement = make_edge(source, target, EdgeRelation::Extends, 0.75);
    replacement.metadata = Some(serde_json::json!({"revision": 2}));
    let replacement_id = replacement.id;
    let created = make_edge(source, other_target, EdgeRelation::Extends, 0.5);
    let created_id = created.id;
    let outcome = store
        .upsert_edges_guarded_observed(vec![
            EdgeUpsertRequest {
                edge: replacement.clone(),
                resurrect: false,
            },
            EdgeUpsertRequest {
                edge: created,
                resurrect: false,
            },
        ])
        .await
        .unwrap();

    assert!(outcome.refusal.is_none());
    assert_eq!(outcome.rows.len(), 2);
    let updated = &outcome.rows[0];
    assert_eq!(updated.disposition, EdgeUpsertDisposition::Updated);
    assert_eq!(updated.edge.id, original_id);
    assert_eq!(updated.edge.created_at, original.created_at);
    assert_eq!(updated.edge.weight, replacement.weight);
    assert_eq!(updated.edge.metadata, replacement.metadata);
    assert_eq!(updated.edge.target_backend, None);
    assert_eq!(
        serde_json::to_value(updated.previous.as_ref().unwrap()).unwrap(),
        serde_json::to_value(&original).unwrap()
    );
    assert_eq!(outcome.rows[1].disposition, EdgeUpsertDisposition::Created);
    assert_eq!(outcome.rows[1].edge.id, created_id);
    assert!(outcome.rows[1].previous.is_none());
    for row in &outcome.rows {
        let persisted = store.get_edge(row.edge.id).await.unwrap().unwrap();
        assert_eq!(
            serde_json::to_value(&persisted).unwrap(),
            serde_json::to_value(&row.edge).unwrap()
        );
    }
    assert!(store.get_edge(replacement_id).await.unwrap().is_none());
}

#[tokio::test]
async fn observed_batch_upsert_later_refusal_preserves_earlier_replacement() {
    for (route, is_file_backed, use_writer_task) in [
        ("memory_fallback", false, false),
        ("file_fallback", true, false),
        ("writer_task", true, true),
    ] {
        for resurrection_refusal in [false, true] {
            let directory = is_file_backed.then(|| tempfile::tempdir().unwrap());
            let pool = Arc::new(
                ConnectionPool::new(PoolConfig {
                    path: directory
                        .as_ref()
                        .map(|dir| dir.path().join("batch-refusal.db")),
                    write_queue_enabled: Some(use_writer_task),
                    write_routing_strict: use_writer_task,
                    ..PoolConfig::for_test()
                })
                .unwrap(),
            );
            {
                let writer = pool.writer().unwrap();
                apply_test_schema(writer.conn());
            }
            let store = SqlGraphStore::new_scoped(Arc::clone(&pool), is_file_backed, "default");
            assert_eq!(
                pool.writer_task_handle().unwrap().is_some(),
                use_writer_task,
                "{route}"
            );
            let source = Uuid::new_v4();
            let target = Uuid::new_v4();
            let refused_target = Uuid::new_v4();
            let other_target = Uuid::new_v4();
            for id in [source, target, refused_target, other_target] {
                insert_live_entity(&pool, id);
            }
            let mut original = make_edge(source, target, EdgeRelation::Extends, 0.25);
            original.metadata = Some(serde_json::json!({"revision": 1}));
            original.target_backend = Some("original-backend".into());
            let original_id = original.id;
            store.upsert_edge(original).await.unwrap();
            let original = store.get_edge(original_id).await.unwrap().unwrap();

            let tombstone = if resurrection_refusal {
                let mut edge = make_edge(target, refused_target, EdgeRelation::Extends, 0.5);
                edge.metadata = Some(serde_json::json!({"tombstone": true}));
                let id = edge.id;
                store.upsert_edge(edge).await.unwrap();
                store.delete_edge(id, DeleteMode::Soft).await.unwrap();
                store.get_edge_including_deleted(id).await.unwrap()
            } else {
                hard_delete_entity(&pool, refused_target);
                None
            };
            let mut replacement = make_edge(source, target, EdgeRelation::Extends, 0.75);
            replacement.metadata = Some(serde_json::json!({"revision": 2}));
            let refused = make_edge(target, refused_target, EdgeRelation::Extends, 0.9);
            let created = make_edge(source, other_target, EdgeRelation::Extends, 1.0);
            let incoming_ids = [replacement.id, refused.id, created.id];

            let acquisitions_before = pool.writer_acquisition_snapshot();
            let outcome = store
                .upsert_edges_guarded_observed(
                    [replacement, refused, created]
                        .into_iter()
                        .map(|edge| EdgeUpsertRequest {
                            edge,
                            resurrect: false,
                        })
                        .collect(),
                )
                .await
                .unwrap();
            let acquisitions_after = pool.writer_acquisition_snapshot();
            assert_eq!(
                (
                    acquisitions_after.pooled_acquisitions
                        - acquisitions_before.pooled_acquisitions,
                    acquisitions_after.standalone_acquisitions
                        - acquisitions_before.standalone_acquisitions,
                    acquisitions_after.writer_task_acquisitions
                        - acquisitions_before.writer_task_acquisitions,
                ),
                match route {
                    "memory_fallback" => (1, 0, 0),
                    "file_fallback" => (0, 1, 0),
                    "writer_task" => (0, 0, 1),
                    _ => unreachable!(),
                },
                "{route}: the refused batch must execute on the selected writer route"
            );

            assert!(outcome.rows.is_empty());
            let refusal = outcome.refusal.expect("later entry must refuse the batch");
            assert_eq!(refusal.entry_index, 1);
            match (refusal.reason, tombstone) {
                (EdgeUpsertRefusal::ResurrectionRequired { edge }, Some(tombstone)) => {
                    assert_eq!(
                        serde_json::to_value(&edge).unwrap(),
                        serde_json::to_value(&tombstone).unwrap()
                    );
                    let persisted = store
                        .get_edge_including_deleted(tombstone.id)
                        .await
                        .unwrap()
                        .unwrap();
                    assert_eq!(
                        serde_json::to_value(&persisted).unwrap(),
                        serde_json::to_value(&tombstone).unwrap()
                    );
                }
                (EdgeUpsertRefusal::MissingEndpoints(missing), None) => {
                    assert!(!missing.source);
                    assert!(missing.target);
                }
                other => panic!("unexpected later refusal: {other:?}"),
            }
            let persisted = store.get_edge(original_id).await.unwrap().unwrap();
            assert_eq!(
                serde_json::to_value(&persisted).unwrap(),
                serde_json::to_value(&original).unwrap(),
                "{route}: preflight must refuse before replacing an earlier live edge"
            );
            for id in incoming_ids {
                assert!(store
                    .get_edge_including_deleted(id)
                    .await
                    .unwrap()
                    .is_none());
            }
            let writer_join = pool.take_writer_task_join();
            drop(store);
            drop(pool);
            if let Some(writer_join) = writer_join {
                writer_join.await.unwrap();
            }
            drop(directory);
        }
    }
}

/// The base `PRIMARY KEY (namespace, id)` alone would let two namespaces
/// hold the same edge id; the list-cursor ledger (`graph_edges_seq`) and the
/// multi-namespace cursor merge in the runtime both key on `id` alone, so a
/// shared id would corrupt cursor ordering across namespaces. The
/// `idx_graph_edges_id_unique` index closes that gap at the write path
/// itself -- a second namespace cannot silently acquire an id already used
/// by another namespace's edge.
#[tokio::test]
async fn test_upsert_edge_rejects_duplicate_id_across_namespaces() {
    let store = setup_memory_store();

    let shared_id = Uuid::new_v4();
    let mut first = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
    first.id = shared_id.into();
    first.namespace = "ns-a".to_string();
    store.upsert_edge(first).await.unwrap();

    let mut second = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
    second.id = shared_id.into();
    second.namespace = "ns-b".to_string();
    let result = store.upsert_edge(second).await;

    assert!(
        result.is_err(),
        "a second namespace inserting an already-used edge id must fail, not silently \
         share the first namespace's list-cursor ledger row"
    );

    let fetched = store.get_edge(shared_id.into()).await.unwrap().unwrap();
    assert_eq!(
        fetched.namespace, "ns-a",
        "the rejected duplicate-id write must leave the original namespace's edge intact"
    );
}

#[tokio::test]
async fn test_delete_edge() {
    let store = setup_memory_store();

    let edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Contains, 1.0);
    let edge_id = edge.id;

    store.upsert_edge(edge).await.unwrap();
    assert!(store.get_edge(edge_id).await.unwrap().is_some());

    let deleted = store.delete_edge(edge_id, DeleteMode::Hard).await.unwrap();
    assert!(deleted);

    assert!(store.get_edge(edge_id).await.unwrap().is_none());

    let deleted_again = store.delete_edge(edge_id, DeleteMode::Hard).await.unwrap();
    assert!(!deleted_again);
}

#[tokio::test]
async fn test_count_edges() {
    let store = setup_memory_store();

    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 0);

    for _ in 0..5 {
        store
            .upsert_edge(make_edge(
                Uuid::new_v4(),
                Uuid::new_v4(),
                EdgeRelation::DependsOn,
                1.0,
            ))
            .await
            .unwrap();
    }

    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 5);
}

/// A soft delete leaves incident edges in place, so an edge can outlive its endpoint. Every
/// reader that walks the graph hydrates endpoints and cannot reach such an edge, so the
/// counts must not report it either.
///
/// The fixture is built so that one plausible wrong implementation fails each arm: an
/// implementation that ignores endpoints fails the drop; one that excludes an endpoint
/// merely for being absent from `entities`/`notes` fails the ghost arm; one that checks
/// only `source_id` fails the inbound arm; one that checks only entities fails the note
/// arm. The live pair is the control, and it is read in the same pass that produces each
/// absence.
#[tokio::test]
async fn edge_counts_skip_edges_whose_endpoint_is_tombstoned() {
    let (pool, store) = setup_memory_store_with_substrates();

    let (out_src, out_dst) = (Uuid::new_v4(), Uuid::new_v4());
    let (in_src, in_dst) = (Uuid::new_v4(), Uuid::new_v4());
    let (live_src, live_dst) = (Uuid::new_v4(), Uuid::new_v4());
    for id in [out_src, out_dst, in_src, in_dst, live_src, live_dst] {
        insert_live_entity(&pool, id);
    }
    let (note_live, note_doomed) = (Uuid::new_v4(), Uuid::new_v4());
    insert_live_entity(&pool, note_live);
    insert_note(&pool, note_doomed, false);
    // Endpoints this database holds in neither table: a different substrate's ids. Absence
    // is not a tombstone, so these stay counted.
    let (ghost_a, ghost_b) = (Uuid::new_v4(), Uuid::new_v4());

    for edge in [
        make_edge(out_src, out_dst, EdgeRelation::Contains, 1.0),
        make_edge(in_src, in_dst, EdgeRelation::Contains, 1.0),
        make_edge(live_src, live_dst, EdgeRelation::Contains, 1.0),
        make_edge(note_doomed, note_live, EdgeRelation::Annotates, 1.0),
        make_edge(ghost_a, ghost_b, EdgeRelation::DependsOn, 1.0),
    ] {
        store.upsert_edge(edge).await.unwrap();
    }

    let namespaces = vec!["default".to_string()];

    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 5);
    let pre: HashMap<_, _> = store
        .count_edges_by_relation()
        .await
        .unwrap()
        .into_iter()
        .collect();
    assert_eq!(
        pre.get(&EdgeRelation::Contains),
        Some(&3),
        "pre-state: all three entity-to-entity edges are counted"
    );

    // Outbound endpoint, then inbound endpoint, then a note endpoint. After each, the live
    // pair and the ghost pair must still be there.
    soft_delete_entity(&pool, out_src);
    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 4);

    soft_delete_entity(&pool, in_dst);
    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 3);

    soft_delete_note(&pool, note_doomed);
    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 2);

    let by_relation: HashMap<_, _> = store
        .count_edges_by_relation()
        .await
        .unwrap()
        .into_iter()
        .collect();
    assert_eq!(
        by_relation.get(&EdgeRelation::Contains),
        Some(&1),
        "only the live entity pair survives; got {by_relation:?}"
    );
    assert_eq!(
        by_relation.get(&EdgeRelation::DependsOn),
        Some(&1),
        "an endpoint absent from both tables is not a tombstone; got {by_relation:?}"
    );
    assert_eq!(
        by_relation.get(&EdgeRelation::Annotates),
        None,
        "the note endpoint's tombstone removes its edge; got {by_relation:?}"
    );

    // The namespace-scoped variants answer the same question and must agree.
    assert_eq!(
        store
            .count_edges_in_namespaces(&namespaces, EdgeFilter::default())
            .await
            .unwrap(),
        2
    );
    let scoped: HashMap<_, _> = store
        .count_edges_by_relation_in_namespaces(&namespaces)
        .await
        .unwrap()
        .into_iter()
        .collect();
    assert_eq!(scoped, by_relation, "scoped and unscoped counts must agree");
}

#[tokio::test]
async fn batched_namespace_edge_counts_exceed_sqlite_variable_limit() {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    apply_test_schema(pool.writer().unwrap().conn());
    let store_a = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "stats-a");
    let store_b = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "stats-b");

    let mut edge_a = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
    edge_a.namespace = "stats-a".to_string();
    let mut deleted_edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Enables, 1.0);
    deleted_edge.namespace = "stats-a".to_string();
    let deleted_edge_id = deleted_edge.id;
    let mut edge_b = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::DependsOn, 1.0);
    edge_b.namespace = "stats-b".to_string();

    store_a.upsert_edge(edge_a).await.unwrap();
    store_a.upsert_edge(deleted_edge).await.unwrap();
    store_b.upsert_edge(edge_b).await.unwrap();
    assert!(store_a
        .delete_edge(deleted_edge_id, DeleteMode::Soft)
        .await
        .unwrap());

    let per_namespace_total = store_a.count_edges(EdgeFilter::default()).await.unwrap()
        + store_b.count_edges(EdgeFilter::default()).await.unwrap();
    let mut per_namespace_relations: HashMap<EdgeRelation, u64> = store_a
        .count_edges_by_relation()
        .await
        .unwrap()
        .into_iter()
        .collect();
    for (relation, count) in store_b.count_edges_by_relation().await.unwrap() {
        *per_namespace_relations.entry(relation).or_insert(0) += count;
    }
    pool.writer()
        .unwrap()
        .conn()
        .set_limit(rusqlite::limits::Limit::SQLITE_LIMIT_VARIABLE_NUMBER, 999)
        .unwrap();
    let mut namespaces = vec!["stats-a".to_string(), "stats-b".to_string()];
    namespaces.extend((0..999).map(|i| format!("empty-{i}")));
    assert_eq!(namespaces.len(), 1_001);

    assert_eq!(
        store_a
            .count_edges_in_namespaces(&namespaces, EdgeFilter::default())
            .await
            .unwrap(),
        per_namespace_total
    );
    assert_eq!(per_namespace_total, 2);
    assert_eq!(
        store_a
            .count_edges_by_relation_in_namespaces(&namespaces)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashMap<_, _>>(),
        per_namespace_relations
    );
}

/// #2089: `query_edges_in_namespaces` used to bind one
/// `?N` SQL variable per visible namespace, so a caller visible in ~998+
/// namespaces would blow past `SQLITE_LIMIT_VARIABLE_NUMBER` (999 by
/// default) before any filter parameter was even added — well within a
/// realistic broad `[actor]` visibility grant. The fix binds the namespace
/// set as a single JSON-array parameter matched via `json_each`, so this
/// must still enumerate correctly (paged, exactly, in order) even with the
/// SQLite variable limit pinned to 999 and 1,001 distinct namespaces in the
/// visibility set — mirroring
/// `batched_namespace_edge_counts_exceed_sqlite_variable_limit` above, but
/// through the paginated listing path rather than the count-only aggregate.
#[tokio::test]
async fn query_edges_in_namespaces_offset_paging_exceeds_sqlite_variable_limit() {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    apply_test_schema(pool.writer().unwrap().conn());
    let store_a = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "list-a");
    let store_b = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "list-b");

    let tied = Utc::now();
    let mut expected: HashSet<Uuid> = HashSet::new();
    for (store, ns) in [(&store_a, "list-a"), (&store_b, "list-b")] {
        for _ in 0..6 {
            let mut edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 0.5);
            edge.namespace = ns.to_string();
            edge.created_at = tied;
            edge.updated_at = tied;
            expected.insert(edge.id.into());
            store.upsert_edge(edge).await.unwrap();
        }
    }
    assert_eq!(expected.len(), 12);

    pool.writer()
        .unwrap()
        .conn()
        .set_limit(rusqlite::limits::Limit::SQLITE_LIMIT_VARIABLE_NUMBER, 999)
        .unwrap();

    let mut namespaces = vec!["list-a".to_string(), "list-b".to_string()];
    namespaces.extend((0..999).map(|i| format!("empty-{i}")));
    assert_eq!(namespaces.len(), 1_001);

    let sort = vec![SortOrder {
        field: EdgeSortField::CreatedAt,
        direction: SortDirection::Asc,
    }];
    let mut seen: Vec<Uuid> = Vec::new();
    let mut offset: u64 = 0;
    loop {
        let page = store_a
            .query_edges_in_namespaces(
                &namespaces,
                EdgeFilter::default(),
                sort.clone(),
                PageRequest { offset, limit: 5 },
            )
            .await
            .unwrap();
        assert_eq!(page.total, None);
        if page.items.is_empty() {
            break;
        }
        for e in &page.items {
            seen.push(e.id.into());
        }
        offset += 5;
    }

    assert_eq!(seen.len(), 12, "offset paging must enumerate every row");
    let distinct: HashSet<Uuid> = seen.iter().copied().collect();
    assert_eq!(distinct.len(), 12, "no row may appear on two pages");
    assert_eq!(
        distinct, expected,
        "enumerated set must equal the seeded set"
    );
}

#[tokio::test]
async fn duplicate_namespace_across_chunk_boundary_is_not_double_counted() {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    apply_test_schema(pool.writer().unwrap().conn());
    let store_a = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "stats-a");

    let mut edge_a1 = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
    edge_a1.namespace = "stats-a".to_string();
    let mut edge_a2 = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Enables, 1.0);
    edge_a2.namespace = "stats-a".to_string();
    store_a.upsert_edge(edge_a1).await.unwrap();
    store_a.upsert_edge(edge_a2).await.unwrap();

    let per_namespace_total = store_a.count_edges(EdgeFilter::default()).await.unwrap();
    assert_eq!(per_namespace_total, 2);
    let per_namespace_relations: HashMap<EdgeRelation, u64> = store_a
        .count_edges_by_relation()
        .await
        .unwrap()
        .into_iter()
        .collect();

    // 501 unique namespaces ("stats-a" + 500 empties), then "stats-a" repeats
    // once more past index 500 — the repeat lands in the second 500-entry
    // chunk, so a dedup bug double-counts "stats-a"'s rows.
    let mut namespaces = vec!["stats-a".to_string()];
    namespaces.extend((0..500).map(|i| format!("empty-{i}")));
    assert_eq!(namespaces.len(), 501);
    namespaces.push("stats-a".to_string());
    assert_eq!(namespaces.len(), 502);

    assert_eq!(
        store_a
            .count_edges_in_namespaces(&namespaces, EdgeFilter::default())
            .await
            .unwrap(),
        per_namespace_total
    );
    assert_eq!(
        store_a
            .count_edges_by_relation_in_namespaces(&namespaces)
            .await
            .unwrap()
            .into_iter()
            .collect::<HashMap<_, _>>(),
        per_namespace_relations
    );
}

// `#[serial(neighbor_select_count)]`: shares the key with the tests that
// assert on the process-wide `NEIGHBOR_SELECT_COUNT` so a concurrent
// `neighbors()` call from this test can't corrupt their count.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_outbound() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();

    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(a, c, EdgeRelation::DependsOn, 0.7))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(d, a, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();

    let query = NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: None,
        min_weight: None,
    };

    let hits = store.neighbors(a, query).await.unwrap();
    assert_eq!(hits.len(), 2);

    let neighbor_ids: Vec<Uuid> = hits.iter().map(|h| h.node_id).collect();
    assert!(neighbor_ids.contains(&b));
    assert!(neighbor_ids.contains(&c));
    assert!(!neighbor_ids.contains(&d));
}

/// Regression guard (ADR-089 context-verb review): a
/// `limit` narrower than the neighbor set must keep the highest-weight edges,
/// not an arbitrary SQLite row-order subset. Before the fix, `neighbors()`
/// applied `LIMIT` with no `ORDER BY`, so a low-weight neighbor could win over
/// a high-weight one purely by insertion order.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_limit_keeps_highest_weight_not_insertion_order() {
    let store = setup_memory_store();

    let centre = Uuid::new_v4();
    let low = Uuid::new_v4();
    let mid = Uuid::new_v4();
    let high = Uuid::new_v4();

    // Insert lowest-weight edge FIRST so insertion order and weight order
    // disagree — if LIMIT ignores ORDER BY, the low-weight edge can win.
    store
        .upsert_edge(make_edge(centre, low, EdgeRelation::Extends, 0.1))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, mid, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, high, EdgeRelation::Extends, 0.9))
        .await
        .unwrap();

    let query = NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: Some(2),
        min_weight: None,
    };

    let hits = store.neighbors(centre, query).await.unwrap();
    assert_eq!(hits.len(), 2, "limit=2 must return exactly 2 neighbors");

    let neighbor_ids: HashSet<Uuid> = hits.iter().map(|h| h.node_id).collect();
    assert!(
        neighbor_ids.contains(&high),
        "highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        neighbor_ids.contains(&mid),
        "second-highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        !neighbor_ids.contains(&low),
        "lowest-weight neighbor must be the one dropped by limit"
    );

    // Weight-descending, node_id-ascending-tiebreak ordering must also hold
    // for the returned rows themselves (ADR-089's neighbor-ordering contract).
    assert!((hits[0].weight - 0.9).abs() < 1e-9);
    assert!((hits[1].weight - 0.5).abs() < 1e-9);
}

/// Correctness parity (ADR-089 context-verb optimization): `neighbors_both_directions`
/// must return exactly the union of a separate `Out` call and a separate `In`
/// call — same node/edge/relation/weight set, same direction tag per hit, and
/// the same global weight-descending/node_id-ascending order — while doing it
/// in one storage query instead of two. Outgoing and incoming weights
/// interleave (0.9, 0.6, 0.3 outgoing vs 0.8, 0.4 incoming) so the assertion
/// proves global post-union ordering, not per-branch order.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_both_directions_matches_two_separate_calls_and_order() {
    let store = setup_memory_store();

    let centre = Uuid::new_v4();
    let out_hi = Uuid::new_v4();
    let out_mid = Uuid::new_v4();
    let out_lo = Uuid::new_v4();
    let in_hi = Uuid::new_v4();
    let in_mid = Uuid::new_v4();

    store
        .upsert_edge(make_edge(centre, out_hi, EdgeRelation::Extends, 0.9))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(in_hi, centre, EdgeRelation::Extends, 0.8))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, out_mid, EdgeRelation::Extends, 0.6))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(in_mid, centre, EdgeRelation::Extends, 0.4))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, out_lo, EdgeRelation::Extends, 0.3))
        .await
        .unwrap();

    let directed = store
        .neighbors_both_directions(
            centre,
            NeighborQuery {
                direction: Direction::Both,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();

    // Global weight-descending order across BOTH directions, not per-branch:
    // 0.9(out) > 0.8(in) > 0.6(out) > 0.4(in) > 0.3(out).
    let got: Vec<(Uuid, f64, Direction)> = directed
        .iter()
        .map(|d| (d.hit.node_id, d.hit.weight, d.direction.clone()))
        .collect();
    assert_eq!(
        got,
        vec![
            (out_hi, 0.9, Direction::Out),
            (in_hi, 0.8, Direction::In),
            (out_mid, 0.6, Direction::Out),
            (in_mid, 0.4, Direction::In),
            (out_lo, 0.3, Direction::Out),
        ],
        "neighbors_both_directions must interleave by global weight DESC, tagging each hit's real direction"
    );

    // Parity: the same result set must be reconstructable from two separate
    // direction-scoped `neighbors()` calls (the pre-optimization behavior).
    let out_hits = store
        .neighbors(
            centre,
            NeighborQuery {
                direction: Direction::Out,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();
    let in_hits = store
        .neighbors(
            centre,
            NeighborQuery {
                direction: Direction::In,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();

    let directed_out: HashSet<(Uuid, Uuid)> = directed
        .iter()
        .filter(|d| d.direction == Direction::Out)
        .map(|d| (d.hit.node_id, d.hit.edge_id))
        .collect();
    let directed_in: HashSet<(Uuid, Uuid)> = directed
        .iter()
        .filter(|d| d.direction == Direction::In)
        .map(|d| (d.hit.node_id, d.hit.edge_id))
        .collect();
    let plain_out: HashSet<(Uuid, Uuid)> =
        out_hits.iter().map(|h| (h.node_id, h.edge_id)).collect();
    let plain_in: HashSet<(Uuid, Uuid)> = in_hits.iter().map(|h| (h.node_id, h.edge_id)).collect();
    assert_eq!(
        directed_out, plain_out,
        "outgoing subset must match a separate Out call"
    );
    assert_eq!(
        directed_in, plain_in,
        "incoming subset must match a separate In call"
    );
}

/// Tight `fanout`/`limit` parity: with a narrowing limit, `neighbors_both_directions`
/// must keep the same top-K set (by global weight) that the pre-optimization
/// two-call-then-merge-then-truncate approach kept.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_both_directions_limit_keeps_global_top_k() {
    let store = setup_memory_store();

    let centre = Uuid::new_v4();
    let out_hi = Uuid::new_v4();
    let in_hi = Uuid::new_v4();
    let out_lo = Uuid::new_v4();
    let in_lo = Uuid::new_v4();

    store
        .upsert_edge(make_edge(centre, out_hi, EdgeRelation::Extends, 0.95))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(in_hi, centre, EdgeRelation::Extends, 0.85))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, out_lo, EdgeRelation::Extends, 0.2))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(in_lo, centre, EdgeRelation::Extends, 0.1))
        .await
        .unwrap();

    let directed = store
        .neighbors_both_directions(
            centre,
            NeighborQuery {
                direction: Direction::Both,
                relations: None,
                limit: Some(2),
                min_weight: None,
            },
        )
        .await
        .unwrap();

    assert_eq!(
        directed.len(),
        2,
        "limit=2 must cap at 2 across both directions"
    );
    let ids: Vec<Uuid> = directed.iter().map(|d| d.hit.node_id).collect();
    assert_eq!(
        ids,
        vec![out_hi, in_hi],
        "the two globally-highest-weight neighbors must survive, in weight-descending order"
    );
    assert_eq!(directed[0].direction, Direction::Out);
    assert_eq!(directed[1].direction, Direction::In);
}

/// Count-falsifiable A/B proof (ADR-089 context-verb optimization): a
/// `direction="both"` neighbor fetch issues exactly ONE storage `neighbors`
/// SELECT via `neighbors_both_directions`, versus TWO if a caller instead
/// issued separate `Out` and `In` calls (the pre-fix `context` handler
/// pattern this replaces — see `fetch_directed_neighbors` in
/// `khive-pack-kg/src/handlers/context.rs`). For N expanded nodes across V
/// visible namespaces this is the `2*N*V -> 1*N*V` reduction described in the
/// verification verdict.
///
/// `#[serial(neighbor_select_count)]`: `NEIGHBOR_SELECT_COUNT` is a process-wide
/// counter — any other test issuing a neighbor SELECT while this one resets
/// and checks it would corrupt the count under the default parallel test runner.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_both_directions_halves_storage_query_count() {
    let store = setup_memory_store();
    let centre = Uuid::new_v4();
    let neighbor = Uuid::new_v4();
    store
        .upsert_edge(make_edge(centre, neighbor, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();

    let query = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: None,
        min_weight: None,
    };

    reset_neighbor_select_count();
    store
        .neighbors_both_directions(centre, query.clone())
        .await
        .unwrap();
    assert_eq!(
        neighbor_select_count(),
        1,
        "one neighbors_both_directions call must issue exactly 1 storage SELECT"
    );

    reset_neighbor_select_count();
    let out_query = NeighborQuery {
        direction: Direction::Out,
        ..query.clone()
    };
    let in_query = NeighborQuery {
        direction: Direction::In,
        ..query
    };
    store.neighbors(centre, out_query).await.unwrap();
    store.neighbors(centre, in_query).await.unwrap();
    assert_eq!(
        neighbor_select_count(),
        2,
        "the old pattern of two direction-scoped neighbors() calls issues 2 SELECTs — \
         exactly the query count neighbors_both_directions halves"
    );
}

/// Reciprocal equal-weight determinism: a
/// node with an Out edge to a neighbor and an In edge from the SAME neighbor
/// at the SAME weight ties on `(weight, node_id)` — the pre-fix `ORDER BY`.
/// Under a tight `limit`, the surviving row must be picked by a deterministic
/// tie-break (the `out` row wins), not by whichever row SQLite happens to
/// return first. Repeated to demonstrate the result is stable across calls.
///
/// `#[serial(neighbor_select_count)]`: shares the key with
/// `test_neighbors_both_directions_halves_storage_query_count` so this test's
/// repeated `neighbors_both_directions` calls can't corrupt that test's count
/// of the process-wide `NEIGHBOR_SELECT_COUNT`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_both_directions_reciprocal_equal_weight_limit_is_deterministic() {
    let store = setup_memory_store();

    let centre = Uuid::new_v4();
    let other = Uuid::new_v4();

    store
        .upsert_edge(make_edge(centre, other, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(other, centre, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();

    let query = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: Some(1),
        min_weight: None,
    };

    for _ in 0..5 {
        let directed = store
            .neighbors_both_directions(centre, query.clone())
            .await
            .unwrap();
        assert_eq!(
            directed.len(),
            1,
            "limit=1 must return exactly one hit even with a reciprocal equal-weight pair"
        );
        assert_eq!(directed[0].hit.node_id, other);
        assert_eq!(
            directed[0].direction,
            Direction::Out,
            "the out-direction row must win the weight/node_id tie deterministically"
        );
    }
}

/// Forward-regression contract for `neighbors_both_directions`'s pre-`LIMIT`
/// tie-break order: `weight DESC, node_id ASC, CASE dir WHEN 'out' THEN 0 ELSE
/// 1 END ASC, edge_id ASC`. Four edges tie on `(weight, node_id)` — two Out
/// and two In, all to/from the same neighbor — with edge_ids chosen so every
/// Out edge_id sorts below every In edge_id, and inserted in an order that
/// does not match the expected result order. That means neither the
/// direction rank nor the `edge_id` tie-break can be satisfied by accidental
/// insertion-order preservation: only the explicit `ORDER BY` can produce the
/// sequence asserted below. A future change that reverses the direction rank
/// (`in` before `out`) or flips `edge_id ASC` to `DESC` yields a different
/// sequence than the one asserted here for either component, so this test
/// fails the moment either tie-break regresses.
///
/// The two edges sharing a direction use different relations (`Extends` /
/// `DependsOn`) purely to satisfy the storage layer's
/// `(namespace, source_id, target_id, relation)` uniqueness constraint — a
/// second edge with the same relation between the same ordered pair upserts
/// onto the first rather than creating a second row. The relation choice is
/// not part of the ordering contract under test.
///
/// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn test_neighbors_both_directions_direction_then_edge_id_is_a_forward_contract() {
    let store = setup_memory_store();

    let centre = Uuid::new_v4();
    let other = Uuid::new_v4();

    let out_lo: Uuid = "00000000-0000-0000-0000-000000000001".parse().unwrap();
    let out_hi: Uuid = "00000000-0000-0000-0000-000000000002".parse().unwrap();
    let in_lo: Uuid = "00000000-0000-0000-0000-000000000003".parse().unwrap();
    let in_hi: Uuid = "00000000-0000-0000-0000-000000000004".parse().unwrap();

    // Inserted out of both the weight/direction tie-break order and the
    // edge_id order, so nothing about the asserted sequence below can pass
    // by coincidentally preserving insertion order.
    for (id, source, target, relation) in [
        (in_lo, other, centre, EdgeRelation::Extends),
        (out_hi, centre, other, EdgeRelation::DependsOn),
        (in_hi, other, centre, EdgeRelation::DependsOn),
        (out_lo, centre, other, EdgeRelation::Extends),
    ] {
        let now = Utc::now();
        store
            .upsert_edge(Edge {
                id: id.into(),
                namespace: "default".to_string(),
                source_id: source,
                target_id: target,
                relation,
                weight: 0.5,
                created_at: now,
                updated_at: now,
                deleted_at: None,
                metadata: None,
                target_backend: None,
            })
            .await
            .unwrap();
    }

    let query = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: None,
        min_weight: None,
    };

    let unlimited = store
        .neighbors_both_directions(centre, query)
        .await
        .unwrap();

    let observed: Vec<(Direction, Uuid)> = unlimited
        .iter()
        .map(|h| (h.direction.clone(), h.hit.edge_id))
        .collect();
    assert_eq!(
        observed,
        vec![
            (Direction::Out, out_lo),
            (Direction::Out, out_hi),
            (Direction::In, in_lo),
            (Direction::In, in_hi),
        ],
        "tied (weight, node_id) rows must order Out-before-In, then edge_id ASC within each direction"
    );

    // The storage layer applies `LIMIT` to this pre-sorted order (the runtime
    // caller only re-sorts after `LIMIT` has already truncated), so survivors
    // under a narrowing limit must be a prefix of the sequence above, not an
    // arbitrary 2-of-4 subset.
    let limited_query = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: Some(2),
        min_weight: None,
    };
    let limited = store
        .neighbors_both_directions(centre, limited_query)
        .await
        .unwrap();
    let limited_observed: Vec<(Direction, Uuid)> = limited
        .iter()
        .map(|h| (h.direction.clone(), h.hit.edge_id))
        .collect();
    assert_eq!(
        limited_observed,
        vec![(Direction::Out, out_lo), (Direction::Out, out_hi)],
        "limit=2 must keep exactly the first two rows of the deterministic order, both Out"
    );
}

#[tokio::test]
async fn test_traverse_depth_2() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();

    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(b, c, EdgeRelation::Extends, 2.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(c, d, EdgeRelation::Extends, 3.0))
        .await
        .unwrap();

    let request = TraversalRequest {
        roots: vec![a],
        options: TraversalOptions::new(2).with_direction(Direction::Out),
        include_roots: true,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();
    assert_eq!(paths.len(), 1);

    let path = &paths[0];
    let node_ids: Vec<Uuid> = path.nodes.iter().map(|n| n.node_id).collect();
    assert!(node_ids.contains(&a));
    assert!(node_ids.contains(&b));
    assert!(node_ids.contains(&c));
    assert!(!node_ids.contains(&d));
}

/// Diamond graph: A→B, A→C, B→D, C→D.
/// D is reachable via two paths at depth 2.  After the fix it must appear
/// exactly once in the result.
#[tokio::test]
async fn test_traverse_dedups_multipath_node() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();

    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(a, c, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(b, d, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(c, d, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let request = TraversalRequest {
        roots: vec![a],
        options: TraversalOptions::new(3).with_direction(Direction::Out),
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();
    assert_eq!(paths.len(), 1);
    let nodes = &paths[0].nodes;

    // D must appear exactly once despite being reachable via both B and C.
    let d_count = nodes.iter().filter(|n| n.node_id == d).count();
    assert_eq!(d_count, 1, "D must appear exactly once (dedup multi-path)");

    // B and C must each appear once as well.
    assert_eq!(nodes.iter().filter(|n| n.node_id == b).count(), 1);
    assert_eq!(nodes.iter().filter(|n| n.node_id == c).count(), 1);
}

/// First-visit BFS keeps the shallowest occurrence of a node, and the
/// `via_edge` recorded for it is the one from that first-visited path.
///
/// Graph: A→B (depth 1), A→C (depth 1), B→D (depth 2), C→D (depth 2).
/// D appears at depth 2 via B or C. Whichever same-depth frontier path the
/// backend enumerates first is the keeper. The test
/// asserts that D has exactly one entry with a non-None `via_edge` — we
/// do NOT assert *which* edge wins because SQLite row order within the
/// same depth level is non-deterministic, but we DO assert stability:
/// running twice gives the same count.
#[tokio::test]
async fn test_traverse_preserves_first_path_metadata() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();

    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(a, c, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(b, d, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(c, d, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let make_request = || TraversalRequest {
        roots: vec![a],
        options: TraversalOptions::new(3).with_direction(Direction::Out),
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths1 = store.traverse(make_request()).await.unwrap();
    let paths2 = store.traverse(make_request()).await.unwrap();

    // Both runs must return the same total node count (dedup is stable).
    let count1 = paths1[0].nodes.len();
    let count2 = paths2[0].nodes.len();
    assert_eq!(
        count1, count2,
        "traverse result count must be stable across calls"
    );

    // D must appear exactly once and carry a via_edge (it was not a root).
    let d_nodes: Vec<_> = paths1[0].nodes.iter().filter(|n| n.node_id == d).collect();
    assert_eq!(d_nodes.len(), 1, "D deduped to one entry");
    assert!(
        d_nodes[0].via_edge.is_some(),
        "kept entry must have a via_edge"
    );
    assert_eq!(d_nodes[0].depth, 2, "D lives at depth 2");
}

/// Multi-root traversal: two independent chains A→B→C and D→E→F.
/// Each root must produce its own GraphPath with the correct node set.
#[tokio::test]
async fn test_traverse_multi_root_independent_chains() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();
    let e = Uuid::new_v4();
    let f = Uuid::new_v4();

    for (src, tgt) in [(a, b), (b, c), (d, e), (e, f)] {
        store
            .upsert_edge(make_edge(src, tgt, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
    }

    let request = TraversalRequest {
        roots: vec![a, d],
        options: TraversalOptions::new(2).with_direction(Direction::Out),
        include_roots: true,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();
    assert_eq!(paths.len(), 2, "one GraphPath per root");

    // Locate paths by root_id.
    let path_a = paths
        .iter()
        .find(|p| p.root_id == a)
        .expect("path for root A");
    let path_d = paths
        .iter()
        .find(|p| p.root_id == d)
        .expect("path for root D");

    let ids_a: HashSet<Uuid> = path_a.nodes.iter().map(|n| n.node_id).collect();
    assert!(ids_a.contains(&a), "root A in its own path");
    assert!(ids_a.contains(&b), "depth-1 B in A's path");
    assert!(ids_a.contains(&c), "depth-2 C in A's path");
    assert!(!ids_a.contains(&d), "root D must not appear in A's path");

    let ids_d: HashSet<Uuid> = path_d.nodes.iter().map(|n| n.node_id).collect();
    assert!(ids_d.contains(&d), "root D in its own path");
    assert!(ids_d.contains(&e), "depth-1 E in D's path");
    assert!(ids_d.contains(&f), "depth-2 F in D's path");
    assert!(!ids_d.contains(&a), "root A must not appear in D's path");
}

/// Multi-root with a shared neighbor: A→C and B→C.  C must appear in BOTH
/// A's path and B's path (per-root isolation, not global dedup).
#[tokio::test]
async fn test_traverse_multi_root_shared_neighbor_appears_in_both() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4(); // shared neighbor

    for (src, tgt) in [(a, c), (b, c)] {
        store
            .upsert_edge(make_edge(src, tgt, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
    }

    let request = TraversalRequest {
        roots: vec![a, b],
        options: TraversalOptions::new(1).with_direction(Direction::Out),
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();
    assert_eq!(paths.len(), 2, "one GraphPath per root");

    for path in &paths {
        let node_ids: HashSet<Uuid> = path.nodes.iter().map(|n| n.node_id).collect();
        assert!(
            node_ids.contains(&c),
            "shared node C must appear in each root's path; root={:?}",
            path.root_id
        );
    }
}

/// A 15-node binary tree at max_depth=3 must retain every shallowest node
/// exactly once under level-synchronous BFS.
#[tokio::test]
async fn test_traverse_binary_tree_result_count() {
    let store = setup_memory_store();

    // Build a complete binary tree of depth 3: root + 2 + 4 + 8 = 15 nodes.
    let nodes: Vec<Uuid> = (0..15).map(|_| Uuid::new_v4()).collect();
    for i in 0..7usize {
        let left = 2 * i + 1;
        let right = 2 * i + 2;
        store
            .upsert_edge(make_edge(nodes[i], nodes[left], EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
        store
            .upsert_edge(make_edge(
                nodes[i],
                nodes[right],
                EdgeRelation::Extends,
                1.0,
            ))
            .await
            .unwrap();
    }

    let request = TraversalRequest {
        roots: vec![nodes[0]],
        options: TraversalOptions::new(3).with_direction(Direction::Out),
        include_roots: true,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();
    assert_eq!(paths.len(), 1);
    // root + depth-1 (2) + depth-2 (4) + depth-3 (8) = 15
    assert_eq!(
        paths[0].nodes.len(),
        15,
        "binary tree depth-3 must yield exactly 15 nodes"
    );
    // Every depth-3 node must have a via_edge.
    for node in paths[0].nodes.iter().filter(|n| n.depth == 3) {
        assert!(
            node.via_edge.is_some(),
            "depth-3 nodes must carry a via_edge"
        );
    }
}

/// ADR-091 Amendment 4: each adjacency statement retains backend-scoped
/// attribution, but the span is released with that statement instead of
/// covering the whole traversal. A commit made while the first cursor is
/// live becomes visible to the next frontier statement, proving there is no
/// operation-wide deferred snapshot.
#[tokio::test]
#[serial(tx_registry)]
async fn graph_traverse_read_span_scoped_to_secondary_backend_visible_only_in_its_own_view() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("traverse_secondary_origin.db");
    let pool_cfg = PoolConfig {
        path: Some(path.clone()),
        ..PoolConfig::for_test()
    };
    let pool = Arc::new(ConnectionPool::new(pool_cfg).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }
    let secondary_identity = match pool.origin() {
        khive_storage::tx_registry::TxOrigin::Database(id) => id,
        other => panic!("expected a file-backed pool to mint a Database origin, got {other:?}"),
    };
    let other_identity = khive_storage::tx_registry::DbIdentity::new(
        pool.canonical_path()
            .unwrap()
            .with_file_name("unrelated-backend.db"),
    );

    let store = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "default");
    let writer_store = SqlGraphStore::new_scoped(Arc::clone(&pool), false, "default");
    let root = Uuid::new_v4();
    let child = Uuid::new_v4();
    let grandchild = Uuid::new_v4();
    store
        .upsert_edge(make_edge(root, child, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    let (reached, proceed) = traverse_snapshot_seam::install(root);

    let request = TraversalRequest {
        roots: vec![root],
        options: TraversalOptions::new(2).with_direction(Direction::Out),
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let traverse_handle = tokio::spawn(async move { store.traverse(request).await });
    tokio::task::spawn_blocking(move || reached.recv().unwrap())
        .await
        .unwrap();

    let secondary_view =
        khive_storage::tx_registry::TxOriginFilter::Secondary(secondary_identity.clone());
    let main_view = khive_storage::tx_registry::TxOriginFilter::Main(other_identity);

    // Ask about the traversal span by label rather than through `oldest_for`.
    // Writes on this same pool open their own `writer_task_tx` spans, and the
    // writer task replies to its caller *before* dropping that handle, so an
    // acknowledged write can still hold an open span on this backend. Such a
    // span is unrelated to what this test asserts, but it is older than the
    // traversal's, so `oldest_for` returns it and the assertion reads a
    // different span's lifetime as the traversal's.
    assert!(
        khive_storage::tx_registry::any_open_labeled(&secondary_view, "graph_traverse_read"),
        "statement-scoped traversal span must be visible while its seam is held"
    );
    assert!(
        !khive_storage::tx_registry::any_open_labeled(&main_view, "graph_traverse_read"),
        "a secondary-origin graph_traverse_read span must never be visible through a different backend's Main view"
    );

    writer_store
        .upsert_edge(make_edge(child, grandchild, EdgeRelation::Extends, 1.0))
        .await
        .expect("WAL writer must commit while the first traversal cursor is live");
    proceed.send(()).unwrap();

    let result = traverse_handle.await.unwrap().unwrap();
    let grandchild_node = result[0]
        .nodes
        .iter()
        .find(|node| node.node_id == grandchild)
        .expect("the next frontier statement must see the concurrent commit");
    assert_eq!(grandchild_node.depth, 2);
    assert!(
        !khive_storage::tx_registry::any_open_labeled(&secondary_view, "graph_traverse_read"),
        "the statement-scoped traversal span must be gone when its bounded query returns"
    );
}

#[tokio::test]
async fn test_metadata_roundtrip() {
    let store = setup_memory_store();

    let src = Uuid::new_v4();
    let tgt = Uuid::new_v4();
    let meta = serde_json::json!({"note": "important link", "confidence": 0.95});
    let now = Utc::now();
    let edge = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Implements,
        weight: 0.9,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: Some(meta.clone()),
        target_backend: None,
    };
    let edge_id = edge.id;

    store.upsert_edge(edge).await.unwrap();

    let fetched = store.get_edge(edge_id).await.unwrap().unwrap();
    assert_eq!(
        fetched.metadata.as_ref(),
        Some(&meta),
        "metadata must survive a write/read roundtrip via get_edge"
    );

    // Also verify via query_edges.
    let page = store
        .query_edges(EdgeFilter::default(), vec![], PageRequest::default())
        .await
        .unwrap();
    let from_query = page
        .items
        .iter()
        .find(|e| e.id == edge_id)
        .expect("edge must appear in query_edges result");
    assert_eq!(
        from_query.metadata.as_ref(),
        Some(&meta),
        "metadata must survive a write/read roundtrip via query_edges"
    );
}

#[tokio::test]
async fn test_upsert_edges_batch() {
    let store = setup_memory_store();

    let edges: Vec<Edge> = (0..10)
        .map(|i| {
            make_edge(
                Uuid::new_v4(),
                Uuid::new_v4(),
                EdgeRelation::Implements,
                i as f64,
            )
        })
        .collect();

    let summary = store.upsert_edges(edges).await.unwrap();
    assert_eq!(summary.attempted, 10);
    assert_eq!(summary.affected, 10);
    assert_eq!(summary.failed, 0);

    assert_eq!(store.count_edges(EdgeFilter::default()).await.unwrap(), 10);
}

// ---- Deduplication test ---------

#[tokio::test]
async fn graph_duplicate_edges_ignored() {
    let store = setup_memory_store();

    let src = Uuid::new_v4();
    let tgt = Uuid::new_v4();

    // Two edges with the same (source_id, target_id, relation) triple but different IDs.
    let now = Utc::now();
    let edge1 = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 1.0,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };
    let edge2 = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 0.5,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };

    store.upsert_edge(edge1).await.unwrap();
    store.upsert_edge(edge2).await.unwrap();

    assert_eq!(
        store.count_edges(EdgeFilter::default()).await.unwrap(),
        1,
        "duplicate (source, target, relation) triple must be ignored; only one edge must exist"
    );
}

// F053 (CRIT): natural-key conflict must DO UPDATE (refresh weight/metadata), not DO NOTHING.
// The second upsert must overwrite weight=0.5; current code keeps weight=1.0.
#[tokio::test]
async fn graph_duplicate_edges_refresh_existing_row() {
    let store = setup_memory_store();
    let src = Uuid::new_v4();
    let tgt = Uuid::new_v4();

    let now = Utc::now();
    let edge1 = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 1.0,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };
    let edge2 = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 0.5,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };

    store.upsert_edge(edge1).await.unwrap();
    store.upsert_edge(edge2).await.unwrap();

    let edges = store
        .query_edges(EdgeFilter::default(), vec![], PageRequest::default())
        .await
        .unwrap();
    assert_eq!(
        edges.items.len(),
        1,
        "duplicate natural key must collapse to one row"
    );
    assert!(
        (edges.items[0].weight - 0.5).abs() < 0.001,
        "F053: natural-key conflict must DO UPDATE (weight=0.5 from second upsert); \
             current DO NOTHING keeps stale weight={}",
        edges.items[0].weight
    );
}

// Regression test: symmetric edges stored via upsert_edge must
// always have source_id < target_id (lexicographic on UUID bytes).
#[tokio::test]
async fn upsert_edge_canonicalizes_symmetric_relation() {
    let store = setup_memory_store();

    // Construct two UUIDs where larger > smaller lexicographically.
    let smaller = Uuid::from_bytes([0x00; 16]);
    let larger = Uuid::from_bytes([0xff; 16]);
    assert!(
        larger > smaller,
        "test setup: larger must sort after smaller"
    );

    // Insert with source > target — the invariant-violating order.
    let edge = make_edge(larger, smaller, EdgeRelation::CompetesWith, 1.0);
    let edge_id = edge.id;
    store.upsert_edge(edge).await.unwrap();

    let stored = store.get_edge(edge_id).await.unwrap().unwrap();
    assert_eq!(
        stored.source_id, smaller,
        "#476: CompetesWith edge must be stored with source_id < target_id"
    );
    assert_eq!(
        stored.target_id, larger,
        "#476: CompetesWith edge must be stored with target_id > source_id"
    );
}

#[tokio::test]
async fn upsert_edges_batch_canonicalizes_symmetric_relation() {
    let store = setup_memory_store();

    let smaller = Uuid::from_bytes([0x11; 16]);
    let larger = Uuid::from_bytes([0xee; 16]);

    // ComposedWith is the other symmetric relation — insert reversed.
    let edge = make_edge(larger, smaller, EdgeRelation::ComposedWith, 0.9);
    let edge_id = edge.id;
    store.upsert_edges(vec![edge]).await.unwrap();

    let stored = store.get_edge(edge_id).await.unwrap().unwrap();
    assert_eq!(
        stored.source_id, smaller,
        "#476: ComposedWith edge must be stored with source_id < target_id (batch path)"
    );
    assert_eq!(
        stored.target_id, larger,
        "#476: ComposedWith edge must be stored with target_id > source_id (batch path)"
    );
}

#[tokio::test]
async fn upsert_edge_non_symmetric_relation_preserves_direction() {
    let store = setup_memory_store();

    // DependsOn is not symmetric — direction must NOT be swapped.
    let src = Uuid::from_bytes([0xff; 16]);
    let tgt = Uuid::from_bytes([0x00; 16]);
    let edge = make_edge(src, tgt, EdgeRelation::DependsOn, 1.0);
    let edge_id = edge.id;
    store.upsert_edge(edge).await.unwrap();

    let stored = store.get_edge(edge_id).await.unwrap().unwrap();
    assert_eq!(
        stored.source_id, src,
        "non-symmetric edge direction must be preserved"
    );
    assert_eq!(
        stored.target_id, tgt,
        "non-symmetric edge direction must be preserved"
    );
}

// ADR-007 PR-A1 regression: upsert must accept edges whose namespace differs
// from the store's construction-time namespace.  Before the fix, `upsert_edge`
// rejected this with InvalidInput; after, it stores the edge with whatever
// namespace the edge carries and get_edge finds it by UUID alone.
#[tokio::test]
async fn upsert_edge_cross_namespace_accepted() {
    // Store is constructed with "local" as its default multi-record query namespace.
    let store = setup_memory_store(); // uses "default" as construction namespace

    let src = Uuid::new_v4();
    let tgt = Uuid::new_v4();
    let now = Utc::now();

    // Edge carries "lambda:leo" — different from store's "default".
    let edge = Edge {
        id: Uuid::new_v4().into(),
        namespace: "lambda:leo".to_string(),
        source_id: src,
        target_id: tgt,
        relation: EdgeRelation::Extends,
        weight: 0.9,
        created_at: now,
        updated_at: now,
        deleted_at: None,
        metadata: None,
        target_backend: None,
    };
    let edge_id = edge.id;

    // Must not error (V1 violation removed).
    store.upsert_edge(edge).await.unwrap();

    // get_edge by UUID must find it regardless of namespace.
    let stored = store.get_edge(edge_id).await.unwrap();
    assert!(
        stored.is_some(),
        "cross-namespace edge must be retrievable by UUID"
    );
    let stored = stored.unwrap();
    assert_eq!(
        stored.namespace, "lambda:leo",
        "namespace column must be preserved as stored"
    );
}

// ADR-007 PR-A1 regression: namespace column is preserved on the record even
// after upsert — not silently overwritten with the store's construction namespace.
#[tokio::test]
async fn upsert_edge_namespace_stored_on_record() {
    let store = setup_memory_store();

    let edge = make_edge(
        Uuid::new_v4(),
        Uuid::new_v4(),
        EdgeRelation::Implements,
        1.0,
    );
    let edge_id = edge.id;
    let ns = edge.namespace.clone();

    store.upsert_edge(edge).await.unwrap();

    let stored = store.get_edge(edge_id).await.unwrap().unwrap();
    assert_eq!(
        stored.namespace, ns,
        "namespace column must survive the write/read roundtrip"
    );
}

// ---- batch_neighbors parity tests (HIGH bug regression guard) ----

/// Build a star graph: centre node with `out_count` outgoing edges and
/// `in_count` incoming edges.  Returns (centre, out_targets, in_sources,
/// out_edge_ids, in_edge_ids).
async fn build_star(
    store: &SqlGraphStore,
    out_count: usize,
    in_count: usize,
) -> (Uuid, Vec<Uuid>, Vec<Uuid>) {
    let centre = Uuid::new_v4();
    let mut out_nodes = Vec::new();
    let mut in_nodes = Vec::new();
    for _ in 0..out_count {
        let tgt = Uuid::new_v4();
        store
            .upsert_edge(make_edge(centre, tgt, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
        out_nodes.push(tgt);
    }
    for _ in 0..in_count {
        let src = Uuid::new_v4();
        store
            .upsert_edge(make_edge(src, centre, EdgeRelation::Extends, 0.8))
            .await
            .unwrap();
        in_nodes.push(src);
    }
    (centre, out_nodes, in_nodes)
}

fn neighbour_set(hits: &[(Uuid, NeighborHit)]) -> HashSet<Uuid> {
    hits.iter().map(|(_, h)| h.node_id).collect()
}

fn single_neighbour_set(hits: &[NeighborHit]) -> HashSet<Uuid> {
    hits.iter().map(|h| h.node_id).collect()
}

fn neighbor_edge_multiset(hits: &[(Uuid, NeighborHit)]) -> HashMap<(Uuid, Uuid), usize> {
    let mut grouped = HashMap::new();
    for (origin, hit) in hits {
        *grouped.entry((*origin, hit.edge_id)).or_default() += 1;
    }
    grouped
}

#[tokio::test]
async fn batch_neighbors_keeps_exact_rows_grouped_by_requested_node() {
    let store = setup_memory_store();
    let root_a = Uuid::new_v4();
    let root_b = Uuid::new_v4();
    let root_without_edges = Uuid::new_v4();
    let joined = make_edge(root_a, root_b, EdgeRelation::Extends, 0.9);
    let self_loop = make_edge(root_b, root_b, EdgeRelation::DependsOn, 0.8);
    let joined_id = Uuid::from(joined.id);
    let self_loop_id = Uuid::from(self_loop.id);
    for edge in [joined, self_loop] {
        store.upsert_edge(edge).await.unwrap();
    }

    let sources = [root_a, root_b, root_without_edges];
    let outgoing_hits = store
        .batch_neighbors(
            &sources,
            NeighborQuery {
                direction: Direction::Out,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();
    assert_eq!(
        neighbor_edge_multiset(&outgoing_hits),
        HashMap::from([((root_a, joined_id), 1), ((root_b, self_loop_id), 1),])
    );

    let incoming_hits = store
        .batch_neighbors(
            &sources,
            NeighborQuery {
                direction: Direction::In,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();
    assert_eq!(
        neighbor_edge_multiset(&incoming_hits),
        HashMap::from([((root_b, joined_id), 1), ((root_b, self_loop_id), 1),])
    );

    let both_hits = store
        .batch_neighbors(
            &sources,
            NeighborQuery {
                direction: Direction::Both,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();
    assert_eq!(
        neighbor_edge_multiset(&both_hits),
        HashMap::from([
            ((root_a, joined_id), 1),
            ((root_b, joined_id), 1),
            ((root_b, self_loop_id), 2),
        ])
    );
    assert!(
        both_hits
            .iter()
            .all(|(origin, _)| *origin != root_without_edges),
        "the requested zero-edge root must not acquire another root's rows"
    );
}

#[tokio::test]
async fn batch_neighbors_preserves_duplicate_requested_sources() {
    let store = setup_memory_store();
    let root = Uuid::new_v4();
    let target = Uuid::new_v4();
    let edge = make_edge(root, target, EdgeRelation::Extends, 1.0);
    let edge_id = Uuid::from(edge.id);
    store.upsert_edge(edge).await.unwrap();

    let hits = store
        .batch_neighbors(
            &[root, root],
            NeighborQuery {
                direction: Direction::Out,
                relations: None,
                limit: None,
                min_weight: None,
            },
        )
        .await
        .unwrap();

    assert_eq!(
        neighbor_edge_multiset(&hits),
        HashMap::from([((root, edge_id), 2)]),
        "the optimized override must match the trait default's per-occurrence output"
    );
}

/// PARITY REGRESSION GUARD — the critical bug (HIGH).
/// For Direction::Both + limit=Some(1), batch_neighbors must return AT MOST
/// `limit` hits per source, not up to 2× (one per direction).
/// Before the fix, Both ran Out and In separately and concatenated, so a node
/// with ≥1 outgoing AND ≥1 incoming edge would yield 2 hits when limit=1.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_both_limit_matches_single_source_neighbors() {
    let store = setup_memory_store();
    // 2 outgoing, 2 incoming from centre
    let (centre, out_nodes, in_nodes) = build_star(&store, 2, 2).await;

    let q_both_limit1 = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: Some(1),
        min_weight: None,
    };

    // single-source neighbors() — ground truth
    let single_hits = store
        .neighbors(centre, q_both_limit1.clone())
        .await
        .unwrap();
    // batch_neighbors with the same query
    let batch_hits = store
        .batch_neighbors(&[centre], q_both_limit1.clone())
        .await
        .unwrap();

    assert_eq!(
        batch_hits.len(),
        single_hits.len(),
        "batch_neighbors Both+limit=1 must return same count as neighbors() \
         (was 2× before fix)"
    );

    // Sanity: single-source also returns 1 when limit=1
    assert_eq!(single_hits.len(), 1, "neighbors() must respect limit=1");

    // Ensure the returned node is one of the actual neighbours.
    let all_neighbours: HashSet<Uuid> = out_nodes.iter().chain(in_nodes.iter()).copied().collect();
    let batch_node_ids: HashSet<Uuid> = batch_hits.iter().map(|(_, h)| h.node_id).collect();
    for nid in &batch_node_ids {
        assert!(
            all_neighbours.contains(nid),
            "batch result must be a real neighbour of centre"
        );
    }
}

/// PARITY: set equality for Out direction, with and without limit.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_out_parity_with_neighbors() {
    let store = setup_memory_store();
    let (centre, out_nodes, _) = build_star(&store, 3, 2).await;

    let q_out = NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: None,
        min_weight: None,
    };

    let single: HashSet<Uuid> =
        single_neighbour_set(&store.neighbors(centre, q_out.clone()).await.unwrap());
    let batch: HashSet<Uuid> = neighbour_set(
        &store
            .batch_neighbors(&[centre], q_out.clone())
            .await
            .unwrap(),
    );
    assert_eq!(batch, single, "Out: batch must equal single-source set");

    let expected: HashSet<Uuid> = out_nodes.iter().copied().collect();
    assert_eq!(
        batch, expected,
        "Out: must return exactly the out-neighbours"
    );
}

/// PARITY: set equality for In direction.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_in_parity_with_neighbors() {
    let store = setup_memory_store();
    let (centre, _, in_nodes) = build_star(&store, 2, 3).await;

    let q_in = NeighborQuery {
        direction: Direction::In,
        relations: None,
        limit: None,
        min_weight: None,
    };

    let single: HashSet<Uuid> =
        single_neighbour_set(&store.neighbors(centre, q_in.clone()).await.unwrap());
    let batch: HashSet<Uuid> = neighbour_set(
        &store
            .batch_neighbors(&[centre], q_in.clone())
            .await
            .unwrap(),
    );
    assert_eq!(batch, single, "In: batch must equal single-source set");

    let expected: HashSet<Uuid> = in_nodes.iter().copied().collect();
    assert_eq!(batch, expected, "In: must return exactly the in-neighbours");
}

/// PARITY: set equality for Both direction, no limit.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_both_parity_no_limit() {
    let store = setup_memory_store();
    let (centre, out_nodes, in_nodes) = build_star(&store, 2, 3).await;

    let q_both = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: None,
        min_weight: None,
    };

    let single: HashSet<Uuid> =
        single_neighbour_set(&store.neighbors(centre, q_both.clone()).await.unwrap());
    let batch: HashSet<Uuid> = neighbour_set(
        &store
            .batch_neighbors(&[centre], q_both.clone())
            .await
            .unwrap(),
    );
    assert_eq!(batch, single, "Both: batch must equal single-source set");

    let expected: HashSet<Uuid> = out_nodes.iter().chain(in_nodes.iter()).copied().collect();
    assert_eq!(batch, expected, "Both: must return all neighbours");
}

/// PARITY: relations filter applies correctly across directions.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_relations_filter_parity() {
    let store = setup_memory_store();
    let centre = Uuid::new_v4();
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    // outgoing: one Extends, one DependsOn
    store
        .upsert_edge(make_edge(centre, a, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, b, EdgeRelation::DependsOn, 1.0))
        .await
        .unwrap();

    let q = NeighborQuery {
        direction: Direction::Out,
        relations: Some(vec![EdgeRelation::Extends]),
        limit: None,
        min_weight: None,
    };

    let single: HashSet<Uuid> =
        single_neighbour_set(&store.neighbors(centre, q.clone()).await.unwrap());
    let batch: HashSet<Uuid> = neighbour_set(&store.batch_neighbors(&[centre], q).await.unwrap());

    assert_eq!(batch, single, "relations filter: batch must match single");
    assert!(
        batch.contains(&a),
        "filtered result must include Extends target"
    );
    assert!(
        !batch.contains(&b),
        "filtered result must exclude DependsOn target"
    );
}

/// PARITY: min_weight filter applies correctly.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_min_weight_filter_parity() {
    let store = setup_memory_store();
    let centre = Uuid::new_v4();
    let heavy = Uuid::new_v4();
    let light = Uuid::new_v4();
    store
        .upsert_edge(make_edge(centre, heavy, EdgeRelation::Extends, 0.9))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre, light, EdgeRelation::Extends, 0.3))
        .await
        .unwrap();

    let q = NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: None,
        min_weight: Some(0.5),
    };

    let single: HashSet<Uuid> =
        single_neighbour_set(&store.neighbors(centre, q.clone()).await.unwrap());
    let batch: HashSet<Uuid> = neighbour_set(&store.batch_neighbors(&[centre], q).await.unwrap());

    assert_eq!(batch, single, "min_weight filter: batch must match single");
    assert!(
        batch.contains(&heavy),
        "must include edge above weight threshold"
    );
    assert!(
        !batch.contains(&light),
        "must exclude edge below weight threshold"
    );
}

/// Regression guard (issue #589): a `limit` narrower than an origin's full
/// neighbor set must keep that origin's highest-weight edges, not an
/// arbitrary SQLite row-order subset. Before the fix, the `ROW_NUMBER()
/// OVER (PARTITION BY origin_id)` window had no `ORDER BY`, so a low-weight
/// neighbor could win over a high-weight one purely by insertion order —
/// and because this is per-origin, one origin's row order could disagree
/// with another's, so the test uses two origins with the low/mid/high
/// insertion order reversed between them.
#[tokio::test]
async fn batch_neighbors_limit_keeps_highest_weight_per_origin_not_insertion_order() {
    let store = setup_memory_store();

    let centre_a = Uuid::new_v4();
    let a_low = Uuid::new_v4();
    let a_mid = Uuid::new_v4();
    let a_high = Uuid::new_v4();

    let centre_b = Uuid::new_v4();
    let b_low = Uuid::new_v4();
    let b_mid = Uuid::new_v4();
    let b_high = Uuid::new_v4();

    // centre_a: insert lowest-weight edge FIRST (insertion order disagrees
    // with weight order).
    store
        .upsert_edge(make_edge(centre_a, a_low, EdgeRelation::Extends, 0.1))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre_a, a_mid, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre_a, a_high, EdgeRelation::Extends, 0.9))
        .await
        .unwrap();

    // centre_b: insert highest-weight edge FIRST — the reverse of centre_a —
    // so a single global insertion-order artifact cannot accidentally pass
    // both origins.
    store
        .upsert_edge(make_edge(centre_b, b_high, EdgeRelation::Extends, 0.9))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre_b, b_mid, EdgeRelation::Extends, 0.5))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(centre_b, b_low, EdgeRelation::Extends, 0.1))
        .await
        .unwrap();

    let query = NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: Some(2),
        min_weight: None,
    };

    let hits = store
        .batch_neighbors(&[centre_a, centre_b], query)
        .await
        .unwrap();

    let a_hits: Vec<&NeighborHit> = hits
        .iter()
        .filter(|(origin, _)| *origin == centre_a)
        .map(|(_, h)| h)
        .collect();
    let b_hits: Vec<&NeighborHit> = hits
        .iter()
        .filter(|(origin, _)| *origin == centre_b)
        .map(|(_, h)| h)
        .collect();

    assert_eq!(
        a_hits.len(),
        2,
        "centre_a: per-origin limit=2 must return exactly 2 neighbors"
    );
    assert_eq!(
        b_hits.len(),
        2,
        "centre_b: per-origin limit=2 must return exactly 2 neighbors"
    );

    let a_ids: HashSet<Uuid> = a_hits.iter().map(|h| h.node_id).collect();
    assert!(
        a_ids.contains(&a_high),
        "centre_a: highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        a_ids.contains(&a_mid),
        "centre_a: second-highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        !a_ids.contains(&a_low),
        "centre_a: lowest-weight neighbor must be the one dropped by limit"
    );

    let b_ids: HashSet<Uuid> = b_hits.iter().map(|h| h.node_id).collect();
    assert!(
        b_ids.contains(&b_high),
        "centre_b: highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        b_ids.contains(&b_mid),
        "centre_b: second-highest-weight neighbor must survive a narrowing limit"
    );
    assert!(
        !b_ids.contains(&b_low),
        "centre_b: lowest-weight neighbor must be the one dropped by limit"
    );
}

// ---- get_edges direct SQLite tests ----

/// get_edges must return all requested edges regardless of request order.
#[tokio::test]
async fn get_edges_order_independent() {
    let store = setup_memory_store();

    let edges: Vec<Edge> = (0..5)
        .map(|i| {
            make_edge(
                Uuid::new_v4(),
                Uuid::new_v4(),
                EdgeRelation::Extends,
                i as f64,
            )
        })
        .collect();
    let ids: Vec<LinkId> = edges.iter().map(|e| e.id).collect();
    for e in edges {
        store.upsert_edge(e).await.unwrap();
    }

    // Request in reversed order
    let mut reversed = ids.clone();
    reversed.reverse();

    let result = store.get_edges(&reversed).await.unwrap();
    let result_ids: HashSet<LinkId> = result.iter().map(|e| e.id).collect();
    let expected_ids: HashSet<LinkId> = ids.iter().copied().collect();
    assert_eq!(
        result_ids, expected_ids,
        "get_edges must return all edges regardless of request order"
    );
}

/// Soft-deleted or nonexistent IDs are silently omitted.
#[tokio::test]
async fn get_edges_omits_deleted_and_missing() {
    let store = setup_memory_store();

    let live = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
    let soft = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::DependsOn, 0.5);
    let ghost_id = LinkId::from(Uuid::new_v4()); // never inserted

    let live_id = live.id;
    let soft_id = soft.id;

    store.upsert_edge(live).await.unwrap();
    store.upsert_edge(soft).await.unwrap();
    // Soft-delete the second edge
    store.delete_edge(soft_id, DeleteMode::Soft).await.unwrap();

    let result = store
        .get_edges(&[live_id, soft_id, ghost_id])
        .await
        .unwrap();

    assert_eq!(result.len(), 1, "only the live edge must be returned");
    assert_eq!(result[0].id, live_id, "returned edge must be the live one");
}

/// get_edges with more than 900 IDs (chunk boundary) returns all live edges.
#[tokio::test]
async fn get_edges_chunk_boundary() {
    let store = setup_memory_store();

    let count = 950usize;
    let edges: Vec<Edge> = (0..count)
        .map(|_| make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0))
        .collect();
    let ids: Vec<LinkId> = edges.iter().map(|e| e.id).collect();
    store.upsert_edges(edges).await.unwrap();

    let result = store.get_edges(&ids).await.unwrap();
    assert_eq!(
        result.len(),
        count,
        "get_edges must return all {count} edges across the chunk boundary"
    );
}

/// batch_neighbors Direction::Both chunk-boundary test.
///
/// Source IDs are bound once as a JSON array, so Direction::Both does not
/// duplicate one SQL parameter per source for each UNION arm. This test crosses
/// the 880-source chunk boundary and verifies that grouping remains exact.
///
/// Correctness: for a random sample of sources, batch result must equal the
/// per-source neighbors() result.
// `#[serial(neighbor_select_count)]`: see note on `test_neighbors_outbound`.
#[tokio::test]
#[serial(neighbor_select_count)]
async fn batch_neighbors_both_chunk_boundary() {
    let store = setup_memory_store();

    // Create 900 source nodes, each with one outgoing and one incoming edge.
    let source_count = 900usize;
    let mut sources: Vec<Uuid> = Vec::with_capacity(source_count);
    for _ in 0..source_count {
        let centre = Uuid::new_v4();
        let out_tgt = Uuid::new_v4();
        let in_src = Uuid::new_v4();
        store
            .upsert_edge(make_edge(centre, out_tgt, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
        store
            .upsert_edge(make_edge(in_src, centre, EdgeRelation::Extends, 0.8))
            .await
            .unwrap();
        sources.push(centre);
    }

    let q_both = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: None,
        min_weight: None,
    };

    // Must not error (would panic with "too many SQL variables" pre-fix).
    let batch_hits = store
        .batch_neighbors(&sources, q_both.clone())
        .await
        .unwrap();

    // Each source has exactly 2 neighbours (one out, one in).
    assert_eq!(
        batch_hits.len(),
        source_count * 2,
        "Both chunk-boundary: must return 2 hits per source (1 out + 1 in)"
    );

    // Spot-check: first, middle, and last source match per-source neighbors().
    for &idx in &[0, source_count / 2, source_count - 1] {
        let src = sources[idx];
        let single: HashSet<Uuid> = store
            .neighbors(src, q_both.clone())
            .await
            .unwrap()
            .into_iter()
            .map(|h| h.node_id)
            .collect();
        let from_batch: HashSet<Uuid> = batch_hits
            .iter()
            .filter(|(origin, _)| *origin == src)
            .map(|(_, h)| h.node_id)
            .collect();
        assert_eq!(
            from_batch, single,
            "spot-check source {idx}: batch result must match neighbors()"
        );
    }

    // Also verify with limit=1: each source gets at most 1 hit total.
    let q_limit = NeighborQuery {
        direction: Direction::Both,
        relations: None,
        limit: Some(1),
        min_weight: None,
    };
    let limited_hits = store.batch_neighbors(&sources, q_limit).await.unwrap();
    assert_eq!(
        limited_hits.len(),
        source_count,
        "Both chunk-boundary with limit=1: must return exactly 1 hit per source"
    );
}

// ---- per-root limit regression ----

/// Regression guard that `limit` is applied independently to each root.
///
/// Graph: A→B and C→D (two independent chains).  With `limit=1` and
/// `include_roots=false`, EVERY root must receive its own capped result —
/// not just the lexicographically-first root_id.
///
/// A single global cap would let only one root receive a path.
#[tokio::test]
async fn test_traverse_per_root_limit_capped_independently() {
    let store = setup_memory_store();

    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();

    // Two independent one-hop chains: A→B and C→D.
    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(c, d, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let request = TraversalRequest {
        roots: vec![a, c],
        options: TraversalOptions {
            max_depth: 2,
            direction: Direction::Out,
            relations: None,
            min_weight: None,
            limit: Some(1),
        },
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let paths = store.traverse(request).await.unwrap();

    // Both roots must produce a path — global SQL LIMIT would drop one.
    assert_eq!(
        paths.len(),
        2,
        "both roots must produce a path even with limit=1 \
         (per-root cap, not global cap)"
    );

    // Each root gets exactly one node.
    for path in &paths {
        assert_eq!(
            path.nodes.len(),
            1,
            "root {:?}: limit=1 must cap to exactly one non-root node",
            path.root_id
        );
    }

    // Correct child per root.
    let path_a = paths.iter().find(|p| p.root_id == a).expect("path for A");
    let path_c = paths.iter().find(|p| p.root_id == c).expect("path for C");
    assert_eq!(path_a.nodes[0].node_id, b, "root A must reach child B");
    assert_eq!(path_c.nodes[0].node_id, d, "root C must reach child D");
}

/// #1444: a dense cyclic component behind the first hop must not be expanded
/// when `limit=1`. The response and measured work are both one row.
#[tokio::test]
async fn traverse_limit_one_bounds_dense_cycle_work() {
    let store = setup_memory_store();
    let root = Uuid::from_u128(1);
    let children = (2_u128..18).map(Uuid::from_u128).collect::<Vec<_>>();
    let mut edges = children
        .iter()
        .map(|child| make_edge(root, *child, EdgeRelation::Extends, 1.0))
        .collect::<Vec<_>>();
    for source in &children {
        for target in &children {
            if source != target {
                edges.push(make_edge(*source, *target, EdgeRelation::VariantOf, 1.0));
            }
        }
    }
    store.upsert_edges(edges).await.unwrap();

    let ctx = khive_storage::usage::UsageContext::new();
    let paths = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: vec![root],
                options: TraversalOptions {
                    max_depth: MAX_TRAVERSAL_DEPTH,
                    direction: Direction::Out,
                    relations: None,
                    min_weight: None,
                    limit: Some(1),
                },
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await
    .unwrap();

    assert_eq!(paths.len(), 1);
    assert_eq!(paths[0].nodes.len(), 1);
    assert_eq!(paths[0].nodes[0].depth, 1);
    assert!(children.contains(&paths[0].nodes[0].node_id));
    assert_eq!(ctx.snapshot()["graph_hops"], 1);
    assert_eq!(ctx.snapshot()["db_round_trips"], 1);
}

#[tokio::test]
async fn traverse_omitted_limit_uses_finite_public_default_during_execution() {
    let store = setup_memory_store();
    let root = Uuid::from_u128(50);
    let edges = (1_u128..=101)
        .map(|offset| {
            make_edge(
                root,
                Uuid::from_u128(1_000 + offset),
                EdgeRelation::Extends,
                1.0,
            )
        })
        .collect();
    store.upsert_edges(edges).await.unwrap();

    let ctx = khive_storage::usage::UsageContext::new();
    let paths = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: vec![root],
                options: TraversalOptions {
                    max_depth: 1,
                    direction: Direction::Out,
                    relations: None,
                    min_weight: None,
                    limit: None,
                },
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await
    .unwrap();

    assert_eq!(
        paths[0].nodes.len(),
        khive_storage::DEFAULT_TRAVERSAL_LIMIT as usize
    );
    assert_eq!(
        ctx.snapshot()["graph_hops"],
        khive_storage::DEFAULT_TRAVERSAL_LIMIT as u64,
        "the default result cap must stop cursor consumption, not truncate afterward"
    );
}

/// Cyclic rows count against work before visited de-duplication. A root with
/// one self-loop per relation must read those bounded duplicates before the
/// first new child, but it retains only that child.
#[tokio::test]
async fn traverse_limit_one_bounds_duplicate_cycle_rows() {
    let store = setup_memory_store();
    let root = Uuid::from_u128(100);
    let child = Uuid::from_u128(101);
    let mut edges = EdgeRelation::ALL
        .iter()
        .map(|relation| make_edge(root, root, *relation, 1.0))
        .collect::<Vec<_>>();
    edges.push(make_edge(root, child, EdgeRelation::VariantOf, 1.0));
    store.upsert_edges(edges).await.unwrap();

    let ctx = khive_storage::usage::UsageContext::new();
    let paths = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: vec![root],
                options: TraversalOptions {
                    max_depth: MAX_TRAVERSAL_DEPTH,
                    direction: Direction::Out,
                    relations: None,
                    min_weight: None,
                    limit: Some(1),
                },
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await
    .unwrap();

    assert_eq!(paths[0].nodes.len(), 1);
    assert_eq!(paths[0].nodes[0].node_id, child);
    assert_eq!(paths[0].nodes[0].depth, 1);
    assert_eq!(
        ctx.snapshot()["graph_hops"],
        (EdgeRelation::ALL.len() + 1) as u64,
        "self-loop rows are bounded work even though the visited set rejects them"
    );
}

/// A result quota filled at depth one must never be displaced by a node from
/// a deeper frontier, regardless of same-depth tie order.
#[tokio::test]
async fn traverse_limit_keeps_shallowest_breadth_first_nodes() {
    let store = setup_memory_store();
    let root = Uuid::from_u128(200);
    let shallow_a = Uuid::from_u128(201);
    let shallow_b = Uuid::from_u128(202);
    let deep = Uuid::from_u128(199);
    for (source, target) in [(root, shallow_a), (root, shallow_b), (shallow_a, deep)] {
        store
            .upsert_edge(make_edge(source, target, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
    }

    let paths = store
        .traverse(TraversalRequest {
            roots: vec![root],
            options: TraversalOptions {
                max_depth: 2,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(2),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: Default::default(),
        })
        .await
        .unwrap();

    assert_eq!(paths[0].nodes.len(), 2);
    assert!(paths[0].nodes.iter().all(|node| node.depth == 1));
    assert!(!paths[0].nodes.iter().any(|node| node.node_id == deep));
}

/// The work ceiling admits exactly the configured rows and rejects the first
/// over-budget probe without returning partial paths. Usage still reports that
/// sentinel row because it came off SQLite's cursor before enforcement.
#[tokio::test]
#[serial(tx_registry)]
async fn traverse_work_budget_boundary_and_error_are_deterministic() {
    let (_dir, store, origin_view) = setup_file_store_with_origin_view();
    let root = Uuid::from_u128(300);
    for target in (301_u128..305).map(Uuid::from_u128) {
        store
            .upsert_edge(make_edge(root, target, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
    }

    let success = store
        .traverse(TraversalRequest {
            roots: vec![root],
            options: TraversalOptions {
                max_depth: 1,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(3),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: TraversalExecutionBudget::new(3, std::time::Duration::from_secs(5)),
        })
        .await
        .unwrap();
    assert_eq!(
        success[0].nodes.len(),
        3,
        "the exact work boundary succeeds"
    );

    let ctx = khive_storage::usage::UsageContext::new();
    let error = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: vec![root],
                options: TraversalOptions {
                    max_depth: 1,
                    direction: Direction::Out,
                    relations: None,
                    min_weight: None,
                    limit: Some(4),
                },
                include_roots: false,
                include_properties: false,
                execution_budget: TraversalExecutionBudget::new(
                    3,
                    std::time::Duration::from_secs(5),
                ),
            })
            .await
    })
    .await
    .expect_err("the fourth adjacency row must exceed a three-row budget");
    match error {
        StorageError::InvalidInput { message, .. } => {
            assert!(message.contains("work budget exceeded after 3 adjacency rows"));
        }
        other => panic!("expected bounded-work InvalidInput, got {other:?}"),
    }
    assert_eq!(ctx.snapshot()["graph_hops"], 4);
    assert!(
        !khive_storage::tx_registry::any_open_labeled(&origin_view, "graph_traverse_read"),
        "an over-budget return must drop its statement-scoped registry span"
    );
}

#[tokio::test]
async fn traverse_expired_deadline_returns_timeout_before_sql() {
    let store = setup_memory_store();
    let result = store
        .traverse(TraversalRequest {
            roots: vec![Uuid::new_v4()],
            options: TraversalOptions::new(1).with_direction(Direction::Out),
            include_roots: false,
            include_properties: false,
            execution_budget: TraversalExecutionBudget::new(1, std::time::Duration::ZERO),
        })
        .await;
    assert!(matches!(result, Err(StorageError::Timeout { .. })));
}

/// The VM progress handler, rather than a frontier-boundary deadline check,
/// interrupts an executing adjacency statement. The operation returns only a
/// timeout (never partial paths), drops its statement registry span, and clears
/// the connection-global handler before the same connection is reused.
#[tokio::test]
#[serial(tx_registry)]
async fn traverse_progress_handler_interrupts_statement_and_is_cleared() {
    let (_dir, store, origin_view) = setup_file_store_with_origin_view();
    let root = Uuid::new_v4();
    let edges = (0..512)
        .map(|_| make_edge(root, Uuid::new_v4(), EdgeRelation::Extends, 1.0))
        .collect();
    store.upsert_edges(edges).await.unwrap();

    let reader = store.pool.reader().unwrap();
    let counted_rows = std::sync::atomic::AtomicU64::new(0);
    let counted_queries = std::sync::atomic::AtomicU64::new(0);
    traverse_progress_seam::install(root);

    let result = run_bounded_traversal(
        reader.conn(),
        vec![root],
        TraversalOptions {
            max_depth: 1,
            direction: Direction::Out,
            relations: None,
            min_weight: None,
            limit: Some(512),
        },
        false,
        "default".to_string(),
        store.pool.origin(),
        TraversalExecutionBudget::default(),
        &counted_rows,
        &counted_queries,
    );

    assert!(
        matches!(result, Err(StorageError::Timeout { .. })),
        "the executing statement must be interrupted as a timeout, got {result:?}"
    );
    assert!(
        traverse_progress_seam::calls(root) >= 1,
        "the test must enter SQLite's VM progress callback"
    );
    assert!(
        !khive_storage::tx_registry::any_open_labeled(&origin_view, "graph_traverse_read"),
        "the interrupted statement must drop its graph_traverse_read span"
    );

    let calls_after_traversal = traverse_progress_seam::calls(root);
    let sum: i64 = reader
        .conn()
        .query_row(
            "WITH RECURSIVE n(x) AS (VALUES(0) UNION ALL SELECT x + 1 FROM n WHERE x < 10000) SELECT sum(x) FROM n",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(sum, 50_005_000);
    assert_eq!(
        traverse_progress_seam::calls(root),
        calls_after_traversal,
        "the traversal progress handler must be removed before connection reuse"
    );
    traverse_progress_seam::uninstall();
}

#[tokio::test]
#[serial(tx_registry)]
async fn outer_request_deadline_composes_with_graph_progress_and_work_budgets() {
    let (_dir, store, _origin_view) = setup_file_store_with_origin_view();
    let root = Uuid::new_v4();
    let edges = (0..512)
        .map(|_| make_edge(root, Uuid::new_v4(), EdgeRelation::Extends, 1.0))
        .collect();
    store.upsert_edges(edges).await.unwrap();

    // A generous outer request deadline must not overwrite graph's own
    // progress predicate. The graph seam remains the first cause.
    traverse_progress_seam::install(root);
    let graph_timeout = crate::scope_request_read_deadline(
        std::time::Duration::from_secs(60),
        store.traverse(TraversalRequest {
            roots: vec![root],
            options: TraversalOptions {
                max_depth: 1,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(512),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: TraversalExecutionBudget::default(),
        }),
    )
    .await;
    assert!(
        matches!(&graph_timeout, Err(StorageError::Timeout { operation }) if operation.contains("traverse")),
        "graph progress cause must survive the outer request guard; got {graph_timeout:?}"
    );
    traverse_progress_seam::uninstall();

    // The same outer guard must also preserve graph's non-time work-budget
    // classification instead of translating it to request timeout.
    let work_limited = crate::scope_request_read_deadline(
        std::time::Duration::from_secs(60),
        store.traverse(TraversalRequest {
            roots: vec![root],
            options: TraversalOptions {
                max_depth: 1,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(512),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: TraversalExecutionBudget::new(3, std::time::Duration::from_secs(5)),
        }),
    )
    .await;
    assert!(
        matches!(&work_limited, Err(StorageError::InvalidInput { message, .. }) if message.contains("work budget exceeded")),
        "graph work-budget classification must survive the outer request guard; got {work_limited:?}"
    );
}

// ---- multi-root == per-root decomposition equivalence tests ----

/// Equivalence fixture: for a small deterministic graph, multi-root
/// `traverse([R0, R1])` must produce the same per-root results as running
/// `traverse([R0])` and `traverse([R1])` independently, across four sections:
///
/// **Section A** – linear chains, Direction::Out, 6-case parameter matrix
///   (include_roots, relation filter, min_weight, finite limit):
///   A → B (w=0.9, Extends) → C (w=0.8) → D (w=0.7)
///   E → F (w=0.6, Extends) → G (w=0.5)
///
/// **Section B** – diamond with shortcut, Direction::Out (depth/via_edge drift):
///   H → VI (w=1.0, Extends) → K (w=1.0, Extends)
///   H → K  (w=0.5, PartOf)  ← shortcut; K is at depth 1 from H via H→K,
///                                          NOT depth 2 via H→VI→K.
///   Multi-root BFS must attribute K's first visit to depth=1, via_edge=H→K edge.
///
/// **Section C** – same diamond, Direction::In (bidirectional traversal):
///   roots [K, N].  K has two distinct incoming edges (H→K and VI→K); the
///   multi-root BFS must assign each one its own via_edge independently.
///
/// **Section D** – same diamond + chain, Direction::Both:
///   roots [VI, N].  VI has both in-edges (H→VI) and out-edges (VI→K).
///
/// M → N (w=1.0, Extends) — independent parallel chain used as the second
///   root in Sections B/C/D.
///
/// Same-depth node ordering is resolved by sorting PathNodes by (depth, node_id)
/// before the per-node comparison — no new ordering contract is introduced in
/// production code.
#[tokio::test]
async fn test_traverse_batch_equals_per_root_decomposition() {
    let store = setup_memory_store();

    // Section A nodes
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    let d = Uuid::new_v4();
    let e = Uuid::new_v4();
    let f = Uuid::new_v4();
    let g = Uuid::new_v4();

    // Section B/C/D nodes
    let h = Uuid::new_v4(); // diamond source
    let vi = Uuid::new_v4(); // diamond intermediate
    let k = Uuid::new_v4(); // convergent node (reachable from h directly AND via vi)
    let m = Uuid::new_v4(); // independent parallel root
    let n = Uuid::new_v4(); // child of m

    // Section A edges
    for (src, tgt, w) in [
        (a, b, 0.9_f64),
        (b, c, 0.8),
        (c, d, 0.7),
        (e, f, 0.6),
        (f, g, 0.5),
    ] {
        store
            .upsert_edge(make_edge(src, tgt, EdgeRelation::Extends, w))
            .await
            .unwrap();
    }

    // Section B/C/D edges
    store
        .upsert_edge(make_edge(h, vi, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(vi, k, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();
    // Shortcut: K is reachable from H in one hop; the depth-2 path via VI→K must
    // be suppressed by BFS first-visit in both multi-root and single-root traversals.
    store
        .upsert_edge(make_edge(h, k, EdgeRelation::PartOf, 0.5))
        .await
        .unwrap();
    store
        .upsert_edge(make_edge(m, n, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    // Sort PathNodes by (depth, node_id) for a stable comparison even when
    // same-depth nodes may appear in different backend-defined orders.
    // This resolves depth-1 tie-breaking without requiring a new ordering
    // contract in production code.
    let sort_nodes = |nodes: &mut Vec<khive_storage::types::PathNode>| {
        nodes.sort_by_key(|n| (n.depth, n.node_id));
    };

    // Direction is Clone but not Copy; TraversalOptions is built per case.
    struct Case {
        include_roots: bool,
        direction: Direction,
        relation_filter: Option<EdgeRelation>,
        min_weight: Option<f64>,
        limit: Option<u32>,
    }

    // run_cases: for each case, assert batched([r0, r1]) == decomposed([r0]) + ([r1]).
    // Takes roots by value so it can be called for each section.
    async fn run_cases(
        store: &SqlGraphStore,
        root0: Uuid,
        root1: Uuid,
        cases: &[Case],
        sort_nodes: &dyn Fn(&mut Vec<khive_storage::types::PathNode>),
    ) {
        for case in cases {
            let opts = TraversalOptions {
                max_depth: 4,
                direction: case.direction.clone(),
                relations: case.relation_filter.map(|r| vec![r]),
                min_weight: case.min_weight,
                limit: case.limit,
            };

            let batched = store
                .traverse(TraversalRequest {
                    roots: vec![root0, root1],
                    options: opts.clone(),
                    include_roots: case.include_roots,
                    include_properties: false,
                    execution_budget: Default::default(),
                })
                .await
                .unwrap();
            let single_0 = store
                .traverse(TraversalRequest {
                    roots: vec![root0],
                    options: opts.clone(),
                    include_roots: case.include_roots,
                    include_properties: false,
                    execution_budget: Default::default(),
                })
                .await
                .unwrap();
            let single_1 = store
                .traverse(TraversalRequest {
                    roots: vec![root1],
                    options: opts,
                    include_roots: case.include_roots,
                    include_properties: false,
                    execution_budget: Default::default(),
                })
                .await
                .unwrap();

            for (root_id, single_result) in [(root0, &single_0), (root1, &single_1)] {
                let batch_path = batched.iter().find(|p| p.root_id == root_id);
                let single_path = single_result.first();

                let label = format!(
                    "root={root_id:?} params=(include_roots={},dir={:?},rel={:?},\
                     min_w={:?},limit={:?})",
                    case.include_roots,
                    case.direction,
                    case.relation_filter,
                    case.min_weight,
                    case.limit,
                );

                match (batch_path, single_path) {
                    (None, None) => {}
                    (Some(bp), Some(sp)) => {
                        let mut bn = bp.nodes.clone();
                        let mut sn = sp.nodes.clone();
                        sort_nodes(&mut bn);
                        sort_nodes(&mut sn);

                        assert_eq!(
                            bn.len(),
                            sn.len(),
                            "{label}: node count mismatch batch={} single={}",
                            bn.len(),
                            sn.len()
                        );
                        for (bi, si) in bn.iter().zip(sn.iter()) {
                            assert_eq!(
                                bi.node_id, si.node_id,
                                "{label}: node_id mismatch at depth {}",
                                bi.depth
                            );
                            assert_eq!(
                                bi.depth, si.depth,
                                "{label}: depth mismatch for node {}",
                                bi.node_id
                            );
                            assert_eq!(
                                bi.via_edge, si.via_edge,
                                "{label}: via_edge mismatch for node {}",
                                bi.node_id
                            );
                        }
                        assert!(
                            (bp.total_weight - sp.total_weight).abs() < 1e-9,
                            "{label}: total_weight mismatch batch={} single={}",
                            bp.total_weight,
                            sp.total_weight
                        );
                    }
                    (None, Some(sp)) => {
                        panic!(
                            "{label}: batch missing path that single found ({} nodes)",
                            sp.nodes.len()
                        );
                    }
                    (Some(bp), None) => {
                        panic!(
                            "{label}: batch has path ({} nodes) that single didn't produce",
                            bp.nodes.len()
                        );
                    }
                }
            }
        }
    }

    // ── Section A: linear chains, Direction::Out ──────────────────────────────
    run_cases(
        &store,
        a,
        e,
        &[
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: true,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: Some(1),
            },
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: Some(2),
            },
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: Some(EdgeRelation::Extends),
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: Some(0.65),
                limit: None,
            },
        ],
        &sort_nodes,
    )
    .await;

    // ── Section B: diamond, Direction::Out ────────────────────────────────────
    // K must appear at depth=1 via the H→K shortcut edge, NOT at depth=2 via
    // H→VI→K. A bug in the multi-root BFS's per-root seen-set tracking would
    // produce the wrong depth or via_edge for K.
    run_cases(
        &store,
        h,
        m,
        &[
            Case {
                include_roots: false,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: true,
                direction: Direction::Out,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
        ],
        &sort_nodes,
    )
    .await;

    // ── Section C: converging node, Direction::In ─────────────────────────────
    // K has two distinct incoming edges (H→K via PartOf and VI→K via Extends).
    // Both must appear independently in K's path; neither must bleed into N's path.
    run_cases(
        &store,
        k,
        n,
        &[
            Case {
                include_roots: false,
                direction: Direction::In,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: true,
                direction: Direction::In,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
        ],
        &sort_nodes,
    )
    .await;

    // ── Section D: middle node, Direction::Both ───────────────────────────────
    // VI has H→VI incoming and VI→K outgoing.  Direction::Both must walk both
    // sides and produce identical results for the batched and single-root cases.
    run_cases(
        &store,
        vi,
        n,
        &[
            Case {
                include_roots: false,
                direction: Direction::Both,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
            Case {
                include_roots: true,
                direction: Direction::Both,
                relation_filter: None,
                min_weight: None,
                limit: None,
            },
        ],
        &sort_nodes,
    )
    .await;
}

/// Regression: `limit=0` with `include_roots=false` must emit NO path for that
/// root, not an empty `GraphPath`.  Pre-fix, the Rust-level truncation reduced
/// the node list to zero but still pushed a `GraphPath { nodes: [] }`.
///
/// This test must FAIL on the commit that introduced the per-root Rust truncation
/// but BEFORE the post-truncation empty guard was added, and PASS after.
#[tokio::test]
async fn test_traverse_limit_zero_include_roots_false_emits_no_path() {
    let store = setup_memory_store();

    let root = Uuid::new_v4();
    let child = Uuid::new_v4();

    store
        .upsert_edge(make_edge(root, child, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let paths = store
        .traverse(TraversalRequest {
            roots: vec![root],
            options: TraversalOptions {
                max_depth: 2,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(0),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: Default::default(),
        })
        .await
        .unwrap();

    assert_eq!(
        paths.len(),
        0,
        "limit=0 + include_roots=false: root has reachable children but no nodes \
         qualify under the cap, so no GraphPath should be emitted at all"
    );
}

#[tokio::test]
async fn traverse_accepts_root_cap_and_rejects_cap_plus_one_before_sql() {
    let store = setup_memory_store();

    let mut roots = Vec::with_capacity(MAX_TRAVERSAL_ROOTS);
    let mut expected_children: std::collections::HashMap<Uuid, Uuid> =
        std::collections::HashMap::with_capacity(MAX_TRAVERSAL_ROOTS);

    for _ in 0..MAX_TRAVERSAL_ROOTS {
        let root = Uuid::new_v4();
        let child = Uuid::new_v4();
        store
            .upsert_edge(make_edge(root, child, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
        roots.push(root);
        expected_children.insert(root, child);
    }

    let paths = store
        .traverse(TraversalRequest {
            roots: roots.clone(),
            options: TraversalOptions {
                max_depth: 1,
                direction: Direction::Out,
                relations: None,
                min_weight: None,
                limit: Some(1),
            },
            include_roots: false,
            include_properties: false,
            execution_budget: Default::default(),
        })
        .await
        .unwrap();

    // Every root must have exactly one reachable node (its direct child).
    assert_eq!(
        paths.len(),
        MAX_TRAVERSAL_ROOTS,
        "the inclusive root cap must return one path per root"
    );

    for path in &paths {
        let expected_child = expected_children[&path.root_id];
        assert_eq!(
            path.nodes.len(),
            1,
            "root {:?} must reach exactly 1 node",
            path.root_id
        );
        assert_eq!(
            path.nodes[0].node_id, expected_child,
            "root {:?} must reach its direct child",
            path.root_id
        );
    }

    roots.push(Uuid::new_v4());
    let ctx = khive_storage::usage::UsageContext::new();
    let result = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots,
                options: TraversalOptions::new(1).with_direction(Direction::Out),
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await;
    assert!(matches!(result, Err(StorageError::InvalidInput { .. })));
    assert_eq!(
        ctx.snapshot(),
        serde_json::json!({}),
        "cap+1 must be rejected before any traversal SQL is issued"
    );
}

/// One-hop, limit-one roots issue one bounded indexed statement and retain one
/// adjacency row each. This is the execution accounting contract after #1444.
#[tokio::test]
async fn traverse_limit_one_counts_one_query_and_row_per_root() {
    let store = setup_memory_store();

    const N: usize = 8;

    let mut roots: Vec<Uuid> = Vec::with_capacity(N);
    for _ in 0..N {
        let root = Uuid::new_v4();
        let child = Uuid::new_v4();
        store
            .upsert_edge(make_edge(root, child, EdgeRelation::Extends, 1.0))
            .await
            .unwrap();
        roots.push(root);
    }

    let ctx = khive_storage::usage::UsageContext::new();
    let paths = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: roots.clone(),
                options: TraversalOptions {
                    max_depth: 1,
                    direction: Direction::Out,
                    relations: None,
                    min_weight: None,
                    limit: Some(1),
                },
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await
    .unwrap();

    assert_eq!(paths.len(), N, "each root must retain its own result quota");

    let snap = ctx.snapshot();
    assert_eq!(
        snap["db_round_trips"], N as u64,
        "one bounded adjacency query per root; got {snap:?}"
    );
    assert_eq!(
        snap["graph_hops"], N as u64,
        "limit=1 must read exactly one unique one-hop row per root; got {snap:?}"
    );
}

/// STORAGE-AUD-003 / #485: PageRequest.offset > i64::MAX must return
/// InvalidInput instead of silently narrowing to a negative i64 offset.
#[tokio::test]
async fn page_offset_over_i64max_rejected() {
    let store = setup_memory_store();
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let result = store
        .query_edges(
            EdgeFilter::default(),
            vec![],
            PageRequest {
                offset: (i64::MAX as u64) + 1,
                limit: 10,
            },
        )
        .await;

    assert!(
        matches!(result, Err(StorageError::InvalidInput { .. })),
        "expected InvalidInput, got {result:?}"
    );
}

/// #1671: offset pages over edges whose primary sort key (`created_at`)
/// repeats must form a deterministic total order — a full paged sweep returns
/// every edge exactly once (no duplicates, no misses across page boundaries).
#[tokio::test]
async fn query_edges_offset_sweep_covers_equal_created_at_exactly_once() {
    let store = setup_memory_store();
    let created_at = DateTime::from_timestamp_micros(1_750_000_000_000_000).unwrap();
    let mut expected_ids = Vec::new();
    let mut edges = Vec::new();
    for _ in 0..211 {
        let mut edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0);
        edge.created_at = created_at;
        expected_ids.push(edge.id);
        edges.push(edge);
    }
    expected_ids.sort_unstable_by(|a, b| Uuid::from(*b).cmp(&Uuid::from(*a)));
    store.upsert_edges(edges).await.unwrap();

    // Default order (created_at DESC) — the empty-sort path.
    let mut actual_ids = Vec::new();
    let page_size = 37_u32;
    let mut offset = 0_u64;
    loop {
        let page = store
            .query_edges(
                EdgeFilter::default(),
                vec![],
                PageRequest {
                    offset,
                    limit: page_size,
                },
            )
            .await
            .unwrap();
        if page.items.is_empty() {
            break;
        }
        offset += page.items.len() as u64;
        actual_ids.extend(page.items.into_iter().map(|edge| edge.id));
    }
    assert_eq!(
        actual_ids, expected_ids,
        "default-order sweep must cover every edge exactly once"
    );

    // Explicit ASC primary sort — tiebreak direction follows the primary key.
    let mut asc_ids = Vec::new();
    let mut offset = 0_u64;
    loop {
        let page = store
            .query_edges(
                EdgeFilter::default(),
                vec![SortOrder {
                    field: EdgeSortField::CreatedAt,
                    direction: SortDirection::Asc,
                }],
                PageRequest {
                    offset,
                    limit: page_size,
                },
            )
            .await
            .unwrap();
        if page.items.is_empty() {
            break;
        }
        offset += page.items.len() as u64;
        asc_ids.extend(page.items.into_iter().map(|edge| edge.id));
    }
    let mut expected_asc = expected_ids.clone();
    expected_asc.reverse();
    assert_eq!(
        asc_ids, expected_asc,
        "ASC sweep must cover every edge exactly once in id ASC order"
    );

    // Multi-field custom sort: every edge has the same created_at and weight
    // (1.0), so only the appended `id` tiebreak — following the last sort key's
    // direction — makes these pages a deterministic total order.
    for (directions, expected) in [
        (vec![SortDirection::Desc, SortDirection::Asc], {
            // expected_ids is id DESC; the appended tiebreak follows the last
            // sort key's direction, so [Desc, Asc] pages in id ASC order.
            let mut reversed = expected_ids.clone();
            reversed.reverse();
            reversed
        }),
        (
            vec![SortDirection::Asc, SortDirection::Desc],
            expected_ids.clone(),
        ),
    ] {
        let sort = vec![
            SortOrder {
                field: EdgeSortField::CreatedAt,
                direction: directions[0].clone(),
            },
            SortOrder {
                field: EdgeSortField::Weight,
                direction: directions[1].clone(),
            },
        ];
        let mut actual = Vec::new();
        let mut offset = 0_u64;
        loop {
            let page = store
                .query_edges(
                    EdgeFilter::default(),
                    sort.clone(),
                    PageRequest {
                        offset,
                        limit: page_size,
                    },
                )
                .await
                .unwrap();
            if page.items.is_empty() {
                break;
            }
            offset += page.items.len() as u64;
            actual.extend(page.items.into_iter().map(|edge| edge.id));
        }
        assert_eq!(
            actual, expected,
            "multi-field sweep with directions {directions:?} must cover every edge exactly once"
        );
    }
}

#[tokio::test]
async fn query_edges_after_exact_multiple_final_page_has_no_next_after() {
    let store = setup_memory_store();

    let edges: Vec<Edge> = (0..4)
        .map(|_| make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 1.0))
        .collect();
    store.upsert_edges(edges).await.unwrap();

    let page1 = store
        .query_edges_after(EdgeFilter::default(), None, 2)
        .await
        .unwrap();
    assert_eq!(page1.items.len(), 2);
    let cursor = page1
        .next_after
        .expect("first page must report a cursor when two rows remain");

    let page2 = store
        .query_edges_after(EdgeFilter::default(), Some(cursor), 2)
        .await
        .unwrap();
    assert_eq!(page2.items.len(), 2);
    assert_eq!(
        page2.next_after, None,
        "an exact-size final page must not report a cursor"
    );
}

#[tokio::test]
async fn traverse_max_depth_over_public_cap_rejected() {
    let store = setup_memory_store();
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    store
        .upsert_edge(make_edge(a, b, EdgeRelation::Extends, 1.0))
        .await
        .unwrap();

    let request = TraversalRequest {
        roots: vec![a],
        options: TraversalOptions::new(MAX_TRAVERSAL_DEPTH + 1).with_direction(Direction::Out),
        include_roots: false,
        include_properties: false,
        execution_budget: Default::default(),
    };

    let result = store.traverse(request).await;
    assert!(
        matches!(result, Err(StorageError::InvalidInput { .. })),
        "expected InvalidInput, got {result:?}"
    );
}

/// ADR-067 Component A entry 3: with `KHIVE_WRITE_QUEUE=1`, `upsert_edges`
/// routes through the WriterTask channel instead of the pool-mutex path, and
/// both edges are actually committed and independently readable back.
///
/// Constructed via a `PoolConfig` literal (`write_queue_enabled: Some(true)`), not
/// the `KHIVE_WRITE_QUEUE` env var — that env var is process-global and this
/// crate's other tests are NOT `#[serial]` against it, so a window where it
/// is set here could leak into a concurrently-scheduled test's own pool
/// construction (ADR-067 Component A).
#[tokio::test]
async fn upsert_edges_routes_through_writer_task_when_flag_enabled() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_graph.db");
    let pool_cfg = PoolConfig {
        path: Some(path.clone()),
        write_queue_enabled: Some(true),
        ..PoolConfig::for_test()
    };
    let pool = Arc::new(ConnectionPool::new(pool_cfg).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }

    let store = SqlGraphStore::new_scoped(Arc::clone(&pool), true, "default");

    let e1 = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 0.6);
    let e2 = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 0.7);
    let id1 = e1.id;
    let id2 = e2.id;

    let summary = store.upsert_edges(vec![e1, e2]).await.unwrap();
    assert_eq!(summary.attempted, 2);
    assert_eq!(summary.affected, 2);
    assert_eq!(summary.failed, 0);

    assert!(store.get_edge(id1).await.unwrap().is_some());
    assert!(store.get_edge(id2).await.unwrap().is_some());
    assert_eq!(
        pool.writer_task_spawn_count(),
        1,
        "the flag-ON path must actually spawn and use the writer task"
    );
}

/// Fork C slice 2: proves the SINGLE-row `upsert_edge` (via `with_writer`,
/// distinct from the already-migrated batch `upsert_edges` above) is
/// actually enqueued on the pool's shared `WriterTaskHandle` channel when
/// `KHIVE_WRITE_QUEUE=1`.
///
/// `graph.rs`'s own flag-off/no-writer-task fallback (`open_standalone_writer`)
/// differs from entity.rs/note.rs's (`pool.try_writer()`), so a wall-clock
/// occupier-timing test is even less trustworthy here — a real file-backed
/// fallback connection opened per call would ALSO contend with the occupier
/// for SQLite's own write lock and could look "queued" by pure accident of
/// file-level locking, independent of whether `with_writer` used the shared
/// channel at all. This test sidesteps that confound entirely by reading
/// `WriterTaskHandle::queue_depth` directly — the live gauge over the exact
/// `mpsc` channel `with_writer`'s writer-task branch must call `send` on —
/// while an occupier deterministically holds the writer task's one drain
/// slot open (parked on a oneshot via `blocking_recv`, not a sleep/timing
/// race).
///
/// Constructed via a `PoolConfig` literal (`write_queue_enabled: Some(true)`), not
/// the `KHIVE_WRITE_QUEUE` env var — see
/// `upsert_edges_routes_through_writer_task_when_flag_enabled`'s doc comment
/// for the race this avoids.
#[tokio::test]
async fn upsert_edge_routes_through_writer_task_when_flag_enabled() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_graph_single.db");
    let pool_cfg = PoolConfig {
        path: Some(path.clone()),
        write_queue_enabled: Some(true),
        ..PoolConfig::for_test()
    };
    let pool = Arc::new(ConnectionPool::new(pool_cfg).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }

    let store = Arc::new(SqlGraphStore::new_scoped(
        Arc::clone(&pool),
        true,
        "default",
    ));

    let writer_task = pool
        .writer_task_handle()
        .unwrap()
        .expect("writer task must be spawned with the flag on for a file-backed pool");

    let (started_tx, started_rx) = tokio::sync::oneshot::channel::<()>();
    let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>();
    let occupier = {
        let writer_task = writer_task.clone();
        tokio::spawn(async move {
            writer_task
                .send(move |_conn| {
                    let _ = started_tx.send(());
                    let _ = release_rx.blocking_recv();
                    Ok::<(), StorageError>(())
                })
                .await
        })
    };

    started_rx
        .await
        .expect("occupier must signal it has started running inside the writer task");
    assert_eq!(
        writer_task.queue_depth(),
        0,
        "channel must start empty once the occupier has been dequeued and is running"
    );

    let edge = make_edge(Uuid::new_v4(), Uuid::new_v4(), EdgeRelation::Extends, 0.42);
    let edge_id = edge.id;

    let store_task = {
        let store = Arc::clone(&store);
        tokio::spawn(async move { store.upsert_edge(edge).await })
    };

    let mut saw_enqueued = false;
    for _ in 0..100 {
        if writer_task.queue_depth() >= 1 {
            saw_enqueued = true;
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
    }
    assert!(
        saw_enqueued,
        "upsert_edge's write request never appeared in the writer task's \
         channel while the occupier held the single drain slot — with_writer \
         is not routing this single-row write through the shared writer task"
    );

    release_tx
        .send(())
        .expect("occupier must still be waiting on the release signal");
    occupier
        .await
        .expect("occupier task must not panic")
        .expect("occupier write must succeed");
    store_task
        .await
        .expect("store task must not panic")
        .expect("upsert_edge must succeed once unblocked");

    let fetched = store.get_edge(edge_id).await.unwrap();
    assert!(
        fetched.is_some(),
        "edge must be committed and readable after queuing behind the occupier"
    );
}

// ---------------------------------------------------------------------------
// #769 — commit-time endpoint guard for canonical `link`/`link_many`
// ---------------------------------------------------------------------------

#[tokio::test]
async fn upsert_edge_guarded_succeeds_when_both_endpoints_exist() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    insert_live_entity(&pool, target);

    let edge = make_edge(source, target, EdgeRelation::Extends, 1.0);
    let edge_id = edge.id;
    let outcome = store.upsert_edge_guarded(edge).await.unwrap();

    assert_eq!(
        outcome,
        khive_storage::GuardedWriteOutcome::Written,
        "guarded write must succeed when both endpoints exist"
    );
    assert!(store.get_edge(edge_id).await.unwrap().is_some());
}

/// Reproduces #769's failure scenario at the layer where the bug actually
/// lived: an endpoint that existed when a caller validated it (mirrored here
/// by `insert_live_entity`) is hard-deleted before the edge write reaches
/// storage (mirrored by `hard_delete_entity`) — the exact window a
/// concurrent MCP request could land in between `KhiveRuntime::link`'s async
/// `validate_edge_relation_endpoints` read and its write. Before #769's fix,
/// `SqlGraphStore::upsert_edge` had no such check and would have inserted
/// this edge unconditionally, leaving it permanently dangling.
#[tokio::test]
async fn upsert_edge_guarded_returns_false_when_target_hard_deleted_before_write() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    insert_live_entity(&pool, target);

    // Simulates a concurrent hard-delete landing between prepare-time
    // validation and this write.
    hard_delete_entity(&pool, target);

    let edge = make_edge(source, target, EdgeRelation::Extends, 1.0);
    let edge_id = edge.id;
    let outcome = store.upsert_edge_guarded(edge).await.unwrap();

    match outcome {
        khive_storage::GuardedWriteOutcome::Refused(missing) => {
            assert!(missing.target, "target must be reported missing");
            assert!(!missing.source, "source was never deleted");
        }
        other => panic!(
            "guarded write must refuse an edge whose target vanished before commit, got {other:?}"
        ),
    }
    assert!(
        store.get_edge(edge_id).await.unwrap().is_none(),
        "no dangling edge may be persisted"
    );
}

#[tokio::test]
async fn upsert_edge_guarded_returns_false_when_source_hard_deleted_before_write() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    insert_live_entity(&pool, target);

    hard_delete_entity(&pool, source);

    let edge = make_edge(source, target, EdgeRelation::Extends, 1.0);
    let edge_id = edge.id;
    let outcome = store.upsert_edge_guarded(edge).await.unwrap();

    match outcome {
        khive_storage::GuardedWriteOutcome::Refused(missing) => {
            assert!(missing.source, "source must be reported missing");
            assert!(!missing.target, "target was never deleted");
        }
        other => panic!(
            "guarded write must refuse an edge whose source vanished before commit, got {other:?}"
        ),
    }
    assert!(store.get_edge(edge_id).await.unwrap().is_none());
}

#[tokio::test]
async fn upsert_edges_guarded_succeeds_when_all_endpoints_exist() {
    let (pool, store) = setup_memory_store_with_substrates();
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    insert_live_entity(&pool, a);
    insert_live_entity(&pool, b);
    insert_live_entity(&pool, c);

    let edges = vec![
        make_edge(a, b, EdgeRelation::Extends, 1.0),
        make_edge(b, c, EdgeRelation::Extends, 1.0),
    ];
    let ids: Vec<_> = edges.iter().map(|e| e.id).collect();

    let outcome = store.upsert_edges_guarded(edges).await.unwrap();
    assert_eq!(outcome.summary.attempted, 2);
    assert_eq!(outcome.summary.affected, 2);
    assert!(outcome.refused.is_none());

    for id in ids {
        assert!(store.get_edge(id).await.unwrap().is_some());
    }
}

/// Batch form of #769's regression: one edge in the batch targets an
/// endpoint that vanished before commit. The whole batch must be rejected —
/// no edge from it, including the ones whose endpoints were still live,
/// may be persisted (all-or-nothing, matching `create_many`'s existing
/// validation-failure contract).
#[tokio::test]
async fn upsert_edges_guarded_writes_nothing_when_one_endpoint_vanishes() {
    let (pool, store) = setup_memory_store_with_substrates();
    let a = Uuid::new_v4();
    let b = Uuid::new_v4();
    let c = Uuid::new_v4();
    insert_live_entity(&pool, a);
    insert_live_entity(&pool, b);
    insert_live_entity(&pool, c);

    // b's endpoint vanishes before the batch write — mirrors a concurrent
    // hard-delete landing between per-spec validation and `link_many`'s
    // single batched write.
    hard_delete_entity(&pool, b);

    let edges = vec![
        make_edge(a, b, EdgeRelation::Extends, 1.0), // dangling endpoint
        make_edge(a, c, EdgeRelation::Extends, 1.0), // both endpoints live
    ];
    let ids: Vec<_> = edges.iter().map(|e| e.id).collect();

    let outcome = store.upsert_edges_guarded(edges).await.unwrap();
    assert_eq!(outcome.summary.attempted, 2);
    assert_eq!(
        outcome.summary.affected, 0,
        "no edge from the batch may be persisted when any endpoint is missing"
    );
    let refusal = outcome
        .refused
        .expect("refused batch entry must be reported");
    assert_eq!(
        refusal.entry_index, 0,
        "the first entry (a, b) is the one with the missing endpoint"
    );
    assert!(
        refusal.missing.target,
        "b (the batch entry's target) must be reported missing"
    );
    assert!(!refusal.missing.source, "a was never deleted");
    assert_eq!(outcome.summary.failed, 2);
    assert_eq!(outcome.summary.errors.len(), 2);
    assert_eq!(outcome.summary.errors[0].item_id, Some(ids[0].to_string()));
    assert_eq!(outcome.summary.errors[1].item_id, Some(ids[1].to_string()));
    assert_eq!(
        outcome.summary.errors[0].class,
        BatchWriteErrorClass::InvalidInput
    );
    assert_eq!(
        outcome.summary.errors[0].retryability,
        BatchWriteRetryability::Permanent
    );
    assert_eq!(
        outcome.summary.errors[1].class,
        BatchWriteErrorClass::BatchAborted
    );
    assert_eq!(
        outcome.summary.errors[1].retryability,
        BatchWriteRetryability::Unknown
    );
    assert_eq!(
        outcome
            .summary
            .error_counts
            .iter()
            .map(|count| count.count)
            .sum::<u64>(),
        outcome.summary.failed
    );

    for id in ids {
        assert!(
            store.get_edge(id).await.unwrap().is_none(),
            "batch must be all-or-nothing: {id:?} must not be persisted"
        );
    }
}

#[tokio::test]
async fn legacy_guarded_batch_later_refusal_classifies_culprit_and_aborted_siblings() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    let other_target = Uuid::new_v4();
    let missing_target = Uuid::new_v4();
    for id in [source, target, other_target] {
        insert_live_entity(&pool, id);
    }
    let mut original = make_edge(source, target, EdgeRelation::Extends, 0.25);
    original.metadata = Some(serde_json::json!({"revision": 1}));
    let original_id = original.id;
    store.upsert_edge(original).await.unwrap();
    let original = store.get_edge(original_id).await.unwrap().unwrap();
    let mut replacement = make_edge(source, target, EdgeRelation::Extends, 0.75);
    replacement.metadata = Some(serde_json::json!({"revision": 2}));
    let edges = vec![
        replacement,
        make_edge(target, missing_target, EdgeRelation::Extends, 0.5),
        make_edge(source, other_target, EdgeRelation::Extends, 1.0),
    ];

    let outcome = store.upsert_edges_guarded(edges.clone()).await.unwrap();
    let refusal = outcome.refused.as_ref().unwrap();
    assert_eq!(refusal.entry_index, 1);
    assert!(!refusal.missing.source);
    assert!(refusal.missing.target);
    let summary = &outcome.summary;
    assert_eq!(summary.attempted, 3);
    assert_eq!(summary.affected, 0);
    assert_eq!(summary.failed, 3);
    assert_eq!(summary.errors.len(), 3);
    assert_eq!(summary.errors_omitted, 0);
    assert!(!summary.errors_truncated);
    assert_eq!(
        summary.first_error,
        format!(
            "batch entry 1: edge endpoint no longer exists at write time: source {target} or target {missing_target}"
        )
    );
    for (index, error) in summary.errors.iter().enumerate() {
        assert_eq!(error.index, index as u64);
        assert_eq!(error.item_id, Some(edges[index].id.to_string()));
        if index == 1 {
            assert_eq!(error.class, BatchWriteErrorClass::InvalidInput);
            assert_eq!(error.retryability, BatchWriteRetryability::Permanent);
            assert_eq!(error.message, summary.first_error);
        } else {
            assert_eq!(error.class, BatchWriteErrorClass::BatchAborted);
            assert_eq!(error.retryability, BatchWriteRetryability::Unknown);
            assert_eq!(
                error.message,
                format!(
                    "batch entry {index} was not written because guarded batch entry 1 was refused"
                )
            );
        }
    }
    assert_eq!(summary.error_counts.len(), 2);
    for count in &summary.error_counts {
        match count.class {
            BatchWriteErrorClass::InvalidInput => assert_eq!(count.count, 1),
            BatchWriteErrorClass::BatchAborted => assert_eq!(count.count, 2),
            other => panic!("unexpected refusal error class: {other:?}"),
        }
    }
    let page = outcome
        .refusal_page(
            &edges,
            None,
            PageRequest {
                offset: 0,
                limit: 10,
            },
        )
        .unwrap();
    assert_eq!(page.total, Some(3));
    assert_eq!(page.items, summary.errors);
    let persisted = store.get_edge(original_id).await.unwrap().unwrap();
    assert_eq!(
        serde_json::to_value(&persisted).unwrap(),
        serde_json::to_value(&original).unwrap()
    );
    for edge in edges {
        assert!(store
            .get_edge_including_deleted(edge.id)
            .await
            .unwrap()
            .is_none());
    }
}

#[tokio::test]
async fn upsert_edges_guarded_preserves_refusal_beyond_summary_cap() {
    let (pool, store) = setup_memory_store_with_substrates();
    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    let missing_target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    insert_live_entity(&pool, target);
    let cap = khive_storage::MAX_BATCH_WRITE_ERROR_DETAILS;
    let mut edges: Vec<_> = (0..cap + 3)
        .map(|_| make_edge(source, target, EdgeRelation::Extends, 1.0))
        .collect();
    edges[cap + 1] = make_edge(source, missing_target, EdgeRelation::Extends, 1.0);
    let ids: Vec<_> = edges.iter().map(|edge| edge.id).collect();

    let outcome = store.upsert_edges_guarded(edges.clone()).await.unwrap();
    let refusal = outcome.refused.as_ref().unwrap();
    assert_eq!(refusal.entry_index, cap + 1);
    assert!(!refusal.missing.source);
    assert!(refusal.missing.target);
    let summary = &outcome.summary;
    assert_eq!(summary.affected, 0);
    assert_eq!(summary.attempted, (cap + 3) as u64);
    assert_eq!(summary.failed, summary.attempted);
    assert_eq!(summary.errors.len(), cap);
    assert_eq!(summary.errors_omitted, 3);
    assert!(summary.errors_truncated);
    assert_eq!(summary.first_error, format!(
        "batch entry {}: edge endpoint no longer exists at write time: source {source} or target {missing_target}",
        cap + 1,
    ));
    for (index, error) in summary.errors.iter().enumerate() {
        assert_eq!(error.index, index as u64);
        assert_eq!(error.item_id, Some(ids[index].to_string()));
        assert_eq!(error.class, BatchWriteErrorClass::BatchAborted);
        assert_eq!(error.retryability, BatchWriteRetryability::Unknown);
        assert_eq!(
            error.message,
            format!(
                "batch entry {index} was not written because guarded batch entry {} was refused",
                cap + 1,
            )
        );
    }
    assert_eq!(summary.error_counts.len(), 2);
    assert_eq!(
        summary.error_counts[0].class,
        BatchWriteErrorClass::InvalidInput
    );
    assert_eq!(summary.error_counts[0].count, 1);
    assert_eq!(
        summary.error_counts[1].class,
        BatchWriteErrorClass::BatchAborted
    );
    assert_eq!(summary.error_counts[1].count, (cap + 2) as u64);
    let mut enumerated = Vec::new();
    while enumerated.len() < edges.len() {
        let page = outcome
            .refusal_page(
                &edges,
                None,
                PageRequest {
                    offset: enumerated.len() as u64,
                    limit: u32::MAX,
                },
            )
            .unwrap();
        assert_eq!(page.total, Some(edges.len() as u64));
        assert!(!page.items.is_empty());
        assert!(page.items.len() <= cap);
        enumerated.extend(page.items);
    }
    assert_eq!(enumerated.len(), edges.len());
    assert_eq!(&enumerated[..cap], summary.errors.as_slice());
    for (index, error) in enumerated.iter().enumerate() {
        assert_eq!(error.index, index as u64);
        assert_eq!(error.item_id, Some(ids[index].to_string()));
        if index == cap + 1 {
            assert_eq!(error.class, BatchWriteErrorClass::InvalidInput);
            assert_eq!(error.retryability, BatchWriteRetryability::Permanent);
            assert_eq!(error.message, summary.first_error);
        } else {
            assert_eq!(error.class, BatchWriteErrorClass::BatchAborted);
            assert_eq!(error.retryability, BatchWriteRetryability::Unknown);
        }
    }
    for id in ids {
        assert!(store.get_edge(id).await.unwrap().is_none());
    }
}

/// On the flag-off, file-backed singleton fallback
/// (`SqlGraphStore::with_writer`'s `is_file_backed` branch, no `WriterTask`),
/// `upsert_edge_guarded` must run its guarded `INSERT` and, when refused,
/// the missing-endpoint probe inside ONE `BEGIN IMMEDIATE` transaction —
/// not as two separate autocommit statements a concurrent writer could
/// interleave between, recreating the endpoint after the insert was
/// refused but before the probe explains why.
///
/// Sabotage-verified that a purely timing-based version of this
/// test (sleeps guessing when the guarded call was "probably" parked on
/// the write lock, then "probably" holding it) still passed with
/// the transaction wrapping removed: a refused, zero-row `INSERT`
/// autocommits — and so releases the write lock — before the guarded
/// call's Rust code even returns from it, and the probe that follows is a
/// plain `SELECT`, which never blocks on a writer in WAL mode. The two
/// statements run back to back on the same thread with no `.await`
/// between them, so there is no real scheduling gap for a racer on
/// another OS thread to win; a sleep-based racer is really racing its own
/// wake-up latency against a same-thread, no-I/O continuation, and loses
/// every time.
///
/// So this version does not race at all. [`insert_probe_seam`] has
/// production code itself (`observed_edge_upsert`, `#[cfg(test)]`-only)
/// park the guarded call immediately before its endpoint probe, keyed on
/// this test's own `(source, target)` pair so unrelated
/// concurrent tests are unaffected. The racer's write is then forced to
/// attempt landing at that exact seam:
///   - unwrapped (pre-fix) code holds no lock at the seam, so the racer's
///     write always lands before the probe resumes — reproducing #769's
///     residual bug (the probe wrongly reports the target as not missing).
///   - wrapped (fixed) code still holds its `BEGIN IMMEDIATE` at the seam,
///     so the racer blocks behind it and cannot land until after the
///     probe (and the guarded call's `COMMIT`) has already run.
#[tokio::test]
async fn upsert_edge_guarded_probe_is_atomic_with_insert_on_file_backed_singleton_path() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("guarded_atomic_singleton.db");

    let pool_cfg = PoolConfig {
        path: Some(path.clone()),
        write_queue_enabled: Some(false),
        ..PoolConfig::for_test()
    };
    let pool = Arc::new(ConnectionPool::new(pool_cfg).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
    }
    assert!(
        pool.writer_task_handle().unwrap().is_none(),
        "this test targets the no-WriterTask singleton fallback"
    );

    let store = Arc::new(SqlGraphStore::new_scoped(
        Arc::clone(&pool),
        true,
        "default",
    ));

    let source = Uuid::new_v4();
    let target = Uuid::new_v4();
    insert_live_entity(&pool, source);
    // target is intentionally never created — it is genuinely missing
    // for the entire test, so any observed "exists" state can only come
    // from the racer's later insert.

    let edge = make_edge(source, target, EdgeRelation::Extends, 1.0);
    let edge_id = edge.id;

    // Install the seam barrier before spawning the guarded call, keyed on
    // the exact (source, target) pair `observed_edge_upsert` canonicalizes
    // to and passes into `insert_probe_seam::hook` (Extends is not a
    // symmetric relation, so canonicalization is a no-op here).
    let (reached_rx, proceed_tx) = insert_probe_seam::install((source, target));

    let guarded_task = {
        let store = Arc::clone(&store);
        tokio::spawn(async move { store.upsert_edge_guarded(edge).await })
    };

    // Deterministic rendezvous: blocks until the guarded call has actually
    // entered the write transaction and is parked before the probe. No sleep, no
    // guess — a real signal sent from production code at that exact point.
    tokio::task::spawn_blocking(move || reached_rx.recv())
        .await
        .expect("waiting for the seam signal must not panic")
        .expect("guarded call must reach the insert-to-probe seam");

    // Now attempt to recreate the missing target from an independent
    // connection while the guarded call is parked exactly at the seam. On
    // unwrapped (pre-fix) code nothing holds the write lock here, so this
    // succeeds immediately; on wrapped (fixed) code the guarded call's own
    // BEGIN IMMEDIATE is still open, so this blocks until it commits.
    let racer_path = path.clone();
    let (racer_started_tx, racer_started_rx) = std::sync::mpsc::sync_channel::<()>(0);
    let racer = tokio::task::spawn_blocking(move || {
        let conn = rusqlite::Connection::open(&racer_path).unwrap();
        conn.busy_timeout(std::time::Duration::from_secs(2))
            .unwrap();
        let _ = racer_started_tx.send(());
        conn.execute(
            "INSERT INTO entities (id, deleted_at) VALUES (?1, NULL)",
            rusqlite::params![target.to_string()],
        )
    });

    // Deterministic signal that the racer has actually dispatched its
    // INSERT to SQLite, then a bounded settle window for it to either
    // complete (unwrapped code: no lock in the way) or genuinely start
    // blocking on the guarded call's still-open transaction (fixed code).
    // This window never has to race the guarded call itself — the guarded
    // call cannot move past the seam until `proceed_tx` fires below.
    tokio::task::spawn_blocking(move || racer_started_rx.recv())
        .await
        .expect("waiting for the racer-started signal must not panic")
        .expect("racer must signal before attempting its INSERT");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    // Release the guarded call. On unwrapped code the racer's write has
    // already landed by now, so the probe below will (wrongly) see the
    // target as present. On wrapped code the racer is still blocked behind
    // the open transaction, so the probe still sees it missing.
    proceed_tx
        .send(())
        .expect("guarded call must still be waiting at the seam");

    let outcome = guarded_task
        .await
        .expect("guarded task must not panic")
        .unwrap();
    racer
        .await
        .expect("racer task must not panic")
        .expect("racer's INSERT must eventually succeed");
    insert_probe_seam::uninstall();

    match outcome {
        khive_storage::GuardedWriteOutcome::Refused(missing) => {
            assert!(
                missing.target,
                "target was missing for the entire guarded write and must be \
                 reported so, even though the racer recreated it immediately \
                 afterward"
            );
            assert!(!missing.source, "source was always live");
        }
        other => panic!(
            "guarded write must refuse an edge whose target never existed \
             during the write, got {other:?}"
        ),
    }
    assert!(
        store.get_edge(edge_id).await.unwrap().is_none(),
        "no dangling edge may be persisted"
    );
}

/// A store whose schema was never created, so statement preparation fails
/// before any query reaches the database. Used to pin the negative case: a
/// read that never got as far as executing must count no round trip.
fn setup_memory_store_without_schema() -> SqlGraphStore {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    SqlGraphStore::new_scoped(pool, false, "default")
}

/// A store holding `good` valid edges out of `node`, plus one row whose
/// `relation` column is not a member of the closed relation enum.
///
/// Written through raw SQL because the typed write path cannot express it —
/// which is the point: the row models a database that has drifted from the
/// binary reading it, the one shape where a query prepares, executes, and
/// returns rows, and then fails partway through converting them. Weights
/// order the valid rows ahead of the corrupt one, so the number of entries
/// storage returns before the failure is fixed rather than incidental.
fn setup_store_with_a_corrupt_relation_row(node: Uuid, good: usize) -> SqlGraphStore {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    {
        let writer = pool.writer().unwrap();
        apply_test_schema(writer.conn());
        let now = Utc::now().timestamp_micros();
        for i in 0..good {
            writer
                .conn()
                .execute(
                    "INSERT INTO graph_edges \
                     (namespace, id, source_id, target_id, relation, weight, created_at, updated_at) \
                     VALUES ('default', ?1, ?2, ?3, 'extends', ?4, ?5, ?5)",
                    rusqlite::params![
                        Uuid::new_v4().to_string(),
                        node.to_string(),
                        Uuid::new_v4().to_string(),
                        1.0 - (i as f64) * 0.01,
                        now,
                    ],
                )
                .unwrap();
        }
        writer
            .conn()
            .execute(
                "INSERT INTO graph_edges \
                 (namespace, id, source_id, target_id, relation, weight, created_at, updated_at) \
                 VALUES ('default', ?1, ?2, ?3, 'not_a_relation', 0.1, ?4, ?4)",
                rusqlite::params![
                    Uuid::new_v4().to_string(),
                    node.to_string(),
                    Uuid::new_v4().to_string(),
                    now,
                ],
            )
            .unwrap();
    }
    SqlGraphStore::new_scoped(pool, false, "default")
}

fn neighbor_query(limit: u32) -> NeighborQuery {
    NeighborQuery {
        direction: Direction::Out,
        relations: None,
        limit: Some(limit),
        min_weight: None,
    }
}

/// `db_round_trips` counts round trips ISSUED. A read that executed and then
/// failed did real database work, and reporting zero for it is the failure
/// mode a consumer of these numbers cannot afford.
///
/// `graph_hops` counts adjacency entries storage RETURNED, which is a
/// statement about the database rather than about our types: all three rows
/// here came back off the cursor, and the third then failed to convert into
/// an `EdgeRelation`. So the expected count is `good + 1`, matching where the
/// traversal path increments — immediately after the row is read, before any
/// parsing. Counting after conversion would attribute our own decoding failure
/// to storage having done less work than it did.
#[tokio::test]
async fn neighbors_counts_the_query_and_the_rows_returned_before_the_failure() {
    let node = Uuid::new_v4();
    let store = setup_store_with_a_corrupt_relation_row(node, 2);
    let ctx = khive_storage::usage::UsageContext::new();

    let result = khive_storage::usage::scope(ctx.clone(), async {
        store.neighbors(node, neighbor_query(10)).await
    })
    .await;

    assert!(
        result.is_err(),
        "the corrupt relation row must fail conversion, or this test proves nothing"
    );

    let usage = ctx.snapshot();
    assert_eq!(
        usage
            .get("db_round_trips")
            .and_then(serde_json::Value::as_u64),
        Some(1),
        "the query executed and must be counted despite the error; got {usage}"
    );
    assert_eq!(
        usage.get("graph_hops").and_then(serde_json::Value::as_u64),
        Some(3),
        "all three rows came back off the cursor before the conversion failed; got {usage}"
    );
}

/// The other direction, and the one an unconditional post-await increment gets
/// wrong: a read whose statement never prepared issued nothing, so it must
/// count nothing. Counting it would inflate the number with work that did not
/// happen, which is the same class of error as dropping work that did.
#[tokio::test]
async fn neighbors_counts_nothing_when_the_statement_never_prepared() {
    let store = setup_memory_store_without_schema();
    let ctx = khive_storage::usage::UsageContext::new();

    let result = khive_storage::usage::scope(ctx.clone(), async {
        store.neighbors(Uuid::new_v4(), neighbor_query(10)).await
    })
    .await;

    assert!(
        result.is_err(),
        "a store with no schema must fail to prepare, or this test proves nothing"
    );

    let usage = ctx.snapshot();
    assert!(
        usage.get("db_round_trips").is_none(),
        "no statement executed, so no round trip may be counted; got {usage}"
    );
    assert!(
        usage.get("graph_hops").is_none(),
        "no rows were returned; got {usage}"
    );
}

/// Both directions again on the batched path, where a chunked read makes the
/// stakes higher: the counters live across the whole chunk loop, so a failure
/// in a later chunk must not erase what earlier chunks already did.
#[tokio::test]
async fn batch_neighbors_counts_the_query_and_the_rows_returned_before_the_failure() {
    let node = Uuid::new_v4();
    let store = setup_store_with_a_corrupt_relation_row(node, 2);
    let ctx = khive_storage::usage::UsageContext::new();

    let sources = vec![node];
    let result = khive_storage::usage::scope(ctx.clone(), async {
        store.batch_neighbors(&sources, neighbor_query(10)).await
    })
    .await;

    assert!(
        result.is_err(),
        "the corrupt relation row must fail conversion, or this test proves nothing"
    );

    let usage = ctx.snapshot();
    assert_eq!(
        usage
            .get("db_round_trips")
            .and_then(serde_json::Value::as_u64),
        Some(1),
        "the batched query executed and must be counted despite the error; got {usage}"
    );
    assert_eq!(
        usage.get("graph_hops").and_then(serde_json::Value::as_u64),
        Some(3),
        "all three rows came back off the cursor before the conversion failed; got {usage}"
    );
}

/// Same rule as the `neighbors` twin, on the traversal path: a frontier
/// statement that never prepared issued nothing to the store, so it must count
/// nothing — the increment sits after `prepare` succeeds, exactly where the
/// neighbors and batch paths put theirs.
#[tokio::test]
async fn traverse_counts_nothing_when_the_statement_never_prepared() {
    let store = setup_memory_store_without_schema();
    let ctx = khive_storage::usage::UsageContext::new();

    let result = khive_storage::usage::scope(ctx.clone(), async {
        store
            .traverse(TraversalRequest {
                roots: vec![Uuid::new_v4()],
                options: TraversalOptions::new(2).with_direction(Direction::Out),
                include_roots: false,
                include_properties: false,
                execution_budget: Default::default(),
            })
            .await
    })
    .await;

    assert!(
        result.is_err(),
        "a store with no schema must fail the traversal, or this test proves nothing"
    );

    let usage = ctx.snapshot();
    assert!(
        usage.get("db_round_trips").is_none(),
        "the frontier statement never prepared, so no round trip may be counted; got {usage}"
    );
}

/// Both directions remain two separately indexed statements. Reintroducing a
/// single OR predicate would turn each bounded frontier expansion into a full
/// namespace scan (#1229).
#[tokio::test]
async fn traversal_neighbor_statements_seek_by_source_and_target_id() {
    let store = setup_memory_store();
    let reader = store.pool.reader().unwrap();
    for (direction, expected_index, endpoint) in [
        (Direction::Out, "idx_graph_edges_ns_src_rel", "source_id=?"),
        (Direction::In, "idx_graph_edges_ns_tgt_rel", "target_id=?"),
    ] {
        let plan_sql = format!(
            "EXPLAIN QUERY PLAN {}",
            traversal_neighbor_sql(direction, 0, false)
        );
        let mut stmt = reader.conn().prepare(&plan_sql).unwrap();
        let details: Vec<String> = stmt
            .query_map(rusqlite::params!["default", "root", 10_i64], |row| {
                row.get(3)
            })
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert!(
            details
                .iter()
                .any(|detail| detail.contains(expected_index) && detail.contains(endpoint)),
            "expected indexed {endpoint} seek via {expected_index}, got: {details:?}"
        );
        assert!(
            !details
                .iter()
                .any(|detail| detail.contains("SCAN graph_edges")),
            "bounded adjacency statements must not scan graph_edges: {details:?}"
        );
    }
}

/// khive#1229 correctness: a hub node with mixed out- and in-edges to its
/// spokes, each spoke extending one hop further to its own tail node.
/// `Direction::Both` at `max_depth=2` from the hub must return every spoke
/// (reached via whichever direction connects it to the hub) AND every tail
/// (reached via the spoke's outgoing edge), regardless of which indexed
/// directional statement found the spoke. This is the exact shape the OR-predicate
/// bug silently returned correct results for at small scale but 380x slower
/// at hub scale — this test locks the result set the perf fix must preserve.
#[tokio::test]
async fn traverse_both_direction_hub_depth_two_returns_full_node_set() {
    let store = setup_memory_store();

    let hub = Uuid::new_v4();
    let mut expected_depth1 = HashSet::new();
    let mut expected_depth2 = HashSet::new();

    const SPOKES: usize = 40;
    for i in 0..SPOKES {
        let spoke = Uuid::new_v4();
        let tail = Uuid::new_v4();
        expected_depth1.insert(spoke);
        expected_depth2.insert(tail);

        if i % 2 == 0 {
            // hub --out--> spoke
            store
                .upsert_edge(make_edge(hub, spoke, EdgeRelation::Extends, 1.0))
                .await
                .unwrap();
        } else {
            // spoke --out--> hub (hub reaches it via the in-arm)
            store
                .upsert_edge(make_edge(spoke, hub, EdgeRelation::Extends, 1.0))
                .await
                .unwrap();
        }
        // spoke --out--> tail, reachable regardless of which arm found spoke
        store
            .upsert_edge(make_edge(spoke, tail, EdgeRelation::VariantOf, 1.0))
            .await
            .unwrap();
    }

    let paths = store
        .traverse(TraversalRequest {
            roots: vec![hub],
            options: TraversalOptions::new(2).with_direction(Direction::Both),
            include_roots: false,
            include_properties: false,
            execution_budget: Default::default(),
        })
        .await
        .unwrap();

    assert_eq!(paths.len(), 1);
    let path = &paths[0];
    assert_eq!(path.root_id, hub);

    let got_depth1: HashSet<Uuid> = path
        .nodes
        .iter()
        .filter(|n| n.depth == 1)
        .map(|n| n.node_id)
        .collect();
    let got_depth2: HashSet<Uuid> = path
        .nodes
        .iter()
        .filter(|n| n.depth == 2)
        .map(|n| n.node_id)
        .collect();

    assert_eq!(
        got_depth1, expected_depth1,
        "every spoke must be reached at depth 1 regardless of edge direction"
    );
    assert_eq!(
        got_depth2, expected_depth2,
        "every tail must be reached at depth 2 through its spoke's out-edge"
    );
    assert_eq!(
        path.nodes.len(),
        expected_depth1.len() + expected_depth2.len(),
        "no duplicate or spurious nodes"
    );
}

/// #2911: a relation breakdown cannot separate structure from provenance,
/// because relations do not determine endpoint bases. The fixture puts one
/// edge in every bucket, including a `note -> note` `supports` edge, which is
/// the case that makes `edges - annotates` the wrong derivation: it is neither
/// annotates nor structure, so subtracting annotates alone still overcounts.
#[tokio::test]
async fn endpoint_base_counts_separate_structure_from_provenance() {
    let (pool, store) = setup_memory_store_with_substrates();

    let (e1, e2, e3, e4) = (
        Uuid::new_v4(),
        Uuid::new_v4(),
        Uuid::new_v4(),
        Uuid::new_v4(),
    );
    for id in [e1, e2, e3, e4] {
        insert_live_entity(&pool, id);
    }
    let (n1, n2, n3) = (Uuid::new_v4(), Uuid::new_v4(), Uuid::new_v4());
    for id in [n1, n2, n3] {
        insert_note(&pool, id, false);
    }
    // Endpoints in neither table. Absence is not a tombstone, so these stay
    // counted, and they must land in `unresolved` rather than a named bucket.
    let (ghost_a, ghost_b) = (Uuid::new_v4(), Uuid::new_v4());

    for edge in [
        make_edge(e1, e2, EdgeRelation::Contains, 1.0),
        make_edge(e3, e4, EdgeRelation::DependsOn, 1.0),
        make_edge(n1, e1, EdgeRelation::Annotates, 1.0),
        make_edge(e2, n2, EdgeRelation::Annotates, 1.0),
        make_edge(n2, n3, EdgeRelation::Supports, 1.0),
        make_edge(ghost_a, ghost_b, EdgeRelation::DependsOn, 1.0),
    ] {
        store.upsert_edge(edge).await.unwrap();
    }

    let counts = store.count_edges_by_endpoint_base().await.unwrap();
    assert_eq!(counts.entity_entity, 2, "{counts:?}");
    assert_eq!(counts.note_entity, 1, "{counts:?}");
    assert_eq!(counts.entity_note, 1, "{counts:?}");
    assert_eq!(counts.note_note, 1, "{counts:?}");
    assert_eq!(counts.unresolved, 1, "{counts:?}");

    // The invariant that makes the breakdown checkable: it partitions the same
    // population `count_edges` reports.
    let total = store.count_edges(EdgeFilter::default()).await.unwrap();
    assert_eq!(total, 6);
    assert_eq!(counts.total(), total, "buckets must partition the total");

    // The derivation a caller would reach for, shown wrong on this fixture.
    let by_relation: HashMap<_, _> = store
        .count_edges_by_relation()
        .await
        .unwrap()
        .into_iter()
        .collect();
    let annotates = by_relation
        .get(&EdgeRelation::Annotates)
        .copied()
        .unwrap_or(0);
    assert_eq!(annotates, 2);
    assert_ne!(
        total - annotates,
        counts.entity_entity,
        "total minus annotates is 4, structure is 2: the note-to-note support edge and the \
         unresolved pair are why that subtraction cannot be the definition"
    );

    // The namespace-scoped variant answers the same question.
    let namespaces = vec!["default".to_string()];
    assert_eq!(
        store
            .count_edges_by_endpoint_base_in_namespaces(&namespaces)
            .await
            .unwrap(),
        counts,
        "scoped and unscoped counts must agree"
    );

    // A tombstoned endpoint leaves its bucket, exactly as it leaves the total.
    soft_delete_note(&pool, n2);
    let after = store.count_edges_by_endpoint_base().await.unwrap();
    assert_eq!(after.entity_note, 0, "{after:?}");
    assert_eq!(after.note_note, 0, "{after:?}");
    assert_eq!(after.entity_entity, 2, "{after:?}");
    assert_eq!(
        after.total(),
        store.count_edges(EdgeFilter::default()).await.unwrap()
    );
}