fix(test-linker): fingerprint-keyed test-edge cache + staleness + reroute (TAP-4080)

packages/tapps-mcp/src/tapps_mcp/project/diff_impact.py

@@ -9,7 +9,7 @@
 from tapps_mcp.project.call_graph import build_call_graph_index
 from tapps_mcp.project.call_graph_queries import resolve_symbol_name
 from tapps_mcp.project.impact_analyzer import analyze_impact, build_import_graph
-from tapps_mcp.project.test_linker import build_test_edges, edges_for_symbols
+from tapps_mcp.project.test_linker import edges_for_symbols, load_or_build_test_edges_for_index
 
 DEFAULT_AFFECTED_TESTS_LIMIT = 20
 DEFAULT_DOC_DRIFT_CALLER_THRESHOLD = 5
@@ -98,7 +98,7 @@ def analyze_diff_impact(
 ) -> dict[str, object]:
     """Rank tests affected by *changed_files* using TESTS edges and import impact."""
     index = build_call_graph_index(project_root)
-    test_edges = build_test_edges(index, project_root=project_root)
+    test_edges = load_or_build_test_edges_for_index(project_root, index)
     graph = build_import_graph(project_root)
 
     ranked: dict[str, RankedTest] = {}
@@ -250,9 +250,9 @@ def build_diff_impact_enrichment(
             "changed_files": changed_paths,
         }
 
-    from tapps_mcp.project.test_linker import build_test_edges, get_tests_for_symbol
+    from tapps_mcp.project.test_linker import get_tests_for_symbol
 
-    test_edges = build_test_edges(index, project_root=project_root)
+    test_edges = load_or_build_test_edges_for_index(project_root, index)
 
     symbols_out: dict[str, dict[str, object]] = {}
     for changed in changed_files:
@@ -393,7 +393,9 @@ def export_test_map(
 ) -> Path:
     """Write TDAD-style static test_map.txt from TESTS edges (TAP-4095)."""
     index = build_call_graph_index(project_root, force_rebuild=force_rebuild)
