diff --git a/src/aak/mcp_server.py b/src/aak/mcp_server.py index f2eca63..51b6dbb 100644 --- a/src/aak/mcp_server.py +++ b/src/aak/mcp_server.py @@ -16,6 +16,9 @@ """ from __future__ import annotations import json +import re +import subprocess +from datetime import datetime, timezone from pathlib import Path from mcp.server.fastmcp import FastMCP @@ -26,6 +29,8 @@ _manifest: ProjectManifest | None = None _repo_root: Path | None = None +SUPPORTED_INDEX_SCHEMA_VERSIONS = {"1.0"} +SHA_RE = re.compile(r"^[0-9a-f]{40}$") def _get_manifest() -> ProjectManifest: @@ -41,6 +46,136 @@ def _read_map() -> dict: return json.load(f) +def _git_value(*args: str) -> str | None: + if _repo_root is None: + return None + try: + result = subprocess.run( + ["git", "-C", str(_repo_root), *args], + capture_output=True, text=True, timeout=5, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + value = result.stdout.strip() + return value if result.returncode == 0 and value else None + + +def _git_success(*args: str) -> bool: + if _repo_root is None: + return False + try: + result = subprocess.run( + ["git", "-C", str(_repo_root), *args], + capture_output=True, text=True, timeout=5, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return False + return result.returncode == 0 + + +def _commit_problem(commit: object, observed: str | None, label: str) -> str | None: + """Validate an exact commit SHA and constrain it to the served checkout.""" + if not isinstance(commit, str) or not SHA_RE.fullmatch(commit): + return f"{label} commit is absent or invalid; a full 40-character SHA is required" + if not _git_success("cat-file", "-e", f"{commit}^{{commit}}"): + return f"{label} commit does not exist in the served repository" + if observed is None or not SHA_RE.fullmatch(observed): + return "observed repository commit could not be verified" + if not _git_success("merge-base", "--is-ancestor", commit, observed): + return f"{label} commit is outside the served branch history" + return None + + +def _index_status(expected_commit: str | None = None) -> dict: + """Compute a fail-closed freshness contract without exposing repo content.""" + manifest = _get_manifest() + observed = _git_value("rev-parse", "HEAD") + expected = expected_commit + branch = _git_value("branch", "--show-current") or "detached" + base = { + "repository": manifest.repo_id, + "branch": branch, + "expectedCommit": expected, + "observedCommit": observed, + "indexedCommit": None, + "sourceSha": None, + "indexedAt": None, + "schemaVersion": None, + "indexedFileCount": None, + "componentCount": None, + "includedScope": list(manifest.source_roots), + "excludedScope": [".git", "node_modules", ".next", "coverage", "dist", "build", "__pycache__", ".venv", "venv"], + "freshness": "unknown", + "reason": "index metadata could not be verified", + "checkedAt": datetime.now(timezone.utc).isoformat(), + } + marker = manifest.map_path.with_suffix(manifest.map_path.suffix + ".indexing") + if marker.exists(): + try: + marker_data = json.loads(marker.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {**base, "freshness": "failed", "reason": "indexing marker is invalid"} + if not isinstance(marker_data, dict) or marker_data.get("status") not in { + "requested", "indexing", "validating", "ready", "publishing", + }: + return {**base, "freshness": "failed", "reason": "indexing marker is invalid"} + return {**base, "freshness": "indexing", "reason": "an isolated index build is in progress"} + try: + data = _read_map() + except (OSError, json.JSONDecodeError) as exc: + return {**base, "freshness": "failed", "reason": f"index backend unavailable or invalid: {type(exc).__name__}"} + source_sha = data.get("sourceSha") + schema = data.get("schemaVersion") + components = data.get("components") + status = { + **base, + "indexedCommit": source_sha, + "sourceSha": source_sha, + "indexedAt": data.get("generatedAt"), + "schemaVersion": schema, + "indexedFileCount": len(components) if isinstance(components, list) else None, + "componentCount": len(components) if isinstance(components, list) else None, + "includedScope": data.get("sourceRoots", base["includedScope"]), + "coverage": data.get("coverage"), + "extractionMethod": data.get("extractionMethod"), + } + if schema not in SUPPORTED_INDEX_SCHEMA_VERSIONS: + return {**status, "freshness": "incompatible", "reason": f"unsupported index schema version: {schema!r}"} + if data.get("repoId") != manifest.repo_id: + return {**status, "freshness": "incompatible", "reason": "index repository does not match the configured project"} + if not isinstance(components, list): + return {**status, "freshness": "incompatible", "reason": "index components field is not an array"} + source_problem = _commit_problem(source_sha, observed, "indexed") + if source_problem: + return {**status, "freshness": "unknown", "reason": source_problem} + expected_problem = _commit_problem(expected, observed, "expected") + if expected_problem: + return {**status, "freshness": "unknown", "reason": expected_problem} + if source_sha != expected: + return {**status, "freshness": "stale", "reason": "indexed commit does not match the expected commit"} + return {**status, "freshness": "fresh", "reason": "indexed commit matches the expected commit"} + + +def _freshness_error(expected_commit: str | None = None) -> dict | None: + status = _index_status(expected_commit) + if status["freshness"] == "fresh": + return None + failed = status["freshness"] == "failed" + return { + "error": { + "code": "INDEX_BACKEND_UNAVAILABLE" if failed else "INDEX_NOT_FRESH", + "message": ( + "architectural conclusion blocked because the index backend is unavailable" + if failed else + "architectural conclusion blocked because the index is not fresh" + ), + "retryable": status["freshness"] in {"indexing", "failed", "stale"}, + }, + "index": status, + "historicalOnly": status["freshness"] == "stale", + } + + def _read_incidents() -> list: manifest = _get_manifest() if not manifest.incidents_path.is_file(): @@ -90,7 +225,7 @@ def _get_component(path: str) -> dict: @mcp.tool() -def search_capability(repoId: str, query: str, limit: int = 20) -> list[dict]: +def search_capability(repoId: str, query: str, limit: int = 20, expectedCommit: str | None = None) -> list[dict]: """Search architecture-map components by substring in path or exports. Use BEFORE writing any new helper, route, or component to confirm @@ -107,11 +242,14 @@ def search_capability(repoId: str, query: str, limit: int = 20) -> list[dict]: err = _check_repo_id(repoId) if err: return [err] + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return [freshness_err] return _search_components(query, limit) @mcp.tool() -def get_component(repoId: str, componentId: str) -> dict: +def get_component(repoId: str, componentId: str, expectedCommit: str | None = None) -> dict: """Return the full architecture-map entry for a component by exact path. Args: @@ -124,11 +262,14 @@ def get_component(repoId: str, componentId: str) -> dict: err = _check_repo_id(repoId) if err: return err + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return freshness_err return _get_component(componentId) @mcp.tool() -def check_duplication(description: str) -> dict: +def check_duplication(description: str, expectedCommit: str | None = None) -> dict: """Check whether a described capability likely already exists. Cross-references known duplication incidents (business terms / accepted @@ -144,6 +285,9 @@ def check_duplication(description: str) -> dict: matchedIncident, when present, points at the canonical target (path + symbol) that should be reused instead of rebuilt. """ + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return freshness_err incidents = _read_incidents() desc_lower = description.lower() @@ -169,12 +313,15 @@ def check_duplication(description: str) -> dict: @mcp.tool() -def list_adrs() -> list[dict]: +def list_adrs(expectedCommit: str | None = None) -> list[dict]: """List the repo's Architecture Decision Records. Returns: List of {slug, title, path}. """ + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return [freshness_err] adr_dir = _adr_dir() if not adr_dir.is_dir(): return [] @@ -196,7 +343,7 @@ def list_adrs() -> list[dict]: @mcp.tool() -def get_adr(slug: str) -> str: +def get_adr(slug: str, expectedCommit: str | None = None) -> str | dict: """Return the full markdown content of an ADR by slug. Args: @@ -205,25 +352,17 @@ def get_adr(slug: str) -> str: Returns: Markdown content, or an "error: ..." string if not found. """ + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return freshness_err path = _adr_dir() / f"{slug}.md" if not path.is_file(): return f"error: ADR not found: {slug}" return path.read_text(encoding="utf-8") -def _map_status() -> dict: - data = _read_map() - return { - "generatedAt": data.get("generatedAt"), - "sourceSha": data.get("sourceSha"), - "coverage": data.get("coverage"), - "componentCount": len(data.get("components", [])), - "extractionMethod": data.get("extractionMethod"), - } - - @mcp.tool() -def get_map_status(repoId: str) -> dict: +def get_map_status(repoId: str, expectedCommit: str | None = None) -> dict: """Return architecture-map metadata: when it was generated, from which commit, coverage, and component count. Use to judge whether the map might be stale before trusting an answer from the other tools. @@ -238,11 +377,11 @@ def get_map_status(repoId: str) -> dict: err = _check_repo_id(repoId) if err: return err - return _map_status() + return _index_status(expectedCommit) @mcp.tool() -def verify_references(repoId: str, paths: list[str], symbols: list[str]) -> dict: +def verify_references(repoId: str, paths: list[str], symbols: list[str], expectedCommit: str | None = None) -> dict: """Cross-check cited file paths and exported symbols against the current architecture map. Use to validate evidence before citing it in a Codex audit or Gemini brief -- "not found" without recorded search coverage is @@ -260,6 +399,9 @@ def verify_references(repoId: str, paths: list[str], symbols: list[str]) -> dict err = _check_repo_id(repoId) if err: return err + freshness_err = _freshness_error(expectedCommit) + if freshness_err: + return freshness_err data = _read_map() components = data.get("components", []) diff --git a/tests/test_duplication_golden_corpus.py b/tests/test_duplication_golden_corpus.py index 4878833..8a148ff 100644 --- a/tests/test_duplication_golden_corpus.py +++ b/tests/test_duplication_golden_corpus.py @@ -78,14 +78,23 @@ def golden_repo(tmp_path, monkeypatch): monkeypatch.setattr(mcp_server_mod, "_manifest", None) monkeypatch.setattr(mcp_server_mod, "_repo_root", tmp_path) - yield tmp_path + expected_commit = "a" * 40 + + def require_expected_commit(value): + assert value == expected_commit + return None + + monkeypatch.setattr(mcp_server_mod, "_freshness_error", require_expected_commit) + yield tmp_path, expected_commit monkeypatch.setattr(mcp_server_mod, "_manifest", None) monkeypatch.setattr(mcp_server_mod, "_repo_root", None) def test_replays_case_chat_incident(golden_repo): + _, expected_commit = golden_repo result = mcp_server_mod.check_duplication( - "preciso implementar chat verticalizado por processo pra nova tela de audiencias" + "preciso implementar chat verticalizado por processo pra nova tela de audiencias", + expectedCommit=expected_commit, ) assert result["matchedIncident"] is not None assert result["matchedIncident"]["canonicalTarget"] == { @@ -94,8 +103,10 @@ def test_replays_case_chat_incident(golden_repo): def test_replays_notification_pipeline_incident(golden_repo): + _, expected_commit = golden_repo result = mcp_server_mod.check_duplication( - "vou mexer no pipeline de notificacoes pra adicionar um canal novo de SMS" + "vou mexer no pipeline de notificacoes pra adicionar um canal novo de SMS", + expectedCommit=expected_commit, ) assert result["matchedIncident"] is not None assert result["matchedIncident"]["canonicalTarget"] == { @@ -104,8 +115,10 @@ def test_replays_notification_pipeline_incident(golden_repo): def test_unrelated_request_matches_no_incident(golden_repo): + _, expected_commit = golden_repo result = mcp_server_mod.check_duplication( - "quero adicionar um botao de exportar CSV na tela de relatorios" + "quero adicionar um botao de exportar CSV na tela de relatorios", + expectedCommit=expected_commit, ) assert result["matchedIncident"] is None @@ -114,7 +127,11 @@ def test_case_chat_candidate_components_surface_both_canonical_symbols(golden_re """The fix for the case-chat incident introduced two canonical symbols (getOrCreateCaseChat for provisioning, openProcessChat for navigation) -- a search for "chat" should surface both, not just one.""" - result = mcp_server_mod.check_duplication("preciso implementar chat verticalizado por processo") + _, expected_commit = golden_repo + result = mcp_server_mod.check_duplication( + "preciso implementar chat verticalizado por processo", + expectedCommit=expected_commit, + ) candidate_paths = {c["path"] for c in result["candidateComponents"]} assert "src/domain/cases/case-chat.ts" in candidate_paths assert "src/lib/process-chat.ts" in candidate_paths diff --git a/tests/test_mcp_freshness.py b/tests/test_mcp_freshness.py new file mode 100644 index 0000000..bdaabc9 --- /dev/null +++ b/tests/test_mcp_freshness.py @@ -0,0 +1,174 @@ +import json +import subprocess +from pathlib import Path + +import pytest + +from aak import mcp_server +from aak.manifest import ProjectManifest + + +def _manifest(repo: Path) -> ProjectManifest: + return ProjectManifest( + schema_version="1.0", repo_id="lexnotify/web", tier=1, + default_branch="main", kit_version="1.0", + source_roots=["app", "src"], + architecture={"map": "docs/architecture/ARCHITECTURE_MAP.json"}, + repo_root=repo, + ) + + +@pytest.fixture +def repo(tmp_path, monkeypatch): + subprocess.run(["git", "init", "-b", "main", str(tmp_path)], check=True, capture_output=True) + subprocess.run(["git", "-C", str(tmp_path), "config", "user.email", "test@example.invalid"], check=True) + subprocess.run(["git", "-C", str(tmp_path), "config", "user.name", "Test"], check=True) + (tmp_path / "tracked.txt").write_text("fixture", encoding="utf-8") + subprocess.run(["git", "-C", str(tmp_path), "add", "tracked.txt"], check=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "-m", "fixture"], check=True, capture_output=True) + map_path = tmp_path / "docs/architecture/ARCHITECTURE_MAP.json" + map_path.parent.mkdir(parents=True) + monkeypatch.setattr(mcp_server, "_repo_root", tmp_path) + monkeypatch.setattr(mcp_server, "_manifest", _manifest(tmp_path)) + return tmp_path, map_path + + +def _head(repo: Path) -> str: + return subprocess.run( + ["git", "-C", str(repo), "rev-parse", "HEAD"], check=True, + capture_output=True, text=True, + ).stdout.strip() + + +def _commit(repo: Path, name: str) -> str: + (repo / name).write_text(name, encoding="utf-8") + subprocess.run(["git", "-C", str(repo), "add", name], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-m", name], check=True, capture_output=True) + return _head(repo) + + +def _write_map(path: Path, source_sha: str | None, **overrides): + doc = { + "schemaVersion": "1.0", "repoId": "lexnotify/web", + "generatedAt": "2026-07-28T00:00:00+00:00", "sourceSha": source_sha, + "extractionMethod": "fixture", "coverage": "fixture", + "components": [{"path": "src/a.ts", "exports": ["a"], "layer": "lib"}], + } + doc.update(overrides) + path.write_text(json.dumps(doc), encoding="utf-8") + + +def test_fresh_index_allows_existing_tool_contracts(repo): + root, map_path = repo + sha = _head(root) + _write_map(map_path, sha) + status = mcp_server.get_map_status("lexnotify/web", sha) + assert status["freshness"] == "fresh" + assert status["sourceSha"] == sha + assert status["expectedCommit"] == sha + assert status["componentCount"] == 1 + assert mcp_server.search_capability("lexnotify/web", "a", expectedCommit=sha)[0]["path"] == "src/a.ts" + assert mcp_server.get_component("lexnotify/web", "src/a.ts", sha)["path"] == "src/a.ts" + + +def test_stale_index_blocks_conclusive_tools(repo): + root, map_path = repo + indexed = _head(root) + expected = _commit(root, "new.txt") + _write_map(map_path, indexed) + result = mcp_server.search_capability("lexnotify/web", "a", expectedCommit=expected) + assert result[0]["error"]["code"] == "INDEX_NOT_FRESH" + assert result[0]["index"]["freshness"] == "stale" + assert result[0]["historicalOnly"] is True + + +@pytest.mark.parametrize("source_sha", [None, "not-a-sha"]) +def test_missing_or_invalid_source_sha_is_unknown(repo, source_sha): + _, map_path = repo + _write_map(map_path, source_sha) + assert mcp_server.get_map_status("lexnotify/web")["freshness"] == "unknown" + + +def test_invalid_expected_sha_is_unknown_and_blocked(repo): + root, map_path = repo + _write_map(map_path, _head(root)) + assert mcp_server.get_map_status("lexnotify/web", "bad")["freshness"] == "unknown" + assert mcp_server.get_component("lexnotify/web", "src/a.ts", "bad")["error"]["code"] == "INDEX_NOT_FRESH" + + +def test_missing_expected_sha_is_unknown_and_blocks_all_conclusive_tools(repo): + root, map_path = repo + _write_map(map_path, _head(root)) + assert mcp_server.get_map_status("lexnotify/web")["freshness"] == "unknown" + assert mcp_server.search_capability("lexnotify/web", "a")[0]["error"]["code"] == "INDEX_NOT_FRESH" + assert mcp_server.get_component("lexnotify/web", "src/a.ts")["error"]["code"] == "INDEX_NOT_FRESH" + assert mcp_server.check_duplication("chat")["error"]["code"] == "INDEX_NOT_FRESH" + assert mcp_server.list_adrs()[0]["error"]["code"] == "INDEX_NOT_FRESH" + assert mcp_server.get_adr("decision")["error"]["code"] == "INDEX_NOT_FRESH" + assert mcp_server.verify_references("lexnotify/web", [], [])["error"]["code"] == "INDEX_NOT_FRESH" + + +@pytest.mark.parametrize("expected", ["a" * 7, "g" * 40, "A" * 40, "0" * 40]) +def test_short_malformed_or_nonexistent_expected_sha_is_unknown(repo, expected): + root, map_path = repo + _write_map(map_path, _head(root)) + assert mcp_server.get_map_status("lexnotify/web", expected)["freshness"] == "unknown" + + +def test_commit_from_another_branch_cannot_be_declared_fresh(repo): + root, map_path = repo + main = _head(root) + subprocess.run(["git", "-C", str(root), "switch", "-c", "other"], check=True, capture_output=True) + other = _commit(root, "other.txt") + subprocess.run(["git", "-C", str(root), "switch", "main"], check=True, capture_output=True) + assert _head(root) == main + _write_map(map_path, other) + status = mcp_server.get_map_status("lexnotify/web", other) + assert status["freshness"] == "unknown" + assert "outside the served branch history" in status["reason"] + + +def test_incompatible_schema_is_blocked(repo): + root, map_path = repo + _write_map(map_path, _head(root), schemaVersion="2.0") + assert mcp_server.get_map_status("lexnotify/web")["freshness"] == "incompatible" + assert mcp_server.verify_references("lexnotify/web", [], [])["error"]["code"] == "INDEX_NOT_FRESH" + + +def test_indexing_marker_never_replaces_last_valid_index(repo): + root, map_path = repo + _write_map(map_path, _head(root)) + map_path.with_suffix(map_path.suffix + ".indexing").write_text( + json.dumps({"status": "indexing"}), encoding="utf-8", + ) + assert mcp_server.get_map_status("lexnotify/web")["freshness"] == "indexing" + assert mcp_server.check_duplication("chat")["error"]["code"] == "INDEX_NOT_FRESH" + + +def test_backend_unavailable_is_failed(repo): + _, map_path = repo + assert not map_path.exists() + status = mcp_server.get_map_status("lexnotify/web") + assert status["freshness"] == "failed" + assert "FileNotFoundError" in status["reason"] + result = mcp_server.get_component("lexnotify/web", "src/a.ts", _head(repo[0])) + assert result["error"]["code"] == "INDEX_BACKEND_UNAVAILABLE" + + +def test_corrupt_indexing_marker_is_failed(repo): + root, map_path = repo + _write_map(map_path, _head(root)) + map_path.with_suffix(map_path.suffix + ".indexing").write_text("building", encoding="utf-8") + status = mcp_server.get_map_status("lexnotify/web", _head(root)) + assert status["freshness"] == "failed" + assert "invalid" in status["reason"] + + +def test_repository_mismatch_fails_before_reading_index(repo): + assert "repoId mismatch" in mcp_server.get_map_status("other/repo")["error"] + + +def test_repo_metadata_mismatch_is_incompatible(repo): + root, map_path = repo + _write_map(map_path, _head(root), repoId="other/repo") + assert mcp_server.get_map_status("lexnotify/web")["freshness"] == "incompatible"