diff --git a/README.md b/README.md index abcd6b3..ca50e1b 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,11 @@ as an installable web app (PWA): - every rig and seat, with **working / idle / waiting for an answer** - open and recently finished queue items - the latest commits on every branch of the repositories your teams work in -- tap a seat to see its screen (refreshes every 5 s), answer a prompt with **1 / 2 / 3**, or type a message to it +- **questions from the agents, answered right on the page:** + - menus and permission prompts are read off the seat's screen and shown as labelled buttons (plus Esc) + - an agent that stopped with a question gets a reply box + - queue items blocked on a human seat are decisions, recorded with `rig queue resolve` so the owner wakes up +- tap a seat to see its screen (refreshes every 5 s) or type a message to it - optional push notifications through [ntfy](https://ntfy.sh): new commits, a seat waiting for an answer (only if it is still waiting on the next poll), a new request to the operator, a lead's task finished @@ -37,6 +41,8 @@ python3 -m unittest # tests | `NTFY_URL` / `NTFY_TOPIC` / `NTFY_TOKEN` | (none) | ntfy server, topic and access token; unset = no push | | `RIG_POCKET_DONE_DEST` | `^orch-lead@` | regex: notify when a task to such a seat closes | | `RIG_POCKET_HUMAN_DEST` | `^operator-agent@kernel$` | regex: notify when such a seat gets a new task | +| `RIG_POCKET_HUMAN_SEATS` | `^operator-human@` | regex: human seats (not scanned for prompts; items blocked on them are decisions) | +| `RIG_POCKET_HUMAN_SEAT` | `operator-human@kernel` | the seat a decision is recorded as | ## Security diff --git a/server.py b/server.py index c1621cc..5768488 100644 --- a/server.py +++ b/server.py @@ -34,15 +34,67 @@ NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "") DONE_DEST = re.compile(os.environ.get("RIG_POCKET_DONE_DEST", r"^orch-lead@")) HUMAN_DEST = re.compile(os.environ.get("RIG_POCKET_HUMAN_DEST", r"^operator-agent@kernel$")) -PROMPT = re.compile(r"Do you want to proceed\?|^\s*❯ 1\. ", re.M) +HUMAN_SEATS = re.compile(os.environ.get("RIG_POCKET_HUMAN_SEATS", r"^operator-human@")) +# the seat a decision is recorded as (rig derives the resolver from the seat environment) +HUMAN_ENV = {"OPENRIG_SESSION_NAME": os.environ.get("RIG_POCKET_HUMAN_SEAT", "operator-human@kernel")} + SEAT = re.compile(r"^[A-Za-z0-9._-]+@[A-Za-z0-9._-]+$") -KEYS = {"1", "2", "3", "4"} +QITEM = re.compile(r"^qitem-[A-Za-z0-9-]+$") +KEYS = {str(n) for n in range(1, 10)} | {"esc"} CLOSED = {"done", "closed", "failed", "canceled", "cancelled"} +OPTION = re.compile(r"^\s*(❯\s*)?(\d)\.\s+(\S.*?)\s*$") +FRAME = re.compile(r"^[\s│┃|╌─━┄┈╭╮╰╯]+|[\s│┃|╌─━┄┈╭╮╰╯]+$") +BULLET = re.compile(r"^\s*●\s?") +INPUT_RULE = re.compile(r"^\s*─{8,}") -def rig(*args, timeout=20): + +def clean(line): + return FRAME.sub("", line).strip() + + +def parse_ask(screen, idle): + """What a seat is waiting for, read off its screen. + + menu: a numbered choice with the cursor (❯) on it (permission prompts, multiple choice questions) + question: the agent stopped and its last message ends in a question + """ + lines = [l for l in screen.splitlines() if not l.startswith("[rig capture:")] + tail = lines[-30:] + opts = [(i, OPTION.match(l)) for i, l in enumerate(tail)] + opts = [(i, m) for i, m in opts if m] + if opts and any(m.group(1) for _, m in opts): + above = [clean(l) for l in tail[max(0, opts[0][0] - 8):opts[0][0]]] + above = [t for t in above if t and not t.startswith(("Tip:", "Esc to"))] + # the question line if there is one ("Do you want to proceed?"), else the nearest heading + title = next((t for t in reversed(above) if t.endswith("?")), above[-1] if above else "") + options = [{"key": m.group(2), "label": m.group(3)} for _, m in opts] + return {"kind": "menu", "title": title, "options": options} + + if not idle: + return None + # the last agent message: from the last "●" line down to the input box rule + end = next((i for i in range(len(lines) - 1, -1, -1) if INPUT_RULE.match(lines[i])), len(lines)) + end = next((i for i in range(end - 1, -1, -1) if INPUT_RULE.match(lines[i])), end) + start = next((i for i in range(end - 1, -1, -1) if BULLET.match(lines[i])), None) + if start is None: + return None + body = [lines[start]] + for l in lines[start + 1:end]: + if l and not l[0].isspace(): # status line (✻ …) or the input box: the message has ended + break + body.append(l) + text = " ".join(BULLET.sub("", l).strip() for l in body if l.strip()) + text = re.sub(r"\s+", " ", text).strip() + if text.endswith("?") and not text.startswith(("Bash(", "Read(", "Edit(", "Write(")): + return {"kind": "question", "text": text[-600:]} + return None + + +def rig(*args, timeout=20, env=None): try: - out = subprocess.run([RIG, *args], capture_output=True, text=True, timeout=timeout) + out = subprocess.run([RIG, *args], capture_output=True, text=True, timeout=timeout, + env={**os.environ, **env} if env else None) return out.stdout except (OSError, subprocess.TimeoutExpired): return "" @@ -87,22 +139,30 @@ def collect(): seats = [] for n in as_list(rig_json("ps", "--nodes", "--rig", r["name"], "--json")): name = n.get("canonicalSessionName", "") - screen = rig("capture", name, "--lines", "25") if name else "" + activity = (n.get("agentActivity") or {}).get("state", "unknown") + # human seats are dashboards, not agents: nothing on them is a question for the person + human = bool(HUMAN_SEATS.search(name)) + screen = rig("capture", name, "--lines", "40") if name and not human else "" + ask = parse_ask(screen, activity == "idle") if screen else None seats.append({ "name": name, - "activity": (n.get("agentActivity") or {}).get("state", "unknown"), + "activity": activity, "status": n.get("sessionStatus"), "work": n.get("assignedWorkCount", 0), "lastActivity": n.get("lastActivity"), - "waiting": bool(PROMPT.search(screen)), + "waiting": ask is not None, + "ask": ask, }) rigs.append({"name": r["name"], "status": r.get("status"), "uptime": r.get("uptime"), "attention": r.get("attentionCount", 0), "seats": seats}) queue = [] for q in as_list(rig_json("queue", "list", "-A", "-a", "--limit", "60", "--json")): - queue.append({"id": q.get("qitemId"), "state": q.get("state"), "dest": q.get("destinationSession"), - "source": q.get("sourceSession"), "summary": q.get("summary"), "updated": q.get("tsUpdated")}) + dest = q.get("destinationSession") or "" + queue.append({"id": q.get("qitemId"), "state": q.get("state"), "dest": dest, + "source": q.get("sourceSession"), "summary": q.get("summary"), "updated": q.get("tsUpdated"), + # a human gate: blocked on a human seat until someone records a decision (rig queue resolve) + "gate": q.get("state") == "blocked" and bool(HUMAN_SEATS.search(q.get("blockedOn") or ""))}) repos = [] for p in repo_paths(): @@ -140,6 +200,7 @@ class Watcher: self.waiting_polls = {} self.notified_waiting = set() self.queue_state = None + self.notified_gates = set() def events(self, snap): out = [] @@ -165,6 +226,11 @@ class Watcher: out.append((f"Opgave {q['state']}: {q['dest']}", q["summary"] or k, 4, "white_check_mark")) self.queue_state = {k: q["state"] for k, q in queue.items()} + for k, q in queue.items(): + if q.get("gate") and k not in self.notified_gates: + out.append(("Beslutning til dig", q["summary"] or k, 4, "raising_hand")) + self.notified_gates = {k for k, q in queue.items() if q.get("gate")} + # a prompt must survive two polls before it pings, so prompts answered quickly stay quiet for r in snap["rigs"]: for s in r["seats"]: @@ -172,12 +238,20 @@ class Watcher: self.waiting_polls[n] = self.waiting_polls.get(n, 0) + 1 if s["waiting"] else 0 if self.waiting_polls[n] >= 2 and n not in self.notified_waiting: self.notified_waiting.add(n) - out.append((f"{n} venter på svar", "Åbn for at se skærmen og svare.", 4, "warning")) + out.append((f"{n} venter på svar", ask_text(s.get("ask")), 4, "warning")) if not s["waiting"]: self.notified_waiting.discard(n) return out +def ask_text(ask): + if not ask: + return "Åbn for at se skærmen og svare." + if ask["kind"] == "menu": + return (ask["title"] + "\n" + "\n".join(f"{o['key']}. {o['label']}" for o in ask["options"])).strip() + return ask["text"] + + STATE = {"snap": None} LOCK = threading.Lock() @@ -202,6 +276,12 @@ def known_seats(): return {s["name"] for r in snap["rigs"] for s in r["seats"]} +def open_gates(): + with LOCK: + snap = STATE["snap"] or {"queue": []} + return {q["id"] for q in snap["queue"] if q.get("gate")} + + TYPES = {".html": "text/html; charset=utf-8", ".js": "text/javascript; charset=utf-8", ".webmanifest": "application/manifest+json", ".svg": "image/svg+xml", ".png": "image/png", ".json": "application/json"} @@ -253,7 +333,7 @@ class Handler(BaseHTTPRequestHandler): self.send(200, data, TYPES.get(f.suffix, "application/octet-stream"), cache) def do_POST(self): - if self.path != "/api/send": + if self.path not in ("/api/send", "/api/resolve"): return self.send(404, '{"error":"not found"}') # a custom header forces a CORS preflight, so other sites cannot post here if self.headers.get("X-Rig-Pocket") != "1" or (ORIGIN and self.headers.get("Origin") != ORIGIN): @@ -262,11 +342,21 @@ class Handler(BaseHTTPRequestHandler): body = json.loads(self.rfile.read(min(int(self.headers.get("Content-Length", 0)), 8192))) except (ValueError, json.JSONDecodeError): return self.send(400, '{"error":"bad json"}') - seat, text, key = body.get("seat", ""), (body.get("text") or "").strip(), body.get("key") + text = (body.get("text") or "").strip() + if self.path == "/api/resolve": + qid = body.get("id", "") + if not QITEM.match(qid) or qid not in open_gates(): + return self.send(404, '{"error":"unknown decision"}') + if not text or len(text) > 2000: + return self.send(400, '{"error":"empty decision"}') + out = rig("queue", "resolve", qid, "--decision", text, env=HUMAN_ENV) + return self.send(200, json.dumps({"ok": True, "result": out.strip()[:500]})) + seat, key = body.get("seat", ""), body.get("key") if not SEAT.match(seat) or seat not in known_seats(): return self.send(404, '{"error":"unknown seat"}') if key in KEYS: - out = rig("send", seat, key, "--dangerously-interact", "--reason", "rig-pocket: answered from phone") + keys = "\x1b" if key == "esc" else key + out = rig("send", seat, keys, "--dangerously-interact", "--reason", "rig-pocket: answered from phone") elif text and len(text) <= 2000: out = rig("send", seat, f"From the person (via rig-pocket on their phone): {text}") else: diff --git a/static/app.js b/static/app.js index 5224d8b..db9aa73 100644 --- a/static/app.js +++ b/static/app.js @@ -41,13 +41,63 @@ async function load() { } } +async function post(path, payload) { + const r = await fetch(path, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'X-Rig-Pocket': '1' }, + body: JSON.stringify(payload), + }); + return r.ok; +} + +// A reply box: textarea + button. onSend gets the text and returns a promise of success. +function replyBox(placeholder, label, onSend) { + const input = $('textarea', { class: 'wt-input', rows: '2', placeholder }); + const status = $('span', { class: 'wt-muted' }); + const go = async () => { + const text = input.value.trim(); + if (!text) return; + status.textContent = 'Sender…'; + if (await onSend(text)) { input.value = ''; status.textContent = 'Sendt'; setTimeout(load, 1500); } else status.textContent = 'Fejl'; + }; + return $('div', { class: 'wt-stack' }, input, + $('div', { class: 'wt-row wt-row--between' }, status, $('button', { class: 'wt-btn wt-btn--primary', onclick: go }, label))); +} + +// What a seat is asking, with the controls to answer it. +function askBlock(s, after) { + const a = s.ask; + if (!a) return null; + const done = (ok) => { if (after) after(); setTimeout(load, 1500); return ok; }; + if (a.kind === 'menu') { + return $('div', { class: 'wt-stack' }, + a.title ? $('span', {}, a.title) : null, + $('div', { class: 'wt-stack' }, + a.options.map((o) => $('button', { class: `wt-btn ${o.key === '1' ? 'wt-btn--primary' : ''}`, + onclick: async (e) => { e.target.disabled = true; done(await post('api/send', { seat: s.name, key: o.key })); } }, + `${o.key}. ${o.label}`)), + $('button', { class: 'wt-btn', onclick: async () => done(await post('api/send', { seat: s.name, key: 'esc' })) }, 'Esc (annullér)'))); + } + return $('div', { class: 'wt-stack' }, + $('blockquote', { class: 'wt-muted' }, a.text), + replyBox('Dit svar…', 'Svar', async (text) => done(await post('api/send', { seat: s.name, text })))); +} + function overview() { const view = []; const waiting = snap.rigs.flatMap((r) => r.seats.filter((s) => s.waiting)); - if (waiting.length) { + const gates = snap.queue.filter((q) => q.gate); + if (waiting.length || gates.length) { view.push($('section', { class: 'wt-card wt-card--raised wt-stack' }, $('h2', { class: 'wt-card-title' }, 'Venter på dig'), - waiting.map((s) => $('a', { class: 'wt-row wt-row--between', href: `#/seat/${s.name}` }, s.name, seatTag(s))))); + gates.map((q) => $('div', { class: 'wt-stack' }, + $('div', { class: 'wt-row wt-row--between' }, $('span', { class: 'wt-tag wt-tag--warn' }, 'beslutning'), + $('span', { class: 'wt-muted' }, `${q.source} · ${ago(q.updated)}`)), + $('span', {}, q.summary || q.id), + replyBox('Din beslutning…', 'Send beslutning', (text) => post('api/resolve', { id: q.id, text })))), + waiting.map((s) => $('div', { class: 'wt-stack' }, + $('a', { class: 'wt-row wt-row--between', href: `#/seat/${s.name}` }, $('strong', {}, s.name), seatTag(s)), + askBlock(s))))); } for (const r of snap.rigs) { @@ -87,15 +137,6 @@ function overview() { return view; } -async function send(seat, payload) { - const r = await fetch('api/send', { - method: 'POST', - headers: { 'Content-Type': 'application/json', 'X-Rig-Pocket': '1' }, - body: JSON.stringify({ seat, ...payload }), - }); - return r.ok; -} - function seatView(name) { const s = snap.rigs.flatMap((r) => r.seats).find((x) => x.name === name); const pre = $('pre', { class: 'wt-pre' }, 'Henter skærm…'); @@ -108,23 +149,12 @@ function seatView(name) { clearInterval(timer); timer = setInterval(refresh, 5000); - const input = $('textarea', { class: 'wt-input', rows: '3', placeholder: `Skriv til ${name}…` }); - const status = $('span', { class: 'wt-muted' }); - const doSend = async (payload) => { - status.textContent = 'Sender…'; - status.textContent = (await send(name, payload)) ? 'Sendt' : 'Fejl'; - setTimeout(refresh, 1200); - }; return [ $('a', { href: '#/', class: 'wt-muted' }, '← Oversigt'), $('div', { class: 'wt-row wt-row--between' }, $('h2', {}, name), s ? seatTag(s) : null), + s && s.ask && s.ask.kind === 'menu' ? $('section', { class: 'wt-card wt-card--raised' }, askBlock(s, () => setTimeout(refresh, 1200))) : null, pre, - s && s.waiting ? $('div', { class: 'wt-row' }, - $('span', { class: 'wt-muted' }, 'Svar på prompten:'), - ['1', '2', '3'].map((k) => $('button', { class: 'wt-btn', onclick: () => doSend({ key: k }) }, k))) : null, - $('div', { class: 'wt-stack' }, input, - $('div', { class: 'wt-row wt-row--between' }, status, - $('button', { class: 'wt-btn wt-btn--primary', onclick: () => { if (input.value.trim()) { doSend({ text: input.value }); input.value = ''; } } }, 'Send'))), + replyBox(`Skriv til ${name}…`, 'Send', async (text) => { const ok = await post('api/send', { seat: name, text }); setTimeout(refresh, 1200); return ok; }), ]; } @@ -133,8 +163,10 @@ function render() { document.getElementById('updated').textContent = ago(snap.updated); const m = location.hash.match(/^#\/seat\/(.+)$/); const view = document.getElementById('view'); - // keep the open seat view (and anything typed) while the overview data refreshes + // keep the open seat view, and never wipe an answer being typed if (m && view.dataset.seat === m[1]) return; + const typing = [...view.querySelectorAll('textarea')].some((t) => t.value || t === document.activeElement); + if (!m && !view.dataset.seat && typing) return; view.dataset.seat = m ? m[1] : ''; if (!m) clearInterval(timer); view.replaceChildren(...(m ? seatView(decodeURIComponent(m[1])) : overview()).filter(Boolean)); diff --git a/test_server.py b/test_server.py index 249bb10..03d57a7 100644 --- a/test_server.py +++ b/test_server.py @@ -50,9 +50,91 @@ class WatcherTest(unittest.TestCase): w.events(snap(waiting=True)) self.assertEqual(len(w.events(snap(waiting=True))), 1) - def test_prompt_pattern(self): - self.assertTrue(server.PROMPT.search(" Do you want to proceed?\n ❯ 1. Yes")) - self.assertFalse(server.PROMPT.search("❯ yes, launch it")) + def test_gate_notifies_once(self): + w = server.Watcher() + w.events(snap()) + gate = {**q("7", "blocked", "operator-human@kernel"), "gate": True} + self.assertEqual(w.events(snap(queue=[gate]))[-1][0], "Beslutning til dig") + self.assertEqual(w.events(snap(queue=[gate])), []) + + +PERMISSION = """[rig capture: operator-agent@kernel: 23 earlier lines not shown] +● Bash(rig queue show qitem-1 --json) + ⎿ Waiting… + +──────────────────────────────────────────────────────────────────────────────── + Bash command + Tip: auto mode handles these prompts for you — choose "switch to auto mode" + below + Check whether the Chromium libraries are installed +╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ + │ dpkg-query -W -f='${Package} ${Status}\\n' libatk1.0-0t64 +╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ + This command requires approval + + Do you want to proceed? + ❯ 1. Yes + 2. Yes, and don’t ask again for: dpkg-query * + 3. Yes, and switch to auto mode · auto mode handles these prompts for you + 4. No + + Esc to cancel · Tab to amend +""" + +QUESTION = """ - Install location: ~/rigs/workshop. + +● Once it's running: I'll give the lead the whole task as a durable work item + (spec file, branch, local commits only, no push). When it's done, I'll report + back the commits and test results. + +● Should I launch workshop like this (all Claude, prompts off, in + /home/rig/work/wpp)? + +✻ Churned for 1m 24s · done 4:26 PM + +──────────────────────────────────────────────────────── operator-agent@kernel ─ +❯ yes, launch workshop all-claude +──────────────────────────────────────────────────────────────────────────────── + ⏵⏵ accept edits on (shift+tab to cycle) +""" + +WORKING = """● Bash(cargo test --locked) + ⎿ Running… + +✻ Crunching… (1m 7s · ↓ 3.3k tokens · thinking) + +──────────────────────────────────────────────────────── dev-build@team ─ +❯ +──────────────────────────────────────────────────────────────────────────────── +""" + + +class ParseAskTest(unittest.TestCase): + def test_permission_prompt(self): + a = server.parse_ask(PERMISSION, idle=False) + self.assertEqual(a["kind"], "menu") + self.assertEqual(a["title"], "Do you want to proceed?") + self.assertEqual([o["key"] for o in a["options"]], ["1", "2", "3", "4"]) + self.assertEqual(a["options"][1]["label"], "Yes, and don’t ask again for: dpkg-query *") + + def test_question_when_idle(self): + a = server.parse_ask(QUESTION, idle=True) + self.assertEqual(a["kind"], "question") + self.assertTrue(a["text"].startswith("Should I launch workshop")) + self.assertTrue(a["text"].endswith("?")) + + def test_question_needs_idle(self): + self.assertIsNone(server.parse_ask(QUESTION, idle=False)) + + def test_working_is_not_a_question(self): + self.assertIsNone(server.parse_ask(WORKING, idle=True)) + self.assertIsNone(server.parse_ask(WORKING, idle=False)) + + def test_numbered_list_without_cursor_is_not_a_menu(self): + self.assertIsNone(server.parse_ask("● Plan:\n 1. engine\n 2. rooms\n", idle=False)) + + def test_ask_text_for_notifications(self): + self.assertIn("2. Yes, and", server.ask_text(server.parse_ask(PERMISSION, idle=False))) def test_seat_names(self): self.assertTrue(server.SEAT.match("dev-qa@ubl-workshop"))