feat(operator): Phase 1 — Claude-driven room bridge (join/read/say)

Invert the agent model: a headless OperatorBridge(Client) owns the
websocket while a Claude Code session drives it via a unix control
socket. Reuses Client SRP/Fernet and AgentBridge's reconnect/serve
shape. Ships an hh-bridge CLI (up/read/say/roster/status/down) with a
seq'd in-RAM inbox + asyncio.Condition long-poll (read --wait) for
in-turn autonomy. ACL/sandbox-status frames are recorded for later
phases. Sandbox drive, delegation and nesting are out of scope here.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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"]

cmd_chat/operator/__main__.py

@@ -0,0 +1,217 @@
+"""CLI for the operator bridge — `python -m cmd_chat.operator <verb>`.
+
+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())

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"(?<!\w)(" + "|".join(toks) + r")(?!\w)", re.IGNORECASE)
+
+    def _is_addressed(self, text: str) -> 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())

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()

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/<session>/
+        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()]

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())