feat(operator): Phase 2 — sandbox drive (exec/write/get) + keystroke relay

Co-located mode: the operator launches its own podman/docker container and
execs into it out-of-band (argv-identical to the native harness), exposed as
`sbx launch|status|down`, `exec`, `write`, `get`. When granted, it can also
drive the room's broker-owned container directly on the same host.

Relay mode primitive: `keys` injects raw bytes into the shared PTY via the
same `_sbx:input` frame a human driver emits — full keyboard control incl. the
stop-vocabulary (ctrl-c/ctrl-d/ctrl-z/ctrl-\, esc, arrows, pager q). A compact
per-session cheat-sheet (`sandbox.KEYS_HELP`, `keys --help-keys`) documents
what each inject does and how to end a stuck program, token-efficiently.

Gating: exec/write/get refuse a host (`local`) inherited from the room; the
room sandbox is only driven when `granted`. Keystrokes are inert until granted.

17 offline tests green; e2e verified against real podman (launch→exec→write→
get round-trip→teardown) and through the CLI socket.

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

tests/test_operator_bridge.py

@@ -14,6 +14,7 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
 from cryptography.fernet import Fernet
 
 from cmd_chat.operator.bridge import OperatorBridge
+from cmd_chat.operator import sandbox as sbx
 
 
 def _bridge(name="oracle", trigger=None):
@@ -195,3 +196,79 @@ def test_dispatch_status_and_roster():
         unknown = await b._dispatch({"op": "frobnicate"})
         assert unknown["ok"] is False
     asyncio.run(go())
+
+
+# ── sandbox: keystroke encoding (the stop-vocabulary) ─────────────────────
+def test_encode_keys_named_and_literal():
+    assert sbx.encode_keys(["ls -la", "enter"]) == b"ls -la\r"
+    assert sbx.encode_keys("ctrl-c") == b"\x03"          # SIGINT
+    assert sbx.encode_keys("ctrl-d") == b"\x04"          # EOF
+    assert sbx.encode_keys(["CTRL-C"]) == b"\x03"        # case-insensitive
+    assert sbx.encode_keys("up") == b"\x1b[A"            # ANSI arrow
+    # 'text:' forces verbatim so a literal word like "enter" can be typed
+    assert sbx.encode_keys(["text:enter"]) == b"enter"
+    assert sbx.encode_keys("hex:1b5b41") == b"\x1b[A"
+    # a bare unknown string is typed as-is
+    assert sbx.encode_keys("q") == b"q"
+
+
+def test_exec_prefix_shapes():
+    assert sbx.exec_prefix("podman", "box") == ["podman", "exec", "-i", "box"]
+    assert sbx.exec_prefix("docker", "box") == ["docker", "exec", "-i", "box"]
+    assert sbx.exec_prefix("multipass", "vm") == ["multipass", "exec", "vm", "--"]
+    assert sbx.exec_prefix("local", "x") == []
+    assert sbx.exec_prefix("podman", "") is None        # unaddressable
+    assert sbx.exec_prefix("qemu", "x") is None
+
+
+# ── sandbox: target gating ────────────────────────────────────────────────
+def test_target_prefers_own_then_granted_room():
+    async def go():
+        b = _bridge("oracle")
+        # nothing yet → a clear error, no target
+        eng, name, err = b._target()
+        assert eng is None and err and "no sandbox" in err
+        # room has a sandbox but we're not granted → refused
+        b.sbx_engine, b.sbx_name = "podman", "hack-house"
+        eng, name, err = b._target()
+        assert eng is None and "granted" in err
+        # granted → drive the room's container directly (co-located exec)
+        b.granted = True
+        eng, name, err = b._target()
+        assert (eng, name, err) == ("podman", "hack-house", None)
+        # our own container always wins
+        b._own_engine, b._own_name = "podman", "hh-op-oracle"
+        eng, name, err = b._target()
+        assert (eng, name, err) == ("podman", "hh-op-oracle", None)
+    asyncio.run(go())
+
+
+def test_keys_requires_connection_and_grant():
+    async def go():
+        b = _bridge("oracle")
+        # not connected
+        r = await b._op_keys({"keys": ["enter"]})
+        assert r["ok"] is False and "not connected" in r["error"]
+        # connected but not granted → inert
+        b._ws = FakeWS()
+        r = await b._op_keys({"keys": ["enter"]})
+        assert r["ok"] is False and "granted" in r["error"]
+        # granted → injects an encrypted _sbx:input frame
+        b.granted = True
+        r = await b._op_keys({"keys": ["ls", "enter"]})
+        assert r["ok"] is True and r["bytes"] == 3
+        frame = json.loads(b.room_fernet.decrypt(b._ws.sent[0].encode()).decode())
+        assert frame["_sbx"] == "input"
+        import base64 as _b64
+        assert _b64.b64decode(frame["b64"]) == b"ls\r"
+    asyncio.run(go())
+
+
+def test_exec_requires_target():
+    async def go():
+        b = _bridge("oracle")
+        r = await b._op_exec({"cmd": "echo hi"})
+        assert r["ok"] is False and "no sandbox" in r["error"]
+        r = await b._op_exec({"cmd": ""})
+        assert r["ok"] is False and "empty" in r["error"]
+    asyncio.run(go())