//! Unit tests for [`super`] — chunk upsert / list / lifecycle / embedding /
//! content-pointer accessors against a tempdir-backed SQLite store.
//!
//! ## Test isolation for the connection cache
//!
//! Because the connection cache is a process-level singleton, tests that want
//! to exercise cache behaviour (same Arc, independent workspaces, circuit
//! breaker, cleanup) must call `clear_connection_cache()` at the start — or
//! be careful to use unique tempdirs that cannot collide with other tests.
//! The call is cheap (a mutex lock + HashMap clear) and harmless for tests
//! that don't need it.

use super::*;
use crate::openhuman::memory_store::chunks::types::chunk_id;
use chrono::TimeZone;
use rusqlite::params;
use tempfile::TempDir;

fn test_config() -> (TempDir, Config) {
    let tmp = TempDir::new().expect("tempdir");
    let mut cfg = Config::default();
    cfg.workspace_dir = tmp.path().to_path_buf();
    (tmp, cfg)
}

fn sample_chunk(source_id: &str, seq: u32, ts_ms: i64) -> Chunk {
    let ts = Utc.timestamp_millis_opt(ts_ms).unwrap();
    Chunk {
        id: chunk_id(SourceKind::Chat, source_id, seq, "test-content"),
        content: format!("content {source_id} {seq}"),
        metadata: Metadata {
            source_kind: SourceKind::Chat,
            source_id: source_id.to_string(),
            owner: "alice@example.com".to_string(),
            timestamp: ts,
            time_range: (ts, ts),
            tags: vec!["eng".into()],
            source_ref: Some(SourceRef::new(format!("slack://{source_id}/{seq}"))),
            path_scope: None,
        },
        token_count: 12,
        seq_in_source: seq,
        created_at: ts,
        partial_message: false,
    }
}

#[test]
fn upsert_then_get() {
    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    assert_eq!(upsert_chunks(&cfg, &[c.clone()]).unwrap(), 1);
    let got = get_chunk(&cfg, &c.id).unwrap().expect("chunk stored");
    assert_eq!(got, c);
}

#[test]
fn upsert_persists_path_scope() {
    let (_tmp, cfg) = test_config();
    let mut c = sample_chunk("notion:conn-1:page-abc", 0, 1_700_000_000_000);
    c.metadata.source_kind = SourceKind::Document;
    c.metadata.path_scope = Some("notion:conn-1".to_string());

    assert_eq!(upsert_chunks(&cfg, &[c.clone()]).unwrap(), 1);

    let got = get_chunk(&cfg, &c.id).unwrap().expect("chunk stored");
    assert_eq!(got.metadata.source_id, "notion:conn-1:page-abc");
    assert_eq!(got.metadata.path_scope.as_deref(), Some("notion:conn-1"));
}

#[test]
fn list_chunks_source_scope_filters_before_limit() {
    // Two disallowed-source chunks have NEWER timestamps (sorted first by DESC),
    // and the single allowed-source chunk is older. With a naive post-limit
    // filter and LIMIT 1 the allowed row would be starved; the before-limit gate
    // inside list_chunks must still surface it.
    let (_tmp, cfg) = test_config();
    let tag = || vec!["memory_sources".to_string(), "chat".to_string()];
    let mut bad1 = sample_chunk("slack:#secret", 0, 3_000);
    bad1.metadata.tags = tag();
    let mut bad2 = sample_chunk("slack:#secret", 1, 2_000);
    bad2.metadata.tags = tag();
    let mut good = sample_chunk("slack:#eng", 0, 1_000);
    good.metadata.tags = tag();
    upsert_chunks(&cfg, &[bad1, bad2, good]).unwrap();

    let mut allowed = std::collections::HashSet::new();
    allowed.insert("slack:#eng".to_string());
    let q = ListChunksQuery {
        limit: Some(1),
        source_scope: Some(allowed),
        ..Default::default()
    };
    let rows = list_chunks(&cfg, &q).unwrap();
    assert_eq!(
        rows.len(),
        1,
        "the allowed-source chunk must survive the gate"
    );
    assert_eq!(rows[0].metadata.source_id, "slack:#eng");

    // No scope → unrestricted: the newest (disallowed) chunk wins under LIMIT 1.
    let unscoped = ListChunksQuery {
        limit: Some(1),
        ..Default::default()
    };
    let rows = list_chunks(&cfg, &unscoped).unwrap();
    assert_eq!(rows[0].metadata.source_id, "slack:#secret");
}

#[test]
fn upsert_is_idempotent() {
    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    assert_eq!(count_chunks(&cfg).unwrap(), 1);
}

#[test]
fn reingest_preserves_existing_embedding() {
    let (_tmp, cfg) = test_config();
    let mut c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    set_chunk_embedding(&cfg, &c.id, &[0.1, 0.2, 0.3]).unwrap();

    c.content = "updated content".into();
    c.token_count = 99;
    upsert_chunks(&cfg, &[c.clone()]).unwrap();

    let embedding = get_chunk_embedding(&cfg, &c.id).unwrap().unwrap();
    assert_eq!(embedding, vec![0.1, 0.2, 0.3]);
    let got = get_chunk(&cfg, &c.id).unwrap().unwrap();
    assert_eq!(got.content, "updated content");
    assert_eq!(got.token_count, 99);
}

#[test]
fn chunk_embeddings_are_scoped_by_model_signature() {
    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();

    set_chunk_embedding_for_signature(
        &cfg,
        &c.id,
        "openai/text-embedding-3-small@1536",
        &[0.1, 0.2],
    )
    .unwrap();
    set_chunk_embedding_for_signature(&cfg, &c.id, "local/bge-small@384", &[0.3, 0.4, 0.5])
        .unwrap();

    assert_eq!(
        get_chunk_embedding_for_signature(&cfg, &c.id, "openai/text-embedding-3-small@1536")
            .unwrap(),
        Some(vec![0.1, 0.2])
    );
    assert_eq!(
        get_chunk_embedding_for_signature(&cfg, &c.id, "local/bge-small@384").unwrap(),
        Some(vec![0.3, 0.4, 0.5])
    );
    assert!(
        get_chunk_embedding_for_signature(&cfg, &c.id, "missing/model@1")
            .unwrap()
            .is_none()
    );

    // #1574 cutover: the public `get_chunk_embedding` now reads the sidecar at
    // the *active* signature (not the legacy column). Nothing was written
    // there yet, so it is absent — graceful, never a cross-space read of the
    // openai/local rows above.
    assert!(get_chunk_embedding(&cfg, &c.id).unwrap().is_none());

    // The public setter targets the active signature and round-trips through
    // the public getter — proves the cutover wiring end to end.
    set_chunk_embedding(&cfg, &c.id, &[0.7, 0.8]).unwrap();
    assert_eq!(
        get_chunk_embedding(&cfg, &c.id).unwrap(),
        Some(vec![0.7, 0.8])
    );

    // ...and the earlier per-signature rows remain independently scoped.
    assert_eq!(
        get_chunk_embedding_for_signature(&cfg, &c.id, "local/bge-small@384").unwrap(),
        Some(vec![0.3, 0.4, 0.5])
    );
}

#[test]
fn list_filters_by_source_kind() {
    let (_tmp, cfg) = test_config();
    let c1 = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    let mut c2 = sample_chunk("gmail:t1", 0, 1_700_000_001_000);
    c2.metadata.source_kind = SourceKind::Email;
    upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap();
    let q = ListChunksQuery {
        source_kind: Some(SourceKind::Email),
        ..Default::default()
    };
    let rows = list_chunks(&cfg, &q).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].metadata.source_kind, SourceKind::Email);
}

#[test]
fn list_filters_by_time_range() {
    let (_tmp, cfg) = test_config();
    let a = sample_chunk("s", 0, 1_700_000_000_000);
    let b = sample_chunk("s", 1, 1_700_000_010_000);
    let c = sample_chunk("s", 2, 1_700_000_020_000);
    upsert_chunks(&cfg, &[a.clone(), b.clone(), c.clone()]).unwrap();
    let q = ListChunksQuery {
        since_ms: Some(1_700_000_005_000),
        until_ms: Some(1_700_000_015_000),
        ..Default::default()
    };
    let rows = list_chunks(&cfg, &q).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, b.id);
}

#[test]
fn list_orders_by_timestamp_desc() {
    let (_tmp, cfg) = test_config();
    let a = sample_chunk("s", 0, 1_700_000_000_000);
    let b = sample_chunk("s", 1, 1_700_000_010_000);
    upsert_chunks(&cfg, &[a.clone(), b.clone()]).unwrap();
    let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].id, b.id); // newest first
    assert_eq!(rows[1].id, a.id);
}

#[test]
fn list_orders_equal_timestamps_by_sequence() {
    let (_tmp, cfg) = test_config();
    let a = sample_chunk("s", 0, 1_700_000_000_000);
    let b = sample_chunk("s", 1, 1_700_000_000_000);
    upsert_chunks(&cfg, &[b.clone(), a.clone()]).unwrap();
    let rows = list_chunks(&cfg, &ListChunksQuery::default()).unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].seq_in_source, 0);
    assert_eq!(rows[1].seq_in_source, 1);
}

#[test]
fn list_limit_is_clamped_to_sane_range() {
    let (_tmp, cfg) = test_config();
    let chunks = (0..3)
        .map(|idx| sample_chunk("s", idx, 1_700_000_000_000 + i64::from(idx)))
        .collect::<Vec<_>>();
    upsert_chunks(&cfg, &chunks).unwrap();

    let zero_limit = list_chunks(
        &cfg,
        &ListChunksQuery {
            limit: Some(0),
            ..Default::default()
        },
    )
    .unwrap();
    assert_eq!(zero_limit.len(), 1);

    let huge_limit = list_chunks(
        &cfg,
        &ListChunksQuery {
            limit: Some(usize::MAX),
            ..Default::default()
        },
    )
    .unwrap();
    assert_eq!(huge_limit.len(), 3);
}

