feat(health): טקסונומיית-בריאות לסוכן — zombie/stalled/working/idle (#222)

מוסיף 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 <noreply@anthropic.com>

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 = "ספריית פסיקה — תור חילוץ"