diff --git a/mcp_server/handlers/consolidation/memory_staleness_pass.py b/mcp_server/handlers/consolidation/memory_staleness_pass.py new file mode 100644 index 00000000..b91fe7cb --- /dev/null +++ b/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 diff --git a/scripts/memory_staleness_revalidate.py b/scripts/memory_staleness_revalidate.py new file mode 100644 index 00000000..4dd0f9dd --- /dev/null +++ b/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()) diff --git a/tests_py/handlers/consolidation/test_memory_staleness_pass.py b/tests_py/handlers/consolidation/test_memory_staleness_pass.py new file mode 100644 index 00000000..6a1b8b3d --- /dev/null +++ b/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