2.3.0 — the index train: local-first recall index, incremental sweep, stemming + alias expansion

One schema-bump release (recall-index schema 3, entity-index schema 2; old
recall indexes rebuild automatically, no vault migration):

- Local-first recall index: the live BM25 index lives in the machine state dir
  (keyed by endpoint hash); capture's upkeep is a zero-network atomic file
  write, no advisory lock — replacing the O(vault) GET/PUT-whole-index round
  trip per write. The vault copy is a snapshot (sweep + session-end) used to
  seed fresh machines / catch up after another client swept
  (ECHO_RECALL_SYNC_HOURS, default 24). The read path never PUTs the vault.
- Incremental sweep: entity + recall meta carry 16-char content hashes;
  `sweep --fast` fetches only new/gone/hash-missing notes plus a rotating
  weekday shard (--all-shards forces everything); deletions drop from both
  indexes; index-only so no --apply gate. `load` auto-runs it past
  ECHO_FAST_SWEEP_DAYS (7); doctor reports index freshness. Obsidian-side
  edits now reach the indexes without manual maintenance.
- Stemming + alias expansion: echo_stem.py (conservative Porter-lite, unit-
  tested families + over-stemming guards) applied at index AND query time;
  a query fuzzy-matching an entity folds its title/alias vocabulary into the
  BM25 query at half weight — expansion can only boost docs containing the
  terms. capture hashes the note's FINAL content (auto-link reordered first).

Eval gold set 8 -> 14 queries (6 paraphrases): recall@5/MRR 1.00/1.00 vs
keyword baseline 0.75/0.79. +3 unit tests, +10 e2e; test harnesses now isolate
ECHO_STATE_DIR. All seven suites green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

echo-memory.plugin.src/skills/echo-memory/scripts/echo_recall.py

@@ -52,9 +52,17 @@ sys.path.insert(0, str(Path(__file__).resolve().parent))
 import echo                      # noqa: E402
 import echo_index as idx_mod     # noqa: E402
 import echo_links as links       # noqa: E402
+import echo_stem                 # noqa: E402
 
 RECALL_INDEX_PATH = "_agent/index/recall-index.json"
-INDEX_SCHEMA = 2                 # bumped: schema 2 adds per-doc meta (weight/updated/status)
+# Schema 3 (2.3, the "index train"): postings are STEMMED (echo_stem), per-doc meta
+# gains a content hash `h` (incremental sweep), and the index carries a `built`
+# stamp + endpoint hash. The LIVE index is LOCAL-FIRST — it lives in the state dir
+# and is updated with zero vault round-trips; the vault copy at RECALL_INDEX_PATH
+# is a snapshot written by sweep/session-end, used to seed fresh machines. Older
+# schemas are discarded and rebuilt (sub-second since the 1.1 network work).
+INDEX_SCHEMA = 3
+SYNC_HOURS = float(os.environ.get("ECHO_RECALL_SYNC_HOURS", "24"))
 
 # Tuning knobs. Defaults are standard BM25 (k1/b) + a 0.6 hop decay over <=2 hops.
 K1 = 1.5
@@ -131,8 +139,11 @@ _SKIP_BASENAMES = {"README.md"}
 
 
 def tokenize(text: str) -> list[str]:
-    """Lowercase word tokens, stopworded. Frontmatter should be stripped by the caller."""
-    return [t for t in _WORD.findall(text.lower()) if t not in _STOP and len(t) > 1]
+    """Lowercase word tokens, stopworded, STEMMED (2.3) — applied identically at
+    index and query time, so 'deployed'/'deployment' meet at 'deploy'.
+    Frontmatter should be stripped by the caller."""
+    return [echo_stem.stem(t) for t in _WORD.findall(text.lower())
+            if t not in _STOP and len(t) > 1]
 
 
 def strip_frontmatter(text: str) -> str:
@@ -152,9 +163,10 @@ class Bm25Index:
         self.df: Counter[str] = Counter()              # term -> #docs containing it
         self.postings: dict[str, dict[str, int]] = {}  # term -> {doc_path: tf}
         self.length: dict[str, int] = {}               # doc_path -> token count
