khive-db 0.9.0

use super::*;
use crate::migrations::run_migrations;
use crate::pool::PoolConfig;
use khive_storage::{Attachment, AttachmentSubstrate, ContentRef};
use std::time::Duration;
use tokio::sync::oneshot;

fn setup_pool() -> Arc<ConnectionPool> {
    let config = PoolConfig {
        path: None,
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    {
        let writer = pool.writer().unwrap();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
    }
    pool
}

const TEST_ATTACHMENTS_DDL: &str = r#"
CREATE TABLE attachments (
    record_uuid TEXT NOT NULL,
    substrate   TEXT NOT NULL CHECK (substrate IN ('entity', 'note')),
    role        TEXT NOT NULL,
    content_ref TEXT NOT NULL,
    media_type  TEXT,
    size_bytes  INTEGER,
    created_at  INTEGER NOT NULL,
    PRIMARY KEY (record_uuid, role)
);
CREATE INDEX idx_attachments_content_ref ON attachments(content_ref);
"#;

fn setup_memory_store() -> SqlEntityStore {
    SqlEntityStore::new(setup_pool(), false)
}

fn setup_memory_store_ns(_ns: &str) -> SqlEntityStore {
    SqlEntityStore::new(setup_pool(), false)
}

fn make_entity(namespace: &str, kind: &str, name: &str) -> Entity {
    let now = chrono::Utc::now().timestamp_micros();
    Entity {
        id: Uuid::new_v4(),
        namespace: namespace.to_string(),
        kind: kind.to_string(),
        entity_type: None,
        name: name.to_string(),
        description: None,
        properties: None,
        tags: Vec::new(),
        created_at: now,
        updated_at: now,
        deleted_at: None,
        merged_into: None,
        merge_event_id: None,
        version: 1,
        content_ref: None,
    }
}

fn content_attachment(record_uuid: Uuid, digest: &str) -> Attachment {
    Attachment {
        record_uuid,
        substrate: AttachmentSubstrate::Entity,
        role: "content".to_string(),
        content_ref: ContentRef::from_hex(digest).expect("canonical content ref"),
        media_type: Some("application/octet-stream".to_string()),
        size_bytes: Some(42),
        created_at: 123,
    }
}

#[test]
fn case_insensitive_candidate_lookup_uses_one_partial_index_seek_per_candidate() {
    let mut conn = rusqlite::Connection::open_in_memory().unwrap();
    run_migrations(&mut conn).unwrap();

    let filter = EntityFilter {
        names_ci: vec!["lora".to_string(), "北京大学".to_string()],
        ..EntityFilter::default()
    };
    let mut lookup_filter = filter.clone();
    lookup_filter.names_ci.clear();
    let (where_sql, mut params) = build_entity_where("local", &lookup_filter);
    let candidate_param_indices: Vec<usize> = filter
        .names_ci
        .iter()
        .map(|candidate| {
            params.push(Box::new(candidate.to_ascii_lowercase()));
            params.len()
        })
        .collect();
    params.push(Box::new(64_i64));
    params.push(Box::new(0_i64));
    let limit_idx = params.len() - 1;
    let offset_idx = params.len();
    let data_sql = build_candidate_entity_query(
        "id",
        &where_sql,
        &candidate_param_indices,
        "created_at DESC",
        limit_idx,
        offset_idx,
    );
    assert!(!data_sql.contains("GROUP BY"));
    assert!(data_sql.contains("FROM candidates"));
    assert!(data_sql.contains("LIMIT 1"));

    let sql = format!("EXPLAIN QUERY PLAN {data_sql}");
    let mut stmt = conn.prepare(&sql).unwrap();
    let param_refs: Vec<&dyn rusqlite::types::ToSql> =
        params.iter().map(|param| param.as_ref()).collect();
    let details: Vec<String> = stmt
        .query_map(param_refs.as_slice(), |row| row.get(3))
        .unwrap()
        .collect::<Result<_, _>>()
        .unwrap();

    assert!(
        details.iter().any(|detail| {
            detail.contains("SEARCH entities USING INDEX idx_entities_namespace_name_ci")
        }),
        "candidate lookup must seek the case-insensitive expression index: {details:?}"
    );
    assert!(
        details
            .iter()
            .any(|detail| detail.contains("CORRELATED SCALAR SUBQUERY")),
        "candidate relation must drive one scalar lookup per name: {details:?}"
    );
    assert!(
        !details
            .iter()
            .any(|detail| detail.contains("SCAN entities")),
        "candidate lookup must not scan entities: {details:?}"
    );

    let index_sql: String = conn
        .query_row(
            "SELECT sql FROM sqlite_master \
             WHERE type = 'index' AND name = 'idx_entities_namespace_name_ci'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert!(
        index_sql.contains("WHERE deleted_at IS NULL"),
        "candidate lookup index must exclude tombstones: {index_sql}"
    );
}

#[tokio::test]
async fn case_insensitive_candidate_lookup_skips_unbounded_total_count() {
    let store = setup_memory_store();
    store
        .upsert_entity(make_entity("local", "concept", "LoRA"))
        .await
        .unwrap();

    let page = store
        .query_entities(
            "local",
            EntityFilter {
                names_ci: vec!["lora".to_string()],
                ..EntityFilter::default()
            },
            PageRequest {
                limit: 64,
                offset: 0,
            },
        )
        .await
        .unwrap();

    assert_eq!(page.items.len(), 1);
    assert_eq!(page.items[0].name, "LoRA");
    assert_eq!(page.total, None);
}

#[tokio::test]
async fn complete_id_lookup_skips_redundant_total_count() {
    let store = setup_memory_store();
    let first = make_entity("local", "concept", "first");
    let second = make_entity("local", "concept", "second");
    let ids = vec![first.id, second.id, Uuid::new_v4()];
    store.upsert_entity(first).await.unwrap();
    store.upsert_entity(second).await.unwrap();

    let page = store
        .query_entities(
            "local",
            EntityFilter {
                ids: ids.clone(),
                ..EntityFilter::default()
            },
            PageRequest {
                limit: ids.len() as u32,
                offset: 0,
            },
        )
        .await
        .unwrap();

    assert_eq!(page.items.len(), 2);
    assert_eq!(page.total, None);
}

#[tokio::test]
async fn case_insensitive_candidate_lookup_caps_folded_names_not_duplicate_rows() {
    let store = setup_memory_store();
    let mut older_b = make_entity("local", "concept", "crowdbeta");
    older_b.created_at = 1;
    older_b.updated_at = 1;
    store.upsert_entity(older_b).await.unwrap();

    for created_at in 2..=258 {
        let mut newer_a = make_entity("local", "concept", "CrowdAlpha");
        newer_a.created_at = created_at;
        newer_a.updated_at = created_at;
        store.upsert_entity(newer_a).await.unwrap();
    }

    let page = store
        .query_entities(
            "local",
            EntityFilter {
                names_ci: vec![
                    "CrowdAlpha".to_string(),
                    "crowdalpha".to_string(),
                    "crowdbeta".to_string(),
                ],
                ..EntityFilter::default()
            },
            PageRequest {
                limit: 256,
                offset: 0,
            },
        )
        .await
        .unwrap();

    let folded_names: Vec<String> = page
        .items
        .iter()
        .map(|entity| entity.name.to_lowercase())
        .collect();
    assert_eq!(folded_names.len(), 2);
    assert!(folded_names.contains(&"crowdalpha".to_string()));
    assert!(folded_names.contains(&"crowdbeta".to_string()));
}

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

    let entity = make_entity("default", "concept", "LoRA");
    let id = entity.id;

    store.upsert_entity(entity).await.unwrap();

    let fetched = store.get_entity(id).await.unwrap();
    assert!(fetched.is_some());
    let fetched = fetched.unwrap();
    assert_eq!(fetched.id, id);
    assert_eq!(fetched.name, "LoRA");
    assert_eq!(fetched.kind, "concept");
}

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

    let props = serde_json::json!({"domain": "fine-tuning", "type": "technique"});
    let entity = Entity::new("default", "concept", "QLoRA")
        .with_description("Quantized LoRA")
        .with_properties(props.clone())
        .with_tags(vec!["fine-tuning".to_string(), "quantization".to_string()]);
    let id = entity.id;

    store.upsert_entity(entity).await.unwrap();

    let fetched = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(fetched.description.as_deref(), Some("Quantized LoRA"));
    assert_eq!(fetched.properties, Some(props));
    assert_eq!(fetched.tags, vec!["fine-tuning", "quantization"]);
}

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

    let entity = make_entity("default", "concept", "to-delete");
    let id = entity.id;
    store.upsert_entity(entity).await.unwrap();

    let deleted = store.delete_entity(id, DeleteMode::Soft).await.unwrap();
    assert!(deleted);

    let fetched = store.get_entity(id).await.unwrap();
    assert!(fetched.is_none());
}

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

    let entity = make_entity("default", "concept", "to-hard-delete");
    let id = entity.id;
    store.upsert_entity(entity).await.unwrap();

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

    let fetched = store.get_entity(id).await.unwrap();
    assert!(fetched.is_none());
}

