diff --git a/README.md b/README.md index ca50e1b..da1ca3f 100644 --- a/README.md +++ b/README.md @@ -11,9 +11,11 @@ as an installable web app (PWA): - 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 +- a seat whose turn has run long without producing tokens is flagged as **hung** (a stalled API stream), with an Esc button +- the container's memory use (cgroup v2) in the header, with a warning near the limit - 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 + (only if it is still waiting on the next poll), a new request to the operator, a lead's task finished, a hung seat, memory running out It reads everything through the documented `rig` CLI (`--json`), so it does not depend on daemon internals. Python 3 standard library only, no build step. @@ -42,6 +44,8 @@ python3 -m unittest # tests | `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_STALL_MIN` / `_STALL_RATE` | `15` / `50` | a turn this many minutes long, under this many tokens per minute, is hung | +| `RIG_POCKET_MEM_WARN` | `0.9` | notify when memory use passes this fraction of the limit | | `RIG_POCKET_HUMAN_SEAT` | `operator-human@kernel` | the seat a decision is recorded as | ## Security diff --git a/server.py b/server.py index 5768488..707d78f 100644 --- a/server.py +++ b/server.py @@ -53,6 +53,39 @@ def clean(line): return FRAME.sub("", line).strip() +SPINNER = re.compile(r"\((?:(\d+)h )?(?:(\d+)m )?(\d+)s · [↓↑] ([\d.]+)(k?) tokens") +STALL_MIN = float(os.environ.get("RIG_POCKET_STALL_MIN", "15")) +STALL_RATE = float(os.environ.get("RIG_POCKET_STALL_RATE", "50")) # tokens per minute +MEM_WARN = float(os.environ.get("RIG_POCKET_MEM_WARN", "0.9")) + + +def stalled(screen): + """Minutes of a turn that has gone on long while barely producing tokens (a hung API stream), else None.""" + m = None + for m in SPINNER.finditer(screen): + pass + if not m: + return None + h, mins, secs, tokens, k = m.groups() + minutes = int(h or 0) * 60 + int(mins or 0) + int(secs) / 60 + tokens = float(tokens) * (1000 if k else 1) + if minutes >= STALL_MIN and tokens / minutes < STALL_RATE: + return round(minutes) + return None + + +def memory(): + """Memory use of this container (cgroup v2), as a fraction of its limit, or None.""" + try: + cur = int(Path("/sys/fs/cgroup/memory.current").read_text()) + limit = Path("/sys/fs/cgroup/memory.max").read_text().strip() + if limit == "max": + limit = next(int(l.split()[1]) * 1024 for l in open("/proc/meminfo") if l.startswith("MemTotal")) + return {"used": cur, "limit": int(limit), "fraction": round(cur / int(limit), 3)} + except (OSError, ValueError, StopIteration): + return None + + def parse_ask(screen, idle): """What a seat is waiting for, read off its screen. @@ -152,6 +185,7 @@ def collect(): "lastActivity": n.get("lastActivity"), "waiting": ask is not None, "ask": ask, + "stalled": stalled(screen) if screen and not ask else None, }) rigs.append({"name": r["name"], "status": r.get("status"), "uptime": r.get("uptime"), "attention": r.get("attentionCount", 0), "seats": seats}) @@ -172,7 +206,7 @@ def collect(): commits.append({"hash": full, "short": short, "when": when, "refs": refs, "subject": subject}) repos.append({"name": os.path.basename(p), "commits": commits}) - return {"updated": time.time(), "rigs": rigs, "queue": queue, "repos": repos} + return {"updated": time.time(), "rigs": rigs, "queue": queue, "repos": repos, "memory": memory()} def notify(title, body, priority=3, tags=""): @@ -201,6 +235,8 @@ class Watcher: self.notified_waiting = set() self.queue_state = None self.notified_gates = set() + self.notified_stalled = set() + self.mem_warned = False def events(self, snap): out = [] @@ -241,6 +277,20 @@ class Watcher: out.append((f"{n} venter på svar", ask_text(s.get("ask")), 4, "warning")) if not s["waiting"]: self.notified_waiting.discard(n) + if s.get("stalled") and n not in self.notified_stalled: + self.notified_stalled.add(n) + out.append((f"{n} ser ud til at hænge", + f"Samme tur i {s['stalled']} min uden at producere noget. Åbn og send Esc.", 4, "hourglass")) + if not s.get("stalled"): + self.notified_stalled.discard(n) + + mem = snap.get("memory") + if mem and mem["fraction"] >= MEM_WARN and not self.mem_warned: + self.mem_warned = True + out.append(("Containeren er ved at løbe tør for hukommelse", + f"{mem['used'] / 2**30:.1f} af {mem['limit'] / 2**30:.1f} GB brugt. Agenterne kan gå i stå.", 5, "rotating_light")) + elif mem and mem["fraction"] < MEM_WARN - 0.1: + self.mem_warned = False return out diff --git a/static/app.js b/static/app.js index db9aa73..3806278 100644 --- a/static/app.js +++ b/static/app.js @@ -20,13 +20,16 @@ const ago = (iso) => { }; const seatTag = (s) => { + if (s.stalled) return $('span', { class: 'wt-tag wt-tag--bad' }, `hænger (${s.stalled} min)`); if (s.waiting) return $('span', { class: 'wt-tag wt-tag--warn' }, 'venter på svar'); if (s.status && s.status !== 'running') return $('span', { class: 'wt-tag wt-tag--bad' }, s.status); if (s.activity === 'running' || s.activity === 'working') return $('span', { class: 'wt-tag wt-tag--accent' }, 'arbejder'); return $('span', { class: 'wt-tag' }, 'ledig'); }; -const QSTATE = { 'in-progress': 'wt-tag--accent', blocked: '', parked: '', done: 'wt-tag--good', failed: 'wt-tag--bad' }; +const gb = (bytes) => (bytes / 2 ** 30).toFixed(1); + +const QSTATE ={ 'in-progress': 'wt-tag--accent', blocked: '', parked: '', done: 'wt-tag--good', failed: 'wt-tag--bad' }; let snap = null; let timer = null; @@ -67,8 +70,13 @@ function replyBox(placeholder, label, onSend) { // 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 && s.stalled) { + return $('div', { class: 'wt-stack' }, + $('span', { class: 'wt-muted' }, `Samme tur i ${s.stalled} min uden at producere noget. Esc afbryder, bagefter kan du bede den fortsætte.`), + $('button', { class: 'wt-btn', onclick: async () => done(await post('api/send', { seat: s.name, key: 'esc' })) }, 'Esc (afbryd turen)')); + } + if (!a) return null; if (a.kind === 'menu') { return $('div', { class: 'wt-stack' }, a.title ? $('span', {}, a.title) : null, @@ -85,7 +93,14 @@ function askBlock(s, after) { function overview() { const view = []; - const waiting = snap.rigs.flatMap((r) => r.seats.filter((s) => s.waiting)); + const mem = snap.memory; + if (mem && mem.fraction >= 0.9) { + view.push($('section', { class: 'wt-card wt-card--raised wt-stack' }, + $('div', { class: 'wt-row wt-row--between' }, $('h2', { class: 'wt-card-title' }, 'Hukommelse'), + $('span', { class: 'wt-tag wt-tag--bad' }, `${Math.round(mem.fraction * 100)} %`)), + $('span', {}, `${gb(mem.used)} af ${gb(mem.limit)} GB brugt. Agenterne kan gå i stå; giv containeren mere RAM eller stop et team.`))); + } + const waiting = snap.rigs.flatMap((r) => r.seats.filter((s) => s.waiting || s.stalled)); const gates = snap.queue.filter((q) => q.gate); if (waiting.length || gates.length) { view.push($('section', { class: 'wt-card wt-card--raised wt-stack' }, @@ -160,7 +175,7 @@ function seatView(name) { function render() { if (!snap) return; - document.getElementById('updated').textContent = ago(snap.updated); + document.getElementById('updated').textContent = ago(snap.updated) + (snap.memory ? ` · ${gb(snap.memory.used)}/${gb(snap.memory.limit)} GB` : ''); const m = location.hash.match(/^#\/seat\/(.+)$/); const view = document.getElementById('view'); // keep the open seat view, and never wipe an answer being typed diff --git a/test_server.py b/test_server.py index 03d57a7..3eb72a3 100644 --- a/test_server.py +++ b/test_server.py @@ -133,6 +133,26 @@ class ParseAskTest(unittest.TestCase): 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_stalled_turn(self): + self.assertEqual(server.stalled("· Ionizing… (34m 31s · ↓ 439 tokens · thinking some more)"), 35) + self.assertIsNone(server.stalled("✶ Levitating… (10m 40s · ↓ 30.1k tokens)")) # short and busy + self.assertIsNone(server.stalled("✻ Doing… (16m 3s · ↓ 3.4k tokens)")) # long but producing + self.assertEqual(server.stalled("* Thinking… (1h 2m 0s · ↓ 900 tokens)"), 62) + self.assertIsNone(server.stalled("✻ Churned for 1m 24s · done 4:26 PM")) + + def test_stall_notifies_once(self): + w = server.Watcher() + s = snap() + s["rigs"][0]["seats"][0]["stalled"] = 34 + self.assertEqual(w.events(s)[-1][0], "dev-qa@team ser ud til at hænge") + self.assertEqual(w.events(s), []) + + def test_memory_warning_once(self): + w = server.Watcher() + s = {**snap(), "memory": {"used": 95, "limit": 100, "fraction": 0.95}} + self.assertTrue(w.events(s)[-1][0].startswith("Containeren")) + self.assertEqual(w.events(s), []) + def test_ask_text_for_notifications(self): self.assertIn("2. Yes, and", server.ask_text(server.parse_ask(PERMISSION, idle=False)))