"""The room is the bus (SPEC pillar #1). Two interchangeable substrates implement the same synchronous ``Room`` facade so ``loop.py`` never knows which it is talking to (mirroring how ``runtime.py`` has Podman/Local): • RealRoom — boots the hack-house relay, connects one authenticated client per member plus a referee/recorder client, and posts *real* end-to-end-encrypted frames. Every utterance round-trips through the zero-knowledge server and is observed off the wire by the referee. This is the project thesis: team deliberation flows through a real encrypted room, not a simulated channel. • LocalBus — an in-memory echo bus with the identical facade. No server, no flakiness; used as the default for deterministic local runs and CI. It records the same transcript a RealRoom would. Both are arena-mode: the orchestrator drives inference and *posts* results into the room. Neither edits ``cmd_chat`` — RealRoom only uses the public Client. """ from __future__ import annotations import asyncio import json import socket import subprocess import sys import threading import time from pathlib import Path REPO = Path(__file__).resolve().parents[4] class Delivery: """Result of a post: did the frame round-trip through the substrate?""" __slots__ = ("delivered", "substrate") def __init__(self, delivered: bool, substrate: str): self.delivered = delivered self.substrate = substrate def __repr__(self): return f"Delivery(delivered={self.delivered}, substrate={self.substrate})" # ── in-memory substrate ──────────────────────────────────────────────────── class LocalBus: """Zero-dependency echo bus. Same facade as RealRoom.""" substrate = "local" def __init__(self): self.log: list[dict] = [] def connect(self) -> None: pass def post(self, actor: str, text: str) -> Delivery: self.log.append({"actor": actor, "text": text, "ts": time.time()}) return Delivery(True, self.substrate) def brief(self, text: str) -> Delivery: return self.post("referee", text) def acl(self, payload: dict) -> Delivery: self.log.append({"actor": "referee", "acl": payload, "ts": time.time()}) return Delivery(True, self.substrate) def close(self) -> None: pass # ── real encrypted-room substrate ────────────────────────────────────────── def _port_open(host: str, port: int, timeout: float = 0.5) -> bool: try: with socket.create_connection((host, port), timeout=timeout): return True except OSError: return False class RealRoom: """Boots the relay and connects N member clients + a referee recorder.""" substrate = "real" def __init__(self, member_names: list[str], *, host: str = "127.0.0.1", port: int = 4677, password: str = "olympics-pass", boot_server: bool = True, quiet: float = 1.2, log_dir: str = "/tmp/hh-olympics"): self.member_names = member_names self.host = host self.port = port self.password = password self.boot_server = boot_server self.quiet = quiet self.log_dir = Path(log_dir) self._srv: subprocess.Popen | None = None self._srv_log = None self._loop: asyncio.AbstractEventLoop | None = None self._thread: threading.Thread | None = None self._clients: dict[str, object] = {} # name -> Client self._ws: dict[str, object] = {} # name -> websocket self._referee = "referee" # -- lifecycle ---------------------------------------------------------- def connect(self) -> None: if str(REPO) not in sys.path: sys.path.insert(0, str(REPO)) from cmd_chat.client.client import Client # noqa: E402 if self.boot_server: self._spawn_server() self._loop = asyncio.new_event_loop() names = [*self.member_names, self._referee] for name in names: c = Client(self.host, self.port, username=name, password=self.password, no_tls=True) c.srp_authenticate() self._clients[name] = c # Run the loop continuously on a background thread. The orchestrator does # long blocking inference between posts; if the loop only ran during # run_until_complete(post) the websockets reader couldn't answer the # server's keepalive ping in those gaps and the relay would drop us # (1011). A forever-loop keeps pings serviced regardless of the main # thread blocking. self._thread = threading.Thread(target=self._loop.run_forever, daemon=True) self._thread.start() self._call(self._open_all()) def _call(self, coro): """Run a coroutine on the background loop from the main thread.""" return asyncio.run_coroutine_threadsafe(coro, self._loop).result() def _spawn_server(self) -> None: self.log_dir.mkdir(parents=True, exist_ok=True) self._srv_log = open(self.log_dir / f"server-{self.port}.log", "w") self._srv = subprocess.Popen( [sys.executable, "cmd_chat.py", "serve", self.host, str(self.port), "--password", self.password, "--no-tls"], cwd=str(REPO), stdout=self._srv_log, stderr=subprocess.STDOUT) deadline = time.time() + 30 while time.time() < deadline: if _port_open(self.host, self.port): return time.sleep(0.2) raise RuntimeError(f"relay server never bound on {self.host}:{self.port}") async def _open_all(self) -> None: import websockets for name, c in self._clients.items(): url = (f"{c.ws_url}/ws/chat?user_id={c.user_id}" f"&ws_token={c.ws_token}") self._ws[name] = await websockets.connect(url) # let presence settle so the referee sees subsequent broadcasts await asyncio.sleep(0.4) # -- posting ------------------------------------------------------------ def _encrypt(self, name: str, text: str) -> str: c = self._clients[name] return c.room_fernet.encrypt(text.encode()).decode() async def _send(self, name: str, text: str) -> None: await self._ws[name].send(self._encrypt(name, text)) async def _observe(self, want_actor: str, want_text: str, deadline: float) -> bool: """Drain the referee ws until we see the actor's frame (round-trip).""" import websockets ref = self._clients[self._referee] ws = self._ws[self._referee] snippet = want_text[:24] while time.time() < deadline: try: raw = await asyncio.wait_for(ws.recv(), timeout=max(0.05, deadline - time.time())) except asyncio.TimeoutError: return False except websockets.ConnectionClosed: return False try: data = json.loads(raw) except json.JSONDecodeError: continue if data.get("type") != "message": continue dec = ref.decrypt_message(data.get("data", {})) if dec.get("username") == want_actor and snippet in dec.get("text", ""): return True return False def _post_sync(self, actor: str, text: str) -> Delivery: async def _do(): await self._send(actor, text) ok = await self._observe(actor, text, time.time() + self.quiet) return ok ok = self._call(_do()) return Delivery(ok, self.substrate) def post(self, actor: str, text: str) -> Delivery: return self._post_sync(actor, text) def brief(self, text: str) -> Delivery: return self._post_sync(self._referee, text) def acl(self, payload: dict) -> Delivery: # referee (acting as owner) broadcasts the ACL control frame return self._post_sync(self._referee, json.dumps(payload)) def close(self) -> None: if self._loop is not None: async def _close(): for ws in self._ws.values(): try: await ws.close() except Exception: # noqa: BLE001 pass try: self._call(_close()) except Exception: # noqa: BLE001 pass self._loop.call_soon_threadsafe(self._loop.stop) if self._thread is not None: self._thread.join(timeout=5) self._loop.close() if self._srv is not None: self._srv.terminate() try: self._srv.wait(timeout=10) except subprocess.TimeoutExpired: self._srv.kill() if self._srv_log is not None: self._srv_log.close() def make_room(kind: str, member_names: list[str], **kw) -> "LocalBus | RealRoom": """Pick a substrate. 'local' = in-memory, 'real' = boot relay + clients.""" if kind == "real": return RealRoom(member_names, **kw) return LocalBus()