fix(mcp): bloquear conclusões quando o índice não estiver fresh

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", [])

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