retoor/tai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Agents can drive the running grr TUI: a private unix control socket in --tui and six tui_* tools (screen, keys, type, click, goto, result) whose calls can be async jobs; documented in grrtui.md, 9 tests

Browse Source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
retoor
2026-10-17 20:00:47 +02:00
co-authored by Claude Sonnet 5.5
parent ad8c241e6e
commit bdb732e918
3 changed files with 399 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
grrtui.md
+14
View File
@@ -138,6 +138,15 @@ Mouse works everywhere: click rows, buttons, tabs, chips and tiles, wheel scroll
In the terminal of a bot computer every other key goes to the shell. Tab still moves focus (the framework's rule for
every terminal widget, see [tui.md](tui.md)), so shell tab completion is not available there.
## Agents driving the TUI
While `--tui` runs it listens on a unix socket, `<TAI_HOME>/tui.sock` (or `TAI_TUI_SOCK`), mode 0600, one JSON line in and
one out. Any tai agent on the machine then has six tools (hidden while no TUI runs): `tui_screen` (route, size, focus and
the screen text; `until` waits for text to appear), `tui_keys`, `tui_type`, `tui_click`, `tui_goto` and `tui_result`.
Every action runs as a job on the UI loop. With `async` the call returns a job id at once and `tui_result` collects it,
waiting up to `wait` seconds; without it the call waits for its own job. Keys are the framework's names (`down`, `enter`,
`escape`, `ctrl+p`, a character); clicks are screen cells. A second TUI does not take over the socket of the first.
## Events
The server sends `bot.update`, `bot.delete`, `chat.update` and `chat.delete` (the lists are read again), `status`
@@ -177,6 +186,11 @@ Every limit is a constant in `tai.py`.
| `GRR_DRAFT_KEEP` | 64 | chats with a kept draft |
| `GRR_LIST_MORE` | 12 | rows Show more adds |
| `GRR_MENU_MAX_H` | 14 | rows of a menu before it scrolls |
| `GRR_CONTROL_JOBS` | 64 | control jobs kept (finished ones drop oldest first) |
| `GRR_CONTROL_LINE` | 65536 | bytes of one control request |
| `GRR_CONTROL_WAIT` | 10 | seconds a control call waits for its job by default |
| `GRR_CONTROL_WAIT_MAX` | 60 | longest wait a control call may ask for |
| `GRR_CONTROL_KEYS_MAX` | 200 | keys in one `tui_keys` call |
## Measured
tai.py
+278
View File
@@ -2087,6 +2087,12 @@ class Glow:
"plexer_read": ("Read", ident, ""),
"plexer_type": ("Type", ident, ""),
"plexer_status": ("Plexer", "", ""),
"tui_screen": ("TUI", "screen", ""),
"tui_keys": ("TUI", "keys", ""),
"tui_type": ("TUI", "type", ""),
"tui_click": ("TUI", "click", ""),
"tui_goto": ("TUI", "goto", ""),
"tui_result": ("TUI", "result", ""),
"fork": ("Fork", task, ""),
"poll": ("Poll", ident, ""),
"ask_user": ("Ask", head("question", 60), ""),
@@ -6161,6 +6167,12 @@ TOOL_SCHEMAS = [
tool_schema("plexer_read", "Only available when a tai-managed terminal session is active. Reads back the last N lines the session's shell has produced (its actual rendered output, not a tool-call summary).", {"lines": {"type": "integer"}}, []),
tool_schema("plexer_type", "Only available when a tai-managed terminal session is active. Types text into the session one character at a time with a delay between keystrokes, for a demo/animated-typing effect, then presses Enter unless enter is set to false.", {"text": {"type": "string"}, "delay": {"type": "number"}, "enter": {"type": "boolean"}}, ["text"]),
tool_schema("plexer_status", "Only available when a tai-managed terminal session is active. Reports which backend is managing the current session (tai's own native pty wrapper, or tmux) and backend-specific details (pid/socket/log, or pane id).", {}, []),
tool_schema("tui_screen", "Only available while a grr TUI (tai.py --tui) runs on this machine. Returns what its screen shows now (route, size, focus, the text). Set until to wait for that text to appear (timeout seconds), set async to get a job id at once and collect it with tui_result.", {"until": {"type": "string"}, "timeout": {"type": "integer"}, "async": {"type": "boolean"}}, []),
tool_schema("tui_keys", "Only available while a grr TUI runs. Presses keys in it like a user: names such as down, up, enter, escape, space, tab, f1, ctrl+n, ctrl+p, or single characters. Async calls return a job id for tui_result.", {"keys": {"type": "array", "items": {"type": "string"}}, "async": {"type": "boolean"}}, ["keys"]),
tool_schema("tui_type", "Only available while a grr TUI runs. Types text into the focused box (a newline presses Enter). Async calls return a job id for tui_result.", {"text": {"type": "string"}, "async": {"type": "boolean"}}, ["text"]),
tool_schema("tui_click", "Only available while a grr TUI runs. Clicks the cell at column x, row y (0 based, as in tui_screen text). Async calls return a job id for tui_result.", {"x": {"type": "integer"}, "y": {"type": "integer"}, "async": {"type": "boolean"}}, ["x", "y"]),
tool_schema("tui_goto", "Only available while a grr TUI runs. Opens a route such as /chat/new, /bots, /settings, /automations, /bot/<id>. Async calls return a job id for tui_result.", {"route": {"type": "string"}, "async": {"type": "boolean"}}, ["route"]),
tool_schema("tui_result", "Only available while a grr TUI runs. Collects an async tui_* job by id, waiting up to wait seconds; the state is running, done or error.", {"id": {"type": "integer"}, "wait": {"type": "integer"}}, ["id"]),
tool_schema("fork", "Spawn a background subagent with its own context that works while you continue. Returns an agent id immediately. Collect its summarized result with poll. Subagents get a smaller step budget and a time limit.", {"task": {"type": "string"}, "timeout": {"type": "integer"}, "profile": {"type": "string"}}, ["task"]),
tool_schema("poll", "Collect a background subagent result by id. Waits up to wait seconds, then reports running or the result.", {"id": {"type": "integer"}, "wait": {"type": "integer"}}, ["id"]),
tool_schema("ask_user", "Ask the user one question mid-turn and wait for their answer. Use this instead of ending your turn with questions in prose: an empty tool_calls list ends the turn at once and one-shot runs then exit, so prose questions are never answered. Pass the question plus optional options; the user picks an option (by number or text) or types free input, and their answer comes back as the tool result. Only for real blockers a human must decide; in autonomous mode and for background workers it is unavailable, so decide yourself and state the assumption.", {"question": {"type": "string"}, "options": {"type": "array"}}, ["question"]),
@@ -6204,6 +6216,12 @@ TOOL_ENV_GATES = {
"plexer_read": lambda: PlexerControllerFactory.detect() is not None,
"plexer_type": lambda: PlexerControllerFactory.detect() is not None,
"plexer_status": lambda: PlexerControllerFactory.detect() is not None,
"tui_screen": lambda: os.path.exists(grr_control_path()),
"tui_keys": lambda: os.path.exists(grr_control_path()),
"tui_type": lambda: os.path.exists(grr_control_path()),
"tui_click": lambda: os.path.exists(grr_control_path()),
"tui_goto": lambda: os.path.exists(grr_control_path()),
"tui_result": lambda: os.path.exists(grr_control_path()),
}
@@ -6232,6 +6250,12 @@ TOOL_TAGS = {
"plexer_read": ("plexer", "terminal", "tmux", "pane", "read", "capture"),
"plexer_type": ("plexer", "terminal", "tmux", "type", "animate", "keystroke", "demo"),
"plexer_status": ("plexer", "terminal", "tmux", "status", "session"),
"tui_screen": ("tui", "gui", "screen", "interface", "ui"),
"tui_keys": ("tui", "gui", "keys", "press", "keyboard", "navigate"),
"tui_type": ("tui", "gui", "type", "input", "text"),
"tui_click": ("tui", "gui", "click", "mouse"),
"tui_goto": ("tui", "gui", "open", "route", "navigate", "goto"),
"tui_result": ("tui", "gui", "result", "async", "job"),
"fork": ("fork", "subagent", "background", "parallel", "delegate", "spawn"),
"poll": ("poll", "collect"),
"ask_user": ("ask", "question", "clarify", "confirm", "choose", "answer", "input"),
@@ -13376,6 +13400,12 @@ class Tools:
"plexer_read": self.run_plexer_read,
"plexer_type": self.run_plexer_type,
"plexer_status": self.run_plexer_status,
"tui_screen": lambda args: self.run_tui("screen", args),
"tui_keys": lambda args: self.run_tui("key", args),
"tui_type": lambda args: self.run_tui("type", args),
"tui_click": lambda args: self.run_tui("click", args),
"tui_goto": lambda args: self.run_tui("goto", args),
"tui_result": lambda args: self.run_tui("result", args),
"fork": self.run_fork,
"poll": self.run_poll,
"ask_user": self.run_ask_user,
@@ -14110,6 +14140,14 @@ class Tools:
return "error: " + str(exc)
return "typed %d chars via %s backend at %.2fs/char" % (len(text), controller.kind, delay)
def run_tui(self, op, args):
"""Drive the running grr TUI through its control socket (see GrrControl)."""
payload = {"op": op}
payload.update({k: v for k, v in args.items() if v is not None})
if op == "screen" and args.get("until"):
payload["wait"] = (args.get("timeout") or GRR_CONTROL_WAIT) + 2
return json.dumps(grr_control_call(payload))
def run_plexer_status(self, args):
controller = PlexerControllerFactory.detect()
if controller is None:
@@ -30074,6 +30112,240 @@ class GrrPaneDeploy(GrrPaneTab):
+ [GrrButtonItem(t("deploy.button"), lambda: None, "secondary", inline=True, disabled=True), GrrNote(t("deploy.note"), center=True)])
GRR_CONTROL_JOBS = 64
GRR_CONTROL_LINE = 65536
GRR_CONTROL_WAIT = 10
GRR_CONTROL_WAIT_MAX = 60
GRR_CONTROL_KEYS_MAX = 200
def grr_control_path():
"""The control socket of the running grr TUI: TAI_TUI_SOCK or <TAI_HOME>/tui.sock."""
home = os.environ.get("TAI_HOME") or os.path.join(os.path.expanduser("~"), ".tai")
return os.environ.get("TAI_TUI_SOCK") or os.path.join(home, "tui.sock")
class GrrControl:
"""Lets agents drive the running grr TUI over a local unix socket (mode
0600), one JSON line in and one out. An operation is a job on the UI
loop: `screen`, `key`, `type`, `click`, `goto` and `result`. Every one can
be async: the call returns a job id at once and `result` collects it
(with a wait). `screen` may wait for text to appear (`until`). At most
GRR_CONTROL_JOBS finished jobs are kept."""
def __init__(self, shell, path=None):
self.shell = shell
self.path = path or grr_control_path()
self.sock = None
self.jobs = collections.OrderedDict()
self.seq = 0
self.lock = threading.Lock()
self.thread = None
self.stopped = False
def start(self):
if os.path.exists(self.path):
probe = socket.socket(socket.AF_UNIX)
probe.settimeout(1)
try:
probe.connect(self.path)
probe.close()
return False
except OSError:
os.unlink(self.path)
os.makedirs(os.path.dirname(self.path), exist_ok=True)
self.sock = socket.socket(socket.AF_UNIX)
self.sock.bind(self.path)
os.chmod(self.path, 0o600)
self.sock.listen(8)
self.thread = threading.Thread(target=self.accept, name="grr-control", daemon=True)
self.thread.start()
return True
def stop(self):
self.stopped = True
if self.sock is not None:
try:
self.sock.close()
except OSError:
pass
try:
os.unlink(self.path)
except OSError:
pass
self.sock = None
def accept(self):
while not self.stopped:
try:
conn, _addr = self.sock.accept()
except OSError:
return
threading.Thread(target=self.serve, args=(conn,), daemon=True).start()
def serve(self, conn):
try:
conn.settimeout(GRR_CONTROL_WAIT_MAX + 5)
data = b""
while b"\n" not in data and len(data) < GRR_CONTROL_LINE:
chunk = conn.recv(4096)
if not chunk:
break
data += chunk
try:
request = json.loads(data.decode("utf-8"))
reply = self.handle(request)
except (ValueError, TypeError) as exc:
reply = {"state": "error", "error": "bad request: %s" % exc}
conn.sendall((json.dumps(reply) + "\n").encode("utf-8"))
except OSError:
pass
finally:
conn.close()
def handle(self, request):
op = str(request.get("op") or "")
wait = max(0.0, min(float(request.get("wait", GRR_CONTROL_WAIT)), GRR_CONTROL_WAIT_MAX))
if op == "result":
with self.lock:
job = self.jobs.get(int(request.get("id", 0)))
if job is None:
return {"state": "error", "error": "no such job"}
else:
if not hasattr(self, "op_" + op):
return {"state": "error", "error": "unknown op %r" % op}
job = self.submit(op, request)
if request.get("async") and op != "result":
return {"id": job["id"], "state": job["state"]}
job["event"].wait(wait)
return self.view(job)
def view(self, job):
out = {"id": job["id"], "state": job["state"]}
if job["state"] == "done":
out["result"] = job["result"]
elif job["state"] == "error":
out["error"] = job["error"]
return out
def submit(self, op, request):
with self.lock:
self.seq += 1
job = {"id": self.seq, "op": op, "state": "running", "event": threading.Event(), "result": None, "error": ""}
self.jobs[job["id"]] = job
while len(self.jobs) > GRR_CONTROL_JOBS:
self.jobs.popitem(last=False)
self.shell.app.run_on_loop(lambda: self.run(job, request))
return job
def finish(self, job, result=None, error=""):
job["result"], job["error"] = result, error
job["state"] = "error" if error else "done"
job["event"].set()
def run(self, job, request):
try:
outcome = getattr(self, "op_" + job["op"])(job, request)
except Exception as exc:
self.finish(job, error="%s: %s" % (type(exc).__name__, exc))
return
if outcome is not self.PENDING:
self.finish(job, outcome)
PENDING = object()
def screen_info(self):
app, shell = self.shell.app, self.shell
app.layout()
focus = app.focused
return {"route": shell.router.current, "cols": app.cols, "rows": app.rows, "signed_in": shell.signed_in,
"focus": type(focus).__name__ if focus is not None else "", "screen": app.render_text()}
def op_screen(self, job, request):
until = request.get("until")
if not until:
return self.screen_info()
deadline = time.time() + max(0.0, min(float(request.get("timeout", GRR_CONTROL_WAIT)), GRR_CONTROL_WAIT_MAX))
def look():
info = self.screen_info()
if until in info["screen"]:
self.finish(job, dict(info, matched=True))
elif time.time() >= deadline:
self.finish(job, dict(info, matched=False))
else:
self.shell.app.after(0.2, look)
look()
return self.PENDING
@staticmethod
def key_event(name):
parts = name.split("+")
base = parts[-1]
mods = parts[:-1]
char = {"space": " ", "enter": "\n", "tab": "\t"}.get(base, base if len(base) == 1 else "")
if mods and len(base) == 1 and "ctrl" in mods:
char = chr(ord(base.lower()) & 0x1f) if base.isalpha() else ""
return TuiKey(key=name.lower() if len(base) > 1 or mods else base, char=char, modifiers=mods)
def op_key(self, job, request):
keys = request.get("keys") or []
if not isinstance(keys, list) or len(keys) > GRR_CONTROL_KEYS_MAX:
raise ValueError("keys must be a list of at most %d names" % GRR_CONTROL_KEYS_MAX)
for name in keys:
self.shell.app.dispatch(self.key_event(str(name)))
self.shell.app.wake()
return {"sent": len(keys)}
def op_type(self, job, request):
text = str(request.get("text") or "")[:GRR_CONTROL_KEYS_MAX * 20]
for ch in text:
name = "enter" if ch == "\n" else ("space" if ch == " " else ch)
self.shell.app.dispatch(TuiKey(key=name, char="\n" if ch == "\n" else ch, modifiers=[]))
self.shell.app.wake()
return {"typed": len(text)}
def op_click(self, job, request):
x, y = int(request["x"]), int(request["y"])
app = self.shell.app
app.layout()
app.dispatch(TuiClick(x=x, y=y, button=0, modifiers=[], pressed=True))
app.dispatch(TuiClick(x=x, y=y, button=0, modifiers=[], pressed=False))
app.wake()
return {"clicked": [x, y]}
def op_goto(self, job, request):
route = str(request.get("route") or "/")
if not route.startswith("/"):
raise ValueError("a route starts with /")
self.shell.router.go(route)
return {"route": self.shell.router.current}
def grr_control_call(payload, path=None, timeout=GRR_CONTROL_WAIT_MAX + 10):
"""Client side: one request to the running TUI, its reply as a dict."""
path = path or grr_control_path()
sock = socket.socket(socket.AF_UNIX)
sock.settimeout(timeout)
try:
sock.connect(path)
sock.sendall((json.dumps(payload) + "\n").encode("utf-8"))
data = b""
while not data.endswith(b"\n"):
chunk = sock.recv(65536)
if not chunk:
break
data += chunk
except OSError as exc:
return {"state": "error", "error": "no running grr TUI answers on %s (%s)" % (path, exc)}
finally:
sock.close()
try:
return json.loads(data.decode("utf-8"))
except ValueError:
return {"state": "error", "error": "unreadable reply"}
GRR_ROUTES = (
("/bots", "GrrBotsView"), ("/bot/:id", "GrrProfileView"), ("/bot/:id/instructions", "GrrInstructionsView"), ("/bot/:id/memories", "GrrMemoriesView"),
("/memory/:id", "GrrMemoryView"), ("/bot/:id/skills", "GrrSkillsView"), ("/skill/:id", "GrrSkillView"), ("/bot/:id/computer", "GrrComputerView"),
@@ -30108,6 +30380,11 @@ def grr_tui():
app = GrrTuiApp()
app.rows, app.cols = app._winsize()
shell = GrrShell(app)
control = GrrControl(shell)
try:
control.start()
except OSError as exc:
print("tui control socket unavailable: %s" % exc, file=sys.stderr)
try:
app.mount(shell.root)
shell.boot()
@@ -30116,6 +30393,7 @@ def grr_tui():
print("tui unavailable: %s" % exc, file=sys.stderr)
return 2
finally:
control.stop()
shell.stop_stream()
test_tai.py
+107
View File
@@ -11606,6 +11606,113 @@ class GrrShellTests(GrrThemeIsolation, unittest.TestCase):
self.assertFalse(view.pane.is_open)
class GrrControlTests(GrrThemeIsolation, unittest.TestCase):
"""Agents drive the running TUI over its unix socket, headless here."""
@classmethod
def setUpClass(cls):
cls.server = GrrFakeServer()
@classmethod
def tearDownClass(cls):
cls.server.close()
def setUp(self):
GrrThemeIsolation.setUp(self)
self.tmp = tempfile.mkdtemp()
session = tai.GrrSession(os.path.join(self.tmp, "g.json"))
session.url = self.server.url
session.accounts = [{"token": "tok", "username": "r", "email": ""}]
session.active = "tok"
self.app = tai.GrrTuiApp()
self.app.rows, self.app.cols = 40, 120
self.shell = tai.GrrShell(self.app, session)
self.shell.client.base = self.server.url
self.app.mount(self.shell.root)
with mock.patch.object(tai.GrrStream, "start", lambda self: None):
self.shell.boot()
self.path = os.path.join(self.tmp, "tui.sock")
self.control = tai.GrrControl(self.shell, self.path)
self.assertTrue(self.control.start())
self.addCleanup(self.control.stop)
def call(self, **payload):
return tai.grr_control_call(payload, path=self.path, timeout=10)
def test_the_socket_is_private_and_a_second_tui_does_not_take_it(self):
self.assertEqual(oct(os.stat(self.path).st_mode & 0o777), "0o600")
self.assertFalse(tai.GrrControl(self.shell, self.path).start())
def test_screen_goto_type_and_keys_drive_the_real_screen(self):
reply = self.call(op="screen")
self.assertEqual(reply["state"], "done")
self.assertIn("Automations", reply["result"]["screen"])
self.assertEqual(self.call(op="goto", route="/chat/new")["result"]["route"], "/chat/new")
self.assertEqual(self.call(op="type", text="hello box")["result"]["typed"], 9)
shown = self.call(op="screen")["result"]
self.assertIn("hello box", shown["screen"])
self.assertEqual(shown["focus"], "GrrComposerBox")
self.call(op="goto", route="/settings")
self.call(op="key", keys=["down", "down", "down", "enter"])
self.assertTrue(self.server.seen("PATCH", "/api/me/settings"))
def test_async_calls_return_an_id_and_result_collects_it(self):
job = self.call(op="goto", route="/library", **{"async": True})
self.assertIn("id", job)
done = self.call(op="result", id=job["id"], wait=5)
self.assertEqual((done["state"], done["result"]["route"]), ("done", "/library"))
self.assertEqual(self.call(op="result", id=9999)["state"], "error")
def test_screen_waits_for_text_and_reports_when_it_never_comes(self):
stop = threading.Event()
def pump():
while not stop.is_set():
self.app._fire_timers(time.time())
time.sleep(0.02)
thread = threading.Thread(target=pump, daemon=True)
thread.start()
self.addCleanup(thread.join)
self.addCleanup(stop.set)
self.call(op="goto", route="/settings")
self.assertTrue(self.call(op="screen", until="settings.auto_review", timeout=3)["result"]["matched"])
self.assertFalse(self.call(op="screen", until="no such words", timeout=1)["result"]["matched"])
def test_clicks_land_on_cells(self):
self.call(op="goto", route="/")
rows = self.call(op="screen")["result"]["screen"].split("\n")
y = next(i for i, line in enumerate(rows) if "Library" in line)
self.call(op="click", x=rows[y].index("Library"), y=y)
self.assertEqual(self.shell.router.current, "/library")
def test_bad_requests_are_errors_not_crashes(self):
self.assertEqual(self.call(op="nope")["state"], "error")
self.assertEqual(self.call(op="goto", route="nowhere")["state"], "error")
self.assertEqual(self.call(op="key", keys=["a"] * (tai.GRR_CONTROL_KEYS_MAX + 1))["state"], "error")
self.assertEqual(self.call(op="click", x="a", y=1)["state"], "error")
def test_finished_jobs_are_capped(self):
for _ in range(tai.GRR_CONTROL_JOBS + 5):
self.call(op="goto", route="/")
self.assertEqual(len(self.control.jobs), tai.GRR_CONTROL_JOBS)
def test_no_tui_means_an_error_reply_and_no_tools(self):
reply = tai.grr_control_call({"op": "screen"}, path=os.path.join(self.tmp, "none.sock"))
self.assertEqual(reply["state"], "error")
with mock.patch.dict(os.environ, {"TAI_TUI_SOCK": os.path.join(self.tmp, "none.sock")}):
self.assertFalse(tai.tool_env_ok("tui_screen"))
with mock.patch.dict(os.environ, {"TAI_TUI_SOCK": self.path}):
self.assertTrue(tai.tool_env_ok("tui_screen"))
def test_the_agent_tool_talks_to_the_socket(self):
with mock.patch.dict(os.environ, {"TAI_TUI_SOCK": self.path}):
tools = tai.Tools.__new__(tai.Tools)
out = json.loads(tai.Tools.run_tui(tools, "goto", {"route": "/projects"}))
self.assertEqual(out["result"]["route"], "/projects")
for name in ("tui_screen", "tui_keys", "tui_type", "tui_click", "tui_goto", "tui_result"):
self.assertTrue(any(sc["function"]["name"] == name for sc in tai.TOOL_SCHEMAS), name)
class GrrSpecTests(unittest.TestCase):
"""grrtui.md is the specification; it must list exactly what the code has."""