Skip to main content

plugmem_core/
memory.rs

1//! The engine: data model state, verbs, journaling and replay
2//! (/03, /05).
3//!
4//! All state lives in the flat structures of `plugmem-arena`; every
5//! mutating verb journals itself through the caller's
6//! [`Storage`] and is replayed through the same
7//! internal `apply_*` path on open — the journal's ids are authoritative,
8//! so replay is deterministic and idempotent over a reapplied tail.
9//!
10//! The engine carries the structural sources (facts, entities, edges,
11//! tags, BM25, temporal) and — when `Config::dim > 0` — the vector layer:
12//! quantized vectors in a flat [`VecPool`], searched as a fourth recall
13//! source and used by similar-detection. `maintain` (compaction) still
14//! lands separately.
15
16use alloc::format;
17use alloc::string::String;
18use alloc::vec::Vec;
19
20use plugmem_arena::{
21    Arena, ArenaCfg, BlobHeap, BlobHeapCfg, BlobId, ChunkPool, ChunkPoolCfg, Interner, ListHandle,
22    ShardMode, Slot, TermId, key,
23};
24
25use crate::config::Config;
26use crate::error::Error;
27use crate::id::{EdgeId, EntityId, FactId, NONE_U32};
28use crate::index::IdListIndex;
29use crate::index::bm25::Bm25Index;
30use crate::index::hnsw::HnswGraph;
31use crate::index::vecpool::VecPool;
32use crate::journal::{JournalScan, Op, scan};
33use crate::model::{
34    EdgeHistorySlot, EdgeSlot, EntityByName, EntityRecord, FactAux, FactRecord, TemporalSlot,
35    VALID_TO_OPEN, close_edge_history_payload, edge_history_key, edge_key, fact_flags,
36};
37use crate::storage::Storage;
38use crate::tokenizer::Tokenizer;
39
40use maintain::TOKENIZER_INDEX_VERSION;
41
42/// Most recent same-entity facts examined by similar-detection.
43const SIMILAR_CANDIDATE_CAP: usize = 32;
44
45mod maintain;
46mod migrations;
47mod persist;
48mod recall;
49mod shards;
50
51pub use maintain::{MaintainReport, MaintenanceMode, MaintenanceOptions};
52pub use recall::{RecallQuery, RecallResult, RecallScratch, RecalledEdge, RecalledFact, source};
53pub use shards::ShardLayout;
54
55/// Input of `remember` and `revise`.
56#[derive(Clone, Copy, Debug)]
57#[cfg_attr(feature = "serde", derive(serde::Serialize))]
58pub struct RememberInput<'a> {
59    /// Host timestamp, unix milliseconds.
60    pub now: u64,
61    /// Fact text, ≤ `Config::max_text` bytes.
62    pub text: &'a str,
63    /// Subject entity name; created lazily on first mention.
64    pub entity: Option<&'a str>,
65    /// Tags, verbatim strings, ≤ 32.
66    pub tags: &'a [&'a str],
67    /// `(rel, target_entity)` pairs, ≤ 16; edges go subject → target.
68    pub links: &'a [(&'a str, &'a str)],
69    /// Optional embedding, `len == Config::dim`; quantized on the way in.
70    /// Requires `dim > 0` and is dropped by the engine's own re-quantized
71    /// replay, so nothing float-nondeterministic reaches the state.
72    pub vector: Option<&'a [f32]>,
73    /// Validity start; defaults to `now`.
74    pub valid_from: Option<u64>,
75    /// Optional metadata as key→value pairs (UTF-8). Passed in any order; the
76    /// engine canonicalizes (sorts keys, rejects duplicates) and stores them as
77    /// one opaque blob (see [`crate::metadata`]). `None` or an empty slice = no
78    /// metadata. The engine never interprets the values.
79    pub metadata: Option<&'a [(&'a str, &'a str)]>,
80}
81
82impl<'a> RememberInput<'a> {
83    /// A minimal input: text only.
84    pub fn text(now: u64, text: &'a str) -> Self {
85        Self {
86            now,
87            text,
88            entity: None,
89            tags: &[],
90            links: &[],
91            vector: None,
92            valid_from: None,
93            metadata: None,
94        }
95    }
96}
97
98/// Input of `link`.
99#[derive(Clone, Copy, Debug)]
100#[cfg_attr(feature = "serde", derive(serde::Serialize))]
101pub struct LinkInput<'a> {
102    /// Host timestamp, unix milliseconds.
103    pub now: u64,
104    /// Source entity name (created lazily).
105    pub src: &'a str,
106    /// Relation term, verbatim (`"works_at"`, …).
107    pub rel: &'a str,
108    /// Destination entity name (created lazily).
109    pub dst: &'a str,
110    /// Optional provenance fact.
111    pub provenance: Option<FactId>,
112}
113
114/// Input of `unlink`.
115#[derive(Clone, Copy, Debug)]
116#[cfg_attr(feature = "serde", derive(serde::Serialize))]
117pub struct UnlinkInput<'a> {
118    /// Host timestamp, unix milliseconds.
119    pub now: u64,
120    /// Source entity name.
121    pub src: &'a str,
122    /// Relation term, verbatim.
123    pub rel: &'a str,
124    /// Destination entity name.
125    pub dst: &'a str,
126}
127
128/// Result of `remember`/`revise`.
129#[derive(Clone, Debug, PartialEq)]
130#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
131pub struct RememberOutcome {
132    /// The new fact's id.
133    pub id: FactId,
134    /// The subject entity (resolved or created), if one was named.
135    pub entity: Option<EntityId>,
136    /// Similar / potentially conflicting **live** facts, best match
137    /// first (≤ 8). The engine never revises on its own — it surfaces
138    /// the candidates, the agent judges: `revise`, keep both, or
139    /// `forget`.
140    pub similar: Vec<Similar>,
141}
142
143/// One similar-fact hint.
144#[derive(Clone, Copy, Debug, PartialEq)]
145#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
146pub struct Similar {
147    /// The existing fact.
148    pub id: FactId,
149    /// Match strength (for the lexical detector: the Jaccard overlap of
150    /// the term sets, in `(similar_jaccard, 1]`).
151    pub score: f32,
152    /// What triggered the hint.
153    pub reason: SimilarReason,
154}
155
156/// Why a fact was flagged as similar.
157#[derive(Clone, Copy, Debug, PartialEq, Eq)]
158#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
159pub enum SimilarReason {
160    /// Same subject entity and a term-set overlap above the configured
161    /// Jaccard threshold.
162    LexicalOverlap,
163    /// Same subject entity and a quantized-vector cosine above the
164    /// configured `similar_cos` threshold.
165    VectorCosine,
166}
167
168/// A read view of one fact.
169#[derive(Clone, Copy, Debug)]
170#[cfg_attr(feature = "serde", derive(serde::Serialize))]
171pub struct FactView<'a> {
172    /// The raw record (temporality, flags, references).
173    pub record: FactRecord,
174    /// The fact text.
175    pub text: &'a str,
176}
177
178/// The per-fact content problem [`Memory::faulty_facts`] attributes to a
179/// fact — the salvage predicate `recover` uses. It mirrors the
180/// content checks [`Memory::verify`] runs, split out per fact so a caller can
181/// drop the individual bad records instead of failing the whole image.
182#[derive(Clone, Copy, Debug, PartialEq, Eq)]
183#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
184pub enum FactFault {
185    /// The fact's stored text is not valid UTF-8.
186    Text,
187    /// The fact is flagged with a vector but its slot is out of range or does
188    /// not name the fact back (the fact↔slot bijection is broken).
189    Vector,
190    /// The fact's metadata blob is out of range or does not decode to a
191    /// well-formed key→value map.
192    Metadata,
193}
194
195/// Engine size counters. Every field is an O(1) read; the
196/// struct is `#[non_exhaustive]` so later stages (database identity
197/// markers, HNSW state) can extend it without a breaking change.
198#[derive(Clone, Copy, Debug, PartialEq, Eq)]
199#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
200#[non_exhaustive]
201pub struct Stats {
202    /// Fact records currently stored (live, closed, and tombstoned facts
203    /// awaiting `maintain`).
204    pub facts: usize,
205    /// Entities.
206    pub entities: usize,
207    /// Interned terms (tokens, tags, relations, normalized names).
208    pub terms: usize,
209    /// Directed edges (each `(src, rel, dst)` counted once; the mirrored
210    /// in-arena is an internal detail).
211    pub edges: usize,
212    /// Historical edge versions, including closed versions.
213    pub edge_versions: usize,
214    /// Quantized vector slots.
215    pub vectors: usize,
216    /// Tombstoned fact records awaiting physical purge.
217    pub tombstones: usize,
218    /// Vector slots already covered by the HNSW graph; slots after this are
219    /// searched as the flat tail.
220    pub hnsw_indexed: u32,
221    /// The next fact id to be assigned. Ids below it are in use or burned
222    /// (forgotten and purged) — never reissued.
223    pub next_fact: u32,
224    /// The next entity id to be assigned.
225    pub next_entity: u32,
226    /// The next edge-version id to be assigned.
227    pub next_edge: u32,
228    /// The database lineage identity ([`Config::db_uuid`]); `0` for an
229    /// unnamed database.
230    pub db_uuid: u128,
231    /// Total bytes held by the engine's pools (arenas, blob heaps, chunk
232    /// pools, the term dictionary and the vector pool).
233    pub pool_bytes: usize,
234    /// How the arenas are currently sharded.
235    ///
236    /// The engine chooses this from what it holds and moves it during
237    /// `maintain`, so it is state to observe rather than a setting to pick —
238    /// see [`Config::shards_facts`](crate::Config::shards_facts).
239    pub shards: ShardLayout,
240}
241
242/// Report of an `open`: what the journal replay found.
243#[derive(Clone, Debug, Default, PartialEq, Eq)]
244#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
245pub struct OpenReport {
246    /// Journal records applied.
247    pub replayed: usize,
248    /// Journal records skipped as already contained in the snapshot.
249    pub skipped: usize,
250    /// A torn tail record was dropped (crash between appends).
251    pub truncated_tail: bool,
252}
253
254/// The memory engine. See the module docs for staging.
255///
256/// The lifetime `'a` is the provenance of the engine's byte pools. Every
257/// owned constructor (`new`/`open`/`from_bytes`) yields a `Memory<'static>`
258/// that owns its bytes; the read-only [`Memory::from_bytes_borrowed`] path
259/// borrows them from an mmap'd snapshot instead of copying.
260pub struct Memory<'a> {
261    cfg: Config,
262    // -- data model --
263    facts: Arena<'a, FactRecord>,
264    fact_aux: Arena<'a, FactAux>,
265    entities: Arena<'a, EntityRecord>,
266    by_name: Arena<'a, EntityByName>,
267    edges_out: Arena<'a, EdgeSlot>,
268    edges_in: Arena<'a, EdgeSlot>,
269    edges_hist_out: Arena<'a, EdgeHistorySlot>,
270    edges_hist_in: Arena<'a, EdgeHistorySlot>,
271    temporal: Arena<'a, TemporalSlot>,
272    /// Fact texts and canonical entity names.
273    texts: BlobHeap<'a>,
274    /// Per-fact metadata blobs (canonical key→value, see [`crate::metadata`]),
275    /// referenced by [`FactAux::meta`]. Empty and inert until a fact carries
276    /// metadata; kept out of `texts` so this cold pool stays non-resident on an
277    /// mmap'd base until `show`/`export` touches it.
278    metas: BlobHeap<'a>,
279    /// Terms: tokens, tags (verbatim), relation names, normalized entity
280    /// names.
281    terms: Interner<'a>,
282    /// Per-fact tag lists (`TermId` values), handles in [`FactAux`].
283    tag_lists: ChunkPool<'a>,
284    // -- indexes --
285    bm25: Bm25Index<'a>,
286    tags_idx: IdListIndex<'a>,
287    entity_facts: IdListIndex<'a>,
288    /// Quantized vectors (empty and inert when `cfg.dim == 0`).
289    vecs: VecPool<'a>,
290    /// HNSW graph over the pool; empty until `maintain` crosses
291    /// `Config::flat_to_hnsw`. Slots past its `indexed` mark are the flat
292    /// tail searched by scan.
293    hnsw: HnswGraph<'a>,
294    // -- id allocation (derived from the arenas on load) --
295    next_fact: u32,
296    next_entity: u32,
297    next_edge: u32,
298    // -- maintenance state --
299    tombstones: usize,
300    bm25_tokenizer_version: u32,
301    // -- reusable scratches --
302    tokenizer: Tokenizer,
303    tf_scratch: Vec<(u32, u8)>,
304    name_scratch: String,
305}
306
307impl<'a> Memory<'a> {
308    /// Creates an empty database.
309    ///
310    /// # Errors
311    ///
312    /// [`Error::ConfigMismatch`] for an invalid config (see
313    /// [`Config::validate`]).
314    pub fn new(cfg: Config) -> Result<Self, Error> {
315        cfg.validate()?;
316        let uni =
317            |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
318        let ord =
319            |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
320        let blob = BlobHeapCfg::new()
321            .with_max_bytes(cfg.max_bytes)
322            .with_max_blob(cfg.max_blob);
323        Ok(Self {
324            facts: Arena::new(uni(cfg.shards_facts))?,
325            fact_aux: Arena::new(uni(cfg.shards_facts))?,
326            entities: Arena::new(uni(cfg.shards_entities))?,
327            by_name: Arena::new(ord(cfg.shards_entities))?,
328            edges_out: Arena::new(ord(cfg.shards_edges))?,
329            edges_in: Arena::new(ord(cfg.shards_edges))?,
330            edges_hist_out: Arena::new(ord(cfg.shards_edges))?,
331            edges_hist_in: Arena::new(ord(cfg.shards_edges))?,
332            temporal: Arena::new(ord(cfg.shards_temporal))?,
333            texts: BlobHeap::new(blob),
334            metas: BlobHeap::new(blob),
335            terms: Interner::new(blob),
336            tag_lists: ChunkPool::new(ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes)),
337            bm25: Bm25Index::new(cfg.shards_postings, cfg.max_bytes)?,
338            tags_idx: IdListIndex::new(cfg.shards_postings, cfg.max_bytes)?,
339            entity_facts: IdListIndex::new(cfg.shards_entities, cfg.max_bytes)?,
340            vecs: VecPool::new(cfg.dim, cfg.max_bytes),
341            hnsw: HnswGraph::new(cfg.hnsw_m, cfg.hnsw_m0, cfg.max_bytes)?,
342            next_fact: 0,
343            next_entity: 0,
344            next_edge: 0,
345            tombstones: 0,
346            bm25_tokenizer_version: maintain::TOKENIZER_INDEX_VERSION,
347            tokenizer: Tokenizer::new(),
348            tf_scratch: Vec::new(),
349            name_scratch: String::new(),
350            cfg,
351        })
352    }
353
354    /// Opens a database from a storage: journal replay over an empty
355    /// state (the snapshot fast-path lands with the section composition;
356    /// a non-empty snapshot is rejected rather than half-read).
357    pub fn open<S: Storage>(store: &mut S, cfg: Config) -> Result<(Self, OpenReport), Error> {
358        let snapshot = store
359            .read_snapshot()
360            .map_err(|e| Error::Storage(format!("{e:?}")))?;
361        let journal = store
362            .read_journal()
363            .map_err(|e| Error::Storage(format!("{e:?}")))?;
364        Self::from_bytes(snapshot.as_deref(), &journal, cfg)
365    }
366
367    /// Opens from raw bytes (the wasm path: the host already read them).
368    pub fn from_bytes(
369        snapshot: Option<&[u8]>,
370        journal: &[u8],
371        cfg: Config,
372    ) -> Result<(Self, OpenReport), Error> {
373        let mut mem = match snapshot {
374            Some(bytes) => Self::load_snapshot(bytes, cfg)?,
375            None => Self::new(cfg)?,
376        };
377        let report = mem.replay(journal)?;
378        Ok((mem, report))
379    }
380
381    /// Opens a database read-only over a borrowed snapshot: the
382    /// engine's byte pools borrow `snapshot` (typically an mmap'd file)
383    /// instead of copying it, so a large database residents only the pages
384    /// actually read. The returned engine is tied to `snapshot`'s lifetime.
385    ///
386    /// A **non-empty journal is rejected**: the read-only handle exposes no
387    /// write verbs, so a snapshot with un-checkpointed journal tail would open
388    /// stale. Checkpoint the database read-write once first — or use
389    /// [`Memory::from_bytes_overlay`], which replays the journal into the
390    /// overlay. The caller (the host) owns the map and the exclusive lock.
391    pub fn from_bytes_borrowed(
392        snapshot: &'a [u8],
393        journal: &[u8],
394        cfg: Config,
395    ) -> Result<Self, Error> {
396        let mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
397        let JournalScan { entries, .. } = scan(journal)?;
398        if !entries.is_empty() {
399            return Err(Error::Invalid(
400                "read-only open requires a checkpointed (empty) journal",
401            ));
402        }
403        Ok(mem)
404    }
405
406    /// Opens a database read-**write** over a borrowed snapshot (the overlay
407    /// write path): like [`Memory::from_bytes_borrowed`] the byte
408    /// pools borrow `snapshot` instead of copying it, but the journal is
409    /// **replayed** and the engine stays fully mutable. Mutations do not clone
410    /// the borrowed base: the flat structures land appends in an owned tail and
411    /// copy only the pages they rewrite (per-page copy-on-write), so a
412    /// multi-gigabyte mapped database is opened and written to while resident
413    /// only in the pages it actually touches.
414    ///
415    /// This is the write sibling of [`Memory::from_bytes`] (which owns its
416    /// bytes) — the same `snapshot + journal replay`, over a borrowed base, and
417    /// it returns the same [`OpenReport`] describing the replay. The returned
418    /// engine is tied to `snapshot`'s lifetime; the caller (the host) owns the
419    /// map and the exclusive lock.
420    pub fn from_bytes_overlay(
421        snapshot: &'a [u8],
422        journal: &[u8],
423        cfg: Config,
424    ) -> Result<(Self, OpenReport), Error> {
425        let mut mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
426        let report = mem.replay(journal)?;
427        Ok((mem, report))
428    }
429
430    /// Applies every journal record on top of the current state (
431    /// replay rules: assigned ids are authoritative; records whose
432    /// assigned id is already present are skipped, which makes reapplying
433    /// a tail idempotent).
434    fn replay(&mut self, journal: &[u8]) -> Result<OpenReport, Error> {
435        let JournalScan {
436            entries,
437            truncated_tail,
438        } = scan(journal)?;
439        let mut report = OpenReport {
440            truncated_tail,
441            ..OpenReport::default()
442        };
443        for entry in entries {
444            let op = Op::decode(entry.op, entry.payload)?;
445            match op {
446                Op::Remember {
447                    now,
448                    valid_from,
449                    entity,
450                    text,
451                    ref tags,
452                    ref links,
453                    ref vector,
454                    ref metadata,
455                    revises,
456                    assigned,
457                } => {
458                    if assigned.0 < self.next_fact {
459                        report.skipped += 1;
460                        continue;
461                    }
462                    if assigned.0 != self.next_fact {
463                        return Err(Error::Corrupt("journal fact ids are not contiguous"));
464                    }
465                    if !vector.is_empty() && vector.len() != self.cfg.dim {
466                        return Err(Error::Corrupt(
467                            "journal vector dimension disagrees with dim",
468                        ));
469                    }
470                    if let Some(target) = revises.some() {
471                        self.check_revisable(target)
472                            .map_err(|_| Error::Corrupt("journal revises an unrevisable fact"))?;
473                    }
474                    self.apply_remember(
475                        &RememberInput {
476                            now,
477                            text,
478                            entity,
479                            tags: &tags.to_vec(),
480                            links: &links.to_vec(),
481                            vector: (!vector.is_empty()).then_some(vector.as_slice()),
482                            valid_from: Some(valid_from),
483                            metadata: (!metadata.is_empty()).then_some(metadata.as_slice()),
484                        },
485                        revises,
486                    )?;
487                    if let Some(target) = revises.some() {
488                        self.close_target(target, valid_from);
489                    }
490                    report.replayed += 1;
491                }
492                Op::Forget { fact, .. } => {
493                    // Idempotent; a missing fact in a valid journal is
494                    // corruption, but re-tombstoning is fine.
495                    match self.apply_forget(fact) {
496                        Ok(_) => report.replayed += 1,
497                        Err(Error::NotFound(_)) => {
498                            return Err(Error::Corrupt("journal forgets an unknown fact"));
499                        }
500                        Err(e) => return Err(e),
501                    }
502                }
503                Op::Link {
504                    now,
505                    src,
506                    rel,
507                    dst,
508                    provenance,
509                } => {
510                    self.apply_link(now, src, rel, dst, provenance)?;
511                    report.replayed += 1;
512                }
513                Op::Unlink { now, src, rel, dst } => {
514                    self.apply_unlink(now, src, rel, dst)?;
515                    report.replayed += 1;
516                }
517                Op::Maintain {
518                    mode,
519                    max_hnsw_inserts,
520                    ..
521                } => {
522                    // Re-execute the compaction deterministically, so a
523                    // replayed image matches one snapshotted after a live
524                    // maintain byte for byte.
525                    let options =
526                        maintain::MaintenanceOptions::from_journal(mode, max_hnsw_inserts)?;
527                    self.replay_maintain_with_options(options)?;
528                    report.replayed += 1;
529                }
530            }
531        }
532        Ok(report)
533    }
534
535    /// Remembers a new fact.
536    pub fn remember<S: Storage>(
537        &mut self,
538        store: &mut S,
539        input: RememberInput<'_>,
540    ) -> Result<RememberOutcome, Error> {
541        self.validate_input(&input)?;
542        let mut outcome = self.apply_remember(&input, FactId::NONE)?;
543        self.find_similar(&mut outcome);
544        self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
545        Ok(outcome)
546    }
547
548    /// Batch import: a sequence of remembers in one call, journaled
549    /// individually. `skip_similar` turns off the
550    /// similar-detection pass — imports don't need hints, and skipping
551    /// them makes a bulk load cheaper.
552    ///
553    /// Stops at the first error; already-applied inputs stay applied and
554    /// journaled (replay reproduces exactly the applied prefix).
555    pub fn remember_batch<S: Storage>(
556        &mut self,
557        store: &mut S,
558        inputs: &[RememberInput<'_>],
559        skip_similar: bool,
560    ) -> Result<Vec<RememberOutcome>, Error> {
561        let mut out = Vec::with_capacity(inputs.len());
562        for input in inputs {
563            self.validate_input(input)?;
564            let mut outcome = self.apply_remember(input, FactId::NONE)?;
565            if !skip_similar {
566                self.find_similar(&mut outcome);
567            }
568            self.journal_remember(store, input, FactId::NONE, outcome.id)?;
569            out.push(outcome);
570        }
571        Ok(out)
572    }
573
574    /// Lexical similar-detection: live facts of the same
575    /// entity whose term sets overlap the new fact's above the
576    /// `similar_jaccard` threshold. Bounded: the entity's most recent
577    /// [`SIMILAR_CANDIDATE_CAP`] facts are compared (a hub's full list is
578    /// not re-tokenized).
579    ///
580    /// Comparing two term sets exactly means having both, and only the new
581    /// fact's is at hand — the candidate's would have to be recovered by
582    /// reading its text and running it back through the tokenizer. Doing that
583    /// for every candidate of every write is what the term-set summary in
584    /// [`DocLenSlot`](crate::index::bm25::DocLenSlot) exists to avoid: it
585    /// bounds the overlap from above, and a bound below the threshold settles
586    /// the question, because Jaccard rises with the intersection. Only a
587    /// candidate that survives the bound is read and tokenized, so the
588    /// answer is the same one the exhaustive comparison gives.
589    ///
590    /// The bound is trusted only while the index was built by the current
591    /// tokenizer. A stale index may hold terms today's tokenizer would not
592    /// produce, which would make the summary describe a different term set
593    /// than the comparison uses; then every candidate is read, exactly as
594    /// before the summary existed.
595    fn find_similar(&mut self, outcome: &mut RememberOutcome) {
596        let Some(entity) = outcome.entity else { return };
597        // The new fact's term set is still in tf_scratch (apply_remember
598        // filled it); snapshot the term ids.
599        let new_terms: Vec<u32> = self.tf_scratch.iter().map(|&(t, _)| t).collect();
600        // The new fact's vector slot, if it has one (for the cosine signal).
601        let new_vec = self
602            .fact(outcome.id)
603            .filter(|r| r.has_vector())
604            .map(|r| r.vector);
605        if new_terms.is_empty() && new_vec.is_none() {
606            return;
607        }
608        // Most recent candidates of the entity (ring over the list).
609        let mut ring = [FactId::NONE; SIMILAR_CANDIDATE_CAP];
610        let mut n = 0usize;
611        for (fact, _) in self.entity_facts.entries(entity.0) {
612            if fact != outcome.id {
613                ring[n % SIMILAR_CANDIDATE_CAP] = fact;
614                n += 1;
615            }
616        }
617        let summaries_trustworthy = self.bm25_tokenizer_version == TOKENIZER_INDEX_VERSION;
618        // Without a vector on the new fact, a lexical overlap is the only hint
619        // a candidate can produce, so one ruled out by the summary contributes
620        // nothing whatever its record says — and its record is the second
621        // arena lookup of the pair this loop would otherwise pay per
622        // candidate. (`new_terms` is non-empty here: the two being empty
623        // together returned above.)
624        let lexical_only = new_vec.is_none();
625        let mut cand_terms: Vec<u32> = Vec::new();
626        for &fact in ring.iter().take(n.min(SIMILAR_CANDIDATE_CAP)) {
627            let may_overlap = !new_terms.is_empty()
628                && self.overlap_possible(fact, &new_terms, summaries_trustworthy);
629            if lexical_only && !may_overlap {
630                continue;
631            }
632            let Some(record) = self.fact(fact) else {
633                continue;
634            };
635            if record.is_tombstone() || record.is_closed() {
636                continue;
637            }
638
639            // Lexical signal: term-set Jaccard over the entity's text.
640            let mut lexical = None;
641            // Deferred validation: a load no longer scans the
642            // text pool, so an unreadable text simply yields no lexical signal.
643            if may_overlap && let Ok(text) = core::str::from_utf8(self.texts.get(record.text)) {
644                cand_terms.clear();
645                let terms = &self.terms;
646                let cand = &mut cand_terms;
647                self.tokenizer.tokenize(text, &mut |token| {
648                    if let Some(term) = terms.lookup(token)
649                        && !cand.contains(&term.0)
650                    {
651                        cand.push(term.0);
652                    }
653                });
654                if !cand_terms.is_empty() {
655                    let both = cand_terms.iter().filter(|t| new_terms.contains(t)).count();
656                    let union = cand_terms.len() + new_terms.len() - both;
657                    let jaccard = both as f32 / union as f32;
658                    if jaccard > self.cfg.similar_jaccard {
659                        lexical = Some(jaccard);
660                    }
661                }
662            }
663
664            // Vector signal: quantized cosine when both facts carry one.
665            let mut vector = None;
666            if let (Some(a), true) = (new_vec, record.has_vector()) {
667                let cos = self.vecs.cosine_slots(a, record.vector);
668                if cos > self.cfg.similar_cos {
669                    vector = Some(cos);
670                }
671            }
672
673            // Keep the stronger signal; a tie prefers the lexical reason.
674            let best = match (lexical, vector) {
675                (Some(l), Some(v)) if v > l => Some((v, SimilarReason::VectorCosine)),
676                (Some(l), _) => Some((l, SimilarReason::LexicalOverlap)),
677                (None, Some(v)) => Some((v, SimilarReason::VectorCosine)),
678                (None, None) => None,
679            };
680            if let Some((score, reason)) = best {
681                outcome.similar.push(Similar {
682                    id: fact,
683                    score,
684                    reason,
685                });
686            }
687        }
688        outcome
689            .similar
690            .sort_unstable_by(|a, b| b.score.total_cmp(&a.score).then(a.id.cmp(&b.id)));
691        outcome.similar.truncate(8);
692    }
693
694    /// Whether `candidate` could possibly overlap `new_terms` above
695    /// `similar_jaccard` — the cheap half of [`Memory::find_similar`].
696    ///
697    /// `false` is a proof, not a guess. The candidate's summary marks a term
698    /// absent only when it really is absent, so the terms it *might* share is
699    /// an upper bound `u` on the true intersection `i`. Jaccard
700    /// `i / (|A| + |B| - i)` is increasing in `i`, so `u` at or below the
701    /// threshold puts the true value there too. `true` means "unknown" and
702    /// sends the caller to the exact comparison — which is also what an
703    /// unsummarized document and an untrustworthy index return.
704    fn overlap_possible(&self, candidate: FactId, new_terms: &[u32], trust_summary: bool) -> bool {
705        if !trust_summary {
706            return true;
707        }
708        let Some(doc) = self.bm25.doc(candidate) else {
709            return true;
710        };
711        if !doc.has_signature() {
712            return true;
713        }
714        let bound = doc.overlap_bound(new_terms);
715        // `overlap_bound` counts a subset of `new_terms`, so the subtraction
716        // is grouped to happen where it provably cannot go below zero. The
717        // union is then at least `distinct`, which `has_signature` just
718        // established is non-zero — no underflow and no division by zero, on a
719        // 64-bit `usize` or a 32-bit one.
720        debug_assert!(
721            bound <= new_terms.len(),
722            "the overlap bound counts query terms, so it cannot exceed them"
723        );
724        let union = (new_terms.len() - bound) + usize::from(doc.distinct);
725        bound as f32 / union as f32 > self.cfg.similar_jaccard
726    }
727
728    /// Revises `target`: closes its validity at the new fact's
729    /// `valid_from` and records the new fact with `revises = target`
730    /// (rule 2).
731    pub fn revise<S: Storage>(
732        &mut self,
733        store: &mut S,
734        target: FactId,
735        input: RememberInput<'_>,
736    ) -> Result<RememberOutcome, Error> {
737        self.validate_input(&input)?;
738        // Check first, mutate last: the target closes only after the new
739        // fact exists, so a mid-operation failure (capacity) never leaves
740        // a closed fact without its successor.
741        self.check_revisable(target)?;
742        let outcome = self.apply_remember(&input, target)?;
743        let valid_from = input.valid_from.unwrap_or(input.now);
744        self.close_target(target, valid_from);
745        self.journal_remember(store, &input, target, outcome.id)?;
746        Ok(outcome)
747    }
748
749    /// Tombstones a fact: it disappears from every query immediately, the
750    /// bytes go with the next `maintain` (rule 3). `Ok(false)`
751    /// when it was already tombstoned.
752    pub fn forget<S: Storage>(
753        &mut self,
754        store: &mut S,
755        now: u64,
756        id: FactId,
757    ) -> Result<bool, Error> {
758        let fresh = self.apply_forget(id)?;
759        let mut entry = Vec::new();
760        Op::Forget { now, fact: id }.encode(&mut entry);
761        store
762            .append_journal(&entry)
763            .map_err(|e| Error::Storage(format!("{e:?}")))?;
764        Ok(fresh)
765    }
766
767    /// Upserts a typed edge between two entities (created lazily);
768    /// re-linking an existing `(src, rel, dst)` updates its provenance.
769    pub fn link<S: Storage>(&mut self, store: &mut S, input: LinkInput<'_>) -> Result<(), Error> {
770        self.apply_link(
771            input.now,
772            input.src,
773            input.rel,
774            input.dst,
775            FactId::from_opt(input.provenance),
776        )?;
777        let mut entry = Vec::new();
778        Op::Link {
779            now: input.now,
780            src: input.src,
781            rel: input.rel,
782            dst: input.dst,
783            provenance: FactId::from_opt(input.provenance),
784        }
785        .encode(&mut entry);
786        store
787            .append_journal(&entry)
788            .map_err(|e| Error::Storage(format!("{e:?}")))?;
789        Ok(())
790    }
791
792    /// Closes the current typed edge between two entities. Returns `false`
793    /// when the edge is already absent.
794    pub fn unlink<S: Storage>(
795        &mut self,
796        store: &mut S,
797        input: UnlinkInput<'_>,
798    ) -> Result<bool, Error> {
799        let fresh = self.apply_unlink(input.now, input.src, input.rel, input.dst)?;
800        let mut entry = Vec::new();
801        Op::Unlink {
802            now: input.now,
803            src: input.src,
804            rel: input.rel,
805            dst: input.dst,
806        }
807        .encode(&mut entry);
808        store
809            .append_journal(&entry)
810            .map_err(|e| Error::Storage(format!("{e:?}")))?;
811        Ok(fresh)
812    }
813
814    /// Returns a fact unless it is tombstoned (closed facts are
815    /// returned — their interval says so).
816    pub fn get(&self, id: FactId) -> Option<FactView<'_>> {
817        let record = self.fact(id)?;
818        if record.is_tombstone() {
819            return None;
820        }
821        // Deferred validation: a load no longer scans the text
822        // pool, so tolerate invalid bytes here — a corrupt text hides the fact
823        // rather than panicking. `verify()` reports it explicitly.
824        let text = core::str::from_utf8(self.texts.get(record.text)).ok()?;
825        Some(FactView { record, text })
826    }
827
828    /// Appends the tag terms of a fact to `out`. A tombstoned or unknown
829    /// fact contributes nothing.
830    pub fn tags_of(&self, id: FactId, out: &mut Vec<TermId>) {
831        let Some(record) = self.fact(id) else { return };
832        if record.is_tombstone() {
833            return;
834        }
835        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
836            return;
837        };
838        for chunk in self.tag_lists.iter(&aux.tags) {
839            for raw in chunk.chunks_exact(4) {
840                out.push(TermId(u32::from_be_bytes(raw.try_into().unwrap())));
841            }
842        }
843    }
844
845    /// Fills `out` with the fact's metadata as key→value pairs in canonical
846    /// (ascending-key) order — the same order the blob was written and the same
847    /// order a host `BTreeMap` yields, so no layer sorts twice. Returns `true`
848    /// when the fact carries metadata (even an empty map is `false`, since an
849    /// empty map is stored as no blob).
850    ///
851    /// A tombstoned or unknown fact, a fact with no metadata blob, or a blob
852    /// that fails to decode (deferred validation, — `verify`
853    /// reports it) all leave `out` empty and return `false`; this accessor never
854    /// panics on bad bytes.
855    pub fn metadata_of<'s>(&'s self, id: FactId, out: &mut Vec<(&'s str, &'s str)>) -> bool {
856        out.clear();
857        let Some(record) = self.fact(id) else {
858            return false;
859        };
860        if record.is_tombstone() {
861            return false;
862        }
863        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
864            return false;
865        };
866        if aux.meta.0 == NONE_U32 || aux.meta.0 >= self.metas.len() as u32 {
867            return false;
868        }
869        crate::metadata::decode(self.metas.get(aux.meta), out).is_ok() && !out.is_empty()
870    }
871
872    /// Resolves an entity by name (normalized), without creating it.
873    pub fn entity(&mut self, name: &str) -> Option<EntityId> {
874        let mut norm = core::mem::take(&mut self.name_scratch);
875        normalize_name(&mut self.tokenizer, name, &mut norm);
876        let found = if norm.is_empty() {
877            None
878        } else {
879            // Only resolve an *existing* term: interning would create it.
880            self.lookup_entity_by_norm(&norm)
881        };
882        self.name_scratch = norm;
883        found
884    }
885
886    /// Resolves the string behind a term id (tags, relations).
887    pub fn term(&self, id: TermId) -> &str {
888        self.terms.resolve(id)
889    }
890
891    /// The canonical (verbatim) name of an entity, or `None` for an
892    /// unknown id. The read-side inverse of [`Memory::entity`]: it lets a
893    /// host export a fact with its subject *name* instead of the internal
894    /// [`EntityId`] carried by [`FactRecord`].
895    pub fn entity_name(&self, id: EntityId) -> Option<&str> {
896        let record = self.entities.get(&id.0.to_be_bytes())?;
897        // Tolerate invalid bytes (deferred validation): an
898        // unreadable name reads as `None`, never a panic.
899        core::str::from_utf8(self.texts.get(record.name)).ok()
900    }
901
902    /// Number of fact records currently stored (tombstoned facts count
903    /// until `maintain` removes them). Purged ids stay burned, so this can
904    /// be below [`Stats::next_fact`].
905    pub fn facts_len(&self) -> usize {
906        self.facts.len()
907    }
908
909    /// Number of entities.
910    pub fn entities_len(&self) -> usize {
911        self.entities.len()
912    }
913
914    /// The engine configuration.
915    pub fn cfg(&self) -> &Config {
916        &self.cfg
917    }
918
919    /// Size counters of the engine. O(1).
920    pub fn stats(&self) -> Stats {
921        Stats {
922            facts: self.facts.len(),
923            entities: self.entities.len(),
924            terms: self.terms.len(),
925            edges: self.edges_out.len(),
926            edge_versions: self.edges_hist_out.len(),
927            vectors: self.vecs.len(),
928            tombstones: self.tombstones,
929            hnsw_indexed: self.hnsw.indexed(),
930            next_fact: self.next_fact,
931            next_entity: self.next_entity,
932            next_edge: self.next_edge,
933            db_uuid: self.cfg.db_uuid,
934            pool_bytes: self.facts.pool_bytes()
935                + self.fact_aux.pool_bytes()
936                + self.entities.pool_bytes()
937                + self.by_name.pool_bytes()
938                + self.edges_out.pool_bytes()
939                + self.edges_in.pool_bytes()
940                + self.edges_hist_out.pool_bytes()
941                + self.edges_hist_in.pool_bytes()
942                + self.temporal.pool_bytes()
943                + self.texts.pool_bytes()
944                + self.terms.pool_bytes()
945                + self.tag_lists.pool_bytes()
946                + self.bm25.pool_bytes()
947                + self.tags_idx.pool_bytes()
948                + self.entity_facts.pool_bytes()
949                + self.vecs.pool_bytes()
950                + self.hnsw.pool_bytes(),
951            shards: shards::ShardLayout::of_config(&self.cfg),
952        }
953    }
954
955    // ---- internals ----
956
957    fn fact(&self, id: FactId) -> Option<FactRecord> {
958        self.facts.get(&id.0.to_be_bytes())
959    }
960
961    /// Size-limit checks shared by remember and revise.
962    fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
963        if input.text.len() > self.cfg.max_text {
964            return Err(Error::TooLarge {
965                what: "text",
966                len: input.text.len(),
967                max: self.cfg.max_text,
968            });
969        }
970        if input.tags.len() > 32 {
971            return Err(Error::TooLarge {
972                what: "tags",
973                len: input.tags.len(),
974                max: 32,
975            });
976        }
977        if input.links.len() > 16 {
978            return Err(Error::TooLarge {
979                what: "links",
980                len: input.links.len(),
981                max: 16,
982            });
983        }
984        if input.tags.iter().any(|t| t.is_empty()) {
985            return Err(Error::Invalid("empty tag"));
986        }
987        if !input.links.is_empty() && input.entity.is_none() {
988            return Err(Error::Invalid("links require a subject entity"));
989        }
990        if let Some(v) = input.vector {
991            if self.cfg.dim == 0 {
992                return Err(Error::Invalid("vector given but dim is 0"));
993            }
994            if v.len() != self.cfg.dim {
995                return Err(Error::DimMismatch {
996                    got: v.len(),
997                    want: self.cfg.dim,
998                });
999            }
1000        }
1001        Ok(())
1002    }
1003
1004    /// `Ok` when `target` exists, is not tombstoned and not yet closed.
1005    fn check_revisable(&self, target: FactId) -> Result<(), Error> {
1006        let record = self.fact(target).ok_or(Error::NotFound(target))?;
1007        if record.is_tombstone() {
1008            return Err(Error::NotFound(target));
1009        }
1010        if record.is_closed() {
1011            return Err(Error::AlreadyClosed(target));
1012        }
1013        Ok(())
1014    }
1015
1016    /// Closes a revision target (validity ends where the successor
1017    /// starts). The caller checked [`Memory::check_revisable`] first.
1018    fn close_target(&mut self, target: FactId, valid_to: u64) {
1019        let record = self.fact(target).expect("checked revisable");
1020        let payload = self
1021            .facts
1022            .payload_mut(&target.0.to_be_bytes())
1023            .expect("record fetched above");
1024        // Payload offsets = slot offsets - KEY_LEN(4): flags at 4..6,
1025        // valid_to at 36..44.
1026        let flags = record.flags | fact_flags::CLOSED;
1027        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1028        payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
1029    }
1030
1031    /// The one execution path of a new fact — called by the public verbs
1032    /// and by replay with identical effects.
1033    fn apply_remember(
1034        &mut self,
1035        input: &RememberInput<'_>,
1036        revises: FactId,
1037    ) -> Result<RememberOutcome, Error> {
1038        let id = FactId(self.next_fact);
1039        let entity = match input.entity {
1040            Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
1041            None => None,
1042        };
1043        let text_id = self.texts.push(input.text.as_bytes())?;
1044
1045        // Tokenize into (term, tf) pairs in the reusable scratch.
1046        let mut tfs = core::mem::take(&mut self.tf_scratch);
1047        tfs.clear();
1048        let terms = &mut self.terms;
1049        let mut intern_err = None;
1050        self.tokenizer.tokenize(input.text, &mut |token| {
1051            if intern_err.is_some() {
1052                return;
1053            }
1054            match terms.intern(token) {
1055                Ok(term) => match tfs.iter_mut().find(|(t, _)| *t == term.0) {
1056                    Some((_, tf)) => *tf = tf.saturating_add(1),
1057                    None => tfs.push((term.0, 1)),
1058                },
1059                Err(e) => intern_err = Some(e),
1060            }
1061        });
1062        if let Some(e) = intern_err {
1063            self.tf_scratch = tfs;
1064            return Err(Error::Arena(e));
1065        }
1066        self.bm25.index_doc(id, &tfs)?;
1067        self.tf_scratch = tfs;
1068
1069        // Metadata: canonicalized (keys sorted, dups rejected) and stored as one
1070        // opaque blob; an absent or empty map leaves the sentinel. Pushed before
1071        // the fact record so a capacity/validation failure aborts the whole op.
1072        let meta = match input.metadata {
1073            Some(pairs) if !pairs.is_empty() => {
1074                self.metas.push(&crate::metadata::encode(pairs)?)?
1075            }
1076            _ => BlobId(NONE_U32),
1077        };
1078
1079        // Tags: interned verbatim, deduplicated, listed on the fact and
1080        // inverted.
1081        let mut aux = FactAux {
1082            id,
1083            tags: ListHandle::EMPTY,
1084            meta,
1085        };
1086        let mut seen_tags: [u32; 32] = [NONE_U32; 32];
1087        let mut seen_cnt = 0usize;
1088        for tag in input.tags {
1089            let term = self.terms.intern(tag)?;
1090            if seen_tags[..seen_cnt].contains(&term.0) {
1091                continue;
1092            }
1093            seen_tags[seen_cnt] = term.0;
1094            seen_cnt += 1;
1095            self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
1096            self.tags_idx.push(term.0, id, 0)?;
1097        }
1098        self.fact_aux.insert(&aux)?;
1099
1100        // Links: subject → target edges with this fact as provenance.
1101        if let Some(src) = entity {
1102            for &(rel, dst_name) in input.links {
1103                let dst = self.resolve_or_create_entity(dst_name, input.now)?;
1104                let rel = self.terms.intern(rel)?;
1105                self.open_edge(input.now, src, rel, dst, id)?;
1106            }
1107            self.entity_facts.push(src.0, id, 0)?;
1108        }
1109
1110        // Vector: quantized into the flat pool; the fact keeps the slot
1111        // index. Pushed before the record so a capacity failure aborts the
1112        // whole op (replay rebuilds consistently, like the other indexes).
1113        let (vector, flags) = match input.vector {
1114            Some(v) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
1115            None => (NONE_U32, 0),
1116        };
1117
1118        let recorded_at = input.now;
1119        let valid_from = input.valid_from.unwrap_or(input.now);
1120        self.facts.insert(&FactRecord {
1121            id,
1122            entity: EntityId::from_opt(entity),
1123            flags,
1124            kind: 0,
1125            text: text_id,
1126            vector,
1127            revises,
1128            recorded_at,
1129            valid_from,
1130            valid_to: VALID_TO_OPEN,
1131        })?;
1132        self.temporal.insert(&TemporalSlot {
1133            recorded_at,
1134            fact: id,
1135        })?;
1136        self.next_fact += 1;
1137        Ok(RememberOutcome {
1138            id,
1139            entity,
1140            similar: Vec::new(),
1141        })
1142    }
1143
1144    fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
1145        let record = self.fact(id).ok_or(Error::NotFound(id))?;
1146        if record.is_tombstone() {
1147            return Ok(false);
1148        }
1149        let payload = self
1150            .facts
1151            .payload_mut(&id.0.to_be_bytes())
1152            .expect("record fetched above");
1153        let flags = record.flags | fact_flags::TOMBSTONE;
1154        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1155        self.tombstones += 1;
1156        Ok(true)
1157    }
1158
1159    fn apply_link(
1160        &mut self,
1161        now: u64,
1162        src: &str,
1163        rel: &str,
1164        dst: &str,
1165        provenance: FactId,
1166    ) -> Result<(), Error> {
1167        let src = self.resolve_or_create_entity(src, now)?;
1168        let dst = self.resolve_or_create_entity(dst, now)?;
1169        let rel = self.terms.intern(rel)?;
1170        self.open_edge(now, src, rel, dst, provenance)
1171    }
1172
1173    fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
1174        let Some(src) = self.lookup_entity_name(src) else {
1175            return Ok(false);
1176        };
1177        let Some(dst) = self.lookup_entity_name(dst) else {
1178            return Ok(false);
1179        };
1180        let Some(rel) = self.terms.lookup(rel) else {
1181            return Ok(false);
1182        };
1183        self.close_current_edge(now, src, rel, dst)
1184    }
1185
1186    fn open_edge(
1187        &mut self,
1188        now: u64,
1189        src: EntityId,
1190        rel: TermId,
1191        dst: EntityId,
1192        fact: FactId,
1193    ) -> Result<(), Error> {
1194        if let Some(current) = self.current_edge(src, rel, dst) {
1195            if current.fact == fact {
1196                return Ok(());
1197            }
1198            self.close_current_edge(now, src, rel, dst)?;
1199        }
1200        let edge = EdgeId(self.next_edge);
1201        let history = EdgeHistorySlot {
1202            a: src,
1203            rel,
1204            b: dst,
1205            edge,
1206            fact,
1207            flags: 0,
1208            kind: 0,
1209            recorded_at: now,
1210            valid_from: now,
1211            valid_to: VALID_TO_OPEN,
1212        };
1213        self.insert_history_edge(history)?;
1214        self.insert_current_edge(src, rel, dst, fact, edge, now)?;
1215        self.next_edge += 1;
1216        Ok(())
1217    }
1218
1219    fn insert_current_edge(
1220        &mut self,
1221        src: EntityId,
1222        rel: TermId,
1223        dst: EntityId,
1224        fact: FactId,
1225        edge: EdgeId,
1226        valid_from: u64,
1227    ) -> Result<(), Error> {
1228        for (arena, a, b) in [
1229            (&mut self.edges_out, src, dst),
1230            (&mut self.edges_in, dst, src),
1231        ] {
1232            let slot = EdgeSlot {
1233                a,
1234                rel,
1235                b,
1236                fact,
1237                edge,
1238                valid_from,
1239            };
1240            if !arena.insert(&slot)? {
1241                let payload = arena
1242                    .payload_mut(&edge_key(a, rel, b))
1243                    .expect("insert reported a duplicate");
1244                let mut full = [0u8; EdgeSlot::SIZE];
1245                slot.write(&mut full);
1246                payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
1247            }
1248        }
1249        Ok(())
1250    }
1251
1252    fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
1253        self.edges_hist_out.insert(&edge)?;
1254        self.edges_hist_in.insert(&EdgeHistorySlot {
1255            a: edge.b,
1256            b: edge.a,
1257            ..edge
1258        })?;
1259        Ok(())
1260    }
1261
1262    /// Closes the open version of `(src, rel, dst)` and drops it from the
1263    /// current graph. Returns `false` when there is no such edge.
1264    ///
1265    /// The current slot names its own history record — `edge` and `valid_from`
1266    /// are that record's key tail — so both mirrors are addressed directly.
1267    fn close_current_edge(
1268        &mut self,
1269        now: u64,
1270        src: EntityId,
1271        rel: TermId,
1272        dst: EntityId,
1273    ) -> Result<bool, Error> {
1274        let Some(current) = self.current_edge(src, rel, dst) else {
1275            return Ok(false);
1276        };
1277        // A version never ends before it began, even if the caller's clock
1278        // moved backwards between the link and the unlink.
1279        let close_at = now.max(current.valid_from);
1280        let out_key = edge_history_key(src, current.valid_from, current.edge);
1281        let in_key = edge_history_key(dst, current.valid_from, current.edge);
1282        close_edge_history_payload(
1283            self.edges_hist_out
1284                .payload_mut(&out_key)
1285                .ok_or(Error::Corrupt("missing outgoing edge history"))?,
1286            close_at,
1287        );
1288        close_edge_history_payload(
1289            self.edges_hist_in
1290                .payload_mut(&in_key)
1291                .ok_or(Error::Corrupt("missing incoming edge history"))?,
1292            close_at,
1293        );
1294        let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
1295        let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
1296        if out_removed != in_removed {
1297            return Err(Error::Corrupt("edge mirrors disagree"));
1298        }
1299        Ok(out_removed)
1300    }
1301
1302    fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
1303        self.edges_out
1304            .get_slot(&edge_key(src, rel, dst))
1305            .map(EdgeSlot::read)
1306    }
1307
1308    /// Looks an entity up by its already-normalized name (read-only:
1309    /// neither the vocabulary nor the arenas change on a miss).
1310    fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
1311        let term = self.terms.lookup(norm)?;
1312        let mut from = [0u8; 8];
1313        key::write_u32(&mut from, term.0);
1314        let mut to = [0u8; 8];
1315        key::write_u32(&mut to, term.0);
1316        to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
1317        self.by_name.range(&from, &to).next().map(|e| e.id)
1318    }
1319
1320    fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
1321        let mut norm = core::mem::take(&mut self.name_scratch);
1322        normalize_name(&mut self.tokenizer, name, &mut norm);
1323        let result = (!norm.is_empty())
1324            .then(|| self.lookup_entity_by_norm(&norm))
1325            .flatten();
1326        self.name_scratch = norm;
1327        result
1328    }
1329
1330    fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
1331        let mut norm = core::mem::take(&mut self.name_scratch);
1332        normalize_name(&mut self.tokenizer, name, &mut norm);
1333        if norm.is_empty() {
1334            self.name_scratch = norm;
1335            return Err(Error::Invalid("entity name has no indexable characters"));
1336        }
1337        let result = (|| {
1338            if let Some(found) = self.lookup_entity_by_norm(&norm) {
1339                return Ok(found);
1340            }
1341            let term = self.terms.intern(&norm)?;
1342            let id = EntityId(self.next_entity);
1343            let name_id = self.texts.push(name.as_bytes())?;
1344            self.entities.insert(&EntityRecord {
1345                id,
1346                name: name_id,
1347                name_term: term,
1348                created_at: now,
1349                flags: 0,
1350            })?;
1351            self.by_name.insert(&EntityByName {
1352                name_term: term,
1353                id,
1354            })?;
1355            self.next_entity += 1;
1356            Ok(id)
1357        })();
1358        self.name_scratch = norm;
1359        result
1360    }
1361
1362    fn journal_remember<S: Storage>(
1363        &mut self,
1364        store: &mut S,
1365        input: &RememberInput<'_>,
1366        revises: FactId,
1367        assigned: FactId,
1368    ) -> Result<(), Error> {
1369        let mut entry = Vec::new();
1370        Op::Remember {
1371            now: input.now,
1372            valid_from: input.valid_from.unwrap_or(input.now),
1373            entity: input.entity,
1374            text: input.text,
1375            tags: input.tags.to_vec(),
1376            links: input.links.to_vec(),
1377            vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
1378            metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
1379            revises,
1380            assigned,
1381        }
1382        .encode(&mut entry);
1383        store
1384            .append_journal(&entry)
1385            .map_err(|e| Error::Storage(format!("{e:?}")))
1386    }
1387}
1388
1389impl core::fmt::Debug for Memory<'_> {
1390    /// Summary only — the contents are the user's memory, not ours to
1391    /// print.
1392    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1393        f.debug_struct("Memory")
1394            .field("facts", &self.facts.len())
1395            .field("entities", &self.entities.len())
1396            .field("terms", &self.terms.len())
1397            .finish()
1398    }
1399}
1400
1401/// Normalizes an entity name: its tokens joined by single spaces
1402/// ("  Проект   PlugMem " → "проект plugmem"). Deterministic and aligned
1403/// with search-time tokenization.
1404fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
1405    out.clear();
1406    tokenizer.tokenize(name, &mut |token| {
1407        if !out.is_empty() {
1408            out.push(' ');
1409        }
1410        out.push_str(token);
1411    });
1412}