#!/usr/bin/env python3 """Watch Voicebox MCP reachability and refresh Hermes gateway when it recovers. Why: Hermes discovers native MCP tools at gateway startup. If the Mac mini Voicebox app or SSH tunnel is down during startup, the running gateway can miss Voicebox tools until the gateway is restarted. This watchdog is quiet when the state is unchanged, alerts on down/up transitions, and restarts the default gateway once when Voicebox recovers. """ from __future__ import annotations import json import subprocess from datetime import datetime, timezone from pathlib import Path STATE_PATH = Path.home() / ".hermes" / "state" / "voicebox_mcp_watchdog.json" LOG_PATH = Path.home() / ".hermes" / "logs" / "voicebox_mcp_watchdog.jsonl" HERMES = "/home/adolforeyna/.hermes/hermes-agent/venv/bin/python" HERMES_MAIN = "hermes_cli.main" MAC_HOST = "adolforeyna@192.168.68.102" VOICEBOX_PORT = "17493" TUNNEL_UNIT = "voicebox-mcp-tunnel.service" GATEWAY_UNIT = "hermes-gateway.service" def run(cmd: list[str], timeout: int = 45) -> subprocess.CompletedProcess[str]: return subprocess.run(cmd, text=True, capture_output=True, timeout=timeout) def probe_voicebox() -> tuple[bool, int, str]: probe = run([HERMES, "-m", HERMES_MAIN, "mcp", "test", "voicebox"], timeout=60) output = (probe.stdout + probe.stderr).strip() ok = probe.returncode == 0 and "Tools discovered" in output return ok, probe.returncode, output def restart_tunnel() -> subprocess.CompletedProcess[str]: return run(["systemctl", "--user", "restart", TUNNEL_UNIT], timeout=30) def ensure_voicebox_app_running() -> subprocess.CompletedProcess[str]: """Start Voicebox on the Mac mini if its local MCP port is down.""" remote_script = f''' set -e if curl -fsS -m 5 -H 'Accept: application/json, text/event-stream' http://127.0.0.1:{VOICEBOX_PORT}/mcp/ >/dev/null 2>&1; then echo already-up exit 0 fi open -a Voicebox for i in 1 2 3 4 5 6 7 8 9 10 11 12; do sleep 5 if curl -fsS -m 5 -H 'Accept: application/json, text/event-stream' http://127.0.0.1:{VOICEBOX_PORT}/mcp/ >/dev/null 2>&1; then echo started exit 0 fi done echo still-down >&2 exit 1 ''' return run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", MAC_HOST, remote_script], timeout=90) def schedule_gateway_restart(event: dict) -> str: try: restart = run([ "systemd-run", "--user", "--quiet", "--on-active=5s", "--unit", "voicebox-mcp-gateway-refresh", "systemctl", "--user", "restart", GATEWAY_UNIT, ], timeout=30) event["gateway_restart_exit"] = restart.returncode if restart.returncode == 0: return "scheduled" event["gateway_restart_error"] = (restart.stderr or restart.stdout).strip()[-500:] return "failed" except subprocess.TimeoutExpired: event["gateway_restart_exit"] = "timeout" return "timeout" def load_state() -> dict: try: return json.loads(STATE_PATH.read_text()) except Exception: return {} def save_state(state: dict) -> None: STATE_PATH.parent.mkdir(parents=True, exist_ok=True) tmp = STATE_PATH.with_suffix(".tmp") tmp.write_text(json.dumps(state, indent=2, sort_keys=True)) tmp.replace(STATE_PATH) def log(event: dict) -> None: LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with LOG_PATH.open("a", encoding="utf-8") as handle: handle.write(json.dumps(event, sort_keys=True) + "\n") def main() -> int: now = datetime.now(timezone.utc).isoformat() state = load_state() prev_status = state.get("status") ok, probe_exit, output = probe_voicebox() recovered_by_watchdog = False repair_steps: list[dict] = [] if not ok: tunnel = restart_tunnel() repair_steps.append({"step": "restart_tunnel", "exit": tunnel.returncode, "tail": (tunnel.stderr or tunnel.stdout).strip()[-300:]}) ok, probe_exit, output = probe_voicebox() if not ok: mac = ensure_voicebox_app_running() repair_steps.append({"step": "ensure_voicebox_app", "exit": mac.returncode, "tail": (mac.stderr or mac.stdout).strip()[-300:]}) if mac.returncode == 0: tunnel = restart_tunnel() repair_steps.append({"step": "restart_tunnel_after_mac", "exit": tunnel.returncode, "tail": (tunnel.stderr or tunnel.stdout).strip()[-300:]}) ok, probe_exit, output = probe_voicebox() recovered_by_watchdog = ok status = "up" if ok else "down" event = { "at": now, "status": status, "previous_status": prev_status, "probe_exit": probe_exit, } if repair_steps: event["repair_steps"] = repair_steps message = "" if status != prev_status: event["transition"] = f"{prev_status or 'unknown'}->{status}" if ok: if prev_status == "down" or recovered_by_watchdog: restart_status = schedule_gateway_restart(event) if restart_status == "scheduled": message = "Voicebox MCP recovered; scheduled a default Hermes gateway restart so Voicebox tools are rediscovered." else: message = "Voicebox MCP recovered, but restarting the default Hermes gateway failed. Check voicebox_mcp_watchdog.jsonl." else: message = "Voicebox MCP is reachable and healthy." else: message = "Voicebox MCP is currently unreachable even after tunnel/app repair attempts." event["probe_tail"] = output[-500:] log(event) else: # Quiet steady state, but keep lightweight state freshness. If the # watchdog had to repair a transient failure, alert and refresh the # gateway because the running native MCP client may have given up. if recovered_by_watchdog: restart_status = schedule_gateway_restart(event) if restart_status == "scheduled": message = "Voicebox MCP had failed but recovered after repair; scheduled a Hermes gateway refresh." else: message = "Voicebox MCP recovered after repair, but gateway refresh failed. Check voicebox_mcp_watchdog.jsonl." if not ok: event["probe_tail"] = output[-500:] log(event) state.update({"status": status, "checked_at": now, "last_probe_exit": probe_exit}) save_state(state) if message: print(message) return 0 if __name__ == "__main__": raise SystemExit(main())