feat(test-linker): add optional disk cache for test edges (TAP-4080)

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

@@ -107,8 +107,27 @@ def get_tests_for_symbol(
 
 
 def load_or_build_test_edges(project_root: Path, *, force_rebuild: bool = False) -> list[TestEdge]:
+    """Load cached test edges or build from call graph.
+
+    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.
+    """
+    if not force_rebuild:
+        from tapps_mcp.project.test_linker_cache import load_test_edges_cache
+
+        cached = load_test_edges_cache(project_root)
+        if cached is not None:
+            return cached
+
     index = build_call_graph_index(project_root, force_rebuild=force_rebuild)
-    return build_test_edges(index, project_root=project_root)
+    edges = build_test_edges(index, project_root=project_root)
+
+    # 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 test_edges_to_dicts(edges: list[TestEdge]) -> list[dict[str, object]]:

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

@@ -0,0 +1,87 @@
+"""Disk cache for test edges (TAP-4080)."""
+
+from __future__ import annotations
+
+from dataclasses import asdict
+from pathlib import Path
+
+import structlog
+
+from tapps_core.cache import AtomicJsonCache, 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"
+
+# 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.
+
+    Returns None on cache miss or read failure; a derived cache treats this
+    as a cache miss and rebuilds.
+    """
+    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))
+        _stats["misses"] += 1
+        return None
+    if not isinstance(raw, list):
+        _stats["misses"] += 1
+        return None
+    edges = _test_edges_from_list(raw)
+    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.
+
+    On write failure, logs a warning and leaves the cache untouched.
+    """
+    path = project_root / TEST_EDGES_CACHE_REL
+    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)
+    except OSError as exc:
+        logger.warning("test_edges_cache_write_failed", path=str(path), error=str(exc))
+
+
+def _test_edges_to_list(edges: list[TestEdge]) -> list[dict[str, object]]:
+    """Serialize TestEdge list to JSON-compatible list of dicts."""
+    return [asdict(e) for e in edges]
+
+
+def _test_edges_from_list(raw: list[object]) -> list[TestEdge] | None:
+    """Deserialize JSON list of dicts back to TestEdge list.
+
+    Returns None if deserialization fails (malformed or wrong structure).
+    """
+    try:
+        edges: list[TestEdge] = []
+        for item in raw:
+            if not isinstance(item, dict):
+                return None
+            edge = TestEdge(
+                test_symbol=str(item.get("test_symbol", "")),
+                test_file=str(item.get("test_file", "")),
+                code_symbol=str(item.get("code_symbol", "")),
+                code_file=str(item.get("code_file", "")),
+                line=int(item.get("line", 0)),
+            )
+            edges.append(edge)
+        return edges
+    except (TypeError, ValueError, KeyError):
+        return None

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

@@ -1,11 +1,12 @@
-"""Tests for TESTS edge linker (TAP-4052)."""
+"""Tests for TESTS edge linker (TAP-4052, TAP-4080)."""
 
 from __future__ import annotations
 
 from pathlib import Path
 
 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
+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
 
 
 def _write(root: Path, rel: str, source: str) -> None:
@@ -68,3 +69,96 @@ def test_support():
         ranked = get_tests_for_symbol(edges, "support", index=index)
         assert len(ranked) == 1
         assert ranked[0]["test_symbol"].endswith("test_support")
+
+    def test_test_edges_cache_save_and_load(self, tmp_path: Path) -> None:
+        _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
+""",
+        )
+        # Build edges and save to cache.
+        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)
+
+        # Verify cache file exists.
+        cache_path = tmp_path / ".tapps-mcp" / "test-edges.json"
+        assert cache_path.is_file()
+
+        # Load from cache and verify contents match.
+        cached_edges = load_test_edges_cache(tmp_path)
+        assert cached_edges is not None
+        assert len(cached_edges) == len(edges)
+        assert cached_edges[0].code_symbol == edges[0].code_symbol
+        assert cached_edges[0].test_symbol == edges[0].test_symbol
+
+    def test_load_or_build_uses_cache(self, tmp_path: Path) -> None:
+        _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
+""",
+        )
+        # 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"
+        assert cache_path.is_file()
+
+        # Second call loads from cache (not rebuilt).
+        edges2 = load_or_build_test_edges(tmp_path)
+        assert len(edges2) == 1
+        assert edges2[0].code_symbol == edges1[0].code_symbol
+
+    def test_load_or_build_force_rebuild_bypasses_cache(self, tmp_path: Path) -> None:
+        _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
+""",
+        )
+        # First call builds and caches.
+        edges1 = load_or_build_test_edges(tmp_path)
+        assert len(edges1) == 1
+
+        # force_rebuild=True should rebuild regardless of cache.
+        edges2 = load_or_build_test_edges(tmp_path, force_rebuild=True)
+        assert len(edges2) == 1
+        assert edges2[0].code_symbol == edges1[0].code_symbol