hermes-agent/wizard-bootstrap/wizard_bootstrap.py

#!/usr/bin/env python3
"""
wizard_bootstrap.py — Wizard Environment Validator

Validates that a new wizard's forge environment is ready:
  1. Python version check (>=3.11)
  2. Core dependencies installed
  3. Gitea authentication
  4. Telegram connectivity
  5. Smoke test (hermes import)

Usage:
    python wizard-bootstrap/wizard_bootstrap.py
    python wizard-bootstrap/wizard_bootstrap.py --fix
    python wizard-bootstrap/wizard_bootstrap.py --json

Exits 0 if all checks pass, 1 if any check fails.
"""

import argparse
import importlib
import json
import os
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional


# ---------------------------------------------------------------------------
# Result model
# ---------------------------------------------------------------------------

@dataclass
class CheckResult:
    name: str
    passed: bool
    message: str
    fix_hint: Optional[str] = None
    detail: Optional[str] = None


@dataclass
class BootstrapReport:
    checks: list[CheckResult] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return all(c.passed for c in self.checks)

    @property
    def failed(self) -> list[CheckResult]:
        return [c for c in self.checks if not c.passed]

    def add(self, result: CheckResult) -> None:
        self.checks.append(result)


# ---------------------------------------------------------------------------
# Individual checks
# ---------------------------------------------------------------------------

def check_python_version() -> CheckResult:
    """Require Python >= 3.11."""
    major, minor, micro = sys.version_info[:3]
    ok = (major, minor) >= (3, 11)
    return CheckResult(
        name="python_version",
        passed=ok,
        message=f"Python {major}.{minor}.{micro}",
        fix_hint="Install Python 3.11+ via uv, pyenv, or your OS package manager.",
    )


def check_core_deps() -> CheckResult:
    """Verify that hermes core Python packages are importable."""
    required = [
        "openai",
        "anthropic",
        "dotenv",
        "yaml",
        "rich",
        "requests",
        "pydantic",
        "prompt_toolkit",
    ]
    missing = []
    for pkg in required:
        # dotenv ships as 'python-dotenv' but imports as 'dotenv'
        try:
            importlib.import_module(pkg)
        except ModuleNotFoundError:
            missing.append(pkg)

    if missing:
        return CheckResult(
            name="core_deps",
            passed=False,
            message=f"Missing packages: {', '.join(missing)}",
            fix_hint="Run: uv pip install -r requirements.txt  (or: pip install -r requirements.txt)",
        )
    return CheckResult(name="core_deps", passed=True, message="All core packages importable")


def check_hermes_importable() -> CheckResult:
    """Smoke-test: import hermes_constants (no side effects)."""
    # Add repo root to sys.path so we can import regardless of cwd
    repo_root = str(Path(__file__).parent.parent)
    if repo_root not in sys.path:
        sys.path.insert(0, repo_root)
    try:
        import hermes_constants  # noqa: F401

        return CheckResult(name="hermes_smoke", passed=True, message="hermes_constants imported OK")
    except Exception as exc:
        return CheckResult(
            name="hermes_smoke",
            passed=False,
            message=f"Import error: {exc}",
            fix_hint="Ensure you are in the hermes-agent repo root and your venv is active.",
        )


def check_gitea_auth() -> CheckResult:
    """Verify Gitea token env var is set and the API responds."""
    token = os.environ.get("GITEA_TOKEN") or os.environ.get("FORGE_TOKEN")
    if not token:
        return CheckResult(
            name="gitea_auth",
            passed=False,
            message="GITEA_TOKEN / FORGE_TOKEN not set",
            fix_hint="Export GITEA_TOKEN=<your-token> in your shell or ~/.hermes/.env",
        )

    # Attempt a lightweight API call — list repos endpoint returns quickly
    forge_url = os.environ.get("FORGE_URL", "https://forge.alexanderwhitestone.com")
    try:
        import requests  # noqa: PLC0415

        resp = requests.get(
            f"{forge_url}/api/v1/repos/search",
            headers={"Authorization": f"token {token}"},
            params={"limit": 1},
            timeout=10,
        )
        if resp.status_code == 200:
            return CheckResult(name="gitea_auth", passed=True, message="Gitea API reachable and token valid")
        return CheckResult(
            name="gitea_auth",
            passed=False,
            message=f"Gitea API returned HTTP {resp.status_code}",
            fix_hint="Check that your GITEA_TOKEN is correct and not expired.",
        )
    except Exception as exc:
        return CheckResult(
            name="gitea_auth",
            passed=False,
            message=f"Gitea API unreachable: {exc}",
            fix_hint="Check network connectivity and FORGE_URL env var.",
        )