#[test]
fn delete_chunks_by_source_removes_chunks_side_rows_and_ingest_gate() {
    let (_tmp, cfg) = test_config();
    let target_a = sample_chunk("slack:c-1", 0, 1_700_000_000_000);
    let target_b = sample_chunk("slack:c-1", 1, 1_700_000_001_000);
    let other = sample_chunk("slack:c-2", 0, 1_700_000_002_000);
    upsert_chunks(&cfg, &[target_a.clone(), target_b.clone(), other.clone()]).unwrap();

    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        for chunk in [&target_a, &target_b, &other] {
            tx.execute(
                "INSERT INTO mem_tree_score (
                    chunk_id, total, token_count_signal, unique_words_signal,
                    metadata_weight, source_weight, interaction_weight,
                    entity_density, dropped, reason, computed_at_ms
                ) VALUES (?1, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0, NULL, 1700000000000)",
                params![chunk.id],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_entity_index (
                    entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms
                ) VALUES (?1, ?2, 'chunk', 'person', 'chat', 0.9, 1700000000000)",
                params![format!("entity:{}", chunk.id), chunk.id],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_chunk_embeddings (
                    chunk_id, model_signature, vector, dim, created_at
                ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)",
                params![chunk.id, vec![1_u8, 2, 3]],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_chunk_reembed_skipped (
                    chunk_id, model_signature, reason, skipped_at_ms
                ) VALUES (?1, 'test/model@3', 'terminal', 1700000000000)",
                params![chunk.id],
            )?;
        }
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Chat,
            "slack:c-1",
            1_700_000_000_000
        )?);
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Chat,
            "slack:c-2",
            1_700_000_000_000
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    let deleted = delete_chunks_by_source(&cfg, SourceKind::Chat, "slack:c-1").unwrap();

    assert_eq!(deleted, 2);
    assert_eq!(count_chunks(&cfg).unwrap(), 1);
    assert!(get_chunk(&cfg, &target_a.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &target_b.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &other.id).unwrap().is_some());
    assert!(!is_source_ingested(&cfg, SourceKind::Chat, "slack:c-1").unwrap());
    assert!(is_source_ingested(&cfg, SourceKind::Chat, "slack:c-2").unwrap());

    with_connection(&cfg, |conn| {
        let count_by_table = |table: &str| -> rusqlite::Result<i64> {
            conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))
        };
        assert_eq!(count_by_table("mem_tree_score")?, 1);
        assert_eq!(count_by_table("mem_tree_entity_index")?, 1);
        assert_eq!(count_by_table("mem_tree_chunk_embeddings")?, 1);
        assert_eq!(count_by_table("mem_tree_chunk_reembed_skipped")?, 1);
        Ok(())
    })
    .unwrap();
}

/// Forget-path (`clear_memory=true`) e2e: deleting the last chunk of a source
/// must cascade-delete its summary tree (tree row + summaries + sidecars +
/// entity-index + unsealed buffer), leave a sibling source untouched, and a
/// queued `Seal` job for the now-gone tree must settle to `Done` (not stick
/// in pending). Mocked connection (tempdir), chunks, tree/summary/buffer, job.
#[tokio::test]
async fn clear_memory_delete_cascades_orphaned_source_tree_and_settles_queued_job() {
    use crate::openhuman::memory_queue::{store as queue_store, types as queue_types};
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::{
        Buffer, SummaryNode, Tree, TreeKind, TreeStatus,
    };

    let (_tmp, cfg) = test_config();
    let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();

    // ---- mocked chunks: gmail:acct (conn-1, disconnecting) + gmail:other (conn-2, survives) ----
    let mk_email = |source_id: &str, seq: u32, owner: &str, ts_ms: i64| {
        let mut c = sample_chunk(source_id, seq, ts_ms);
        c.metadata.source_kind = SourceKind::Email;
        c.metadata.owner = owner.to_string();
        c
    };
    let a0 = mk_email("gmail:acct", 0, "gmail-sync:conn-1", 1_700_000_000_000);
    let a1 = mk_email("gmail:acct", 1, "gmail-sync:conn-1", 1_700_000_001_000);
    let b0 = mk_email("gmail:other", 0, "gmail-sync:conn-2", 1_700_000_002_000);
    upsert_chunks(&cfg, &[a0.clone(), a1.clone(), b0.clone()]).unwrap();

    // ---- mocked source trees (scope == source_id), each with summary + sidecars + entity-index + buffer ----
    let mk_tree = |id: &str, scope: &str| Tree {
        id: id.into(),
        kind: TreeKind::Source,
        scope: scope.into(),
        root_id: None,
        max_level: 1,
        status: TreeStatus::Active,
        created_at: ts,
        last_sealed_at: Some(ts),
    };
    tree_store::insert_tree(&cfg, &mk_tree("tree-acct", "gmail:acct")).unwrap();
    tree_store::insert_tree(&cfg, &mk_tree("tree-other", "gmail:other")).unwrap();

    let mk_summary = |id: &str, tree_id: &str, children: Vec<String>| SummaryNode {
        id: id.into(),
        tree_id: tree_id.into(),
        tree_kind: TreeKind::Source,
        level: 1,
        parent_id: None,
        child_ids: children,
        content: format!("summary for {tree_id}"),
        token_count: 3,
        entities: vec![],
        topics: vec![],
        time_range_start: ts,
        time_range_end: ts,
        score: 0.5,
        sealed_at: ts,
        deleted: false,
        embedding: None,
        doc_id: None,
        version_ms: None,
    };

    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;

        tree_store::insert_summary_tx(
            &tx,
            &mk_summary("sum-acct", "tree-acct", vec![a0.id.clone(), a1.id.clone()]),
            None,
            "test/model@3",
        )?;
        tree_store::insert_summary_tx(
            &tx,
            &mk_summary("sum-other", "tree-other", vec![b0.id.clone()]),
            None,
            "test/model@3",
        )?;

        // summary sidecars: embeddings for both summaries, reembed-skip only for sum-acct.
        for sid in ["sum-acct", "sum-other"] {
            tx.execute(
                "INSERT INTO mem_tree_summary_embeddings (
                    summary_id, model_signature, vector, dim, created_at
                ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)",
                params![sid, vec![1_u8, 2, 3]],
            )?;
        }
        tx.execute(
            "INSERT INTO mem_tree_summary_reembed_skipped (
                summary_id, model_signature, reason, skipped_at_ms
            ) VALUES ('sum-acct', 'test/model@3', 'terminal', 1700000000000)",
            [],
        )?;

        // tree-keyed entity-index rows (summary nodes) for each tree.
        for (sid, tree_id) in [("sum-acct", "tree-acct"), ("sum-other", "tree-other")] {
            tx.execute(
                "INSERT INTO mem_tree_entity_index (
                    entity_id, node_id, node_kind, entity_kind, surface,
                    score, timestamp_ms, tree_id, is_user
                ) VALUES (?1, ?2, 'summary', 'person', 'email', 0.9, 1700000000000, ?3, 0)",
                params![format!("entity:{sid}"), sid, tree_id],
            )?;
        }

        // unsealed buffers (the "queue" frontier) referencing the chunk ids.
        tree_store::upsert_buffer_tx(
            &tx,
            &Buffer {
                tree_id: "tree-acct".into(),
                level: 0,
                item_ids: vec![a0.id.clone(), a1.id.clone()],
                token_sum: 24,
                oldest_at: Some(ts),
            },
        )?;
        tree_store::upsert_buffer_tx(
            &tx,
            &Buffer {
                tree_id: "tree-other".into(),
                level: 0,
                item_ids: vec![b0.id.clone()],
                token_sum: 12,
                oldest_at: Some(ts),
            },
        )?;

        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Email,
            "gmail:acct",
            1_700_000_000_000
        )?);
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Email,
            "gmail:other",
            1_700_000_000_000
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    // ---- mocked job: a Seal queued for the tree that's about to be deleted ----
    let seal_payload = queue_types::SealPayload {
        tree_id: "tree-acct".into(),
        level: 0,
        force_now_ms: None,
    };
    let job_id = queue_store::enqueue(&cfg, &queue_types::NewJob::seal(&seal_payload).unwrap())
        .unwrap()
        .expect("seal job enqueued");

    // ---- act: disconnect conn-1 with clear_memory=true → delete its chunks ----
    let deleted = delete_chunks_by_owner(&cfg, SourceKind::Email, "gmail-sync:conn-1").unwrap();
    assert_eq!(deleted, 2);

    // chunks: acct gone, other survives.
    assert!(get_chunk(&cfg, &a0.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &a1.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &b0.id).unwrap().is_some());

    // the orphaned source tree is gone; the sibling tree is untouched.
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:acct")
            .unwrap()
            .is_none()
    );
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:other")
            .unwrap()
            .is_some()
    );

    // exactly the tree-acct rows are cascaded away across every dependent table.
    with_connection(&cfg, |conn| {
        let count = |sql: &str| -> rusqlite::Result<i64> { conn.query_row(sql, [], |r| r.get(0)) };
        assert_eq!(count("SELECT COUNT(*) FROM mem_tree_trees")?, 1);
        assert_eq!(count("SELECT COUNT(*) FROM mem_tree_summaries")?, 1);
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_summary_embeddings")?,
            1
        );
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_summary_reembed_skipped")?,
            0
        );
        assert_eq!(count("SELECT COUNT(*) FROM mem_tree_buffers")?, 1);
        assert_eq!(count("SELECT COUNT(*) FROM mem_tree_entity_index")?, 1);
        // and what survives belongs to tree-other.
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = 'tree-other'")?,
            1
        );
        Ok(())
    })
    .unwrap();

    // ---- the queued Seal job settles to Done (tree missing), not stuck pending ----
    let claimed = queue_store::claim_next(&cfg, queue_store::DEFAULT_LOCK_DURATION_MS)
        .unwrap()
        .expect("seal job claimable");
    assert_eq!(claimed.kind, queue_types::JobKind::Seal);
    let outcome = crate::openhuman::memory_queue::handlers::handle_job(&cfg, &claimed)
        .await
        .expect("handle_job ok");
    assert!(
        matches!(outcome, queue_types::JobOutcome::Done),
        "seal over a deleted tree must no-op to Done, got {outcome:?}"
    );
    queue_store::mark_done(&cfg, &claimed).unwrap();
    assert_eq!(
        queue_store::get_job(&cfg, &job_id).unwrap().unwrap().status,
        queue_types::JobStatus::Done
    );
}

