From 2202835d40c0ebebdec57f92e36d6b26ef0686f9 Mon Sep 17 00:00:00 2001 From: michaelk1957 Date: Tue, 16 Jun 2026 03:00:00 +0300 Subject: [PATCH] Add memory_tree_delete_source RPC (#3664) Co-authored-by: Claude Opus 4.8 (1M context) --- src/openhuman/memory/read_rpc/admin.rs | 84 ++- src/openhuman/memory/read_rpc/mod.rs | 6 +- src/openhuman/memory/read_rpc/types.rs | 12 + src/openhuman/memory/schema/definitions.rs | 36 ++ src/openhuman/memory/schema/handlers.rs | 12 + src/openhuman/memory/schema/registry.rs | 5 + src/openhuman/memory_store/chunks/store.rs | 105 ++++ .../memory_store/chunks/store_tests.rs | 559 ++++++++++++++++++ 8 files changed, 815 insertions(+), 4 deletions(-) diff --git a/src/openhuman/memory/read_rpc/admin.rs b/src/openhuman/memory/read_rpc/admin.rs index 175b8f3cb..04bf74738 100644 --- a/src/openhuman/memory/read_rpc/admin.rs +++ b/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 Result, 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)) +} diff --git a/src/openhuman/memory/read_rpc/mod.rs b/src/openhuman/memory/read_rpc/mod.rs index 4655a2c32..a53af7d2c 100644 --- a/src/openhuman/memory/read_rpc/mod.rs +++ b/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, diff --git a/src/openhuman/memory/read_rpc/types.rs b/src/openhuman/memory/read_rpc/types.rs index 31aae285e..173487096 100644 --- a/src/openhuman/memory/read_rpc/types.rs +++ b/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 { diff --git a/src/openhuman/memory/schema/definitions.rs b/src/openhuman/memory/schema/definitions.rs index 27b2df808..4b37c1c1a 100644 --- a/src/openhuman/memory/schema/definitions.rs +++ b/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", diff --git a/src/openhuman/memory/schema/handlers.rs b/src/openhuman/memory/schema/handlers.rs index bc7d8f96b..64dd9065c 100644 --- a/src/openhuman/memory/schema/handlers.rs +++ b/src/openhuman/memory/schema/handlers.rs @@ -149,6 +149,18 @@ pub(super) fn handle_delete_chunk(params: Map) -> ControllerFutur }) } +pub(super) fn handle_delete_source(params: Map) -> 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::(Value::Object(params))?; + to_json(read_rpc::delete_source_rpc(&config, req.source_id).await?) + }) +} + pub(super) fn handle_graph_export(params: Map) -> ControllerFuture { Box::pin(async move { #[derive(serde::Deserialize, Default)] diff --git a/src/openhuman/memory/schema/registry.rs b/src/openhuman/memory/schema/registry.rs index eafd736ac..fc918362b 100644 --- a/src/openhuman/memory/schema/registry.rs +++ b/src/openhuman/memory/schema/registry.rs @@ -23,6 +23,7 @@ pub fn all_controller_schemas() -> Vec { 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 { 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, diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs index 5372ecb03..402edee81 100644 --- a/src/openhuman/memory_store/chunks/store.rs +++ b/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 { + use crate::openhuman::memory_store::trees::store as tree_store; + use crate::openhuman::memory_store::trees::types::TreeKind; + + let mut content_paths: Vec = 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 = { + 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::>>()? + }; + 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}; diff --git a/src/openhuman/memory_store/chunks/store_tests.rs b/src/openhuman/memory_store/chunks/store_tests.rs index 51b533f97..c50052d42 100644 --- a/src/openhuman/memory_store/chunks/store_tests.rs +++ b/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) { + 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 { + 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 { + 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 { 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")); +}