#[tokio::test]
async fn test_query_entities_basic() {
    let store = setup_memory_store_ns("ns1");

    for name in &["Alpha", "Beta", "Gamma"] {
        store
            .upsert_entity(make_entity("ns1", "concept", name))
            .await
            .unwrap();
    }
    store
        .upsert_entity(make_entity("ns1", "document", "Paper1"))
        .await
        .unwrap();

    let page = store
        .query_entities(
            "ns1",
            EntityFilter::default(),
            PageRequest {
                offset: 0,
                limit: 10,
            },
        )
        .await
        .unwrap();
    assert_eq!(page.items.len(), 4);
    assert_eq!(page.total, Some(4));

    // Filter by kind
    let concepts = store
        .query_entities(
            "ns1",
            EntityFilter {
                kinds: vec!["concept".to_string()],
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(concepts.items.len(), 3);
}

#[tokio::test]
async fn test_query_by_name_prefix() {
    let store = setup_memory_store_ns("ns1");

    // "Alpha" and "AlphaGo" both start with "Alpha"; "Beta" does not
    for &name in &["Alpha", "AlphaGo", "Beta"] {
        store
            .upsert_entity(make_entity("ns1", "concept", name))
            .await
            .unwrap();
    }

    let result = store
        .query_entities(
            "ns1",
            EntityFilter {
                name_prefix: Some("Alpha".to_string()),
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(result.items.len(), 2);
    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(names.contains(&"Alpha"), "Alpha not found in {names:?}");
    assert!(names.contains(&"AlphaGo"), "AlphaGo not found in {names:?}");
    assert!(!names.contains(&"Beta"));
}

#[tokio::test]
async fn test_query_by_name_prefix_escapes_underscore_wildcard() {
    let store = setup_memory_store_ns("ns1");

    // The real entity's name contains a literal underscore.
    store
        .upsert_entity(make_entity("ns1", "concept", "a_b"))
        .await
        .unwrap();

    // 150 decoys that would match the *unescaped* LIKE pattern "a_b%"
    // (`_` as a wildcard matches any single character) but must not match
    // once the wildcard is escaped. Created after "a_b" so a page ordered
    // by created_at DESC with LIMIT 100 would push "a_b" out unless exact
    // matches are also ranked first.
    for i in 0..150 {
        store
            .upsert_entity(make_entity("ns1", "concept", &format!("aXb-{i:03}")))
            .await
            .unwrap();
    }

    let result = store
        .query_entities(
            "ns1",
            EntityFilter {
                name_prefix: Some("a_b".to_string()),
                ..Default::default()
            },
            PageRequest {
                offset: 0,
                limit: 100,
            },
        )
        .await
        .unwrap();

    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(
        names.contains(&"a_b"),
        "exact match 'a_b' must survive escaping and page ordering despite 150 wildcard-matching decoys; got {names:?}"
    );
    assert!(
        !names.iter().any(|n| n.starts_with("aXb")),
        "escaped '_' must not match decoy names like 'aXb-000'; got {names:?}"
    );
}

#[tokio::test]
async fn test_query_by_name_prefix_escapes_percent_wildcard() {
    let store = setup_memory_store_ns("ns1");

    // The real entity's name contains a literal percent sign.
    store
        .upsert_entity(make_entity("ns1", "concept", "50%off"))
        .await
        .unwrap();

    // 150 decoys that would match the *unescaped* LIKE pattern "50%off%"
    // (`%` as a wildcard matches any sequence, including across "-") but
    // must not match once the wildcard is escaped.
    for i in 0..150 {
        store
            .upsert_entity(make_entity("ns1", "concept", &format!("50-off-{i:03}")))
            .await
            .unwrap();
    }

    let result = store
        .query_entities(
            "ns1",
            EntityFilter {
                name_prefix: Some("50%off".to_string()),
                ..Default::default()
            },
            PageRequest {
                offset: 0,
                limit: 100,
            },
        )
        .await
        .unwrap();

    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(
        names.contains(&"50%off"),
        "exact match '50%off' must survive escaping and page ordering despite 150 wildcard-matching decoys; got {names:?}"
    );
    assert!(
        !names.iter().any(|n| n.starts_with("50-off-")),
        "escaped '%' must not match decoy names like '50-off-000'; got {names:?}"
    );
}

// Regression for a gap on #818/#834: the two tests above prove
// escaping keeps decoys out of the WHERE clause entirely, so they pass even
// without the exact-match-first ORDER BY CASE. This test isolates that
// ordering: every decoy genuinely matches the LIKE pattern "Base%" (no
// wildcard characters, escaping plays no role) and is newer than the exact
// match, so a page ordered only by `created_at DESC` would fill entirely with
// decoys and strand "Base" past the LIMIT 100 boundary. It fails if the
// `ORDER BY CASE WHEN LOWER(name) = ... THEN 0 ELSE 1 END` clause is removed.
#[tokio::test]
async fn test_query_by_name_prefix_exact_match_ranked_before_many_matching_decoys() {
    let store = setup_memory_store_ns("ns1");

    store
        .upsert_entity(make_entity("ns1", "concept", "Base"))
        .await
        .unwrap();

    for i in 0..150 {
        store
            .upsert_entity(make_entity("ns1", "concept", &format!("Base-{i:03}")))
            .await
            .unwrap();
    }

    let result = store
        .query_entities(
            "ns1",
            EntityFilter {
                name_prefix: Some("Base".to_string()),
                ..Default::default()
            },
            PageRequest {
                offset: 0,
                limit: 100,
            },
        )
        .await
        .unwrap();

    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(
        names.contains(&"Base"),
        "exact match 'Base' must be ranked ahead of 150 newer, genuinely prefix-matching \
         decoys within a LIMIT 100 page; got {names:?}"
    );
}

#[tokio::test]
async fn test_count_entities() {
    let store = setup_memory_store_ns("ns1");

    for _ in 0..5 {
        store
            .upsert_entity(make_entity("ns1", "concept", "X"))
            .await
            .unwrap();
    }

    let count = store
        .count_entities("ns1", EntityFilter::default())
        .await
        .unwrap();
    assert_eq!(count, 5);

    // Namespace is the caller's responsibility — querying "ns2" returns 0
    // because no entities were inserted in that namespace.
    let count_other = store
        .count_entities("ns2", EntityFilter::default())
        .await
        .unwrap();
    assert_eq!(count_other, 0);
}

#[tokio::test]
async fn count_entities_normalizes_case_insensitive_names() {
    let store = setup_memory_store();
    store
        .upsert_entity(make_entity("local", "concept", "LoRA"))
        .await
        .unwrap();

    let count = store
        .count_entities(
            "local",
            EntityFilter {
                names_ci: vec!["LORA".to_string()],
                ..EntityFilter::default()
            },
        )
        .await
        .unwrap();

    assert_eq!(count, 1);
}

#[tokio::test]
async fn batched_namespace_entity_count_exceeds_sqlite_variable_limit() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(Arc::clone(&pool), false);
    let live_a = make_entity("stats-a", "concept", "live-a");
    let deleted_a = make_entity("stats-a", "concept", "deleted-a");
    let deleted_a_id = deleted_a.id;
    let live_b = make_entity("stats-b", "document", "live-b");

    store.upsert_entity(live_a).await.unwrap();
    store.upsert_entity(deleted_a).await.unwrap();
    store.upsert_entity(live_b).await.unwrap();
    assert!(store
        .delete_entity(deleted_a_id, DeleteMode::Soft)
        .await
        .unwrap());

    let per_namespace_total = store
        .count_entities("stats-a", EntityFilter::default())
        .await
        .unwrap()
        + store
            .count_entities("stats-b", EntityFilter::default())
            .await
            .unwrap();
    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);

    let filter = EntityFilter {
        namespaces: namespaces.clone(),
        ..EntityFilter::default()
    };
    assert_eq!(
        store.count_entities("stats-a", filter).await.unwrap(),
        per_namespace_total
    );

    let kind_filter = EntityFilter {
        namespaces,
        kinds: vec!["document".to_string()],
        ..EntityFilter::default()
    };
    assert_eq!(
        store.count_entities("stats-a", kind_filter).await.unwrap(),
        1
    );
    assert_eq!(per_namespace_total, 2);
}

#[tokio::test]
async fn duplicate_namespace_across_chunk_boundary_is_not_double_counted() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(Arc::clone(&pool), false);

    store
        .upsert_entity(make_entity("stats-a", "concept", "live-a-1"))
        .await
        .unwrap();
    store
        .upsert_entity(make_entity("stats-a", "concept", "live-a-2"))
        .await
        .unwrap();

    let per_namespace_total = store
        .count_entities("stats-a", EntityFilter::default())
        .await
        .unwrap();
    assert_eq!(per_namespace_total, 2);

    // 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);

    let filter = EntityFilter {
        namespaces,
        ..EntityFilter::default()
    };
    assert_eq!(
        store.count_entities("stats-a", filter).await.unwrap(),
        per_namespace_total
    );
}

#[tokio::test]
async fn test_batch_upsert() {
    let store = setup_memory_store_ns("batch_ns");

    let entities: Vec<Entity> = (0..10)
        .map(|i| make_entity("batch_ns", "concept", &format!("entity_{i}")))
        .collect();

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

    let count = store
        .count_entities("batch_ns", EntityFilter::default())
        .await
        .unwrap();
    assert_eq!(count, 10);
}

/// One store, two namespaces — each query sees only its own.
#[tokio::test]
async fn test_namespace_isolation() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(Arc::clone(&pool), false);

    store
        .upsert_entity(make_entity("ns_a", "concept", "EntityA"))
        .await
        .unwrap();
    store
        .upsert_entity(make_entity("ns_b", "concept", "EntityB"))
        .await
        .unwrap();

    // Namespace is the caller's responsibility — pass it in the query.
    let count_a = store
        .count_entities("ns_a", EntityFilter::default())
        .await
        .unwrap();
    let count_b = store
        .count_entities("ns_b", EntityFilter::default())
        .await
        .unwrap();

    assert_eq!(count_a, 1);
    assert_eq!(count_b, 1);

    let page_a = store
        .query_entities("ns_a", EntityFilter::default(), PageRequest::default())
        .await
        .unwrap();
    assert_eq!(page_a.items[0].name, "EntityA");

    let page_b = store
        .query_entities("ns_b", EntityFilter::default(), PageRequest::default())
        .await
        .unwrap();
    assert_eq!(page_b.items[0].name, "EntityB");
}