def check_telegram_connectivity() -> CheckResult:
    """Verify Telegram bot token is set and the Bot API responds."""
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    if not token:
        return CheckResult(
            name="telegram",
            passed=False,
            message="TELEGRAM_BOT_TOKEN not set",
            fix_hint="Export TELEGRAM_BOT_TOKEN=<token> in your shell or ~/.hermes/.env",
        )

    try:
        import requests  # noqa: PLC0415

        resp = requests.get(
            f"https://api.telegram.org/bot{token}/getMe",
            timeout=10,
        )
        if resp.status_code == 200:
            data = resp.json()
            username = data.get("result", {}).get("username", "?")
            return CheckResult(
                name="telegram",
                passed=True,
                message=f"Telegram bot @{username} reachable",
            )
        return CheckResult(
            name="telegram",
            passed=False,
            message=f"Telegram API returned HTTP {resp.status_code}",
            fix_hint="Check that TELEGRAM_BOT_TOKEN is valid.",
        )
    except Exception as exc:
        return CheckResult(
            name="telegram",
            passed=False,
            message=f"Telegram unreachable: {exc}",
            fix_hint="Check network connectivity.",
        )


def check_env_vars() -> CheckResult:
    """Check that at least one LLM provider key is configured."""
    provider_keys = [
        "OPENROUTER_API_KEY",
        "ANTHROPIC_API_KEY",
        "ANTHROPIC_TOKEN",
        "OPENAI_API_KEY",
        "GLM_API_KEY",
        "KIMI_API_KEY",
        "MINIMAX_API_KEY",
    ]
    found = [k for k in provider_keys if os.environ.get(k)]
    if found:
        return CheckResult(
            name="llm_provider",
            passed=True,
            message=f"LLM provider key(s) present: {', '.join(found)}",
        )
    return CheckResult(
        name="llm_provider",
        passed=False,
        message="No LLM provider API key found",
        fix_hint=(
            "Set at least one of: OPENROUTER_API_KEY, ANTHROPIC_API_KEY, OPENAI_API_KEY "
            "in ~/.hermes/.env or your shell."
        ),
    )


def check_hermes_home() -> CheckResult:
    """Verify HERMES_HOME directory exists and is writable."""
    hermes_home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
    if not hermes_home.exists():
        return CheckResult(
            name="hermes_home",
            passed=False,
            message=f"HERMES_HOME does not exist: {hermes_home}",
            fix_hint="Run 'hermes setup' or create the directory manually.",
        )
    if not os.access(hermes_home, os.W_OK):
        return CheckResult(
            name="hermes_home",
            passed=False,
            message=f"HERMES_HOME not writable: {hermes_home}",
            fix_hint=f"Fix permissions: chmod u+w {hermes_home}",
        )
    return CheckResult(
        name="hermes_home",
        passed=True,
        message=f"HERMES_HOME OK: {hermes_home}",
    )


# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------

def _load_dotenv_if_available() -> None:
    """Load ~/.hermes/.env so token checks work without manual export."""
    hermes_home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
    env_path = hermes_home / ".env"
    if env_path.exists():
        try:
            from dotenv import load_dotenv  # noqa: PLC0415

            load_dotenv(env_path, override=False)
        except Exception:
            pass  # dotenv not installed yet — that's fine


def run_all_checks() -> BootstrapReport:
    report = BootstrapReport()
    _load_dotenv_if_available()

    checks = [
        check_python_version,
        check_core_deps,
        check_hermes_importable,
        check_hermes_home,
        check_env_vars,
        check_gitea_auth,
        check_telegram_connectivity,
    ]
    for fn in checks:
        result = fn()
        report.add(result)

    return report


# ---------------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------------

_GREEN = "\033[32m"
_RED = "\033[31m"
_YELLOW = "\033[33m"
_BOLD = "\033[1m"
_RESET = "\033[0m"


def _render_terminal(report: BootstrapReport) -> None:
    print(f"\n{_BOLD}=== Wizard Bootstrap — Environment Check ==={_RESET}\n")
    for check in report.checks:
        icon = f"{_GREEN}✓{_RESET}" if check.passed else f"{_RED}✗{_RESET}"
        label = check.name.replace("_", " ").title()
        print(f"  {icon}  {_BOLD}{label}{_RESET}: {check.message}")
        if not check.passed and check.fix_hint:
            print(f"       {_YELLOW}→ {check.fix_hint}{_RESET}")
        if check.detail:
            print(f"       {check.detail}")

    total = len(report.checks)
    passed = sum(1 for c in report.checks if c.passed)
    print()
    if report.passed:
        print(f"{_GREEN}{_BOLD}All {total} checks passed. Forge is ready.{_RESET}\n")
    else:
        failed = total - passed
        print(
            f"{_RED}{_BOLD}{failed}/{total} check(s) failed.{_RESET} "
            f"Resolve the issues above before going online.\n"
        )


def _render_json(report: BootstrapReport) -> None:
    out = {
        "passed": report.passed,
        "summary": {
            "total": len(report.checks),
            "passed": sum(1 for c in report.checks if c.passed),
            "failed": sum(1 for c in report.checks if not c.passed),
        },
        "checks": [
            {
                "name": c.name,
                "passed": c.passed,
                "message": c.message,
                "fix_hint": c.fix_hint,
                "detail": c.detail,
            }
            for c in report.checks
        ],
    }
    print(json.dumps(out, indent=2))


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="Validate the forge wizard environment."
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output results as JSON",
    )
    args = parser.parse_args()

    report = run_all_checks()

    if args.json:
        _render_json(report)
    else:
        _render_terminal(report)

    sys.exit(0 if report.passed else 1)


if __name__ == "__main__":
    main()