diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py index 1ecf18e..3d6d6bb 100644 --- a/cmd_chat/operator/__main__.py +++ b/cmd_chat/operator/__main__.py @@ -105,6 +105,21 @@ def _build_parser() -> argparse.ArgumentParser: help="actually launch (default is a dry-run plan)") sp.add_argument("--session", default=None) + sc = sub.add_parser("screen", help="print the relayed sandbox terminal buffer") + sc.add_argument("--tail", type=int, default=None, help="last N bytes only") + sc.add_argument("--ansi", action="store_true", help="keep ANSI codes (raw)") + sc.add_argument("--session", default=None) + + wt = sub.add_parser("watch", help="block until a stop condition fires") + wt.add_argument("--for", dest="pattern", default=None, + help="regex to match (stops on first hit)") + wt.add_argument("--in", dest="source", choices=["events", "screen"], + default="events", help="where to match (default: events)") + wt.add_argument("--idle", type=float, default=None, + help="stop after this many seconds of quiet") + wt.add_argument("--timeout", type=float, default=30.0) + wt.add_argument("--session", default=None) + ky = sub.add_parser("keys", help="inject keystrokes into the room's shared PTY") ky.add_argument("keys", nargs="*", help="tokens: literal text, named keys (enter, ctrl-c, esc, " @@ -295,6 +310,25 @@ def _run_get(args) -> int: return 0 +def _run_screen(args) -> int: + resp = _client_request(args, {"op": "screen", "tail": args.tail, + "ansi": args.ansi}) + if not resp.get("ok"): + print(resp.get("error", "screen failed"), file=sys.stderr) + return 1 + sys.stdout.write(resp.get("text", "")) + return 0 + + +def _run_watch(args) -> int: + resp = _client_request(args, {"op": "watch", "for": args.pattern, + "in": args.source, "idle": args.idle, + "timeout": args.timeout}, + read_timeout=args.timeout + 5.0) + print(json.dumps(resp)) + return 0 if resp.get("ok") else 1 + + def _run_spawn(args) -> int: resp = _client_request(args, { "op": "spawn", "objective": args.objective, @@ -351,6 +385,10 @@ def main(argv: list[str] | None = None) -> int: return _run_keys(args) if verb == "spawn": return _run_spawn(args) + if verb == "screen": + return _run_screen(args) + if verb == "watch": + return _run_watch(args) if verb in ("roster", "status", "down"): return _run_simple(args, verb) return 2 diff --git a/cmd_chat/operator/bridge.py b/cmd_chat/operator/bridge.py index 7b8533d..1ac5435 100644 --- a/cmd_chat/operator/bridge.py +++ b/cmd_chat/operator/bridge.py @@ -36,6 +36,7 @@ class OperatorBridge(Client): _RECONNECT_MAX_BACKOFF = 30.0 _RECONNECT_STABLE_SECS = 30.0 MAX_EVENTS = 2000 # in-RAM inbox ring; full history lives in inbox.jsonl + TERM_CAP = 65536 # rolling relayed-PTY buffer (bytes) kept for `screen` def __init__(self, server: str, port: int, name: str, password: str | None = None, insecure: bool = False, @@ -66,6 +67,12 @@ class OperatorBridge(Client): self._own_engine: str | None = None self._own_name: str = "" + # Relay terminal buffer (Phase 6): rolling raw PTY bytes the broker + # relays as `_sbx:data`. Lets the operator *read* a remote/broker-owned + # sandbox it drives via `keys` — the output half of relay mode. + self._term = bytearray() + self._term_seq = 0 # bumps on each chunk so `watch` can poll for growth + # Live websocket (None while reconnecting); set in _connect_and_serve. self._ws = None self._stop = asyncio.Event() @@ -143,13 +150,37 @@ class OperatorBridge(Client): await self._emit("message", **{"from": sender}, text=text, addressed=self._is_addressed(text)) + def _absorb_term(self, b64: str) -> None: + """Append a relayed PTY chunk to the rolling terminal buffer (capped). + High-volume, so we don't emit an inbox event per chunk — `watch`/`screen` + poll `_term_seq` instead.""" + if not b64: + return + try: + chunk = base64.b64decode(b64) + except (ValueError, TypeError): + return + self._term += chunk + if len(self._term) > self.TERM_CAP: + del self._term[:-self.TERM_CAP] + self._term_seq += 1 + + def _screen_text(self, tail: int | None, ansi: bool) -> str: + raw = bytes(self._term[-tail:] if tail else self._term) + s = raw.decode(errors="replace") + return s if ansi else sbx.strip_ansi(s) + 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).""" + """Surface ACL + sandbox-status changes into the inbox (awareness only) + and absorb relayed PTY output (`_sbx:data`) into the terminal buffer. + Ignore other high-volume control frames (_ai, _ft).""" try: frame = json.loads(text) except json.JSONDecodeError: return + if frame.get("_sbx") == "data": + self._absorb_term(frame.get("b64", "")) + return if frame.get("_sbx") == "status": ready = frame.get("state") == "ready" self.sbx_engine = frame.get("engine") if ready else None @@ -307,6 +338,10 @@ class OperatorBridge(Client): return await self._op_keys(req) if op == "spawn": return await self._op_spawn(req) + if op == "screen": + return await self._op_screen(req) + if op == "watch": + return await self._op_watch(req) if op == "down": asyncio.get_running_loop().call_soon(self._begin_shutdown) return {"ok": True, "stopping": True} @@ -474,6 +509,65 @@ class OperatorBridge(Client): await self._emit("keys", bytes=len(data)) return {"ok": True, "bytes": len(data)} + # ── relay read + stop-condition engine (Phase 6) ───────────────────── + async def _op_screen(self, req: dict) -> dict: + """Return the current relayed terminal buffer — the output half of relay + mode (drive with `keys`, read with `screen`). ANSI-stripped by default; + `tail` bounds the bytes returned.""" + tail = req.get("tail") + tail = int(tail) if tail else None + ansi = bool(req.get("ansi", False)) + return {"ok": True, "text": self._screen_text(tail, ansi), + "bytes": len(self._term), "seq": self._term_seq} + + async def _op_watch(self, req: dict) -> dict: + """Block until a stop condition fires, then report which one. Formalizes + the `/loop` autonomy stops so a driver can `keys` then wait for a result. + + Conditions (first to fire wins): + - ``for`` regex matched in the chosen source ``in`` = "events" (default, + matches event text) or "screen" (matches relayed terminal output); + - ``idle`` seconds with no new event/output (quiescence); + - ``timeout`` seconds elapsed (the hard ceiling). + """ + pattern = req.get("for") + source = req.get("in", "events") + timeout = float(req.get("timeout", 30.0)) + idle = req.get("idle") + idle = float(idle) if idle is not None else None + match_since = int(req.get("since", self.seq)) # fixed scan cursor + try: + rx = re.compile(pattern) if pattern else None + except re.error as e: + return {"ok": False, "error": f"bad regex: {e}"} + + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + last_change = loop.time() + prev_seq, prev_term = self.seq, self._term_seq + while True: + if rx and source == "screen": + m = rx.search(self._screen_text(None, False)) + if m: + return {"ok": True, "reason": "match", "where": "screen", + "match": m.group(0), "seq": self.seq} + if rx and source == "events": + for e in self.events: + if e["seq"] <= match_since: + continue + if rx.search(str(e.get("text", "")) or json.dumps(e)): + return {"ok": True, "reason": "match", "where": "events", + "event": e, "seq": self.seq} + now = loop.time() + if self.seq != prev_seq or self._term_seq != prev_term: + last_change, prev_seq, prev_term = now, self.seq, self._term_seq + if idle is not None and now - last_change >= idle: + return {"ok": True, "reason": "idle", "idle": idle, "seq": self.seq} + if now >= deadline: + return {"ok": True, "reason": "timeout", "matched": False, + "seq": self.seq} + await asyncio.sleep(min(0.3, max(0.0, deadline - now))) + # ── recursive spawn (Phase 5) ──────────────────────────────────────── async def _op_spawn(self, req: dict) -> dict: """Stand up a nested Claude Code operator against another room. diff --git a/cmd_chat/operator/sandbox.py b/cmd_chat/operator/sandbox.py index 517787b..90090f2 100644 --- a/cmd_chat/operator/sandbox.py +++ b/cmd_chat/operator/sandbox.py @@ -22,6 +22,24 @@ Nothing here opens a websocket or holds state; the bridge wires these in. from __future__ import annotations import asyncio +import re + +# CSI / OSC / single-char escape sequences + carriage returns, so relayed PTY +# bytes read as plain text. We strip rather than interpret: the operator wants +# to grep the output, not render a faithful screen. +_ANSI_RE = re.compile( + r"\x1b\[[0-9;?]*[ -/]*[@-~]" # CSI … (colours, cursor moves) + r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC … (title sets) BEL/ST-terminated + r"|\x1b[@-Z\\-_]" # two-char escapes +) + + +def strip_ansi(text: str) -> str: + """Drop ANSI/control noise from relayed terminal output. Collapses bare + carriage returns (so progress redraws don't stack) but keeps newlines.""" + clean = _ANSI_RE.sub("", text) + clean = clean.replace("\r\n", "\n").replace("\r", "\n") + return "".join(c for c in clean if c == "\n" or c == "\t" or c >= " ") # Match the native harness so behaviour is identical across both drivers. EXEC_TIMEOUT = 60.0 # NATIVE_TOOL_TIMEOUT diff --git a/skills/hh-operator/SKILL.md b/skills/hh-operator/SKILL.md index 06a8951..4f9e5a0 100644 --- a/skills/hh-operator/SKILL.md +++ b/skills/hh-operator/SKILL.md @@ -72,12 +72,18 @@ $HH sbx down # tear your container do When the **room** has a sandbox and you're granted, `exec`/`write`/`get` target it directly (you're co-located on the host) — no launch needed. -**Keystroke relay** — drive the room's *shared* terminal live, like a human: +**Keystroke relay** — drive the room's *shared* terminal live, like a human. +The relay loop is **type → wait → read**: ```bash -$HH keys "ls -la" enter # type + run -$HH keys ctrl-c # interrupt the running program -$HH keys --help-keys # print the full vocabulary +$HH keys "make build" enter # type + run +$HH watch --for "BUILD (SUCCESS|FAIL)" --in screen --timeout 120 # wait +$HH screen # read the relayed terminal (ansi-stripped) +$HH keys ctrl-c # interrupt the running program +$HH keys --help-keys # print the full vocabulary ``` +`watch` is the stop-condition engine — it blocks until a regex matches (in +`screen` output or chat `events`), or `--idle N` quiet, or `--timeout` — then +reports which fired. This is how you stay autonomous without busy-polling. ### Keystroke cheat-sheet (how to *stop* things matters most) diff --git a/tests/test_operator_bridge.py b/tests/test_operator_bridge.py index aa4cc6d..0f4f7b3 100644 --- a/tests/test_operator_bridge.py +++ b/tests/test_operator_bridge.py @@ -363,3 +363,67 @@ def test_spawn_refuses_without_objective_or_room_or_budget(): "room": {"host": "h", "port": 7}}) assert r["ok"] is False and "budget exhausted" in r["error"] asyncio.run(go()) + + +# ── Phase 6: relay read (screen) + stop-condition engine (watch) ────────── +def test_strip_ansi(): + assert sbx.strip_ansi("\x1b[31mred\x1b[0m text") == "red text" + assert sbx.strip_ansi("a\rb") == "a\nb" # bare CR → newline + assert sbx.strip_ansi("\x1b]0;title\x07ok") == "ok" # OSC title strip + + +def _sbx_data(b, raw: bytes): + """A broker PTY-relay frame carrying raw terminal bytes.""" + import base64 as _b64 + payload = json.dumps({"_sbx": "data", "b64": _b64.b64encode(raw).decode()}) + return _msg_frame(b, "broker", payload) + + +def test_sbx_data_absorbed_into_screen(): + async def go(): + b = _bridge("oracle") + await b._handle_frame(None, _sbx_data(b, b"\x1b[32mroot@box\x1b[0m:~# ls\r\n")) + # PTY relay must NOT surface as a chat message + assert [e for e in b.events if e["kind"] == "message"] == [] + r = await b._op_screen({}) + assert "root@box:~# ls" in r["text"] # ansi-stripped + raw = await b._op_screen({"ansi": True}) + assert "\x1b[32m" in raw["text"] # raw keeps codes + asyncio.run(go()) + + +def test_watch_matches_screen(): + async def go(): + b = _bridge("oracle") + + async def driver(): + await asyncio.sleep(0.05) + await b._handle_frame(None, _sbx_data(b, b"...building...\n")) + await asyncio.sleep(0.05) + await b._handle_frame(None, _sbx_data(b, b"BUILD SUCCESS\n")) + + watcher = b._op_watch({"for": "BUILD (SUCCESS|FAIL)", "in": "screen", + "timeout": 5}) + res, _ = await asyncio.gather(watcher, driver()) + assert res["reason"] == "match" and res["match"] == "BUILD SUCCESS" + asyncio.run(go()) + + +def test_watch_matches_event_and_idle_and_timeout(): + async def go(): + b = _bridge("oracle") + + async def speak(): + await asyncio.sleep(0.05) + await b._emit("message", **{"from": "alice"}, text="please stop now", + addressed=True) + res, _ = await asyncio.gather( + b._op_watch({"for": r"\bstop\b", "in": "events", "timeout": 5}), speak()) + assert res["reason"] == "match" and res["event"]["from"] == "alice" + # idle: nothing happens → quiescence fires fast + res = await b._op_watch({"idle": 0.2, "timeout": 5}) + assert res["reason"] == "idle" + # timeout: a pattern that never appears + res = await b._op_watch({"for": "never-ever", "timeout": 0.3}) + assert res["reason"] == "timeout" and res["matched"] is False + asyncio.run(go())