leetcrypt
df8f1881d8
Fourth benchmark axis: teams of LLM agents deliberate in a room, implement code in an isolated VM, and are scored deterministically on correctness/speed. - Multi-language adapter (python/js/go/rust/bash) via MultiPL-E continuation mode - Append-only JSONL ledger with status tracking (ok/dnf/killed/error) so budget-exhausted or crashed runs still record a row (fixes selection bias) - Model-aware wall-clock scaling (U-shaped by param count; 3x for reasoning) - Self-owned SIGALRM/SIGTERM watchdog (RunTimeout: BaseException so broad except Exception handlers in the infer/completion path can't swallow it) - Seed forwarded to Ollama sampler + markdown-fence stripping in completion.py
243 lines
9.3 KiB
Python
243 lines
9.3 KiB
Python
"""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()
|