plugmem_core/

memory.rs

//! The engine: data model state, verbs, journaling and replay
//! (/03, /05).
//!
//! All state lives in the flat structures of `plugmem-arena`; every
//! mutating verb journals itself through the caller's
//! [`Storage`] and is replayed through the same
//! internal `apply_*` path on open — the journal's ids are authoritative,
//! so replay is deterministic and idempotent over a reapplied tail.
//!
//! The engine carries the structural sources (facts, entities, edges,
//! tags, BM25, temporal) and — when `Config::dim > 0` — the vector layer:
//! quantized vectors in a flat [`VecPool`], searched as a fourth recall
//! source and used by similar-detection. `maintain` (compaction) still
//! lands separately.

use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec::Vec;

use plugmem_arena::{
    Arena, ArenaCfg, BlobHeap, BlobHeapCfg, BlobId, ChunkPool, ChunkPoolCfg, Interner, ListHandle,
    ShardMode, Slot, TermId, key,
};

use crate::config::Config;
use crate::error::Error;
use crate::id::{EdgeId, EntityId, FactId, NONE_U32};
use crate::index::IdListIndex;
use crate::index::bm25::Bm25Index;
use crate::index::hnsw::HnswGraph;
use crate::index::vecpool::VecPool;
use crate::journal::{JournalScan, Op, scan};
use crate::model::{
    EdgeHistorySlot, EdgeSlot, EntityByName, EntityRecord, FactAux, FactRecord, TemporalSlot,
    VALID_TO_OPEN, close_edge_history_payload, edge_history_key, edge_key, fact_flags,
};
use crate::storage::Storage;
use crate::tokenizer::Tokenizer;

use maintain::TOKENIZER_INDEX_VERSION;

/// Most recent same-entity facts examined by similar-detection.
const SIMILAR_CANDIDATE_CAP: usize = 32;

/// Tags one fact may carry.
///
/// Structural, not a policy knob: `apply_remember` deduplicates into a stack
/// array of exactly this length, and `change_catalog_for_fact` reads the tags
/// back into another one to keep the catalog counts balanced. Every path that
/// reaches `apply_remember` — the verbs through `validate_input`, replay
/// through its own corruption check — has to have established the bound first.
const MAX_TAGS: usize = 32;

mod maintain;
mod migrations;
mod persist;
mod recall;
mod reembed;
mod shards;
mod tags;

pub use maintain::{MaintainReport, MaintenanceMode, MaintenanceOptions};
pub use recall::{RecallQuery, RecallResult, RecallScratch, RecalledEdge, RecalledFact, source};
pub use reembed::{ReembedError, ReembedReport};
pub use shards::ShardLayout;
pub use tags::{DEFAULT_TAG_PAGE_LIMIT, MAX_TAG_PAGE_LIMIT, TagPage, TagQuery, TagSummary};

/// Maximum UTF-8 byte length of a persisted embedding-space identity.
pub const MAX_VECTOR_SPACE_ID_BYTES: usize = 256;

/// Input of `remember` and `revise`.
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct RememberInput<'a> {
    /// Host timestamp, unix milliseconds.
    pub now: u64,
    /// Fact text, ≤ `Config::max_text` bytes.
    pub text: &'a str,
    /// Subject entity name; created lazily on first mention.
    pub entity: Option<&'a str>,
    /// Tags, verbatim strings, ≤ 32.
    pub tags: &'a [&'a str],
    /// `(rel, target_entity)` pairs, ≤ 16; edges go subject → target.
    pub links: &'a [(&'a str, &'a str)],
    /// Optional embedding, `len == Config::dim`; quantized on the way in.
    /// Requires `dim > 0` and is dropped by the engine's own re-quantized
    /// replay, so nothing float-nondeterministic reaches the state.
    pub vector: Option<&'a [f32]>,
    /// Validity start; defaults to `now`.
    pub valid_from: Option<u64>,
    /// Optional metadata as key→value pairs (UTF-8). Passed in any order; the
    /// engine canonicalizes (sorts keys, rejects duplicates) and stores them as
    /// one opaque blob (see `crate::metadata`). `None` or an empty slice = no
    /// metadata. The engine never interprets the values.
    pub metadata: Option<&'a [(&'a str, &'a str)]>,
}

impl<'a> RememberInput<'a> {
    /// A minimal input: text only.
    pub fn text(now: u64, text: &'a str) -> Self {
        Self {
            now,
            text,
            entity: None,
            tags: &[],
            links: &[],
            vector: None,
            valid_from: None,
            metadata: None,
        }
    }
}

/// Input of `link`.
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct LinkInput<'a> {
    /// Host timestamp, unix milliseconds.
    pub now: u64,
    /// Source entity name (created lazily).
    pub src: &'a str,
    /// Relation term, verbatim (`"works_at"`, …).
    pub rel: &'a str,
    /// Destination entity name (created lazily).
    pub dst: &'a str,
    /// Optional provenance fact.
    pub provenance: Option<FactId>,
}

/// Input of `unlink`.
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct UnlinkInput<'a> {
    /// Host timestamp, unix milliseconds.
    pub now: u64,
    /// Source entity name.
    pub src: &'a str,
    /// Relation term, verbatim.
    pub rel: &'a str,
    /// Destination entity name.
    pub dst: &'a str,
}

/// Result of `remember`/`revise`.
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct RememberOutcome {
    /// The new fact's id.
    pub id: FactId,
    /// The subject entity (resolved or created), if one was named.
    pub entity: Option<EntityId>,
    /// Similar / potentially conflicting **live** facts, best match
    /// first (≤ 8). The engine never revises on its own — it surfaces
    /// the candidates, the agent judges: `revise`, keep both, or
    /// `forget`.
    pub similar: Vec<Similar>,
}

/// Result of [`Memory::remember_guarded`].
///
/// The two variants are intentionally disjoint: a blocked call has no fact id
/// because it did not allocate one, mutate an index, or append a journal
/// record. A caller that decides the candidates are compatible can make that
/// choice explicit by following with ordinary [`Memory::remember`].
///
/// `Stored` additionally reports whether a check happened at all - see
/// [`GuardedRememberOutcome::Stored::checked`]. Without it the two answers
/// "nothing similar was found" and "nothing could be compared" are the same
/// value, and a caller reaching for this verb specifically to avoid duplicates
/// has no way to learn it did not get one.
#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "serde", serde(tag = "status", rename_all = "snake_case"))]
pub enum GuardedRememberOutcome {
    /// No candidate crossed the configured similarity thresholds; the fact was
    /// stored normally.
    Stored {
        /// The committed fact.
        outcome: RememberOutcome,
        /// Whether the similarity detector actually had candidates to compare
        /// against.
        ///
        /// `false` means the write was stored **unguarded**: the detector is
        /// scoped to the fact's entity and walks that entity's most recent
        /// live facts, so an input carrying no `entity` has no candidate set
        /// and cannot block anything, now or after any number of later
        /// writes. The fact is stored, exactly as an ordinary
        /// [`Memory::remember`] would have stored it.
        ///
        /// `true` says a comparison took place, including the case where the
        /// entity is new and its candidate set was empty - that is a real
        /// "nothing similar is stored", and the same call will be compared
        /// properly from the entity's second fact onwards.
        ///
        /// This is not a failure and not a fallback: comparing against every
        /// live fact instead is precisely the unbounded work the per-entity
        /// candidate cap exists to avoid. It is reported because the
        /// alternative is silence, and silence here reads as "checked, and it
        /// was clear".
        checked: bool,
    },
    /// At least one live same-entity fact crossed a similarity threshold; no
    /// mutation was made.
    Blocked {
        /// Similar / potentially conflicting live facts, best match first
        /// (at most eight).
        similar: Vec<Similar>,
    },
}

/// Result of removing one tag from every current fact.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct RemoveTagReport {
    /// Current facts superseded by an otherwise-identical revision without
    /// the tag.
    pub affected: u32,
}

/// One similar-fact hint.
#[derive(Clone, Copy, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct Similar {
    /// The existing fact.
    pub id: FactId,
    /// Match strength (for the lexical detector: the Jaccard overlap of
    /// the term sets, in `(similar_jaccard, 1]`).
    pub score: f32,
    /// What triggered the hint.
    pub reason: SimilarReason,
}

/// Why a fact was flagged as similar.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum SimilarReason {
    /// Same subject entity and a term-set overlap above the configured
    /// Jaccard threshold.
    LexicalOverlap,
    /// Same subject entity and a quantized-vector cosine above the
    /// configured `similar_cos` threshold.
    VectorCosine,
}

/// Reusable write-side buffers for similarity detection. Kept outside the
/// persistent pools: after warm-up guarded writes do not allocate scratch on
/// every call, and none of these bytes enters a snapshot.
#[derive(Debug, Default)]
struct SimilarityScratch {
    /// Incoming terms already present in the interner.
    new_terms: Vec<u32>,
    /// Unknown incoming tokens concatenated for exact deduplication without
    /// growing the interner during a read-only guard check.
    unknown: String,
    /// Byte ranges in `unknown`, one per distinct unknown token.
    unknown_ranges: Vec<(usize, usize)>,
    /// Candidate term ids, reused across the bounded scan.
    candidate_terms: Vec<u32>,
    /// Quantized incoming vector slot, never appended to the vector pool.
    vector: Vec<u8>,
}