#[tokio::test]
async fn test_query_by_tags() {
    let store = setup_memory_store_ns("tags_ns");

    let mut e1 = make_entity("tags_ns", "concept", "Tagged1");
    e1.tags = vec!["rust".to_string(), "systems".to_string()];
    let mut e2 = make_entity("tags_ns", "concept", "Tagged2");
    e2.tags = vec!["python".to_string(), "ml".to_string()];
    let mut e3 = make_entity("tags_ns", "concept", "Tagged3");
    e3.tags = vec!["rust".to_string(), "ml".to_string()];

    store.upsert_entity(e1).await.unwrap();
    store.upsert_entity(e2).await.unwrap();
    store.upsert_entity(e3).await.unwrap();

    // Filter by "rust" tag — should match Tagged1 and Tagged3
    let result = store
        .query_entities(
            "tags_ns",
            EntityFilter {
                tags_any: vec!["rust".to_string()],
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(result.items.len(), 2);
    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(names.contains(&"Tagged1"));
    assert!(names.contains(&"Tagged3"));
    assert!(!names.contains(&"Tagged2"));

    // Filter by "ml" tag — should match Tagged2 and Tagged3
    let result = store
        .query_entities(
            "tags_ns",
            EntityFilter {
                tags_any: vec!["ml".to_string()],
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(result.items.len(), 2);

    // Filter by both "rust" and "python" (union) — should match all three
    let result = store
        .query_entities(
            "tags_ns",
            EntityFilter {
                tags_any: vec!["rust".to_string(), "python".to_string()],
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(result.items.len(), 3);
}

#[tokio::test]
async fn test_query_by_ids() {
    let store = setup_memory_store_ns("ns1");

    let e1 = make_entity("ns1", "concept", "E1");
    let e2 = make_entity("ns1", "concept", "E2");
    let e3 = make_entity("ns1", "concept", "E3");
    let ids = vec![e1.id, e3.id];

    store.upsert_entity(e1).await.unwrap();
    store.upsert_entity(e2).await.unwrap();
    store.upsert_entity(e3).await.unwrap();

    let result = store
        .query_entities(
            "ns1",
            EntityFilter {
                ids,
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();
    assert_eq!(result.items.len(), 2);
    let names: Vec<&str> = result.items.iter().map(|e| e.name.as_str()).collect();
    assert!(names.contains(&"E1"));
    assert!(names.contains(&"E3"));
    assert!(!names.contains(&"E2"));
}

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

    let entity =
        Entity::new("default", "document", "ResearchPaper").with_entity_type(Some("paper"));
    let id = entity.id;

    store.upsert_entity(entity).await.unwrap();

    let fetched = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(fetched.entity_type, Some("paper".to_string()));
    assert_eq!(fetched.kind, "document");
    assert_eq!(fetched.name, "ResearchPaper");
}

#[tokio::test]
async fn insert_entity_if_absent_preserves_the_winning_row() {
    let store = setup_memory_store();
    let winner = make_entity("default", "concept", "winner");
    let mut loser = winner.clone();
    loser.name = "loser".to_string();
    loser.description = Some("must not replace the winner".to_string());

    assert!(store.insert_entity_if_absent(winner.clone()).await.unwrap());
    assert!(!store.insert_entity_if_absent(loser).await.unwrap());

    let persisted = store.get_entity(winner.id).await.unwrap().unwrap();
    assert_eq!(persisted.name, "winner");
    assert_eq!(persisted.description, None);
}

#[tokio::test]
async fn test_query_by_kind_and_entity_type() {
    let store = setup_memory_store_ns("et_ns");

    let typed = Entity::new("et_ns", "person", "Researcher").with_entity_type(Some("researcher"));
    let untyped = make_entity("et_ns", "person", "Generic");

    store.upsert_entity(typed).await.unwrap();
    store.upsert_entity(untyped).await.unwrap();

    let result = store
        .query_entities(
            "et_ns",
            EntityFilter {
                entity_types: vec!["researcher".to_string()],
                ..Default::default()
            },
            PageRequest::default(),
        )
        .await
        .unwrap();

    assert_eq!(result.items.len(), 1);
    assert_eq!(result.items[0].name, "Researcher");
    assert_eq!(result.items[0].entity_type, Some("researcher".to_string()));
}

#[test]
fn legacy_entity_type_fallback_seeks_both_type_indexes() {
    let mut conn = rusqlite::Connection::open_in_memory().unwrap();
    run_migrations(&mut conn).unwrap();
    for filter in [
        EntityFilter {
            kinds: vec!["concept".into()],
            entity_types: vec!["algorithm".into()],
            legacy_entity_type_fallback: true,
            ..Default::default()
        },
        EntityFilter {
            entity_types_by_kind: [("concept".into(), vec!["algorithm".into()])].into(),
            legacy_entity_type_fallback: true,
            ..Default::default()
        },
    ] {
        for ordered in [false, true] {
            let (where_sql, mut params) = build_entity_where("local", &filter);
            let query = if ordered {
                params.push(Box::new(2_i64));
                params.push(Box::new(1_i64));
                let limit_idx = params.len() - 1;
                let offset_idx = params.len();
                format!(
                    "SELECT {ENTITY_SELECT_COLUMNS} FROM entities{where_sql} \
                     ORDER BY created_at DESC, id DESC LIMIT ?{limit_idx} OFFSET ?{offset_idx}"
                )
            } else {
                format!("SELECT COUNT(*) FROM entities{where_sql}")
            };
            let mut stmt = conn
                .prepare(&format!("EXPLAIN QUERY PLAN {query}"))
                .unwrap();
            let param_refs: Vec<&dyn rusqlite::types::ToSql> =
                params.iter().map(|param| param.as_ref()).collect();
            let details: Vec<String> = stmt
                .query_map(param_refs.as_slice(), |row| row.get(3))
                .unwrap()
                .collect::<Result<_, _>>()
                .unwrap();
            assert!(
                details.iter().any(|detail| {
                    detail.contains("SEARCH entities")
                        && detail.contains("idx_entities_kind_entity_type")
                        && detail.contains("namespace=? AND kind=? AND entity_type=?")
                }),
                "typed rows must retain the third-column seek: {query}; {details:?}"
            );
            assert!(
                details.iter().any(|detail| {
                    detail.contains("SEARCH entities")
                        && detail.contains("idx_entities_legacy_type")
                        && detail.contains("namespace=? AND kind=? AND <expr>=?")
                }),
                "legacy rows must seek the partial expression index: {query}; {details:?}"
            );
        }
    }
}

#[tokio::test]
async fn legacy_entity_type_union_preserves_offset_seek_and_kind_filters() {
    let store = setup_memory_store();
    let mut legacy = Entity::new("local", "concept", "Alpha-one")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    legacy.id = Uuid::from_u128(1);
    legacy.created_at = 500;
    let mut typed = Entity::new("local", "concept", "Alpha").with_entity_type(Some("algorithm"));
    typed.id = Uuid::from_u128(2);
    typed.created_at = 100;
    let mut team = Entity::new("team", "concept", "Alpha-team").with_entity_type(Some("algorithm"));
    team.id = Uuid::from_u128(5);
    team.created_at = 600;
    let mut legacy_tie = Entity::new("local", "concept", "Alpha-three")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    legacy_tie.id = Uuid::from_u128(3);
    legacy_tie.created_at = 400;
    let mut typed_tie = Entity::new("local", "concept", "Alpha-four")
        .with_entity_type(Some("algorithm"))
        .with_properties(serde_json::json!({"type": "technique"}));
    typed_tie.id = Uuid::from_u128(4);
    typed_tie.created_at = 400;
    for entity in [&legacy, &typed, &team, &legacy_tie, &typed_tie] {
        store.upsert_entity(entity.clone()).await.unwrap();
    }
    legacy.description = Some("Revised legacy description".into());
    store.upsert_entity(legacy.clone()).await.unwrap();
    for description in ["First typed revision", "Second typed revision"] {
        typed.description = Some(description.into());
        store.upsert_entity(typed.clone()).await.unwrap();
    }
    let mut deleted = Entity::new("local", "concept", "Alpha-deleted")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    deleted.deleted_at = Some(deleted.created_at);
    for entity in [
        deleted,
        Entity::new("local", "concept", "Alpha-overridden")
            .with_entity_type(Some("technique"))
            .with_properties(serde_json::json!({"type": "algorithm"})),
        Entity::new("local", "document", "Alpha-document")
            .with_properties(serde_json::json!({"type": "algorithm"})),
        Entity::new("foreign", "concept", "Alpha-foreign").with_entity_type(Some("algorithm")),
    ] {
        store.upsert_entity(entity).await.unwrap();
    }
    let filter = EntityFilter {
        namespaces: vec!["local".into(), "team".into()],
        kinds: vec!["concept".into()],
        entity_types: vec!["algorithm".into()],
        legacy_entity_type_fallback: true,
        name_prefix: Some("Alpha".into()),
        ..Default::default()
    };
    let mut grouped = filter.clone();
    grouped.kinds.clear();
    grouped.entity_types.clear();
    grouped.entity_types_by_kind = [("concept".into(), vec!["algorithm".into()])].into();
    for filter in [filter, grouped] {
        assert_eq!(
            store
                .count_entities("unused", filter.clone())
                .await
                .unwrap(),
            5
        );
        let mut offset_rows = Vec::new();
        for offset in 0..5 {
            let page = store
                .query_entities("unused", filter.clone(), PageRequest { limit: 1, offset })
                .await
                .unwrap();
            assert_eq!(page.total, Some(5));
            assert_eq!(page.items.len(), 1);
            offset_rows.push((page.items[0].id, page.items[0].version));
        }
        assert_eq!(
            offset_rows,
            vec![
                (typed.id, 3),
                (team.id, 1),
                (legacy.id, 2),
                (typed_tie.id, 1),
                (legacy_tie.id, 1),
            ]
        );
        let first = store
            .query_entities_after("unused", filter.clone(), None, 2)
            .await
            .unwrap();
        assert_eq!(
            first
                .items
                .iter()
                .map(|entity| (entity.id, entity.version))
                .collect::<Vec<_>>(),
            vec![(legacy.id, 2), (typed.id, 3)]
        );
        assert_eq!(
            first.next_after,
            Some(SeekCursor {
                sequence: 2,
                id: typed.id,
            })
        );
        let second = store
            .query_entities_after("unused", filter.clone(), first.next_after, 2)
            .await
            .unwrap();
        assert_eq!(
            second
                .items
                .iter()
                .map(|entity| (entity.id, entity.version))
                .collect::<Vec<_>>(),
            vec![(team.id, 1), (legacy_tie.id, 1)]
        );
        assert_eq!(
            second.next_after,
            Some(SeekCursor {
                sequence: 4,
                id: legacy_tie.id,
            })
        );
        let third = store
            .query_entities_after("unused", filter, second.next_after, 2)
            .await
            .unwrap();
        assert_eq!(
            third
                .items
                .iter()
                .map(|entity| (entity.id, entity.version))
                .collect::<Vec<_>>(),
            vec![(typed_tie.id, 1)]
        );
        assert!(third.next_after.is_none());
    }
}

#[tokio::test]
async fn legacy_entity_type_filters_skip_invalid_json_and_json5() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(Arc::clone(&pool), false);
    let invalid_properties = [
        "not-json",
        r#"{"type":"algorithm""#,
        r#"{type:'algorithm'}"#,
        r#"{"type":"algorithm",}"#,
    ];
    for (index, properties) in invalid_properties.iter().enumerate() {
        let mut entity = Entity::new("local", "concept", "Invalid legacy properties");
        entity.id = Uuid::from_u128(index as u128 + 1);
        store.upsert_entity(entity.clone()).await.unwrap();
        // Typed writes cannot represent malformed JSON or JSON5-only text.
        pool.writer()
            .unwrap()
            .conn()
            .execute(
                "UPDATE entities SET properties = ?1, version = version + 1 WHERE id = ?2",
                rusqlite::params![properties, entity.id.to_string()],
            )
            .unwrap();
    }
    let mut legacy = Entity::new("local", "concept", "Legacy")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    legacy.id = Uuid::from_u128(10);
    legacy.created_at = 10;
    let mut typed = Entity::new("local", "concept", "Typed")
        .with_entity_type(Some("algorithm"))
        .with_properties(serde_json::json!({"type": "technique"}));
    typed.id = Uuid::from_u128(11);
    typed.created_at = 20;
    let overridden = Entity::new("local", "concept", "Overridden")
        .with_entity_type(Some("technique"))
        .with_properties(serde_json::json!({"type": "algorithm"}));
    for entity in [&legacy, &typed, &overridden] {
        store.upsert_entity(entity.clone()).await.unwrap();
    }

    for indexed in [true, false] {
        if !indexed {
            pool.writer()
                .unwrap()
                .conn()
                .execute_batch("DROP INDEX idx_entities_legacy_type")
                .unwrap();
        }
        for filter in [
            EntityFilter {
                kinds: vec!["concept".into()],
                entity_types: vec!["algorithm".into()],
                legacy_entity_type_fallback: true,
                ..Default::default()
            },
            EntityFilter {
                entity_types_by_kind: [("concept".into(), vec!["algorithm".into()])].into(),
                legacy_entity_type_fallback: true,
                ..Default::default()
            },
        ] {
            assert_eq!(
                store.count_entities("local", filter.clone()).await.unwrap(),
                2,
                "indexed={indexed}"
            );
            for (offset, expected_id) in (0_u64..).zip([typed.id, legacy.id]) {
                let page = store
                    .query_entities("local", filter.clone(), PageRequest { offset, limit: 1 })
                    .await
                    .unwrap();
                assert_eq!(page.total, Some(2), "indexed={indexed}");
                assert_eq!(page.items.len(), 1, "indexed={indexed}");
                assert_eq!(page.items[0].id, expected_id, "indexed={indexed}");
            }
            let first = store
                .query_entities_after("local", filter.clone(), None, 1)
                .await
                .unwrap();
            assert_eq!(first.items.len(), 1, "indexed={indexed}");
            assert_eq!(first.items[0].id, legacy.id, "indexed={indexed}");
            assert_eq!(
                first.next_after,
                Some(SeekCursor {
                    sequence: 5,
                    id: legacy.id,
                }),
                "indexed={indexed}"
            );
            let second = store
                .query_entities_after("local", filter, first.next_after, 1)
                .await
                .unwrap();
            assert_eq!(second.items.len(), 1, "indexed={indexed}");
            assert_eq!(second.items[0].id, typed.id, "indexed={indexed}");
            assert!(second.next_after.is_none(), "indexed={indexed}");
        }
    }

    for (index, properties) in invalid_properties.iter().enumerate() {
        let stored: (String, i64) = pool
            .writer()
            .unwrap()
            .conn()
            .query_row(
                "SELECT properties, version FROM entities WHERE id = ?1",
                [Uuid::from_u128(index as u128 + 1).to_string()],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(stored, (properties.to_string(), 2));
    }
}

#[tokio::test]
async fn test_legacy_entity_type_filter_is_opt_in_and_preserves_column_precedence() {
    let store = setup_memory_store();
    let legacy = Entity::new("local", "concept", "Legacy")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    let typed = Entity::new("local", "concept", "Typed")
        .with_entity_type(Some("algorithm"))
        .with_properties(serde_json::json!({"type": "technique"}));
    let overridden = Entity::new("local", "concept", "Overridden")
        .with_entity_type(Some("technique"))
        .with_properties(serde_json::json!({"type": "algorithm"}));
    for entity in [&legacy, &typed, &overridden] {
        store.upsert_entity(entity.clone()).await.unwrap();
    }
    for properties in [
        serde_json::json!({}),
        serde_json::json!({"type": null}),
        serde_json::json!({"type": 7}),
        serde_json::json!({"type": true}),
        serde_json::json!({"type": ["algorithm"]}),
        serde_json::json!({"type": {"name": "algorithm"}}),
        serde_json::json!({"type": "Algorithm"}),
    ] {
        store
            .upsert_entity(Entity::new("local", "concept", "Unmatched").with_properties(properties))
            .await
            .unwrap();
    }
    store
        .upsert_entity(Entity::new("local", "concept", "NoProperties"))
        .await
        .unwrap();
    store
        .upsert_entity(
            Entity::new("foreign", "concept", "Foreign")
                .with_properties(serde_json::json!({"type": "algorithm"})),
        )
        .await
        .unwrap();
    let mut deleted = Entity::new("local", "concept", "Deleted")
        .with_properties(serde_json::json!({"type": "algorithm"}));
    deleted.deleted_at = Some(deleted.created_at);
    store.upsert_entity(deleted).await.unwrap();

    let exact = EntityFilter {
        entity_types: vec!["algorithm".into()],
        ..Default::default()
    };
    let control = store
        .query_entities("local", exact.clone(), PageRequest::default())
        .await
        .unwrap();
    assert_eq!(control.items.len(), 1);
    assert_eq!(control.items[0].id, typed.id);
    assert_eq!(control.total, Some(1));

    let fallback = EntityFilter {
        legacy_entity_type_fallback: true,
        ..exact
    };
    let non_string_filter = EntityFilter {
        entity_types: vec![
            "7".into(),
            "1".into(),
            "[\"algorithm\"]".into(),
            "{\"name\":\"algorithm\"}".into(),
        ],
        ..fallback.clone()
    };
    assert_eq!(
        store
            .count_entities("local", non_string_filter)
            .await
            .unwrap(),
        0
    );
    assert_eq!(
        store
            .count_entities("local", fallback.clone())
            .await
            .unwrap(),
        2
    );
    let mut page_ids = Vec::new();
    for offset in 0..2 {
        let page = store
            .query_entities("local", fallback.clone(), PageRequest { offset, limit: 1 })
            .await
            .unwrap();
        assert_eq!(page.total, Some(2));
        assert_eq!(page.items.len(), 1);
        page_ids.push(page.items[0].id);
    }
    page_ids.sort_unstable();
    let mut expected = vec![legacy.id, typed.id];
    expected.sort_unstable();
    assert_eq!(page_ids, expected);

    let first = store
        .query_entities_after("local", fallback.clone(), None, 1)
        .await
        .unwrap();
    assert_eq!(first.items.len(), 1);
    assert_eq!(first.items[0].id, legacy.id);
    let second = store
        .query_entities_after("local", fallback.clone(), first.next_after, 1)
        .await
        .unwrap();
    assert_eq!(second.items.len(), 1);
    assert_eq!(second.items[0].id, typed.id);
    assert!(second.next_after.is_none());
    let unchanged = store.get_entity(legacy.id).await.unwrap().unwrap();
    assert!(unchanged.entity_type.is_none());
    assert_eq!(unchanged.properties, legacy.properties);

    let mut cleared = overridden;
    cleared.entity_type = None;
    store.upsert_entity(cleared).await.unwrap();
    assert_eq!(store.count_entities("local", fallback).await.unwrap(), 3);
}

/// UUID is globally unique (id TEXT PRIMARY KEY). Upserting the same UUID in a
/// different namespace overwrites the row (INSERT OR REPLACE). get_entity by ID
/// returns whichever namespace currently owns that UUID.
#[tokio::test]
async fn test_same_id_upsert_replaces_row() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(Arc::clone(&pool), false);

    let shared_id = Uuid::new_v4();
    let now = chrono::Utc::now().timestamp_micros();

    let entity_a = Entity {
        id: shared_id,
        namespace: "ns_a".to_string(),
        kind: "concept".to_string(),
        entity_type: None,
        name: "SharedInA".to_string(),
        description: None,
        properties: None,
        tags: Vec::new(),
        created_at: now,
        updated_at: now,
        deleted_at: None,
        merged_into: None,
        merge_event_id: None,
        version: 1,
        content_ref: None,
    };
    store.upsert_entity(entity_a).await.unwrap();

    // At this point the row is in ns_a.
    let fetched = store.get_entity(shared_id).await.unwrap().unwrap();
    assert_eq!(fetched.namespace, "ns_a");
    assert_eq!(fetched.name, "SharedInA");

    // Upsert same UUID into ns_b — INSERT OR REPLACE replaces the row.
    let entity_b = Entity {
        id: shared_id,
        namespace: "ns_b".to_string(),
        kind: "concept".to_string(),
        entity_type: None,
        name: "SharedInB".to_string(),
        description: None,
        properties: None,
        tags: Vec::new(),
        created_at: now,
        updated_at: now,
        deleted_at: None,
        merged_into: None,
        merge_event_id: None,
        version: 1,
        content_ref: None,
    };
    store.upsert_entity(entity_b).await.unwrap();

    // Now the row is in ns_b — get_entity returns ns_b regardless of which namespace
    // you query from (namespace is caller's responsibility).
    let fetched = store.get_entity(shared_id).await.unwrap().unwrap();
    assert_eq!(fetched.namespace, "ns_b");
    assert_eq!(fetched.name, "SharedInB");

    // ns_a now has 0 entities; ns_b has 1.
    let count_a = store
        .count_entities("ns_a", EntityFilter::default())
        .await
        .unwrap();
    let count_b = store
        .count_entities("ns_b", EntityFilter::default())
        .await
        .unwrap();
    assert_eq!(count_a, 0);
    assert_eq!(count_b, 1);
}

/// 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_ns("ns1");
    store
        .upsert_entity(make_entity("ns1", "concept", "Alpha"))
        .await
        .unwrap();

    let result = store
        .query_entities(
            "ns1",
            EntityFilter::default(),
            PageRequest {
                offset: (i64::MAX as u64) + 1,
                limit: 10,
            },
        )
        .await;

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

/// #1671: offset pages over entities whose primary sort key (`created_at`)
/// repeats must form a deterministic total order — a full paged sweep returns
/// every entity exactly once (no duplicates, no misses across page boundaries).
#[tokio::test]
async fn query_entities_offset_sweep_covers_equal_created_at_exactly_once() {
    let store = setup_memory_store_ns("ns1");
    let created_at = 1_750_000_000_000_000_i64;
    let mut expected_ids = Vec::new();
    for index in 0..211 {
        let mut entity = make_entity("ns1", "concept", &format!("sweep-{index:03}"));
        entity.created_at = created_at;
        expected_ids.push(entity.id);
        store.upsert_entity(entity).await.unwrap();
    }
    expected_ids.sort_unstable_by(|a, b| b.cmp(a));

    let mut actual_ids = Vec::new();
    let page_size = 37_u32;
    let mut offset = 0_u64;
    loop {
        let page = store
            .query_entities(
                "ns1",
                EntityFilter::default(),
                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(|entity| entity.id));
    }

    assert_eq!(
        actual_ids, expected_ids,
        "sweep must cover every entity exactly once"
    );
}

/// #1671: the name-prefix branch ranks exact matches first, then orders by
/// `created_at DESC` — that branch needs the same `id` tiebreak so offset
/// pages over equal `created_at` rows are a deterministic total order.
///
/// One row is named exactly `Alpha` (the prefix) so the sweep also exercises
/// the exact-match `CASE WHEN LOWER(name) = ?` branch under ties: it must
/// sort ahead of every prefix match.
#[tokio::test]
async fn query_entities_name_prefix_offset_sweep_covers_equal_created_at_exactly_once() {
    let store = setup_memory_store_ns("ns1");
    let created_at = 1_750_000_000_000_000_i64;
    let mut expected_ids = Vec::new();
    for index in 0..113 {
        let mut entity = make_entity("ns1", "concept", &format!("Alpha-{index:03}"));
        entity.created_at = created_at;
        expected_ids.push(entity.id);
        store.upsert_entity(entity).await.unwrap();
    }
    // An exact match for the prefix must rank first despite sharing the same
    // `created_at` as every prefix-matching row.
    let mut exact_entity = make_entity("ns1", "concept", "Alpha");
    exact_entity.created_at = created_at;
    let exact_id = exact_entity.id;
    store.upsert_entity(exact_entity).await.unwrap();
    expected_ids.sort_unstable_by(|a, b| b.cmp(a));
    expected_ids.insert(0, exact_id);

    let mut actual_ids = Vec::new();
    let page_size = 23_u32;
    let mut offset = 0_u64;
    loop {
        let page = store
            .query_entities(
                "ns1",
                EntityFilter {
                    name_prefix: Some("Alpha".to_string()),
                    ..Default::default()
                },
                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(|entity| entity.id));
    }

    assert_eq!(
        actual_ids.first(),
        Some(&exact_id),
        "the exact-match row must sort into the exact-match-first group, ahead of prefix matches"
    );
    assert_eq!(
        actual_ids, expected_ids,
        "name-prefix sweep must cover every entity exactly once"
    );
}

// =============================================================================
// ADR-067 slice 1: WriterTask-routed `upsert_entities` (KHIVE_WRITE_QUEUE=1)
// =============================================================================

/// Flag-ON mechanism: with `KHIVE_WRITE_QUEUE=1`, `upsert_entities` routes
/// through the WriterTask channel instead of the pool-mutex path, and the
/// `BatchWriteSummary` fields (attempted/affected/failed/first_error) survive
/// the trip through the type-erased channel intact, and both rows are
/// actually committed and independently readable back through the store.
///
/// 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_entities_routes_through_writer_task_when_flag_enabled() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_entities.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();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
    }

    let store = SqlEntityStore::new(Arc::clone(&pool), true);

    let e1 = make_entity("default", "concept", "LoRA");
    let e2 = make_entity("default", "concept", "QLoRA");
    let id1 = e1.id;
    let id2 = e2.id;

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

    let fetched1 = store.get_entity(id1).await.unwrap();
    assert!(
        fetched1.is_some(),
        "entity 1 must be committed and readable"
    );
    assert_eq!(fetched1.unwrap().name, "LoRA");

    let fetched2 = store.get_entity(id2).await.unwrap();
    assert!(
        fetched2.is_some(),
        "entity 2 must be committed and readable"
    );
    assert_eq!(fetched2.unwrap().name, "QLoRA");
}

/// Flag-OFF regression (explicit): with the flag at its default (off), the
/// store never spawns a writer task, and `upsert_entities` still returns a
/// correct `BatchWriteSummary` via the legacy pool-mutex path — the same
/// shape `test_batch_upsert` above already covers, restated here to
/// document the flag-off/flag-on pairing for ADR-067 slice 1.
#[tokio::test]
async fn upsert_entities_legacy_path_unchanged_when_flag_is_off() {
    let store = setup_memory_store();

    let e1 = make_entity("default", "concept", "LoRA");
    let e2 = make_entity("default", "concept", "QLoRA");

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

/// ADR-067 Component A's whole point is ONE writer owning the write
/// connection for a DB file — a per-store writer task would let concurrent
/// stores over the same pool spawn independent writer connections that
/// contend with each other at `BEGIN IMMEDIATE`, so the migrated path would
/// race itself instead of eliminating write contention. Constructing
/// several `SqlEntityStore`s over the SAME pool with the flag on must spawn
/// the writer task exactly once; every store resolves to a clone of the one
/// pool-owned handle (`ConnectionPool::writer_task_handle`).
///
/// Constructed via a `PoolConfig` literal (`write_queue_enabled: Some(true)`), not
/// the `KHIVE_WRITE_QUEUE` env var — see the sibling
/// `upsert_entities_routes_through_writer_task_when_flag_enabled` test's doc
/// comment for the race this avoids.
#[tokio::test]
async fn multiple_stores_over_one_pool_share_a_single_writer_task() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_shared_writer.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();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
    }

    // Three independent stores over the same pool, each resolving the
    // write-queue flag on construction and asking the pool for its writer
    // task — none of them must trigger a second spawn.
    let _store1 = SqlEntityStore::new(Arc::clone(&pool), true);
    let _store2 = SqlEntityStore::new(Arc::clone(&pool), true);
    let _store3 = SqlEntityStore::new(Arc::clone(&pool), true);

    assert_eq!(
        pool.writer_task_spawn_count(),
        1,
        "N stores constructed over one pool must spawn the writer task \
         exactly once — one writer task per pool (per DB file), not one \
         per store"
    );
}

/// Full-slice single-writer guarantee (ADR-067 Component A, Fork C slice 2):
/// with every MIGRATE-listed write path routed through the writer task, drive
/// CONCURRENT writes across entity, note, and graph stores (entries 2/3/6 —
/// er, 1/2/3) plus `SqlBridge`'s unmigrated `writer()` (entry 8, still routes
/// its self-contained `execute_batch` through the task per entry 10) over ONE
/// pool. Asserts exactly one writer task is ever spawned and every write
/// actually lands, proving the migrated paths coexist over a single DB file
/// without contending at `BEGIN IMMEDIATE` or racing the writer task's own
/// spawn-once guarantee.
///
/// Entry 9 (`SqlBridge::begin_tx`, formerly a genuine design fork on its own
/// standalone connection) is retired as of ADR-099 D5 — `begin_tx` and
/// `SqlTransaction` no longer exist on `SqlAccess` (their one production
/// caller, session-mirror ingest, converted to `atomic_unit`), so this test
/// no longer exercises it.
///
/// Constructed via a `PoolConfig` literal (`write_queue_enabled: Some(true)`), not
/// the `KHIVE_WRITE_QUEUE` env var — see the sibling
/// `upsert_entities_routes_through_writer_task_when_flag_enabled` test's doc
/// comment for the race this avoids.
#[tokio::test]
async fn concurrent_writes_across_all_migrated_stores_share_one_writer_task() {
    use crate::stores::graph::SqlGraphStore;
    use crate::stores::note::SqlNoteStore;
    use khive_storage::note::Note;
    use khive_storage::types::{Edge, SqlStatement, SqlValue};
    use khive_storage::{GraphStore as _, NoteStore as _, SqlAccess as _};
    use khive_types::EdgeRelation;

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_all_paths_shared_writer.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();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
        crate::stores::note::ensure_notes_schema(writer.conn()).unwrap();
        crate::stores::graph::ensure_graph_schema(writer.conn()).unwrap();
    }

    let entity_store = Arc::new(SqlEntityStore::new(Arc::clone(&pool), true));
    let note_store = Arc::new(SqlNoteStore::new(Arc::clone(&pool), true));
    let graph_store = Arc::new(SqlGraphStore::new_scoped(
        Arc::clone(&pool),
        true,
        "default",
    ));
    let bridge = crate::sql_bridge::SqlBridge::new(Arc::clone(&pool), true);

    assert_eq!(
        pool.writer_task_spawn_count(),
        1,
        "entity + note + graph stores plus SqlBridge over one pool must still \
         share exactly one writer task"
    );

    let entity = make_entity("default", "concept", "WriterTaskConcurrency");
    let entity_id = entity.id;

    let note = Note::new("default", "observation", "concurrent writer task note");
    let note_id = note.id;

    let edge_src = Uuid::new_v4();
    let edge_tgt = Uuid::new_v4();
    let now = chrono::Utc::now();
    let edge = Edge {
        id: Uuid::new_v4().into(),
        namespace: "default".to_string(),
        source_id: edge_src,
        target_id: edge_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;

    // Entry 10 (SqliteWriter::execute_batch, migrated): a raw INSERT issued
    // through SqlBridge's file-backed writer() handle.
    let batch_row_id = Uuid::new_v4();
    let batch_src = Uuid::new_v4();
    let batch_tgt = Uuid::new_v4();
    let now_micros = chrono::Utc::now().timestamp_micros();
    let insert_stmt = SqlStatement {
        sql: "INSERT INTO graph_edges (namespace, id, source_id, target_id, relation, \
              weight, created_at, updated_at, deleted_at, metadata, target_backend) \
              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, NULL, NULL, NULL)"
            .to_string(),
        params: vec![
            SqlValue::Text("default".to_string()),
            SqlValue::Text(batch_row_id.to_string()),
            SqlValue::Text(batch_src.to_string()),
            SqlValue::Text(batch_tgt.to_string()),
            SqlValue::Text("extends".to_string()),
            SqlValue::Float(0.5),
            SqlValue::Integer(now_micros),
            SqlValue::Integer(now_micros),
        ],
        label: Some("test_execute_batch".to_string()),
    };

    let entity_fut = {
        let entity_store = Arc::clone(&entity_store);
        async move { entity_store.upsert_entity(entity).await }
    };
    let note_fut = {
        let note_store = Arc::clone(&note_store);
        async move { note_store.upsert_note(note).await }
    };
    let edge_fut = {
        let graph_store = Arc::clone(&graph_store);
        async move { graph_store.upsert_edge(edge).await }
    };
    let batch_fut = async {
        let mut writer = bridge.writer().await.unwrap();
        writer.execute_batch(vec![insert_stmt]).await
    };

    let (entity_res, note_res, edge_res, batch_res) =
        tokio::join!(entity_fut, note_fut, edge_fut, batch_fut);

    entity_res.unwrap();
    note_res.unwrap();
    edge_res.unwrap();
    batch_res.unwrap();

    assert!(entity_store.get_entity(entity_id).await.unwrap().is_some());
    assert!(note_store.get_note(note_id).await.unwrap().is_some());
    assert!(graph_store.get_edge(edge_id).await.unwrap().is_some());

    assert_eq!(
        pool.writer_task_spawn_count(),
        1,
        "concurrent writes across every migrated path must not trigger a \
         second writer task spawn"
    );
}

// =============================================================================
// Fork C slice 2: single-row `upsert_entity` (via `with_writer`) routes
// through the WriterTask when the flag is on
// =============================================================================

/// Proves `upsert_entity` (the SINGLE-row path, going through the
/// `with_writer` helper directly rather than the already-migrated
/// `upsert_entities` batch path) is actually enqueued on the pool's shared
/// `WriterTaskHandle` channel when `KHIVE_WRITE_QUEUE=1` — not inferred from
/// wall-clock contention (real SQLite file-level locking would serialize a
/// competing writer connection against an in-flight transaction regardless
/// of which Rust-level path issued it, which makes elapsed-time alone a
/// vacuous signal here), but read directly off `WriterTaskHandle::queue_depth`,
/// the live gauge over the exact `mpsc` channel `with_writer`'s writer-task
/// branch must call `send` on.
///
/// Technique: submit an occupier request directly on the SAME
/// `WriterTaskHandle` the store resolved at construction. The occupier signals
/// a oneshot once it is actually running inside the writer task's single
/// drain slot, then parks on a second oneshot (`blocking_recv`, valid here
/// because this closure runs inside the writer task's `spawn_blocking`) until
/// the test releases it — deterministically holding that one drain slot open
/// for as long as the test needs, no sleep/timing race. While the slot is
/// held, call `store.upsert_entity` on a separate task and poll
/// `queue_depth()`: a version that genuinely routes this call through
/// `writer_task.send(..)` must show `queue_depth() >= 1` (the request sitting
/// in the channel behind the still-running occupier) before the occupier is
/// released; a version that left `upsert_entity`/`with_writer` on the legacy
/// `pool.try_writer()` path never touches this channel at all, so
/// `queue_depth()` would stay `0` for the full poll window — this is what
/// makes the test non-vacuous (verified: reverting `with_writer` to always
/// take the legacy branch makes this assertion fail, see khive-db PR history
/// for Fork C slice 2).
///
/// Constructed via a `PoolConfig` literal (`write_queue_enabled: Some(true)`), not
/// the `KHIVE_WRITE_QUEUE` env var — see
/// `upsert_entities_routes_through_writer_task_when_flag_enabled`'s doc
/// comment for the race this avoids.
#[tokio::test]
async fn upsert_entity_routes_through_writer_task_when_flag_enabled() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("write_queue_entity_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();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
    }

    let store = Arc::new(SqlEntityStore::new(Arc::clone(&pool), true));

    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) = oneshot::channel::<()>();
    let (release_tx, release_rx) = 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 entity = make_entity("default", "concept", "RoPE");
    let entity_id = entity.id;

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

    let mut saw_enqueued = false;
    for _ in 0..100 {
        if writer_task.queue_depth() >= 1 {
            saw_enqueued = true;
            break;
        }
        tokio::time::sleep(Duration::from_millis(5)).await;
    }
    assert!(
        saw_enqueued,
        "upsert_entity'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_entity must succeed once unblocked");

    let fetched = store.get_entity(entity_id).await.unwrap();
    assert!(
        fetched.is_some(),
        "entity must be committed and readable after queuing behind the occupier"
    );
    assert_eq!(fetched.unwrap().name, "RoPE");
}

// ── attachment-backed content_ref projection ───────────────────────────────

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

    let digest = "a".repeat(64);
    let entity = Entity::new("default", "document", "SourcePdf");
    let id = entity.id;

    store
        .upsert_entity_with_attachments(entity, vec![content_attachment(id, &digest)])
        .await
        .unwrap();

    let fetched = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(fetched.content_ref, Some(digest));
}

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

    let entity = Entity::new("default", "concept", "NoBlob");
    let id = entity.id;
    store.upsert_entity(entity).await.unwrap();

    let fetched = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(fetched.content_ref, None);
}

