diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py index f5f4cfe..b4b35e9 100644 --- a/cmd_chat/operator/__main__.py +++ b/cmd_chat/operator/__main__.py @@ -58,6 +58,36 @@ def _build_parser() -> argparse.ArgumentParser: s.add_argument("text", nargs="+") s.add_argument("--session", default=None) + # ── sandbox drive ─────────────────────────────────────────────────── + sb = sub.add_parser("sbx", help="manage the operator's own container") + sb.add_argument("action", choices=["launch", "status", "down"], nargs="?", + default="status") + sb.add_argument("--engine", default=None, help="podman|docker (default: auto)") + sb.add_argument("--image", default=None, help="container image (default per engine)") + sb.add_argument("--name", default=None, help="container name (default: hh-op-)") + sb.add_argument("--session", default=None) + + ex = sub.add_parser("exec", help="run a shell command in the sandbox") + ex.add_argument("cmd", nargs="+") + ex.add_argument("--session", default=None) + + wr = sub.add_parser("write", help="write a file in the sandbox (content from stdin or --text)") + wr.add_argument("path") + wr.add_argument("--text", default=None, help="inline content (else read stdin)") + wr.add_argument("--session", default=None) + + gt = sub.add_parser("get", help="read a file out of the sandbox") + gt.add_argument("path") + gt.add_argument("--out", default=None, help="write to this local path (else stdout)") + gt.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, " + "up…), text:, hex:") + ky.add_argument("--help-keys", action="store_true", help="print the key vocabulary") + ky.add_argument("--session", default=None) + for v in ("roster", "status", "down"): sub.add_parser(v).add_argument("--session", default=None) return ap @@ -191,6 +221,67 @@ def _run_say(args) -> int: return 0 +def _run_sbx(args) -> int: + resp = _client_request(args, {"op": "sbx", "action": args.action, + "engine": args.engine, "image": args.image, + "name": args.name}) + print(json.dumps(resp)) + return 0 if resp.get("ok") else 1 + + +def _run_exec(args) -> int: + cmd = " ".join(args.cmd) + resp = _client_request(args, {"op": "exec", "cmd": cmd}) + out = resp.get("output", "") + if out: + sys.stdout.write(out if out.endswith("\n") else out + "\n") + if not resp.get("ok"): + if "error" in resp: + print(resp["error"], file=sys.stderr) + return resp.get("rc", 1) or 1 + return 0 + + +def _run_write(args) -> int: + text = args.text if args.text is not None else sys.stdin.read() + resp = _client_request(args, {"op": "write", "path": args.path, "content": text}) + if not resp.get("ok"): + print(resp.get("error", "write failed"), file=sys.stderr) + return 1 + print(f"wrote {resp.get('path')} ({resp.get('bytes')} bytes)", file=sys.stderr) + return 0 + + +def _run_get(args) -> int: + import base64 + resp = _client_request(args, {"op": "get", "path": args.path}) + if not resp.get("ok"): + print(resp.get("error", "get failed"), file=sys.stderr) + return 1 + raw = base64.b64decode(resp.get("b64", "")) + if args.out: + Path(args.out).write_bytes(raw) + print(f"wrote {args.out} ({len(raw)} bytes)", file=sys.stderr) + else: + sys.stdout.buffer.write(raw) + return 0 + + +def _run_keys(args) -> int: + from . import sandbox as sbx + if args.help_keys: + print(sbx.KEYS_HELP) + return 0 + if not args.keys: + print("no keys given (try --help-keys)", file=sys.stderr) + return 2 + resp = _client_request(args, {"op": "keys", "keys": args.keys}) + if not resp.get("ok"): + print(resp.get("error", "keys failed"), file=sys.stderr) + return 1 + return 0 + + def _run_simple(args, op: str) -> int: resp = _client_request(args, {"op": op}) print(json.dumps(resp)) @@ -208,6 +299,16 @@ def main(argv: list[str] | None = None) -> int: return _run_read(args) if verb == "say": return _run_say(args) + if verb == "sbx": + return _run_sbx(args) + if verb == "exec": + return _run_exec(args) + if verb == "write": + return _run_write(args) + if verb == "get": + return _run_get(args) + if verb == "keys": + return _run_keys(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 05d10bc..224b0ad 100644 --- a/cmd_chat/operator/bridge.py +++ b/cmd_chat/operator/bridge.py @@ -25,6 +25,7 @@ import time import websockets from ..client.client import Client +from . import sandbox as sbx from .session import Session @@ -56,6 +57,12 @@ class OperatorBridge(Client): self.sbx_engine: str | None = None self.sbx_name: str = "" + # Co-located sandbox the operator launched itself (Phase 2). When set we + # exec straight into it (out-of-band); independent of the room's broker + # sandbox, which we can also drive when co-located + granted. + self._own_engine: str | None = None + self._own_name: str = "" + # Live websocket (None while reconnecting); set in _connect_and_serve. self._ws = None self._stop = asyncio.Event() @@ -285,6 +292,16 @@ class OperatorBridge(Client): return await self._op_read(req) if op == "say": return await self._op_say(req) + if op == "sbx": + return await self._op_sbx(req) + if op == "exec": + return await self._op_exec(req) + if op == "write": + return await self._op_write(req) + if op == "get": + return await self._op_get(req) + if op == "keys": + return await self._op_keys(req) if op == "down": asyncio.get_running_loop().call_soon(self._begin_shutdown) return {"ok": True, "stopping": True} @@ -317,6 +334,141 @@ class OperatorBridge(Client): await self._emit("sent", **{"from": self.name}, text=text, addressed=False) return {"ok": True} + # ── sandbox drive (Phase 2) ────────────────────────────────────────── + def _target(self) -> tuple[str | None, str, str | None]: + """Pick the sandbox to exec into and report why. + + Returns ``(engine, name, error)``. Prefer the operator's *own* container + (always ours to drive). Else, if the room granted us drive AND its broker + sandbox is a kind we can exec directly (we're co-located on this host), + target that. ``local`` is refused here — execing on the host is opt-in + only via our own `local` launch, never inherited from the room.""" + if self._own_engine: + return self._own_engine, self._own_name, None + if self.granted and self.sbx_engine in ("docker", "podman", "multipass") \ + and self.sbx_name: + return self.sbx_engine, self.sbx_name, None + if self.sbx_engine and not self.granted: + return None, "", "not a granted driver (owner must `/grant`)" + return None, "", "no sandbox — `sbx launch` one or join a room with one" + + async def _op_sbx(self, req: dict) -> dict: + """Manage the operator's own co-located container: launch / status / down.""" + action = req.get("action", "status") + if action == "launch": + engine = req.get("engine") or sbx.pick_engine() + if not engine: + return {"ok": False, "error": "no container engine (need podman or docker)"} + image = req.get("image") or sbx.default_image(engine) + name = req.get("name") or f"hh-op-{self.name}" + ok, msg = await sbx.launch_container(engine, name, image) + if ok: + self._own_engine, self._own_name = engine, name + await self._emit("sandbox", state="own-ready", engine=engine, + name=name, image=image) + return {"ok": ok, "engine": engine, "name": name, "image": image, + "message": msg} + if action == "down": + if not self._own_engine: + return {"ok": False, "error": "no own sandbox to tear down"} + ok, msg = await sbx.teardown_container(self._own_engine, self._own_name) + if ok: + await self._emit("sandbox", state="own-down", + engine=self._own_engine, name=self._own_name) + self._own_engine, self._own_name = None, "" + return {"ok": ok, "message": msg} + # status + engine, name, err = self._target() + return {"ok": True, "target_engine": engine, "target_name": name, + "reason": err, "own_engine": self._own_engine, + "own_name": self._own_name, "room_engine": self.sbx_engine, + "room_name": self.sbx_name, "granted": self.granted} + + async def _op_exec(self, req: dict) -> dict: + """Run a shell command in the target sandbox, capture combined output + + exit code. The command runs under `sh -c` (a real shell, intended) inside + the container — the container is the blast radius.""" + cmd = str(req.get("cmd", "")) + if not cmd: + return {"ok": False, "error": "empty cmd"} + engine, name, err = self._target() + if err: + return {"ok": False, "error": err} + prefix = sbx.exec_prefix(engine, name) + if prefix is None: + return {"ok": False, "error": f"can't address {engine}/{name}"} + out, rc = await sbx.exec_capture(prefix + ["sh", "-c", cmd]) + await self._emit("exec", engine=engine, name=name, cmd=cmd, rc=rc) + return {"ok": rc == 0, "rc": rc, "output": out, "engine": engine, "name": name} + + async def _op_write(self, req: dict) -> dict: + """Write a file in the target sandbox. Content arrives on stdin (never + shell-parsed); the path is a positional arg. `mkdir -p` the parent first + so an absolute path into a new dir works. Mirrors the native harness.""" + path = str(req.get("path", "")).strip() + if not path: + return {"ok": False, "error": "missing path"} + content = req.get("content", "") + data = content.encode() if isinstance(content, str) else bytes(content) + engine, name, err = self._target() + if err: + return {"ok": False, "error": err} + prefix = sbx.exec_prefix(engine, name) + if prefix is None: + return {"ok": False, "error": f"can't address {engine}/{name}"} + out, rc = await sbx.exec_capture( + prefix + ["sh", "-c", 'mkdir -p "$(dirname "$1")" && cat > "$1"', + "hh-write", path], + stdin=data) + await self._emit("write", engine=engine, name=name, path=path, rc=rc) + return {"ok": rc == 0, "rc": rc, "path": path, "bytes": len(data), + "output": out if (out or "").strip() else ""} + + async def _op_get(self, req: dict) -> dict: + """Read a file out of the target sandbox. Returns base64 (binary-safe); + the content is plumbing, so it's uncapped (unlike exec output).""" + path = str(req.get("path", "")).strip() + if not path: + return {"ok": False, "error": "missing path"} + engine, name, err = self._target() + if err: + return {"ok": False, "error": err} + prefix = sbx.exec_prefix(engine, name) + if prefix is None: + return {"ok": False, "error": f"can't address {engine}/{name}"} + raw, rc = await sbx.exec_capture(prefix + ["cat", "--", path], text=False) + if rc != 0: + return {"ok": False, "rc": rc, + "error": f"read failed (exit={rc}): {raw.decode(errors='replace')[:200]}"} + return {"ok": True, "rc": 0, "path": path, + "b64": base64.b64encode(raw).decode(), "bytes": len(raw)} + + async def _op_keys(self, req: dict) -> dict: + """Relay raw keystrokes into the room's *shared* PTY — the same + `_sbx:input` frame a human driver's terminal emits. The broker writes our + bytes to the shared shell iff we're a granted driver, so this is how the + operator drives a broker-owned sandbox (and tests interactive programs: + ctrl-c to stop, q to quit a pager, esc to leave vim). See + `sandbox.KEYS_HELP` for the byte vocabulary.""" + if self._ws is None: + return {"ok": False, "error": "not connected to room"} + if not self.granted: + return {"ok": False, "error": "not a granted driver — keystrokes inert until `/grant`"} + spec = req.get("keys") + if spec in (None, "", []): + return {"ok": False, "error": "no keys", "help": sbx.KEYS_HELP} + try: + data = sbx.encode_keys(spec) + except ValueError as e: + return {"ok": False, "error": f"bad key spec: {e}"} + frame = json.dumps({"_sbx": "input", "b64": base64.b64encode(data).decode()}) + try: + await self._ws.send(self.room_fernet.encrypt(frame.encode()).decode()) + except Exception as e: # noqa: BLE001 + return {"ok": False, "error": f"send failed: {type(e).__name__}: {e}"} + await self._emit("keys", bytes=len(data)) + return {"ok": True, "bytes": len(data)} + def _begin_shutdown(self) -> None: self._stop.set() self.running = False diff --git a/cmd_chat/operator/sandbox.py b/cmd_chat/operator/sandbox.py new file mode 100644 index 0000000..517787b --- /dev/null +++ b/cmd_chat/operator/sandbox.py @@ -0,0 +1,187 @@ +"""Sandbox primitives for the operator bridge — pure, dependency-light helpers. + +Two ways the operator touches a sandbox: + +1. **Co-located exec** (this module's `exec_*` / `launch_*`): the daemon owns a + container it started itself (`podman`/`docker run -d … sleep infinity`) and + runs commands *out-of-band* with `engine exec -i`. Output is captured and + returned — clean, scriptable, invisible to the room. Mirrors the native + harness's exec path (cmd_chat/agent/bridge.py:640-676) argv-for-argv so the + blast radius and quoting rules are identical. + +2. **Relay keystrokes** (`encode_keys`): when the *room's* shared sandbox is + owned by the Rust broker, the only way in is the same `_sbx:input` keystroke + frame a human driver's terminal emits. The broker writes our bytes to the + shared PTY iff we're a granted driver. That path needs a precise byte + vocabulary — including how to *stop* a running program — which lives here so + it's one tested table, reused by the bridge and documented once per session. + +Nothing here opens a websocket or holds state; the bridge wires these in. +""" + +from __future__ import annotations + +import asyncio + +# Match the native harness so behaviour is identical across both drivers. +EXEC_TIMEOUT = 60.0 # NATIVE_TOOL_TIMEOUT +OUTPUT_CAP = 4096 # NATIVE_OUTPUT_CAP (bytes of combined stdout+stderr) + + +# ── engine / image selection ─────────────────────────────────────────────── +def pick_engine() -> str | None: + """Prefer podman (rootless, daemonless, no sudo) then docker. None if neither + is on PATH — the caller surfaces that as a clear 'no engine' error.""" + import shutil + + for eng in ("podman", "docker"): + if shutil.which(eng): + return eng + return None + + +def default_image(engine: str) -> str: + """Match the Rust/agent defaults: Podman→Kali, Docker→Parrot.""" + return "kalilinux/kali-rolling" if engine == "podman" else "parrotsec/core" + + +# ── co-located exec ───────────────────────────────────────────────────────── +def exec_prefix(engine: str | None, name: str | None) -> list[str] | None: + """argv prefix that runs a program *inside* the named sandbox, or None if we + can't address it. The real program + args are appended as separate argv + elements (never a shell-interpolated string), so untrusted room text can't + inject shell metacharacters outside an explicit `sh -c`. `-i` keeps stdin + open so `write` can pipe file content in. Identical shape to the native + harness (bridge.py:640).""" + if engine in ("docker", "podman"): + return [engine, "exec", "-i", name] if name else None + if engine == "multipass": + return ["multipass", "exec", name, "--"] if name else None + if engine == "local": + return [] # host shell — explicit, opt-in exception + return None + + +async def exec_capture(argv: list[str], stdin: bytes | None = None, + text: bool = True) -> tuple[str | bytes, int]: + """Run an exec argv, capture combined stdout+stderr and the exit code, + time-bounded. `text=True` decodes + byte-caps for display; `text=False` + returns raw bytes uncapped (for `get`, where truncation would corrupt the + file). The container/VM is the blast radius; `local` is the host by choice.""" + try: + proc = await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.PIPE if stdin is not None else asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + except (FileNotFoundError, OSError) as e: + return (f"[exec failed: {e}]" if text else b""), 127 + try: + out, _ = await asyncio.wait_for(proc.communicate(input=stdin), timeout=EXEC_TIMEOUT) + except asyncio.TimeoutError: + proc.kill() + return ("[command timed out]" if text else b""), 124 + rc = proc.returncode if proc.returncode is not None else -1 + if not text: + return out, rc + s = out.decode(errors="replace") + if len(s) > OUTPUT_CAP: + s = s[:OUTPUT_CAP] + "\n[output truncated]" + return s, rc + + +async def launch_container(engine: str, name: str, image: str) -> tuple[bool, str]: + """Start a persistent throwaway container we can exec into: remove any stale + one, then `run -d --name … --hostname … -w /root sleep infinity` + (argv-identical to hh/src/sbx.rs:779). Returns (ok, message).""" + await exec_capture([engine, "rm", "-f", name]) # ignore "no such container" + out, rc = await exec_capture( + [engine, "run", "-d", "--name", name, "--hostname", name, + "-w", "/root", image, "sleep", "infinity"]) + if rc != 0: + tail = (out or "").strip().splitlines()[-1:] or [""] + return False, f"{engine} run failed (exit={rc}): {tail[0]}" + return True, f"launched {engine} container '{name}' from {image}" + + +async def teardown_container(engine: str, name: str) -> tuple[bool, str]: + out, rc = await exec_capture([engine, "rm", "-f", name]) + if rc != 0: + return False, f"{engine} rm failed (exit={rc}): {(out or '').strip()[:200]}" + return True, f"removed {engine} container '{name}'" + + +# ── relay keystrokes (shared-PTY drive) ───────────────────────────────────── +# +# A terminal is just a byte stream. Named keys below are the bytes a real +# keyboard sends; the bridge base64-wraps them into a `{"_sbx":"input"}` frame. +# The control characters are the *how to stop things* vocabulary — being able to +# interrupt a hung program is as essential as starting one, especially in tests. +NAMED_KEYS: dict[str, bytes] = { + # line endings / whitespace + "enter": b"\r", "return": b"\r", "newline": b"\n", "tab": b"\t", + "space": b" ", "backspace": b"\x7f", + # the stop/exit family — the most important keys for staying unstuck + "ctrl-c": b"\x03", # SIGINT — interrupt the foreground program + "ctrl-d": b"\x04", # EOF — end stdin / exit an empty shell + "ctrl-z": b"\x1a", # SIGTSTP — suspend to the background (then `fg`/`bg`) + "ctrl-\\": b"\x1c", # SIGQUIT — hard quit + core dump when SIGINT is ignored + "esc": b"\x1b", # leave vim insert-mode, dismiss menus, abort readline + # screen / job control niceties + "ctrl-l": b"\x0c", # clear/redraw the screen + "ctrl-u": b"\x15", # kill the input line (clear a half-typed command) + "ctrl-a": b"\x01", # start-of-line (also tmux/screen prefix) + "ctrl-e": b"\x05", # end-of-line + "ctrl-k": b"\x0b", # kill to end-of-line + "ctrl-w": b"\x17", # delete the previous word + # arrows / paging — ANSI sequences (history, menus, pagers) + "up": b"\x1b[A", "down": b"\x1b[B", "right": b"\x1b[C", "left": b"\x1b[D", + "home": b"\x1b[H", "end": b"\x1b[F", + "pageup": b"\x1b[5~", "pagedown": b"\x1b[6~", + "delete": b"\x1b[3~", +} + +# Token-efficient cheat-sheet the skill/status surfaces once per session, so a +# Claude operator knows the stop-vocabulary without re-deriving it every turn. +KEYS_HELP = ( + "send-keys vocabulary (bytes → shared PTY; granted drivers only):\n" + " literal text types it verbatim (no trailing newline)\n" + " enter/tab/space/esc \\r \\t ' ' \\x1b\n" + " ctrl-c interrupt the running program (SIGINT) — your main 'stop'\n" + " ctrl-d EOF: end stdin / exit an empty shell\n" + " ctrl-z suspend to background (resume with `fg`); ctrl-\\ = SIGQUIT\n" + " ctrl-u clear a half-typed line; ctrl-l clears the screen\n" + " up/down/left/right home end pageup pagedown delete backspace\n" + " q quits most pagers (less/man); :q quits vim\n" + "End a stuck command with ctrl-c; if that's ignored, ctrl-\\ then ctrl-c again." +) + + +def encode_keys(spec) -> bytes: + """Resolve a key spec into the raw bytes to inject. + + Accepts either a list of tokens or a single string: + - a NAMED_KEYS name (case-insensitive) → that control sequence + - 'text:...' → the rest, verbatim (preserves a + literal name like 'enter') + - 'hex:1b5b41' / '\\x1b' → raw bytes from hex + - anything else → typed verbatim as UTF-8 + + A bare string with no recognised token is typed as-is, so the common case + (`send-keys "ls -la" enter`) is one obvious call. Naming is explicit on + purpose: nothing here guesses, so an operator always knows what it sent.""" + tokens = spec if isinstance(spec, (list, tuple)) else [spec] + out = bytearray() + for tok in tokens: + s = "" if tok is None else str(tok) + low = s.lower() + if low in NAMED_KEYS: + out += NAMED_KEYS[low] + elif s.startswith("text:"): + out += s[5:].encode() + elif low.startswith("hex:"): + out += bytes.fromhex(s[4:]) + else: + out += s.encode() + return bytes(out) diff --git a/tests/test_operator_bridge.py b/tests/test_operator_bridge.py index d69ed7e..131ae8c 100644 --- a/tests/test_operator_bridge.py +++ b/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())