diff --git a/cmd_chat/operator/__init__.py b/cmd_chat/operator/__init__.py new file mode 100644 index 0000000..a059b45 --- /dev/null +++ b/cmd_chat/operator/__init__.py @@ -0,0 +1,11 @@ +"""Operator bridge — a Claude Code session drives a hack-house room as a client. + +The room becomes an API: a headless daemon (`OperatorBridge`) owns the +websocket, and the `hh-bridge` CLI (`python -m cmd_chat.operator`) reads/sends +through a local unix socket. See `bridge.py` for the protocol reuse map. +""" + +from .bridge import OperatorBridge +from .session import Session + +__all__ = ["OperatorBridge", "Session"] diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py new file mode 100644 index 0000000..f5f4cfe --- /dev/null +++ b/cmd_chat/operator/__main__.py @@ -0,0 +1,217 @@ +"""CLI for the operator bridge — `python -m cmd_chat.operator `. + +Verbs +----- + serve HOST PORT USER run the daemon (normally spawned by `up`, not by hand) + up HOST PORT USER start the daemon detached and wait for it to connect + read [--wait] print new inbox events as JSONL (cursor-tracked) + say TEXT… send a chat line to the room + roster list who's in the room + status daemon + connection state + down stop the daemon, leave the room + +Phase 1: join + read + say. The reusable skill wrapper comes in a later phase; +for now a Claude Code session can drive everything straight from Bash. +""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +from pathlib import Path + +from .cli_client import BridgeUnreachable, request +from .session import Session, resolve + + +def _add_conn_args(p: argparse.ArgumentParser) -> None: + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("user", help="room display name to join as (also the session id)") + p.add_argument("--password", default=None) + p.add_argument("--no-tls", action="store_true", help="plain ws/http") + p.add_argument("--insecure", action="store_true", help="skip TLS verify") + p.add_argument("--session", default=None, help="session id (default: USER)") + p.add_argument("--trigger", default=None, + help="extra phrase that marks a message 'addressed' to us " + "(the operator name / @name always count)") + + +def _build_parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(prog="cmd_chat.operator") + sub = ap.add_subparsers(dest="verb", required=True) + + _add_conn_args(sub.add_parser("serve", help="run the daemon (foreground)")) + _add_conn_args(sub.add_parser("up", help="start the daemon detached")) + + r = sub.add_parser("read", help="print new inbox events") + r.add_argument("--session", default=None) + r.add_argument("--since", type=int, default=None, + help="seq to read after (default: saved cursor)") + r.add_argument("--wait", action="store_true", help="long-poll until an event arrives") + r.add_argument("--timeout", type=float, default=30.0) + r.add_argument("--all", action="store_true", help="ignore cursor, dump full ring") + + s = sub.add_parser("say", help="send a chat line") + s.add_argument("text", nargs="+") + s.add_argument("--session", default=None) + + for v in ("roster", "status", "down"): + sub.add_parser(v).add_argument("--session", default=None) + return ap + + +# ── daemon ─────────────────────────────────────────────────────────────── +def _run_serve(args) -> int: + from .bridge import OperatorBridge # heavy imports only on the daemon path + bridge = OperatorBridge( + args.host, args.port, name=args.user, password=args.password, + insecure=args.insecure, no_tls=args.no_tls, + session=args.session, trigger=args.trigger) + try: + bridge.run() + except KeyboardInterrupt: + pass + return 0 + + +def _run_up(args) -> int: + import subprocess + + sess = Session(args.session or args.user) + sess.ensure_dir() + + # Already alive? + if sess.sock_path.exists(): + try: + st = request(sess.sock_path, {"op": "status"}, read_timeout=5) + if st.get("connected"): + print(f"session '{sess.name}' already up (connected)") + return 0 + except BridgeUnreachable: + sess.cleanup() # stale socket → fall through and respawn + + sess.write_meta(host=args.host, port=args.port, user=args.user, + trigger=args.trigger or "", no_tls=args.no_tls, + started=time.time()) + + cmd = [sys.executable, "-m", "cmd_chat.operator", "serve", + args.host, str(args.port), args.user] + if args.password is not None: + cmd += ["--password", args.password] + if args.no_tls: + cmd.append("--no-tls") + if args.insecure: + cmd.append("--insecure") + if args.session: + cmd += ["--session", args.session] + if args.trigger: + cmd += ["--trigger", args.trigger] + + log = open(sess.log_path, "a") # noqa: SIM115 — handed to the child + proc = subprocess.Popen(cmd, stdout=log, stderr=log, stdin=subprocess.DEVNULL, + start_new_session=True, close_fds=True) + + # Wait for the daemon to connect (or report a fatal-looking error). + deadline = time.time() + 15.0 + last_err = None + while time.time() < deadline: + if proc.poll() is not None: + print(f"daemon exited early (code {proc.returncode}); see {sess.log_path}", + file=sys.stderr) + return 1 + try: + st = request(sess.sock_path, {"op": "status"}, read_timeout=3) + except BridgeUnreachable: + time.sleep(0.3) + continue + if st.get("connected"): + print(f"session '{sess.name}' up — connected to {args.host}:{args.port} " + f"as '{args.user}' (pid {proc.pid})") + return 0 + # Connected socket but not in-room yet: peek for an error event. + try: + rd = request(sess.sock_path, {"op": "read", "since": 0}, read_timeout=3) + errs = [e for e in rd.get("events", []) if e.get("event") == "error"] + if errs: + last_err = errs[-1].get("reason") + except BridgeUnreachable: + pass + time.sleep(0.4) + + msg = f"session '{sess.name}' started (pid {proc.pid}) but not connected yet" + if last_err: + msg += f" — last error: {last_err}" + print(msg + f"; tail {sess.log_path}", file=sys.stderr) + return 1 + + +# ── client verbs ───────────────────────────────────────────────────────── +def _client_request(args, obj: dict, read_timeout: float = 35.0) -> dict: + sess = resolve(getattr(args, "session", None)) + if not sess.sock_path.exists(): + print(f"no live bridge for session '{sess.name}' " + f"(run `up` first)", file=sys.stderr) + sys.exit(2) + try: + return request(sess.sock_path, obj, read_timeout=read_timeout) + except BridgeUnreachable as e: + print(str(e), file=sys.stderr) + sys.exit(2) + + +def _run_read(args) -> int: + sess = resolve(getattr(args, "session", None)) + if args.all: + since = 0 + elif args.since is not None: + since = args.since + else: + since = sess.read_cursor() + timeout = args.timeout + resp = _client_request(args, {"op": "read", "since": since, + "wait": args.wait, "timeout": timeout}, + read_timeout=timeout + 5.0) + events = resp.get("events", []) + for ev in events: + print(json.dumps(ev)) + if events and not args.all and args.since is None: + sess.write_cursor(events[-1]["seq"]) + return 0 + + +def _run_say(args) -> int: + text = " ".join(args.text) + resp = _client_request(args, {"op": "say", "text": text}) + if not resp.get("ok"): + print(resp.get("error", "say failed"), file=sys.stderr) + return 1 + return 0 + + +def _run_simple(args, op: str) -> int: + resp = _client_request(args, {"op": op}) + print(json.dumps(resp)) + return 0 if resp.get("ok") else 1 + + +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + verb = args.verb + if verb == "serve": + return _run_serve(args) + if verb == "up": + return _run_up(args) + if verb == "read": + return _run_read(args) + if verb == "say": + return _run_say(args) + if verb in ("roster", "status", "down"): + return _run_simple(args, verb) + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/cmd_chat/operator/bridge.py b/cmd_chat/operator/bridge.py new file mode 100644 index 0000000..05d10bc --- /dev/null +++ b/cmd_chat/operator/bridge.py @@ -0,0 +1,324 @@ +"""OperatorBridge — a headless room client a Claude Code session drives. + +Inverts the `cmd_chat/agent` model: instead of the room calling a model per +message, this daemon owns the websocket and exposes the room as a local API. +A Claude Code session pumps it through the `hh-bridge` CLI (`read`/`say`/…), +which talks to a unix control socket served in this same event loop. + +Phase 1 scope: join + roster + read (with long-poll) + say. Sandbox drive, +delegation and nesting are later phases — `_perm:acl`/`_sbx:status` frames are +*recorded* into the inbox for awareness but not acted on yet. + +Reuses the protocol wholesale from `Client` (SRP, room key, encrypt/decrypt) +and mirrors `AgentBridge`'s reconnect/serve shape (cmd_chat/agent/bridge.py). +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import os +import re +import time + +import websockets + +from ..client.client import Client +from .session import Session + + +class OperatorBridge(Client): + _PING_INTERVAL = 20.0 + _PING_TIMEOUT = 60.0 + _RECONNECT_MAX_BACKOFF = 30.0 + _RECONNECT_STABLE_SECS = 30.0 + MAX_EVENTS = 2000 # in-RAM inbox ring; full history lives in inbox.jsonl + + def __init__(self, server: str, port: int, name: str, + password: str | None = None, insecure: bool = False, + no_tls: bool = False, session: str | None = None, + trigger: str | None = None): + super().__init__(server, port, username=name, password=password, + insecure=insecure, no_tls=no_tls) + self.name = name + self.trigger = (trigger or "").strip() + self.session = Session(session or name) + + # Inbox: monotonically-seq'd events + an async condition to wake waiters. + self.events: list[dict] = [] + self.seq = 0 + self.cond = asyncio.Condition() + + # Mirrored room awareness (recorded only in Phase 1). + self.granted = False + self.can_sudo = False + self.sbx_engine: str | None = None + self.sbx_name: str = "" + + # Live websocket (None while reconnecting); set in _connect_and_serve. + self._ws = None + self._stop = asyncio.Event() + self._seeded = False # backfill room history only on first init + self._inbox_fp = None # append handle to inbox.jsonl + self._addr_re = self._build_addr_re() + + # ── addressing ─────────────────────────────────────────────────────── + def _build_addr_re(self) -> re.Pattern: + toks = [re.escape(self.name), re.escape("@" + self.name)] + if self.trigger: + toks.append(re.escape(self.trigger)) + return re.compile(r"(? bool: + return bool(self._addr_re.search(text or "")) + + # ── inbox ──────────────────────────────────────────────────────────── + async def _emit(self, kind: str, **fields) -> None: + async with self.cond: + self.seq += 1 + ev = {"seq": self.seq, "ts": round(time.time(), 3), "kind": kind, **fields} + self.events.append(ev) + if len(self.events) > self.MAX_EVENTS: + self.events = self.events[-self.MAX_EVENTS:] + if self._inbox_fp is not None: + try: + self._inbox_fp.write(json.dumps(ev) + "\n") + self._inbox_fp.flush() + except OSError: + pass + self.cond.notify_all() + + # ── frame handling ─────────────────────────────────────────────────── + async def _handle_frame(self, ws, raw) -> None: + try: + data = json.loads(raw) + except json.JSONDecodeError: + return + mtype = data.get("type") + if mtype == "init": + self.users = data.get("users", []) + await self._emit("roster", users=self._roster()) + if not self._seeded: + for m in data.get("messages", []): + dec = self.decrypt_message(dict(m)) + text = dec.get("text", "") + sender = dec.get("username", "?") + if (not text or text == "[decrypt failed]" + or text.startswith('{"_') or sender == self.name): + continue + await self._emit("message", **{"from": sender}, text=text, + addressed=False, backfill=True) + self._seeded = True + return + if mtype == "roster": + self.users = data.get("users", []) + await self._emit("roster", users=self._roster()) + return + if mtype == "user_left": + left = data.get("user_id") + self.users = [u for u in self.users if u.get("user_id") != left] + await self._emit("roster", users=self._roster()) + return + if mtype != "message": + return + msg = self.decrypt_message(data.get("data", {})) + text = msg.get("text", "") + sender = msg.get("username", "?") + if sender == self.name: + return + if text.startswith('{"_'): + await self._record_control(text) + return + await self._emit("message", **{"from": sender}, text=text, + addressed=self._is_addressed(text)) + + async def _record_control(self, text: str) -> None: + """Surface ACL + sandbox-status changes into the inbox (awareness only). + Ignore high-volume / irrelevant control frames (_sbx:data, _ai, _ft).""" + try: + frame = json.loads(text) + except json.JSONDecodeError: + return + if frame.get("_sbx") == "status": + ready = frame.get("state") == "ready" + self.sbx_engine = frame.get("engine") if ready else None + self.sbx_name = (frame.get("name") or "") if ready else "" + await self._emit("sandbox", state=frame.get("state"), + engine=self.sbx_engine, name=self.sbx_name, + backend=frame.get("backend")) + elif frame.get("_perm") == "acl": + self.granted = self.name in frame.get("drivers", []) + self.can_sudo = self.name in frame.get("sudoers", []) + await self._emit("acl", owner=frame.get("owner"), + granted=self.granted, can_sudo=self.can_sudo, + drivers=frame.get("drivers", [])) + + def _roster(self) -> list[str]: + return [u.get("username", "?") for u in self.users] + + # ── connection lifecycle (mirrors AgentBridge) ─────────────────────── + async def run_async(self) -> None: + self.session.ensure_dir() + self.session.cleanup() # drop a stale socket from a prior run + self._inbox_fp = open(self.session.inbox_path, "a") # noqa: SIM115 + server = await asyncio.start_unix_server( + self._serve_control, path=str(self.session.sock_path)) + try: + os.chmod(self.session.sock_path, 0o600) + except OSError: + pass + await self._emit("system", event="starting", + host=self.server, port=self.port, user=self.name) + try: + await self._reconnect_loop() + finally: + server.close() + try: + await server.wait_closed() + except Exception: # noqa: BLE001 + pass + if self._inbox_fp is not None: + self._inbox_fp.close() + self.session.cleanup() + + async def _reconnect_loop(self) -> None: + backoff = 1.0 + first = True + while not self._stop.is_set(): + loop = asyncio.get_running_loop() + started = loop.time() + try: + await self._connect_and_serve(reconnect=not first) + except (KeyboardInterrupt, asyncio.CancelledError): + raise + except (websockets.ConnectionClosed, OSError) as e: + await self._emit("system", event="reconnecting", + reason=type(e).__name__) + except Exception as e: # noqa: BLE001 — auth/transient: report + retry + await self._emit("system", event="error", + reason=f"{type(e).__name__}: {e}") + else: + if not self._stop.is_set(): + await self._emit("system", event="reconnecting", + reason="closed") + if self._stop.is_set(): + break + if loop.time() - started >= self._RECONNECT_STABLE_SECS: + backoff = 1.0 + first = False + try: + await asyncio.wait_for(self._stop.wait(), timeout=backoff) + except asyncio.TimeoutError: + pass + backoff = min(backoff * 2, self._RECONNECT_MAX_BACKOFF) + + async def _connect_and_serve(self, reconnect: bool) -> None: + self.srp_authenticate() # fresh token each attempt; old one died on drop + url = f"{self.ws_url}/ws/chat?user_id={self.user_id}&ws_token={self.ws_token}" + async with websockets.connect( + url, ssl=self._ws_ssl_context(), + ping_interval=self._PING_INTERVAL, ping_timeout=self._PING_TIMEOUT, + ) as ws: + self._ws = ws + self.running = True + announce = (f"{self.name} (operator) " + f"{'back online' if reconnect else 'online'} — Claude Code bridge") + await ws.send(self.room_fernet.encrypt(announce.encode()).decode()) + await self._emit("system", event="connected", reconnect=reconnect) + try: + await self._serve(ws) + finally: + self._ws = None + self.running = False + + async def _serve(self, ws) -> None: + async for raw in ws: + if not self.running: + break + try: + await self._handle_frame(ws, raw) + except (websockets.ConnectionClosed, asyncio.CancelledError): + raise + except Exception as e: # noqa: BLE001 — one bad frame mustn't drop us + await self._emit("system", event="frame_error", + reason=f"{type(e).__name__}: {e}") + + # ── control socket (the CLI talks here) ────────────────────────────── + async def _serve_control(self, reader: asyncio.StreamReader, + writer: asyncio.StreamWriter) -> None: + try: + line = await reader.readline() + if not line: + return + try: + req = json.loads(line.decode()) + except json.JSONDecodeError: + resp = {"ok": False, "error": "bad json"} + else: + resp = await self._dispatch(req) + writer.write((json.dumps(resp) + "\n").encode()) + await writer.drain() + except (ConnectionResetError, BrokenPipeError): + pass + finally: + try: + writer.close() + except Exception: # noqa: BLE001 + pass + + async def _dispatch(self, req: dict) -> dict: + op = req.get("op") + if op == "ping": + return {"ok": True, "pong": True, "name": self.name, + "connected": self._ws is not None} + if op == "status": + return {"ok": True, "name": self.name, + "connected": self._ws is not None, + "granted": self.granted, "can_sudo": self.can_sudo, + "sbx_engine": self.sbx_engine, "sbx_name": self.sbx_name, + "users": self._roster(), "seq": self.seq, + "host": self.server, "port": self.port} + if op == "roster": + return {"ok": True, "users": self._roster()} + if op == "read": + return await self._op_read(req) + if op == "say": + return await self._op_say(req) + if op == "down": + asyncio.get_running_loop().call_soon(self._begin_shutdown) + return {"ok": True, "stopping": True} + return {"ok": False, "error": f"unknown op {op!r}"} + + async def _op_read(self, req: dict) -> dict: + since = int(req.get("since", 0)) + wait = bool(req.get("wait", False)) + timeout = float(req.get("timeout", 30.0)) + async with self.cond: + new = [e for e in self.events if e["seq"] > since] + if not new and wait: + try: + await asyncio.wait_for(self.cond.wait(), timeout) + except asyncio.TimeoutError: + pass + new = [e for e in self.events if e["seq"] > since] + return {"ok": True, "events": new, "seq": self.seq} + + async def _op_say(self, req: dict) -> dict: + text = str(req.get("text", "")) + if not text: + return {"ok": False, "error": "empty text"} + if self._ws is None: + return {"ok": False, "error": "not connected to room"} + try: + await self._ws.send(self.room_fernet.encrypt(text.encode()).decode()) + except Exception as e: # noqa: BLE001 + return {"ok": False, "error": f"send failed: {type(e).__name__}: {e}"} + await self._emit("sent", **{"from": self.name}, text=text, addressed=False) + return {"ok": True} + + def _begin_shutdown(self) -> None: + self._stop.set() + self.running = False + if self._ws is not None: + asyncio.create_task(self._ws.close()) diff --git a/cmd_chat/operator/cli_client.py b/cmd_chat/operator/cli_client.py new file mode 100644 index 0000000..4349784 --- /dev/null +++ b/cmd_chat/operator/cli_client.py @@ -0,0 +1,45 @@ +"""Tiny synchronous AF_UNIX client for the operator control socket. + +The control protocol is newline-delimited JSON: the CLI sends one request +object terminated by ``\\n`` and reads one response object terminated by ``\\n``. +Kept dependency-free and asyncio-free so the CLI verbs stay fast and simple. +""" + +from __future__ import annotations + +import json +import socket +from pathlib import Path + + +class BridgeUnreachable(RuntimeError): + """The daemon socket is missing or not answering.""" + + +def request(sock_path: str | Path, obj: dict, read_timeout: float = 35.0) -> dict: + """Send one request, return the decoded response. + + ``read_timeout`` must exceed a long-poll ``read --wait --timeout`` window so + the client doesn't give up before the daemon replies. + """ + path = str(sock_path) + s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + s.settimeout(read_timeout) + try: + try: + s.connect(path) + except (FileNotFoundError, ConnectionRefusedError) as e: + raise BridgeUnreachable(f"no bridge at {path} ({e.__class__.__name__})") from e + s.sendall((json.dumps(obj) + "\n").encode()) + buf = bytearray() + while b"\n" not in buf: + chunk = s.recv(65536) + if not chunk: + break + buf.extend(chunk) + line, _, _ = bytes(buf).partition(b"\n") + if not line: + raise BridgeUnreachable("bridge closed the connection without replying") + return json.loads(line.decode()) + finally: + s.close() diff --git a/cmd_chat/operator/session.py b/cmd_chat/operator/session.py new file mode 100644 index 0000000..b56111f --- /dev/null +++ b/cmd_chat/operator/session.py @@ -0,0 +1,123 @@ +"""Session filesystem layout for the operator bridge. + +Each running daemon owns a session directory under a per-user runtime root: + + ${XDG_RUNTIME_DIR:-/tmp}/hh-bridge// + control.sock unix socket the CLI verbs talk to + inbox.jsonl append-only event log (durability/debug; truth is in-RAM) + meta.json {host, port, user, pid, trigger, no_tls, started} + daemon.log daemon stdout/stderr + cursor last seq the CLI `read` consumed (client-side bookmark) + +The session name defaults to the room display name (the `user` we join as), so +one operator-per-name maps to one session dir. `--session` overrides it when a +single box runs several bridges. +""" + +from __future__ import annotations + +import json +import os +import re +from pathlib import Path + + +def runtime_root() -> Path: + base = os.environ.get("XDG_RUNTIME_DIR") or "/tmp" + return Path(base) / "hh-bridge" + + +def _safe(name: str) -> str: + """Reduce a session label to a filesystem-safe slug (no path escapes).""" + slug = re.sub(r"[^A-Za-z0-9._-]", "_", name.strip()) + return slug or "default" + + +class Session: + """Resolves the paths for one bridge session; never opens anything itself.""" + + def __init__(self, name: str): + self.name = _safe(name) + self.dir = runtime_root() / self.name + + # ── paths ──────────────────────────────────────────────────────────── + @property + def sock_path(self) -> Path: + return self.dir / "control.sock" + + @property + def inbox_path(self) -> Path: + return self.dir / "inbox.jsonl" + + @property + def meta_path(self) -> Path: + return self.dir / "meta.json" + + @property + def log_path(self) -> Path: + return self.dir / "daemon.log" + + @property + def cursor_path(self) -> Path: + return self.dir / "cursor" + + # ── lifecycle helpers ──────────────────────────────────────────────── + def ensure_dir(self) -> None: + # 0700: the socket grants room-send rights, so keep it user-private. + self.dir.mkdir(parents=True, exist_ok=True) + try: + os.chmod(self.dir, 0o700) + except OSError: + pass + + def write_meta(self, **fields) -> None: + self.ensure_dir() + self.meta_path.write_text(json.dumps(fields, indent=2)) + + def read_meta(self) -> dict: + try: + return json.loads(self.meta_path.read_text()) + except (OSError, json.JSONDecodeError): + return {} + + def read_cursor(self) -> int: + try: + return int(self.cursor_path.read_text().strip() or "0") + except (OSError, ValueError): + return 0 + + def write_cursor(self, seq: int) -> None: + self.ensure_dir() + self.cursor_path.write_text(str(seq)) + + def cleanup(self) -> None: + """Remove the socket so a stale path never confuses the next `up`.""" + for p in (self.sock_path,): + try: + p.unlink() + except OSError: + pass + + +def resolve(name: str | None) -> Session: + """Pick the session to act on. + + Explicit name → that session. Otherwise, if exactly one live session dir + exists (has a control.sock), use it; if none/many, fall back to "default" + (so a clear error surfaces when the CLI can't reach a socket). + """ + if name: + return Session(name) + root = runtime_root() + if root.is_dir(): + live = [d for d in root.iterdir() if (d / "control.sock").exists()] + if len(live) == 1: + return Session(live[0].name) + return Session("default") + + +def list_sessions() -> list[Session]: + root = runtime_root() + if not root.is_dir(): + return [] + return [Session(d.name) for d in sorted(root.iterdir()) if d.is_dir()] diff --git a/tests/test_operator_bridge.py b/tests/test_operator_bridge.py new file mode 100644 index 0000000..d69ed7e --- /dev/null +++ b/tests/test_operator_bridge.py @@ -0,0 +1,197 @@ +"""Offline tests for the operator bridge (no server, no real websocket). + +Each test builds the bridge *inside* its own ``asyncio.run`` so the bridge's +``asyncio.Condition`` binds to that test's loop (it is loop-bound on first use). +""" + +import sys +import asyncio +import json +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from cryptography.fernet import Fernet + +from cmd_chat.operator.bridge import OperatorBridge + + +def _bridge(name="oracle", trigger=None): + b = OperatorBridge("h", 1, name=name, password="pw", no_tls=True, trigger=trigger) + b.room_fernet = Fernet(Fernet.generate_key()) + return b + + +def _enc(b, text): + return b.room_fernet.encrypt(text.encode()).decode() + + +def _msg_frame(b, sender, text): + """A server 'message' envelope carrying an encrypted line from `sender`.""" + return json.dumps({"type": "message", + "data": {"username": sender, "text": _enc(b, text)}}) + + +class FakeWS: + def __init__(self): + self.sent = [] + + async def send(self, data): + self.sent.append(data) + + async def close(self): + pass + + +# ── addressing ──────────────────────────────────────────────────────────── +def test_is_addressed(): + b = _bridge("oracle", trigger="@bot") + assert b._is_addressed("hey oracle you there") + assert b._is_addressed("@oracle ping") + assert b._is_addressed("ORACLE up?") # case-insensitive + assert b._is_addressed("ask @bot to help") # custom trigger + assert not b._is_addressed("oracleX is a var") # word boundary + assert not b._is_addressed("just chatting") + + +# ── inbox: seq + long-poll wake ─────────────────────────────────────────── +def test_emit_seq_monotonic_and_ring(): + async def go(): + b = _bridge() + for i in range(5): + await b._emit("message", **{"from": "a"}, text=f"m{i}", addressed=False) + seqs = [e["seq"] for e in b.events] + assert seqs == [1, 2, 3, 4, 5] + assert b.seq == 5 + asyncio.run(go()) + + +def test_read_long_poll_wakes_on_emit(): + async def go(): + b = _bridge() + + async def reader(): + return await b._op_read({"since": 0, "wait": True, "timeout": 5}) + + async def writer(): + await asyncio.sleep(0.05) + await b._emit("message", **{"from": "alice"}, text="hi oracle", addressed=True) + + rd, _ = await asyncio.gather(reader(), writer()) + assert rd["ok"] and len(rd["events"]) == 1 + assert rd["events"][0]["text"] == "hi oracle" + assert rd["events"][0]["addressed"] is True + asyncio.run(go()) + + +def test_read_since_filters(): + async def go(): + b = _bridge() + for i in range(3): + await b._emit("message", **{"from": "a"}, text=str(i), addressed=False) + rd = await b._op_read({"since": 2, "wait": False}) + assert [e["seq"] for e in rd["events"]] == [3] + asyncio.run(go()) + + +# ── frame routing ───────────────────────────────────────────────────────── +def test_message_frame_emits_addressed(): + async def go(): + b = _bridge("oracle") + await b._handle_frame(None, _msg_frame(b, "alice", "hey oracle help")) + msgs = [e for e in b.events if e["kind"] == "message"] + assert len(msgs) == 1 + assert msgs[0]["from"] == "alice" and msgs[0]["addressed"] is True + asyncio.run(go()) + + +def test_own_message_ignored(): + async def go(): + b = _bridge("oracle") + await b._handle_frame(None, _msg_frame(b, "oracle", "/say echo")) + assert [e for e in b.events if e["kind"] == "message"] == [] + asyncio.run(go()) + + +def test_control_acl_recorded(): + async def go(): + b = _bridge("oracle") + ctrl = json.dumps({"_perm": "acl", "owner": "alice", + "drivers": ["alice", "oracle"], "sudoers": ["alice"]}) + await b._handle_frame(None, _msg_frame(b, "alice", ctrl)) + assert b.granted is True + acl = [e for e in b.events if e["kind"] == "acl"] + assert acl and acl[0]["granted"] is True + # an ACL control frame must NOT surface as a chat 'message' + assert [e for e in b.events if e["kind"] == "message"] == [] + asyncio.run(go()) + + +def test_control_sandbox_status_recorded(): + async def go(): + b = _bridge("oracle") + ctrl = json.dumps({"_sbx": "status", "state": "ready", + "engine": "podman", "name": "hack-house", "backend": "podman"}) + await b._handle_frame(None, _msg_frame(b, "alice", ctrl)) + assert b.sbx_engine == "podman" and b.sbx_name == "hack-house" + assert any(e["kind"] == "sandbox" for e in b.events) + asyncio.run(go()) + + +def test_init_backfill_once(): + async def go(): + b = _bridge("oracle") + init = json.dumps({ + "type": "init", + "users": [{"username": "alice"}, {"username": "oracle"}], + "messages": [ + {"username": "alice", "text": _enc(b, "older line")}, + {"username": "oracle", "text": _enc(b, "my own old line")}, # skipped + ], + }) + await b._handle_frame(None, init) + backfilled = [e for e in b.events if e.get("backfill")] + assert len(backfilled) == 1 and backfilled[0]["from"] == "alice" + assert backfilled[0]["addressed"] is False + # a reconnect re-sends init; we must not duplicate history + await b._handle_frame(None, init) + assert len([e for e in b.events if e.get("backfill")]) == 1 + asyncio.run(go()) + + +# ── control ops ─────────────────────────────────────────────────────────── +def test_say_requires_connection(): + async def go(): + b = _bridge() + resp = await b._op_say({"text": "hello"}) + assert resp["ok"] is False and "not connected" in resp["error"] + asyncio.run(go()) + + +def test_say_sends_and_records(): + async def go(): + b = _bridge() + ws = FakeWS() + b._ws = ws + resp = await b._op_say({"text": "yes operator online"}) + assert resp["ok"] is True + assert len(ws.sent) == 1 + # round-trips through the room key + assert b.room_fernet.decrypt(ws.sent[0].encode()).decode() == "yes operator online" + sent = [e for e in b.events if e["kind"] == "sent"] + assert sent and sent[0]["text"] == "yes operator online" + asyncio.run(go()) + + +def test_dispatch_status_and_roster(): + async def go(): + b = _bridge("oracle") + b.users = [{"username": "oracle"}, {"username": "alice"}] + st = await b._dispatch({"op": "status"}) + assert st["ok"] and st["name"] == "oracle" and st["connected"] is False + assert set(st["users"]) == {"oracle", "alice"} + ping = await b._dispatch({"op": "ping"}) + assert ping["pong"] is True + unknown = await b._dispatch({"op": "frobnicate"}) + assert unknown["ok"] is False + asyncio.run(go())