/// #1: the cascade must also delete the summary's **on-disk content file**, not
/// just the row — otherwise a `clear_memory` delete leaves the summarised text
/// orphaned on disk.
#[test]
fn clear_memory_delete_removes_orphaned_summary_content_file() {
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::{SummaryNode, Tree, TreeKind, TreeStatus};

    let (_tmp, cfg) = test_config();
    let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();

    let mut c = sample_chunk("gmail:acct", 0, 1_700_000_000_000);
    c.metadata.source_kind = SourceKind::Email;
    c.metadata.owner = "gmail-sync:conn-1".to_string();
    upsert_chunks(&cfg, &[c.clone()]).unwrap();

    tree_store::insert_tree(
        &cfg,
        &Tree {
            id: "tree-acct".into(),
            kind: TreeKind::Source,
            scope: "gmail:acct".into(),
            root_id: None,
            max_level: 1,
            status: TreeStatus::Active,
            created_at: ts,
            last_sealed_at: Some(ts),
        },
    )
    .unwrap();

    // A real on-disk summary content file under the memory tree content root.
    let rel = "summaries/gmail_acct/L1/sum-acct.md";
    let abs = cfg.memory_tree_content_root().join(rel);
    std::fs::create_dir_all(abs.parent().unwrap()).unwrap();
    std::fs::write(&abs, "summarised email body").unwrap();
    assert!(abs.exists());

    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        tree_store::insert_summary_tx(
            &tx,
            &SummaryNode {
                id: "sum-acct".into(),
                tree_id: "tree-acct".into(),
                tree_kind: TreeKind::Source,
                level: 1,
                parent_id: None,
                child_ids: vec![c.id.clone()],
                content: "preview".into(),
                token_count: 3,
                entities: vec![],
                topics: vec![],
                time_range_start: ts,
                time_range_end: ts,
                score: 0.5,
                sealed_at: ts,
                deleted: false,
                embedding: None,
                doc_id: None,
                version_ms: None,
            },
            None,
            "test/model@3",
        )?;
        tx.execute(
            "UPDATE mem_tree_summaries SET content_path = ?1 WHERE id = 'sum-acct'",
            params![rel],
        )?;
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Email,
            "gmail:acct",
            1_700_000_000_000
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    delete_chunks_by_owner(&cfg, SourceKind::Email, "gmail-sync:conn-1").unwrap();

    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:acct")
            .unwrap()
            .is_none()
    );
    assert!(
        !abs.exists(),
        "orphaned summary content file must be removed from disk"
    );
}

/// #2: the safety property — deleting one connection's chunks must NOT delete
/// the source tree while ANOTHER connection still owns chunks for the same
/// account (source not yet orphaned).
#[test]
fn clear_memory_delete_keeps_tree_when_another_connection_still_owns_chunks() {
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::{Buffer, Tree, TreeKind, TreeStatus};

    let (_tmp, cfg) = test_config();
    let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();

    // Same account `gmail:acct`, two connections (owners).
    let mut a = sample_chunk("gmail:acct", 0, 1_700_000_000_000);
    a.metadata.source_kind = SourceKind::Email;
    a.metadata.owner = "gmail-sync:conn-1".to_string();
    let mut b = sample_chunk("gmail:acct", 1, 1_700_000_001_000);
    b.metadata.source_kind = SourceKind::Email;
    b.metadata.owner = "gmail-sync:conn-2".to_string();
    upsert_chunks(&cfg, &[a.clone(), b.clone()]).unwrap();

    tree_store::insert_tree(
        &cfg,
        &Tree {
            id: "tree-acct".into(),
            kind: TreeKind::Source,
            scope: "gmail:acct".into(),
            root_id: None,
            max_level: 1,
            status: TreeStatus::Active,
            created_at: ts,
            last_sealed_at: Some(ts),
        },
    )
    .unwrap();
    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        tree_store::upsert_buffer_tx(
            &tx,
            &Buffer {
                tree_id: "tree-acct".into(),
                level: 0,
                item_ids: vec![a.id.clone(), b.id.clone()],
                token_sum: 24,
                oldest_at: Some(ts),
            },
        )?;
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Email,
            "gmail:acct",
            1_700_000_000_000
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    // Disconnect ONLY conn-1.
    let deleted = delete_chunks_by_owner(&cfg, SourceKind::Email, "gmail-sync:conn-1").unwrap();
    assert_eq!(deleted, 1);

    // conn-1's chunk is gone, conn-2's remains → source still has chunks →
    // the tree (and its buffer + ingest gate) MUST survive.
    assert!(get_chunk(&cfg, &a.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &b.id).unwrap().is_some());
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:acct")
            .unwrap()
            .is_some(),
        "tree must survive while another connection still owns chunks"
    );
    assert!(is_source_ingested(&cfg, SourceKind::Email, "gmail:acct").unwrap());
    with_connection(&cfg, |conn| {
        let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_buffers", [], |r| r.get(0))?;
        assert_eq!(n, 1);
        Ok(())
    })
    .unwrap();
}

/// #3: queued `Extract` / `AppendBuffer` jobs that reference a chunk deleted
/// out from under them settle to `Done` (warn-and-skip), not stuck pending.
#[tokio::test]
async fn queued_jobs_for_deleted_chunk_settle_to_done() {
    use crate::openhuman::memory_queue::{store as queue_store, types as queue_types};

    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    delete_chunks_by_source(&cfg, SourceKind::Chat, "slack:#eng").unwrap();
    assert!(get_chunk(&cfg, &c.id).unwrap().is_none());

    queue_store::enqueue(
        &cfg,
        &queue_types::NewJob::extract_chunk(&queue_types::ExtractChunkPayload {
            chunk_id: c.id.clone(),
        })
        .unwrap(),
    )
    .unwrap();
    queue_store::enqueue(
        &cfg,
        &queue_types::NewJob::append_buffer(&queue_types::AppendBufferPayload {
            node: queue_types::NodeRef::Leaf {
                chunk_id: c.id.clone(),
            },
            target: queue_types::AppendTarget::Source {
                source_id: "slack:#eng".into(),
            },
        })
        .unwrap(),
    )
    .unwrap();

    for _ in 0..2 {
        let job = queue_store::claim_next(&cfg, queue_store::DEFAULT_LOCK_DURATION_MS)
            .unwrap()
            .expect("job claimable");
        let outcome = crate::openhuman::memory_queue::handlers::handle_job(&cfg, &job)
            .await
            .expect("handle_job ok");
        assert!(
            matches!(outcome, queue_types::JobOutcome::Done),
            "{:?} over a deleted chunk must settle Done, got {outcome:?}",
            job.kind
        );
        queue_store::mark_done(&cfg, &job).unwrap();
    }
}

#[test]
fn delete_chunks_by_owner_preserves_other_owners_for_same_source() {
    let (_tmp, cfg) = test_config();
    let mut target = sample_chunk("slack:shared", 0, 1_700_000_000_000);
    target.metadata.owner = "slack-sync:c-1".to_string();
    let mut same_source_other_owner = sample_chunk("slack:shared", 1, 1_700_000_001_000);
    same_source_other_owner.metadata.owner = "slack-sync:c-2".to_string();
    let mut target_other_source = sample_chunk("slack:c-1-only", 0, 1_700_000_002_000);
    target_other_source.metadata.owner = "slack-sync:c-1".to_string();
    upsert_chunks(
        &cfg,
        &[
            target.clone(),
            same_source_other_owner.clone(),
            target_other_source.clone(),
        ],
    )
    .unwrap();
    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Chat,
            "slack:shared",
            1_700_000_000_000
        )?);
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Chat,
            "slack:c-1-only",
            1_700_000_000_000
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    let deleted = delete_chunks_by_owner(&cfg, SourceKind::Chat, "slack-sync:c-1").unwrap();

    assert_eq!(deleted, 2);
    assert!(get_chunk(&cfg, &target.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &target_other_source.id).unwrap().is_none());
    assert!(get_chunk(&cfg, &same_source_other_owner.id)
        .unwrap()
        .is_some());
    assert!(is_source_ingested(&cfg, SourceKind::Chat, "slack:shared").unwrap());
    assert!(!is_source_ingested(&cfg, SourceKind::Chat, "slack:c-1-only").unwrap());
}

#[test]
fn delete_chunks_by_source_removes_safe_content_files_but_rejects_escape_paths() {
    let (_tmp, cfg) = test_config();
    let safe = sample_chunk("slack:c-1", 0, 1_700_000_000_000);
    let unsafe_chunk = sample_chunk("slack:c-1", 1, 1_700_000_001_000);
    upsert_chunks(&cfg, &[safe.clone(), unsafe_chunk.clone()]).unwrap();

    let content_root = cfg.memory_tree_content_root();
    let safe_rel = "chunks/safe.md";
    let safe_path = content_root.join(safe_rel);
    std::fs::create_dir_all(safe_path.parent().unwrap()).unwrap();
    std::fs::write(&safe_path, "safe").unwrap();

    let outside_path = content_root.parent().unwrap().join("outside.md");
    std::fs::write(&outside_path, "outside").unwrap();

    with_connection(&cfg, |conn| {
        conn.execute(
            "UPDATE mem_tree_chunks SET content_path = ?1 WHERE id = ?2",
            params![safe_rel, safe.id],
        )?;
        conn.execute(
            "UPDATE mem_tree_chunks SET content_path = ?1 WHERE id = ?2",
            params!["../outside.md", unsafe_chunk.id],
        )?;
        Ok(())
    })
    .unwrap();

    let deleted = delete_chunks_by_source(&cfg, SourceKind::Chat, "slack:c-1").unwrap();

    assert_eq!(deleted, 2);
    assert!(!safe_path.exists());
    assert!(outside_path.exists());
}

#[cfg(unix)]
#[test]
fn delete_chunks_by_source_removes_symlink_entry_not_target_file() {
    let (_tmp, cfg) = test_config();
    let linked_chunk = sample_chunk("slack:c-1", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[linked_chunk.clone()]).unwrap();

    let content_root = cfg.memory_tree_content_root();
    let target_path = content_root.join("chunks/target.md");
    let link_rel = "chunks/link.md";
    let link_path = content_root.join(link_rel);
    std::fs::create_dir_all(target_path.parent().unwrap()).unwrap();
    std::fs::write(&target_path, "target").unwrap();
    std::os::unix::fs::symlink("target.md", &link_path).unwrap();

    with_connection(&cfg, |conn| {
        conn.execute(
            "UPDATE mem_tree_chunks SET content_path = ?1 WHERE id = ?2",
            params![link_rel, linked_chunk.id],
        )?;
        Ok(())
    })
    .unwrap();

    let deleted = delete_chunks_by_source(&cfg, SourceKind::Chat, "slack:c-1").unwrap();

    assert_eq!(deleted, 1);
    assert!(target_path.exists());
    assert!(!link_path.exists());
}

#[test]
fn missing_chunk_returns_none() {
    let (_tmp, cfg) = test_config();
    assert!(get_chunk(&cfg, "nonexistent").unwrap().is_none());
}

#[test]
fn empty_batch_is_noop() {
    let (_tmp, cfg) = test_config();
    assert_eq!(upsert_chunks(&cfg, &[]).unwrap(), 0);
    assert_eq!(count_chunks(&cfg).unwrap(), 0);
}