#[derive(Clone, Copy)]
enum SimilarVector<'s> {
    None,
    Stored(u32),
    Encoded(&'s [u8]),
}

#[derive(Clone, Copy)]
struct SimilarityQuery<'s> {
    entity: EntityId,
    exclude: Option<FactId>,
    terms: &'s [u32],
    term_count: usize,
    vector: SimilarVector<'s>,
}

/// A read view of one fact.
#[derive(Clone, Copy, Debug)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct FactView<'a> {
    /// The raw record (temporality, flags, references).
    pub record: FactRecord,
    /// The fact text.
    pub text: &'a str,
}

/// The per-fact content problem [`Memory::faulty_facts`] attributes to a
/// fact — the salvage predicate `recover` uses. It mirrors the
/// content checks [`Memory::verify`] runs, split out per fact so a caller can
/// drop the individual bad records instead of failing the whole image.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum FactFault {
    /// The fact's stored text is not valid UTF-8.
    Text,
    /// The fact is flagged with a vector but its slot is out of range or does
    /// not name the fact back (the fact↔slot bijection is broken).
    Vector,
    /// The fact's metadata blob is out of range or does not decode to a
    /// well-formed key→value map.
    Metadata,
}

/// Engine size counters. Every field is an O(1) read; the
/// struct is `#[non_exhaustive]` so later stages (database identity
/// markers, HNSW state) can extend it without a breaking change.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
#[non_exhaustive]
pub struct Stats {
    /// Fact records currently stored (live, closed, and tombstoned facts
    /// awaiting `maintain`).
    pub facts: usize,
    /// Entities.
    pub entities: usize,
    /// Interned terms (tokens, tags, relations, normalized names).
    pub terms: usize,
    /// Directed edges (each `(src, rel, dst)` counted once; the mirrored
    /// in-arena is an internal detail).
    pub edges: usize,
    /// Historical edge versions, including closed versions.
    pub edge_versions: usize,
    /// Quantized vector slots.
    pub vectors: usize,
    /// Tombstoned fact records awaiting physical purge.
    pub tombstones: usize,
    /// Vector slots already covered by the HNSW graph; slots after this are
    /// searched as the flat tail.
    pub hnsw_indexed: u32,
    /// The next fact id to be assigned. Ids below it are in use or burned
    /// (forgotten and purged) — never reissued.
    pub next_fact: u32,
    /// The next entity id to be assigned.
    pub next_entity: u32,
    /// The next edge-version id to be assigned.
    pub next_edge: u32,
    /// The database lineage identity ([`Config::db_uuid`]); `0` for an
    /// unnamed database.
    pub db_uuid: u128,
    /// Total bytes held by the engine's pools (arenas, blob heaps, chunk
    /// pools, the term dictionary and the vector pool).
    pub pool_bytes: usize,
    /// How the arenas are currently sharded.
    ///
    /// The engine chooses this from what it holds and moves it during
    /// `maintain`, so it is state to observe rather than a setting to pick —
    /// see [`Config::shards_facts`](crate::Config::shards_facts).
    pub shards: ShardLayout,
}

/// Report of an `open`: what the journal replay found.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct OpenReport {
    /// Journal records applied.
    pub replayed: usize,
    /// Journal records skipped as already contained in the snapshot.
    pub skipped: usize,
    /// A torn tail record was dropped (crash between appends).
    pub truncated_tail: bool,
}

/// The memory engine. See the module docs for staging.
///
/// The lifetime `'a` is the provenance of the engine's byte pools. Every
/// owned constructor (`new`/`open`/`from_bytes`) yields a `Memory<'static>`
/// that owns its bytes; the read-only [`Memory::from_bytes_borrowed`] path
/// borrows them from an mmap'd snapshot instead of copying.
pub struct Memory<'a> {
    cfg: Config,
    // -- data model --
    facts: Arena<'a, FactRecord>,
    fact_aux: Arena<'a, FactAux>,
    entities: Arena<'a, EntityRecord>,
    by_name: Arena<'a, EntityByName>,
    edges_out: Arena<'a, EdgeSlot>,
    edges_in: Arena<'a, EdgeSlot>,
    edges_hist_out: Arena<'a, EdgeHistorySlot>,
    edges_hist_in: Arena<'a, EdgeHistorySlot>,
    temporal: Arena<'a, TemporalSlot>,
    /// Fact texts and canonical entity names.
    texts: BlobHeap<'a>,
    /// Per-fact metadata blobs (canonical key→value, see `crate::metadata`),
    /// referenced by [`FactAux::meta`]. Empty and inert until a fact carries
    /// metadata; kept out of `texts` so this cold pool stays non-resident on an
    /// mmap'd base until `show`/`export` touches it.
    metas: BlobHeap<'a>,
    /// Terms: tokens, tags (verbatim), relation names, normalized entity
    /// names.
    terms: Interner<'a>,
    /// Per-fact tag lists (`TermId` values), handles in [`FactAux`].
    tag_lists: ChunkPool<'a>,
    // -- indexes --
    bm25: Bm25Index<'a>,
    tags_idx: IdListIndex<'a>,
    /// Current tag counts: merged snapshot base plus changes since it.
    tag_catalog: tags::TagCatalog,
    entity_facts: IdListIndex<'a>,
    /// Quantized vectors (empty and inert when `cfg.dim == 0`).
    vecs: VecPool<'a>,
    /// HNSW graph over the pool; empty until `maintain` crosses
    /// `Config::flat_to_hnsw`. Slots past its `indexed` mark are the flat
    /// tail searched by scan.
    hnsw: HnswGraph<'a>,
    /// Human-readable identity of the model/vector space that produced every
    /// slot in `vecs`. `None` is legacy/untracked state, never permission to
    /// mix a configured model into a non-empty pool.
    vector_space: Option<String>,
    // -- id allocation (derived from the arenas on load) --
    next_fact: u32,
    next_entity: u32,
    next_edge: u32,
    // -- maintenance state --
    tombstones: usize,
    bm25_tokenizer_version: u32,
    // -- reusable scratches --
    tokenizer: Tokenizer,
    tf_scratch: Vec<(u32, u8)>,
    /// Open-addressing `term -> index into `tf_scratch` + 1` table (0 is
    /// empty), so accumulating a document's term frequencies costs one probe
    /// per token instead of a scan of the terms already seen. Grown to the
    /// largest document indexed and reused; see [`tally_term`].
    tf_probe: Vec<u32>,
    name_scratch: String,
    similarity_scratch: SimilarityScratch,
}

impl<'a> Memory<'a> {
    /// Creates an empty database.
    ///
    /// # Errors
    ///
    /// [`Error::ConfigMismatch`] for an invalid config (see
    /// [`Config::validate`]).
    pub fn new(cfg: Config) -> Result<Self, Error> {
        cfg.validate()?;
        let uni =
            |shards: usize| ArenaCfg::new(shards, ShardMode::Uniform).with_max_bytes(cfg.max_bytes);
        let ord =
            |shards: usize| ArenaCfg::new(shards, ShardMode::Ordered).with_max_bytes(cfg.max_bytes);
        let blob = BlobHeapCfg::new()
            .with_max_bytes(cfg.max_bytes)
            .with_max_blob(cfg.max_blob);
        Ok(Self {
            facts: Arena::new(uni(cfg.shards_facts))?,
            fact_aux: Arena::new(uni(cfg.shards_facts))?,
            entities: Arena::new(uni(cfg.shards_entities))?,
            by_name: Arena::new(ord(cfg.shards_entities))?,
            edges_out: Arena::new(ord(cfg.shards_edges))?,
            edges_in: Arena::new(ord(cfg.shards_edges))?,
            edges_hist_out: Arena::new(ord(cfg.shards_edges))?,
            edges_hist_in: Arena::new(ord(cfg.shards_edges))?,
            temporal: Arena::new(ord(cfg.shards_temporal))?,
            texts: BlobHeap::new(blob),
            metas: BlobHeap::new(blob),
            terms: Interner::new(blob),
            tag_lists: ChunkPool::new(ChunkPoolCfg::new().with_max_bytes(cfg.max_bytes)),
            bm25: Bm25Index::new(cfg.shards_postings, cfg.max_bytes)?,
            tags_idx: IdListIndex::new(cfg.shards_postings, cfg.max_bytes)?,
            tag_catalog: tags::TagCatalog::new(),
            entity_facts: IdListIndex::new(cfg.shards_entities, cfg.max_bytes)?,
            vecs: VecPool::new(cfg.dim, cfg.max_bytes),
            hnsw: HnswGraph::new(cfg.hnsw_m, cfg.hnsw_m0, cfg.max_bytes)?,
            vector_space: None,
            next_fact: 0,
            next_entity: 0,
            next_edge: 0,
            tombstones: 0,
            bm25_tokenizer_version: maintain::TOKENIZER_INDEX_VERSION,
            tokenizer: Tokenizer::new(),
            tf_scratch: Vec::new(),
            tf_probe: Vec::new(),
            name_scratch: String::new(),
            similarity_scratch: SimilarityScratch::default(),
            cfg,
        })
    }