-        self.meta: dict[str, dict] = {}                # doc_path -> {w, u, s} priors
+        self.meta: dict[str, dict] = {}                # doc_path -> {w, u, s, h} priors
         self.n_docs = 0
         self.avg_len = 0.0
+        self.built = ""                                # ISO stamp of last full (re)build
 
     def _recompute(self) -> None:
         self.n_docs = len(self.length)
@@ -165,12 +177,19 @@ class Bm25Index:
             self.remove(path)
         toks = tokenize(strip_frontmatter(raw_text))
         self.length[path] = len(toks)
-        self.meta[path] = doc_meta(path, raw_text)
+        meta = doc_meta(path, raw_text)
+        # Content hash: the incremental sweep's change detector (2.3).
+        import hashlib
+        meta["h"] = hashlib.sha1(raw_text.encode("utf-8", "replace")).hexdigest()[:16]
+        self.meta[path] = meta
         for term, tf in Counter(toks).items():
             self.postings.setdefault(term, {})[path] = tf
             self.df[term] += 1
         self._recompute()
 
+    def content_hash(self, path: str) -> str | None:
+        return (self.meta.get(path) or {}).get("h")
+
     def remove(self, path: str) -> None:
         if path not in self.length:
             return
@@ -192,11 +211,17 @@ class Bm25Index:
                 * freshness(m.get("u"), today_s)
                 * STATUS_FACTOR.get(m.get("s", ""), 1.0))
 
-    def score(self, query: str, limit: int = 10,
-              today_s: str | None = None) -> list[tuple[str, float]]:
+    def score(self, query: str, limit: int = 10, today_s: str | None = None,
+              extra_terms: dict[str, float] | None = None) -> list[tuple[str, float]]:
+        """BM25 x priors. `extra_terms` (term -> weight) are alias-expansion terms
+        (2.3): they contribute at reduced weight and can only boost documents that
+        actually contain them — expansion never invents corpus-free hits."""
         today_s = today_s or echo.today()
+        weighted: dict[str, float] = {t: 1.0 for t in set(tokenize(query))}
+        for t, w in (extra_terms or {}).items():
+            weighted.setdefault(t, w)
         scores: Counter[str] = Counter()
-        for term in set(tokenize(query)):
+        for term, weight in weighted.items():
             posting = self.postings.get(term)
             if not posting:
                 continue
@@ -204,26 +229,29 @@ class Bm25Index:
             for path, tf in posting.items():
                 dl = self.length.get(path, 0)
                 denom = tf + K1 * (1 - B + B * (dl / self.avg_len if self.avg_len else 1))
-                scores[path] += idf * (tf * (K1 + 1) / denom if denom else 0)
+                scores[path] += weight * idf * (tf * (K1 + 1) / denom if denom else 0)
         fused = {p: s * self._doc_factor(p, today_s) for p, s in scores.items()}
         ranked = sorted(fused.items(), key=lambda kv: -kv[1])[:limit]
         return [(p, s) for p, s in ranked if s > MIN_SCORE]
 
     # ---- serialization (compact; rebuildable, so the format can change freely) ----
     def to_json(self) -> dict:
-        return {"schema": INDEX_SCHEMA, "n_docs": self.n_docs, "avg_len": self.avg_len,
-                "length": self.length, "postings": self.postings, "meta": self.meta}
+        return {"schema": INDEX_SCHEMA, "built": self.built, "n_docs": self.n_docs,
+                "avg_len": self.avg_len, "length": self.length,
+                "postings": self.postings, "meta": self.meta}
 
     @classmethod
     def from_json(cls, d: dict) -> "Bm25Index":
         if d.get("schema") != INDEX_SCHEMA:
             return cls()  # older/newer format -> empty; recall's cold path rebuilds
+            # (a pre-stemming index MUST NOT be reused: its postings are unstemmed)
         ix = cls()
         ix.length = d.get("length", {})
         ix.postings = d.get("postings", {})
         ix.meta = d.get("meta", {})
         ix.n_docs = d.get("n_docs", len(ix.length))
         ix.avg_len = d.get("avg_len", 0.0)
