379 lines
11 KiB
Python
379 lines
11 KiB
Python
|
|
#!/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()
|