google_meet

1"""Remote node server — hosts the Meet bot on another machine (``hermes meet node run``).2 3WebSocket endpoint accepting token-signed RPC requests dispatched to ``process_manager``.4Token: 32 hex chars minted on first boot, persisted at ``$HERMES_HOME/workspace/meetings/5node_token.json`` so approved gateways survive restarts; the operator copies it to the gateway6via ``hermes meet node approve <name> <url> <token>``. ``websockets`` is imported lazily.7"""8 9from __future__ import annotations10 11import asyncio12import contextlib13import json14import secrets15import time16from pathlib import Path17from typing import Any, Dict, Optional18 19from hermes_constants import get_hermes_home20from plugins.google_meet._jsonfile import read_json21from utils import atomic_json_write22from plugins.google_meet.node import protocol as _proto23 24_START_BOT_KEYS = ("url", "guest_name", "duration", "headed", "auth_state", "session_id", "out_dir")25 26 27class _RpcError(Exception):28    """Handler-level protocol error; sent verbatim as an error envelope."""29 30 31def _rpc_start_bot(payload: Dict[str, Any], pm) -> Dict[str, Any]:32    # Whitelist kwargs we pass through to pm.start.33    kwargs = {k: payload[k] for k in _START_BOT_KEYS if k in payload}34    if "url" not in kwargs:35        raise _RpcError("missing 'url' in payload")36    return pm.start(**kwargs)37 38 39def _rpc_say(payload: Dict[str, Any], pm) -> Dict[str, Any]:40    # The bot-side consumer only exists in realtime mode: ok=True means "enqueued", not "spoken".41    text = payload.get("text", "")42    active = pm._read_active()43    enqueued = False44    if active and active.get("out_dir"):45        with contextlib.suppress(OSError):46            queue = Path(active["out_dir"]) / "say_queue.jsonl"47            queue.parent.mkdir(parents=True, exist_ok=True)48            with queue.open("a", encoding="utf-8") as fh:49                fh.write(json.dumps({"text": text, "ts": time.time()}) + "\n")50            enqueued = True51    return {"ok": True, "enqueued": enqueued, "text": text}52 53 54# request type → fn(payload, pm) returning the response payload.55_RPC = {56    "start_bot": _rpc_start_bot,57    "stop": lambda p, pm: pm.stop(reason=p.get("reason", "requested")),58    "status": lambda p, pm: pm.status(),59    "transcript": lambda p, pm: pm.transcript(last=p.get("last")),60    "say": _rpc_say}61 62 63class NodeServer:64    """WebSocket server that executes meet bot RPCs locally."""65 66    def __init__(self, host: str = "127.0.0.1", port: int = 18789, token_path: Optional[Path] = None,67                 display_name: str = "hermes-meet-node") -> None:68        self.host = host69        self.port = port70        self.display_name = display_name71        self.token_path = Path(token_path) if token_path is not None else (72            Path(get_hermes_home()) / "workspace" / "meetings" / "node_token.json")73        self._token: Optional[str] = None74 75    def ensure_token(self) -> str:76        """Return the persisted shared secret, generating one on first use."""77        if self._token:78            return self._token79        data = read_json(self.token_path)80        tok = data.get("token") if isinstance(data, dict) else None81        if not (isinstance(tok, str) and tok):82            tok = secrets.token_hex(16)  # 32 hex chars83            # Owner-only: the token grants full RPC access to the meet bot.84            atomic_json_write(self.token_path, {"token": tok, "generated_at": time.time()}, mode=0o600)85        self._token = tok86        return tok87 88    async def _handle_request(self, msg: Dict[str, Any]) -> Dict[str, Any]:89        """Validate + dispatch one decoded request; always returns an envelope, never raises.90        Envelope ``error`` is for auth/protocol failures and pm crashes; pm's own ``ok``/``error``91        results travel inside a normal response payload."""92        ok, reason = _proto.validate_request(msg, self.ensure_token())93        if not ok:94            return _proto.make_error(str(msg.get("id") or ""), reason)95        req_id, t = msg["id"], msg["type"]96        if t == "ping":97            return {"type": "pong", "id": req_id,98                    "payload": {"display_name": self.display_name, "ts": time.time()}}99        handler = _RPC.get(t)100        if handler is None:101            return _proto.make_error(req_id, f"unhandled type: {t!r}")102        # Import lazily so test mocks can monkeypatch freely.103        from plugins.google_meet import process_manager as pm104        try:105            return _proto.make_response(req_id, handler(msg["payload"], pm))106        except _RpcError as exc:107            return _proto.make_error(req_id, str(exc))108        except Exception as exc:  # noqa: BLE001 — surface any pm crash to client109            return _proto.make_error(req_id, f"{type(exc).__name__}: {exc}")110 111    async def serve(self) -> None:112        """Run the WebSocket server until cancelled (wrap in ``asyncio.run``)."""113        try:114            import websockets  # type: ignore115        except ImportError as exc:116            raise RuntimeError("NodeServer.serve requires the 'websockets' package. "117                               "Install it with: pip install websockets") from exc118        self.ensure_token()119 120        async def _handler(ws):121            async for raw in ws:122                try:123                    msg = _proto.decode(raw)124                except ValueError as exc:125                    await ws.send(_proto.encode(_proto.make_error("", f"decode: {exc}")))126                    continue127                await ws.send(_proto.encode(await self._handle_request(msg)))128 129        async with websockets.serve(_handler, self.host, self.port):130            await asyncio.Future()  # run until cancelled131