Trilltechnician/hack-house
1
3
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(ai): robust TUI driving — fix stale-input + agent-restart keystrokes

Browse Source 
The runner cleared the input with only 6 backspaces and fired one unverified
Enter, so a dropped keystroke left a half-typed prompt that corrupted the next
send and lingered after exit. Worse, online/grant detection counted chat events
via `capture-pane -S` — but this is a full-screen alt-screen app whose scrollback
returns stale/empty frames, so detection was unreliable and the restart loop kept
dismissing healthy-but-slow spawns into a churn cycle.

New bench/tui.py exposes verified primitives shared by the runner and a restart
CLID:
  * clear_input / submit — backspace-clear and Enter until the input box reads
    empty (the box has no line-editing; Ctrl-A/U/K arrive as literal letters)
  * capture() now reads only the VISIBLE viewport (no -S) — the live screen is
    the only trustworthy source
  * agent_online() reads the present-tense clergy roster, not scrolled-away chat
  * restart_agent() stops/starts/grants with a generous 180s online wait (cold
    /ai start reloads the model and takes 60-90s on CPU) and retries only a
    genuinely hung spawn

run.py now delegates send/clear/online-check to tui and clears the box on exit.

    python bench/tui.py restart <model>   # one-shot reliable restart+grant

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
leetcrypt
2026-06-13 09:31:47 -07:00
parent c61413c648
commit d211a75616
2 changed files with 195 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bench/run.py
+15 -26
View File
@@ -33,44 +33,28 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import tasks as T # noqa: E402
from tui import agent_online, capture, clear_input, send_command # noqa: E402
def sh(cmd, **kw):
return subprocess.run(cmd, capture_output=True, text=True, **kw)
def tmux(*args):
return sh(["tmux", *args])
def capture(target, lines=4000):
"""Full scrollback of the TUI pane as text."""
r = tmux("capture-pane", "-t", target, "-p", "-S", f"-{lines}")
return r.stdout
def podman(container, snippet):
"""Run a POSIX-sh snippet in the container; return (rc, output)."""
r = sh(["podman", "exec", container, "sh", "-c", snippet])
return r.returncode, (r.stdout + r.stderr).strip()
def agent_online(target, model):
return f"{model} (ai) online" in capture(target) or "online — ollama" in capture(target)
def is_thinking(buf):
return "is thinking" in buf
def send_task(target, model, prompt):
"""Clear the input box, then type the /ai command and submit it."""
# The TUI has no kill-line; clear with backspaces (harmless when empty).
tmux("send-keys", "-t", target, *(["BSpace"] * 6))
time.sleep(0.3)
tmux("send-keys", "-t", target, "-l", f"/ai {model} !{prompt}")
time.sleep(0.3)
tmux("send-keys", "-t", target, "Enter")
"""Robustly clear the input box, then type and submit the /ai command. The
box is fully backspace-cleared and the Enter is verified (see tui.py), so a
dropped keystroke can't leave a half-typed prompt to corrupt the next task."""
send_command(target, f"/ai {model} !{prompt}")
def scrape_summary(target, user):
@@ -197,11 +181,16 @@ def main():
print(f"running {len(selected)} task(s) against {args.model} "
f"(container={args.container}, tui={args.target})")
rows = []
for i, task in enumerate(selected, 1):
print(f"[{i}/{len(selected)}] {task.id} … ", end="", flush=True)
r = run_task(args, task)
print(f"{r['result']} ({r['secs']}s)")
rows.append(r)
try:
for i, task in enumerate(selected, 1):
print(f"[{i}/{len(selected)}] {task.id} … ", end="", flush=True)
r = run_task(args, task)
print(f"{r['result']} ({r['secs']}s)")
rows.append(r)
finally:
# Leave the input box empty so the next action (e.g. an agent restart)
# isn't corrupted by a lingering prompt.
clear_input(args.target)
print_table(rows)
bench/tui.py
+180
View File
@@ -0,0 +1,180 @@
#!/usr/bin/env python3
"""Robust primitives for driving the hack-house Rust TUI headlessly via tmux.
The TUI is a full-screen ratatui app whose message box has **no line-editing
shortcuts** — Ctrl-A/U/K arrive as the literal letters a/u/k, so the only way to
clear the input is Backspace. A single `Enter` also doesn't always register, so
every primitive here VERIFIES the on-screen input line and retries instead of
firing keystrokes blind. Shared by `run.py` (task sending) and the agent-restart
CLI below so both stop leaking stale text into the box.
Online/grant detection counts occurrences in the scrollback and waits for the
count to *increase* — never just "is the needle present" — because a previous
`/ai start <same-model>` leaves a stale "online" line that a naive check would
match immediately.
CLI:
python bench/tui.py restart <model> [--target hh-house:2]
python bench/tui.py clear [--target hh-house:2]
python bench/tui.py send '<text>' [--target hh-house:2]
"""
import argparse
import re
import subprocess
import sys
import time
DEFAULT_TARGET = "hh-house:2"
# The message box bottom line looks like: │> some text │
# Capture the text between the prompt and the box's right border.
_INPUT_RE = re.compile(r"│>\s?(.*)│\s*$")
def tmux(*args):
return subprocess.run(["tmux", *args], capture_output=True, text=True)
def capture(target, lines=None):
"""The current VISIBLE screen of the TUI pane.
Deliberately NO `-S`/scrollback: this is a full-screen alt-screen app, so
`capture-pane -S -N` returns stale or empty buffer frames (it once reported
an agent in the clergy that the live screen had already dropped). Only the
live viewport is trustworthy. The `lines` arg is accepted for call-site
compatibility but ignored."""
return tmux("capture-pane", "-t", target, "-p").stdout
def input_text(target):
"""Current contents of the message input box (best-effort, '' when empty)."""
for line in reversed(capture(target, lines=60).splitlines()):
m = _INPUT_RE.search(line)
if m:
return m.group(1).rstrip()
return ""
def clear_input(target, tries=12, burst=120):
"""Backspace the input box until it reads empty. Harmless when already empty."""
for _ in range(tries):
if not input_text(target):
return True
tmux("send-keys", "-t", target, *(["BSpace"] * burst))
time.sleep(0.25)
return not input_text(target)
def submit(target, tries=5):
"""Press Enter until the input box empties (i.e. the line was actually sent)."""
for _ in range(tries):
tmux("send-keys", "-t", target, "Enter")
time.sleep(0.6)
if not input_text(target):
return True
return not input_text(target)
def send_command(target, text):
"""Clear the box, type `text`, and submit it with verification. Returns bool."""
clear_input(target)
tmux("send-keys", "-t", target, "-l", text)
time.sleep(0.3)
return submit(target)
def _columns(target):
"""Split the two side-by-side panels. The TUI is a full-screen alt-screen
app, so `capture-pane` only ever returns the VISIBLE viewport (no
scrollback) — counting chat events is therefore unreliable, but the clergy
roster on the right is present-tense and doesn't scroll. Returns
(chat_text, clergy_text)."""
chat, clergy = [], []
for line in capture(target).splitlines():
parts = line.split("│")
if len(parts) >= 5: # │ chat ││ clergy │
chat.append(parts[1])
clergy.append(parts[3])
return "\n".join(chat), "\n".join(clergy)
def agent_online(target, model):
"""True iff `model` is currently listed in the clergy roster panel — the
one signal that reflects live state instead of scrolled-away history."""
return model in _columns(target)[1]
def wait_state(target, predicate, timeout, poll=2.0):
deadline = time.time() + timeout
while time.time() < deadline:
if predicate():
return True
time.sleep(poll)
return False
def restart_agent(target, model, online_timeout=180, start_tries=2):
"""Stop the current agent, start `model`, and grant it drive — each step
verified against the live clergy roster so a hung spawn is retried rather
than a healthy one dismissed. Returns True only if the new agent is online
AND holds drive.
`online_timeout` is deliberately generous: a cold `/ai start` reloads the
model into Ollama and routinely takes 60–90 s on the CPU-only box, so a
tight window would kill a slow-but-healthy spawn and churn it into a stuck
summon (exactly what the old aggressive retry caused)."""
send_command(target, "/ai stop")
wait_state(target, lambda: not agent_online(target, model), 20)
started = False
for _ in range(start_tries):
send_command(target, f"/ai start {model}")
if wait_state(target, lambda: agent_online(target, model), online_timeout):
started = True
break
# spawn hung — dismiss the stuck summon and try once more
send_command(target, "/ai stop")
wait_state(target, lambda: not agent_online(target, model), 20)
if not started:
return False
time.sleep(2) # let the roster settle before granting
send_command(target, "/grant ai")
granted = f"granted drive to all AI agents: {model}"
return wait_state(target, lambda: granted in _columns(target)[0], 20)
def main():
ap = argparse.ArgumentParser(description="Drive the hack-house TUI via tmux.")
ap.add_argument("action", choices=["restart", "clear", "send", "show"])
ap.add_argument("arg", nargs="?", help="model (restart) or text (send)")
ap.add_argument("--target", default=DEFAULT_TARGET, help="tmux window of the TUI")
a = ap.parse_args()
if a.action == "clear":
ok = clear_input(a.target)
print("input clear" if ok else "! input still has text")
return 0 if ok else 1
if a.action == "show":
print(repr(input_text(a.target)))
return 0
if a.action == "send":
if not a.arg:
print("send needs text", file=sys.stderr)
return 2
ok = send_command(a.target, a.arg)
print("sent" if ok else "! could not submit (box not empty)")
return 0 if ok else 1
if a.action == "restart":
if not a.arg:
print("restart needs a model", file=sys.stderr)
return 2
ok = restart_agent(a.target, a.arg)
print(f"{a.arg} online+drive" if ok else f"! {a.arg} restart failed")
return 0 if ok else 1
return 2
if __name__ == "__main__":
raise SystemExit(main())