From aeb2137def7f4ce5a504c6a5a2f2f6855a83efcf Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 31 Aug 2026 16:42:33 +0530 Subject: [PATCH 1/3] feat: five contract doors for the openhuman engine shed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit openhuman#5560 removes tinycortex and tinymemory-core from the host build, reaching the engine only through this contract over the TinyBus module. Five host surfaces had no door; this adds them, all as defaulted trait methods so every other driver keeps compiling and an older pinned artifact answers Unsupported instead of failing to load. MemoryTree: - summarise — the archivist recap fold. Owned wire twins of SummaryInput / SummaryContext / SummaryOutput (the usage-carrying tinymemory-core one); tree_kind travels as an open string per the retrieval-hit precedent, noting the hazard is symmetric because here it rides a request. - root_summaries_with_caps — the system-prompt tree summaries, tuple named as RootSummary. Infallible engine-side; only the default body errors. MemoryChunks: - chunk_score — the score-row read behind the dashboard's score breakdown, with ChunkScore/ChunkScoreSignals carried faithfully (including the two fields that always read back empty, documented) and DEFAULT_DROP_THRESHOLD published so the host stops hardcoding 0.3. - source_ingest_status — the per-configured-source sync counters. This is NOT source_totals: chunks_pending spans three tables no chunk member exposes, a zero-chunk source must still get a row, and the grouping key is the registry entry, not the chunk ingest key — the host supplies the prefix, the engine counts. The three-table pending predicate now lives once, in core, shared by the old path and the new door; the new door's LIKE prefix is escaped (the old path's unescaped `_` over-match is preserved there, proven by test, so nothing shipped changes). MemoryMaintenance: - degraded_state — the cheap atomics read behind pipeline_status, separate from diagnose because the hot path must not run the doctor. diagnose itself needed nothing: the provider already implements it; its degraded mapping is now shared with the new member so the two cannot disagree. METHODS grows 126 -> 131, appended in wire order; the module serves all five and its sequence/manifest assertions pin them. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-api/src/provider/chunks.rs | 107 +++++- crates/tinymemory-api/src/provider/content.rs | 127 +++++++ crates/tinymemory-api/src/provider/mod.rs | 8 +- crates/tinymemory-api/src/provider/records.rs | 52 ++- crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 17 +- crates/tinymemory-bus/src/names_tests.rs | 51 +++ crates/tinymemory-bus/src/provider/chunks.rs | 244 ++++++++++++++ .../src/provider/chunks_tests.rs | 108 ++++++ crates/tinymemory-bus/src/tree.rs | 226 +++++++++++++ crates/tinymemory-bus/src/tree_tests.rs | 132 +++++++- crates/tinymemory-core/src/sources/status.rs | 164 ++++++--- .../src/sources/status_tests.rs | 108 ++++++ crates/tinymemory-module/src/lib.rs | 10 + crates/tinymemory-module/src/service/mod.rs | 139 +++++++- crates/tinymemory-module/tests/module_e2e.rs | 3 + .../tinymemory-tinycortex/src/engine/mod.rs | 313 +++++++++++++++++- .../tinymemory-tinycortex/src/engine/test.rs | 70 +++- .../tests/full_provider_conformance.rs | 168 ++++++++++ 19 files changed, 1980 insertions(+), 69 deletions(-) create mode 100644 crates/tinymemory-bus/src/provider/chunks_tests.rs diff --git a/crates/tinymemory-api/src/provider/chunks.rs b/crates/tinymemory-api/src/provider/chunks.rs index feff6fdf..0284ccec 100644 --- a/crates/tinymemory-api/src/provider/chunks.rs +++ b/crates/tinymemory-api/src/provider/chunks.rs @@ -42,7 +42,8 @@ use crate::provider::types::SourceScope; // able to name them without compiling this trait — and re-exported here so // every historical path keeps resolving and the types stay the same types. pub use tinymemory_bus::provider::chunks::{ - ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, SourceTotal, + ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, ChunkScore, ChunkScoreSignals, + SourceIngestQuery, SourceIngestStatus, SourceTotal, DEFAULT_DROP_THRESHOLD, }; /// Direct read access to the chunk tier. @@ -238,4 +239,108 @@ pub trait MemoryChunks: Send + Sync { chunk_ids: &[String], model_signature: &str, ) -> Result, MemoryError>; + + /// One chunk's admission decision, and the signals it was reached from. + /// + /// The scorer's own row: what each signal measured, what they summed to, + /// whether the chunk was kept, and why. A diagnostic read for "this + /// document is in memory and that one is not" — not an input to ranking, + /// which [`MemoryRetrieval`](super::MemoryRetrieval) owns and which happens + /// per query rather than once at ingest. + /// + /// # Why the driver has to answer this + /// + /// The decision is a row in the driver's own score table, written at + /// admission time under the policy in force then. Nothing in the chunk tier + /// records it: [`Self::chunk_detail`] can say a chunk exists and is marked + /// dropped, and cannot say what it scored or which signal it failed on. A + /// caller cannot re-derive it either — re-running the scorer today would + /// answer under today's policy, and produce a number the store never used. + /// + /// # `None` is "never scored", not "scored zero" + /// + /// A chunk with no score row was not judged, which is a different fact from + /// a chunk judged uninteresting and kept anyway. Collapsing the two — by + /// defaulting to a zeroed [`ChunkScore`] — reports a verdict that was never + /// reached, on a screen whose entire purpose is to explain verdicts. + /// + /// # Errors + /// + /// [`MemoryError::Unsupported`] from a driver that implements this family + /// but keeps no admission record. That is the honest answer for a driver + /// that admits everything, and it is deliberately not defaulted to + /// `Ok(None)`: "this driver does not score" and "this chunk was not scored" + /// are different answers, and only the first is true of every chunk. + /// + /// Otherwise backend failures; an unknown chunk id yields `Ok(None)`, the + /// same as a known one with no score row — a caller inspecting a chunk it + /// just listed cannot tell those apart and does not need to. + async fn chunk_score(&self, chunk_id: &str) -> Result, MemoryError> { + let _ = chunk_id; + Err(MemoryError::unsupported(Capability::Chunks)) + } + + /// How far ingest has got for each of the sources named in + /// `source_prefixes`. + /// + /// One row per query, in the order asked, echoing + /// [`SourceIngestQuery::source_id`] so a caller can pair them by value + /// rather than by position. **A query whose prefix matches nothing still + /// gets a row**, zero-filled — see below. + /// + /// # Why the caller supplies the prefix + /// + /// Because the caller is the only party that can. The prefix is derived + /// from a configured source's kind, toolkit and connection id, which live + /// in the host's source registry; a driver asked to derive it would need + /// that registry, which is precisely the coupling this contract exists to + /// remove. So the host states the key and the driver counts the rows under + /// it — each side answering from what it actually holds. + /// + /// # Why this is not [`Self::source_totals`] + /// + /// Three differences, and a caller that substituted one for the other would + /// get a result that renders as a healthy store. + /// + /// 1. [`SourceTotal`] has no pending count and none can be derived from it. + /// The predicate spans the embedding sidecar and the re-embed skip + /// ledger as well as the chunk's own lifecycle column, so a caller + /// reading only the chunk tier reports nothing in flight — which is what + /// a finished sync looks like. + /// 2. `source_totals` returns the groups that *exist*. A configured source + /// that has never synced forms no group, so it vanishes from the answer + /// rather than appearing idle — and a source missing from a dashboard + /// reads as one that was never set up. + /// 3. [`SourceTotal::source_id`] is the ingest key the chunk rows carry; + /// [`SourceIngestQuery::source_id`] is the registry entry a user + /// configured. For a connector source the two share no substring, so + /// matching them up is not a formatting difference a caller can paper + /// over. + /// + /// # Freshness is deliberately absent + /// + /// An `Active`/`Recent`/`Idle` label is arithmetic over + /// [`SourceIngestStatus::last_chunk_at_ms`] and the current time. Answering + /// it here would freeze the driver's clock into the reply, so a panel + /// rendering the label a minute later would show how fresh the source was + /// when the driver looked. The caller has the timestamp and its own clock. + /// + /// # Errors + /// + /// [`MemoryError::Unsupported`] from a driver that implements this family + /// but tracks no per-source ingest state — not defaulted to zero-filled + /// rows, which would report every configured source as never synced. + /// + /// Otherwise backend failures, for the whole batch rather than per row: the + /// counts come from one store, so a read that fails fails for all of them, + /// and a partial answer would be indistinguishable from a set of genuinely + /// empty sources. An empty `source_prefixes` yields an empty vector without + /// touching the store. + async fn source_ingest_status( + &self, + source_prefixes: &[SourceIngestQuery], + ) -> Result, MemoryError> { + let _ = source_prefixes; + Err(MemoryError::unsupported(Capability::Chunks)) + } } diff --git a/crates/tinymemory-api/src/provider/content.rs b/crates/tinymemory-api/src/provider/content.rs index 83baf139..4a76f275 100644 --- a/crates/tinymemory-api/src/provider/content.rs +++ b/crates/tinymemory-api/src/provider/content.rs @@ -25,6 +25,15 @@ use crate::provider::types::{IngestItem, IngestOutcome, SourceScope}; use crate::tree::{IngestRequest, QueryResult, SummaryForest, TreeLeaf, TreeStatus}; use crate::types::{NamespaceDocumentInput, NamespaceRetrievalContext, StoredMemoryDocument}; +// The value types the summariser door exchanges. They are defined in +// `tinymemory-bus` — they cross the module boundary, and a host that only makes +// calls must be able to name them without compiling this trait — and +// re-exported here so the family's vocabulary is reachable from the family, the +// same arrangement `provider::chunks` uses. They are re-exported at +// `crate::tree` too, alongside the rest of the tree vocabulary; both paths name +// the same items, not twins of them. +pub use tinymemory_bus::tree::{RootSummary, SummaryContext, SummaryInput, SummaryOutput}; + /// Bulk content ingestion — the driver owns chunking and embedding. /// /// The distinction from [`crate::provider::MemoryCore::store`] is ownership of @@ -410,4 +419,122 @@ pub trait MemoryTree: Send + Sync { async fn flush_source_tree(&self, _source_scope: &str) -> Result { Err(MemoryError::unsupported(Capability::Tree)) } + + /// Fold `inputs` into one parent summary, using the driver's own chat + /// provider, and report what that call cost. + /// + /// This is the LLM step of a seal, exposed on its own. Everything else in + /// this family either writes content ([`Self::append`]), navigates what is + /// already sealed, or asks the driver to run a whole seal/cascade pass + /// ([`Self::seal`], [`Self::cascade`], [`Self::flush_source_tree`]). This + /// one does a single fold and hands the text back, which is what a caller + /// driving its own cascade needs and what none of the others can be made to + /// answer: they return tree *state*, and the summary they produced is never + /// in it. + /// + /// # Why the provider is the driver's and not the caller's + /// + /// The summariser is configured where the engine is — model, temperature, + /// output language, rate card. A caller reaching memory over a module has + /// none of those, so a fold it performed itself would use a different model + /// than every fold the scheduler performs, and the two would disagree about + /// the shape of a summary in the same tree. Passing the configuration + /// across instead is not an option: no signature in this contract names a + /// config type, for the reasons in [`crate::provider`]. + /// + /// The consequence is that the usage numbers on [`SummaryOutput`] are the + /// only record of the spend. Nothing on the caller's side saw the request. + /// + /// # The budgets and the ask are inputs, not hints + /// + /// A driver applies [`SummaryContext`]'s three token budgets exactly as + /// given and selects its prompt from [`SummaryContext::ask`]. It does not + /// substitute its own defaults for a budget it finds implausible: the + /// caller owns the level it is sealing and therefore owns the budget for + /// it, and a driver that quietly widened one would produce a node that + /// overruns the level above. See that type for what each budget bounds and + /// what a zero does. + /// + /// # Errors + /// + /// [`MemoryError::Unsupported`] from a driver with a tree family but no + /// provider-backed summariser — deliberately not an empty summary, which a + /// caller would seal as a real, blank node. + /// + /// [`MemoryError::Invalid`] for a [`SummaryContext::tree_kind`] the driver + /// does not recognise. Refusing beats folding under a guessed kind: the + /// summary is written either way and nothing afterwards records which + /// prompt produced it. + /// + /// Otherwise a backend failure, which here includes the provider call — a + /// model that errors, times out, or refuses. That is a real and recurring + /// outcome rather than an exceptional one, and the caller is expected to + /// have a deterministic fallback for it; the driver does not silently + /// substitute one, because a caller cannot tell a fallback summary from a + /// model's own work once it is in the tree. + /// + /// Nothing to fold is **not** an error: an empty slice, or one whose inputs + /// are all blank, returns a default [`SummaryOutput`] with empty content and + /// no usage. That is the same idempotence [`Self::seal`] has, and it is what + /// lets a cascade call this unconditionally at every level. + async fn summarise( + &self, + _inputs: &[SummaryInput], + _context: &SummaryContext, + ) -> Result { + Err(MemoryError::unsupported(Capability::Tree)) + } + + /// Every namespace's root summary, truncated to a per-namespace cap and a + /// total cap. + /// + /// The top of the markdown time tree, read across all namespaces at once. + /// Its caller is a prompt builder: this is the block of standing context a + /// host puts in front of a model, which is why the bounds are in + /// **characters** rather than in rows, and why the truncation happens + /// driver-side rather than after the read. A caller that fetched whole + /// roots and clipped them itself would pay for text it then threw away, and + /// would clip at a boundary the driver did not choose. + /// + /// # Why not [`Self::summary_forest`] + /// + /// Different tier and different shape. The forest walks the *sealed summary + /// forest* — one tree per ingest source, levelled by seal generation — and + /// returns structure: ids, parents, children, no bodies. This returns the + /// **markdown time tree**'s root body, one per namespace, and bodies are + /// the entire point. [`crate::tree`] describes why the two live side by + /// side. + /// + /// # Bounds + /// + /// `per_namespace_cap` clips each namespace's body; `total_cap` stops the + /// walk once the accumulated bodies reach it, so the last body included may + /// itself be clipped short of its own cap. Both are applied in namespace + /// order, which is stable and alphabetical — so a total cap that binds + /// drops the *tail* of the namespace list rather than sampling across it, + /// exactly the reading [`SummaryForest::truncated`] warns about. A caller + /// that needs a particular namespace represented cannot rely on a small + /// total cap to include it. + /// + /// A clipped body ends in a `[... truncated]` marker; see + /// [`RootSummary::body`]. + /// + /// # Errors + /// + /// [`MemoryError::Unsupported`] from a driver with a tree family but no + /// markdown time tree. + /// + /// Otherwise this is deliberately hard to fail. The read is a best-effort + /// filesystem scan: a namespace whose root cannot be read is skipped and + /// the rest are returned, and a workspace with no tree at all is an empty + /// vector. That is the engine's own behaviour and the door does not + /// manufacture an error it never produced — the result is a prompt block, + /// and one unreadable namespace is worth less than failing the turn. + async fn root_summaries_with_caps( + &self, + _per_namespace_cap: usize, + _total_cap: usize, + ) -> Result, MemoryError> { + Err(MemoryError::unsupported(Capability::Tree)) + } } diff --git a/crates/tinymemory-api/src/provider/mod.rs b/crates/tinymemory-api/src/provider/mod.rs index 7b74eb7b..2d7a658d 100644 --- a/crates/tinymemory-api/src/provider/mod.rs +++ b/crates/tinymemory-api/src/provider/mod.rs @@ -90,9 +90,13 @@ pub use tinymemory_bus::provider::{diagnosis, types}; pub use audit::{audit_provider, CapabilityAudit}; pub use chunks::{ - ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, MemoryChunks, SourceTotal, + ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, ChunkScore, ChunkScoreSignals, + MemoryChunks, SourceIngestQuery, SourceIngestStatus, SourceTotal, DEFAULT_DROP_THRESHOLD, +}; +pub use content::{ + MemoryDocuments, MemoryIngest, MemoryTree, RootSummary, SummaryContext, SummaryInput, + SummaryOutput, }; -pub use content::{MemoryDocuments, MemoryIngest, MemoryTree}; pub use diagnosis::{ DegradedCapabilities, Diagnosis, DiagnosisCounters, DiagnosisFailure, DiagnosisStage, }; diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs index fec952f3..1eca486a 100644 --- a/crates/tinymemory-api/src/provider/records.rs +++ b/crates/tinymemory-api/src/provider/records.rs @@ -19,7 +19,7 @@ use async_trait::async_trait; use crate::capabilities::Capability; use crate::error::MemoryError; use crate::goals::GoalsDoc; -use crate::provider::diagnosis::Diagnosis; +use crate::provider::diagnosis::{DegradedCapabilities, Diagnosis}; use crate::provider::types::{ FlushOutcome, ForgetOutcome, ForgetSelector, IngestOutcome, MaintenanceReport, PurgeOutcome, QueueFailure, QueueStats, ResetOutcome, SourceItem, StoreStats, @@ -470,4 +470,54 @@ pub trait MemoryMaintenance: Send + Sync { async fn diagnose(&self) -> Result { Err(MemoryError::unsupported(Capability::Maintenance)) } + + /// Which capabilities are currently running in a reduced mode. + /// + /// The degradation flags on their own: semantic recall fallen back to + /// recency, extraction producing no structure, the storage path unusable — + /// and the cause of the most severe of those, when the driver knows it. + /// + /// # Why this is not [`Self::diagnose`] with the rest thrown away + /// + /// Cost, and the difference is not marginal. A [`Diagnosis`] is a full + /// pass: it counts chunks, counts jobs in three states, measures extraction + /// coverage over the whole store, and inspects the configuration of every + /// pipeline stage. This is a read of flags the pipeline sets as it runs — + /// no query, no configuration walk, nothing that touches storage. + /// + /// That matters because of who calls it. A diagnosis is asked for once, + /// deliberately, by someone looking at a problem. Degradation is polled: it + /// is what a status indicator shows continuously, and driving that from a + /// full pass would put an aggregate query over the chunk table on a + /// repeating timer. The two members exist so a caller can ask the cheap + /// question without paying for the expensive one — and, just as important, + /// so it is not tempted to poll the expensive one and cache the answer, + /// which is how a status light ends up reporting a degradation that cleared + /// minutes ago. + /// + /// [`Diagnosis::degraded`] carries the same shape, from the same source, so + /// a caller that has just run a diagnosis has no reason to call this too. + /// + /// # Why a caller cannot compute it + /// + /// The flags are set inside the driver, by the embed and extract stages, as + /// they fail. Nothing observable from outside distinguishes a recall that + /// ranked semantically from one that fell back to recency — both return + /// rows, in an order, with no marker on them. A caller with no engine would + /// report an all-clear, which is not a stale answer but a confidently wrong + /// one. + /// + /// # Errors + /// + /// [`MemoryError::Unsupported`] from a driver that tracks no degradation + /// state. Deliberately not defaulted to + /// [`DegradedCapabilities::default()`], which is all-clear: a driver that + /// has never looked would report that everything is fine, and the whole + /// purpose of this member is to be believed when it says that. + /// + /// Otherwise backend failures — though a driver reading in-process flags + /// has no failure path and should not invent one. + async fn degraded_state(&self) -> Result { + Err(MemoryError::unsupported(Capability::Maintenance)) + } } diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index f644259c..4e0c2e4a 100644 --- a/crates/tinymemory-bus/src/lib.rs +++ b/crates/tinymemory-bus/src/lib.rs @@ -2,7 +2,7 @@ //! the members that carry them. //! //! TinyMemory ships as a loadable `TinyBus` module: `crates/tinymemory-module` -//! exports one object with 123 members on it, built as a `cdylib`. A host that +//! exports one object (`tinymemory_bus::METHODS.len()` members) built as a `cdylib`. A host that //! loads it — OpenHuman — can call into it but cannot `use` anything out of it, //! so the payload vocabulary has to be published as an ordinary library. This //! is that library. diff --git a/crates/tinymemory-bus/src/names.rs b/crates/tinymemory-bus/src/names.rs index 065f588a..e5b4f7c5 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -98,6 +98,10 @@ pub mod methods { pub const SUMMARY_FOREST: &str = "SummaryForest"; /// `RecentLeaves` — the newest leaves and the summaries that sealed them. pub const RECENT_LEAVES: &str = "RecentLeaves"; + /// `Summarise` — fold summary inputs into one parent summary. + pub const SUMMARISE: &str = "Summarise"; + /// `RootSummaries` — every namespace's root summary, capped. + pub const ROOT_SUMMARIES: &str = "RootSummaries"; // Entities, relations and the namespaced key/value store. /// `Entities` — entities. @@ -219,6 +223,10 @@ pub mod methods { pub const LIST_CHUNK_DETAILS: &str = "ListChunkDetails"; /// `SourceTotals` — one row per source, with what it contributed. pub const SOURCE_TOTALS: &str = "SourceTotals"; + /// `ChunkScore` — one chunk's admission decision and the signals behind it. + pub const CHUNK_SCORE: &str = "ChunkScore"; + /// `SourceIngestStatus` — per configured source, how far ingest has got. + pub const SOURCE_INGEST_STATUS: &str = "SourceIngestStatus"; // The scored retrieval surface. /// `FastRetrieve` — fast retrieve. @@ -285,6 +293,8 @@ pub mod methods { // report. /// `Diagnose` — the typed, per-stage pipeline diagnosis. pub const DIAGNOSE: &str = "Diagnose"; + /// `DegradedState` — the degradation flags alone, without a diagnosis. + pub const DEGRADED_STATE: &str = "DegradedState"; // Syncs the driver runs itself: the manual trigger, the persisted state, // and what past runs cost. @@ -331,7 +341,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; 126] = [ +pub const METHODS: [&str; 131] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -458,6 +468,11 @@ pub const METHODS: [&str; 126] = [ methods::EXTRACT_ENTITIES, methods::EMBED_TEXT, methods::EMBEDDER_SLUG, + methods::SUMMARISE, + methods::ROOT_SUMMARIES, + methods::DEGRADED_STATE, + methods::CHUNK_SCORE, + methods::SOURCE_INGEST_STATUS, ]; #[cfg(test)] diff --git a/crates/tinymemory-bus/src/names_tests.rs b/crates/tinymemory-bus/src/names_tests.rs index 14656e67..b0ebb212 100644 --- a/crates/tinymemory-bus/src/names_tests.rs +++ b/crates/tinymemory-bus/src/names_tests.rs @@ -65,3 +65,54 @@ fn the_constants_and_the_table_are_the_same_set() { "WorkflowIdentityMatches" ); } + +#[test] +fn the_host_shed_members_are_spelled_as_the_module_derives_them() { + // These three exist because a host is removing its direct engine link and + // has nowhere else to ask. A typo in one of them is not a compile error on + // either side — it is an `UnknownMethod` the first time a status panel or a + // chunk inspector is opened against a released module — so the spellings + // are pinned here rather than only read off the table. + assert_eq!(methods::DEGRADED_STATE, "DegradedState"); + assert_eq!(methods::CHUNK_SCORE, "ChunkScore"); + assert_eq!(methods::SOURCE_INGEST_STATUS, "SourceIngestStatus"); + assert!(METHODS.contains(&methods::DEGRADED_STATE)); + assert!(METHODS.contains(&methods::CHUNK_SCORE)); + assert!(METHODS.contains(&methods::SOURCE_INGEST_STATUS)); +} + +#[test] +fn the_newest_members_are_appended_rather_than_filed_with_their_family() { + // Member order is wire order: the module compares its served members + // against this table as a *sequence*, so a new member filed beside its + // family renumbers every member after it. That is invisible here and shows + // up as the wrong method being invoked on a host built against a different + // release, which is why the tail order is asserted and not just membership. + let tail = &METHODS[METHODS.len() - 3..]; + assert_eq!( + tail, + [ + methods::DEGRADED_STATE, + methods::CHUNK_SCORE, + methods::SOURCE_INGEST_STATUS, + ] + ); +} + +#[test] +fn the_summariser_door_holds_the_wire_slots_it_was_released_in() { + // `Summarise` and `RootSummaries` are the two members a host reaches for + // once it stops linking the engine, so their spellings are pinned here as + // well as read off the table — a typo in either is an `UnknownMethod` the + // first time a seal runs against a released module, not a compile error. + assert_eq!(methods::SUMMARISE, "Summarise"); + assert_eq!(methods::ROOT_SUMMARIES, "RootSummaries"); + + // Their *positions* are pinned too, and by absolute index rather than from + // the end. Member order is wire order, so a member inserted ahead of these + // renumbers both and every member after them; asserting from the tail would + // move silently under the next append, which is exactly the edit this is + // here to catch. + assert_eq!(METHODS[126], methods::SUMMARISE); + assert_eq!(METHODS[127], methods::ROOT_SUMMARIES); +} diff --git a/crates/tinymemory-bus/src/provider/chunks.rs b/crates/tinymemory-bus/src/provider/chunks.rs index 55195a2e..bf9c7b29 100644 --- a/crates/tinymemory-bus/src/provider/chunks.rs +++ b/crates/tinymemory-bus/src/provider/chunks.rs @@ -29,6 +29,19 @@ //! there, just filed under a name the caller did not ask for. That is a real //! failure mode with a real precedent, and it is silent; see //! `docs/specs/2026-08-13-memory-module-port.md` §3. +//! +//! # Two of these reads are diagnostic rather than retrieval +//! +//! [`ChunkScore`] and [`SourceIngestStatus`] answer "why is this here" and "how +//! far has this got", not "what is relevant". They are in this family because +//! both are keyed by the chunk tier — one by a chunk id, the other by the +//! prefix its ingest key carries — and because a caller that has the chunks +//! surface is exactly the caller with a browser to render them in. +//! +//! Neither is derivable from the reads above it. A score row lives in a table +//! of its own; a pending count spans three. That is the whole reason they are +//! members rather than arithmetic a host does over a chunk page, and each +//! type's own docs give the specific version of the argument. use serde::{Deserialize, Serialize}; @@ -279,3 +292,234 @@ pub struct SourceTotal { /// [`StoreStats`]: crate::provider::types::StoreStats pub most_recent_ms: i64, } + +/// The admission score below which the engine's default policy tombstones a +/// chunk, so a caller can draw the same line the scorer drew. +/// +/// # Why the number crosses at all +/// +/// A browser rendering [`ChunkScore::total`] as a bar wants to show where the +/// keep/drop boundary sits, and a host-side copy of `0.3` is the kind of copy +/// that goes wrong silently: the driver retunes, every row on the screen keeps +/// its label, and only the line drawn under them is stale. There is nothing to +/// notice, because the rows still render. +/// +/// # It is the default, not the effective threshold +/// +/// A driver whose scoring policy was tuned admits at its own number, and this +/// constant does not follow it. [`ChunkScore::dropped`] is the verdict the +/// engine actually reached for that chunk; this is the reference line a gauge +/// is drawn against. A caller that recomputes `dropped` from `total` against +/// this constant will disagree with the driver on a tuned store — and the +/// driver is the one that was there. +pub const DEFAULT_DROP_THRESHOLD: f32 = 0.3; + +/// The per-signal breakdown behind a chunk's admission score. +/// +/// Every field is one term of the weighted sum that produced +/// [`ChunkScore::total`], carried unweighted: the caller sees what each signal +/// measured, not what the policy paid for it. That is the useful half for +/// "why was this dropped" — a chunk that failed on length and a chunk that +/// failed on a source it does not trust have the same total and different +/// stories. +/// +/// # Two fields read back empty, by construction +/// +/// [`Self::llm_importance`] is an admission-time input that the engine's score +/// table has no column for, so a stored row answers `0.0` for it however +/// strongly the extractor rated the chunk. It is carried anyway rather than +/// dropped from the shape, because a driver that *does* persist it should have +/// somewhere to put it, and because a field that is absent from the type reads +/// to the next caller as a signal that does not exist rather than as one this +/// store does not keep. The same applies to +/// [`ChunkScore::llm_importance_reason`]. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct ChunkScoreSignals { + /// Length signal derived from the chunk's token count. + #[serde(default)] + pub token_count: f32, + /// Lexical-diversity signal derived from the count of distinct words. + #[serde(default)] + pub unique_words: f32, + /// Contribution from structural or front-matter metadata on the source. + #[serde(default)] + pub metadata_weight: f32, + /// Contribution from the source's provenance or authority. + #[serde(default)] + pub source_weight: f32, + /// Direct-engagement signal from user interaction with the chunk. + #[serde(default)] + pub interaction: f32, + /// Signal proportional to the density of extracted entities in the chunk. + #[serde(default)] + pub entity_density: f32, + /// LLM-derived importance rating in `[0.0, 1.0]`. + /// + /// `0.0` both when no LLM signal was available and when the driver does not + /// persist one — see the type's own note. A caller must not read a zero + /// here as "the model found this unimportant". + #[serde(default)] + pub llm_importance: f32, +} + +/// One chunk's admission decision and the signals it was reached from. +/// +/// The row the scorer wrote when it decided whether to keep the chunk. It is a +/// **diagnostic** read — "why is this here", or "why is this not" — and not +/// part of any ranking: [`crate::provider::retrieval`] owns relevance, and this +/// owns admission, which happened once, at ingest, against the whole store's +/// policy rather than against a query. +/// +/// # Why the whole row crosses when a caller reads three fields +/// +/// The first caller wants [`Self::total`], [`Self::dropped`] and the signals. +/// The second wants [`Self::reason`] beside them, because "dropped" with no +/// rationale is a verdict without an argument; the third wants +/// [`Self::computed_at_ms`], because a score from before the policy changed +/// explains a row the current policy would have kept. Each of those is a +/// widening of a shipped wire type, and the alternative — carrying the row the +/// engine already assembled — costs nothing: the columns are read in one +/// `query_row` whether or not they are returned. +/// +/// # A missing row is not a zero score +/// +/// `MemoryChunks::chunk_score` answers `None` for a chunk the scorer never +/// wrote a row for, which is a different fact from a chunk that scored `0.0` +/// and was kept anyway. A caller rendering "score: 0" for both reports a +/// judgement that was never made. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ChunkScore { + /// The chunk this rationale belongs to. + pub chunk_id: String, + /// The aggregate admission score — the single scalar the keep/drop + /// decision was taken on, and the one [`DEFAULT_DROP_THRESHOLD`] is the + /// reference line for. + pub total: f32, + /// The per-signal breakdown that produced [`Self::total`]. + #[serde(default)] + pub signals: ChunkScoreSignals, + /// Whether the chunk failed admission and was tombstoned rather than kept. + /// + /// The driver's own verdict, not a re-derivation. See + /// [`DEFAULT_DROP_THRESHOLD`] for why the two can disagree. + pub dropped: bool, + /// The recorded rationale for the keep or drop, when one was written. + /// + /// Operator-facing prose in the driver's words, never localised and never + /// parsed. `None` means no rationale was recorded, not that there was none. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reason: Option, + /// When the score was computed, epoch milliseconds. + /// + /// Carried because admission is a decision taken *at a time*, under the + /// policy in force then. A row scored before a retune is the explanation + /// for a chunk today's policy would have judged differently, and without + /// this field that explanation is unavailable. + pub computed_at_ms: i64, + /// The LLM's one-line explanation for its importance rating. + /// + /// `None` from a driver that does not persist it — which is every driver + /// backed by the engine's current score table, for the reason + /// [`ChunkScoreSignals::llm_importance`] gives. Carried so a driver that + /// keeps it has somewhere to put it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub llm_importance_reason: Option, +} + +/// One configured source to ask ingest progress about. +/// +/// Two identifiers, because they are two different things and conflating them +/// is the bug this type exists to prevent. [`Self::source_id`] is the id in the +/// *host's* source registry — the thing a user configured, named and can +/// disable. [`Self::chunk_id_prefix`] is the key the ingest path stamps on the +/// chunk rows it writes, which is derived from the first but is not equal to +/// it: the engine's readers key chunks `mem_src:{source id}:{item}`, and its +/// connector sync keys them `{toolkit}:{connection id}:{document id}`, which +/// does not contain the registry id at all. +/// +/// # Why the host supplies the prefix rather than the driver deriving it +/// +/// The derivation is host policy over host state. It reads the source's kind, +/// its toolkit and its connection id out of the registry the host owns, and a +/// driver asked to redo it would need that registry — which is the coupling +/// this contract exists to remove. So the host, which already holds the entry, +/// states the prefix, and the driver answers only the question it can answer +/// from its own tables: how many rows are under this key. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceIngestQuery { + /// The configured source's id, echoed back on the matching + /// [`SourceIngestStatus`] so a caller can pair the rows without relying on + /// order. + pub source_id: String, + /// The literal prefix of the chunk rows' ingest key for this source. + /// + /// A **literal** prefix, matched as text: a driver must treat any pattern + /// metacharacter in it as itself. The convention is the one the engine's + /// own readers write, `mem_src:{source id}:` and `{toolkit}:{connection + /// id}:`, including the trailing separator — without it a source keyed + /// `mem_src:src_a:` also counts the chunks of `mem_src:src_ab:`. + pub chunk_id_prefix: String, +} + +/// How far one configured source's ingest has got. +/// +/// # Why this is not [`SourceTotal`] +/// +/// [`SourceTotal`] is a `GROUP BY` over the chunk rows: it describes the groups +/// that *exist*. This is a per-question answer about the sources a host has +/// *configured*, and the two differ in exactly the places a status panel +/// depends on. +/// +/// A source that has never synced has no chunk rows, so it forms no group and +/// is simply absent from `source_totals` — a dashboard built on that loses the +/// row rather than showing it idle, which reads as "this source is not +/// configured" instead of "this source has done nothing yet". So a row comes +/// back for **every** query, zero-filled when the prefix matches nothing. +/// +/// It also carries [`Self::chunks_pending`], which no grouping of the chunk +/// table can produce: the predicate spans the embedding sidecar and the +/// re-embed skip ledger as well as the chunk row's own lifecycle column. A +/// caller that substituted a chunk count for it would report a store with +/// nothing in flight, which is the same answer a healthy store gives. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SourceIngestStatus { + /// The configured source id from the query this row answers. + pub source_id: String, + /// Chunk rows the driver holds under the source's prefix. + /// + /// Every row, whatever its state — this is "how much has landed", and + /// [`Self::chunks_pending`] is the part of it that is not finished yet, not + /// a disjoint bucket. A caller showing progress renders + /// `synced - pending` of `synced`, never `synced + pending`. + pub chunks_synced: u64, + /// Chunk rows still in flight: no embedding, not dropped by the lifecycle, + /// and not recorded as deliberately skipped for re-embedding. + /// + /// All three exits are terminal and all three count as resolved. The + /// negative form matters: "pending" is *not resolved*, so a driver that + /// checks only for a missing embedding reports every dropped and every + /// skipped chunk as eternally in flight, and a healthy source then shows + /// work that never completes. + pub chunks_pending: u64, + /// Source time of the newest chunk under the prefix, epoch milliseconds. + /// + /// `None` when the source has no chunks at all — unlike + /// [`SourceTotal::most_recent_ms`], which is not optional precisely because + /// a group cannot exist without one. Here the row exists because the source + /// was *asked about*, so "never" is a real answer and has to be + /// representable. + /// + /// # Freshness is deliberately not on the wire + /// + /// An `Active` / `Recent` / `Idle` label is arithmetic over this field and + /// the current time — under 30 seconds, under 5 minutes, anything else — + /// and putting it here would freeze the driver's clock into the answer. A + /// caller rendering a label a minute after the call would show one derived + /// from when the driver looked, not from now. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_chunk_at_ms: Option, +} + +#[cfg(test)] +#[path = "chunks_tests.rs"] +mod tests; diff --git a/crates/tinymemory-bus/src/provider/chunks_tests.rs b/crates/tinymemory-bus/src/provider/chunks_tests.rs new file mode 100644 index 00000000..0ac962d6 --- /dev/null +++ b/crates/tinymemory-bus/src/provider/chunks_tests.rs @@ -0,0 +1,108 @@ +//! Tests for the chunk-family value types. +//! +//! What is worth pinning here is not the shape of a struct — the compiler has +//! that — but the two decisions a later slice could silently reverse: that a +//! score row's diagnostic fields survive a round trip through a peer that does +//! not know them, and that the ingest-status row can represent a source with +//! nothing in it. Both failures render as a plausible screen rather than as an +//! error. + +// A failed assertion in a test is a panic either way; `unwrap`/`expect` here say +// what the invariant was. Same allowance the crate's other test modules take. +#![allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] + +use super::*; + +#[test] +fn the_drop_threshold_is_the_engines_own_number() { + // Pinned as a literal rather than derived, because the point of carrying it + // is that a caller and the scorer draw the same line. A change here is a + // change to what every rendered gauge means, and it should have to be + // typed. + assert!((DEFAULT_DROP_THRESHOLD - 0.3).abs() < f32::EPSILON); +} + +#[test] +fn a_score_row_round_trips_every_field() { + // The row is carried whole rather than narrowed to what today's caller + // reads, so the test is the whole row: a field quietly dropped from the + // wire would still decode, as its default, and read as a store that + // recorded nothing rather than as a type that forgot to ask. + let score = ChunkScore { + chunk_id: "chunk-1".to_string(), + total: 0.42, + signals: ChunkScoreSignals { + token_count: 0.1, + unique_words: 0.2, + metadata_weight: 0.3, + source_weight: 0.4, + interaction: 0.5, + entity_density: 0.6, + llm_importance: 0.7, + }, + dropped: true, + reason: Some("below the admission threshold".to_string()), + computed_at_ms: 1_700_000_000_000, + llm_importance_reason: Some("boilerplate footer".to_string()), + }; + let round_tripped: ChunkScore = + serde_json::from_value(serde_json::to_value(&score).expect("serialize score")) + .expect("decode score"); + assert_eq!(round_tripped, score); +} + +#[test] +fn a_score_row_from_a_store_that_keeps_no_llm_signal_still_decodes() { + // The engine's score table has no column for either LLM field, so a real + // row omits both. They are on the type for a driver that does keep them, + // which is only safe if their absence is not a decode failure. + let score: ChunkScore = serde_json::from_value(serde_json::json!({ + "chunk_id": "chunk-1", + "total": 0.9, + "signals": { "token_count": 0.5 }, + "dropped": false, + "computed_at_ms": 1_700_000_000_000_i64, + })) + .expect("decode a row with no LLM signal"); + assert_eq!(score.llm_importance_reason, None); + assert!(score.signals.llm_importance.abs() < f32::EPSILON); + assert!((score.signals.token_count - 0.5).abs() < f32::EPSILON); + assert_eq!(score.reason, None); +} + +#[test] +fn an_ingest_status_can_say_a_source_has_never_synced() { + // The gap this type exists for. `SourceTotal` cannot express it — a group + // with no rows is not a zero row, it is an absent one — and a dashboard + // built on that loses the source instead of showing it idle. + let never_synced = SourceIngestStatus { + source_id: "src_new".to_string(), + chunks_synced: 0, + chunks_pending: 0, + last_chunk_at_ms: None, + }; + let encoded = serde_json::to_value(&never_synced).expect("serialize status"); + assert!( + encoded.get("last_chunk_at_ms").is_none(), + "a never-synced source omits the timestamp rather than sending a zero one" + ); + let round_tripped: SourceIngestStatus = serde_json::from_value(encoded).expect("decode status"); + assert_eq!(round_tripped, never_synced); +} + +#[test] +fn the_two_source_identifiers_are_kept_apart() { + // The registry id and the chunk key are different strings for a connector + // source — the chunk key does not contain the registry id at all — so a + // type that carried one field would force the caller to send whichever the + // other end did not want. + let query = SourceIngestQuery { + source_id: "src_gmail_work".to_string(), + chunk_id_prefix: "gmail:conn-1:".to_string(), + }; + let round_tripped: SourceIngestQuery = + serde_json::from_value(serde_json::to_value(&query).expect("serialize query")) + .expect("decode query"); + assert_eq!(round_tripped, query); + assert_ne!(round_tripped.source_id, round_tripped.chunk_id_prefix); +} diff --git a/crates/tinymemory-bus/src/tree.rs b/crates/tinymemory-bus/src/tree.rs index cd3bc93f..860c9f8d 100644 --- a/crates/tinymemory-bus/src/tree.rs +++ b/crates/tinymemory-bus/src/tree.rs @@ -372,6 +372,232 @@ pub fn leaf_preview(content: &str) -> String { .collect() } +// ───────────────────────────────────────────────────────────────────────────── +// The summariser door, and the roots it eventually produces. +// ───────────────────────────────────────────────────────────────────────────── +// +// Everything above describes a tree that has already been sealed. The three +// shapes below describe the *act* of sealing one level of it — N contributions +// from level `n` folded into the single node that lands at level `n + 1` — and +// the fourth describes the top of what that folding leaves behind. +// +// # Why the fold has to be a contract member +// +// It is the one step in the tree pipeline that is neither deterministic nor +// local. It costs an inference call, that call is billed, and the provider +// making it is configured on the *driver's* side of the boundary. A host that +// reaches memory only over the module has no chat provider of its own to fold +// with and no rate card to charge against, so the fold has to happen where the +// provider is — and the usage has to come back attached to the text, because +// nothing downstream can recompute it from the words. +// +// That is why [`SummaryOutput`] here carries seven fields rather than the four +// the engine's own deterministic summarisers return: `input_tokens`, +// `output_tokens` and `charged_amount_usd` exist only because a provider was +// paid, and a caller doing cost accounting cannot derive them. +// +// # Every budget crosses, and none of them is defaulted +// +// [`SummaryContext`] carries three separate token numbers and they are not +// interchangeable. `token_budget` clamps the *output*; `input_token_budget` is +// the whole context the fold may occupy; `overhead_reserve_tokens` is the +// prompt and formatting headroom withheld from the sources. The driver divides +// what is left after the other two among the inputs, so a caller that omits +// one and lets it default to zero does not get a slightly different summary — +// it gets every source clamped to nothing and a fold with no evidence in it. +// They are required fields for that reason, not optional ones with a sensible +// fallback: there is no sensible fallback. +// +// [`SummaryContext::ask`] is load-bearing in the same way and in the opposite +// direction: its presence selects an entirely different system prompt. A +// flavoured tree folded without its ask produces a generic digest where the +// caller expected a profile, and nothing in the output says which prompt ran. +// +// # `tree_kind` crosses as a string, not as an enum +// +// The engine's `TreeKind` is `#[non_exhaustive]` and has already grown a fourth +// variant (`flavoured`). A closed enum on the wire would mean the first payload +// naming a kind this build predates fails to *deserialize*, taking the whole +// frame with it — an unfamiliar label degrading into a hard decode failure. +// [`crate::provider::retrieval`] documents that argument at length for +// `RetrievalHit::tree_kind` and `EntityMatch::kind`; this is the same rule +// applied to the same enum, and the hazard is symmetric here because this field +// travels in a *request*: a newer caller naming a newer kind must not make an +// older driver fail to parse the call. +// +// Known values today are `source`, `topic`, `global` and `flavoured`. Unlike +// the response-side fields, though, this one is validated on arrival — a driver +// that cannot map the string onto a kind it understands answers +// [`MemoryError::Invalid`](crate::error::MemoryError::Invalid) naming it, +// because folding under the wrong kind silently mislabels a summary and nothing +// afterwards can tell that it happened. + +/// One contribution being folded — a raw leaf at level 0 on its way to level 1, +/// or a lower-level summary on its way to the level above it. +/// +/// Owned throughout, where the engine's own twin borrows: this shape is +/// serialized into a frame, and a borrow cannot outlive the call that decoded +/// it. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct SummaryInput { + /// Machine-readable id of the contributing leaf or lower-level summary. + /// + /// It is written into the prompt as the provenance marker for its block, so + /// it is not decorative: two inputs sharing an id produce a fold whose + /// sources cannot be told apart afterwards. + pub id: String, + /// Raw text being folded into the parent summary. + /// + /// Clamped driver-side to this input's share of the context budget; an + /// input whose content is blank after trimming is dropped from the prompt + /// entirely rather than contributing an empty block. + pub content: String, + /// Approximate token count of [`content`](Self::content). + pub token_count: u32, + /// Canonical entity ids attached to this input. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub entities: Vec, + /// Topic labels attached to this input. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub topics: Vec, + /// Start of the time window this input covers (inclusive). + pub time_range_start: DateTime, + /// End of the time window this input covers (inclusive). + pub time_range_end: DateTime, + /// Importance weight; higher-scoring inputs are folded first and are least + /// likely to be dropped under budget pressure. + /// + /// The ordering is the driver's, and it is applied to the whole slice + /// before any clamping, so the order the caller sends inputs in does not + /// change the result — the scores do. + pub score: f32, +} + +/// Per-seal context: which tree and level is being sealed, and under what +/// budgets. +/// +/// The engine's twin borrows its strings from the tree it was built over; this +/// one owns them, for the reason [`SummaryInput`] gives. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SummaryContext { + /// Machine-readable id of the tree being sealed. + pub tree_id: String, + /// Wire kind of the tree: `source`, `topic`, `global`, `flavoured`, …. + /// + /// A string rather than an enum, and validated by the driver rather than by + /// serde — see the section note above this type for both halves of that + /// decision. + pub tree_kind: String, + /// Level the produced summary lands at; inputs come from `target_level - 1`. + pub target_level: u32, + /// Maximum approximate tokens the produced summary may occupy. + /// + /// Both an instruction and a clamp: it is stated in the prompt *and* + /// enforced on the returned text, so a provider that overruns it is + /// truncated rather than trusted. + pub token_budget: u32, + /// Total input/context budget available to this fold. + /// + /// What is left of it after [`token_budget`](Self::token_budget) and + /// [`overhead_reserve_tokens`](Self::overhead_reserve_tokens) are withheld + /// is divided evenly among the inputs. A value smaller than those two + /// leaves every input a share of zero, which is a fold over nothing. + pub input_token_budget: u32, + /// Prompt and formatting headroom withheld from source inputs. + pub overhead_reserve_tokens: u32, + /// Natural-language ask that steers the fold, for flavoured trees. + /// + /// `Some` selects a flavour-directed system prompt that distils the inputs + /// into a running profile answering the ask; `None` selects the generic + /// folding prompt every other tree kind uses. An ask that is present but + /// blank reads as `None`. + /// + /// This is not a hint the driver may drop. Sending `None` for a tree that + /// has an ask produces a well-formed summary of the wrong kind, and the + /// response carries no field that says which prompt ran. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ask: Option, +} + +/// The folded summary, with what the provider charged to produce it. +/// +/// # Why this shape and not the engine's four-field one +/// +/// The engine has two summary outputs. The deterministic in-crate summarisers +/// return content, tokens, entities and topics; the provider-backed fold +/// returns those plus the usage the call incurred. This is the second one, +/// deliberately: a fold that crosses the module boundary is always the billed +/// one — the deterministic path never leaves the driver — so dropping the usage +/// fields here would lose the only copy of a number the host meters spend +/// against. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct SummaryOutput { + /// Folded summary text, clamped to the seal's token budget. + /// + /// Empty when there was nothing to fold — every input blank, or the slice + /// itself empty. That is a successful no-op rather than an error: a seal + /// over an empty buffer is idempotent for the same reason + /// `MemoryTree::seal` is. + #[serde(default, skip_serializing_if = "String::is_empty")] + pub content: String, + /// Approximate token count of [`content`](Self::content). + #[serde(default)] + pub token_count: u32, + /// Canonical entity ids for the summary. + /// + /// Emitted empty by the provider fold: entity labelling happens separately, + /// at seal time, under the tree's own label strategy. Carried anyway so a + /// driver whose summariser does extract them has somewhere to put them. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub entities: Vec, + /// Topic labels for the summary; empty on the same terms as + /// [`entities`](Self::entities). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub topics: Vec, + /// Prompt tokens the provider reported for this fold, or `0` when it + /// reported no usage at all. + /// + /// Zero is "not reported", not "free". Providers differ in whether they + /// return usage, and a caller that treats zero as a measured floor + /// under-counts spend rather than over-counting it. + #[serde(default)] + pub input_tokens: u64, + /// Completion tokens the provider reported, on the same terms as + /// [`input_tokens`](Self::input_tokens). + #[serde(default)] + pub output_tokens: u64, + /// What the provider said the call cost, in USD. + /// + /// `None` when the provider quoted nothing, and also when it quoted zero — + /// a zero charge is indistinguishable from an unpriced one, so it is + /// reported as absent rather than as a free call the caller would then add + /// to a running total. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub charged_amount_usd: Option, +} + +/// One namespace's root summary, as a bounded read returns it. +/// +/// A named shape rather than the engine's `(String, String, DateTime)` tuple. +/// The tuple is unambiguous at the one call site that builds it and is nothing +/// but positional on the wire, where the two `String`s are the same type and +/// swapping them produces a payload that decodes cleanly and means the +/// opposite. Naming the fields is what makes that a compile error instead of a +/// mislabelled memory tab. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RootSummary { + /// The namespace the summary is the root of. + pub namespace: String, + /// The root summary text, already truncated to the caller's caps. + /// + /// A body that was cut carries a trailing `[... truncated]` marker, so a + /// caller can tell a clipped summary from a short one without comparing + /// lengths against the caps it asked for. + pub body: String, + /// When the root node was last written. + pub updated_at: DateTime, +} + #[cfg(test)] #[path = "tree_tests.rs"] mod tests; diff --git a/crates/tinymemory-bus/src/tree_tests.rs b/crates/tinymemory-bus/src/tree_tests.rs index bb9965fa..67b5c684 100644 --- a/crates/tinymemory-bus/src/tree_tests.rs +++ b/crates/tinymemory-bus/src/tree_tests.rs @@ -1,4 +1,9 @@ -//! Tests for the markdown time-tree node types. +//! Tests for the markdown time-tree node types, and for the summariser door's +//! wire shapes. + +// A failed assertion in a test is a panic either way; `unwrap`/`expect` here say +// what the invariant was. Same allowance the crate's other test modules take. +#![allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] use super::*; use chrono::TimeZone; @@ -84,3 +89,128 @@ fn node_level_roundtrip() { assert_eq!(NodeLevel::from_str_label(level.as_str()), Some(level)); } } + +#[test] +fn a_summary_context_decodes_a_tree_kind_this_build_has_never_heard_of() { + // The reason `tree_kind` is a `String`. `TreeKind` is `#[non_exhaustive]` + // and has already grown a fourth variant, so a closed enum here would turn + // the first payload naming a fifth into a *decode* failure that takes the + // whole frame with it — a summarise call that fails outright rather than a + // label nothing recognises. Asserted with an invented kind rather than with + // `flavoured`, because `flavoured` would pass even against a closed enum + // that already knows it. + let raw = serde_json::json!({ + "tree_id": "tree-1", + "tree_kind": "a-kind-invented-after-this-build", + "target_level": 2, + "token_budget": 800, + "input_token_budget": 6_000, + "overhead_reserve_tokens": 400, + }); + let context: SummaryContext = + serde_json::from_value(raw).expect("an unknown kind still parses"); + assert_eq!(context.tree_kind, "a-kind-invented-after-this-build"); + assert_eq!(context.ask, None, "an absent ask is the generic fold"); +} + +#[test] +fn a_summary_context_round_trips_all_three_budgets_and_its_ask() { + // Each of these is load-bearing on its own: `token_budget` clamps the + // output, `input_token_budget` is the whole context, and + // `overhead_reserve_tokens` is withheld from the sources before the rest is + // divided. A field that silently failed to cross would not error — it would + // default to zero and produce a fold over nothing, which reads downstream + // as a model that returned little. + let context = SummaryContext { + tree_id: "tree-7".to_string(), + tree_kind: "flavoured".to_string(), + target_level: 3, + token_budget: 900, + input_token_budget: 8_000, + overhead_reserve_tokens: 512, + ask: Some("how does this person write".to_string()), + }; + let encoded = serde_json::to_string(&context).unwrap(); + let decoded: SummaryContext = serde_json::from_str(&encoded).unwrap(); + assert_eq!(decoded, context); + + // The ask changes which system prompt runs, so its absence has to be + // distinguishable from its presence rather than encoded as an empty string. + let generic = SummaryContext { + ask: None, + ..context + }; + let payload: serde_json::Value = serde_json::to_value(&generic).unwrap(); + assert!( + payload.get("ask").is_none(), + "an absent ask is absent from the payload, not an empty one" + ); +} + +#[test] +fn a_summary_input_keeps_its_score_and_its_window() { + let start = Utc.with_ymd_and_hms(2024, 3, 15, 10, 0, 0).unwrap(); + let end = Utc.with_ymd_and_hms(2024, 3, 15, 11, 0, 0).unwrap(); + let input = SummaryInput { + id: "chunk-1".to_string(), + content: "the body being folded".to_string(), + token_count: 5, + entities: vec!["person:ada".to_string()], + topics: vec!["release".to_string()], + time_range_start: start, + time_range_end: end, + score: 0.75, + }; + let decoded: SummaryInput = + serde_json::from_str(&serde_json::to_string(&input).unwrap()).unwrap(); + assert_eq!(decoded, input); + // The score orders the fold and decides what survives budget pressure, so + // it must cross as the float it is rather than being rounded on the way. + assert!((decoded.score - 0.75).abs() < f32::EPSILON); +} + +#[test] +fn a_summary_output_reports_no_usage_as_zero_and_no_charge_as_absent() { + // The default is what a fold with nothing to fold returns, and it has to + // decode from a payload that omits every optional field — an older peer's + // shape, and also the cheapest thing a driver can send. + let empty: SummaryOutput = serde_json::from_str("{}").unwrap(); + assert_eq!(empty, SummaryOutput::default()); + assert!(empty.content.is_empty()); + assert_eq!(empty.input_tokens, 0); + assert_eq!( + empty.charged_amount_usd, None, + "an unpriced call is absent, not a zero a caller would add to a total" + ); + + let billed = SummaryOutput { + content: "folded".to_string(), + token_count: 2, + entities: Vec::new(), + topics: Vec::new(), + input_tokens: 1_200, + output_tokens: 300, + charged_amount_usd: Some(0.0042), + }; + let decoded: SummaryOutput = + serde_json::from_str(&serde_json::to_string(&billed).unwrap()).unwrap(); + assert_eq!(decoded, billed); +} + +#[test] +fn a_root_summary_travels_by_name_so_its_two_strings_cannot_be_swapped() { + // The whole reason this is not the engine's `(String, String, DateTime)` + // tuple. Positionally, `namespace` and `body` are the same type: a producer + // that emitted them the other way round would encode cleanly, decode + // cleanly, and put a whole summary where a namespace label belongs. + let summary = RootSummary { + namespace: "team".to_string(), + body: "what the team did\n\n[... truncated]".to_string(), + updated_at: Utc.with_ymd_and_hms(2024, 3, 15, 14, 0, 0).unwrap(), + }; + let payload = serde_json::to_value(&summary).unwrap(); + assert_eq!(payload["namespace"], "team"); + assert_eq!(payload["body"], "what the team did\n\n[... truncated]"); + let decoded: RootSummary = serde_json::from_value(payload).unwrap(); + assert_eq!(decoded, summary); +} diff --git a/crates/tinymemory-core/src/sources/status.rs b/crates/tinymemory-core/src/sources/status.rs index c04f2467..a2a9ad59 100644 --- a/crates/tinymemory-core/src/sources/status.rs +++ b/crates/tinymemory-core/src/sources/status.rs @@ -21,6 +21,8 @@ //! recorded in `mem_tree_chunk_reembed_skipped`. Both are terminal, and both //! count as resolved. +use anyhow::Result; +use rusqlite::Connection; use serde::Serialize; use crate::sources::types::{MemorySourceEntry, SourceKind}; @@ -45,61 +47,133 @@ pub struct SourceStatus { pub freshness: FreshnessLabel, } +/// What one source's chunks amount to, before a freshness label is put on them. +/// +/// The counting half of a [`SourceStatus`], split out because two callers need +/// exactly it and nothing else: this module's own [`source_status`], and the +/// memory contract's `MemoryChunks::source_ingest_status`, which answers for a +/// registry it does not own and therefore cannot name a source's freshness +/// vocabulary either. +/// +/// Freshness is deliberately not here. It is arithmetic over +/// [`Self::last_chunk_at_ms`] and a clock, and folding it in would make every +/// caller inherit the moment the count was taken as the moment the label was +/// computed. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct IngestCounts { + /// Chunk rows the store holds under the prefix, whatever their state. + pub chunks_synced: u64, + /// Rows still in flight: no embedding, not dropped, not skipped. + pub chunks_pending: u64, + /// Source time of the newest row under the prefix, epoch milliseconds. + pub last_chunk_at_ms: Option, +} + +/// The one definition of what a source's chunks count as. +/// +/// "Pending" is **not resolved**, and a chunk resolves three ways: it has an +/// embedding, it was dropped by the lifecycle, or it was deliberately recorded +/// as skipped for re-embedding. All three are terminal. This is the engine's +/// own predicate from `list_sync_statuses`, kept identical so the per-source +/// view and the per-provider one cannot disagree about the same chunk — and +/// written once here so the two callers of it cannot drift apart the way the +/// second copy of [`source_id_prefix`] did. +/// +/// `?1` is a `LIKE` pattern, not a literal prefix: the wildcard is the +/// caller's to place, and `ESCAPE` is declared so a caller that means a `%` or +/// a `_` literally can say so. [`source_id_prefix`] deliberately does not +/// escape — its ids are generated and its patterns are the ones this module +/// has always sent. +const INGEST_COUNTS_SQL: &str = "SELECT \ + COUNT(*), \ + SUM(CASE WHEN EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_embeddings e \ + WHERE e.chunk_id = c.id) \ + OR c.lifecycle_status = 'dropped' \ + OR EXISTS ( \ + SELECT 1 FROM mem_tree_chunk_reembed_skipped s \ + WHERE s.chunk_id = c.id) \ + THEN 0 ELSE 1 END), \ + MAX(c.timestamp_ms) \ + FROM mem_tree_chunks c \ + WHERE c.source_id LIKE ?1 ESCAPE '\\'"; + +/// Count the chunks one `source_id LIKE` pattern selects, on an open +/// connection. +/// +/// Always yields a row. The query is a bare aggregate with no `GROUP BY`, so a +/// pattern matching nothing returns one row of `(0, NULL, NULL)` rather than no +/// rows — which is what lets a caller ask about a source that has never synced +/// and get zeroes instead of an absence. +fn ingest_counts_on_connection(conn: &Connection, pattern: &str) -> Result { + let (synced, pending, last_ts): (i64, i64, Option) = + conn.query_row(INGEST_COUNTS_SQL, [pattern], |r| { + Ok(( + r.get(0)?, + // `SUM` over no rows is `NULL`, not `0`. + r.get::<_, Option>(1)?.unwrap_or(0), + r.get(2)?, + )) + })?; + + Ok(IngestCounts { + chunks_synced: synced.max(0) as u64, + chunks_pending: pending.max(0) as u64, + last_chunk_at_ms: last_ts, + }) +} + +/// Counts for several `source_id LIKE` patterns, in the order asked. +/// +/// One connection for the batch and one statement per pattern. Synchronous +/// SQLite work: an async caller runs it on a blocking thread, as +/// [`source_status`] does. +/// +/// Surfaces real query errors rather than degrading, so status telemetry cannot +/// report a healthy zero-row state over a database that is actually broken. An +/// empty `patterns` opens no connection at all. +/// +/// # Errors +/// +/// Any failure opening the chunk store or running the count. +pub fn ingest_counts_for_patterns( + config: &Config, + patterns: &[String], +) -> Result> { + if patterns.is_empty() { + return Ok(Vec::new()); + } + with_connection(config, |conn| { + patterns + .iter() + .map(|pattern| ingest_counts_on_connection(conn, pattern)) + .collect() + }) +} + /// Compute status for one source. pub async fn source_status( config: &Config, source: &MemorySourceEntry, ) -> Result { let cfg = config.to_arc(); - let source_clone = source.clone(); + let source_id = source.id.clone(); + let pattern = source_id_prefix(source); tokio::task::spawn_blocking(move || { - with_connection(&*cfg, |conn| { - let prefix = source_id_prefix(&source_clone); - - // Surface real query errors so status telemetry doesn't lie about - // a healthy zero-row state when the DB is actually broken. - // - // "Pending" is "not resolved", and a chunk resolves three ways: - // it has an embedding, it was dropped by the lifecycle, or it was - // deliberately skipped for re-embedding. This is the engine's own - // predicate from `list_sync_statuses`, kept identical so the - // per-source view and the per-provider one cannot disagree about - // the same chunk. - let (synced, pending, last_ts): (i64, i64, Option) = conn.query_row( - "SELECT \ - COUNT(*), \ - SUM(CASE WHEN EXISTS ( \ - SELECT 1 FROM mem_tree_chunk_embeddings e \ - WHERE e.chunk_id = c.id) \ - OR c.lifecycle_status = 'dropped' \ - OR EXISTS ( \ - SELECT 1 FROM mem_tree_chunk_reembed_skipped s \ - WHERE s.chunk_id = c.id) \ - THEN 0 ELSE 1 END), \ - MAX(c.timestamp_ms) \ - FROM mem_tree_chunks c \ - WHERE c.source_id LIKE ?1", - [&prefix], - |r| { - Ok(( - r.get(0)?, - r.get::<_, Option>(1)?.unwrap_or(0), - r.get(2)?, - )) - }, - )?; + let counts = ingest_counts_for_patterns(&*cfg, std::slice::from_ref(&pattern)) + .map_err(|e| format!("source_status: {e}"))?; + // One pattern in, one row out — the batch always answers per pattern. + let counts = counts.first().copied().unwrap_or_default(); - let now_ms = chrono::Utc::now().timestamp_millis(); - Ok(SourceStatus { - source_id: source_clone.id.clone(), - chunks_synced: synced.max(0) as u64, - chunks_pending: pending.max(0) as u64, - last_chunk_at_ms: last_ts, - freshness: FreshnessLabel::from_age_ms(last_ts, now_ms), - }) + let now_ms = chrono::Utc::now().timestamp_millis(); + Ok(SourceStatus { + source_id, + chunks_synced: counts.chunks_synced, + chunks_pending: counts.chunks_pending, + last_chunk_at_ms: counts.last_chunk_at_ms, + freshness: FreshnessLabel::from_age_ms(counts.last_chunk_at_ms, now_ms), }) - .map_err(|e| format!("source_status: {e}")) }) .await .map_err(|e| format!("source_status join: {e}"))? diff --git a/crates/tinymemory-core/src/sources/status_tests.rs b/crates/tinymemory-core/src/sources/status_tests.rs index 0ebe08fa..88365ae9 100644 --- a/crates/tinymemory-core/src/sources/status_tests.rs +++ b/crates/tinymemory-core/src/sources/status_tests.rs @@ -158,3 +158,111 @@ async fn a_source_with_no_chunks_reports_zeroes() { assert_eq!(status.last_chunk_at_ms, None); assert_eq!(status.freshness, FreshnessLabel::Idle); } + +/// A pattern that matches nothing still gets a row. +/// +/// This is the difference between the counting surface and a `GROUP BY` over +/// the chunk table: a group with no rows is *absent*, so a caller building a +/// dashboard from groups loses a never-synced source instead of showing it +/// idle. The batch answers per pattern, in order, so the caller can pair rows +/// with the sources it asked about. +#[tokio::test] +async fn the_batch_answers_one_row_per_pattern_including_the_empty_ones() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace.path().join("workspace"); + let config = host.to_arc(); + + crate::store::chunks::store::upsert_chunks( + &*config, + &[chunk("chunk-batch-1", "mem_src:src_batch:item-1")], + ) + .expect("upsert chunks"); + + let patterns = vec![ + "mem_src:src_batch:%".to_string(), + "mem_src:src_never_synced:%".to_string(), + "mem_src:src_batch:%".to_string(), + ]; + let counts = ingest_counts_for_patterns(&*config, &patterns).expect("counts"); + + assert_eq!( + counts.len(), + patterns.len(), + "one row per pattern, in order" + ); + assert_eq!(counts[0].chunks_synced, 1); + assert_eq!( + counts[1], + IngestCounts::default(), + "a source that has never synced reports zeroes, not an absent row" + ); + assert_eq!(counts[1].last_chunk_at_ms, None); + assert_eq!( + counts[2].chunks_synced, 1, + "the batch does not consume rows" + ); +} + +/// An empty ask is answered without opening the store. +#[test] +fn an_empty_batch_touches_nothing() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + // Deliberately a workspace that was never created: if the empty case + // reached the store this would fail rather than return an empty vector. + let mut host = TestHostConfig::default(); + host.workspace_dir = std::path::PathBuf::from("/nonexistent/tinymemory/status/batch"); + let config = host.to_arc(); + + let counts = ingest_counts_for_patterns(&*config, &[]).expect("empty batch"); + assert!(counts.is_empty()); +} + +/// `_` and `%` in a pattern are the caller's to escape. +/// +/// The `ESCAPE` clause is what lets the memory contract's +/// `source_ingest_status` honour its own promise that a chunk-id prefix is +/// matched literally — without it, a source keyed `src_a` also counts the +/// chunks of any source whose id differs only where the underscore is. +/// [`source_id_prefix`] deliberately does not escape, so this pins the clause +/// rather than the existing caller's use of it. +#[tokio::test] +async fn an_escaped_wildcard_matches_itself() { + use tinymemory_api::host::test_support::TestHostConfig; + use tinymemory_api::host::MemoryHostConfig; + + crate::test_seams::init(); + let workspace = tempfile::tempdir().expect("workspace"); + let mut host = TestHostConfig::default(); + host.workspace_dir = workspace.path().join("workspace"); + let config = host.to_arc(); + + crate::store::chunks::store::upsert_chunks( + &*config, + &[ + chunk("chunk-underscore", "mem_src:src_a:item-1"), + chunk("chunk-collider", "mem_src:srcXa:item-1"), + ], + ) + .expect("upsert chunks"); + + let unescaped = ingest_counts_for_patterns(&*config, &["mem_src:src_a:%".to_string()]) + .expect("unescaped counts"); + assert_eq!( + unescaped[0].chunks_synced, 2, + "an unescaped `_` is a single-character wildcard, which is why the contract escapes" + ); + + let escaped = ingest_counts_for_patterns(&*config, &[r"mem_src:src\_a:%".to_string()]) + .expect("escaped counts"); + assert_eq!( + escaped[0].chunks_synced, 1, + "an escaped `_` matches only itself" + ); +} diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 926d1a15..9808732d 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -673,6 +673,16 @@ mod exports { "ExtractEntities", "EmbedText", "EmbedderSlug", + // The summariser door, and the roots folding leaves behind. + "Summarise", + "RootSummaries", + // The three doors a host opens once it stops linking the engine + // itself: the cheap degradation poll beside the full diagnosis, the + // scorer's verdict on one chunk, and per-configured-source ingest + // progress — none of which any earlier member can answer. + "DegradedState", + "ChunkScore", + "SourceIngestStatus", ], signals = [], // The host's embedder is deliberately NOT declared as `requires`. That diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 92766b96..a8d5082c 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -35,6 +35,8 @@ //! StorageKinds() -> [String] //! ListChunkDetails(query, scope) -> [ChunkListRow] //! SourceTotals(limit, scope) -> [SourceTotal] +//! ChunkScore(chunk_id) -> Option +//! SourceIngestStatus(source_prefixes) -> [SourceIngestStatus] //! //! ListActiveFacets() / ListAllFacets() -> [ProfileFacet] //! GetFacet(key) / FacetsByType(type) -> facet(s) @@ -53,6 +55,8 @@ //! //! SummaryForest(limit, scope) -> SummaryForest //! RecentLeaves(limit, scope) -> [TreeLeaf] +//! Summarise(inputs, context) -> SummaryOutput +//! RootSummaries(per_namespace_cap, total_cap) -> [RootSummary] //! //! TopEntities(kind, limit) -> [EntityOccurrence] //! ChunkEntities(chunk_ids, kinds) -> [ChunkEntityOccurrence] @@ -63,6 +67,7 @@ //! //! FlushSourceTree(source_scope) -> u64 //! Diagnose() -> Diagnosis +//! DegradedState() -> DegradedCapabilities //! //! RunConnectionSync(toolkit, connection_id) -> SyncRunOutcome //! BootstrapConnection(toolkit, connection_id) -> () @@ -163,9 +168,10 @@ use tinymemory_api::provider::types::{ // imported: they are supertraits of `MemoryProvider`, so their methods are // already callable on the trait object. use tinymemory_api::provider::chunks::{ - ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, SourceTotal, + ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, ChunkScore, SourceIngestQuery, + SourceIngestStatus, SourceTotal, }; -use tinymemory_api::provider::diagnosis::Diagnosis; +use tinymemory_api::provider::diagnosis::{DegradedCapabilities, Diagnosis}; use tinymemory_api::provider::episodic::{ConversationSegment, EpisodicEvent, EpisodicTurn}; use tinymemory_api::provider::people::{ AddressBookSeedOutcome, PersonHandle, PersonInteraction, PersonRecord, PersonScore, @@ -186,7 +192,10 @@ use tinymemory_api::provider::sync::{ use tinymemory_api::provider::MemoryProvider; use tinymemory_api::recall::OwnedRecallOpts; use tinymemory_api::tool_memory::ToolMemoryRule; -use tinymemory_api::tree::{IngestRequest, QueryResult, SummaryForest, TreeLeaf, TreeStatus}; +use tinymemory_api::tree::{ + IngestRequest, QueryResult, RootSummary, SummaryContext, SummaryForest, SummaryInput, + SummaryOutput, TreeLeaf, TreeStatus, +}; use tinymemory_api::types::{ GraphRelationRecord, MemoryCategory, MemoryEntry, MemoryKvRecord, MemoryTaint, NamespaceDocumentInput, NamespaceMemoryHit, NamespaceRetrievalContext, NamespaceSummary, @@ -1921,6 +1930,130 @@ impl MemoryService { .await .map_err(|error| into_bus_error(&error)) } + + /// Fold summary inputs into one parent summary, through the driver's chat + /// provider. + /// + /// Appended here rather than filed beside `Seal` for the reason + /// `count_chunks` gives above: member order is wire order. + /// + /// The longest-running member of the tree family: one provider call, over + /// the network, priced at the driver's rate. It is a call rather than a job + /// for the same reason `RebuildFromRawArchive` is — the module holds no + /// notion of a caller's request, so a fire-and-forget fold would have + /// nowhere to report the summary it produced. + /// + /// Not size-checked. The response is one summary, clamped driver-side to + /// the `token_budget` the caller itself supplied, so no input can make it + /// exceed a frame. The *request* can be large — it carries every input's + /// body — and that bound is the caller's: it chose how many inputs to fold. + async fn summarise( + &self, + inputs: Vec, + context: SummaryContext, + ) -> BusResult { + require_family!(self, as_tree, Capability::Tree) + .summarise(&inputs, &context) + .await + .map_err(|error| into_bus_error(&error)) + } + + /// Every namespace's root summary, capped per namespace and in total. + /// + /// Appended here for the reason above. The member name is deliberately + /// shorter than the trait method it forwards to + /// (`MemoryTree::root_summaries_with_caps`): the caps are visible in the + /// signature on both sides, and a wire name is a string a host spells by + /// hand, so it carries only what distinguishes the call. + /// + /// Size-checked even though `total_cap` already bounds the payload in + /// characters, because that bound is the *caller's* number and nothing + /// stops it being larger than a frame. A named refusal telling the caller + /// to lower it beats a response the host cannot decode. + async fn root_summaries( + &self, + per_namespace_cap: usize, + total_cap: usize, + ) -> BusResult> { + let summaries = require_family!(self, as_tree, Capability::Tree) + .root_summaries_with_caps(per_namespace_cap, total_cap) + .await + .map_err(|error| into_bus_error(&error))?; + ensure_response_fits(&summaries, "RootSummaries")?; + Ok(summaries) + } + + /// Which capabilities are currently running in a reduced mode. + /// + /// Appended here for the reason `count_chunks` gives above: member order is + /// wire order. + /// + /// Beside `Diagnose` rather than inside it, and the difference is the price. + /// `Diagnose` runs the driver's whole diagnostic pass — an aggregate scan of + /// the chunk table, three job counts, an extraction-coverage measurement and + /// a walk of the pipeline configuration. This reads the flags the pipeline + /// set as it ran. A status indicator polls the second; only a human asks for + /// the first. + /// + /// Not size-checked: three booleans and at most one classified cause. + async fn degraded_state(&self) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .degraded_state() + .await + .map_err(|error| into_bus_error(&error)) + } + + /// One chunk's admission decision and the signals behind it. + /// + /// Appended here for the reason above. + /// + /// A diagnostic read — "why is this in memory, and why is that not" — and + /// not an input to ranking, which the retrieval family owns. `None` is a + /// chunk that was never scored, which is a different fact from one that + /// scored zero; the driver must not collapse them and neither may a caller. + /// + /// Not size-checked. The response is one row of numbers plus, at most, the + /// driver's own short rationale for the verdict. + async fn chunk_score(&self, chunk_id: String) -> BusResult> { + require_family!(self, as_chunks, Capability::Chunks) + .chunk_score(&chunk_id) + .await + .map_err(|error| into_bus_error(&error)) + } + + /// How far ingest has got for each configured source the caller names. + /// + /// Appended here for the reason above. + /// + /// The caller supplies the chunk-id prefix per source because deriving it + /// needs the host's source registry — the source's kind, its toolkit, its + /// connection id — which is state the driver does not have and this contract + /// exists to stop it reaching for. The driver answers only what it can read + /// from its own tables: how many rows sit under that key, and how many of + /// them are still in flight. + /// + /// A row comes back for every query, zero-filled when the prefix matches + /// nothing. That is the whole reason this is not `SourceTotals`, which + /// returns the groups that exist and therefore drops a source that has never + /// synced — off a dashboard, where an absent row reads as a source that was + /// never configured. + /// + /// Neither the prefixes nor the ids are logged: a connector prefix carries a + /// connection id, which is user data. + /// + /// Size-checked, because the caller chooses how many sources to ask about + /// and the rows are small but unbounded in number. + async fn source_ingest_status( + &self, + source_prefixes: Vec, + ) -> BusResult> { + let rows = require_family!(self, as_chunks, Capability::Chunks) + .source_ingest_status(&source_prefixes) + .await + .map_err(|error| into_bus_error(&error))?; + ensure_response_fits(&rows, "SourceIngestStatus")?; + Ok(rows) + } } /// The response-size ceiling for a method that returns a list of entries. diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index e4dc6a36..207a7120 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -717,6 +717,9 @@ const EXPECTED_METHODS: &[&str] = &[ "ExtractEntities", "EmbedText", "EmbedderSlug", + // The tree family's summariser door and its root read. + "Summarise", + "RootSummaries", ]; #[tokio::test] diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 13bcf05b..48fd1420 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -42,8 +42,8 @@ use tinymemory_api::provider::types::{ #[cfg(feature = "memory-git")] use tinymemory_api::provider::types::{ChangeKind, DiffReport, SnapshotRef, SourceChange}; use tinymemory_api::provider::{ - AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, - CodingSessionIngestReport, CodingSessionIngestRequest, CodingSessionSource, + AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkListRow, ChunkQuery, ChunkScore, + ChunkScoreSignals, CodingSessionIngestReport, CodingSessionIngestRequest, CodingSessionSource, ConversationSegment, CoverWindowQuery, DegradedCapabilities, Diagnosis, DiagnosisCounters, DiagnosisFailure, DiagnosisStage, EntityMatch, EpisodicEvent, EpisodicTurn, EventKind, FacetType, FastRetrieveQuery, MemoryChunks, MemoryCodingSessions, MemoryCore, MemoryDiff, @@ -52,13 +52,15 @@ use tinymemory_api::provider::{ MemoryRecall, MemoryRetrieval, MemoryScoring, MemorySourceSink, MemorySourceSync, MemoryToolMemory, MemoryTree, PersonHandle, PersonInteraction, PersonRecord, PersonScore, ProfileFacet, RankedPerson, RawArchiveCoverage, RawRebuildOutcome, ResolvedPerson, - RetrievalHit, RetrievalResponse, SourceRetrievalQuery, SourceSyncState, SourceSyncStatus, - SourceTotal, SyncAuditEntry, SyncFreshness, SyncRunOutcome, UserState, + RetrievalHit, RetrievalResponse, SourceIngestQuery, SourceIngestStatus, SourceRetrievalQuery, + SourceSyncState, SourceSyncStatus, SourceTotal, SyncAuditEntry, SyncFreshness, SyncRunOutcome, + UserState, }; use tinymemory_api::recall::OwnedRecallOpts; use tinymemory_api::tool_memory::ToolMemoryRule; use tinymemory_api::tree::{ - IngestRequest, QueryResult, SummaryForest, TreeLeaf, TreeStatus, TreeSummary, + IngestRequest, QueryResult, RootSummary, SummaryContext, SummaryForest, SummaryInput, + SummaryOutput, TreeLeaf, TreeStatus, TreeSummary, }; use tinymemory_api::types::{ GraphRelationRecord, MemoryCategory, MemoryEntry, MemoryKvRecord, MemoryTaint, @@ -68,7 +70,14 @@ use tinymemory_api::types::{ // The KV read paths must address rows by the same canonical form the // write-path shim stores them under — see `MemoryGraph::kv_get` below. use tinymemory_core::store::safety::canonical_identifier; +use tinymemory_core::store::trees::TreeKind; use tinymemory_core::store::{MemoryClient, MemoryClientRef}; +// The engine's own summariser twins, aliased because the contract's owned wire +// types share their names. Same shape either side of the seam; the difference +// is that these borrow and those do not. +use tinymemory_core::tree::summarise::{ + SummaryContext as EngineSummaryContext, SummaryInput as EngineSummaryInput, +}; /// The concrete, credential-free host configuration available inside a module. #[derive(Debug, Clone)] @@ -1299,6 +1308,123 @@ impl MemoryTree for TinycortexProvider { .map_err(|error| Self::other("flush the source tree", error))?; Ok(u64::try_from(sealed.len()).unwrap_or(u64::MAX)) } + + async fn summarise( + &self, + inputs: &[SummaryInput], + context: &SummaryContext, + ) -> Result { + // The kind arrives as an open string and is mapped back here, at the + // edge, because the engine's own signature takes the enum. An + // unrecognised value is a caller mistake and is refused rather than + // defaulted: `TreeKind` chooses the labelling policy a seal writes + // under, so folding a `flavoured` tree as a `source` one produces a + // well-formed summary filed against the wrong policy, and nothing in + // the tree afterwards records that the substitution happened. + let tree_kind = TreeKind::parse(&context.tree_kind).map_err(MemoryError::Invalid)?; + + // Field-by-field rather than a serde round-trip: the engine's twins + // derive neither `Serialize` nor `Deserialize`, and writing the mapping + // out means a field added to either side is a compile error here rather + // than a value that silently stops crossing. + let engine_inputs: Vec = inputs + .iter() + .map(|input| EngineSummaryInput { + id: input.id.clone(), + content: input.content.clone(), + token_count: input.token_count, + entities: input.entities.clone(), + topics: input.topics.clone(), + time_range_start: input.time_range_start, + time_range_end: input.time_range_end, + score: input.score, + }) + .collect(); + + // Every budget and the ask are carried through exactly as the caller + // stated them. Nothing is clamped, defaulted or second-guessed on the + // way in: the caller owns the level being sealed, so it owns the budget + // for it, and the engine already treats these three numbers as one + // arithmetic (see `prepare_summary_prompt`) that a partial substitution + // would silently unbalance. + let engine_context = EngineSummaryContext { + tree_id: context.tree_id.as_str(), + tree_kind, + target_level: context.target_level, + token_budget: context.token_budget, + input_token_budget: context.input_token_budget, + overhead_reserve_tokens: context.overhead_reserve_tokens, + ask: context.ask.as_deref(), + }; + + // Not on a blocking thread, unlike almost everything else in this file: + // this is an outbound provider call that awaits the network, so it + // yields its worker between steps. `spawn_blocking` would hold a + // blocking-pool thread idle for the whole round trip. + let output = tinymemory_core::tree::summarise::summarise( + &self.config, + &engine_inputs, + &engine_context, + ) + .await + .map_err(|error| Self::other("summarise tree inputs", error))?; + + // The error above is propagated rather than turned into the engine's + // deterministic `fallback_summary`. The fallback belongs to the caller + // that owns the cascade — it is what the engine's own seal path + // substitutes when a summariser errors — and a driver that applied it + // here would hand back a concatenation the caller could not tell apart + // from a model's work. + Ok(SummaryOutput { + content: output.content, + token_count: output.token_count, + entities: output.entities, + topics: output.topics, + input_tokens: output.input_tokens, + output_tokens: output.output_tokens, + charged_amount_usd: output.charged_amount_usd, + }) + } + + async fn root_summaries_with_caps( + &self, + per_namespace_cap: usize, + total_cap: usize, + ) -> Result, MemoryError> { + // The workspace root comes from this driver's own configuration, never + // from the caller: a path argument here would be both a configuration + // crossing the contract and an unbounded filesystem read addressed by + // whoever placed the call. + let rows = blocking( + self.config.clone(), + "collect root summaries", + move |config| { + // Infallible by construction — the engine swallows a failed + // scan into an empty vector — so the `anyhow` wrapper here is + // the `blocking` helper's shape, not a hidden error path. + Ok( + tinymemory_core::tree::tree_runtime::store::collect_root_summaries_with_caps( + &config.workspace_dir, + per_namespace_cap, + total_cap, + ), + ) + }, + ) + .await?; + + // The tuple is named on the way out and nowhere else: positional + // `(namespace, body, updated_at)` is exactly the shape that survives a + // swap of its two `String`s without complaint. + Ok(rows + .into_iter() + .map(|(namespace, body, updated_at)| RootSummary { + namespace, + body, + updated_at, + }) + .collect()) + } } /// Validate entity-kind wire strings and re-emit them in the index's spelling. @@ -2265,12 +2391,7 @@ impl MemoryMaintenance for TinycortexProvider { }) .collect(), first_blocking_cause: report.first_blocking_cause.as_ref().map(diagnosis_failure), - degraded: DegradedCapabilities { - semantic_recall: report.degraded.semantic_recall, - structure: report.degraded.structure, - storage: report.degraded.storage, - cause: report.degraded.cause.as_ref().map(diagnosis_failure), - }, + degraded: degraded_capabilities(&report.degraded), counters: DiagnosisCounters { total_chunks: report.counters.total_chunks, jobs_ready: report.counters.jobs_ready, @@ -2280,6 +2401,56 @@ impl MemoryMaintenance for TinycortexProvider { }, }) } + + /// The degradation flags on their own, without the diagnosis around them. + /// + /// The same three booleans and the same cause + /// [`MemoryMaintenance::diagnose`] reports, read from the same place — the + /// process-global atomics the embed, extract and storage stages set as they + /// fail — and crossed by the same function, so the two members cannot + /// disagree about what is degraded. + /// + /// # Why it does not delegate to `diagnose` + /// + /// Because that would defeat the point of the member. `async_run_doctor` + /// counts every chunk in the store, counts jobs in three states, measures + /// extraction coverage across the whole chunk table, and walks the routing + /// and scheduler configuration — all on a blocking thread, because it is + /// enough SQLite work to hold one. This reads three atomics and three more + /// for their causes: no query, no thread hop, nothing that can fail. A + /// status light polling the diagnosis would put an aggregate scan of the + /// chunk table on a repeating timer. + /// + /// It is not `async` work at all, and is deliberately not wrapped in + /// `spawn_blocking` the way this file's storage reads are: dispatching a + /// blocking task to load six atomics costs more than the load does. + /// + /// It cannot fail, for the same reason [`MemoryMaintenance::diagnose`] + /// cannot: there is no fallible step to map. `Ok` here is the honest + /// answer, not a swallowed error. + async fn degraded_state(&self) -> Result { + Ok(degraded_capabilities( + &tinymemory_core::tree::health::current_degraded_state(), + )) + } +} + +/// Carry the engine's degradation snapshot across as the contract's shape. +/// +/// Shared by [`MemoryMaintenance::diagnose`] and +/// [`MemoryMaintenance::degraded_state`] rather than written out twice. The two +/// members answer the same question at different prices, so a caller can +/// reasonably compare their answers — and two copies of a four-field mapping +/// are two copies that can disagree about which flag is which. +fn degraded_capabilities( + degraded: &tinymemory_core::tree::health::DegradedState, +) -> DegradedCapabilities { + DegradedCapabilities { + semantic_recall: degraded.semantic_recall, + structure: degraded.structure, + storage: degraded.storage, + cause: degraded.cause.as_ref().map(diagnosis_failure), + } } /// Carry one engine pipeline failure across as the contract's own shape. @@ -3278,6 +3449,126 @@ impl MemoryChunks for TinycortexProvider { embeddings.sort_by(|a, b| a.chunk_id.cmp(&b.chunk_id)); Ok(embeddings) } + + async fn chunk_score(&self, chunk_id: &str) -> Result, MemoryError> { + let id = chunk_id.to_string(); + let row = blocking(self.config.clone(), "read chunk score", move |config| { + tinymemory_core::tree::score::store::get_score(config, &id) + }) + .await?; + + // Absence stays absence. The engine answers `None` both for a chunk it + // has never heard of and for one it holds but never scored, and neither + // is a zero score — see the contract's own note on why collapsing them + // reports a verdict that was never reached. + Ok(row.map(|row| ChunkScore { + chunk_id: row.chunk_id, + total: row.total, + signals: ChunkScoreSignals { + token_count: row.signals.token_count, + unique_words: row.signals.unique_words, + metadata_weight: row.signals.metadata_weight, + source_weight: row.signals.source_weight, + interaction: row.signals.interaction, + entity_density: row.signals.entity_density, + // Carried rather than dropped, and always `0.0` from this + // engine: `mem_tree_score` has no column for it, so the row it + // reads back cannot hold what the extractor rated the chunk. + // The `total` it contributed to is what survived. + llm_importance: row.signals.llm_importance, + }, + dropped: row.dropped, + reason: row.reason, + computed_at_ms: row.computed_at_ms, + // Same story as `llm_importance`: diagnostic only, no column, + // always `None` from a stored row. + llm_importance_reason: row.llm_importance_reason, + })) + } + + async fn source_ingest_status( + &self, + source_prefixes: &[SourceIngestQuery], + ) -> Result, MemoryError> { + // Answered without opening the store, which is not merely an + // optimisation: the contract says an empty ask yields an empty answer, + // and a driver that opened a database to discover that would fail a + // caller with nothing configured on a store that has never been built. + if source_prefixes.is_empty() { + return Ok(Vec::new()); + } + + let patterns: Vec = source_prefixes + .iter() + .map(|query| like_prefix_pattern(&query.chunk_id_prefix)) + .collect(); + let asked = patterns.len(); + let counts = blocking( + self.config.clone(), + "read source ingest status", + move |config| { + tinymemory_core::sources::status::ingest_counts_for_patterns(config, &patterns) + }, + ) + .await?; + + // The engine promises a row per pattern — the query is a bare aggregate + // with no `GROUP BY`, so a pattern matching nothing still returns one. + // Checked rather than trusted because the failure mode of zipping two + // lists of different lengths is a silent truncation, and the rows it + // would drop are exactly the never-synced sources this member exists to + // report. + if counts.len() != asked { + return Err(Self::other( + "read source ingest status", + format!( + "asked for {asked} sources and the engine answered {}", + counts.len() + ), + )); + } + + Ok(source_prefixes + .iter() + .zip(counts) + .map(|(query, counts)| SourceIngestStatus { + // Echoed from the query, not derived from the chunk rows: the + // registry id and the ingest key are different identifiers, and + // for a connector source they share no substring. + source_id: query.source_id.clone(), + chunks_synced: counts.chunks_synced, + chunks_pending: counts.chunks_pending, + last_chunk_at_ms: counts.last_chunk_at_ms, + }) + .collect()) + } +} + +/// Turn a literal chunk-id prefix into the `LIKE` pattern that selects it. +/// +/// The contract calls [`SourceIngestQuery::chunk_id_prefix`] a *literal* +/// prefix, so honouring it means escaping the pattern metacharacters before +/// appending the wildcard. Without that a source keyed `mem_src:src_a:` also +/// counts the chunks of any source whose id differs only where the underscore +/// is — `_` is `LIKE`'s single-character wildcard, and every generated source +/// id contains one. The count would be wrong in the direction that looks +/// healthy: too many chunks, attributed to the wrong source. +/// +/// `\` is escaped along with `%` and `_`, and pairs with the `ESCAPE '\'` +/// clause the engine's counting query declares. This is the same construction — +/// and the same argument — as the engine's own `like_contains_pattern`, which +/// exists because a source id containing `_` had already been observed to match +/// more than it should. +fn like_prefix_pattern(prefix: &str) -> String { + let mut pattern = String::with_capacity(prefix.len() + 1); + for character in prefix.chars() { + if matches!(character, '\\' | '%' | '_') { + pattern.push('\\'); + } + pattern.push(character); + } + pattern.push('%'); + pattern } #[async_trait] diff --git a/crates/tinymemory-tinycortex/src/engine/test.rs b/crates/tinymemory-tinycortex/src/engine/test.rs index 7a4455e0..ee753928 100644 --- a/crates/tinymemory-tinycortex/src/engine/test.rs +++ b/crates/tinymemory-tinycortex/src/engine/test.rs @@ -23,9 +23,10 @@ use tinymemory_api::provider::types::IngestItem; use tinymemory_api::types::MemoryTaint; use super::{ - advertised_capabilities, audit_entry, diagnosis_failure, facet_type_to_engine, - handle_to_contract, handle_to_engine, parse_person_id, refuse_composio_dispatch, - scope_to_engine, validate_ingest_item, EngineRuntimeConfig, + advertised_capabilities, audit_entry, degraded_capabilities, diagnosis_failure, + facet_type_to_engine, handle_to_contract, handle_to_engine, like_prefix_pattern, + parse_person_id, refuse_composio_dispatch, scope_to_engine, validate_ingest_item, + EngineRuntimeConfig, }; fn ingest_item(content: &str, mime: Option<&str>, taint: MemoryTaint) -> IngestItem { @@ -512,3 +513,66 @@ fn set_memory_sources_json_writes_through_to_the_registry_file() { entries ); } + +#[test] +fn a_degradation_snapshot_crosses_every_flag_and_its_cause() { + // `Diagnose` and `DegradedState` answer the same question at different + // prices, so a caller can reasonably compare them. One mapping, asserted + // field by field, is what makes that comparison safe — a transposed pair + // here would have the two members disagree about which capability is + // reduced, and both would still look like plausible answers. + use tinymemory_core::tree::health::{DegradedState, FailureCode, PipelineFailure}; + + let degraded = DegradedState { + semantic_recall: true, + structure: false, + storage: true, + cause: Some(PipelineFailure::new(FailureCode::StorageUnavailable)), + }; + let crossed = degraded_capabilities(°raded); + assert!(crossed.semantic_recall); + assert!(!crossed.structure); + assert!(crossed.storage); + assert_eq!( + crossed.cause.as_ref().map(|failure| failure.code.as_str()), + Some(FailureCode::StorageUnavailable.as_str()) + ); + + // Nothing degraded is nothing to explain: a cause carried over a cleared + // set of flags would put a remediation on a panel with no fault on it. + let clear = degraded_capabilities(&DegradedState::default()); + assert_eq!( + clear, + tinymemory_api::provider::DegradedCapabilities::default() + ); + assert_eq!(clear.cause, None); +} + +#[test] +fn a_chunk_id_prefix_is_matched_literally() { + // The contract calls the prefix literal, so the driver has to make `LIKE` + // agree. Every generated source id contains an underscore, which is `LIKE`'s + // single-character wildcard — left unescaped, `mem_src:src_a:` would also + // count another source's chunks, and the count would be wrong in the + // direction that looks healthy. + assert_eq!( + like_prefix_pattern("mem_src:src_a:"), + r"mem\_src:src\_a:%", + "every underscore is escaped, not left as a wildcard" + ); + assert_eq!( + like_prefix_pattern("gmail:conn-1:"), + "gmail:conn-1:%", + "a prefix with no metacharacter gains only the trailing wildcard" + ); + assert_eq!( + like_prefix_pattern("100%_of\\it"), + r"100\%\_of\\it%", + "the escape character itself is escaped, as the ESCAPE clause requires" + ); + assert_eq!( + like_prefix_pattern(""), + "%", + "an empty prefix matches everything, which is what an empty prefix means" + ); +} diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index e977a9e5..825f0149 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -3388,3 +3388,171 @@ async fn occurrences_read_in_a_batch_stay_attached_to_their_own_chunk() { Err(MemoryError::Invalid(_)) )); } + +/// The summariser door, on the two paths that need no provider. +/// +/// The fold itself cannot be asserted here — it is an outbound model call, and +/// this suite configures none — but the two decisions the driver makes *before* +/// it reaches one can be, and both are the ones a caller trips over: an empty +/// fold must be a successful no-op rather than an error a cascade has to +/// special-case, and a tree kind the engine does not have must be refused +/// rather than folded under a guessed one. +#[tokio::test(flavor = "multi_thread")] +async fn the_summariser_refuses_an_unknown_tree_kind_and_folds_nothing_without_a_provider() { + use tinymemory_api::error::MemoryError; + use tinymemory_api::provider::{MemoryProvider, SummaryContext, SummaryInput}; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let tree = provider.as_tree().expect("Tree"); + + let context = |kind: &str| SummaryContext { + tree_id: "tree-1".into(), + tree_kind: kind.into(), + target_level: 1, + token_budget: 400, + input_token_budget: 4_000, + overhead_reserve_tokens: 200, + ask: None, + }; + + // An unrecognised kind is refused, and refused *first* — before any + // provider is built, which is why this assertion holds with none + // configured. A default to `source` would fold a flavoured tree under the + // wrong labelling policy and leave nothing behind that says so. + assert!( + matches!( + tree.summarise(&[], &context("a-kind-this-engine-never-had")) + .await, + Err(MemoryError::Invalid(_)) + ), + "an unknown tree kind is a caller mistake, not a silent substitution" + ); + + // Nothing to fold is a successful no-op: the prompt builder finds no + // content, so no provider is reached and the default output comes back. + // This is the idempotence a cascade relies on to call the door at every + // level unconditionally. + let empty = tree + .summarise(&[], &context("source")) + .await + .expect("an empty fold is not an error"); + assert!(empty.content.is_empty()); + assert_eq!(empty.token_count, 0); + assert_eq!(empty.input_tokens, 0); + assert_eq!(empty.output_tokens, 0); + assert_eq!( + empty.charged_amount_usd, None, + "a fold that reached no provider was not billed" + ); + + // Inputs that are all blank are the same case, and it is worth pinning + // separately: the emptiness is decided after trimming, inside the engine, + // not by the slice being empty here. + let at = chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("timestamp"); + let blank = SummaryInput { + id: "chunk-1".into(), + content: " \n\t ".into(), + token_count: 0, + entities: Vec::new(), + topics: Vec::new(), + time_range_start: at, + time_range_end: at, + score: 1.0, + }; + let blank_fold = tree + .summarise(std::slice::from_ref(&blank), &context("source")) + .await + .expect("a fold over blank inputs is not an error"); + assert!(blank_fold.content.is_empty()); + + // Every kind the engine actually has is accepted, including the fourth one + // it grew after this contract was written — the reason the field crosses as + // a string rather than as a closed enum. + for kind in ["source", "topic", "global", "flavoured"] { + tree.summarise(&[], &context(kind)) + .await + .expect("a kind the engine has parses"); + } +} + +/// The root-summary read: stable order, both caps, and an empty workspace. +#[tokio::test(flavor = "multi_thread")] +async fn the_root_summary_read_caps_each_namespace_and_then_the_whole_block() { + use tinymemory_api::provider::MemoryProvider; + use tinymemory_core::tree::tree_runtime::{ + derive_parent_id, estimate_tokens, level_from_node_id, TreeNode, + }; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + let config = provider_config(workspace.path(), serde_json::Value::Null); + let tree = provider.as_tree().expect("Tree"); + + // A workspace with no tree at all is an empty block rather than an error: + // the caller is building a prompt, and "nothing to add" is an answer. + assert!(tree + .root_summaries_with_caps(1_000, 10_000) + .await + .expect("an empty workspace is not an error") + .is_empty()); + + const ALPHA: &str = "alpha root summary"; + const BETA: &str = "beta root summary"; + let at = chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("timestamp"); + // Seeded through the engine's own writer rather than by hand, so the test + // cannot pass against a file layout the reader does not actually use. + for (namespace, summary) in [("alpha", ALPHA), ("beta", BETA)] { + tinymemory_core::tree::tree_runtime::store::write_node( + &config, + &TreeNode { + node_id: "root".into(), + namespace: namespace.into(), + level: level_from_node_id("root"), + parent_id: derive_parent_id("root"), + summary: summary.into(), + token_count: estimate_tokens(summary), + child_count: 0, + created_at: at, + updated_at: at, + metadata: None, + }, + ) + .expect("write a root node"); + } + + let all = tree + .root_summaries_with_caps(1_000, 10_000) + .await + .expect("root summaries"); + assert_eq!(all.len(), 2); + assert_eq!( + all[0].namespace, "alpha", + "namespaces come back in stable sorted order, which is what makes a \ + binding total cap predictable" + ); + assert_eq!(all[0].body, ALPHA); + assert_eq!(all[0].updated_at, at); + assert_eq!(all[1].namespace, "beta"); + assert_eq!(all[1].body, BETA); + + // The per-namespace cap clips a body and marks it, so a caller can tell a + // clipped summary from a short one without re-deriving the cap. + let clipped = tree + .root_summaries_with_caps(5, 10_000) + .await + .expect("clipped root summaries"); + assert_eq!(clipped.len(), 2); + assert!(clipped[0].body.starts_with("alpha")); + assert!(clipped[0].body.ends_with("[... truncated]")); + + // The total cap stops the walk. It drops the *tail* of the namespace list + // rather than sampling across it, which is exactly what a caller must not + // read as "these are all the namespaces". + let bounded = tree + .root_summaries_with_caps(1_000, ALPHA.chars().count()) + .await + .expect("bounded root summaries"); + assert_eq!(bounded.len(), 1); + assert_eq!(bounded[0].namespace, "alpha"); +} From 4af18174d92557dfbdb19e0d136c4d0cdb6c1c55 Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 31 Aug 2026 16:58:44 +0530 Subject: [PATCH 2/3] fix(review): manifest drift list, trimmed lint allow, error contracts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three review findings from #122, one of which was also the red CI lane: - module_e2e's EXPECTED_METHODS — the test's own hardcoded manifest copy — gained Summarise/RootSummaries but not the other three new members, so the_manifest_declares_every_method_the_module_serves failed on exactly ["ChunkScore", "DegradedState", "SourceIngestStatus"]. All five are now listed. This list is deliberately a second copy (drift witness), which is why it fails instead of following names.rs automatically. - chunks_tests.rs allowed clippy::unwrap_used and clippy::panic while using only expect; the allowance now matches what the file actually trips. - source_status/status_list document their failure contracts: per-source store failures degrade that row to zeroed counts rather than failing the batch (verified against the loop, not asserted from memory), and a prefix matching nothing is an answer, not an error. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-bus/src/provider/chunks_tests.rs | 6 +++--- crates/tinymemory-core/src/sources/status.rs | 13 +++++++++++++ crates/tinymemory-module/tests/module_e2e.rs | 5 +++++ 3 files changed, 21 insertions(+), 3 deletions(-) diff --git a/crates/tinymemory-bus/src/provider/chunks_tests.rs b/crates/tinymemory-bus/src/provider/chunks_tests.rs index 0ac962d6..2bfc4d7a 100644 --- a/crates/tinymemory-bus/src/provider/chunks_tests.rs +++ b/crates/tinymemory-bus/src/provider/chunks_tests.rs @@ -7,9 +7,9 @@ //! nothing in it. Both failures render as a plausible screen rather than as an //! error. -// A failed assertion in a test is a panic either way; `unwrap`/`expect` here say -// what the invariant was. Same allowance the crate's other test modules take. -#![allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)] +// A failed assertion in a test is a panic either way; `expect` here says what +// the invariant was. Only the lint this file actually trips is allowed. +#![allow(clippy::expect_used)] use super::*; diff --git a/crates/tinymemory-core/src/sources/status.rs b/crates/tinymemory-core/src/sources/status.rs index a2a9ad59..430ecf6a 100644 --- a/crates/tinymemory-core/src/sources/status.rs +++ b/crates/tinymemory-core/src/sources/status.rs @@ -152,6 +152,13 @@ pub fn ingest_counts_for_patterns( } /// Compute status for one source. +/// +/// # Errors +/// +/// Fails when the chunk store cannot be opened or the count query fails +/// (storage unavailable, corrupt store), or when the blocking task is +/// cancelled at shutdown. A source whose prefix matches nothing is NOT an +/// error — it answers with zeroed counts. pub async fn source_status( config: &Config, source: &MemorySourceEntry, @@ -180,6 +187,12 @@ pub async fn source_status( } /// Compute status for all configured sources (one SQL roundtrip per source). +/// +/// # Errors +/// +/// Fails only when the source registry itself cannot be read. A per-source +/// store failure does not fail the batch — that row degrades to zeroed +/// counts, so one bad source cannot blank the whole dashboard. pub async fn status_list(config: &Config) -> Result, String> { let sources = crate::sources::registry::list_sources().await?; let mut out = Vec::with_capacity(sources.len()); diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index 207a7120..ef0e7106 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -720,6 +720,11 @@ const EXPECTED_METHODS: &[&str] = &[ // The tree family's summariser door and its root read. "Summarise", "RootSummaries", + // The maintenance hot-path read and the two chunk-family doors that + // landed beside it. + "DegradedState", + "ChunkScore", + "SourceIngestStatus", ]; #[tokio::test] From 1bd984d257fde76670c881c11a52cfbb57028ece Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 31 Aug 2026 17:11:26 +0530 Subject: [PATCH 3/3] fix(review): the LIKE-escape helper lives beside the query it pairs with MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tinysweeper flagged ingest_counts_for_patterns for taking raw LIKE patterns. Half the finding does not hold — the pattern is a bound parameter, so it can never terminate or extend the SQL — but the kernel does: a pub function whose argument is a pattern invites a future caller to pass an unescaped id, and `%`/`_` then over-match silently. The sanctioned escape (like_prefix_pattern) moves from the tinycortex driver into core, directly beside the ESCAPE '\' query it pairs with, and the driver's copy becomes a name for it. The pattern contract is now documented on the function itself: bound therefore injection-free, over-match therefore escape-first, and the helper to do it one line up. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-core/src/sources/status.rs | 29 +++++++++++++++++++ .../tinymemory-tinycortex/src/engine/mod.rs | 19 ++++-------- 2 files changed, 34 insertions(+), 14 deletions(-) diff --git a/crates/tinymemory-core/src/sources/status.rs b/crates/tinymemory-core/src/sources/status.rs index 430ecf6a..2d4055d6 100644 --- a/crates/tinymemory-core/src/sources/status.rs +++ b/crates/tinymemory-core/src/sources/status.rs @@ -125,6 +125,26 @@ fn ingest_counts_on_connection(conn: &Connection, pattern: &str) -> Result String { + let mut pattern = String::with_capacity(prefix.len() + 1); + for character in prefix.chars() { + if matches!(character, '\\' | '%' | '_') { + pattern.push('\\'); + } + pattern.push(character); + } + pattern.push('%'); + pattern +} + /// One connection for the batch and one statement per pattern. Synchronous /// SQLite work: an async caller runs it on a blocking thread, as /// [`source_status`] does. @@ -136,6 +156,15 @@ fn ingest_counts_on_connection(conn: &Connection, pattern: &str) -> Result String { - let mut pattern = String::with_capacity(prefix.len() + 1); - for character in prefix.chars() { - if matches!(character, '\\' | '%' | '_') { - pattern.push('\\'); - } - pattern.push(character); - } - pattern.push('%'); - pattern + tinymemory_core::sources::status::like_prefix_pattern(prefix) } #[async_trait]