+        ix.built = d.get("built", "")
         ix.df = Counter({term: len(posting) for term, posting in ix.postings.items()})
         return ix
 
@@ -271,27 +299,84 @@ def _indexable(path: str) -> bool:
 
 
 # ------------------------------------------------------------------- persistence
-def load_index() -> Bm25Index:
-    status, body = echo.request("GET", echo.vault_url(RECALL_INDEX_PATH))
-    if status == 404:
-        return Bm25Index()
-    echo.check(status, body, "recall-index load")
+# LOCAL-FIRST (2.3). The live index is a machine-local file — updating it on capture
+# is a zero-network atomic file write instead of GET-whole-index -> PUT-whole-index
+# under the vault lock (the old per-write O(vault) round trip). The vault copy is a
+# SNAPSHOT: written by sweep (and best-effort at session end), read only to seed a
+# fresh machine or to catch up after another client swept (checked at most every
+# ECHO_RECALL_SYNC_HOURS). Staleness across clients is tolerable — recall degrades
+# gracefully and the fast sweep trues it up.
+
+def _local_path() -> Path:
+    import hashlib
+    import echo_queue
+    h = hashlib.sha1((echo.BASE or "").encode()).hexdigest()[:12]
+    return echo_queue.state_dir() / f"recall-index-{h}.json"
+
+
+def _load_local() -> Bm25Index | None:
+    p = _local_path()
+    if not p.exists():
+        return None
     try:
+        return Bm25Index.from_json(json.loads(p.read_text(encoding="utf-8")))
+    except (json.JSONDecodeError, KeyError, TypeError, OSError):
+        return None
+
+
+def save_local(ix: Bm25Index) -> None:
+    p = _local_path()
+    p.parent.mkdir(parents=True, exist_ok=True)
+    tmp = p.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(ix.to_json(), ensure_ascii=False), encoding="utf-8")
+    os.replace(tmp, p)  # atomic: a crash mid-write can't corrupt the live index
+
+
+def _fetch_snapshot() -> Bm25Index | None:
+    try:
+        status, body = echo.request("GET", echo.vault_url(RECALL_INDEX_PATH))
+        if status != 200:
+            return None
         return Bm25Index.from_json(json.loads(body))
-    except (json.JSONDecodeError, KeyError, TypeError):
-        return Bm25Index()
+    except Exception:  # noqa: BLE001 — the snapshot is an optimization, never required
+        return None
 
 
-def save_index(ix: Bm25Index) -> None:
+def save_snapshot(ix: Bm25Index) -> None:
+    """PUT the vault snapshot — sweep + session-end only, never the per-capture path."""
     body = json.dumps(ix.to_json(), ensure_ascii=False).encode("utf-8")
     status, b = echo.request("PUT", echo.vault_url(RECALL_INDEX_PATH), data=body,
                              headers={"Content-Type": "application/json"})