#[test]
fn schema_has_content_path_and_content_sha256_columns() {
    // Verify that with_connection applies additive migrations for content
    // pointers and source grouping scope on a fresh DB.
    let (_tmp, cfg) = test_config();
    with_connection(&cfg, |conn| {
        let mut has_path_scope = false;
        let mut has_content_path = false;
        let mut has_content_sha256 = false;
        let mut stmt = conn.prepare("PRAGMA table_info(mem_tree_chunks)")?;
        let names: Vec<String> = stmt
            .query_map(params![], |row| row.get::<_, String>(1))?
            .filter_map(|r| r.ok())
            .collect();
        for name in &names {
            if name == "path_scope" {
                has_path_scope = true;
            }
            if name == "content_path" {
                has_content_path = true;
            }
            if name == "content_sha256" {
                has_content_sha256 = true;
            }
        }
        assert!(
            has_path_scope,
            "mem_tree_chunks must have path_scope column after migration; found: {names:?}"
        );
        assert!(
            has_content_path,
            "mem_tree_chunks must have content_path column after migration; found: {names:?}"
        );
        assert!(
            has_content_sha256,
            "mem_tree_chunks must have content_sha256 column after migration; found: {names:?}"
        );
        Ok(())
    })
    .unwrap();
}

/// Regression: OPENHUMAN-TAURI-HH / -ZM / -MB.
///
/// Before this fix, N `tree_jobs_worker` tasks racing into
/// `with_connection` on a cold workspace would trigger a WAL/SHM
/// cold-start code — 14 (CANTOPEN), 1546 (IOERR_TRUNCATE), or a
/// `-shm` code (4618 SHMOPEN / 4874 SHMSIZE / 5386 SHMMAP) — surfaced as
/// `Failed to initialize memory_tree schema`. The mutex-gated init set
/// in `store::open_and_init_with_retry` serialises the WAL+SHM
/// bootstrap so only one thread runs `apply_schema` per DB path.
///
/// Asserts:
/// 1. All N concurrent callers return `Ok` (no races, no surfaced errors).
/// 2. `apply_schema` runs exactly once for the shared path even though
///    8 threads hit a cold DB simultaneously.
#[test]
fn with_connection_serialises_concurrent_schema_init() {
    use std::sync::atomic::Ordering;

    let (_tmp, cfg) = test_config();
    let db_path = cfg.workspace_dir.join("memory_tree").join("chunks.db");
    let errors = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));

    let threads: Vec<_> = (0..8)
        .map(|_| {
            let cfg = cfg.clone();
            let errors = errors.clone();
            std::thread::spawn(move || {
                if with_connection(&cfg, |_| Ok(())).is_err() {
                    errors.fetch_add(1, Ordering::Relaxed);
                }
            })
        })
        .collect();
    for t in threads {
        t.join().expect("worker thread panicked");
    }

    assert_eq!(
        errors.load(Ordering::Relaxed),
        0,
        "concurrent with_connection callers must all succeed"
    );
    let applied = super::schema_apply_count_for_path_for_tests(&db_path);
    assert_eq!(
        applied, 1,
        "apply_schema must run exactly once per DB path under concurrent init; ran {applied} times"
    );
}

/// Directly pins the `is_transient_cold_start` classifier — the
/// gatekeeper for the retry loop in `open_and_init_with_retry`. The
/// concurrent-init test above only exercises it indirectly (and only
/// if a transient happens to fire on the dev box). A targeted test
/// catches regressions if the match arms are edited.
#[test]
fn is_transient_cold_start_classifies_known_extended_codes() {
    use rusqlite::ffi;
    use rusqlite::ErrorCode;

    // The WAL/SHM cold-start codes that fire under contention. All must
    // classify as transient → retried. (4618 SHMOPEN is the macOS failure;
    // 5386 is the real SHMMAP; 4874 is SHMSIZE — all of the `-shm` family.)
    for extended in [
        14,   // CANTOPEN
        1546, // IOERR_TRUNCATE
        4618, // IOERR_SHMOPEN
        4874, // IOERR_SHMSIZE
        5386, // IOERR_SHMMAP
        8714, // IOERR_IN_PAGE
    ] {
        let err = anyhow::Error::from(rusqlite::Error::SqliteFailure(
            ffi::Error {
                code: ErrorCode::SystemIoFailure,
                extended_code: extended,
            },
            None,
        ));
        assert!(
            super::is_transient_cold_start(&err),
            "extended_code {extended} must classify as transient cold-start"
        );
    }

    // SQLITE_BUSY (extended code 5) is a real lock-contention signal,
    // NOT a cold-start race — the caller handles it via `busy_timeout`
    // not via this retry loop. Must NOT classify.
    let busy = anyhow::Error::from(rusqlite::Error::SqliteFailure(
        ffi::Error {
            code: ErrorCode::DatabaseBusy,
            extended_code: 5,
        },
        None,
    ));
    assert!(
        !super::is_transient_cold_start(&busy),
        "DatabaseBusy must not be classified as cold-start transient"
    );

    // Non-SQLite error in the chain — must not classify.
    let other: anyhow::Error = anyhow::anyhow!("not a sqlite error");
    assert!(
        !super::is_transient_cold_start(&other),
        "non-SQLite errors must not classify as transient cold-start"
    );
}

/// Regression: `PRAGMA foreign_keys` is connection-local in SQLite and
/// must be re-set on every `Connection::open`. After the schema-init
/// refactor, the pragma moved out of `SCHEMA` (which only runs on
/// first init per path) into `open_connection`. Verify both the
/// cold-init path and the fast path return a connection with FK on.
#[test]
fn with_connection_keeps_foreign_keys_on_for_every_call() {
    let (_tmp, cfg) = test_config();
    // First call — exercises apply_schema + open_connection.
    let fk_on_first: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row("PRAGMA foreign_keys;", params![], |r| r.get::<_, i64>(0))?)
    })
    .unwrap();
    assert_eq!(
        fk_on_first, 1,
        "foreign_keys must be ON on first connection"
    );
    // Second call — fast path (schema init skipped); pragma must still be set.
    let fk_on_second: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row("PRAGMA foreign_keys;", params![], |r| r.get::<_, i64>(0))?)
    })
    .unwrap();
    assert_eq!(
        fk_on_second, 1,
        "foreign_keys must be ON on fast-path (post-init) connection"
    );
}

/// #1574 §7: the one-shot, version-gated legacy→sidecar migration copies a
/// legacy `embedding` blob whose dimensionality matches the active embedder
/// into the per-model sidecar at the active signature, skips dim-mismatched
/// rows (left for the §6 re-embed), keeps the legacy column, and runs exactly
/// once (PRAGMA user_version gate).
#[test]
fn legacy_embeddings_migrate_to_sidecar_once() {
    let (_tmp, cfg) = test_config();
    let c_match = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    let c_mismatch = sample_chunk("slack:#eng", 1, 1_700_000_000_001);
    // First open runs the (no-op) migration and sets user_version = 1.
    upsert_chunks(&cfg, &[c_match.clone(), c_mismatch.clone()]).unwrap();

    // Resolve the active signature/dims exactly as the migration does —
    // base-independent, never hard-coded (see the brittle-literal lesson).
    let (p, m, dims) = crate::openhuman::memory_store::effective_embedding_settings(
        &cfg.memory,
        cfg.workload_local_model("embeddings").as_deref(),
    );
    let sig = crate::openhuman::embeddings::format_embedding_signature(&p, &m, dims);
    let match_vec = vec![0.25f32; dims];
    let mismatch_vec = vec![0.5f32; dims + 1];

    // Simulate a pre-#1574 DB: legacy columns populated, migration not yet
    // run. On entry user_version is 1 (from upsert above) so the migration
    // is skipped here; the closure then resets the gate to 0.
    with_connection(&cfg, |conn| {
        conn.execute(
            "UPDATE mem_tree_chunks SET embedding = ?1 WHERE id = ?2",
            params![embedding_to_blob(&match_vec), c_match.id],
        )?;
        conn.execute(
            "UPDATE mem_tree_chunks SET embedding = ?1 WHERE id = ?2",
            params![embedding_to_blob(&mismatch_vec), c_mismatch.id],
        )?;
        conn.pragma_update(None, "user_version", 0i64)?;
        Ok(())
    })
    .unwrap();

    // Evict the cached connection so the next open sees user_version = 0
    // and re-runs migrate_legacy_embeddings_to_sidecar.
    invalidate_connection(&cfg);

    // The next store open (this getter's `with_connection`) sees
    // user_version = 0 and runs the migration before returning.
    assert_eq!(
        get_chunk_embedding_for_signature(&cfg, &c_match.id, &sig).unwrap(),
        Some(match_vec.clone()),
        "matching-dim legacy row must be copied to the sidecar at the active sig"
    );
    assert!(
        get_chunk_embedding_for_signature(&cfg, &c_mismatch.id, &sig)
            .unwrap()
            .is_none(),
        "dim-mismatched legacy row must be skipped (left for §6 re-embed)"
    );

    with_connection(&cfg, |conn| {
        let legacy: Option<Vec<u8>> = conn
            .query_row(
                "SELECT embedding FROM mem_tree_chunks WHERE id = ?1",
                params![c_match.id],
                |r| r.get(0),
            )
            .unwrap();
        assert!(
            legacy.is_some(),
            "legacy column must be KEPT post-migration (vN+1 drops it later)"
        );
        let v: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        // A full init runs every one-shot migration in sequence, so the gate
        // lands on the latest version (the global/topic purge), not just the
        // embedding migration's.
        assert_eq!(
            v, GLOBAL_TOPIC_PURGE_MIGRATION_VERSION,
            "version gate must be set to the latest migration"
        );
        Ok(())
    })
    .unwrap();

    // Idempotent: subsequent opens are no-ops (gate set); sidecar unchanged.
    with_connection(&cfg, |_| Ok(())).unwrap();
    assert_eq!(
        get_chunk_embedding_for_signature(&cfg, &c_match.id, &sig).unwrap(),
        Some(match_vec)
    );
}

// ── Connection cache tests (#2206) ───────────────────────────────────────────

/// Two `with_connection` calls for the same workspace must return the same
/// cached `Arc` (pointer identity proves no re-init happened).
#[test]
fn connection_cache_returns_same_arc_for_same_workspace() {
    clear_connection_cache();
    let (_tmp, cfg) = test_config();

    let arc1 = get_or_init_connection(&cfg).expect("first get_or_init");
    let arc2 = get_or_init_connection(&cfg).expect("second get_or_init");
    assert!(
        Arc::ptr_eq(&arc1, &arc2),
        "expected the same Arc from the connection cache on the second call"
    );
}

