Add memory_tree_delete_source RPC (#3664)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/openhuman/memory/read_rpc/admin.rs

@@ -2,10 +2,16 @@ use anyhow::{Context, Result};
 use rusqlite::params;
 
 use crate::openhuman::config::Config;
-use crate::openhuman::memory_store::chunks::store::with_connection;
+use crate::openhuman::memory_store::chunks::store::{
+    delete_chunks_by_source, delete_orphaned_source_tree, with_connection,
+};
+use crate::openhuman::memory_store::chunks::types::SourceKind;
 use crate::rpc::RpcOutcome;
 
-use super::types::{FlushNowResponse, FlushSourceTreeResponse, ResetTreeResponse, WipeAllResponse};
+use super::types::{
+    DeleteSourceResponse, FlushNowResponse, FlushSourceTreeResponse, ResetTreeResponse,
+    WipeAllResponse,
+};
 
 // ── wipe_all ─────────────────────────────────────────────────────────────
 
@@ -345,3 +351,77 @@ pub async fn flush_now_rpc(config: &Config) -> Result<RpcOutcome<FlushNowRespons
     );
     Ok(RpcOutcome::single_log(resp, log))
 }
+
+// ── delete_source ──────────────────────────────────────────────────────────
+
+/// Fully delete one document source by its **exact** `source_id`.
+///
+/// Unlike [`super::entities::delete_chunk_rpc`] (which removes a single chunk and
+/// leaves the source tree intact), this wraps
+/// [`delete_chunks_by_source`] so the whole logical source is purged: every
+/// chunk plus its score / entity-index / embedding / reembed-skip side rows and
+/// chunk content files, the ingest dedup gate, and — when the source becomes
+/// fully orphaned — its source summary tree via `delete_tree_cascade_tx`
+/// (summaries, summary embeddings + reembed-skip, tree-keyed entity-index,
+/// buffers, the tree row, and summary content files). This prevents stale
+/// summaries of a deleted note/event/meeting from resurfacing in semantic
+/// recall.
+///
+/// Matching is **exact** (never a prefix), so sibling sources sharing a prefix
+/// are untouched. Idempotent: an unknown `source_id` removes nothing and returns
+/// `deleted = false`.
+///
+/// Legacy cleanup: a source whose chunks were already removed earlier by the
+/// per-chunk path keeps a now-stale summary tree (which `delete_chunks_by_source`
+/// won't touch, since it only cascades trees for chunks it deletes in the same
+/// call). After the chunk delete we therefore also run
+/// [`delete_orphaned_source_tree`], which cascades the source-scoped tree and
+/// clears the bare + versioned (`{source_id}@{version_ms}`) ingest gates iff zero
+/// chunks remain — so calling delete_source over such a source finishes the job.
+///
+/// Scope: this deletes the exact document source and its **source-scoped** orphan
+/// tree. It intentionally does NOT tear down shared collection / `path_scope`
+/// trees (e.g. Notion `notion:{connection}`), which may summarise many documents;
+/// per-document pruning inside a shared collection summary is out of scope here.
+pub async fn delete_source_rpc(
+    config: &Config,
+    source_id: String,
+) -> Result<RpcOutcome<DeleteSourceResponse>, String> {
+    let source_id = source_id.trim().to_string();
+    if source_id.is_empty() {
+        return Err("delete_source: source_id must be a non-empty string".to_string());
+    }
+    let cfg = config.clone();
+    let id = source_id.clone();
+    let (chunks_removed, tree_cleaned) =
+        tokio::task::spawn_blocking(move || -> Result<(usize, bool)> {
+            // Exact-match document delete; cascade logic lives in the store.
+            let removed = delete_chunks_by_source(&cfg, SourceKind::Document, &id)
+                .context("delete_chunks_by_source during delete_source")?;
+            // Finish off any legacy partial delete: cascade a now-orphaned tree
+            // (no-op when chunks were just deleted — the tree is already gone —
+            // or when there is no stale tree).
+            let tree_cleaned = delete_orphaned_source_tree(&cfg, SourceKind::Document, &id)
+                .context("delete_orphaned_source_tree during delete_source")?;
+            Ok((removed, tree_cleaned))
+        })
+        .await
+        .map_err(|e| format!("delete_source join error: {e}"))?
+        .map_err(|e| format!("delete_source: {e:#}"))?;
+
+    let resp = DeleteSourceResponse {
+        // `deleted` is true if we removed chunks OR cleaned a stale orphaned tree
+        // (the legacy-cleanup case has chunks_removed == 0 but still did work).
+        deleted: chunks_removed > 0 || tree_cleaned,
+        chunks_removed: chunks_removed as u64,
+    };
+    let log = format!(
+        // Redact the source id: it can embed user-linked identifiers.
+        "memory_tree::read: delete_source source_id_hash={} deleted={} chunks_removed={} tree_cleaned={}",
+        crate::openhuman::memory::util::redact::redact(&source_id),
+        resp.deleted,
+        resp.chunks_removed,
+        tree_cleaned
+    );
+    Ok(RpcOutcome::single_log(resp, log))
+}