-    echo.check(status, b, "recall-index save")
+    echo.check(status, b, "recall-index snapshot save")
 
 
-def rebuild(prefetched: dict | None = None) -> Bm25Index:
-    """Full rebuild from every indexable entity note. Called by sweep.py and as the
-    cold-cache path inside recall(). Saves and returns the index.
+def load_index() -> Bm25Index:
+    """The live index: local file first; the vault snapshot only seeds a fresh
+    machine or replaces a local copy that a newer sweep elsewhere has superseded
+    (checked when the local build stamp is older than SYNC_HOURS)."""
+    local = _load_local()
+    if local and local.n_docs:
+        try:
+            age_ok = local.built and (
+                _dt.datetime.now(_dt.timezone.utc)
+                - _dt.datetime.strptime(local.built, "%Y-%m-%dT%H:%M:%SZ").replace(
+                    tzinfo=_dt.timezone.utc)
+            ).total_seconds() < SYNC_HOURS * 3600
+        except ValueError:
+            age_ok = False
+        if age_ok:
+            return local
+    snap = _fetch_snapshot()
+    if snap and snap.n_docs and (not local or (snap.built or "") > (local.built or "")):
+        save_local(snap)
+        return snap
+    return local or Bm25Index()
+
+
+def rebuild(prefetched: dict | None = None, snapshot: bool = False) -> Bm25Index:
+    """Full rebuild from every indexable note. Saves the LOCAL index always; writes
+    the vault snapshot only when `snapshot=True` (sweep) — the read path never
+    surprises the vault with a PUT.
 
     `prefetched` (a {path: text} map, e.g. from echo.read_many) lets sweep.py reuse the
     content it already pulled instead of walking and re-fetching the whole vault again."""
@@ -299,27 +384,30 @@ def rebuild(prefetched: dict | None = None) -> Bm25Index:
     if prefetched is not None:
         items = ((p, t) for p, t in prefetched.items() if _indexable(p))
     else:
-        items = ((p, _get(p)) for p in _walk() if _indexable(p))
+        items = echo.read_many([p for p in _walk() if _indexable(p)]).items()
     for path, text in items:
         if text is not None:
             ix.add(path, text)   # add() strips frontmatter + extracts doc meta itself
-    save_index(ix)
+    ix.built = echo.now_iso()
+    save_local(ix)
+    if snapshot:
+        save_snapshot(ix)
     return ix
 
 
 def update_note(path: str, text: str) -> None:
-    """Best-effort incremental upkeep for a single note (mirrors how entities.json is
-    maintained on capture). Never raises into the caller. The load->save is wrapped in the
-    advisory lock with a fresh re-read (H3), so concurrent captures can't clobber the
-    recall index either."""
+    """Best-effort incremental upkeep for a single note — a LOCAL atomic file write,
+    no vault round-trip, no advisory lock (2.3). Maintains an existing local index
+    only: with no local index yet, the next recall's cold path (or a sweep) builds
+    one, and a one-note index must never shadow the snapshot. Never raises."""
     try:
         if not _indexable(path):
-            return  # only entity notes are part of the recall corpus
-        import echo_concurrency
-        with echo_concurrency.vault_lock():
-            ix = load_index()                      # fresh read inside the lock
-            ix.add(path, text)                     # raw text: add() strips + extracts meta
-            save_index(ix)
+            return  # only corpus notes are part of the recall index
+        ix = _load_local()
+        if ix is None or not ix.n_docs:
+            return
+        ix.add(path, text)                     # raw text: add() strips + extracts meta
+        save_local(ix)
     except Exception as exc:  # noqa: BLE001 — upkeep must never fail a capture
         print(f"echo_recall: index update skipped ({exc})", file=sys.stderr)
 
@@ -405,13 +493,27 @@ def recall_op(query, limit: int = 8) -> dict:
     index = idx_mod.load()
     nmap = idx_mod.name_map(index)
 
+    # --- alias query expansion (2.3): a query phrased by an entity's alias or a
+    # partial name folds that entity's title/alias vocabulary into the BM25 query at
+    # half weight — "the ECHO plugin" also scores `echo-memory` terms. Expansion can
+    # only boost documents that actually contain the terms.
+    extra_terms: dict[str, float] = {}
+    q_toks = set(tokenize(q))
+    for _slug, e, sc in idx_mod.fuzzy_candidates(index, q)[:2]:
+        if sc < 0.5:
+            continue
+        for name in (e.get("title", ""), _slug, *e.get("aliases", [])):
+            for tok in tokenize(str(name).replace("-", " ")):
+                if tok not in q_toks:
+                    extra_terms[tok] = 0.5
+
     # --- lexical layer (BM25); rebuild the cache on a cold miss --------------
     lexical: list[tuple[str, float]] = []
     try:
         ix = load_index()
         if ix.n_docs == 0:
             ix = rebuild()
-        lexical = ix.score(q, limit=limit, today_s=today_s)
+        lexical = ix.score(q, limit=limit, today_s=today_s, extra_terms=extra_terms)
     except echo.EchoError as exc:
         print(f"recall: BM25 index unavailable ({exc}); falling back to server search",
               file=sys.stderr)