perf(hot-path): persistent BM25F index, flat bulk ingest, batched write groups (#1135)

CHANGELOG/v4.md

@@ -15,6 +15,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Hot-path performance overhaul ([#1135](https://github.com/robotrocketscience/aelfrice/issues/1135)).** Measured audit of the retrieval + ingest hot paths, fixed structurally (no ranking change — retrieval outputs are byte-identical on a fixed store):
+  - *BM25F index persists across processes.* The default-ON BM25F lane rebuilt its index from scratch on every `retrieve()` — the `UserPromptSubmit` hook is a fresh process per prompt, so every prompt re-tokenized and Porter-stemmed the whole corpus (~584 ms at 5k beliefs, linear in corpus size). The built index is now written to a `<db-path>.bm25f` sidecar stamped with a new durable `store_generation` counter (bumped in the same transaction as every belief/edge content mutation) and reloaded in low milliseconds while the stamp matches; any content change invalidates it. Long-running processes (MCP server) additionally reuse one in-memory cache per store instead of rebuilding — and no longer leak an invalidation callback per query.
+  - *Bulk ingest is no longer O(n²).* Each ingested turn snapshotted the full belief-id set and re-read its log rows to tell inserts from corroborations; the derivation worker now reports per-row `(belief_id, was_inserted)` outcomes, and the worker's unstamped-log scan got a partial index (the log grows monotonically; the unstamped set stays tiny). Per-turn cost is now flat with respect to corpus size.
+  - *Write groups batch into single transactions.* A new `MemoryStore.transaction()` context manager suppresses per-call commits (measured 33× cheaper than commit-per-row for a 200-insert group); wired into the ingest turn (~8 commits → 1, and a turn is now crash-atomic), the hook's retrieval-audit / injection-event / touch loops (~45–60 commits per prompt with hits → a handful), and the deferred-feedback enqueue. `PRAGMA synchronous=NORMAL` now pairs with WAL as documented.
+  - *Smaller structural fixes.* Partial index for the locked-belief tier (ran up to 3× per retrieve as a full-table scan) and an index for the `edges.type` existence probe; the `UserPromptSubmit` hook opens the store once per prompt instead of 4–6 times; `.aelfrice.toml` is parsed once per file version instead of ~24× per retrieve (the directory walk still runs, so config file creation/deletion is honoured); the v1.2 origin backfill (two full-table UPDATE statements) is now a marker-gated one-shot instead of running on every store open. The BM25 serialize format is v2 (float64 scalars) so a loaded index scores byte-identically to a built one; no v1 blobs existed.
+
 ### Fixed
 
 - **Reader-facing docs: prose de-formularization + residual default-flip drift ([#1141](https://github.com/robotrocketscience/aelfrice/issues/1141)).** Style pass over README and the concepts docs (drop repeated bold-led bullet formulas, unbold sentence-lead paragraphs, fix a "two recovery angles" lead on a three-item list). Content fixes found on the same sweep, continuing #1137: `ARCHITECTURE.md` still showed the temporal-spine lane default-OFF (default-ON since the #1107 Phase-2 cutover), the mirror hook as env-inert (consent-gated since v4.0, #1089), and agent-context as unreleased (shipped v4.0.0); `HARNESS_INTEGRATION.md` predated the claude-memory mirror and claimed the two stores never merge (one-way consent-gated mirror since v3.7/#985 + v4.0/#1089); `RELEASING.md` said the current line is v3.x; `MCP.md`'s "CLI-only" phrasing aligned with the corrected COMMANDS.md claim.

src/aelfrice/bm25.py

@@ -39,9 +39,13 @@
 from __future__ import annotations
 
 import io
+import os
 import re
+import sys
+import tempfile
 from dataclasses import dataclass, field
 from functools import lru_cache
+from pathlib import Path
 from typing import Final
 
 import numpy as np
@@ -99,7 +103,10 @@
 # changes incompatibly. The format is documented in
 # `BM25Index.serialize` / `BM25Index.deserialize`.
 _SERIALIZE_MAGIC: Final[bytes] = b"AELFBM25"
-_SERIALIZE_VERSION: Final[int] = 1
+# v2 (#1135): k1/b/avgdl widened float32 -> float64 so a deserialised
+# index scores byte-identically to a fresh build. No v1 blobs exist in
+# the wild — nothing called serialize() before the sidecar cache.
+_SERIALIZE_VERSION: Final[int] = 2
 
 
 def tokenize(text: str) -> list[str]:
@@ -414,12 +421,12 @@
         """Return a deterministic byte representation of the index.
 
         Same inputs (store contents + same `anchor_weight`) round-trip
-        to identical bytes, satisfying AC7. Format::
+        to identical bytes, satisfying AC7. Format (v2)::
 
             magic              8 bytes  b"AELFBM25"
             version            uint32   _SERIALIZE_VERSION
             anchor_weight      int32
-            k1, b, avgdl       float32 x 3
+            k1, b, avgdl       float64 x 3
             n_docs, n_terms    uint64 x 2
             belief_ids         length-prefixed UTF-8 strings
             vocabulary terms   length-prefixed UTF-8 strings
@@ -432,14 +439,21 @@
 
         Vocabulary terms are written in column-index order, which
         matches the sorted-ASC order produced by `build()`.
+
+        v2 (#1135) widened k1/b/avgdl from float32 to float64: `build()`
+        keeps them as Python floats, and the sidecar cache requires a
+        deserialised index to score byte-identically to a fresh build —
+        a float32 round-trip perturbed the low-order bits of every
+        score. dl/idf/tf stay float32 (already float32 in the built
+        index, so their round-trip is exact).
         """
         buf = io.BytesIO()
         buf.write(_SERIALIZE_MAGIC)
         buf.write(np.uint32(_SERIALIZE_VERSION).tobytes())
         buf.write(np.int32(self.anchor_weight).tobytes())
-        buf.write(np.float32(self.k1).tobytes())
-        buf.write(np.float32(self.b).tobytes())
-        buf.write(np.float32(self.avgdl).tobytes())
+        buf.write(np.float64(self.k1).tobytes())
+        buf.write(np.float64(self.b).tobytes())
+        buf.write(np.float64(self.avgdl).tobytes())
 
         n_docs = len(self.belief_ids)
         n_terms = len(self.vocabulary)
@@ -501,9 +515,9 @@
                 f"expected {_SERIALIZE_VERSION}"
             )
         anchor_weight = int(_read(np.dtype(np.int32), 1)[0])
-        k1 = float(_read(np.dtype(np.float32), 1)[0])
-        b = float(_read(np.dtype(np.float32), 1)[0])
-        avgdl = float(_read(np.dtype(np.float32), 1)[0])
+        k1 = float(_read(np.dtype(np.float64), 1)[0])
+        b = float(_read(np.dtype(np.float64), 1)[0])
+        avgdl = float(_read(np.dtype(np.float64), 1)[0])
         n_docs = int(_read(np.dtype(np.uint64), 1)[0])
         n_terms = int(_read(np.dtype(np.uint64), 1)[0])
 
@@ -550,6 +564,22 @@
         )
 
 
+# Sidecar file framing (#1135). The payload after the header is the
+# `BM25Index.serialize()` blob, which carries its own magic + version.
+_SIDECAR_MAGIC: Final[bytes] = b"AELFB25S"
+_SIDECAR_VERSION: Final[int] = 1
+_SIDECAR_SUFFIX: Final[str] = ".bm25f"
+
+
+def sidecar_path_for(store: MemoryStore) -> Path | None:
+    """The persistent-index sidecar path for `store`, or None for
+    in-memory stores (nothing to persist against)."""
+    db_path = store.db_path
+    if db_path == ":memory:":
+        return None
+    return Path(db_path + _SIDECAR_SUFFIX)
+
+
 @dataclass
 class BM25IndexCache:
     """Lazy, invalidation-aware wrapper around a single `BM25Index`.
@@ -558,6 +588,19 @@
     construction, so any belief / edge mutation drops the cached
     index. The next `get()` rebuilds.
 
+    #1135: for on-disk stores the built index is also persisted to a
+    sidecar file (`<db-path>.bm25f`) stamped with the store's durable
+    generation counter and scope id. A fresh process (the
+    UserPromptSubmit hook is one per prompt) deserialises the sidecar
+    instead of re-tokenising + re-stemming the whole corpus — measured
+    584 ms build vs low-ms load at 5k beliefs. Staleness is decided by
+    the stamp: any belief/edge content mutation bumps the generation
+    in the same transaction (see `MemoryStore._commit_mutation`), so a
+    matching stamp proves the blob reflects current content. Loads and
+    writes are fail-soft — a missing, corrupt, foreign (scope-id
+    mismatch), stale, or parameter-mismatched sidecar falls back to a
+    build; an unwritable sidecar is skipped silently.
+
     Per-instance: two caches pointing at different stores never share
     state. Thread safety is the caller's responsibility (matches the
     contract of `aelfrice.retrieval.RetrievalCache`).
@@ -568,6 +611,7 @@
     k1: float = DEFAULT_K1
     b: float = DEFAULT_B
     _index: BM25Index | None = field(default=None, init=False, repr=False)
+    _generation: int | None = field(default=None, init=False, repr=False)
     _subscribed: bool = field(default=False, init=False, repr=False)
 
     def __post_init__(self) -> None:
@@ -576,19 +620,129 @@
             self._subscribed = True
 
     def get(self) -> BM25Index:
-        """Return the current index, building or rebuilding as needed."""
+        """Return the current index; load the sidecar or build as needed."""
+        if self._index is not None and self._generation is not None:
+            # Revalidate against the durable counter: the in-process
+            # invalidation callback only covers own-process mutations,
+            # so without this a long-running process (MCP server) would
+            # never see a sibling process's writes (the default-on
+            # ingest hooks). One indexed point-read per get(); the
+            # pre-#1135 behavior was a full rebuild per query.
+            if self.store.store_generation() != self._generation:
+                self._index = None
         if self._index is None:
+            self._index = self._load_sidecar()
+        if self._index is None:
+            # Read the stamp BEFORE building: a mutation that lands
+            # during the build makes the stamp stale, so the next
+            # reader rebuilds rather than trusting a torn snapshot.
+            generation = self.store.store_generation()
             self._index = BM25Index.build(
                 self.store,
                 anchor_weight=self.anchor_weight,
                 k1=self.k1,
                 b=self.b,
             )
+            self._write_sidecar(self._index, generation)
+            self._generation = generation
         return self._index
 
     def invalidate(self) -> None:
-        """Drop the cached index. Wired to the store mutation hook."""
+        """Drop the cached index. Wired to the store mutation hook.
+
+        The sidecar file is left in place — its generation stamp no
+        longer matches after the mutation, so every reader treats it
+        as stale; the next `get()` rebuild overwrites it.
+        """
         self._index = None
+        self._generation = None
+
+    # --- Sidecar persistence (#1135) ----------------------------------
+
+    def _load_sidecar(self) -> BM25Index | None:
+        """Deserialise a valid sidecar, or None on any miss/mismatch."""
+        path = sidecar_path_for(self.store)
+        if path is None:
+            return None
+        try:
+            blob = path.read_bytes()
+            header_len = len(_SIDECAR_MAGIC) + 4 + 8 + 4
+            if len(blob) < header_len:
+                return None
+            if blob[: len(_SIDECAR_MAGIC)] != _SIDECAR_MAGIC:
+                return None
+            off = len(_SIDECAR_MAGIC)
+            version = int(np.frombuffer(blob, np.uint32, 1, off)[0])
+            if version != _SIDECAR_VERSION:
+                return None
+            off += 4
+            generation = int(np.frombuffer(blob, np.uint64, 1, off)[0])
+            off += 8
+            scope_len = int(np.frombuffer(blob, np.uint32, 1, off)[0])
+            off += 4
+            scope = blob[off:off + scope_len].decode("utf-8")
+            off += scope_len
+            # Scope id catches a swapped-in different DB at the same
+            # path; the generation stamp catches every content
+            # mutation on this DB.
+            if scope != self.store.local_scope_id:
+                return None
+            if generation != self.store.store_generation():
+                return None
+            index = BM25Index.deserialize(blob[off:])
+            if index.anchor_weight != self.anchor_weight:
+                return None
+            # v2 stores k1/b as float64, so the round-trip is exact;
+            # compare at full precision so a nearly-equal config never
+            # reuses another config's sidecar.
+            if index.k1 != self.k1:
+                return None
+            if index.b != self.b:
+                return None
+            self._generation = generation
+            return index
+        except Exception:  # noqa: BLE001 — any bad sidecar => rebuild
+            return None
+
+    def _write_sidecar(self, index: BM25Index, generation: int) -> None:
+        """Atomically persist `index` stamped with `generation`.
+
+        Best-effort: any failure (read-only dir, disk full) is traced
+        to stderr and swallowed — persistence is an optimisation, not
+        a correctness requirement. `os.replace` of a same-directory
+        temp file keeps concurrent readers safe: they see either the
+        old blob or the new one, never a torn write.
+        """
+        path = sidecar_path_for(self.store)
+        if path is None:
+            return
+        try:
+            scope = self.store.local_scope_id.encode("utf-8")
+            buf = io.BytesIO()
+            buf.write(_SIDECAR_MAGIC)
+            buf.write(np.uint32(_SIDECAR_VERSION).tobytes())
+            buf.write(np.uint64(generation).tobytes())
+            buf.write(np.uint32(len(scope)).tobytes())
+            buf.write(scope)
+            buf.write(index.serialize())
+            fd, tmp_name = tempfile.mkstemp(
+                prefix=path.name + ".", dir=str(path.parent),
+            )
+            try:
+                with os.fdopen(fd, "wb") as f:
+                    f.write(buf.getvalue())
+                os.replace(tmp_name, str(path))
+            except BaseException:
+                try:
+                    os.unlink(tmp_name)
+                except OSError:
+                    pass
+                raise
+        except Exception as exc:  # noqa: BLE001 — persistence is optional
+            print(
+                f"aelfrice bm25: sidecar write failed (non-fatal): {exc}",
+                file=sys.stderr,
+            )
 
 
 __all__ = [

src/aelfrice/deferred_feedback.py

@@ -226,13 +226,16 @@ def enqueue_retrieval_exposures(
         return 0
     ts = now if now is not None else _utc_now_iso()
     n = 0
-    for bid in belief_ids:
-        store.enqueue_deferred_feedback(
-            bid,
-            event_type=EVENT_RETRIEVAL_EXPOSURE,
-            enqueued_at=ts,
-        )
-        n += 1
+    # #1135: one commit for the batch instead of one per row — this
+    # runs inside every retrieve() with N surfaced beliefs.
+    with store.transaction():
+        for bid in belief_ids:
+            store.enqueue_deferred_feedback(
+                bid,
+                event_type=EVENT_RETRIEVAL_EXPOSURE,
+                enqueued_at=ts,
+            )
+            n += 1
     return n
 
 

src/aelfrice/derivation_worker.py

@@ -41,7 +41,7 @@
 from __future__ import annotations
 
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Final
 
 from aelfrice.derivation import DerivationInput, RouteOverrides, derive
@@ -130,13 +130,24 @@ class WorkerResult:
         (orphan recovery + new derivations both contribute).
     `rows_skipped_no_belief` counts rows where `derive()` returned no
         belief (classifier marked persist=False).
+    `outcomes` (#1135) maps every log row id stamped by THIS invocation
+        to `(belief_id, was_inserted)` — `belief_id` is None when
+        `derive()` produced no belief, `was_inserted` is True only for
+        brand-new canonical rows (False for corroborations and no-belief
+        stamps). Lets callers that just wrote log rows learn their
+        per-row fate without re-reading the log or snapshotting the
+        belief-id set — the pre-#1135 `set(list_belief_ids())` diff in
+        `_ingest_turn_ids` was a full-table scan per turn.
     """
 
     rows_scanned: int = 0
     beliefs_inserted: int = 0
     beliefs_corroborated: int = 0
     rows_stamped: int = 0
     rows_skipped_no_belief: int = 0
+    outcomes: dict[str, tuple[str | None, bool]] = field(
+        default_factory=dict,
+    )
 
 
 _TRANSCRIPT_CALL_SITE: str = CORROBORATION_SOURCE_TRANSCRIPT_INGEST
@@ -258,12 +269,14 @@ def _process_row(
         # Stamp the row with an explicit empty list so a subsequent
         # worker pass treats it as covered (vs ambiguous NULL = unstamped).
         store.update_ingest_derived_ids(log_id, derived_belief_ids=[])
+        acc.outcomes[log_id] = (None, False)
         return WorkerResult(
             rows_scanned=rows_scanned,
             beliefs_inserted=acc.beliefs_inserted,
             beliefs_corroborated=acc.beliefs_corroborated,
             rows_stamped=acc.rows_stamped + 1,
             rows_skipped_no_belief=acc.rows_skipped_no_belief + 1,
+            outcomes=acc.outcomes,
         )
 
     corroboration_source = _resolve_corroboration_source(row)
@@ -322,6 +335,7 @@ def _process_row(
         derived_edge_ids=derived_edge_ids if derived_edge_ids else None,
     )
 
+    acc.outcomes[log_id] = (actual_id, was_inserted)
     return WorkerResult(
         rows_scanned=rows_scanned,
         beliefs_inserted=acc.beliefs_inserted + (1 if was_inserted else 0),
@@ -330,4 +344,5 @@ def _process_row(
         ),
         rows_stamped=acc.rows_stamped + 1,
         rows_skipped_no_belief=acc.rows_skipped_no_belief,
+        outcomes=acc.outcomes,
     )