""" Reyna family ESP32 Voice Gateway - durable Hermes plugin Survives `hermes update` which wipes in-tree patches to gateway/platforms/api_server.py. What it does: - On Hermes import (plugin load), checks if api_server.py has ESP32 routes. - If missing (after update), auto-restores from bundled patch + logs actionable warning. - Exposes tools for manual verify/restore from inside Hermes chat (via slash or tool). The actual voice logic stays in api_server.py (patched file) because it's 1600+ lines deeply integrated with APIServerAdapter internals (ResponseStore, _run_agent, TTS, transcription, screen MCP). Extracting as runtime monkeypatch would be brittle. The durable strategy is: 1. Patch file lives in this repo + bundled in plugin as `patch/*.patch` 2. Plugin's register() does self-heal check (fast ~10ms) 3. Gateway auto-restarts handled by operator (can't restart self from inside gateway - see skill pitfalls) Future: true runtime injection plugin that patches APIServerAdapter.connect to add routes without touching file — feasible but needs tracking _handle_esp32_* methods. Kept separate as v2. """ import logging import subprocess import sys from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) PLUGIN_DIR = Path(__file__).parent PATCH_DIR = PLUGIN_DIR / "patch" # repo root for dev checkout (symlinked plugin) REPO_PATCH = PLUGIN_DIR.parent / "patch" / "api_server_esp32_live_draft_with_llm.patch" def _find_hermes_agent_dir() -> Optional[Path]: # Try standard locations candidates = [ Path.home() / ".hermes" / "hermes-agent", Path(__file__).resolve().parents[3], # if running from within hermes-agent/plugins/esp32-... ] # Also check HERMES_AGENT_DIR env import os env_dir = os.getenv("HERMES_AGENT_DIR") if env_dir: candidates.insert(0, Path(env_dir)) # pip install location try: import hermes_cli candidates.append(Path(hermes_cli.__file__).parents[1]) except Exception: pass for c in candidates: if (c / "gateway" / "platforms" / "api_server.py").exists(): return c return None def _api_server_has_esp32() -> bool: agent_dir = _find_hermes_agent_dir() if not agent_dir: return False api_file = agent_dir / "gateway" / "platforms" / "api_server.py" try: txt = api_file.read_text()[:5000] # top matter + first routes # quick check if "ESP32_AUDIO_MAX_BYTES" not in api_file.read_text()[:20000]: return False # full route registration present? content = api_file.read_text() return "/api/esp32/voice" in content and "_handle_esp32_voice_ws" in content except Exception: return False def _find_patch_file() -> Optional[Path]: # Search order: bundled patch in plugin, repo patch, hermes-home fallback for p in [ PATCH_DIR / "api_server_esp32_live_draft_with_llm.patch", REPO_PATCH, Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch", Path.home() / ".hermes" / "esp32_voice_patch.patch", PATCH_DIR / "esp32_live_draft_with_llm_patch.patch", ]: if p.exists(): return p # Also search inside plugin dir itself (dev checkout) for pattern in ["*.patch"]: for f in PATCH_DIR.glob(pattern): if f.stat().st_size > 1000: return f return None def _try_apply_patch(patch_path: Path, agent_dir: Path) -> bool: """Apply patch via git apply. Returns True if applied or already applied.""" api_file = agent_dir / "gateway" / "platforms" / "api_server.py" if not api_file.exists(): logger.warning("[esp32-voice-gateway] api_server.py not found at %s", api_file) return False # Check if git git_dir = agent_dir / ".git" if not git_dir.exists(): logger.warning("[esp32-voice-gateway] not a git repo: %s", agent_dir) return False # Try git apply --check first try: # if already patched, check will fail with "already applied" type – we detect via has_esp32 already res = subprocess.run( ["git", "apply", "--check", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=15, ) if res.returncode != 0: # Could be already applied or conflict. Log but try to see if content already has marker if "ESP32_AUDIO_MAX_BYTES" not in api_file.read_text()[:20000]: logger.warning("[esp32-voice-gateway] patch check failed: %s %s", res.stdout[:500], res.stderr[:800]) # Try --3way if possible res2 = subprocess.run( ["git", "apply", "--3way", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=30, ) if res2.returncode != 0: logger.error("[esp32-voice-gateway] git apply --3way failed: %s %s", res2.stdout[-500:], res2.stderr[-1000:]) return False else: # already has marker, consider applied logger.info("[esp32-voice-gateway] patch already present (has ESP32 marker)") return True else: res_apply = subprocess.run( ["git", "apply", str(patch_path)], cwd=str(agent_dir), capture_output=True, text=True, timeout=30, ) if res_apply.returncode != 0: logger.error("[esp32-voice-gateway] git apply failed: %s %s", res_apply.stdout[-500:], res_apply.stderr[-1000:]) return False # Verify if _api_server_has_esp32(): logger.info("[esp32-voice-gateway] Successfully restored ESP32 routes from %s -> %s", patch_path, api_file) return True else: logger.warning("[esp32-voice-gateway] Patch applied but route marker still missing – need manual check") return False except Exception as e: logger.exception("[esp32-voice-gateway] Exception applying patch %s: %s", patch_path, e) return False def restore_if_needed() -> dict: """Check and auto-restore if needed. Safe to call multiple times.""" result = {"had_routes": False, "restored": False, "patch": None, "agent_dir": None, "error": None} try: agent_dir = _find_hermes_agent_dir() result["agent_dir"] = str(agent_dir) if agent_dir else None if not agent_dir: result["error"] = "hermes-agent dir not found" return result if _api_server_has_esp32(): result["had_routes"] = True return result # missing -> try restore patch_file = _find_patch_file() result["patch"] = str(patch_file) if patch_file else None if not patch_file: result["error"] = "no patch file found (checked patch/*.patch + ~/.hermes/esp32*.patch)" logger.error("[esp32-voice-gateway] Routes missing after update and no patch file found!") return result logger.warning("[esp32-voice-gateway] ESP32 routes missing (likely after `hermes update`) – auto-restoring from %s", patch_file) ok = _try_apply_patch(patch_file, agent_dir) result["restored"] = ok if ok: # also copy patch to ~/.hermes for next fallback try: dest = Path.home() / ".hermes" / "esp32_live_draft_with_llm_patch.patch" if patch_file.resolve() != dest.resolve(): import shutil shutil.copy2(patch_file, dest) except Exception: pass return result except Exception as e: result["error"] = str(e) logger.exception("[esp32-voice-gateway] restore_if_needed failed") return result # --- Plugin registration --- def register(ctx): """Hermes plugin entry point.""" # On import, do fast self-heal check (10ms if routes present) info = restore_if_needed() # Register a tool so user can ask inside Hermes: "check ESP32 voice gateway" import json def _tool_check_esp32_gateway(args, **kwargs) -> str: res = restore_if_needed() # also run live curl checks if gateway running checks = {} try: import socket # try to find api_server port from config port = 8642 try: from hermes_cli.config import load_config cfg = load_config() g_port = ((cfg.get("gateway") or {}).get("api_server") or {}).get("port") p_port = ((cfg.get("platforms") or {}).get("api_server") or {}).get("port") raw_port = g_port or p_port or port port = int(raw_port) except Exception: pass # quick TCP check s = socket.socket() s.settimeout(1.0) try: s.connect(("127.0.0.1", port)) checks["tcp"] = f"127.0.0.1:{port} open" except Exception as e: checks["tcp"] = f"closed: {e}" finally: s.close() except Exception as e: checks["tcp_error"] = str(e) return json.dumps({ "had_routes_before": info.get("had_routes"), "restored": info.get("restored"), "patch_used": info.get("patch"), "agent_dir": info.get("agent_dir"), "error": info.get("error"), "has_routes_now": _api_server_has_esp32(), "checks": checks, "verify_curl": f'curl -H "Authorization: Bearer $API_SERVER_KEY" http://127.0.0.1:8642/api/esp32/voice/ws -v 2>&1 | grep "< HTTP" -> want 400 not 404' }, indent=2) def _tool_restore_esp32_gateway(args, **kwargs) -> str: res = restore_if_needed() # force reapply even if has routes if requested force = bool(args.get("force")) if force: patch = _find_patch_file() agent_dir = _find_hermes_agent_dir() if patch and agent_dir: # reset file first then reapply? just apply ok = _try_apply_patch(patch, agent_dir) res["force_restored"] = ok return json.dumps(res, indent=2) ctx.register_tool( name="esp32_voice_gateway_status", toolset="esp32_voice", schema={ "name": "esp32_voice_gateway_status", "description": "Check if ESP32 voice gateway routes (POST /api/esp32/voice + WS /api/esp32/voice/ws) are present after hermes update. Auto-restores from bundled patch if missing.", "parameters": {"type": "object", "properties": {}, "required": []}, }, handler=_tool_check_esp32_gateway, description="Check ESP32 voice gateway route status and auto-restore after update", ) ctx.register_tool( name="esp32_voice_gateway_restore", toolset="esp32_voice", schema={ "name": "esp32_voice_gateway_restore", "description": "Force restore ESP32 voice gateway routes from bundled patch after hermes update wipes api_server.py", "parameters": { "type": "object", "properties": { "force": {"type": "boolean", "description": "Force re-apply even if routes seem present"}, }, "required": [], }, }, handler=_tool_restore_esp32_gateway, description="Force restore ESP32 routes from patch", ) def _on_startup(session_id=None, **kw): # Gateway hook – log status on startup for visibility try: if _api_server_has_esp32(): logger.info("[esp32-voice-gateway] Routes OK: /api/esp32/voice + /api/esp32/voice/ws present") else: logger.warning("[esp32-voice-gateway] Routes MISSING at startup – will try restore on next tool call; if you just ran `hermes update`, run: ~/Projects/hermes-esp32-voice-gateway/scripts/restore_after_update.py && hermes gateway restart") except Exception: pass try: ctx.register_hook("on_session_start", _on_startup) except Exception: pass logger.info("[esp32-voice-gateway] Plugin loaded (had_routes=%s restored=%s has_now=%s)", info.get("had_routes"), info.get("restored"), _api_server_has_esp32())