/// Two configs pointing at different tempdirs must produce independent
/// connections (separate Arc pointers, no cross-contamination).
#[test]
fn connection_cache_uses_separate_connections_for_different_workspaces() {
    clear_connection_cache();
    let (_tmp1, cfg1) = test_config();
    let (_tmp2, cfg2) = test_config();

    let arc1 = get_or_init_connection(&cfg1).expect("workspace 1");
    let arc2 = get_or_init_connection(&cfg2).expect("workspace 2");
    assert!(
        !Arc::ptr_eq(&arc1, &arc2),
        "different workspaces must have independent connections"
    );

    // Sanity: each DB is usable independently.
    let c = sample_chunk("s", 0, 1_700_000_000_000);
    upsert_chunks(&cfg1, &[c.clone()]).unwrap();
    assert_eq!(count_chunks(&cfg1).unwrap(), 1);
    assert_eq!(count_chunks(&cfg2).unwrap(), 0);
}

/// Pointing the DB path at a *file* (not a directory) makes it impossible to
/// create the DB, so `get_or_init_connection` must fail. After
/// `CB_THRESHOLD` failures the circuit breaker trips and subsequent calls
/// return an error immediately without touching the filesystem.
#[test]
fn circuit_breaker_trips_after_threshold() {
    clear_connection_cache();
    let tmp = TempDir::new().expect("tempdir");

    // Create a regular file where the memory_tree *directory* would be —
    // this prevents `create_dir_all` from succeeding.
    let blocker = tmp.path().join(DB_DIR);
    std::fs::write(&blocker, b"not a dir").expect("write blocker file");

    let mut cfg = Config::default();
    cfg.workspace_dir = tmp.path().to_path_buf();

    // First CB_THRESHOLD calls should all fail (can't create dir over a file).
    for i in 0..CB_THRESHOLD {
        let result = get_or_init_connection(&cfg);
        assert!(
            result.is_err(),
            "call {i}: expected error before breaker trips"
        );
    }

    // The CB_THRESHOLD+1'th call should be rejected immediately by the
    // circuit breaker (error message contains "circuit breaker").
    let cb_err = get_or_init_connection(&cfg)
        .expect_err("expected circuit breaker error on call after threshold");
    let msg = format!("{cb_err:#}").to_ascii_lowercase();
    assert!(
        msg.contains("circuit breaker"),
        "expected circuit breaker message, got: {msg}"
    );
}

/// `try_cleanup_stale_files` removes `.db-shm` and `.db-wal` side-files that
/// exist alongside the main DB file.
#[test]
fn stale_shm_cleanup_removes_files() {
    let tmp = TempDir::new().expect("tempdir");
    let db_path = tmp.path().join("chunks.db");

    // Create the main DB file and the two stale side-files.
    std::fs::write(&db_path, b"").expect("create db file");
    let shm = tmp.path().join("chunks.db-shm");
    let wal = tmp.path().join("chunks.db-wal");
    std::fs::write(&shm, b"stale shm").expect("create shm");
    std::fs::write(&wal, b"stale wal").expect("create wal");

    assert!(shm.exists(), "shm must exist before cleanup");
    assert!(wal.exists(), "wal must exist before cleanup");

    let cleaned = try_cleanup_stale_files(&db_path);
    assert!(
        cleaned,
        "cleanup should return true when files were removed"
    );
    assert!(!shm.exists(), "shm must be removed");
    assert!(!wal.exists(), "wal must be removed");
}

/// memory_tree must run the TRUNCATE rollback journal — never WAL. WAL's
/// `-shm`/`-wal` machinery is the source of the cold-start IOERR_SHMMAP /
/// IOERR_TRUNCATE failures (Sentry TAURI-RUST-EV / TAURI-RUST-X1), and the
/// single cached connection gains nothing from WAL's reader concurrency.
#[test]
fn memory_tree_uses_truncate_journal_not_wal() {
    let (_tmp, cfg) = test_config();

    with_connection(&cfg, |conn| {
        let mode: String = conn.query_row("PRAGMA journal_mode", [], |r| r.get(0))?;
        assert!(
            mode.eq_ignore_ascii_case("truncate"),
            "memory_tree journal_mode must be TRUNCATE, got '{mode}'"
        );
        let sync: i64 = conn.query_row("PRAGMA synchronous", [], |r| r.get(0))?;
        assert_eq!(sync, 2, "rollback journal requires synchronous=FULL (2)");
        Ok(())
    })
    .expect("with_connection");

    // A `-shm` shared-memory side-file is only ever created under WAL.
    let shm = cfg.workspace_dir.join("memory_tree").join("chunks.db-shm");
    assert!(
        !shm.exists(),
        "no -shm file must exist under TRUNCATE journal"
    );
}

/// A database a prior (WAL-mode) release left behind must migrate cleanly to
/// TRUNCATE on the next open, with the `-wal`/`-shm` side-files gone.
#[test]
fn existing_wal_db_migrates_to_truncate() {
    let (_tmp, cfg) = test_config();
    let db_path = cfg.workspace_dir.join("memory_tree").join("chunks.db");
    std::fs::create_dir_all(db_path.parent().unwrap()).expect("mkdir");

    // Simulate the old release: open the DB in WAL mode and commit a row so
    // the WAL marker is persisted in the database header.
    {
        let conn = rusqlite::Connection::open(&db_path).expect("open wal db");
        let mode: String = conn
            .query_row("PRAGMA journal_mode=WAL", [], |r| r.get(0))
            .expect("set wal");
        assert!(mode.eq_ignore_ascii_case("wal"), "precondition: db in WAL");
        conn.execute_batch("CREATE TABLE legacy_marker(x); INSERT INTO legacy_marker VALUES (1);")
            .expect("seed");
    } // connection dropped — the header still records WAL

    // Clear any cached connection for isolation, then open via with_connection.
    clear_connection_cache();
    with_connection(&cfg, |conn| {
        let mode: String = conn.query_row("PRAGMA journal_mode", [], |r| r.get(0))?;
        assert!(
            mode.eq_ignore_ascii_case("truncate"),
            "WAL db must migrate to TRUNCATE on open, got '{mode}'"
        );
        // Data written under WAL must survive the checkpoint-and-switch — the
        // migration must not lose committed rows.
        let marker: i64 = conn.query_row("SELECT x FROM legacy_marker", [], |r| r.get(0))?;
        assert_eq!(marker, 1, "row committed under WAL must survive migration");
        Ok(())
    })
    .expect("with_connection migrates");

    assert!(
        !db_path.with_file_name("chunks.db-shm").exists(),
        "-shm must be gone after WAL→TRUNCATE migration"
    );
    assert!(
        !db_path.with_file_name("chunks.db-wal").exists(),
        "-wal must be gone after WAL→TRUNCATE migration"
    );
}

#[test]
fn clear_chunk_reembed_skipped_is_idempotent() {
    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    let sig = tree_active_signature(&cfg);
    mark_chunk_reembed_skipped(&cfg, &c.id, &sig, "test orphan").unwrap();
    clear_chunk_reembed_skipped(&cfg, &c.id, &sig).unwrap();
    clear_chunk_reembed_skipped(&cfg, &c.id, &sig).unwrap();
    let count: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped
              WHERE chunk_id = ?1 AND model_signature = ?2",
            params![c.id, sig],
            |r| r.get(0),
        )?)
    })
    .unwrap();
    assert_eq!(count, 0);
}

#[test]
fn clear_reembed_skipped_for_signature_removes_all_tombstones_for_sig() {
    let (_tmp, cfg) = test_config();
    let c1 = sample_chunk("slack:#a", 0, 1_700_000_000_000);
    let c2 = sample_chunk("slack:#b", 1, 1_700_000_000_001);
    upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap();
    let sig = tree_active_signature(&cfg);
    let other_sig = "provider=other;model=x;dims=8";
    mark_chunk_reembed_skipped(&cfg, &c1.id, &sig, "r1").unwrap();
    mark_chunk_reembed_skipped(&cfg, &c2.id, &sig, "r2").unwrap();
    mark_chunk_reembed_skipped(&cfg, &c1.id, other_sig, "other").unwrap();
    let summary_id = "summary-bulk-clear-test";
    with_connection(&cfg, |conn| {
        conn.execute(
            "INSERT OR IGNORE INTO mem_tree_trees (id, kind, scope, created_at_ms)
             VALUES ('tree-bulk-clear', 'source', 'bulk-clear', 0)",
            [],
        )?;
        conn.execute(
            "INSERT INTO mem_tree_summaries (
                id, tree_id, tree_kind, level, child_ids_json, content, token_count,
                entities_json, topics_json, time_range_start_ms, time_range_end_ms,
                score, sealed_at_ms, deleted
             ) VALUES (?1, 'tree-bulk-clear', 'source', 0, '[]', 'x', 1, '[]', '[]', 0, 0, 0.0, 0, 0)",
            params![summary_id],
        )?;
        Ok(())
    })
    .unwrap();
    crate::openhuman::memory_store::trees::store::mark_summary_reembed_skipped(
        &cfg,
        summary_id,
        &sig,
        "summary tombstone",
    )
    .unwrap();

    let deleted = clear_reembed_skipped_for_signature(&cfg, &sig).unwrap();
    assert_eq!(deleted, 3);

    let remaining_chunks: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped WHERE model_signature = ?1",
            params![sig],
            |r| r.get(0),
        )?)
    })
    .unwrap();
    assert_eq!(remaining_chunks, 0);

    let remaining_summaries: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_summary_reembed_skipped WHERE model_signature = ?1",
            params![sig],
            |r| r.get(0),
        )?)
    })
    .unwrap();
    assert_eq!(remaining_summaries, 0);

    let other_kept: i64 = with_connection(&cfg, |conn| {
        Ok(conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped
              WHERE chunk_id = ?1 AND model_signature = ?2",
            params![c1.id, other_sig],
            |r| r.get(0),
        )?)
    })
    .unwrap();
    assert_eq!(other_kept, 1);
}

#[test]
fn validate_reembed_skip_key_rejects_empty_and_oversized() {
    assert!(validate_reembed_skip_key("chunk_id", "  ").is_err());
    let huge = "a".repeat(REEMBED_SKIP_KEY_MAX_LEN + 1);
    assert!(validate_reembed_skip_key("chunk_id", &huge).is_err());
    assert!(validate_reembed_skip_key("chunk_id", "ok\0bad").is_err());
    assert_eq!(
        validate_reembed_skip_key("chunk_id", "  trimmed  ").unwrap(),
        "trimmed"
    );
}