    /// Opens a database from a storage: journal replay over an empty
    /// state (the snapshot fast-path lands with the section composition;
    /// a non-empty snapshot is rejected rather than half-read).
    pub fn open<S: Storage>(store: &mut S, cfg: Config) -> Result<(Self, OpenReport), Error> {
        let snapshot = store
            .read_snapshot()
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        let journal = store
            .read_journal()
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        Self::from_bytes(snapshot.as_deref(), &journal, cfg)
    }

    /// Opens from raw bytes (the wasm path: the host already read them).
    pub fn from_bytes(
        snapshot: Option<&[u8]>,
        journal: &[u8],
        cfg: Config,
    ) -> Result<(Self, OpenReport), Error> {
        let mut mem = match snapshot {
            Some(bytes) => Self::load_snapshot(bytes, cfg)?,
            None => Self::new(cfg)?,
        };
        let report = mem.replay(journal)?;
        Ok((mem, report))
    }

    /// Opens a database read-only over a borrowed snapshot: the
    /// engine's byte pools borrow `snapshot` (typically an mmap'd file)
    /// instead of copying it, so a large database residents only the pages
    /// actually read. The returned engine is tied to `snapshot`'s lifetime.
    ///
    /// A **non-empty journal is rejected**: the read-only handle exposes no
    /// write verbs, so a snapshot with un-checkpointed journal tail would open
    /// stale. Checkpoint the database read-write once first — or use
    /// [`Memory::from_bytes_overlay`], which replays the journal into the
    /// overlay. The caller (the host) owns the map and the exclusive lock.
    pub fn from_bytes_borrowed(
        snapshot: &'a [u8],
        journal: &[u8],
        cfg: Config,
    ) -> Result<Self, Error> {
        let mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
        let JournalScan { entries, .. } = scan(journal)?;
        if !entries.is_empty() {
            return Err(Error::Invalid(
                "read-only open requires a checkpointed (empty) journal",
            ));
        }
        Ok(mem)
    }

    /// Opens a database read-**write** over a borrowed snapshot (the overlay
    /// write path): like [`Memory::from_bytes_borrowed`] the byte
    /// pools borrow `snapshot` instead of copying it, but the journal is
    /// **replayed** and the engine stays fully mutable. Mutations do not clone
    /// the borrowed base: the flat structures land appends in an owned tail and
    /// copy only the pages they rewrite (per-page copy-on-write), so a
    /// multi-gigabyte mapped database is opened and written to while resident
    /// only in the pages it actually touches.
    ///
    /// This is the write sibling of [`Memory::from_bytes`] (which owns its
    /// bytes) — the same `snapshot + journal replay`, over a borrowed base, and
    /// it returns the same [`OpenReport`] describing the replay. The returned
    /// engine is tied to `snapshot`'s lifetime; the caller (the host) owns the
    /// map and the exclusive lock.
    pub fn from_bytes_overlay(
        snapshot: &'a [u8],
        journal: &[u8],
        cfg: Config,
    ) -> Result<(Self, OpenReport), Error> {
        let mut mem = Self::load_snapshot_borrowed(snapshot, cfg)?;
        let report = mem.replay(journal)?;
        Ok((mem, report))
    }

    /// Applies every journal record on top of the current state (
    /// replay rules: assigned ids are authoritative; records whose
    /// assigned id is already present are skipped, which makes reapplying
    /// a tail idempotent).
    fn replay(&mut self, journal: &[u8]) -> Result<OpenReport, Error> {
        let JournalScan {
            entries,
            truncated_tail,
        } = scan(journal)?;
        let mut report = OpenReport {
            truncated_tail,
            ..OpenReport::default()
        };
        for entry in entries {
            let op = Op::decode(entry.op, entry.payload)?;
            match op {
                Op::Remember {
                    now,
                    valid_from,
                    entity,
                    text,
                    ref tags,
                    ref links,
                    ref vector,
                    ref metadata,
                    revises,
                    assigned,
                } => {
                    if assigned.0 < self.next_fact {
                        report.skipped += 1;
                        continue;
                    }
                    if assigned.0 != self.next_fact {
                        return Err(Error::Corrupt("journal fact ids are not contiguous"));
                    }
                    if !vector.is_empty() && vector.len() != self.cfg.dim {
                        return Err(Error::Corrupt(
                            "journal vector dimension disagrees with dim",
                        ));
                    }
                    // The record count is a `u8` on the wire, so a damaged
                    // journal can name up to 255 tags where a fact holds
                    // [`MAX_TAGS`]. Replay does not run `validate_input` — it
                    // reapplies what the verbs already accepted — so the one
                    // bound `apply_remember` indexes an array with has to be
                    // re-established here, beside the other corruption checks.
                    if tags.len() > MAX_TAGS {
                        return Err(Error::Corrupt("journal record carries too many tags"));
                    }
                    if let Some(target) = revises.some() {
                        self.check_revisable(target)
                            .map_err(|_| Error::Corrupt("journal revises an unrevisable fact"))?;
                    }
                    self.apply_remember(
                        &RememberInput {
                            now,
                            text,
                            entity,
                            tags: &tags.to_vec(),
                            links: &links.to_vec(),
                            vector: (!vector.is_empty()).then_some(vector.as_slice()),
                            valid_from: Some(valid_from),
                            metadata: (!metadata.is_empty()).then_some(metadata.as_slice()),
                        },
                        revises,
                        None,
                    )?;
                    if let Some(target) = revises.some() {
                        self.close_target(target, valid_from);
                    }
                    report.replayed += 1;
                }
                Op::Forget { fact, .. } => {
                    // Idempotent; a missing fact in a valid journal is
                    // corruption, but re-tombstoning is fine.
                    match self.apply_forget(fact) {
                        Ok(_) => report.replayed += 1,
                        Err(Error::NotFound(_)) => {
                            return Err(Error::Corrupt("journal forgets an unknown fact"));
                        }
                        Err(e) => return Err(e),
                    }
                }
                Op::Link {
                    now,
                    src,
                    rel,
                    dst,
                    provenance,
                } => {
                    self.apply_link(now, src, rel, dst, provenance)?;
                    report.replayed += 1;
                }
                Op::Unlink { now, src, rel, dst } => {
                    self.apply_unlink(now, src, rel, dst)?;
                    report.replayed += 1;
                }
                Op::RemoveTag { now, tag } => {
                    self.apply_remove_tag(now, tag)?;
                    report.replayed += 1;
                }
                Op::SetVectorSpace { space } => {
                    self.apply_set_vector_space(space)?;
                    report.replayed += 1;
                }
                Op::Maintain {
                    mode,
                    max_hnsw_inserts,
                    ..
                } => {
                    // Re-execute the compaction deterministically, so a
                    // replayed image matches one snapshotted after a live
                    // maintain byte for byte.
                    let options =
                        maintain::MaintenanceOptions::from_journal(mode, max_hnsw_inserts)?;
                    self.replay_maintain_with_options(options)?;
                    report.replayed += 1;
                }
            }
        }
        Ok(report)
    }

    /// The semantic identity persisted with the current vector pool.
    ///
    /// `None` means either that no automatic vector has been written yet, or
    /// that this is a legacy image whose old model was not recorded.
    pub fn vector_space(&self) -> Option<&str> {
        self.vector_space.as_deref()
    }

    /// Assigns `space` to an empty vector pool and journals the assignment.
    /// Repeating the same claim is a no-op; claiming a different or untracked
    /// non-empty pool is refused. Explicit reembed is the only operation that
    /// may replace a non-empty pool's identity.
    pub fn claim_vector_space<S: Storage>(
        &mut self,
        store: &mut S,
        space: &str,
    ) -> Result<bool, Error> {
        let needs_claim = self.check_vector_space_claim(space)?;
        if !needs_claim {
            return Ok(false);
        }
        let mut entry = Vec::new();
        Op::SetVectorSpace { space }.encode(&mut entry);
        store
            .append_journal(&entry)
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        self.vector_space = Some(space.into());
        Ok(true)
    }

    /// Validates a vector-space claim without changing state. `true` means an
    /// empty pool needs its first persisted identity.
    fn check_vector_space_claim(&self, space: &str) -> Result<bool, Error> {
        Self::validate_vector_space(space)?;
        if let Some(stored) = &self.vector_space {
            if stored == space {
                return Ok(false);
            }
            if !self.vecs.is_empty() {
                return Err(Error::VectorSpaceMismatch {
                    stored: stored.clone(),
                    requested: space.into(),
                });
            }
        }
        if !self.vecs.is_empty() {
            return Err(Error::UntrackedVectorSpace);
        }
        Ok(true)
    }

    fn apply_set_vector_space(&mut self, space: &str) -> Result<(), Error> {
        Self::validate_vector_space(space)?;
        if let Some(stored) = &self.vector_space {
            if stored == space {
                return Ok(());
            }
            if !self.vecs.is_empty() {
                return Err(Error::Corrupt(
                    "journal changes an established vector space",
                ));
            }
        }
        if !self.vecs.is_empty() {
            return Err(Error::Corrupt(
                "journal assigns a vector space after vector records",
            ));
        }
        self.vector_space = Some(space.into());
        Ok(())
    }

