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, ToString};
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
45/// Tags one fact may carry.
46///
47/// Structural, not a policy knob: `apply_remember` deduplicates into a stack
48/// array of exactly this length, and `change_catalog_for_fact` reads the tags
49/// back into another one to keep the catalog counts balanced. Every path that
50/// reaches `apply_remember` — the verbs through `validate_input`, replay
51/// through its own corruption check — has to have established the bound first.
52const MAX_TAGS: usize = 32;
53
54mod maintain;
55mod migrations;
56mod persist;
57mod recall;
58mod reembed;
59mod shards;
60mod tags;
61
62pub use maintain::{MaintainReport, MaintenanceMode, MaintenanceOptions};
63pub use recall::{RecallQuery, RecallResult, RecallScratch, RecalledEdge, RecalledFact, source};
64pub use reembed::{ReembedError, ReembedReport};
65pub use shards::ShardLayout;
66pub use tags::{DEFAULT_TAG_PAGE_LIMIT, MAX_TAG_PAGE_LIMIT, TagPage, TagQuery, TagSummary};
67
68/// Maximum UTF-8 byte length of a persisted embedding-space identity.
69pub const MAX_VECTOR_SPACE_ID_BYTES: usize = 256;
70
71/// Input of `remember` and `revise`.
72#[derive(Clone, Copy, Debug)]
73#[cfg_attr(feature = "serde", derive(serde::Serialize))]
74pub struct RememberInput<'a> {
75    /// Host timestamp, unix milliseconds.
76    pub now: u64,
77    /// Fact text, ≤ `Config::max_text` bytes.
78    pub text: &'a str,
79    /// Subject entity name; created lazily on first mention.
80    pub entity: Option<&'a str>,
81    /// Tags, verbatim strings, ≤ 32.
82    pub tags: &'a [&'a str],
83    /// `(rel, target_entity)` pairs, ≤ 16; edges go subject → target.
84    pub links: &'a [(&'a str, &'a str)],
85    /// Optional embedding, `len == Config::dim`; quantized on the way in.
86    /// Requires `dim > 0` and is dropped by the engine's own re-quantized
87    /// replay, so nothing float-nondeterministic reaches the state.
88    pub vector: Option<&'a [f32]>,
89    /// Validity start; defaults to `now`.
90    pub valid_from: Option<u64>,
91    /// Optional metadata as key→value pairs (UTF-8). Passed in any order; the
92    /// engine canonicalizes (sorts keys, rejects duplicates) and stores them as
93    /// one opaque blob (see `crate::metadata`). `None` or an empty slice = no
94    /// metadata. The engine never interprets the values.
95    pub metadata: Option<&'a [(&'a str, &'a str)]>,
96}
97
98impl<'a> RememberInput<'a> {
99    /// A minimal input: text only.
100    pub fn text(now: u64, text: &'a str) -> Self {
101        Self {
102            now,
103            text,
104            entity: None,
105            tags: &[],
106            links: &[],
107            vector: None,
108            valid_from: None,
109            metadata: None,
110        }
111    }
112}
113
114/// Input of `link`.
115#[derive(Clone, Copy, Debug)]
116#[cfg_attr(feature = "serde", derive(serde::Serialize))]
117pub struct LinkInput<'a> {
118    /// Host timestamp, unix milliseconds.
119    pub now: u64,
120    /// Source entity name (created lazily).
121    pub src: &'a str,
122    /// Relation term, verbatim (`"works_at"`, …).
123    pub rel: &'a str,
124    /// Destination entity name (created lazily).
125    pub dst: &'a str,
126    /// Optional provenance fact.
127    pub provenance: Option<FactId>,
128}
129
130/// Input of `unlink`.
131#[derive(Clone, Copy, Debug)]
132#[cfg_attr(feature = "serde", derive(serde::Serialize))]
133pub struct UnlinkInput<'a> {
134    /// Host timestamp, unix milliseconds.
135    pub now: u64,
136    /// Source entity name.
137    pub src: &'a str,
138    /// Relation term, verbatim.
139    pub rel: &'a str,
140    /// Destination entity name.
141    pub dst: &'a str,
142}
143
144/// Result of `remember`/`revise`.
145#[derive(Clone, Debug, PartialEq)]
146#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
147pub struct RememberOutcome {
148    /// The new fact's id.
149    pub id: FactId,
150    /// The subject entity (resolved or created), if one was named.
151    pub entity: Option<EntityId>,
152    /// Similar / potentially conflicting **live** facts, best match
153    /// first (≤ 8). The engine never revises on its own — it surfaces
154    /// the candidates, the agent judges: `revise`, keep both, or
155    /// `forget`.
156    pub similar: Vec<Similar>,
157}
158
159/// Result of [`Memory::remember_guarded`].
160///
161/// The two variants are intentionally disjoint: a blocked call has no fact id
162/// because it did not allocate one, mutate an index, or append a journal
163/// record. A caller that decides the candidates are compatible can make that
164/// choice explicit by following with ordinary [`Memory::remember`].
165///
166/// `Stored` additionally reports whether a check happened at all - see
167/// [`GuardedRememberOutcome::Stored::checked`]. Without it the two answers
168/// "nothing similar was found" and "nothing could be compared" are the same
169/// value, and a caller reaching for this verb specifically to avoid duplicates
170/// has no way to learn it did not get one.
171#[derive(Clone, Debug, PartialEq)]
172#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
173#[cfg_attr(feature = "serde", serde(tag = "status", rename_all = "snake_case"))]
174pub enum GuardedRememberOutcome {
175    /// No candidate crossed the configured similarity thresholds; the fact was
176    /// stored normally.
177    Stored {
178        /// The committed fact.
179        outcome: RememberOutcome,
180        /// Whether the similarity detector actually had candidates to compare
181        /// against.
182        ///
183        /// `false` means the write was stored **unguarded**: the detector is
184        /// scoped to the fact's entity and walks that entity's most recent
185        /// live facts, so an input carrying no `entity` has no candidate set
186        /// and cannot block anything, now or after any number of later
187        /// writes. The fact is stored, exactly as an ordinary
188        /// [`Memory::remember`] would have stored it.
189        ///
190        /// `true` says a comparison took place, including the case where the
191        /// entity is new and its candidate set was empty - that is a real
192        /// "nothing similar is stored", and the same call will be compared
193        /// properly from the entity's second fact onwards.
194        ///
195        /// This is not a failure and not a fallback: comparing against every
196        /// live fact instead is precisely the unbounded work the per-entity
197        /// candidate cap exists to avoid. It is reported because the
198        /// alternative is silence, and silence here reads as "checked, and it
199        /// was clear".
200        checked: bool,
201    },
202    /// At least one live same-entity fact crossed a similarity threshold; no
203    /// mutation was made.
204    Blocked {
205        /// Similar / potentially conflicting live facts, best match first
206        /// (at most eight).
207        similar: Vec<Similar>,
208    },
209}
210
211/// Result of removing one tag from every current fact.
212#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
213#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
214pub struct RemoveTagReport {
215    /// Current facts superseded by an otherwise-identical revision without
216    /// the tag.
217    pub affected: u32,
218}
219
220/// One similar-fact hint.
221#[derive(Clone, Copy, Debug, PartialEq)]
222#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
223pub struct Similar {
224    /// The existing fact.
225    pub id: FactId,
226    /// Match strength (for the lexical detector: the Jaccard overlap of
227    /// the term sets, in `(similar_jaccard, 1]`).
228    pub score: f32,
229    /// What triggered the hint.
230    pub reason: SimilarReason,
231}
232
233/// Why a fact was flagged as similar.
234#[derive(Clone, Copy, Debug, PartialEq, Eq)]
235#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
236pub enum SimilarReason {
237    /// Same subject entity and a term-set overlap above the configured
238    /// Jaccard threshold.
239    LexicalOverlap,
240    /// Same subject entity and a quantized-vector cosine above the
241    /// configured `similar_cos` threshold.
242    VectorCosine,
243}
244
245/// Reusable write-side buffers for similarity detection. Kept outside the
246/// persistent pools: after warm-up guarded writes do not allocate scratch on
247/// every call, and none of these bytes enters a snapshot.
248#[derive(Debug, Default)]
249struct SimilarityScratch {
250    /// Incoming terms already present in the interner.
251    new_terms: Vec<u32>,
252    /// Unknown incoming tokens concatenated for exact deduplication without
253    /// growing the interner during a read-only guard check.
254    unknown: String,
255    /// Byte ranges in `unknown`, one per distinct unknown token.
256    unknown_ranges: Vec<(usize, usize)>,
257    /// Candidate term ids, reused across the bounded scan.
258    candidate_terms: Vec<u32>,
259    /// Quantized incoming vector slot, never appended to the vector pool.
260    vector: Vec<u8>,
261}
262
263#[derive(Clone, Copy)]
264enum SimilarVector<'s> {
265    None,
266    Stored(u32),
267    Encoded(&'s [u8]),
268}
269
270#[derive(Clone, Copy)]
271struct SimilarityQuery<'s> {
272    entity: EntityId,
273    exclude: Option<FactId>,
274    terms: &'s [u32],
275    term_count: usize,
276    vector: SimilarVector<'s>,
277}
278
279/// A read view of one fact.
280#[derive(Clone, Copy, Debug)]
281#[cfg_attr(feature = "serde", derive(serde::Serialize))]
282pub struct FactView<'a> {
283    /// The raw record (temporality, flags, references).
284    pub record: FactRecord,
285    /// The fact text.
286    pub text: &'a str,
287}
288
289/// The per-fact content problem [`Memory::faulty_facts`] attributes to a
290/// fact — the salvage predicate `recover` uses. It mirrors the
291/// content checks [`Memory::verify`] runs, split out per fact so a caller can
292/// drop the individual bad records instead of failing the whole image.
293#[derive(Clone, Copy, Debug, PartialEq, Eq)]
294#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
295pub enum FactFault {
296    /// The fact's stored text is not valid UTF-8.
297    Text,
298    /// The fact is flagged with a vector but its slot is out of range or does
299    /// not name the fact back (the fact↔slot bijection is broken).
300    Vector,
301    /// The fact's metadata blob is out of range or does not decode to a
302    /// well-formed key→value map.
303    Metadata,
304}
305
306/// Engine size counters. Every field is an O(1) read; the
307/// struct is `#[non_exhaustive]` so later stages (database identity
308/// markers, HNSW state) can extend it without a breaking change.
309#[derive(Clone, Copy, Debug, PartialEq, Eq)]
310#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
311#[non_exhaustive]
312pub struct Stats {
313    /// Fact records currently stored (live, closed, and tombstoned facts
314    /// awaiting `maintain`).
315    pub facts: usize,
316    /// Entities.
317    pub entities: usize,
318    /// Interned terms (tokens, tags, relations, normalized names).
319    pub terms: usize,
320    /// Directed edges (each `(src, rel, dst)` counted once; the mirrored
321    /// in-arena is an internal detail).
322    pub edges: usize,
323    /// Historical edge versions, including closed versions.
324    pub edge_versions: usize,
325    /// Quantized vector slots.
326    pub vectors: usize,
327    /// Tombstoned fact records awaiting physical purge.
328    pub tombstones: usize,
329    /// Vector slots already covered by the HNSW graph; slots after this are
330    /// searched as the flat tail.
331    pub hnsw_indexed: u32,
332    /// The next fact id to be assigned. Ids below it are in use or burned
333    /// (forgotten and purged) — never reissued.
334    pub next_fact: u32,
335    /// The next entity id to be assigned.
336    pub next_entity: u32,
337    /// The next edge-version id to be assigned.
338    pub next_edge: u32,
339    /// The database lineage identity ([`Config::db_uuid`]); `0` for an
340    /// unnamed database.
341    pub db_uuid: u128,
342    /// Total bytes held by the engine's pools (arenas, blob heaps, chunk
343    /// pools, the term dictionary and the vector pool).
344    pub pool_bytes: usize,
345    /// How the arenas are currently sharded.
346    ///
347    /// The engine chooses this from what it holds and moves it during
348    /// `maintain`, so it is state to observe rather than a setting to pick —
349    /// see [`Config::shards_facts`](crate::Config::shards_facts).
350    pub shards: ShardLayout,
351}
352
353/// Report of an `open`: what the journal replay found.
354#[derive(Clone, Debug, Default, PartialEq, Eq)]
355#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
356pub struct OpenReport {
357    /// Journal records applied.
358    pub replayed: usize,
359    /// Journal records skipped as already contained in the snapshot.
360    pub skipped: usize,
361    /// A torn tail record was dropped (crash between appends).
362    pub truncated_tail: bool,
363}
364
365/// The memory engine. See the module docs for staging.
366///
367/// The lifetime `'a` is the provenance of the engine's byte pools. Every
368/// owned constructor (`new`/`open`/`from_bytes`) yields a `Memory<'static>`
369/// that owns its bytes; the read-only [`Memory::from_bytes_borrowed`] path
370/// borrows them from an mmap'd snapshot instead of copying.
371pub struct Memory<'a> {
372    cfg: Config,
373    // -- data model --
374    facts: Arena<'a, FactRecord>,
375    fact_aux: Arena<'a, FactAux>,
376    entities: Arena<'a, EntityRecord>,
377    by_name: Arena<'a, EntityByName>,
378    edges_out: Arena<'a, EdgeSlot>,
379    edges_in: Arena<'a, EdgeSlot>,
380    edges_hist_out: Arena<'a, EdgeHistorySlot>,
381    edges_hist_in: Arena<'a, EdgeHistorySlot>,
382    temporal: Arena<'a, TemporalSlot>,
383    /// Fact texts and canonical entity names.
384    texts: BlobHeap<'a>,
385    /// Per-fact metadata blobs (canonical key→value, see `crate::metadata`),
386    /// referenced by [`FactAux::meta`]. Empty and inert until a fact carries
387    /// metadata; kept out of `texts` so this cold pool stays non-resident on an
388    /// mmap'd base until `show`/`export` touches it.
389    metas: BlobHeap<'a>,
390    /// Terms: tokens, tags (verbatim), relation names, normalized entity
391    /// names.
392    terms: Interner<'a>,
393    /// Per-fact tag lists (`TermId` values), handles in [`FactAux`].
394    tag_lists: ChunkPool<'a>,
395    // -- indexes --
396    bm25: Bm25Index<'a>,
397    tags_idx: IdListIndex<'a>,
398    /// Current tag counts: merged snapshot base plus changes since it.
399    tag_catalog: tags::TagCatalog,
400    entity_facts: IdListIndex<'a>,
401    /// Quantized vectors (empty and inert when `cfg.dim == 0`).
402    vecs: VecPool<'a>,
403    /// HNSW graph over the pool; empty until `maintain` crosses
404    /// `Config::flat_to_hnsw`. Slots past its `indexed` mark are the flat
405    /// tail searched by scan.
406    hnsw: HnswGraph<'a>,
407    /// Human-readable identity of the model/vector space that produced every
408    /// slot in `vecs`. `None` is legacy/untracked state, never permission to
409    /// mix a configured model into a non-empty pool.
410    vector_space: Option<String>,
411    // -- id allocation (derived from the arenas on load) --
412    next_fact: u32,
413    next_entity: u32,
414    next_edge: u32,
415    // -- maintenance state --
416    tombstones: usize,
417    bm25_tokenizer_version: u32,
418    // -- reusable scratches --
419    tokenizer: Tokenizer,
420    tf_scratch: Vec<(u32, u8)>,
421    /// Open-addressing `term -> index into `tf_scratch` + 1` table (0 is
422    /// empty), so accumulating a document's term frequencies costs one probe
423    /// per token instead of a scan of the terms already seen. Grown to the
424    /// largest document indexed and reused; see [`tally_term`].
425    tf_probe: Vec<u32>,
426    name_scratch: String,
427    similarity_scratch: SimilarityScratch,
428}
429
430impl<'a> Memory<'a> {
431    /// Creates an empty database.
432    ///
433    /// # Errors
434    ///
435    /// [`Error::ConfigMismatch`] for an invalid config (see
436    /// [`Config::validate`]).
437    pub fn new(cfg: Config) -> Result<Self, Error> {
438        cfg.validate()?;
439        let uni =
440            |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
441        let ord =
442            |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
443        let blob = BlobHeapCfg::new()
444            .with_max_bytes(cfg.max_bytes)
445            .with_max_blob(cfg.max_blob);
446        Ok(Self {
447            facts: Arena::new(uni(cfg.shards_facts))?,
448            fact_aux: Arena::new(uni(cfg.shards_facts))?,
449            entities: Arena::new(uni(cfg.shards_entities))?,
450            by_name: Arena::new(ord(cfg.shards_entities))?,
451            edges_out: Arena::new(ord(cfg.shards_edges))?,
452            edges_in: Arena::new(ord(cfg.shards_edges))?,
453            edges_hist_out: Arena::new(ord(cfg.shards_edges))?,
454            edges_hist_in: Arena::new(ord(cfg.shards_edges))?,
455            temporal: Arena::new(ord(cfg.shards_temporal))?,
456            texts: BlobHeap::new(blob),
457            metas: BlobHeap::new(blob),
458            terms: Interner::new(blob),
459            tag_lists: ChunkPool::new(ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes)),
460            bm25: Bm25Index::new(cfg.shards_postings, cfg.max_bytes)?,
461            tags_idx: IdListIndex::new(cfg.shards_postings, cfg.max_bytes)?,
462            tag_catalog: tags::TagCatalog::new(),
463            entity_facts: IdListIndex::new(cfg.shards_entities, cfg.max_bytes)?,
464            vecs: VecPool::new(cfg.dim, cfg.max_bytes),
465            hnsw: HnswGraph::new(cfg.hnsw_m, cfg.hnsw_m0, cfg.max_bytes)?,
466            vector_space: None,
467            next_fact: 0,
468            next_entity: 0,
469            next_edge: 0,
470            tombstones: 0,
471            bm25_tokenizer_version: maintain::TOKENIZER_INDEX_VERSION,
472            tokenizer: Tokenizer::new(),
473            tf_scratch: Vec::new(),
474            tf_probe: Vec::new(),
475            name_scratch: String::new(),
476            similarity_scratch: SimilarityScratch::default(),
477            cfg,
478        })
479    }
480
481    /// Opens a database from a storage: journal replay over an empty
482    /// state (the snapshot fast-path lands with the section composition;
483    /// a non-empty snapshot is rejected rather than half-read).
484    pub fn open<S: Storage>(store: &mut S, cfg: Config) -> Result<(Self, OpenReport), Error> {
485        let snapshot = store
486            .read_snapshot()
487            .map_err(|e| Error::Storage(format!("{e:?}")))?;
488        let journal = store
489            .read_journal()
490            .map_err(|e| Error::Storage(format!("{e:?}")))?;
491        Self::from_bytes(snapshot.as_deref(), &journal, cfg)
492    }
493
494    /// Opens from raw bytes (the wasm path: the host already read them).
495    pub fn from_bytes(
496        snapshot: Option<&[u8]>,
497        journal: &[u8],
498        cfg: Config,
499    ) -> Result<(Self, OpenReport), Error> {
500        let mut mem = match snapshot {
501            Some(bytes) => Self::load_snapshot(bytes, cfg)?,
502            None => Self::new(cfg)?,
503        };
504        let report = mem.replay(journal)?;
505        Ok((mem, report))
506    }
507
508    /// Opens a database read-only over a borrowed snapshot: the
509    /// engine's byte pools borrow `snapshot` (typically an mmap'd file)
510    /// instead of copying it, so a large database residents only the pages
511    /// actually read. The returned engine is tied to `snapshot`'s lifetime.
512    ///
513    /// A **non-empty journal is rejected**: the read-only handle exposes no
514    /// write verbs, so a snapshot with un-checkpointed journal tail would open
515    /// stale. Checkpoint the database read-write once first — or use
516    /// [`Memory::from_bytes_overlay`], which replays the journal into the
517    /// overlay. The caller (the host) owns the map and the exclusive lock.
518    pub fn from_bytes_borrowed(
519        snapshot: &'a [u8],
520        journal: &[u8],
521        cfg: Config,
522    ) -> Result<Self, Error> {
523        let mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
524        let JournalScan { entries, .. } = scan(journal)?;
525        if !entries.is_empty() {
526            return Err(Error::Invalid(
527                "read-only open requires a checkpointed (empty) journal",
528            ));
529        }
530        Ok(mem)
531    }
532
533    /// Opens a database read-**write** over a borrowed snapshot (the overlay
534    /// write path): like [`Memory::from_bytes_borrowed`] the byte
535    /// pools borrow `snapshot` instead of copying it, but the journal is
536    /// **replayed** and the engine stays fully mutable. Mutations do not clone
537    /// the borrowed base: the flat structures land appends in an owned tail and
538    /// copy only the pages they rewrite (per-page copy-on-write), so a
539    /// multi-gigabyte mapped database is opened and written to while resident
540    /// only in the pages it actually touches.
541    ///
542    /// This is the write sibling of [`Memory::from_bytes`] (which owns its
543    /// bytes) — the same `snapshot + journal replay`, over a borrowed base, and
544    /// it returns the same [`OpenReport`] describing the replay. The returned
545    /// engine is tied to `snapshot`'s lifetime; the caller (the host) owns the
546    /// map and the exclusive lock.
547    pub fn from_bytes_overlay(
548        snapshot: &'a [u8],
549        journal: &[u8],
550        cfg: Config,
551    ) -> Result<(Self, OpenReport), Error> {
552        let mut mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
553        let report = mem.replay(journal)?;
554        Ok((mem, report))
555    }
556
557    /// Applies every journal record on top of the current state (
558    /// replay rules: assigned ids are authoritative; records whose
559    /// assigned id is already present are skipped, which makes reapplying
560    /// a tail idempotent).
561    fn replay(&mut self, journal: &[u8]) -> Result<OpenReport, Error> {
562        let JournalScan {
563            entries,
564            truncated_tail,
565        } = scan(journal)?;
566        let mut report = OpenReport {
567            truncated_tail,
568            ..OpenReport::default()
569        };
570        for entry in entries {
571            let op = Op::decode(entry.op, entry.payload)?;
572            match op {
573                Op::Remember {
574                    now,
575                    valid_from,
576                    entity,
577                    text,
578                    ref tags,
579                    ref links,
580                    ref vector,
581                    ref metadata,
582                    revises,
583                    assigned,
584                } => {
585                    if assigned.0 < self.next_fact {
586                        report.skipped += 1;
587                        continue;
588                    }
589                    if assigned.0 != self.next_fact {
590                        return Err(Error::Corrupt("journal fact ids are not contiguous"));
591                    }
592                    if !vector.is_empty() && vector.len() != self.cfg.dim {
593                        return Err(Error::Corrupt(
594                            "journal vector dimension disagrees with dim",
595                        ));
596                    }
597                    // The record count is a `u8` on the wire, so a damaged
598                    // journal can name up to 255 tags where a fact holds
599                    // [`MAX_TAGS`]. Replay does not run `validate_input` — it
600                    // reapplies what the verbs already accepted — so the one
601                    // bound `apply_remember` indexes an array with has to be
602                    // re-established here, beside the other corruption checks.
603                    if tags.len() > MAX_TAGS {
604                        return Err(Error::Corrupt("journal record carries too many tags"));
605                    }
606                    if let Some(target) = revises.some() {
607                        self.check_revisable(target)
608                            .map_err(|_| Error::Corrupt("journal revises an unrevisable fact"))?;
609                    }
610                    self.apply_remember(
611                        &RememberInput {
612                            now,
613                            text,
614                            entity,
615                            tags: &tags.to_vec(),
616                            links: &links.to_vec(),
617                            vector: (!vector.is_empty()).then_some(vector.as_slice()),
618                            valid_from: Some(valid_from),
619                            metadata: (!metadata.is_empty()).then_some(metadata.as_slice()),
620                        },
621                        revises,
622                        None,
623                    )?;
624                    if let Some(target) = revises.some() {
625                        self.close_target(target, valid_from);
626                    }
627                    report.replayed += 1;
628                }
629                Op::Forget { fact, .. } => {
630                    // Idempotent; a missing fact in a valid journal is
631                    // corruption, but re-tombstoning is fine.
632                    match self.apply_forget(fact) {
633                        Ok(_) => report.replayed += 1,
634                        Err(Error::NotFound(_)) => {
635                            return Err(Error::Corrupt("journal forgets an unknown fact"));
636                        }
637                        Err(e) => return Err(e),
638                    }
639                }
640                Op::Link {
641                    now,
642                    src,
643                    rel,
644                    dst,
645                    provenance,
646                } => {
647                    self.apply_link(now, src, rel, dst, provenance)?;
648                    report.replayed += 1;
649                }
650                Op::Unlink { now, src, rel, dst } => {
651                    self.apply_unlink(now, src, rel, dst)?;
652                    report.replayed += 1;
653                }
654                Op::RemoveTag { now, tag } => {
655                    self.apply_remove_tag(now, tag)?;
656                    report.replayed += 1;
657                }
658                Op::SetVectorSpace { space } => {
659                    self.apply_set_vector_space(space)?;
660                    report.replayed += 1;
661                }
662                Op::Maintain {
663                    mode,
664                    max_hnsw_inserts,
665                    ..
666                } => {
667                    // Re-execute the compaction deterministically, so a
668                    // replayed image matches one snapshotted after a live
669                    // maintain byte for byte.
670                    let options =
671                        maintain::MaintenanceOptions::from_journal(mode, max_hnsw_inserts)?;
672                    self.replay_maintain_with_options(options)?;
673                    report.replayed += 1;
674                }
675            }
676        }
677        Ok(report)
678    }
679
680    /// The semantic identity persisted with the current vector pool.
681    ///
682    /// `None` means either that no automatic vector has been written yet, or
683    /// that this is a legacy image whose old model was not recorded.
684    pub fn vector_space(&self) -> Option<&str> {
685        self.vector_space.as_deref()
686    }
687
688    /// Assigns `space` to an empty vector pool and journals the assignment.
689    /// Repeating the same claim is a no-op; claiming a different or untracked
690    /// non-empty pool is refused. Explicit reembed is the only operation that
691    /// may replace a non-empty pool's identity.
692    pub fn claim_vector_space<S: Storage>(
693        &mut self,
694        store: &mut S,
695        space: &str,
696    ) -> Result<bool, Error> {
697        let needs_claim = self.check_vector_space_claim(space)?;
698        if !needs_claim {
699            return Ok(false);
700        }
701        let mut entry = Vec::new();
702        Op::SetVectorSpace { space }.encode(&mut entry);
703        store
704            .append_journal(&entry)
705            .map_err(|e| Error::Storage(format!("{e:?}")))?;
706        self.vector_space = Some(space.into());
707        Ok(true)
708    }
709
710    /// Validates a vector-space claim without changing state. `true` means an
711    /// empty pool needs its first persisted identity.
712    fn check_vector_space_claim(&self, space: &str) -> Result<bool, Error> {
713        Self::validate_vector_space(space)?;
714        if let Some(stored) = &self.vector_space {
715            if stored == space {
716                return Ok(false);
717            }
718            if !self.vecs.is_empty() {
719                return Err(Error::VectorSpaceMismatch {
720                    stored: stored.clone(),
721                    requested: space.into(),
722                });
723            }
724        }
725        if !self.vecs.is_empty() {
726            return Err(Error::UntrackedVectorSpace);
727        }
728        Ok(true)
729    }
730
731    fn apply_set_vector_space(&mut self, space: &str) -> Result<(), Error> {
732        Self::validate_vector_space(space)?;
733        if let Some(stored) = &self.vector_space {
734            if stored == space {
735                return Ok(());
736            }
737            if !self.vecs.is_empty() {
738                return Err(Error::Corrupt(
739                    "journal changes an established vector space",
740                ));
741            }
742        }
743        if !self.vecs.is_empty() {
744            return Err(Error::Corrupt(
745                "journal assigns a vector space after vector records",
746            ));
747        }
748        self.vector_space = Some(space.into());
749        Ok(())
750    }
751
752    pub(super) fn validate_vector_space(space: &str) -> Result<(), Error> {
753        if space.is_empty() {
754            return Err(Error::Invalid("vector space must not be empty"));
755        }
756        if space.len() > MAX_VECTOR_SPACE_ID_BYTES {
757            return Err(Error::TooLarge {
758                what: "vector space",
759                len: space.len(),
760                max: MAX_VECTOR_SPACE_ID_BYTES,
761            });
762        }
763        if space.bytes().any(|b| b < b' ' || b == 0x7f) {
764            return Err(Error::Invalid(
765                "vector space must not contain control bytes",
766            ));
767        }
768        Ok(())
769    }
770
771    /// Remembers a new fact.
772    pub fn remember<S: Storage>(
773        &mut self,
774        store: &mut S,
775        input: RememberInput<'_>,
776    ) -> Result<RememberOutcome, Error> {
777        self.validate_input(&input)?;
778        let mut outcome = self.apply_remember(&input, FactId::NONE, None)?;
779        self.find_similar(&mut outcome);
780        self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
781        Ok(outcome)
782    }
783
784    /// Stores a fact only when the ordinary `remember` similarity detector
785    /// finds no live same-entity candidate above its configured lexical or
786    /// vector threshold.
787    ///
788    /// The check and possible mutation are one engine operation. A blocked
789    /// call does not allocate a fact id, touch an index, or append to the
790    /// journal. The detector is exactly the one that fills
791    /// [`RememberOutcome::similar`] after an unconditional [`remember`](Self::remember);
792    /// hybrid [`recall`](Self::recall) ranking is deliberately not involved.
793    pub fn remember_guarded<S: Storage>(
794        &mut self,
795        store: &mut S,
796        input: RememberInput<'_>,
797    ) -> Result<GuardedRememberOutcome, Error> {
798        self.remember_guarded_with_vector_space(store, input, None)
799    }
800
801    /// Host integration for [`remember_guarded`](Self::remember_guarded):
802    /// claims an automatic embedder's semantic vector space only if the fact
803    /// will actually be stored. A blocked call therefore remains entirely
804    /// non-mutating, including vector-space metadata and its journal record.
805    #[doc(hidden)]
806    pub fn remember_guarded_with_vector_space<S: Storage>(
807        &mut self,
808        store: &mut S,
809        input: RememberInput<'_>,
810        vector_space: Option<&str>,
811    ) -> Result<GuardedRememberOutcome, Error> {
812        self.validate_input(&input)?;
813        if let Some(space) = vector_space {
814            self.check_vector_space_claim(space)?;
815        }
816        // Whether a check was POSSIBLE, which is a property of the call rather
817        // than of what the memory currently holds: the detector is scoped to
818        // the input's entity, so an input naming none has no candidate set now
819        // and would have none however many facts were stored later. An input
820        // that does name one is checked - against an empty set when it is that
821        // entity's first fact, which is a real "nothing similar" rather than an
822        // absent comparison.
823        let checked = input.entity.is_some();
824        let similar = self.find_similar_input(&input)?;
825        if !similar.is_empty() {
826            return Ok(GuardedRememberOutcome::Blocked { similar });
827        }
828        if let Some(space) = vector_space {
829            self.claim_vector_space(store, space)?;
830        }
831        let outcome = self.apply_remember(&input, FactId::NONE, None)?;
832        self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
833        Ok(GuardedRememberOutcome::Stored { outcome, checked })
834    }
835
836    /// Batch import: a sequence of remembers in one call, journaled
837    /// individually. `skip_similar` turns off the
838    /// similar-detection pass — imports don't need hints, and skipping
839    /// them makes a bulk load cheaper.
840    ///
841    /// Stops at the first error; already-applied inputs stay applied and
842    /// journaled (replay reproduces exactly the applied prefix).
843    pub fn remember_batch<S: Storage>(
844        &mut self,
845        store: &mut S,
846        inputs: &[RememberInput<'_>],
847        skip_similar: bool,
848    ) -> Result<Vec<RememberOutcome>, Error> {
849        let mut out = Vec::with_capacity(inputs.len());
850        for input in inputs {
851            self.validate_input(input)?;
852            let mut outcome = self.apply_remember(input, FactId::NONE, None)?;
853            if !skip_similar {
854                self.find_similar(&mut outcome);
855            }
856            self.journal_remember(store, input, FactId::NONE, outcome.id)?;
857            out.push(outcome);
858        }
859        Ok(out)
860    }
861
862    /// Runs similarity detection against an incoming, not-yet-stored fact.
863    /// Unknown tokens are counted in the Jaccard union but never interned:
864    /// a blocked call is observationally a read and cannot grow vocabulary.
865    fn find_similar_input(&mut self, input: &RememberInput<'_>) -> Result<Vec<Similar>, Error> {
866        let Some(name) = input.entity else {
867            return Ok(Vec::new());
868        };
869        let Some(entity) = self.lookup_entity_name(name) else {
870            return Ok(Vec::new());
871        };
872
873        let mut scratch = core::mem::take(&mut self.similarity_scratch);
874        scratch.new_terms.clear();
875        scratch.unknown.clear();
876        scratch.unknown_ranges.clear();
877        scratch.candidate_terms.clear();
878        scratch.vector.clear();
879
880        let terms = &self.terms;
881        let new_terms = &mut scratch.new_terms;
882        let unknown = &mut scratch.unknown;
883        let unknown_ranges = &mut scratch.unknown_ranges;
884        self.tokenizer.tokenize(input.text, &mut |token| {
885            if let Some(term) = terms.lookup(token) {
886                if !new_terms.contains(&term.0) {
887                    new_terms.push(term.0);
888                }
889                return;
890            }
891            if unknown_ranges
892                .iter()
893                .any(|&(start, end)| &unknown[start..end] == token)
894            {
895                return;
896            }
897            let start = unknown.len();
898            unknown.push_str(token);
899            unknown_ranges.push((start, unknown.len()));
900        });
901        let new_term_count = scratch.new_terms.len() + scratch.unknown_ranges.len();
902
903        let mut similar = Vec::new();
904        let result = (|| {
905            let encoded = match input.vector {
906                Some(vector) => {
907                    self.vecs
908                        .encode_slot_into(FactId::NONE, vector, &mut scratch.vector)?;
909                    SimilarVector::Encoded(&scratch.vector)
910                }
911                None => SimilarVector::None,
912            };
913            self.scan_similar(
914                SimilarityQuery {
915                    entity,
916                    exclude: None,
917                    terms: &scratch.new_terms,
918                    term_count: new_term_count,
919                    vector: encoded,
920                },
921                &mut scratch.candidate_terms,
922                &mut similar,
923            );
924            Ok::<(), Error>(())
925        })();
926        self.similarity_scratch = scratch;
927        result.map(|()| similar)
928    }
929
930    /// Similarity detection for an already-stored ordinary remember. This
931    /// feeds the same scanner as `remember_guarded`, but reuses the term ids and
932    /// quantized vector slot the write has just produced.
933    fn find_similar(&mut self, outcome: &mut RememberOutcome) {
934        let Some(entity) = outcome.entity else { return };
935        let new_vec = self
936            .fact(outcome.id)
937            .filter(|record| record.has_vector())
938            .map(|record| record.vector);
939        let mut scratch = core::mem::take(&mut self.similarity_scratch);
940        scratch.new_terms.clear();
941        scratch
942            .new_terms
943            .extend(self.tf_scratch.iter().map(|&(term, _)| term));
944        let new_term_count = scratch.new_terms.len();
945        self.scan_similar(
946            SimilarityQuery {
947                entity,
948                exclude: Some(outcome.id),
949                terms: &scratch.new_terms,
950                term_count: new_term_count,
951                vector: new_vec.map_or(SimilarVector::None, SimilarVector::Stored),
952            },
953            &mut scratch.candidate_terms,
954            &mut outcome.similar,
955        );
956        self.similarity_scratch = scratch;
957    }
958
959    /// Lexical similar-detection: live facts of the same
960    /// entity whose term sets overlap the new fact's above the
961    /// `similar_jaccard` threshold. Bounded: the entity's most recent
962    /// [`SIMILAR_CANDIDATE_CAP`] facts are compared (a hub's full list is
963    /// not re-tokenized).
964    ///
965    /// Comparing two term sets exactly means having both, and only the new
966    /// fact's is at hand — the candidate's would have to be recovered by
967    /// reading its text and running it back through the tokenizer. Doing that
968    /// for every candidate of every write is what the term-set summary in
969    /// [`DocLenSlot`](crate::index::bm25::DocLenSlot) exists to avoid: it
970    /// bounds the overlap from above, and a bound below the threshold settles
971    /// the question, because Jaccard rises with the intersection. Only a
972    /// candidate that survives the bound is read and tokenized, so the
973    /// answer is the same one the exhaustive comparison gives.
974    ///
975    /// The bound is trusted only while the index was built by the current
976    /// tokenizer. A stale index may hold terms today's tokenizer would not
977    /// produce, which would make the summary describe a different term set
978    /// than the comparison uses; then every candidate is read, exactly as
979    /// before the summary existed.
980    fn scan_similar(
981        &mut self,
982        query: SimilarityQuery<'_>,
983        candidate_terms: &mut Vec<u32>,
984        out: &mut Vec<Similar>,
985    ) {
986        out.clear();
987        if query.term_count == 0 && matches!(query.vector, SimilarVector::None) {
988            return;
989        }
990        // Most recent candidates of the entity (ring over the list).
991        let mut ring = [FactId::NONE; SIMILAR_CANDIDATE_CAP];
992        let mut n = 0usize;
993        for (fact, _) in self.entity_facts.entries(query.entity.0) {
994            if query.exclude == Some(fact) {
995                continue;
996            }
997            ring[n % SIMILAR_CANDIDATE_CAP] = fact;
998            n += 1;
999        }
1000        let summaries_trustworthy = self.bm25_tokenizer_version == TOKENIZER_INDEX_VERSION;
1001        // Without a vector on the new fact, a lexical overlap is the only hint
1002        // a candidate can produce, so one ruled out by the summary contributes
1003        // nothing whatever its record says — and its record is the second
1004        // arena lookup of the pair this loop would otherwise pay per
1005        // candidate. (`new_terms` is non-empty here: the two being empty
1006        // together returned above.)
1007        let lexical_only = matches!(query.vector, SimilarVector::None);
1008        for &fact in ring.iter().take(n.min(SIMILAR_CANDIDATE_CAP)) {
1009            let may_overlap = !query.terms.is_empty()
1010                && self.overlap_possible(
1011                    fact,
1012                    query.terms,
1013                    query.term_count,
1014                    summaries_trustworthy,
1015                );
1016            if lexical_only && !may_overlap {
1017                continue;
1018            }
1019            let Some(record) = self.fact(fact) else {
1020                continue;
1021            };
1022            if record.is_tombstone() || record.is_closed() {
1023                continue;
1024            }
1025
1026            // Lexical signal: term-set Jaccard over the entity's text.
1027            let mut lexical = None;
1028            // Deferred validation: a load no longer scans the
1029            // text pool, so an unreadable text simply yields no lexical signal.
1030            if may_overlap && let Ok(text) = core::str::from_utf8(self.texts.get(record.text)) {
1031                candidate_terms.clear();
1032                let terms = &self.terms;
1033                let cand = &mut *candidate_terms;
1034                self.tokenizer.tokenize(text, &mut |token| {
1035                    if let Some(term) = terms.lookup(token)
1036                        && !cand.contains(&term.0)
1037                    {
1038                        cand.push(term.0);
1039                    }
1040                });
1041                if !candidate_terms.is_empty() {
1042                    let both = candidate_terms
1043                        .iter()
1044                        .filter(|term| query.terms.contains(term))
1045                        .count();
1046                    let union = candidate_terms.len() + query.term_count - both;
1047                    let jaccard = both as f32 / union as f32;
1048                    if jaccard > self.cfg.similar_jaccard {
1049                        lexical = Some(jaccard);
1050                    }
1051                }
1052            }
1053
1054            // Vector signal: quantized cosine when both facts carry one.
1055            let mut vector = None;
1056            if record.has_vector() {
1057                let cos = match query.vector {
1058                    SimilarVector::None => 0.0,
1059                    SimilarVector::Stored(slot) => self.vecs.cosine_slots(slot, record.vector),
1060                    SimilarVector::Encoded(encoded) => {
1061                        self.vecs.cosine_encoded_slot(encoded, record.vector)
1062                    }
1063                };
1064                if cos > self.cfg.similar_cos {
1065                    vector = Some(cos);
1066                }
1067            }
1068
1069            // Keep the stronger signal; a tie prefers the lexical reason.
1070            let best = match (lexical, vector) {
1071                (Some(l), Some(v)) if v > l => Some((v, SimilarReason::VectorCosine)),
1072                (Some(l), _) => Some((l, SimilarReason::LexicalOverlap)),
1073                (None, Some(v)) => Some((v, SimilarReason::VectorCosine)),
1074                (None, None) => None,
1075            };
1076            if let Some((score, reason)) = best {
1077                out.push(Similar {
1078                    id: fact,
1079                    score,
1080                    reason,
1081                });
1082            }
1083        }
1084        out.sort_unstable_by(|a, b| b.score.total_cmp(&a.score).then(a.id.cmp(&b.id)));
1085        out.truncate(8);
1086    }
1087
1088    /// Whether `candidate` could possibly overlap `new_terms` above
1089    /// `similar_jaccard` — the cheap half of [`Memory::find_similar`].
1090    ///
1091    /// `false` is a proof, not a guess. The candidate's summary marks a term
1092    /// absent only when it really is absent, so the terms it *might* share is
1093    /// an upper bound `u` on the true intersection `i`. Jaccard
1094    /// `i / (|A| + |B| - i)` is increasing in `i`, so `u` at or below the
1095    /// threshold puts the true value there too. `true` means "unknown" and
1096    /// sends the caller to the exact comparison — which is also what an
1097    /// unsummarized document and an untrustworthy index return.
1098    fn overlap_possible(
1099        &self,
1100        candidate: FactId,
1101        new_terms: &[u32],
1102        new_term_count: usize,
1103        trust_summary: bool,
1104    ) -> bool {
1105        if !trust_summary {
1106            return true;
1107        }
1108        let Some(doc) = self.bm25.doc(candidate) else {
1109            return true;
1110        };
1111        if !doc.has_signature() {
1112            return true;
1113        }
1114        let bound = doc.overlap_bound(new_terms);
1115        // `overlap_bound` counts a subset of `new_terms`, so the subtraction
1116        // is grouped to happen where it provably cannot go below zero. The
1117        // union is then at least `distinct`, which `has_signature` just
1118        // established is non-zero — no underflow and no division by zero, on a
1119        // 64-bit `usize` or a 32-bit one.
1120        debug_assert!(
1121            bound <= new_terms.len(),
1122            "the overlap bound counts query terms, so it cannot exceed them"
1123        );
1124        let union = (new_term_count - bound) + usize::from(doc.distinct);
1125        bound as f32 / union as f32 > self.cfg.similar_jaccard
1126    }
1127
1128    /// Revises `target`: closes its validity at the new fact's
1129    /// `valid_from` and records the new fact with `revises = target`
1130    /// (rule 2).
1131    pub fn revise<S: Storage>(
1132        &mut self,
1133        store: &mut S,
1134        target: FactId,
1135        input: RememberInput<'_>,
1136    ) -> Result<RememberOutcome, Error> {
1137        self.validate_input(&input)?;
1138        // Check first, mutate last: the target closes only after the new
1139        // fact exists, so a mid-operation failure (capacity) never leaves
1140        // a closed fact without its successor.
1141        self.check_revisable(target)?;
1142        let outcome = self.apply_remember(&input, target, None)?;
1143        let valid_from = input.valid_from.unwrap_or(input.now);
1144        self.close_target(target, valid_from);
1145        self.journal_remember(store, &input, target, outcome.id)?;
1146        Ok(outcome)
1147    }
1148
1149    /// Tombstones a fact: it disappears from every query immediately, the
1150    /// bytes go with the next `maintain` (rule 3). `Ok(false)`
1151    /// when it was already tombstoned.
1152    pub fn forget<S: Storage>(
1153        &mut self,
1154        store: &mut S,
1155        now: u64,
1156        id: FactId,
1157    ) -> Result<bool, Error> {
1158        let fresh = self.apply_forget(id)?;
1159        let mut entry = Vec::new();
1160        Op::Forget { now, fact: id }.encode(&mut entry);
1161        store
1162            .append_journal(&entry)
1163            .map_err(|e| Error::Storage(format!("{e:?}")))?;
1164        Ok(fresh)
1165    }
1166
1167    /// Removes `tag` from every current fact without deleting knowledge.
1168    ///
1169    /// Each affected fact is closed and replaced by an otherwise-identical
1170    /// successor without the tag. Historical `as_of` queries keep seeing the
1171    /// old classification; current recall and [`Memory::list_tags`] stop
1172    /// seeing it immediately. The interned string is stable historical residue
1173    /// and is not physically reclaimed.
1174    pub fn remove_tag<S: Storage>(
1175        &mut self,
1176        store: &mut S,
1177        now: u64,
1178        tag: &str,
1179    ) -> Result<RemoveTagReport, Error> {
1180        if tag.is_empty() {
1181            return Err(Error::Invalid("empty tag"));
1182        }
1183        let report = self.apply_remove_tag(now, tag)?;
1184        if report.affected != 0 {
1185            let mut entry = Vec::new();
1186            Op::RemoveTag { now, tag }.encode(&mut entry);
1187            store
1188                .append_journal(&entry)
1189                .map_err(|e| Error::Storage(format!("{e:?}")))?;
1190        }
1191        Ok(report)
1192    }
1193
1194    /// Upserts a typed edge between two entities (created lazily);
1195    /// re-linking an existing `(src, rel, dst)` updates its provenance.
1196    pub fn link<S: Storage>(&mut self, store: &mut S, input: LinkInput<'_>) -> Result<(), Error> {
1197        self.apply_link(
1198            input.now,
1199            input.src,
1200            input.rel,
1201            input.dst,
1202            FactId::from_opt(input.provenance),
1203        )?;
1204        let mut entry = Vec::new();
1205        Op::Link {
1206            now: input.now,
1207            src: input.src,
1208            rel: input.rel,
1209            dst: input.dst,
1210            provenance: FactId::from_opt(input.provenance),
1211        }
1212        .encode(&mut entry);
1213        store
1214            .append_journal(&entry)
1215            .map_err(|e| Error::Storage(format!("{e:?}")))?;
1216        Ok(())
1217    }
1218
1219    /// Closes the current typed edge between two entities. Returns `false`
1220    /// when the edge is already absent.
1221    pub fn unlink<S: Storage>(
1222        &mut self,
1223        store: &mut S,
1224        input: UnlinkInput<'_>,
1225    ) -> Result<bool, Error> {
1226        let fresh = self.apply_unlink(input.now, input.src, input.rel, input.dst)?;
1227        let mut entry = Vec::new();
1228        Op::Unlink {
1229            now: input.now,
1230            src: input.src,
1231            rel: input.rel,
1232            dst: input.dst,
1233        }
1234        .encode(&mut entry);
1235        store
1236            .append_journal(&entry)
1237            .map_err(|e| Error::Storage(format!("{e:?}")))?;
1238        Ok(fresh)
1239    }
1240
1241    /// Returns a fact unless it is tombstoned (closed facts are
1242    /// returned — their interval says so).
1243    pub fn get(&self, id: FactId) -> Option<FactView<'_>> {
1244        let record = self.fact(id)?;
1245        if record.is_tombstone() {
1246            return None;
1247        }
1248        // Deferred validation: a load no longer scans the text
1249        // pool, so tolerate invalid bytes here — a corrupt text hides the fact
1250        // rather than panicking. `verify()` reports it explicitly.
1251        let text = core::str::from_utf8(self.texts.get(record.text)).ok()?;
1252        Some(FactView { record, text })
1253    }
1254
1255    /// Appends the tag terms of a fact to `out`. A tombstoned or unknown
1256    /// fact contributes nothing.
1257    pub fn tags_of(&self, id: FactId, out: &mut Vec<TermId>) {
1258        let Some(record) = self.fact(id) else { return };
1259        if record.is_tombstone() {
1260            return;
1261        }
1262        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
1263            return;
1264        };
1265        for chunk in self.tag_lists.iter(&aux.tags) {
1266            for raw in chunk.chunks_exact(4) {
1267                out.push(TermId(u32::from_be_bytes(raw.try_into().unwrap())));
1268            }
1269        }
1270    }
1271
1272    /// Returns one bounded, stable page of current tags in lexical order.
1273    ///
1274    /// A cursor is tied to the tag catalog state and prefix. If a concurrent
1275    /// mutation changes any current tag count, continuing returns
1276    /// [`Error::StaleCursor`] rather than silently skipping or duplicating a
1277    /// tag.
1278    pub fn list_tags(&self, query: TagQuery<'_>) -> Result<TagPage, Error> {
1279        self.tag_catalog.page(&self.terms, self.cfg.db_uuid, query)
1280    }
1281
1282    /// Fills `out` with the fact's metadata as key→value pairs in canonical
1283    /// (ascending-key) order — the same order the blob was written and the same
1284    /// order a host `BTreeMap` yields, so no layer sorts twice. Returns `true`
1285    /// when the fact carries metadata (even an empty map is `false`, since an
1286    /// empty map is stored as no blob).
1287    ///
1288    /// A tombstoned or unknown fact, a fact with no metadata blob, or a blob
1289    /// that fails to decode (deferred validation, — `verify`
1290    /// reports it) all leave `out` empty and return `false`; this accessor never
1291    /// panics on bad bytes.
1292    pub fn metadata_of<'s>(&'s self, id: FactId, out: &mut Vec<(&'s str, &'s str)>) -> bool {
1293        out.clear();
1294        let Some(record) = self.fact(id) else {
1295            return false;
1296        };
1297        if record.is_tombstone() {
1298            return false;
1299        }
1300        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
1301            return false;
1302        };
1303        if aux.meta.0 == NONE_U32 || aux.meta.0 >= self.metas.len() as u32 {
1304            return false;
1305        }
1306        crate::metadata::decode(self.metas.get(aux.meta), out).is_ok() && !out.is_empty()
1307    }
1308
1309    /// Resolves an entity by name (normalized), without creating it.
1310    pub fn entity(&mut self, name: &str) -> Option<EntityId> {
1311        let mut norm = core::mem::take(&mut self.name_scratch);
1312        normalize_name(&mut self.tokenizer, name, &mut norm);
1313        let found = if norm.is_empty() {
1314            None
1315        } else {
1316            // Only resolve an *existing* term: interning would create it.
1317            self.lookup_entity_by_norm(&norm)
1318        };
1319        self.name_scratch = norm;
1320        found
1321    }
1322
1323    /// Resolves the string behind a term id (tags, relations).
1324    pub fn term(&self, id: TermId) -> &str {
1325        self.terms.resolve(id)
1326    }
1327
1328    /// The canonical (verbatim) name of an entity, or `None` for an
1329    /// unknown id. The read-side inverse of [`Memory::entity`]: it lets a
1330    /// host export a fact with its subject *name* instead of the internal
1331    /// [`EntityId`] carried by [`FactRecord`].
1332    pub fn entity_name(&self, id: EntityId) -> Option<&str> {
1333        let record = self.entities.get(&id.0.to_be_bytes())?;
1334        // Tolerate invalid bytes (deferred validation): an
1335        // unreadable name reads as `None`, never a panic.
1336        core::str::from_utf8(self.texts.get(record.name)).ok()
1337    }
1338
1339    /// Visits every currently-open edge once, as
1340    /// `(source name, relation, destination name, provenance)`.
1341    ///
1342    /// Reads the out-arena only: each edge is mirrored in both arenas, and the
1343    /// out-arena holds it keyed `(src, rel, dst)`, which is the direction a
1344    /// caller wants to write back. Closed versions live in the history arenas
1345    /// and are not visited — this is the *current* graph, matching what a fact
1346    /// export does for facts.
1347    ///
1348    /// Names are borrowed from the interner and the text pool, so a caller that
1349    /// writes them straight out allocates nothing per edge. An edge whose
1350    /// endpoint name is unreadable is skipped rather than reported with a
1351    /// placeholder: deferred validation means a damaged image must not turn
1352    /// into plausible-looking output.
1353    ///
1354    /// `visit` returning `false` stops the walk, so a caller can bound its own
1355    /// work without materializing the graph.
1356    pub fn edges_each(&self, mut visit: impl FnMut(&str, &str, &str, FactId) -> bool) {
1357        for slot in self.edges_out.iter() {
1358            let (Some(src), Some(dst)) = (self.entity_name(slot.a), self.entity_name(slot.b))
1359            else {
1360                continue;
1361            };
1362            if !visit(src, self.term(slot.rel), dst, slot.fact) {
1363                return;
1364            }
1365        }
1366    }
1367
1368    /// Number of fact records currently stored (tombstoned facts count
1369    /// until `maintain` removes them). Purged ids stay burned, so this can
1370    /// be below [`Stats::next_fact`].
1371    pub fn facts_len(&self) -> usize {
1372        self.facts.len()
1373    }
1374
1375    /// Number of entities.
1376    pub fn entities_len(&self) -> usize {
1377        self.entities.len()
1378    }
1379
1380    /// The engine configuration.
1381    pub fn cfg(&self) -> &Config {
1382        &self.cfg
1383    }
1384
1385    /// Size counters of the engine. O(1).
1386    pub fn stats(&self) -> Stats {
1387        Stats {
1388            facts: self.facts.len(),
1389            entities: self.entities.len(),
1390            terms: self.terms.len(),
1391            edges: self.edges_out.len(),
1392            edge_versions: self.edges_hist_out.len(),
1393            vectors: self.vecs.len(),
1394            tombstones: self.tombstones,
1395            hnsw_indexed: self.hnsw.indexed(),
1396            next_fact: self.next_fact,
1397            next_entity: self.next_entity,
1398            next_edge: self.next_edge,
1399            db_uuid: self.cfg.db_uuid,
1400            pool_bytes: self.facts.pool_bytes()
1401                + self.fact_aux.pool_bytes()
1402                + self.entities.pool_bytes()
1403                + self.by_name.pool_bytes()
1404                + self.edges_out.pool_bytes()
1405                + self.edges_in.pool_bytes()
1406                + self.edges_hist_out.pool_bytes()
1407                + self.edges_hist_in.pool_bytes()
1408                + self.temporal.pool_bytes()
1409                + self.texts.pool_bytes()
1410                + self.terms.pool_bytes()
1411                + self.tag_lists.pool_bytes()
1412                + self.bm25.pool_bytes()
1413                + self.tags_idx.pool_bytes()
1414                + self.tag_catalog.pool_bytes()
1415                + self.entity_facts.pool_bytes()
1416                + self.vecs.pool_bytes()
1417                + self.hnsw.pool_bytes(),
1418            shards: shards::ShardLayout::of_config(&self.cfg),
1419        }
1420    }
1421
1422    // ---- internals ----
1423
1424    fn fact(&self, id: FactId) -> Option<FactRecord> {
1425        self.facts.get(&id.0.to_be_bytes())
1426    }
1427
1428    /// Size-limit checks shared by remember and revise.
1429    fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
1430        if input.text.len() > self.cfg.max_text {
1431            return Err(Error::TooLarge {
1432                what: "text",
1433                len: input.text.len(),
1434                max: self.cfg.max_text,
1435            });
1436        }
1437        if input.tags.len() > MAX_TAGS {
1438            return Err(Error::TooLarge {
1439                what: "tags",
1440                len: input.tags.len(),
1441                max: MAX_TAGS,
1442            });
1443        }
1444        if input.links.len() > 16 {
1445            return Err(Error::TooLarge {
1446                what: "links",
1447                len: input.links.len(),
1448                max: 16,
1449            });
1450        }
1451        if input.tags.iter().any(|t| t.is_empty()) {
1452            return Err(Error::Invalid("empty tag"));
1453        }
1454        if !input.links.is_empty() && input.entity.is_none() {
1455            return Err(Error::Invalid("links require a subject entity"));
1456        }
1457        if let Some(v) = input.vector {
1458            if self.cfg.dim == 0 {
1459                return Err(Error::Invalid("vector given but dim is 0"));
1460            }
1461            if v.len() != self.cfg.dim {
1462                return Err(Error::DimMismatch {
1463                    got: v.len(),
1464                    want: self.cfg.dim,
1465                });
1466            }
1467        }
1468        Ok(())
1469    }
1470
1471    /// `Ok` when `target` exists, is not tombstoned and not yet closed.
1472    fn check_revisable(&self, target: FactId) -> Result<(), Error> {
1473        let record = self.fact(target).ok_or(Error::NotFound(target))?;
1474        if record.is_tombstone() {
1475            return Err(Error::NotFound(target));
1476        }
1477        if record.is_closed() {
1478            return Err(Error::AlreadyClosed(target));
1479        }
1480        Ok(())
1481    }
1482
1483    /// Closes a revision target (validity ends where the successor
1484    /// starts). The caller checked [`Memory::check_revisable`] first.
1485    fn close_target(&mut self, target: FactId, valid_to: u64) {
1486        let record = self.fact(target).expect("checked revisable");
1487        let payload = self
1488            .facts
1489            .payload_mut(&target.0.to_be_bytes())
1490            .expect("record fetched above");
1491        // Payload offsets = slot offsets - KEY_LEN(4): flags at 4..6,
1492        // valid_to at 36..44.
1493        let flags = record.flags | fact_flags::CLOSED;
1494        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1495        payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
1496        self.change_catalog_for_fact(target, -1);
1497    }
1498
1499    /// The one execution path of a new fact — called by the public verbs
1500    /// and by replay with identical effects.
1501    fn apply_remember(
1502        &mut self,
1503        input: &RememberInput<'_>,
1504        revises: FactId,
1505        copy_vector: Option<u32>,
1506    ) -> Result<RememberOutcome, Error> {
1507        let id = FactId(self.next_fact);
1508        let entity = match input.entity {
1509            Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
1510            None => None,
1511        };
1512        let text_id = self.texts.push(input.text.as_bytes())?;
1513
1514        // Tokenize into (term, tf) pairs in the reusable scratch.
1515        let mut tfs = core::mem::take(&mut self.tf_scratch);
1516        let mut probe = core::mem::take(&mut self.tf_probe);
1517        tfs.clear();
1518        probe.fill(0);
1519        let terms = &mut self.terms;
1520        let mut intern_err = None;
1521        self.tokenizer.tokenize(input.text, &mut |token| {
1522            if intern_err.is_some() {
1523                return;
1524            }
1525            match terms.intern(token) {
1526                Ok(term) => tally_term(&mut tfs, &mut probe, term.0),
1527                Err(e) => intern_err = Some(e),
1528            }
1529        });
1530        self.tf_probe = probe;
1531        if let Some(e) = intern_err {
1532            self.tf_scratch = tfs;
1533            return Err(Error::Arena(e));
1534        }
1535        self.bm25.index_doc(id, &tfs)?;
1536        self.tf_scratch = tfs;
1537
1538        // Metadata: canonicalized (keys sorted, dups rejected) and stored as one
1539        // opaque blob; an absent or empty map leaves the sentinel. Pushed before
1540        // the fact record so a capacity/validation failure aborts the whole op.
1541        let meta = match input.metadata {
1542            Some(pairs) if !pairs.is_empty() => {
1543                self.metas.push(&crate::metadata::encode(pairs)?)?
1544            }
1545            _ => BlobId(NONE_U32),
1546        };
1547
1548        // Tags: interned verbatim, deduplicated, listed on the fact and
1549        // inverted.
1550        let mut aux = FactAux {
1551            id,
1552            tags: ListHandle::EMPTY,
1553            meta,
1554        };
1555        let mut seen_tags: [u32; MAX_TAGS] = [NONE_U32; MAX_TAGS];
1556        let mut seen_cnt = 0usize;
1557        for tag in input.tags {
1558            let term = self.terms.intern(tag)?;
1559            if seen_tags[..seen_cnt].contains(&term.0) {
1560                continue;
1561            }
1562            seen_tags[seen_cnt] = term.0;
1563            seen_cnt += 1;
1564            self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
1565            self.tags_idx.push(term.0, id, 0)?;
1566        }
1567        self.fact_aux.insert(&aux)?;
1568
1569        // Links: subject → target edges with this fact as provenance.
1570        if let Some(src) = entity {
1571            for &(rel, dst_name) in input.links {
1572                let dst = self.resolve_or_create_entity(dst_name, input.now)?;
1573                let rel = self.terms.intern(rel)?;
1574                self.open_edge(input.now, src, rel, dst, id)?;
1575            }
1576            self.entity_facts.push(src.0, id, 0)?;
1577        }
1578
1579        // Vector: quantized into the flat pool; the fact keeps the slot
1580        // index. Pushed before the record so a capacity failure aborts the
1581        // whole op (replay rebuilds consistently, like the other indexes).
1582        let (vector, flags) = match (input.vector, copy_vector) {
1583            (Some(v), None) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
1584            (None, Some(source)) => (
1585                self.vecs.clone_slot_for_fact(id, source)?,
1586                fact_flags::HAS_VECTOR,
1587            ),
1588            (None, None) => (NONE_U32, 0),
1589            (Some(_), Some(_)) => unreachable!("retag does not provide a raw vector"),
1590        };
1591
1592        let recorded_at = input.now;
1593        let valid_from = input.valid_from.unwrap_or(input.now);
1594        self.facts.insert(&FactRecord {
1595            id,
1596            entity: EntityId::from_opt(entity),
1597            flags,
1598            kind: 0,
1599            text: text_id,
1600            vector,
1601            revises,
1602            recorded_at,
1603            valid_from,
1604            valid_to: VALID_TO_OPEN,
1605        })?;
1606        self.temporal.insert(&TemporalSlot {
1607            recorded_at,
1608            fact: id,
1609        })?;
1610        for &term in &seen_tags[..seen_cnt] {
1611            self.tag_catalog.change(&self.terms, TermId(term), 1);
1612        }
1613        self.next_fact += 1;
1614        Ok(RememberOutcome {
1615            id,
1616            entity,
1617            similar: Vec::new(),
1618        })
1619    }
1620
1621    fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
1622        let record = self.fact(id).ok_or(Error::NotFound(id))?;
1623        if record.is_tombstone() {
1624            return Ok(false);
1625        }
1626        let payload = self
1627            .facts
1628            .payload_mut(&id.0.to_be_bytes())
1629            .expect("record fetched above");
1630        let flags = record.flags | fact_flags::TOMBSTONE;
1631        payload[4..6].copy_from_slice(&flags.to_be_bytes());
1632        self.tombstones += 1;
1633        if !record.is_closed() {
1634            self.change_catalog_for_fact(id, -1);
1635        }
1636        Ok(true)
1637    }
1638
1639    fn apply_remove_tag(&mut self, now: u64, tag: &str) -> Result<RemoveTagReport, Error> {
1640        let Some(term) = self.terms.lookup(tag) else {
1641            return Ok(RemoveTagReport::default());
1642        };
1643        let targets: Vec<FactId> = self
1644            .tags_idx
1645            .entries(term.0)
1646            .map(|(id, _)| id)
1647            .filter(|&id| {
1648                self.fact(id)
1649                    .is_some_and(|record| !record.is_tombstone() && !record.is_closed())
1650            })
1651            .collect();
1652        let mut affected = 0u32;
1653        for target in targets {
1654            self.retag_without(now, target, tag)?;
1655            affected = affected.saturating_add(1);
1656        }
1657        Ok(RemoveTagReport { affected })
1658    }
1659
1660    fn retag_without(&mut self, now: u64, target: FactId, removed: &str) -> Result<(), Error> {
1661        let record = self.fact(target).ok_or(Error::NotFound(target))?;
1662        let view = self.get(target).ok_or(Error::NotFound(target))?;
1663        let text = view.text.to_string();
1664        let entity = record
1665            .entity
1666            .some()
1667            .and_then(|id| self.entity_name(id))
1668            .map(ToString::to_string);
1669
1670        let mut tag_terms = Vec::new();
1671        self.tags_of(target, &mut tag_terms);
1672        let tags: Vec<String> = tag_terms
1673            .into_iter()
1674            .map(|term| self.term(term))
1675            .filter(|name| *name != removed)
1676            .map(ToString::to_string)
1677            .collect();
1678        let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();
1679
1680        let mut metadata = Vec::new();
1681        self.metadata_of(target, &mut metadata);
1682        let metadata: Vec<(String, String)> = metadata
1683            .into_iter()
1684            .map(|(key, value)| (key.to_string(), value.to_string()))
1685            .collect();
1686        let metadata_refs: Vec<(&str, &str)> = metadata
1687            .iter()
1688            .map(|(key, value)| (key.as_str(), value.as_str()))
1689            .collect();
1690        let vector = (record.flags & fact_flags::HAS_VECTOR != 0).then_some(record.vector);
1691        let input = RememberInput {
1692            now,
1693            text: &text,
1694            entity: entity.as_deref(),
1695            tags: &tag_refs,
1696            links: &[],
1697            vector: None,
1698            valid_from: Some(now),
1699            metadata: (!metadata_refs.is_empty()).then_some(metadata_refs.as_slice()),
1700        };
1701        self.apply_remember(&input, target, vector)?;
1702        self.close_target(target, now);
1703        Ok(())
1704    }
1705
1706    /// Applies one current-count change for every tag attached to `id`.
1707    /// Facts accept at most 32 distinct tags, so this uses fixed stack storage
1708    /// and never allocates on revise/forget.
1709    fn change_catalog_for_fact(&mut self, id: FactId, delta: i32) {
1710        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
1711            return;
1712        };
1713        let mut ids = [NONE_U32; MAX_TAGS];
1714        let mut len = 0usize;
1715        for chunk in self.tag_lists.iter(&aux.tags) {
1716            for raw in chunk.chunks_exact(4) {
1717                if len == ids.len() {
1718                    break;
1719                }
1720                ids[len] = u32::from_be_bytes(raw.try_into().unwrap());
1721                len += 1;
1722            }
1723        }
1724        for &term in &ids[..len] {
1725            self.tag_catalog.change(&self.terms, TermId(term), delta);
1726        }
1727    }
1728
1729    fn apply_link(
1730        &mut self,
1731        now: u64,
1732        src: &str,
1733        rel: &str,
1734        dst: &str,
1735        provenance: FactId,
1736    ) -> Result<(), Error> {
1737        let src = self.resolve_or_create_entity(src, now)?;
1738        let dst = self.resolve_or_create_entity(dst, now)?;
1739        let rel = self.terms.intern(rel)?;
1740        self.open_edge(now, src, rel, dst, provenance)
1741    }
1742
1743    fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
1744        let Some(src) = self.lookup_entity_name(src) else {
1745            return Ok(false);
1746        };
1747        let Some(dst) = self.lookup_entity_name(dst) else {
1748            return Ok(false);
1749        };
1750        let Some(rel) = self.terms.lookup(rel) else {
1751            return Ok(false);
1752        };
1753        self.close_current_edge(now, src, rel, dst)
1754    }
1755
1756    fn open_edge(
1757        &mut self,
1758        now: u64,
1759        src: EntityId,
1760        rel: TermId,
1761        dst: EntityId,
1762        fact: FactId,
1763    ) -> Result<(), Error> {
1764        if let Some(current) = self.current_edge(src, rel, dst) {
1765            if current.fact == fact {
1766                return Ok(());
1767            }
1768            self.close_current_edge(now, src, rel, dst)?;
1769        }
1770        let edge = EdgeId(self.next_edge);
1771        let history = EdgeHistorySlot {
1772            a: src,
1773            rel,
1774            b: dst,
1775            edge,
1776            fact,
1777            flags: 0,
1778            kind: 0,
1779            recorded_at: now,
1780            valid_from: now,
1781            valid_to: VALID_TO_OPEN,
1782        };
1783        self.insert_history_edge(history)?;
1784        self.insert_current_edge(src, rel, dst, fact, edge, now)?;
1785        self.next_edge += 1;
1786        Ok(())
1787    }
1788
1789    fn insert_current_edge(
1790        &mut self,
1791        src: EntityId,
1792        rel: TermId,
1793        dst: EntityId,
1794        fact: FactId,
1795        edge: EdgeId,
1796        valid_from: u64,
1797    ) -> Result<(), Error> {
1798        for (arena, a, b) in [
1799            (&mut self.edges_out, src, dst),
1800            (&mut self.edges_in, dst, src),
1801        ] {
1802            let slot = EdgeSlot {
1803                a,
1804                rel,
1805                b,
1806                fact,
1807                edge,
1808                valid_from,
1809            };
1810            if !arena.insert(&slot)? {
1811                let payload = arena
1812                    .payload_mut(&edge_key(a, rel, b))
1813                    .expect("insert reported a duplicate");
1814                let mut full = [0u8; EdgeSlot::SIZE];
1815                slot.write(&mut full);
1816                payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
1817            }
1818        }
1819        Ok(())
1820    }
1821
1822    fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
1823        self.edges_hist_out.insert(&edge)?;
1824        self.edges_hist_in.insert(&EdgeHistorySlot {
1825            a: edge.b,
1826            b: edge.a,
1827            ..edge
1828        })?;
1829        Ok(())
1830    }
1831
1832    /// Closes the open version of `(src, rel, dst)` and drops it from the
1833    /// current graph. Returns `false` when there is no such edge.
1834    ///
1835    /// The current slot names its own history record — `edge` and `valid_from`
1836    /// are that record's key tail — so both mirrors are addressed directly.
1837    fn close_current_edge(
1838        &mut self,
1839        now: u64,
1840        src: EntityId,
1841        rel: TermId,
1842        dst: EntityId,
1843    ) -> Result<bool, Error> {
1844        let Some(current) = self.current_edge(src, rel, dst) else {
1845            return Ok(false);
1846        };
1847        // A version never ends before it began, even if the caller's clock
1848        // moved backwards between the link and the unlink.
1849        let close_at = now.max(current.valid_from);
1850        let out_key = edge_history_key(src, current.valid_from, current.edge);
1851        let in_key = edge_history_key(dst, current.valid_from, current.edge);
1852        close_edge_history_payload(
1853            self.edges_hist_out
1854                .payload_mut(&out_key)
1855                .ok_or(Error::Corrupt("missing outgoing edge history"))?,
1856            close_at,
1857        );
1858        close_edge_history_payload(
1859            self.edges_hist_in
1860                .payload_mut(&in_key)
1861                .ok_or(Error::Corrupt("missing incoming edge history"))?,
1862            close_at,
1863        );
1864        let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
1865        let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
1866        if out_removed != in_removed {
1867            return Err(Error::Corrupt("edge mirrors disagree"));
1868        }
1869        Ok(out_removed)
1870    }
1871
1872    fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
1873        self.edges_out
1874            .get_slot(&edge_key(src, rel, dst))
1875            .map(EdgeSlot::read)
1876    }
1877
1878    /// Looks an entity up by its already-normalized name (read-only:
1879    /// neither the vocabulary nor the arenas change on a miss).
1880    fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
1881        let term = self.terms.lookup(norm)?;
1882        let mut from = [0u8; 8];
1883        key::write_u32(&mut from, term.0);
1884        let mut to = [0u8; 8];
1885        key::write_u32(&mut to, term.0);
1886        to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
1887        self.by_name.range(&from, &to).next().map(|e| e.id)
1888    }
1889
1890    fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
1891        let mut norm = core::mem::take(&mut self.name_scratch);
1892        normalize_name(&mut self.tokenizer, name, &mut norm);
1893        let result = (!norm.is_empty())
1894            .then(|| self.lookup_entity_by_norm(&norm))
1895            .flatten();
1896        self.name_scratch = norm;
1897        result
1898    }
1899
1900    fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
1901        let mut norm = core::mem::take(&mut self.name_scratch);
1902        normalize_name(&mut self.tokenizer, name, &mut norm);
1903        if norm.is_empty() {
1904            self.name_scratch = norm;
1905            return Err(Error::Invalid("entity name has no indexable characters"));
1906        }
1907        let result = (|| {
1908            if let Some(found) = self.lookup_entity_by_norm(&norm) {
1909                return Ok(found);
1910            }
1911            let term = self.terms.intern(&norm)?;
1912            let id = EntityId(self.next_entity);
1913            let name_id = self.texts.push(name.as_bytes())?;
1914            self.entities.insert(&EntityRecord {
1915                id,
1916                name: name_id,
1917                name_term: term,
1918                created_at: now,
1919                flags: 0,
1920            })?;
1921            self.by_name.insert(&EntityByName {
1922                name_term: term,
1923                id,
1924            })?;
1925            self.next_entity += 1;
1926            Ok(id)
1927        })();
1928        self.name_scratch = norm;
1929        result
1930    }
1931
1932    fn journal_remember<S: Storage>(
1933        &mut self,
1934        store: &mut S,
1935        input: &RememberInput<'_>,
1936        revises: FactId,
1937        assigned: FactId,
1938    ) -> Result<(), Error> {
1939        let mut entry = Vec::new();
1940        Op::Remember {
1941            now: input.now,
1942            valid_from: input.valid_from.unwrap_or(input.now),
1943            entity: input.entity,
1944            text: input.text,
1945            tags: input.tags.to_vec(),
1946            links: input.links.to_vec(),
1947            vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
1948            metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
1949            revises,
1950            assigned,
1951        }
1952        .encode(&mut entry);
1953        store
1954            .append_journal(&entry)
1955            .map_err(|e| Error::Storage(format!("{e:?}")))
1956    }
1957}
1958
1959impl core::fmt::Debug for Memory<'_> {
1960    /// Summary only — the contents are the user's memory, not ours to
1961    /// print.
1962    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1963        f.debug_struct("Memory")
1964            .field("facts", &self.facts.len())
1965            .field("entities", &self.entities.len())
1966            .field("terms", &self.terms.len())
1967            .finish()
1968    }
1969}
1970
1971/// Normalizes an entity name: its tokens joined by single spaces
1972/// ("  Проект   PlugMem " → "проект plugmem"). Deterministic and aligned
1973/// with search-time tokenization.
1974fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
1975    out.clear();
1976    tokenizer.tokenize(name, &mut |token| {
1977        if !out.is_empty() {
1978            out.push(' ');
1979        }
1980        out.push_str(token);
1981    });
1982}
1983
1984/// Probe index for `term` in a table of `mask + 1` slots. Fibonacci hashing —
1985/// the same constant the arena shards with, so term ids scatter without a
1986/// second hash family.
1987#[inline]
1988fn probe_at(term: u32, mask: usize) -> usize {
1989    (u64::from(term).wrapping_mul(0x9E37_79B9_7F4A_7C15) >> 32) as usize & mask
1990}
1991
1992/// Records one occurrence of `term` in `tfs`, using `probe` as a
1993/// `term -> index + 1` open-addressing table (0 is empty).
1994///
1995/// This replaces a scan of the terms already seen, which made indexing a
1996/// document quadratic in its length — at the 4 KiB `max_text` ceiling, on the
1997/// order of 10^5 comparisons for one `remember`.
1998///
1999/// **A term keeps the slot its first occurrence claimed**, so `tfs` comes out
2000/// in first-occurrence order exactly as the scan left it and `index_doc`
2001/// receives byte-identical input.
2002fn tally_term(tfs: &mut Vec<(u32, u8)>, probe: &mut Vec<u32>, term: u32) {
2003    // Half full at most: probes stay short and the loop below terminates.
2004    // Growth rehashes from `tfs`, which is the authoritative order.
2005    if (tfs.len() + 1) * 2 > probe.len() {
2006        let grown = (probe.len() * 2).max(64);
2007        probe.clear();
2008        probe.resize(grown, 0);
2009        let mask = grown - 1;
2010        for (i, &(seen, _)) in tfs.iter().enumerate() {
2011            let mut at = probe_at(seen, mask);
2012            while probe[at] != 0 {
2013                at = (at + 1) & mask;
2014            }
2015            probe[at] = i as u32 + 1;
2016        }
2017    }
2018    let mask = probe.len() - 1;
2019    let mut at = probe_at(term, mask);
2020    loop {
2021        match probe[at] {
2022            0 => {
2023                probe[at] = tfs.len() as u32 + 1;
2024                tfs.push((term, 1));
2025                return;
2026            }
2027            slot => {
2028                let entry = &mut tfs[slot as usize - 1];
2029                if entry.0 == term {
2030                    entry.1 = entry.1.saturating_add(1);
2031                    return;
2032                }
2033            }
2034        }
2035        at = (at + 1) & mask;
2036    }
2037}