feat(operator): Phase 6 — relay-mode read (screen) + stop-condition engine (watch)

Completes the remote-drive triad for a broker-owned sandbox the operator can't
exec into directly: keys (in) → watch (wait) → screen (out).

- _sbx:data PTY-relay frames are absorbed into a capped rolling terminal buffer
  (not surfaced as chat); `screen` returns it ansi-stripped (or raw). sandbox.
  strip_ansi handles CSI/OSC/CR noise so output greps cleanly.
- `watch` is a formal stop-condition engine: blocks until a regex matches in the
  screen buffer or chat events, or an idle-quiescence window, or a hard timeout,
  then reports which fired — the autonomy loop's principled wait.
- Skill doctrine updated with the type→wait→read relay loop.

28 offline tests green.

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

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

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.

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