mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
fix(composio): complete connection-disconnect cleanup (config entry + memory tree) (#3140)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
sanil-23
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
511ad84bf4
commit
f24adaca57
@@ -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()),
|
||||
|
||||
@@ -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<String>| 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<String>| 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()
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -137,6 +137,35 @@ pub async fn remove_source(id: &str) -> Result<bool, String> {
|
||||
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<usize, String> {
|
||||
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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<String>| SummaryNode {
|
||||
id: id.into(),
|
||||
tree_id: tree_id.into(),
|
||||
tree_kind: TreeKind::Source,
|
||||
level: 1,
|
||||
parent_id: None,
|
||||
child_ids: children,
|
||||
content: format!("summary for {tree_id}"),
|
||||
token_count: 3,
|
||||
entities: vec![],
|
||||
topics: vec![],
|
||||
time_range_start: ts,
|
||||
time_range_end: ts,
|
||||
score: 0.5,
|
||||
sealed_at: ts,
|
||||
deleted: false,
|
||||
embedding: None,
|
||||
};
|
||||
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
|
||||
tree_store::insert_summary_tx(
|
||||
&tx,
|
||||
&mk_summary("sum-acct", "tree-acct", vec![a0.id.clone(), a1.id.clone()]),
|
||||
None,
|
||||
"test/model@3",
|
||||
)?;
|
||||
tree_store::insert_summary_tx(
|
||||
&tx,
|
||||
&mk_summary("sum-other", "tree-other", vec![b0.id.clone()]),
|
||||
None,
|
||||
"test/model@3",
|
||||
)?;
|
||||
|
||||
// summary sidecars: embeddings for both summaries, reembed-skip only for sum-acct.
|
||||
for sid in ["sum-acct", "sum-other"] {
|
||||
tx.execute(
|
||||
"INSERT INTO mem_tree_summary_embeddings (
|
||||
summary_id, model_signature, vector, dim, created_at
|
||||
) VALUES (?1, 'test/model@3', ?2, 3, 1700000000.0)",
|
||||
params![sid, vec![1_u8, 2, 3]],
|
||||
)?;
|
||||
}
|
||||
tx.execute(
|
||||
"INSERT INTO mem_tree_summary_reembed_skipped (
|
||||
summary_id, model_signature, reason, skipped_at_ms
|
||||
) VALUES ('sum-acct', 'test/model@3', 'terminal', 1700000000000)",
|
||||
[],
|
||||
)?;
|
||||
|
||||
// tree-keyed entity-index rows (summary nodes) for each tree.
|
||||
for (sid, tree_id) in [("sum-acct", "tree-acct"), ("sum-other", "tree-other")] {
|
||||
tx.execute(
|
||||
"INSERT INTO mem_tree_entity_index (
|
||||
entity_id, node_id, node_kind, entity_kind, surface,
|
||||
score, timestamp_ms, tree_id, is_user
|
||||
) VALUES (?1, ?2, 'summary', 'person', 'email', 0.9, 1700000000000, ?3, 0)",
|
||||
params![format!("entity:{sid}"), sid, tree_id],
|
||||
)?;
|
||||
}
|
||||
|
||||
// unsealed buffers (the "queue" frontier) referencing the chunk ids.
|
||||
tree_store::upsert_buffer_tx(
|
||||
&tx,
|
||||
&Buffer {
|
||||
tree_id: "tree-acct".into(),
|
||||
level: 0,
|
||||
item_ids: vec![a0.id.clone(), a1.id.clone()],
|
||||
token_sum: 24,
|
||||
oldest_at: Some(ts),
|
||||
},
|
||||
)?;
|
||||
tree_store::upsert_buffer_tx(
|
||||
&tx,
|
||||
&Buffer {
|
||||
tree_id: "tree-other".into(),
|
||||
level: 0,
|
||||
item_ids: vec![b0.id.clone()],
|
||||
token_sum: 12,
|
||||
oldest_at: Some(ts),
|
||||
},
|
||||
)?;
|
||||
|
||||
assert!(claim_source_ingest_tx(
|
||||
&tx,
|
||||
SourceKind::Email,
|
||||
"gmail:acct",
|
||||
1_700_000_000_000
|
||||
)?);
|
||||
assert!(claim_source_ingest_tx(
|
||||
&tx,
|
||||
SourceKind::Email,
|
||||
"gmail:other",
|
||||
1_700_000_000_000
|
||||
)?);
|
||||
tx.commit()?;
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
// ---- mocked job: a Seal queued for the tree that's about to be deleted ----
|
||||
let seal_payload = queue_types::SealPayload {
|
||||
tree_id: "tree-acct".into(),
|
||||
level: 0,
|
||||
force_now_ms: None,
|
||||
};
|
||||
let job_id = queue_store::enqueue(&cfg, &queue_types::NewJob::seal(&seal_payload).unwrap())
|
||||
.unwrap()
|
||||
.expect("seal job enqueued");
|
||||
|
||||
// ---- act: disconnect conn-1 with clear_memory=true → delete its chunks ----
|
||||
let deleted = delete_chunks_by_owner(&cfg, SourceKind::Email, "gmail-sync:conn-1").unwrap();
|
||||
assert_eq!(deleted, 2);
|
||||
|
||||
// chunks: acct gone, other survives.
|
||||
assert!(get_chunk(&cfg, &a0.id).unwrap().is_none());
|
||||
assert!(get_chunk(&cfg, &a1.id).unwrap().is_none());
|
||||
assert!(get_chunk(&cfg, &b0.id).unwrap().is_some());
|
||||
|
||||
// the orphaned source tree is gone; the sibling tree is untouched.
|
||||
assert!(
|
||||
tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:acct")
|
||||
.unwrap()
|
||||
.is_none()
|
||||
);
|
||||
assert!(
|
||||
tree_store::get_tree_by_scope(&cfg, TreeKind::Source, "gmail:other")
|
||||
.unwrap()
|
||||
.is_some()
|
||||
);
|
||||
|
||||
// exactly the tree-acct rows are cascaded away across every dependent table.
|
||||
with_connection(&cfg, |conn| {
|
||||
let count = |sql: &str| -> rusqlite::Result<i64> { conn.query_row(sql, [], |r| r.get(0)) };
|
||||
assert_eq!(count("SELECT COUNT(*) FROM mem_tree_trees")?, 1);
|
||||
assert_eq!(count("SELECT COUNT(*) FROM mem_tree_summaries")?, 1);
|
||||
assert_eq!(
|
||||
count("SELECT COUNT(*) FROM mem_tree_summary_embeddings")?,
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
count("SELECT COUNT(*) FROM mem_tree_summary_reembed_skipped")?,
|
||||
0
|
||||
);
|
||||
assert_eq!(count("SELECT COUNT(*) FROM mem_tree_buffers")?, 1);
|
||||
assert_eq!(count("SELECT COUNT(*) FROM mem_tree_entity_index")?, 1);
|
||||
// and what survives belongs to tree-other.
|
||||
assert_eq!(
|
||||
count("SELECT COUNT(*) FROM mem_tree_summaries WHERE tree_id = 'tree-other'")?,
|
||||
1
|
||||
);
|
||||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
// ---- the queued Seal job settles to Done (tree missing), not stuck pending ----
|
||||
let claimed = queue_store::claim_next(&cfg, queue_store::DEFAULT_LOCK_DURATION_MS)
|
||||
.unwrap()
|
||||
.expect("seal job claimable");
|
||||
assert_eq!(claimed.kind, queue_types::JobKind::Seal);
|
||||
let outcome = crate::openhuman::memory_queue::handlers::handle_job(&cfg, &claimed)
|
||||
.await
|
||||
.expect("handle_job ok");
|
||||
assert!(
|
||||
matches!(outcome, queue_types::JobOutcome::Done),
|
||||
"seal over a deleted tree must no-op to Done, got {outcome:?}"
|
||||
);
|
||||
queue_store::mark_done(&cfg, &claimed).unwrap();
|
||||
assert_eq!(
|
||||
queue_store::get_job(&cfg, &job_id).unwrap().unwrap().status,
|
||||
queue_types::JobStatus::Done
|
||||
);
|
||||
}
|
||||
|
||||
/// #1: the cascade must also delete the summary's **on-disk content file**, not
|
||||
/// just the row — otherwise a `clear_memory` delete leaves the summarised text
|
||||
/// orphaned on disk.
|
||||
#[test]
|
||||
fn clear_memory_delete_removes_orphaned_summary_content_file() {
|
||||
use crate::openhuman::memory_store::trees::store as tree_store;
|
||||
use crate::openhuman::memory_store::trees::types::{SummaryNode, Tree, TreeKind, TreeStatus};
|
||||
|
||||
let (_tmp, cfg) = test_config();
|
||||
let ts = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
|
||||
|
||||
let mut c = sample_chunk("gmail:acct", 0, 1_700_000_000_000);
|
||||
c.metadata.source_kind = SourceKind::Email;
|
||||
c.metadata.owner = "gmail-sync:conn-1".to_string();
|
||||
upsert_chunks(&cfg, &[c.clone()]).unwrap();
|
||||
|
||||
tree_store::insert_tree(
|
||||
&cfg,
|
||||
&Tree {
|
||||
id: "tree-acct".into(),
|
||||
kind: TreeKind::Source,
|
||||
scope: "gmail:acct".into(),
|
||||
root_id: None,
|
||||
max_level: 1,
|
||||
status: TreeStatus::Active,
|
||||
created_at: ts,
|
||||
last_sealed_at: Some(ts),
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
// A real on-disk summary content file under the memory tree content root.
|
||||
let rel = "summaries/gmail_acct/L1/sum-acct.md";
|
||||
let abs = cfg.memory_tree_content_root().join(rel);
|
||||
std::fs::create_dir_all(abs.parent().unwrap()).unwrap();
|
||||
std::fs::write(&abs, "summarised email body").unwrap();
|
||||
assert!(abs.exists());
|
||||
|
||||
with_connection(&cfg, |conn| {
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
tree_store::insert_summary_tx(
|
||||
&tx,
|
||||
&SummaryNode {
|
||||
id: "sum-acct".into(),
|
||||
tree_id: "tree-acct".into(),
|
||||
tree_kind: TreeKind::Source,
|
||||
level: 1,
|
||||
parent_id: None,
|
||||
child_ids: vec![c.id.clone()],
|
||||
content: "preview".into(),
|
||||
token_count: 3,
|
||||
entities: vec![],
|
||||
topics: vec![],
|
||||
time_range_start: ts,
|
||||
time_range_end: ts,
|
||||
score: 0.5,
|
||||
sealed_at: ts,
|
||||
deleted: false,
|
||||
embedding: None,
|
||||
},
|
||||
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();
|
||||
|
||||
@@ -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<TreeCascadeDeletion> {
|
||||
// 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<String> = {
|
||||
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::<rusqlite::Result<Vec<_>>>()
|
||||
.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<String>,
|
||||
}
|
||||
|
||||
/// 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<Option<Tree>> {
|
||||
with_connection(config, |conn| get_tree_by_scope_conn(conn, kind, scope))
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user