// ---------- get_chunks_batch ----------
//
// Contract: equivalent to looping `get_chunk` per id but in
// `O(ceil(n / MAX_FETCH_BATCH))` SQLite round-trips. The map carries
// only ids that exist; missing ids are silently absent (same as the
// per-row helper returning Ok(None)).

#[test]
fn get_chunks_batch_returns_present_ids_in_map() {
    let (_tmp, cfg) = test_config();
    let c1 = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    let c2 = sample_chunk("slack:#eng", 1, 1_700_000_000_000);
    let c3 = sample_chunk("slack:#ops", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c1.clone(), c2.clone(), c3.clone()]).unwrap();

    let ids = vec![c1.id.clone(), c2.id.clone(), c3.id.clone()];
    let map = get_chunks_batch(&cfg, &ids).unwrap();
    assert_eq!(map.len(), 3);
    assert_eq!(map.get(&c1.id), Some(&c1));
    assert_eq!(map.get(&c2.id), Some(&c2));
    assert_eq!(map.get(&c3.id), Some(&c3));
}

#[test]
fn get_chunks_batch_empty_input_and_missing_ids() {
    // Empty input: empty map (no SQL issued).
    let (_tmp, cfg) = test_config();
    let empty = get_chunks_batch(&cfg, &[]).unwrap();
    assert!(empty.is_empty());

    // Missing ids: silently absent (mirrors per-row Ok(None)).
    // `fetch_leaves` relies on this so partial-result detection
    // (`hits.len() < ids.len()`) keeps working unchanged.
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    let ids = vec![
        c.id.clone(),
        "ghost:no-such-1".into(),
        "ghost:no-such-2".into(),
    ];
    let map = get_chunks_batch(&cfg, &ids).unwrap();
    assert_eq!(map.len(), 1);
    assert_eq!(map.get(&c.id), Some(&c));
    assert!(map.get("ghost:no-such-1").is_none());
    assert!(map.get("ghost:no-such-2").is_none());
}

// ---------- get_chunk_embeddings_for_signature_batch ----------
//
// Contract: equivalent to looping `get_chunk_embedding_for_signature`
// per id, but in O(ceil(n / MAX_EMBEDDING_BATCH)) round-trips instead
// of O(n). The map contains only ids that have a vector under the
// requested signature; absent rows are silently dropped (same as the
// per-row helper returning Ok(None)).

#[test]
fn batch_embedding_lookup_returns_only_signature_scoped_rows() {
    let (_tmp, cfg) = test_config();
    let c1 = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    let c2 = sample_chunk("slack:#eng", 1, 1_700_000_000_000);
    let c3 = sample_chunk("slack:#eng", 2, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c1.clone(), c2.clone(), c3.clone()]).unwrap();

    let sig_a = "openai/text-embedding-3-small@1536";
    let sig_b = "local/bge-small@384";
    set_chunk_embedding_for_signature(&cfg, &c1.id, sig_a, &[0.1, 0.2]).unwrap();
    set_chunk_embedding_for_signature(&cfg, &c2.id, sig_a, &[0.3, 0.4]).unwrap();
    set_chunk_embedding_for_signature(&cfg, &c3.id, sig_b, &[0.5, 0.6, 0.7]).unwrap();

    let ids = vec![c1.id.clone(), c2.id.clone(), c3.id.clone()];
    let map_a = get_chunk_embeddings_for_signature_batch(&cfg, &ids, sig_a).unwrap();
    assert_eq!(map_a.len(), 2, "only c1 and c2 are under sig_a");
    assert_eq!(map_a.get(&c1.id).cloned(), Some(vec![0.1, 0.2]));
    assert_eq!(map_a.get(&c2.id).cloned(), Some(vec![0.3, 0.4]));
    assert!(map_a.get(&c3.id).is_none(), "c3 has only sig_b");

    let map_b = get_chunk_embeddings_for_signature_batch(&cfg, &ids, sig_b).unwrap();
    assert_eq!(map_b.len(), 1);
    assert_eq!(map_b.get(&c3.id).cloned(), Some(vec![0.5, 0.6, 0.7]));
}

#[test]
fn batch_embedding_lookup_empty_input_returns_empty_map() {
    let (_tmp, cfg) = test_config();
    let map = get_chunk_embeddings_for_signature_batch(&cfg, &[], "any/sig@1").unwrap();
    assert!(map.is_empty());
}

#[test]
fn batch_embedding_lookup_unknown_ids_absent_from_map() {
    // Pre-batch contract: per-row helper returned Ok(None) for missing
    // chunks. Batch helper must mirror that — missing ids absent from
    // the map, present ids carry their vector. The retrieval rerank
    // path depends on this so absent rows get the
    // (NEG_INFINITY, false) sink-to-bottom treatment.
    let (_tmp, cfg) = test_config();
    let c = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c.clone()]).unwrap();
    let sig = "openai/text-embedding-3-small@1536";
    set_chunk_embedding_for_signature(&cfg, &c.id, sig, &[0.1]).unwrap();

    let ids = vec![
        c.id.clone(),
        "ghost:no-such-chunk-1".into(),
        "ghost:no-such-chunk-2".into(),
    ];
    let map = get_chunk_embeddings_for_signature_batch(&cfg, &ids, sig).unwrap();
    assert_eq!(map.len(), 1);
    assert_eq!(map.get(&c.id).cloned(), Some(vec![0.1]));
}

#[test]
fn batch_embedding_lookup_splits_id_list_above_per_batch_threshold() {
    // Validates the `chunks(MAX_EMBEDDING_BATCH)` window loop in
    // `get_chunk_embeddings_for_signature_batch`. We pass > 500 ids in
    // one call; the helper must internally split them into multiple
    // `IN (...)` queries and merge results into a single map. 3 of the
    // 501 ids actually carry embeddings; the other 498 are unknown
    // strings and must be absent from the returned map (no error).
    let (_tmp, cfg) = test_config();
    let c1 = sample_chunk("slack:#a", 0, 1_700_000_000_000);
    let c2 = sample_chunk("slack:#b", 0, 1_700_000_000_000);
    let c3 = sample_chunk("slack:#c", 0, 1_700_000_000_000);
    upsert_chunks(&cfg, &[c1.clone(), c2.clone(), c3.clone()]).unwrap();
    let sig = "openai/text-embedding-3-small@1536";
    set_chunk_embedding_for_signature(&cfg, &c1.id, sig, &[1.0]).unwrap();
    set_chunk_embedding_for_signature(&cfg, &c2.id, sig, &[2.0]).unwrap();
    set_chunk_embedding_for_signature(&cfg, &c3.id, sig, &[3.0]).unwrap();

    // Build 501 ids: 3 real + 498 ghosts. The 501-element vec crosses
    // the 500-per-batch boundary, forcing two `IN (...)` queries.
    let mut ids: Vec<String> = (0..498).map(|i| format!("ghost:{i}")).collect();
    ids.push(c1.id.clone());
    ids.push(c2.id.clone());
    ids.push(c3.id.clone());
    assert_eq!(ids.len(), 501);

    let map = get_chunk_embeddings_for_signature_batch(&cfg, &ids, sig).unwrap();
    assert_eq!(map.len(), 3, "only the 3 real ids should be present");
    assert_eq!(map.get(&c1.id).cloned(), Some(vec![1.0]));
    assert_eq!(map.get(&c2.id).cloned(), Some(vec![2.0]));
    assert_eq!(map.get(&c3.id).cloned(), Some(vec![3.0]));
}

/// The one-shot purge migration deletes global + topic trees (rows, summaries,
/// buffers, jobs, and on-disk summary folders) while leaving source trees and
/// non-retired jobs untouched, and runs exactly once (PRAGMA user_version gate).
#[test]
fn global_topic_purge_removes_only_global_and_topic() {
    let (_tmp, cfg) = test_config();
    // First open initialises the schema and runs both migrations (sets
    // user_version = 2).
    upsert_chunks(&cfg, &[sample_chunk("slack:#eng", 0, 1_700_000_000_000)]).unwrap();

    // On-disk: a legacy per-day global folder, the singleton global folder, a
    // topic folder, and a source folder that must survive.
    let summaries = cfg
        .memory_tree_content_root()
        .join("wiki")
        .join("summaries");
    for d in [
        "global-2026-05-28",
        "global",
        "topic-alice",
        "source-slack-eng",
    ] {
        std::fs::create_dir_all(summaries.join(d).join("L0")).unwrap();
    }

    with_connection(&cfg, |conn| {
        // Seed one tree of each kind, each with a summary.
        for (id, kind) in [
            ("source:s1", "source"),
            ("global:g1", "global"),
            ("topic:t1", "topic"),
        ] {
            conn.execute(
                "INSERT INTO mem_tree_trees (id, kind, scope, max_level, status, created_at_ms) \
                 VALUES (?1, ?2, ?2, 0, 'active', 0)",
                params![id, kind],
            )?;
            conn.execute(
                "INSERT INTO mem_tree_summaries \
                 (id, tree_id, tree_kind, level, content, token_count, \
                  time_range_start_ms, time_range_end_ms, sealed_at_ms) \
                 VALUES (?1, ?2, ?3, 0, 'x', 1, 0, 0, 0)",
                params![format!("sum-{id}"), id, kind],
            )?;
        }
        // Seed retired + surviving job rows.
        for (jid, kind) in [
            ("j1", "topic_route"),
            ("j2", "digest_daily"),
            ("j3", "extract_chunk"),
        ] {
            conn.execute(
                "INSERT INTO mem_tree_jobs (id, kind, payload_json, available_at_ms, created_at_ms) \
                 VALUES (?1, ?2, '{}', 0, 0)",
                params![jid, kind],
            )?;
        }
        // Re-arm the gate so the purge runs against the seeded rows.
        conn.pragma_update(None, "user_version", 1i64)?;
        super::purge_global_topic_trees(conn, &cfg)?;

        // Trees: only the source tree survives.
        let trees: i64 =
            conn.query_row("SELECT COUNT(*) FROM mem_tree_trees", [], |r| r.get(0))?;
        assert_eq!(trees, 1, "only the source tree should remain");
        let kind: String =
            conn.query_row("SELECT kind FROM mem_tree_trees", [], |r| r.get(0))?;
        assert_eq!(kind, "source");

        // Summaries: only the source summary survives.
        let summaries_left: i64 =
            conn.query_row("SELECT COUNT(*) FROM mem_tree_summaries", [], |r| r.get(0))?;
        assert_eq!(summaries_left, 1);

        // Jobs: retired kinds gone, extract_chunk kept.
        let jobs_left: Vec<String> = {
            let mut stmt = conn.prepare("SELECT kind FROM mem_tree_jobs ORDER BY kind")?;
            let rows = stmt.query_map([], |r| r.get::<_, String>(0))?;
            rows.collect::<rusqlite::Result<_>>()?
        };
        assert_eq!(jobs_left, vec!["extract_chunk".to_string()]);

        // Gate advanced — a second run is a no-op.
        let version: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
        assert_eq!(version, 2);
        Ok(())
    })
    .unwrap();

    // On-disk: global*/topic-* folders gone, source-* kept.
    assert!(!summaries.join("global-2026-05-28").exists());
    assert!(!summaries.join("global").exists());
    assert!(!summaries.join("topic-alice").exists());
    assert!(
        summaries.join("source-slack-eng").exists(),
        "source summary folder must survive the purge"
    );
}

