deep_research/scripts/deploy_check.py

#!/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
CODEX_TEMPLATE = REPO_ROOT / "codex_adapter_templates" / "codex"
CODEX_ROOT = REPO_ROOT / ".codex"
OPENCODE_ROOT = REPO_ROOT / ".opencode"
AGENTS_SKILLS = REPO_ROOT / ".agents" / "skills"
OPENCODE_SKILLS = OPENCODE_ROOT / "skills"

REQUIRED_PATHS = [
    "AGENTS.md",
    "README.md",
    "PLAN.md",
    ".opencode/opencode.json",
    ".codex/config.toml",
    ".codex/agents/dr-pm.toml",
    ".codex/commands/dr-run.md",
    ".agents/skills/search-strategy/SKILL.md",
    "scripts/dr.py",
    "scripts/install_codex_adapter.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", ".opencode", ".agents/skills"])
    for item in REQUIRED_PATHS:
        if item.startswith((".codex/", ".opencode/", ".agents/")) and item not in tracked:
            warnings.append(f"not tracked by git: {item}")

    skill_files = sorted(AGENTS_SKILLS.glob("*/SKILL.md"))
    if len(skill_files) < 10:
        issues.append(f"expected at least 10 Codex skills, found {len(skill_files)}")

    check_toml(CODEX_ROOT / "config.toml", issues)
    for path in sorted((CODEX_ROOT / "agents").glob("*.toml")):
        check_toml(path, issues)

    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 files tracked: {sum(1 for p in tracked if p.startswith('.codex/'))}")
    print(f"  opencode files tracked: {sum(1 for p in tracked if p.startswith('.opencode/'))}")
    print(f"  codex skills: {len(skill_files)}")

    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) -> int:
    try:
        codex_written = copy_tree_contents(CODEX_TEMPLATE, CODEX_ROOT, force=force)
        skills_written = copy_tree_contents(OPENCODE_SKILLS, AGENTS_SKILLS, 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 files written: {len(codex_written)}")
    print(f"  .agents skills written: {len(skills_written)}")
    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="copy Codex templates and skills into hidden dirs")
    parser.add_argument("--force", action="store_true", help="overwrite existing files during --repair")
    args = parser.parse_args()

    if args.repair:
        return repair(force=args.force)
    return check_deployment()


if __name__ == "__main__":
    raise SystemExit(main())