src/openhuman/memory/read_rpc/mod.rs

@@ -17,7 +17,9 @@ pub mod types;
 pub mod vault;
 
 // Re-export everything so consumers and the test file keep working with `use super::*;`
-pub use admin::{flush_now_rpc, flush_source_tree_rpc, reset_tree_rpc, wipe_all_rpc};
+pub use admin::{
+    delete_source_rpc, flush_now_rpc, flush_source_tree_rpc, reset_tree_rpc, wipe_all_rpc,
+};
 pub use chunks::{
     display_name_for_source, list_chunks_rpc, list_sources_rpc, read_chunk_row, recall_rpc,
     search_rpc,
@@ -30,7 +32,7 @@ pub use graph::{
     graph_export_rpc, sanitize_basename, GraphEdge, GraphExportResponse, GraphMode, GraphNode,
 };
 pub use types::{
-    ChunkFilter, ChunkRow, DeleteChunkResponse, EntityRef, FlushNowResponse,
+    ChunkFilter, ChunkRow, DeleteChunkResponse, DeleteSourceResponse, EntityRef, FlushNowResponse,
     FlushSourceTreeResponse, ListChunksResponse, ObsidianVaultStatusResponse, RecallResponse,
     ResetTreeResponse, ScoreBreakdown, ScoreSignal, Source, VaultHealthCheckResponse,
     WipeAllResponse,

src/openhuman/memory/read_rpc/types.rs

@@ -113,6 +113,18 @@ pub struct DeleteChunkResponse {
     pub entity_index_rows_removed: u32,
 }
 
+/// Response shape for [`delete_source_rpc`].
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct DeleteSourceResponse {
+    /// True when the call did real work: at least one chunk was removed, OR a
+    /// stale orphaned source tree was cleaned up (legacy partial-delete case,
+    /// where `chunks_removed == 0`).
+    pub deleted: bool,
+    /// Number of chunk rows removed for the source (may be 0 when only a stale
+    /// orphaned tree/gate was cleaned).
+    pub chunks_removed: u64,
+}
+
 /// Response shape for [`wipe_all_rpc`].
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct WipeAllResponse {

src/openhuman/memory/schema/definitions.rs

@@ -366,6 +366,42 @@ pub fn schemas(function: &str) -> ControllerSchema {
                 },
             ],
         },
+        "delete_source" => ControllerSchema {
+            namespace: NAMESPACE,
+            function: "delete_source",
+            description: "Fully delete one document source by its EXACT source_id: every chunk \
+                 plus its score / entity-index / embedding / reembed-skip side rows and chunk \
+                 content files, the ingest dedup gates (bare source_id AND versioned \
+                 source_id@version), and (when the source becomes fully orphaned) its \
+                 source-scoped summary tree — summaries, summary embeddings + reembed-skip, \
+                 tree entity-index, buffers, the tree row, and summary content files. Unlike \
+                 delete_chunk this cascades, so stale summaries of the deleted source cannot \
+                 resurface in recall, and it also finishes legacy partial deletes (chunks already \
+                 gone, tree/gate left behind). Exact match only (never a prefix); shared \
+                 collection/path_scope trees that summarise multiple documents are left intact. \
+                 Idempotent — an unknown source_id returns deleted=false.",
+            inputs: vec![FieldSchema {
+                name: "source_id",
+                ty: TypeSchema::String,
+                comment: "Exact source id to remove (e.g. a Telegram note/event/meeting id).",
+                required: true,
+            }],
+            outputs: vec![
+                FieldSchema {
+                    name: "deleted",
+                    ty: TypeSchema::Bool,
+                    comment: "True when the call did real work: chunks were removed OR a stale \
+                              orphaned source tree was cleaned (legacy case, chunks_removed=0).",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "chunks_removed",
+                    ty: TypeSchema::U64,
+                    comment: "Number of chunk rows removed for the source.",
+                    required: true,
+                },
+            ],
+        },
         "wipe_all" => ControllerSchema {
             namespace: NAMESPACE,
             function: "wipe_all",

src/openhuman/memory/schema/handlers.rs

@@ -149,6 +149,18 @@ pub(super) fn handle_delete_chunk(params: Map<String, Value>) -> ControllerFutur
     })
 }
 