// ── extraction_coverage (#002 FR-010 / US5) ──────────────────────────────

#[test]
fn extraction_coverage_empty_store_is_zero() {
    let (_tmp, cfg) = test_config();
    assert_eq!(extraction_coverage(&cfg).unwrap(), 0.0);
}

#[test]
fn extraction_coverage_reflects_indexed_fraction() {
    let (_tmp, cfg) = test_config();
    // Two chunks; index an entity for only the first → coverage 0.5.
    let c1 = sample_chunk("slack:#eng", 0, 1_700_000_000_000);
    let c2 = sample_chunk("slack:#eng", 1, 1_700_000_001_000);
    upsert_chunks(&cfg, &[c1.clone(), c2.clone()]).unwrap();

    with_connection(&cfg, |conn| {
        conn.execute(
            "INSERT INTO mem_tree_entity_index
                (entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms)
             VALUES (?1, ?2, 'leaf', 'person', 'Alice', 0.9, 1)",
            params!["person:Alice", c1.id],
        )?;
        Ok(())
    })
    .unwrap();

    let cov = extraction_coverage(&cfg).unwrap();
    assert!((cov - 0.5).abs() < 1e-6, "expected 0.5, got {cov}");

    // Index the second chunk too → full coverage.
    with_connection(&cfg, |conn| {
        conn.execute(
            "INSERT INTO mem_tree_entity_index
                (entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms)
             VALUES (?1, ?2, 'leaf', 'person', 'Bob', 0.9, 2)",
            params!["person:Bob", c2.id],
        )?;
        Ok(())
    })
    .unwrap();
    assert!((extraction_coverage(&cfg).unwrap() - 1.0).abs() < 1e-6);
}

// ── memory_tree_delete_source RPC ────────────────────────────────────────────
// These prove the new `delete_source_rpc` is a FULL source-level delete (not a
// chunk-only delete): it must cascade through every dependent table, the ingest
// gate, and the source summary tree, remove content files, and leave stale
// summaries unable to resurface in recall — while sibling sources are untouched.

/// Seed a fully-formed Document source (chunks + side rows + content files +
/// source tree + summary + sidecars + buffer + ingest gate) for `source_id`.
/// Returns the chunk ids created. Used by the delete_source tests below.
#[cfg(test)]
async fn seed_full_document_source(
    cfg: &Config,
    source_id: &str,
    tree_id: &str,
    summary_id: &str,
    base_ts_ms: i64,
) -> (Vec<String>, String, String) {
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::{
        Buffer, SummaryNode, Tree, TreeKind, TreeStatus,
    };

    let ts = Utc.timestamp_millis_opt(base_ts_ms).unwrap();
    let mk_doc = |seq: u32, ts_ms: i64| {
        let mut c = sample_chunk(source_id, seq, ts_ms);
        c.metadata.source_kind = SourceKind::Document;
        c
    };
    let c0 = mk_doc(0, base_ts_ms);
    let c1 = mk_doc(1, base_ts_ms + 1000);
    upsert_chunks(cfg, &[c0.clone(), c1.clone()]).unwrap();

    // Real on-disk chunk + summary content files under the content root.
    let content_root = cfg.memory_tree_content_root();
    let chunk_rel = format!("document/{tree_id}/c0.md");
    let summary_rel = format!("summaries/{tree_id}/L1/{summary_id}.md");
    for rel in [&chunk_rel, &summary_rel] {
        let abs = content_root.join(rel);
        std::fs::create_dir_all(abs.parent().unwrap()).unwrap();
        std::fs::write(&abs, "body").unwrap();
    }

    let mk_summary = SummaryNode {
        id: summary_id.into(),
        tree_id: tree_id.into(),
        tree_kind: TreeKind::Source,
        level: 1,
        parent_id: None,
        child_ids: vec![c0.id.clone(), c1.id.clone()],
        content: format!("summary text for {source_id}"),
        token_count: 3,
        entities: vec![],
        topics: vec![],
        time_range_start: ts,
        time_range_end: ts,
        score: 0.5,
        sealed_at: ts,
        deleted: false,
        embedding: None,
        doc_id: None,
        version_ms: None,
    };
    tree_store::insert_tree(
        cfg,
        &Tree {
            id: tree_id.into(),
            kind: TreeKind::Source,
            scope: source_id.into(),
            root_id: None,
            max_level: 1,
            status: TreeStatus::Active,
            created_at: ts,
            last_sealed_at: Some(ts),
        },
    )
    .unwrap();

    with_connection(cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        for chunk in [&c0, &c1] {
            tx.execute(
                "INSERT INTO mem_tree_score (
                    chunk_id, total, token_count_signal, unique_words_signal,
                    metadata_weight, source_weight, interaction_weight,
                    entity_density, dropped, reason, computed_at_ms
                ) VALUES (?1, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 1.0, 0, NULL, 1700000000000)",
                params![chunk.id],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_entity_index (
                    entity_id, node_id, node_kind, entity_kind, surface, score, timestamp_ms
                ) VALUES (?1, ?2, 'chunk', 'person', 'doc', 0.9, 1700000000000)",
                params![format!("entity:{}", chunk.id), chunk.id],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_chunk_embeddings (
                    chunk_id, model_signature, vector, dim, created_at
                ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)",
                params![chunk.id, vec![1_u8, 2, 3]],
            )?;
            tx.execute(
                "INSERT INTO mem_tree_chunk_reembed_skipped (
                    chunk_id, model_signature, reason, skipped_at_ms
                ) VALUES (?1, 'test/model@3', 'terminal', 1700000000000)",
                params![chunk.id],
            )?;
        }
        // point chunk c0 at its on-disk content file.
        tx.execute(
            "UPDATE mem_tree_chunks SET content_path = ?1 WHERE id = ?2",
            params![chunk_rel, c0.id],
        )?;

        tree_store::insert_summary_tx(&tx, &mk_summary, None, "test/model@3")?;
        tx.execute(
            "UPDATE mem_tree_summaries SET content_path = ?1 WHERE id = ?2",
            params![summary_rel, summary_id],
        )?;
        tx.execute(
            "INSERT INTO mem_tree_summary_embeddings (
                summary_id, model_signature, vector, dim, created_at
            ) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)",
            params![summary_id, vec![1_u8, 2, 3]],
        )?;
        tx.execute(
            "INSERT INTO mem_tree_summary_reembed_skipped (
                summary_id, model_signature, reason, skipped_at_ms
            ) VALUES (?1, 'test/model@3', 'terminal', 1700000000000)",
            params![summary_id],
        )?;
        tx.execute(
            "INSERT INTO mem_tree_entity_index (
                entity_id, node_id, node_kind, entity_kind, surface,
                score, timestamp_ms, tree_id, is_user
            ) VALUES (?1, ?2, 'summary', 'person', 'doc', 0.9, 1700000000000, ?3, 0)",
            params![format!("entity:{summary_id}"), summary_id, tree_id],
        )?;
        tree_store::upsert_buffer_tx(
            &tx,
            &Buffer {
                tree_id: tree_id.into(),
                level: 0,
                item_ids: vec![c0.id.clone(), c1.id.clone()],
                token_sum: 24,
                oldest_at: Some(ts),
            },
        )?;
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Document,
            source_id,
            base_ts_ms
        )?);
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    (vec![c0.id.clone(), c1.id], chunk_rel, summary_rel)
}