-    test_edges = build_test_edges(index, project_root=project_root)
+    test_edges = load_or_build_test_edges_for_index(
+        project_root, index, force_rebuild=force_rebuild
+    )
     target = output_path or (project_root / "test_map.txt")
     lines = [
         "# TappsMCP test_map — code symbol -> test file (TDAD static artifact)",

packages/tapps-mcp/src/tapps_mcp/project/test_linker.py

@@ -61,8 +61,7 @@ def build_test_edges(
             )
 
     edges.sort(key=lambda e: (e.code_symbol, e.test_symbol))
-    if project_root is not None:
-        _ = project_root  # reserved for future cache fingerprinting
+    _ = project_root  # accepted for signature symmetry with the cached path
     return edges
 
 
@@ -106,29 +105,58 @@ def get_tests_for_symbol(
     return sorted(ranked.values(), key=lambda item: str(item.get("test_file", "")))
 
 
-def load_or_build_test_edges(project_root: Path, *, force_rebuild: bool = False) -> list[TestEdge]:
-    """Load cached test edges or build from call graph.
+def load_or_build_test_edges_for_index(
+    project_root: Path,
+    index: CallGraphIndex,
+    *,
+    force_rebuild: bool = False,
+) -> list[TestEdge]:
+    """Return TESTS edges for an already-built *index*, using the disk cache.
 
-    Unless *force_rebuild* is True, attempts to load from disk cache first
-    (TAP-4080: optional disk cache via AtomicJsonCache). Cache miss triggers
-    a full rebuild.
+    The cache is keyed by ``index.fingerprint`` (TAP-4080): a hit is served only
+    when the cached fingerprint matches the current call graph; on a fingerprint
+    mismatch or ``force_rebuild`` the edges are rebuilt from *index* and the cache
+    is rewritten. This is the fast path for ``diff_impact`` call sites that
+    already hold the index, so it never re-parses the graph.
     """
-    if not force_rebuild:
-        from tapps_mcp.project.test_linker_cache import load_test_edges_cache
-
-        cached = load_test_edges_cache(project_root)
+    from tapps_mcp.project.test_linker_cache import (
+        load_test_edges_cache,
+        save_test_edges_cache,
+    )
+
+    fingerprint = index.fingerprint
+    if not force_rebuild and fingerprint:
+        cached = load_test_edges_cache(project_root, fingerprint)
         if cached is not None:
             return cached
 
-    index = build_call_graph_index(project_root, force_rebuild=force_rebuild)
     edges = build_test_edges(index, project_root=project_root)
+    if fingerprint:
+        save_test_edges_cache(project_root, fingerprint, edges)
+    return edges
 
-    # Write cache for next call (TAP-4080).
-    from tapps_mcp.project.test_linker_cache import save_test_edges_cache
 
-    save_test_edges_cache(project_root, edges)
-    return edges
+def load_or_build_test_edges(project_root: Path, *, force_rebuild: bool = False) -> list[TestEdge]:
+    """Load cached test edges or build from the call graph.
+
+    Builds (or loads) the call-graph index, then delegates to
+    :func:`load_or_build_test_edges_for_index`, which serves the fingerprint-keyed
+    disk cache when fresh (TAP-4080) and rebuilds on a fingerprint mismatch.
+    """
+    index = build_call_graph_index(project_root, force_rebuild=force_rebuild)
+    return load_or_build_test_edges_for_index(
+        project_root, index, force_rebuild=force_rebuild
+    )
 
 
 def test_edges_to_dicts(edges: list[TestEdge]) -> list[dict[str, object]]:
     return [asdict(e) for e in edges]
+
+
+# Eager side-effect import so ``register_cache_stats("test_edges", ...)`` fires
+# on any import of this module (TAP-4080 criterion): ``test_linker`` is imported
+# on the ``diff_impact`` and call-graph paths that a normal ``tapps_stats`` run
+# exercises, so "test_edges" reliably appears in ``tapps_stats.caches`` — not
+# only after a lazy ``load_or_build`` call. Placed at module end to avoid a
+# definition-time cycle (``test_linker_cache`` imports ``TestEdge`` from here).
+from tapps_mcp.project import test_linker_cache as _test_linker_cache  # noqa: F401

packages/tapps-mcp/src/tapps_mcp/project/test_linker_cache.py

@@ -1,4 +1,15 @@
-"""Disk cache for test edges (TAP-4080)."""
+"""Fingerprint-keyed disk cache for test edges (TAP-4080, ADR-0029).
+
+TESTS edges are a deterministic derivative of the call-graph index, so this
+cache follows the same content-fingerprint model as the call-graph cache
+(``call_graph_cache.py``): the cached payload embeds the call-graph fingerprint
+it was built from, and :class:`FingerprintStaleness` rejects it the moment the
+current fingerprint differs. No TTL, no clock — fresh exactly while the call
+graph is unchanged.
+
+Built on the ADR-0029 substrate (``AtomicJsonCache`` + ``FingerprintStaleness``);
+no hand-rolled atomic-write or staleness code.
+"""
 
 from __future__ import annotations
 
@@ -7,54 +18,75 @@
 
 import structlog
 
-from tapps_core.cache import AtomicJsonCache, register_cache_stats
+from tapps_core.cache import AtomicJsonCache, FingerprintStaleness, register_cache_stats
 from tapps_mcp.project.test_linker import TestEdge
 
 logger = structlog.get_logger(__name__)
 
-TEST_EDGES_CACHE_REL = ".tapps-mcp/test-edges.json"
+# Fingerprint-keyed cache path (TAP-4080). Distinct from the pre-TAP-4080
+# fixed-path file — this one is validated against the call-graph fingerprint.
+TEST_EDGES_CACHE_REL = ".tapps-mcp/test-edges-index.json"
 
 # ADR-0029 / TAP-4561: unified cache-stats counters (test edge load hits/misses).
 _stats: dict[str, int] = {"hits": 0, "misses": 0}
 register_cache_stats("test_edges", lambda: dict(_stats))
 
 
-def load_test_edges_cache(project_root: Path) -> list[TestEdge] | None:
-    """Load cached test edges from disk, or None if cache is missing/unreadable.
+def _reset_test_edges_stats() -> None:
+    """Reset hit/miss counters (test isolation — conftest ``_reset_caches``)."""
+    _stats["hits"] = 0
+    _stats["misses"] = 0
+
+
+def load_test_edges_cache(project_root: Path, fingerprint: str) -> list[TestEdge] | None:
+    """Load cached test edges when fresh for *fingerprint*, else ``None``.
 
-    Returns None on cache miss or read failure; a derived cache treats this
-    as a cache miss and rebuilds.
+    Returns ``None`` on a missing file, an unreadable/malformed payload, or a
+    fingerprint mismatch (the cache was built from a stale call graph). The
+    caller treats every ``None`` as a miss and rebuilds — a stale cache is never
+    served.
     """
     path = project_root / TEST_EDGES_CACHE_REL
     if not path.is_file():
         _stats["misses"] += 1
         return None
     raw = AtomicJsonCache.read_json(path)
-    if raw is None:
-        logger.warning("test_edges_cache_read_failed", path=str(path))
+    if not isinstance(raw, dict):
+        if raw is not None:
+            logger.warning("test_edges_cache_read_failed", path=str(path))
         _stats["misses"] += 1
         return None
-    if not isinstance(raw, list):
+
+    cached_fp = raw.get("fingerprint")
+    if not isinstance(cached_fp, str) or FingerprintStaleness(fingerprint).is_stale(cached_fp):
+        # Stale: cached edges came from a different call-graph fingerprint.
+        _stats["misses"] += 1
+        return None
+
+    raw_edges = raw.get("edges")
+    if not isinstance(raw_edges, list):
         _stats["misses"] += 1
         return None
-    edges = _test_edges_from_list(raw)
+    edges = _test_edges_from_list(raw_edges)
     if edges is None:
         _stats["misses"] += 1
         return None
     _stats["hits"] += 1
     return edges
 
 
-def save_test_edges_cache(project_root: Path, edges: list[TestEdge]) -> None:
-    """Save test edges to disk atomically.
+def save_test_edges_cache(
+    project_root: Path, fingerprint: str, edges: list[TestEdge]
+) -> None:
+    """Persist *edges* keyed by *fingerprint*, atomically.
 
     On write failure, logs a warning and leaves the cache untouched.
     """
     path = project_root / TEST_EDGES_CACHE_REL
+    payload = {"fingerprint": fingerprint, "edges": _test_edges_to_list(edges)}
     try:
         path.parent.mkdir(parents=True, exist_ok=True)
-        # sort_keys/indent preserve the exact prior byte layout (ADR-0029 pilot).
-        AtomicJsonCache.write_json(path, _test_edges_to_list(edges), indent=2, sort_keys=True)
+        AtomicJsonCache.write_json(path, payload, indent=2, sort_keys=True)
     except OSError as exc:
         logger.warning("test_edges_cache_write_failed", path=str(path), error=str(exc))
 

packages/tapps-mcp/tests/conftest.py

@@ -520,6 +520,10 @@ def _clear_test_singleton_caches() -> None:
 
     _clear_content_cache()
 
+    from tapps_mcp.project.test_linker_cache import _reset_test_edges_stats
+
+    _reset_test_edges_stats()
+
 
 @pytest.fixture(autouse=True)
 def _reset_caches() -> Generator[None, None, None]:

packages/tapps-mcp/tests/unit/test_test_linker.py

@@ -4,9 +4,19 @@
 
 from pathlib import Path
 
+from tapps_core.cache import collect_cache_stats
 from tapps_mcp.project.call_graph import build_call_graph_index
-from tapps_mcp.project.test_linker import build_test_edges, edges_for_symbols, get_tests_for_symbol, load_or_build_test_edges
-from tapps_mcp.project.test_linker_cache import load_test_edges_cache, save_test_edges_cache
+from tapps_mcp.project.test_linker import (
+    build_test_edges,
+    edges_for_symbols,
+    get_tests_for_symbol,
+    load_or_build_test_edges,
+)
+from tapps_mcp.project.test_linker_cache import (
+    TEST_EDGES_CACHE_REL,
+    load_test_edges_cache,
+    save_test_edges_cache,
+)
 
 
 def _write(root: Path, rel: str, source: str) -> None:
@@ -89,17 +99,17 @@ def test_support():
     assert support() == 1
 """,
         )
-        # Build edges and save to cache.
+        # Build edges and save to cache keyed by the call-graph fingerprint.
         index = build_call_graph_index(tmp_path, force_rebuild=True)
         edges = build_test_edges(index, project_root=tmp_path)
-        save_test_edges_cache(tmp_path, edges)
+        save_test_edges_cache(tmp_path, index.fingerprint, edges)
 
-        # Verify cache file exists.
-        cache_path = tmp_path / ".tapps-mcp" / "test-edges.json"
+        # Verify cache file exists at the fingerprint-keyed path.
+        cache_path = tmp_path / TEST_EDGES_CACHE_REL
         assert cache_path.is_file()
 
-        # Load from cache and verify contents match.
-        cached_edges = load_test_edges_cache(tmp_path)
+        # Load from cache (same fingerprint) and verify contents match.
+        cached_edges = load_test_edges_cache(tmp_path, index.fingerprint)
         assert cached_edges is not None
         assert len(cached_edges) == len(edges)
         assert cached_edges[0].code_symbol == edges[0].code_symbol
@@ -127,7 +137,7 @@ def test_support():
         # First call builds and caches.
         edges1 = load_or_build_test_edges(tmp_path)
         assert len(edges1) == 1
-        cache_path = tmp_path / ".tapps-mcp" / "test-edges.json"
+        cache_path = tmp_path / TEST_EDGES_CACHE_REL
         assert cache_path.is_file()
 
         # Second call loads from cache (not rebuilt).
@@ -162,3 +172,84 @@ def test_support():
         edges2 = load_or_build_test_edges(tmp_path, force_rebuild=True)
         assert len(edges2) == 1
         assert edges2[0].code_symbol == edges1[0].code_symbol
+
+    def test_cache_invalidates_on_fingerprint_change(self, tmp_path: Path) -> None:
+        """A cache built from an old call graph is a MISS once the fingerprint changes.
+
+        Same fingerprint => hit; a different (changed call graph) fingerprint =>
+        no stale edges served, so the caller rebuilds fresh (TAP-4080).
+        """
+        _write(
+            tmp_path,
+            "demo/helper.py",
+            """
+def support():
+    return 1
+""",
+        )
+        _write(
+            tmp_path,
+            "tests/test_helper.py",
+            """
+from demo.helper import support
+
+def test_support():
+    assert support() == 1
+""",
+        )
+        index = build_call_graph_index(tmp_path, force_rebuild=True)
+        edges = build_test_edges(index, project_root=tmp_path)
+        assert index.fingerprint  # keyed by a real fingerprint
+        save_test_edges_cache(tmp_path, index.fingerprint, edges)
+
+        # Same fingerprint -> cache HIT.
+        assert load_test_edges_cache(tmp_path, index.fingerprint) is not None
+
+        # Changed call graph -> different fingerprint -> MISS (no stale serve).
+        stale = load_test_edges_cache(tmp_path, index.fingerprint + "-changed")
+        assert stale is None
+
+        # A full rebuild path with a genuinely changed graph returns fresh edges
+        # and rewrites the cache under the new fingerprint.
+        _write(
+            tmp_path,
+            "demo/helper.py",
+            """
+def support():
+    return 1
+
+def support_two():
+    return 2
+""",
+        )
+        _write(
+            tmp_path,
+            "tests/test_helper.py",
+            """
+from demo.helper import support, support_two
+
+def test_support():
+    assert support() == 1
+
+def test_support_two():
+    assert support_two() == 2
+""",
+        )
+        new_index = build_call_graph_index(tmp_path, force_rebuild=True)
+        assert new_index.fingerprint != index.fingerprint
+        fresh = load_or_build_test_edges(tmp_path)
+        assert {e.code_symbol for e in fresh} == {
+            "demo.helper.support",
+            "demo.helper.support_two",
+        }
+        # Cache now serves the fresh edges under the NEW fingerprint.
+        assert load_test_edges_cache(tmp_path, new_index.fingerprint) is not None
+
+    def test_stats_registered_in_collect_cache_stats(self) -> None:
+        """`test_edges` appears in the unified cache-stats surface (tapps_stats.caches)."""
+        # Importing test_linker (done at module top) eagerly registers the provider.
+        import tapps_mcp.project.test_linker  # noqa: F401
+
+        stats = collect_cache_stats()
+        assert "test_edges" in stats
+        assert set(stats["test_edges"]) >= {"hits", "misses"}