From 07610bdd4205deb1e767d1417bd080c81c063ae4 Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 24 Aug 2026 14:08:33 +0530 Subject: [PATCH 1/4] Give recency recall a member of its own, because the obvious substitute is wrong MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two OpenHuman handlers — `memory.recall_context` and `memory.recall_memories` — still call the engine directly because the contract has no recency path. `MemoryRetrieval::recall_namespace_scored` looks like the twin and is not. The two share a prefix (`load_documents_for_scope` + `kv_records_for_scope`) and diverge after it. The scored path ranks candidates against the query; handed an empty string it still runs the ranking, with nothing to rank against. It does not degrade to recency — it returns hits ordered by a similarity signal computed from nothing. That makes the substitution worse than a missing method: it compiles, returns plausible hits, and quietly changes what the user gets back. So this adds `recall_namespace_recent(namespace, limit)`, resolving to the engine's `recall_namespace_memories`, which orders by freshness and priority. Same `NamespaceMemoryHit` shape as the scored path, so a host re-ranking on engine signals treats both uniformly. Required rather than defaulted, like every other member of this family. A default returning an empty vector would be indistinguishable from a namespace with nothing in it — the silent-empty failure this contract has been bitten by before. The null provider refuses with `Unsupported(Retrieval)`, as its siblings do. Registered in all four lists — the interface block, the module manifest, `METHODS` with its length (94 -> 95), and the loader's `EXPECTED_METHODS`. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-api/src/null.rs | 8 +++ .../tinymemory-api/src/provider/retrieval.rs | 33 +++++++++++ crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 5 +- crates/tinymemory-module/src/lib.rs | 1 + crates/tinymemory-module/src/service/mod.rs | 13 ++++ crates/tinymemory-module/src/service/test.rs | 2 +- crates/tinymemory-module/tests/module_e2e.rs | 1 + .../tinymemory-tinycortex/src/engine/mod.rs | 19 ++++++ .../tests/full_provider_conformance.rs | 59 +++++++++++++++++++ 10 files changed, 140 insertions(+), 3 deletions(-) diff --git a/crates/tinymemory-api/src/null.rs b/crates/tinymemory-api/src/null.rs index e94a11b9..c7e28fe7 100644 --- a/crates/tinymemory-api/src/null.rs +++ b/crates/tinymemory-api/src/null.rs @@ -612,6 +612,14 @@ impl MemoryRetrieval for NullMemoryProvider { unsupported(Capability::Retrieval) } + async fn recall_namespace_recent( + &self, + _namespace: &str, + _limit: usize, + ) -> Result, MemoryError> { + unsupported(Capability::Retrieval) + } + async fn search_entities( &self, _query: &str, diff --git a/crates/tinymemory-api/src/provider/retrieval.rs b/crates/tinymemory-api/src/provider/retrieval.rs index 666396d4..a87586ac 100644 --- a/crates/tinymemory-api/src/provider/retrieval.rs +++ b/crates/tinymemory-api/src/provider/retrieval.rs @@ -178,6 +178,39 @@ pub trait MemoryRetrieval: Send + Sync { exclude_session_id: Option<&str>, ) -> Result, MemoryError>; + /// Namespace recall ordered by **recency**, with no query to rank against. + /// + /// # Why this is not [`Self::recall_namespace_scored`] with an empty query + /// + /// It looks like the same call with one argument left blank, and it is not. + /// The two share a prefix — loading the namespace's documents and key-value + /// records — and diverge after it. The scored path ranks candidates against + /// the query text; handed an empty string it still runs the ranking, with + /// nothing to rank against, and returns hits ordered by a similarity signal + /// computed from nothing. + /// + /// This path never ranks. It orders by freshness and priority, which is + /// what a caller asking "what is in this namespace" means, and what a + /// context-assembly step needs when there is no user query yet. + /// + /// The substitution is dangerous precisely because it compiles, returns + /// plausible hits, and quietly changes what the user gets back. A caller + /// that *has* a real query wants the scored path; one that does not wants + /// this. + /// + /// Hits carry the same [`NamespaceMemoryHit`] shape as the scored path, so + /// a host re-ranking on engine signals treats the two uniformly. + /// + /// # Errors + /// + /// Backend failures; an unknown namespace yields an empty vector, which is + /// a true statement about it rather than a fault. + async fn recall_namespace_recent( + &self, + namespace: &str, + limit: usize, + ) -> Result, MemoryError>; + /// Free-text search over the entity index. /// /// `kinds` filters by classification; `None` matches every kind. This is diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index cc00d88d..f560a896 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 94 members on it, built as a `cdylib`. A host that +//! exports one object with 95 members on it, 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 a9f71615..a42200d8 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -161,6 +161,8 @@ pub mod methods { /// `BackfillInProgress` — whether a re-embedding backfill is still running /// anywhere in the driver's process. pub const BACKFILL_IN_PROGRESS: &str = "BackfillInProgress"; + /// `RecallNamespaceRecent` — namespace recall ordered by recency, no query. + pub const RECALL_NAMESPACE_RECENT: &str = "RecallNamespaceRecent"; // The people store: ranking, handles, scores and interactions. /// `ListPeople` — list people. @@ -250,7 +252,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; 94] = [ +pub const METHODS: [&str; 95] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -307,6 +309,7 @@ pub const METHODS: [&str; 94] = [ methods::QUEUE_STATS, methods::LATEST_QUEUE_FAILURE, methods::BACKFILL_IN_PROGRESS, + methods::RECALL_NAMESPACE_RECENT, methods::LIST_PEOPLE, methods::GET_PERSON, methods::RESOLVE_HANDLE, diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 5812ff27..9ce9c759 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -312,6 +312,7 @@ mod exports { "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "RecallNamespaceRecent", ], 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 e0c75f8d..a3b82e56 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -922,6 +922,19 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn recall_namespace_recent( + &self, + namespace: String, + limit: usize, + ) -> BusResult> { + let hits = require_family!(self, as_retrieval, Capability::Retrieval) + .recall_namespace_recent(&namespace, limit) + .await + .map_err(|error| into_bus_error(&error))?; + ensure_response_fits(&hits, "RecallNamespaceRecent")?; + Ok(hits) + } + // ── People ────────────────────────────────────────────────────────────── /// Known people, ranked by closeness. diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 45bc41dc..5798bcb7 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -544,7 +544,7 @@ fn the_served_members_are_exactly_the_published_contract() { .map(|member| (*member).to_string()) .collect(); - // Reported as differences rather than as a 94-element inequality, so the + // Reported as differences rather than as a 95-element inequality, so the // failure names the method that moved instead of printing both lists. let missing: Vec<&String> = served.iter().filter(|m| !published.contains(m)).collect(); assert!( diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index 9e751c56..f0b8eff9 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -676,6 +676,7 @@ const EXPECTED_METHODS: &[&str] = &[ "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "RecallNamespaceRecent", ]; #[tokio::test] diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index b460709e..d4cf05a1 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -2039,6 +2039,25 @@ impl MemoryRetrieval for TinycortexProvider { Self::cross(&hits, "convert namespace hits") } + async fn recall_namespace_recent( + &self, + namespace: &str, + limit: usize, + ) -> Result, MemoryError> { + // `recall_namespace_memories`, deliberately — NOT + // `query_namespace_hits_excluding_session` with an empty query. The two + // share `load_documents_for_scope` + `kv_records_for_scope` and diverge + // after it: one ranks against the query, this one scores freshness and + // priority. Passing "" to the scored path does not degrade to recency. + let hits = self + .client + .unified_handle() + .recall_namespace_memories(namespace, u32::try_from(limit).unwrap_or(u32::MAX)) + .await + .map_err(|error| Self::other("recall namespace recent", error))?; + Self::cross(&hits, "convert namespace hits") + } + async fn search_entities( &self, query: &str, diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 67fbb889..f3ab202f 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -1197,6 +1197,65 @@ async fn ingest_chunks_and_retrieval_cover_success_and_validation_without_networ ); } +/// Recency recall answers when there is no query to rank against. +/// +/// This is the member's whole reason for existing. `recall_namespace_scored` +/// looks like the same call with the query left blank, and handing it `""` +/// does not degrade to recency — it runs the ranking path against nothing. A +/// context-assembly step that has not seen a user query yet needs the +/// namespace's contents ordered by freshness, which is what this returns. +/// +/// What is pinned: writes are visible through it, and an unknown namespace is +/// an empty answer rather than an error — a true statement about that +/// namespace, not a fault the caller can act on. +#[tokio::test(flavor = "multi_thread")] +async fn recency_recall_answers_without_a_query() { + use tinymemory_api::provider::{MemoryCore, MemoryProvider}; + use tinymemory_api::types::{MemoryCategory, MemoryTaint}; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + + for (key, content) in [ + ("first", "the earliest note in this namespace"), + ("second", "a later note about something else entirely"), + ] { + provider + .store( + "global", + key, + content, + MemoryCategory::Core, + None, + MemoryTaint::default(), + ) + .await + .expect("store"); + } + + let retrieval = provider.as_retrieval().expect("Retrieval"); + + let recent = retrieval + .recall_namespace_recent("global", 5) + .await + .expect("recency recall"); + assert!( + !recent.is_empty(), + "a caller with no query still gets the namespace's contents back" + ); + assert!( + recent.len() <= 5, + "the limit is honoured: asked for 5, got {}", + recent.len() + ); + + let empty = retrieval + .recall_namespace_recent("a-namespace-nothing-was-written-to", 5) + .await + .expect("an unknown namespace is not an error"); + assert!(empty.is_empty()); +} + #[tokio::test(flavor = "multi_thread")] async fn tree_entities_and_maintenance_execute_real_workspace_transitions() { use tinymemory_api::provider::MemoryProvider; From 238b17994b20f939da59c3533690d2414346c415 Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 24 Aug 2026 14:37:08 +0530 Subject: [PATCH 2/4] Give the queue's two compound operations members, and move their SQL here MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpenHuman's `reset_tree` and `flush_now` handlers are the last queue call sites, and neither is a thin call the existing families can express. Both drive raw SQL against `mem_tree_*` from the host — tables this engine owns — so giving them contract members means the logic moves here rather than a signature being added over it. `flush_pending` answers a "flush now" control. Deduplication is the driver's, keyed on date and three-hour block, so two presses inside one window schedule the work once. `FlushOutcome` carries the buffer count beside `enqueued` because either number alone misleads: `enqueued: false` is ambiguous between "nothing to flush" and "already scheduled", and a caller showing the first when it is the second is lying about state the user is watching. `reset_derived_index` discards summaries, buffers, entity indexes and trees, then schedules their re-derivation. It is one operation on purpose — deleting the derived rows without queueing the rebuild leaves a store that answers structural queries with nothing and looks healthy doing it. `mem_tree_chunks` is deliberately not in the table list: the chunks are the source, they are never deleted, and the conformance test asserts the chunk count is unchanged across a reset. A reset that took them too would be data loss wearing the word "reset". Two transactions rather than one, also on purpose: the first drops the job table, so re-enqueueing inside it would race its own truncation. `ResetOutcome` carries three counts because they are not a ratio — rows deleted, chunks returned to scope, and jobs scheduled are independent, and collapsing any pair loses the ability to tell "nothing to re-derive" from "re-derivation was not scheduled". Jobs can be fewer than chunks because the enqueue is keyed. The null provider refuses both rather than taking the trait defaults. The defaults are right for a real driver with nothing buffered or nothing derived; they are wrong for a provider that stores nothing, where "flushed nothing" and "reset nothing" would read as work done rather than as a driver that cannot do it. Registration order matters and the test proved it: the interface serves these with their family, so `METHODS`, the manifest and `EXPECTED_METHODS` list them before `RecallNamespaceRecent`. `the_served_members_are_exactly_the_published_contract` compares sequences, not sets, and caught the mismatch on the first run. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-api/src/null.rs | 17 ++- crates/tinymemory-api/src/provider/mod.rs | 6 +- crates/tinymemory-api/src/provider/records.rs | 51 +++++++- crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 8 +- crates/tinymemory-bus/src/provider/types.rs | 35 ++++++ crates/tinymemory-module/src/lib.rs | 2 + crates/tinymemory-module/src/service/mod.rs | 19 ++- crates/tinymemory-module/src/service/test.rs | 2 +- crates/tinymemory-module/tests/module_e2e.rs | 2 + .../tinymemory-tinycortex/src/engine/mod.rs | 115 +++++++++++++++++- .../tests/full_provider_conformance.rs | 93 ++++++++++++++ 12 files changed, 339 insertions(+), 13 deletions(-) diff --git a/crates/tinymemory-api/src/null.rs b/crates/tinymemory-api/src/null.rs index c7e28fe7..446ca6c0 100644 --- a/crates/tinymemory-api/src/null.rs +++ b/crates/tinymemory-api/src/null.rs @@ -56,8 +56,8 @@ use crate::error::MemoryError; use crate::goals::GoalsDoc; use crate::health::MemoryHealth; use crate::provider::types::{ - DiffReport, EntityHit, ExportPage, ExportRecord, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, SnapshotRef, SourceItem, SourceScope, + DiffReport, EntityHit, ExportPage, ExportRecord, FlushOutcome, ImportOutcome, IngestItem, + IngestOutcome, MaintenanceReport, ResetOutcome, SnapshotRef, SourceItem, SourceScope, }; use crate::provider::{ AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, CoverWindowQuery, EntityMatch, @@ -478,6 +478,19 @@ impl MemoryMaintenance for NullMemoryProvider { async fn doctor(&self) -> Result { unsupported(Capability::Maintenance) } + + // The trait defaults these to an empty outcome, which is the right answer + // for a real driver that simply has nothing buffered or nothing derived. + // It is the wrong answer here: both *mutate*, and this provider stores + // nothing, so "flushed nothing" and "reset nothing" would read as work + // done rather than as a driver that cannot do it. + async fn flush_pending(&self) -> Result { + unsupported(Capability::Maintenance) + } + + async fn reset_derived_index(&self) -> Result { + unsupported(Capability::Maintenance) + } } #[async_trait] diff --git a/crates/tinymemory-api/src/provider/mod.rs b/crates/tinymemory-api/src/provider/mod.rs index 32d2c861..7ca7701a 100644 --- a/crates/tinymemory-api/src/provider/mod.rs +++ b/crates/tinymemory-api/src/provider/mod.rs @@ -91,7 +91,7 @@ pub use retrieval::{ RetrievalNodeKind, RetrievalResponse, SourceRetrievalQuery, }; pub use types::{ - ChangeKind, DiffReport, EntityHit, EntityRef, ExportPage, ExportRecord, ImportOutcome, - IngestItem, IngestOutcome, MaintenanceReport, SnapshotRef, SourceChange, SourceItem, - SourceScope, + ChangeKind, DiffReport, EntityHit, EntityRef, ExportPage, ExportRecord, FlushOutcome, + ImportOutcome, IngestItem, IngestOutcome, MaintenanceReport, ResetOutcome, SnapshotRef, + SourceChange, SourceItem, SourceScope, }; diff --git a/crates/tinymemory-api/src/provider/records.rs b/crates/tinymemory-api/src/provider/records.rs index 0e3038f2..952d3f7a 100644 --- a/crates/tinymemory-api/src/provider/records.rs +++ b/crates/tinymemory-api/src/provider/records.rs @@ -19,7 +19,8 @@ use async_trait::async_trait; use crate::error::MemoryError; use crate::goals::GoalsDoc; use crate::provider::types::{ - IngestOutcome, MaintenanceReport, QueueFailure, QueueStats, SourceItem, StoreStats, + FlushOutcome, IngestOutcome, MaintenanceReport, QueueFailure, QueueStats, ResetOutcome, + SourceItem, StoreStats, }; use crate::tool_memory::ToolMemoryRule; use crate::types::MemoryTaint; @@ -263,4 +264,52 @@ pub trait MemoryMaintenance: Send + Sync { async fn backfill_in_progress(&self) -> Result { Ok(false) } + + /// Flush buffered work that is old enough to be written out. + /// + /// The caller is a "flush now" control: a user who does not want to wait + /// for the scheduled window. Whether a flush is *scheduled* is the + /// driver's business — this asks it to consider the buffers now, and + /// reports what it found and whether it acted. + /// + /// Deduplication is the driver's, not the caller's. Two flushes inside one + /// window must not schedule the work twice, and the second reports + /// `enqueued: false` with a truthful `stale_buffers` — which is why both + /// numbers are on [`FlushOutcome`] rather than a bare bool. + /// + /// Defaulted to an empty outcome: a driver with nothing buffered has + /// nothing to flush, which is true of it rather than a refusal. + /// + /// # Errors + /// + /// Backend failures only. + async fn flush_pending(&self) -> Result { + Ok(FlushOutcome::default()) + } + + /// Drop everything derived from stored content and schedule its + /// re-derivation. + /// + /// Summaries, buffers, entity indexes and the trees over them are all + /// *derived* — recomputable from the chunks they were built from. This + /// discards them and queues the work to build them again. **Nothing a + /// caller wrote is deleted**, which is the invariant that makes it safe to + /// offer as an operator control at all; a driver that cannot promise that + /// must refuse rather than implement this. + /// + /// Necessarily one operation. Deleting the derived rows without scheduling + /// re-derivation leaves a store that answers structural queries with + /// nothing and looks healthy doing it, and the two halves are not + /// separately useful. + /// + /// Defaulted to an empty outcome, for a driver with nothing derived. + /// + /// # Errors + /// + /// Backend failures only. A driver that keeps derived state it cannot + /// rebuild should answer [`MemoryError::Unsupported`] rather than delete + /// it. + async fn reset_derived_index(&self) -> Result { + Ok(ResetOutcome::default()) + } } diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index f560a896..c5346e7e 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 95 members on it, built as a `cdylib`. A host that +//! exports one object with 97 members on it, 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 a42200d8..3980f543 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -163,6 +163,10 @@ pub mod methods { pub const BACKFILL_IN_PROGRESS: &str = "BackfillInProgress"; /// `RecallNamespaceRecent` — namespace recall ordered by recency, no query. pub const RECALL_NAMESPACE_RECENT: &str = "RecallNamespaceRecent"; + /// `FlushPending` — flush buffered work old enough to be written out. + pub const FLUSH_PENDING: &str = "FlushPending"; + /// `ResetDerivedIndex` — drop derived state and schedule its rebuild. + pub const RESET_DERIVED_INDEX: &str = "ResetDerivedIndex"; // The people store: ranking, handles, scores and interactions. /// `ListPeople` — list people. @@ -252,7 +256,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; 95] = [ +pub const METHODS: [&str; 97] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -309,6 +313,8 @@ pub const METHODS: [&str; 95] = [ methods::QUEUE_STATS, methods::LATEST_QUEUE_FAILURE, methods::BACKFILL_IN_PROGRESS, + methods::FLUSH_PENDING, + methods::RESET_DERIVED_INDEX, methods::RECALL_NAMESPACE_RECENT, methods::LIST_PEOPLE, methods::GET_PERSON, diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index ab8a6b47..d98bcb03 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -489,6 +489,41 @@ pub struct QueueFailure { pub last_success_ms: Option, } +/// What a flush of pending buffered work did, and what was pending. +/// +/// Both numbers, because either alone misleads. `enqueued: false` with +/// `stale_buffers: 0` means there was nothing to do; `enqueued: false` with +/// `stale_buffers: 3` means the driver deduplicated against work it had +/// already scheduled — the same answer for opposite reasons, and a caller +/// showing "nothing to flush" in the second case is wrong. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct FlushOutcome { + /// Whether this call scheduled work. `false` when an equivalent flush is + /// already scheduled — a deduplication, not a failure. + pub enqueued: bool, + /// Buffers old enough to be flushed, at the moment the driver looked. + pub stale_buffers: u64, +} + +/// What resetting the derived index deleted, requeued and scheduled. +/// +/// Three numbers rather than a `MaintenanceReport`'s two, because they are not +/// a ratio: rows deleted, chunks put back in scope, and jobs scheduled to +/// re-derive from them are three independent counts, and collapsing any pair +/// loses the ability to tell "nothing to re-derive" from "re-derivation was +/// not scheduled". +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ResetOutcome { + /// Rows removed from the derived tables. + pub rows_deleted: u64, + /// Source chunks returned to the pool the index is derived from. + pub chunks_requeued: u64, + /// Re-derivation jobs scheduled. Lower than `chunks_requeued` when some + /// were already queued — the enqueue is keyed, so a duplicate is a no-op + /// rather than a second job. + pub jobs_enqueued: u64, +} + #[cfg(test)] #[path = "types_tests.rs"] mod tests; diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index 9ce9c759..e05a2f25 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -312,6 +312,8 @@ mod exports { "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "FlushPending", + "ResetDerivedIndex", "RecallNamespaceRecent", ], signals = [], diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index a3b82e56..f643c885 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -120,8 +120,9 @@ use tinymemory_api::error::MemoryError; use tinymemory_api::goals::GoalsDoc; use tinymemory_api::health::MemoryHealth; use tinymemory_api::provider::types::{ - DiffReport, EntityHit, ExportPage, ExportRecord, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, QueueFailure, QueueStats, SnapshotRef, SourceItem, SourceScope, StoreStats, + DiffReport, EntityHit, ExportPage, ExportRecord, FlushOutcome, ImportOutcome, IngestItem, + IngestOutcome, MaintenanceReport, QueueFailure, QueueStats, ResetOutcome, SnapshotRef, + SourceItem, SourceScope, StoreStats, }; // `MemoryCore`, `MemoryRecall` and `MemoryPortability` are deliberately not // imported: they are supertraits of `MemoryProvider`, so their methods are @@ -922,6 +923,20 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn flush_pending(&self) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .flush_pending() + .await + .map_err(|error| into_bus_error(&error)) + } + + async fn reset_derived_index(&self) -> BusResult { + require_family!(self, as_maintenance, Capability::Maintenance) + .reset_derived_index() + .await + .map_err(|error| into_bus_error(&error)) + } + async fn recall_namespace_recent( &self, namespace: String, diff --git a/crates/tinymemory-module/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 5798bcb7..02bfb2b2 100644 --- a/crates/tinymemory-module/src/service/test.rs +++ b/crates/tinymemory-module/src/service/test.rs @@ -544,7 +544,7 @@ fn the_served_members_are_exactly_the_published_contract() { .map(|member| (*member).to_string()) .collect(); - // Reported as differences rather than as a 95-element inequality, so the + // Reported as differences rather than as a 97-element inequality, so the // failure names the method that moved instead of printing both lists. let missing: Vec<&String> = served.iter().filter(|m| !published.contains(m)).collect(); assert!( diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index f0b8eff9..e7d9a1ed 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -676,6 +676,8 @@ const EXPECTED_METHODS: &[&str] = &[ "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "FlushPending", + "ResetDerivedIndex", "RecallNamespaceRecent", ]; diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index d4cf05a1..52f89490 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -32,8 +32,9 @@ use tinymemory_api::host::{ }; use tinymemory_api::mandatory::MemoryTraitProvider; use tinymemory_api::provider::types::{ - EntityHit, EntityRef, ExportPage, ExportRecord, ImportOutcome, IngestItem, IngestOutcome, - MaintenanceReport, QueueFailure, QueueStats, SourceItem, SourceScope, StoreStats, + EntityHit, EntityRef, ExportPage, ExportRecord, FlushOutcome, ImportOutcome, IngestItem, + IngestOutcome, MaintenanceReport, QueueFailure, QueueStats, ResetOutcome, SourceItem, + SourceScope, StoreStats, }; // Diff-family value types, used only by the `MemoryDiff` impl below — which is // compiled out without the git-backed snapshot store. @@ -1348,6 +1349,116 @@ impl MemoryMaintenance for TinycortexProvider { .await } + async fn flush_pending(&self) -> Result { + blocking( + self.config.clone(), + "flush pending buffers", + move |config| { + let now = chrono::Utc::now(); + let stale = tinymemory_core::store::trees::store::list_stale_buffers(config, now)?; + let stale_buffers = u64::try_from(stale.len()).unwrap_or(u64::MAX); + + // `max_age_secs: 0` is what "now" means here: consider every buffer + // rather than only those past the scheduled age. + let payload = tinymemory_core::queue::types::FlushStalePayload { + max_age_secs: Some(0), + }; + // The key is date + three-hour block, so a second flush inside the + // same window deduplicates against the first instead of scheduling + // the work twice. `enqueue` answering `None` is that deduplication, + // not a failure — which is why the outcome carries the buffer count + // beside it, so a caller can tell "nothing to do" from "already + // scheduled". + let date_iso = now.format("%Y-%m-%d").to_string(); + let hour_block = chrono::Timelike::hour(&now) / 3; + let job = tinymemory_core::queue::types::NewJob::flush_stale( + &payload, &date_iso, hour_block, + )?; + let enqueued = tinymemory_core::queue::store::enqueue(config, &job)?.is_some(); + if enqueued { + tinymemory_core::queue::wake_workers(); + } + Ok(FlushOutcome { + enqueued, + stale_buffers, + }) + }, + ) + .await + } + + async fn reset_derived_index(&self) -> Result { + blocking(self.config.clone(), "reset derived index", move |config| { + // Everything here is derived from `mem_tree_chunks`, which is NOT + // in the list and is never deleted. That is the invariant the + // contract promises: nothing a caller wrote is lost, only what was + // computed from it. + const DERIVED_TABLES: &[&str] = &[ + "mem_tree_summaries", + "mem_tree_buffers", + "mem_tree_jobs", + "mem_tree_entity_index", + "mem_tree_trees", + ]; + let rows_deleted = + tinymemory_core::store::chunks::store::with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + let mut total: u64 = 0; + for table in DERIVED_TABLES { + total += tx.execute(&format!("DELETE FROM {table}"), [])? as u64; + } + tx.commit()?; + Ok(total) + })?; + + // Second transaction, deliberately. The delete above drops the job + // table, so re-enqueueing in the same one would race its own + // truncation. + let (chunks_requeued, jobs_enqueued) = + tinymemory_core::store::chunks::store::with_connection(config, |conn| { + let tx = conn.unchecked_transaction()?; + let chunks_requeued = tx.execute( + "UPDATE mem_tree_chunks SET lifecycle_status = 'pending_extraction'", + [], + )? as u64; + let chunk_ids: Vec = { + let mut stmt = tx.prepare("SELECT id FROM mem_tree_chunks")?; + // Bound rather than returned directly: the rows borrow + // `stmt`, which the block would otherwise drop first. + let rows = stmt + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + rows + }; + let mut jobs_enqueued: u64 = 0; + for chunk_id in &chunk_ids { + let payload = tinymemory_core::queue::types::ExtractChunkPayload { + chunk_id: chunk_id.clone(), + }; + let job = tinymemory_core::queue::types::NewJob::extract_chunk(&payload)?; + // Keyed, so a chunk already queued is a no-op rather + // than a duplicate job — which is why this count can be + // lower than `chunks_requeued`. + if tinymemory_core::queue::store::enqueue_tx(&tx, &job)?.is_some() { + jobs_enqueued += 1; + } + } + tx.commit()?; + Ok((chunks_requeued, jobs_enqueued)) + })?; + + // The work is scheduled; nothing drains it until a worker is woken. + tinymemory_core::queue::wake_workers(); + + Ok(ResetOutcome { + rows_deleted, + chunks_requeued, + jobs_enqueued, + }) + }) + .await + } + async fn backfill_in_progress(&self) -> Result { // A process-global the backfill chain owns, not a column — and not one // this engine can narrow, since `tinymemory_core::queue` tracks the diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index f3ab202f..5872f8f9 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -1197,6 +1197,99 @@ async fn ingest_chunks_and_retrieval_cover_success_and_validation_without_networ ); } +/// Flushing twice inside one window schedules the work once, and says so. +/// +/// The deduplication is the driver's, keyed on date and three-hour block, so a +/// user hitting "flush now" twice does not get the work queued twice. What +/// makes that safe to surface is the buffer count riding alongside: without +/// it, `enqueued: false` is ambiguous between "nothing to flush" and "already +/// scheduled", and a caller showing the first when it is the second is lying +/// about state the user is watching. +#[tokio::test(flavor = "multi_thread")] +async fn flushing_twice_in_a_window_schedules_the_work_once() { + use tinymemory_api::provider::{MemoryMaintenance, MemoryProvider}; + use tinymemory_api::tree::IngestRequest; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + + provider + .as_tree() + .expect("Tree") + .append(IngestRequest { + namespace: "project".into(), + content: "something buffered and waiting to be written out".into(), + timestamp: Some(chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("ts")), + metadata: None, + }) + .await + .expect("append"); + + let first = provider.flush_pending().await.expect("first flush"); + assert!( + first.enqueued, + "the first flush in a window schedules the work" + ); + + let second = provider.flush_pending().await.expect("second flush"); + assert!( + !second.enqueued, + "the second deduplicates against the first rather than queueing it twice" + ); + assert_eq!( + second.stale_buffers, first.stale_buffers, + "and still reports what is pending, so `enqueued: false` is not mistaken \ + for an empty queue" + ); +} + +/// Resetting the derived index keeps every chunk it was derived from. +/// +/// This is the invariant that makes the operation safe to expose at all. +/// Summaries, buffers, entity indexes and trees are recomputable; the chunks +/// are the source and are never deleted. A reset that took them too would be +/// data loss wearing the word "reset". +#[tokio::test(flavor = "multi_thread")] +async fn resetting_the_derived_index_keeps_the_chunks_it_derives_from() { + use tinymemory_api::provider::{MemoryMaintenance, MemoryProvider}; + use tinymemory_api::tree::IngestRequest; + + let workspace = tempfile::tempdir().expect("workspace"); + let provider = provider_over(workspace.path()); + + provider + .as_tree() + .expect("Tree") + .append(IngestRequest { + namespace: "project".into(), + content: "content the derived index is built from".into(), + timestamp: Some(chrono::DateTime::from_timestamp(1_700_000_000, 0).expect("ts")), + metadata: None, + }) + .await + .expect("append"); + + let before = provider.store_stats().await.expect("stats before"); + + let outcome = provider + .reset_derived_index() + .await + .expect("reset the derived index"); + + let after = provider.store_stats().await.expect("stats after"); + assert_eq!( + after.chunks, before.chunks, + "the reset deletes what was derived, never the source it was derived from" + ); + assert!( + outcome.jobs_enqueued <= outcome.chunks_requeued, + "the enqueue is keyed, so scheduled jobs cannot exceed requeued chunks: \ + {} jobs for {} chunks", + outcome.jobs_enqueued, + outcome.chunks_requeued + ); +} + /// Recency recall answers when there is no query to rank against. /// /// This is the member's whole reason for existing. `recall_namespace_scored` From 5c4bab807264f56ee5cf0a9fd764c9cbd0a7438d Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 24 Aug 2026 17:23:17 +0530 Subject: [PATCH 3/4] Carry chat attribution faithfully, and give extracted events a member MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two additions the archivist migration cannot ship without, batched here so one release serves the whole OpenHuman PR. `IngestItem` gains `author`, `channel_label` and `platform`, all optional and serde-defaulted. The chat mapping used to manufacture all three: every message was attributed to the batch's owner, the display label collapsed into the dedupe key, and the platform string was rewritten to the enum's name. For single-speaker sources that was merely redundant; for an agent session — owner = the session the memory belongs to, author = the speaking role — it destroys role attribution in the stored transcript, and the platform rewrite would silently change what is on disk for a caller that has always written its own value. Each field falls back to exactly the old behaviour when absent, so existing callers store byte-identical rows; conformance now sets all three so the mapping is exercised rather than trusted. `MemoryEpisodic::insert_event` records one extracted event against its segment. Events are segment-derived episodic artifacts — a summariser reads a closed segment and records the durable facts it found — and the record arrives fully formed because the extraction policy is the caller's, not the driver's. The id is an upsert key, so re-running extraction replaces its own rows rather than duplicating them; conformance pins that. `EventKind` mirrors the engine's `event_log` vocabulary on the wire. Registered in all four lists (METHODS 97 -> 98), and the member checks earned their keep again: the first run failed with "declared in the manifest but not served" because the forwarder edit had not landed. Co-Authored-By: Claude Opus 5 --- crates/tinymemory-api/src/null_tests.rs | 3 + .../tinymemory-api/src/provider/episodic.rs | 14 +++- crates/tinymemory-api/src/provider/mod.rs | 2 +- crates/tinymemory-bus/src/lib.rs | 2 +- crates/tinymemory-bus/src/names.rs | 5 +- .../tinymemory-bus/src/provider/episodic.rs | 59 ++++++++++++++++ crates/tinymemory-bus/src/provider/types.rs | 23 +++++++ crates/tinymemory-documents/src/ingest/mod.rs | 3 + crates/tinymemory-module/src/lib.rs | 1 + crates/tinymemory-module/src/service/mod.rs | 9 ++- crates/tinymemory-module/tests/module_e2e.rs | 7 ++ .../tinymemory-tinycortex/src/engine/mod.rs | 68 ++++++++++++++++--- .../tinymemory-tinycortex/src/engine/test.rs | 3 + .../tests/full_provider_conformance.rs | 42 ++++++++++++ 14 files changed, 227 insertions(+), 14 deletions(-) diff --git a/crates/tinymemory-api/src/null_tests.rs b/crates/tinymemory-api/src/null_tests.rs index 9339047d..9ae79428 100644 --- a/crates/tinymemory-api/src/null_tests.rs +++ b/crates/tinymemory-api/src/null_tests.rs @@ -251,6 +251,9 @@ fn every_optional_method_fails_with_its_advertised_family_name() { tags: Vec::new(), taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }; assert_unsupported(block_on(driver.ingest_document(ingest)), Capability::Ingest); diff --git a/crates/tinymemory-api/src/provider/episodic.rs b/crates/tinymemory-api/src/provider/episodic.rs index aaaeaab1..8b8d3cc6 100644 --- a/crates/tinymemory-api/src/provider/episodic.rs +++ b/crates/tinymemory-api/src/provider/episodic.rs @@ -48,7 +48,9 @@ use crate::error::MemoryError; // — 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 // every historical path keeps resolving and the types stay the same types. -pub use tinymemory_bus::provider::episodic::{ConversationSegment, EpisodicTurn}; +pub use tinymemory_bus::provider::episodic::{ + ConversationSegment, EpisodicEvent, EpisodicTurn, EventKind, +}; /// The turn-by-turn conversation record. /// @@ -148,6 +150,16 @@ pub trait MemoryEpisodic: Send + Sync { /// # Errors /// /// Backend failures only. + /// Record one extracted event against its segment. + /// + /// Keyed on `event_id`, so re-running extraction over the same segment + /// replaces its own rows rather than duplicating them. + /// + /// # Errors + /// + /// Backend failures only. + async fn insert_event(&self, event: &EpisodicEvent) -> Result<(), MemoryError>; + async fn upsert_segment_embedding( &self, segment_id: &str, diff --git a/crates/tinymemory-api/src/provider/mod.rs b/crates/tinymemory-api/src/provider/mod.rs index 7ca7701a..35146e06 100644 --- a/crates/tinymemory-api/src/provider/mod.rs +++ b/crates/tinymemory-api/src/provider/mod.rs @@ -77,7 +77,7 @@ pub use audit::{audit_provider, CapabilityAudit}; pub use chunks::{ChunkDetail, ChunkEmbedding, ChunkQuery, MemoryChunks}; pub use content::{MemoryDocuments, MemoryIngest, MemoryTree}; pub use driver::MemoryProvider; -pub use episodic::{ConversationSegment, EpisodicTurn, MemoryEpisodic}; +pub use episodic::{ConversationSegment, EpisodicEvent, EpisodicTurn, EventKind, MemoryEpisodic}; pub use knowledge::{MemoryDiff, MemoryEntities, MemoryGraph, INBOUND_SCAN_LIMIT}; pub use mandatory::{MemoryCore, MemoryPortability, MemoryRecall}; pub use people::{ diff --git a/crates/tinymemory-bus/src/lib.rs b/crates/tinymemory-bus/src/lib.rs index c5346e7e..7114f8a0 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 97 members on it, built as a `cdylib`. A host that +//! exports one object with 98 members on it, 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 3980f543..73e1db71 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -249,6 +249,8 @@ pub mod methods { pub const SET_SEGMENT_SUMMARY: &str = "SetSegmentSummary"; /// `UpsertSegmentEmbedding` — upsert segment embedding. pub const UPSERT_SEGMENT_EMBEDDING: &str = "UpsertSegmentEmbedding"; + /// `InsertEvent` — record one extracted event against its segment. + pub const INSERT_EVENT: &str = "InsertEvent"; } /// Every member name, in the order the module declares them. @@ -256,7 +258,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; 97] = [ +pub const METHODS: [&str; 98] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -342,6 +344,7 @@ pub const METHODS: [&str; 97] = [ methods::CLOSE_SEGMENT, methods::SET_SEGMENT_SUMMARY, methods::UPSERT_SEGMENT_EMBEDDING, + methods::INSERT_EVENT, methods::UPSERT_FACET, methods::UPSERT_PROVIDER_FACET, methods::SET_FACET_USER_STATE, diff --git a/crates/tinymemory-bus/src/provider/episodic.rs b/crates/tinymemory-bus/src/provider/episodic.rs index 5a02312a..d461a14f 100644 --- a/crates/tinymemory-bus/src/provider/episodic.rs +++ b/crates/tinymemory-bus/src/provider/episodic.rs @@ -111,3 +111,62 @@ pub struct ConversationSegment { /// Whether the segment is still open. pub open: bool, } + +/// What kind of durable fact an extracted event records. +/// +/// Mirrors the engine's `event_log.event_type` vocabulary; the wire carries +/// the same lowercase identifiers. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum EventKind { + /// A stated fact about the world or the user. + Fact, + /// A decision that was made. + Decision, + /// A commitment somebody took on. + Commitment, + /// A preference the user expressed. + Preference, + /// A question left open. + Question, + /// Something anticipated to happen. + Foresight, +} + +/// One extracted event, ready to be recorded against its segment. +/// +/// Events are segment-derived episodic artifacts — a summariser or heuristic +/// reads a closed segment and records the durable facts it found. The driver +/// owns the table; the caller owns the extraction policy, which is why the +/// record arrives fully formed rather than as text to extract from. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct EpisodicEvent { + /// Caller-assigned id; an upsert key, so a re-run replaces its own rows. + pub event_id: String, + /// Segment the event was extracted from. + pub segment_id: String, + /// Session that segment belongs to. + pub session_id: String, + /// Namespace the event is scoped to. + pub namespace: String, + /// What kind of fact this is. + pub kind: EventKind, + /// The event, in prose. + pub content: String, + /// Who or what the event is about, when extraction identified one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub subject: Option, + /// A time the prose refers to, verbatim, when extraction found one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub timestamp_ref: Option, + /// Extraction confidence in `[0, 1]`. + pub confidence: f64, + /// Embedding for the content, when the caller computed one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub embedding: Option>, + /// Turn ids the event was derived from, encoded by the caller. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source_turn_ids: Option, + /// When the event was recorded, seconds since the epoch. + pub created_at: f64, +} diff --git a/crates/tinymemory-bus/src/provider/types.rs b/crates/tinymemory-bus/src/provider/types.rs index d98bcb03..a71baee3 100644 --- a/crates/tinymemory-bus/src/provider/types.rs +++ b/crates/tinymemory-bus/src/provider/types.rs @@ -137,6 +137,29 @@ pub struct IngestItem { /// Labels carried through from the source. Ingest does not interpret them. #[serde(default)] pub tags: Vec, + /// Who spoke this item, when that is not [`Self::owner`]. + /// + /// A chat batch from an agent session has owner = the session the memory + /// belongs to and author = the speaking role (`user`, `assistant`). The + /// previous mapping collapsed the two — every message attributed to the + /// owner — which destroys role attribution in the stored transcript. + /// Absent means "the owner spoke", which is true of the single-speaker + /// sources this field predates. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub author: Option, + /// Display label for the conversation, when it is not [`Self::source_id`]. + /// + /// `source_id` is the dedupe key and may be a constant ("all agent + /// sessions share one tree source"); the label is what a human reads in a + /// summary. Absent means the id is readable enough to double as the label. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub channel_label: Option, + /// Platform string to store verbatim, when [`DataSource::as_str`] is not + /// it. Migrating a caller that has always written a bespoke platform value + /// must not silently rewrite what is on disk; absent keeps the enum's + /// name, which is right for every new caller. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub platform: Option, /// Provenance taint. The **host** stamps this; a driver must persist what it /// is given and must never assign or upgrade it. #[serde(default)] diff --git a/crates/tinymemory-documents/src/ingest/mod.rs b/crates/tinymemory-documents/src/ingest/mod.rs index c7dbe4cd..2d511ca4 100644 --- a/crates/tinymemory-documents/src/ingest/mod.rs +++ b/crates/tinymemory-documents/src/ingest/mod.rs @@ -133,6 +133,9 @@ impl<'a> DocumentIntake<'a> { tags: request.tags.clone(), taint: request.taint, path_scope: None, + author: None, + channel_label: None, + platform: None, }; let outcome = ingest.ingest_document(item).await?; Ok(IntakeReceipt { diff --git a/crates/tinymemory-module/src/lib.rs b/crates/tinymemory-module/src/lib.rs index e05a2f25..6ff563b6 100644 --- a/crates/tinymemory-module/src/lib.rs +++ b/crates/tinymemory-module/src/lib.rs @@ -225,6 +225,7 @@ mod exports { "CloseSegment", "SetSegmentSummary", "UpsertSegmentEmbedding", + "InsertEvent", "Store", "Get", "Forget", diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index f643c885..48057eae 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -128,7 +128,7 @@ 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, ChunkQuery}; -use tinymemory_api::provider::episodic::{ConversationSegment, EpisodicTurn}; +use tinymemory_api::provider::episodic::{ConversationSegment, EpisodicEvent, EpisodicTurn}; use tinymemory_api::provider::people::{ AddressBookSeedOutcome, PersonHandle, PersonInteraction, PersonRecord, PersonScore, RankedPerson, ResolvedPerson, @@ -1254,6 +1254,13 @@ impl MemoryService { .map_err(|error| into_bus_error(&error)) } + async fn insert_event(&self, event: EpisodicEvent) -> BusResult<()> { + require_family!(self, as_episodic, Capability::Episodic) + .insert_event(&event) + .await + .map_err(|error| into_bus_error(&error)) + } + async fn upsert_facet(&self, facet: ProfileFacet) -> BusResult<()> { require_family!(self, as_profile, Capability::Profile) .upsert_facet(&facet) diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index e7d9a1ed..daa51858 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -591,6 +591,7 @@ const EXPECTED_METHODS: &[&str] = &[ "CloseSegment", "SetSegmentSummary", "UpsertSegmentEmbedding", + "InsertEvent", "Store", "Get", "Forget", @@ -1167,6 +1168,9 @@ async fn ingest_and_chunks_round_trip(bus: &tinybus::Proxy) -> String { tags: vec!["coverage".into()], taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }; let outcome: IngestOutcome = bus .call("IngestDocument", (ingest,)) @@ -1385,6 +1389,9 @@ async fn maintenance_and_diff_round_trip(bus: &tinybus::Proxy) { tags: vec!["coverage".into()], taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }; let _: IngestOutcome = bus .call("IngestDocument", (changed,)) diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index 52f89490..ea9340ab 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -42,12 +42,13 @@ use tinymemory_api::provider::types::{ use tinymemory_api::provider::types::{ChangeKind, DiffReport, SnapshotRef, SourceChange}; use tinymemory_api::provider::{ AddressBookSeedOutcome, ChunkDetail, ChunkEmbedding, ChunkQuery, ConversationSegment, - CoverWindowQuery, EntityMatch, EpisodicTurn, FacetType, FastRetrieveQuery, MemoryChunks, - MemoryCore, MemoryDiff, MemoryDocuments, MemoryEntities, MemoryEpisodic, MemoryGoals, - MemoryGraph, MemoryIngest, MemoryMaintenance, MemoryPeople, MemoryPortability, MemoryProfile, - MemoryProvider, MemoryRecall, MemoryRetrieval, MemorySourceSink, MemoryToolMemory, MemoryTree, - PersonHandle, PersonInteraction, PersonRecord, PersonScore, ProfileFacet, RankedPerson, - ResolvedPerson, RetrievalHit, RetrievalResponse, SourceRetrievalQuery, UserState, + CoverWindowQuery, EntityMatch, EpisodicEvent, EpisodicTurn, EventKind, FacetType, + FastRetrieveQuery, MemoryChunks, MemoryCore, MemoryDiff, MemoryDocuments, MemoryEntities, + MemoryEpisodic, MemoryGoals, MemoryGraph, MemoryIngest, MemoryMaintenance, MemoryPeople, + MemoryPortability, MemoryProfile, MemoryProvider, MemoryRecall, MemoryRetrieval, + MemorySourceSink, MemoryToolMemory, MemoryTree, PersonHandle, PersonInteraction, PersonRecord, + PersonScore, ProfileFacet, RankedPerson, ResolvedPerson, RetrievalHit, RetrievalResponse, + SourceRetrievalQuery, UserState, }; use tinymemory_api::recall::OwnedRecallOpts; use tinymemory_api::tool_memory::ToolMemoryRule; @@ -490,7 +491,16 @@ impl MemoryIngest for TinycortexProvider { let source_id = first.source_id.clone(); let owner = first.owner.clone(); let tags = first.tags.clone(); - let platform = first.source.as_str().to_string(); + // The three widened fields default to what this mapping always did, so + // a caller that never sets them stores byte-identical rows. + let platform = first + .platform + .clone() + .unwrap_or_else(|| first.source.as_str().to_string()); + let channel_label = first + .channel_label + .clone() + .unwrap_or_else(|| source_id.clone()); for item in &messages { validate_ingest_item(item)?; if item.source_id != source_id { @@ -501,12 +511,16 @@ impl MemoryIngest for TinycortexProvider { } let batch = tinycortex::memory::ingest::canonicalize::chat::ChatBatch { platform, - channel_label: source_id.clone(), + channel_label, messages: messages .into_iter() .map( |item| tinycortex::memory::ingest::canonicalize::chat::ChatMessage { - author: item.owner, + // The speaking role when the caller distinguishes it; + // the owner otherwise. Attributing every message to + // the owner is what destroyed role attribution for + // multi-speaker batches. + author: item.author.unwrap_or(item.owner), timestamp: item.timestamp.unwrap_or_else(Utc::now), text: item.content, source_ref: item.source_ref.map(|source_ref| source_ref.value), @@ -2211,6 +2225,18 @@ impl MemoryRetrieval for TinycortexProvider { // handle over the client's connection, not an open — so it is fetched inside // the blocking closure rather than held across an await. +fn event_kind_to_engine(kind: EventKind) -> tinymemory_core::store::events::EventType { + use tinymemory_core::store::events::EventType as Engine; + match kind { + EventKind::Fact => Engine::Fact, + EventKind::Decision => Engine::Decision, + EventKind::Commitment => Engine::Commitment, + EventKind::Preference => Engine::Preference, + EventKind::Question => Engine::Question, + EventKind::Foresight => Engine::Foresight, + } +} + fn facet_type_to_engine( facet_type: FacetType, ) -> tinymemory_core::store::namespace_store::profile::FacetType { @@ -2529,6 +2555,30 @@ impl MemoryEpisodic for TinycortexProvider { .map_err(|e| Self::other("set_segment_summary", e)) } + async fn insert_event(&self, event: &EpisodicEvent) -> Result<(), MemoryError> { + let conn = self.client.profile_conn(); + let record = tinymemory_core::store::events::EventRecord { + event_id: event.event_id.clone(), + segment_id: event.segment_id.clone(), + session_id: event.session_id.clone(), + namespace: event.namespace.clone(), + event_type: event_kind_to_engine(event.kind), + content: event.content.clone(), + subject: event.subject.clone(), + timestamp_ref: event.timestamp_ref.clone(), + confidence: event.confidence, + embedding: event.embedding.clone(), + source_turn_ids: event.source_turn_ids.clone(), + created_at: event.created_at, + }; + tokio::task::spawn_blocking(move || { + tinymemory_core::store::events::event_insert(&conn, &record) + }) + .await + .map_err(|error| Self::other("insert event", error))? + .map_err(|error| Self::other("insert event", error)) + } + async fn upsert_segment_embedding( &self, segment_id: &str, diff --git a/crates/tinymemory-tinycortex/src/engine/test.rs b/crates/tinymemory-tinycortex/src/engine/test.rs index c10c0b09..7d27a3c8 100644 --- a/crates/tinymemory-tinycortex/src/engine/test.rs +++ b/crates/tinymemory-tinycortex/src/engine/test.rs @@ -40,6 +40,9 @@ fn ingest_item(content: &str, mime: Option<&str>, taint: MemoryTaint) -> IngestI tags: Vec::new(), taint, path_scope: None, + author: None, + channel_label: None, + platform: None, } } diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 5872f8f9..19de752b 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -180,6 +180,9 @@ async fn maintenance_diagnostics_read_the_store_rather_than_their_defaults() { tags: Vec::new(), taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }) .await .expect("ingest a document"); @@ -993,6 +996,32 @@ async fn people_profile_and_episodic_lifecycles_are_real_and_typed() { .upsert_segment_embedding("seg-1", "noop:8", &[0.0; 8], 15.0) .await .expect("upsert segment embedding"); + // An extracted event lands against its segment through the contract, and + // the id is an upsert key: re-recording under the same id replaces the row + // rather than duplicating it, so a re-run of extraction is idempotent. + use tinymemory_api::provider::{EpisodicEvent, EventKind}; + let event = EpisodicEvent { + event_id: "evt-1".into(), + segment_id: "seg-1".into(), + session_id: "session-1".into(), + namespace: "global".into(), + kind: EventKind::Decision, + content: "the test decided to remember".into(), + subject: Some("the test".into()), + timestamp_ref: None, + confidence: 0.9, + embedding: None, + source_turn_ids: Some(turn_id.to_string()), + created_at: 16.0, + }; + episodic.insert_event(&event).await.expect("insert event"); + episodic + .insert_event(&EpisodicEvent { + content: "the test revised its decision".into(), + ..event + }) + .await + .expect("re-insert under the same id"); assert!(episodic .open_segment("session-1") .await @@ -1023,6 +1052,9 @@ async fn ingest_chunks_and_retrieval_cover_success_and_validation_without_networ tags: Vec::new(), taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }; assert!(matches!( ingest.ingest_document(invalid).await, @@ -1050,6 +1082,9 @@ async fn ingest_chunks_and_retrieval_cover_success_and_validation_without_networ tags: vec!["coverage".into()], taint: MemoryTaint::Internal, path_scope: None, + author: None, + channel_label: None, + platform: None, }) .await .expect("successful deterministic ingest"); @@ -1070,6 +1105,13 @@ async fn ingest_chunks_and_retrieval_cover_success_and_validation_without_networ tags: vec!["chat".into()], taint: MemoryTaint::Internal, path_scope: None, + // The widened trio: the speaking role is not the owner, the label + // is not the dedupe key, and the platform string is the caller's + // own. Set here so the mapping that used to collapse all three is + // exercised by conformance rather than trusted. + author: Some("assistant".into()), + channel_label: Some("Agent session #1".into()), + platform: Some("agent".into()), }]) .await .expect("successful chat ingest"); From 8282679f5801406f69257d0441843203336ebd40 Mon Sep 17 00:00:00 2001 From: Shanu Date: Mon, 24 Aug 2026 17:30:13 +0530 Subject: [PATCH 4/4] Carry the per-session sequence through the segment members MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The archivist migration surfaced a linkage the contract silently dropped: the engine's `segment_create` and `segment_append_turn` take a per-session `seq`, and the driver was pinning it to `None` with a comment declaring it "not part of this contract". It has to be. The md-backed archivist store rounds timestamps to milliseconds, which can move a fast turn just before its segment's higher-precision start time — the sequence is the identity that survives the rounding, and segment selection prefers it. A host migrated onto members that cannot carry it would silently degrade every segment filter to the timestamp fallback. `create_segment` gains `start_seq: Option`, `append_turn` gains `seq: Option`, and `ConversationSegment` carries the pair back out (serde-defaulted, so older payloads still decode). Conformance pins the round trip: create with `Some(1)`, append with `Some(2)`, read both back off `open_segment`. Changing two members' arity is safe here for the same reason it usually is not: no released artifact will ever face a host built against the widened signatures — hosts pin exact digests and re-pin in lockstep with the release that carries this. Co-Authored-By: Claude Opus 5 --- .../tinymemory-api/src/provider/episodic.rs | 6 +++++ .../tinymemory-bus/src/provider/episodic.rs | 10 +++++++ crates/tinymemory-module/src/service/mod.rs | 5 +++- crates/tinymemory-module/tests/module_e2e.rs | 19 ++++++++++--- .../tinymemory-tinycortex/src/engine/mod.rs | 27 +++++++++++++------ .../tests/full_provider_conformance.rs | 10 +++++-- 6 files changed, 62 insertions(+), 15 deletions(-) diff --git a/crates/tinymemory-api/src/provider/episodic.rs b/crates/tinymemory-api/src/provider/episodic.rs index 8b8d3cc6..6eef33a6 100644 --- a/crates/tinymemory-api/src/provider/episodic.rs +++ b/crates/tinymemory-api/src/provider/episodic.rs @@ -91,12 +91,17 @@ pub trait MemoryEpisodic: Send + Sync { /// # Errors /// /// Backend failures only. + #[allow( + clippy::too_many_arguments, + reason = "mirrors the engine row it creates; a params struct would be its only caller's" + )] async fn create_segment( &self, segment_id: &str, session_id: &str, namespace: &str, start_episodic_id: i64, + start_seq: Option, start_timestamp: f64, now: f64, ) -> Result<(), MemoryError>; @@ -110,6 +115,7 @@ pub trait MemoryEpisodic: Send + Sync { &self, segment_id: &str, episodic_id: i64, + seq: Option, timestamp: f64, now: f64, ) -> Result<(), MemoryError>; diff --git a/crates/tinymemory-bus/src/provider/episodic.rs b/crates/tinymemory-bus/src/provider/episodic.rs index d461a14f..6f3f359e 100644 --- a/crates/tinymemory-bus/src/provider/episodic.rs +++ b/crates/tinymemory-bus/src/provider/episodic.rs @@ -110,6 +110,16 @@ pub struct ConversationSegment { pub embedding: Option>, /// Whether the segment is still open. pub open: bool, + /// Stable per-session sequence of the first user turn, when the backing + /// store assigns one. The md-backed archivist store rounds timestamps to + /// milliseconds, so a fast turn can sort before its segment's + /// higher-precision start time — the sequence is the identity that + /// survives that, and segment selection prefers it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub start_seq: Option, + /// Sequence of the last appended user turn, likewise. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub end_seq: Option, } /// What kind of durable fact an extracted event records. diff --git a/crates/tinymemory-module/src/service/mod.rs b/crates/tinymemory-module/src/service/mod.rs index 48057eae..4c6da20e 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -1189,6 +1189,7 @@ impl MemoryService { session_id: String, namespace: String, start_episodic_id: i64, + start_seq: Option, start_timestamp: f64, now: f64, ) -> BusResult<()> { @@ -1198,6 +1199,7 @@ impl MemoryService { &session_id, &namespace, start_episodic_id, + start_seq, start_timestamp, now, ) @@ -1210,11 +1212,12 @@ impl MemoryService { &self, segment_id: String, episodic_id: i64, + seq: Option, timestamp: f64, now: f64, ) -> BusResult<()> { require_family!(self, as_episodic, Capability::Episodic) - .append_turn(&segment_id, episodic_id, timestamp, now) + .append_turn(&segment_id, episodic_id, seq, timestamp, now) .await .map_err(|error| into_bus_error(&error)) } diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index daa51858..0f858604 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -1112,13 +1112,24 @@ async fn episodic_round_trip(bus: &tinybus::Proxy) { .expect("SessionTurns"); bus.call::<()>( "CreateSegment", - ("seg-1", "session-1", "global", turn_id, 10.0_f64, 10.0_f64), + ( + "seg-1", + "session-1", + "global", + turn_id, + Option::::None, + 10.0_f64, + 10.0_f64, + ), ) .await .expect("CreateSegment"); - bus.call::<()>("AppendTurn", ("seg-1", turn_id, 10.0_f64, 11.0_f64)) - .await - .expect("AppendTurn"); + bus.call::<()>( + "AppendTurn", + ("seg-1", turn_id, Option::::None, 10.0_f64, 11.0_f64), + ) + .await + .expect("AppendTurn"); let _: Option = bus .call("OpenSegment", ("session-1",)) .await diff --git a/crates/tinymemory-tinycortex/src/engine/mod.rs b/crates/tinymemory-tinycortex/src/engine/mod.rs index ea9340ab..1887388d 100644 --- a/crates/tinymemory-tinycortex/src/engine/mod.rs +++ b/crates/tinymemory-tinycortex/src/engine/mod.rs @@ -2470,12 +2470,17 @@ impl MemoryEpisodic for TinycortexProvider { Ok(segment.map(segment_to_contract)) } + #[allow( + clippy::too_many_arguments, + reason = "trait signature; see the contract's rationale" + )] async fn create_segment( &self, segment_id: &str, session_id: &str, namespace: &str, start_episodic_id: i64, + start_seq: Option, start_timestamp: f64, now: f64, ) -> Result<(), MemoryError> { @@ -2492,9 +2497,7 @@ impl MemoryEpisodic for TinycortexProvider { &session_id, &namespace, start_episodic_id, - // Per-session seq numbering is the archivist store's, and it is - // not part of this contract; legacy rows carry `None` too. - None, + start_seq, start_timestamp, now, ) @@ -2508,6 +2511,7 @@ impl MemoryEpisodic for TinycortexProvider { &self, segment_id: &str, episodic_id: i64, + seq: Option, timestamp: f64, now: f64, ) -> Result<(), MemoryError> { @@ -2518,7 +2522,7 @@ impl MemoryEpisodic for TinycortexProvider { &conn, &segment_id, episodic_id, - None, + seq, timestamp, now, ) @@ -2620,10 +2624,15 @@ fn episodic_to_contract(entry: tinymemory_core::store::fts5::EpisodicEntry) -> E /// Engine segment row -> contract segment. /// -/// Written out rather than derived: the engine row carries several fields the -/// contract deliberately does not expose (`topic_keywords`, the seq numbers, -/// `created_at`), and a blanket conversion would quietly start shipping them if -/// the contract ever grew a matching name. +/// Written out rather than derived: the engine row carries fields the contract +/// deliberately does not expose (`topic_keywords`, `created_at`), and a +/// blanket conversion would quietly start shipping them if the contract ever +/// grew a matching name. +/// +/// The seq pair used to be on that withheld list; it is contract vocabulary +/// now, because segment selection prefers it — the md-backed archivist store +/// rounds timestamps to milliseconds, and the sequence is the identity that +/// survives the rounding. fn segment_to_contract( segment: tinymemory_core::store::segments::ConversationSegment, ) -> ConversationSegment { @@ -2640,6 +2649,8 @@ fn segment_to_contract( summary: segment.summary, embedding: segment.embedding, open: matches!(segment.status, SegmentStatus::Open), + start_seq: segment.start_seq, + end_seq: segment.end_seq, } } diff --git a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs index 19de752b..cb8641f5 100644 --- a/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs +++ b/crates/tinymemory-tinycortex/tests/full_provider_conformance.rs @@ -971,11 +971,11 @@ async fn people_profile_and_episodic_lifecycles_are_real_and_typed() { assert_eq!(turns[0].id, Some(turn_id)); assert_eq!(turns[0].cost_microdollars, 0, "negative costs clamp"); episodic - .create_segment("seg-1", "session-1", "global", turn_id, 10.0, 10.0) + .create_segment("seg-1", "session-1", "global", turn_id, Some(1), 10.0, 10.0) .await .expect("create segment"); episodic - .append_turn("seg-1", turn_id, 10.0, 11.0) + .append_turn("seg-1", turn_id, Some(2), 10.0, 11.0) .await .expect("append turn"); let segment = episodic @@ -984,6 +984,12 @@ async fn people_profile_and_episodic_lifecycles_are_real_and_typed() { .expect("open segment") .expect("segment present"); assert_eq!(segment.turn_count, 2); + assert_eq!( + (segment.start_seq, segment.end_seq), + (Some(1), Some(2)), + "the per-session sequence pair survives the contract round trip — \ + segment selection prefers it over ms-rounded timestamps" + ); episodic .close_segment("seg-1", 13.0) .await