"""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()