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    /// Visits every currently-open edge once, as
903    /// `(source name, relation, destination name, provenance)`.
904    ///
905    /// Reads the out-arena only: each edge is mirrored in both arenas, and the
906    /// out-arena holds it keyed `(src, rel, dst)`, which is the direction a
907    /// caller wants to write back. Closed versions live in the history arenas
908    /// and are not visited — this is the *current* graph, matching what a fact
909    /// export does for facts.
910    ///
911    /// Names are borrowed from the interner and the text pool, so a caller that
912    /// writes them straight out allocates nothing per edge. An edge whose
913    /// endpoint name is unreadable is skipped rather than reported with a
914    /// placeholder: deferred validation means a damaged image must not turn
915    /// into plausible-looking output.
916    ///
917    /// `visit` returning `false` stops the walk, so a caller can bound its own
918    /// work without materializing the graph.
919    pub fn edges_each(&self, mut visit: impl FnMut(&str, &str, &str, FactId) -> bool) {
920        for slot in self.edges_out.iter() {
921            let (Some(src), Some(dst)) = (self.entity_name(slot.a), self.entity_name(slot.b))
922            else {
923                continue;
924            };
925            if !visit(src, self.term(slot.rel), dst, slot.fact) {
926                return;
927            }
928        }
929    }
930
931    /// Number of fact records currently stored (tombstoned facts count
932    /// until `maintain` removes them). Purged ids stay burned, so this can
933    /// be below [`Stats::next_fact`].
934    pub fn facts_len(&self) -> usize {
935        self.facts.len()
936    }
937
938    /// Number of entities.
939    pub fn entities_len(&self) -> usize {
940        self.entities.len()
941    }
942
943    /// The engine configuration.
944    pub fn cfg(&self) -> &Config {
945        &self.cfg
946    }
947
948    /// Size counters of the engine. O(1).
949    pub fn stats(&self) -> Stats {
950        Stats {
951            facts: self.facts.len(),
952            entities: self.entities.len(),
953            terms: self.terms.len(),
954            edges: self.edges_out.len(),
955            edge_versions: self.edges_hist_out.len(),
956            vectors: self.vecs.len(),
957            tombstones: self.tombstones,
958            hnsw_indexed: self.hnsw.indexed(),
959            next_fact: self.next_fact,
960            next_entity: self.next_entity,
961            next_edge: self.next_edge,
962            db_uuid: self.cfg.db_uuid,
963            pool_bytes: self.facts.pool_bytes()
964                + self.fact_aux.pool_bytes()
965                + self.entities.pool_bytes()
966                + self.by_name.pool_bytes()
967                + self.edges_out.pool_bytes()
968                + self.edges_in.pool_bytes()
969                + self.edges_hist_out.pool_bytes()
970                + self.edges_hist_in.pool_bytes()
971                + self.temporal.pool_bytes()
972                + self.texts.pool_bytes()
973                + self.terms.pool_bytes()
974                + self.tag_lists.pool_bytes()
975                + self.bm25.pool_bytes()
976                + self.tags_idx.pool_bytes()
977                + self.entity_facts.pool_bytes()
978                + self.vecs.pool_bytes()
979                + self.hnsw.pool_bytes(),
980            shards: shards::ShardLayout::of_config(&self.cfg),
981        }
982    }
983
984    // ---- internals ----
985
986    fn fact(&self, id: FactId) -> Option<FactRecord> {
987        self.facts.get(&id.0.to_be_bytes())
988    }
989
990    /// Size-limit checks shared by remember and revise.
991    fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
992        if input.text.len() > self.cfg.max_text {
993            return Err(Error::TooLarge {
994                what: "text",
995                len: input.text.len(),
996                max: self.cfg.max_text,
997            });
998        }
999        if input.tags.len() > 32 {
1000            return Err(Error::TooLarge {
1001                what: "tags",
1002                len: input.tags.len(),
1003                max: 32,
1004            });
1005        }
1006        if input.links.len() > 16 {
1007            return Err(Error::TooLarge {
1008                what: "links",
1009                len: input.links.len(),
1010                max: 16,
1011            });
1012        }
1013        if input.tags.iter().any(|t| t.is_empty()) {
1014            return Err(Error::Invalid("empty tag"));
1015        }
1016        if !input.links.is_empty() && input.entity.is_none() {
1017            return Err(Error::Invalid("links require a subject entity"));
1018        }
1019        if let Some(v) = input.vector {
1020            if self.cfg.dim == 0 {
1021                return Err(Error::Invalid("vector given but dim is 0"));
1022            }
1023            if v.len() != self.cfg.dim {
1024                return Err(Error::DimMismatch {
1025                    got: v.len(),
1026                    want: self.cfg.dim,
1027                });
1028            }
1029        }
1030        Ok(())
1031    }
1032
1033    /// `Ok` when `target` exists, is not tombstoned and not yet closed.
1034    fn check_revisable(&self, target: FactId) -> Result<(), Error> {
1035        let record = self.fact(target).ok_or(Error::NotFound(target))?;
1036        if record.is_tombstone() {
1037            return Err(Error::NotFound(target));
1038        }
1039        if record.is_closed() {
1040            return Err(Error::AlreadyClosed(target));
1041        }
1042        Ok(())
1043    }
1044
1045    /// Closes a revision target (validity ends where the successor
1046    /// starts). The caller checked [`Memory::check_revisable`] first.
1047    fn close_target(&mut self, target: FactId, valid_to: u64) {
1048        let record = self.fact(target).expect("checked revisable");
1049        let payload = self
1050            .facts
1051            .payload_mut(&target.0.to_be_bytes())
1052            .expect("record fetched above");
1053        // Payload offsets = slot offsets - KEY_LEN(4): flags at 4..6,
1054        // valid_to at 36..44.
1055        let flags = record.flags | fact_flags::CLOSED;
1056        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1057        payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
1058    }
1059
1060    /// The one execution path of a new fact — called by the public verbs
1061    /// and by replay with identical effects.
1062    fn apply_remember(
1063        &mut self,
1064        input: &RememberInput<'_>,
1065        revises: FactId,
1066    ) -> Result<RememberOutcome, Error> {
1067        let id = FactId(self.next_fact);
1068        let entity = match input.entity {
1069            Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
1070            None => None,
1071        };
1072        let text_id = self.texts.push(input.text.as_bytes())?;
1073
1074        // Tokenize into (term, tf) pairs in the reusable scratch.
1075        let mut tfs = core::mem::take(&mut self.tf_scratch);
1076        tfs.clear();
1077        let terms = &mut self.terms;
1078        let mut intern_err = None;
1079        self.tokenizer.tokenize(input.text, &mut |token| {
1080            if intern_err.is_some() {
1081                return;
1082            }
1083            match terms.intern(token) {
1084                Ok(term) => match tfs.iter_mut().find(|(t, _)| *t == term.0) {
1085                    Some((_, tf)) => *tf = tf.saturating_add(1),
1086                    None => tfs.push((term.0, 1)),
1087                },
1088                Err(e) => intern_err = Some(e),
1089            }
1090        });
1091        if let Some(e) = intern_err {
1092            self.tf_scratch = tfs;
1093            return Err(Error::Arena(e));
1094        }
1095        self.bm25.index_doc(id, &tfs)?;
1096        self.tf_scratch = tfs;
1097
1098        // Metadata: canonicalized (keys sorted, dups rejected) and stored as one
1099        // opaque blob; an absent or empty map leaves the sentinel. Pushed before
1100        // the fact record so a capacity/validation failure aborts the whole op.
1101        let meta = match input.metadata {
1102            Some(pairs) if !pairs.is_empty() => {
1103                self.metas.push(&crate::metadata::encode(pairs)?)?
1104            }
1105            _ => BlobId(NONE_U32),
1106        };
1107
1108        // Tags: interned verbatim, deduplicated, listed on the fact and
1109        // inverted.
1110        let mut aux = FactAux {
1111            id,
1112            tags: ListHandle::EMPTY,
1113            meta,
1114        };
1115        let mut seen_tags: [u32; 32] = [NONE_U32; 32];
1116        let mut seen_cnt = 0usize;
1117        for tag in input.tags {
1118            let term = self.terms.intern(tag)?;
1119            if seen_tags[..seen_cnt].contains(&term.0) {
1120                continue;
1121            }
1122            seen_tags[seen_cnt] = term.0;
1123            seen_cnt += 1;
1124            self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
1125            self.tags_idx.push(term.0, id, 0)?;
1126        }
1127        self.fact_aux.insert(&aux)?;
1128
1129        // Links: subject → target edges with this fact as provenance.
1130        if let Some(src) = entity {
1131            for &(rel, dst_name) in input.links {
1132                let dst = self.resolve_or_create_entity(dst_name, input.now)?;
1133                let rel = self.terms.intern(rel)?;
1134                self.open_edge(input.now, src, rel, dst, id)?;
1135            }
1136            self.entity_facts.push(src.0, id, 0)?;
1137        }
1138
1139        // Vector: quantized into the flat pool; the fact keeps the slot
1140        // index. Pushed before the record so a capacity failure aborts the
1141        // whole op (replay rebuilds consistently, like the other indexes).
1142        let (vector, flags) = match input.vector {
1143            Some(v) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
1144            None => (NONE_U32, 0),
1145        };
1146
1147        let recorded_at = input.now;
1148        let valid_from = input.valid_from.unwrap_or(input.now);
1149        self.facts.insert(&FactRecord {
1150            id,
1151            entity: EntityId::from_opt(entity),
1152            flags,
1153            kind: 0,
1154            text: text_id,
1155            vector,
1156            revises,
1157            recorded_at,
1158            valid_from,
1159            valid_to: VALID_TO_OPEN,
1160        })?;
1161        self.temporal.insert(&TemporalSlot {
1162            recorded_at,
1163            fact: id,
1164        })?;
1165        self.next_fact += 1;
1166        Ok(RememberOutcome {
1167            id,
1168            entity,
1169            similar: Vec::new(),
1170        })
1171    }
1172
1173    fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
1174        let record = self.fact(id).ok_or(Error::NotFound(id))?;
1175        if record.is_tombstone() {
1176            return Ok(false);
1177        }
1178        let payload = self
1179            .facts
1180            .payload_mut(&id.0.to_be_bytes())
1181            .expect("record fetched above");
1182        let flags = record.flags | fact_flags::TOMBSTONE;
1183        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1184        self.tombstones += 1;
1185        Ok(true)
1186    }
1187
1188    fn apply_link(
1189        &mut self,
1190        now: u64,
1191        src: &str,
1192        rel: &str,
1193        dst: &str,
1194        provenance: FactId,
1195    ) -> Result<(), Error> {
1196        let src = self.resolve_or_create_entity(src, now)?;
1197        let dst = self.resolve_or_create_entity(dst, now)?;
1198        let rel = self.terms.intern(rel)?;
1199        self.open_edge(now, src, rel, dst, provenance)
1200    }
1201
1202    fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
1203        let Some(src) = self.lookup_entity_name(src) else {
1204            return Ok(false);
1205        };
1206        let Some(dst) = self.lookup_entity_name(dst) else {
1207            return Ok(false);
1208        };
1209        let Some(rel) = self.terms.lookup(rel) else {
1210            return Ok(false);
1211        };
1212        self.close_current_edge(now, src, rel, dst)
1213    }
1214
1215    fn open_edge(
1216        &mut self,
1217        now: u64,
1218        src: EntityId,
1219        rel: TermId,
1220        dst: EntityId,
1221        fact: FactId,
1222    ) -> Result<(), Error> {
1223        if let Some(current) = self.current_edge(src, rel, dst) {
1224            if current.fact == fact {
1225                return Ok(());
1226            }
1227            self.close_current_edge(now, src, rel, dst)?;
1228        }
1229        let edge = EdgeId(self.next_edge);
1230        let history = EdgeHistorySlot {
1231            a: src,
1232            rel,
1233            b: dst,
1234            edge,
1235            fact,
1236            flags: 0,
1237            kind: 0,
1238            recorded_at: now,
1239            valid_from: now,
1240            valid_to: VALID_TO_OPEN,
1241        };
1242        self.insert_history_edge(history)?;
1243        self.insert_current_edge(src, rel, dst, fact, edge, now)?;
1244        self.next_edge += 1;
1245        Ok(())
1246    }
1247
1248    fn insert_current_edge(
1249        &mut self,
1250        src: EntityId,
1251        rel: TermId,
1252        dst: EntityId,
1253        fact: FactId,
1254        edge: EdgeId,
1255        valid_from: u64,
1256    ) -> Result<(), Error> {
1257        for (arena, a, b) in [
1258            (&mut self.edges_out, src, dst),
1259            (&mut self.edges_in, dst, src),
1260        ] {
1261            let slot = EdgeSlot {
1262                a,
1263                rel,
1264                b,
1265                fact,
1266                edge,
1267                valid_from,
1268            };
1269            if !arena.insert(&slot)? {
1270                let payload = arena
1271                    .payload_mut(&edge_key(a, rel, b))
1272                    .expect("insert reported a duplicate");
1273                let mut full = [0u8; EdgeSlot::SIZE];
1274                slot.write(&mut full);
1275                payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
1276            }
1277        }
1278        Ok(())
1279    }
1280
1281    fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
1282        self.edges_hist_out.insert(&edge)?;
1283        self.edges_hist_in.insert(&EdgeHistorySlot {
1284            a: edge.b,
1285            b: edge.a,
1286            ..edge
1287        })?;
1288        Ok(())
1289    }
1290
1291    /// Closes the open version of `(src, rel, dst)` and drops it from the
1292    /// current graph. Returns `false` when there is no such edge.
1293    ///
1294    /// The current slot names its own history record — `edge` and `valid_from`
1295    /// are that record's key tail — so both mirrors are addressed directly.
1296    fn close_current_edge(
1297        &mut self,
1298        now: u64,
1299        src: EntityId,
1300        rel: TermId,
1301        dst: EntityId,
1302    ) -> Result<bool, Error> {
1303        let Some(current) = self.current_edge(src, rel, dst) else {
1304            return Ok(false);
1305        };
1306        // A version never ends before it began, even if the caller's clock
1307        // moved backwards between the link and the unlink.
1308        let close_at = now.max(current.valid_from);
1309        let out_key = edge_history_key(src, current.valid_from, current.edge);
1310        let in_key = edge_history_key(dst, current.valid_from, current.edge);
1311        close_edge_history_payload(
1312            self.edges_hist_out
1313                .payload_mut(&out_key)
1314                .ok_or(Error::Corrupt("missing outgoing edge history"))?,
1315            close_at,
1316        );
1317        close_edge_history_payload(
1318            self.edges_hist_in
1319                .payload_mut(&in_key)
1320                .ok_or(Error::Corrupt("missing incoming edge history"))?,
1321            close_at,
1322        );
1323        let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
1324        let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
1325        if out_removed != in_removed {
1326            return Err(Error::Corrupt("edge mirrors disagree"));
1327        }
1328        Ok(out_removed)
1329    }
1330
1331    fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
1332        self.edges_out
1333            .get_slot(&edge_key(src, rel, dst))
1334            .map(EdgeSlot::read)
1335    }
1336
1337    /// Looks an entity up by its already-normalized name (read-only:
1338    /// neither the vocabulary nor the arenas change on a miss).
1339    fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
1340        let term = self.terms.lookup(norm)?;
1341        let mut from = [0u8; 8];
1342        key::write_u32(&mut from, term.0);
1343        let mut to = [0u8; 8];
1344        key::write_u32(&mut to, term.0);
1345        to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
1346        self.by_name.range(&from, &to).next().map(|e| e.id)
1347    }
1348
1349    fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
1350        let mut norm = core::mem::take(&mut self.name_scratch);
1351        normalize_name(&mut self.tokenizer, name, &mut norm);
1352        let result = (!norm.is_empty())
1353            .then(|| self.lookup_entity_by_norm(&norm))
1354            .flatten();
1355        self.name_scratch = norm;
1356        result
1357    }
1358
1359    fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
1360        let mut norm = core::mem::take(&mut self.name_scratch);
1361        normalize_name(&mut self.tokenizer, name, &mut norm);
1362        if norm.is_empty() {
1363            self.name_scratch = norm;
1364            return Err(Error::Invalid("entity name has no indexable characters"));
1365        }
1366        let result = (|| {
1367            if let Some(found) = self.lookup_entity_by_norm(&norm) {
1368                return Ok(found);
1369            }
1370            let term = self.terms.intern(&norm)?;
1371            let id = EntityId(self.next_entity);
1372            let name_id = self.texts.push(name.as_bytes())?;
1373            self.entities.insert(&EntityRecord {
1374                id,
1375                name: name_id,
1376                name_term: term,
1377                created_at: now,
1378                flags: 0,
1379            })?;
1380            self.by_name.insert(&EntityByName {
1381                name_term: term,
1382                id,
1383            })?;
1384            self.next_entity += 1;
1385            Ok(id)
1386        })();
1387        self.name_scratch = norm;
1388        result
1389    }
1390
1391    fn journal_remember<S: Storage>(
1392        &mut self,
1393        store: &mut S,
1394        input: &RememberInput<'_>,
1395        revises: FactId,
1396        assigned: FactId,
1397    ) -> Result<(), Error> {
1398        let mut entry = Vec::new();
1399        Op::Remember {
1400            now: input.now,
1401            valid_from: input.valid_from.unwrap_or(input.now),
1402            entity: input.entity,
1403            text: input.text,
1404            tags: input.tags.to_vec(),
1405            links: input.links.to_vec(),
1406            vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
1407            metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
1408            revises,
1409            assigned,
1410        }
1411        .encode(&mut entry);
1412        store
1413            .append_journal(&entry)
1414            .map_err(|e| Error::Storage(format!("{e:?}")))
1415    }
1416}
1417
1418impl core::fmt::Debug for Memory<'_> {
1419    /// Summary only — the contents are the user's memory, not ours to
1420    /// print.
1421    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1422        f.debug_struct("Memory")
1423            .field("facts", &self.facts.len())
1424            .field("entities", &self.entities.len())
1425            .field("terms", &self.terms.len())
1426            .finish()
1427    }
1428}
1429
1430/// Normalizes an entity name: its tokens joined by single spaces
1431/// ("  Проект   PlugMem " → "проект plugmem"). Deterministic and aligned
1432/// with search-time tokenization.
1433fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
1434    out.clear();
1435    tokenizer.tokenize(name, &mut |token| {
1436        if !out.is_empty() {
1437            out.push(' ');
1438        }
1439        out.push_str(token);
1440    });
1441}