#[tokio::test]
async fn test_content_ref_survives_query_entities() {
    let store = setup_memory_store_ns("blob_ns");
    let digest = "b".repeat(64);
    let entity = Entity::new("blob_ns", "document", "QueriedPdf");
    let id = entity.id;
    store
        .upsert_entity_with_attachments(entity, vec![content_attachment(id, &digest)])
        .await
        .unwrap();

    let page = store
        .query_entities("blob_ns", EntityFilter::default(), PageRequest::default())
        .await
        .unwrap();
    assert_eq!(page.items.len(), 1);
    assert_eq!(page.items[0].content_ref, Some(digest));
}

#[tokio::test]
async fn entity_upserts_ignore_the_response_only_content_ref_projection() {
    let store = setup_memory_store();
    let digest = "c".repeat(64);
    let mut projected = Entity::new("default", "document", "Batch1");
    projected.content_ref = Some(digest);
    let entities = vec![projected, Entity::new("default", "document", "Batch2")];
    let ids: Vec<Uuid> = entities.iter().map(|e| e.id).collect();

    let summary = store.upsert_entities(entities).await.unwrap();
    assert_eq!(summary.affected, 2);

    let with_ref = store.get_entity(ids[0]).await.unwrap().unwrap();
    assert_eq!(with_ref.content_ref, None);
    let without_ref = store.get_entity(ids[1]).await.unwrap().unwrap();
    assert_eq!(without_ref.content_ref, None);
}

