velesdb_memory/service.rs
1//! The memory service: five operations over the in-core Agent Memory SDK.
2
3use std::collections::{HashMap, HashSet};
4#[cfg(feature = "persistence")]
5use std::path::Path;
6
7use serde_json::{Map, Value};
8
9/// Structured metadata attached to a memory (the `ColumnStore` facet): exact-match
10/// fields like `project`, `author`, `type`, `status`, `date`. `content` and
11/// `_veles_expires_at` are reserved keys. [`crate::storage::AUTO_DATE_FIELD`]
12/// (`_veles_date`) is auto-populated by [`MemoryService::remember_with_ttl`]
13/// with today's date unless already present — see that method's docs.
14pub type Metadata = Map<String, Value>;
15
16use crate::clock;
17use crate::embedder::Embedder;
18use crate::error::MemoryError;
19use crate::extract::{ExtractedAttribute, ExtractedRelation, Extractor};
20use crate::id;
21use crate::model::{
22 ColumnFilter, EntityProfile, EntityRelation, Explanation, Link, MemoryEdge, MemoryNode,
23 Recollection, RememberedExtraction, UnrelateOutcome,
24};
25#[cfg(feature = "persistence")]
26use crate::storage::NativeStore;
27use crate::storage::{is_reserved_key, strip_reserved_keys, MemoryStore, AUTO_DATE_FIELD};
28
29/// [`MemoryService::recall_fused`] and its helpers — split out to keep this
30/// file under the crate's 500-NLOC-per-file budget, same pattern as
31/// `velesdb-core`'s `database/*.rs` split. A child module of `service`, so it
32/// shares full access to `MemoryService`'s private fields and methods.
33#[path = "fused_recall.rs"]
34mod fused_recall;
35
36/// [`MemoryService::feedback`] and the recall re-ranking it drives (RL Memory).
37/// A child module of `service`, like [`fused_recall`], so it uses
38/// `MemoryService`'s private `store` directly. Gated on `persistence`: it
39/// builds on `velesdb-core`'s agent SDK (`ReinforcementStrategy`), itself
40/// behind that feature, and a durable learned confidence is meaningless on the
41/// in-memory (WASM) backend.
42#[cfg(feature = "persistence")]
43#[path = "reinforce.rs"]
44mod reinforce;
45
46/// The context compiler's memory bridge (`compile_context`,
47/// `retrieve_context_source`, `context_savings`, working contexts). A child
48/// module of `service`, like [`fused_recall`], so it reuses the private
49/// `store_fact`/`HUB_FIELD` system-fact machinery — compiler system facts
50/// (sources, events, working contexts) are hub-marked so they never surface
51/// in normal recall.
52#[cfg(feature = "context")]
53#[path = "context/memory_bridge.rs"]
54mod memory_bridge;
55
56/// Reserved metadata key marking an entity hub auto-created by
57/// [`MemoryService::remember_extracted`] (value `true`). Namespaced under the
58/// system `_veles_` prefix so it can never collide with a caller's own metadata,
59/// and rejected from caller-supplied metadata/filters (see [`is_reserved_key`]).
60/// Hubs are internal graph scaffolding — they connect facts that share a topic —
61/// so they are excluded from unfiltered recall and from `why` seeds.
62///
63/// Re-exported from [`crate::storage`] rather than spelled out again here: it
64/// is one of the five markers [`crate::storage::INTERNAL_MARKER_FIELDS`]
65/// excludes from `recall_where`, and a second literal could drift from that
66/// list without any test noticing.
67use crate::storage::HUB_FIELD;
68/// Salt mixed into a hub's stable id so the hub id space is disjoint from
69/// natural fact ids: a caller fact whose text happens to equal a hub's display
70/// content (`Entity: rust`) can never collide with, or overwrite, the hub.
71const HUB_ID_SALT: &str = "\u{0}_veles_entity_hub\u{0}";
72/// Edge label a hub uses to point back at a fact it tags (the hub → fact
73/// direction). [`fused_recall`] reads this to recognise which edges in a
74/// `why()` walk crossed a hub, so it can weight the reached fact by that
75/// hub's specificity instead of a flat constant.
76const MENTIONS_RELATION: &str = "mentions";
77/// Edge label a fact uses to point at a hub it is tagged with (the fact → hub
78/// direction) — [`MENTIONS_RELATION`]'s bipartite twin, written by
79/// [`MemoryService::remember_extracted`]'s `wire_entity`.
80const ABOUT_RELATION: &str = "about";
81
82/// Local-first agent memory backed by a single `VelesDB` instance.
83///
84/// Generic over the [`Embedder`] so production can use an on-device model while
85/// tests use a deterministic, network-free one, and over the [`MemoryStore`]
86/// backend `S` so the same orchestration runs over the native, file-backed
87/// engine (the default — nothing changes for existing callers) or any other
88/// backend that implements the trait (e.g. an in-memory one for WASM).
89///
90/// Two definitions, `persistence`-gated: the default type parameter itself
91/// references [`NativeStore`], which doesn't exist as a type at all without
92/// the feature, so a `persistence`-free build (e.g. `velesdb-wasm`) drops the
93/// default and every caller names its own [`MemoryStore`] backend explicitly.
94#[cfg(feature = "persistence")]
95pub struct MemoryService<E: Embedder, S: MemoryStore = NativeStore> {
96 store: S,
97 embedder: E,
98 autograph: Option<crate::extract::DynExtractor>,
99 autograph_queue: AutographQueue,
100}
101#[cfg(not(feature = "persistence"))]
102pub struct MemoryService<E: Embedder, S: MemoryStore> {
103 store: S,
104 embedder: E,
105 autograph: Option<crate::extract::DynExtractor>,
106 autograph_queue: AutographQueue,
107}
108
109/// One deferred autograph: the stored fact a background worker will read for
110/// entities, edges and attributes (#1846).
111// The fields are read only on the worker path, which `spawn_autograph_worker`
112// cfg-gates off wasm32 (no threads there) — without this the wasm check dies
113// on dead_code under -D warnings.
114#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
115struct AutographJob {
116 fact_id: u64,
117 fact: String,
118}
119
120/// The decoupling state of [`MemoryService::autograph`] (#1846).
121///
122/// Empty by default: every construction path starts with autograph running
123/// INLINE, exactly as before — the WASM binding has no threads, library
124/// consumers keep the synchronous contract, and every existing test stays
125/// meaningful. [`MemoryService::spawn_autograph_worker`] fills `tx`, after
126/// which `remember` only ENQUEUES: the caller stops paying the generation on
127/// its response path — 46 s measured for a 12-word fact on the production
128/// daemon, versus a 0.12 s embedding — and the worker wires the graph behind.
129///
130/// `dropped` counts enrichments refused by a FULL queue or skipped by a
131/// closing worker. Non-negotiably visible: a burst that outruns the
132/// extractor loses graph structure, and a loss nobody can see is the exact
133/// defect class #1820 closed for responses.
134///
135/// `closing` is the shutdown latch: the handle's drop raises it BEFORE
136/// removing the sender, so the worker finishes the job in flight and SKIPS
137/// what is still queued (counted, one aggregated warning) instead of
138/// draining a queue of generations — 64 × a 46 s model would hold the
139/// daemon's exit for tens of minutes. Re-armed by each spawn.
140#[derive(Default)]
141// `closing` is read only by the worker/drop path, absent on wasm32 — same
142// rationale as `AutographJob` above.
143#[cfg_attr(target_arch = "wasm32", allow(dead_code))]
144struct AutographQueue {
145 tx: parking_lot::Mutex<Option<std::sync::mpsc::SyncSender<AutographJob>>>,
146 dropped: std::sync::atomic::AtomicU64,
147 closing: std::sync::atomic::AtomicBool,
148}
149
150/// Join guard for the background autograph worker.
151///
152/// Dropping it raises the closing latch, takes the sender out of the
153/// service, and JOINS the worker: the job in flight completes, the
154/// still-queued ones are SKIPPED — counted in the drop counter, one
155/// aggregated warning — and only then does the drop return. Tests get
156/// determinism; the daemon's shutdown waits for at most ONE generation,
157/// never a queue of them.
158pub struct AutographWorkerHandle {
159 close_queue: Option<Box<dyn FnOnce() + Send + Sync>>,
160 join: Option<std::thread::JoinHandle<()>>,
161}
162
163impl Drop for AutographWorkerHandle {
164 fn drop(&mut self) {
165 if let Some(close) = self.close_queue.take() {
166 close();
167 }
168 if let Some(join) = self.join.take() {
169 let _ = join.join();
170 }
171 }
172}
173
174#[cfg(feature = "persistence")]
175impl<E: Embedder> MemoryService<E, NativeStore> {
176 /// Open (or create) a native, file-backed memory store at `path`, using
177 /// `embedder` for text vectorization. The store never leaves this directory.
178 ///
179 /// # Errors
180 /// Returns [`MemoryError`] if the store cannot be opened or the agent
181 /// memory cannot be initialized for the embedder's dimension.
182 pub fn open<P: AsRef<Path>>(path: P, embedder: E) -> Result<Self, MemoryError> {
183 let store = NativeStore::open(path, embedder.dimension())?;
184 Ok(Self {
185 store,
186 embedder,
187 autograph: None,
188 autograph_queue: AutographQueue::default(),
189 })
190 }
191}
192
193impl<E: Embedder, S: MemoryStore> MemoryService<E, S> {
194 /// Build a service directly over a `store` backend, bypassing
195 /// [`Self::open`]'s filesystem-specific setup — the constructor a
196 /// non-native backend (e.g. `velesdb-wasm`'s in-memory store) uses.
197 pub fn with_store(store: S, embedder: E) -> Self {
198 Self {
199 store,
200 embedder,
201 autograph: None,
202 autograph_queue: AutographQueue::default(),
203 }
204 }
205
206 /// Turn on **autograph**: every [`Self::remember`] additionally reads the
207 /// stored fact for entities, entity→entity edges and entity attributes,
208 /// and wires them — so the knowledge graph builds itself from ordinary
209 /// `remember` calls, with no separate [`Self::remember_extracted`].
210 ///
211 /// Opt-in, and off unless this is called. It runs in one of two modes:
212 /// **inline** by default — the enrichment costs one generation per
213 /// `remember`, on the caller's write path, which is a real latency and
214 /// availability change: a memory write that silently depends on a local
215 /// model being up is not a default anyone should inherit — or
216 /// **decoupled** when [`Self::spawn_autograph_worker`] is active, where
217 /// `remember` returns as soon as the fact is durably stored and the
218 /// derived edges lag by one generation (an `entity`/`why` read issued
219 /// immediately after may not see them yet; the fact itself is always
220 /// immediately readable).
221 ///
222 /// The caller's fact is stored **verbatim and first**. Autograph only
223 /// *adds* structure around it; it never rewrites or replaces what the
224 /// caller asked to remember.
225 #[must_use]
226 pub fn with_autograph(mut self, extractor: crate::extract::DynExtractor) -> Self {
227 self.autograph = Some(extractor);
228 self
229 }
230
231 /// Remember a `fact`, optionally tagging it with structured `metadata`
232 /// (`ColumnStore` facet) and linking it to existing memories (graph facet).
233 /// Returns the stable id of the fact (idempotent on identical content).
234 ///
235 /// The stored metadata is auto-stamped with today's date under
236 /// [`crate::storage::AUTO_DATE_FIELD`] unless `metadata` already carries
237 /// that key — see [`Self::remember_with_ttl`] (this method's only caller)
238 /// for the full contract.
239 ///
240 /// Every link is validated — target existence AND relation label —
241 /// *before* the fact is stored, so bad link input never leaves the fact
242 /// half-written. If an edge write itself fails afterwards (e.g. a target
243 /// expiring concurrently), a freshly-created fact is rolled back; a
244 /// re-remembered fact keeps its updated payload (re-remembering updates
245 /// metadata by design, and deleting it would destroy prior state).
246 /// Concurrent `remember`s of identical content are last-writer-wins,
247 /// not transactional.
248 ///
249 /// # Errors
250 /// Returns [`MemoryError::EmptyFact`] for empty/whitespace facts,
251 /// [`MemoryError::FactTooLarge`] if the fact exceeds
252 /// [`crate::limits::MAX_EMBEDDABLE_TEXT_BYTES`],
253 /// [`MemoryError::SelfRelation`] if a link points the fact at itself,
254 /// [`MemoryError::ReservedKey`] if `metadata` names a reserved key
255 /// (`content` or any `_veles_`-prefixed system key, [`crate::storage::AUTO_DATE_FIELD`]
256 /// excepted),
257 /// [`MemoryError::MetadataTooLarge`] if `metadata` exceeds
258 /// [`crate::limits::MAX_METADATA_BYTES`],
259 /// [`MemoryError::UnknownMemory`] if a link points at a missing memory,
260 /// [`MemoryError::InvalidRelation`] for a bad relation label,
261 /// [`MemoryError::RollbackFailed`] if an edge write failed and the
262 /// compensating delete also failed (the fact remains stored),
263 /// or a storage error if persistence fails.
264 pub fn remember(
265 &self,
266 fact: &str,
267 links: &[Link],
268 metadata: Option<&Metadata>,
269 ) -> Result<u64, MemoryError> {
270 self.remember_with_ttl(fact, links, metadata, None)
271 }
272
273 /// Like [`Self::remember`], but the fact **expires after `ttl_seconds`**.
274 ///
275 /// The expiry is a durable TTL — persisted with the fact (reserved
276 /// `_veles_expires_at` payload field), so it survives a process restart, and
277 /// expired facts stop being recalled. `None` stores the fact permanently,
278 /// exactly like [`Self::remember`]; an explicit `Some(0)` is **refused**
279 /// ([`MemoryError::ZeroTtl`]) rather than silently normalised to
280 /// "permanent", which is the opposite of what a caller writing `0` means.
281 /// Metadata and a TTL combine: the metadata is written and the expiry
282 /// preserved.
283 ///
284 /// The stored metadata is **auto-stamped with today's date** under
285 /// [`crate::storage::AUTO_DATE_FIELD`] (`_veles_date`, a `YYYYMMDD`
286 /// integer read from the system clock at write time — see
287 /// [`crate::clock::today_ymd`]) whenever `metadata` doesn't already carry
288 /// that key; an explicit value in `metadata` (e.g. to date a fact
289 /// retroactively) is never overwritten. No clock is available on
290 /// `wasm32-unknown-unknown`, so that target stamps nothing and `metadata`
291 /// passes through unchanged. This is the ONE place in the crate that
292 /// reads wall-clock time on the write path — the context compiler
293 /// (`compile_context` and friends) stays clock-free and deterministic,
294 /// unaffected by this stamp (it never re-derives a date from `now()`,
295 /// only ever reads whatever a fact already carries).
296 ///
297 /// Because [`Self::remember_extracted`] stores each extracted fact via
298 /// [`Self::remember`] (which delegates here), it gets the same auto-stamp
299 /// for free — entity hubs it also creates go through [`Self::store_fact`]
300 /// directly and are never stamped, since they are internal graph
301 /// scaffolding, not caller facts.
302 ///
303 /// # Errors
304 /// Same as [`Self::remember`].
305 pub fn remember_with_ttl(
306 &self,
307 fact: &str,
308 links: &[Link],
309 metadata: Option<&Metadata>,
310 ttl_seconds: Option<u64>,
311 ) -> Result<u64, MemoryError> {
312 self.remember_inner(fact, links, metadata, ttl_seconds, true)
313 }
314
315 /// The shared write path. `run_autograph` is false for the one caller that
316 /// has ALREADY extracted the passage — [`Self::remember_extracted`] — so a
317 /// service with autograph on does not run a second generation per stored
318 /// fact, re-deriving what it just computed.
319 fn remember_inner(
320 &self,
321 fact: &str,
322 links: &[Link],
323 metadata: Option<&Metadata>,
324 ttl_seconds: Option<u64>,
325 run_autograph: bool,
326 ) -> Result<u64, MemoryError> {
327 let fact = fact.trim();
328 self.validate_write(fact, links, metadata, ttl_seconds)?;
329 let fact_id = id::stable_id(fact);
330 reject_self_links(fact_id, links)?;
331 let existed_before = !links.is_empty() && self.store.get(fact_id)?.is_some();
332 self.write_fact(fact_id, fact, metadata, ttl_seconds)?;
333 self.link_or_rollback(fact_id, links, existed_before)?;
334 self.autograph_if(run_autograph, fact_id, fact);
335 Ok(fact_id)
336 }
337
338 /// Every deterministic rejection, before anything is written: a blank or
339 /// over-long fact, an explicit zero TTL, reserved or oversized metadata,
340 /// and each link's label and target. Run as one pass so a bad input never
341 /// leaves a half-written fact behind.
342 fn validate_write(
343 &self,
344 fact: &str,
345 links: &[Link],
346 metadata: Option<&Metadata>,
347 ttl_seconds: Option<u64>,
348 ) -> Result<(), MemoryError> {
349 validate_fact(fact)?;
350 reject_zero_ttl(ttl_seconds)?;
351 reject_reserved_keys(metadata)?;
352 reject_oversized_metadata(metadata)?;
353 self.validate_links(links)
354 }
355
356 /// Embed the fact and persist it with its date-stamped metadata and TTL.
357 fn write_fact(
358 &self,
359 fact_id: u64,
360 fact: &str,
361 metadata: Option<&Metadata>,
362 ttl_seconds: Option<u64>,
363 ) -> Result<(), MemoryError> {
364 let embedding = self.embedder.embed(fact)?;
365 let stamped = stamp_with_today(metadata);
366 // `ttl_seconds` is already known positive-or-absent: `validate_write`
367 // refuses an explicit `Some(0)` before any of this runs.
368 self.store_fact(fact_id, fact, &embedding, stamped.as_ref(), ttl_seconds)
369 }
370
371 /// Validate EVERY link property — relation label and target existence —
372 /// before any write, so all deterministic link failures happen while
373 /// nothing has been stored or overwritten yet.
374 fn validate_links(&self, links: &[Link]) -> Result<(), MemoryError> {
375 for link in links {
376 validate_relation(&link.relation)?;
377 }
378 self.ensure_link_targets_exist(links)
379 }
380
381 /// Write the edges, undoing a freshly-created fact if one of them fails.
382 ///
383 /// Links are fully pre-validated by [`Self::validate_links`], so an edge
384 /// write can only fail here on a race (e.g. a target's TTL lapsing since
385 /// the pre-check). Roll a FRESH fact back (delete cascades any edges
386 /// already created); a fact that existed before the call is kept —
387 /// deleting it would destroy prior state, and its updated payload stands
388 /// per re-remember's update semantics. The existence probe and the delete
389 /// are not one atomic unit: a concurrent remember of identical content
390 /// between them is last-writer-wins (documented on [`Self::remember`]).
391 fn link_or_rollback(
392 &self,
393 fact_id: u64,
394 links: &[Link],
395 existed_before: bool,
396 ) -> Result<(), MemoryError> {
397 let Err(cause) = self.relate_links(fact_id, links) else {
398 return Ok(());
399 };
400 if existed_before {
401 return Err(cause);
402 }
403 match self.store.delete(fact_id) {
404 Ok(()) => Err(cause),
405 Err(rollback) => Err(MemoryError::RollbackFailed {
406 cause: Box::new(cause),
407 rollback: Box::new(rollback),
408 }),
409 }
410 }
411
412 /// Run [`Self::autograph`] only when this write path asked for it — the
413 /// branch lives here rather than in the write path itself.
414 ///
415 /// With a worker spawned ([`Self::spawn_autograph_worker`]), the job is
416 /// ENQUEUED and this returns immediately: the enrichment leaves the
417 /// caller's response path (#1846). A FULL queue drops the job, counted
418 /// in [`Self::autograph_dropped`] — losing structure is recoverable by
419 /// re-remembering, stalling every write behind a slow model is not. A
420 /// disconnected queue (worker gone) falls back inline, so the graph
421 /// keeps building even if the worker died.
422 fn autograph_if(&self, run: bool, fact_id: u64, fact: &str) {
423 if !run {
424 return;
425 }
426 let guard = self.autograph_queue.tx.lock();
427 if let Some(tx) = guard.as_ref() {
428 use std::sync::mpsc::TrySendError;
429 match tx.try_send(AutographJob {
430 fact_id,
431 fact: fact.to_owned(),
432 }) {
433 Ok(()) => return,
434 Err(TrySendError::Full(_)) => {
435 self.autograph_queue
436 .dropped
437 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
438 #[cfg(feature = "mcp")]
439 tracing::warn!(
440 fact_id,
441 "autograph queue full: enrichment dropped — the fact is \
442 stored, its graph structure is not; re-remembering \
443 rebuilds it"
444 );
445 return;
446 }
447 Err(TrySendError::Disconnected(_)) => {
448 // fall through to the inline path below
449 }
450 }
451 }
452 drop(guard);
453 self.autograph(fact_id, fact);
454 }
455
456 /// How many autograph enrichments a FULL queue refused since this
457 /// service was built (#1846). The facts themselves were stored; only
458 /// their graph wiring was skipped, and re-remembering a fact rebuilds it.
459 #[must_use]
460 pub fn autograph_dropped(&self) -> u64 {
461 self.autograph_queue
462 .dropped
463 .load(std::sync::atomic::Ordering::Relaxed)
464 }
465
466 /// Whether the background autograph queue is OPEN — a worker is spawned
467 /// and `remember` enqueues instead of running the enrichment inline.
468 /// Turns false the moment a worker handle's drop closes the queue.
469 #[must_use]
470 pub fn autograph_queue_open(&self) -> bool {
471 self.autograph_queue.tx.lock().is_some()
472 }
473
474 /// Whether an autograph extractor is configured at all.
475 #[must_use]
476 pub fn has_autograph(&self) -> bool {
477 self.autograph.is_some()
478 }
479
480 /// The total number of live tracked facts, internal entity hubs included
481 /// — the store's [`MemoryStore::count`], relayed for `memory_status`.
482 #[must_use]
483 pub fn fact_count(&self) -> usize {
484 self.store.count()
485 }
486
487 /// The total number of graph edges, when the backend can say —
488 /// [`MemoryStore::edge_count`], relayed for `memory_status`. `None`
489 /// means "cannot say", never "zero": the two answers tell a caller
490 /// different things about `why()`.
491 #[must_use]
492 pub fn edge_count(&self) -> Option<usize> {
493 self.store.edge_count()
494 }
495
496 /// One page of the store's facts, for auditing — "what does my agent
497 /// know?" — which `recall` structurally cannot answer: it ranks by
498 /// resemblance to a query, and what resembles nothing you thought to
499 /// ask stays invisible.
500 ///
501 /// The store hands back raw pages ([`MemoryStore::list`]); the
502 /// visibility policy is applied here, once, for every backend: internal
503 /// entity hubs are skipped unless `include_internal` (they are the
504 /// graph's scaffolding, not the user's facts), reserved `_veles_*` keys
505 /// are stripped exactly as `recall` strips them (the auto-stamped date
506 /// survives — an audit legitimately asks WHEN), and `filter` keeps only
507 /// facts whose metadata equals every given key. A filtered page may
508 /// come back sparse — the cursor still advances over what was skipped,
509 /// so the WALK stays exhaustive.
510 ///
511 /// # Errors
512 /// Returns [`MemoryError`] if the backend cannot enumerate or the walk
513 /// fails.
514 pub fn list(
515 &self,
516 cursor: Option<u64>,
517 limit: usize,
518 filter: Option<&Metadata>,
519 include_internal: bool,
520 ) -> Result<(Vec<crate::model::ListedMemory>, Option<u64>), MemoryError> {
521 let limit = crate::limits::clamp_recall_limit(limit.max(1));
522 let (page, next) = self.store.list(cursor, limit)?;
523 let memories = page
524 .into_iter()
525 .filter_map(|fact| audited(fact, filter, include_internal))
526 .collect();
527 Ok((memories, next))
528 }
529}
530
531#[cfg(not(target_arch = "wasm32"))]
532impl<E, S> MemoryService<E, S>
533where
534 E: Embedder + Send + Sync + 'static,
535 S: MemoryStore + Send + Sync + 'static,
536{
537 /// The autograph worker's whole life, run on the spawned thread: ends
538 /// when every sender is gone — i.e. when the handle's drop takes the
539 /// sender back out of the service. Once the closing latch is up,
540 /// still-queued jobs are SKIPPED: the exit pays for the job in flight,
541 /// never for the queue.
542 fn autograph_worker_loop(&self, rx: &std::sync::mpsc::Receiver<AutographJob>) {
543 let mut skipped_on_close: u64 = 0;
544 for job in rx {
545 if self
546 .autograph_queue
547 .closing
548 .load(std::sync::atomic::Ordering::Acquire)
549 {
550 skipped_on_close += 1;
551 continue;
552 }
553 self.autograph(job.fact_id, &job.fact);
554 }
555 if skipped_on_close > 0 {
556 self.autograph_queue
557 .dropped
558 .fetch_add(skipped_on_close, std::sync::atomic::Ordering::Relaxed);
559 // ONE aggregated line, not one per job (#1834's rule).
560 #[cfg(feature = "mcp")]
561 tracing::warn!(
562 skipped = skipped_on_close,
563 "autograph worker closing: queued enrichments skipped — \
564 the facts are stored, their graph structure is not; \
565 re-remembering rebuilds it"
566 );
567 }
568 }
569
570 /// Move autograph off the response path: spawn ONE background worker
571 /// consuming a bounded queue, so `remember` returns as soon as the fact
572 /// is durably stored and the graph is wired behind (#1846).
573 ///
574 /// Measured motivation: with the production extractor, an inline
575 /// autograph held every `remember` for 46-52 s while the embedding cost
576 /// 0.12 s — and the MCP client timed out mid-generation, making a stored
577 /// fact indistinguishable from a lost one (#1839).
578 ///
579 /// The read-after-write contract changes, deliberately and visibly: an
580 /// `entity()` issued right after `remember` may not see the new edges
581 /// yet. The fact itself is always readable immediately — only the
582 /// DERIVED structure lags by one generation.
583 ///
584 /// One worker on purpose: the store is single-writer, and a second
585 /// in-flight generation would only add contention, not throughput.
586 /// `capacity` bounds the queue ([`crate::limits::MAX_AUTOGRAPH_QUEUE`]
587 /// is the daemon's choice); a full queue DROPS new enrichments, counted
588 /// by [`Self::autograph_dropped`] and logged — never silent, never
589 /// blocking the write path.
590 ///
591 /// # Errors
592 /// Returns [`MemoryError::Extract`] when a worker is already spawned for
593 /// this service — two workers would race the single-writer store for no
594 /// gain — or when the OS refuses the thread.
595 pub fn spawn_autograph_worker(
596 self: &std::sync::Arc<Self>,
597 capacity: usize,
598 ) -> Result<AutographWorkerHandle, MemoryError> {
599 let (tx, rx) = std::sync::mpsc::sync_channel::<AutographJob>(capacity);
600 {
601 let mut guard = self.autograph_queue.tx.lock();
602 if guard.is_some() {
603 return Err(MemoryError::Extract(crate::extract::ExtractError::Backend(
604 "autograph worker already spawned for this service".to_owned(),
605 )));
606 }
607 *guard = Some(tx);
608 // Re-arm the shutdown latch under the same lock that installs
609 // the sender: a previous worker's close must not poison this one
610 // into skipping every job it will ever receive.
611 self.autograph_queue
612 .closing
613 .store(false, std::sync::atomic::Ordering::Release);
614 }
615 let worker_service = std::sync::Arc::clone(self);
616 let join = std::thread::Builder::new()
617 .name("velesdb-autograph".to_owned())
618 .spawn(move || worker_service.autograph_worker_loop(&rx))
619 .map_err(|err| {
620 MemoryError::Extract(crate::extract::ExtractError::Backend(format!(
621 "spawn autograph worker: {err}"
622 )))
623 })?;
624 let closer_service = std::sync::Arc::clone(self);
625 Ok(AutographWorkerHandle {
626 close_queue: Some(Box::new(move || {
627 // Latch FIRST, sender out second: the worker observes the
628 // latch no later than the queue's end, so it cannot start
629 // draining jobs the shutdown meant to skip.
630 closer_service
631 .autograph_queue
632 .closing
633 .store(true, std::sync::atomic::Ordering::Release);
634 closer_service.autograph_queue.tx.lock().take();
635 })),
636 join: Some(join),
637 })
638 }
639}
640
641impl<E: Embedder, S: MemoryStore> MemoryService<E, S> {
642 /// Autograph one just-stored fact: read the entities, entity→entity edges
643 /// and attributes it states, and wire them around it.
644 ///
645 /// **Deliberately infallible.** The caller's fact is already durably
646 /// stored by the time this runs, and the caller asked to remember a fact —
647 /// not to run a model. Propagating an extraction failure would turn a
648 /// successful write into a reported error, and an agent that sees
649 /// `remember` fail will sensibly retry it, re-running the generation and
650 /// failing again. So a model that is down, slow, or talking nonsense costs
651 /// the *graph enrichment* and nothing else: the memory is kept, the id is
652 /// returned, and the next `remember` tries again.
653 ///
654 /// The trade-off is that a persistently broken extractor degrades silently
655 /// to plain `remember`. That is the right way round — losing structure is
656 /// recoverable by re-remembering, losing the fact is not.
657 fn autograph(&self, fact_id: u64, fact: &str) {
658 let Some(extractor) = self.autograph.as_ref() else {
659 return;
660 };
661 let Ok(mut extraction) = extractor.extract_graph(fact) else {
662 return;
663 };
664 // Privacy invariant: autograph derives structure FROM a stored fact, so
665 // if the fact no longer exists, none of its derived structure may be
666 // created. With a background worker a job can sit queued for
667 // minutes-to-hours and its generation runs for tens of seconds, so a
668 // `forget` issued in between must win — a permanent deletion cannot be
669 // undone by a stale enrichment resurrecting the entity hubs. Re-check
670 // once the generation has returned, before any wiring: this closes the
671 // whole queue-plus-generation window, the common case, completely.
672 if !self.fact_exists(fact_id) {
673 return;
674 }
675 crate::extract::orient_kinship(fact, &mut extraction.relations);
676 let mut entity_ids: HashMap<String, u64> = HashMap::new();
677 let mut edges: HashSet<(u64, u64, String)> = HashSet::new();
678 // The caller's fact is the node the topics attach to — the extracted
679 // facts are NOT stored as separate memories here, which is what
680 // separates autograph from `remember_extracted`: one `remember` call
681 // must still produce exactly one caller-visible memory.
682 for extracted in &extraction.facts {
683 let _ = self.wire_entities(fact_id, &extracted.entities, &mut entity_ids, &mut edges);
684 }
685 // A concurrent `forget` can still race the wiring writes above between
686 // the first check and here. Re-check once more before the hub↔hub and
687 // attribute writes — neither of which references the source fact, so
688 // `ensure_exists` cannot catch a retired fact for them — leaving a
689 // residual window of a single already-committed generation, not the
690 // whole job.
691 if !self.fact_exists(fact_id) {
692 return;
693 }
694 let _ = self.wire_relations(&extraction.relations, &mut entity_ids, &mut edges);
695 let _ = self.wire_attributes(&extraction.attributes, &mut entity_ids);
696 }
697
698 /// Cheap "is this fact still stored?" probe gating [`Self::autograph`]'s
699 /// wiring: the same `store.get` existence check [`Self::forget`] and
700 /// [`Self::ensure_exists`] use. A store read error answers `false` —
701 /// autograph must never fabricate structure for a fact it cannot prove is
702 /// still there, and a missing (or unprovable) fact is a clean skip, never
703 /// an error on this deliberately-infallible path.
704 fn fact_exists(&self, fact_id: u64) -> bool {
705 matches!(self.store.get(fact_id), Ok(Some(_)))
706 }
707
708 /// Create each outgoing link from `fact_id`.
709 ///
710 /// Precondition: every label was already validated by
711 /// [`Self::remember_with_ttl`]'s pre-write pass (its only caller) —
712 /// no re-check here, so the validation rule lives in exactly one
713 /// place on this path.
714 fn relate_links(&self, fact_id: u64, links: &[Link]) -> Result<(), MemoryError> {
715 for link in links {
716 self.store.relate(fact_id, link.target, &link.relation)?;
717 }
718 Ok(())
719 }
720
721 /// Remember a passage of raw `text` by running it through an [`Extractor`]
722 /// and storing every fact it yields, **auto-wiring the fact↔entity graph**.
723 ///
724 /// This is the commodity on top of [`Self::remember`]'s bring-your-own-links
725 /// core: each extracted fact is stored (tagged with `metadata`), each salient
726 /// topic becomes a deduplicated hub memory, and every fact is linked to its
727 /// topics with a bidirectional `about`/`mentions` edge. Two facts sharing a
728 /// topic therefore become reachable from one another, so [`Self::why`] has a
729 /// real graph to traverse with no manual `relate()`.
730 ///
731 /// Entity hubs are content-addressed, so the same topic seen across many
732 /// calls collapses onto one hub. Returns the ids of the stored facts (entity
733 /// hubs excluded), in extraction order, plus how many facts were skipped
734 /// for exceeding the embeddable cap — one unusable fact must not cost the
735 /// others, the policy every other stage of this pipeline already follows
736 /// (a malformed triple is skipped, a blank entity is skipped).
737 ///
738 /// # Errors
739 /// Returns [`MemoryError::EmptyFact`] for empty/whitespace `text`,
740 /// [`MemoryError::Extract`] if extraction fails, [`MemoryError::ReservedKey`]
741 /// if `metadata` names a reserved key, [`MemoryError::MetadataTooLarge`] if
742 /// `metadata` exceeds [`crate::limits::MAX_METADATA_BYTES`], or a storage
743 /// error if persistence fails. A fact past
744 /// [`crate::limits::MAX_EMBEDDABLE_TEXT_BYTES`] is NOT an error: it is
745 /// counted in [`RememberedExtraction::skipped_over_cap`] and the call
746 /// carries on.
747 pub fn remember_extracted<X: Extractor>(
748 &self,
749 text: &str,
750 extractor: &X,
751 metadata: Option<&Metadata>,
752 ) -> Result<RememberedExtraction, MemoryError> {
753 let text = text.trim();
754 if text.is_empty() {
755 return Err(MemoryError::EmptyFact);
756 }
757 let mut extraction = extractor.extract_graph(text)?;
758 crate::extract::orient_kinship(text, &mut extraction.relations);
759 let mut entity_ids: HashMap<String, u64> = HashMap::new();
760 let mut edges: HashSet<(u64, u64, String)> = HashSet::new();
761 let outcome =
762 self.store_extracted_facts(&extraction.facts, metadata, &mut entity_ids, &mut edges)?;
763 self.wire_relations(&extraction.relations, &mut entity_ids, &mut edges)?;
764 self.wire_attributes(&extraction.attributes, &mut entity_ids)?;
765 Ok(outcome)
766 }
767
768 /// Look up everything known about a named entity: the attributes merged
769 /// onto its hub, and the typed edges leaving it.
770 ///
771 /// This is the *read* side of the auto-built graph, and it exists because
772 /// entity hubs are deliberately invisible to [`Self::recall`] and
773 /// [`Self::recall_where`] — a hub ranking for its own topic would evict a
774 /// real fact from the caller's results. Without this accessor an attribute
775 /// merged onto a hub would be stored correctly and yet be unreachable
776 /// through every public read path: the worst kind of feature, one that
777 /// looks done and silently returns nothing.
778 ///
779 /// `name` is canonicalized exactly like an extracted entity (trimmed,
780 /// lowercased), so the caller may pass `"Theo Durand"` and reach the node
781 /// built from `"theo durand"`. Returns `None` when no hub exists for the
782 /// name — nothing has ever mentioned that entity.
783 ///
784 /// # Errors
785 /// Returns [`MemoryError`] if the store lookup fails.
786 pub fn entity_profile(&self, name: &str) -> Result<Option<EntityProfile>, MemoryError> {
787 let key = canonical_entity_name(name);
788 if key.is_empty() {
789 return Ok(None);
790 }
791 let id = id::stable_id(&format!("{HUB_ID_SALT}{key}"));
792 if self.store.get(id)?.is_none() {
793 return Ok(None);
794 }
795 // Reserved system keys (the hub flag itself) are scaffolding, not
796 // attributes the caller ever wrote — strip them exactly as every other
797 // caller-facing read path does.
798 let (relations, relations_truncated) = self.outgoing_entity_relations(id)?;
799 let (relations_in, relations_in_truncated) = self.incoming_entity_relations(id)?;
800 Ok(Some(EntityProfile {
801 id,
802 name: key,
803 attributes: strip_reserved_keys(self.store.get_metadata(id)?).unwrap_or_default(),
804 relations,
805 relations_in,
806 relations_truncated,
807 relations_in_truncated,
808 }))
809 }
810
811 /// The typed edges leaving `id`, resolved to their target's content, and
812 /// whether that list is a partial view.
813 ///
814 /// Scaffolding edges (`mentions`, and `about` for symmetry — a hub never
815 /// has an outgoing `about`) are dropped: they point at the facts that
816 /// tagged this entity, not at a statement *about* it.
817 fn outgoing_entity_relations(
818 &self,
819 id: u64,
820 ) -> Result<(Vec<EntityRelation>, bool), MemoryError> {
821 let scanned = self
822 .store
823 .relations_bounded(id, crate::limits::MAX_ENTITY_SCAN_EDGES)?;
824 self.resolve_entity_relations(scanned, |edge| edge.to)
825 }
826
827 /// The typed edges pointing at `id`, resolved to their SOURCE's content —
828 /// for an incoming edge the far end is where it comes *from* — and
829 /// whether that list is a partial view.
830 ///
831 /// Without these, a question is only answerable from one side: the graph
832 /// holds `camille --soeur de--> theo`, so reading Theo's outgoing edges
833 /// never finds Camille. The edge exists, it simply leaves the other node.
834 ///
835 /// The scaffolding filter is the incoming mirror of
836 /// [`Self::outgoing_entity_relations`]'s: `about` edges are dropped (they
837 /// are the fact → hub half of the `about`/`mentions` pair), and `mentions`
838 /// with them for symmetry.
839 fn incoming_entity_relations(
840 &self,
841 id: u64,
842 ) -> Result<(Vec<EntityRelation>, bool), MemoryError> {
843 let scanned = self
844 .store
845 .incoming_relations_bounded(id, crate::limits::MAX_ENTITY_SCAN_EDGES)?;
846 self.resolve_entity_relations(scanned, |edge| edge.from)
847 }
848
849 /// Shared resolver for both edge directions: skip the bipartite
850 /// scaffolding labels, resolve each edge's far end (`far_end` picks which
851 /// endpoint that is) to its stored content — at most
852 /// [`crate::limits::MAX_ENTITY_RELATIONS`] of them — and say whether the
853 /// result is a partial view (#1820).
854 ///
855 /// Truncated when either budget bit: the store's raw scan window
856 /// ([`crate::limits::MAX_ENTITY_SCAN_EDGES`]) left edges unread, or a
857 /// typed edge past the resolution cap was seen and dropped. Both cuts
858 /// are the same honest signal — "there is more than this view shows".
859 fn resolve_entity_relations(
860 &self,
861 scanned: crate::model::BoundedMemoryEdges,
862 far_end: impl Fn(&MemoryEdge) -> u64,
863 ) -> Result<(Vec<EntityRelation>, bool), MemoryError> {
864 let mut relations = Vec::new();
865 let mut truncated = scanned.truncated;
866 for edge in scanned.edges {
867 if edge.relation == MENTIONS_RELATION || edge.relation == ABOUT_RELATION {
868 continue;
869 }
870 if relations.len() >= crate::limits::MAX_ENTITY_RELATIONS {
871 truncated = true;
872 break;
873 }
874 let far = far_end(&edge);
875 let content = self.store.get(far)?.map(|(content, _)| content);
876 relations.push(EntityRelation {
877 predicate: edge.relation,
878 target_id: far,
879 target: content.unwrap_or_default(),
880 });
881 }
882 Ok((relations, truncated))
883 }
884
885 /// Wire each extracted `subject -[predicate]-> object` triple as a typed
886 /// edge between the two entity hubs.
887 ///
888 /// This is the step that turns the bipartite fact↔topic graph into a real
889 /// knowledge graph. The hubs are resolved through [`Self::entity_hub`], so
890 /// an endpoint naming an entity some earlier passage already introduced
891 /// reuses that entity's existing node rather than forking a parallel one —
892 /// hub ids are content-addressed, so this holds across calls and sessions.
893 ///
894 /// Only the stated direction is written. Inferring the converse
895 /// (`father of` ⇒ `child of`) would mean inventing a label the passage
896 /// never used, and an inverted vocabulary nobody can predict is worse than
897 /// an absent edge: `why()` walks outgoing edges, so a wrong direction
898 /// silently misroutes every later traversal.
899 ///
900 /// A malformed triple is skipped, not fatal — one unusable predicate must
901 /// not cost the caller the facts stored alongside it.
902 fn wire_relations(
903 &self,
904 relations: &[ExtractedRelation],
905 entity_ids: &mut HashMap<String, u64>,
906 edges: &mut HashSet<(u64, u64, String)>,
907 ) -> Result<(), MemoryError> {
908 for relation in relations {
909 if validate_relation(&relation.predicate).is_err() {
910 continue;
911 }
912 let subject_id = self.entity_hub(&relation.subject, entity_ids)?;
913 let object_id = self.entity_hub(&relation.object, entity_ids)?;
914 if subject_id == object_id {
915 continue;
916 }
917 self.add_edge(subject_id, object_id, &relation.predicate, edges)?;
918 }
919 Ok(())
920 }
921
922 /// Merge each extracted attribute into its entity hub's `ColumnStore`
923 /// metadata, so `recall_where` can filter on it (`age >= 15`).
924 ///
925 /// The write goes through `update_metadata`, which **merges** rather than
926 /// replaces. That is the whole point: learning "Theo has a sister" after
927 /// "Theo is 15" must not erase the age. Re-storing the hub payload wholesale
928 /// would silently drop every attribute learned in an earlier session.
929 ///
930 /// Values keep the JSON type the extractor produced. `recall_where`
931 /// compares type-strictly with no coercion, so an age stored as `"15"`
932 /// would never match a numeric filter — no error, just a permanent silent
933 /// miss.
934 ///
935 /// Reserved keys are skipped: a model emitting `content` or a `_veles_`
936 /// key must never be able to overwrite the hub's own content or its
937 /// system flags.
938 fn wire_attributes(
939 &self,
940 attributes: &[ExtractedAttribute],
941 entity_ids: &mut HashMap<String, u64>,
942 ) -> Result<(), MemoryError> {
943 let mut per_entity: HashMap<String, Metadata> = HashMap::new();
944 for attribute in attributes {
945 if is_reserved_key(&attribute.key) {
946 continue;
947 }
948 per_entity
949 .entry(attribute.entity.clone())
950 .or_default()
951 .insert(attribute.key.clone(), attribute.value.clone());
952 }
953 for (entity, meta) in per_entity {
954 if meta.is_empty() {
955 continue;
956 }
957 reject_oversized_metadata(Some(&meta))?;
958 let hub_id = self.entity_hub(&entity, entity_ids)?;
959 self.store.update_metadata(hub_id, &meta)?;
960 }
961 Ok(())
962 }
963
964 /// Store each extracted fact and wire it to its topics, returning their ids.
965 ///
966 /// Goes through the no-autograph path: the passage was ALREADY extracted by
967 /// the caller, so re-running a generation per stored fact would re-derive
968 /// what was just computed.
969 fn store_extracted_facts(
970 &self,
971 facts: &[crate::extract::ExtractedFact],
972 metadata: Option<&Metadata>,
973 entity_ids: &mut HashMap<String, u64>,
974 edges: &mut HashSet<(u64, u64, String)>,
975 ) -> Result<RememberedExtraction, MemoryError> {
976 let mut ids = Vec::with_capacity(facts.len());
977 let mut skipped_over_cap = 0;
978 for fact in facts {
979 let content = fact.text.trim();
980 if content.is_empty() {
981 continue;
982 }
983 // An over-cap fact is skipped, not fatal: aborting here used to
984 // leave the previous iterations persisted with no rollback, no
985 // graph wiring, and no ids returned — the worst of every world.
986 // Every OTHER error still aborts: they signal bad caller input
987 // (reserved keys, oversized metadata) or a failing store, where
988 // carrying on would compound the damage.
989 let fact_id = match self.remember_inner(content, &[], metadata, None, false) {
990 Ok(id) => id,
991 Err(MemoryError::FactTooLarge { .. }) => {
992 skipped_over_cap += 1;
993 continue;
994 }
995 Err(error) => return Err(error),
996 };
997 ids.push(fact_id);
998 self.wire_entities(fact_id, &fact.entities, entity_ids, edges)?;
999 }
1000 Ok(RememberedExtraction {
1001 ids,
1002 skipped_over_cap,
1003 })
1004 }
1005
1006 /// Link `fact_id` to each of its topics with a deduplicated edge in *both*
1007 /// directions. `why()` only follows outgoing edges, so the fact→topic edge
1008 /// alone leaves hubs as dead ends; the topic→fact edge is what lets a walk
1009 /// hop from one fact, through a shared topic, to its sibling facts.
1010 fn wire_entities(
1011 &self,
1012 fact_id: u64,
1013 entities: &[String],
1014 entity_ids: &mut HashMap<String, u64>,
1015 edges: &mut HashSet<(u64, u64, String)>,
1016 ) -> Result<(), MemoryError> {
1017 for entity in entities {
1018 // Skip blank or punctuation-only topics: they would persist as junk
1019 // hubs (`Entity: -`) yet can never carry a meaningful multi-hop link.
1020 if entity.chars().any(char::is_alphanumeric) {
1021 self.wire_entity(fact_id, entity, entity_ids, edges)?;
1022 }
1023 }
1024 Ok(())
1025 }
1026
1027 /// Wire one topic to `fact_id`: resolve its hub, then add the deduplicated
1028 /// `about`/`mentions` pair (skipping a hub that is the fact itself).
1029 fn wire_entity(
1030 &self,
1031 fact_id: u64,
1032 entity: &str,
1033 entity_ids: &mut HashMap<String, u64>,
1034 edges: &mut HashSet<(u64, u64, String)>,
1035 ) -> Result<(), MemoryError> {
1036 let entity_id = self.entity_hub(entity, entity_ids)?;
1037 if entity_id == fact_id {
1038 return Ok(());
1039 }
1040 self.add_edge(fact_id, entity_id, ABOUT_RELATION, edges)?;
1041 self.add_edge(entity_id, fact_id, MENTIONS_RELATION, edges)?;
1042 Ok(())
1043 }
1044
1045 /// Create the edge `from -> to` labelled `label`, unless `edges` already
1046 /// records that triple for this call (in-call dedup only). `relate`
1047 /// derives the edge id from `(from, relation, to)`
1048 /// ([`crate::wire::hash_edge_id`] upstream in core) and is itself an O(1)
1049 /// idempotent no-op against an already-persisted edge, so there is
1050 /// nothing left to preload from the store — a prior preload here made
1051 /// every write to a hub with `k` existing edges cost O(k), turning `n`
1052 /// writes to the same hub into O(n²).
1053 fn add_edge(
1054 &self,
1055 from: u64,
1056 to: u64,
1057 label: &str,
1058 edges: &mut HashSet<(u64, u64, String)>,
1059 ) -> Result<(), MemoryError> {
1060 if edges.insert((from, to, label.to_string())) {
1061 self.relate(from, to, label)?;
1062 }
1063 Ok(())
1064 }
1065
1066 /// Get or create the hub memory for a topic, caching its id per call. The
1067 /// hub id is a deterministic function of the (normalized) topic, so the same
1068 /// topic resolves to the same hub across calls — never a duplicate.
1069 fn entity_hub(
1070 &self,
1071 entity: &str,
1072 entity_ids: &mut HashMap<String, u64>,
1073 ) -> Result<u64, MemoryError> {
1074 let key = entity.trim().to_lowercase();
1075 if let Some(&id) = entity_ids.get(&key) {
1076 return Ok(id);
1077 }
1078 let id = self.remember_hub(&key)?;
1079 entity_ids.insert(key, id);
1080 Ok(id)
1081 }
1082
1083 /// Idempotently store the hub memory for topic `key`. The id is salted so the
1084 /// hub id space is disjoint from natural fact ids (no caller fact can collide
1085 /// with or overwrite a hub), while the stored content stays human-readable.
1086 /// Marked with the reserved [`HUB_FIELD`] so recall and `why` seeds exclude
1087 /// it; goes straight to [`Self::store_fact`] to bypass the caller-facing
1088 /// reserved-key rejection in [`Self::remember`].
1089 fn remember_hub(&self, key: &str) -> Result<u64, MemoryError> {
1090 let id = id::stable_id(&format!("{HUB_ID_SALT}{key}"));
1091 // An existing hub is left exactly as it is. Re-storing it would rewrite
1092 // the payload to the bare hub marker and destroy every attribute merged
1093 // onto it by an earlier call — learning "Theo has a sister" would erase
1094 // "Theo is 15", because a later sentence re-resolves the same hub. The
1095 // content is a pure function of `key`, so there is nothing to refresh;
1096 // skipping also avoids re-embedding a hub on every single mention.
1097 if self.store.get(id)?.is_some() {
1098 return Ok(id);
1099 }
1100 let content = format!("Entity: {key}");
1101 let embedding = self.embedder.embed(&content)?;
1102 let mut meta = Map::new();
1103 meta.insert(HUB_FIELD.to_string(), Value::Bool(true));
1104 // Topic hubs are graph anchors — they never expire.
1105 self.store_fact(id, &content, &embedding, Some(&meta), None)?;
1106 Ok(id)
1107 }
1108
1109 /// Fail with [`MemoryError::UnknownMemory`] unless memory `id` exists.
1110 fn ensure_exists(&self, id: u64) -> Result<(), MemoryError> {
1111 if self.store.get(id)?.is_none() {
1112 return Err(MemoryError::UnknownMemory(id));
1113 }
1114 Ok(())
1115 }
1116
1117 /// Fail unless every link target already exists (keeps `remember` atomic).
1118 fn ensure_link_targets_exist(&self, links: &[Link]) -> Result<(), MemoryError> {
1119 for link in links {
1120 self.ensure_exists(link.target)?;
1121 }
1122 Ok(())
1123 }
1124
1125 /// Store a fact with any combination of metadata and a durable TTL.
1126 fn store_fact(
1127 &self,
1128 id: u64,
1129 fact: &str,
1130 embedding: &[f32],
1131 metadata: Option<&Metadata>,
1132 ttl_seconds: Option<u64>,
1133 ) -> Result<(), MemoryError> {
1134 match (metadata, ttl_seconds) {
1135 (Some(meta), Some(ttl)) => {
1136 // ONE write, not two. The previous `store_with_ttl` then
1137 // `update_metadata` pair left the fact live and expiring
1138 // between the calls: a short TTL could lapse in the gap and
1139 // the metadata write then failed with `NotFound(... is
1140 // expired ...)` — the caller got an error on a fact that was
1141 // valid when they asked for it. Observed with a 1 s TTL on a
1142 // loaded machine. Every TTL'd write takes this arm, since the
1143 // auto date stamp means `metadata` is always `Some`.
1144 self.store
1145 .store_with_metadata_and_ttl(id, fact, embedding, meta, ttl)?;
1146 }
1147 (Some(meta), None) => self.store.store_with_metadata(id, fact, embedding, meta)?,
1148 (None, Some(ttl)) => self.store.store_with_ttl(id, fact, embedding, ttl)?,
1149 (None, None) => self.store.store(id, fact, embedding)?,
1150 }
1151 Ok(())
1152 }
1153
1154 /// Recall up to `k` memories semantically similar to `query` (vector facet),
1155 /// optionally narrowed to an exact-match metadata `filter` (`ColumnStore`
1156 /// facet) — e.g. `{ "project": "veles", "status": "resolved" }`.
1157 ///
1158 /// A highly selective filter may return fewer than `k` hits even when more
1159 /// matches exist — raise `k` for fuller coverage with a narrow filter.
1160 ///
1161 /// Entity hubs created by [`Self::remember_extracted`] are never returned:
1162 /// they are internal graph scaffolding, not facts the caller stored.
1163 ///
1164 /// Each hit carries its caller metadata (`Recollection::metadata`, `None`
1165 /// when the fact carries none) — store a date field (e.g. `occurred_at`)
1166 /// and it round-trips here, so a caller can sort the result into a
1167 /// chronological, date-stamped context without `recall_where`'s explicit
1168 /// filters. One extra, single batched lookup covers every returned hit.
1169 ///
1170 /// # Errors
1171 /// Returns [`MemoryError`] if the semantic query or the metadata lookup fails.
1172 pub fn recall(
1173 &self,
1174 query: &str,
1175 k: usize,
1176 filter: Option<&Metadata>,
1177 ) -> Result<Vec<Recollection>, MemoryError> {
1178 let query = query.trim();
1179 if query.is_empty() {
1180 return Ok(Vec::new());
1181 }
1182 reject_reserved_keys(filter)?;
1183 let embedding = self.embedder.embed(query)?;
1184 let hits = self.search(&embedding, k, filter)?;
1185 let ids: Vec<u64> = hits.iter().map(|(id, _, _)| *id).collect();
1186 // One raw batched payload lookup (reserved keys included), reused for
1187 // BOTH the RL re-rank and the caller-facing metadata below — a single
1188 // round trip, not one per concern.
1189 let payloads = self.store.get_metadata_batch(&ids)?;
1190 // RL Memory: re-order the recalled set by learned confidence. Facts
1191 // that never received `feedback` keep their similarity order exactly.
1192 #[cfg(feature = "persistence")]
1193 let (hits, payloads) = Self::rl_rerank(hits, payloads);
1194 Ok(hits
1195 .into_iter()
1196 .zip(payloads)
1197 .map(|((id, score, content), payload)| Recollection {
1198 id,
1199 score,
1200 content,
1201 metadata: strip_reserved_keys(payload),
1202 })
1203 .collect())
1204 }
1205
1206 /// Vector search for up to `k` ids, optionally narrowed by a metadata
1207 /// `filter`. Shared by [`Self::recall`] and [`Self::why`].
1208 fn search(
1209 &self,
1210 embedding: &[f32],
1211 k: usize,
1212 filter: Option<&Metadata>,
1213 ) -> Result<Vec<(u64, f32, String)>, MemoryError> {
1214 match filter {
1215 // An include filter already excludes hubs: a hub's payload
1216 // carries only reserved keys (`content`, `_veles_hub`), and
1217 // reserved keys are rejected from caller filters, so a non-empty
1218 // filter can never match a hub. An EMPTY-but-present filter (`Some({})`, the
1219 // natural `{}` idiom at the JS boundary) matches every payload —
1220 // hubs included — so it must take the hub-excluding path below,
1221 // exactly like an absent filter (same `Some({})` ≡ `None`
1222 // convention as `recall_fused`'s graph-side `matches_filter`).
1223 Some(meta) if !meta.is_empty() => self.store.query_filtered(embedding, k, meta, 0),
1224 // Unfiltered recall must still drop entity hubs explicitly, or a hub
1225 // like `Entity: rust` would rank for the topic and evict a real fact.
1226 _ => self
1227 .store
1228 .query_excluding(embedding, k, &hub_exclude_filter()),
1229 }
1230 }
1231
1232 /// Fused recall: semantic `NEAR` search combined with structured
1233 /// `ColumnStore` predicates over metadata columns — ranges and comparisons,
1234 /// not just the equality of [`Self::recall`]. One query spanning the vector
1235 /// and column facets (e.g. "most similar facts **with `timestamp` in this
1236 /// window**"), which a vector-only or equality-only recall cannot express.
1237 ///
1238 /// Filter *values* are bound as query parameters (never interpolated), so
1239 /// they cannot inject; filter *field names* are validated to be plain
1240 /// identifiers. Results come back in similarity order.
1241 ///
1242 /// **Caller memories only.** The store also holds internal scaffolding —
1243 /// the entity hubs of [`Self::remember_extracted`] and the context
1244 /// compiler's four artefact classes (stored sources, compilation events,
1245 /// working contexts, and the per-project working-context index). They sit
1246 /// in the same collection as caller facts and are excluded from every
1247 /// result here, whatever the predicate.
1248 ///
1249 /// That exclusion is applied by the backend against
1250 /// [`crate::storage::INTERNAL_MARKER_FIELDS`]; it is NOT a consequence of
1251 /// those facts being unfilterable. A caller cannot write a filter naming a
1252 /// reserved key, but `field ne value` MATCHES a fact that has no such
1253 /// field at all — and scaffolding has none of the caller's columns, so
1254 /// before #1737 every `ne` predicate returned all of it.
1255 ///
1256 /// # Errors
1257 /// Returns [`MemoryError::InvalidFilter`] if a filter field is not a plain
1258 /// identifier, [`MemoryError::Embed`] if the query cannot be embedded, or a
1259 /// storage error if the query fails. An empty query or `k == 0` yields `[]`.
1260 pub fn recall_where(
1261 &self,
1262 query: &str,
1263 k: usize,
1264 filters: &[ColumnFilter],
1265 ) -> Result<Vec<Recollection>, MemoryError> {
1266 let query = query.trim();
1267 if query.is_empty() || k == 0 {
1268 return Ok(Vec::new());
1269 }
1270 // No column predicates = a plain recall: route through [`Self::recall`]
1271 // so entity hubs stay excluded — `query_columnar` with an empty filter
1272 // set is a bare vector search that would rank internal `Entity:` hub
1273 // scaffolding as results (same `[]` ≡ unfiltered convention as
1274 // `search`'s empty-map handling).
1275 if filters.is_empty() {
1276 return self.recall(query, k, None);
1277 }
1278 let embedding = self.embedder.embed(query)?;
1279 self.store.query_columnar(&embedding, k, filters)
1280 }
1281
1282 /// Create a typed edge `from -> to`. Returns the edge id.
1283 ///
1284 /// Both endpoints are validated to exist first, so the tool reports an
1285 /// unknown id as client input (`UnknownMemory`) rather than a generic
1286 /// storage fault — and the graph never gains an edge dangling off a memory
1287 /// that was never stored.
1288 ///
1289 /// A self-loop (`from == to`) is refused: it states nothing, and `why`
1290 /// traverses it like any other edge, so it only adds noise to the
1291 /// evidence trail. The same rule covers [`Self::remember`]'s `links`.
1292 ///
1293 /// # Errors
1294 /// Returns [`MemoryError::InvalidRelation`] for a bad label,
1295 /// [`MemoryError::SelfRelation`] if both endpoints are the same memory,
1296 /// [`MemoryError::UnknownMemory`] if either endpoint is missing, or
1297 /// a storage error if the edge cannot be created.
1298 pub fn relate(&self, from: u64, to: u64, relation: &str) -> Result<u64, MemoryError> {
1299 validate_relation(relation)?;
1300 if from == to {
1301 return Err(MemoryError::SelfRelation(from));
1302 }
1303 self.ensure_exists(from)?;
1304 self.ensure_exists(to)?;
1305 self.store.relate(from, to, relation)
1306 }
1307
1308 /// Remove the edge(s) `from -relation-> to`: [`Self::relate`]'s exact
1309 /// undo (issue #1661), so a mistaken edge no longer costs the facts at
1310 /// its endpoints. Neither the facts nor any entity hub are touched —
1311 /// collecting an orphaned hub stays [`Self::forget`]'s job.
1312 ///
1313 /// Idempotent: an absent edge is `found: false`, not an error, so a
1314 /// cleanup is replayable. It refuses exactly what `relate` refuses
1315 /// (empty label, self-loop), and deliberately does NOT require the
1316 /// endpoints to exist — the edge of a forgotten fact is already gone,
1317 /// and reporting that as an error would break replay.
1318 ///
1319 /// Scope: the store does not distinguish an explicit edge from one the
1320 /// autograph derived from a passage, so `unrelate` removes both alike.
1321 /// To correct an autograph edge, prefer `forget` + `remember` of the
1322 /// source fact — otherwise a later `remember` of the same passage can
1323 /// rebuild the edge removed here.
1324 ///
1325 /// # Errors
1326 /// Returns [`MemoryError::InvalidRelation`] for a bad label,
1327 /// [`MemoryError::SelfRelation`] if both endpoints are the same memory,
1328 /// or a storage error if lookup or removal fails.
1329 pub fn unrelate(
1330 &self,
1331 from: u64,
1332 to: u64,
1333 relation: &str,
1334 ) -> Result<UnrelateOutcome, MemoryError> {
1335 validate_relation(relation)?;
1336 if from == to {
1337 return Err(MemoryError::SelfRelation(from));
1338 }
1339 let removed = self.remove_matching_edges(from, to, relation)?;
1340 Ok(UnrelateOutcome {
1341 found: removed > 0,
1342 removed,
1343 })
1344 }
1345
1346 /// [`Self::unrelate`]'s removal pass: resolve `from`'s outgoing edges and
1347 /// delete every one matching `(to, relation)` by its id, counting them.
1348 fn remove_matching_edges(
1349 &self,
1350 from: u64,
1351 to: u64,
1352 relation: &str,
1353 ) -> Result<usize, MemoryError> {
1354 let mut removed = 0usize;
1355 for edge in self.store.relations(from)? {
1356 if edge.to == to && edge.relation == relation && self.store.unrelate(edge.id)? {
1357 removed += 1;
1358 }
1359 }
1360 Ok(removed)
1361 }
1362
1363 /// Forget (delete) the memory with `fact_id`. Returns whether a memory
1364 /// actually existed under that id — the underlying store's `delete` is a
1365 /// silent no-op on an unknown id (matching most backends' idempotent
1366 /// delete semantics), which is indistinguishable from a real deletion
1367 /// unless existence is checked first. Every surface that exposes
1368 /// `forget` (MCP, Node, WASM, Python) forwards this so a caller can tell
1369 /// "I removed something" from "that id was a typo".
1370 ///
1371 /// The delete always runs, even when `get` reports the id absent: `get`
1372 /// filters TTL-expired facts, and an expired-but-unpurged row must still
1373 /// be reclaimed (the caller is told `false` — the memory was already
1374 /// gone from its perspective). Existence check and delete are two store
1375 /// calls, not one atomic operation: two concurrent forgets of one id may
1376 /// both report `true`.
1377 ///
1378 /// # Errors
1379 /// Returns [`MemoryError`] if the existence check or the deletion fails.
1380 pub fn forget(&self, fact_id: u64) -> Result<bool, MemoryError> {
1381 let found = self.store.get(fact_id)?.is_some();
1382 // Read the fact's hubs BEFORE the delete: afterwards its edges are gone
1383 // and there is no way back to the entities it created.
1384 let hubs = self.hubs_linked_from(fact_id)?;
1385 self.store.delete(fact_id)?;
1386 self.collect_orphan_hubs(&hubs)?;
1387 Ok(found)
1388 }
1389
1390 /// The entity hubs `fact_id` points at.
1391 ///
1392 /// Hubs are recognised by the reserved [`HUB_FIELD`] marker rather than by
1393 /// the edge label, so a caller's own `relate` to a hub is seen too.
1394 fn hubs_linked_from(&self, fact_id: u64) -> Result<Vec<u64>, MemoryError> {
1395 let mut hubs = Vec::new();
1396 for edge in self.store.relations(fact_id)? {
1397 if self.is_hub(edge.to)? {
1398 hubs.push(edge.to);
1399 }
1400 }
1401 Ok(hubs)
1402 }
1403
1404 /// Delete every hub in `hubs` that no surviving fact mentions any more.
1405 ///
1406 /// An entity outlives the fact that introduced it as long as another fact
1407 /// still refers to it — forgetting "Theo is 15" must not erase Theo while
1408 /// "Theo has a sister" is still stored. Only a hub whose every `mentions`
1409 /// target is gone is itself removed, so entities do not accumulate as
1410 /// unreachable scaffolding once the facts behind them are retracted.
1411 fn collect_orphan_hubs(&self, hubs: &[u64]) -> Result<(), MemoryError> {
1412 for &hub in hubs {
1413 if !self.hub_still_mentioned(hub)? {
1414 self.store.delete(hub)?;
1415 }
1416 }
1417 Ok(())
1418 }
1419
1420 /// Whether anything alive still needs `hub`.
1421 ///
1422 /// Two references count, and the second is why this reads BOTH
1423 /// directions (issue #1662):
1424 ///
1425 /// - an outgoing `mentions` edge to a live fact — the pair
1426 /// [`Self::wire_entities`] writes, the ordinary case;
1427 /// - an incoming edge from a live NON-HUB fact — what a caller's own
1428 /// `relate` writes, and it writes one direction only. Relating a fact
1429 /// to a hub is reachable (`entity()` hands out the hub id), so reading
1430 /// outgoing edges alone swept hubs from under live callers' edges,
1431 /// losing them in silence.
1432 ///
1433 /// Incoming edges from another HUB are deliberately ignored: hub↔hub
1434 /// edges exist (`wire_relations` writes them), and counting them would
1435 /// let two hubs keep each other alive forever — a leak whose outcome
1436 /// depends on collection order, which is worse than the bug being fixed.
1437 fn hub_still_mentioned(&self, hub: u64) -> Result<bool, MemoryError> {
1438 for edge in self.store.relations(hub)? {
1439 if edge.relation == MENTIONS_RELATION && self.store.get(edge.to)?.is_some() {
1440 return Ok(true);
1441 }
1442 }
1443 self.hub_has_live_referent(hub)
1444 }
1445
1446 /// Whether a live non-hub fact points AT `hub` — see
1447 /// [`Self::hub_still_mentioned`] for why hub→hub edges do not count.
1448 fn hub_has_live_referent(&self, hub: u64) -> Result<bool, MemoryError> {
1449 for edge in self.store.incoming_relations(hub)? {
1450 if self.store.get(edge.from)?.is_some() && !self.is_hub(edge.from)? {
1451 return Ok(true);
1452 }
1453 }
1454 Ok(false)
1455 }
1456
1457 /// Whether `id` is an entity hub (carries the reserved [`HUB_FIELD`]).
1458 fn is_hub(&self, id: u64) -> Result<bool, MemoryError> {
1459 Ok(self
1460 .store
1461 .get_metadata(id)?
1462 .is_some_and(|meta| meta.contains_key(HUB_FIELD)))
1463 }
1464
1465 /// Explain a `decision`: find the best-matching memory (optionally scoped to
1466 /// a metadata `filter`, e.g. the current project), then walk its typed links
1467 /// up to `max_hops` away — fusing the vector, `ColumnStore`, and graph facets.
1468 ///
1469 /// Returns an empty [`Explanation`] when nothing matches the decision.
1470 ///
1471 /// # Errors
1472 /// Returns [`MemoryError`] if recall or graph traversal fails.
1473 pub fn why(
1474 &self,
1475 decision: &str,
1476 max_hops: usize,
1477 filter: Option<&Metadata>,
1478 ) -> Result<Explanation, MemoryError> {
1479 let decision = decision.trim();
1480 if decision.is_empty() {
1481 return Ok(Explanation::default());
1482 }
1483 reject_reserved_keys(filter)?;
1484 let embedding = self.embedder.embed(decision)?;
1485 let seeds = self.search(&embedding, 1, filter)?;
1486 let Some((seed_id, _score, seed_content)) = seeds.into_iter().next() else {
1487 return Ok(Explanation::default());
1488 };
1489 self.traverse(seed_id, seed_content, max_hops)
1490 }
1491
1492 /// Breadth-first walk over outgoing links from `seed_id`, collecting nodes
1493 /// and edges up to `max_hops` away.
1494 fn traverse(
1495 &self,
1496 seed_id: u64,
1497 seed_content: String,
1498 max_hops: usize,
1499 ) -> Result<Explanation, MemoryError> {
1500 let mut explanation = Explanation {
1501 nodes: vec![MemoryNode {
1502 id: seed_id,
1503 content: seed_content,
1504 hop: 0,
1505 }],
1506 edges: Vec::new(),
1507 truncated: false,
1508 };
1509 let mut visited: HashSet<u64> = HashSet::from([seed_id]);
1510 let mut frontier = vec![seed_id];
1511 let mut next: Vec<u64> = Vec::new();
1512 'hops: for hop in 1..=max_hops {
1513 next.clear();
1514 for node_id in frontier.drain(..) {
1515 // Both width budgets, checked here AND inside `expand`: this
1516 // check alone would let the expansion that crosses the line
1517 // finish its node — up to MAX_WHY_NODE_DEGREE nodes past the
1518 // "ceiling", which a review measured at 522 of a promised 500.
1519 if explanation.nodes.len() >= crate::limits::MAX_WHY_NODES
1520 || explanation.edges.len() >= crate::limits::MAX_WHY_EDGES
1521 {
1522 // Unexpanded frontier work remained — the response is a
1523 // partial view and must SAY so (#1820); whether the rest
1524 // held anything unseen is exactly what the budget forbids
1525 // finding out, so the cautious true is the honest one.
1526 explanation.truncated = true;
1527 break 'hops; // width budget spent — depth left in max_hops is moot
1528 }
1529 self.expand(node_id, hop, &mut explanation, &mut visited, &mut next)?;
1530 }
1531 if next.is_empty() {
1532 break;
1533 }
1534 std::mem::swap(&mut frontier, &mut next);
1535 }
1536 Ok(explanation)
1537 }
1538
1539 /// Expand a single node: enqueue unseen targets and record edges, following
1540 /// at most [`crate::limits::MAX_WHY_NODE_DEGREE`] outgoing edges — an entity
1541 /// hub's degree scales with the whole store, so an unbounded walk here would
1542 /// dump its entire neighborhood into one response (issue #1743). An edge is
1543 /// only recorded once its target is a resolved node, so the subgraph never
1544 /// contains an edge pointing at a node absent from `nodes` (e.g. a forgotten
1545 /// target whose edge outlived it).
1546 fn expand(
1547 &self,
1548 node_id: u64,
1549 hop: usize,
1550 explanation: &mut Explanation,
1551 visited: &mut HashSet<u64>,
1552 next: &mut Vec<u64>,
1553 ) -> Result<(), MemoryError> {
1554 // The bounded read pushes the per-node budget into the store's own
1555 // index scan: the old full fetch materialized a super-node's whole
1556 // degree before `.take()` could apply — O(store size) transient
1557 // allocation at a single hop, the cost half of #1743 that #1820
1558 // closes. The store also reports whether the degree exceeded the
1559 // budget, which is what makes the cut OBSERVABLE.
1560 let bounded = self
1561 .store
1562 .relations_bounded(node_id, crate::limits::MAX_WHY_NODE_DEGREE)?;
1563 if bounded.truncated {
1564 explanation.truncated = true;
1565 }
1566 for edge in bounded.edges {
1567 // The budgets are ceilings, not suggestions: once either is spent,
1568 // this node's expansion stops MID-NODE rather than finishing. The
1569 // caller's check between nodes cannot provide that — an expansion
1570 // that crosses the line would otherwise add its whole degree.
1571 if explanation.nodes.len() >= crate::limits::MAX_WHY_NODES
1572 || explanation.edges.len() >= crate::limits::MAX_WHY_EDGES
1573 {
1574 // An edge was in hand and not followed — an exact cut, not
1575 // a conservative one.
1576 explanation.truncated = true;
1577 break;
1578 }
1579 let target = edge.to;
1580 if !visited.contains(&target) {
1581 let Some((content, _embedding)) = self.store.get(target)? else {
1582 continue; // target no longer exists → drop the dangling edge too
1583 };
1584 visited.insert(target);
1585 explanation.nodes.push(MemoryNode {
1586 id: target,
1587 content,
1588 hop,
1589 });
1590 next.push(target);
1591 }
1592 explanation.edges.push(edge);
1593 }
1594 Ok(())
1595 }
1596}
1597
1598/// The metadata filter that excludes entity hubs from unfiltered recall and
1599/// `why` seeds — the negative counterpart [`MemoryService::search`] applies so
1600/// internal `_veles_hub` scaffolding never surfaces as a result.
1601fn hub_exclude_filter() -> Metadata {
1602 let mut exclude = Map::new();
1603 exclude.insert(HUB_FIELD.to_string(), Value::Bool(true));
1604 exclude
1605}
1606
1607/// Reject caller-supplied metadata/filters that name a reserved key.
1608fn reject_reserved_keys(metadata: Option<&Metadata>) -> Result<(), MemoryError> {
1609 let Some(meta) = metadata else {
1610 return Ok(());
1611 };
1612 for key in meta.keys() {
1613 if is_reserved_key(key) {
1614 return Err(MemoryError::ReservedKey(key.clone()));
1615 }
1616 }
1617 Ok(())
1618}
1619
1620/// Reject caller-supplied metadata over [`crate::limits::MAX_METADATA_BYTES`]
1621/// — the `DoS` guard every `remember` path shares (see
1622/// [`MemoryError::MetadataTooLarge`]).
1623fn reject_oversized_metadata(metadata: Option<&Metadata>) -> Result<(), MemoryError> {
1624 let Some(meta) = metadata else {
1625 return Ok(());
1626 };
1627 let bytes = crate::limits::metadata_bytes(meta);
1628 if bytes > crate::limits::MAX_METADATA_BYTES {
1629 return Err(MemoryError::MetadataTooLarge {
1630 bytes,
1631 max: crate::limits::MAX_METADATA_BYTES,
1632 });
1633 }
1634 Ok(())
1635}
1636
1637/// Normalise a TTL supplied as *configuration*: `Some(0)` (and `None`) mean
1638/// "no TTL policy" — the fact is stored permanently. Any positive value is
1639/// kept as-is.
1640///
1641/// Deliberately NOT applied to `remember`'s own `ttl_seconds` any more: an
1642/// explicit per-call `0` is an intent about one fact ("expire it"), and
1643/// silently turning that into "permanent" is the opposite (see
1644/// [`reject_zero_ttl`]). A compile policy's `source_ttl_seconds`, on the
1645/// other hand, is a knob about a whole server, where `0` reading as "no
1646/// policy" is the ordinary, unsurprising meaning.
1647///
1648/// Gated on `context`: since `remember` stopped calling it, the compile
1649/// policy in `context::memory_bridge` is its only caller, and a build
1650/// without that feature saw dead code — which `-D warnings` turns into a
1651/// failed build, not a warning.
1652#[cfg(feature = "context")]
1653pub(crate) fn positive_ttl(ttl_seconds: Option<u64>) -> Option<u64> {
1654 ttl_seconds.filter(|&seconds| seconds > 0)
1655}
1656
1657/// The canonical form of an entity name: trimmed and lowercased, exactly as
1658/// an extracted entity is keyed.
1659///
1660/// Public because a lookup MISS has to echo it too: an adapter that answered
1661/// `name: ""` when nothing matched left a caller running several lookups
1662/// unable to pair a response with its question (issue #1654). Hit and miss go
1663/// through this one function, so the two can never drift.
1664/// The audit's per-fact visibility policy, in one place: `None` skips the
1665/// fact (internal scaffolding under the default view, or a metadata filter
1666/// miss), `Some` carries what the caller may see — reserved keys stripped
1667/// exactly as recall strips them, or the raw payload under
1668/// `include_internal`.
1669fn audited(
1670 fact: crate::storage::RawListedFact,
1671 filter: Option<&Metadata>,
1672 include_internal: bool,
1673) -> Option<crate::model::ListedMemory> {
1674 if !include_internal && crate::storage::is_internal_scaffolding(&fact.payload) {
1675 return None;
1676 }
1677 let matches = filter.is_none_or(|wanted| {
1678 wanted
1679 .iter()
1680 .all(|(key, value)| fact.payload.get(key) == Some(value))
1681 });
1682 if !matches {
1683 return None;
1684 }
1685 let metadata = if include_internal {
1686 (!fact.payload.is_empty()).then_some(fact.payload)
1687 } else {
1688 strip_reserved_keys(Some(fact.payload))
1689 };
1690 Some(crate::model::ListedMemory {
1691 id: fact.id,
1692 content: fact.content,
1693 metadata,
1694 })
1695}
1696
1697#[must_use]
1698pub fn canonical_entity_name(name: &str) -> String {
1699 name.trim().to_lowercase()
1700}
1701
1702/// Refuse a fact that cannot be stored as written: blank, or past the size an
1703/// embedding model still accepts.
1704///
1705/// The size check runs BEFORE [`MemoryService::write_fact`] calls the
1706/// embedder, so an over-long fact is reported with its own size and the cap
1707/// instead of whatever the backend says — issue #1654 saw `ollama embeddings
1708/// call failed`, which names neither.
1709fn validate_fact(fact: &str) -> Result<(), MemoryError> {
1710 if fact.is_empty() {
1711 return Err(MemoryError::EmptyFact);
1712 }
1713 validate_embeddable(fact)
1714}
1715
1716/// Refuse a text past the size an embedding model still accepts.
1717///
1718/// Extracted from [`validate_fact`] so every path that embeds CALLER content
1719/// answers to the same cap: `remember` refuses (this function), while the
1720/// context bridge truncates via [`embeddable_prefix`] — but neither may hand
1721/// the backend an oversized text and relay its raw failure, which is how
1722/// issue #1654's `ollama embeddings call failed` (naming neither size nor
1723/// cap) reached users.
1724pub(crate) fn validate_embeddable(text: &str) -> Result<(), MemoryError> {
1725 if text.len() > crate::limits::MAX_EMBEDDABLE_TEXT_BYTES {
1726 return Err(MemoryError::FactTooLarge {
1727 bytes: text.len(),
1728 max: crate::limits::MAX_EMBEDDABLE_TEXT_BYTES,
1729 });
1730 }
1731 Ok(())
1732}
1733
1734/// The longest prefix of `text` that fits the embeddable cap without
1735/// splitting a UTF-8 character.
1736///
1737/// For content that must be STORED whole but whose vector only serves
1738/// similarity search (context sources: retrieval is hash-addressed, the
1739/// vector is a ranking aid), truncating the *embedded* text is the correct
1740/// trade — refusing would fail a legitimate compile, and a placeholder
1741/// vector would remove the source from semantic recall entirely.
1742///
1743/// Gated on `context`: the compiler's source writer is its only caller, so a
1744/// build without that feature sees dead code, which CI's `-D warnings`
1745/// turns into a failed build. Same shape as `positive_ttl` — and only the
1746/// per-feature ISOLATION loop catches it, never a feature combination.
1747#[cfg(feature = "context")]
1748pub(crate) fn embeddable_prefix(text: &str) -> &str {
1749 let cap = crate::limits::MAX_EMBEDDABLE_TEXT_BYTES;
1750 if text.len() <= cap {
1751 return text;
1752 }
1753 let mut end = cap;
1754 while end > 0 && !text.is_char_boundary(end) {
1755 end -= 1;
1756 }
1757 &text[..end]
1758}
1759
1760/// Refuse an explicit per-call TTL of `0`.
1761///
1762/// `0` used to be normalised to "no expiry", so a caller who meant "expire
1763/// immediately" silently got a **permanent** fact — the opposite intent, with
1764/// no signal (issue #1654). A TTL supplied as *configuration*
1765/// (`McpServer::with_default_ttl`, a compile policy's `source_ttl_seconds`)
1766/// still reads `0` as "no TTL policy": that is a default about a whole
1767/// server, not an intent about one fact, and it is deliberately untouched.
1768fn reject_zero_ttl(ttl_seconds: Option<u64>) -> Result<(), MemoryError> {
1769 if ttl_seconds == Some(0) {
1770 return Err(MemoryError::ZeroTtl);
1771 }
1772 Ok(())
1773}
1774
1775/// Refuse a `remember` link that points the fact at itself.
1776///
1777/// The same rule [`MemoryService::relate`] enforces, applied to the other way
1778/// a self-loop can be created: re-remembering existing content yields its
1779/// existing id, so a caller CAN name that id as a link target. Without this
1780/// the `relate` guard would only close half the door.
1781fn reject_self_links(fact_id: u64, links: &[Link]) -> Result<(), MemoryError> {
1782 if links.iter().any(|link| link.target == fact_id) {
1783 return Err(MemoryError::SelfRelation(fact_id));
1784 }
1785 Ok(())
1786}
1787
1788/// [`MemoryService::remember_with_ttl`]'s auto-date stamp: `metadata` with
1789/// today's date added under [`AUTO_DATE_FIELD`], unless `metadata` already
1790/// names that key (an explicit, possibly retroactive, caller value is never
1791/// overwritten) or no clock is available ([`clock::today_ymd`] returns `None`
1792/// on `wasm32-unknown-unknown`). Returns an owned map either way, `None` only
1793/// when there is nothing to store at all (no caller metadata AND no clock).
1794fn stamp_with_today(metadata: Option<&Metadata>) -> Option<Metadata> {
1795 if metadata.is_some_and(|meta| meta.contains_key(AUTO_DATE_FIELD)) {
1796 return metadata.cloned();
1797 }
1798 let Some(today) = clock::today_ymd() else {
1799 return metadata.cloned();
1800 };
1801 let mut stamped = metadata.cloned().unwrap_or_default();
1802 stamped.insert(AUTO_DATE_FIELD.to_owned(), Value::from(today));
1803 Some(stamped)
1804}
1805
1806/// Maximum byte length for a relation label (prevents oversized graph edge labels
1807/// from reaching the storage layer).
1808const MAX_RELATION_BYTES: usize = 512;
1809
1810/// Validate a caller-supplied relation label: non-empty, within the size cap, and
1811/// containing only printable, non-control ASCII characters (32–126) or non-ASCII
1812/// Unicode. This prevents null bytes and control characters from reaching the
1813/// storage layer while permitting natural-language labels like `"decided_in"` or
1814/// `"is a friend of"`.
1815fn validate_relation(label: &str) -> Result<(), MemoryError> {
1816 if label.is_empty() {
1817 return Err(MemoryError::InvalidRelation(
1818 "relation label must not be empty".to_owned(),
1819 ));
1820 }
1821 if label.len() > MAX_RELATION_BYTES {
1822 return Err(MemoryError::InvalidRelation(format!(
1823 "relation label exceeds maximum of {MAX_RELATION_BYTES} bytes ({} given)",
1824 label.len()
1825 )));
1826 }
1827 if label.chars().any(|c| c.is_ascii_control()) {
1828 return Err(MemoryError::InvalidRelation(
1829 "relation label must not contain ASCII control characters".to_owned(),
1830 ));
1831 }
1832 Ok(())
1833}