diff --git a/src/openhuman/composio/ops.rs b/src/openhuman/composio/ops.rs index 198a58175..4b5634202 100644 --- a/src/openhuman/composio/ops.rs +++ b/src/openhuman/composio/ops.rs @@ -497,6 +497,30 @@ pub async fn composio_delete_connection( ); } } + // Prune the local memory_sources registry entry for this connection. + // The registry keys composio sources by `connection_id` and the + // reconciler only ever upserts, so a deleted connection's + // `[[memory_sources]]` entry is otherwise orphaned forever (and on + // reconnect the backend mints a fresh `connection_id`, leaving the stale + // one stranded). Best-effort: the backend connection is already gone, so + // a config-save failure must not fail the whole delete — log and move on. + match crate::openhuman::memory_sources::registry::remove_composio_source_by_connection_id( + connection_id, + ) + .await + { + Ok(0) => {} + Ok(removed) => tracing::debug!( + connection_id = %connection_id, + removed, + "[composio] pruned memory_sources entry after connection deletion" + ), + Err(e) => tracing::warn!( + connection_id = %connection_id, + error = %e, + "[composio] failed to prune memory_sources entry after connection deletion (non-fatal)" + ), + } crate::core::event_bus::publish_global( crate::core::event_bus::DomainEvent::ComposioConnectionDeleted { toolkit: toolkit.unwrap_or_else(|| "unknown".to_string()), diff --git a/src/openhuman/composio/ops_tests.rs b/src/openhuman/composio/ops_tests.rs index 70a881597..5e7d92813 100644 --- a/src/openhuman/composio/ops_tests.rs +++ b/src/openhuman/composio/ops_tests.rs @@ -522,6 +522,246 @@ async fn composio_delete_connection_clear_memory_deletes_slack_source() { assert_eq!(remaining[0].metadata.source_id, "slack:c2"); } +/// #4: full path through the REAL `composio_delete_connection` handler +/// (clear_memory=true, mock backend) — deleting a connection's last chunk must +/// cascade away its source summary tree AND the summary's on-disk content file, +/// not just the chunk rows. The tree is a real `get_or_create_source_tree`; the +/// content file sits at the production `content_path` location. +#[tokio::test] +async fn composio_delete_connection_clear_memory_cascades_source_tree_and_content_file() { + use crate::openhuman::memory::tree_source::registry::get_or_create_source_tree; + use crate::openhuman::memory_store::trees::store as tree_store; + use crate::openhuman::memory_store::trees::types::{SummaryNode, TreeKind}; + use rusqlite::params; + + let app = Router::new() + .route( + "/agent-integrations/composio/connections", + get(|| async { + Json(json!({ + "success": true, + "data": {"connections": [ + {"id":"c1","toolkit":"slack","status":"ACTIVE"} + ]} + })) + }), + ) + .route( + "/agent-integrations/composio/connections/{id}", + axum::routing::delete(|Path(_id): Path| async move { + Json(json!({"success": true, "data": {"deleted": true}})) + }), + ); + let base = start_mock_backend(app).await; + let tmp = tempfile::tempdir().unwrap(); + let config = config_with_backend(&tmp, base); + + // One slack chunk for connection c1 → source_id `slack:c1`. + let chunk = sample_memory_chunk(SourceKind::Chat, "slack:c1", 0); + memory_tree_store::upsert_chunks(&config, &[chunk.clone()]).expect("seed chunk"); + + // Real source tree for that source + a summary whose content file lives at + // the production content-root location. + let tree = get_or_create_source_tree(&config, "slack:c1").expect("source tree"); + let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap(); + let rel = "summaries/slack_c1/L1/sum-1.md"; + let abs = config.memory_tree_content_root().join(rel); + std::fs::create_dir_all(abs.parent().unwrap()).unwrap(); + std::fs::write(&abs, "summarised slack body").unwrap(); + + memory_tree_store::with_connection(&config, |conn| { + let tx = conn.unchecked_transaction()?; + tree_store::insert_summary_tx( + &tx, + &SummaryNode { + id: "sum-1".into(), + tree_id: tree.id.clone(), + tree_kind: TreeKind::Source, + level: 1, + parent_id: None, + child_ids: vec![chunk.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, + }, + None, + "test/model@3", + )?; + tx.execute( + "UPDATE mem_tree_summaries SET content_path = ?1 WHERE id = 'sum-1'", + params![rel], + )?; + tx.commit()?; + Ok(()) + }) + .expect("seed summary + content file pointer"); + + // sanity: tree + on-disk file exist before the disconnect. + assert!( + tree_store::get_tree_by_scope(&config, TreeKind::Source, "slack:c1") + .unwrap() + .is_some() + ); + assert!(abs.exists()); + + // ---- act: the REAL handler, clear_memory=true ---- + let outcome = composio_delete_connection(&config, "c1", true) + .await + .unwrap(); + assert!(outcome.value.deleted); + assert_eq!(outcome.value.memory_chunks_deleted, 1); + + // chunk, source tree, summary row, AND on-disk content file are all gone. + assert!(memory_tree_store::get_chunk(&config, &chunk.id) + .unwrap() + .is_none()); + assert!( + tree_store::get_tree_by_scope(&config, TreeKind::Source, "slack:c1") + .unwrap() + .is_none() + ); + memory_tree_store::with_connection(&config, |conn| { + let n: i64 = conn.query_row("SELECT COUNT(*) FROM mem_tree_summaries", [], |r| r.get(0))?; + assert_eq!(n, 0); + Ok(()) + }) + .unwrap(); + assert!( + !abs.exists(), + "summary content file must be removed via the real handler cascade" + ); +} + +/// #4 (full live seal): like the above, but the summary + on-disk file are +/// produced by the REAL `seal_one_level` pipeline (staged chunk body → +/// summarise → `stage_summary`), not hand-written. Then the REAL +/// `composio_delete_connection(clear_memory=true)` handler must cascade the +/// tree, the summary row, AND the seal-produced content file away. +#[tokio::test] +async fn composio_delete_connection_clear_memory_cascades_live_sealed_tree_and_file() { + use crate::openhuman::memory::tree_source::registry::get_or_create_source_tree; + use crate::openhuman::memory_store::chunks::store::{ + get_summary_content_pointers, upsert_staged_chunks_tx, + }; + use crate::openhuman::memory_store::content::stage_chunks; + use crate::openhuman::memory_store::trees::store as tree_store; + use crate::openhuman::memory_store::trees::types::{Buffer, TreeKind}; + use crate::openhuman::memory_tree::tree::bucket_seal::{seal_one_level, LabelStrategy}; + + let app = Router::new() + .route( + "/agent-integrations/composio/connections", + get(|| async { + Json(json!({ + "success": true, + "data": {"connections": [ + {"id":"c1","toolkit":"slack","status":"ACTIVE"} + ]} + })) + }), + ) + .route( + "/agent-integrations/composio/connections/{id}", + axum::routing::delete(|Path(_id): Path| async move { + Json(json!({"success": true, "data": {"deleted": true}})) + }), + ); + let base = start_mock_backend(app).await; + let tmp = tempfile::tempdir().unwrap(); + let mut config = config_with_backend(&tmp, base); + // Force the inert embedder so the real seal's summary-embed step doesn't + // reach a live endpoint. `config_with_backend` stores a cloud session + + // api_url, so the factory would otherwise build a *cloud* embedder against + // the mock (no embeddings route). `embeddings_provider = "none"` is the + // actual switch that selects `InertEmbedder`. + config.embeddings_provider = Some("none".to_string()); + config.memory_tree.embedding_endpoint = None; + config.memory_tree.embedding_model = None; + config.memory_tree.embedding_strict = false; + + // Real chunk for slack:c1 WITH its body staged to disk, so the seal's + // `hydrate_leaf_inputs` → `read_chunk_body` can resolve it. + let chunk = sample_memory_chunk(SourceKind::Chat, "slack:c1", 0); + memory_tree_store::upsert_chunks(&config, &[chunk.clone()]).expect("seed chunk"); + let staged = stage_chunks( + &config.memory_tree_content_root(), + std::slice::from_ref(&chunk), + ) + .expect("stage chunk body"); + memory_tree_store::with_connection(&config, |conn| { + let tx = conn.unchecked_transaction()?; + upsert_staged_chunks_tx(&tx, &staged)?; + tx.commit()?; + Ok(()) + }) + .expect("record staged chunk pointer"); + + // Run the REAL seal — produces a genuine summary row + on-disk file. + let tree = get_or_create_source_tree(&config, "slack:c1").expect("source tree"); + let buf = Buffer { + tree_id: tree.id.clone(), + level: 0, + item_ids: vec![chunk.id.clone()], + token_sum: i64::from(chunk.token_count), + oldest_at: Some(chunk.metadata.time_range.0), + }; + let summary_id = seal_one_level(&config, &tree, &buf, &LabelStrategy::Empty, false) + .await + .expect("real seal produces a summary"); + + // The seal wrote a real on-disk content file for the summary. + let (rel, _sha) = get_summary_content_pointers(&config, &summary_id) + .unwrap() + .expect("seal staged a summary content file"); + let abs = { + let mut p = config.memory_tree_content_root(); + for c in rel.split('/') { + p.push(c); + } + p + }; + assert!( + abs.exists(), + "seal must have written a summary file on disk" + ); + assert!( + tree_store::get_tree_by_scope(&config, TreeKind::Source, "slack:c1") + .unwrap() + .is_some() + ); + + // ---- act: REAL handler, clear_memory=true ---- + let outcome = composio_delete_connection(&config, "c1", true) + .await + .unwrap(); + assert!(outcome.value.deleted); + assert_eq!(outcome.value.memory_chunks_deleted, 1); + + // chunk, tree, summary row, and the seal-produced file are all gone. + assert!(memory_tree_store::get_chunk(&config, &chunk.id) + .unwrap() + .is_none()); + assert!( + tree_store::get_tree_by_scope(&config, TreeKind::Source, "slack:c1") + .unwrap() + .is_none() + ); + assert!(tree_store::get_summary(&config, &summary_id) + .unwrap() + .is_none()); + assert!( + !abs.exists(), + "seal-produced summary file must be removed via the real handler cascade" + ); +} + #[tokio::test] async fn composio_delete_connection_clear_memory_keeps_other_gmail_connections() { let app = Router::new() diff --git a/src/openhuman/memory_sources/mod.rs b/src/openhuman/memory_sources/mod.rs index 2c680b110..565d0917c 100644 --- a/src/openhuman/memory_sources/mod.rs +++ b/src/openhuman/memory_sources/mod.rs @@ -25,8 +25,9 @@ pub mod sync; pub mod types; pub use registry::{ - add_source, get_source, list_enabled_by_kind, list_sources, remove_source, update_source, - upsert_composio_source, MemorySourcePatch, + add_source, get_source, list_enabled_by_kind, list_sources, + remove_composio_source_by_connection_id, remove_source, update_source, upsert_composio_source, + MemorySourcePatch, }; pub use schemas::{ all_controller_schemas as all_memory_sources_controller_schemas, diff --git a/src/openhuman/memory_sources/registry.rs b/src/openhuman/memory_sources/registry.rs index da21dc8fd..98d0afd48 100644 --- a/src/openhuman/memory_sources/registry.rs +++ b/src/openhuman/memory_sources/registry.rs @@ -137,6 +137,35 @@ pub async fn remove_source(id: &str) -> Result { Ok(removed) } +/// Remove every composio source bound to `connection_id` — the disconnect path. +/// +/// Mirrors [`upsert_composio_source`], which keys composio sources on +/// `connection_id`. [`remove_source`] keys on the `src_*` id, which the +/// connection-delete flow doesn't have, so this is the connection-keyed +/// counterpart. Returns the number of entries removed (0 if none matched). +pub async fn remove_composio_source_by_connection_id(connection_id: &str) -> Result { + let mut config = config_rpc::load_config_with_timeout().await?; + let before = config.memory_sources.len(); + config.memory_sources.retain(|s| { + !(s.kind == SourceKind::Composio && s.connection_id.as_deref() == Some(connection_id)) + }); + let removed = before - config.memory_sources.len(); + + if removed > 0 { + tracing::info!( + connection_id = %connection_id, + removed, + "[memory_sources] removed composio source(s) on connection disconnect" + ); + config + .save() + .await + .map_err(|e| format!("failed to save config: {e:#}"))?; + } + + Ok(removed) +} + /// Upsert a composio source — used by the auto-registration path. /// If a source with the same `connection_id` already exists, updates /// the label; otherwise inserts a new entry. diff --git a/src/openhuman/memory_store/chunks/store.rs b/src/openhuman/memory_store/chunks/store.rs index 122f8e47b..ea470622c 100644 --- a/src/openhuman/memory_store/chunks/store.rs +++ b/src/openhuman/memory_store/chunks/store.rs @@ -967,6 +967,36 @@ fn delete_chunks_by_source_filter( )?; } + // A fully-orphaned source has zero chunks left, so its summary tree + // now summarises deleted content — and its unsealed buffer holds + // dangling chunk ids. Cascade-delete the tree (summaries + sidecars + // + entity-index + buffer + tree row) so a `clear_memory` delete is + // complete and stale summaries can't resurface in retrieval. Source + // trees use the chunk `source_id` verbatim as their scope, so we + // match on that. Same tx as the chunk delete → atomic. + for source_id in &orphaned_deleted_sources { + if let Some(tree) = + crate::openhuman::memory_store::trees::store::get_tree_by_scope_conn( + &tx, + crate::openhuman::memory_store::trees::types::TreeKind::Source, + source_id, + )? + { + let cascade = crate::openhuman::memory_store::trees::store::delete_tree_cascade_tx( + &tx, &tree.id, + )?; + // Defer the summary content-file removal to the same + // post-commit sweep as the chunk files. + content_paths.extend(cascade.content_paths); + log::debug!( + "[memory::chunk_store] {op}: orphaned source_id_hash={} → deleted source tree tree_id={} summaries={}", + redact_value(source_id), + tree.id, + cascade.removed_summaries, + ); + } + } + let deleted = chunks.len(); tx.commit()?; Ok(deleted) diff --git a/src/openhuman/memory_store/chunks/store_tests.rs b/src/openhuman/memory_store/chunks/store_tests.rs index 06cf8c465..d09cde830 100644 --- a/src/openhuman/memory_store/chunks/store_tests.rs +++ b/src/openhuman/memory_store/chunks/store_tests.rs @@ -297,6 +297,444 @@ fn delete_chunks_by_source_removes_chunks_side_rows_and_ingest_gate() { .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| 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, + }; + + 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 { 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, + }, + 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(); diff --git a/src/openhuman/memory_store/trees/store.rs b/src/openhuman/memory_store/trees/store.rs index 1d4ad71a1..283ccac81 100644 --- a/src/openhuman/memory_store/trees/store.rs +++ b/src/openhuman/memory_store/trees/store.rs @@ -69,6 +69,72 @@ pub(crate) fn insert_tree_conn(conn: &Connection, tree: &Tree) -> Result<()> { Ok(()) } +/// Hard-delete one tree and every dependent row, within an existing tx. +/// +/// Cascade order mirrors [`crate::openhuman::memory_store::chunks::store`]'s +/// global/topic purge: summary sidecars (`summary_embeddings` / +/// `summary_reembed_skipped`, keyed by `summary_id`) first, then +/// `entity_index` + `buffers` (keyed by `tree_id`), then the `summaries`, +/// then the tree row. Used by the chunk-delete path when a source's last +/// chunk is removed, so the now-contentless summary tree (and its unsealed +/// buffer) doesn't outlive the data it summarised. Returns the number of +/// summary rows removed. +pub(crate) fn delete_tree_cascade_tx( + tx: &Transaction<'_>, + tree_id: &str, +) -> Result { + // Collect the on-disk content-file paths BEFORE deleting the summary rows + // — sealed summaries stage their body to `content_path` under the memory + // tree content root (see `bucket_seal::seal_one_level` → `stage_summary`). + // The caller removes these files after the tx commits (mirroring + // `remove_chunk_content_files`), so a `clear_memory` delete doesn't leave + // the summarised text orphaned on disk. + let content_paths: Vec = { + let mut stmt = tx.prepare( + "SELECT content_path FROM mem_tree_summaries \ + WHERE tree_id = ?1 AND content_path IS NOT NULL AND content_path <> ''", + )?; + let rows = stmt.query_map(params![tree_id], |row| row.get::<_, String>(0))?; + rows.collect::>>() + .context("collect summary content paths for tree cascade delete")? + }; + + tx.execute( + "DELETE FROM mem_tree_summary_embeddings WHERE summary_id IN \ + (SELECT id FROM mem_tree_summaries WHERE tree_id = ?1)", + params![tree_id], + )?; + tx.execute( + "DELETE FROM mem_tree_summary_reembed_skipped WHERE summary_id IN \ + (SELECT id FROM mem_tree_summaries WHERE tree_id = ?1)", + params![tree_id], + )?; + tx.execute( + "DELETE FROM mem_tree_entity_index WHERE tree_id = ?1", + params![tree_id], + )?; + let removed_summaries = tx.execute( + "DELETE FROM mem_tree_summaries WHERE tree_id = ?1", + params![tree_id], + )?; + tx.execute( + "DELETE FROM mem_tree_buffers WHERE tree_id = ?1", + params![tree_id], + )?; + tx.execute("DELETE FROM mem_tree_trees WHERE id = ?1", params![tree_id])?; + Ok(TreeCascadeDeletion { + removed_summaries, + content_paths, + }) +} + +/// Outcome of [`delete_tree_cascade_tx`]: how many summary rows were removed +/// and the on-disk content-file paths the caller must delete post-commit. +pub(crate) struct TreeCascadeDeletion { + pub removed_summaries: usize, + pub content_paths: Vec, +} + /// Fetch a tree by `(kind, scope)`. Returns `None` if no such tree exists. pub fn get_tree_by_scope(config: &Config, kind: TreeKind, scope: &str) -> Result> { with_connection(config, |conn| get_tree_by_scope_conn(conn, kind, scope)) diff --git a/tests/memory_sync_sources_raw_coverage_e2e.rs b/tests/memory_sync_sources_raw_coverage_e2e.rs index 976aaac0a..c195bfde4 100644 --- a/tests/memory_sync_sources_raw_coverage_e2e.rs +++ b/tests/memory_sync_sources_raw_coverage_e2e.rs @@ -21,8 +21,9 @@ use openhuman_core::openhuman::credentials::{ }; use openhuman_core::openhuman::memory_sources::readers::SourceReader; use openhuman_core::openhuman::memory_sources::{ - add_source, get_source, list_enabled_by_kind, list_sources, remove_source, update_source, - upsert_composio_source, MemorySourceEntry, MemorySourcePatch, SourceKind, + add_source, get_source, list_enabled_by_kind, list_sources, + remove_composio_source_by_connection_id, remove_source, update_source, upsert_composio_source, + MemorySourceEntry, MemorySourcePatch, SourceKind, }; use openhuman_core::openhuman::memory_sync::composio::bus::{ ComposioConfigChangedSubscriber, ComposioConnectionCreatedSubscriber, ComposioTriggerSubscriber, @@ -202,6 +203,80 @@ async fn memory_sources_registry_persists_crud_and_composio_upserts() { assert_eq!(all[0].id, first.id); } +#[tokio::test] +async fn remove_composio_source_by_connection_id_prunes_on_disconnect_and_survives_reconnect() { + let _guard = env_lock(); + let tmp = TempDir::new().expect("tempdir"); + let config = config_in(&tmp); + let _workspace = EnvGuard::set_path("OPENHUMAN_WORKSPACE", tmp.path()); + let _home = EnvGuard::set_path("HOME", tmp.path()); + let _backend = EnvGuard::unset("BACKEND_URL"); + persist_config(&config).await; + + // Two live composio connections plus an unrelated folder source. + let gmail_old = upsert_composio_source("gmail", "conn-old", "Gmail · conn-old") + .await + .expect("insert gmail"); + upsert_composio_source("slack", "conn-slack", "Slack") + .await + .expect("insert slack"); + let mut folder = source(SourceKind::Folder, "src_folder_disc"); + folder.path = Some(tmp.path().join("notes").to_string_lossy().into_owned()); + folder.glob = Some("**/*.md".to_string()); + add_source(folder.clone()).await.expect("add folder"); + + // No-match is a no-op (returns 0, removes nothing). + assert_eq!( + remove_composio_source_by_connection_id("conn-does-not-exist") + .await + .expect("no-match remove"), + 0 + ); + assert_eq!(list_sources().await.expect("list").len(), 3); + + // Disconnect: prune ONLY the matching composio source, by connection_id. + assert_eq!( + remove_composio_source_by_connection_id("conn-old") + .await + .expect("prune on disconnect"), + 1 + ); + let after_disconnect = list_sources().await.expect("list after disconnect"); + assert_eq!(after_disconnect.len(), 2); + assert!( + after_disconnect.iter().all(|s| s.id != gmail_old.id), + "old gmail entry must be gone" + ); + assert!( + after_disconnect + .iter() + .any(|s| s.connection_id.as_deref() == Some("conn-slack")), + "the other composio connection must be untouched" + ); + assert!( + after_disconnect.iter().any(|s| s.id == folder.id), + "non-composio folder source must be untouched" + ); + + // Reconnect: backend mints a NEW connection_id for the same Gmail account. + // upsert inserts a fresh entry; no stale duplicate is left behind. + let gmail_new = upsert_composio_source("gmail", "conn-new", "Gmail · conn-new") + .await + .expect("reconnect gmail"); + assert_ne!(gmail_new.id, gmail_old.id); + let final_sources = list_sources().await.expect("final list"); + let gmail_entries: Vec<_> = final_sources + .iter() + .filter(|s| s.toolkit.as_deref() == Some("gmail")) + .collect(); + assert_eq!( + gmail_entries.len(), + 1, + "exactly one gmail source after reconnect — no orphan" + ); + assert_eq!(gmail_entries[0].connection_id.as_deref(), Some("conn-new")); +} + #[tokio::test] async fn rss_reader_lists_reads_and_reports_feed_errors_from_loopback() { let _guard = env_lock();