#[tokio::test]
async fn entity_and_multiple_attachments_roll_back_as_one_unit() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(pool.clone(), false);
    let entity = Entity::new("default", "document", "Atomic");
    let id = entity.id;
    pool.writer()
        .unwrap()
        .conn()
        .execute_batch(
            "CREATE TRIGGER reject_second_attachment BEFORE INSERT ON attachments \
             WHEN NEW.role = 'reject' BEGIN SELECT RAISE(ABORT, 'injected'); END;",
        )
        .unwrap();
    let mut rejected = content_attachment(id, &"d".repeat(64));
    rejected.role = "reject".to_string();

    store
        .upsert_entity_with_attachments(
            entity,
            vec![content_attachment(id, &"c".repeat(64)), rejected],
        )
        .await
        .expect_err("second attachment failure must abort the unit");

    assert!(store.get_entity(id).await.unwrap().is_none());
    let count: i64 = pool
        .reader()
        .unwrap()
        .conn()
        .query_row(
            "SELECT COUNT(*) FROM attachments WHERE record_uuid = ?1",
            [id.to_string()],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(count, 0);
}

#[tokio::test]
async fn entity_soft_delete_retains_attachments_and_hard_delete_removes_them() {
    let pool = setup_pool();
    let store = SqlEntityStore::new(pool.clone(), false);
    let entity = Entity::new("default", "document", "Delete");
    let id = entity.id;
    store
        .upsert_entity_with_attachments(entity, vec![content_attachment(id, &"e".repeat(64))])
        .await
        .unwrap();

    assert!(store.delete_entity(id, DeleteMode::Soft).await.unwrap());
    let retained: i64 = pool
        .reader()
        .unwrap()
        .conn()
        .query_row(
            "SELECT COUNT(*) FROM attachments WHERE record_uuid = ?1",
            [id.to_string()],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(retained, 1);

    assert!(store.delete_entity(id, DeleteMode::Hard).await.unwrap());
    let removed: i64 = pool
        .reader()
        .unwrap()
        .conn()
        .query_row(
            "SELECT COUNT(*) FROM attachments WHERE record_uuid = ?1",
            [id.to_string()],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(removed, 0);
}

/// #1656: a seek-cursor walk with an active `kind` filter must return every
/// matching row, matching the count an offset-based query would return for
/// the same filter, even when matching rows are sparse relative to the
/// sequence range being walked.
#[tokio::test]
async fn cursor_kind_filter_returns_records() {
    let store = setup_memory_store_ns("ns1");

    // Insert a large run of "document" entities first, then a sparse run of
    // "concept" entities, so a naive seq-first probe window can land
    // entirely on non-matching rows.
    for i in 0..40 {
        store
            .upsert_entity(make_entity("ns1", "document", &format!("Doc{i}")))
            .await
            .unwrap();
    }
    let mut concept_ids = Vec::new();
    for i in 0..5 {
        let entity = make_entity("ns1", "concept", &format!("Concept{i}"));
        concept_ids.push(entity.id);
        store.upsert_entity(entity).await.unwrap();
    }

    let filter = EntityFilter {
        kinds: vec!["concept".to_string()],
        ..Default::default()
    };

    let mut walked_ids: Vec<Uuid> = Vec::new();
    let mut after = None;
    // A cursor that stops advancing must fail here rather than loop forever.
    for _ in 0..=concept_ids.len() {
        let page = store
            .query_entities_after("ns1", filter.clone(), after, 2)
            .await
            .unwrap();
        for entity in &page.items {
            walked_ids.push(entity.id);
        }
        after = page.next_after;
        if after.is_none() {
            break;
        }
    }
    assert!(after.is_none(), "cursor did not reach the last page");

    assert_eq!(
        walked_ids.len(),
        concept_ids.len(),
        "cursor walk with kind filter must return every matching row"
    );
    let walked_set: std::collections::HashSet<Uuid> = walked_ids.into_iter().collect();
    for id in &concept_ids {
        assert!(
            walked_set.contains(id),
            "cursor walk missing expected concept entity {id}"
        );
    }

    // Control: offset-based query for the same filter must return the same set.
    let offset_page = store
        .query_entities("ns1", filter, PageRequest::default())
        .await
        .unwrap();
    assert_eq!(offset_page.items.len(), concept_ids.len());
}

/// #1847: construction is synchronous and may happen before a Tokio runtime
/// exists. A file-backed store must refresh its missing construction-time
/// handle at write time; otherwise this batch silently takes the legacy
/// pool-writer path even though the queue is enabled.
#[test]
fn batch_write_refreshes_writer_task_after_construction_outside_runtime() {
    assert!(tokio::runtime::Handle::try_current().is_err());

    let dir = tempfile::tempdir().unwrap();
    let pool = Arc::new(
        ConnectionPool::new(PoolConfig {
            path: Some(dir.path().join("entity-late-writer-task.db")),
            write_queue_enabled: Some(true),
            ..PoolConfig::for_test()
        })
        .unwrap(),
    );
    {
        let writer = pool.writer().unwrap();
        writer
            .conn()
            .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
            .unwrap();
    }
    let store = Arc::new(SqlEntityStore::new(Arc::clone(&pool), true));

    tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async move {
            let writer_task = pool
                .writer_task_handle()
                .unwrap()
                .expect("file-backed pool must spawn its writer task inside the runtime");
            let (started_tx, started_rx) = oneshot::channel::<()>();
            let (release_tx, release_rx) = 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.unwrap();

            let write = {
                let store = Arc::clone(&store);
                tokio::spawn(async move {
                    store
                        .upsert_entities(vec![
                            make_entity("default", "concept", "late-a"),
                            make_entity("default", "concept", "late-b"),
                        ])
                        .await
                })
            };
            let mut saw_enqueued = false;
            for _ in 0..100 {
                if writer_task.queue_depth() >= 1 {
                    saw_enqueued = true;
                    break;
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }

            release_tx.send(()).unwrap();
            occupier.await.unwrap().unwrap();
            write.await.unwrap().unwrap();
            assert!(
                saw_enqueued,
                "batch write bypassed the queue after construction cached no runtime handle"
            );
        });
}

/// Both refusal arms of the pooled-reader checkout, exercised through a typed
/// store. The typed-store paths previously inverted the pair: genuine pool
/// exhaustion surfaced as a non-retryable Driver failure while a cancelled
/// request surfaced as the retryable AdmissionTimeout — so a saturated read
/// was never retried and an abandoned one was.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn pooled_entity_read_classifies_exhaustion_and_cancellation_distinctly() {
    let dir = tempfile::tempdir().unwrap();
    let config = PoolConfig {
        path: Some(dir.path().join("entity_checkout_classes.db")),
        max_readers: 1,
        checkout_timeout: Duration::from_millis(200),
        ..PoolConfig::default()
    };
    let pool = Arc::new(ConnectionPool::new(config).unwrap());
    pool.writer()
        .unwrap()
        .conn()
        .execute_batch(&format!("{ENTITIES_DDL}\n{TEST_ATTACHMENTS_DDL}"))
        .unwrap();
    // Production-shaped file-backed reads must contend on the same bounded
    // reader pool. Before ADR-165 Slice 2 they opened a fresh standalone
    // connection here, silently bypassing the sole pooled reader held below.
    let store = SqlEntityStore::new(Arc::clone(&pool), true);
    let held_reader = pool.reader().expect("hold the sole pooled reader");

    // Admission expiry with no cancellation: `checkout_timeout` runs to
    // exhaustion and must surface as the retryable AdmissionTimeout.
    let exhausted = store.get_entity(Uuid::new_v4()).await.unwrap_err();
    assert!(
        matches!(exhausted, StorageError::AdmissionTimeout { .. }),
        "pool exhaustion through a typed store must be the retryable \
         AdmissionTimeout, got {exhausted:?}"
    );

    // Cancellation before checkout: must return promptly as the non-retryable
    // Timeout, never AdmissionTimeout, and never wait out `checkout_timeout`.
    let cancel_store = SqlEntityStore::new(Arc::clone(&pool), true);
    let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
    let waiting = tokio::spawn(khive_storage::scope_request_read_cancellation(
        cancel_rx,
        async move { cancel_store.get_entity(Uuid::new_v4()).await },
    ));
    tokio::task::yield_now().await;
    cancel_tx.send(true).unwrap();
    let cancelled = tokio::time::timeout(Duration::from_millis(100), waiting)
        .await
        .expect("cancelled checkout waited for the pool checkout timeout")
        .expect("checkout task panicked")
        .unwrap_err();
    assert!(
        matches!(cancelled, StorageError::Timeout { .. }),
        "cancellation before checkout through a typed store must be the \
         non-retryable Timeout, got {cancelled:?}"
    );

    drop(held_reader);
    let reader = pool.reader_acquisition_snapshot();
    assert_eq!(reader.pooled_checkouts, 1);
    assert_eq!(reader.checkout_timeouts, 1);
    assert_eq!(reader.completed_pooled_checkouts, 1);
    assert_eq!(reader.active_pooled_checkouts, 0);
    assert_eq!(
        reader.standalone_opens, 0,
        "neither saturation nor cancellation may fall back to a standalone reader"
    );
}

#[tokio::test]
async fn issue2673_entity_versions_cover_typed_storage_writers() {
    let store = setup_memory_store();
    let entity = Entity::new("local", "concept", "Versioned");
    let id = entity.id;
    store.upsert_entity(entity.clone()).await.unwrap();
    assert_eq!(store.get_entity(id).await.unwrap().unwrap().version, 1);

    // Upsert must increment the stored counter even when passed an old clone.
    store.upsert_entity(entity.clone()).await.unwrap();
    assert_eq!(store.get_entity(id).await.unwrap().unwrap().version, 2);
    let batch = store
        .upsert_entities(vec![entity.clone(), entity.clone()])
        .await
        .unwrap();
    assert_eq!((batch.affected, batch.failed), (2, 0));
    assert_eq!(store.get_entity(id).await.unwrap().unwrap().version, 4);

    assert!(!store.insert_entity_if_absent(entity).await.unwrap());
    let before = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(
        before.version, 4,
        "losing conditional insert is not a write"
    );
    let mut changed = before.clone();
    changed.name = "CAS winner".into();
    changed.updated_at += 1;
    assert!(store
        .replace_entity_if_unchanged(changed.clone(), before.updated_at, before.deleted_at)
        .await
        .unwrap());
    let winner = store.get_entity(id).await.unwrap().unwrap();
    assert_eq!(winner.version, 5);
    assert!(!store
        .replace_entity_if_unchanged(changed, before.updated_at, before.deleted_at)
        .await
        .unwrap());
    assert_eq!(
        serde_json::to_value(store.get_entity(id).await.unwrap().unwrap()).unwrap(),
        serde_json::to_value(&winner).unwrap()
    );

    // A same-timestamp upsert still invalidates the persisted revision.
    let mut stale = winner.clone();
    stale.updated_at += 1;
    stale.name = "must not land".into();
    store.upsert_entity(winner.clone()).await.unwrap();
    assert!(!store
        .replace_entity_if_unchanged(stale, winner.updated_at, winner.deleted_at)
        .await
        .unwrap());
    assert_eq!(store.get_entity(id).await.unwrap().unwrap().version, 6);

    let current = store.get_entity(id).await.unwrap().unwrap();
    store
        .upsert_entity_with_attachments(current, vec![content_attachment(id, &"a".repeat(64))])
        .await
        .unwrap();
    assert_eq!(store.get_entity(id).await.unwrap().unwrap().version, 7);
    assert!(store.delete_entity(id, DeleteMode::Soft).await.unwrap());
    assert_eq!(
        store
            .get_entity_including_deleted(id)
            .await
            .unwrap()
            .unwrap()
            .version,
        8
    );
    assert!(!store.delete_entity(id, DeleteMode::Soft).await.unwrap());
    assert_eq!(
        store
            .get_entity_including_deleted(id)
            .await
            .unwrap()
            .unwrap()
            .version,
        8
    );

    let fresh = Entity::new("local", "concept", "Conditional");
    assert!(store.insert_entity_if_absent(fresh.clone()).await.unwrap());
    assert_eq!(
        store.get_entity(fresh.id).await.unwrap().unwrap().version,
        1
    );
}