#!/usr/bin/env python3 """Check and repair a Deep Research deployment checkout.""" from __future__ import annotations import argparse import os import shutil import subprocess import sys from pathlib import Path try: import tomllib except ModuleNotFoundError: # pragma: no cover - Python < 3.11 fallback. tomllib = None # type: ignore[assignment] REPO_ROOT = Path(__file__).resolve().parent.parent if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from scripts.deploy_adapters import default_codex_home, deploy_codex from scripts.runtime.skills import SkillRegistry CODEX_TEMPLATE = REPO_ROOT / "codex_adapter_templates" / "codex" AGENT_SKILLS = REPO_ROOT / ".agent" / "skills" REQUIRED_PATHS = [ "AGENTS.md", "GEMINI.md", "README.md", "PLAN.md", ".agent/agents.md", ".agent/rules/deep-research-antigravity.md", ".agent/workflows/deep-research-native.md", ".agent/skills/search-strategy/SKILL.md", "codex_adapter_templates/codex/config.toml", "codex_adapter_templates/codex/agents/dr-pm.toml", "codex_adapter_templates/codex/commands/dr-run.md", "skills/deep-research/SKILL.md", "skills/document-ingest/SKILL.md", "scripts/dr.py", "scripts/deploy_adapters.py", "scripts/export_antigravity_workspace.py", ] REQUIRED_ENV_KEYS = [ "ZENMUX_API_KEY", "TAVILY_API_KEY", "BRAVE_API_KEY", "EXA_API_KEY", ] def rel(path: Path) -> str: return str(path.relative_to(REPO_ROOT)) def copy_tree_contents(src: Path, dst: Path, *, force: bool) -> list[Path]: written: list[Path] = [] if not src.exists(): raise RuntimeError(f"source not found: {rel(src)}") dst.mkdir(parents=True, exist_ok=True) for item in src.rglob("*"): target = dst / item.relative_to(src) if item.is_dir(): target.mkdir(parents=True, exist_ok=True) continue if target.exists() and not force: continue target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(item, target) written.append(target) return written def parse_env(path: Path) -> dict[str, str]: values: dict[str, str] = {} if not path.exists(): return values for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) values[key.strip()] = value.strip().strip('"').strip("'") return values def git_tracked(paths: list[str]) -> set[str]: proc = subprocess.run( ["git", "ls-files", *paths], cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False, ) return set(proc.stdout.splitlines()) def check_toml(path: Path, issues: list[str]) -> None: if not path.exists(): return if tomllib is None: issues.append("Python tomllib unavailable; skip TOML parse checks") return try: tomllib.loads(path.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 - deployment diagnostics. issues.append(f"{rel(path)} TOML parse failed: {exc}") def check_deployment() -> int: issues: list[str] = [] warnings: list[str] = [] for item in REQUIRED_PATHS: if not (REPO_ROOT / item).exists(): issues.append(f"missing required path: {item}") tracked = git_tracked(["codex_adapter_templates", ".agent", "skills"]) legacy_tracked = git_tracked([".codex", ".opencode", ".claude", ".gemini", ".agents"]) if legacy_tracked: warnings.append("legacy platform adapter files are still tracked; prefer Antigravity clean workspace or remove them from the branch") for item in REQUIRED_PATHS: if item.startswith(("codex_adapter_templates/", ".agent/")) and item not in tracked: warnings.append(f"not tracked by git: {item}") skills = SkillRegistry().list() if len(skills) < 10: issues.append(f"expected at least 10 Codex skills, found {len(skills)}") check_toml(CODEX_TEMPLATE / "config.toml", issues) for path in sorted((CODEX_TEMPLATE / "agents").glob("*.toml")): check_toml(path, issues) codex_home = default_codex_home() if not (codex_home / "commands" / "dr-run.md").exists(): warnings.append(f"Codex adapter not deployed to {codex_home}; run scripts/deploy_adapters.py codex") env_values = {key: os.environ.get(key, "") for key in REQUIRED_ENV_KEYS} env_values.update({k: v for k, v in parse_env(REPO_ROOT / "secrets.env").items() if not env_values.get(k)}) missing_env = [key for key in REQUIRED_ENV_KEYS if not env_values.get(key)] if missing_env: warnings.append("missing optional/required API keys for full automation: " + ", ".join(missing_env)) print("Deep Research deployment check") print(f" repo: {REPO_ROOT}") print(f" codex template files tracked: {sum(1 for p in tracked if p.startswith('codex_adapter_templates/'))}") print(f" legacy platform adapter files tracked: {len(legacy_tracked)}") print(f" codex home: {codex_home}") print(f" antigravity .agent files tracked: {sum(1 for p in tracked if p.startswith('.agent/'))}") print(f" codex skills: {len(skills)}") if warnings: print("\nWarnings:") for item in warnings: print(f" - {item}") if issues: print("\nIssues:") for item in issues: print(f" - {item}") return 1 print("\nDeployment check passed.") return 0 def repair(force: bool, codex_home: Path | None) -> int: try: codex_result = deploy_codex(target=codex_home, force=force) except PermissionError as exc: print(f"repair failed: permission denied: {exc}", file=sys.stderr) return 1 except RuntimeError as exc: print(f"repair failed: {exc}", file=sys.stderr) return 1 print("Repair completed.") print(f" Codex home files written: {len(codex_result.written)}") print(f" Codex home: {codex_result.target}") print(f" .agent skills source: {AGENT_SKILLS}") return check_deployment() def main() -> int: parser = argparse.ArgumentParser(description="Check or repair Deep Research deployment files") parser.add_argument("--repair", action="store_true", help="deploy Codex templates outside the repo and sync skills") parser.add_argument("--force", action="store_true", help="overwrite existing files during --repair") parser.add_argument("--codex-home", type=Path, help="Codex home target for --repair; defaults to $CODEX_HOME or ~/.codex") args = parser.parse_args() if args.repair: return repair(force=args.force, codex_home=args.codex_home) return check_deployment() if __name__ == "__main__": raise SystemExit(main())