feat(staleness): auto-fire is_stale via a bounded file-existence revalidation pass

mcp_server/handlers/consolidation/memory_staleness_pass.py

@@ -0,0 +1,112 @@
+"""Bounded file-existence staleness re-validation pass (fleet-watch #110).
+
+``is_stale`` is the signal the injection banners now surface (age · grade ·
+stale), but it only ever got *set* on file grounds by the manual
+``validate_memory`` tool — so a memory referencing a file that was moved or
+deleted stayed ``is_stale=FALSE`` until someone ran the tool by hand.
+harness-comparison rev.2 measured exactly this: Harness B served facts months
+stale with no stale flag.
+
+This pass makes the flag fire automatically: it pages non-stale,
+file-referencing memories, re-checks whether their referenced paths still
+exist, and marks ``is_stale`` via the same pure assessment
+(``core.staleness.assess_staleness``) and store method (``mark_memory_stale``)
+the tool uses.
+
+Deliberately **mark-only**: it never de-stales (rehabilitates) a memory. The
+active-forgetting circuit (``consolidation/forgetting.py``, Rac1) also writes
+``is_stale`` for non-file reasons; auto-rehabilitation here could fight it, so
+de-staling stays with the explicit, human-invoked ``validate_memory`` tool.
+Existence only — content-change detection (a file that still exists but diverged
+from what the memory claims) needs per-ref content hashing and is a separate
+#110 seam.
+
+Composition root: pure decision (``assess_staleness``) + an injected filesystem
+resolver + store I/O. Script-invoked (``scripts/memory_staleness_revalidate.py``),
+bounded per run, NOT on the commit critical path or the hot consolidate cycle.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Protocol
+
+from mcp_server.core.staleness import assess_staleness, extract_file_references
+
+logger = logging.getLogger(__name__)
+
+# Per-run scan cap — bounds one run's FS+DB cost. A run that hits the cap
+# resumes from the id cursor on the next invocation.
+# source: mirrors DEFAULT_MEMORY_DOMAIN_BACKFILL_LIMIT = 5000
+#   (memory_domain_backfill_pass.py) — a chosen per-run bound, not a measured
+#   value; adjust with the corpus size.
+DEFAULT_STALENESS_SCAN_LIMIT = 5000
+# Page size for the id-cursor scan.
+# source: get_all_memories_for_validation default page (pg_store_queries.py:94)
+#   is 1000; reused here for parity with the existing validation read path.
+_PAGE = 1000
+
+ResolveExistingFn = Callable[[list[str], str], set[str]]
+
+
+class _StaleStore(Protocol):
+    def get_all_memories_for_validation(
+        self, limit: int, *, after_id: int, include_stale: bool
+    ) -> list[dict[str, Any]]: ...
+
+    def mark_memory_stale(self, memory_id: int, stale: bool = True) -> None: ...
+
+
+def _mark_if_missing(
+    store: _StaleStore,
+    mem: dict,
+    resolve_existing: ResolveExistingFn,
+    threshold: float,
+) -> bool:
+    """Mark one memory stale if its file refs no longer resolve.
+
+    Skips memories with no file refs (their staleness, if any, is not
+    file-derived) and already-stale rows (idempotent). Returns True if it
+    marked the memory stale this call.
+    """
+    content = mem.get("content", "")
+    refs = extract_file_references(content)
+    if not refs or mem.get("is_stale"):
+        return False
+    existing = resolve_existing(refs, mem.get("directory_context", "") or "")
+    report = assess_staleness(
+        mem["id"], content, existing_paths=existing, threshold=threshold
+    )
+    if report.is_stale:
+        store.mark_memory_stale(mem["id"], True)
+        return True
+    return False
+
+
+def revalidate_staleness(
+    store: _StaleStore,
+    resolve_existing: ResolveExistingFn,
+    *,
+    limit: int = DEFAULT_STALENESS_SCAN_LIMIT,
+    threshold: float = 0.5,
+) -> dict[str, int]:
+    """Page non-stale, file-referencing memories and set is_stale on missing refs."""
+    counts = {"scanned": 0, "marked_stale": 0}
+    after_id = 0
+    while counts["scanned"] < limit:
+        page = store.get_all_memories_for_validation(
+            limit=min(_PAGE, limit - counts["scanned"]),
+            after_id=after_id,
+            include_stale=False,
+        )
+        if not page:
+            break
+        for mem in page:
+            after_id = max(after_id, int(mem["id"]))
+            counts["scanned"] += 1
+            if _mark_if_missing(store, mem, resolve_existing, threshold):
+                counts["marked_stale"] += 1
+        if len(page) < _PAGE:
+            break
+    logger.info("staleness revalidation: %s", counts)
+    return counts

scripts/memory_staleness_revalidate.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Re-validate file-existence staleness for memories — fleet-watch #110.
+
+Runs ``handlers.consolidation.memory_staleness_pass`` against the shared store:
+for every non-stale, file-referencing memory whose referenced paths no longer
+resolve on disk, sets ``is_stale=TRUE`` (mark-only; never de-stales — see the
+pass docstring). This makes the staleness the injection banners surface
+(age · grade · stale) actually fire, instead of waiting for a manual
+``validate_memory`` run.
+
+Usage
+-----
+
+Dry-run (default) — report what would be marked, write nothing::
+
+    uv run python scripts/memory_staleness_revalidate.py
+
+Apply the change to the DB::
+
+    uv run python scripts/memory_staleness_revalidate.py --apply
+
+Idempotent: a re-run skips rows already marked stale (``include_stale=False``).
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+if str(_REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(_REPO_ROOT))
+from mcp_server.handlers.consolidation.memory_staleness_pass import (  # noqa: E402
+    DEFAULT_STALENESS_SCAN_LIMIT,
+    revalidate_staleness,
+)
+from mcp_server.handlers.validate_memory import _resolve_existing_paths  # noqa: E402
+from mcp_server.infrastructure.memory_config import get_memory_settings  # noqa: E402
+from mcp_server.infrastructure.memory_store import get_shared_store  # noqa: E402
+
+
+class _DryRunStore:
+    """Wraps the real store; reads pass through, mark writes are suppressed."""
+
+    def __init__(self, inner):
+        self._inner = inner
+
+    def get_all_memories_for_validation(self, limit, *, after_id, include_stale):
+        return self._inner.get_all_memories_for_validation(
+            limit, after_id=after_id, include_stale=include_stale
+        )
+
+    def mark_memory_stale(self, memory_id, stale=True):
+        pass  # dry-run: count via the pass's return value, write nothing
+
+
+def _parse_args(argv):
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--apply", action="store_true", help="Write is_stale to the DB.")
+    p.add_argument("--limit", type=int, default=DEFAULT_STALENESS_SCAN_LIMIT)
+    p.add_argument("--threshold", type=float, default=0.5)
+    return p.parse_args(argv)
+
+
+def main(argv=None) -> int:
+    args = _parse_args(argv)
+    settings = get_memory_settings()
+    store = get_shared_store(settings.DB_PATH, settings.EMBEDDING_DIM)
+    target = store if args.apply else _DryRunStore(store)
+    counts = revalidate_staleness(
+        target, _resolve_existing_paths, limit=args.limit, threshold=args.threshold
+    )
+    mode = "APPLIED" if args.apply else "DRY-RUN (no writes)"
+    label = "marked_stale" if args.apply else "would_mark_stale"
+    print(f"{mode}: scanned={counts['scanned']} {label}={counts['marked_stale']}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests_py/handlers/consolidation/test_memory_staleness_pass.py

@@ -0,0 +1,79 @@
+"""Unit tests for the file-existence staleness re-validation pass (#110).
+
+DB-free: a fake store records mark_memory_stale calls and an injected resolver
+stands in for the filesystem. Verifies the mark-only, skip-no-refs,
+skip-already-stale, and cursor-paging behavior.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp_server.handlers.consolidation.memory_staleness_pass import (
+    revalidate_staleness,
+)
+
+
+class _FakeStore:
+    def __init__(self, memories: list[dict]):
+        self._mems = memories
+        self.marked: list[tuple[int, bool]] = []
+
+    def get_all_memories_for_validation(
+        self, limit: int, *, after_id: int, include_stale: bool
+    ) -> list[dict[str, Any]]:
+        rows = [
+            m
+            for m in self._mems
+            if m["id"] > after_id and (include_stale or not m.get("is_stale"))
+        ]
+        rows.sort(key=lambda m: m["id"])
+        return rows[:limit]
+
+    def mark_memory_stale(self, memory_id: int, stale: bool = True) -> None:
+        self.marked.append((memory_id, stale))
+        for m in self._mems:
+            if m["id"] == memory_id:
+                m["is_stale"] = stale
+
+
+def _resolve(refs: list[str], _base: str) -> set[str]:
+    # Everything resolves except paths that mention "gone".
+    return {r for r in refs if "gone" not in r}
+
+
+def test_marks_only_memory_with_missing_ref() -> None:
+    memories = [
+        {"id": 1, "content": "see src/here.py and src/gone.py", "is_stale": False},
+        {"id": 2, "content": "see src/here.py", "is_stale": False},
+        {"id": 3, "content": "prose with no file references at all", "is_stale": False},
+        {"id": 4, "content": "see src/gone.py", "is_stale": True},  # already stale
+    ]
+    store = _FakeStore(memories)
+
+    counts = revalidate_staleness(store, _resolve, threshold=0.5)
+
+    # Only memory 1 (a missing ref) is newly marked; 2 resolves, 3 has no refs,
+    # 4 is already stale (excluded by include_stale=False).
+    assert store.marked == [(1, True)]
+    assert counts["marked_stale"] == 1
+    assert counts["scanned"] == 3
+
+
+def test_never_destales() -> None:
+    # A memory whose refs all resolve is left untouched — no de-stale write,
+    # so the pass can never fight the active-forgetting circuit.
+    memories = [{"id": 1, "content": "see src/here.py", "is_stale": False}]
+    store = _FakeStore(memories)
+    revalidate_staleness(store, _resolve)
+    assert store.marked == []
+
+
+def test_respects_scan_limit() -> None:
+    memories = [
+        {"id": i, "content": "see src/gone.py", "is_stale": False} for i in range(1, 11)
+    ]
+    store = _FakeStore(memories)
+    counts = revalidate_staleness(store, _resolve, limit=4)
+    assert counts["scanned"] == 4
+    assert counts["marked_stale"] == 4