fix(mcp): add active IDE vs global config-drift diagnostic (Closes #672)

docs/mcp-config-drift-runbook.md

@@ -0,0 +1,70 @@
+# MCP Config Drift Diagnostic & Sanctioned Repair Runbook (#672)
+
+This document describes the diagnostic framework for detecting configuration drift between the active IDE MCP configuration (`~/.gemini/antigravity-ide/mcp_config.json`) and the offline/global canonical configuration (`~/.gemini/config/mcp_config.json`), and establishes the **sanctioned repair runbook**.
+
+## Background & Problem Statement
+
+Offline tools like `test_mcp_conn.py` test the global configuration (`~/.gemini/config/mcp_config.json`) via `subprocess.Popen`. However, the active IDE/client namespace uses `~/.gemini/antigravity-ide/mcp_config.json`. When required Gitea role servers (`gitea-author`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, `gitea-controller`, `gitea-tools`) are missing or carry mismatched profile environments in the active IDE config:
+
+1. Offline tests pass (`test_mcp_conn.py` green).
+2. The IDE client returns `EOF` / `transport closed` when attempting role-scoped mutations.
+3. Operators misdiagnose missing server definitions as stale runtimes, leading to forbidden `pkill` attempts (#630) or `mtime` hacks (#655).
+
+## Diagnostic Tool: `mcp_config_drift.py`
+
+Run the diagnostic tool directly to compare configurations:
+
+```bash
+python3 mcp_config_drift.py --json
+```
+
+Or specify custom config locations:
+
+```bash
+python3 mcp_config_drift.py \
+  --active-config ~/.gemini/antigravity-ide/mcp_config.json \
+  --global-config ~/.gemini/config/mcp_config.json
+```
+
+### Key Diagnostic Outputs
+
+- `in_sync`: Boolean indicating if all required Gitea role servers exist in the active IDE config with matching profile declarations.
+- `missing_role_servers`: List of role servers present in global config but missing from active IDE config.
+- `profile_mismatches`: List of profile environment mismatches per server.
+- `reasons`: Explicit, human-readable list of drift causes.
+
+All returned payloads automatically redact secret tokens, DSNs, Authorization headers, and private keys.
+
+---
+
+## Sanctioned Repair Path (Step-by-Step)
+
+When `mcp_config_drift.py` reports drift (`in_sync: false`), execute the following **sanctioned repair steps**:
+
+1. **Backup Active IDE Config:**
+   ```bash
+   cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak
+   ```
+2. **Patch Active IDE Config:**
+   Copy the missing Gitea role server JSON blocks (`gitea-author`, `gitea-reviewer`, etc.) from `~/.gemini/config/mcp_config.json` into `~/.gemini/antigravity-ide/mcp_config.json`.
+3. **Reconnect via IDE/Client:**
+   Use the IDE / client UI reconnection control (or restart the IDE client app).
+4. **Verify Active Namespace Health:**
+   Invoke `gitea_whoami` (and optional `gitea_resolve_task_capability`) through the active IDE client on each required role namespace.
+
+---
+
+## FORBIDDEN Repair Actions (#630 / #655)
+
+The following actions are **strictly forbidden** for config drift repair:
+
+- ❌ **`pkill` or manual daemon process kill commands:** Process kills cause contamination and break active session leases.
+- ❌ **`mtime` touch edits:** Artificial mtime modifications mask stale runtimes without updating configuration.
+- ❌ **Source code edits:** Mutating python tool logic to bypass missing server entries.
+- ❌ **Session-state edits:** Direct database or lock-file state mutation.
+
+---
+
+## Final Report Guidelines
+
+A workflow final report **must not** rely on offline `test_mcp_conn.py` output alone. Final reports must include active-config evidence from live `gitea_whoami` calls on the active IDE namespaces.

mcp_config_drift.py

@@ -0,0 +1,237 @@
+"""Antigravity IDE vs Global MCP Config Drift Diagnostic (#672).
+
+Diagnoses config drift between the active IDE MCP configuration
+(e.g. ``~/.gemini/antigravity-ide/mcp_config.json``) and the offline/global
+canonical configuration (e.g. ``~/.gemini/config/mcp_config.json``).
+
+Hard rules (#672 / #630 / #655):
+* Distinguish offline/global success from active IDE namespace availability.
+* Never print tokens, DSNs, Authorization headers, or secret-bearing env vars.
+* Sanctioned repair path is: backup active config -> patch active config from canonical
+  -> reconnect through IDE/client -> verify with live ``gitea_whoami``.
+* FORBIDDEN: ``pkill``, mtime edits, source edits, or session-state edits for repair.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from webui import console_redaction
+
+DEFAULT_ACTIVE_IDE_CONFIG = "~/.gemini/antigravity-ide/mcp_config.json"
+DEFAULT_GLOBAL_CONFIG = "~/.gemini/config/mcp_config.json"
+
+REQUIRED_GITEA_ROLE_SERVERS = (
+    "gitea-author",
+    "gitea-reviewer",
+    "gitea-merger",
+    "gitea-reconciler",
+    "gitea-controller",
+    "gitea-tools",
+)
+
+SANCTIONED_REPAIR_RUNBOOK: tuple[str, ...] = (
+    "1. Backup active IDE config: cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak",
+    "2. Patch active IDE config: copy required missing Gitea role server entries from global config (~/.gemini/config/mcp_config.json) into active IDE config.",
+    "3. Reconnect via IDE/client UI or client restart (do NOT use host process kill).",
+    "4. Verify active namespace health using live gitea_whoami and gitea_resolve_task_capability on each role namespace.",
+    "FORBIDDEN REPAIR PATHS: pkill / host process kill, mtime touch edits, source code edits, or session-state edits.",
+)
+
+
+def resolve_config_path(path_str: str) -> Path:
+    """Expand user and resolve absolute path."""
+    return Path(os.path.expanduser(path_str)).resolve()
+
+
+def load_mcp_config(config_path: str | Path) -> tuple[dict[str, Any] | None, str | None]:
+    """Load and parse JSON MCP configuration from file.
+
+    Returns (config_dict, error_message).
+    """
+    resolved = resolve_config_path(str(config_path))
+    if not resolved.exists():
+        return None, f"file_not_found: {resolved}"
+    try:
+        with open(resolved, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        if not isinstance(data, dict):
+            return None, f"invalid_schema: root is not a JSON object in {resolved}"
+        return data, None
+    except Exception as exc:
+        return None, f"unreadable_json: {exc} in {resolved}"
+
+
+def extract_mcp_servers(config: dict[str, Any] | None) -> dict[str, dict[str, Any]]:
+    """Extract the mcpServers or mcp_servers mapping safely."""
+    if not config:
+        return {}
+    servers = config.get("mcpServers") or config.get("mcp_servers") or {}
+    if isinstance(servers, dict):
+        return {str(k): v for k, v in servers.items() if isinstance(v, dict)}
+    return {}
+
+
+def _safe_redact_server_config(srv_cfg: dict[str, Any]) -> dict[str, Any]:
+    """Redact secrets from environment variables and command line args."""
+    safe = {}
+    if "command" in srv_cfg:
+        safe["command"] = str(srv_cfg["command"])
+    if "args" in srv_cfg and isinstance(srv_cfg["args"], list):
+        safe["args"] = [console_redaction.redact_text(str(a)) for a in srv_cfg["args"]]
+    if "env" in srv_cfg and isinstance(srv_cfg["env"], dict):
+        safe_env = {}
+        for k, v in srv_cfg["env"].items():
+            if any(secret_kw in k.lower() for secret_kw in ("token", "secret", "pass", "key", "auth")):
+                safe_env[k] = "[REDACTED]"
+            else:
+                safe_env[k] = console_redaction.redact_text(str(v))
+        safe["env"] = safe_env
+    return safe
+
+
+def analyze_config_drift(
+    active_config_path: str = DEFAULT_ACTIVE_IDE_CONFIG,
+    global_config_path: str = DEFAULT_GLOBAL_CONFIG,
+) -> dict[str, Any]:
+    """Analyze MCP configuration drift between active IDE config and global config.
+
+    Returns structured diagnostic output.
+    """
+    active_resolved = resolve_config_path(active_config_path)
+    global_resolved = resolve_config_path(global_config_path)
+
+    active_cfg, active_err = load_mcp_config(active_resolved)
+    global_cfg, global_err = load_mcp_config(global_resolved)
+
+    active_servers = extract_mcp_servers(active_cfg)
+    global_servers = extract_mcp_servers(global_cfg)
+
+    missing_role_servers: list[str] = []
+    present_role_servers: list[str] = []
+    profile_mismatches: list[dict[str, Any]] = []
+    reasons: list[str] = []
+
+    if active_err:
+        reasons.append(f"Active IDE config error: {active_err}")
+    if global_err:
+        reasons.append(f"Global canonical config error: {global_err}")
+
+    # Check Gitea role servers
+    for srv_name in REQUIRED_GITEA_ROLE_SERVERS:
+        in_active = srv_name in active_servers
+        in_global = srv_name in global_servers
+
+        if in_active:
+            present_role_servers.append(srv_name)
+        elif in_global:
+            missing_role_servers.append(srv_name)
+            reasons.append(
+                f"Missing Gitea role server '{srv_name}' in active IDE config ({active_resolved})"
+            )
+
+        if in_active and in_global:
+            # Compare profiles & environments
+            act_env = active_servers[srv_name].get("env", {}) if isinstance(active_servers[srv_name], dict) else {}
+            glo_env = global_servers[srv_name].get("env", {}) if isinstance(global_servers[srv_name], dict) else {}
+
+            act_prof = act_env.get("GITEA_MCP_PROFILE") or act_env.get("GITEA_PROFILE_NAME")
+            glo_prof = glo_env.get("GITEA_MCP_PROFILE") or glo_env.get("GITEA_PROFILE_NAME")
+
+            if act_prof != glo_prof:
+                mismatch_item = {
+                    "server": srv_name,
+                    "active_profile": act_prof,
+                    "global_profile": glo_prof,
+                }
+                profile_mismatches.append(mismatch_item)
+                reasons.append(
+                    f"Profile mismatch for '{srv_name}': active='{act_prof}' != global='{glo_prof}'"
+                )
+
+    in_sync = bool(
+        not active_err
+        and not global_err
+        and not missing_role_servers
+        and not profile_mismatches
+    )
+
+    report = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "in_sync": in_sync,
+        "active_config_path": str(active_resolved),
+        "active_config_exists": active_cfg is not None,
+        "global_config_path": str(global_resolved),
+        "global_config_exists": global_cfg is not None,
+        "required_role_servers": list(REQUIRED_GITEA_ROLE_SERVERS),
+        "present_role_servers": present_role_servers,
+        "missing_role_servers": missing_role_servers,
+        "profile_mismatches": profile_mismatches,
+        "reasons": reasons,
+        "sanctioned_repair_runbook": list(SANCTIONED_REPAIR_RUNBOOK),
+        "forbidden_repair_methods": [
+            "pkill / host process kill",
+            "mtime touch edits",
+            "source code edits",
+            "session-state edits",
+        ],
+    }
+
+    return console_redaction.redact_payload(report)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Diagnose Gitea MCP role server config drift between active IDE and global config."
+    )
+    parser.add_argument(
+        "--active-config",
+        default=DEFAULT_ACTIVE_IDE_CONFIG,
+        help="Path to active IDE MCP config JSON",
+    )
+    parser.add_argument(
+        "--global-config",
+        default=DEFAULT_GLOBAL_CONFIG,
+        help="Path to global/canonical MCP config JSON",
+    )
+    parser.add_argument(
+        "--json", action="store_true", help="Print raw JSON report"
+    )
+
+    args = parser.parse_args()
+
+    report = analyze_config_drift(args.active_config, args.global_config)
+
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print("=== MCP Config Drift Diagnostic Report ===")
+        print(f"Timestamp: {report['timestamp']}")
+        print(f"In Sync: {report['in_sync']}")
+        print(f"Active IDE Config: {report['active_config_path']} (exists={report['active_config_exists']})")
+        print(f"Global Config: {report['global_config_path']} (exists={report['global_config_exists']})")
+        print(f"Present Role Servers: {', '.join(report['present_role_servers']) if report['present_role_servers'] else 'None'}")
+        print(f"Missing Role Servers: {', '.join(report['missing_role_servers']) if report['missing_role_servers'] else 'None'}")
+        if report['profile_mismatches']:
+            print("Profile Mismatches:")
+            for m in report['profile_mismatches']:
+                print(f"  - {m['server']}: active={m['active_profile']} vs global={m['global_profile']}")
+        if report['reasons']:
+            print("Drift Reasons:")
+            for r in report['reasons']:
+                print(f"  - {r}")
+        print("\nSanctioned Repair Runbook:")
+        for step in report['sanctioned_repair_runbook']:
+            print(f"  {step}")
+
+    sys.exit(0 if report["in_sync"] else 1)
+
+
+if __name__ == "__main__":
+    main()

tests/test_mcp_config_drift.py

@@ -0,0 +1,149 @@
+"""Unit tests for mcp_config_drift.py (#672)."""
+
+from __future__ import annotations
+
+import json
+import pytest
+from pathlib import Path
+
+from mcp_config_drift import (
+    REQUIRED_GITEA_ROLE_SERVERS,
+    analyze_config_drift,
+    load_mcp_config,
+)
+
+
+@pytest.fixture
+def sample_global_config() -> dict:
+    return {
+        "mcpServers": {
+            "gitea-author": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author", "SENTRY_AUTH_TOKEN": "secret-token-999"},
+            },
+            "gitea-reviewer": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reviewer"},
+            },
+            "gitea-merger": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-merger"},
+            },
+            "gitea-reconciler": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-reconciler"},
+            },
+            "gitea-controller": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-controller"},
+            },
+            "gitea-tools": {
+                "command": "python3",
+                "args": ["gitea_mcp_server.py"],
+                "env": {"GITEA_MCP_PROFILE": "prgs-author"},
+            },
+        }
+    }
+
+
+def write_json(path: Path, data: dict) -> str:
+    path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return str(path)
+
+
+def test_drift_detection_in_sync(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is True
+    assert report["missing_role_servers"] == []
+    assert report["profile_mismatches"] == []
+    assert set(report["present_role_servers"]) == set(REQUIRED_GITEA_ROLE_SERVERS)
+
+
+def test_drift_detection_missing_author(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-author"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-author" in report["missing_role_servers"]
+    assert "gitea-author" not in report["present_role_servers"]
+
+
+def test_drift_detection_missing_reviewer(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    del active_config["mcpServers"]["gitea-reviewer"]
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert "gitea-reviewer" in report["missing_role_servers"]
+
+
+def test_drift_detection_profile_mismatch(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    active_config = json.loads(json.dumps(sample_global_config))
+    active_config["mcpServers"]["gitea-author"]["env"]["GITEA_MCP_PROFILE"] = "dadeschools-author"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, active_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+
+    assert report["in_sync"] is False
+    assert len(report["profile_mismatches"]) == 1
+    mismatch = report["profile_mismatches"][0]
+    assert mismatch["server"] == "gitea-author"
+    assert mismatch["active_profile"] == "dadeschools-author"
+    assert mismatch["global_profile"] == "prgs-author"
+
+
+def test_secret_redaction_in_drift_report(tmp_path, sample_global_config):
+    glob_file = tmp_path / "global_mcp.json"
+    act_file = tmp_path / "active_mcp.json"
+
+    write_json(glob_file, sample_global_config)
+    write_json(act_file, sample_global_config)
+
+    report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
+    serialized = str(report)
+
+    assert "secret-token-999" not in serialized
+
+
+def test_sanctioned_runbook_forbids_pkill():
+    report = analyze_config_drift(active_config_path="/nonexistent/path/active.json", global_config_path="/nonexistent/path/global.json")
+
+    runbook_text = " ".join(report["sanctioned_repair_runbook"]).lower()
+    forbidden_text = " ".join(report["forbidden_repair_methods"]).lower()
+
+    assert "pkill" in forbidden_text
+    assert "mtime" in forbidden_text
+    assert "source" in forbidden_text
+    assert "session-state" in forbidden_text