+pub(super) fn handle_delete_source(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        #[derive(serde::Deserialize)]
+        struct Req {
+            source_id: String,
+        }
+        let config = config_rpc::load_config_with_timeout().await?;
+        let req = parse_value::<Req>(Value::Object(params))?;
+        to_json(read_rpc::delete_source_rpc(&config, req.source_id).await?)
+    })
+}
+
 pub(super) fn handle_graph_export(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         #[derive(serde::Deserialize, Default)]

src/openhuman/memory/schema/registry.rs

@@ -23,6 +23,7 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("top_entities"),
         schemas("chunk_score"),
         schemas("delete_chunk"),
+        schemas("delete_source"),
         schemas("graph_export"),
         schemas("obsidian_vault_status"),
         schemas("vault_health_check"),
@@ -90,6 +91,10 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("delete_chunk"),
             handler: handle_delete_chunk,
         },
+        RegisteredController {
+            schema: schemas("delete_source"),
+            handler: handle_delete_source,
+        },
         RegisteredController {
             schema: schemas("graph_export"),
             handler: handle_graph_export,

src/openhuman/memory_store/chunks/store.rs

@@ -1153,6 +1153,111 @@ fn delete_chunks_by_source_filter(
     Ok(deleted)
 }
 
+/// Finish off an orphaned **Source** (one with zero chunks remaining): clear its
+/// ingest dedup gates and cascade-delete its source-scoped summary tree.
+///
+/// `delete_chunks_by_source` only cascades the tree for sources whose chunks it
+/// deletes in the same call; a source whose chunks were already removed earlier
+/// (e.g. by the per-chunk `delete_chunk` path) keeps a now-stale summary tree
+/// that can still resurface in recall. This cleans up exactly that **legacy
+/// partial-delete** state.
+///
+/// Specifically, when no chunks remain it:
+/// - removes the ingest dedup gates for the source — both the bare `source_id`
+///   AND any versioned `{source_id}@{version_ms}` gates (matched Rust-side with
+///   exact/prefix comparison, never SQL `LIKE`/`GLOB`, to avoid metachar pitfalls);
+/// - cascades the **source-scoped** tree (scope == `source_id`) if present.
+///
+/// Scoped-collection conservatism: a document ingested under a shared collection
+/// `path_scope` (e.g. Notion `notion:{connection}`) lives in a tree scoped by that
+/// `path_scope`, NOT by this `source_id`, so `get_tree_by_scope(Source,
+/// source_id)` returns `None` and such shared trees are left intact — deleting one
+/// document must never tear down a tree that summarises many documents.
+///
+/// Returns `true` when a source-scoped tree was removed (drives the RPC's
+/// `deleted` flag). No-op-safe to call unconditionally after
+/// `delete_chunks_by_source`.
+pub fn delete_orphaned_source_tree(
+    config: &Config,
+    source_kind: SourceKind,
+    source_id: &str,
+) -> Result<bool> {
+    use crate::openhuman::memory_store::trees::store as tree_store;
+    use crate::openhuman::memory_store::trees::types::TreeKind;
+
+    let mut content_paths: Vec<String> = Vec::new();
+    let tree_cascaded = with_connection(config, |conn| {
+        let tx = conn.unchecked_transaction()?;
+        let remaining: i64 = tx.query_row(
+            "SELECT COUNT(*) FROM mem_tree_chunks WHERE source_kind = ?1 AND source_id = ?2",
+            params![source_kind.as_str(), source_id],
+            |r| r.get(0),
+        )?;
+        if remaining > 0 {
+            // Source still has chunks — not orphaned; leave its live tree + gates.
+            log::debug!(
+                "[memory::chunk_store] delete_orphaned_source_tree: source_id_hash={} still has {remaining} chunk(s) — no-op",
+                redact_value(source_id),
+            );
+            return Ok(false);
+        }
+
+        // Clear ALL ingest dedup gates for this source: the bare source_id and any
+        // versioned `{source_id}@{version_ms}` gates. Filter in Rust (exact or
+        // `source_id@` prefix) so `_`/`%`/glob chars in ids are treated literally.
+        let versioned_prefix = format!("{source_id}@");
+        let gate_ids: Vec<String> = {
+            let mut stmt = tx.prepare(
+                "SELECT source_id FROM mem_tree_ingested_sources WHERE source_kind = ?1",
+            )?;
+            let rows = stmt.query_map(params![source_kind.as_str()], |r| r.get::<_, String>(0))?;
+            rows.filter_map(|row| match row {
+                Ok(s) if s == source_id || s.starts_with(&versioned_prefix) => Some(Ok(s)),
+                Ok(_) => None,
+                Err(e) => Some(Err(e)),
+            })
+            .collect::<rusqlite::Result<Vec<_>>>()?
+        };
+        for gid in &gate_ids {
+            tx.execute(
+                "DELETE FROM mem_tree_ingested_sources WHERE source_kind = ?1 AND source_id = ?2",
+                params![source_kind.as_str(), gid],
+            )?;
+        }
+
+        // Cascade the source-scoped orphan tree if one exists. Shared
+        // collection/path_scope trees are not keyed by this source_id (see fn
+        // docs), so they are intentionally left untouched.
+        let cascaded = if let Some(tree) =
+            tree_store::get_tree_by_scope_conn(&tx, TreeKind::Source, source_id)?
+        {
+            let cascade = tree_store::delete_tree_cascade_tx(&tx, &tree.id)?;
+            content_paths.extend(cascade.content_paths);
+            log::debug!(
+                    "[memory::chunk_store] delete_orphaned_source_tree: source_id_hash={} → removed stale tree_id={} summaries={} gates_cleared={}",
+                    redact_value(source_id),
+                    tree.id,
+                    cascade.removed_summaries,
+                    gate_ids.len(),
+                );
+            true
+        } else {
+            log::debug!(
+                    "[memory::chunk_store] delete_orphaned_source_tree: source_id_hash={} has no source-scoped tree (gates_cleared={}); shared/collection trees left intact",
+                    redact_value(source_id),
+                    gate_ids.len(),
+                );
+            false
+        };
+        tx.commit()?;
+        Ok(cascaded)
+    })?;
+    if tree_cascaded {
+        remove_chunk_content_files(config, &content_paths);
+    }
+    Ok(tree_cascaded)
+}
+
 fn remove_chunk_content_files(config: &Config, content_paths: &[String]) {
     use std::path::{Component, Path};
 

src/openhuman/memory_store/chunks/store_tests.rs

@@ -1757,3 +1757,562 @@ fn extraction_coverage_reflects_indexed_fraction() {
     .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"));
+}