From f663cbe9782bdd545bd3f361bcdd66b8a17d2e4e Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 3 Sep 2026 23:15:01 +0530 Subject: [PATCH 1/5] feat(maintenance): backfill stored connector documents into the memory tree MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #134 fixed the routing: connector items reach `mem_tree_chunks` as they sync. It could not fix the records already on disk, and neither can a re-sync — the per-item sync gate treats an ingested document as done, so it fetches nothing and creates no tree rows. Those memories stay fully embedded in the namespace store and invisible to tree recall, the memory graph and the source row's ingest status, and recovery today means removing and re-adding the account. `MemoryMaintenance::backfill_connector_trees` walks the connector namespaces and feeds each stored document through the funnel the sync path already uses, so a backfilled row and a freshly-synced row are the same row. The walk lives in `tinymemory-core::backfill` next to that funnel and the provider is a thin conversion: reconstructing `{toolkit}:{connection_id}:{item_id}` at a second call site is exactly what caused openhuman#6007. Namespaces are rebuilt from the source registry rather than parsed out of `list_namespaces`, because the registry is what the writers used and cannot drift, while parsing a namespace back into halves has to guess where the toolkit ends. Legacy `skill-` documents record no connection at all — `store_skill_sync` takes an `_integration_id` it never persists — so they are attached only where the registry holds exactly one connection for the toolkit, and skipped by name where it holds several. A wrong attribution in a memory system is worse than a missing one. `ingest_connector_item_into_tree` now answers `Option` instead of `()`. Without it the backfill cannot tell "just treed this" from "the tree already had it", and that distinction is the difference between reporting progress and reporting nothing. Both existing callers drop the payload, and the funnel's own test now asserts `None` on a blank scope rather than discarding the value. The member is appended at the tail of `METHODS`, not filed beside `FlushPending` where its family sits. Member order is wire order, so filing it with its family would renumber every member after it and make a host built against v1.13.8 invoke the wrong method — silently. The append-only guard caught that; the slot is now pinned at 142. Idempotent by construction: the ingest pipeline answers `already_ingested` when its transaction persists nothing, so a second pass writes nothing and an interrupted pass loses nothing. `limit` therefore bounds cost rather than carrying a cursor. `dry_run` reports what a pass would examine while writing nothing, because a full pass is one read and one set of chunk embeddings per document and that cost should be visible before it is paid. Nothing calls this automatically. The null driver refuses, as every other mutating Maintenance member does: "backfilled nothing" must not read as work done by a driver that stores nothing. Refs tinyhumansai/openhuman#6012, tinyhumansai/openhuman#6007 --- crates/tinymemory-api/src/null.rs | 12 +- crates/tinymemory-api/src/provider/mod.rs | 8 +- crates/tinymemory-api/src/provider/records.rs | 39 ++- crates/tinymemory-bus/src/names.rs | 11 +- crates/tinymemory-bus/src/names_tests.rs | 8 +- crates/tinymemory-bus/src/provider/types.rs | 43 +++ crates/tinymemory-core/src/backfill.rs | 273 ++++++++++++++++++ crates/tinymemory-core/src/backfill_tests.rs | 235 +++++++++++++++ crates/tinymemory-core/src/engine/sync.rs | 15 +- .../tinymemory-core/src/engine/sync_tests.rs | 9 +- crates/tinymemory-core/src/lib.rs | 1 + crates/tinymemory-module/src/service/mod.rs | 10 + .../tinymemory-tinycortex/src/engine/mod.rs | 35 ++- 13 files changed, 681 insertions(+), 18 deletions(-) create mode 100644 crates/tinymemory-core/src/backfill.rs create mode 100644 crates/tinymemory-core/src/backfill_tests.rs diff --git a/crates/tinymemory-api/src/null.rs b/crates/tinymemory-api/src/null.rs index 9ac12b0d..675c5d92 100644 --- a/crates/tinymemory-api/src/null.rs +++ b/crates/tinymemory-api/src/null.rs @@ -58,8 +58,9 @@ use crate::health::MemoryHealth; use crate::learning::LearningCandidate; use crate::operations::{AnswerRequest, AnswerResponse, RawMemoryEvent}; use crate::provider::types::{ - DiffReport, EntityHit, ExportPage, ExportRecord, FlushOutcome, ImportOutcome, IngestItem, - IngestOutcome, MaintenanceReport, ResetOutcome, SnapshotRef, SourceItem, SourceScope, + BackfillTreesOutcome, BackfillTreesRequest, DiffReport, EntityHit, ExportPage, ExportRecord, + FlushOutcome, ImportOutcome, IngestItem, IngestOutcome, MaintenanceReport, ResetOutcome, + SnapshotRef, SourceItem, SourceScope, }; use crate::provider::{ AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, CodingSessionIngestReport, @@ -543,6 +544,13 @@ impl MemoryMaintenance for NullMemoryProvider { unsupported(Capability::Maintenance) } + async fn backfill_connector_trees( + &self, + _request: BackfillTreesRequest, + ) -> Result { + unsupported(Capability::Maintenance) + } + async fn reset_derived_index(&self) -> Result { unsupported(Capability::Maintenance) } diff --git a/crates/tinymemory-api/src/provider/mod.rs b/crates/tinymemory-api/src/provider/mod.rs index daf018d1..ad7b72cc 100644 --- a/crates/tinymemory-api/src/provider/mod.rs +++ b/crates/tinymemory-api/src/provider/mod.rs @@ -131,8 +131,8 @@ pub use sync::{ SyncAuditEntry, SyncFreshness, SyncRunOutcome, }; pub use types::{ - ChangeKind, ChunkEntityOccurrence, DiffReport, EntityHit, EntityOccurrence, EntityRef, - ExportPage, ExportRecord, FlushOutcome, ForgetOutcome, ForgetSelector, ImportOutcome, - IngestItem, IngestOutcome, MaintenanceReport, PurgeOutcome, ResetOutcome, SnapshotRef, - SourceChange, SourceItem, SourceScope, + BackfillTreesOutcome, BackfillTreesRequest, ChangeKind, ChunkEntityOccurrence, DiffReport, + EntityHit, EntityOccurrence, EntityRef, ExportPage, ExportRecord, FlushOutcome, ForgetOutcome, + ForgetSelector, ImportOutcome, IngestItem, IngestOutcome, MaintenanceReport, PurgeOutcome, + ResetOutcome, SnapshotRef, SourceChange, SourceItem, SourceScope, }; diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs index 1eca486a..f1f66417 100644 --- a/crates/tinymemory-api/src/provider/records.rs +++ b/crates/tinymemory-api/src/provider/records.rs @@ -21,8 +21,9 @@ use crate::error::MemoryError; use crate::goals::GoalsDoc; use crate::provider::diagnosis::{DegradedCapabilities, Diagnosis}; use crate::provider::types::{ - FlushOutcome, ForgetOutcome, ForgetSelector, IngestOutcome, MaintenanceReport, PurgeOutcome, - QueueFailure, QueueStats, ResetOutcome, SourceItem, StoreStats, + BackfillTreesOutcome, BackfillTreesRequest, FlushOutcome, ForgetOutcome, ForgetSelector, + IngestOutcome, MaintenanceReport, PurgeOutcome, QueueFailure, QueueStats, ResetOutcome, + SourceItem, StoreStats, }; use crate::tool_memory::ToolMemoryRule; use crate::types::MemoryTaint; @@ -338,6 +339,40 @@ pub trait MemoryMaintenance: Send + Sync { Ok(FlushOutcome::default()) } + /// Re-file already-stored connector documents into the memory tree (#6012). + /// + /// openhuman#6007 fixed the routing for items synced *from now on*. It could + /// not fix the records already stored: the per-item sync gate treats an + /// ingested document as done, so re-syncing fetches nothing and creates no + /// tree rows. Those memories stay fully embedded in the document store and + /// invisible to every tree-backed surface until something re-files them. + /// + /// Idempotent, and by construction rather than by bookkeeping — the ingest + /// gate answers `already_ingested` for a document the tree already holds, so + /// a second pass writes nothing and an interrupted pass loses nothing. That + /// is also why `limit` bounds cost rather than carrying a cursor. + /// + /// Expensive on purpose to call explicitly: a pass is one read and one set + /// of chunk embeddings per document. A driver must not run this on its own + /// initiative — spending a user's embedding budget unasked is its own bug + /// (openhuman#5324). + /// + /// Defaulted to an empty outcome, matching [`Self::flush_pending`]: a driver + /// with no connector documents has nothing to re-file, which is a fact about + /// it rather than a refusal. A driver that stores nothing at all should + /// override and refuse instead, so "backfilled nothing" cannot read as work + /// done. + /// + /// # Errors + /// + /// Backend failures only. + async fn backfill_connector_trees( + &self, + _request: BackfillTreesRequest, + ) -> Result { + Ok(BackfillTreesOutcome::default()) + } + /// Drop everything derived from stored content and schedule its /// re-derivation. /// diff --git a/crates/tinymemory-bus/src/names.rs b/crates/tinymemory-bus/src/names.rs index 69ff8f68..0221ec6c 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -212,6 +212,10 @@ pub mod methods { pub const RECALL_NAMESPACE_RECENT: &str = "RecallNamespaceRecent"; /// `FlushPending` — flush buffered work old enough to be written out. pub const FLUSH_PENDING: &str = "FlushPending"; + /// `BackfillConnectorTrees` — file already-stored connector documents into + /// the memory tree, for records synced before the routing fix + /// (openhuman#6007) reached them. + pub const BACKFILL_CONNECTOR_TREES: &str = "BackfillConnectorTrees"; /// `ResetDerivedIndex` — drop derived state and schedule its rebuild. pub const RESET_DERIVED_INDEX: &str = "ResetDerivedIndex"; /// `PurgeAll` — erase every row the driver holds. @@ -373,7 +377,7 @@ pub mod methods { /// The order matters: `tinybus`'s `Interface::members()` returns declaration /// order, and the module compares the two sequences directly rather than as /// sets, so a reordering is caught alongside an addition or a removal. -pub const METHODS: [&str; 142] = [ +pub const METHODS: [&str; 143] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -516,6 +520,11 @@ pub const METHODS: [&str; 142] = [ methods::INGEST_EVENT, methods::ANSWER, methods::OVERRIDE_SCHEDULER_GATE, + // Connector-backfill round (openhuman#6012): appended at the tail, per + // this table's append-only rule — filing it beside `FlushPending`, where + // its family lives, would renumber every member after it and invoke the + // wrong method on a host built against an earlier release. + methods::BACKFILL_CONNECTOR_TREES, ]; #[cfg(test)] diff --git a/crates/tinymemory-bus/src/names_tests.rs b/crates/tinymemory-bus/src/names_tests.rs index 568852a6..e6b04e6d 100644 --- a/crates/tinymemory-bus/src/names_tests.rs +++ b/crates/tinymemory-bus/src/names_tests.rs @@ -104,6 +104,12 @@ fn the_newest_members_are_appended_rather_than_filed_with_their_family() { // append-only rule. assert_eq!(METHODS[141], methods::OVERRIDE_SCHEDULER_GATE); assert_eq!(methods::OVERRIDE_SCHEDULER_GATE, "OverrideSchedulerGate"); + // Connector-backfill round (openhuman#6012): slot 142. Its family sits at + // 137 (`FlushPending`), and filing it there is exactly what this test + // exists to catch — it renumbers 137 onward and a host built against + // v1.13.8 then invokes the wrong member. + assert_eq!(METHODS[142], methods::BACKFILL_CONNECTOR_TREES); + assert_eq!(methods::BACKFILL_CONNECTOR_TREES, "BackfillConnectorTrees"); } #[test] @@ -144,7 +150,7 @@ fn the_runtime_tree_doors_hold_the_wire_slots_they_were_released_in() { // reason the summariser-door test above gives: member order is wire order, // and an assertion measured from the end moves silently under the next // append — which is exactly the edit this exists to catch. - assert_eq!(METHODS.len(), 142); + assert_eq!(METHODS.len(), 143); assert_eq!(METHODS[131], methods::RUNTIME_BUFFER_WRITE); assert_eq!(METHODS[132], methods::RUNTIME_READ_NODE); assert_eq!(METHODS[133], methods::RUNTIME_READ_CHILDREN); diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index f9b40c0d..faa3acf5 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -781,6 +781,49 @@ pub struct FlushOutcome { pub stale_buffers: u64, } +/// What one connector-tree backfill pass examined and wrote (#6012). +/// +/// Four counters rather than one, because "did nothing" has three very +/// different causes a caller has to be able to tell apart: the tree already +/// held everything (`already_present`), nothing could be addressed +/// (`skipped`), or there was nothing to look at (`scanned: 0`). Collapsing +/// them would make an account whose scope could not be resolved read exactly +/// like one that is fully backfilled. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillTreesOutcome { + /// Documents examined this pass. + pub scanned: u64, + /// Documents that produced new memory-tree rows. + pub ingested: u64, + /// Documents the tree already held. Not a failure: this is what makes a + /// repeated pass readable as "nothing left to do". + pub already_present: u64, + /// Documents left alone — no resolvable scope, or a tolerated failure. + /// Never filed under a guess. + pub skipped: u64, + /// Whether the pass stopped on its limit with documents still unexamined. + /// The caller resumes by calling again; there is no cursor to carry. + pub more_pending: bool, + /// Bounded, human-readable reasons behind `skipped`. + pub notes: Vec, +} + +/// How much of the backfill to attempt, and whether to write at all. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BackfillTreesRequest { + /// Documents to examine at most. `None` leaves the bound to the driver. + /// + /// A bound rather than a cursor because the work is idempotent: the pass + /// re-reads what it already treed and the ingest gate answers + /// `already_ingested`, so resuming is just calling again. + pub limit: Option, + /// Report what a real pass would examine, and write nothing. + /// + /// The honest way to show an operator the size of the job before they pay + /// for it — a full pass is one read and one embedding per document. + pub dry_run: bool, +} + /// What resetting the derived index deleted, requeued and scheduled. /// /// Three numbers rather than a `MaintenanceReport`'s two, because they are not diff --git a/crates/tinymemory-core/src/backfill.rs b/crates/tinymemory-core/src/backfill.rs new file mode 100644 index 00000000..57215c68 --- /dev/null +++ b/crates/tinymemory-core/src/backfill.rs @@ -0,0 +1,273 @@ +//! Re-file already-stored connector documents into the memory tree (#6012). +//! +//! openhuman#6007 fixed the *routing*: connector items now reach +//! `mem_tree_chunks` as they are synced. It did nothing for the records already +//! on disk, and it cannot — the per-item sync gate treats an ingested document +//! as done, so re-syncing fetches nothing and creates no tree rows. On the +//! profile that reported the bug that is ~3000 documents, fully embedded in the +//! namespace store and invisible to every tree-backed surface. +//! +//! This walks the connector namespaces and feeds each stored document through +//! the same funnel the sync path uses, so a backfilled row and a freshly-synced +//! one are the same row. It deliberately reads +//! [`crate::engine::ingest_connector_item_into_tree`] rather than re-deriving +//! the `{toolkit}:{connection_id}` identity: two call sites owning one rule is +//! exactly what produced #6007. +//! +//! # Idempotent by construction, not by bookkeeping +//! +//! The ingest pipeline answers `already_ingested` when its transaction persists +//! nothing, so running this twice writes nothing the second time. There is no +//! watermark to keep and no way for an interrupted run to corrupt anything — +//! the worst case is repeated work. `limit` exists to bound *cost*, not to +//! guarantee correctness. +//! +//! # Why it costs what it costs +//! +//! `list_documents` carries no `content` column, so each document needs its own +//! read, and each ingest embeds its chunks. A full pass over a large mailbox is +//! thousands of reads and thousands of embeddings — which is why nothing calls +//! this automatically. It is an operator action with a `dry_run` preview, not +//! something that should fire on upgrade and quietly spend a user's embedding +//! budget (openhuman#5324). + +use std::collections::BTreeMap; + +use crate::sources::SourceKind; +use crate::store::MemoryClientRef; +use crate::Config; + +/// Documents examined per pass when the caller names no bound. +/// +/// Deliberately modest: a pass is resumable (just call again), and a caller +/// that wants the whole account can say so. The default protects the operator +/// who clicks once without reading the cost note above. +pub const DEFAULT_BACKFILL_LIMIT: u64 = 500; + +/// How many distinct skip reasons to carry back before truncating. +/// +/// The notes are for a human deciding what to do next, and the same reason +/// repeated a thousand times tells them nothing the first one did not. +const MAX_NOTES: usize = 20; + +/// What one backfill pass examined and wrote. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct BackfillReport { + /// Documents examined. + pub scanned: u64, + /// Documents that produced new memory-tree rows. + pub ingested: u64, + /// Documents the tree already held. Not a failure — this is the counter + /// that makes a repeated run readable as "nothing left to do". + pub already_present: u64, + /// Documents left alone: no resolvable scope, or a tolerated read/ingest + /// failure. Never filed under a guess. + pub skipped: u64, + /// Whether the pass stopped on its limit with documents still unexamined. + pub more_pending: bool, + /// Bounded, human-readable reasons behind `skipped`. + pub notes: Vec, +} + +impl BackfillReport { + fn note(&mut self, reason: String) { + if self.notes.len() < MAX_NOTES && !self.notes.contains(&reason) { + self.notes.push(reason); + } + } +} + +/// One namespace to sweep, and the tree scope its documents belong to. +struct Target { + namespace: String, + toolkit: String, + connection_id: String, +} + +/// Walk the connector namespaces, feeding stored documents into the memory tree. +/// +/// `dry_run` reports what a real pass would examine without reading any content +/// or writing anything, which is the only honest way to show an operator the +/// size of the job before they pay for it. +pub async fn backfill_connector_trees( + config: &Config, + client: &MemoryClientRef, + limit: Option, + dry_run: bool, +) -> anyhow::Result { + let limit = limit.unwrap_or(DEFAULT_BACKFILL_LIMIT); + let mut report = BackfillReport::default(); + let targets = resolve_targets(config, &mut report)?; + + // Tolerated (non-corrupt) per-document failures, counted the same way the + // connector sync counts them. A corrupt store still aborts: it fails every + // later document identically, so continuing would burn the whole limit + // producing the same error (openhuman#5820). + let failures = std::sync::atomic::AtomicU32::new(0); + + 'targets: for target in targets { + let listed = match client.list_documents(Some(&target.namespace)).await { + Ok(listed) => listed, + Err(error) => { + report.note(format!( + "{}: could not be listed ({error})", + target.namespace + )); + continue; + } + }; + let documents = listed + .get("documents") + .and_then(serde_json::Value::as_array) + .cloned() + .unwrap_or_default(); + + for document in documents { + if report.scanned >= limit { + report.more_pending = true; + break 'targets; + } + let Some(key) = document.get("key").and_then(serde_json::Value::as_str) else { + // A document row with no key cannot be read back or addressed + // in the tree; counting it as skipped keeps `scanned` honest. + report.skipped = report.skipped.saturating_add(1); + report.note(format!( + "{}: a document row carries no key", + target.namespace + )); + continue; + }; + report.scanned = report.scanned.saturating_add(1); + if dry_run { + continue; + } + + let stored = match client.get_document(&target.namespace, key).await { + Ok(Some(stored)) => stored, + // Listed but unreadable: it was deleted between the list and + // the read, or the row is damaged. Neither is worth failing the + // pass over. + Ok(None) => { + report.skipped = report.skipped.saturating_add(1); + continue; + } + Err(error) => { + report.skipped = report.skipped.saturating_add(1); + report.note(format!( + "{}: a document could not be read ({error})", + target.namespace + )); + continue; + } + }; + + match crate::engine::ingest_connector_item_into_tree( + config, + &target.toolkit, + &target.connection_id, + key, + &stored.title, + &stored.content, + ) + .await + { + Ok(Some(result)) if result.already_ingested => { + report.already_present = report.already_present.saturating_add(1); + } + Ok(Some(_)) => report.ingested = report.ingested.saturating_add(1), + // The funnel refused the scope. It was built from the registry + // above, so this is close to unreachable — but counting it is + // cheaper than assuming it cannot happen. + Ok(None) => report.skipped = report.skipped.saturating_add(1), + Err(error) => { + let rendered = format!("{error:#}"); + crate::corruption::escalate_or_count( + "connector tree backfill", + config, + error, + &failures, + )?; + report.skipped = report.skipped.saturating_add(1); + report.note(format!( + "{}: an ingest failed ({rendered})", + target.namespace + )); + } + } + } + } + + tracing::info!( + scanned = report.scanned, + ingested = report.ingested, + already_present = report.already_present, + skipped = report.skipped, + more_pending = report.more_pending, + dry_run, + "[tinycortex:backfill] connector tree backfill pass complete" + ); + Ok(report) +} + +/// The namespaces worth sweeping, derived from the source registry. +/// +/// Built from the registry rather than from `list_namespaces`, because the +/// registry is what the *writers* used: `accept_source_items` composes +/// `source:{toolkit}:{connection_id}` from the same row, so reconstructing it +/// the same way cannot drift. Parsing a namespace string back into its halves +/// would have to guess where the toolkit ends. +/// +/// The legacy `skill-{toolkit}` namespaces are the awkward half. +/// `store_skill_sync` took an `_integration_id` it never persisted, so those +/// pre-migration documents record no connection at all — and the tree scope +/// needs one. Where the registry holds exactly one connection for the toolkit +/// there is only one answer and it is used; where it holds several there is no +/// way to tell which account a document came from, and a wrong attribution in a +/// memory system is worse than a missing one, so they are skipped and named. +fn resolve_targets(config: &Config, report: &mut BackfillReport) -> anyhow::Result> { + let sources = crate::sources::registry::list_sources_in(config).map_err(anyhow::Error::msg)?; + + let mut targets = Vec::new(); + let mut by_toolkit: BTreeMap> = BTreeMap::new(); + + for source in sources.iter().filter(|s| s.kind == SourceKind::Composio) { + let (Some(toolkit), Some(connection_id)) = + (source.toolkit.as_deref(), source.connection_id.as_deref()) + else { + continue; + }; + let toolkit = toolkit.trim().to_ascii_lowercase(); + let connection_id = connection_id.trim().to_string(); + if toolkit.is_empty() || connection_id.is_empty() { + continue; + } + targets.push(Target { + namespace: format!("source:{toolkit}:{connection_id}"), + toolkit: toolkit.clone(), + connection_id: connection_id.clone(), + }); + by_toolkit.entry(toolkit).or_default().push(connection_id); + } + + for (toolkit, connections) in &by_toolkit { + match connections.as_slice() { + [only] => targets.push(Target { + namespace: format!("skill-{toolkit}"), + toolkit: toolkit.clone(), + connection_id: only.clone(), + }), + several => report.note(format!( + "skill-{toolkit}: skipped — {} connections are registered for this toolkit and \ + the pre-migration documents record none, so the account they belong to cannot \ + be determined", + several.len() + )), + } + } + + Ok(targets) +} + +#[cfg(test)] +#[path = "backfill_tests.rs"] +mod tests; diff --git a/crates/tinymemory-core/src/backfill_tests.rs b/crates/tinymemory-core/src/backfill_tests.rs new file mode 100644 index 00000000..e629c85d --- /dev/null +++ b/crates/tinymemory-core/src/backfill_tests.rs @@ -0,0 +1,235 @@ +use std::sync::Arc; + +use tinymemory_api::host::test_support::TestHostConfig; +use tinymemory_api::host::MemoryHostConfig; + +use crate::sources::MemorySourceEntry; +use crate::store::{MemoryClient, MemoryClientRef}; + +fn composio_source(id: &str, toolkit: &str, connection_id: &str) -> MemorySourceEntry { + serde_json::from_value(serde_json::json!({ + "id": id, + "kind": "composio", + "label": "Test connector", + "enabled": true, + "toolkit": toolkit, + "connection_id": connection_id, + })) + .expect("a valid composio source entry") +} + +/// A workspace with a registry, a store client, and the stub seams installed. +/// +/// Opening the client starts the ingestion queue, so every caller has to be a +/// `#[tokio::test]` even when the body itself does no awaiting. +fn workspace( + sources: &[MemorySourceEntry], +) -> (tempfile::TempDir, Arc, MemoryClientRef) { + crate::test_seams::init(); + let dir = tempfile::tempdir().expect("workspace"); + let workspace_dir = dir.path().join("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace_dir.clone(); + // The source registry is written beside the host's config file, so the + // default's empty `config_path` has to be given a real one here. + host.config_path = dir.path().join("config.toml"); + let config = host.to_arc(); + crate::sources::registry::replace_sources_in(&*config, sources).expect("write the registry"); + let client: MemoryClientRef = Arc::new( + MemoryClient::from_workspace_dir(workspace_dir).expect("memory client initialises"), + ); + (dir, config, client) +} + +async fn store_document(client: &MemoryClientRef, namespace: &str, key: &str, body: &str) { + client + .put_doc(tinymemory_api::types::NamespaceDocumentInput { + namespace: namespace.to_string(), + key: key.to_string(), + title: "Quarterly planning".into(), + content: body.to_string(), + source_type: "composio".into(), + priority: "medium".into(), + tags: vec!["gmail".into()], + metadata: serde_json::json!({}), + category: "core".into(), + session_id: None, + document_id: None, + taint: tinymemory_api::types::MemoryTaint::ExternalSync, + }) + .await + .expect("store a connector document"); +} + +/// The namespaces are rebuilt from the registry the writers used, so a +/// connected toolkit yields both its current namespace and — because it has +/// exactly one connection — its pre-migration `skill-` one. +#[tokio::test] +async fn targets_are_rebuilt_from_the_registry_including_the_legacy_namespace() { + let (_dir, config, _client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]); + let mut report = super::BackfillReport::default(); + let targets = super::resolve_targets(&*config, &mut report).expect("resolve targets"); + + let namespaces: Vec<&str> = targets.iter().map(|t| t.namespace.as_str()).collect(); + assert!( + namespaces.contains(&"source:gmail:conn-1"), + "the current namespace must be swept: {namespaces:?}" + ); + assert!( + namespaces.contains(&"skill-gmail"), + "one connection means the legacy namespace is unambiguous: {namespaces:?}" + ); + assert!( + report.notes.is_empty(), + "nothing was skipped, so nothing should be reported: {:?}", + report.notes + ); +} + +/// Two accounts on one toolkit make the legacy documents unattributable, and a +/// wrong attribution in a memory system is worse than a missing one. The +/// current namespaces are still swept — only the shared `skill-` one is not. +#[tokio::test] +async fn the_legacy_namespace_is_skipped_when_a_toolkit_has_several_connections() { + let (_dir, config, _client) = workspace(&[ + composio_source("src_a", "gmail", "conn-1"), + composio_source("src_b", "gmail", "conn-2"), + ]); + let mut report = super::BackfillReport::default(); + let targets = super::resolve_targets(&*config, &mut report).expect("resolve targets"); + + let namespaces: Vec<&str> = targets.iter().map(|t| t.namespace.as_str()).collect(); + assert!( + !namespaces.contains(&"skill-gmail"), + "an ambiguous legacy namespace must not be guessed at: {namespaces:?}" + ); + assert!( + namespaces.contains(&"source:gmail:conn-1") && namespaces.contains(&"source:gmail:conn-2"), + "both current namespaces stay addressable: {namespaces:?}" + ); + assert!( + report.notes.iter().any(|n| n.contains("skill-gmail")), + "the skip must say which namespace and why: {:?}", + report.notes + ); +} + +/// The point of the whole issue: a document stored before the routing fix +/// reaches the memory tree, under the identity the sync path would have given +/// it — and a second pass writes nothing, because the ingest gate recognises it. +#[tokio::test] +async fn a_stored_document_is_filed_into_the_tree_and_never_twice() { + let (_dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]); + store_document( + &client, + "source:gmail:conn-1", + "msg-1", + "Let's finalise the Q3 roadmap and align on the launch date.", + ) + .await; + + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + 0, + "the document store alone must leave the tree empty — that IS openhuman#6007" + ); + + let first = super::backfill_connector_trees(&*config, &client, None, false) + .await + .expect("backfill"); + assert_eq!( + first.ingested, 1, + "the stored document must be treed: {first:?}" + ); + assert_eq!( + first.already_present, 0, + "nothing was there before: {first:?}" + ); + + // The identity has to match what the sync path writes, because that is the + // prefix OpenHuman counts a Composio source's ingest by. + let treed = crate::store::chunks::store::list_chunks( + &*config, + &tinycortex::memory::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(), + "backfilled rows must carry the per-item connector source id" + ); + assert!( + treed + .iter() + .all(|chunk| chunk.metadata.path_scope.as_deref() == Some("gmail:conn-1")), + "backfilled rows must carry the platform-prefixed path_scope, or retrieval never \ + resolves them" + ); + + let chunks_after_first = + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"); + + // Idempotence is the property that makes this safe to re-run and safe to + // interrupt: no watermark, just an ingest gate that recognises its own work. + let second = super::backfill_connector_trees(&*config, &client, None, false) + .await + .expect("second backfill"); + assert_eq!( + second.ingested, 0, + "a second pass must write nothing: {second:?}" + ); + assert_eq!( + second.already_present, 1, + "and must say why it wrote nothing: {second:?}" + ); + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + chunks_after_first, + "a repeated pass must not duplicate chunks" + ); +} + +/// The preview an operator gets before paying for a full pass: it counts what +/// it would examine and writes nothing at all. +#[tokio::test] +async fn a_dry_run_counts_without_writing() { + let (_dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]); + store_document(&client, "source:gmail:conn-1", "msg-1", "Q3 roadmap.").await; + + let report = super::backfill_connector_trees(&*config, &client, None, true) + .await + .expect("dry run"); + + assert_eq!( + report.scanned, 1, + "a dry run still reports the size of the job" + ); + assert_eq!(report.ingested, 0, "a dry run must not ingest"); + assert_eq!( + crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), + 0, + "a dry run must leave the tree untouched" + ); +} + +/// `limit` bounds cost, and says so rather than looking like a finished pass. +#[tokio::test] +async fn a_bounded_pass_reports_that_more_is_pending() { + let (_dir, config, client) = workspace(&[composio_source("src_gmail", "gmail", "conn-1")]); + for key in ["msg-1", "msg-2", "msg-3"] { + store_document(&client, "source:gmail:conn-1", key, "Q3 roadmap.").await; + } + + let report = super::backfill_connector_trees(&*config, &client, Some(2), true) + .await + .expect("bounded dry run"); + + assert_eq!(report.scanned, 2, "the limit is respected: {report:?}"); + assert!( + report.more_pending, + "a pass that stopped on its limit must not read as complete: {report:?}" + ); +} diff --git a/crates/tinymemory-core/src/engine/sync.rs b/crates/tinymemory-core/src/engine/sync.rs index a7af19d6..3d696123 100644 --- a/crates/tinymemory-core/src/engine/sync.rs +++ b/crates/tinymemory-core/src/engine/sync.rs @@ -131,6 +131,14 @@ impl HostSyncAdapter { /// `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. +/// +/// Answers `Ok(None)` when the item was skipped for want of a scope, and +/// `Ok(Some(result))` when it reached the pipeline. Callers that only care +/// whether it failed drop the payload; the backfill (#6012) reads +/// `already_ingested` off it to tell a document it has just treed from one the +/// tree already held. That distinction has to come from here rather than from a +/// second call site re-deriving the scope rules, because re-deriving them is +/// what caused openhuman#6007 in the first place. pub async fn ingest_connector_item_into_tree( config: &Config, toolkit: &str, @@ -138,7 +146,7 @@ pub async fn ingest_connector_item_into_tree( item_id: &str, title: &str, content: &str, -) -> anyhow::Result<()> { +) -> 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 @@ -149,7 +157,7 @@ pub async fn ingest_connector_item_into_tree( item_id = %item_id, "[tinycortex:sync] skipping memory-tree ingest: item has no toolkit/connection scope" ); - return Ok(()); + return Ok(None); } let tree_scope = format!("{toolkit}:{connection_id}"); let source_id = format!("{tree_scope}:{item_id}"); @@ -170,7 +178,7 @@ pub async fn ingest_connector_item_into_tree( Some(tree_scope), ) .await - .map(|_| ()) + .map(Some) .map_err(|error| anyhow::anyhow!("memory-tree ingest failed for source `{source_id}`: {error}")) } @@ -203,6 +211,7 @@ pub async fn ingest_connector_item_tolerated( if let Err(error) = ingest_connector_item_into_tree(config, toolkit, connection_id, item_id, title, content) .await + .map(|_| ()) { let rendered = format!("{error:#}"); crate::corruption::escalate_or_count("connector tree ingest", config, error, counter)?; diff --git a/crates/tinymemory-core/src/engine/sync_tests.rs b/crates/tinymemory-core/src/engine/sync_tests.rs index 35189b36..1411f4d1 100644 --- a/crates/tinymemory-core/src/engine/sync_tests.rs +++ b/crates/tinymemory-core/src/engine/sync_tests.rs @@ -742,7 +742,7 @@ async fn the_shared_funnel_skips_either_blank_scope_half() { (" ", "conn-1", "toolkit"), ("gmail", " ", "connection_id"), ] { - super::ingest_connector_item_into_tree( + let outcome = super::ingest_connector_item_into_tree( &*config, toolkit, connection_id, @@ -754,6 +754,13 @@ async fn the_shared_funnel_skips_either_blank_scope_half() { .unwrap_or_else(|error| { panic!("a blank {blank_half} must be skipped, not an error: {error:#}") }); + // `None` IS the skip contract (#6012): the backfill tells "skipped for + // want of a scope" from "ingested" by this, so a skip that started + // answering `Some` would silently be counted as work done. + assert!( + outcome.is_none(), + "a blank {blank_half} must report a skip (`None`), not an ingest" + ); assert_eq!( crate::store::chunks::store::count_chunks(&*config).expect("count chunks"), diff --git a/crates/tinymemory-core/src/lib.rs b/crates/tinymemory-core/src/lib.rs index 079670fe..4bb24fb8 100644 --- a/crates/tinymemory-core/src/lib.rs +++ b/crates/tinymemory-core/src/lib.rs @@ -31,6 +31,7 @@ /// why its return types are shaped the way they are. pub type Config = dyn tinymemory_api::host::MemoryHostConfig; +pub mod backfill; pub mod chat; pub mod chat_host; pub mod config_loader; diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 9ccbd0ad..7132ed6d 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -1026,6 +1026,16 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn backfill_connector_trees( + &self, + request: BackfillTreesRequest, + ) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .backfill_connector_trees(request) + .await + .map_err(|error| into_bus_error(&error)) + } + async fn reset_derived_index(&self) -> BusResult { require_family!(self, as_maintenance, Capability::Maintenance) .reset_derived_index() diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 9334974c..de4a0b3d 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -32,10 +32,10 @@ use tinymemory_api::host::{ }; use tinymemory_api::mandatory::MemoryTraitProvider; use tinymemory_api::provider::types::{ - ChunkEntityOccurrence, EntityHit, EntityOccurrence, EntityRef, ExportPage, ExportRecord, - FlushOutcome, ForgetOutcome, ForgetSelector, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, PurgeOutcome, QueueFailure, QueueStats, ResetOutcome, SourceItem, - SourceScope, StoreStats, + BackfillTreesOutcome, BackfillTreesRequest, ChunkEntityOccurrence, EntityHit, EntityOccurrence, + EntityRef, ExportPage, ExportRecord, FlushOutcome, ForgetOutcome, ForgetSelector, + ImportOutcome, IngestItem, IngestOutcome, MaintenanceReport, PurgeOutcome, QueueFailure, + QueueStats, ResetOutcome, SourceItem, SourceScope, StoreStats, }; // Diff-family value types, used only by the `MemoryDiff` impl below — which is // compiled out without the git-backed snapshot store. @@ -2692,6 +2692,33 @@ impl MemoryMaintenance for TinycortexProvider { .await } + async fn backfill_connector_trees( + &self, + request: BackfillTreesRequest, + ) -> Result { + // The walk, the registry read and the scope rules all live in + // `tinymemory-core`, next to the funnel the connector sync uses. This + // adapter only carries the shape across: a backfilled row and a + // freshly-synced one have to be the same row, and they can only stay + // that way while one function writes both (openhuman#6007). + let report = tinymemory_core::backfill::backfill_connector_trees( + &self.config, + &self.client, + request.limit, + request.dry_run, + ) + .await + .map_err(|error| Self::other("backfill connector trees", error))?; + Ok(BackfillTreesOutcome { + scanned: report.scanned, + ingested: report.ingested, + already_present: report.already_present, + skipped: report.skipped, + more_pending: report.more_pending, + notes: report.notes, + }) + } + async fn reset_derived_index(&self) -> Result { blocking(self.config.clone(), "reset derived index", move |config| { // Everything here is derived from `mem_tree_chunks`, which is NOT From 4898f93a67401f58ab2a99e0c030b20b3df2661c Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 3 Sep 2026 23:34:47 +0530 Subject: [PATCH 2/5] fix(module): wire the backfill member into the module's own workspace MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The module is excluded from the root workspace and carries its own Cargo.lock, so `cargo check --workspace`, `cargo clippy --all-targets --all-features` and `cargo test --all-features` at the repo root never compile it. All three were green with this crate broken in two ways. - The contract types were never imported into `service/mod.rs`. The identical import was needed in `null.rs` and in the tinycortex provider, where the root workspace surfaced it immediately; this copy stayed broken and invisible. - The forwarder went in beside `flush_pending`, where its family sits. `#[tinybus::interface]` derives member order from the impl block, and `the_served_members_are_exactly_the_published_contract` compares that order positionally against `tinymemory_bus::METHODS`. So the same append-only rule that governs the wire table governs this block: a member inserted mid-impl renumbers every member after it. Moved to the tail, with the reason recorded where the next person will be tempted to tidy it back. The manifest in `lib.rs` also had to learn the name. `#[tinybus::interface]` serves whatever the impl declares, and `every_served_method_is_declared_in_the_manifest` fails on a member that is served but undeclared — which is the state where no host can call it. That list is compared as a set, so it stays grouped with its family; only the two ordered lists are append-only. Refs tinyhumansai/openhuman#6012 --- crates/tinymemory-module/src/lib.rs | 6 ++++ crates/tinymemory-module/src/service/mod.rs | 34 ++++++++++++--------- 2 files changed, 26 insertions(+), 14 deletions(-) diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index b56ef58d..afa4bb2e 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -646,6 +646,12 @@ mod exports { "LatestQueueFailure", "BackfillInProgress", "FlushPending", + // Re-files connector documents stored before the routing fix + // (openhuman#6007) into the memory tree. Declared beside its + // family here because this list is compared as a SET; the + // wire-order table in `tinymemory_bus::METHODS` is the one + // that is append-only. + "BackfillConnectorTrees", "ResetDerivedIndex", "PurgeAll", "RecallNamespaceRecent", diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 7132ed6d..3a822dc4 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -169,10 +169,10 @@ use tinymemory_api::health::MemoryHealth; use tinymemory_api::learning::LearningCandidate; use tinymemory_api::operations::{AnswerRequest, AnswerResponse, RawMemoryEvent}; use tinymemory_api::provider::types::{ - ChunkEntityOccurrence, DiffReport, EntityHit, EntityOccurrence, ExportPage, ExportRecord, - FlushOutcome, ForgetOutcome, ForgetSelector, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, PurgeOutcome, QueueFailure, QueueStats, ResetOutcome, SnapshotRef, - SourceItem, SourceScope, StoreStats, + BackfillTreesOutcome, BackfillTreesRequest, ChunkEntityOccurrence, DiffReport, EntityHit, + EntityOccurrence, ExportPage, ExportRecord, FlushOutcome, ForgetOutcome, ForgetSelector, + ImportOutcome, IngestItem, IngestOutcome, MaintenanceReport, PurgeOutcome, QueueFailure, + QueueStats, ResetOutcome, SnapshotRef, SourceItem, SourceScope, StoreStats, }; // `MemoryCore`, `MemoryRecall` and `MemoryPortability` are deliberately not // imported: they are supertraits of `MemoryProvider`, so their methods are @@ -1026,16 +1026,6 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } - async fn backfill_connector_trees( - &self, - request: BackfillTreesRequest, - ) -> BusResult { - require_family!(self, as_maintenance, Capability::Maintenance) - .backfill_connector_trees(request) - .await - .map_err(|error| into_bus_error(&error)) - } - async fn reset_derived_index(&self) -> BusResult { require_family!(self, as_maintenance, Capability::Maintenance) .reset_derived_index() @@ -2249,6 +2239,22 @@ impl MemoryService { ); Ok(()) } + + // Appended at the tail, not filed beside `flush_pending` where its family + // sits. `#[tinybus::interface]` derives member order from this block, and + // `the_served_members_are_exactly_the_published_contract` compares that + // order positionally against `tinymemory_bus::METHODS` — which is + // append-only for the same reason: a member inserted mid-list renumbers + // every member after it (openhuman#6012). + async fn backfill_connector_trees( + &self, + request: BackfillTreesRequest, + ) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .backfill_connector_trees(request) + .await + .map_err(|error| into_bus_error(&error)) + } } /// The response-size ceiling for a method that returns a list of entries. From 06e2030d6c47e9a88c209e7748b4d76a3c08deac Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 3 Sep 2026 23:41:22 +0530 Subject: [PATCH 3/5] fix(core): route the backfill test's chunk query through the seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `engine-containment.sh` (#18 §C1) fails a `tinymemory-core` file outside `src/engine/` that names tinycortex in code, and the new backfill test named `tinycortex::memory::chunks::ListChunksQuery` directly. The import was copied from `engine/sync_tests.rs`, where that spelling is legal precisely because that file is inside the seam. Copying it out of the seam is what broke containment — the same query is reachable as `crate::store::chunks::ListChunksQuery`, which is how the other non-engine callers name it. Refs tinyhumansai/openhuman#6012 --- crates/tinymemory-core/src/backfill_tests.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/tinymemory-core/src/backfill_tests.rs b/crates/tinymemory-core/src/backfill_tests.rs index e629c85d..7a7bb164 100644 --- a/crates/tinymemory-core/src/backfill_tests.rs +++ b/crates/tinymemory-core/src/backfill_tests.rs @@ -148,9 +148,14 @@ async fn a_stored_document_is_filed_into_the_tree_and_never_twice() { // The identity has to match what the sync path writes, because that is the // prefix OpenHuman counts a Composio source's ingest by. + // Named through `crate::store::chunks`, not `tinycortex::…`. This file sits + // outside `src/engine/`, which is the seam, and `engine-containment.sh` + // fails a core file outside it that reaches the engine in code. The sibling + // in `engine/sync_tests.rs` may spell it the other way because it is inside + // the seam — copying an import across that boundary is what broke here. let treed = crate::store::chunks::store::list_chunks( &*config, - &tinycortex::memory::chunks::ListChunksQuery { + &crate::store::chunks::ListChunksQuery { source_id: Some("gmail:conn-1:msg-1".into()), limit: Some(8), ..Default::default() From 67334beaae42385624af0c982a1a0b088e3155c5 Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 4 Sep 2026 00:01:54 +0530 Subject: [PATCH 4/5] test(module): drive the backfill through the port, with a real store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The module workspace runs only its own tests, so `tinymemory-core`'s backfill suite never executes in that lane while its coverage gate still measures core's production source. `backfill.rs` was landing there as ~270 uncovered lines. Adding it to the ignore-regex would have cleared the gate and hidden a real gap, so this covers it instead — and the test earns its place independently: nothing else exercised service -> provider -> core for this member, which is the path a host actually calls. The gate was pointing at something true. The document is written straight through the store client rather than with `accept_source_items`, deliberately. That member now trees on the way in, so using it would leave the backfill nothing to do; a document sitting in the namespace store with no tree row IS the state this feature repairs. Idempotence is asserted here as well as in core, because a host will offer this as a button and someone will press it twice. Refs tinyhumansai/openhuman#6012 --- crates/tinymemory-module/src/service/test.rs | 108 +++++++++++++++++++ 1 file changed, 108 insertions(+) diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index dc6bc17e..7252f3df 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -1041,3 +1041,111 @@ async fn override_member_opens_a_window_that_outranks_a_paused_gate() { assert!(matches!(gate::current_policy(), Policy::Paused { .. })); gate::clear_scheduler_gate(); } + +/// openhuman#6012: the backfill reaches core *through this port*, with a real +/// store underneath — not only in `tinymemory-core`'s own suite. +/// +/// Worth having here for two independent reasons. The module workspace runs +/// only its own tests, so core's backfill suite never executes in this lane +/// while the coverage gate still measures core's production source. And more to +/// the point: nothing else exercises service → provider → core for this member, +/// which is the path a host actually calls. +/// +/// The document is written straight through the store client, deliberately. +/// Storing it with `accept_source_items` would tree it on the way in — that is +/// what #134 fixed — and then there would be nothing left for a backfill to do. +/// A document in the namespace store with no tree row *is* the state this +/// feature exists to repair. +#[tokio::test] +async fn the_backfill_door_files_a_stored_connector_document_through_the_port() { + use tinymemory_api::provider::types::BackfillTreesRequest; + + let workspace = tempfile::tempdir().expect("tempdir"); + let connection = test_connection().await; + let mut config = test_config(workspace.path()); + // The source registry is written beside the host's config file, so the + // default's absent path has to be given a real one or the walk finds no + // namespaces to sweep. + config.config_path = Some(workspace.path().join("config.toml")); + let _embedding_host = EmbeddingHostRestore::install(connection, &config); + + let client = std::sync::Arc::new( + tinymemory_core::store::MemoryClient::from_workspace_dir(workspace.path().to_path_buf()) + .expect("open the workspace store"), + ); + + // One connected account, so the namespace is derivable and the legacy + // `skill-` one is unambiguous rather than skipped. + let host = tinymemory_tinycortex::engine::EngineRuntimeConfig::from(&config); + let source: tinymemory_core::sources::MemorySourceEntry = + serde_json::from_value(serde_json::json!({ + "id": "src_gmail", + "kind": "composio", + "label": "Gmail", + "enabled": true, + "toolkit": "gmail", + "connection_id": "conn-1", + })) + .expect("a valid composio source entry"); + tinymemory_core::sources::registry::replace_sources_in(&host, &[source]) + .expect("write the source registry"); + + client + .put_doc(tinymemory_api::types::NamespaceDocumentInput { + namespace: "source:gmail:conn-1".to_string(), + key: "msg-1".to_string(), + title: "Quarterly planning".into(), + content: "Let's finalise the Q3 roadmap and align on the launch date.".into(), + source_type: "composio".into(), + priority: "medium".into(), + tags: vec!["gmail".into()], + metadata: serde_json::json!({}), + category: "core".into(), + session_id: None, + document_id: None, + taint: tinymemory_api::types::MemoryTaint::ExternalSync, + }) + .await + .expect("store a connector document the way the sync path stored one"); + + let service = super::MemoryService::new(std::sync::Arc::new(crate::provider::provider( + &config, + std::sync::Arc::clone(&client), + ))); + + let report = service + .backfill_connector_trees(BackfillTreesRequest { + limit: None, + dry_run: false, + }) + .await + .expect("the backfill door answers"); + + assert_eq!( + report.ingested, 1, + "the stored document must reach the tree through this port: {report:?}" + ); + assert_eq!( + report.already_present, 0, + "nothing was treed before this ran: {report:?}" + ); + + // Idempotence, asserted here as well as in core, because it is the property + // that makes this safe for a host to offer as a button someone can press + // twice. + let again = service + .backfill_connector_trees(BackfillTreesRequest { + limit: None, + dry_run: false, + }) + .await + .expect("a second pass answers"); + assert_eq!( + again.ingested, 0, + "a second pass must write nothing: {again:?}" + ); + assert_eq!( + again.already_present, 1, + "and must say why it wrote nothing: {again:?}" + ); +} From 5b955792d486a84cbdfa556af64c8e75e09bb74b Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 4 Sep 2026 00:14:30 +0530 Subject: [PATCH 5/5] test(module): declare the backfill member in the e2e's expected list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `the_manifest_declares_every_method_the_module_serves` compares the module's manifest against `EXPECTED_METHODS`, a fourth hand-maintained list of member names — after `tinymemory_bus::METHODS`, the `#[tinybus::interface]` impl block, and the manifest in `lib.rs`. It lives in `tests/`, so `cargo test --lib` never ran it. That is the same class of miss as the module workspace itself: a scope the obvious command silently steps over, green locally and red in CI. Compared as a `BTreeSet`, so this entry sits with its family. Only the two positional lists are append-only, and the comment says which is which so the next member does not have to rediscover the difference the hard way. Refs tinyhumansai/openhuman#6012 --- crates/tinymemory-module/tests/module_e2e.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index 55ecc85a..524c21db 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -694,6 +694,10 @@ const EXPECTED_METHODS: &[&str] = &[ "LatestQueueFailure", "BackfillInProgress", "FlushPending", + // openhuman#6012. Compared as a set (`BTreeSet`), so this sits with its + // family rather than at the tail — unlike `tinymemory_bus::METHODS` and the + // `#[tinybus::interface]` impl block, which are positional and append-only. + "BackfillConnectorTrees", "ResetDerivedIndex", "PurgeAll", "RecallNamespaceRecent",