    pub(super) fn validate_vector_space(space: &str) -> Result<(), Error> {
        if space.is_empty() {
            return Err(Error::Invalid("vector space must not be empty"));
        }
        if space.len() > MAX_VECTOR_SPACE_ID_BYTES {
            return Err(Error::TooLarge {
                what: "vector space",
                len: space.len(),
                max: MAX_VECTOR_SPACE_ID_BYTES,
            });
        }
        if space.bytes().any(|b| b < b' ' || b == 0x7f) {
            return Err(Error::Invalid(
                "vector space must not contain control bytes",
            ));
        }
        Ok(())
    }

    /// Remembers a new fact.
    pub fn remember<S: Storage>(
        &mut self,
        store: &mut S,
        input: RememberInput<'_>,
    ) -> Result<RememberOutcome, Error> {
        self.validate_input(&input)?;
        let mut outcome = self.apply_remember(&input, FactId::NONE, None)?;
        self.find_similar(&mut outcome);
        self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
        Ok(outcome)
    }

    /// Stores a fact only when the ordinary `remember` similarity detector
    /// finds no live same-entity candidate above its configured lexical or
    /// vector threshold.
    ///
    /// The check and possible mutation are one engine operation. A blocked
    /// call does not allocate a fact id, touch an index, or append to the
    /// journal. The detector is exactly the one that fills
    /// [`RememberOutcome::similar`] after an unconditional [`remember`](Self::remember);
    /// hybrid [`recall`](Self::recall) ranking is deliberately not involved.
    pub fn remember_guarded<S: Storage>(
        &mut self,
        store: &mut S,
        input: RememberInput<'_>,
    ) -> Result<GuardedRememberOutcome, Error> {
        self.remember_guarded_with_vector_space(store, input, None)
    }

    /// Host integration for [`remember_guarded`](Self::remember_guarded):
    /// claims an automatic embedder's semantic vector space only if the fact
    /// will actually be stored. A blocked call therefore remains entirely
    /// non-mutating, including vector-space metadata and its journal record.
    #[doc(hidden)]
    pub fn remember_guarded_with_vector_space<S: Storage>(
        &mut self,
        store: &mut S,
        input: RememberInput<'_>,
        vector_space: Option<&str>,
    ) -> Result<GuardedRememberOutcome, Error> {
        self.validate_input(&input)?;
        if let Some(space) = vector_space {
            self.check_vector_space_claim(space)?;
        }
        // Whether a check was POSSIBLE, which is a property of the call rather
        // than of what the memory currently holds: the detector is scoped to
        // the input's entity, so an input naming none has no candidate set now
        // and would have none however many facts were stored later. An input
        // that does name one is checked - against an empty set when it is that
        // entity's first fact, which is a real "nothing similar" rather than an
        // absent comparison.
        let checked = input.entity.is_some();
        let similar = self.find_similar_input(&input)?;
        if !similar.is_empty() {
            return Ok(GuardedRememberOutcome::Blocked { similar });
        }
        if let Some(space) = vector_space {
            self.claim_vector_space(store, space)?;
        }
        let outcome = self.apply_remember(&input, FactId::NONE, None)?;
        self.journal_remember(store, &input, FactId::NONE, outcome.id)?;
        Ok(GuardedRememberOutcome::Stored { outcome, checked })
    }

