From 5a6b926431ccc8a464c1b8641beb370d1d66497d Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:55:13 -0700 Subject: [PATCH 01/14] perf(store): pair WAL with PRAGMA synchronous=NORMAL (#1135) WAL was set at open but synchronous stayed at the FULL default, paying a full fsync per commit. NORMAL is the documented WAL pairing: fsync at checkpoint, not per commit; app crashes lose nothing. Measured ~2x cheaper per-commit on the ingest write path. --- src/aelfrice/store.py | 7 +++++++ tests/test_worktree_concurrency.py | 12 ++++++++++++ 2 files changed, 19 insertions(+) diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index e16e3e707..2893c6c3e 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -1019,6 +1019,13 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # WAL only meaningful on-disk; harmless on :memory:. try: self._conn.execute("PRAGMA journal_mode=WAL") + # The documented WAL pairing: fsync on checkpoint, not on + # every commit. Durability window is the WAL — an app crash + # loses nothing; an OS crash can lose the tail of the WAL, + # which for a memory store is re-derivable (ingest_log is + # append-only and re-ingest is idempotent). Measured ~2x + # cheaper per commit than the FULL default (#1135). + self._conn.execute("PRAGMA synchronous=NORMAL") except sqlite3.DatabaseError: pass # Block up to 5s waiting for a write lock instead of failing diff --git a/tests/test_worktree_concurrency.py b/tests/test_worktree_concurrency.py index 64aa726e7..e04c0b7b6 100644 --- a/tests/test_worktree_concurrency.py +++ b/tests/test_worktree_concurrency.py @@ -119,6 +119,18 @@ def test_wal_mode_is_on(tmp_path: Path) -> None: assert mode == "wal" +def test_synchronous_normal_is_set(tmp_path: Path) -> None: + """A fresh on-disk store pairs WAL with synchronous=NORMAL (#1135).""" + db = tmp_path / "sync.db" + store = MemoryStore(str(db)) + try: + cur = store._conn.execute("PRAGMA synchronous") # type: ignore[attr-defined] + level = cur.fetchone()[0] + finally: + store.close() + assert level == 1, f"expected synchronous=NORMAL (1), got {level}" + + def test_busy_timeout_is_set(tmp_path: Path) -> None: """busy_timeout is non-zero (must wait for a write lock, not fail).""" db = tmp_path / "bt.db" From 434e8b71fbcd641631a7961ca7b0e9867422430d Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:56:39 -0700 Subject: [PATCH 02/14] perf(store): gate the v1.2 origin backfill behind a schema_meta marker (#1135) _BACKFILL_STATEMENTS (two full-table UPDATEs) ran unguarded on every store open, unlike every other one-shot pass. Contemporary writers set origin explicitly, so the flip only matters once per legacy DB. Marker follows the entity-backfill convention; the pass rides the existing open commit. --- src/aelfrice/store.py | 28 ++++++++++++++++-- tests/test_v1_to_v1x_migration.py | 48 +++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 2893c6c3e..1ba76d585 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -594,6 +594,14 @@ def _check_insert_belief_authority() -> None: # re-extract entities for every existing belief on first open. SCHEMA_META_ENTITY_BACKFILL: Final[str] = "entity_backfill_complete" +# #1135: marker for the v1.2 origin backfill (_BACKFILL_STATEMENTS). +# The two UPDATEs ran unguarded on every open — two full-table write +# statements per open, compounding with the hook's multi-open pattern. +# Contemporary writers set origin explicitly (derive() routes, +# cli lock upgrade), so the flip only ever matters once per legacy DB. +# ISO timestamp on completion; absence triggers the pass on next open. +SCHEMA_META_ORIGIN_BACKFILL: Final[str] = "origin_backfill_complete" + # v1.5.0 #204 federation forward-compat. Stable per-DB scope id, # generated on first v1.5+ open and persisted in `schema_meta`. # Today aelfrice is single-scope per DB; the local scope id is @@ -1055,8 +1063,24 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: raise for stmt in _POST_MIGRATION_INDEXES: self._conn.execute(stmt) - for stmt in _BACKFILL_STATEMENTS: - self._conn.execute(stmt) + # #1135: one-shot. Ran unguarded on every open pre-v4.2; the + # marker matches the other schema_meta-gated passes. Rides the + # single open commit below. + marker = self._conn.execute( + "SELECT value FROM schema_meta WHERE key = ?", + (SCHEMA_META_ORIGIN_BACKFILL,), + ).fetchone() + if marker is None: + for stmt in _BACKFILL_STATEMENTS: + self._conn.execute(stmt) + self._conn.execute( + "INSERT OR REPLACE INTO schema_meta (key, value) " + "VALUES (?, ?)", + ( + SCHEMA_META_ORIGIN_BACKFILL, + datetime.now(timezone.utc).isoformat(), + ), + ) self._conn.commit() self._invalidation_callbacks: list[Callable[[], None]] = [] # v1.5.0 #204 federation forward-compat. Resolve (or diff --git a/tests/test_v1_to_v1x_migration.py b/tests/test_v1_to_v1x_migration.py index 98e526fb5..9ee0e5c20 100644 --- a/tests/test_v1_to_v1x_migration.py +++ b/tests/test_v1_to_v1x_migration.py @@ -118,6 +118,54 @@ def test_v1_0_store_accepts_new_writes_after_migration(tmp_path: Path) -> None: store.close() +def test_origin_backfill_runs_once_behind_marker(tmp_path: Path) -> None: + """#1135: the v1.2 origin backfill is schema_meta-gated one-shot. + + First open flips legacy (origin=unknown, lock_level=user) rows to + user_stated and stamps the marker. Rows crafted after that open are + NOT flipped by later opens — the two UPDATE statements no longer run + on every open. + """ + db = tmp_path / "v1_0.db" + _seed_v1_0_store(db) + # Give legacy1 a user lock so the backfill has an eligible row. + raw = sqlite3.connect(str(db)) + raw.execute( + "UPDATE beliefs SET lock_level = 'user', " + "locked_at = '2025-01-03T00:00:00+00:00' WHERE id = 'legacy1'" + ) + raw.commit() + raw.close() + + s1 = MemoryStore(str(db)) + try: + b = s1.get_belief("legacy1") + assert b is not None + assert b.origin == "user_stated" + from aelfrice.store import SCHEMA_META_ORIGIN_BACKFILL + assert s1.get_schema_meta(SCHEMA_META_ORIGIN_BACKFILL) is not None + finally: + s1.close() + + # Craft a post-marker eligible row directly; the gated pass must + # leave it alone on the next open. + raw = sqlite3.connect(str(db)) + raw.execute( + "UPDATE beliefs SET origin = 'unknown' WHERE id = 'legacy1'" + ) + raw.commit() + raw.close() + s2 = MemoryStore(str(db)) + try: + b = s2.get_belief("legacy1") + assert b is not None + assert b.origin == "unknown", ( + "origin backfill re-ran despite completion marker" + ) + finally: + s2.close() + + def test_migration_idempotent_on_re_open(tmp_path: Path) -> None: db = tmp_path / "v1_0.db" _seed_v1_0_store(db) From a38e1940515973a94044cdb3fb1889ff62297390 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 10:57:51 -0700 Subject: [PATCH 03/14] perf(store): add hot-path indexes + named columns on the unstamped scan (#1135) Three query shapes ran unindexed on every hot-path invocation: - derivation worker's unstamped ingest_log scan (O(total log) per turn on a monotonically growing table) -> partial index on the NULL stamp - list_locked_beliefs L0 tier (up to 3x per retrieve, full-table scan) -> partial index matching the ORDER BY - has_edge_type() probe (per non-empty query since the #1064 lane flip; miss case scanned all edges) -> index on edges(type) The unstamped scan also names its columns instead of SELECT *. --- src/aelfrice/store.py | 20 +++++++- tests/test_store_hot_path_indexes.py | 74 ++++++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 tests/test_store_hot_path_indexes.py diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 1ba76d585..a7e52068a 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -757,6 +757,21 @@ def _check_insert_belief_authority() -> None: # branch of the hook filter. "CREATE INDEX IF NOT EXISTS idx_beliefs_project_context " "ON beliefs(project_context)", + # #1135: partial index for the derivation worker's unstamped scan. + # ingest_log grows monotonically (one row per sentence, never + # deleted) while the unstamped set stays tiny, so without this the + # per-turn `WHERE derived_belief_ids IS NULL` scan is O(total log). + "CREATE INDEX IF NOT EXISTS idx_ingest_log_unstamped " + "ON ingest_log(id) WHERE derived_belief_ids IS NULL", + # #1135: partial index for list_locked_beliefs — the L0 tier runs + # up to three times per retrieve and was a full-table scan. Column + # order matches the query's ORDER BY (locked_at DESC, id ASC). + "CREATE INDEX IF NOT EXISTS idx_beliefs_locked " + "ON beliefs(locked_at DESC, id) WHERE lock_level != 'none'", + # #1135: has_edge_type() probes fire per non-empty query since the + # #1064 temporal-spine lane flip; without an index on type the + # miss case scans the whole edges table. + "CREATE INDEX IF NOT EXISTS idx_edges_type ON edges(type)", ) # One-shot backfill for v1.0/v1.1 stores opening on v1.2+. Each row @@ -3682,7 +3697,10 @@ def list_unstamped_ingest_log(self) -> list[dict[str, object]]: same shape as `get_ingest_log_entry`. """ cur = self._conn.execute( - "SELECT * FROM ingest_log " + "SELECT id, ts, source_kind, source_path, raw_text, raw_meta, " + " derived_belief_ids, derived_edge_ids, " + " classifier_version, rule_set_hash, session_id " + "FROM ingest_log " "WHERE derived_belief_ids IS NULL " "ORDER BY id" ) diff --git a/tests/test_store_hot_path_indexes.py b/tests/test_store_hot_path_indexes.py new file mode 100644 index 000000000..cfc7243e8 --- /dev/null +++ b/tests/test_store_hot_path_indexes.py @@ -0,0 +1,74 @@ +"""#1135 hot-path indexes: unstamped ingest_log scan, locked-belief +tier, edge-type probe. + +Asserts both existence (sqlite_master) and that the planner actually +uses each index for the exact query shape the hot path issues — an +index that exists but is not chosen is a silent regression. +""" +from __future__ import annotations + +from aelfrice.store import MemoryStore + + +def _index_names(store: MemoryStore) -> set[str]: + cur = store._conn.execute( # type: ignore[attr-defined] + "SELECT name FROM sqlite_master WHERE type = 'index'" + ) + return {str(r["name"]) for r in cur.fetchall()} + + +def _plan(store: MemoryStore, sql: str) -> str: + cur = store._conn.execute( # type: ignore[attr-defined] + "EXPLAIN QUERY PLAN " + sql + ) + return " | ".join(str(r["detail"]) for r in cur.fetchall()) + + +def test_hot_path_indexes_exist() -> None: + store = MemoryStore(":memory:") + try: + names = _index_names(store) + assert "idx_ingest_log_unstamped" in names + assert "idx_beliefs_locked" in names + assert "idx_edges_type" in names + finally: + store.close() + + +def test_unstamped_scan_uses_partial_index() -> None: + store = MemoryStore(":memory:") + try: + plan = _plan( + store, + "SELECT id FROM ingest_log " + "WHERE derived_belief_ids IS NULL ORDER BY id", + ) + assert "idx_ingest_log_unstamped" in plan, plan + finally: + store.close() + + +def test_locked_beliefs_query_uses_partial_index() -> None: + store = MemoryStore(":memory:") + try: + plan = _plan( + store, + "SELECT * FROM beliefs b " + "WHERE b.lock_level != 'none' AND b.valid_to IS NULL " + "ORDER BY b.locked_at DESC, b.id ASC", + ) + assert "idx_beliefs_locked" in plan, plan + finally: + store.close() + + +def test_edge_type_probe_uses_index() -> None: + store = MemoryStore(":memory:") + try: + plan = _plan( + store, + "SELECT 1 FROM edges WHERE type = 'TEMPORAL_NEXT' LIMIT 1", + ) + assert "idx_edges_type" in plan, plan + finally: + store.close() From 12f5bc189ad364775f7470e79dd43d4ac93f48ec Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:00:21 -0700 Subject: [PATCH 04/14] perf(ingest): worker reports per-row outcomes; drop per-turn belief-set snapshot (#1135) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _ingest_turn_ids snapshotted set(store.list_belief_ids()) every turn (full-table scan, linear in corpus size) solely to distinguish new inserts from corroborations — a fact insert_or_corroborate already returns. WorkerResult now carries outcomes: {log_id: (belief_id, was_inserted)}; ingest reads its rows' fate from there, which also removes the per-sentence get_ingest_log_entry re-read. Fallback to the log row covers the sibling-process race (a row stamped elsewhere is not a brand-new insert of ours). Bulk-ingest per-turn cost is now flat w.r.t. corpus size. --- src/aelfrice/derivation_worker.py | 17 +++++++++- src/aelfrice/ingest.py | 52 +++++++++++++++++++------------ tests/test_derivation_worker.py | 24 ++++++++++++++ 3 files changed, 72 insertions(+), 21 deletions(-) diff --git a/src/aelfrice/derivation_worker.py b/src/aelfrice/derivation_worker.py index 69cc1768f..9965f95ac 100644 --- a/src/aelfrice/derivation_worker.py +++ b/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,6 +130,14 @@ 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 @@ -137,6 +145,9 @@ class WorkerResult: 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, ) diff --git a/src/aelfrice/ingest.py b/src/aelfrice/ingest.py index a0b018ed8..db43a4938 100644 --- a/src/aelfrice/ingest.py +++ b/src/aelfrice/ingest.py @@ -218,11 +218,6 @@ def _ingest_turn_ids( return [] ts = created_at or _now_utc_iso() - # Snapshot the canonical belief set so we can identify which derived - # ids in this turn correspond to brand-new inserts (vs corroborations - # of already-known beliefs). Preserves the pre-#264 public contract - # that `ingest_turn` returns the count of newly-inserted beliefs. - ids_before: set[str] = set(store.list_belief_ids()) # #888: when the caller supplies a `role` (typically 'user' from # ingest_jsonl after the #785 assistant-row gate), stamp it on the @@ -250,26 +245,43 @@ def _ingest_turn_ids( # Worker is idempotent and scans all unstamped rows; calling it once # at end-of-turn is the per-batch invocation pattern from the spec. - run_worker(store) - - # Resolve each log_id to its canonical belief id once, in input - # order. Used twice: (a) for the public return value (newly - # inserted beliefs, deduped), (b) for the #809 intra-turn edge - # wiring below (per-sentence belief id, position-preserving). + worker_result = run_worker(store) + + # Resolve each log_id to its per-row fate, in input order. #1135: + # the worker reports `(belief_id, was_inserted)` per stamped row, + # replacing the pre/post `set(list_belief_ids())` diff (a full-table + # scan per turn) and the per-log-id re-read. `was_inserted` is True + # only for brand-new canonical rows, which preserves the pre-#264 + # public contract that `ingest_turn` returns the count of + # newly-inserted beliefs. Used twice: (a) for the public return + # value (newly inserted beliefs, deduped), (b) for the #809 + # intra-turn edge wiring below (per-sentence belief id, + # position-preserving). log_belief_ids: list[str | None] = [] inserted: list[str] = [] seen: set[str] = set() for log_id in log_ids: - entry = store.get_ingest_log_entry(log_id) - bid: str | None = None - if entry is not None: - ids = entry.get("derived_belief_ids") or [] - if isinstance(ids, list) and ids: - head = ids[0] - if isinstance(head, str): - bid = head + outcome = worker_result.outcomes.get(log_id) + bid: str | None + was_inserted: bool + if outcome is not None: + bid, was_inserted = outcome + else: + # A sibling process stamped this row in the window between + # record_ingest and our run_worker pass. Fall back to the + # log row; a row stamped elsewhere is by definition not a + # brand-new insert of ours. + was_inserted = False + bid = None + entry = store.get_ingest_log_entry(log_id) + if entry is not None: + ids = entry.get("derived_belief_ids") or [] + if isinstance(ids, list) and ids: + head = ids[0] + if isinstance(head, str): + bid = head log_belief_ids.append(bid) - if bid is not None and bid not in ids_before and bid not in seen: + if bid is not None and was_inserted and bid not in seen: seen.add(bid) inserted.append(bid) diff --git a/tests/test_derivation_worker.py b/tests/test_derivation_worker.py index f7ae22dd7..9e0e79048 100644 --- a/tests/test_derivation_worker.py +++ b/tests/test_derivation_worker.py @@ -182,3 +182,27 @@ def test_worker_processes_multiple_rows_in_one_pass( row = store.get_ingest_log_entry(lid) assert row is not None assert isinstance(row["derived_belief_ids"], list) + + +def test_worker_outcomes_report_per_row_fate(store: MemoryStore) -> None: + """Hypothesis (#1135): `WorkerResult.outcomes` maps every stamped + log id to (belief_id, was_inserted) — True for a brand-new insert, + False for a corroboration of the same content, (None, False) for a + no-belief stamp. Falsifiable by a missing log id, a wrong belief + id, or was_inserted=True on the corroboration row.""" + new_id = _record_unstamped(store, "The scheduler retries three times.") + dup_id = _record_unstamped(store, "The scheduler retries three times.") + + result = run_worker(store) + + assert set(result.outcomes) == {new_id, dup_id} + new_bid, new_was_inserted = result.outcomes[new_id] + dup_bid, dup_was_inserted = result.outcomes[dup_id] + assert new_was_inserted is True + assert dup_was_inserted is False + assert new_bid == dup_bid # same content -> same canonical belief + assert isinstance(new_bid, str) + assert store.get_belief(new_bid) is not None + + # A second pass sees no unstamped rows -> empty outcomes. + assert run_worker(store).outcomes == {} From 2c9a4c231cf03aecb046e1b6ddef4e83edd4e356 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:02:37 -0700 Subject: [PATCH 05/14] perf(store): transaction() context manager for write-group batching (#1135) Every mutating method committed per call (~8 commits per ingested turn, ~45-60 per hook prompt with hits). transaction() suppresses the per-call commits via a depth counter and issues one commit at outermost exit; measured 33x cheaper than commit-per-row for a 200-insert group. Invalidation callbacks defer to one post-commit fire so derived caches never read uncommitted state; exceptions roll the whole group back. No call sites wired yet - behavior outside transaction() is byte-identical. --- src/aelfrice/store.py | 135 ++++++++++++++++++++---------- tests/test_store_transaction.py | 140 ++++++++++++++++++++++++++++++++ 2 files changed, 233 insertions(+), 42 deletions(-) create mode 100644 tests/test_store_transaction.py diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index a7e52068a..1e51f89da 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -17,6 +17,7 @@ import re import secrets import sqlite3 +from contextlib import contextmanager from datetime import datetime, timezone from typing import TYPE_CHECKING, Callable, Final, Iterable, Iterator @@ -1037,6 +1038,13 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # without a repo identity keep the pre-#970 cross-context behaviour. # db_paths._open_store() injects the value derived from the DB path. self._project_context_default: str = project_context_default + # #1135 write-group batching. Non-zero depth suppresses the + # per-call commits issued by mutation methods (via `_commit`); + # `transaction()` commits once at outermost exit. Must exist + # before the schema battery below — the one-shot migration + # helpers commit through `_commit` too. + self._txn_depth: int = 0 + self._pending_invalidation: bool = False self._conn: sqlite3.Connection = sqlite3.connect(path) self._conn.row_factory = sqlite3.Row # WAL only meaningful on-disk; harmless on :memory:. @@ -1096,7 +1104,7 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: datetime.now(timezone.utc).isoformat(), ), ) - self._conn.commit() + self._commit() self._invalidation_callbacks: list[Callable[[], None]] = [] # v1.5.0 #204 federation forward-compat. Resolve (or # generate) the local scope id BEFORE any belief/edge @@ -1181,9 +1189,52 @@ def add_invalidation_callback(self, fn: Callable[[], None]) -> None: self._invalidation_callbacks.append(fn) def _fire_invalidation(self) -> None: + # Inside a transaction() the mutation is not yet visible to + # readers; defer to one post-commit fire so callbacks never + # observe uncommitted state (#1135). + if self._txn_depth > 0: + self._pending_invalidation = True + return for fn in self._invalidation_callbacks: fn() + # --- Write-group batching (#1135) ------------------------------------ + + def _commit(self) -> None: + """Commit unless inside `transaction()` (outermost commit wins).""" + if self._txn_depth == 0: + self._conn.commit() + + @contextmanager + def transaction(self) -> Iterator[None]: + """Group multiple mutating calls into one SQLite transaction. + + Per-call commits inside the block are suppressed; one commit is + issued at outermost exit. Measured 33x cheaper than + commit-per-row for a 200-insert group (#1135). Invalidation + callbacks fire once, after that commit, so derived caches never + read uncommitted state. Reentrant: nested blocks join the + outermost transaction. On exception the outermost block rolls + the whole group back and re-raises; a caller that swallows an + inner block's exception keeps the outer group's prior writes + (the rollback happens only at depth zero). + """ + self._txn_depth += 1 + try: + yield + except BaseException: + self._txn_depth -= 1 + if self._txn_depth == 0: + self._conn.rollback() + self._pending_invalidation = False + raise + self._txn_depth -= 1 + if self._txn_depth == 0: + self._conn.commit() + if self._pending_invalidation: + self._pending_invalidation = False + self._fire_invalidation() + # --- Schema meta (v1.3+ migration tracker) --------------------------- def get_schema_meta(self, key: str) -> str | None: @@ -1204,7 +1255,7 @@ def set_schema_meta(self, key: str, value: str) -> None: "INSERT OR REPLACE INTO schema_meta (key, value) VALUES (?, ?)", (key, value), ) - self._conn.commit() + self._commit() # --- v1.5.0 #204 version-vector helpers ------------------------------ @@ -1333,7 +1384,7 @@ def _maybe_backfill_version_vectors(self) -> int: (scope,), ) edge_inserted = edge_cur.rowcount or 0 - self._conn.commit() + self._commit() self.set_schema_meta( SCHEMA_META_VERSION_VECTOR_BACKFILL, datetime.now(timezone.utc).isoformat(), @@ -1367,7 +1418,7 @@ def _maybe_backfill_project_context(self) -> int: (self._project_context_default, BELIEF_SCOPE_PROJECT, LOCK_USER), ) stamped = cur.rowcount or 0 - self._conn.commit() + self._commit() self.set_schema_meta( SCHEMA_META_PROJECT_CONTEXT_BACKFILL, datetime.now(timezone.utc).isoformat(), @@ -1392,7 +1443,7 @@ def _maybe_backfill_log_version_vectors(self) -> int: (scope,), ) inserted = cur.rowcount or 0 - self._conn.commit() + self._commit() self.set_schema_meta( SCHEMA_META_LOG_VERSION_VECTOR_BACKFILL, datetime.now(timezone.utc).isoformat(), @@ -1470,7 +1521,7 @@ def _maybe_synthesize_legacy_log_rows(self) -> int: ) self._bump_log_version(log_id) inserted += 1 - self._conn.commit() + self._commit() self.set_schema_meta( SCHEMA_META_LEGACY_LOG_SYNTH, datetime.now(timezone.utc).isoformat(), @@ -2111,7 +2162,7 @@ def _maybe_rehash_speculative_v2(self) -> int: ) rewritten += 1 - self._conn.commit() + self._commit() self.set_schema_meta( SCHEMA_META_SPECULATIVE_HASH_V2_COMPLETE, datetime.now(timezone.utc).isoformat(), @@ -2229,7 +2280,7 @@ def insert_belief(self, b: Belief) -> None: ) self._write_belief_entities(b.id, b.content) self._bump_belief_version(b.id) - self._conn.commit() + self._commit() self._fire_invalidation() def get_belief_by_content_hash(self, content_hash: str) -> Belief | None: @@ -2329,7 +2380,7 @@ def update_belief(self, b: Belief) -> None: ) self._write_belief_entities(b.id, b.content) self._bump_belief_version(b.id) - self._conn.commit() + self._commit() self._fire_invalidation() def delete_belief(self, belief_id: str) -> None: @@ -2342,7 +2393,7 @@ def delete_belief(self, belief_id: str) -> None: self._conn.execute( "DELETE FROM belief_entities WHERE belief_id = ?", (belief_id,) ) - self._conn.commit() + self._commit() self._fire_invalidation() def soft_delete_belief(self, belief_id: str, ts: str | None = None) -> None: @@ -2367,7 +2418,7 @@ def soft_delete_belief(self, belief_id: str, ts: str | None = None) -> None: "DELETE FROM beliefs_fts WHERE id = ?", (belief_id,) ) self._bump_belief_version(belief_id) - self._conn.commit() + self._commit() self._fire_invalidation() def restore_belief(self, belief_id: str) -> bool: @@ -2406,7 +2457,7 @@ def restore_belief(self, belief_id: str) -> bool: (belief_id, row["content"]), ) self._bump_belief_version(belief_id) - self._conn.commit() + self._commit() self._fire_invalidation() return True @@ -2423,7 +2474,7 @@ def update_last_confirmed_at(self, belief_id: str, ts_iso: str) -> None: (ts_iso, belief_id), ) self._bump_belief_version(belief_id) - self._conn.commit() + self._commit() self._fire_invalidation() def list_review_candidates(self, *, limit: int = 10) -> list[Belief]: @@ -3057,7 +3108,7 @@ def stamp_retrieved( f"UPDATE beliefs SET last_retrieved_at = ? WHERE id IN ({placeholders})", (ts, *ids), ) - self._conn.commit() + self._commit() return cur.rowcount or 0 # --- Feedback history ------------------------------------------------ @@ -3080,7 +3131,7 @@ def insert_feedback_event( """, (belief_id, valence, source, created_at), ) - self._conn.commit() + self._commit() rowid = cur.lastrowid if rowid is None: raise RuntimeError("feedback_history insert returned no rowid") @@ -3165,7 +3216,7 @@ def delete_orphan_feedback_events(self) -> int: ) """ ) - self._conn.commit() + self._commit() return cur.rowcount if cur.rowcount is not None else 0 # --- Belief corroborations (v1.5+, #190) ----------------------------- @@ -3262,7 +3313,7 @@ def record_corroboration( """, (belief_id, ts, source_type, session_id, source_path_hash), ) - self._conn.commit() + self._commit() def count_corroborations(self, belief_id: str) -> int: """Return the count of belief_corroborations rows for one belief. @@ -3346,7 +3397,7 @@ def link_belief_to_document( """, (belief_id, doc_uri, anchor_type, position_hint, ts), ) - self._conn.commit() + self._commit() cur = self._conn.execute( """ SELECT belief_id, doc_uri, anchor_type, position_hint, created_at @@ -3459,7 +3510,7 @@ def upsert_category( """, (name, 1 if always_on else 0, trigger_json, default_lock, now), ) - self._conn.commit() + self._commit() got = self.get_category(name) assert got is not None # just wrote it return got @@ -3489,7 +3540,7 @@ def set_category_trigger(self, name: str, trigger_json: str) -> bool: "UPDATE categories SET trigger_json = ? WHERE name = ?", (trigger_json, name), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def delete_category(self, name: str) -> bool: @@ -3498,7 +3549,7 @@ def delete_category(self, name: str) -> bool: cur = self._conn.execute( "DELETE FROM categories WHERE name = ?", (name,) ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def assign_belief_to_category( @@ -3520,7 +3571,7 @@ def assign_belief_to_category( "(belief_id, category_name, created_at) VALUES (?, ?, ?)", (belief_id, category_name, now), ) - self._conn.commit() + self._commit() # `INSERT OR IGNORE` silently drops FK violations, so if the belief # or category was deleted between the checks above and this write, # the row never landed. Confirm membership and surface the race @@ -3546,7 +3597,7 @@ def unassign_belief_from_category( "WHERE belief_id = ? AND category_name = ?", (belief_id, category_name), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def get_categories_for_belief(self, belief_id: str) -> list[str]: @@ -3632,7 +3683,7 @@ def record_ingest( ), ) self._bump_log_version(log_id) - self._conn.commit() + self._commit() return log_id def update_ingest_derived_ids( @@ -3663,7 +3714,7 @@ def update_ingest_derived_ids( f"UPDATE ingest_log SET {', '.join(sets)} WHERE id = ?", params, ) - self._conn.commit() + self._commit() def get_ingest_log_entry(self, log_id: str) -> dict[str, object] | None: """Return a dict view of one ingest_log row, or None if missing. @@ -3748,7 +3799,7 @@ def enqueue_deferred_feedback( """, (belief_id, enqueued_at, event_type), ) - self._conn.commit() + self._commit() rowid = cur.lastrowid if rowid is None: raise RuntimeError( @@ -3836,7 +3887,7 @@ def record_injection_event( (session_id, turn_id, belief_id, injected_at, source, encoded_consumers), ) - self._conn.commit() + self._commit() rowid = cur.lastrowid if rowid is None: raise RuntimeError( @@ -3926,7 +3977,7 @@ def update_injection_referenced( """, (referenced, referenced_at, event_id), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 # ----- #816 belief_touches (hot-path v1) ------------------------- @@ -3997,7 +4048,7 @@ def record_touch( """, (belief_id, session_id, fire_idx, event_kind), ) - self._conn.commit() + self._commit() def read_touch_set_in_window( self, @@ -4404,7 +4455,7 @@ def set_retention_class(self, belief_id: str, retention_class: str) -> None: (retention_class, belief_id), ) self._bump_belief_version(belief_id) - self._conn.commit() + self._commit() self._fire_invalidation() def count_beliefs_by_type(self) -> dict[str, int]: @@ -4628,7 +4679,7 @@ class is unknown. (key, float(static_default), int(half_life_seconds), int(now_ts), encoded), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def update_meta_belief( @@ -4702,7 +4753,7 @@ def update_meta_belief( "UPDATE meta_beliefs SET last_updated_ts = ? WHERE key = ?", (int(now_ts), key), ) - self._conn.commit() + self._commit() return True def read_meta_belief_state(self, key: str) -> MetaBeliefState | None: @@ -4790,7 +4841,7 @@ def insert_edge(self, e: Edge) -> None: (e.src, e.dst, e.type, e.weight, e.anchor_text), ) self._bump_edge_version(e.src, e.dst, e.type) - self._conn.commit() + self._commit() self._fire_invalidation() def get_edge(self, src: str, dst: str, type_: str) -> Edge | None: @@ -4808,7 +4859,7 @@ def update_edge(self, e: Edge) -> None: (e.weight, e.anchor_text, e.src, e.dst, e.type), ) self._bump_edge_version(e.src, e.dst, e.type) - self._conn.commit() + self._commit() self._fire_invalidation() def delete_edge(self, src: str, dst: str, type_: str) -> None: @@ -4816,7 +4867,7 @@ def delete_edge(self, src: str, dst: str, type_: str) -> None: "DELETE FROM edges WHERE src = ? AND dst = ? AND type = ?", (src, dst, type_), ) - self._conn.commit() + self._commit() self._fire_invalidation() def delete_edges_by_type(self, type_: str) -> int: @@ -4831,7 +4882,7 @@ def delete_edges_by_type(self, type_: str) -> int: "DELETE FROM edges WHERE type = ?", (type_,) ) removed = cur.rowcount - self._conn.commit() + self._commit() self._fire_invalidation() return removed @@ -5197,7 +5248,7 @@ def insert_onboard_session(self, s: OnboardSession) -> None: s.created_at, s.completed_at, ), ) - self._conn.commit() + self._commit() def get_onboard_session(self, session_id: str) -> OnboardSession | None: cur = self._conn.execute( @@ -5226,7 +5277,7 @@ def complete_onboard_session( """, (ONBOARD_STATE_COMPLETED, completed_at, session_id), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def count_onboard_sessions(self, state: str | None = None) -> int: @@ -5258,7 +5309,7 @@ def insert_onboard_rejection( """, (belief_id, text, source, rejected_at), ) - self._conn.commit() + self._commit() def delete_onboard_rejection(self, belief_id: str) -> bool: """Drop a rejection ledger entry. Returns True if a row was @@ -5270,7 +5321,7 @@ def delete_onboard_rejection(self, belief_id: str) -> bool: "DELETE FROM onboard_rejections WHERE belief_id = ?", (belief_id,), ) - self._conn.commit() + self._commit() return cur.rowcount > 0 def is_onboard_rejected(self, belief_id: str) -> bool: @@ -5341,7 +5392,7 @@ def create_session( """, (s.id, s.started_at, s.completed_at, s.model, s.project_context), ) - self._conn.commit() + self._commit() return s def complete_session(self, session_id: str) -> None: @@ -5356,7 +5407,7 @@ def complete_session(self, session_id: str) -> None: "UPDATE sessions SET completed_at = ? WHERE id = ?", (datetime.now(timezone.utc).isoformat(), session_id), ) - self._conn.commit() + self._commit() def get_session(self, session_id: str) -> Session | None: cur = self._conn.execute( diff --git a/tests/test_store_transaction.py b/tests/test_store_transaction.py new file mode 100644 index 000000000..6c1af513e --- /dev/null +++ b/tests/test_store_transaction.py @@ -0,0 +1,140 @@ +"""#1135 write-group batching: `MemoryStore.transaction()`. + +Contract under test: per-call commits inside the block are suppressed +and issued once at outermost exit; exceptions roll the whole group +back; invalidation callbacks fire once, after the commit; nested +blocks join the outermost transaction. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest + +from aelfrice.models import BELIEF_FACTUAL, LOCK_NONE, Belief +from aelfrice.store import MemoryStore + + +def _mk_belief(bid: str) -> Belief: + return Belief( + id=bid, + content=f"content for {bid}", + content_hash="h_" + bid, + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + created_at="2026-07-21T00:00:00Z", + last_retrieved_at=None, + ) + + +def test_transaction_groups_writes_into_one_commit(tmp_path: Path) -> None: + db = tmp_path / "txn.db" + store = MemoryStore(str(db)) + try: + with store.transaction(): + for i in range(5): + store.insert_belief(_mk_belief(f"t{i}")) + # Mid-transaction: a second connection must not see the + # uncommitted rows (proves the inner commits were + # suppressed, not just batched by chance). + other = MemoryStore(str(db)) + try: + assert other.count_beliefs() == 0 + finally: + other.close() + assert store.count_beliefs() == 5 + # Post-commit: the rows are durable for other connections. + other = MemoryStore(str(db)) + try: + assert other.count_beliefs() == 5 + finally: + other.close() + finally: + store.close() + + +def test_transaction_rolls_back_on_exception(tmp_path: Path) -> None: + store = MemoryStore(str(tmp_path / "txn.db")) + try: + with pytest.raises(RuntimeError): + with store.transaction(): + store.insert_belief(_mk_belief("doomed")) + raise RuntimeError("boom") + assert store.count_beliefs() == 0 + assert store.get_belief("doomed") is None + # The store remains usable after the rollback. + store.insert_belief(_mk_belief("after")) + assert store.count_beliefs() == 1 + finally: + store.close() + + +def test_invalidation_fires_once_after_commit(tmp_path: Path) -> None: + store = MemoryStore(str(tmp_path / "txn.db")) + try: + fired: list[int] = [] + store.add_invalidation_callback( + lambda: fired.append(store.count_beliefs()) + ) + with store.transaction(): + store.insert_belief(_mk_belief("a")) + store.insert_belief(_mk_belief("b")) + assert fired == [] # deferred while the group is open + # One post-commit fire; the callback observed committed state. + assert fired == [2] + finally: + store.close() + + +def test_invalidation_not_fired_after_rollback(tmp_path: Path) -> None: + store = MemoryStore(str(tmp_path / "txn.db")) + try: + fired: list[bool] = [] + store.add_invalidation_callback(lambda: fired.append(True)) + with pytest.raises(RuntimeError): + with store.transaction(): + store.insert_belief(_mk_belief("gone")) + raise RuntimeError("boom") + assert fired == [] + finally: + store.close() + + +def test_nested_transactions_join_outermost(tmp_path: Path) -> None: + db = tmp_path / "txn.db" + store = MemoryStore(str(db)) + try: + with store.transaction(): + store.insert_belief(_mk_belief("outer")) + with store.transaction(): + store.insert_belief(_mk_belief("inner")) + # Inner exit must NOT have committed yet. + other = MemoryStore(str(db)) + try: + assert other.count_beliefs() == 0 + finally: + other.close() + assert store.count_beliefs() == 2 + finally: + store.close() + + +def test_mutations_outside_transaction_commit_per_call( + tmp_path: Path, +) -> None: + """The legacy path is unchanged: no transaction() means each + mutating call commits immediately.""" + db = tmp_path / "txn.db" + store = MemoryStore(str(db)) + try: + store.insert_belief(_mk_belief("solo")) + other = MemoryStore(str(db)) + try: + assert other.count_beliefs() == 1 + finally: + other.close() + finally: + store.close() From c7581422f9ad88e11600c5bcacd119506831ec25 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:12:44 -0700 Subject: [PATCH 06/14] perf(ingest,hook): batch hot-path write groups into single transactions (#1135) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire store.transaction() into the four commit-per-row hot spots: - ingest turn (record_ingest per sentence + worker insert/stamp per row + edge inserts): one write group per turn, which also makes the turn atomic — a crash mid-turn leaves no unstamped orphans - hook record_retrieval (apply_feedback per hit + stamp_retrieved): one group per retrieval; an un-committable group (closed handle) reports 0 rows written, preserving the non-blocking contract - hook injection_events + belief_touches loops: one commit per batch - deferred-feedback exposure enqueue: one commit per batch transaction()'s rollback path no longer masks the original exception when the rollback itself fails on a dead handle. --- src/aelfrice/deferred_feedback.py | 17 ++- src/aelfrice/hook.py | 58 ++++---- src/aelfrice/hook_search.py | 56 +++++--- src/aelfrice/ingest.py | 226 +++++++++++++++--------------- src/aelfrice/store.py | 8 +- 5 files changed, 198 insertions(+), 167 deletions(-) diff --git a/src/aelfrice/deferred_feedback.py b/src/aelfrice/deferred_feedback.py index 5b6c0cc9f..afbae2ec1 100644 --- a/src/aelfrice/deferred_feedback.py +++ b/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 diff --git a/src/aelfrice/hook.py b/src/aelfrice/hook.py index e17c9d855..3202d6531 100644 --- a/src/aelfrice/hook.py +++ b/src/aelfrice/hook.py @@ -1504,21 +1504,23 @@ def _record_touches( store = MemoryStore(str(p)) try: # Current turn's injection set — forward-only, no ring replay. - for bid in belief_ids: - if not bid: - continue - try: - store.record_touch( - belief_id=bid, - session_id=session_id, - fire_idx=fire_idx, - event_kind=TOUCH_EVENT_KIND_INJECTION, - ) - except Exception: - # Same per-row tolerance for the current set: - # extremely unlikely but possible (a belief - # deleted between retrieval and the touch write). - continue + # #1135: one commit for the batch instead of one per touch. + with store.transaction(): + for bid in belief_ids: + if not bid: + continue + try: + store.record_touch( + belief_id=bid, + session_id=session_id, + fire_idx=fire_idx, + event_kind=TOUCH_EVENT_KIND_INJECTION, + ) + except Exception: + # Same per-row tolerance for the current set: + # extremely unlikely but possible (a belief + # deleted between retrieval and the touch write). + continue finally: store.close() except Exception as exc: @@ -1561,18 +1563,20 @@ def _record_injection_events( injected_at = datetime.now(timezone.utc).isoformat() store = MemoryStore(str(p)) try: - for h in hits: - bid = getattr(h, "id", None) - if not bid: - continue - store.record_injection_event( - session_id=session_id, - turn_id=turn_id, - belief_id=bid, - injected_at=injected_at, - source=source, - active_consumers=active_consumers, - ) + # #1135: one commit for the batch instead of one per event. + with store.transaction(): + for h in hits: + bid = getattr(h, "id", None) + if not bid: + continue + store.record_injection_event( + session_id=session_id, + turn_id=turn_id, + belief_id=bid, + injected_at=injected_at, + source=source, + active_consumers=active_consumers, + ) finally: store.close() except Exception as exc: diff --git a/src/aelfrice/hook_search.py b/src/aelfrice/hook_search.py index 7ebb4a0a8..b561f523f 100644 --- a/src/aelfrice/hook_search.py +++ b/src/aelfrice/hook_search.py @@ -123,26 +123,38 @@ def record_retrieval( update_posterior: bool = _exposure_updates_posterior() written: int = 0 stamped_ids: list[str] = [] - for b in beliefs: - try: - apply_feedback( - store, - b.id, - valence, - source, - update_posterior=update_posterior, - ) - written += 1 - stamped_ids.append(b.id) - except Exception: # non-blocking: log and continue - traceback.print_exc(file=serr) - # Mirror the audit row to beliefs.last_retrieved_at so downstream - # consumers (decay moderation, recency-aware ranking, telemetry) get - # an O(1) read instead of having to join feedback_history. Same - # best-effort posture as the loop above. - if stamped_ids: - try: - store.stamp_retrieved(stamped_ids) - except Exception: - traceback.print_exc(file=serr) + # #1135: one write group per retrieval instead of a commit per hit + # (~15 commits per hook prompt pre-batching). Per-row failures are + # still swallowed individually — the surviving rows commit together. + # The outer try preserves the non-blocking contract when the group + # itself cannot commit (e.g. the store handle was closed): nothing + # persisted, so report zero rows written. + try: + with store.transaction(): + for b in beliefs: + try: + apply_feedback( + store, + b.id, + valence, + source, + update_posterior=update_posterior, + ) + written += 1 + stamped_ids.append(b.id) + except Exception: # non-blocking: log and continue + traceback.print_exc(file=serr) + # Mirror the audit row to beliefs.last_retrieved_at so + # downstream consumers (decay moderation, recency-aware + # ranking, telemetry) get an O(1) read instead of having to + # join feedback_history. Same best-effort posture as the + # loop above. + if stamped_ids: + try: + store.stamp_retrieved(stamped_ids) + except Exception: + traceback.print_exc(file=serr) + except Exception: # non-blocking: the whole group rolled back + traceback.print_exc(file=serr) + return 0 return written diff --git a/src/aelfrice/ingest.py b/src/aelfrice/ingest.py index db43a4938..8d3e480cc 100644 --- a/src/aelfrice/ingest.py +++ b/src/aelfrice/ingest.py @@ -231,116 +231,122 @@ def _ingest_turn_ids( if role is not None: raw_meta["role"] = role - log_ids: list[str] = [] - for sentence in full_sentences: - log_id = store.record_ingest( - source_kind=INGEST_SOURCE_TRANSCRIPT, - source_path=source, - raw_text=sentence, - session_id=session_id, - ts=ts, - raw_meta=raw_meta, - ) - log_ids.append(log_id) - - # Worker is idempotent and scans all unstamped rows; calling it once - # at end-of-turn is the per-batch invocation pattern from the spec. - worker_result = run_worker(store) - - # Resolve each log_id to its per-row fate, in input order. #1135: - # the worker reports `(belief_id, was_inserted)` per stamped row, - # replacing the pre/post `set(list_belief_ids())` diff (a full-table - # scan per turn) and the per-log-id re-read. `was_inserted` is True - # only for brand-new canonical rows, which preserves the pre-#264 - # public contract that `ingest_turn` returns the count of - # newly-inserted beliefs. Used twice: (a) for the public return - # value (newly inserted beliefs, deduped), (b) for the #809 - # intra-turn edge wiring below (per-sentence belief id, - # position-preserving). - log_belief_ids: list[str | None] = [] - inserted: list[str] = [] - seen: set[str] = set() - for log_id in log_ids: - outcome = worker_result.outcomes.get(log_id) - bid: str | None - was_inserted: bool - if outcome is not None: - bid, was_inserted = outcome - else: - # A sibling process stamped this row in the window between - # record_ingest and our run_worker pass. Fall back to the - # log row; a row stamped elsewhere is by definition not a - # brand-new insert of ours. - was_inserted = False - bid = None - entry = store.get_ingest_log_entry(log_id) - if entry is not None: - ids = entry.get("derived_belief_ids") or [] - if isinstance(ids, list) and ids: - head = ids[0] - if isinstance(head, str): - bid = head - log_belief_ids.append(bid) - if bid is not None and was_inserted and bid not in seen: - seen.add(bid) - inserted.append(bid) - - # #809: wire intra-turn DERIVED_FROM edges between consecutive - # full-length beliefs whose original-prose ordering was separated - # by one or more sub-floor clauses. Edge direction matches the - # inter-turn DERIVED_FROM convention in `ingest_jsonl` (src is the - # later belief, dst is the earlier one — "this is derived from - # that earlier one"). Anchor_text is the joined sub-floor clauses, - # truncated to ANCHOR_TEXT_MAX_LEN. - for i in range(1, len(log_belief_ids)): - between = subfloor_between[i] - if not between: - continue - prior_bid = log_belief_ids[i - 1] - curr_bid = log_belief_ids[i] - if prior_bid is None or curr_bid is None or prior_bid == curr_bid: - continue - anchor = " | ".join(between)[:ANCHOR_TEXT_MAX_LEN] - if store.get_edge(curr_bid, prior_bid, EDGE_DERIVED_FROM) is not None: - continue - store.insert_edge(Edge( - src=curr_bid, dst=prior_bid, - type=EDGE_DERIVED_FROM, weight=1.0, - anchor_text=anchor, - )) - - # #988/#1000: optionally build the semantic-edge substrate at ingest. - # Writes CONTRADICTS edges across the store for this turn's new beliefs - # so the graph lanes (HRR-expand #981, BFS) are no longer inert on - # benchmark corpora. Default-OFF (is_auto_relationship_detection_enabled): - # when off, this branch is never entered and ingest is byte-identical to - # today. Gated on `inserted` so corroboration-only turns skip the audit. - # The CONTRADICTS build is incremental (delta-scoped, #1000): only pairs - # touching at least one belief from `inserted` are evaluated, which is - # provably equivalent to a full-store audit for discovering new edges - # while avoiding the O(n²) re-scan of the whole store every turn. - if inserted: - from aelfrice.relationship_detector import ( - is_auto_relationship_detection_enabled, - write_semantic_edges, - ) - - if is_auto_relationship_detection_enabled(): - write_semantic_edges(store, new_belief_ids=inserted) - - # #1064: optionally chain this turn's new beliefs into the - # per-session temporal spine (TEMPORAL_NEXT, src = successor, - # dst = predecessor). Default-OFF (is_temporal_spine_write_enabled): - # when off, this branch is never entered and ingest is - # byte-identical to today. Gated on `inserted` so - # corroboration-only turns skip the predecessor lookups. - from aelfrice.temporal_spine import ( - is_temporal_spine_write_enabled, - write_temporal_spine, - ) - - if is_temporal_spine_write_enabled(): - write_temporal_spine(store, new_belief_ids=inserted) + # #1135: one write group per turn. Pre-batching this was ~8 + # commits per 2-sentence turn (record_ingest per sentence, the + # worker's insert/corroborate + stamp per row, edge inserts). + # One transaction also makes the turn atomic: a crash mid-turn + # leaves no unstamped orphan rows. + with store.transaction(): + log_ids: list[str] = [] + for sentence in full_sentences: + log_id = store.record_ingest( + source_kind=INGEST_SOURCE_TRANSCRIPT, + source_path=source, + raw_text=sentence, + session_id=session_id, + ts=ts, + raw_meta=raw_meta, + ) + log_ids.append(log_id) + + # Worker is idempotent and scans all unstamped rows; calling it once + # at end-of-turn is the per-batch invocation pattern from the spec. + worker_result = run_worker(store) + + # Resolve each log_id to its per-row fate, in input order. #1135: + # the worker reports `(belief_id, was_inserted)` per stamped row, + # replacing the pre/post `set(list_belief_ids())` diff (a full-table + # scan per turn) and the per-log-id re-read. `was_inserted` is True + # only for brand-new canonical rows, which preserves the pre-#264 + # public contract that `ingest_turn` returns the count of + # newly-inserted beliefs. Used twice: (a) for the public return + # value (newly inserted beliefs, deduped), (b) for the #809 + # intra-turn edge wiring below (per-sentence belief id, + # position-preserving). + log_belief_ids: list[str | None] = [] + inserted: list[str] = [] + seen: set[str] = set() + for log_id in log_ids: + outcome = worker_result.outcomes.get(log_id) + bid: str | None + was_inserted: bool + if outcome is not None: + bid, was_inserted = outcome + else: + # A sibling process stamped this row in the window between + # record_ingest and our run_worker pass. Fall back to the + # log row; a row stamped elsewhere is by definition not a + # brand-new insert of ours. + was_inserted = False + bid = None + entry = store.get_ingest_log_entry(log_id) + if entry is not None: + ids = entry.get("derived_belief_ids") or [] + if isinstance(ids, list) and ids: + head = ids[0] + if isinstance(head, str): + bid = head + log_belief_ids.append(bid) + if bid is not None and was_inserted and bid not in seen: + seen.add(bid) + inserted.append(bid) + + # #809: wire intra-turn DERIVED_FROM edges between consecutive + # full-length beliefs whose original-prose ordering was separated + # by one or more sub-floor clauses. Edge direction matches the + # inter-turn DERIVED_FROM convention in `ingest_jsonl` (src is the + # later belief, dst is the earlier one — "this is derived from + # that earlier one"). Anchor_text is the joined sub-floor clauses, + # truncated to ANCHOR_TEXT_MAX_LEN. + for i in range(1, len(log_belief_ids)): + between = subfloor_between[i] + if not between: + continue + prior_bid = log_belief_ids[i - 1] + curr_bid = log_belief_ids[i] + if prior_bid is None or curr_bid is None or prior_bid == curr_bid: + continue + anchor = " | ".join(between)[:ANCHOR_TEXT_MAX_LEN] + if store.get_edge(curr_bid, prior_bid, EDGE_DERIVED_FROM) is not None: + continue + store.insert_edge(Edge( + src=curr_bid, dst=prior_bid, + type=EDGE_DERIVED_FROM, weight=1.0, + anchor_text=anchor, + )) + + # #988/#1000: optionally build the semantic-edge substrate at ingest. + # Writes CONTRADICTS edges across the store for this turn's new beliefs + # so the graph lanes (HRR-expand #981, BFS) are no longer inert on + # benchmark corpora. Default-OFF (is_auto_relationship_detection_enabled): + # when off, this branch is never entered and ingest is byte-identical to + # today. Gated on `inserted` so corroboration-only turns skip the audit. + # The CONTRADICTS build is incremental (delta-scoped, #1000): only pairs + # touching at least one belief from `inserted` are evaluated, which is + # provably equivalent to a full-store audit for discovering new edges + # while avoiding the O(n²) re-scan of the whole store every turn. + if inserted: + from aelfrice.relationship_detector import ( + is_auto_relationship_detection_enabled, + write_semantic_edges, + ) + + if is_auto_relationship_detection_enabled(): + write_semantic_edges(store, new_belief_ids=inserted) + + # #1064: optionally chain this turn's new beliefs into the + # per-session temporal spine (TEMPORAL_NEXT, src = successor, + # dst = predecessor). Default-OFF (is_temporal_spine_write_enabled): + # when off, this branch is never entered and ingest is + # byte-identical to today. Gated on `inserted` so + # corroboration-only turns skip the predecessor lookups. + from aelfrice.temporal_spine import ( + is_temporal_spine_write_enabled, + write_temporal_spine, + ) + + if is_temporal_spine_write_enabled(): + write_temporal_spine(store, new_belief_ids=inserted) return inserted diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 1e51f89da..7c45d5a0b 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -1225,8 +1225,14 @@ def transaction(self) -> Iterator[None]: except BaseException: self._txn_depth -= 1 if self._txn_depth == 0: - self._conn.rollback() self._pending_invalidation = False + try: + self._conn.rollback() + except sqlite3.Error: + # A rollback that itself fails (e.g. the handle was + # closed mid-group) must not mask the original + # exception propagating below. + pass raise self._txn_depth -= 1 if self._txn_depth == 0: From b332b7d75f866e86a2f53cf5a23755a78db7337c Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:20:22 -0700 Subject: [PATCH 07/14] perf(bm25): persist the BM25F index to a generation-stamped sidecar (#1135) BM25F is default-ON but no caller passed a bm25f_cache, so every retrieve() re-tokenized + Porter-stemmed the whole corpus and rebuilt the CSR matrix (584 ms at 5k beliefs; the UPS hook is a fresh process per prompt, and even the MCP server rebuilt per query while leaking one invalidation callback per call). - MemoryStore gains a durable store_generation counter, bumped inside the same transaction as every belief/edge content mutation (the 11 _fire_invalidation sites, now _commit_mutation). Feedback/touch writes don't bump it - they change ranking inputs, not index content. Seed is read-first so opens stay lock-free. - BM25IndexCache.get() loads .bm25f when its generation + scope-id stamp and k1/b/anchor_weight match, else builds and atomically rewrites it (os.replace; fail-soft both ways; :memory: skips persistence). - serialize() v2 widens k1/b/avgdl to float64: a loaded index now scores byte-identically to a fresh build (float32 round-trip perturbed low-order score bits). No v1 blobs exist - nothing called serialize() before this. - retrieve() reuses one store-scoped cache instead of constructing a fresh one per call. --- src/aelfrice/bm25.py | 162 ++++++++++++++++++++++++++++++--- src/aelfrice/retrieval.py | 32 ++++++- src/aelfrice/store.py | 109 +++++++++++++++++----- tests/test_bm25_sidecar.py | 182 +++++++++++++++++++++++++++++++++++++ 4 files changed, 451 insertions(+), 34 deletions(-) create mode 100644 tests/test_bm25_sidecar.py diff --git a/src/aelfrice/bm25.py b/src/aelfrice/bm25.py index 77b8c3bb8..f1b1bcd88 100644 --- a/src/aelfrice/bm25.py +++ b/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 @@ def _stem(token: str) -> str: # 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 @@ def serialize(self) -> bytes: """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 @@ def serialize(self) -> bytes: 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 @@ def _read(dtype: np.dtype, count: int) -> np.ndarray: 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 @@ def _read_string() -> str: ) +# 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 @@ class BM25IndexCache: 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 (`.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`). @@ -576,20 +619,117 @@ def __post_init__(self) -> None: 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 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) 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 + # --- 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 + # k1/b round-trip through float32 in the blob; compare in + # float32 to avoid spurious rebuilds. + if np.float32(index.k1) != np.float32(self.k1): + return None + if np.float32(index.b) != np.float32(self.b): + return None + 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__ = [ "DEFAULT_ANCHOR_WEIGHT", diff --git a/src/aelfrice/retrieval.py b/src/aelfrice/retrieval.py index cd233798a..6fa69869d 100644 --- a/src/aelfrice/retrieval.py +++ b/src/aelfrice/retrieval.py @@ -2647,6 +2647,31 @@ def _origin_priority(origin: str) -> int: ) +def _store_scoped_bm25f_cache( + store: MemoryStore, + *, + anchor_weight: int, +) -> BM25IndexCache: + """One process-lifetime `BM25IndexCache` per store (#1135). + + Replaces the pre-#1135 per-retrieve construction, which leaked one + invalidation-callback subscription per query and threw away the + built index between calls on long-running processes (MCP server). + The cache lives on `store._bm25f_shared_cache` so its lifetime is + the store's. A changed `anchor_weight` (the meta-belief consumer + can move it between calls) drops the cached index; the sidecar + check in `BM25IndexCache.get()` compares weights independently. + """ + cache = store._bm25f_shared_cache # noqa: SLF001 — slot owned here + if not isinstance(cache, BM25IndexCache): + cache = BM25IndexCache(store, anchor_weight=anchor_weight) + store._bm25f_shared_cache = cache # noqa: SLF001 + elif cache.anchor_weight != anchor_weight: + cache.anchor_weight = anchor_weight + cache.invalidate() + return cache + + def _l1_hits( store: MemoryStore, query: str, @@ -2735,8 +2760,13 @@ def _l1_hits( # anchor_weight through the meta-belief consumer; an explicit # caller-supplied cache is honoured as-is (the bench harness # and unit tests pin specific anchor_weights via the cache). + # #1135: the fallback cache is store-scoped and reused across + # calls — constructing one per retrieve leaked an invalidation + # callback per query and rebuilt the index every time on + # long-running processes; with the sidecar (bm25.py) a fresh + # hook process loads the persisted index instead of building. if bm25f_cache is None: - cache = BM25IndexCache( + cache = _store_scoped_bm25f_cache( store, anchor_weight=resolve_bm25f_anchor_weight_with_meta( store, now_ts=int(time.time()), diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 7c45d5a0b..91dc6e01c 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -595,6 +595,15 @@ def _check_insert_belief_authority() -> None: # re-extract entities for every existing belief on first open. SCHEMA_META_ENTITY_BACKFILL: Final[str] = "entity_backfill_complete" +# #1135: durable mutation counter for cross-process derived-state +# caches (the persistent BM25F sidecar). Bumped in the same SQLite +# transaction as every belief/edge content mutation — exactly the +# mutations that fire the in-process invalidation registry — so a +# cache blob stamped with generation G is valid iff the store still +# reads G. Feedback/touch/corroboration writes do NOT bump it (they +# change ranking inputs, not index content). +SCHEMA_META_STORE_GENERATION: Final[str] = "store_generation" + # #1135: marker for the v1.2 origin backfill (_BACKFILL_STATEMENTS). # The two UPDATEs ran unguarded on every open — two full-table write # statements per open, compounding with the hook's multi-open pattern. @@ -1045,6 +1054,9 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # helpers commit through `_commit` too. self._txn_depth: int = 0 self._pending_invalidation: bool = False + # #1135: retained for sidecar-file placement (persistent BM25F + # index). ":memory:" disables persistence. + self._db_path: str = path self._conn: sqlite3.Connection = sqlite3.connect(path) self._conn.row_factory = sqlite3.Row # WAL only meaningful on-disk; harmless on :memory:. @@ -1089,6 +1101,22 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # #1135: one-shot. Ran unguarded on every open pre-v4.2; the # marker matches the other schema_meta-gated passes. Rides the # single open commit below. + # #1135: seed the durable mutation counter. Read-first: an + # unconditional INSERT OR IGNORE would take the write lock on + # every open even when the row exists, blocking behind any + # concurrent writer's open transaction. Rides the single open + # commit below. Two connections racing the first-ever seed both + # pass the SELECT; OR IGNORE makes the second insert a no-op. + seeded = self._conn.execute( + "SELECT 1 FROM schema_meta WHERE key = ?", + (SCHEMA_META_STORE_GENERATION,), + ).fetchone() + if seeded is None: + self._conn.execute( + "INSERT OR IGNORE INTO schema_meta (key, value) " + "VALUES (?, '0')", + (SCHEMA_META_STORE_GENERATION,), + ) marker = self._conn.execute( "SELECT value FROM schema_meta WHERE key = ?", (SCHEMA_META_ORIGIN_BACKFILL,), @@ -1156,6 +1184,13 @@ def __init__(self, path: str, *, project_context_default: str = "") -> None: # content-hash UNIQUE swap so any same-key collisions raise # cleanly. Idempotent via SCHEMA_META marker. self._maybe_rehash_speculative_v2() + # #1135: process-lifetime BM25F cache slot, managed by + # `aelfrice.retrieval._store_scoped_bm25f_cache`. Typed loosely + # to avoid a store->bm25 import cycle. One cache per store also + # means one invalidation-callback subscription per store — the + # pre-#1135 per-retrieve construction leaked one callback per + # query on long-running processes (MCP server). + self._bm25f_shared_cache: object | None = None # #655 read-only federation. Peer DBs are opened on demand via # `federation.open_peer_connection`; the deps list itself is # cached eagerly so `aelf health` can report missing peers @@ -1205,6 +1240,47 @@ def _commit(self) -> None: if self._txn_depth == 0: self._conn.commit() + def _commit_mutation(self) -> None: + """Commit a belief/edge content mutation. + + Bumps the durable store generation in the same transaction as + the mutation (so cross-process caches keyed on the generation + can never observe a stale-but-matching pair), then commits and + fires the in-process invalidation registry. Inside + `transaction()` the bump rides the outer commit and the fire is + deferred, like any other write. + """ + self._conn.execute( + "UPDATE schema_meta SET value = CAST(value AS INTEGER) + 1 " + "WHERE key = ?", + (SCHEMA_META_STORE_GENERATION,), + ) + self._commit() + self._fire_invalidation() + + @property + def db_path(self) -> str: + """The path this store was opened with (":memory:" for tests).""" + return self._db_path + + def store_generation(self) -> int: + """Durable belief/edge mutation counter (#1135). + + Monotonically increasing across processes; bumped inside the + same transaction as every content mutation. Missing key (a DB + created before v4.2 that has not been reopened) reads as 0. + """ + row = self._conn.execute( + "SELECT value FROM schema_meta WHERE key = ?", + (SCHEMA_META_STORE_GENERATION,), + ).fetchone() + if row is None: + return 0 + try: + return int(row["value"]) + except (TypeError, ValueError): + return 0 + @contextmanager def transaction(self) -> Iterator[None]: """Group multiple mutating calls into one SQLite transaction. @@ -2286,8 +2362,7 @@ def insert_belief(self, b: Belief) -> None: ) self._write_belief_entities(b.id, b.content) self._bump_belief_version(b.id) - self._commit() - self._fire_invalidation() + self._commit_mutation() def get_belief_by_content_hash(self, content_hash: str) -> Belief | None: """Look up a belief by its content_hash. Returns None if not found. @@ -2386,8 +2461,7 @@ def update_belief(self, b: Belief) -> None: ) self._write_belief_entities(b.id, b.content) self._bump_belief_version(b.id) - self._commit() - self._fire_invalidation() + self._commit_mutation() def delete_belief(self, belief_id: str) -> None: self._conn.execute("DELETE FROM beliefs WHERE id = ?", (belief_id,)) @@ -2399,8 +2473,7 @@ def delete_belief(self, belief_id: str) -> None: self._conn.execute( "DELETE FROM belief_entities WHERE belief_id = ?", (belief_id,) ) - self._commit() - self._fire_invalidation() + self._commit_mutation() def soft_delete_belief(self, belief_id: str, ts: str | None = None) -> None: """Set `valid_to` on a belief to soft-delete it (v2.1 #548 wonder GC). @@ -2424,8 +2497,7 @@ def soft_delete_belief(self, belief_id: str, ts: str | None = None) -> None: "DELETE FROM beliefs_fts WHERE id = ?", (belief_id,) ) self._bump_belief_version(belief_id) - self._commit() - self._fire_invalidation() + self._commit_mutation() def restore_belief(self, belief_id: str) -> bool: """Clear ``valid_to`` on a soft-deleted belief, returning it to active. @@ -2463,8 +2535,7 @@ def restore_belief(self, belief_id: str) -> bool: (belief_id, row["content"]), ) self._bump_belief_version(belief_id) - self._commit() - self._fire_invalidation() + self._commit_mutation() return True def update_last_confirmed_at(self, belief_id: str, ts_iso: str) -> None: @@ -2480,8 +2551,7 @@ def update_last_confirmed_at(self, belief_id: str, ts_iso: str) -> None: (ts_iso, belief_id), ) self._bump_belief_version(belief_id) - self._commit() - self._fire_invalidation() + self._commit_mutation() def list_review_candidates(self, *, limit: int = 10) -> list[Belief]: """Return active beliefs to surface in the weekly review file (#936). @@ -4461,8 +4531,7 @@ def set_retention_class(self, belief_id: str, retention_class: str) -> None: (retention_class, belief_id), ) self._bump_belief_version(belief_id) - self._commit() - self._fire_invalidation() + self._commit_mutation() def count_beliefs_by_type(self) -> dict[str, int]: """Return a mapping of belief type → count across all beliefs.""" @@ -4847,8 +4916,7 @@ def insert_edge(self, e: Edge) -> None: (e.src, e.dst, e.type, e.weight, e.anchor_text), ) self._bump_edge_version(e.src, e.dst, e.type) - self._commit() - self._fire_invalidation() + self._commit_mutation() def get_edge(self, src: str, dst: str, type_: str) -> Edge | None: cur = self._conn.execute( @@ -4865,16 +4933,14 @@ def update_edge(self, e: Edge) -> None: (e.weight, e.anchor_text, e.src, e.dst, e.type), ) self._bump_edge_version(e.src, e.dst, e.type) - self._commit() - self._fire_invalidation() + self._commit_mutation() def delete_edge(self, src: str, dst: str, type_: str) -> None: self._conn.execute( "DELETE FROM edges WHERE src = ? AND dst = ? AND type = ?", (src, dst, type_), ) - self._commit() - self._fire_invalidation() + self._commit_mutation() def delete_edges_by_type(self, type_: str) -> int: """Delete every edge of ``type_``; return how many were removed. @@ -4888,8 +4954,7 @@ def delete_edges_by_type(self, type_: str) -> int: "DELETE FROM edges WHERE type = ?", (type_,) ) removed = cur.rowcount - self._commit() - self._fire_invalidation() + self._commit_mutation() return removed def edges_from(self, src: str) -> list[Edge]: diff --git a/tests/test_bm25_sidecar.py b/tests/test_bm25_sidecar.py new file mode 100644 index 000000000..fb73a90cb --- /dev/null +++ b/tests/test_bm25_sidecar.py @@ -0,0 +1,182 @@ +"""#1135 persistent BM25F sidecar: `BM25IndexCache` load/persist. + +Contract under test: a built index is persisted next to the DB and a +fresh cache (fresh process stand-in) loads it instead of rebuilding; +any content mutation invalidates via the durable generation stamp; +mismatched parameters, corrupt blobs, and in-memory stores all fall +back to a build; a loaded index scores byte-identically to a built +one (the retrieval byte-identity AC depends on this). +""" +from __future__ import annotations + +from pathlib import Path + +from aelfrice.bm25 import BM25Index, BM25IndexCache, sidecar_path_for +from aelfrice.models import BELIEF_FACTUAL, LOCK_NONE, Belief +from aelfrice.store import MemoryStore + + +def _mk_belief(bid: str, content: str) -> Belief: + return Belief( + id=bid, + content=content, + content_hash="h_" + bid, + alpha=1.0, + beta=1.0, + type=BELIEF_FACTUAL, + lock_level=LOCK_NONE, + locked_at=None, + created_at="2026-07-21T00:00:00Z", + last_retrieved_at=None, + ) + + +def _seed(store: MemoryStore) -> None: + store.insert_belief(_mk_belief("b1", "the quick brown fox jumps")) + store.insert_belief(_mk_belief("b2", "sqlite stores beliefs durably")) + store.insert_belief(_mk_belief("b3", "the fox likes sqlite")) + + +def test_sidecar_written_on_build(tmp_path: Path) -> None: + store = MemoryStore(str(tmp_path / "m.db")) + try: + _seed(store) + cache = BM25IndexCache(store) + cache.get() + sidecar = sidecar_path_for(store) + assert sidecar is not None and sidecar.is_file() + assert sidecar.stat().st_size > 0 + finally: + store.close() + + +def test_fresh_cache_loads_sidecar_without_building( + tmp_path: Path, monkeypatch, +) -> None: + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + BM25IndexCache(store).get() # builds + persists + finally: + store.close() + + # Fresh store + fresh cache = fresh hook process. A build here + # means the sidecar was not honoured. + builds: list[int] = [] + real_build = BM25Index.build + + def counting_build(*args: object, **kwargs: object) -> BM25Index: + builds.append(1) + return real_build(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(BM25Index, "build", counting_build) + store2 = MemoryStore(str(db)) + try: + idx = BM25IndexCache(store2).get() + assert builds == [], "sidecar present and current, but build ran" + assert idx.score("fox", top_k=3) # loaded index actually works + finally: + store2.close() + + +def test_loaded_index_scores_identical_to_built(tmp_path: Path) -> None: + """Byte-identity AC: the sidecar path must not perturb ranking.""" + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + built = BM25IndexCache(store).get() + loaded = BM25IndexCache(store)._load_sidecar() + assert loaded is not None + for query in ("fox", "sqlite beliefs", "quick brown sqlite"): + assert built.score(query, top_k=10) == loaded.score( + query, top_k=10, + ) + finally: + store.close() + + +def test_mutation_invalidates_sidecar_via_generation( + tmp_path: Path, +) -> None: + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + BM25IndexCache(store).get() + gen_at_build = store.store_generation() + store.insert_belief(_mk_belief("b4", "a brand new belief row")) + assert store.store_generation() > gen_at_build + # A fresh cache must reject the stale sidecar and rebuild — + # the new belief has to be retrievable. + idx = BM25IndexCache(store)._load_sidecar() + assert idx is None, "stale sidecar accepted after mutation" + rebuilt = BM25IndexCache(store).get() + assert any( + bid == "b4" for bid, _ in rebuilt.score("brand new belief", top_k=5) + ) + finally: + store.close() + + +def test_anchor_weight_mismatch_rejects_sidecar(tmp_path: Path) -> None: + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + BM25IndexCache(store, anchor_weight=3).get() + assert BM25IndexCache(store, anchor_weight=5)._load_sidecar() is None + finally: + store.close() + + +def test_corrupt_sidecar_falls_back_to_build(tmp_path: Path) -> None: + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + cache = BM25IndexCache(store) + cache.get() + sidecar = sidecar_path_for(store) + assert sidecar is not None + sidecar.write_bytes(b"garbage not an index blob") + fresh = BM25IndexCache(store) + assert fresh._load_sidecar() is None + idx = fresh.get() # must build, not raise + assert idx.score("fox", top_k=3) + finally: + store.close() + + +def test_memory_store_has_no_sidecar(tmp_path: Path) -> None: + store = MemoryStore(":memory:") + try: + _seed(store) + assert sidecar_path_for(store) is None + BM25IndexCache(store).get() # no crash, no file + assert list(tmp_path.iterdir()) == [] + finally: + store.close() + + +def test_generation_bump_rides_mutation_transaction( + tmp_path: Path, +) -> None: + """The durable counter moves with every content mutation and holds + still for non-content writes (feedback/touches don't reindex).""" + store = MemoryStore(str(tmp_path / "m.db")) + try: + g0 = store.store_generation() + store.insert_belief(_mk_belief("b1", "content one")) + g1 = store.store_generation() + assert g1 == g0 + 1 + with store.transaction(): + store.insert_belief(_mk_belief("b2", "content two")) + store.insert_belief(_mk_belief("b3", "content three")) + g2 = store.store_generation() + assert g2 == g1 + 2 + store.stamp_retrieved(["b1"]) # ranking input, not index content + assert store.store_generation() == g2 + finally: + store.close() From 73641712a22fcc91fc1805a2a60bdd9b8eaa2f08 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:22:00 -0700 Subject: [PATCH 08/14] perf(retrieval): memoize .aelfrice.toml parse per (mtime, size) (#1135) ~24 resolver call sites fall through to the TOML rung per retrieve(), each re-reading and re-parsing the file. The directory walk (stat calls) still runs per call so created/deleted config files are honoured, but the read + tomllib parse is cached per file until its mtime_ns/size changes. Malformed-file stderr traces now print once per file version instead of once per resolver call. --- src/aelfrice/retrieval.py | 108 +++++++++++++++++------------- tests/test_retrieval_toml_memo.py | 36 ++++++++++ 2 files changed, 98 insertions(+), 46 deletions(-) create mode 100644 tests/test_retrieval_toml_memo.py diff --git a/src/aelfrice/retrieval.py b/src/aelfrice/retrieval.py index 6fa69869d..05273ff70 100644 --- a/src/aelfrice/retrieval.py +++ b/src/aelfrice/retrieval.py @@ -893,6 +893,62 @@ def _env_hrr_persist_override() -> bool | None: return None +# #1135: memoized `.aelfrice.toml` parse. ~24 resolver call sites fall +# through to the TOML rung per retrieve(), and each used to re-read + +# re-parse the file. The directory walk (a handful of stat calls) +# still runs per call — so a config file created, deleted, or moved +# mid-process is honoured — but the read + tomllib parse is cached per +# file until its (mtime_ns, size) changes. Value None records a +# malformed / unreadable / section-less parse, so the error path is +# cached too (and its stderr trace prints once per file version +# instead of once per resolver call). +_TOML_SECTION_CACHE: dict[str, tuple[int, int, dict[str, Any] | None]] = {} + + +def _parsed_retrieval_section(candidate: Path) -> dict[str, Any] | None: + """Return `candidate`'s `[retrieval]` table, or None when the file + is unreadable, malformed, or has no such table. Memoized on the + file's (mtime_ns, size); tolerant — never raises.""" + serr: IO[str] = sys.stderr + try: + st = candidate.stat() + except OSError as exc: + print( + f"aelfrice retrieval: cannot read {candidate}: {exc}", + file=serr, + ) + return None + cache_key = str(candidate) + hit = _TOML_SECTION_CACHE.get(cache_key) + if ( + hit is not None + and hit[0] == st.st_mtime_ns + and hit[1] == st.st_size + ): + return hit[2] + section: dict[str, Any] | None = None + try: + raw = candidate.read_bytes() + parsed: dict[str, Any] = tomllib.loads( + raw.decode("utf-8", errors="replace"), + ) + section_obj: Any = parsed.get(RETRIEVAL_SECTION, {}) + if isinstance(section_obj, dict): + section = section_obj + except OSError as exc: + print( + f"aelfrice retrieval: cannot read {candidate}: {exc}", + file=serr, + ) + except tomllib.TOMLDecodeError as exc: + print( + f"aelfrice retrieval: malformed TOML in {candidate}: {exc}", + file=serr, + ) + _TOML_SECTION_CACHE[cache_key] = (st.st_mtime_ns, st.st_size, section) + return section + + def _read_toml_flag_for( key: str, start: Path | None = None, @@ -912,30 +968,10 @@ def _read_toml_flag_for( seen.add(current) candidate = current / CONFIG_FILENAME if candidate.is_file(): - try: - raw = candidate.read_bytes() - except OSError as exc: - print( - f"aelfrice retrieval: cannot read {candidate}: {exc}", - file=serr, - ) + section = _parsed_retrieval_section(candidate) + if section is None or key not in section: return None - try: - parsed: dict[str, Any] = tomllib.loads( - raw.decode("utf-8", errors="replace"), - ) - except tomllib.TOMLDecodeError as exc: - print( - f"aelfrice retrieval: malformed TOML in {candidate}: {exc}", - file=serr, - ) - return None - section_obj: Any = parsed.get(RETRIEVAL_SECTION, {}) - if not isinstance(section_obj, dict): - return None - if key not in section_obj: # type: ignore[operator] - return None - value: Any = section_obj[key] # type: ignore[index] + value: Any = section[key] if isinstance(value, bool): return value print( @@ -969,30 +1005,10 @@ def _read_toml_float_for( seen.add(current) candidate = current / CONFIG_FILENAME if candidate.is_file(): - try: - raw = candidate.read_bytes() - except OSError as exc: - print( - f"aelfrice retrieval: cannot read {candidate}: {exc}", - file=serr, - ) - return None - try: - parsed: dict[str, Any] = tomllib.loads( - raw.decode("utf-8", errors="replace"), - ) - except tomllib.TOMLDecodeError as exc: - print( - f"aelfrice retrieval: malformed TOML in {candidate}: {exc}", - file=serr, - ) - return None - section_obj: Any = parsed.get(RETRIEVAL_SECTION, {}) - if not isinstance(section_obj, dict): - return None - if key not in section_obj: # type: ignore[operator] + section = _parsed_retrieval_section(candidate) + if section is None or key not in section: return None - value: Any = section_obj[key] # type: ignore[index] + value: Any = section[key] # bool is a subclass of int -- reject it explicitly so # `posterior_weight = true` reads as malformed rather # than silently coercing to 1.0. diff --git a/tests/test_retrieval_toml_memo.py b/tests/test_retrieval_toml_memo.py new file mode 100644 index 000000000..44f97eb4e --- /dev/null +++ b/tests/test_retrieval_toml_memo.py @@ -0,0 +1,36 @@ +"""#1135 TOML parse memo: cached per (mtime_ns, size), rewrite honoured.""" +from __future__ import annotations + +from pathlib import Path + +import aelfrice.retrieval as retrieval + + +def test_toml_flag_rewrite_is_honoured(tmp_path: Path) -> None: + cfg = tmp_path / ".aelfrice.toml" + cfg.write_text("[retrieval]\nuse_bfs = true\n") + assert retrieval._read_toml_flag_for("use_bfs", start=tmp_path) is True + cfg.write_text("[retrieval]\nuse_bfs = false\n") + assert retrieval._read_toml_flag_for("use_bfs", start=tmp_path) is False + cfg.unlink() + assert retrieval._read_toml_flag_for("use_bfs", start=tmp_path) is None + + +def test_toml_parse_is_memoized(tmp_path: Path, monkeypatch) -> None: + cfg = tmp_path / ".aelfrice.toml" + cfg.write_text("[retrieval]\nposterior_weight = 0.4\n") + parses: list[int] = [] + import tomllib + + real_loads = tomllib.loads + + def counting_loads(*args: object, **kwargs: object) -> object: + parses.append(1) + return real_loads(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(retrieval.tomllib, "loads", counting_loads) + for _ in range(10): + assert retrieval._read_toml_float_for( + "posterior_weight", start=tmp_path, + ) == 0.4 + assert len(parses) <= 1, f"parsed {len(parses)}x for an unchanged file" From 9b22b3f4ef2ccaaa45e96664be6338439480d65b Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:27:40 -0700 Subject: [PATCH 09/14] perf(hook): open the store once per UserPromptSubmit prompt (#1135) The UPS process opened MemoryStore 4-6 times per prompt (relevance sweep, retrieval, injection events, touches, plus session-start on first prompt), each replaying the schema battery (~25-40 ms/prompt of pure re-open). user_prompt_submit now opens one handle after payload parse and threads it through those helpers; each keeps its legacy self-open fallback for direct callers and tests, and the default-off lanes (sentiment, category boost, phantom blocks) are unchanged. A failed shared open degrades to the fallback path, preserving per-helper fail-softness. --- src/aelfrice/hook.py | 122 +++++++++++++++++++++++++++++++------------ 1 file changed, 88 insertions(+), 34 deletions(-) diff --git a/src/aelfrice/hook.py b/src/aelfrice/hook.py index 3202d6531..2d5415558 100644 --- a/src/aelfrice/hook.py +++ b/src/aelfrice/hook.py @@ -33,10 +33,11 @@ import time import tomllib import traceback +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import IO, Any, Final, cast +from typing import IO, Any, Final, Iterator, cast try: from aelfrice.db_paths import active_project_context, db_path @@ -794,6 +795,12 @@ def user_prompt_submit( file=serr, ) return 0 + # #1135: one store handle for the whole prompt. The helpers below + # each used to open their own (4-6 opens per prompt, each replaying + # the schema battery). Opened lazily after the payload parses; None + # (open failure or in-memory DB) lets every helper fall back to its + # legacy self-open path, preserving per-helper fail-softness. + ups_store: MemoryStore | None = None try: # TTL-gated background update check, completely detached, never # blocks the hook. Statusline reads the cache it writes. @@ -822,6 +829,13 @@ def user_prompt_submit( payload_cwd = Path(cwd_field) except Exception: payload_cwd = None + try: + p = db_path() + if str(p) != ":memory:": + p.parent.mkdir(parents=True, exist_ok=True) + ups_store = MemoryStore(str(p)) + except Exception: + ups_store = None # #578: detect first prompt of a new session and build the # sub-block if needed. Fail-soft: any error in # detection or block-building leaves session_start_block="" so @@ -835,7 +849,7 @@ def user_prompt_submit( try: if is_session_first_prompt(session_id): session_start_block = _retrieve_session_start_block( - serr, cwd=payload_cwd, + serr, cwd=payload_cwd, store=ups_store, ) cadence_resume_block = _maybe_read_cadence_resume(serr) if cadence_resume_block: @@ -897,7 +911,9 @@ def user_prompt_submit( # retrieval so the shifted posteriors are visible to the # half-life / anchor-weight / etc. consumers that fire below. # Fail-soft, like sentiment-feedback. - _sweep_relevance_signal(session_id=session_id, stderr=serr) + _sweep_relevance_signal( + session_id=session_id, stderr=serr, store=ups_store, + ) # #674: prompt-shape gate — skip BM25 for system envelopes and # trivial acks, preserving any session-start block unchanged. gate_skip = False @@ -956,7 +972,7 @@ def user_prompt_submit( # Fail-soft: surface the trace, retrieve on prompt. traceback.print_exc(file=serr) retrieval_query = prompt - hits = _retrieve(retrieval_query, budget) + hits = _retrieve(retrieval_query, budget, store=ups_store) # #858 defect 3: drop hits whose stored project_context is # non-empty AND does not match the active in-process # context. '' on either side means "no filter": legacy @@ -1024,6 +1040,7 @@ def user_prompt_submit( source="ups", active_consumers=get_active_meta_belief_consumers(), stderr=serr, + store=ups_store, ) # total_chars measured post-collapse (what is actually injected). total_chars = sum(len(h.content) for h in hits) @@ -1113,6 +1130,7 @@ def user_prompt_submit( belief_ids=injected_ids, fire_idx=_next_fire - 1, stderr=serr, + store=ups_store, ) elif gate_skip: # Gate fired, no BM25 hits. Emit rebuild_log with empty hits @@ -1175,6 +1193,12 @@ def user_prompt_submit( sout.write(promotion_block) except Exception: # non-blocking: surface but do not fail traceback.print_exc(file=serr) + finally: + if ups_store is not None: + try: + ups_store.close() + except Exception: + pass return 0 @@ -1339,6 +1363,7 @@ def _sweep_relevance_signal( *, session_id: str | None, stderr: IO[str] | None = None, + store: MemoryStore | None = None, ) -> None: """Score prior turns' pending ``injection_events`` against the assistant transcript and update each active consumer's @@ -1375,11 +1400,9 @@ def _sweep_relevance_signal( score_references, ) - p = db_path() - if str(p) == ":memory:": - return - store = MemoryStore(str(p)) - try: + with _store_handle(store) as store: + if store is None: + return pending = store.list_pending_injection_events(session_id) if not pending: return @@ -1428,8 +1451,6 @@ def _sweep_relevance_signal( referenced=int(referenced), referenced_at=now_iso, ) - finally: - store.close() except Exception as exc: print( f"aelfrice: relevance sweeper failed (non-fatal): {exc}", @@ -1460,6 +1481,7 @@ def _record_touches( belief_ids: list[str], fire_idx: int, stderr: IO[str] | None = None, + store: MemoryStore | None = None, ) -> None: """Append one ``belief_touches`` row per injected belief. @@ -1498,11 +1520,9 @@ def _record_touches( from aelfrice.hot_path import ( # noqa: PLC0415 TOUCH_EVENT_KIND_INJECTION, ) - p = db_path() - if str(p) == ":memory:": - return - store = MemoryStore(str(p)) - try: + with _store_handle(store) as store: + if store is None: + return # Current turn's injection set — forward-only, no ring replay. # #1135: one commit for the batch instead of one per touch. with store.transaction(): @@ -1521,8 +1541,6 @@ def _record_touches( # extremely unlikely but possible (a belief # deleted between retrieval and the touch write). continue - finally: - store.close() except Exception as exc: print( f"aelfrice: UPS belief_touches emit failed " @@ -1539,6 +1557,7 @@ def _record_injection_events( source: str, active_consumers: list[str], stderr: IO[str] | None = None, + store: MemoryStore | None = None, ) -> None: """Append one ``injection_events`` row per injected belief. @@ -1557,12 +1576,10 @@ def _record_injection_events( if not session_id or not hits: return try: - p = db_path() - if str(p) == ":memory:": - return injected_at = datetime.now(timezone.utc).isoformat() - store = MemoryStore(str(p)) - try: + with _store_handle(store) as store: + if store is None: + return # #1135: one commit for the batch instead of one per event. with store.transaction(): for h in hits: @@ -1577,8 +1594,6 @@ def _record_injection_events( source=source, active_consumers=active_consumers, ) - finally: - store.close() except Exception as exc: print( f"aelfrice: UPS injection_events emit failed " @@ -1737,19 +1752,28 @@ def _build_conversation_aware_query( return " ".join(parts) -def _retrieve(prompt: str, token_budget: int) -> list[Belief]: +def _retrieve( + prompt: str, + token_budget: int, + *, + store: MemoryStore | None = None, +) -> list[Belief]: """Run retrieval for the given prompt and return the raw hit list. Separating retrieval from formatting lets callers inspect the hits (for telemetry, optional dedup, etc.) before the string is built. Returns an empty list when the store is absent or retrieval yields - nothing. + nothing. A caller-supplied `store` (#1135: the per-prompt shared + handle) is used as-is and left open; without one the legacy + open-per-call behaviour applies. """ - store = _open_store() - try: + if store is not None: return search_for_prompt(store, prompt, token_budget=token_budget) + owned = _open_store() + try: + return search_for_prompt(owned, prompt, token_budget=token_budget) finally: - store.close() + owned.close() def _filter_by_project_context(hits: list[Belief]) -> list[Belief]: @@ -1925,6 +1949,30 @@ def _open_store() -> MemoryStore: return MemoryStore(str(p)) +@contextmanager +def _store_handle(store: MemoryStore | None) -> Iterator[MemoryStore | None]: + """Yield `store` unchanged, or open a fresh one that closes on exit. + + #1135: the UserPromptSubmit flow opens one store per prompt and + threads it through its helpers; each helper keeps its legacy + self-open for callers (and tests) that pass no handle. Yields None + when no handle was passed AND the DB is in-memory — matching the + per-helper ":memory:" skip guards this replaces. + """ + if store is not None: + yield store + return + p = db_path() + if str(p) == ":memory:": + yield None + return + fresh = MemoryStore(str(p)) + try: + yield fresh + finally: + fresh.close() + + # --------------------------------------------------------------------------- # Sentiment-feedback hook lane (#606) # --------------------------------------------------------------------------- @@ -2587,8 +2635,12 @@ def _retrieve_session_start_block( stderr: IO[str] | None = None, *, cwd: Path | None = None, + store: MemoryStore | None = None, ) -> str: - """Open the store, build the session-start sub-block, close the store. + """Build the session-start sub-block. + + Uses the caller-supplied `store` when given (#1135: the per-prompt + shared handle, left open); otherwise opens and closes its own. `cwd` is forwarded to `_build_session_start_subblock` so the resolver (#887) uses the payload's cwd, not the @@ -2599,11 +2651,13 @@ def _retrieve_session_start_block( """ serr = stderr if stderr is not None else sys.stderr try: - store = _open_store() - try: + if store is not None: return _build_session_start_subblock(store, cwd=cwd) + owned = _open_store() + try: + return _build_session_start_subblock(owned, cwd=cwd) finally: - store.close() + owned.close() except Exception as exc: print( f"aelfrice: session-start sub-block build failed (non-fatal): {exc}", From a9d1007ad937157fa886b57fef38e5791a59c03d Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:28:44 -0700 Subject: [PATCH 10/14] docs(changelog): #1135 hot-path performance overhaul entry --- CHANGELOG/v4.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG/v4.md b/CHANGELOG/v4.md index 8c8e4e34b..3fe9ee146 100644 --- a/CHANGELOG/v4.md +++ b/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 `.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 UPDATEs) 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. From 9ba43676ebf2f97bbabd2582999bba0c9cca1b76 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:31:43 -0700 Subject: [PATCH 11/14] fix(migrate): apply the v1.2 origin catch-up during legacy-row conversion (#1135) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The catch-up used to be re-applied by the target store on every open; now that it is a marker-gated one-shot, the target's marker is already stamped (on an empty store) before legacy rows land, so migrated locked/correction rows stayed at origin=unknown. Apply the same mapping in _read_legacy_beliefs — migrate output is unchanged from pre-#1135 (regression: test_apply_preserves_origin_from_legacy_row, #224 contract). --- src/aelfrice/migrate.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/aelfrice/migrate.py b/src/aelfrice/migrate.py index 1335b119b..233f1c28a 100644 --- a/src/aelfrice/migrate.py +++ b/src/aelfrice/migrate.py @@ -37,9 +37,12 @@ from aelfrice.db_paths import repo_identity_from_db_path from aelfrice.models import ( + BELIEF_CORRECTION, BELIEF_SCOPE_PROJECT, LOCK_USER, ORIGIN_UNKNOWN, + ORIGIN_USER_CORRECTED, + ORIGIN_USER_STATED, Belief, Edge, ) @@ -117,6 +120,19 @@ def _read_legacy_beliefs( and lock_level != LOCK_USER ): pc = source_identity + # #1135: the v1.2 origin catch-up (promotion_path.md § 1) used + # to be re-applied by the target store on every open; it is a + # marker-gated one-shot now, and the target's marker is already + # stamped by the time legacy rows land. Apply the same mapping + # during conversion instead, so migrate output is unchanged: + # a still-'unknown' locked row was user-stated, a + # still-'unknown' correction row was user-corrected. + origin = row["origin"] if has_origin else ORIGIN_UNKNOWN + if origin == ORIGIN_UNKNOWN: + if lock_level == LOCK_USER: + origin = ORIGIN_USER_STATED + elif row["type"] == BELIEF_CORRECTION: + origin = ORIGIN_USER_CORRECTED out.append(Belief( id=row["id"], content=row["content"], @@ -128,7 +144,7 @@ def _read_legacy_beliefs( locked_at=row["locked_at"], created_at=row["created_at"], last_retrieved_at=row["last_retrieved_at"], - origin=row["origin"] if has_origin else ORIGIN_UNKNOWN, + origin=origin, scope=scope, project_context=pc, )) From c7f2f5668431f2d79dba9475f2f99b0363eaa459 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:37:29 -0700 Subject: [PATCH 12/14] style: reword 'UPDATEs' to satisfy the typos gate --- CHANGELOG/v4.md | 2 +- src/aelfrice/store.py | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/CHANGELOG/v4.md b/CHANGELOG/v4.md index 3fe9ee146..bdf72d965 100644 --- a/CHANGELOG/v4.md +++ b/CHANGELOG/v4.md @@ -19,7 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - *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 `.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 UPDATEs) 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. + - *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 diff --git a/src/aelfrice/store.py b/src/aelfrice/store.py index 91dc6e01c..ba9e0b15d 100644 --- a/src/aelfrice/store.py +++ b/src/aelfrice/store.py @@ -605,8 +605,9 @@ def _check_insert_belief_authority() -> None: SCHEMA_META_STORE_GENERATION: Final[str] = "store_generation" # #1135: marker for the v1.2 origin backfill (_BACKFILL_STATEMENTS). -# The two UPDATEs ran unguarded on every open — two full-table write -# statements per open, compounding with the hook's multi-open pattern. +# The two UPDATE statements ran unguarded on every open — two +# full-table writes per open, compounding with the hook's multi-open +# pattern. # Contemporary writers set origin explicitly (derive() routes, # cli lock upgrade), so the flip only ever matters once per legacy DB. # ISO timestamp on completion; absence triggers the pass on next open. From 2592c2f8d440982aaf1dffe270093ca0a16a1086 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:59:08 -0700 Subject: [PATCH 13/14] fix(bm25): revalidate cached index against the durable generation on get() (#1135) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-process invalidation callback only covers own-process mutations, so a resident process (MCP server) holding the store-scoped cache would never observe sibling-process writes (default-on ingest hooks) — a freshness regression vs the pre-#1135 rebuild-per-query behavior. One indexed schema_meta point-read per get(); on mismatch the cache drops the index and reloads the (possibly sibling-refreshed) sidecar before rebuilding. --- src/aelfrice/bm25.py | 13 +++++++++++++ tests/test_bm25_sidecar.py | 25 +++++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/src/aelfrice/bm25.py b/src/aelfrice/bm25.py index f1b1bcd88..0c2f6346d 100644 --- a/src/aelfrice/bm25.py +++ b/src/aelfrice/bm25.py @@ -611,6 +611,7 @@ class BM25IndexCache: 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: @@ -620,6 +621,15 @@ def __post_init__(self) -> None: def get(self) -> BM25Index: """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: @@ -634,6 +644,7 @@ def get(self) -> BM25Index: b=self.b, ) self._write_sidecar(self._index, generation) + self._generation = generation return self._index def invalidate(self) -> None: @@ -644,6 +655,7 @@ def invalidate(self) -> None: as stale; the next `get()` rebuild overwrites it. """ self._index = None + self._generation = None # --- Sidecar persistence (#1135) ---------------------------------- @@ -686,6 +698,7 @@ def _load_sidecar(self) -> BM25Index | None: return None if np.float32(index.b) != np.float32(self.b): return None + self._generation = generation return index except Exception: # noqa: BLE001 — any bad sidecar => rebuild return None diff --git a/tests/test_bm25_sidecar.py b/tests/test_bm25_sidecar.py index fb73a90cb..8b898d09e 100644 --- a/tests/test_bm25_sidecar.py +++ b/tests/test_bm25_sidecar.py @@ -180,3 +180,28 @@ def test_generation_bump_rides_mutation_transaction( assert store.store_generation() == g2 finally: store.close() + +def test_resident_cache_sees_sibling_process_writes(tmp_path: Path) -> None: + """A long-lived cache revalidates the durable generation on get(). + + The in-process invalidation callback only covers the cache's own + store handle; a write arriving through a second handle (stand-in + for a sibling process, e.g. an ingest hook firing while an MCP + server is resident) must still invalidate the cached index. + """ + db = tmp_path / "m.db" + resident = MemoryStore(str(db)) + sibling = MemoryStore(str(db)) + try: + _seed(resident) + cache = BM25IndexCache(resident) + cache.get() + sibling.insert_belief(_mk_belief("b4", "a sibling process write")) + refreshed = cache.get() + assert any( + bid == "b4" + for bid, _ in refreshed.score("sibling process write", top_k=5) + ), "resident cache served a stale index after a sibling write" + finally: + resident.close() + sibling.close() From d603cd21d3ff790c3d5be7e89454c1e5bf519349 Mon Sep 17 00:00:00 2001 From: rrs <276464689+robotrocketscience@users.noreply.github.com> Date: Tue, 21 Jul 2026 12:15:12 -0700 Subject: [PATCH 14/14] fix(bm25): compare sidecar k1/b at full precision (#1135) The v2 sidecar serialises k1/b as float64, so the round-trip is exact and the float32-narrowed comparison could let two configs that differ only below float32 precision share a sidecar. Compare exactly and pin with a nextafter mismatch test for each parameter. --- src/aelfrice/bm25.py | 9 +++++---- tests/test_bm25_sidecar.py | 16 ++++++++++++++++ 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/src/aelfrice/bm25.py b/src/aelfrice/bm25.py index 0c2f6346d..c1e5739a9 100644 --- a/src/aelfrice/bm25.py +++ b/src/aelfrice/bm25.py @@ -692,11 +692,12 @@ def _load_sidecar(self) -> BM25Index | None: index = BM25Index.deserialize(blob[off:]) if index.anchor_weight != self.anchor_weight: return None - # k1/b round-trip through float32 in the blob; compare in - # float32 to avoid spurious rebuilds. - if np.float32(index.k1) != np.float32(self.k1): + # 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 np.float32(index.b) != np.float32(self.b): + if index.b != self.b: return None self._generation = generation return index diff --git a/tests/test_bm25_sidecar.py b/tests/test_bm25_sidecar.py index 8b898d09e..6422109ff 100644 --- a/tests/test_bm25_sidecar.py +++ b/tests/test_bm25_sidecar.py @@ -9,6 +9,7 @@ """ from __future__ import annotations +import math from pathlib import Path from aelfrice.bm25 import BM25Index, BM25IndexCache, sidecar_path_for @@ -131,6 +132,21 @@ def test_anchor_weight_mismatch_rejects_sidecar(tmp_path: Path) -> None: store.close() +def test_k1_b_mismatch_rejects_sidecar(tmp_path: Path) -> None: + db = tmp_path / "m.db" + store = MemoryStore(str(db)) + try: + _seed(store) + base = BM25IndexCache(store) + base.get() + bumped_k1 = math.nextafter(base.k1, math.inf) + assert BM25IndexCache(store, k1=bumped_k1)._load_sidecar() is None + bumped_b = math.nextafter(base.b, math.inf) + assert BM25IndexCache(store, b=bumped_b)._load_sidecar() is None + finally: + store.close() + + def test_corrupt_sidecar_falls_back_to_build(tmp_path: Path) -> None: db = tmp_path / "m.db" store = MemoryStore(str(db))