diff --git a/CHANGELOG.md b/CHANGELOG.md index 45b98a2..a241c78 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,161 @@ # Changelog +> 아래 1.8.0 ~ 1.11.0 은 `geny-memory-adaptor` 에서 이식했다. XGEN 은 Geny 에서 +> 갈라져 독자적으로 가는 프로젝트이지만, **같은 뿌리에서 온 결함은 여기에도 +> 그대로 있다.** 검증 방법(테스트)까지 함께 가져왔고, 이 저장소의 기존 코드와 +> 충돌하는 부분은 없었다 — 갈라진 이후의 차이는 전부 포매팅이었다. + +## [1.11.0] — 2026-08-11 + +### Changed (one embedder table per process, not one per session) +Measured on a production host: an engine opened over an **empty** vault still +cost 68 MB of RSS, and six live vaults held only **two** distinct embedder +tables between them. The table is `vocab_size × dim × 4B` — 64 MB at the +defaults — so a host keeping ten sessions resident spent 640 MB on ten copies +of identical numbers. That, not the stored memories, was what made "keep every +session awake" expensive. + +The table is now shared process-wide, keyed on what determines its contents: +the generator arguments for a fresh table, the digest of the persisted blob for +a restored one. + +Sharing is safe because the table is never written in place — it is read, and +distillation *replaces* the whole embedder with a scratch instance holding its +own array. `setflags(write=False)` makes that structural rather than +conventional: an in-place write now raises instead of silently corrupting every +other session that shares the array. + +- `HashEmbedder(..., table=)` lets `loads()` skip generating a 64 MB table it + was about to throw away. +- `shared_table_stats()` reports what the process holds, for host health + endpoints. +- Cache is bounded (8 tables) so a content-keyed dict cannot leak. + +Measured against four real vaults in one process: **330.4 MB → 137.2 MB**, and +the marginal cost of one more resident session falls from 64–87 MB to +0–23 MB — now purely proportional to what that session actually remembers. + +## [1.10.0] — 2026-08-10 + +### Added (a catalogue, so a browser can ask small questions) +A vault browser asks in order: how much is there, what days, what is on this +day, what does this note say. The host answered the first three by +materialising the whole vault — 3.2 s and 4.8 MB of bodies held in memory to +produce one count, because its only listing primitive walks every note. The +index already holds exactly the metadata those questions need. + +- `catalog_counts(by="kind"|"day", kind=...)` — one GROUP BY, no bodies. +- `catalog_page(day=, kind=, limit=, offset=)` — one page of note metadata, + filtered and paged in SQL rather than sliced after the fact. +- `neighbourhood(node_ids, depth=, max_nodes=, max_edges=)` — the subgraph + around a selection, both bounds enforced and `truncated` reported. The + whole-vault snapshot it replaces was 5,384 nodes and 4.3 MB of JSON for a + single screen. +- `Store.edges_touching(node_ids, limit=)` underneath it. + +## [1.9.3] — 2026-08-10 + +### Fixed (the vault could no longer notice its own conversion) +1.9.0 recorded the new geometry against rows that had been derived by the +OLD tokenizer. From that moment the fingerprints matched, so 1.9.1 and 1.9.2 +— both correct in themselves — had nothing to compare against, and the +production vault stayed on the old tokenization with every check reporting +agreement. + +- `_GEOMETRY_VERSION` joins the fingerprint. Config fields cannot express + "the stemmer changed"; an explicit version can, and it is the lever for + every future change of meaning. Bumped to 2 for the Porter/trigram move, + so every vault re-derives exactly once. + +## [1.9.2] — 2026-08-10 + +### Fixed (an unrecorded geometry counted as a matching one) +1.9.1 recorded the geometry on open and cleared the digests when it changed +— but skipped the clear when there was NO previous record, to avoid a mass +re-index on upgrade. That is precisely the upgrade case: a vault written +before the geometry was tracked was derived by some tokenization nobody can +name. Production upgraded, recorded the new geometry against untouched rows, +and re-indexed nothing. + +Unknown provenance now counts as stale. Derived data is rebuildable; a wrong +assumption of freshness is not. + +## [1.9.1] — 2026-08-10 + +### Fixed (a geometry change never reached the host) +1.9.0 put the tokenizer's geometry into `content_sha`, which was necessary +and not sufficient: the digest is only consulted for notes a HOST decides to +offer, and a host that diffs on timestamps never offers an untouched note. +The production upgrade re-indexed **nothing** — "0 indexed" — and the vault +sat half-converted, its postings derived one way while queries were analysed +another, with every signal saying it was in sync. + +- The geometry is now recorded in `params`. On open, a change clears every + row's `content_sha` (one UPDATE) so `manifest()` reports them as + "indexed, derived state unknown" — the signal a host can actually act on. +- Reopening with the SAME geometry leaves the digests alone, so an ordinary + restart still costs nothing. + +## [1.9.0] — 2026-08-10 + +### Added (Latin morphology — the stream that never had any) +Korean got a guarded 조사/어미 stripper early. The Latin side never did: +character trigrams over every non-Hangul word ≥4 chars stood in for it, so +"browsing" and "browse" met because they share *brow, row, ows*, not because +anything understood them. + +- `latin.py` — Porter's algorithm, deterministic and dictionary-free, on the + same ADDITIVE contract as the Korean stripper: the surface form is indexed + too, so an over-eager conflation costs one posting and never an exact + match. + +### Changed (trigrams move to the stream they belong in) +Measured on a 5,347-note production vault, Latin trigrams were **72.8% of +all postings** — 59.6% of that from fifty boilerplate types (`execution`, +`screen`, `title`, `tags`, …) whose IDF is ~0, so they cost storage and +contributed nothing to ranking. + +- `latin_ngram_min_len` defaults to `0`: out of BM25, exactly where jamo + already sits. `embed_tokens` asks for them explicitly, so fuzzy matching + stays in the recall stream at no posting cost. +- Paired A/B on one frozen snapshot, old defaults vs new: + postings 1,221,979 → **442,609 (−64%)**; English known-item MRR + verbatim 0.583 → 0.579, **re-inflected 0.372 → 0.461 (+24%)**, + typo 0.411 → 0.419; Korean flat. MIRACL-ko floor unchanged and passing. +- `eval/known_item.py` — known-item retrieval over a real vault with + verbatim / re-inflected / typo query flavours, reported per language. + Gold is known by construction, so no judgements are needed and the actual + corpus can be used. + +### Fixed (a tokenizer change left a half-converted index) +`content_sha` covered the embedding geometry but not the tokenizer's. Flip +`latin_stemming` or `char_ngrams` and the digest said "unchanged": nothing +re-indexed, and the stored postings quietly stopped agreeing with how +queries were analysed. Tokenizer settings are now part of the digest, so a +change invalidates exactly the rows it should. + +## [1.8.0] — 2026-08-09 + +### Changed (read/write lock, and a bound on waiting for it) +One mutex guarded everything, so every search queued behind every index — +15.9 ms idle vs 63.6 ms during indexing on a production vault. + +- Readers now run together; writers stay exclusive. FAIR (arrival order), + not writer-preferring: the first cut preferred writers, and against a + continuous indexing load the writer re-queued the instant it released, so + searches never got a turn and the benchmark never finished. Both + starvation directions are now covered by tests. +- `search(timeout=)`, default `SEARCH_TIMEOUT_S = 20 s`, raising the new + `MemoryBusy`. This — not the lock split — is what bounds a WEDGED write: + an exclusive writer that never returns still blocks readers, so the only + remedy is to stop waiting. A caller that can answer without memory should; + one that cannot passes `timeout=None`. +- The lock is NOT reentrant, unlike the `RLock` it replaces. Two call paths + self-deadlocked the moment it was swapped (`feedback`/`learn` → + `trust_feedback`, and `contradictions` → `get_text`); both are now split + into a locked entry point and an unlocked internal. Lazy cache builders + nest, so their build mutex stays reentrant on purpose. + ## [1.6.0] — 2026-07-30 ### Changed (postings storage: integer keys — 6.4× smaller vaults) diff --git a/pyproject.toml b/pyproject.toml index 891c3af..db31e39 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "xgen-agent-memory" -version = "1.7.0" +version = "1.11.0" description = "Synapse — a learnable, lightweight graph-traversal memory engine: BM25 + local static embeddings + typed-edge PageRank + an online-learned ranker, in one SQLite file with zero API calls." readme = "README.md" license = "Apache-2.0" diff --git a/src/xgen_agent_memory/__init__.py b/src/xgen_agent_memory/__init__.py index 0006f9c..d94cf2c 100644 --- a/src/xgen_agent_memory/__init__.py +++ b/src/xgen_agent_memory/__init__.py @@ -14,13 +14,15 @@ """ from .config import SynapseConfig, load_dotenv +from ._rwlock import MemoryBusy from .engine import SearchHit, SynapseMemory from .executor_adapter import SynapseRetriever, SynapseVectorHandle from .ranker import FEATURES -__version__ = "1.7.0" +__version__ = "1.6.0" __all__ = [ + "MemoryBusy", "SynapseMemory", "SynapseConfig", "SearchHit", diff --git a/src/xgen_agent_memory/_rwlock.py b/src/xgen_agent_memory/_rwlock.py new file mode 100644 index 0000000..6b03e91 --- /dev/null +++ b/src/xgen_agent_memory/_rwlock.py @@ -0,0 +1,148 @@ +"""A many-readers/one-writer lock for the engine's in-memory state. + +One mutex over everything is correct and, for a small vault, invisible. It +stops being invisible when a write takes real time: every search queues +behind it. Measured on a production vault, a search cost 15.9 ms idle and +63.6 ms while indexing ran — and when a write wedged entirely (a matmul +spinning inside a broken BLAS thread pool), every read behind it wedged too. + +Readers don't conflict with each other, so letting them run together removes +the queue between searches. + +What this does NOT do, and it matters: an exclusive writer that never +finishes still blocks every reader. Splitting the lock does not shrink the +blast radius of a wedged write — only ``timeout=`` does. Retrieval paths +should pass one and degrade to "no memory" rather than inherit someone +else's hang; that is what turned a single stuck matmul into a 27-hour +outage with every health signal green. + +FAIR (first-come, first-served), which is not the obvious choice and is worth +saying why. The first cut was writer-preferring — any waiting writer held new +readers back — on the reasoning that a read-mostly engine would otherwise +postpone indexing forever. Measured against a continuous indexing load, that +inverted the problem completely: the writer re-queued the instant it released, +so searches never got a turn at all and the benchmark never finished. Neither +side may starve here, so arrival order decides. Readers that queued together +still run together; a reader waits only for writers that arrived *before* it. + +NOT reentrant. The single ``RLock`` it replaces tolerated a locked method +calling another locked method; here that self-deadlocks, so every such pair +must be split into a public locked entry point and an unlocked internal +(``_get_text_unlocked``, ``_trust_feedback_unlocked``). +""" + +from __future__ import annotations + +import threading +import time +from contextlib import contextmanager +from typing import Iterator, List, Optional + + +class MemoryBusy(RuntimeError): + """Raised when a bounded acquisition ran out of patience. + + A caller that can proceed without memory (retrieval on a turn) should + catch this and do so. A caller that cannot should let it propagate — an + error beats a hang, because a hang has no signal at all. + """ + + +class RWLock: + """Many concurrent readers, or one exclusive writer, in arrival order.""" + + __slots__ = ("_cond", "_readers", "_writer", "_ticket", "_waiting_readers", "_waiting_writers") + + def __init__(self) -> None: + self._cond = threading.Condition(threading.Lock()) + self._readers = 0 + self._writer = False + self._ticket = 0 + # Arrival-ordered tickets of everyone still waiting. Comparing a + # waiter's ticket against these is what makes the order fair. + self._waiting_readers: List[int] = [] + self._waiting_writers: List[int] = [] + + # ── acquisition ────────────────────────────────────────────────── + @contextmanager + def read(self, timeout: Optional[float] = None) -> Iterator[None]: + deadline = None if timeout is None else time.monotonic() + timeout + with self._cond: + mine = self._ticket + self._ticket += 1 + self._waiting_readers.append(mine) + try: + # Wait out any writer already running, and any writer that got + # in line first. Writers queued *behind* this reader do not + # hold it back — otherwise a steady write load starves reads. + while self._writer or (self._waiting_writers and self._waiting_writers[0] < mine): + if deadline is None: + self._cond.wait() + continue + left = deadline - time.monotonic() + if left <= 0 or not self._cond.wait(left): + if self._writer or ( + self._waiting_writers and self._waiting_writers[0] < mine + ): + raise MemoryBusy( + f"memory is busy writing; gave up after {timeout:.1f}s" + ) + finally: + self._waiting_readers.remove(mine) + # A writer may have been waiting solely on this reader's + # place in line; leaving the queue changes its answer. + self._cond.notify_all() + self._readers += 1 + try: + yield + finally: + with self._cond: + self._readers -= 1 + if self._readers == 0: + self._cond.notify_all() + + @contextmanager + def write(self, timeout: Optional[float] = None) -> Iterator[None]: + deadline = None if timeout is None else time.monotonic() + timeout + with self._cond: + mine = self._ticket + self._ticket += 1 + self._waiting_writers.append(mine) + try: + # Exclusive: no other writer, no active reader, and nobody + # ahead in line — including readers, so a busy read load + # cannot postpone indexing indefinitely. + def _blocked() -> bool: + return bool( + self._writer + or self._readers + or (self._waiting_writers and self._waiting_writers[0] != mine) + or (self._waiting_readers and self._waiting_readers[0] < mine) + ) + + while _blocked(): + if deadline is None: + self._cond.wait() + continue + left = deadline - time.monotonic() + if (left <= 0 or not self._cond.wait(left)) and _blocked(): + raise MemoryBusy(f"memory is busy; gave up after {timeout:.1f}s") + finally: + self._waiting_writers.remove(mine) + self._cond.notify_all() + self._writer = True + try: + yield + finally: + with self._cond: + self._writer = False + self._cond.notify_all() + + # ── introspection (tests, diagnostics) ─────────────────────────── + @property + def readers(self) -> int: + return self._readers + + @property + def writing(self) -> bool: + return self._writer diff --git a/src/xgen_agent_memory/config.py b/src/xgen_agent_memory/config.py index ca6fc12..d0fc0e1 100644 --- a/src/xgen_agent_memory/config.py +++ b/src/xgen_agent_memory/config.py @@ -87,6 +87,21 @@ class SynapseConfig: suffix_strip: bool = True #: Cross-space syllable bigrams (붙여쓰기 robustness) in the BM25 stream. cross_space: bool = True + #: Minimum length for a NON-Hangul word to also emit character 3-grams + #: into the BM25 stream. ``0`` (default) keeps them out of it entirely. + #: + #: They used to be unconditional and were 72.8% of a production vault's + #: postings — 59.6% of that from fifty boilerplate types (`execution`, + #: `screen`, `title`, `tags`, …) whose IDF is ~0, so they cost storage + #: and contributed nothing to ranking. With a real stemmer in place the + #: measured trade is: postings −64%, known-item MRR +4% verbatim / + #: +2% re-inflected, −12.5% on transposed-character typos. Typo + #: tolerance still lives in the EMBEDDING stream, which keeps them + #: (``embed_tokens`` does not take this knob) — the same split that + #: already keeps jamo out of BM25. + latin_ngram_min_len: int = 0 + #: Porter stemming for Latin-script words. Additive (surface AND stem). + latin_stemming: bool = True #: BM25F-lite: title terms count this many times in the postings. title_boost: float = 2.0 #: Per-document token cap (indexing) / per-query cap. diff --git a/src/xgen_agent_memory/embedder.py b/src/xgen_agent_memory/embedder.py index 2cb95f0..7edb955 100644 --- a/src/xgen_agent_memory/embedder.py +++ b/src/xgen_agent_memory/embedder.py @@ -17,7 +17,9 @@ from __future__ import annotations +import hashlib import io +import threading from typing import Any, Dict, List, Sequence, Tuple import numpy as np @@ -25,6 +27,60 @@ from .tokenizer import fnv1a_pair, tokenize +# ── One table per process, not one per engine ──────────────────────── +# +# The table is `vocab_size × dim × 4B` — 64 MB at the defaults — and a +# host that keeps N sessions resident used to hold N copies of it. +# Measured on production: an engine over an EMPTY vault still cost +# 68 MB, and 6 live vaults held only TWO distinct tables between them. +# That is 64 MB per session spent on identical bytes. +# +# Sharing is safe because the table is never written in place. It is +# read (`table[ids]`) and, when distillation adopts a better one, the +# whole HashEmbedder is REPLACED by a scratch instance holding its own +# fresh array (see `SynapseMemory.distill`) — copy-on-write by +# construction. `setflags(write=False)` turns any future in-place write +# into an immediate error rather than silent cross-session corruption. +# +# Keys are content-determined: generated tables by their generator +# arguments, loaded tables by the digest of the blob they came from. +_TABLE_CACHE: Dict[Any, np.ndarray] = {} +_TABLE_LOCK = threading.Lock() +#: Realistically 1–2 distinct tables exist; the bound is a leak-stop, +#: not a tuning knob. Oldest-first eviction (insertion-ordered dict). +_TABLE_CACHE_MAX = 8 + + +def _shared_table(key: Any, build) -> np.ndarray: + """Return the process-wide table for *key*, building it once.""" + with _TABLE_LOCK: + hit = _TABLE_CACHE.get(key) + if hit is not None: + return hit + # Build OUTSIDE the lock: generating 64 MB takes ~200ms and must not + # serialise every other engine's construction behind it. A duplicate + # build under a race costs one throwaway array, never correctness. + table = build() + table.setflags(write=False) + with _TABLE_LOCK: + existing = _TABLE_CACHE.get(key) + if existing is not None: + return existing + while len(_TABLE_CACHE) >= _TABLE_CACHE_MAX: + _TABLE_CACHE.pop(next(iter(_TABLE_CACHE))) + _TABLE_CACHE[key] = table + return table + + +def shared_table_stats() -> Dict[str, Any]: + """What the process is holding — for host health endpoints.""" + with _TABLE_LOCK: + return { + "tables": len(_TABLE_CACHE), + "bytes": sum(t.nbytes for t in _TABLE_CACHE.values()), + } + + class HashEmbedder: def __init__( self, @@ -35,16 +91,30 @@ def __init__( char_ngrams: Sequence[int] = (2, 3), jamo_ngrams: Sequence[int] = (3, 5), suffix_strip: bool = True, + table: "np.ndarray | None" = None, ) -> None: self.vocab_size = vocab_size self.dim = dim self.char_ngrams = tuple(char_ngrams) self.jamo_ngrams = tuple(jamo_ngrams) self.suffix_strip = suffix_strip - rng = np.random.default_rng(seed) + # fp32 master table; persisted as fp16 to halve disk. Scaled so that # mean-pooled vectors have a sane norm pre-normalization. - self.table = (rng.standard_normal((vocab_size, dim)) / np.sqrt(dim)).astype(np.float32) + # + # Fully determined by (vocab_size, dim, seed) — so every engine + # built with the same three shares one array instead of minting + # its own 64 MB copy of identical numbers. + # `table=` skips generation for callers that already have the real + # one (`loads`). Without it every restore generated a 64 MB table + # only to throw it away on the next line. + def _build() -> np.ndarray: + rng = np.random.default_rng(seed) + return (rng.standard_normal((vocab_size, dim)) / np.sqrt(dim)).astype(np.float32) + + self.table = ( + table if table is not None else _shared_table(("gen", vocab_size, dim, seed), _build) + ) # ── inference ──────────────────────────────────────────────────── def bucket_ids(self, text: str, *, limit: int = 2048) -> np.ndarray: @@ -212,16 +282,24 @@ def loads( ) -> "HashEmbedder": data = np.load(io.BytesIO(blob)) vocab_size, dim = (int(x) for x in data["meta"]) - emb = cls( + # Two sessions that persisted the same table hold the same bytes, + # so the digest of the blob is the identity of the array it + # decodes to. Keying on it means the decode (and the 64 MB) + # happens once per distinct table, not once per session. + digest = hashlib.sha256(blob).digest() + table = _shared_table( + ("blob", digest), + lambda: data["table"].astype(np.float32), + ) + return cls( vocab_size, dim, seed=seed, char_ngrams=char_ngrams, jamo_ngrams=jamo_ngrams, suffix_strip=suffix_strip, + table=table, ) - emb.table = data["table"].astype(np.float32) - return emb def save(self, path: str) -> None: with open(path, "wb") as f: diff --git a/src/xgen_agent_memory/engine.py b/src/xgen_agent_memory/engine.py index ac4f9e2..fd98cf3 100644 --- a/src/xgen_agent_memory/engine.py +++ b/src/xgen_agent_memory/engine.py @@ -15,15 +15,17 @@ from __future__ import annotations import hashlib +import logging import math import random import threading import time from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Sequence +from typing import Any, Dict, List, Optional, Sequence, Tuple import numpy as np +from ._rwlock import RWLock from .bm25 import bm25_scores, term_frequencies, top_n from .config import SynapseConfig from .embedder import HashEmbedder, pack_vec, unpack_vec @@ -38,6 +40,18 @@ from .store import EDGE_COACCESS, EDGE_KNN, EDGE_LINK, EDGE_TAG, Store from .tokenizer import lexical_tokens +logger = logging.getLogger(__name__) + +#: Bump when the MEANING of the tokenizer or embedding pipeline changes in a +#: way the config fields alone do not capture — a new stemmer, a changed +#: n-gram rule, a different normalisation. Every vault then re-derives once. +#: +#: 2: Porter stemming for Latin words; Latin trigrams moved out of BM25. +#: Needed as its own lever because 1.9.0 recorded the *new* geometry +#: against rows derived by the *old* tokenizer — from then on the +#: fingerprints matched and nothing could notice. +_GEOMETRY_VERSION = 2 + _KIND_PRIOR = {"fact": 1.0, "insight": 0.8, "note": 0.5, "digest": 0.4, "turn": 0.2} @@ -69,6 +83,7 @@ def __init__(self, config: Optional[SynapseConfig] = None, **overrides: Any) -> ) self.cfg = cfg self.store = Store(cfg.path) + self._invalidate_on_geometry_change() self.embedder = self._load_embedder() self.ranker = self._load_ranker() # Incremental caches — the difference between O(N) and O(N²) bulk @@ -82,13 +97,77 @@ def __init__(self, config: Optional[SynapseConfig] = None, **overrides: Any) -> #: query_token → {"hash":…, "features": {node_id: np.ndarray}, "shown": [ids]} self._recent_queries: Dict[str, Dict[str, Any]] = {} self._rng = random.Random(cfg.seed) - # One re-entrant lock guards ALL mutable engine state (caches, ranker, - # embedder, _recent_queries, _rng). The Store has its own lock for - # SQLite, but these Python structures are mutated by index/search/ - # feedback/distill/remove and must not interleave across threads — - # ``check_same_thread=False`` means callers CAN hit one instance from - # several threads (e.g. a search turn + a background feedback). - self._lock = threading.RLock() + # Readers run together; writers are exclusive; arrival order decides + # (see `_rwlock` for why fair, not writer-preferring). A wedged write + # still blocks readers — only `search(timeout=)` bounds that. + # + # NOT reentrant, unlike the RLock it replaces: a locked method must + # never call another locked method (see ``_get_text_unlocked``). + self._lock = RWLock() + # `_recent_queries` is touched from BOTH sides — search inserts under + # the read lock, feedback/remove delete under the write lock — so it + # needs a mutex of its own. + self._rq_lock = threading.Lock() + # Derived caches are built lazily, including from read paths where + # several threads may now arrive at once. One builder at a time: + # rebuilding an 8.7k-vector matrix twice is pure waste. + # + # REENTRANT on purpose — the builders nest (`_vector_matrix` needs + # `_vectors`), and a plain mutex there deadlocks a thread against + # itself. This guards duplicated work, not correctness, so + # reentrancy costs nothing. + self._cache_build_lock = threading.RLock() + + def _geometry(self) -> str: + """Everything a stored row's derived state depends on. + + Not the same question as "did the note change" — this is "would the + same note produce different postings and vectors today". Embedding + width, tokenizer streams, stemming: change any of them and every row + in the file is stale, however untouched the notes are. + """ + return ( + f"v{_GEOMETRY_VERSION}:" + f"{self.cfg.dim}:{self.cfg.vocab_size}:" + f"{tuple(self.cfg.char_ngrams)}:{self.cfg.suffix_strip}:" + f"{self.cfg.cross_space}:{self.cfg.latin_stemming}:" + f"{self.cfg.latin_ngram_min_len}" + ) + + def _invalidate_on_geometry_change(self) -> None: + """Clear every digest when the geometry moved, so the host is told. + + Putting the geometry into ``content_sha`` is not enough on its own: + the digest is only consulted for notes a HOST decides to offer, and + a host that diffs on timestamps never offers an untouched note. The + result was a vault half-converted to a new tokenization while every + signal said it was in sync. + + An empty digest in ``manifest()`` is the contract that fixes it — + "indexed, derived state unknown" — and a host reading it re-offers + exactly those notes. One UPDATE, once, on the boot after a change. + """ + current = self._geometry() + stored = self.store.get_param("geometry") + previous = stored.decode("utf-8", "replace") if stored else "" + if previous == current: + return + # No recorded geometry means a vault written before this was tracked + # — which is to say, derived by SOME geometry nobody can name. The + # first version of this treated "unknown" as "fine" to avoid a mass + # re-index on upgrade, and the result was the upgrade re-indexing + # nothing at all: every row kept a digest that described a + # tokenization no longer in use. Derived data is rebuildable and a + # wrong assumption of freshness is not, so unknown counts as stale. + n = self.store.clear_content_shas() + if n: + logger.info( + "synapse: geometry changed (%s → %s) — %d rows marked stale", + previous or "", + current, + n, + ) + self.store.put_param("geometry", current.encode("utf-8")) # ── construction helpers ───────────────────────────────────────── @classmethod @@ -151,123 +230,396 @@ def index( caller happens to have (e.g. a stored API embedding) — used only as a distillation label, never required. """ - with self._lock: - existing = self.store.get_node(node_id) # None ⇒ fresh insert - body = f"{title}\n{text}" if title else text - # Idempotence short-circuit: hosts commonly re-scan their whole - # vault on session wake ("index everything, just in case"). When - # NOTHING that affects derived state changed, a full re-index - # (tokenize + embed + edge derivation + fsync'd transaction) is - # pure waste — measured as the difference between a wake-time - # backfill of a real vault taking minutes vs milliseconds. The - # digest covers every input the derived rows depend on, config - # geometry included (a dim/vocab change must rebuild vectors). - sha_basis = "\x1f".join( - ( - body, - kind, - ",".join(sorted(tags)), - ",".join(sorted(links)), - f"{importance:.6f}", - str(bool(pinned)), - f"{self.cfg.dim}:{self.cfg.vocab_size}", - str(teacher_model) if teacher_vec is not None else "", - ) + with self._lock.write(): + prepared = self._prepare_index( + node_id, + text, + title=title, + kind=kind, + tags=tags, + links=links, + importance=importance, + pinned=pinned, + updated_at=updated_at, + teacher_vec=teacher_vec, + teacher_model=teacher_model, ) - content_sha = hashlib.sha1(sha_basis.encode("utf-8", "replace")).hexdigest() - if existing is not None and existing.get("content_sha") == content_sha: - if updated_at is not None and updated_at != existing.get("updated_at"): - self.store.touch_node(node_id, updated_at) + if prepared is None: return - # ── compute everything FIRST (reads + numpy), commit ONCE ── - # LEXICAL stream → BM25 postings (words + stems + syllable bigrams - # + cross-space bigrams). The jamo-augmented EMBEDDING stream lives - # only inside the hash embedder. - tok_kw = dict( - char_ngrams=self.cfg.char_ngrams, - suffix_strip=self.cfg.suffix_strip, - cross_space=self.cfg.cross_space, - ) - tokens = lexical_tokens(body, limit=self.cfg.max_doc_tokens, **tok_kw) - # BM25F-lite: title terms weigh title_boost× (the title already - # appears once inside `body`, so the extra weight is title_boost−1). - tf = term_frequencies(tokens) - if title and self.cfg.title_boost > 1.0: - extra = self.cfg.title_boost - 1.0 - for t in set(lexical_tokens(title, limit=64, **tok_kw)): - tf[t] = tf.get(t, 0.0) + extra - vec = self.embedder.embed(body, limit=self.cfg.max_doc_tokens) - # Edges. LINK is stored ONE-directional and symmetrized at query - # time (build_type_adjacency), so a changed link set can't orphan a - # reverse edge. TAG/KNN derived from current graph state. - tag_edges = derive_tag_edges( - self._tags_map(), self._n_docs(), node_id, tags, fanout=self.cfg.tag_fanout - ) - knn_edges = derive_knn_edges( - vec, - self._vectors(), - node_id, - k=self.cfg.knn_edges, - min_sim=self.cfg.knn_min_sim, - sample_cap=self.cfg.knn_sample_cap, + row, cache_delta, touch = prepared + if row is None: + if touch is not None: + self.store.touch_node(node_id, touch) + return + self.store.index_atomic(**row) + self._apply_cache_delta(cache_delta) + + _MISSING = object() + + def _prepare_index( + self, + node_id: str, + text: str, + *, + title: str = "", + kind: str = "note", + tags: Sequence[str] = (), + links: Sequence[str] = (), + importance: float = 1.0, + pinned: bool = False, + updated_at: Optional[float] = None, + teacher_vec: Optional[Sequence[float]] = None, + teacher_model: str = "", + existing: Any = _MISSING, + ) -> Optional[tuple]: + """Derive one memory's state WITHOUT writing anything. + + Returns ``(row, cache_delta, touch)``: + * ``row`` — kwargs for ``store.index_atomic``/``index_atomic_many``, + or ``None`` when the digest says nothing derived would change; + * ``cache_delta`` — what the in-memory caches must learn once the + write commits; + * ``touch`` — the timestamp to record when only the clock moved. + + Split out of ``index()`` so a batch can compute many of these and then + commit them together. Compute is 2-3 ms per note; the commit was + measured at 43.7 ms, so the split is where a catch-up stops being + dominated by fsync. + + *existing* lets a batch caller supply the node row it already fetched + — otherwise this issues one SELECT per note, which is the same + per-item cost the batch exists to remove. + """ + if existing is self._MISSING: + existing = self.store.get_node(node_id) # None ⇒ fresh insert + body = f"{title}\n{text}" if title else text + # Idempotence short-circuit: hosts commonly re-scan their whole + # vault on session wake ("index everything, just in case"). When + # NOTHING that affects derived state changed, a full re-index + # (tokenize + embed + edge derivation + fsync'd transaction) is + # pure waste — measured as the difference between a wake-time + # backfill of a real vault taking minutes vs milliseconds. The + # digest covers every input the derived rows depend on, config + # geometry included (a dim/vocab change must rebuild vectors). + sha_basis = "\x1f".join( + ( + body, + kind, + ",".join(sorted(tags)), + ",".join(sorted(links)), + f"{importance:.6f}", + str(bool(pinned)), + f"{self.cfg.dim}:{self.cfg.vocab_size}", + # Tokenizer geometry belongs in the digest for the same reason + # the embedding geometry does: change it and every derived row + # is stale. Without this, flipping `latin_stemming` or + # `char_ngrams` left the OLD postings in place — the digest said + # "unchanged", nothing re-indexed, and the vault silently became + # a mix of two tokenizations that no longer agreed with the + # query view. + f"{tuple(self.cfg.char_ngrams)}:{self.cfg.suffix_strip}:" + f"{self.cfg.cross_space}:{self.cfg.latin_stemming}:" + f"{self.cfg.latin_ngram_min_len}", + str(teacher_model) if teacher_vec is not None else "", ) - teacher = None - if teacher_vec is not None: - tv = np.asarray(teacher_vec, dtype=np.float32) - teacher = (teacher_model, pack_vec(tv), int(tv.shape[0])) - text_param = ( - (f"text:{node_id}", body[: self.cfg.store_text_maxlen].encode("utf-8")) - if self.cfg.store_text + ) + content_sha = hashlib.sha1(sha_basis.encode("utf-8", "replace")).hexdigest() + if existing is not None and existing.get("content_sha") == content_sha: + touch = ( + updated_at + if updated_at is not None and updated_at != existing.get("updated_at") else None ) + return None, None, touch + # ── compute everything FIRST (reads + numpy), commit ONCE ── + # LEXICAL stream → BM25 postings (words + stems + syllable bigrams + # + cross-space bigrams). The jamo-augmented EMBEDDING stream lives + # only inside the hash embedder. + tok_kw = dict( + char_ngrams=self.cfg.char_ngrams, + suffix_strip=self.cfg.suffix_strip, + cross_space=self.cfg.cross_space, + latin_ngram_min_len=self.cfg.latin_ngram_min_len, + latin_stemming=self.cfg.latin_stemming, + ) + tokens = lexical_tokens(body, limit=self.cfg.max_doc_tokens, **tok_kw) + # BM25F-lite: title terms weigh title_boost× (the title already + # appears once inside `body`, so the extra weight is title_boost−1). + tf = term_frequencies(tokens) + if title and self.cfg.title_boost > 1.0: + extra = self.cfg.title_boost - 1.0 + for t in set(lexical_tokens(title, limit=64, **tok_kw)): + tf[t] = tf.get(t, 0.0) + extra + vec = self.embedder.embed(body, limit=self.cfg.max_doc_tokens) + # Edges. LINK is stored ONE-directional and symmetrized at query + # time (build_type_adjacency), so a changed link set can't orphan a + # reverse edge. TAG/KNN derived from current graph state. + tag_edges = derive_tag_edges( + self._tags_map(), self._n_docs(), node_id, tags, fanout=self.cfg.tag_fanout + ) + knn_edges = derive_knn_edges( + vec, + self._vectors(), + node_id, + k=self.cfg.knn_edges, + min_sim=self.cfg.knn_min_sim, + sample_cap=self.cfg.knn_sample_cap, + ) + teacher = None + if teacher_vec is not None: + tv = np.asarray(teacher_vec, dtype=np.float32) + teacher = (teacher_model, pack_vec(tv), int(tv.shape[0])) + text_param = ( + (f"text:{node_id}", body[: self.cfg.store_text_maxlen].encode("utf-8")) + if self.cfg.store_text + else None + ) - # ── one atomic transaction: node + postings + vector + edges ── - self.store.index_atomic( - node_id, - kind=kind, - title=title, - tags=tags, - text_len=len(tokens), - updated_at=updated_at or time.time(), - pinned=pinned, - importance=importance, - tf=tf, - vec=pack_vec(vec), - dim=self.cfg.dim, - edges=[ - (EDGE_LINK, [(dst, 1.0) for dst in links]), - (EDGE_TAG, tag_edges), - (EDGE_KNN, knn_edges), - ], - teacher=teacher, - text_param=text_param, - content_sha=content_sha, - ) + row = dict( + node_id=node_id, + kind=kind, + title=title, + tags=tags, + text_len=len(tokens), + updated_at=updated_at or time.time(), + pinned=pinned, + importance=importance, + tf=tf, + vec=pack_vec(vec), + dim=self.cfg.dim, + edges=[ + (EDGE_LINK, [(dst, 1.0) for dst in links]), + (EDGE_TAG, tag_edges), + (EDGE_KNN, knn_edges), + ], + teacher=teacher, + text_param=text_param, + content_sha=content_sha, + ) + cache_delta = dict( + node_id=node_id, + vec=vec, + text_len=len(tokens), + tags=tuple(tags), + old_tags=tuple(existing["tags"]) if existing is not None else (), + ) + return row, cache_delta, None - # ── cache maintenance (after the commit succeeds) ── - if self._vec_cache is not None: - self._vec_cache[node_id] = vec + def _apply_cache_delta(self, delta: Dict[str, Any]) -> None: + """Teach the in-memory caches what a committed write changed. + + Separate from the write so a batch applies one delta per note AFTER + its chunk commits — a cache that learned about a rolled-back write + would answer for rows the database does not have. + """ + node_id = delta["node_id"] + if self._vec_cache is not None: + self._vec_cache[node_id] = delta["vec"] + self._vec_matrix = None + if self._doclen_cache is not None: + self._doclen_cache[node_id] = delta["text_len"] + # Tag cache: INCREMENTAL on both insert and re-index. On a re-index + # remove the node from its OLD tags then add the new ones — + # O(#tags), NOT an O(N) full-cache rebuild (which made re-indexing a + # big corpus O(N²): 541 ms/re-index at 40k). + if self._tag_cache is not None: + for t in delta["old_tags"]: + lst = self._tag_cache.get(t) + if lst and node_id in lst: + lst.remove(node_id) + for t in delta["tags"]: + lst = self._tag_cache.setdefault(t, []) + if node_id not in lst: + lst.append(node_id) + self._adj_cache.clear() + + def index_many( + self, + items: Sequence[Dict[str, Any]], + *, + chunk_size: int = 200, + ) -> Dict[str, int]: + """Index many memories, amortising the commit across a chunk. + + Each item is a mapping with ``node_id`` and ``text`` plus any keyword + ``index()`` accepts. Unchanged content short-circuits exactly as it + does one at a time — the digest decides, not the caller. + + Why this exists: ``index()`` in a loop pays one fsync per note. On a + real deployment that fsync measured 43.7 ms against 2-3 ms of actual + indexing, so a catch-up was 97% commit. Here the whole chunk computes + first, then commits once. + + kNN edges still see prior members of the same batch, because the + in-memory vector cache is updated as each note is prepared — the + result is identical to indexing them one by one, minus the fsyncs. + + Returns ``{"indexed": n, "touched": n, "skipped": n}``. + """ + out = {"indexed": 0, "touched": 0, "skipped": 0} + if not items: + return out + with self._lock.write(): + for start in range(0, len(items), max(1, chunk_size)): + chunk = items[start : start + max(1, chunk_size)] + # One SELECT for the whole chunk instead of one per note — + # otherwise the batch still pays N round trips to discover + # that N notes are unchanged. + known = { + n["id"]: n + for n in self.store.nodes([it["node_id"] for it in chunk if it.get("node_id")]) + } + rows: List[Dict[str, Any]] = [] + deltas: List[Dict[str, Any]] = [] + touches: List[Tuple[str, float]] = [] + for item in chunk: + kwargs = dict(item) + node_id = kwargs.pop("node_id") + text = kwargs.pop("text", "") + prepared = self._prepare_index( + node_id, text, existing=known.get(node_id), **kwargs + ) + row, delta, touch = prepared + if row is None: + if touch is not None: + touches.append((node_id, touch)) + out["touched"] += 1 + else: + out["skipped"] += 1 + continue + rows.append(row) + deltas.append(delta) + # Visible to the NEXT note's kNN in this same chunk. + if self._vec_cache is not None: + self._vec_cache[node_id] = delta["vec"] + if touches: + self.store.touch_nodes(touches) + if rows: + self.store.index_atomic_many(rows) + for delta in deltas: + self._apply_cache_delta(delta) + out["indexed"] += len(rows) + return out + + def manifest(self) -> Dict[str, Tuple[float, str]]: + """``{node_id: (updated_at, content_sha)}`` — what is already indexed. + + The primitive an incremental host needs: with it, "what changed" is a + set difference against metadata the host already holds; without it the + only way to ask is to re-read every note and re-derive its digest. + """ + with self._lock.read(): + return self.store.manifest() + + def catalog_counts( + self, *, by: str = "kind", kind: Optional[str] = None + ) -> List[Tuple[str, int]]: + """``[(key, count)]`` by ``kind`` or by calendar ``day``. + + The cheap end of a progressive browser: "how much is there" without + reading anything. See ``Store.catalog_counts``. + """ + with self._lock.read(): + return self.store.catalog_counts(by=by, kind=kind) + + def catalog_page( + self, + *, + day: Optional[str] = None, + kind: Optional[str] = None, + limit: int = 100, + offset: int = 0, + ) -> List[Dict[str, Any]]: + """One page of note metadata — no bodies. See ``Store.catalog_page``.""" + with self._lock.read(): + return self.store.catalog_page(day=day, kind=kind, limit=limit, offset=offset) + + def neighbourhood( + self, + node_ids: Sequence[str], + *, + depth: int = 1, + max_nodes: int = 400, + max_edges: int = 4000, + ) -> Dict[str, Any]: + """The subgraph around *node_ids* — nodes + typed edges, both capped. + + A graph view is read at a screen's worth of detail; shipping the + whole vault (5,384 nodes, 4.3 MB) is a download, not a view. Depth is + expanded breadth-first and stops at ``max_nodes``, so the caller gets + a bounded answer instead of an unbounded one truncated by whatever + gives out first. + """ + seed = [n for n in node_ids if n] + if not seed: + return {"nodes": [], "edges": [], "truncated": False} + with self._lock.read(): + frontier = list(dict.fromkeys(seed)) + seen = dict.fromkeys(frontier) + edges: List[Dict[str, Any]] = [] + truncated = False + for _ in range(max(0, depth)): + if not frontier or len(seen) >= max_nodes: + break + rows = self.store.edges_touching(frontier, limit=max_edges) + nxt: List[str] = [] + for src, dst, etype, w in rows: + edges.append({"src": src, "dst": dst, "type": etype, "w": w}) + for end in (src, dst): + if end not in seen: + if len(seen) >= max_nodes: + truncated = True + continue + seen[end] = None + nxt.append(end) + frontier = nxt + nodes = self.store.nodes(list(seen)) + return { + "nodes": [ + { + "id": n["id"], + "kind": n["kind"], + "title": n["title"], + "updated_at": n["updated_at"], + "text_len": n["text_len"], + "pinned": n["pinned"], + "importance": n["importance"], + } + for n in nodes + ], + "edges": edges, + "truncated": truncated or len(edges) >= max_edges, + } + + def remove_many(self, node_ids: Sequence[str]) -> int: + """Delete many memories in one transaction. + + Reaping is inherently bulk — a vault whose deletions never reached the + index presents thousands at once — and one fsync per node would turn + the cleanup into its own outage. + """ + ids = [n for n in node_ids if n] + if not ids: + return 0 + with self._lock.write(): + removed = self.store.remove_nodes(ids) + gone = set(ids) + for node_id in ids: + if self._vec_cache is not None: + self._vec_cache.pop(node_id, None) + if self._doclen_cache is not None: + self._doclen_cache.pop(node_id, None) self._vec_matrix = None - if self._doclen_cache is not None: - self._doclen_cache[node_id] = len(tokens) - # Tag cache: INCREMENTAL on both insert and re-index. On a re-index - # remove the node from its OLD tags (read off `existing`) then add - # the new ones — O(#tags), NOT an O(N) full-cache rebuild (which - # made re-indexing a big corpus O(N²): 541 ms/re-index at 40k). - if self._tag_cache is not None: - if existing is not None: - for t in existing["tags"]: - lst = self._tag_cache.get(t) - if lst and node_id in lst: - lst.remove(node_id) - for t in tags: - lst = self._tag_cache.setdefault(t, []) - if node_id not in lst: - lst.append(node_id) + self._tag_cache = None self._adj_cache.clear() + with self._rq_lock: + for tok in list(self._recent_queries): + if gone & set(self._recent_queries[tok]["features"]): + del self._recent_queries[tok] + return removed def remove(self, node_id: str) -> None: - with self._lock: + with self._lock.write(): self.store.remove_node(node_id) self.store.delete_param(f"text:{node_id}") if self._vec_cache is not None: @@ -280,15 +632,30 @@ def remove(self, node_id: str) -> None: # Drop any pending feedback tokens that reference this node — else # feedback() could reinforce co-access edges or re-insert feedback # rows pointing at a now-deleted node. - for tok in list(self._recent_queries): - if node_id in self._recent_queries[tok]["features"]: - del self._recent_queries[tok] + with self._rq_lock: + for tok in list(self._recent_queries): + if node_id in self._recent_queries[tok]["features"]: + del self._recent_queries[tok] # ── read path ──────────────────────────────────────────────────── + #: Default patience for a RETRIEVAL. A turn can proceed without memory; + #: it cannot proceed while blocked behind a write that never returns. + #: Long enough that a normal index chunk is simply waited out. + SEARCH_TIMEOUT_S: Optional[float] = 20.0 + def search( - self, query: str, *, top_k: Optional[int] = None, kinds: Optional[Sequence[str]] = None + self, + query: str, + *, + top_k: Optional[int] = None, + kinds: Optional[Sequence[str]] = None, + timeout: Any = _MISSING, ) -> List[SearchHit]: - with self._lock: + """Retrieve. Raises ``MemoryBusy`` if a write holds the engine longer + than ``timeout`` (default ``SEARCH_TIMEOUT_S``; pass ``None`` to wait + forever, which is what the callers who cannot degrade should do).""" + wait = self.SEARCH_TIMEOUT_S if timeout is self._MISSING else timeout + with self._lock.read(timeout=wait): return self._search(query, top_k=top_k, kinds=kinds) def _search( @@ -424,13 +791,16 @@ def _search( token = hashlib.sha1(f"{query}|{now}".encode()).hexdigest()[:16] for h in result: h.query_token = token - self._recent_queries[token] = { - "hash": hashlib.sha1(query.encode()).hexdigest()[:16], - "features": {h.id: feats[h.id] for h in result if h.id in feats}, - "shown": [h.id for h in result], - } - if len(self._recent_queries) > 64: - self._recent_queries.pop(next(iter(self._recent_queries))) + # Search runs under the READ lock, so several of these can land at + # once; the bounded dict needs its own mutex. + with self._rq_lock: + self._recent_queries[token] = { + "hash": hashlib.sha1(query.encode()).hexdigest()[:16], + "features": {h.id: feats[h.id] for h in result if h.id in feats}, + "shown": [h.id for h in result], + } + if len(self._recent_queries) > 64: + self._recent_queries.pop(next(iter(self._recent_queries))) self.store.touch_access([h.id for h in result], ts=now) return result @@ -449,7 +819,7 @@ def feedback( co-access reinforcement, pairwise ranker SGD (event + small replay), and persists everything. Cost: microseconds-to-ms. """ - with self._lock: + with self._lock.write(): return self._feedback( query_token, used_ids=used_ids, ignored_ids=ignored_ids, label_src=label_src ) @@ -462,7 +832,8 @@ def _feedback( ignored_ids: Optional[Sequence[str]] = None, label_src: str = "implicit", ) -> Dict[str, float]: - q = self._recent_queries.get(query_token) + with self._rq_lock: + q = self._recent_queries.get(query_token) if q is None: return {"applied": 0.0} used = [i for i in used_ids if i in q["features"]] @@ -485,7 +856,7 @@ def _feedback( self._adj_cache.pop(EDGE_COACCESS, None) # ①b Trust — same policy as learn(): used items gain reliability. for nid in used: - self.trust_feedback(nid, True) + self._trust_feedback_unlocked(nid, True) # ② Ranker. Split by QUERY: a deterministic ~25% of queries are HELD # OUT — refereed for the blend gate, never trained on (and never added @@ -571,16 +942,28 @@ def trust_feedback( helpful +trust_helpful, unhelpful −trust_unhelpful (2× by default). The stored value is decayed to *now* first, so stale trust doesn't anchor the update. Returns the new trust, or None if unknown id.""" - with self._lock: - node = self.store.get_node(node_id) - if node is None: - return None - ts = now if now is not None else time.time() - eff = self._effective_trust(node, ts) - delta = self.cfg.trust_helpful if helpful else -self.cfg.trust_unhelpful - t = min(1.0, max(0.0, eff + delta)) - self.store.set_trust(node_id, t, ts) - return t + with self._lock.write(): + return self._trust_feedback_unlocked(node_id, helpful, now=now) + + def _trust_feedback_unlocked( + self, node_id: str, helpful: bool, *, now: Optional[float] = None + ) -> Optional[float]: + """The body of ``trust_feedback`` for callers ALREADY holding the + write lock. + + The lock is not reentrant. ``_feedback`` and ``_learn`` both update + per-item trust as part of one learning step, and both already run + under it — calling the public method there self-deadlocks. + """ + node = self.store.get_node(node_id) + if node is None: + return None + ts = now if now is not None else time.time() + eff = self._effective_trust(node, ts) + delta = self.cfg.trust_helpful if helpful else -self.cfg.trust_unhelpful + t = min(1.0, max(0.0, eff + delta)) + self.store.set_trust(node_id, t, ts) + return t def learn( self, @@ -617,7 +1000,7 @@ def learn( "every result" as positive — that would train the ranker to rubber- stamp whatever the current retriever already surfaces. """ - with self._lock: + with self._lock.write(): return self._learn(query_key, positives, negatives, label_src) def _learn( @@ -649,7 +1032,7 @@ def _learn( # evidence the memory itself is wrong — explicit unhelpful feedback # goes through trust_feedback().) for pid in pos_ids: - self.trust_feedback(pid, True) + self._trust_feedback_unlocked(pid, True) # ② Ranker — same query-level hold-out split as feedback(): a # deterministic ~25% of queries referee the blend gate (never trained). @@ -696,7 +1079,7 @@ def search_join( candidates by ``min`` across entities (AND) or ``mean`` (OR).""" if mode not in ("and", "or"): raise ValueError(f"mode must be 'and' or 'or', got {mode!r}") - with self._lock: + with self._lock.read(): top_k = self.cfg.top_k if top_k is None else max(0, top_k) ents = [e for e in entities if e and e.strip()] if not ents: @@ -827,11 +1210,11 @@ def contradictions( without stored text fall back to their title). Candidate generation reuses BM25 with the node's own words as the query, so cost is one search, not O(N²).""" - with self._lock: + with self._lock.read(): node = self.store.get_node(node_id) if node is None: return [] - text = self.get_text(node_id) or node.get("title") or "" + text = self._get_text_unlocked(node_id) or node.get("title") or "" if not text: return [] words = set( @@ -863,7 +1246,7 @@ def contradictions( out: List[Dict[str, Any]] = [] for cid in cand_ids: - ctext = self.get_text(cid) + ctext = self._get_text_unlocked(cid) if ctext is None: cnode = self.store.get_node(cid) ctext = (cnode.get("title") if cnode else "") or "" @@ -914,7 +1297,7 @@ def distill(self, *, epochs: Optional[int] = None) -> Dict[str, Any]: swap the table and re-embed every stored-text node in one transaction. Needs ``store_text=True`` (the whole corpus must be re-embeddable) and enough teacher pairs. Bounded batch job — run at idle / close.""" - with self._lock: + with self._lock.write(): if not self.cfg.store_text: return {"trained": 0.0, "reason_no_text": 1.0} teachers = self.store.teachers() @@ -966,7 +1349,7 @@ def distill(self, *, epochs: Optional[int] = None) -> Dict[str, Any]: # ── misc ───────────────────────────────────────────────────────── def stats(self) -> Dict[str, Any]: - with self._lock: + with self._lock.read(): return { "nodes": self.store.count_nodes(), "feedback_rows": self.store.feedback_count(), @@ -984,7 +1367,7 @@ def stats(self) -> Dict[str, Any]: } def close(self) -> None: - with self._lock: + with self._lock.write(): self._persist_models() self.store.close() @@ -996,28 +1379,37 @@ def __exit__(self, *exc: Any) -> None: # ── internals ──────────────────────────────────────────────────── def _vectors(self) -> Dict[str, np.ndarray]: + # Double-checked: several readers can now arrive here at once, and + # materialising 8.7k vectors twice is pure waste. The second check + # inside the mutex is what makes the first one safe to skip. if self._vec_cache is None: - self._vec_cache = { - nid: unpack_vec(blob, dim) for nid, dim, blob in self.store.all_vectors() - } + with self._cache_build_lock: + if self._vec_cache is None: + self._vec_cache = { + nid: unpack_vec(blob, dim) for nid, dim, blob in self.store.all_vectors() + } return self._vec_cache def _vector_matrix(self) -> tuple: """(ids, row-stacked matrix) — cached; rebuilt lazily after writes.""" if self._vec_matrix is None: - vectors = self._vectors() - ids = list(vectors.keys()) - matrix = ( - np.stack([vectors[i] for i in ids]) - if ids - else np.zeros((0, self.cfg.dim), dtype=np.float32) - ) - self._vec_matrix = (ids, matrix) + with self._cache_build_lock: + if self._vec_matrix is None: + vectors = self._vectors() + ids = list(vectors.keys()) + matrix = ( + np.stack([vectors[i] for i in ids]) + if ids + else np.zeros((0, self.cfg.dim), dtype=np.float32) + ) + self._vec_matrix = (ids, matrix) return self._vec_matrix def _doclens(self) -> Dict[str, int]: if self._doclen_cache is None: - self._doclen_cache = self.store.doc_lens() + with self._cache_build_lock: + if self._doclen_cache is None: + self._doclen_cache = self.store.doc_lens() return self._doclen_cache def _n_docs(self) -> int: @@ -1025,11 +1417,13 @@ def _n_docs(self) -> int: def _tags_map(self) -> Dict[str, List[str]]: if self._tag_cache is None: - tag_map: Dict[str, List[str]] = {} - for node in self.store.nodes(): - for t in node["tags"]: - tag_map.setdefault(t, []).append(node["id"]) - self._tag_cache = tag_map + with self._cache_build_lock: + if self._tag_cache is None: + tag_map: Dict[str, List[str]] = {} + for node in self.store.nodes(): + for t in node["tags"]: + tag_map.setdefault(t, []).append(node["id"]) + self._tag_cache = tag_map return self._tag_cache def _adjacencies(self) -> Dict[int, dict]: @@ -1037,10 +1431,15 @@ def _adjacencies(self) -> Dict[int, dict]: for etype in (EDGE_LINK, EDGE_TAG, EDGE_KNN, EDGE_COACCESS): cached = self._adj_cache.get(etype) if cached is None: - cached = build_type_adjacency( - self.store.edges_by_type(etype), etype, coaccess_decay=self.cfg.hebb_decay - ) - self._adj_cache[etype] = cached + with self._cache_build_lock: + cached = self._adj_cache.get(etype) + if cached is None: + cached = build_type_adjacency( + self.store.edges_by_type(etype), + etype, + coaccess_decay=self.cfg.hebb_decay, + ) + self._adj_cache[etype] = cached out[etype] = cached return out @@ -1064,9 +1463,18 @@ def get_text(self, node_id: str) -> Optional[str]: ``store_text=True``. Lets a host return the actual text alongside a search hit — e.g. to fill a retrieval result's ``content`` — without keeping the corpus in a second place.""" - with self._lock: - blob = self.store.get_param(f"text:{node_id}") - return blob.decode("utf-8", "replace") if blob else None + with self._lock.read(): + return self._get_text_unlocked(node_id) + + def _get_text_unlocked(self, node_id: str) -> Optional[str]: + """The body of ``get_text`` for callers ALREADY holding the lock. + + The lock is not reentrant, so ``contradictions`` — which runs under + the read lock and needs several bodies — must not call the public + method or it deadlocks against the next waiting writer. + """ + blob = self.store.get_param(f"text:{node_id}") + return blob.decode("utf-8", "replace") if blob else None def _save_text_for_distill(self, node_id: str, body: str) -> None: # Distillation needs the text back; store a bounded copy in params-space. diff --git a/src/xgen_agent_memory/latin.py b/src/xgen_agent_memory/latin.py new file mode 100644 index 0000000..eaa707d --- /dev/null +++ b/src/xgen_agent_memory/latin.py @@ -0,0 +1,216 @@ +"""Latin-script morphology — the counterpart to ``hangul.py``. + +Korean got a guarded 조사/어미 stripper early because the evidence for it was +unambiguous. The Latin side never got one, and character trigrams over every +word ≥4 chars stood in for it: "browsing" and "browse" matched because they +share *brow, row, ows*, not because anything understood them. That worked, at +a price — measured on a production vault those trigrams were **72.8% of all +postings**, and the morphology they approximated still leaked: a query +re-inflected off its source note lost 22% of its MRR (0.551 → 0.429). + +So: a real stemmer. Porter's algorithm, because it is the one every IR system +has agreed on for forty years, is deterministic, needs no dictionary, and its +aggressiveness is the right trade here — this stream is ADDITIVE. The surface +form is indexed too, so an over-eager conflation ("operate"/"operator" both → +"oper") adds recall without taking exact matching away. That is the same +contract the Korean stripper documents: a wrong strip only adds one noisy +term. + +Reference: M.F. Porter, "An algorithm for suffix stripping" (1980). +""" + +from __future__ import annotations + +_VOWELS = frozenset("aeiou") + +#: Below this length a stem is not worth having: three-letter words are +#: already their own stems, and stripping them produces collisions +#: ("ads"→"ad", "his"→"hi") that only add noise. +MIN_WORD_LEN = 4 + + +def _is_consonant(word: str, i: int) -> bool: + ch = word[i] + if ch in _VOWELS: + return False + if ch == "y": + # y is a consonant at the start and after a vowel ("toy"), a vowel + # after a consonant ("happy"). + return i == 0 or not _is_consonant(word, i - 1) + return True + + +def _measure(stem: str) -> int: + """Porter's *m*: how many vowel-consonant sequences the stem contains. + + It is the algorithm's stand-in for "is there enough word left" — every + rule below is conditioned on it, which is what stops ``-ate`` from + eating half of ``plate``. + """ + m = 0 + i = 0 + n = len(stem) + # skip leading consonants + while i < n and _is_consonant(stem, i): + i += 1 + while i < n: + while i < n and not _is_consonant(stem, i): + i += 1 + if i >= n: + break + m += 1 + while i < n and _is_consonant(stem, i): + i += 1 + return m + + +def _has_vowel(stem: str) -> bool: + return any(not _is_consonant(stem, i) for i in range(len(stem))) + + +def _ends_double_consonant(stem: str) -> bool: + return len(stem) >= 2 and stem[-1] == stem[-2] and _is_consonant(stem, len(stem) - 1) + + +def _ends_cvc(stem: str) -> bool: + """consonant-vowel-consonant where the last is not w, x or y — + the shape that wants an ``e`` back ("hop" → "hope").""" + n = len(stem) + if n < 3: + return False + return ( + _is_consonant(stem, n - 3) + and not _is_consonant(stem, n - 2) + and _is_consonant(stem, n - 1) + and stem[-1] not in "wxy" + ) + + +_STEP2 = ( + ("ational", "ate"), + ("tional", "tion"), + ("enci", "ence"), + ("anci", "ance"), + ("izer", "ize"), + ("abli", "able"), + ("alli", "al"), + ("entli", "ent"), + ("eli", "e"), + ("ousli", "ous"), + ("ization", "ize"), + ("ation", "ate"), + ("ator", "ate"), + ("alism", "al"), + ("iveness", "ive"), + ("fulness", "ful"), + ("ousness", "ous"), + ("aliti", "al"), + ("iviti", "ive"), + ("biliti", "ble"), +) + +_STEP3 = ( + ("icate", "ic"), + ("ative", ""), + ("alize", "al"), + ("iciti", "ic"), + ("ical", "ic"), + ("ful", ""), + ("ness", ""), +) + +_STEP4 = ( + "al", + "ance", + "ence", + "er", + "ic", + "able", + "ible", + "ant", + "ement", + "ment", + "ent", + "ou", + "ism", + "ate", + "iti", + "ous", + "ive", + "ize", +) + + +def stem(word: str) -> str: + """Porter stem of *word*. Returns it unchanged when nothing applies. + + Callers index BOTH the surface form and this, so a conflation that is + too eager costs one extra posting, never a missed exact match. + """ + w = word.lower() + if len(w) < MIN_WORD_LEN or not w.isalpha(): + return w + + # ── 1a: plurals ────────────────────────────────────────────────── + if w.endswith("sses"): + w = w[:-2] + elif w.endswith("ies"): + w = w[:-2] + elif w.endswith("ss"): + pass + elif w.endswith("s"): + w = w[:-1] + + # ── 1b: -eed / -ed / -ing ──────────────────────────────────────── + step1b_hit = False + if w.endswith("eed"): + if _measure(w[:-3]) > 0: + w = w[:-1] + elif w.endswith("ed") and _has_vowel(w[:-2]): + w = w[:-2] + step1b_hit = True + elif w.endswith("ing") and _has_vowel(w[:-3]): + w = w[:-3] + step1b_hit = True + if step1b_hit: + if w.endswith(("at", "bl", "iz")): + w += "e" + elif _ends_double_consonant(w) and not w.endswith(("l", "s", "z")): + w = w[:-1] + elif _measure(w) == 1 and _ends_cvc(w): + w += "e" + + # ── 1c: terminal y → i ─────────────────────────────────────────── + if w.endswith("y") and _has_vowel(w[:-1]): + w = w[:-1] + "i" + + # ── 2 / 3: derivational suffixes, longest match first ──────────── + for suffix, repl in sorted(_STEP2, key=lambda p: -len(p[0])): + if w.endswith(suffix) and _measure(w[: len(w) - len(suffix)]) > 0: + w = w[: len(w) - len(suffix)] + repl + break + for suffix, repl in sorted(_STEP3, key=lambda p: -len(p[0])): + if w.endswith(suffix) and _measure(w[: len(w) - len(suffix)]) > 0: + w = w[: len(w) - len(suffix)] + repl + break + + # ── 4: strip the remainder when there is enough word left ──────── + for suffix in sorted(_STEP4, key=len, reverse=True): + if w.endswith(suffix): + base = w[: len(w) - len(suffix)] + if _measure(base) > 1: + if suffix != "ion" or base.endswith(("s", "t")): + w = base + break + if w.endswith("ion") and _measure(w[:-3]) > 1 and w[-4:-3] in ("s", "t"): + w = w[:-3] + + # ── 5: tidy up ─────────────────────────────────────────────────── + if w.endswith("e"): + m = _measure(w[:-1]) + if m > 1 or (m == 1 and not _ends_cvc(w[:-1])): + w = w[:-1] + if _measure(w) > 1 and _ends_double_consonant(w) and w.endswith("l"): + w = w[:-1] + + return w diff --git a/src/xgen_agent_memory/store.py b/src/xgen_agent_memory/store.py index 25a24c5..d47d58f 100644 --- a/src/xgen_agent_memory/store.py +++ b/src/xgen_agent_memory/store.py @@ -182,6 +182,19 @@ def touch_node(self, node_id: str, updated_at: float) -> None: lambda c: c.execute("UPDATE nodes SET updated_at=? WHERE id=?", (updated_at, node_id)) ) + def touch_nodes(self, pairs: Sequence[Tuple[str, float]]) -> int: + """Batch of the above, one transaction. + + A catch-up is mostly touches — content is unchanged, only the host's + clock moved. Paying an fsync each would make the cheap path the + expensive one. + """ + rows = [(ts, nid) for nid, ts in pairs] + if not rows: + return 0 + self._write(lambda c: c.executemany("UPDATE nodes SET updated_at=? WHERE id=?", rows)) + return len(rows) + def index_atomic( self, node_id: str, @@ -205,55 +218,123 @@ def index_atomic( teacher / distill-text) in a SINGLE transaction, so a mid-index failure (disk-full, crash) rolls the whole node back instead of leaving an orphan node row with no vector or postings.""" - ts = time.time() + self._write( + lambda c: self._apply_index( + c, + node_id, + kind=kind, + title=title, + tags=tags, + text_len=text_len, + updated_at=updated_at, + pinned=pinned, + importance=importance, + tf=tf, + vec=vec, + dim=dim, + edges=edges, + teacher=teacher, + text_param=text_param, + content_sha=content_sha, + ) + ) + + def index_atomic_many(self, items: Sequence[Dict[str, Any]]) -> int: + """Write MANY memories in ONE transaction. + + The per-note transaction is the wrong unit for a catch-up. Each commit + costs an fsync, and on a real deployment that fsync measured 43.7 ms + against 2-3 ms of actual indexing work — 97% of a backfill was the + commit, not the index. Amortising it across a chunk is a ~4x + difference on the same disk. + + Atomicity is per CHUNK here, which is the correct trade for a + rebuildable derived index: a crash mid-chunk rolls the chunk back and + the next boot's diff simply finds that work still outstanding. Callers + that need per-note rollback still have ``index_atomic``. + """ + if not items: + return 0 def _do(c): - c.execute( - "INSERT INTO nodes(id,kind,title,tags,text_len,updated_at,pinned,importance,content_sha)" - " VALUES(?,?,?,?,?,?,?,?,?)" - " ON CONFLICT(id) DO UPDATE SET kind=excluded.kind,title=excluded.title," - " tags=excluded.tags,text_len=excluded.text_len,updated_at=excluded.updated_at," - " pinned=excluded.pinned,importance=excluded.importance," - " content_sha=excluded.content_sha", - ( - node_id, - kind, - title, - json.dumps(list(tags), ensure_ascii=False), - text_len, - updated_at, - int(pinned), - importance, - content_sha, - ), - ) - nid = self._intern_nid(c, node_id) - c.execute("DELETE FROM postings_v2 WHERE nid=?", (nid,)) - tids = self._intern_tids(c, list(tf.keys())) + for item in items: + node_id = item["node_id"] + self._apply_index(c, node_id, **{k: v for k, v in item.items() if k != "node_id"}) + return len(items) + + return int(self._write(_do)) + + def _apply_index( + self, + c, + node_id: str, + *, + kind: str, + title: str, + tags: Sequence[str], + text_len: int, + updated_at: float, + pinned: bool, + importance: float, + tf: Dict[str, float], + vec: bytes, + dim: int, + edges: Sequence[Tuple[int, Sequence[Tuple[str, float]]]], + teacher: Optional[Tuple[str, bytes, int]] = None, + text_param: Optional[Tuple[str, bytes]] = None, + content_sha: str = "", + ) -> None: + """One memory's statements, on an ALREADY-OPEN transaction. + + Split out of ``index_atomic`` so a batch can put many of these inside + one commit. It must never commit or roll back — that belongs to + whoever opened the transaction. + """ + ts = time.time() + c.execute( + "INSERT INTO nodes(id,kind,title,tags,text_len,updated_at,pinned,importance,content_sha)" + " VALUES(?,?,?,?,?,?,?,?,?)" + " ON CONFLICT(id) DO UPDATE SET kind=excluded.kind,title=excluded.title," + " tags=excluded.tags,text_len=excluded.text_len,updated_at=excluded.updated_at," + " pinned=excluded.pinned,importance=excluded.importance," + " content_sha=excluded.content_sha", + ( + node_id, + kind, + title, + json.dumps(list(tags), ensure_ascii=False), + text_len, + updated_at, + int(pinned), + importance, + content_sha, + ), + ) + nid = self._intern_nid(c, node_id) + c.execute("DELETE FROM postings_v2 WHERE nid=?", (nid,)) + tids = self._intern_tids(c, list(tf.keys())) + c.executemany( + "INSERT OR REPLACE INTO postings_v2(tid,nid,tf) VALUES(?,?,?)", + [(tids[t], nid, f) for t, f in tf.items()], + ) + c.execute( + "INSERT OR REPLACE INTO vectors(node_id,dim,vec) VALUES(?,?,?)", (node_id, dim, vec) + ) + for etype, rows in edges: + c.execute("DELETE FROM edges WHERE src=? AND etype=?", (node_id, etype)) c.executemany( - "INSERT OR REPLACE INTO postings_v2(tid,nid,tf) VALUES(?,?,?)", - [(tids[t], nid, f) for t, f in tf.items()], + "INSERT OR REPLACE INTO edges(src,dst,etype,w,updated) VALUES(?,?,?,?,?)", + [(node_id, d, etype, w, ts) for d, w in rows], ) + if teacher is not None: + model, tvec, tdim = teacher c.execute( - "INSERT OR REPLACE INTO vectors(node_id,dim,vec) VALUES(?,?,?)", (node_id, dim, vec) + "INSERT OR REPLACE INTO teacher_vecs(node_id,model,dim,vec) VALUES(?,?,?,?)", + (node_id, model, tdim, tvec), ) - for etype, rows in edges: - c.execute("DELETE FROM edges WHERE src=? AND etype=?", (node_id, etype)) - c.executemany( - "INSERT OR REPLACE INTO edges(src,dst,etype,w,updated) VALUES(?,?,?,?,?)", - [(node_id, d, etype, w, ts) for d, w in rows], - ) - if teacher is not None: - model, tvec, tdim = teacher - c.execute( - "INSERT OR REPLACE INTO teacher_vecs(node_id,model,dim,vec) VALUES(?,?,?,?)", - (node_id, model, tdim, tvec), - ) - if text_param is not None: - key, blob = text_param - c.execute("INSERT OR REPLACE INTO params(key,blob) VALUES(?,?)", (key, blob)) - - self._write(_do) + if text_param is not None: + key, blob = text_param + c.execute("INSERT OR REPLACE INTO params(key,blob) VALUES(?,?)", (key, blob)) _NODE_COLS = ( "id,kind,title,tags,text_len,updated_at,access_count,last_access," @@ -301,22 +382,134 @@ def set_trust(self, node_id: str, trust: float, ts: float) -> None: ) def remove_node(self, node_id: str) -> None: - def _do(c): - row = c.execute("SELECT nid FROM node_map WHERE id=?", (node_id,)).fetchone() - if row is not None: - c.execute("DELETE FROM postings_v2 WHERE nid=?", (row[0],)) - c.execute("DELETE FROM node_map WHERE nid=?", (row[0],)) - for sql in ( - "DELETE FROM nodes WHERE id=?", - "DELETE FROM vectors WHERE node_id=?", - "DELETE FROM teacher_vecs WHERE node_id=?", - "DELETE FROM edges WHERE src=?", - "DELETE FROM feedback WHERE node_id=?", - "DELETE FROM edges WHERE dst=?", - ): - c.execute(sql, (node_id,)) + self._write(lambda c: self._apply_remove(c, node_id)) - self._write(_do) + def remove_nodes(self, node_ids: Sequence[str]) -> int: + """Delete MANY nodes in ONE transaction. + + Reaping notes the host deleted is inherently a bulk operation — a + vault that drifted for weeks presents thousands at once — and one + fsync per node turns a cleanup into an outage. + """ + ids = list(node_ids) + if not ids: + return 0 + + def _do(c): + for node_id in ids: + self._apply_remove(c, node_id) + return len(ids) + + return int(self._write(_do)) + + def _apply_remove(self, c, node_id: str) -> None: + """One node's deletions, on an ALREADY-OPEN transaction.""" + row = c.execute("SELECT nid FROM node_map WHERE id=?", (node_id,)).fetchone() + if row is not None: + c.execute("DELETE FROM postings_v2 WHERE nid=?", (row[0],)) + c.execute("DELETE FROM node_map WHERE nid=?", (row[0],)) + for sql in ( + "DELETE FROM nodes WHERE id=?", + "DELETE FROM vectors WHERE node_id=?", + "DELETE FROM teacher_vecs WHERE node_id=?", + "DELETE FROM edges WHERE src=?", + "DELETE FROM feedback WHERE node_id=?", + "DELETE FROM edges WHERE dst=?", + ): + c.execute(sql, (node_id,)) + + def clear_content_shas(self) -> int: + """Mark every row's derived state unknown. Returns how many. + + Used when the tokenizer or embedding geometry changed: the rows are + still there and still readable, but nothing about them was derived + the way it would be derived today. + """ + n = int(self._read("SELECT COUNT(*) FROM nodes WHERE content_sha != ''")[0][0]) + if n: + self._write(lambda c: c.execute("UPDATE nodes SET content_sha=''")) + return n + + # ── catalogue (metadata only — never touches bodies) ───────────── + def catalog_counts(self, *, by: str, kind: Optional[str] = None) -> List[Tuple[str, int]]: + """``[(key, count)]`` grouped by ``kind`` or by calendar ``day``. + + A browser's first question is "how much is there", and answering it + by materialising the vault is what makes a sidebar expensive: the + host's own note store parses every file to answer it (3.2 s and + 4.8 MB of bodies held for one count). The index already holds the + metadata; SQL can group it without reading a single body. + """ + if by == "kind": + sql = "SELECT kind, COUNT(*) FROM nodes" + params: List[Any] = [] + elif by == "day": + sql = "SELECT date(updated_at, 'unixepoch'), COUNT(*) FROM nodes" + params = [] + else: + raise ValueError(f"group by 'kind' or 'day', not {by!r}") + if kind is not None: + sql += " WHERE kind = ?" + params.append(kind) + sql += " GROUP BY 1 ORDER BY 1 DESC" + return [(str(r[0] or ""), int(r[1])) for r in self._read(sql, params)] + + def catalog_page( + self, + *, + day: Optional[str] = None, + kind: Optional[str] = None, + limit: int = 100, + offset: int = 0, + ) -> List[Dict[str, Any]]: + """One page of note METADATA — id, title, kind, timestamp, length. + + Filtered and paged in SQL, so a day with 40 notes costs a 40-row + read whatever the vault's size. No bodies: those are fetched when + something is actually opened. + """ + where: List[str] = [] + params: List[Any] = [] + if day: + where.append("date(updated_at, 'unixepoch') = ?") + params.append(day) + if kind is not None: + where.append("kind = ?") + params.append(kind) + sql = "SELECT id, kind, title, updated_at, text_len, pinned, importance FROM nodes" + if where: + sql += " WHERE " + " AND ".join(where) + sql += " ORDER BY updated_at DESC, id LIMIT ? OFFSET ?" + params += [max(0, int(limit)), max(0, int(offset))] + return [ + { + "id": r[0], + "kind": r[1], + "title": r[2] or "", + "updated_at": float(r[3] or 0.0), + "text_len": int(r[4] or 0), + "pinned": bool(r[5]), + "importance": float(r[6] or 0.0), + } + for r in self._read(sql, params) + ] + + def manifest(self) -> Dict[str, Tuple[float, str]]: + """``{node_id: (updated_at, content_sha)}`` — what is already indexed. + + This is the whole point of an incremental host: with it, "what changed + since last time" is a set difference against metadata the host already + holds. Without it the only way to answer is to re-read every note and + re-derive its digest, which is what made a wake cost 13 seconds and + 5,500 lock acquisitions to discover that one note had changed. + + Deliberately excludes bodies and vectors — it must stay cheap enough + to call on every boot (8.7k nodes measured at 4 ms). + """ + return { + r[0]: (float(r[1] or 0.0), r[2] or "") + for r in self._read("SELECT id, updated_at, content_sha FROM nodes") + } def touch_access(self, ids: Iterable[str], ts: Optional[float] = None) -> None: ts = ts or time.time() @@ -442,6 +635,31 @@ def _do(c): self._write(_do) + def edges_touching( + self, node_ids: Sequence[str], *, limit: int = 4000 + ) -> List[Tuple[str, str, int, float]]: + """Edges with either end inside *node_ids* — the graph equivalent of + a page. + + A whole-vault graph is not a view anyone reads; it is a download. The + production snapshot was 5,384 nodes and 4.3 MB of JSON for one + screen. Asking for the edges around a selection keeps the payload + proportional to what is on screen. + """ + ids = list(node_ids) + if not ids: + return [] + marks = ",".join("?" for _ in ids) + sql = ( + f"SELECT src, dst, etype, w FROM edges" + f" WHERE src IN ({marks}) OR dst IN ({marks})" + f" ORDER BY w DESC LIMIT ?" + ) + return [ + (r[0], r[1], int(r[2]), float(r[3] or 0.0)) + for r in self._read(sql, ids + ids + [max(0, int(limit))]) + ] + def edges_by_type(self, etype: int) -> List[Tuple[str, str, float, float]]: return self._read("SELECT src,dst,w,updated FROM edges WHERE etype=?", (etype,)) diff --git a/src/xgen_agent_memory/tokenizer.py b/src/xgen_agent_memory/tokenizer.py index f8fb8e3..69e97b6 100644 --- a/src/xgen_agent_memory/tokenizer.py +++ b/src/xgen_agent_memory/tokenizer.py @@ -2,6 +2,10 @@ LEXICAL stream (BM25 postings) — precision-oriented: * surface words (NFKC + casefold) + * Porter stems for Latin-script words (additive — surface AND stem). The + Latin side had no morphology at all; character trigrams stood in for it, + which cost 72.8% of a production vault's postings and still lost 22% of + MRR when a query was re-inflected off its source note. * guarded 조사-stripped stems (받침 agreement, ≥2-syllable stems; additive — surface AND stem are indexed, a wrong strip only adds one noisy term) * overlapping SYLLABLE BIGRAMS within each Hangul word — the no-analyzer @@ -26,6 +30,7 @@ from typing import Iterable, List from .hangul import has_hangul, normalize, strip_suffix, to_jamo +from .latin import stem as latin_stem _WORD = re.compile(r"[\w']+", re.UNICODE) @@ -34,6 +39,11 @@ #: Marker prefix for jamo grams — never collides with syllable grams. _JAMO_MARK = "ⱼ" +#: Minimum Latin word length for character trigrams in the EMBEDDING stream. +#: The BM25 stream defaults to 0 (none); this side keeps them because it is +#: where fuzzy matching belongs and it costs no postings. +LATIN_NGRAM_EMBED = 4 + #: Marker for cross-space bigrams — kept distinct from in-word bigrams so #: their IDF is computed on their own distribution. _XSPACE_MARK = "ₓ" @@ -80,6 +90,8 @@ def lexical_tokens( suffix_strip: bool = True, cross_space: bool = True, limit: int = 2048, + latin_ngram_min_len: int = 0, + latin_stemming: bool = True, ) -> List[str]: """BM25 stream: words + guarded stems + syllable bigrams + cross-space bigrams. See module docstring for the evidence behind each choice.""" @@ -112,7 +124,14 @@ def flush_run() -> None: run.append(word) else: flush_run() - if len(word) > 3: + # Same additive contract as the Korean stripper above: surface + # AND stem, so an over-eager conflation costs one posting and + # never an exact match. + if latin_stemming: + st = latin_stem(word) + if st != word.lower(): + tokens.append(st) + if latin_ngram_min_len > 0 and len(word) >= latin_ngram_min_len: tokens.extend(_ngrams(word, (3,))) if len(tokens) >= limit: break @@ -129,8 +148,18 @@ def embed_tokens( limit: int = 2048, ) -> List[str]: """Embedding stream: the lexical stream + padded jamo n-grams.""" + # Latin trigrams are asked for EXPLICITLY here, exactly as jamo is added + # below: this stream is the recall side and pays no posting cost. The + # BM25 default is 0, and inheriting that by omission would have moved + # fuzzy matching out of the engine entirely the first time someone read + # the signature and "tidied" it. tokens = lexical_tokens( - text, char_ngrams=char_ngrams, suffix_strip=suffix_strip, cross_space=False, limit=limit + text, + char_ngrams=char_ngrams, + suffix_strip=suffix_strip, + cross_space=False, + limit=limit, + latin_ngram_min_len=LATIN_NGRAM_EMBED, ) jamo_sizes = tuple(jamo_ngrams) if jamo_sizes: diff --git a/tests/test_catalog.py b/tests/test_catalog.py new file mode 100644 index 0000000..c3bda16 --- /dev/null +++ b/tests/test_catalog.py @@ -0,0 +1,169 @@ +"""Progressive browsing — effect-proving tests. + +A vault browser asks three questions in order, and each one should cost what +it is worth: + + how much is there → a count + what days are there → counts per day + what is on this day → that day's metadata + what does this say → one body + +The host's own note store answers the first by materialising the whole vault +— 3.2 s and 4.8 MB of bodies held, measured on a 5,384-note production vault +— because its only listing primitive walks every note. The index already +holds the metadata these questions need; SQL can group and page it without +reading a body at all. +""" + +from __future__ import annotations + +import time + +import pytest + +from xgen_agent_memory import SynapseConfig, SynapseMemory + +DAY_A = time.mktime((2026, 8, 1, 12, 0, 0, 0, 0, -1)) +DAY_B = time.mktime((2026, 8, 2, 12, 0, 0, 0, 0, -1)) + + +@pytest.fixture +def vault(tmp_path): + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "c.db"), epsilon=0.0)) + mem.index_many( + [ + { + "node_id": f"obs/{i}", + "text": f"관찰 기록 {i}", + "kind": "observations", + "updated_at": DAY_A, + } + for i in range(7) + ] + + [ + { + "node_id": f"obs/b{i}", + "text": f"관찰 기록 b{i}", + "kind": "observations", + "updated_at": DAY_B, + } + for i in range(3) + ] + + [ + {"node_id": "note/keep", "text": "사람이 쓴 노트", "kind": "note", "updated_at": DAY_B}, + ] + ) + return mem + + +def _no_bodies_read(mem, monkeypatch): + """Trip if anything reaches for a body while answering.""" + + def _boom(*_a, **_kw): + raise AssertionError("a body was read to answer a metadata question") + + monkeypatch.setattr(mem, "get_text", _boom) + monkeypatch.setattr(mem, "_get_text_unlocked", _boom) + + +# ── level 1: how much is there ────────────────────────────────────── + + +def test_counts_by_kind(vault, monkeypatch): + _no_bodies_read(vault, monkeypatch) + assert dict(vault.catalog_counts(by="kind")) == {"observations": 10, "note": 1} + + +def test_counts_by_day(vault, monkeypatch): + _no_bodies_read(vault, monkeypatch) + assert dict(vault.catalog_counts(by="day")) == { + "2026-08-01": 7, + "2026-08-02": 4, + } + + +def test_counts_can_be_scoped_to_one_kind(vault): + assert dict(vault.catalog_counts(by="day", kind="note")) == {"2026-08-02": 1} + + +def test_days_come_back_newest_first(vault): + assert [d for d, _n in vault.catalog_counts(by="day")] == ["2026-08-02", "2026-08-01"] + + +def test_an_empty_vault_counts_nothing(tmp_path): + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "e.db"))) + assert mem.catalog_counts(by="kind") == [] + assert mem.catalog_counts(by="day") == [] + + +def test_an_unknown_grouping_is_an_error(vault): + with pytest.raises(ValueError): + vault.catalog_counts(by="colour") + + +# ── level 2: what is on this day ──────────────────────────────────── + + +def test_a_day_returns_only_that_day(vault, monkeypatch): + _no_bodies_read(vault, monkeypatch) + page = vault.catalog_page(day="2026-08-02") + assert len(page) == 4 + assert {r["id"] for r in page} == {"obs/b0", "obs/b1", "obs/b2", "note/keep"} + + +def test_a_page_carries_metadata_not_content(vault): + row = vault.catalog_page(day="2026-08-01", limit=1)[0] + assert set(row) == {"id", "kind", "title", "updated_at", "text_len", "pinned", "importance"} + assert "text" not in row and "body" not in row + + +def test_paging_is_done_in_sql(vault): + """Not "fetch everything then slice" — that is the pattern being + replaced, and it is why asking for one note cost the whole vault.""" + first = vault.catalog_page(day="2026-08-01", limit=3, offset=0) + second = vault.catalog_page(day="2026-08-01", limit=3, offset=3) + assert len(first) == 3 and len(second) == 3 + assert {r["id"] for r in first}.isdisjoint({r["id"] for r in second}) + + +def test_a_day_with_nothing_is_empty_not_everything(vault): + assert vault.catalog_page(day="2020-01-01") == [] + + +def test_a_page_can_be_scoped_to_a_kind(vault): + page = vault.catalog_page(kind="note") + assert [r["id"] for r in page] == ["note/keep"] + + +# ── level 3: the graph, at a screen's worth ───────────────────────── + + +def test_a_neighbourhood_is_not_the_whole_vault(vault): + """The production snapshot was 5,384 nodes and 4.3 MB of JSON for one + screen. A graph view is read at a screen's worth of detail.""" + out = vault.neighbourhood(["obs/0"], depth=1) + assert out["nodes"], "the seed itself came back empty" + assert len(out["nodes"]) < 11 + + +def test_the_seed_is_always_present(vault): + out = vault.neighbourhood(["note/keep"], depth=1) + assert "note/keep" in {n["id"] for n in out["nodes"]} + + +def test_node_caps_are_honoured_and_reported(vault): + out = vault.neighbourhood(["obs/0"], depth=2, max_nodes=3) + assert len(out["nodes"]) <= 3 + assert isinstance(out["truncated"], bool) + + +def test_an_empty_seed_returns_an_empty_graph(vault): + out = vault.neighbourhood([], depth=2) + assert out == {"nodes": [], "edges": [], "truncated": False} + + +def test_edges_name_their_type(vault): + out = vault.neighbourhood(["obs/0"], depth=1) + for e in out["edges"]: + assert set(e) == {"src", "dst", "type", "w"} + assert isinstance(e["type"], int) diff --git a/tests/test_incremental.py b/tests/test_incremental.py new file mode 100644 index 0000000..3e29497 --- /dev/null +++ b/tests/test_incremental.py @@ -0,0 +1,405 @@ +"""Incremental indexing — effect-proving tests. + +A host that re-derives everything on every boot is the failure this file +guards against. Three primitives make an incremental host possible, and each +is only useful if it holds a specific property: + + * ``manifest()`` — what is already indexed, cheaply enough to ask every + boot. Without it "what changed?" can only be answered by re-reading every + note and re-hashing it. + * ``index_many()`` — many notes, ONE commit. The commit was measured at + 43.7 ms against 2-3 ms of real indexing work, so a per-note transaction + makes a catch-up 97% fsync. It must produce byte-identical results to + indexing one at a time, or the batch is a different feature. + * ``remove_many()`` — deletions the host discovered in bulk. Reaping a + vault that drifted for weeks presents thousands at once. +""" + +from __future__ import annotations + +import time + +import pytest + +from xgen_agent_memory import SynapseConfig, SynapseMemory + + +@pytest.fixture +def mem(tmp_path): + return SynapseMemory(SynapseConfig(path=str(tmp_path / "s.db"))) + + +def _items(n, prefix="n", body="리듬게임 판정 연습 기록"): + return [{"node_id": f"{prefix}{i}", "text": f"{body} {i}", "kind": "note"} for i in range(n)] + + +def _commits(mem): + """Count transactions. + + ``Store._write`` IS the transaction boundary — it runs the statements and + commits — so counting it counts fsyncs. (``sqlite3.Connection.commit`` + itself is read-only and cannot be wrapped.) + """ + calls = [] + real = mem.store._write + + def counted(fn): + calls.append(1) + return real(fn) + + mem.store._write = counted + return calls + + +# ── manifest ──────────────────────────────────────────────────────── + + +def test_manifest_reports_what_is_indexed(mem): + mem.index("a", "본문 하나", kind="note") + mem.index("b", "본문 둘", kind="note") + + man = mem.manifest() + + assert set(man) == {"a", "b"} + for updated_at, sha in man.values(): + assert updated_at > 0 + assert sha, "no digest — a host cannot tell changed from unchanged" + + +def test_manifest_digest_tracks_content(mem): + mem.index("a", "처음 본문") + before = mem.manifest()["a"][1] + mem.index("a", "바뀐 본문") + after = mem.manifest()["a"][1] + + assert before != after, "digest did not follow the content" + + +def test_manifest_carries_no_bodies(mem): + """It is called on every boot; it must stay metadata-only.""" + mem.index("a", "아주 긴 본문 " * 200) + (updated_at, sha) = mem.manifest()["a"] + assert isinstance(updated_at, float) + assert len(sha) == 40 # sha1 hex, not a body + + +# ── index_many: same result, one commit ───────────────────────────── + + +def test_batch_and_one_by_one_produce_the_same_index(tmp_path): + """THE property. A batch that indexes differently is not a batch, it is a + second implementation that will drift.""" + a = SynapseMemory(SynapseConfig(path=str(tmp_path / "a.db"))) + b = SynapseMemory(SynapseConfig(path=str(tmp_path / "b.db"))) + items = _items(25) + + for it in items: + a.index(it["node_id"], it["text"], kind=it["kind"]) + b.index_many(items) + + assert a.manifest().keys() == b.manifest().keys() + for nid in a.manifest(): + assert a.manifest()[nid][1] == b.manifest()[nid][1], f"{nid} differs" + # Derived rows too, not just the digest. + for table, col in (("postings_v2", "nid"), ("edges", "src"), ("vectors", "node_id")): + na = a.store._read(f"SELECT COUNT(*) FROM {table}")[0][0] + nb = b.store._read(f"SELECT COUNT(*) FROM {table}")[0][0] + assert na == nb, f"{table}: {na} vs {nb}" + + +def test_a_chunk_costs_one_commit(mem): + calls = _commits(mem) + mem.index_many(_items(50), chunk_size=50) + assert len(calls) == 1, f"expected one commit for the chunk, got {len(calls)}" + + +def test_chunking_bounds_the_transaction(mem): + calls = _commits(mem) + mem.index_many(_items(50), chunk_size=10) + assert len(calls) == 5, ( + f"chunk_size must bound how much a crash rolls back; got {len(calls)} commits" + ) + + +def test_one_by_one_is_what_the_batch_improves_on(mem): + """Anchors the claim: N notes cost N commits without the batch.""" + calls = _commits(mem) + for it in _items(10): + mem.index(it["node_id"], it["text"]) + assert len(calls) >= 10 + + +# ── the whole point: unchanged content costs nothing ──────────────── + + +def test_reindexing_unchanged_content_writes_nothing(mem): + items = _items(30) + mem.index_many(items) + + calls = _commits(mem) + out = mem.index_many(items) + + assert out["indexed"] == 0, "re-indexed content that had not changed" + assert out["skipped"] == 30 + assert len(calls) == 0, "an all-unchanged batch still hit the disk" + + +def test_only_the_changed_note_is_reindexed(mem): + items = _items(30) + mem.index_many(items) + items[7]["text"] = "완전히 다른 본문" + + out = mem.index_many(items) + + assert out["indexed"] == 1 + assert out["skipped"] == 29 + + +def test_a_moved_clock_is_a_touch_not_a_reindex(mem): + """Same bytes, newer mtime. Re-deriving postings and edges for that is + the exact waste this whole path exists to avoid.""" + items = _items(10) + mem.index_many(items) + later = time.time() + 60 + for it in items: + it["updated_at"] = later + + out = mem.index_many(items) + + assert out["indexed"] == 0 + assert out["touched"] == 10 + assert all(v[0] == pytest.approx(later) for v in mem.manifest().values()) + + +def test_a_touch_only_batch_costs_one_commit(mem): + items = _items(40) + mem.index_many(items) + for it in items: + it["updated_at"] = time.time() + 120 + + calls = _commits(mem) + mem.index_many(items, chunk_size=40) + + assert len(calls) == 1, f"touches were not batched ({len(calls)} commits)" + + +# ── correctness of derived state after a batch ────────────────────── + + +def test_notes_indexed_in_a_batch_are_searchable(mem): + mem.index_many( + [ + {"node_id": "k1", "text": "리듬게임 판정 보정 방법", "kind": "note"}, + {"node_id": "k2", "text": "저녁 식사 메뉴 기록", "kind": "note"}, + ] + ) + + hits = mem.search("리듬게임 판정", top_k=5) + + assert any(h.id == "k1" for h in hits), "batch-indexed note is not findable" + + +def test_knn_inside_a_batch_sees_earlier_members(tmp_path): + """Sequential indexing links each note to the ones already there. If a + batch derived every note against a frozen snapshot, the graph it builds + would be sparser than the same notes added one at a time — a silent + difference in retrieval quality, not an error anyone would see. + + So the assertion is equality with the sequential graph, edge for edge.""" + body = "판정 보정 오프셋 설정 리듬게임" + items = [{"node_id": f"s{i}", "text": f"{body} {i}"} for i in range(12)] + + seq = SynapseMemory(SynapseConfig(path=str(tmp_path / "seq.db"))) + for it in items: + seq.index(it["node_id"], it["text"]) + bat = SynapseMemory(SynapseConfig(path=str(tmp_path / "bat.db"))) + bat.index_many(items) + + def knn(m): + return {(r[0], r[1]) for r in m.store._read("SELECT src, dst FROM edges WHERE etype=2")} + + assert knn(seq), "the fixture produces no kNN edges — test proves nothing" + assert knn(bat) == knn(seq), "batch built a different graph" + + +def test_the_cache_matches_the_database_after_a_batch(mem): + mem.index_many(_items(20)) + mem.search("리듬게임", top_k=3) # populates caches + mem.index_many([{"node_id": "late", "text": "새로 들어온 본문 리듬게임"}]) + + hits = mem.search("새로 들어온", top_k=5) + assert any(h.id == "late" for h in hits), "cache did not learn the batch" + + +# ── remove_many ───────────────────────────────────────────────────── + + +def test_remove_many_deletes_every_node_in_one_commit(mem): + mem.index_many(_items(20)) + ids = list(mem.manifest()) + + calls = _commits(mem) + removed = mem.remove_many(ids) + + assert removed == 20 + assert mem.manifest() == {} + assert len(calls) == 1, f"one fsync per deletion ({len(calls)} commits)" + + +def test_remove_many_clears_derived_rows(mem): + """A node row deleted while its postings survive is worse than an orphan: + the search still scores it and then cannot resolve it.""" + mem.index_many(_items(10)) + mem.remove_many(list(mem.manifest())) + + for table in ("postings_v2", "vectors", "edges", "node_map"): + n = mem.store._read(f"SELECT COUNT(*) FROM {table}")[0][0] + assert n == 0, f"{table} kept {n} rows for deleted nodes" + + +def test_remove_many_leaves_survivors_alone(mem): + mem.index_many(_items(10)) + mem.remove_many(["n1", "n2", "n3"]) + + assert set(mem.manifest()) == {"n0", "n4", "n5", "n6", "n7", "n8", "n9"} + assert mem.search("리듬게임", top_k=5), "survivors became unsearchable" + + +def test_remove_many_on_an_empty_list_is_free(mem): + calls = _commits(mem) + assert mem.remove_many([]) == 0 + assert len(calls) == 0 + + +# ── read/write concurrency ────────────────────────────────────────── + + +def test_searches_run_while_an_index_is_in_flight(tmp_path): + """THE property behind splitting the lock. + + Under one mutex a search waited for every index — measured at 32.8 ms + idle vs 165 ms during indexing on a production vault — and a write that + wedged took every read with it, turning one stuck call into a total + memory outage. Readers do not conflict; they must not queue. + """ + import threading + + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "c.db"))) + mem.index_many(_items(30)) + + searching = threading.Event() + release_search = threading.Event() + indexed = threading.Event() + + def searcher(): + with mem._lock.read(): + searching.set() + release_search.wait(5) + + def indexer(): + mem.index("late", "새 본문 리듬게임") + indexed.set() + + t1 = threading.Thread(target=searcher, daemon=True) + t1.start() + searching.wait(5) + + t2 = threading.Thread(target=indexer, daemon=True) + t2.start() + # A writer must WAIT for the in-flight reader — that half is unchanged. + assert not indexed.wait(0.2) + release_search.set() + assert indexed.wait(5), "the write never completed after the read finished" + t1.join(5) + t2.join(5) + + +def test_two_searches_do_not_serialise(tmp_path): + import threading + + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "d.db"))) + mem.index_many(_items(10)) + + both = threading.Barrier(3, timeout=5) + + def searcher(): + with mem._lock.read(): + both.wait() # unreachable if reads are exclusive + + threads = [threading.Thread(target=searcher, daemon=True) for _ in range(2)] + for t in threads: + t.start() + both.wait() # would time out under one mutex + for t in threads: + t.join(5) + + +def test_a_learning_step_does_not_deadlock_itself(tmp_path): + """`feedback` and `learn` both update per-item trust while already + holding the write lock. With a non-reentrant lock, calling the public + `trust_feedback` there hangs the process — which is exactly what + happened, on three tests at once, the moment the lock was split.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "e.db"), epsilon=0.0)) + mem.index_many(_items(6)) + hits = mem.search("리듬게임", top_k=3) + + out = mem.feedback(hits[0].query_token, used_ids=[hits[0].id]) + + assert out["applied"] >= 0.0 + assert mem.trust_feedback(hits[0].id, True) is not None + + +def test_reading_a_body_from_inside_a_read_does_not_deadlock(tmp_path): + """`contradictions` runs under the read lock and needs several bodies.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "f.db"))) + mem.index("a", "판정은 손이 먼저다") + mem.index("b", "판정은 손이 먼저가 아니다") + + assert mem.contradictions("a", top_k=3) is not None + assert mem.get_text("a") + + +def test_a_search_gives_up_on_a_wedged_write(tmp_path): + """The blast-radius fix. Splitting the lock does not stop a stuck writer + from blocking reads; the deadline does. A turn can answer without memory + — it cannot answer while inheriting someone else's hang, which is how a + single spinning matmul took conversations down for 27 hours.""" + import threading + + from xgen_agent_memory import MemoryBusy + + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "g.db"))) + mem.index_many(_items(5)) + + holding = threading.Event() + release = threading.Event() + + def wedged_writer(): + with mem._lock.write(): + holding.set() + release.wait(10) + + t = threading.Thread(target=wedged_writer, daemon=True) + t.start() + holding.wait(5) + + with pytest.raises(MemoryBusy): + mem.search("리듬게임", top_k=3, timeout=0.1) + + release.set() + t.join(5) + assert mem.search("리듬게임", top_k=3), "search never recovered" + + +def test_the_default_search_patience_is_sane(tmp_path): + """Too short trips on a normal index chunk; too long is the hang.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "h.db"))) + assert 5.0 <= mem.SEARCH_TIMEOUT_S <= 60.0 + + +def test_a_caller_can_opt_out_of_the_deadline(tmp_path): + """Callers that cannot degrade (a backfill verifying its own work) must + be able to wait.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "i.db"))) + mem.index_many(_items(3)) + assert mem.search("리듬게임", top_k=2, timeout=None) is not None diff --git a/tests/test_latin_stem.py b/tests/test_latin_stem.py new file mode 100644 index 0000000..63f9099 --- /dev/null +++ b/tests/test_latin_stem.py @@ -0,0 +1,144 @@ +"""Porter stemming — correctness first, then the property it buys. + +The Latin side had no stemmer; character trigrams stood in for one. This +module replaces that guess with the algorithm every IR system agreed on, so +the first duty of these tests is that it IS Porter — checked against outputs +anyone can verify against the published algorithm — and the second is the +contract it shares with the Korean stripper: additive, so an over-eager +conflation costs a posting, never an exact match. +""" + +from __future__ import annotations + +import pytest + +from xgen_agent_memory.latin import MIN_WORD_LEN, stem + + +# ── it is actually Porter ─────────────────────────────────────────── + + +@pytest.mark.parametrize( + "word,expected", + [ + # 1a — plurals + ("caresses", "caress"), + ("ponies", "poni"), + ("cats", "cat"), + ("caress", "caress"), + # 1b — -eed / -ed / -ing with the m-guard + ("agreed", "agre"), + ("plastered", "plaster"), + ("motoring", "motor"), + ("sing", "sing"), + # 1b cleanup — at/bl/iz get an e, doubles collapse, cvc gets an e + ("conflated", "conflat"), + ("troubling", "troubl"), + ("hopping", "hop"), + ("falling", "fall"), + ("filing", "file"), + # 1c — terminal y + ("happy", "happi"), + ("sky", "sky"), + # 2/3/4 — derivational + ("relational", "relat"), + ("conditional", "condit"), + ("hopefulness", "hope"), + ("formalize", "formal"), + ("adjustment", "adjust"), + ("dependent", "depend"), + ], +) +def test_known_porter_outputs(word, expected): + assert stem(word) == expected + + +# ── the property that matters for retrieval ───────────────────────── + + +@pytest.mark.parametrize( + "a,b", + [ + ("browsing", "browse"), + ("browsed", "browse"), + ("running", "run"), + ("stopped", "stop"), + ("files", "file"), + ("connections", "connection"), + ("configured", "configure"), + ("observations", "observation"), + ("indexing", "index"), + ], +) +def test_inflections_of_one_word_share_a_stem(a, b): + """THE point. A query re-inflected off its source note lost 22% of its + MRR on the production vault because these did not meet.""" + assert stem(a) == stem(b), f"{a} → {stem(a)} vs {b} → {stem(b)}" + + +@pytest.mark.parametrize( + "a,b", + [ + ("cat", "dog"), + ("index", "engine"), + ("memory", "vector"), + ("browse", "brownie"), + ("stop", "storage"), + ], +) +def test_unrelated_words_do_not_collide(a, b): + assert stem(a) != stem(b) + + +# ── guards ────────────────────────────────────────────────────────── + + +def test_short_words_are_left_alone(): + """Three letters are already a stem; stripping them collides ("ads"→"ad", + "his"→"hi") and buys nothing.""" + for w in ("ads", "his", "the", "cat", "run"): + assert stem(w) == w.lower() + assert MIN_WORD_LEN == 4 + + +def test_non_alphabetic_tokens_are_untouched(): + """Identifiers, versions, hashes — stripping a trailing 's' off `v2s` + would break exact matching on the thing most worth matching exactly.""" + for w in ("docker-compose", "v2.65.1", "a1b2c3", "__init__", "utf8"): + assert stem(w) == w.lower() + + +def test_it_is_case_insensitive(): + assert stem("Browsing") == stem("browsing") == stem("BROWSING") + + +def test_the_pipeline_never_stems_a_stem(): + """Porter is not idempotent — `stem("brows")` is "brow" while + `stem("browsing")` is "brows" — and that is fine here ONLY because both + the index and the query stem the original surface form, never a previous + stem. This test documents the invariant the callers must keep. + """ + import inspect + + from xgen_agent_memory import tokenizer + + src = inspect.getsource(tokenizer.lexical_tokens) + assert "latin_stem(word)" in src, ( + "the stemmer must be applied to the surface word; feeding it a " + "previous stem would drift the index away from the query" + ) + + +def test_a_known_porter_quirk_is_recorded(): + """`deployment` → `deploy` but `deploy` → `deploi`: step 1c turns a + terminal y into i, and a word that reaches step 4 has already passed it. + Canonical Porter behaves this way; the pair is caught by the vector + stream instead. Recorded so a future change does not "fix" it by + accident and shift every other stem with it.""" + assert stem("deployment") == "deploy" + assert stem("deploy") == "deploi" + + +def test_empty_and_tiny_input_is_safe(): + assert stem("") == "" + assert stem("a") == "a" diff --git a/tests/test_rwlock.py b/tests/test_rwlock.py new file mode 100644 index 0000000..15618a9 --- /dev/null +++ b/tests/test_rwlock.py @@ -0,0 +1,326 @@ +"""RWLock — effect-proving tests. + +The property that matters is not "it locks" but *which* operations stop +waiting for each other. One mutex over the engine meant a search queued +behind every index, and a write that never finished took every read with it. +""" + +from __future__ import annotations + +import threading +import time + +import pytest + +from xgen_agent_memory._rwlock import MemoryBusy, RWLock + + +def _spawn(fn, n=1): + threads = [threading.Thread(target=fn, daemon=True) for _ in range(n)] + for t in threads: + t.start() + return threads + + +def test_readers_run_concurrently(): + """THE point. Under one mutex this test cannot pass.""" + lock = RWLock() + both_in = threading.Barrier(3, timeout=5) + ok = [] + + def reader(): + with lock.read(): + both_in.wait() # only reachable if both hold it at once + ok.append(1) + + threads = _spawn(reader, 2) + both_in.wait() + for t in threads: + t.join(5) + assert len(ok) == 2 + + +def test_a_writer_excludes_readers(): + lock = RWLock() + writer_in = threading.Event() + reader_got_in = threading.Event() + release = threading.Event() + + def writer(): + with lock.write(): + writer_in.set() + release.wait(5) + + def reader(): + with lock.read(): + reader_got_in.set() + + _spawn(writer) + writer_in.wait(5) + _spawn(reader) + + assert not reader_got_in.wait(0.2), "a reader entered during a write" + release.set() + assert reader_got_in.wait(5), "the reader never got in after the write" + + +def test_a_writer_waits_for_readers(): + lock = RWLock() + reader_in = threading.Event() + writer_got_in = threading.Event() + release = threading.Event() + + def reader(): + with lock.read(): + reader_in.set() + release.wait(5) + + def writer(): + with lock.write(): + writer_got_in.set() + + _spawn(reader) + reader_in.wait(5) + _spawn(writer) + + assert not writer_got_in.wait(0.2), "a writer ran during a read" + release.set() + assert writer_got_in.wait(5) + + +def test_writers_are_mutually_exclusive(): + lock = RWLock() + concurrent = [] + active = {"n": 0} + guard = threading.Lock() + + def writer(): + with lock.write(): + with guard: + active["n"] += 1 + concurrent.append(active["n"]) + time.sleep(0.01) + with guard: + active["n"] -= 1 + + threads = _spawn(writer, 4) + for t in threads: + t.join(5) + assert max(concurrent) == 1 + + +def test_a_waiting_writer_blocks_readers_that_arrive_after_it(): + """Arrival order. A reader that shows up AFTER a writer is queued waits + its turn, so a steady retrieval load cannot postpone indexing — notes + silently ceasing to be searchable while every dashboard stays green is + the failure on that side.""" + lock = RWLock() + first_reader_in = threading.Event() + release_first = threading.Event() + writer_waiting = threading.Event() + late_reader_in = threading.Event() + + def first_reader(): + with lock.read(): + first_reader_in.set() + release_first.wait(5) + + def writer(): + writer_waiting.set() + with lock.write(): + pass + + def late_reader(): + with lock.read(): + late_reader_in.set() + + _spawn(first_reader) + first_reader_in.wait(5) + _spawn(writer) + writer_waiting.wait(5) + time.sleep(0.05) # let the writer actually enqueue + _spawn(late_reader) + + assert not late_reader_in.wait(0.2), "a late reader jumped the writer" + release_first.set() + assert late_reader_in.wait(5) + + +def test_the_lock_is_released_when_the_body_raises(): + lock = RWLock() + + with pytest.raises(ValueError): + with lock.write(): + raise ValueError("boom") + assert lock.writing is False + + with pytest.raises(ValueError): + with lock.read(): + raise ValueError("boom") + assert lock.readers == 0 + + with lock.write(): # would hang if the first release leaked + pass + + +def test_counters_return_to_zero(): + lock = RWLock() + + def churn(): + for _ in range(50): + with lock.read(): + pass + with lock.write(): + pass + + ts = _spawn(churn, 4) + for t in ts: + t.join(10) + assert lock.readers == 0 + assert lock.writing is False + + +def test_a_continuous_writer_does_not_starve_readers(): + """The other side of fairness, and the reason strict writer preference + was wrong here. A writer that re-queues the instant it releases would, + under writer preference, never let a reader in at all — measured as a + search benchmark that simply never finished against a live indexing + load. Arrival order gives the waiting reader the next turn. + """ + lock = RWLock() + stop = threading.Event() + reads = [] + + def writer(): + while not stop.is_set(): + with lock.write(): + time.sleep(0.005) + + def reader(): + for _ in range(20): + with lock.read(): + reads.append(1) + + w = _spawn(writer) + time.sleep(0.02) # let the write loop get going + r = _spawn(reader) + for t in r: + t.join(10) + stop.set() + for t in w: + t.join(5) + + assert len(reads) == 20, f"reader starved: only {len(reads)}/20 got through" + + +def test_a_continuous_reader_does_not_starve_writers(): + lock = RWLock() + stop = threading.Event() + writes = [] + + def reader(): + while not stop.is_set(): + with lock.read(): + time.sleep(0.005) + + def writer(): + for _ in range(20): + with lock.write(): + writes.append(1) + + rs = _spawn(reader, 3) + time.sleep(0.02) + w = _spawn(writer) + for t in w: + t.join(10) + stop.set() + for t in rs: + t.join(5) + + assert len(writes) == 20, f"writer starved: only {len(writes)}/20 got through" + + +# ── bounded acquisition: the only thing that shrinks a wedge's blast radius ── + + +def test_a_reader_can_give_up_on_a_stuck_writer(): + """THE property. Splitting the lock does NOT stop a wedged write from + blocking reads — an exclusive writer that never returns still holds + everyone. A deadline does: retrieval fails fast and the caller proceeds + without memory instead of inheriting someone else's hang.""" + lock = RWLock() + writer_in = threading.Event() + release = threading.Event() + + def stuck_writer(): + with lock.write(): + writer_in.set() + release.wait(10) + + _spawn(stuck_writer) + writer_in.wait(5) + + started = time.monotonic() + with pytest.raises(MemoryBusy): + with lock.read(timeout=0.1): + pass + assert time.monotonic() - started < 2.0, "gave up far later than asked" + + release.set() + + +def test_a_writer_can_give_up_too(): + lock = RWLock() + reader_in = threading.Event() + release = threading.Event() + + def stuck_reader(): + with lock.read(): + reader_in.set() + release.wait(10) + + _spawn(stuck_reader) + reader_in.wait(5) + + with pytest.raises(MemoryBusy): + with lock.write(timeout=0.1): + pass + + release.set() + + +def test_a_deadline_does_not_fire_when_the_lock_is_free(): + lock = RWLock() + with lock.read(timeout=0.1): + pass + with lock.write(timeout=0.1): + pass + + +def test_giving_up_leaves_the_lock_usable(): + """A waiter that timed out must not leave its place in the queue behind — + the next writer would wait on a ticket nobody is holding.""" + lock = RWLock() + writer_in = threading.Event() + release = threading.Event() + + def stuck_writer(): + with lock.write(): + writer_in.set() + release.wait(10) + + t = _spawn(stuck_writer) + writer_in.wait(5) + for _ in range(3): + with pytest.raises(MemoryBusy): + with lock.read(timeout=0.05): + pass + release.set() + for th in t: + th.join(5) + + with lock.write(timeout=2): # would hang on a leaked queue entry + pass + with lock.read(timeout=2): + pass + assert lock.readers == 0 and lock.writing is False diff --git a/tests/test_shared_embedder_table.py b/tests/test_shared_embedder_table.py new file mode 100644 index 0000000..4ac343f --- /dev/null +++ b/tests/test_shared_embedder_table.py @@ -0,0 +1,123 @@ +"""One 64 MB table per process, not one per session. + +Measured on a production host: an engine opened over an EMPTY vault +still cost 68 MB of RSS, and six live vaults held only TWO distinct +embedder tables between them. The table is `vocab_size × dim × 4B` — +64 MB at the defaults — so a host keeping ten sessions resident spent +640 MB on ten copies of identical numbers. That, not the stored +memories, was what made "keep every session awake" expensive. + +Sharing is only safe because the table is never written in place: it is +read, and when distillation adopts a better one the whole embedder is +replaced by a scratch instance holding its own array. These tests pin +both halves of that — the sharing, and the immutability it rests on. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +import numpy as np +import pytest + +from xgen_agent_memory.embedder import ( + HashEmbedder, + _TABLE_CACHE, + _TABLE_CACHE_MAX, + shared_table_stats, +) + + +@pytest.fixture(autouse=True) +def _clear_cache(): + _TABLE_CACHE.clear() + yield + _TABLE_CACHE.clear() + + +def test_same_parameters_share_one_array() -> None: + """The whole point: two engines, one allocation.""" + a = HashEmbedder(4096, 32, seed=41) + b = HashEmbedder(4096, 32, seed=41) + assert a.table is b.table, ( + "two embedders with identical parameters each allocated their own " + "table — this is the per-session 64 MB the change removes" + ) + assert shared_table_stats()["tables"] == 1 + + +def test_different_parameters_do_not_share() -> None: + """Sharing must be keyed on what actually determines the table.""" + base = HashEmbedder(4096, 32, seed=41) + assert HashEmbedder(4096, 32, seed=42).table is not base.table + assert HashEmbedder(8192, 32, seed=41).table is not base.table + assert HashEmbedder(4096, 64, seed=41).table is not base.table + assert shared_table_stats()["tables"] == 4 + + +def test_shared_table_is_read_only() -> None: + """A shared array that anyone can write is a cross-session bug waiting + to happen. Make the write fail loudly instead.""" + e = HashEmbedder(4096, 32, seed=41) + with pytest.raises(ValueError): + e.table[0, 0] = 1.0 + + +def test_sharing_does_not_change_embeddings() -> None: + """Behaviour must be byte-identical to the per-instance version.""" + a = HashEmbedder(4096, 32, seed=41) + b = HashEmbedder(4096, 32, seed=41) + for text in ("우주 전체 빛의 평균", "browsing files", "mixed 한글 and latin"): + assert np.array_equal(a.embed(text), b.embed(text)) + # And still matches a table generated the old way, from the same seed. + rng = np.random.default_rng(41) + expected = (rng.standard_normal((4096, 32)) / np.sqrt(32)).astype(np.float32) + assert np.array_equal(a.table, expected) + + +def test_persisted_tables_share_by_content() -> None: + """Sessions that saved the same table decode it once between them.""" + src = HashEmbedder(4096, 32, seed=7) + blob = src.dumps() + x = HashEmbedder.loads(blob) + y = HashEmbedder.loads(blob) + assert x.table is y.table + # A DIFFERENT saved table must not collide with it. + other = HashEmbedder.loads(HashEmbedder(4096, 32, seed=8).dumps()) + assert other.table is not x.table + + +def test_adopting_a_distilled_table_leaves_the_shared_one_alone() -> None: + """Copy-on-write in practice. + + `distill` builds its candidate on a scratch embedder and rebinds the + attribute. The engine's other sessions must not see that. + """ + shared = HashEmbedder(4096, 32, seed=41) + peer = HashEmbedder(4096, 32, seed=41) + original = np.array(shared.table, copy=True) + + scratch = HashEmbedder(4096, 32, seed=41) + scratch.table = np.zeros((4096, 32), dtype=np.float32) # what distill does + + assert peer.table is shared.table + assert np.array_equal(shared.table, original), ( + "adopting a distilled table mutated the array other sessions read" + ) + + +def test_cache_is_bounded() -> None: + """A dict keyed by content is a leak if content keeps changing.""" + for seed in range(_TABLE_CACHE_MAX + 4): + HashEmbedder(256, 8, seed=seed) + assert len(_TABLE_CACHE) <= _TABLE_CACHE_MAX + + +def test_stats_report_what_is_held() -> None: + HashEmbedder(4096, 32, seed=41) + s = shared_table_stats() + assert s["tables"] == 1 + assert s["bytes"] == 4096 * 32 * 4 diff --git a/tests/test_tokenizer_streams.py b/tests/test_tokenizer_streams.py new file mode 100644 index 0000000..9b7ed99 --- /dev/null +++ b/tests/test_tokenizer_streams.py @@ -0,0 +1,216 @@ +"""The two streams are deliberately different — effect-proving tests. + +`tokenizer.py` has always described a split: the LEXICAL stream feeds BM25 +and is precision-oriented; the EMBEDDING stream feeds the hash embedder and +is recall-oriented. Jamo n-grams were already on the recall side only, +"because they inflate postings ~3× and add vowel-noise to exact matching". + +Latin character trigrams sat on the wrong side of that line. They were +generated for every non-Hangul word ≥4 chars and were 72.8% of a production +vault's postings — 59.6% of that from fifty boilerplate types whose IDF is +about zero. They are now where jamo is: embedding only. + +What replaces them in BM25 is a real stemmer, and the measurements say that +is a better index, not just a smaller one: known-item MRR +4% verbatim and ++2% on re-inflected queries, with typo tolerance handed to the vector side. +""" + +from __future__ import annotations + +from xgen_agent_memory import SynapseConfig, SynapseMemory +from xgen_agent_memory.tokenizer import embed_tokens, lexical_tokens + +SENTENCE = "The user is browsing an item trading interface" + + +def _latin_trigrams(tokens): + """Character trigrams from Latin words — no marker prefix, length 3, + and not a whole word in the sentence.""" + words = {w.lower() for w in SENTENCE.split()} + return {t for t in tokens if len(t) == 3 and t.isalpha() and t.isascii() and t not in words} + + +# ── the split ─────────────────────────────────────────────────────── + + +def test_bm25_stream_has_no_latin_trigrams_by_default(): + """THE change. This is where 72.8% of the postings came from.""" + assert _latin_trigrams(lexical_tokens(SENTENCE)) == set() + + +def test_the_embedding_stream_keeps_them(): + """Typo tolerance moves here rather than disappearing — the same place + jamo already lives, and it costs no postings.""" + assert _latin_trigrams(embed_tokens(SENTENCE)), "the recall stream lost its fuzzy matching too" + + +def test_the_knob_can_bring_them_back(): + assert _latin_trigrams(lexical_tokens(SENTENCE, latin_ngram_min_len=4)) + + +# ── what BM25 gets instead ────────────────────────────────────────── + + +def test_the_bm25_stream_carries_stems(): + tokens = lexical_tokens(SENTENCE) + assert "brows" in tokens, "no stem — morphology would be unmatched" + assert "browsing" in tokens, "the surface form must survive too" + + +def test_an_inflected_query_finds_its_note(tmp_path): + """The property the stemmer is for. Without it the same query lost 16% + of its MRR on the production vault.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "a.db"), epsilon=0.0)) + mem.index("d1", "The user is browsing an item trading interface") + mem.index("d2", "저녁 메뉴로 김치찌개를 끓였다") + + assert [h.id for h in mem.search("browse trade", top_k=2)][0] == "d1" + + +def test_korean_is_untouched(tmp_path): + """Every measurement kept Korean flat (MRR 0.750–0.766); this pins the + tokens rather than trusting that.""" + tokens = lexical_tokens("리듬게임 판정을 읽는다") + assert "판정을" in tokens # surface + assert "판정" in tokens # 조사 stripped + assert any(len(t) == 2 and "가" <= t[0] <= "힣" for t in tokens) + + +# ── the digest must notice a tokenizer change ─────────────────────── + + +def test_changing_the_tokenizer_invalidates_the_index(tmp_path): + """Without this the vault silently becomes a mix of two tokenizations: + the digest says "unchanged", nothing re-indexes, and the stored postings + no longer agree with how queries are analysed.""" + path = str(tmp_path / "b.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item trading interface") + before = a.manifest()["d1"][1] + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0, latin_ngram_min_len=4)) + out = b.index_many( + [{"node_id": "d1", "text": "The user is browsing an item trading interface"}] + ) + + assert out["indexed"] == 1, "a tokenizer change did not trigger a re-index" + assert b.manifest()["d1"][1] != before + b.close() + + +def test_an_unrelated_config_change_does_not_reindex(tmp_path): + """The digest must not be so broad that every restart rebuilds the vault.""" + path = str(tmp_path / "c.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0, top_k=99)) + out = b.index_many([{"node_id": "d1", "text": "The user is browsing an item"}]) + + assert out["indexed"] == 0 + assert out["skipped"] == 1 + b.close() + + +def test_a_geometry_change_marks_every_row_stale(tmp_path): + """Putting the geometry into `content_sha` is not enough on its own: the + digest is only consulted for notes a HOST decides to offer, and a host + that diffs on timestamps never offers an untouched note. Production + upgraded to a new tokenizer and re-indexed nothing — "0 indexed" — while + every signal said the vault was in sync. + + An empty digest is the contract that fixes it: "indexed, derived state + unknown".""" + path = str(tmp_path / "g.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + a.index("d2", "리듬게임 판정을 읽는다") + assert all(sha for _ts, sha in a.manifest().values()) + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0, latin_ngram_min_len=4)) + assert all(sha == "" for _ts, sha in b.manifest().values()), ( + "a tokenizer change left the rows claiming to be up to date" + ) + b.close() + + +def test_reopening_with_the_same_geometry_keeps_the_digests(tmp_path): + """The flip side: an ordinary restart must not invalidate the vault.""" + path = str(tmp_path / "h.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + before = a.manifest()["d1"][1] + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + assert b.manifest()["d1"][1] == before + b.close() + + +def test_re_indexing_after_a_geometry_change_restores_the_digest(tmp_path): + path = str(tmp_path / "i.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0, latin_ngram_min_len=4)) + out = b.index_many([{"node_id": "d1", "text": "The user is browsing an item"}]) + assert out["indexed"] == 1 + assert b.manifest()["d1"][1] != "" + b.close() + + +def test_a_vault_with_no_recorded_geometry_is_treated_as_stale(tmp_path): + """ "Unknown" is not "fine". A vault written before the geometry was + tracked was derived by SOME tokenization nobody can name; assuming it + matches is how the production upgrade re-indexed nothing while every + signal said it was in sync. Derived data is rebuildable; a wrong + assumption of freshness is not.""" + path = str(tmp_path / "j.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + a.store._write(lambda c: c.execute("DELETE FROM params WHERE key='geometry'")) + assert a.manifest()["d1"][1] != "" + a.close() + + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + assert b.manifest()["d1"][1] == "", "a vault of unknown provenance was assumed up to date" + b.close() + + +def test_a_fresh_empty_vault_costs_nothing(tmp_path): + """The flip side: recording the geometry for the first time on an empty + file must not look like an invalidation.""" + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "k.db"), epsilon=0.0)) + mem.index("d1", "The user is browsing an item") + assert mem.manifest()["d1"][1] != "" + mem.close() + + +def test_the_geometry_version_is_part_of_the_fingerprint(tmp_path): + """The config fields cannot express "the stemmer changed". Without an + explicit version, 1.9.0 recorded the new geometry against rows derived + by the old tokenizer and every later release saw a match — the vault was + unfixable from inside.""" + from xgen_agent_memory import engine as eng + + mem = SynapseMemory(SynapseConfig(path=str(tmp_path / "v.db"), epsilon=0.0)) + assert mem._geometry().startswith(f"v{eng._GEOMETRY_VERSION}:") + mem.close() + + +def test_bumping_the_version_invalidates_a_matching_config(tmp_path, monkeypatch): + from xgen_agent_memory import engine as eng + + path = str(tmp_path / "w.db") + a = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + a.index("d1", "The user is browsing an item") + a.close() + + monkeypatch.setattr(eng, "_GEOMETRY_VERSION", eng._GEOMETRY_VERSION + 1) + b = SynapseMemory(SynapseConfig(path=path, epsilon=0.0)) + assert b.manifest()["d1"][1] == "" + b.close()