From af2b4b965675765f7b7a96f7dc07a0fc0d37dbf8 Mon Sep 17 00:00:00 2001 From: PureTensor Date: Sat, 17 Oct 2026 21:19:02 +0100 Subject: [PATCH 1/2] v0.1.5: large vaults: incremental keyword index, deferred compaction under watch, remembered model-load failures Measured on Linux (fox-n1), synthetic vault of 50,000 chunks (20,000), v0.1.4 vs this commit: - next search after a save (serve reload): 7.5 s -> 0.8 s (2.9 s -> 0.3 s) - end of each `puremind watch` run: 0.5 s -> 0.002 s (0.22 s -> 0.002 s); the 0.5 s compaction now runs once per quiet spell - keyword query: under 2 ms -> 14-27 ms (the price of query-time scoring) Reload: bm25s.BM25().index (6.5 s at 50k) is replaced by retrieval.Lexicon, BM25 scored at query time exactly as bm25s scores it (Lucene variant, k1 1.5, b 0.75, repeated query tokens count again), from per-chunk term ids and counts. LexiconBuilder keeps those for the chunks of the last snapshot, keyed by a blake2b digest of the text (chunk ids are reused after a delete), so a reload tokenizes only new chunks. It prunes text that left the index, gives each Lexicon its own vocabulary dict, and drops unused words once they are half the vocabulary. bm25s.tokenize is still used, so tokens are unchanged. Ties are now ordered by position. Store.load_index joins the vector blobs into one buffer. Watch: index_vault(purge=False) records 'purge_pending' in meta and only checkpoints; cmd_watch compacts on its first run, when watch() calls its new quiet hook 60 s after a run with no change since, at the latest 300 s after the first deferral, and on exit. `puremind index` compacts every run and pays a debt left by a crashed watch. purge() and clear() clear the record. Models: _LoadOnce remembers a ModelLoadError for LOAD_RETRY_AFTER (60 s) per instance; calls within it, including queued ones, get a fresh ModelLoadError with the same message and cause, so the indexer's cause check is unchanged. Tests: parity with bm25s scores on random corpora, incremental vs fresh builds, tokenization of new text only, pruning, vocabulary compaction with an older lexicon still answering, failed builds, tie order, a reload after one save tokenizing one chunk, purge_pending lifecycle, deferred compaction in index_vault and cmd_watch (first run, quiet, exit, cap), the quiet hook, and the model-load cool-down including overlapping callers. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011zd8G1cibJoQPVbZEpVLYs --- CHANGELOG.md | 8 ++ README.md | 2 +- SECURITY.md | 2 +- src/puremind_mac/__init__.py | 2 +- src/puremind_mac/cli.py | 35 +++++- src/puremind_mac/indexer.py | 19 +++- src/puremind_mac/models.py | 95 ++++++++++------ src/puremind_mac/retrieval.py | 198 ++++++++++++++++++++++++++-------- src/puremind_mac/service.py | 8 +- src/puremind_mac/store.py | 22 +++- src/puremind_mac/watcher.py | 17 ++- tests/test_concurrency.py | 9 +- tests/test_indexer.py | 13 +++ tests/test_models.py | 54 ++++++++++ tests/test_retrieval.py | 115 +++++++++++++++++++- tests/test_service.py | 20 ++++ tests/test_store.py | 20 ++++ tests/test_watcher.py | 94 +++++++++++++++- 18 files changed, 633 insertions(+), 100 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73fc661..5e10e6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 0.1.5 (unreleased) + +Faster with large vaults. Measured on Linux with a synthetic vault of 50,000 chunks (20,000 in brackets), against 0.1.4: + +- After a note changes, a running `puremind serve` reloads the index on its next search in 0.8 s (0.3 s), down from 7.5 s (2.9 s). The keyword index is no longer built again from all the text: only chunks the previous snapshot did not hold are tokenized, and the BM25 scores are computed when a search asks for them. This also applies to every search during `puremind index --rebuild`, which changes the index note by note. Scores are the same as before; chunks with exactly equal scores are now listed in index order, and a keyword search costs 15 to 30 ms instead of a few. The first search of a new server process takes about the same time as before. +- `puremind watch` no longer compacts the keyword index after every save, which took 0.5 s (0.2 s) and rewrote the whole keyword index each time. It now compacts once the vault has been quiet for a minute, at the latest five minutes after the change, and when it stops. The text of deleted and edited notes still leaves the index file at the end of the run that notices it; only their words stay in the keyword index until the compaction. A watch that crashes leaves a note in the index, so the next `puremind index` or `puremind watch` compacts. `puremind index` compacts at the end of every run, as before. +- A model that fails to load is not tried again for a minute. Every search used to try again, which could mean waiting for the Hugging Face Hub or reading the weights each time, and searches that overlapped tried one after another. They now get the same error at once. + ## 0.1.4 (2026-10-17) - A new embedding model, or a new `embedder_revision`, now needs `puremind index --rebuild`. `puremind index` without it stops with an error and changes nothing; 0.1.3 and earlier emptied the index and embedded every note again. A `puremind watch` still running on the old configuration, and a fresh `puremind index` on the new one, used to wipe each other's index in turn, with a full re-embed each time. Now the stale watch stops with an error and a hint to restart it, and the index stays as it is. An index that holds no notes takes a new model without `--rebuild`. diff --git a/README.md b/README.md index e5b53a4..fca2bb8 100644 --- a/README.md +++ b/README.md @@ -106,7 +106,7 @@ The reranker costs about 0.2 seconds per candidate on a CPU, so with the default Once both models are in the Hugging Face cache, pureMind makes no network requests: models load from the cache at their pinned revisions, and Hugging Face telemetry is turned off unless you have set `HF_HUB_DISABLE_TELEMETRY` yourself. The only network use is the one-time download, which sends your Hugging Face token if you have one configured. Set `HF_HUB_OFFLINE=1` to forbid downloads altogether; a model that is not downloaded then fails at once with a clear message. A cached copy that lacks a file the models need (after an interrupted download, or one made by another tool) is completed on the next run with network access, and reported by name when offline. -The index is a second copy of your notes' text. It is created readable only by you (mode 0600 in a 0700 directory), and an index, `~/.puremind` directory or `config.toml` that an earlier version left more open is tightened when pureMind next uses it. When a note is deleted or edited, the index run that notices it removes the old text from the index file and its `-wal` file; if another process is reading the index at that moment, the `-wal` file is emptied by a later run. Text deleted while you used 0.1.1 or earlier is removed when 0.1.2 first opens the index. Copies of the index in backups or snapshots still hold it; to remove everything, delete `~/.puremind/index.sqlite` and the `-wal` and `-shm` files beside it, then run `puremind index` again. +The index is a second copy of your notes' text. It is created readable only by you (mode 0600 in a 0700 directory), and an index, `~/.puremind` directory or `config.toml` that an earlier version left more open is tightened when pureMind next uses it. When a note is deleted or edited, the index run that notices it removes the old text from the index file and its `-wal` file; if another process is reading the index at that moment, the `-wal` file is emptied by a later run. The keyword index keeps the old words a little longer under `puremind watch`: it is compacted once the vault has been quiet for a minute, and at the latest five minutes after the change, rather than after every save. A watch that stops first compacts on its way out, and one that crashes leaves a note in the index so the next `puremind index` or `puremind watch` does it. `puremind index` compacts at the end of every run. Text deleted while you used 0.1.1 or earlier is removed when 0.1.2 first opens the index. Copies of the index in backups or snapshots still hold it; to remove everything, delete `~/.puremind/index.sqlite` and the `-wal` and `-shm` files beside it, then run `puremind index` again. ## Command line diff --git a/SECURITY.md b/SECURITY.md index 083266d..76c51e2 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -20,7 +20,7 @@ Security fixes go into the latest 0.1.x release. Upgrade to it before reporting. - **What `puremind_get` serves.** Only the notes the indexer reads: regular `.md` and `.markdown` files of at most 2 MiB that, after resolving symlinks, lie inside the vault, outside hidden folders, `node_modules` and `__pycache__`. Absolute paths, `..`, NUL characters and paths over 1,024 characters are refused before the file system is touched, and refusals never contain absolute paths. A path is tried as given and then without surrounding white space, and each spelling must pass every check. - **Network use.** The only network access is the one-time download of the two models from Hugging Face, at pinned revisions, which sends your Hugging Face token if one is configured. Once they are cached, models load from the cache and nothing contacts the network. pureMind turns Hugging Face telemetry off unless you have set `HF_HUB_DISABLE_TELEMETRY` yourself. Set `HF_HUB_OFFLINE=1` to forbid downloads altogether. A cached copy that lacks a file the models need is completed by a download, or refused when offline, so a partial copy is never loaded as if it were complete. - **Model loading.** A model name is a Hugging Face id unless it starts with `/`, `./`, `../` or `~`, so a directory in the working directory is never loaded by accident. Models load with `trust_remote_code` off, pureMind downloads only safetensors weights (no pickle files), and `puremind serve` runs from the pureMind home directory. `sentence-transformers` 5.6.0 or newer is required because older releases could run code from a local model directory (CVE-2026-68770). -- **The index holds your notes' text.** `~/.puremind/index.sqlite` is a second copy of every note's text. It and its `-wal` and `-shm` files are created mode 0600 in a 0700 directory, and existing files you own, including `~/.puremind` and `config.toml`, are tightened when pureMind uses them. The index run that notices a deleted or edited note removes the old text from the file and its `-wal` file (SQLite zeroes the freed pages, and the keyword index is compacted so it drops the deleted words). If another process is reading the index at that moment, the `-wal` file is emptied by a later run. Text deleted under 0.1.1 or earlier is removed when 0.1.2 first opens the index. Backups and snapshots of the index keep whatever they captured. To purge everything, delete the three files and re-index. +- **The index holds your notes' text.** `~/.puremind/index.sqlite` is a second copy of every note's text. It and its `-wal` and `-shm` files are created mode 0600 in a 0700 directory, and existing files you own, including `~/.puremind` and `config.toml`, are tightened when pureMind uses them. The index run that notices a deleted or edited note removes the old text from the file and its `-wal` file (SQLite zeroes the freed pages). The keyword index is compacted so it drops the deleted words at the end of that run under `puremind index`; under `puremind watch` this waits until the vault has been quiet for a minute, five minutes at most, or until the watch stops, and a watch that crashes leaves a note in the index so the next `index` or `watch` run does it. If another process is reading the index at that moment, the `-wal` file is emptied by a later run. Text deleted under 0.1.1 or earlier is removed when 0.1.2 first opens the index. Backups and snapshots of the index keep whatever they captured. To purge everything, delete the three files and re-index. - **Hostile vault content.** The indexer only reads the vault. Symlink loops, pipes, undecodable names, oversized files, a cap of 10,000 chunks per note and a 200-character cap on each heading in a breadcrumb bound the work a vault can cause. A note is checked again just before it is read, so a symlink swapped in during a run is not followed. - **Notes are untrusted input to an agent.** Note text returned to an agent is data, not instructions, but a model may still follow instructions written in a note. Keep untrusted material out of the vault you connect to your agents. - **MCP SDK advisories.** The published advisories for the `mcp` package releases this project allows concern its HTTP, SSE and WebSocket transports, or its experimental task handlers (CVE-2026-52870, mcp 1.23.0 to 1.27.1). pureMind uses stdio only and never enables tasks, so none of them apply. diff --git a/src/puremind_mac/__init__.py b/src/puremind_mac/__init__.py index 8a08aad..98b86f1 100644 --- a/src/puremind_mac/__init__.py +++ b/src/puremind_mac/__init__.py @@ -1,3 +1,3 @@ """pureMind for Mac: local memory for AI coding agents, served over MCP.""" -__version__ = "0.1.4" +__version__ = "0.1.5" diff --git a/src/puremind_mac/cli.py b/src/puremind_mac/cli.py index 62c238e..b43b44e 100644 --- a/src/puremind_mac/cli.py +++ b/src/puremind_mac/cli.py @@ -18,6 +18,7 @@ import sqlite3 import sys import threading +import time from collections.abc import Mapping, Sequence from pathlib import Path @@ -114,6 +115,13 @@ def cmd_index(args, cfg: Config) -> int: return 0 +# Under `puremind watch` the keyword index is compacted (Store.purge: about half a second at 50,000 +# chunks) once the vault has been quiet this long after a run, not after every save... +PURGE_WHEN_QUIET = 60.0 +# ...but at the latest this long after the first run that left words of removed text in it. +PURGE_AT_LEAST_EVERY = 300.0 + + def cmd_watch(args, cfg: Config) -> int: from .indexer import ModelMismatch, VaultMismatch, index_vault from .watcher import watch @@ -122,25 +130,46 @@ def cmd_watch(args, cfg: Config) -> int: svc = _service(dataclasses.replace(cfg, vault=vault)) stop = threading.Event() refused: list[ValueError] = [] + owed_since: list[float | None] = [None] # when a run first left the keyword index to compact + first = [True] def run() -> None: # always called on this thread, one run at a time + # The first run compacts, and so pays what a watch that crashed left owing. + now = time.monotonic() + purge = first[0] or (owed_since[0] is not None and now - owed_since[0] >= PURGE_AT_LEAST_EVERY) + first[0] = False try: - stats = index_vault(vault, svc.store, svc.embedder) + stats = index_vault(vault, svc.store, svc.embedder, purge=purge) except (VaultMismatch, ModelMismatch) as exc: # A symlinked vault that now names another directory, or an index rebuilt for another # model since this watch read its config: every later run would fail the same way. refused.append(exc) stop.set() return + if purge: + owed_since[0] = None + elif stats.purge_deferred and owed_since[0] is None: + owed_since[0] = now print(stats.summary(), flush=True) for item in stats.skipped: print(f" skipped: {_one_line(item)}", file=sys.stderr, flush=True) + def compact() -> None: + if svc.store.purge_pending(): + svc.store.purge() + owed_since[0] = None + print(f"watching {_one_line(str(vault.resolve()))} (Ctrl-C to stop)", flush=True) try: - watch(vault, run, delay=args.debounce, stop=stop) + watch(vault, run, delay=args.debounce, stop=stop, quiet=compact, quiet_after=PURGE_WHEN_QUIET) finally: - svc.store.close() + try: + compact() # skipped by a crash or power cut: the record in the index carries it to the next run + except Exception as exc: # noqa: BLE001 - must not hide why watching ended + print(f"warning: the keyword index was not compacted ({_one_line(exc)}); the next run does it", + file=sys.stderr) + finally: + svc.store.close() if refused: if isinstance(refused[0], ModelMismatch): return _fail( diff --git a/src/puremind_mac/indexer.py b/src/puremind_mac/indexer.py index e2f93e3..a94692f 100644 --- a/src/puremind_mac/indexer.py +++ b/src/puremind_mac/indexer.py @@ -61,6 +61,7 @@ class IndexStats: skipped: list[str] = field(default_factory=list) removal_skipped: str = "" # why notes missing from the scan were not removed from the index stopped: bool = False # the run was asked to stop before it finished + purge_deferred: bool = False # the keyword index still holds words of text this run removed def summary(self) -> str: s = ( @@ -420,6 +421,7 @@ def index_vault( rebuild: bool = False, *, should_stop: Callable[[], bool] | None = None, + purge: bool = True, ) -> IndexStats: """Bring the index in line with the vault. @@ -429,6 +431,12 @@ def index_vault( vectors from two models cannot be searched together. Notes cut by an earlier chunker are cut and embedded again. ``should_stop`` is polled between notes; once it returns True the run ends early without removing anything. + + Text a run removes or replaces leaves the tables at once. Its words leave the keyword index + when the index is compacted (``Store.purge``), which takes about half a second at 50,000 + chunks. With ``purge`` the run compacts at its end, also paying a compaction an earlier run + put off; without it, the run only records that one is owed (``stats.purge_deferred``) and the + caller pays it later, as ``puremind watch`` does. """ vault = vault.expanduser().resolve() if not vault.is_dir(): @@ -488,7 +496,16 @@ def index_vault( for rel in gone: store.delete_file(rel) stats.removed += 1 - if (stats.removed or stats.updated) and not store.purge(): # clear() purges by itself + emptied = True + if stats.removed or stats.updated: # clear() purges by itself + if purge: + emptied = store.purge() + else: + emptied = store.defer_purge() + stats.purge_deferred = True + elif purge and store.purge_pending(): # owed by a `puremind watch` that stopped before paying it + emptied = store.purge() + if not emptied: log.warning( "another process was reading the index, so text deleted or replaced in this run stays in " "%s-wal until the next run", store.path, diff --git a/src/puremind_mac/models.py b/src/puremind_mac/models.py index f8a76d2..e9f615e 100644 --- a/src/puremind_mac/models.py +++ b/src/puremind_mac/models.py @@ -17,7 +17,8 @@ import logging import os import threading -from collections.abc import Sequence +import time +from collections.abc import Callable, Sequence from pathlib import Path from typing import TYPE_CHECKING, Protocol @@ -39,6 +40,10 @@ log = logging.getLogger("puremind.models") +# How long a model that failed to load is not tried again: each try can wait seconds for the Hub +# or read gigabytes of weights, and searches that overlap would otherwise each try in turn. +LOAD_RETRY_AFTER = 60.0 + # Revision used when none is configured for a default model. A new embedder revision here makes # `index` and `watch` refuse every existing index until `puremind index --rebuild`, so a release # that changes it must say so. @@ -272,6 +277,35 @@ def resolve_model(name: str, revision: str | None = None, *, namespace: str | No return os.path.abspath(path) +class _LoadOnce: + """Loads a model on first use, one thread at a time, and remembers a failure for a while. + + Within LOAD_RETRY_AFTER seconds of a failed load, every caller gets the same error at once, + with the same cause, instead of trying again. The memory is per instance, so a new process + always tries. + """ + + def __init__(self, load: Callable[[], object]): + self._load = load + self._model = None + self._failed: tuple[float, ModelLoadError] | None = None + self._lock = threading.Lock() + + def get(self): + with self._lock: + if self._model is None: + if self._failed is not None and time.monotonic() - self._failed[0] < LOAD_RETRY_AFTER: + error = self._failed[1] + raise ModelLoadError(str(error)) from error.__cause__ + try: + self._model = self._load() + except ModelLoadError as exc: + self._failed = (time.monotonic(), exc) + raise + self._failed = None + return self._model + + def _normalise(v: np.ndarray) -> np.ndarray: import numpy as np @@ -311,28 +345,25 @@ def __init__( self._revision = revision # What resolve_model loads: the pinned commit of a default model when none is given. self.revision = None if is_local_path(model) else revision or PINNED_REVISIONS.get(model) - self._model = None - self._lock = threading.Lock() + self._model = _LoadOnce(self._open) # Searches overlap on MCP worker threads, but one model must not run two calls at once: its # fast tokenizer raises "Already borrowed" when two threads use it, and devices such as mps # are not safe for concurrent forward passes on one model. self._busy = threading.Lock() - def _load(self): - with self._lock: - if self._model is None: - path = resolve_model(self.name, self._revision, namespace="sentence-transformers") - try: - from sentence_transformers import SentenceTransformer + def _open(self): + path = resolve_model(self.name, self._revision, namespace="sentence-transformers") + try: + from sentence_transformers import SentenceTransformer - m = SentenceTransformer( - path, device=pick_device(self._device), trust_remote_code=False, local_files_only=True - ) - m.max_seq_length = self._max_seq - except Exception as exc: - raise ModelLoadError(f"could not load model {model_label(self.name)} ({type(exc).__name__})") from exc - self._model = m - return self._model + m = SentenceTransformer(path, device=pick_device(self._device), trust_remote_code=False, local_files_only=True) + m.max_seq_length = self._max_seq + except Exception as exc: + raise ModelLoadError(f"could not load model {model_label(self.name)} ({type(exc).__name__})") from exc + return m + + def _load(self): + return self._model.get() def embed_documents(self, texts: Sequence[str]) -> np.ndarray: import numpy as np @@ -375,25 +406,23 @@ def __init__(self, model: str, device: str = "auto", query_chars: int = 1000, *, self._device = device self._query_chars = query_chars self._revision = revision - self._model = None - self._lock = threading.Lock() + self._model = _LoadOnce(self._open) self._busy = threading.Lock() # one call at a time, as for the embedder - def _load(self): - with self._lock: - if self._model is None: - path = resolve_model(self.name, self._revision) - try: - from sentence_transformers import CrossEncoder + def _open(self): + path = resolve_model(self.name, self._revision) + try: + from sentence_transformers import CrossEncoder - m = CrossEncoder( - path, max_length=512, device=pick_device(self._device), - trust_remote_code=False, local_files_only=True, - ) - except Exception as exc: - raise ModelLoadError(f"could not load model {model_label(self.name)} ({type(exc).__name__})") from exc - self._model = m - return self._model + m = CrossEncoder( + path, max_length=512, device=pick_device(self._device), trust_remote_code=False, local_files_only=True + ) + except Exception as exc: + raise ModelLoadError(f"could not load model {model_label(self.name)} ({type(exc).__name__})") from exc + return m + + def _load(self): + return self._model.get() def score(self, query: str, passages: Sequence[str]) -> list[float]: import numpy as np diff --git a/src/puremind_mac/retrieval.py b/src/puremind_mac/retrieval.py index 4586d86..c7814c1 100644 --- a/src/puremind_mac/retrieval.py +++ b/src/puremind_mac/retrieval.py @@ -8,9 +8,10 @@ from __future__ import annotations +import hashlib import re import threading -from collections.abc import Sequence +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass import numpy as np @@ -31,6 +32,9 @@ _TOKEN_PATTERN = rf"(?u)\b\w\w+\b|[{_CJK}]" _UNBUILT = object() # BM25 index not built yet; None means built, with nothing to match +# bm25s's defaults (its Lucene variant), with which the retrieval benchmark was run. +BM25_K1 = 1.5 +BM25_B = 0.75 def _bigrams(run: re.Match[str]) -> str: @@ -58,6 +62,134 @@ def _segment_cjk(text: str, unigrams: bool = False) -> str: return _CJK_RUN.sub(_unigrams_and_bigrams if unigrams else _bigrams, text) +def tokenize(docs: list[str], query: bool = False): + """bm25s tokens (a ``Tokenized``) of ``docs``: English-stemmed words without stopwords, CJK + runs cut as described in ``_segment_cjk``.""" + import bm25s + import Stemmer + + # A PyStemmer instance is stateful and must not be used by two threads at once, so + # every call makes its own (construction is free). + return bm25s.tokenize( + [_segment_cjk(d, unigrams=not query) for d in docs], + token_pattern=_TOKEN_PATTERN, + stopwords="en", + stemmer=Stemmer.Stemmer("english"), + show_progress=False, + ) + + +def _words(tokens) -> list[list[str]]: + """The token strings of each document in a bm25s ``Tokenized``.""" + spelled = {i: word for word, i in tokens.vocab.items()} + return [[spelled[i] for i in ids] for ids in tokens.ids] + + +class Lexicon: + """BM25 over a fixed list of documents, scored at query time as bm25s scores it. + + Each document is given as its term ids and their counts, so making one is a few array + operations and needs no index of its own: a snapshot of a large vault is ready at once. + Read-only once made, so several threads may search it. + """ + + def __init__(self, vocab: Mapping[str, int], docs: Sequence[tuple[np.ndarray, np.ndarray]]): + self.vocab = vocab # never changed afterwards: a newer Lexicon gets a dict of its own + self.size = len(docs) + widths = np.array([len(terms) for terms, _ in docs], dtype=np.int64) + self._terms = np.concatenate([terms for terms, _ in docs]) if docs else np.zeros(0, np.int32) + counts = np.concatenate([n for _, n in docs]).astype(np.float64) if docs else np.zeros(0) + self._docs = np.repeat(np.arange(self.size, dtype=np.int32), widths) + lengths = np.bincount(self._docs, weights=counts, minlength=self.size) # tokens per document + df = np.bincount(self._terms, minlength=len(vocab)).astype(np.float64) + self._idf = np.log(1 + (self.size - df + 0.5) / (df + 0.5)) + norm = BM25_K1 * (1 - BM25_B + BM25_B * lengths / max(lengths.mean() if self.size else 0.0, 1e-12)) + # The term-frequency part of each entry's score, in float32 as bm25s keeps its scores. + self._tfc = (counts / (counts + norm[self._docs])).astype(np.float32) + + def __bool__(self) -> bool: + return self._terms.size > 0 + + def scores(self, words: Sequence[str]) -> np.ndarray: + """The BM25 score of every document for the query tokens ``words``. A repeated token counts + again and a token no document has adds nothing, as in bm25s.""" + ids = [self.vocab[w] for w in words if w in self.vocab] + if not ids: + return np.zeros(self.size) + query, repeats = np.unique(np.asarray(ids, dtype=np.int32), return_counts=True) + hit = np.flatnonzero(np.isin(self._terms, query, kind="table")) + terms = self._terms[hit] + weight = self._idf[terms] * repeats[np.searchsorted(query, terms)] * self._tfc[hit] + return np.bincount(self._docs[hit], weights=weight, minlength=self.size) + + def top(self, words: Sequence[str], k: int) -> list[int]: + """Positions of the ``k`` best documents with a positive score, best first, ties in position order.""" + if k <= 0: + return [] + scores = self.scores(words) + found = np.flatnonzero(scores > 0) + return [int(j) for j in found[np.lexsort((found, -scores[found]))][:k]] + + +def _digest(text: str) -> bytes: + return hashlib.blake2b(text.encode("utf-8", "surrogatepass"), digest_size=16).digest() + + +class LexiconBuilder: + """Makes the Lexicon of each new snapshot, tokenizing only text the previous one did not hold. + + It keeps the term ids and counts of the last snapshot's documents, keyed by a digest of their + text (chunk ids are reused after a delete), and nothing of text that has left the index. Not + thread-safe: the caller makes one Lexicon at a time. + """ + + def __init__(self) -> None: + self._vocab: dict[str, int] = {} + self._docs: dict[bytes, tuple[np.ndarray, np.ndarray]] = {} + self._compact_at = 4096 # vocabulary size at which terms no document uses are looked for + + def build(self, texts: Sequence[str], tokenizer: Callable | None = None) -> Lexicon | None: + """The Lexicon of ``texts``, or None if none of them has a single token. ``tokenizer`` + defaults to ``tokenize``.""" + keys = [_digest(t) for t in texts] + docs = {key: self._docs[key] for key in keys if key in self._docs} + new = {key: text for key, text in zip(keys, texts) if key not in docs} + vocab = self._vocab + if new: + tokens = (tokenizer or tokenize)(list(new.values())) + if any(word not in vocab for word in tokens.vocab): + vocab = dict(vocab) # the Lexicon already made keeps the dict it was made with + for word in tokens.vocab: + vocab.setdefault(word, len(vocab)) + ours = np.zeros(len(tokens.vocab), dtype=np.int32) + for word, i in tokens.vocab.items(): + ours[i] = vocab[word] + for key, ids in zip(new, tokens.ids): + terms, counts = np.unique(ours[np.asarray(ids, dtype=np.int64)], return_counts=True) + docs[key] = (terms.astype(np.int32), counts.astype(np.int32)) + if len(vocab) >= self._compact_at: + vocab, docs = _compact(vocab, docs) + self._compact_at = max(4096, 2 * len(vocab)) + self._vocab, self._docs = vocab, docs # only now: a failed build leaves the last state intact + lexicon = Lexicon(vocab, [docs[key] for key in keys]) + return lexicon if lexicon else None + + +def _compact( + vocab: dict[str, int], docs: dict[bytes, tuple[np.ndarray, np.ndarray]] +) -> tuple[dict[str, int], dict[bytes, tuple[np.ndarray, np.ndarray]]]: + """Drop the terms no document uses any more (words of deleted or edited text), if they are at + least half the vocabulary, and number the rest again.""" + used = np.zeros(len(vocab), dtype=bool) + for terms, _ in docs.values(): + used[terms] = True + if used.sum() * 2 > len(vocab): + return vocab, docs + renumber = np.cumsum(used, dtype=np.int64).astype(np.int32) - 1 # keeps the order of the ids + vocab = {word: int(renumber[i]) for word, i in vocab.items() if used[i]} + return vocab, {key: (renumber[terms], counts) for key, (terms, counts) in docs.items()} + + def rrf_fuse(rankings: Sequence[Sequence[int]], k: int = 60) -> list[tuple[int, float]]: """Reciprocal rank fusion. Ranks are 1-based; ties keep first-seen order.""" fused: dict[int, float] = {} @@ -76,68 +208,42 @@ class Scored: class HybridIndex: """An in-memory corpus: texts plus an (n, dim) matrix of normalised vectors. - Searches may overlap on several threads. The BM25 index is built on first use, by one - thread, and is visible to the others only once complete; everything else is read-only. + Searches may overlap on several threads. ``lexicon`` is the BM25 index of ``texts`` when the + caller has made it (see LexiconBuilder); otherwise it is made on first use, by one thread, + and is visible to the others only once complete. Everything else is read-only. """ - def __init__(self, texts: Sequence[str], vectors: np.ndarray | None): + def __init__(self, texts: Sequence[str], vectors: np.ndarray | None, lexicon: Lexicon | None | object = _UNBUILT): self.texts = list(texts) self.vectors = vectors - self._bm25: object = _UNBUILT + self._lex = lexicon self._build_lock = threading.Lock() def __len__(self) -> int: return len(self.texts) def _tokenize(self, docs: list[str], query: bool = False): - import bm25s - import Stemmer - - # A PyStemmer instance is stateful and must not be used by two threads at once, so - # every call makes its own (construction is free). - return bm25s.tokenize( - [_segment_cjk(d, unigrams=not query) for d in docs], - token_pattern=_TOKEN_PATTERN, - stopwords="en", - stemmer=Stemmer.Stemmer("english"), - show_progress=False, - ) - - def _lexicon(self): - """The BM25 index, built once; None if no chunk contains a token.""" - bm25 = self._bm25 - if bm25 is _UNBUILT: + return tokenize(docs, query) + + def _lexicon(self) -> Lexicon | None: + """The BM25 index, made once; None if no chunk contains a token.""" + lex = self._lex + if lex is _UNBUILT: with self._build_lock: - bm25 = self._bm25 - if bm25 is _UNBUILT: - bm25 = self._build_bm25() - self._bm25 = bm25 # published only once complete - return bm25 - - def _build_bm25(self): - import bm25s - - tokens = self._tokenize(self.texts) - if not tokens.vocab: # only emoji, one-letter words or stopwords: bm25s cannot index that - return None - bm25 = bm25s.BM25() - bm25.index(tokens, show_progress=False) - return bm25 + lex = self._lex + if lex is _UNBUILT: + lex = LexiconBuilder().build(self.texts, self._tokenize) + self._lex = lex # published only once complete + return lex def lexical(self, query: str, k: int) -> list[int]: """Positions of the top-k BM25 matches with a positive score.""" if not self.texts or k <= 0: return [] - bm25 = self._lexicon() - if bm25 is None: - return [] - try: - res, scores = bm25.retrieve( - self._tokenize([query], query=True), k=min(k, len(self.texts)), show_progress=False - ) - except (ValueError, IndexError): # query reduced to nothing (all stopwords) + lex = self._lexicon() + if lex is None: return [] - return [int(j) for j, s in zip(res[0], scores[0]) if s > 0] + return lex.top(_words(self._tokenize([query], query=True))[0], min(k, len(self.texts))) def dense(self, qvec: np.ndarray, k: int) -> list[int]: if self.vectors is None or len(self.vectors) == 0 or k <= 0: diff --git a/src/puremind_mac/service.py b/src/puremind_mac/service.py index d1d6a9e..d9568bc 100644 --- a/src/puremind_mac/service.py +++ b/src/puremind_mac/service.py @@ -15,7 +15,7 @@ model_key, model_label, ) -from .retrieval import HybridIndex +from .retrieval import HybridIndex, LexiconBuilder from .store import ChunkRow, Store @@ -66,6 +66,7 @@ def __init__( ) self._lock = threading.Lock() self._snap = _Snapshot(-1, (), HybridIndex([], None)) + self._lexicons = LexiconBuilder() # used under _lock only def _loaded(self) -> _Snapshot: """The current snapshot, reloaded from the store if its generation has moved on.""" @@ -73,7 +74,10 @@ def _loaded(self) -> _Snapshot: gen = self.store.generation() # read first: a write racing the load only costs one extra reload if gen != self._snap.gen: chunks, vectors = self.store.load_index() - index = HybridIndex([c.content for c in chunks], vectors if chunks else None) + texts = [c.content for c in chunks] + # Only chunks the last snapshot did not hold are tokenized, so a reload after one + # saved note costs little even in a large vault. + index = HybridIndex(texts, vectors if chunks else None, self._lexicons.build(texts)) self._snap = _Snapshot(gen, tuple(chunks), index) return self._snap diff --git a/src/puremind_mac/store.py b/src/puremind_mac/store.py index b1f3de0..b77b1ea 100644 --- a/src/puremind_mac/store.py +++ b/src/puremind_mac/store.py @@ -375,8 +375,22 @@ def purge(self) -> bool: self._check_writable() with self.db: self.db.execute("INSERT INTO chunks_fts(chunks_fts) VALUES ('optimize')") + self.db.execute("DELETE FROM meta WHERE key = 'purge_pending'") return self.checkpoint() + def defer_purge(self) -> bool: + """Note in the index that ``purge`` is still owed, so that a process that stops before + paying it (a crash, a power cut) leaves the debt to the next one, and empty the -wal file. + The deleted chunks' text is already gone from the tables; only the keyword index still + holds their words. Returns what ``checkpoint`` returns.""" + self._check_writable() + with self.db: + self.db.execute("INSERT INTO meta(key, value) VALUES ('purge_pending', '1') ON CONFLICT(key) DO NOTHING") + return self.checkpoint() + + def purge_pending(self) -> bool: + return self.get_meta("purge_pending") is not None + def __enter__(self) -> Self: return self @@ -468,6 +482,7 @@ def clear(self) -> None: with self.db: self.db.execute("DELETE FROM files") self.db.execute("INSERT INTO chunks_fts(chunks_fts) VALUES ('rebuild')") # empties the keyword index outright + self.db.execute("DELETE FROM meta WHERE key = 'purge_pending'") self._bump() if self.path != ":memory:": with self.db: # no transaction is open here, which VACUUM needs @@ -492,8 +507,11 @@ def load_index(self) -> tuple[list[ChunkRow], np.ndarray]: chunks = [ChunkRow(r["id"], r["path"], r["chunk_index"], r["heading_path"], r["content"]) for r in rows] if not rows: return chunks, np.zeros((0, 0), dtype=np.float32) - mat = np.stack([np.frombuffer(r["vec"], dtype=np.float16) for r in rows]).astype(np.float32) - return chunks, mat + dims = {r["dim"] for r in rows} + if len(dims) != 1 or any(len(r["vec"]) != 2 * r["dim"] for r in rows): + raise ValueError("the vectors in the index do not all have the same length; run `puremind index --rebuild`") + mat = np.frombuffer(b"".join(r["vec"] for r in rows), dtype=np.float16).reshape(len(rows), dims.pop()) + return chunks, mat.astype(np.float32) def fts_search(self, query: str, limit: int = 50) -> list[tuple[int, float]]: """FTS5 lexical search. Returns (chunk_id, bm25 score) with larger meaning better.""" diff --git a/src/puremind_mac/watcher.py b/src/puremind_mac/watcher.py index f1dda36..c1c96fd 100644 --- a/src/puremind_mac/watcher.py +++ b/src/puremind_mac/watcher.py @@ -160,6 +160,8 @@ def watch( delay: float = 2.0, stop: threading.Event | None = None, rescan: float = RESCAN, + quiet: Callable[[], None] | None = None, + quiet_after: float = 60.0, ) -> None: """Run ``run_index`` once, then again each time the vault has been quiet for ``delay`` seconds. @@ -168,8 +170,10 @@ def watch( except the first, whose error propagates. If the vault directory is replaced or made again (a remount, or a sync tool swapping it), the new directory is watched and indexed; while it is missing or cannot be read, nothing runs. After ``rescan`` seconds without a - run the vault is indexed anyway, in case the observer missed an event. Returns when - ``stop`` is set. Ctrl-C raises KeyboardInterrupt here after the observer has been shut down. + run the vault is indexed anyway, in case the observer missed an event. ``quiet`` is called + once ``quiet_after`` seconds after a run when no change has come in since, on the same + thread; it fails as a run does. Returns when ``stop`` is set. Ctrl-C raises + KeyboardInterrupt here after the observer has been shut down. """ stop = stop or threading.Event() settle = Settle(delay) @@ -204,7 +208,15 @@ def deaf() -> bool: if not stop.is_set(): run_index() last_run = time.monotonic() + quiet_due = last_run + quiet_after if quiet else None while not stop.is_set(): + if quiet_due is not None and time.monotonic() >= quiet_due and settle.remaining() is None: + quiet_due = None + try: + quiet() + except Exception as exc: # keep watching, as after a failed run + log.error("%s", _one_line(exc)) + log.debug("quiet-time work failed", exc_info=True) current = _identity(root) recreated = current is not None and current == identity and armed and deaf() if current != identity or recreated or (current is not None and not armed): @@ -239,6 +251,7 @@ def deaf() -> bool: log.error("index run failed: %s", _one_line(exc)) log.debug("index run failed", exc_info=True) last_run = time.monotonic() + quiet_due = last_run + quiet_after if quiet else None finally: observer.stop() observer.join(timeout=5) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index eb12fa8..fc25de1 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -16,6 +16,7 @@ import pytest from fakes import FakeReranker +from puremind_mac import retrieval from puremind_mac.indexer import index_vault from puremind_mac.retrieval import HybridIndex from puremind_mac.server import build_server @@ -55,16 +56,16 @@ def run_threads(threads, timeout=60): def slow_corpus_tokenizer(monkeypatch, delay=0.15): """Make building a BM25 index slow, so every other thread arrives while it is being built.""" - real = HybridIndex._tokenize + real = retrieval.tokenize builds = [] - def tokenize(self, docs, query=False): + def tokenize(docs, query=False): if not query: # the corpus builds.append(len(docs)) time.sleep(delay) - return real(self, docs, query) + return real(docs, query) - monkeypatch.setattr(HybridIndex, "_tokenize", tokenize) + monkeypatch.setattr(retrieval, "tokenize", tokenize) return builds diff --git a/tests/test_indexer.py b/tests/test_indexer.py index d86ed87..b621e2f 100644 --- a/tests/test_indexer.py +++ b/tests/test_indexer.py @@ -793,3 +793,16 @@ def test_a_reader_holding_the_log_gets_a_warning_instead_of_a_stall(vault, priva assert stats.removed == 1 and took < 10 # the checkpoint does not wait out the 30 s lock timeout assert any("until the next run" in r.getMessage() for r in caplog.records) assert private_index.checkpoint() # with the reader gone, the log is emptied + + +def test_a_run_can_leave_the_keyword_index_compaction_to_the_next_purging_run(vault, private_index, embedder): + home = Path(private_index.path).parent + (vault / "secret.md").write_text("# Secret\n\nthe code is zqxcanary7f3a9\n") + index_vault(vault, private_index, embedder) + (vault / "secret.md").unlink() + stats = index_vault(vault, private_index, embedder, purge=False) + assert stats.removed == 1 and stats.purge_deferred and private_index.purge_pending() + assert not any("zqxcanary7f3a9" in c.content for c in private_index.load_index()[0]) # the text itself is gone + stats = index_vault(vault, private_index, embedder) # nothing changed, but the compaction is owed + assert (stats.removed, stats.updated, stats.purge_deferred) == (0, 0, False) + assert not private_index.purge_pending() and _copies(home, b"zqxcanary7f3a9") == 0 diff --git a/tests/test_models.py b/tests/test_models.py index dc1c729..a7a189c 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -8,6 +8,8 @@ import socket import subprocess import sys +import threading +import time import types import urllib.request from pathlib import Path @@ -615,3 +617,55 @@ def test_real_models_rank_the_relevant_passage_first(monkeypatch): reranker = CrossEncoderReranker(DEFAULT_RERANKER, os.environ.get("PUREMIND_DEVICE", "cpu")) good, bad = reranker.score("how often do I descale the espresso machine", ["Descale the boiler every three months.", "Plant tomatoes in May."]) assert 0.0 <= bad < good <= 1.0 + + +# ---- a model that failed to load is not tried again at once + + +@pytest.mark.parametrize("make", [SentenceTransformerEmbedder, CrossEncoderReranker]) +def test_a_failed_load_is_remembered_and_tried_again_later(stubs, monkeypatch, make): + monkeypatch.setenv("HF_HUB_OFFLINE", "1") # not cached and offline: every try fails + model = make(DEFAULT_EMBEDDER if make is SentenceTransformerEmbedder else DEFAULT_RERANKER) + use = (lambda: model.embed_query("q")) if make is SentenceTransformerEmbedder else (lambda: model.score("q", ["p"])) + errors = [] + for _ in range(3): + with pytest.raises(ModelLoadError) as excinfo: + use() + errors.append(excinfo.value) + assert len(stubs.calls) == 1 # one look at the cache, not three + assert len({str(e) for e in errors}) == 1 and "not downloaded yet" in str(errors[0]) + assert all(type(e.__cause__) is type(errors[0].__cause__) for e in errors) # the cause line stays the same + monkeypatch.setattr(models, "LOAD_RETRY_AFTER", 0.0) + with pytest.raises(ModelLoadError): + use() + assert len(stubs.calls) == 2 # after the cool-down it tries again + monkeypatch.delenv("HF_HUB_OFFLINE") + stubs.cached.add((DEFAULT_EMBEDDER, DEFAULT_EMBEDDER_REVISION)) + stubs.cached.add((DEFAULT_RERANKER, DEFAULT_RERANKER_REVISION)) + use() # and once the model is there, it loads + + +def test_overlapping_calls_during_a_failing_load_try_it_once(): + calls = [] + + def load(): + calls.append(1) + time.sleep(0.2) + raise ModelLoadError("could not load model x") from OSError("disk") + + once = models._LoadOnce(load) + errors = [] + + def call(): + try: + once.get() + except ModelLoadError as exc: + errors.append(exc) + + threads = [threading.Thread(target=call) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(20) + assert len(calls) == 1 and len(errors) == 8 + assert all(isinstance(e.__cause__, OSError) for e in errors) diff --git a/tests/test_retrieval.py b/tests/test_retrieval.py index b48439d..42c9fc8 100644 --- a/tests/test_retrieval.py +++ b/tests/test_retrieval.py @@ -9,11 +9,14 @@ import Stemmer from fakes import FakeEmbedder, FakeReranker +from puremind_mac import retrieval from puremind_mac.retrieval import ( _CJK, _CJK_RANGES, HybridIndex, + LexiconBuilder, _segment_cjk, + _words, rrf_fuse, ) @@ -311,4 +314,114 @@ def test_english_lexical_ranking_is_unchanged(): idx, _ = make_index() for q in ("espresso grinder", "tomato garden compost", "the build pipeline", "boiler"): res, scores = legacy.retrieve(legacy_tokenize([q]), k=len(DOCS), show_progress=False) - assert idx.lexical(q, len(DOCS)) == [int(j) for j, s in zip(res[0], scores[0]) if s > 0] + ranked = sorted(zip(scores[0].tolist(), res[0].tolist()), key=lambda t: (-t[0], t[1])) # ties by position + assert idx.lexical(q, len(DOCS)) == [j for s, j in ranked if s > 0] + + +# ---- the BM25 index scores as bm25s does, and is rebuilt from what changed + +WORDS = [ + "the", "a", "of", "and", "espresso", "grinder", "boiler", "tomatoes", "garden", "compost", "pipeline", + "containers", "backflush", "weekly", "raised", "beds", "descale", "months", "beans", "burr", "setting", + "migrate", "hermetic", "build", "kubernetes", "cluster", "deploy", + *(f"term{i}" for i in range(60)), "数据库", "服务器", "猫", "서울", "サーバー", +] + + +def random_texts(rng, n): + texts = [] + for _ in range(n): + size = rng.choice([0, 1, 3, 12, 40, 120]) + texts.append(" ".join(rng.choice(WORDS) for _ in range(size))) + return texts + + +def query_words(query): + return _words(retrieval.tokenize([query], query=True))[0] + + +QUERIES = ["espresso", "espresso espresso grinder", "the of", "term3 term3 term7 boiler", "unknownword", "猫 服务器", + "tomatoes garden compost raised beds pipeline", ""] + + +@pytest.mark.parametrize("seed", range(6)) +def test_lexicon_scores_match_bm25s(seed): + rng = random.Random(seed) + texts = random_texts(rng, 200) + ["espresso grinder"] # at least one token + reference = bm25s.BM25() + reference.index(retrieval.tokenize(texts), show_progress=False) + lexicon = LexiconBuilder().build(texts) + for query in QUERIES: + words = query_words(query) + ours = lexicon.scores(words) + theirs = reference.get_scores(words) if words else np.zeros(len(texts)) + assert np.allclose(ours, theirs, rtol=1e-5, atol=1e-6), query + + +def test_a_lexicon_built_from_the_previous_one_scores_like_a_fresh_build(): + rng = random.Random(7) + builder = LexiconBuilder() + before = random_texts(rng, 150) + builder.build(before) + after = before[20:] + random_texts(rng, 30) # some chunks gone, some new + after[5] += " espresso" # one edited + built, fresh = builder.build(after), LexiconBuilder().build(after) + for query in QUERIES: + words = query_words(query) + assert np.allclose(built.scores(words), fresh.scores(words), rtol=1e-12, atol=0) + assert built.top(words, 20) == fresh.top(words, 20) + + +def test_the_builder_tokenizes_only_text_it_has_not_seen(): + seen = [] + + def counting(docs, query=False): + seen.append(list(docs)) + return retrieval.tokenize(docs, query) + + builder = LexiconBuilder() + builder.build(["alpha one", "beta two", "alpha one"], counting) + assert seen == [["alpha one", "beta two"]] # a repeated text is tokenized once + builder.build(["alpha one", "beta two", "gamma three"], counting) + assert seen[-1] == ["gamma three"] + builder.build(["gamma three", "alpha one"], counting) + assert len(seen) == 2 # nothing new + + +def test_the_builder_keeps_nothing_of_text_that_left_the_index(): + builder = LexiconBuilder() + builder.build(["the code is zqxcanary", "garden plan"]) + lexicon = builder.build(["garden plan"]) + assert len(builder._docs) == 1 + assert lexicon.top(query_words("zqxcanary"), 5) == [] + + +def test_unused_words_are_dropped_and_an_older_lexicon_keeps_working(): + builder = LexiconBuilder() + first = None + for round_ in range(4): # every round replaces the whole vault with 3,000 words not seen before + texts = [" ".join(f"w{round_}x{i}y{j}" for j in range(10)) for i in range(300)] + lexicon = builder.build(texts) + assert lexicon.top(query_words(texts[3]), 1) == [3] + assert len(builder._vocab) <= 6000 # words of earlier rounds do not pile up + if first is None: + first, first_texts = lexicon, texts + assert first.top(query_words(first_texts[7]), 1) == [7] # a search still holding the first snapshot + + +def test_a_failed_build_leaves_the_builder_as_it_was(): + builder = LexiconBuilder() + builder.build(["alpha beta"]) + + def failing(docs, query=False): + raise MemoryError("simulated") + + with pytest.raises(MemoryError): + builder.build(["alpha beta", "gamma delta"], failing) + lexicon = builder.build(["alpha beta", "gamma delta"]) + assert lexicon.top(query_words("gamma"), 5) == [1] and lexicon.top(query_words("alpha"), 5) == [0] + + +def test_ties_are_ordered_by_position(): + lexicon = LexiconBuilder().build(["espresso machine", "espresso boiler", "espresso cups", "garden"]) + assert lexicon.top(query_words("espresso"), 3) == [0, 1, 2] diff --git a/tests/test_service.py b/tests/test_service.py index de1c6e6..c942fd6 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -104,3 +104,23 @@ def test_an_index_built_at_another_embedder_revision_is_refused(cfg, store, vaul embedder.revision = "rev-b" with pytest.raises(IndexMismatch, match="revision rev-a but revision rev-b"): Service(cfg, store=store, embedder=embedder, reranker=reranker).search("espresso", rerank=False) + + +def test_a_reload_after_one_saved_note_tokenizes_only_that_note(cfg, store, vault, embedder, reranker, monkeypatch): + from puremind_mac import retrieval + + index_vault(vault, store, embedder) + corpus = [] + real = retrieval.tokenize + monkeypatch.setattr( + retrieval, "tokenize", lambda docs, query=False: (corpus.append(len(docs)) if not query else None, real(docs, query))[1] + ) + svc = Service(cfg, store=store, embedder=embedder, reranker=reranker) + svc.search("espresso", rerank=False) + assert corpus == [store.stats()["chunks"]] + (vault / "projects" / "garden.md").write_text("# Garden plan\n\nPlant zucchini in June.\n") + index_vault(vault, store, embedder) + hits = svc.search("zucchini", rerank=False) + assert corpus[1:] == [1] # the one new chunk; the others come from the previous snapshot + assert hits[0].path == "projects/garden.md" and "zucchini" in hits[0].content + assert not any("tomatoes" in h.content for h in svc.search("tomatoes", rerank=False)) diff --git a/tests/test_store.py b/tests/test_store.py index 8fda3df..480ee92 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -698,3 +698,23 @@ def test_an_old_index_whose_column_another_process_just_added_still_opens(tmp_pa real = store_module._has_chunker_column monkeypatch.setattr(store_module, "_has_chunker_column", lambda conn: next(checks, None) or real(conn)) Store(path).close() + + +def test_vectors_of_different_lengths_are_refused(store): + store.replace_file("a.md", "h", 0.0, 1, "A", [("", "alpha")], np.ones((1, 4), dtype=np.float32)) + store.replace_file("b.md", "h", 0.0, 1, "B", [("", "beta")], np.ones((1, 8), dtype=np.float32)) + with pytest.raises(ValueError, match="same length"): + store.load_index() + + +def test_a_deferred_purge_is_recorded_until_a_purge_or_clear(store): + store.replace_file("a.md", "h", 0.0, 1, "A", [("", "alpha")], np.ones((1, 4), dtype=np.float32)) + assert not store.purge_pending() + store.defer_purge() + store.defer_purge() + assert store.purge_pending() + store.purge() + assert not store.purge_pending() + store.defer_purge() + store.clear() + assert not store.purge_pending() diff --git a/tests/test_watcher.py b/tests/test_watcher.py index 2368e43..b21ca6b 100644 --- a/tests/test_watcher.py +++ b/tests/test_watcher.py @@ -437,7 +437,7 @@ def test_puremind_watch_prints_what_it_skipped(vault, tmp_path, monkeypatch, cap monkeypatch.setattr("puremind_mac.indexer.MAX_FILE_BYTES", 50) stub = SimpleNamespace(name="stub", embed_documents=lambda texts: np.ones((len(texts), 4), dtype="float32")) monkeypatch.setattr("puremind_mac.service.SentenceTransformerEmbedder", lambda *a, **k: stub) - monkeypatch.setattr("puremind_mac.watcher.watch", lambda vault, run_index, delay, stop=None: run_index()) # one run + monkeypatch.setattr("puremind_mac.watcher.watch", lambda vault, run_index, delay, stop=None, **kw: run_index()) # one run (vault / "evil\x1b]0;TITLE\x07.md").write_text("x" * 100) assert cli.main(["watch", str(vault), "--debounce", "0.1"]) == 0 out = capsys.readouterr() @@ -461,7 +461,7 @@ def test_puremind_watch_stops_when_a_symlinked_vault_names_another_directory(tmp stub = SimpleNamespace(name="stub", embed_documents=lambda texts: np.ones((len(texts), 4), dtype="float32")) monkeypatch.setattr("puremind_mac.service.SentenceTransformerEmbedder", lambda *a, **k: stub) - def two_runs(vault, run_index, delay, stop): + def two_runs(vault, run_index, delay, stop, **kw): run_index() link.unlink() link.symlink_to(second) # the index belongs to the first directory: every later run is refused @@ -491,7 +491,7 @@ def stub(name): monkeypatch.setattr("puremind_mac.service.SentenceTransformerEmbedder", lambda *a, **k: stub("old-model")) - def two_runs(vault, run_index, delay, stop): + def two_runs(vault, run_index, delay, stop, **kw): run_index() store = Store(home / "index.sqlite") try: # `puremind index --rebuild` with the new config, while this watch still has the old one @@ -513,6 +513,94 @@ def two_runs(vault, run_index, delay, stop): store.close() +def _stub_models(monkeypatch, name="stub"): + import numpy as np + + stub = SimpleNamespace(name=name, embed_documents=lambda texts: np.ones((len(texts), 4), dtype="float32")) + monkeypatch.setattr("puremind_mac.service.SentenceTransformerEmbedder", lambda *a, **k: stub) + + +def _compaction_owed(db): + from puremind_mac.store import Store + + store = Store(db) + try: + return store.purge_pending() + finally: + store.close() + + +def test_puremind_watch_compacts_on_its_first_run_when_quiet_and_on_exit(vault, tmp_path, monkeypatch): + from puremind_mac import cli + from puremind_mac.store import Store + + home = tmp_path / "home" + monkeypatch.setenv("PUREMIND_HOME", str(home)) + monkeypatch.delenv("PUREMIND_DB", raising=False) + _stub_models(monkeypatch) + assert cli.main(["index", str(vault)]) == 0 + store = Store(home / "index.sqlite") + store.defer_purge() # what a watch that crashed leaves behind + store.close() + db, note, seen = home / "index.sqlite", vault / "projects" / "garden.md", [] + + def fake_watch(vault_, run_index, delay, stop, quiet, quiet_after): + assert quiet_after == cli.PURGE_WHEN_QUIET + run_index() # the first run pays what was owed + seen.append(_compaction_owed(db)) + note.write_text("# Garden plan\n\nPlant zucchini in June.\n") + run_index() # a save: the compaction is put off + seen.append(_compaction_owed(db)) + quiet() # the vault has been quiet for a while + seen.append(_compaction_owed(db)) + note.write_text("# Garden plan\n\nPlant leeks in March.\n") + run_index() + seen.append(_compaction_owed(db)) # owed again, and paid when watching stops + + monkeypatch.setattr("puremind_mac.watcher.watch", fake_watch) + assert cli.main(["watch", str(vault)]) == 0 + assert seen == [False, True, False, True] + assert not _compaction_owed(db) + + +def test_puremind_watch_compacts_anyway_when_the_vault_never_goes_quiet(vault, tmp_path, monkeypatch): + from puremind_mac import cli + + home = tmp_path / "home" + monkeypatch.setenv("PUREMIND_HOME", str(home)) + monkeypatch.delenv("PUREMIND_DB", raising=False) + _stub_models(monkeypatch) + monkeypatch.setattr(cli, "PURGE_AT_LEAST_EVERY", 0.0) + db, note, seen = home / "index.sqlite", vault / "projects" / "garden.md", [] + + def fake_watch(vault_, run_index, delay, stop, quiet, quiet_after): + run_index() + for crop in ("zucchini", "leeks"): # saves that keep coming, never a quiet spell + note.write_text(f"# Garden plan\n\nPlant {crop}.\n") + run_index() + seen.append(_compaction_owed(db)) + + monkeypatch.setattr("puremind_mac.watcher.watch", fake_watch) + assert cli.main(["watch", str(vault)]) == 0 + assert seen == [True, False] # the second save is past the limit, so its run compacts + + +def test_quiet_work_runs_once_after_each_quiet_spell(tmp_path): + vault = tmp_path / "vault" + vault.mkdir() + runs, quiets = [], [] + stop, thread = _watch_in_thread(vault, lambda: runs.append(1), quiet=lambda: quiets.append(len(runs)), quiet_after=0.3) + try: + assert wait_for(lambda: quiets == [1]) # after the first run + time.sleep(0.8) + assert quiets == [1] # once per quiet spell, not on every tick + (vault / "a.md").write_text("# a\n") + assert wait_for(lambda: quiets == [1, 2]) + finally: + stop.set() + thread.join(PATIENT) + + def test_puremind_watch_names_a_missing_vault(tmp_path, monkeypatch, capsys): from puremind_mac import cli From 87ce50508ef5d1e67c47421c449e5ae5b59cae3c Mon Sep 17 00:00:00 2001 From: PureTensor Date: Sat, 17 Oct 2026 21:28:01 +0100 Subject: [PATCH 2/2] v0.1.5: quiet-hook test tolerates the extra run FSEvents can report at startup Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011zd8G1cibJoQPVbZEpVLYs --- tests/test_watcher.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/tests/test_watcher.py b/tests/test_watcher.py index b21ca6b..6398470 100644 --- a/tests/test_watcher.py +++ b/tests/test_watcher.py @@ -591,11 +591,14 @@ def test_quiet_work_runs_once_after_each_quiet_spell(tmp_path): runs, quiets = [], [] stop, thread = _watch_in_thread(vault, lambda: runs.append(1), quiet=lambda: quiets.append(len(runs)), quiet_after=0.3) try: - assert wait_for(lambda: quiets == [1]) # after the first run + # FSEvents may report the vault's own creation after watching starts, which costs a second run. + assert wait_for(lambda: quiets and quiets[-1] == len(runs)) # after the last run time.sleep(0.8) - assert quiets == [1] # once per quiet spell, not on every tick + assert len(set(quiets)) == len(quiets) # once per quiet spell: never twice without a run between + before = len(quiets) (vault / "a.md").write_text("# a\n") - assert wait_for(lambda: quiets == [1, 2]) + assert wait_for(lambda: len(quiets) > before and quiets[-1] == len(runs)) + assert len(set(quiets)) == len(quiets) and quiets == sorted(quiets) finally: stop.set() thread.join(PATIENT)