From bfc352c7b5e21d2167216f506b7508c0aadc478c Mon Sep 17 00:00:00 2001 From: Chaim Date: Tue, 7 Jul 2026 03:05:17 +0000 Subject: [PATCH] =?UTF-8?q?feat(health):=20=D7=98=D7=A7=D7=A1=D7=95=D7=A0?= =?UTF-8?q?=D7=95=D7=9E=D7=99=D7=99=D7=AA-=D7=91=D7=A8=D7=99=D7=90=D7=95?= =?UTF-8?q?=D7=AA=20=D7=9C=D7=A1=D7=95=D7=9B=D7=9F=20=E2=80=94=20zombie/st?= =?UTF-8?q?alled/working/idle=20(#222)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit מוסיף web/agent_health.py — מסווג-טהור classify_issue_health (agent-agnostic, testable) שממפה כל issue פתוח+משויך-סוכן למצב-בריאות אחד, ו-get_agent_health ב-paperclip_client שגוזר את הפרימיטיבים מ-heartbeat_runs (run חי=finished_at null בחלון) + agent_wakeup_requests (total + recovery-markers בחלון). מרכיב על אירועי #219 (אותם reasons של recovery). - zombie = פתוח+משויך-סוכן, אין run חי, יש wakeup-recovery → הstranded-child שאובחן ידנית (reference_recovery_loop_stranded_child) עולה אוטומטית. - stalled = ננער חוזר בלי progress; working = run חי; idle = המתנה בנונית. - Port: pc_get_agent_health (read-only, לא עטוף-טלמטריה לפי הכלל); endpoint GET /api/operations/agents/health (worst-first, +counts). - 7 בדיקות pytest (מסווג-טהור + fetch fake-asyncpg). אומת חי read-only מול Paperclip DB: 6 issues→idle נכון, אפס zombie. Invariants: מקיים G12 (מסווג אגנוסטי; fetch מהמעטפת; המגע מהשער), G2 (מקור-בריאות יחיד). UI ב-/operations = follow-up מגודר-שער-עיצוב (feedback_claude_design_gate). Co-Authored-By: Claude Opus 4.8 --- web/agent_health.py | 79 ++++++++++++++++++++++ web/agent_platform_port.py | 2 + web/app.py | 13 ++++ web/paperclip_client.py | 98 ++++++++++++++++++++++++++++ web/tests/test_agent_health.py | 115 +++++++++++++++++++++++++++++++++ 5 files changed, 307 insertions(+) create mode 100644 web/agent_health.py create mode 100644 web/tests/test_agent_health.py diff --git a/web/agent_health.py b/web/agent_health.py new file mode 100644 index 0000000..5b3908c --- /dev/null +++ b/web/agent_health.py @@ -0,0 +1,79 @@ +"""Agent health taxonomy (TaskMaster #222) — one crisp state per stuck issue. + +Classifies each open, agent-assigned issue into a single health state so the +stranded-child / recovery-loop cases we diagnose by hand today +([[reference_recovery_loop_stranded_child]], [[reference_paperclip_recovery_loops]]) +surface automatically in the dashboard — the ``gt feed --problems`` idea from +Gastown, grounded in *our* failure modes. + +Two layers, split for testability (same shape as web.agent_telemetry): +- :func:`classify_issue_health` is a **pure function** over three primitives + (live-run flag + wakeup counts). No DB, no platform symbols — trivially + unit-tested and platform-agnostic (INV-G12). +- The Paperclip-specific fetch that derives those primitives from + ``heartbeat_runs`` / ``agent_wakeup_requests`` lives in the shell + (``web.paperclip_client.get_agent_health``); this module only decides *how to + label* what the fetch measured, so the taxonomy survives a platform swap. + +The wakeup/recovery signals it reads are the very ones #219 telemetry now emits +(``agent.wakeup`` with recovery reasons) — this is the consumer that turns that +stream into an at-a-glance verdict. +""" +from __future__ import annotations + +# Severity order (worst first) — the dashboard sorts problems by this. +HEALTH_STATES: tuple[str, ...] = ("zombie", "stalled", "working", "idle") + +# Substrings of Paperclip wakeup ``reason`` values that mark a recovery-loop +# wake (as opposed to a genuine new task). Sourced from the recovery-loop +# forensics in memory: an issue that keeps getting these while agent-owned is a +# zombie — assigned + "active" but nothing is really executing. +RECOVERY_REASON_MARKERS: tuple[str, ...] = ( + "source_scoped_recovery", + "stranded_assigned", + "issue_continuation", + "issue_reopened", + "missing_disposition", + "successful_run_missing_state", + "waiting_on_review", +) + +# ≥ this many wakeups inside the fetch window, with no recovery marker and no +# live run, reads as "poked repeatedly but not progressing" = stalled. +STALL_WAKEUP_THRESHOLD = 3 + + +def is_recovery_reason(reason: str | None) -> bool: + """True if a wakeup ``reason`` is a recovery-loop marker (not a new task).""" + if not reason: + return False + return any(marker in reason for marker in RECOVERY_REASON_MARKERS) + + +def classify_issue_health( + *, + has_live_run: bool, + recovery_wakeups: int, + total_wakeups: int, +) -> str: + """Label one open, agent-assigned issue. Pure — no I/O. + + - ``working`` — a live run is executing right now. + - ``zombie`` — no live run but recovery-loop wakeups are firing: the issue + looks agent-owned + active yet nothing is really running (the stranded + child we chase manually). + - ``stalled`` — no live run, no recovery markers, but poked ``>=`` + :data:`STALL_WAKEUP_THRESHOLD` times in-window (repeated wakes, no + progress). + - ``idle`` — open + assigned but quiet: benign waiting. + + Called only for issues that are already open and agent-assigned; unassigned + or closed issues are not health-tracked (the fetch filters them out). + """ + if has_live_run: + return "working" + if recovery_wakeups >= 1: + return "zombie" + if total_wakeups >= STALL_WAKEUP_THRESHOLD: + return "stalled" + return "idle" diff --git a/web/agent_platform_port.py b/web/agent_platform_port.py index 600f503..9b46271 100644 --- a/web/agent_platform_port.py +++ b/web/agent_platform_port.py @@ -40,6 +40,7 @@ from web.paperclip_client import ( archive_project as pc_archive_project, create_project as pc_create_project, create_workflow_issue as pc_create_workflow_issue, + get_agent_health as pc_get_agent_health, get_agents_for_case as pc_get_agents_for_case, get_agents_for_company as pc_get_agents, get_case_issues as pc_get_case_issues, @@ -177,6 +178,7 @@ __all__ = [ "pc_reap_stale_interactions", "pc_escalate_issue", # agent-run observability + control (live view + smart management) + "pc_get_agent_health", "pc_list_live_runs", "pc_get_run_log", "pc_get_run_events", diff --git a/web/app.py b/web/app.py index c200bb1..3abc9bb 100644 --- a/web/app.py +++ b/web/app.py @@ -62,6 +62,7 @@ from web.agent_platform_port import ( pc_create_project, pc_create_workflow_issue, pc_escalate_issue, + pc_get_agent_health, pc_get_agents, pc_get_agents_for_case, pc_get_case_issues, @@ -7531,6 +7532,18 @@ async def operations_agents(): } +@app.get("/api/operations/agents/health") +async def operations_agent_health(): + """Per-issue agent health taxonomy (#222): zombie / stalled / working / idle. + + Read-only. Surfaces the stranded-child / recovery-loop cases (``zombie``) + automatically — the ``gt feed --problems`` idea grounded in our failure + modes — so they no longer need hand-querying the Paperclip DB. Consumed by + the /operations dashboard (UI is design-gated, follow-up). + """ + return await pc_get_agent_health() + + @app.get("/api/operations/agents/runs/{run_id}/log") async def operations_agent_run_log(run_id: str): """Full output log (NDJSON stream) of one heartbeat run.""" diff --git a/web/paperclip_client.py b/web/paperclip_client.py index a1aa5be..5c48968 100644 --- a/web/paperclip_client.py +++ b/web/paperclip_client.py @@ -1034,6 +1034,104 @@ async def reap_stale_interactions() -> dict: return {"ok": True, "cancelled": len(rows)} +# Windows for the agent-health taxonomy (#222). A run with no finished_at older +# than the live window is treated as dead (not "working"); wakeups are counted +# over the recovery window to spot loops. +_HEALTH_LIVE_RUN_WINDOW = "30 minutes" +_HEALTH_WAKEUP_WINDOW = "2 hours" + + +async def get_agent_health() -> dict: + """Classify every open, agent-assigned issue into a health state (#222). + + Read-only. Derives, per issue, the three primitives the pure classifier + (:func:`web.agent_health.classify_issue_health`) needs, from Paperclip's + ``heartbeat_runs`` (a live run = ``finished_at IS NULL`` within the live + window) and ``agent_wakeup_requests`` (total + recovery-marker counts within + the recovery window). Scoped to the two legal-ai companies. + + Surfaces the stranded-child / recovery-loop cases (``zombie``) automatically + instead of by hand-querying the DB. Returns items sorted worst-first. + """ + from web.agent_health import ( # local import: keep the shell → agnostic dep inward + HEALTH_STATES, + classify_issue_health, + is_recovery_reason, + ) + + company_ids = list(COMPANIES.values()) + conn = await asyncpg.connect(PAPERCLIP_DB_URL) + try: + issues = await conn.fetch( + """SELECT i.id, i.identifier, i.status, i.assignee_agent_id, + a.name AS agent_name + FROM issues i + JOIN agents a ON a.id = i.assignee_agent_id + WHERE i.company_id = ANY($1::uuid[]) + AND i.assignee_agent_id IS NOT NULL + AND i.status IN ('backlog','todo','in_progress','blocked','in_review')""", + company_ids, + ) + if not issues: + return {"ok": True, "items": [], "counts": {s: 0 for s in HEALTH_STATES}} + + agent_ids = list({str(r["assignee_agent_id"]) for r in issues}) + + live_rows = await conn.fetch( + f"""SELECT DISTINCT agent_id FROM heartbeat_runs + WHERE agent_id = ANY($1::uuid[]) + AND finished_at IS NULL + AND started_at > now() - interval '{_HEALTH_LIVE_RUN_WINDOW}'""", + agent_ids, + ) + live_agents = {str(r["agent_id"]) for r in live_rows} + + wake_rows = await conn.fetch( + f"""SELECT agent_id, payload->>'issueId' AS issue_id, reason + FROM agent_wakeup_requests + WHERE agent_id = ANY($1::uuid[]) + AND requested_at > now() - interval '{_HEALTH_WAKEUP_WINDOW}'""", + agent_ids, + ) + finally: + await conn.close() + + total_by_issue: dict[str, int] = {} + recovery_by_issue: dict[str, int] = {} + for w in wake_rows: + iid = w["issue_id"] + if not iid: + continue + total_by_issue[iid] = total_by_issue.get(iid, 0) + 1 + if is_recovery_reason(w["reason"]): + recovery_by_issue[iid] = recovery_by_issue.get(iid, 0) + 1 + + items = [] + counts = {s: 0 for s in HEALTH_STATES} + for r in issues: + iid = str(r["id"]) + state = classify_issue_health( + has_live_run=str(r["assignee_agent_id"]) in live_agents, + recovery_wakeups=recovery_by_issue.get(iid, 0), + total_wakeups=total_by_issue.get(iid, 0), + ) + counts[state] += 1 + items.append({ + "issue_id": iid, + "identifier": r["identifier"], + "status": r["status"], + "agent_id": str(r["assignee_agent_id"]), + "agent_name": r["agent_name"], + "health": state, + "wakeups": total_by_issue.get(iid, 0), + "recovery_wakeups": recovery_by_issue.get(iid, 0), + }) + + order = {s: i for i, s in enumerate(HEALTH_STATES)} + items.sort(key=lambda it: (order[it["health"]], -it["recovery_wakeups"])) + return {"ok": True, "items": items, "counts": counts} + + # Singleton project for the precedent-library extraction queue. One issue per # uploaded precedent — assigned to the CEO who runs the local-MCP extractor. _LIBRARY_PROJECT_NAME = "ספריית פסיקה — תור חילוץ" diff --git a/web/tests/test_agent_health.py b/web/tests/test_agent_health.py new file mode 100644 index 0000000..865170b --- /dev/null +++ b/web/tests/test_agent_health.py @@ -0,0 +1,115 @@ +"""Tests for #222 — agent health taxonomy (pure classifier + Paperclip fetch). + +Pure-classifier tests need no deps. The fetch test fakes an asyncpg connection +(routing the three queries by content) so it never touches the live DB. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # web/ + +from agent_health import ( # noqa: E402 + STALL_WAKEUP_THRESHOLD, + classify_issue_health, + is_recovery_reason, +) + + +# ── pure classifier ──────────────────────────────────────────────────────── +def test_live_run_is_working_regardless_of_wakeups(): + assert classify_issue_health(has_live_run=True, recovery_wakeups=9, total_wakeups=9) == "working" + + +def test_recovery_wakeup_without_live_run_is_zombie(): + # the stranded-child signature: assigned+open, nothing running, recovery firing + assert classify_issue_health(has_live_run=False, recovery_wakeups=1, total_wakeups=1) == "zombie" + + +def test_repeated_wakeups_no_recovery_is_stalled(): + assert ( + classify_issue_health( + has_live_run=False, recovery_wakeups=0, total_wakeups=STALL_WAKEUP_THRESHOLD + ) + == "stalled" + ) + + +def test_quiet_open_issue_is_idle(): + assert classify_issue_health(has_live_run=False, recovery_wakeups=0, total_wakeups=1) == "idle" + + +def test_is_recovery_reason_markers(): + assert is_recovery_reason("source_scoped_recovery_action") + assert is_recovery_reason("issue_continuation_needed") + assert is_recovery_reason("issue_reopened_via_comment") + assert not is_recovery_reason("start_workflow_8125-09-24") + assert not is_recovery_reason(None) + + +# ── Paperclip fetch (fake DB) ─────────────────────────────────────────────── +os.environ.setdefault("PAPERCLIP_DB_URL", "postgres://x:x@127.0.0.1:54329/paperclip") +pc = pytest.importorskip("paperclip_client", reason="web deps unavailable") + + +class _FakeConn: + def __init__(self, issues, live, wakeups): + self._issues, self._live, self._wakeups = issues, live, wakeups + + async def fetch(self, query, *args): + if "FROM issues" in query: + return self._issues + if "heartbeat_runs" in query: + return self._live + if "agent_wakeup_requests" in query: + return self._wakeups + raise AssertionError(f"unexpected query: {query[:40]}") + + async def close(self): + pass + + +def test_get_agent_health_classifies_and_sorts(monkeypatch): + # zombie agent A (recovery wake, no live run); working agent B (live run); + # idle agent C (quiet). + issues = [ + {"id": "iA", "identifier": "CMP-1", "status": "blocked", "assignee_agent_id": "agA", "agent_name": "writer"}, + {"id": "iB", "identifier": "CMP-2", "status": "in_progress", "assignee_agent_id": "agB", "agent_name": "analyst"}, + {"id": "iC", "identifier": "CMP-3", "status": "todo", "assignee_agent_id": "agC", "agent_name": "qa"}, + ] + live = [{"agent_id": "agB"}] + wakeups = [ + {"agent_id": "agA", "issue_id": "iA", "reason": "source_scoped_recovery_action"}, + {"agent_id": "agC", "issue_id": "iC", "reason": "start_workflow_x"}, + ] + + async def _connect(_url): + return _FakeConn(issues, live, wakeups) + + monkeypatch.setattr(pc.asyncpg, "connect", _connect) + + result = asyncio.run(pc.get_agent_health()) + assert result["ok"] is True + by_issue = {it["issue_id"]: it["health"] for it in result["items"]} + assert by_issue == {"iA": "zombie", "iB": "working", "iC": "idle"} + assert result["counts"]["zombie"] == 1 + assert result["counts"]["working"] == 1 + assert result["counts"]["idle"] == 1 + # worst-first: zombie leads + assert result["items"][0]["health"] == "zombie" + assert result["items"][0]["recovery_wakeups"] == 1 + + +def test_get_agent_health_empty(monkeypatch): + async def _connect(_url): + return _FakeConn([], [], []) + + monkeypatch.setattr(pc.asyncpg, "connect", _connect) + result = asyncio.run(pc.get_agent_health()) + assert result == {"ok": True, "items": [], "counts": {"zombie": 0, "stalled": 0, "working": 0, "idle": 0}}