#[tokio::test]
async fn delete_source_rpc_purges_document_source_fully() {
    use crate::openhuman::memory::read_rpc::{delete_source_rpc, list_chunks_rpc, recall_rpc};
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::TreeKind;

    let (_tmp, cfg) = test_config();
    let target = "telegram-note-A";
    let sibling = "telegram-note-B";
    let (target_ids, target_chunk_file, target_summary_file) =
        seed_full_document_source(&cfg, target, "tree-A", "sum-A", 1_700_000_000_000).await;
    let (sibling_ids, sibling_chunk_file, sibling_summary_file) =
        seed_full_document_source(&cfg, sibling, "tree-B", "sum-B", 1_700_000_100_000).await;

    let content_root = cfg.memory_tree_content_root();
    // Pre-conditions: both sources fully present on disk + in DB.
    assert!(content_root.join(&target_chunk_file).exists());
    assert!(content_root.join(&target_summary_file).exists());

    // ---- act ----
    let out = delete_source_rpc(&cfg, target.to_string())
        .await
        .expect("delete_source ok")
        .value;
    assert!(out.deleted);
    assert_eq!(out.chunks_removed, 2);

    // JSON response shape: { deleted: bool, chunks_removed: u64 }.
    let json = serde_json::to_value(&out).unwrap();
    assert_eq!(json.get("deleted").and_then(|v| v.as_bool()), Some(true));
    assert_eq!(json.get("chunks_removed").and_then(|v| v.as_u64()), Some(2));

    // 1. target chunks gone; sibling chunk survives.
    for id in &target_ids {
        assert!(get_chunk(&cfg, id).unwrap().is_none(), "chunk {id} remains");
    }
    for id in &sibling_ids {
        assert!(get_chunk(&cfg, id).unwrap().is_some(), "sibling {id} gone");
    }

    // 2–11. every dependent table for the target is empty; sibling rows remain.
    with_connection(&cfg, |conn| {
        let count = |sql: &str, p: &str| -> rusqlite::Result<i64> {
            conn.query_row(sql, params![p], |r| r.get(0))
        };
        // chunk side rows keyed by the target chunk ids
        for id in &target_ids {
            assert_eq!(
                count(
                    "SELECT COUNT(*) FROM mem_tree_score WHERE chunk_id = ?1",
                    id
                )?,
                0
            );
            assert_eq!(
                count(
                    "SELECT COUNT(*) FROM mem_tree_entity_index WHERE node_id = ?1",
                    id
                )?,
                0
            );
            assert_eq!(
                count(
                    "SELECT COUNT(*) FROM mem_tree_chunk_embeddings WHERE chunk_id = ?1",
                    id
                )?,
                0
            );
            assert_eq!(
                count(
                    "SELECT COUNT(*) FROM mem_tree_chunk_reembed_skipped WHERE chunk_id = ?1",
                    id
                )?,
                0
            );
        }
        // source tree rows (scope/tree-id == tree-A) gone
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = ?1",
                "tree-A"
            )?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_summary_embeddings WHERE summary_id = ?1",
                "sum-A"
            )?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_summary_reembed_skipped WHERE summary_id = ?1",
                "sum-A"
            )?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_entity_index WHERE tree_id = ?1",
                "tree-A"
            )?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_buffers WHERE tree_id = ?1",
                "tree-A"
            )?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_trees WHERE id = ?1",
                "tree-A"
            )?,
            0
        );
        // sibling tree intact
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = ?1",
                "tree-B"
            )?,
            1
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_trees WHERE id = ?1",
                "tree-B"
            )?,
            1
        );
        Ok(())
    })
    .unwrap();

    // 6. ingest dedup gate cleared for target, retained for sibling.
    assert!(!is_source_ingested(&cfg, SourceKind::Document, target).unwrap());
    assert!(is_source_ingested(&cfg, SourceKind::Document, sibling).unwrap());
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, target)
            .unwrap()
            .is_none()
    );

    // 12–13. target content files removed; sibling content files remain.
    assert!(!content_root.join(&target_chunk_file).exists());
    assert!(!content_root.join(&target_summary_file).exists());
    assert!(content_root.join(&sibling_chunk_file).exists());
    assert!(content_root.join(&sibling_summary_file).exists());

    // 14. recall no longer surfaces the deleted source (no summary/chunk left to
    // rank). Tolerant of minimal-config recall backends: if it returns, none of
    // the hits may belong to the deleted source.
    if let Ok(rc) = recall_rpc(&cfg, "summary text".to_string(), 10).await {
        assert!(
            rc.value.chunks.iter().all(|c| c.source_id != target),
            "deleted source must not resurface in recall"
        );
    }

    // 16. second delete is idempotent.
    let again = delete_source_rpc(&cfg, target.to_string())
        .await
        .unwrap()
        .value;
    assert!(!again.deleted);
    assert_eq!(again.chunks_removed, 0);

    // 15. re-ingesting the same source_id works again (gate cleared) and writes chunks.
    let mut fresh = sample_chunk(target, 0, 1_700_000_500_000);
    fresh.metadata.source_kind = SourceKind::Document;
    assert_eq!(upsert_chunks(&cfg, &[fresh.clone()]).unwrap(), 1);
    let listed = list_chunks_rpc(&cfg, Default::default())
        .await
        .unwrap()
        .value;
    assert!(listed.chunks.iter().any(|c| c.source_id == target));
    // The dedup gate was cleared by delete, so re-ingest can claim it again.
    // Commit the claim (a rolled-back tx would prove nothing) and verify it
    // actually persisted.
    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        assert!(
            claim_source_ingest_tx(&tx, SourceKind::Document, target, 1_700_000_500_000)?,
            "ingest gate must be re-claimable after delete_source cleared it"
        );
        tx.commit()?;
        Ok(())
    })
    .unwrap();
    assert!(
        is_source_ingested(&cfg, SourceKind::Document, target).unwrap(),
        "re-claimed ingest gate must persist"
    );
}

/// Versioned document sources store the ingest gate as `{source_id}@{version_ms}`
/// in addition to (or instead of) the bare id. `delete_source` must clear both.
#[tokio::test]
async fn delete_source_rpc_clears_versioned_ingest_gates() {
    use crate::openhuman::memory::read_rpc::delete_source_rpc;

    let (_tmp, cfg) = test_config();
    let sid = "notion:conn-1:page-xyz";
    let versioned = format!("{sid}@1700000000000");

    let mut c = sample_chunk(sid, 0, 1_700_000_000_000);
    c.metadata.source_kind = SourceKind::Document;
    upsert_chunks(&cfg, &[c.clone()]).unwrap();

    // Seed both a bare gate and a versioned gate for the source.
    with_connection(&cfg, |conn| {
        let tx = conn.unchecked_transaction()?;
        assert!(claim_source_ingest_tx(
            &tx,
            SourceKind::Document,
            sid,
            1_700_000_000_000
        )?);
        tx.execute(
            "INSERT INTO mem_tree_ingested_sources (source_kind, source_id, ingested_at_ms)
             VALUES ('document', ?1, 1700000000000)",
            params![versioned],
        )?;
        tx.commit()?;
        Ok(())
    })
    .unwrap();

    let gate_count = |conn: &rusqlite::Connection| -> rusqlite::Result<i64> {
        conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_ingested_sources
              WHERE source_kind = 'document' AND (source_id = ?1 OR source_id LIKE ?2)",
            params![sid, format!("{sid}@%")],
            |r| r.get(0),
        )
    };
    with_connection(&cfg, |conn| {
        assert_eq!(gate_count(conn)?, 2, "both gates seeded");
        Ok(())
    })
    .unwrap();

    let out = delete_source_rpc(&cfg, sid.to_string())
        .await
        .unwrap()
        .value;
    assert!(out.deleted);
    assert_eq!(out.chunks_removed, 1);

    assert!(!is_source_ingested(&cfg, SourceKind::Document, sid).unwrap());
    with_connection(&cfg, |conn| {
        assert_eq!(
            gate_count(conn)?,
            0,
            "bare AND versioned ingest gates must be cleared"
        );
        Ok(())
    })
    .unwrap();
}

#[tokio::test]
async fn delete_source_rpc_unknown_id_is_idempotent() {
    use crate::openhuman::memory::read_rpc::delete_source_rpc;
    let (_tmp, cfg) = test_config();
    let out = delete_source_rpc(&cfg, "telegram-note-does-not-exist".to_string())
        .await
        .unwrap()
        .value;
    assert!(!out.deleted);
    assert_eq!(out.chunks_removed, 0);
}

#[tokio::test]
async fn delete_source_rpc_rejects_empty_source_id() {
    use crate::openhuman::memory::read_rpc::delete_source_rpc;
    let (_tmp, cfg) = test_config();
    assert!(delete_source_rpc(&cfg, "   ".to_string()).await.is_err());
}

/// Legacy partial delete: chunks were already removed earlier (e.g. by the bot's
/// old per-chunk `delete_chunk` loop), leaving an orphaned summary tree + dedup
/// gate. `delete_source_rpc` must finish the job and remove the stale tree.
#[tokio::test]
async fn delete_source_rpc_cleans_legacy_partial_delete() {
    use crate::openhuman::memory::read_rpc::{delete_chunk_rpc, delete_source_rpc};
    use crate::openhuman::memory_store::trees::store as tree_store;
    use crate::openhuman::memory_store::trees::types::TreeKind;

    let (_tmp, cfg) = test_config();
    let target = "telegram-event-legacy";
    let (chunk_ids, _chunk_file, summary_file) =
        seed_full_document_source(&cfg, target, "tree-legacy", "sum-legacy", 1_700_000_000_000)
            .await;

    // ---- simulate the OLD per-chunk delete loop: remove only the chunks ----
    for id in &chunk_ids {
        assert!(
            delete_chunk_rpc(&cfg, id.clone())
                .await
                .unwrap()
                .value
                .deleted
        );
    }

    // pre-condition: chunks gone, but the summary tree + gate are still stale.
    for id in &chunk_ids {
        assert!(get_chunk(&cfg, id).unwrap().is_none());
    }
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, target)
            .unwrap()
            .is_some()
    );
    assert!(is_source_ingested(&cfg, SourceKind::Document, target).unwrap());
    with_connection(&cfg, |conn| {
        let n: i64 = conn.query_row(
            "SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = 'tree-legacy'",
            [],
            |r| r.get(0),
        )?;
        assert_eq!(n, 1, "stale summary must exist before delete_source");
        Ok(())
    })
    .unwrap();

    // ---- act: delete_source must finish the legacy cleanup ----
    let out = delete_source_rpc(&cfg, target.to_string())
        .await
        .unwrap()
        .value;
    // chunks were already gone, but a stale tree was cleaned → deleted=true.
    assert!(out.deleted);
    assert_eq!(out.chunks_removed, 0);

    // ---- assert: the stale tree / summaries / sidecars / gate are now gone ----
    assert!(
        tree_store::get_tree_by_scope(&cfg, TreeKind::Source, target)
            .unwrap()
            .is_none()
    );
    assert!(!is_source_ingested(&cfg, SourceKind::Document, target).unwrap());
    with_connection(&cfg, |conn| {
        let count = |sql: &str| -> rusqlite::Result<i64> { conn.query_row(sql, [], |r| r.get(0)) };
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = 'tree-legacy'")?,
            0
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM mem_tree_summary_embeddings WHERE summary_id = 'sum-legacy'"
            )?,
            0
        );
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_buffers WHERE tree_id = 'tree-legacy'")?,
            0
        );
        assert_eq!(
            count("SELECT COUNT(*) FROM mem_tree_trees WHERE id = 'tree-legacy'")?,
            0
        );
        Ok(())
    })
    .unwrap();
    // the summary content file is removed from disk too.
    assert!(!cfg.memory_tree_content_root().join(&summary_file).exists());

    // idempotent: a second delete_source now finds nothing.
    let again = delete_source_rpc(&cfg, target.to_string())
        .await
        .unwrap()
        .value;
    assert!(!again.deleted);
    assert_eq!(again.chunks_removed, 0);
}

#[test]
fn delete_source_registered_in_schema_and_controllers() {
    use crate::openhuman::memory::schema::{all_controller_schemas, all_registered_controllers};
    let schema = all_controller_schemas()
        .into_iter()
        .find(|s| s.function == "delete_source")
        .expect("delete_source schema present");
    assert_eq!(schema.namespace, "memory_tree"); // => openhuman.memory_tree_delete_source
    assert!(schema.inputs.iter().any(|f| f.name == "source_id"));
    assert!(schema.outputs.iter().any(|f| f.name == "deleted"));
    assert!(schema.outputs.iter().any(|f| f.name == "chunks_removed"));
    assert!(all_registered_controllers()
        .iter()
        .any(|c| c.schema.function == "delete_source"));
}
