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..7a7bb164 --- /dev/null +++ b/crates/tinymemory-core/src/backfill_tests.rs @@ -0,0 +1,240 @@ +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. + // 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, + &crate::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(), + "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/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 9ccbd0ad..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 @@ -2239,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. 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:?}" + ); +} 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", 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