#!/usr/bin/env python3 """ Gateway watchdog — silent check, restart if dead. Cron contract: prints NOTHING when the gateway is healthy, prints a one-line note when it had to restart something. With no_agent=true cron delivery, empty stdout = no Rob ping. Non-empty = Rob gets a heads-up that the gateway died. Also writes PID-change events to ~/AppData/Local/hermes/logs/gateway_events.jsonl so a downstream cron (provider_watchdog) can alert on model/provider changes that happen inside the gateway. Design: - 2-minute startup grace period after manual start (prevents thrash on legit restarts). - Uses both PID file and process check to avoid acting on a stale PID. - Calls `hermes gateway start` via subprocess. Returns silently if start fails (logs to stderr only, not stdout). """ import json import os import subprocess import sys import time from pathlib import Path PID_FILE = Path.home() / "AppData" / "Local" / "hermes" / "gateway.pid" STATE_FILE = Path.home() / "AppData" / "Local" / "hermes" / "gateway_state.json" EVENTS_FILE = Path.home() / "AppData" / "Local" / "hermes" / "logs" / "gateway_events.jsonl" GRACE_SECONDS = 120 def read_pid() -> int | None: """Read PID from gateway.pid (JSON format with 'pid' key).""" if not PID_FILE.exists(): return None try: data = json.loads(PID_FILE.read_text(encoding="utf-8")) pid = data.get("pid") return int(pid) if pid else None except (json.JSONDecodeError, ValueError, OSError): return None def pid_alive(pid: int) -> bool: """Check if a Windows PID is still running.""" if pid is None: return False try: # tasklist with /FI is the portable way; /NH strips the header. result = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}", "/NH"], capture_output=True, text=True, timeout=10, ) return str(pid) in result.stdout except (subprocess.TimeoutExpired, OSError): return False def state_running() -> bool: """Read gateway_state.json to confirm state == 'running'.""" if not STATE_FILE.exists(): return False try: data = json.loads(STATE_FILE.read_text(encoding="utf-8")) return data.get("gateway_state") == "running" except (json.JSONDecodeError, ValueError, OSError): return False def state_age_seconds() -> float: """How old is the gateway_state.json updated_at? -1 if unreadable. Diagnostic only — not used as a health signal because the gateway doesn't heartbeat the state file when platforms are idle. """ if not STATE_FILE.exists(): return -1.0 try: mtime = STATE_FILE.stat().st_mtime return time.time() - mtime except OSError: return -1.0 def is_healthy() -> bool: """Gateway is healthy if: pid file present, PID alive, state file says running. We deliberately do NOT use state-file mtime as a health signal — the gateway only updates state on platform events, so it goes stale whenever the bots are idle (which is most of the time). Liveness of the PID is the real signal. """ pid = read_pid() if not pid_alive(pid): return False if not state_running(): return False return True def start_gateway() -> bool: """Spawn a fresh gateway process. Returns True if it appears to have started. Tries `hermes gateway start` first (clean path). If that's blocked (because the watchdog is running inside an existing gateway session), falls back to direct pythonw.exe spawn with the same command line the gateway uses for itself. """ # Path 1: clean CLI try: result = subprocess.run( ["hermes", "gateway", "start"], capture_output=True, text=True, timeout=30, ) if result.returncode == 0 and read_pid() and pid_alive(read_pid()): return True except (subprocess.TimeoutExpired, OSError): pass # Path 2: direct spawn — replicates the gateway's own startup. # The state file from the last successful run tells us the exact argv. try: if STATE_FILE.exists(): data = json.loads(STATE_FILE.read_text(encoding="utf-8")) argv = data.get("argv") if argv and len(argv) >= 2: # pythonw.exe is the Python launcher; main.py is the entry. # Use the same Python the hermes CLI uses. pythonw = ( Path.home() / "AppData" / "Local" / "hermes" / "hermes-agent" / "venv" / "Scripts" / "pythonw.exe" ) if not pythonw.exists(): # Fallback to venv python.exe pythonw = pythonw.with_name("python.exe") subprocess.Popen( [str(pythonw)] + list(argv[1:]), creationflags=getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0), close_fds=True, ) # Give it a moment to write its pid file for _ in range(10): time.sleep(0.5) if read_pid() and pid_alive(read_pid()): return True except (OSError, json.JSONDecodeError) as e: print(f"direct spawn failed: {e}", file=sys.stderr) return False def main() -> int: new_pid = read_pid() if is_healthy(): # Detect PID change vs last run — write a log event so provider_watchdog # can alert if the gateway was restarted externally (by Rob or by the OS). _detect_pid_change(new_pid) return 0 # silent — healthy # Don't thrash: if the PID file exists and points to a dead process that's # less than GRACE_SECONDS old, give it time to come up. pid = read_pid() if pid and pid_alive(pid) is False and state_age_seconds() < GRACE_SECONDS: return 0 # silent — within grace period if start_gateway(): new_pid = read_pid() if new_pid and pid_alive(new_pid): _log_event("restart", f"Gateway was down — restarted (new PID {new_pid})") print(f"Gateway was down — restarted (new PID {new_pid})") else: _log_event("restart_unconfirmed", "Gateway was down — restart invoked but PID not confirmed") print("Gateway was down — restart invoked but PID not confirmed") return 0 else: _log_event("restart_failed", "Gateway down and restart failed") print("Gateway down and restart failed", file=sys.stderr) return 1 _LAST_PID_FILE = Path.home() / "AppData" / "Local" / "hermes" / "gateway_watchdog.last_pid" def _detect_pid_change(current_pid: int | None) -> None: """Track PID transitions across watchdog runs. On first run, records the current PID. On subsequent runs, if the PID changed (gateway was restarted by Rob, by the OS, or by something else), log a `pid_change` event. This catches restarts the watchdog itself didn't perform. """ if current_pid is None: return try: if _LAST_PID_FILE.exists(): last_pid = int(_LAST_PID_FILE.read_text().strip() or "0") else: last_pid = 0 if last_pid and last_pid != current_pid: _log_event( "pid_change", f"Gateway PID changed: {last_pid} → {current_pid} (restarted externally)", ) _LAST_PID_FILE.write_text(str(current_pid), encoding="utf-8") except (ValueError, OSError): pass def _log_event(event_type: str, message: str) -> None: """Append a JSON line to gateway_events.jsonl for downstream watchdogs.""" try: EVENTS_FILE.parent.mkdir(parents=True, exist_ok=True) with EVENTS_FILE.open("a", encoding="utf-8") as f: f.write( json.dumps( { "ts": time.time(), "iso": time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime()), "type": event_type, "message": message, } ) + "\n" ) except OSError: pass # event logging is best-effort; never break the watchdog itself if __name__ == "__main__": sys.exit(main())