diff --git a/crates/tinymemory-api/src/null.rs b/crates/tinymemory-api/src/null.rs index e94a11b..446ca6c 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] @@ -612,6 +625,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/null_tests.rs b/crates/tinymemory-api/src/null_tests.rs index 9339047..9ae7942 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 aaaeaab..6eef33a 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. /// @@ -89,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>; @@ -108,6 +115,7 @@ pub trait MemoryEpisodic: Send + Sync { &self, segment_id: &str, episodic_id: i64, + seq: Option, timestamp: f64, now: f64, ) -> Result<(), MemoryError>; @@ -148,6 +156,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 32d2c86..35146e0 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::{ @@ -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 0e3038f..952d3f7 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-api/src/provider/retrieval.rs b/crates/tinymemory-api/src/provider/retrieval.rs index 666396d..a87586a 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 cc00d88..7114f8a 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 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 a9f7161..73e1db7 100644 --- a/crates/tinymemory-bus/src/names.rs +++ b/crates/tinymemory-bus/src/names.rs @@ -161,6 +161,12 @@ 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"; + /// `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. @@ -243,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. @@ -250,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; 94] = [ +pub const METHODS: [&str; 98] = [ methods::DRIVER_ID, methods::CAPABILITIES, methods::HEALTH, @@ -307,6 +315,9 @@ pub const METHODS: [&str; 94] = [ 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, methods::RESOLVE_HANDLE, @@ -333,6 +344,7 @@ pub const METHODS: [&str; 94] = [ 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 5a02312..6f3f359 100644 --- a/crates/tinymemory-bus/src/provider/episodic.rs +++ b/crates/tinymemory-bus/src/provider/episodic.rs @@ -110,4 +110,73 @@ 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. +/// +/// 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 ab8a6b4..a71baee 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)] @@ -489,6 +512,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-documents/src/ingest/mod.rs b/crates/tinymemory-documents/src/ingest/mod.rs index c7dbe4c..2d511ca 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 5812ff2..6ff563b 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", @@ -312,6 +313,9 @@ mod exports { "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "FlushPending", + "ResetDerivedIndex", + "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 e0c75f8..4c6da20 100644 --- a/crates/tinymemory-module/src/service/mod.rs +++ b/crates/tinymemory-module/src/service/mod.rs @@ -120,14 +120,15 @@ 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 // 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, @@ -922,6 +923,33 @@ 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, + 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. @@ -1161,6 +1189,7 @@ impl MemoryService { session_id: String, namespace: String, start_episodic_id: i64, + start_seq: Option, start_timestamp: f64, now: f64, ) -> BusResult<()> { @@ -1170,6 +1199,7 @@ impl MemoryService { &session_id, &namespace, start_episodic_id, + start_seq, start_timestamp, now, ) @@ -1182,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)) } @@ -1226,6 +1257,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/src/service/test.rs b/crates/tinymemory-module/src/service/test.rs index 45bc41d..02bfb2b 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 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 9e751c5..0f85860 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", @@ -676,6 +677,9 @@ const EXPECTED_METHODS: &[&str] = &[ "QueueStats", "LatestQueueFailure", "BackfillInProgress", + "FlushPending", + "ResetDerivedIndex", + "RecallNamespaceRecent", ]; #[tokio::test] @@ -1108,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 @@ -1164,6 +1179,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,)) @@ -1382,6 +1400,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 b460709..1887388 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. @@ -41,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; @@ -489,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 { @@ -500,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), @@ -1348,6 +1363,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 @@ -2039,6 +2164,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, @@ -2081,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 { @@ -2314,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> { @@ -2336,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, ) @@ -2352,6 +2511,7 @@ impl MemoryEpisodic for TinycortexProvider { &self, segment_id: &str, episodic_id: i64, + seq: Option, timestamp: f64, now: f64, ) -> Result<(), MemoryError> { @@ -2362,7 +2522,7 @@ impl MemoryEpisodic for TinycortexProvider { &conn, &segment_id, episodic_id, - None, + seq, timestamp, now, ) @@ -2399,6 +2559,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, @@ -2440,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 { @@ -2460,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/src/engine/test.rs b/crates/tinymemory-tinycortex/src/engine/test.rs index c10c0b0..7d27a3c 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 67fbb88..cb8641f 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"); @@ -968,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 @@ -981,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 @@ -993,6 +1002,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 +1058,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 +1088,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 +1111,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"); @@ -1197,6 +1245,158 @@ 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` +/// 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;