    /// Batch import: a sequence of remembers in one call, journaled
    /// individually. `skip_similar` turns off the
    /// similar-detection pass — imports don't need hints, and skipping
    /// them makes a bulk load cheaper.
    ///
    /// Stops at the first error; already-applied inputs stay applied and
    /// journaled (replay reproduces exactly the applied prefix).
    pub fn remember_batch<S: Storage>(
        &mut self,
        store: &mut S,
        inputs: &[RememberInput<'_>],
        skip_similar: bool,
    ) -> Result<Vec<RememberOutcome>, Error> {
        let mut out = Vec::with_capacity(inputs.len());
        for input in inputs {
            self.validate_input(input)?;
            let mut outcome = self.apply_remember(input, FactId::NONE, None)?;
            if !skip_similar {
                self.find_similar(&mut outcome);
            }
            self.journal_remember(store, input, FactId::NONE, outcome.id)?;
            out.push(outcome);
        }
        Ok(out)
    }

    /// Runs similarity detection against an incoming, not-yet-stored fact.
    /// Unknown tokens are counted in the Jaccard union but never interned:
    /// a blocked call is observationally a read and cannot grow vocabulary.
    fn find_similar_input(&mut self, input: &RememberInput<'_>) -> Result<Vec<Similar>, Error> {
        let Some(name) = input.entity else {
            return Ok(Vec::new());
        };
        let Some(entity) = self.lookup_entity_name(name) else {
            return Ok(Vec::new());
        };

        let mut scratch = core::mem::take(&mut self.similarity_scratch);
        scratch.new_terms.clear();
        scratch.unknown.clear();
        scratch.unknown_ranges.clear();
        scratch.candidate_terms.clear();
        scratch.vector.clear();

        let terms = &self.terms;
        let new_terms = &mut scratch.new_terms;
        let unknown = &mut scratch.unknown;
        let unknown_ranges = &mut scratch.unknown_ranges;
        self.tokenizer.tokenize(input.text, &mut |token| {
            if let Some(term) = terms.lookup(token) {
                if !new_terms.contains(&term.0) {
                    new_terms.push(term.0);
                }
                return;
            }
            if unknown_ranges
                .iter()
                .any(|&(start, end)| &unknown[start..end] == token)
            {
                return;
            }
            let start = unknown.len();
            unknown.push_str(token);
            unknown_ranges.push((start, unknown.len()));
        });
        let new_term_count = scratch.new_terms.len() + scratch.unknown_ranges.len();

        let mut similar = Vec::new();
        let result = (|| {
            let encoded = match input.vector {
                Some(vector) => {
                    self.vecs
                        .encode_slot_into(FactId::NONE, vector, &mut scratch.vector)?;
                    SimilarVector::Encoded(&scratch.vector)
                }
                None => SimilarVector::None,
            };
            self.scan_similar(
                SimilarityQuery {
                    entity,
                    exclude: None,
                    terms: &scratch.new_terms,
                    term_count: new_term_count,
                    vector: encoded,
                },
                &mut scratch.candidate_terms,
                &mut similar,
            );
            Ok::<(), Error>(())
        })();
        self.similarity_scratch = scratch;
        result.map(|()| similar)
    }

    /// Similarity detection for an already-stored ordinary remember. This
    /// feeds the same scanner as `remember_guarded`, but reuses the term ids and
    /// quantized vector slot the write has just produced.
    fn find_similar(&mut self, outcome: &mut RememberOutcome) {
        let Some(entity) = outcome.entity else { return };
        let new_vec = self
            .fact(outcome.id)
            .filter(|record| record.has_vector())
            .map(|record| record.vector);
        let mut scratch = core::mem::take(&mut self.similarity_scratch);
        scratch.new_terms.clear();
        scratch
            .new_terms
            .extend(self.tf_scratch.iter().map(|&(term, _)| term));
        let new_term_count = scratch.new_terms.len();
        self.scan_similar(
            SimilarityQuery {
                entity,
                exclude: Some(outcome.id),
                terms: &scratch.new_terms,
                term_count: new_term_count,
                vector: new_vec.map_or(SimilarVector::None, SimilarVector::Stored),
            },
            &mut scratch.candidate_terms,
            &mut outcome.similar,
        );
        self.similarity_scratch = scratch;
    }

    /// Lexical similar-detection: live facts of the same
    /// entity whose term sets overlap the new fact's above the
    /// `similar_jaccard` threshold. Bounded: the entity's most recent
    /// [`SIMILAR_CANDIDATE_CAP`] facts are compared (a hub's full list is
    /// not re-tokenized).
    ///
    /// Comparing two term sets exactly means having both, and only the new
    /// fact's is at hand — the candidate's would have to be recovered by
    /// reading its text and running it back through the tokenizer. Doing that
    /// for every candidate of every write is what the term-set summary in
    /// [`DocLenSlot`](crate::index::bm25::DocLenSlot) exists to avoid: it
    /// bounds the overlap from above, and a bound below the threshold settles
    /// the question, because Jaccard rises with the intersection. Only a
    /// candidate that survives the bound is read and tokenized, so the
    /// answer is the same one the exhaustive comparison gives.
    ///
    /// The bound is trusted only while the index was built by the current
    /// tokenizer. A stale index may hold terms today's tokenizer would not
    /// produce, which would make the summary describe a different term set
    /// than the comparison uses; then every candidate is read, exactly as
    /// before the summary existed.
    fn scan_similar(
        &mut self,
        query: SimilarityQuery<'_>,
        candidate_terms: &mut Vec<u32>,
        out: &mut Vec<Similar>,
    ) {
        out.clear();
        if query.term_count == 0 && matches!(query.vector, SimilarVector::None) {
            return;
        }
        // Most recent candidates of the entity (ring over the list).
        let mut ring = [FactId::NONE; SIMILAR_CANDIDATE_CAP];
        let mut n = 0usize;
        for (fact, _) in self.entity_facts.entries(query.entity.0) {
            if query.exclude == Some(fact) {
                continue;
            }
            ring[n % SIMILAR_CANDIDATE_CAP] = fact;
            n += 1;
        }
        let summaries_trustworthy = self.bm25_tokenizer_version == TOKENIZER_INDEX_VERSION;
        // Without a vector on the new fact, a lexical overlap is the only hint
        // a candidate can produce, so one ruled out by the summary contributes
        // nothing whatever its record says — and its record is the second
        // arena lookup of the pair this loop would otherwise pay per
        // candidate. (`new_terms` is non-empty here: the two being empty
        // together returned above.)
        let lexical_only = matches!(query.vector, SimilarVector::None);
        for &fact in ring.iter().take(n.min(SIMILAR_CANDIDATE_CAP)) {
            let may_overlap = !query.terms.is_empty()
                && self.overlap_possible(
                    fact,
                    query.terms,
                    query.term_count,
                    summaries_trustworthy,
                );
            if lexical_only && !may_overlap {
                continue;
            }
            let Some(record) = self.fact(fact) else {
                continue;
            };
            if record.is_tombstone() || record.is_closed() {
                continue;
            }

            // Lexical signal: term-set Jaccard over the entity's text.
            let mut lexical = None;
            // Deferred validation: a load no longer scans the
            // text pool, so an unreadable text simply yields no lexical signal.
            if may_overlap && let Ok(text) = core::str::from_utf8(self.texts.get(record.text)) {
                candidate_terms.clear();
                let terms = &self.terms;
                let cand = &mut *candidate_terms;
                self.tokenizer.tokenize(text, &mut |token| {
                    if let Some(term) = terms.lookup(token)
                        && !cand.contains(&term.0)
                    {
                        cand.push(term.0);
                    }
                });
                if !candidate_terms.is_empty() {
                    let both = candidate_terms
                        .iter()
                        .filter(|term| query.terms.contains(term))
                        .count();
                    let union = candidate_terms.len() + query.term_count - both;
                    let jaccard = both as f32 / union as f32;
                    if jaccard > self.cfg.similar_jaccard {
                        lexical = Some(jaccard);
                    }
                }
            }

            // Vector signal: quantized cosine when both facts carry one.
            let mut vector = None;
            if record.has_vector() {
                let cos = match query.vector {
                    SimilarVector::None => 0.0,
                    SimilarVector::Stored(slot) => self.vecs.cosine_slots(slot, record.vector),
                    SimilarVector::Encoded(encoded) => {
                        self.vecs.cosine_encoded_slot(encoded, record.vector)
                    }
                };
                if cos > self.cfg.similar_cos {
                    vector = Some(cos);
                }
            }

            // Keep the stronger signal; a tie prefers the lexical reason.
            let best = match (lexical, vector) {
                (Some(l), Some(v)) if v > l => Some((v, SimilarReason::VectorCosine)),
                (Some(l), _) => Some((l, SimilarReason::LexicalOverlap)),
                (None, Some(v)) => Some((v, SimilarReason::VectorCosine)),
                (None, None) => None,
            };
            if let Some((score, reason)) = best {
                out.push(Similar {
                    id: fact,
                    score,
                    reason,
                });
            }
        }
        out.sort_unstable_by(|a, b| b.score.total_cmp(&a.score).then(a.id.cmp(&b.id)));
        out.truncate(8);
    }

    /// Whether `candidate` could possibly overlap `new_terms` above
    /// `similar_jaccard` — the cheap half of [`Memory::find_similar`].
    ///
    /// `false` is a proof, not a guess. The candidate's summary marks a term
    /// absent only when it really is absent, so the terms it *might* share is
    /// an upper bound `u` on the true intersection `i`. Jaccard
    /// `i / (|A| + |B| - i)` is increasing in `i`, so `u` at or below the
    /// threshold puts the true value there too. `true` means "unknown" and
    /// sends the caller to the exact comparison — which is also what an
    /// unsummarized document and an untrustworthy index return.
    fn overlap_possible(
        &self,
        candidate: FactId,
        new_terms: &[u32],
        new_term_count: usize,
        trust_summary: bool,
    ) -> bool {
        if !trust_summary {
            return true;
        }
        let Some(doc) = self.bm25.doc(candidate) else {
            return true;
        };
        if !doc.has_signature() {
            return true;
        }
        let bound = doc.overlap_bound(new_terms);
        // `overlap_bound` counts a subset of `new_terms`, so the subtraction
        // is grouped to happen where it provably cannot go below zero. The
        // union is then at least `distinct`, which `has_signature` just
        // established is non-zero — no underflow and no division by zero, on a
        // 64-bit `usize` or a 32-bit one.
        debug_assert!(
            bound <= new_terms.len(),
            "the overlap bound counts query terms, so it cannot exceed them"
        );
        let union = (new_term_count - bound) + usize::from(doc.distinct);
        bound as f32 / union as f32 > self.cfg.similar_jaccard
    }

    /// Revises `target`: closes its validity at the new fact's
    /// `valid_from` and records the new fact with `revises = target`
    /// (rule 2).
    pub fn revise<S: Storage>(
        &mut self,
        store: &mut S,
        target: FactId,
        input: RememberInput<'_>,
    ) -> Result<RememberOutcome, Error> {
        self.validate_input(&input)?;
        // Check first, mutate last: the target closes only after the new
        // fact exists, so a mid-operation failure (capacity) never leaves
        // a closed fact without its successor.
        self.check_revisable(target)?;
        let outcome = self.apply_remember(&input, target, None)?;
        let valid_from = input.valid_from.unwrap_or(input.now);
        self.close_target(target, valid_from);
        self.journal_remember(store, &input, target, outcome.id)?;
        Ok(outcome)
    }

    /// Tombstones a fact: it disappears from every query immediately, the
    /// bytes go with the next `maintain` (rule 3). `Ok(false)`
    /// when it was already tombstoned.
    pub fn forget<S: Storage>(
        &mut self,
        store: &mut S,
        now: u64,
        id: FactId,
    ) -> Result<bool, Error> {
        let fresh = self.apply_forget(id)?;
        let mut entry = Vec::new();
        Op::Forget { now, fact: id }.encode(&mut entry);
        store
            .append_journal(&entry)
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        Ok(fresh)
    }

    /// Removes `tag` from every current fact without deleting knowledge.
    ///
    /// Each affected fact is closed and replaced by an otherwise-identical
    /// successor without the tag. Historical `as_of` queries keep seeing the
    /// old classification; current recall and [`Memory::list_tags`] stop
    /// seeing it immediately. The interned string is stable historical residue
    /// and is not physically reclaimed.
    pub fn remove_tag<S: Storage>(
        &mut self,
        store: &mut S,
        now: u64,
        tag: &str,
    ) -> Result<RemoveTagReport, Error> {
        if tag.is_empty() {
            return Err(Error::Invalid("empty tag"));
        }
        let report = self.apply_remove_tag(now, tag)?;
        if report.affected != 0 {
            let mut entry = Vec::new();
            Op::RemoveTag { now, tag }.encode(&mut entry);
            store
                .append_journal(&entry)
                .map_err(|e| Error::Storage(format!("{e:?}")))?;
        }
        Ok(report)
    }

    /// Upserts a typed edge between two entities (created lazily);
    /// re-linking an existing `(src, rel, dst)` updates its provenance.
    pub fn link<S: Storage>(&mut self, store: &mut S, input: LinkInput<'_>) -> Result<(), Error> {
        self.apply_link(
            input.now,
            input.src,
            input.rel,
            input.dst,
            FactId::from_opt(input.provenance),
        )?;
        let mut entry = Vec::new();
        Op::Link {
            now: input.now,
            src: input.src,
            rel: input.rel,
            dst: input.dst,
            provenance: FactId::from_opt(input.provenance),
        }
        .encode(&mut entry);
        store
            .append_journal(&entry)
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        Ok(())
    }

    /// Closes the current typed edge between two entities. Returns `false`
    /// when the edge is already absent.
    pub fn unlink<S: Storage>(
        &mut self,
        store: &mut S,
        input: UnlinkInput<'_>,
    ) -> Result<bool, Error> {
        let fresh = self.apply_unlink(input.now, input.src, input.rel, input.dst)?;
        let mut entry = Vec::new();
        Op::Unlink {
            now: input.now,
            src: input.src,
            rel: input.rel,
            dst: input.dst,
        }
        .encode(&mut entry);
        store
            .append_journal(&entry)
            .map_err(|e| Error::Storage(format!("{e:?}")))?;
        Ok(fresh)
    }

    /// Returns a fact unless it is tombstoned (closed facts are
    /// returned — their interval says so).
    pub fn get(&self, id: FactId) -> Option<FactView<'_>> {
        let record = self.fact(id)?;
        if record.is_tombstone() {
            return None;
        }
        // Deferred validation: a load no longer scans the text
        // pool, so tolerate invalid bytes here — a corrupt text hides the fact
        // rather than panicking. `verify()` reports it explicitly.
        let text = core::str::from_utf8(self.texts.get(record.text)).ok()?;
        Some(FactView { record, text })
    }

