From 9b61391f4bd9f5e4ecac284082f4714230de5894 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 3 Sep 2026 21:07:51 +0530 Subject: [PATCH] fix(sources): route connector items into the memory tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `MemorySourceSink::accept_source_items` wrote namespace documents and vector chunks but never `mem_tree_chunks`, so a Composio sync landed thousands of documents and embeddings while every tree-backed surface — tree recall, the Memory Tree graph, the source row's ingest status — still reported zero. None of those read `memory_docs`. This is the #5473 reconnect restored on the connector path. The migration to `accept_source_items` bypassed the `SkillDocSink` path that carried it rather than extending it, so the scope rules are now one shared function both paths call instead of a method on the old path's adapter: - `ingest_connector_item_into_tree` owns the `{toolkit}:{connection_id}` tree scope, the `{...}:{item_id}` per-item key and the blank-half guard. That key pair is what OpenHuman's `source_id_prefix` counts a Composio source by, so a drift empties the source status and the graph together, silently. - `ingest_connector_item_tolerated` owns the failure policy, so `corruption::escalate_or_count` stays `pub(crate)` and no call site can implement a weaker one. A tree failure is counted and warned, never fatal: propagating lets one poisonous item stall a connection and re-bill the Composio page on every retry. Corruption still aborts the run. Gated on `source_kind == "composio"` *and* a non-empty `{toolkit}:{connection}` split. Both halves: OpenHuman derives a toolkit-shaped ingest prefix only for Composio and keys every other kind `mem_src:{id}:`, so treeing a non-Composio source would write rows nothing can count — the same invisible-write bug in a different costume. `forget_source` also had to learn the per-item key. It deleted chunks by exact source id, which never matches `{source_id}:{item_id}`, so once items are treed a user who disconnected Gmail would keep every synced message retrievable. The prefix sweep is unconditional: it matches nothing for a source that never had per-item rows, and a shape gate here could drift out of step with the write gate, which is the direction that leaks. Tests: the non-connector source asserts *zero* tree rows (that is the gate); a connector source asserts the exact per-item id and `path_scope`, then asserts `forget_source` takes those rows with it. The pre-existing #5473 test is unchanged and still green, which is what makes the refactor behavior-neutral. Refs openhuman#6007 --- crates/tinymemory-core/src/engine/mod.rs | 5 +- crates/tinymemory-core/src/engine/sync.rs | 227 +++++++++++------- .../tinymemory-core/src/engine/sync_tests.rs | 65 ++++- .../tinymemory-tinycortex/src/engine/mod.rs | 106 +++++++- .../tests/full_provider_conformance.rs | 112 +++++++++ 5 files changed, 415 insertions(+), 100 deletions(-) diff --git a/crates/tinymemory-core/src/engine/mod.rs b/crates/tinymemory-core/src/engine/mod.rs index 39e37469..ff4e8745 100644 --- a/crates/tinymemory-core/src/engine/mod.rs +++ b/crates/tinymemory-core/src/engine/mod.rs @@ -65,8 +65,9 @@ pub use seal::{ }; pub use summariser::HostSummariser; pub use sync::{ - estimate_cost_usd, needs_rebuild, raw_coverage, read_audit_log, rebuild_tree_from_raw, - run_github_sync, run_source_pipeline, sync_context, HostSyncAdapter, RawCoverage, RawFileRef, + estimate_cost_usd, ingest_connector_item_into_tree, ingest_connector_item_tolerated, + needs_rebuild, raw_coverage, read_audit_log, rebuild_tree_from_raw, run_github_sync, + run_source_pipeline, sync_context, HostSyncAdapter, RawCoverage, RawFileRef, RealCostAccumulator, RebuildOutcome, SourcePipelineFailure, HOST_SYNC_STATE_NAMESPACE, }; // Crate-private seam for `crate::sources::sync` (openhuman#5820); not host surface. diff --git a/crates/tinymemory-core/src/engine/sync.rs b/crates/tinymemory-core/src/engine/sync.rs index 9f4497ea..a7af19d6 100644 --- a/crates/tinymemory-core/src/engine/sync.rs +++ b/crates/tinymemory-core/src/engine/sync.rs @@ -94,68 +94,127 @@ impl HostSyncAdapter { self.tree_ingest_failures .load(std::sync::atomic::Ordering::Relaxed) } +} - /// Reconnect a synced Composio document to the memory tree (#5473). - /// - /// The TinyCortex migration (#4794) dropped the per-provider tree-ingest - /// half of the connector sync: synced items reached the `skill-` - /// document store but never `mem_tree_chunks`, so connector memories fell - /// out of tree-backed recall. This routes each synced item through the - /// engine's document ingest — the same L0-chunk path local folder sources - /// use via [`LocalDocumentSink`] — additively alongside the skill store. - /// - /// Scope naming matches the tree retrieval contract: the tree scope - /// (`path_scope`) is `"{toolkit}:{connection_id}"` so `query_source` resolves - /// it by platform prefix (`gmail:` → email, `slack:` → chat, …), while the - /// per-item `source_id` carries the document id so each message admits - /// independently rather than colliding on one dedup key. - /// - /// `ingest_document` writes the L0 chunk rows synchronously and enqueues the - /// summary seal on the async extract worker. Retrieval (`query_source`) reads - /// sealed summaries, so an item becomes retrievable once its buffer seals — - /// on the token threshold or the time-based `flush_stale_buffers` — and the - /// seal degrades to a fallback summary when no LLM is available. - async fn ingest_document_into_memory_tree( - &self, - config: &Config, - document: &SkillDocument, - ) -> anyhow::Result<()> { - let toolkit = document.toolkit.trim().to_ascii_lowercase(); - let connection_id = document.connection_id.trim(); - // A blank toolkit/connection would yield a scope with no platform prefix - // (`":conn"`), which no retrieval kind matches; skip rather than write an - // unreachable tree. The skill store still holds the item. - if toolkit.is_empty() || connection_id.is_empty() { - tracing::debug!( - document_id = %document.document_id, - "[tinycortex:sync] skipping memory-tree ingest: item has no toolkit/connection scope" - ); - return Ok(()); - } - let tree_scope = format!("{toolkit}:{connection_id}"); - let source_id = format!("{tree_scope}:{}", document.document_id); - let owner = format!("{toolkit}-sync:{connection_id}"); - let input = tinycortex::memory::ingest::canonicalize::document::DocumentInput { - provider: format!("composio:{toolkit}"), - title: document.title.clone(), - body: document.content.clone(), - modified_at: chrono::Utc::now(), - source_ref: Some(document.document_id.clone()), - }; - crate::ingest_pipeline::ingest_document_with_scope( - config, - &source_id, - &owner, - vec![toolkit], - input, - Some(tree_scope), - ) - .await - .map(|_| ()) - .map_err(|error| { - anyhow::anyhow!("memory-tree ingest failed for source `{source_id}`: {error}") - }) +/// Reconnect one synced connector item to the memory tree (#5473). +/// +/// The TinyCortex migration (#4794) dropped the per-provider tree-ingest half of +/// the connector sync: synced items reached the `skill-` document store +/// but never `mem_tree_chunks`, so connector memories fell out of tree-backed +/// recall. This routes each synced item through the engine's document ingest — +/// the same L0-chunk path local folder sources use via [`LocalDocumentSink`] — +/// additively alongside whichever store holds the item. +/// +/// A `pub` free function rather than a method, because owning these rules in ONE +/// place is the actual fix for openhuman#6007. #5473 put them on +/// [`HostSyncAdapter`], the *old* `SkillDocSink` path's adapter. The connector +/// migration then built a second ingest path — `MemorySourceSink::accept_source_items` +/// in `tinymemory-tinycortex` — which wrote namespace documents and vector chunks +/// but never learned the tree half, so Gmail synced, embedded, and stayed +/// invisible to every tree-backed surface. A fix that patched only that second +/// call site would leave the same trap for a third. Both paths now call this. +/// +/// Scope naming matches the tree retrieval contract: the tree scope +/// (`path_scope`) is `"{toolkit}:{connection_id}"` so `query_source` resolves it +/// by platform prefix (`gmail:` → email, `slack:` → chat, …), while the per-item +/// `source_id` carries the item id so each message admits independently rather +/// than colliding on one dedup key. That pair is *also* the literal prefix +/// OpenHuman counts a Composio source's ingest by (`"{toolkit}:{connection_id}:"`, +/// its `source_id_prefix`), so a drift here empties the source row's status and +/// the memory graph together — silently, because the documents and vectors are +/// still written. +/// +/// `ingest_document_with_scope` writes the L0 chunk rows synchronously and +/// enqueues the summary seal on the async extract worker. Retrieval +/// (`query_source`) reads sealed summaries, so an item becomes retrievable once +/// its buffer seals — on the token threshold or the time-based +/// `flush_stale_buffers` — and the seal degrades to a fallback summary when no +/// LLM is available. Chunk rows existing while recall is still thin is that +/// latency, not a second bug. +pub async fn ingest_connector_item_into_tree( + config: &Config, + toolkit: &str, + connection_id: &str, + item_id: &str, + title: &str, + content: &str, +) -> anyhow::Result<()> { + let toolkit = toolkit.trim().to_ascii_lowercase(); + let connection_id = connection_id.trim(); + // A blank toolkit/connection would yield a scope with no platform prefix + // (`":conn"`), which no retrieval kind matches; skip rather than write an + // unreachable tree. The caller's own store still holds the item. + if toolkit.is_empty() || connection_id.is_empty() { + tracing::debug!( + item_id = %item_id, + "[tinycortex:sync] skipping memory-tree ingest: item has no toolkit/connection scope" + ); + return Ok(()); } + let tree_scope = format!("{toolkit}:{connection_id}"); + let source_id = format!("{tree_scope}:{item_id}"); + let owner = format!("{toolkit}-sync:{connection_id}"); + let input = tinycortex::memory::ingest::canonicalize::document::DocumentInput { + provider: format!("composio:{toolkit}"), + title: title.to_string(), + body: content.to_string(), + modified_at: chrono::Utc::now(), + source_ref: Some(item_id.to_string()), + }; + crate::ingest_pipeline::ingest_document_with_scope( + config, + &source_id, + &owner, + vec![toolkit], + input, + Some(tree_scope), + ) + .await + .map(|_| ()) + .map_err(|error| anyhow::anyhow!("memory-tree ingest failed for source `{source_id}`: {error}")) +} + +/// [`ingest_connector_item_into_tree`] plus the failure policy every connector +/// sync path needs, so no path has to reach for `crate::corruption` itself. +/// +/// The tree is a secondary index over a store that has *already* committed, so an +/// ordinary failure here must NOT abort the sync run. Most providers do not +/// tolerate scope errors, so a propagated error becomes a run-aborting `Err` in +/// the orchestrator, and one deterministically-poisonous item then stalls the +/// whole connection and re-fetches the page — real Composio spend — on every +/// retry. Count it, warn, and continue: the per-item source gate re-attempts the +/// item on a later sync, and an operator rebuild can backfill. +/// +/// Corruption is the exception (openhuman#5820). A malformed `chunks.db` fails +/// every later item identically, so it escalates through the shared recovery and +/// aborts the run — there is nothing per-item about it. That split belongs to +/// `crate::corruption::escalate_or_count`, which stays `pub(crate)` deliberately: +/// callers reach the *policy* through this function rather than the primitive, so +/// a new call site cannot quietly implement a weaker one. +pub async fn ingest_connector_item_tolerated( + config: &Config, + toolkit: &str, + connection_id: &str, + item_id: &str, + title: &str, + content: &str, + counter: &std::sync::atomic::AtomicU32, +) -> anyhow::Result<()> { + if let Err(error) = + ingest_connector_item_into_tree(config, toolkit, connection_id, item_id, title, content) + .await + { + let rendered = format!("{error:#}"); + crate::corruption::escalate_or_count("connector tree ingest", config, error, counter)?; + tracing::warn!( + toolkit = %toolkit, + connection_id = %connection_id, + item_id = %item_id, + error = %rendered, + "[tinycortex:sync] memory-tree ingest failed; the item's own store write is retained" + ); + } + Ok(()) } /// Read persisted sync audit records for best-effort RPC and reporting surfaces. @@ -491,41 +550,25 @@ impl SkillDocSink for HostSyncAdapter { .await .map_err(anyhow::Error::msg)?; - // #5473: additively reconnect the synced item to the memory tree. This - // is a best-effort secondary index over the skill store, which is the - // source of truth and has already committed above. An ordinary failure - // here must NOT abort the connector sync: most providers do not - // tolerate scope errors, so the orchestrator turns a `store` error - // into a run-aborting `Err` — propagating would let one - // deterministically-poisonous item stall the whole connection and - // re-fetch the page (Composio spend) on every retry. Log, count, and - // continue; the per-item source gate re-attempts the item on a later - // sync, and an operator rebuild can backfill. Corruption is the - // exception (openhuman#5820): a malformed `chunks.db` fails every - // later item identically, so it escalates through the shared recovery - // and aborts the run — there is nothing per-item about it. + // #5473: additively reconnect the synced item to the memory tree. The + // skill store above is the source of truth and has already committed; + // `ingest_connector_item_tolerated` owns both the scope rules and the + // best-effort-except-corruption policy, so this path and the connector + // path in `tinymemory-tinycortex` cannot drift apart again (openhuman#6007). + // // The config-less adapter (`sync_context`) has no ingest pipeline and is // not on the connector sync path, so it skips tree ingest entirely. if let Some(config) = self.config.as_deref() { - if let Err(error) = self - .ingest_document_into_memory_tree(config, &document) - .await - { - let rendered = format!("{error:#}"); - crate::corruption::escalate_or_count( - "connector tree ingest", - config, - error, - &self.tree_ingest_failures, - )?; - tracing::warn!( - toolkit = %document.toolkit, - connection_id = %document.connection_id, - document_id = %document.document_id, - error = %rendered, - "[tinycortex:sync] memory-tree ingest failed; skill store retained" - ); - } + ingest_connector_item_tolerated( + config, + &document.toolkit, + &document.connection_id, + &document.document_id, + &document.title, + &document.content, + &self.tree_ingest_failures, + ) + .await?; } Ok(()) } diff --git a/crates/tinymemory-core/src/engine/sync_tests.rs b/crates/tinymemory-core/src/engine/sync_tests.rs index c9ff08fd..35189b36 100644 --- a/crates/tinymemory-core/src/engine/sync_tests.rs +++ b/crates/tinymemory-core/src/engine/sync_tests.rs @@ -545,10 +545,16 @@ async fn tree_ingest_failure_is_tolerated_and_skill_store_is_retained() { // ingest resolves its store path elsewhere), this fires rather than the // test silently passing without exercising the tolerance path. assert!( - adapter - .ingest_document_into_memory_tree(&*config, &document) - .await - .is_err(), + super::ingest_connector_item_into_tree( + &*config, + &document.toolkit, + &document.connection_id, + &document.document_id, + &document.title, + &document.content, + ) + .await + .is_err(), "the broken tree-ingest workspace must make ingest fail" ); @@ -705,3 +711,54 @@ async fn blank_scope_item_is_skipped_for_memory_tree_ingest() { "an item without a toolkit/connection scope must be skipped for tree ingest" ); } + +/// The shared funnel's blank-scope guard, on the half the sink-path test above +/// cannot reach. +/// +/// [`blank_scope_item_is_skipped_for_memory_tree_ingest`] covers a blank +/// *toolkit* through `SkillDocSink::store`. A blank *connection_id* is the other +/// way to form an unreachable scope (`"gmail:"`), and no sink path can produce +/// one — a `SkillDocument` always carries its connection. Since openhuman#6007 +/// the funnel is also called directly by the connector path's +/// `accept_source_items`, whose `source_id` is split at the first colon, so both +/// halves are now reachable from a caller and the guard belongs to the funnel +/// rather than to either call site. +/// +/// A skip is `Ok`, not an error: the caller's own store holds the item, and +/// failing the sync over an item that simply cannot be scoped would stall the +/// whole connection. +#[tokio::test] +async fn the_shared_funnel_skips_either_blank_scope_half() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace.path().join("workspace"); + let config = host.to_arc(); + + for (toolkit, connection_id, blank_half) in [ + (" ", "conn-1", "toolkit"), + ("gmail", " ", "connection_id"), + ] { + super::ingest_connector_item_into_tree( + &*config, + toolkit, + connection_id, + "msg-1", + "Quarterly planning", + "Let's finalise the Q3 roadmap.", + ) + .await + .unwrap_or_else(|error| { + panic!("a blank {blank_half} must be skipped, not an error: {error:#}") + }); + + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + 0, + "a blank {blank_half} forms an unreachable tree scope and must write no chunks" + ); + } +} diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index d5c92c8d..9334974c 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -335,6 +335,12 @@ pub struct TinycortexProvider { mandatory: MemoryTraitProvider, client: MemoryClientRef, config: EngineRuntimeConfig, + /// Items whose namespace-document write committed but whose (non-corrupt) + /// memory-tree ingest failed — the tolerated warns in + /// [`MemorySourceSink::accept_source_items`] (openhuman#6007). The same + /// counter `HostSyncAdapter` keeps for the old sink path, so the shared + /// funnel's corruption escalation has somewhere to record a tolerated miss. + tree_ingest_failures: std::sync::atomic::AtomicU32, } impl TinycortexProvider { @@ -351,6 +357,7 @@ impl TinycortexProvider { mandatory, client, config, + tree_ingest_failures: std::sync::atomic::AtomicU32::new(0), } } @@ -2147,6 +2154,35 @@ fn parse_source_kind( tinymemory_core::store::chunks::SourceKind::parse(kind).map_err(MemoryError::Invalid) } +/// The source kind OpenHuman's connector sync labels its batches with, and the +/// only kind whose items are keyed for the memory tree. +const COMPOSIO_SOURCE_KIND: &str = "composio"; + +/// The `{toolkit}:{connection_id}` halves of a connector-keyed source id, for a +/// source whose items belong in the memory tree (openhuman#6007). +/// +/// Both conditions, not either. The kind gate is the load-bearing half: +/// OpenHuman derives a toolkit-shaped ingest prefix (`source_id_prefix`) only for +/// `SourceKind::Composio` and keys every other kind `mem_src:{id}:`, so treeing a +/// non-Composio source would write rows that no status or graph surface can count +/// — the same invisible-write bug this fixes, wearing a different source kind. +/// The shape gate is the `":conn"` guard the shared funnel documents: a blank half +/// yields a scope with no platform prefix, which no retrieval kind resolves. +/// +/// `split_once` splits on the FIRST colon deliberately, so a connection id that +/// contains one lands wholly in `connection_id` and the derived per-item key stays +/// `{toolkit}:{connection_id}:{item_id}` — the literal prefix OpenHuman counts by. +fn connector_tree_scope<'a>(source_kind: &str, source_id: &'a str) -> Option<(&'a str, &'a str)> { + if source_kind != COMPOSIO_SOURCE_KIND { + return None; + } + let (toolkit, connection_id) = source_id.split_once(':')?; + if toolkit.trim().is_empty() || connection_id.trim().is_empty() { + return None; + } + Some((toolkit, connection_id)) +} + #[async_trait] impl MemorySourceSink for TinycortexProvider { async fn accept_source_items( @@ -2159,6 +2195,9 @@ impl MemorySourceSink for TinycortexProvider { let items_len = items.len(); let namespace = format!("source:{source_id}"); let mut outcome = IngestOutcome::default(); + // Resolved once: whether a source's items belong in the memory tree is a + // property of the source, not of the item. + let tree_scope = connector_tree_scope(source_kind, source_id); for item in items { if item.item_id.trim().is_empty() { return Err(MemoryError::Invalid( @@ -2170,6 +2209,12 @@ impl MemorySourceSink for TinycortexProvider { } else { item.title.clone() }; + // Captured before these fields move into `NamespaceDocumentInput`, + // because the tree ingest below has to receive the same body the + // document store does. Only a tree-scoped source pays the clones — + // `tree_scope` is `None` for every other kind. + let tree_item = + tree_scope.map(|_| (item.item_id.clone(), title.clone(), item.content.clone())); let input = NamespaceDocumentInput { namespace: namespace.clone(), key: item.item_id, @@ -2195,6 +2240,44 @@ impl MemorySourceSink for TinycortexProvider { Ok(id) => { outcome.written = outcome.written.saturating_add(1); outcome.ids.push(id); + // openhuman#6007: the namespace document is the source of + // truth and has just committed — now feed the same item to + // the memory tree, which is what tree-backed recall, the + // Memory Tree graph and the source row's ingest status all + // read. None of those look at `memory_docs`, so a sync + // without this writes documents and embeddings and still + // reports zero: exactly the "Gmail synced nothing" report. + // + // This restores the #5473 reconnect that the connector + // migration bypassed; the funnel is shared with the old sink + // path so the two cannot drift apart again. + // + // Best-effort by contract — see the policy on + // `ingest_connector_item_tolerated`. Only store corruption + // returns `Err`, and aborting is right there: it fails every + // later item identically. + if let (Some((toolkit, connection_id)), Some((item_id, title, content))) = + (tree_scope, tree_item) + { + tinymemory_core::engine::ingest_connector_item_tolerated( + &self.config, + toolkit, + connection_id, + &item_id, + &title, + &content, + &self.tree_ingest_failures, + ) + .await + .map_err(|error| { + MemoryError::Other(anyhow::anyhow!( + "memory-tree store is corrupt after {} of {} item(s) \ + were written: {error:#}", + outcome.written, + items_len + )) + })?; + } } // A write failure is NOT `skipped`. The contract defines that // field as "units the driver recognised as already present" @@ -2236,11 +2319,30 @@ impl MemorySourceSink for TinycortexProvider { let source_id = source_id.to_string(); let chunks = blocking(self.config.clone(), "clear source chunks", move |config| { use tinymemory_core::store::chunks::{ - delete_chunks_by_source, delete_orphaned_source_tree, SourceKind, + delete_chunks_by_source, delete_chunks_by_source_prefix, + delete_orphaned_source_tree, SourceKind, }; let removed = delete_chunks_by_source(config, SourceKind::Document, &source_id)?; + // openhuman#6007: connector items are treed under a PER-ITEM key + // (`{toolkit}:{connection_id}:{item_id}`), and the delete above + // matches a source id *exactly*. Without this sweep, forgetting a + // Composio source clears its documents and leaves every synced + // message retrievable in the tree — a user who disconnects Gmail + // would keep getting answers out of their mail. Unconditional on + // purpose: the prefix matches nothing for a source that never had + // per-item rows, and gating it on the id's shape could drift out of + // step with the write gate, which is the direction that leaks. + // + // The bare-id `delete_orphaned_source_tree` below still lands: the + // items' shared `path_scope` IS this source id, so the summary tree + // over them is keyed by it. + let per_item = delete_chunks_by_source_prefix( + config, + SourceKind::Document, + &format!("{source_id}:"), + )?; delete_orphaned_source_tree(config, SourceKind::Document, &source_id)?; - Ok(removed) + Ok(removed.saturating_add(per_item)) }) .await?; Ok(u64::try_from(documents.saturating_add(chunks)).unwrap_or(u64::MAX)) diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 632230d4..3b8e8258 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -752,6 +752,21 @@ async fn document_source_graph_goals_and_tool_rule_state_transitions_round_trip( .await .expect("accept source item"); assert_eq!(outcome.written, 1); + // openhuman#6007's gate, asserted at the non-connector source: a kind that + // is not `composio` must reach the document store and NOTHING in the memory + // tree. OpenHuman keys a non-Composio source's ingest `mem_src:{id}:` and + // derives a toolkit-shaped prefix only for Composio, so treeing this row + // would write chunks that no status or graph surface can count — the same + // invisible-write bug #6007 fixes, wearing a different source kind. + assert_eq!( + tinymemory_core::store::chunks::count_chunks(&provider_config( + workspace.path(), + serde_json::Value::Null + )) + .expect("count chunks"), + 0, + "a non-connector source must not write memory-tree chunks (openhuman#6007)" + ); assert_eq!( source .forget_source("drive-1") @@ -3904,3 +3919,100 @@ async fn the_flavour_door_answers_none_until_a_body_exists_then_serves_the_whole ); assert!(served.starts_with("---\n")); } + +/// openhuman#6007: a connector sync must land in the memory tree, not only in the +/// namespace document store. +/// +/// The tree is what tree-backed recall, the Memory Tree graph and the source +/// row's ingest status all read; none of them look at `memory_docs`. Before this +/// fix a Gmail sync wrote thousands of documents and vector chunks and every +/// tree-backed surface still reported zero — "Gmail synced nothing", with the +/// data sitting right there. This is the #5473 guarantee restored on the +/// *connector* path, which the migration to `accept_source_items` bypassed. +/// +/// The identity is asserted exactly, not loosely, because it is what OpenHuman +/// counts a Composio source's ingest by (`source_id_prefix` → +/// `"{toolkit}:{connection_id}:"`) and a drift is silent in both directions. +#[tokio::test(flavor = "multi_thread")] +async fn a_connector_sync_reaches_the_memory_tree_and_is_forgotten_with_its_source() { + use tinymemory_api::provider::types::SourceItem; + use tinymemory_api::provider::MemoryProvider; + use tinymemory_api::types::MemoryTaint; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let config = provider_config(workspace.path(), serde_json::Value::Null); + let source = provider.as_sources().expect("SourceSink"); + + // Precondition, so the counts below are attributable to this sync rather + // than to pre-existing state. + assert_eq!( + tinymemory_core::store::chunks::count_chunks(&config).expect("count chunks"), + 0, + "a fresh workspace must start with an empty memory tree" + ); + + let outcome = source + .accept_source_items( + "gmail:conn-1", + "composio", + vec![SourceItem { + item_id: "msg-1".into(), + title: "Quarterly planning".into(), + content: "Let's finalise the Q3 roadmap and align on the launch date.".into(), + mime: Some("text/plain".into()), + url: Some("https://example.invalid/msg-1".into()), + updated_at_ms: Some(42), + tags: vec!["gmail".into()], + }], + MemoryTaint::ExternalSync, + ) + .await + .expect("accept a connector item"); + assert_eq!( + outcome.written, 1, + "`written` counts namespace documents; the tree is a secondary index over \ + them and must not inflate the caller's written count" + ); + + // The per-item key is `{toolkit}:{connection_id}:{item_id}` so each message + // admits independently instead of colliding on one dedup key, and the shared + // `path_scope` is `{toolkit}:{connection_id}` — the platform prefix tree + // retrieval resolves by (`gmail:` → email). + let treed = tinymemory_core::store::chunks::list_chunks( + &config, + &tinymemory_core::store::chunks::ListChunksQuery { + source_id: Some("gmail:conn-1:msg-1".into()), + limit: Some(8), + ..Default::default() + }, + ) + .expect("list chunks by source id"); + assert!( + !treed.is_empty(), + "a connector sync must add memory-tree chunks keyed by the deterministic \ + per-item connector source id (openhuman#6007)" + ); + assert!( + treed + .iter() + .all(|chunk| chunk.metadata.path_scope.as_deref() == Some("gmail:conn-1")), + "connector chunks must carry the `{{toolkit}}:{{connection_id}}` tree scope so \ + query_source resolves them (gmail → email)" + ); + + // Forgetting the source must take the per-item rows with it. The delete that + // already existed matches a source id EXACTLY, so without the prefix sweep a + // user who disconnects Gmail keeps every synced message retrievable in the + // tree — the documents go and the memories stay. + source + .forget_source("gmail:conn-1") + .await + .expect("forget the connector source"); + assert_eq!( + tinymemory_core::store::chunks::count_chunks(&config).expect("count chunks"), + 0, + "forgetting a connector source must remove its per-item tree rows too \ + (openhuman#6007)" + ); +}