hermes-esp32-voice-gateway/scripts/restore_after_update.py

#!/usr/bin/env python3
"""
Restore ESP32 voice gateway after `hermes update`.

Usage:
  ./scripts/restore_after_update.py
  python3 scripts/restore_after_update.py --verify
  python3 scripts/restore_after_update.py --force

What it does:
- Checks if gateway/platforms/api_server.py has ESP32 routes
- If missing, applies patch/api_server_esp32_live_draft_with_llm.patch via git apply
- Verifies routes return 400 not 404 (400 means route exists but missing WS upgrade / empty body)
- Copies patch to ~/.hermes/ fallback location
- Prints next steps (gateway restart)

Env:
  HERMES_AGENT_DIR - override hermes-agent checkout path
"""

import argparse
import subprocess
import sys
from pathlib import Path

def find_agent_dir() -> Path | None:
    import os
    candidates = []
    env_dir = os.getenv("HERMES_AGENT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    candidates.append(Path.home() / ".hermes" / "hermes-agent")
    alt_path = Path(__file__).resolve().parents[2] / ".hermes" / "hermes-agent"
    if alt_path.exists():
        candidates.append(alt_path)
    try:
        import hermes_cli  # type: ignore
        candidates.append(Path(hermes_cli.__file__).parents[1])
    except Exception:
        pass
    for c in candidates:
        if c and (c / "gateway" / "platforms" / "api_server.py").exists():
            return c
    return None

def has_routes(api_file: Path) -> bool:
    try:
        txt = api_file.read_text()
        return "ESP32_AUDIO_MAX_BYTES" in txt[:20000] and "/api/esp32/voice" in txt and "_handle_esp32_voice_ws" in txt
    except Exception:
        return False

def find_patch(repo_root: Path) -> Path | None:
    for p in [
        repo_root / "patch" / "api_server_esp32_live_draft_with_llm.patch",
        repo_root / "hermes_plugin" / "patch" / "api_server_esp32_live_draft_with_llm.patch",
        Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch",
        Path.home() / ".hermes" / "esp32_voice_patch.patch",
    ]:
        if p.exists() and p.stat().st_size > 1000:
            return p
    # any in patch/
    for f in (repo_root / "patch").glob("*.patch"):
        if f.stat().st_size > 1000:
            return f
    return None

def apply_patch(patch_path: Path, agent_dir: Path, force: bool=False) -> bool:
    api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
    print(f"[restore] patch={patch_path} agent_dir={agent_dir} api_file={api_file} force={force}")
    if not (agent_dir / ".git").exists():
        print(f"[restore] {agent_dir} is not a git repo")
        return False
    if has_routes(api_file) and not force:
        print("[restore] routes already present, skipping (use --force to reapply)")
        return True
    # Try apply
    if force:
        # reset file to HEAD first? we do checkout
        try:
            subprocess.run(["git", "checkout", "HEAD", "--", "gateway/platforms/api_server.py"], cwd=str(agent_dir), timeout=15)
        except Exception:
            pass
    # Check
    chk = subprocess.run(["git", "apply", "--check", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=15)
    if chk.returncode != 0:
        print(f"[restore] git apply --check output:\n{chk.stdout[:1000]}\n{chk.stderr[:1000]}")
        # Try 3way if not forced yet
        if not force:
            print("[restore] trying git apply --3way")
            r = subprocess.run(["git", "apply", "--3way", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=30)
            print(r.stdout[-500:] if r.stdout else "", r.stderr[-1000:] if r.stderr else "")
            if r.returncode == 0:
                print("[restore] 3way apply ok")
                return has_routes(api_file)
        else:
            r = subprocess.run(["git", "apply", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=30)
            print(r.stdout[-500:] if r.stdout else "", r.stderr[-1000:] if r.stderr else "")
            return r.returncode == 0 and has_routes(api_file)
    else:
        r = subprocess.run(["git", "apply", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=30)
        if r.returncode != 0:
            print(f"[restore] apply failed: {r.stdout} {r.stderr}")
            return False
        print(f"[restore] applied OK")
        return has_routes(api_file)

def verify_curly_brace_imports(agent_dir: Path) -> bool:
    """Guard against missing top imports ipaddress/urlparse (bug that caused 500 loop 2026-07-14)"""
    api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
    txt = api_file.read_text()
    has_ip = "import ipaddress" in txt
    has_url = "from urllib.parse import urlparse" in txt
    print(f"[verify] top imports: ipaddress={has_ip} urlparse={has_url}")
    if not has_ip or not has_url:
        print("[verify] MISSING critical imports – this will cause 500 import loop (see skill pitfalls). Fix by ensuring patch includes them.")
        return False
    return True

def compile_check(agent_dir: Path) -> bool:
    api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
    r = subprocess.run([sys.executable, "-m", "py_compile", str(api_file)], capture_output=True, text=True, timeout=15)
    if r.returncode != 0:
        print(f"[compile] failed: {r.stderr}")
        return False
    print("[compile] OK")
    return True

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--force", action="store_true", help="Force reapply even if routes present")
    parser.add_argument("--verify", action="store_true", help="Only verify, don't apply")
    args = parser.parse_args()

    repo_root = Path(__file__).resolve().parents[1]
    agent_dir = find_agent_dir()
    if not agent_dir:
        print("Could not find hermes-agent dir (checked ~/.hermes/hermes-agent, HERMES_AGENT_DIR, pip location)")
        sys.exit(2)
    api_file = agent_dir / "gateway" / "platforms" / "api_server.py"
    print(f"Agent dir: {agent_dir}")
    print(f"API file: {api_file}")
    print(f"Has routes: {has_routes(api_file)}")

    if args.verify:
        ok = verify_curly_brace_imports(agent_dir) and compile_check(agent_dir) and has_routes(api_file)
        sys.exit(0 if ok else 1)

    patch = find_patch(repo_root)
    if not patch:
        print(f"No patch found. Looked in {repo_root}/patch/ and ~/.hermes/*.patch")
        sys.exit(2)
    print(f"Using patch: {patch} ({patch.stat().st_size} bytes)")

    ok = apply_patch(patch, agent_dir, force=args.force)
    if not ok:
        print("[restore] FAILED to restore")
        sys.exit(1)

    verify_curly_brace_imports(agent_dir)
    compile_check(agent_dir)

    # Copy to ~/.hermes fallback
    try:
        dest = Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch"
        if patch.resolve() != dest.resolve():
            import shutil
            shutil.copy2(patch, dest)
            print(f"[restore] copied patch to {dest} for fallback")
    except Exception as e:
        print(f"[restore] failed to copy fallback: {e}")

    print("""
[restore] DONE. Next:
  1. hermes gateway restart (from a DIFFERENT shell, not from inside Hermes chat - see skill pitfalls)
     Use: systemctl --user restart hermes-gateway hermes-gateway-kids --no-pager || hermes gateway restart
  2. If stuck in deactivating:
     pkill -9 -f "hermes_cli.main gateway run"; systemctl --user reset-failed hermes-gateway*; systemctl --user start hermes-gateway
  3. Verify routes (should be 400 not 404):
     API_KEY=$(grep -E API_SERVER_KEY .env -h ~/.hermes/.env ~/.hermes/profiles/*/ .env 2>/dev/null | head -1 | cut -d= -f2 | tr -d '\"'\\'')
     curl -s -o /dev/null -w "%{http_code}\\n" -H "Authorization: Bearer $API_KEY" http://127.0.0.1:8642/api/esp32/voice/ws
     curl -s -o /dev/null -w "%{http_code}\\n" -H "Authorization: Bearer $API_KEY" -X POST http://127.0.0.1:8642/api/esp32/voice --data-binary @/dev/null
     # Both should print 400. 404 means routes still missing.
""")

if __name__ == "__main__":
    main()