    /// Appends the tag terms of a fact to `out`. A tombstoned or unknown
    /// fact contributes nothing.
    pub fn tags_of(&self, id: FactId, out: &mut Vec<TermId>) {
        let Some(record) = self.fact(id) else { return };
        if record.is_tombstone() {
            return;
        }
        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
            return;
        };
        for chunk in self.tag_lists.iter(&aux.tags) {
            for raw in chunk.chunks_exact(4) {
                out.push(TermId(u32::from_be_bytes(raw.try_into().unwrap())));
            }
        }
    }

    /// Returns one bounded, stable page of current tags in lexical order.
    ///
    /// A cursor is tied to the tag catalog state and prefix. If a concurrent
    /// mutation changes any current tag count, continuing returns
    /// [`Error::StaleCursor`] rather than silently skipping or duplicating a
    /// tag.
    pub fn list_tags(&self, query: TagQuery<'_>) -> Result<TagPage, Error> {
        self.tag_catalog.page(&self.terms, self.cfg.db_uuid, query)
    }

    /// Fills `out` with the fact's metadata as key→value pairs in canonical
    /// (ascending-key) order — the same order the blob was written and the same
    /// order a host `BTreeMap` yields, so no layer sorts twice. Returns `true`
    /// when the fact carries metadata (even an empty map is `false`, since an
    /// empty map is stored as no blob).
    ///
    /// A tombstoned or unknown fact, a fact with no metadata blob, or a blob
    /// that fails to decode (deferred validation, — `verify`
    /// reports it) all leave `out` empty and return `false`; this accessor never
    /// panics on bad bytes.
    pub fn metadata_of<'s>(&'s self, id: FactId, out: &mut Vec<(&'s str, &'s str)>) -> bool {
        out.clear();
        let Some(record) = self.fact(id) else {
            return false;
        };
        if record.is_tombstone() {
            return false;
        }
        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
            return false;
        };
        if aux.meta.0 == NONE_U32 || aux.meta.0 >= self.metas.len() as u32 {
            return false;
        }
        crate::metadata::decode(self.metas.get(aux.meta), out).is_ok() && !out.is_empty()
    }

    /// Resolves an entity by name (normalized), without creating it.
    pub fn entity(&mut self, name: &str) -> Option<EntityId> {
        let mut norm = core::mem::take(&mut self.name_scratch);
        normalize_name(&mut self.tokenizer, name, &mut norm);
        let found = if norm.is_empty() {
            None
        } else {
            // Only resolve an *existing* term: interning would create it.
            self.lookup_entity_by_norm(&norm)
        };
        self.name_scratch = norm;
        found
    }

    /// Resolves the string behind a term id (tags, relations).
    pub fn term(&self, id: TermId) -> &str {
        self.terms.resolve(id)
    }

    /// The canonical (verbatim) name of an entity, or `None` for an
    /// unknown id. The read-side inverse of [`Memory::entity`]: it lets a
    /// host export a fact with its subject *name* instead of the internal
    /// [`EntityId`] carried by [`FactRecord`].
    pub fn entity_name(&self, id: EntityId) -> Option<&str> {
        let record = self.entities.get(&id.0.to_be_bytes())?;
        // Tolerate invalid bytes (deferred validation): an
        // unreadable name reads as `None`, never a panic.
        core::str::from_utf8(self.texts.get(record.name)).ok()
    }

    /// Visits every currently-open edge once, as
    /// `(source name, relation, destination name, provenance)`.
    ///
    /// Reads the out-arena only: each edge is mirrored in both arenas, and the
    /// out-arena holds it keyed `(src, rel, dst)`, which is the direction a
    /// caller wants to write back. Closed versions live in the history arenas
    /// and are not visited — this is the *current* graph, matching what a fact
    /// export does for facts.
    ///
    /// Names are borrowed from the interner and the text pool, so a caller that
    /// writes them straight out allocates nothing per edge. An edge whose
    /// endpoint name is unreadable is skipped rather than reported with a
    /// placeholder: deferred validation means a damaged image must not turn
    /// into plausible-looking output.
    ///
    /// `visit` returning `false` stops the walk, so a caller can bound its own
    /// work without materializing the graph.
    pub fn edges_each(&self, mut visit: impl FnMut(&str, &str, &str, FactId) -> bool) {
        for slot in self.edges_out.iter() {
            let (Some(src), Some(dst)) = (self.entity_name(slot.a), self.entity_name(slot.b))
            else {
                continue;
            };
            if !visit(src, self.term(slot.rel), dst, slot.fact) {
                return;
            }
        }
    }

    /// Number of fact records currently stored (tombstoned facts count
    /// until `maintain` removes them). Purged ids stay burned, so this can
    /// be below [`Stats::next_fact`].
    pub fn facts_len(&self) -> usize {
        self.facts.len()
    }

    /// Number of entities.
    pub fn entities_len(&self) -> usize {
        self.entities.len()
    }

    /// The engine configuration.
    pub fn cfg(&self) -> &Config {
        &self.cfg
    }

    /// Size counters of the engine. O(1).
    pub fn stats(&self) -> Stats {
        Stats {
            facts: self.facts.len(),
            entities: self.entities.len(),
            terms: self.terms.len(),
            edges: self.edges_out.len(),
            edge_versions: self.edges_hist_out.len(),
            vectors: self.vecs.len(),
            tombstones: self.tombstones,
            hnsw_indexed: self.hnsw.indexed(),
            next_fact: self.next_fact,
            next_entity: self.next_entity,
            next_edge: self.next_edge,
            db_uuid: self.cfg.db_uuid,
            pool_bytes: self.facts.pool_bytes()
                + self.fact_aux.pool_bytes()
                + self.entities.pool_bytes()
                + self.by_name.pool_bytes()
                + self.edges_out.pool_bytes()
                + self.edges_in.pool_bytes()
                + self.edges_hist_out.pool_bytes()
                + self.edges_hist_in.pool_bytes()
                + self.temporal.pool_bytes()
                + self.texts.pool_bytes()
                + self.terms.pool_bytes()
                + self.tag_lists.pool_bytes()
                + self.bm25.pool_bytes()
                + self.tags_idx.pool_bytes()
                + self.tag_catalog.pool_bytes()
                + self.entity_facts.pool_bytes()
                + self.vecs.pool_bytes()
                + self.hnsw.pool_bytes(),
            shards: shards::ShardLayout::of_config(&self.cfg),
        }
    }

    // ---- internals ----

    fn fact(&self, id: FactId) -> Option<FactRecord> {
        self.facts.get(&id.0.to_be_bytes())
    }

    /// Size-limit checks shared by remember and revise.
    fn validate_input(&self, input: &RememberInput<'_>) -> Result<(), Error> {
        if input.text.len() > self.cfg.max_text {
            return Err(Error::TooLarge {
                what: "text",
                len: input.text.len(),
                max: self.cfg.max_text,
            });
        }
        if input.tags.len() > MAX_TAGS {
            return Err(Error::TooLarge {
                what: "tags",
                len: input.tags.len(),
                max: MAX_TAGS,
            });
        }
        if input.links.len() > 16 {
            return Err(Error::TooLarge {
                what: "links",
                len: input.links.len(),
                max: 16,
            });
        }
        if input.tags.iter().any(|t| t.is_empty()) {
            return Err(Error::Invalid("empty tag"));
        }
        if !input.links.is_empty() && input.entity.is_none() {
            return Err(Error::Invalid("links require a subject entity"));
        }
        if let Some(v) = input.vector {
            if self.cfg.dim == 0 {
                return Err(Error::Invalid("vector given but dim is 0"));
            }
            if v.len() != self.cfg.dim {
                return Err(Error::DimMismatch {
                    got: v.len(),
                    want: self.cfg.dim,
                });
            }
        }
        Ok(())
    }

    /// `Ok` when `target` exists, is not tombstoned and not yet closed.
    fn check_revisable(&self, target: FactId) -> Result<(), Error> {
        let record = self.fact(target).ok_or(Error::NotFound(target))?;
        if record.is_tombstone() {
            return Err(Error::NotFound(target));
        }
        if record.is_closed() {
            return Err(Error::AlreadyClosed(target));
        }
        Ok(())
    }

    /// Closes a revision target (validity ends where the successor
    /// starts). The caller checked [`Memory::check_revisable`] first.
    fn close_target(&mut self, target: FactId, valid_to: u64) {
        let record = self.fact(target).expect("checked revisable");
        let payload = self
            .facts
            .payload_mut(&target.0.to_be_bytes())
            .expect("record fetched above");
        // Payload offsets = slot offsets - KEY_LEN(4): flags at 4..6,
        // valid_to at 36..44.
        let flags = record.flags | fact_flags::CLOSED;
        payload[4..6].copy_from_slice(&flags.to_be_bytes());
        payload[36..44].copy_from_slice(&valid_to.to_be_bytes());
        self.change_catalog_for_fact(target, -1);
    }

    /// The one execution path of a new fact — called by the public verbs
    /// and by replay with identical effects.
    fn apply_remember(
        &mut self,
        input: &RememberInput<'_>,
        revises: FactId,
        copy_vector: Option<u32>,
    ) -> Result<RememberOutcome, Error> {
        let id = FactId(self.next_fact);
        let entity = match input.entity {
            Some(name) => Some(self.resolve_or_create_entity(name, input.now)?),
            None => None,
        };
        let text_id = self.texts.push(input.text.as_bytes())?;

        // Tokenize into (term, tf) pairs in the reusable scratch.
        let mut tfs = core::mem::take(&mut self.tf_scratch);
        let mut probe = core::mem::take(&mut self.tf_probe);
        tfs.clear();
        probe.fill(0);
        let terms = &mut self.terms;
        let mut intern_err = None;
        self.tokenizer.tokenize(input.text, &mut |token| {
            if intern_err.is_some() {
                return;
            }
            match terms.intern(token) {
                Ok(term) => tally_term(&mut tfs, &mut probe, term.0),
                Err(e) => intern_err = Some(e),
            }
        });
        self.tf_probe = probe;
        if let Some(e) = intern_err {
            self.tf_scratch = tfs;
            return Err(Error::Arena(e));
        }
        self.bm25.index_doc(id, &tfs)?;
        self.tf_scratch = tfs;

        // Metadata: canonicalized (keys sorted, dups rejected) and stored as one
        // opaque blob; an absent or empty map leaves the sentinel. Pushed before
        // the fact record so a capacity/validation failure aborts the whole op.
        let meta = match input.metadata {
            Some(pairs) if !pairs.is_empty() => {
                self.metas.push(&crate::metadata::encode(pairs)?)?
            }
            _ => BlobId(NONE_U32),
        };

        // Tags: interned verbatim, deduplicated, listed on the fact and
        // inverted.
        let mut aux = FactAux {
            id,
            tags: ListHandle::EMPTY,
            meta,
        };
        let mut seen_tags: [u32; MAX_TAGS] = [NONE_U32; MAX_TAGS];
        let mut seen_cnt = 0usize;
        for tag in input.tags {
            let term = self.terms.intern(tag)?;
            if seen_tags[..seen_cnt].contains(&term.0) {
                continue;
            }
            seen_tags[seen_cnt] = term.0;
            seen_cnt += 1;
            self.tag_lists.push(&mut aux.tags, &term.0.to_be_bytes())?;
            self.tags_idx.push(term.0, id, 0)?;
        }
        self.fact_aux.insert(&aux)?;

        // Links: subject → target edges with this fact as provenance.
        if let Some(src) = entity {
            for &(rel, dst_name) in input.links {
                let dst = self.resolve_or_create_entity(dst_name, input.now)?;
                let rel = self.terms.intern(rel)?;
                self.open_edge(input.now, src, rel, dst, id)?;
            }
            self.entity_facts.push(src.0, id, 0)?;
        }

        // Vector: quantized into the flat pool; the fact keeps the slot
        // index. Pushed before the record so a capacity failure aborts the
        // whole op (replay rebuilds consistently, like the other indexes).
        let (vector, flags) = match (input.vector, copy_vector) {
            (Some(v), None) => (self.vecs.push(id, v)?, fact_flags::HAS_VECTOR),
            (None, Some(source)) => (
                self.vecs.clone_slot_for_fact(id, source)?,
                fact_flags::HAS_VECTOR,
            ),
            (None, None) => (NONE_U32, 0),
            (Some(_), Some(_)) => unreachable!("retag does not provide a raw vector"),
        };

        let recorded_at = input.now;
        let valid_from = input.valid_from.unwrap_or(input.now);
        self.facts.insert(&FactRecord {
            id,
            entity: EntityId::from_opt(entity),
            flags,
            kind: 0,
            text: text_id,
            vector,
            revises,
            recorded_at,
            valid_from,
            valid_to: VALID_TO_OPEN,
        })?;
        self.temporal.insert(&TemporalSlot {
            recorded_at,
            fact: id,
        })?;
        for &term in &seen_tags[..seen_cnt] {
            self.tag_catalog.change(&self.terms, TermId(term), 1);
        }
        self.next_fact += 1;
        Ok(RememberOutcome {
            id,
            entity,
            similar: Vec::new(),
        })
    }

    fn apply_forget(&mut self, id: FactId) -> Result<bool, Error> {
        let record = self.fact(id).ok_or(Error::NotFound(id))?;
        if record.is_tombstone() {
            return Ok(false);
        }
        let payload = self
            .facts
            .payload_mut(&id.0.to_be_bytes())
            .expect("record fetched above");
        let flags = record.flags | fact_flags::TOMBSTONE;
        payload[4..6].copy_from_slice(&flags.to_be_bytes());
        self.tombstones += 1;
        if !record.is_closed() {
            self.change_catalog_for_fact(id, -1);
        }
        Ok(true)
    }

    fn apply_remove_tag(&mut self, now: u64, tag: &str) -> Result<RemoveTagReport, Error> {
        let Some(term) = self.terms.lookup(tag) else {
            return Ok(RemoveTagReport::default());
        };
        let targets: Vec<FactId> = self
            .tags_idx
            .entries(term.0)
            .map(|(id, _)| id)
            .filter(|&id| {
                self.fact(id)
                    .is_some_and(|record| !record.is_tombstone() && !record.is_closed())
            })
            .collect();
        let mut affected = 0u32;
        for target in targets {
            self.retag_without(now, target, tag)?;
            affected = affected.saturating_add(1);
        }
        Ok(RemoveTagReport { affected })
    }

    fn retag_without(&mut self, now: u64, target: FactId, removed: &str) -> Result<(), Error> {
        let record = self.fact(target).ok_or(Error::NotFound(target))?;
        let view = self.get(target).ok_or(Error::NotFound(target))?;
        let text = view.text.to_string();
        let entity = record
            .entity
            .some()
            .and_then(|id| self.entity_name(id))
            .map(ToString::to_string);

        let mut tag_terms = Vec::new();
        self.tags_of(target, &mut tag_terms);
        let tags: Vec<String> = tag_terms
            .into_iter()
            .map(|term| self.term(term))
            .filter(|name| *name != removed)
            .map(ToString::to_string)
            .collect();
        let tag_refs: Vec<&str> = tags.iter().map(String::as_str).collect();

        let mut metadata = Vec::new();
        self.metadata_of(target, &mut metadata);
        let metadata: Vec<(String, String)> = metadata
            .into_iter()
            .map(|(key, value)| (key.to_string(), value.to_string()))
            .collect();
        let metadata_refs: Vec<(&str, &str)> = metadata
            .iter()
            .map(|(key, value)| (key.as_str(), value.as_str()))
            .collect();
        let vector = (record.flags & fact_flags::HAS_VECTOR != 0).then_some(record.vector);
        let input = RememberInput {
            now,
            text: &text,
            entity: entity.as_deref(),
            tags: &tag_refs,
            links: &[],
            vector: None,
            valid_from: Some(now),
            metadata: (!metadata_refs.is_empty()).then_some(metadata_refs.as_slice()),
        };
        self.apply_remember(&input, target, vector)?;
        self.close_target(target, now);
        Ok(())
    }

    /// Applies one current-count change for every tag attached to `id`.
    /// Facts accept at most 32 distinct tags, so this uses fixed stack storage
    /// and never allocates on revise/forget.
    fn change_catalog_for_fact(&mut self, id: FactId, delta: i32) {
        let Some(aux) = self.fact_aux.get(&id.0.to_be_bytes()) else {
            return;
        };
        let mut ids = [NONE_U32; MAX_TAGS];
        let mut len = 0usize;
        for chunk in self.tag_lists.iter(&aux.tags) {
            for raw in chunk.chunks_exact(4) {
                if len == ids.len() {
                    break;
                }
                ids[len] = u32::from_be_bytes(raw.try_into().unwrap());
                len += 1;
            }
        }
        for &term in &ids[..len] {
            self.tag_catalog.change(&self.terms, TermId(term), delta);
        }
    }

    fn apply_link(
        &mut self,
        now: u64,
        src: &str,
        rel: &str,
        dst: &str,
        provenance: FactId,
    ) -> Result<(), Error> {
        let src = self.resolve_or_create_entity(src, now)?;
        let dst = self.resolve_or_create_entity(dst, now)?;
        let rel = self.terms.intern(rel)?;
        self.open_edge(now, src, rel, dst, provenance)
    }

    fn apply_unlink(&mut self, now: u64, src: &str, rel: &str, dst: &str) -> Result<bool, Error> {
        let Some(src) = self.lookup_entity_name(src) else {
            return Ok(false);
        };
        let Some(dst) = self.lookup_entity_name(dst) else {
            return Ok(false);
        };
        let Some(rel) = self.terms.lookup(rel) else {
            return Ok(false);
        };
        self.close_current_edge(now, src, rel, dst)
    }

    fn open_edge(
        &mut self,
        now: u64,
        src: EntityId,
        rel: TermId,
        dst: EntityId,
        fact: FactId,
    ) -> Result<(), Error> {
        if let Some(current) = self.current_edge(src, rel, dst) {
            if current.fact == fact {
                return Ok(());
            }
            self.close_current_edge(now, src, rel, dst)?;
        }
        let edge = EdgeId(self.next_edge);
        let history = EdgeHistorySlot {
            a: src,
            rel,
            b: dst,
            edge,
            fact,
            flags: 0,
            kind: 0,
            recorded_at: now,
            valid_from: now,
            valid_to: VALID_TO_OPEN,
        };
        self.insert_history_edge(history)?;
        self.insert_current_edge(src, rel, dst, fact, edge, now)?;
        self.next_edge += 1;
        Ok(())
    }

    fn insert_current_edge(
        &mut self,
        src: EntityId,
        rel: TermId,
        dst: EntityId,
        fact: FactId,
        edge: EdgeId,
        valid_from: u64,
    ) -> Result<(), Error> {
        for (arena, a, b) in [
            (&mut self.edges_out, src, dst),
            (&mut self.edges_in, dst, src),
        ] {
            let slot = EdgeSlot {
                a,
                rel,
                b,
                fact,
                edge,
                valid_from,
            };
            if !arena.insert(&slot)? {
                let payload = arena
                    .payload_mut(&edge_key(a, rel, b))
                    .expect("insert reported a duplicate");
                let mut full = [0u8; EdgeSlot::SIZE];
                slot.write(&mut full);
                payload.copy_from_slice(&full[EdgeSlot::KEY_LEN..]);
            }
        }
        Ok(())
    }

    fn insert_history_edge(&mut self, edge: EdgeHistorySlot) -> Result<(), Error> {
        self.edges_hist_out.insert(&edge)?;
        self.edges_hist_in.insert(&EdgeHistorySlot {
            a: edge.b,
            b: edge.a,
            ..edge
        })?;
        Ok(())
    }

    /// Closes the open version of `(src, rel, dst)` and drops it from the
    /// current graph. Returns `false` when there is no such edge.
    ///
    /// The current slot names its own history record — `edge` and `valid_from`
    /// are that record's key tail — so both mirrors are addressed directly.
    fn close_current_edge(
        &mut self,
        now: u64,
        src: EntityId,
        rel: TermId,
        dst: EntityId,
    ) -> Result<bool, Error> {
        let Some(current) = self.current_edge(src, rel, dst) else {
            return Ok(false);
        };
        // A version never ends before it began, even if the caller's clock
        // moved backwards between the link and the unlink.
        let close_at = now.max(current.valid_from);
        let out_key = edge_history_key(src, current.valid_from, current.edge);
        let in_key = edge_history_key(dst, current.valid_from, current.edge);
        close_edge_history_payload(
            self.edges_hist_out
                .payload_mut(&out_key)
                .ok_or(Error::Corrupt("missing outgoing edge history"))?,
            close_at,
        );
        close_edge_history_payload(
            self.edges_hist_in
                .payload_mut(&in_key)
                .ok_or(Error::Corrupt("missing incoming edge history"))?,
            close_at,
        );
        let out_removed = self.edges_out.remove(&edge_key(src, rel, dst));
        let in_removed = self.edges_in.remove(&edge_key(dst, rel, src));
        if out_removed != in_removed {
            return Err(Error::Corrupt("edge mirrors disagree"));
        }
        Ok(out_removed)
    }

    fn current_edge(&self, src: EntityId, rel: TermId, dst: EntityId) -> Option<EdgeSlot> {
        self.edges_out
            .get_slot(&edge_key(src, rel, dst))
            .map(EdgeSlot::read)
    }

    /// Looks an entity up by its already-normalized name (read-only:
    /// neither the vocabulary nor the arenas change on a miss).
    fn lookup_entity_by_norm(&self, norm: &str) -> Option<EntityId> {
        let term = self.terms.lookup(norm)?;
        let mut from = [0u8; 8];
        key::write_u32(&mut from, term.0);
        let mut to = [0u8; 8];
        key::write_u32(&mut to, term.0);
        to[4..].copy_from_slice(&u32::MAX.to_be_bytes());
        self.by_name.range(&from, &to).next().map(|e| e.id)
    }

    fn lookup_entity_name(&mut self, name: &str) -> Option<EntityId> {
        let mut norm = core::mem::take(&mut self.name_scratch);
        normalize_name(&mut self.tokenizer, name, &mut norm);
        let result = (!norm.is_empty())
            .then(|| self.lookup_entity_by_norm(&norm))
            .flatten();
        self.name_scratch = norm;
        result
    }

    fn resolve_or_create_entity(&mut self, name: &str, now: u64) -> Result<EntityId, Error> {
        let mut norm = core::mem::take(&mut self.name_scratch);
        normalize_name(&mut self.tokenizer, name, &mut norm);
        if norm.is_empty() {
            self.name_scratch = norm;
            return Err(Error::Invalid("entity name has no indexable characters"));
        }
        let result = (|| {
            if let Some(found) = self.lookup_entity_by_norm(&norm) {
                return Ok(found);
            }
            let term = self.terms.intern(&norm)?;
            let id = EntityId(self.next_entity);
            let name_id = self.texts.push(name.as_bytes())?;
            self.entities.insert(&EntityRecord {
                id,
                name: name_id,
                name_term: term,
                created_at: now,
                flags: 0,
            })?;
            self.by_name.insert(&EntityByName {
                name_term: term,
                id,
            })?;
            self.next_entity += 1;
            Ok(id)
        })();
        self.name_scratch = norm;
        result
    }

    fn journal_remember<S: Storage>(
        &mut self,
        store: &mut S,
        input: &RememberInput<'_>,
        revises: FactId,
        assigned: FactId,
    ) -> Result<(), Error> {
        let mut entry = Vec::new();
        Op::Remember {
            now: input.now,
            valid_from: input.valid_from.unwrap_or(input.now),
            entity: input.entity,
            text: input.text,
            tags: input.tags.to_vec(),
            links: input.links.to_vec(),
            vector: input.vector.map(<[f32]>::to_vec).unwrap_or_default(),
            metadata: input.metadata.map(<[_]>::to_vec).unwrap_or_default(),
            revises,
            assigned,
        }
        .encode(&mut entry);
        store
            .append_journal(&entry)
            .map_err(|e| Error::Storage(format!("{e:?}")))
    }
}

impl core::fmt::Debug for Memory<'_> {
    /// Summary only — the contents are the user's memory, not ours to
    /// print.
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.debug_struct("Memory")
            .field("facts", &self.facts.len())
            .field("entities", &self.entities.len())
            .field("terms", &self.terms.len())
            .finish()
    }
}

/// Normalizes an entity name: its tokens joined by single spaces
/// ("  Проект   PlugMem " → "проект plugmem"). Deterministic and aligned
/// with search-time tokenization.
fn normalize_name(tokenizer: &mut Tokenizer, name: &str, out: &mut String) {
    out.clear();
    tokenizer.tokenize(name, &mut |token| {
        if !out.is_empty() {
            out.push(' ');
        }
        out.push_str(token);
    });
}

/// Probe index for `term` in a table of `mask + 1` slots. Fibonacci hashing —
/// the same constant the arena shards with, so term ids scatter without a
/// second hash family.
#[inline]
fn probe_at(term: u32, mask: usize) -> usize {
    (u64::from(term).wrapping_mul(0x9E37_79B9_7F4A_7C15) >> 32) as usize & mask
}

/// Records one occurrence of `term` in `tfs`, using `probe` as a
/// `term -> index + 1` open-addressing table (0 is empty).
///
/// This replaces a scan of the terms already seen, which made indexing a
/// document quadratic in its length — at the 4 KiB `max_text` ceiling, on the
/// order of 10^5 comparisons for one `remember`.
///
/// **A term keeps the slot its first occurrence claimed**, so `tfs` comes out
/// in first-occurrence order exactly as the scan left it and `index_doc`
/// receives byte-identical input.
fn tally_term(tfs: &mut Vec<(u32, u8)>, probe: &mut Vec<u32>, term: u32) {
    // Half full at most: probes stay short and the loop below terminates.
    // Growth rehashes from `tfs`, which is the authoritative order.
    if (tfs.len() + 1) * 2 > probe.len() {
        let grown = (probe.len() * 2).max(64);
        probe.clear();
        probe.resize(grown, 0);
        let mask = grown - 1;
        for (i, &(seen, _)) in tfs.iter().enumerate() {
            let mut at = probe_at(seen, mask);
            while probe[at] != 0 {
                at = (at + 1) & mask;
            }
            probe[at] = i as u32 + 1;
        }
    }
    let mask = probe.len() - 1;
    let mut at = probe_at(term, mask);
    loop {
        match probe[at] {
            0 => {
                probe[at] = tfs.len() as u32 + 1;
                tfs.push((term, 1));
                return;
            }
            slot => {
                let entry = &mut tfs[slot as usize - 1];
                if entry.0 == term {
                    entry.1 = entry.1.saturating_add(1);
                    return;
                }
            }
        }
        at = (at + 1) & mask;
    }
}