feat(watchdog): auto-escalation של zombie-loops ל-chair — סגירת הלולאה (#218+#222)
מרכיב את הפרוסות שמוזגו לכלל closed-loop: watchdog שרת-צדי (_zombie_escalation_loop ב-lifespan, כל 15 דק') שגוזר בריאות per-issue (#222) ומסלים כל zombie מתמשך (לולאת-recovery, ≥ZOMBIE_ESCALATE_MIN_RECOVERY=2 יקיצות-שחזור) ל-chair דרך escalate_issue הloop-safe (#218). הטלמטריה agent.escalated (#219) נדלקת מה-Port. - מדוע שרת-צדי: סוכן תקוע-בלולאה לא יכול להסלים-עצמו אמין; ה-watchdog (מקביל ל-Deacon של Gastown, אבל דרך ה-reaper הקיים ולא tier חדש) עושה זאת. - אידמפוטנטי מבנייה: issue מוסלם הופך agent-unowned → לא zombie בsweep הבא. - החלטת-ההסלמה (סף+severity) = helper טהור zombie_escalation ב-agent_health (testable). severity: ≥5 יקיצות→high, אחרת→medium. - kill-switch: AGENT_ZOMBIE_AUTO_ESCALATE=0 להשבתה (הבריאות עדיין נצפית, escalate ידני עדיין זמין ב-endpoint). ברירת-מחדל ON — הלולאות שורפות budget והתרופה מוכחת. - 4 בדיקות policy חדשות (11 סה"כ). dry-run חי: 6 issues→0 hzombies→0 הסלמות (blast-radius מינימלי על deploy). Invariants: מקיים G2 (מרחיב את מנגנון-ה-reaper הקיים, לא sweep מקביל; escalate דרך המסלול הקנוני), G12 (הכל ב-web/, המגע מהשער), §6 (כשל-sweep נלכד ולא-קטלני). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -77,3 +77,34 @@ def classify_issue_health(
|
|||||||
if total_wakeups >= STALL_WAKEUP_THRESHOLD:
|
if total_wakeups >= STALL_WAKEUP_THRESHOLD:
|
||||||
return "stalled"
|
return "stalled"
|
||||||
return "idle"
|
return "idle"
|
||||||
|
|
||||||
|
|
||||||
|
# A zombie needs at least this many recovery-marker wakeups before the watchdog
|
||||||
|
# auto-escalates it — filters a single transient recovery wake from a genuine,
|
||||||
|
# budget-burning loop. Escalation reassigns the issue to the chair (via
|
||||||
|
# escalate_issue), so an escalated issue is agent-unowned next sweep and never
|
||||||
|
# re-escalates (idempotent by construction).
|
||||||
|
ZOMBIE_ESCALATE_MIN_RECOVERY = 2
|
||||||
|
|
||||||
|
|
||||||
|
def zombie_escalation(item: dict) -> tuple[str, str] | None:
|
||||||
|
"""Decide whether a health item warrants auto-escalation to the chair.
|
||||||
|
|
||||||
|
Pure (no I/O) so the watchdog's policy is unit-tested. Returns
|
||||||
|
``(severity, reason)`` for a persistent zombie, else ``None``. Severity
|
||||||
|
scales with loop intensity — a long-running loop (``>=5`` recovery wakes)
|
||||||
|
is ``high``, a fresh one ``medium``. Only ``zombie`` items escalate: a
|
||||||
|
``stalled``/``working``/``idle`` issue is not a recovery loop.
|
||||||
|
"""
|
||||||
|
if item.get("health") != "zombie":
|
||||||
|
return None
|
||||||
|
recovery = int(item.get("recovery_wakeups", 0))
|
||||||
|
if recovery < ZOMBIE_ESCALATE_MIN_RECOVERY:
|
||||||
|
return None
|
||||||
|
severity = "high" if recovery >= 5 else "medium"
|
||||||
|
reason = (
|
||||||
|
f"auto-escalation (watchdog): זוהתה לולאת-recovery — {recovery} יקיצות-שחזור "
|
||||||
|
f"על {item.get('identifier')} ({item.get('agent_name')}); ה-issue נמסר לחיים "
|
||||||
|
f"במקום להמשיך לשרוף budget."
|
||||||
|
)
|
||||||
|
return severity, reason
|
||||||
|
|||||||
50
web/app.py
50
web/app.py
@@ -124,18 +124,66 @@ async def _interaction_reaper_loop():
|
|||||||
logger.warning("interaction reaper sweep failed", exc_info=True)
|
logger.warning("interaction reaper sweep failed", exc_info=True)
|
||||||
|
|
||||||
|
|
||||||
|
# Watchdog kill-switch — set AGENT_ZOMBIE_AUTO_ESCALATE=0 to observe health
|
||||||
|
# without acting (escalation stays available manually via the endpoint).
|
||||||
|
_ZOMBIE_AUTO_ESCALATE = os.environ.get("AGENT_ZOMBIE_AUTO_ESCALATE", "1") != "0"
|
||||||
|
|
||||||
|
|
||||||
|
async def _zombie_escalation_loop():
|
||||||
|
"""Watchdog: auto-escalate persistent recovery-loop zombies to the chair (#218+#222).
|
||||||
|
|
||||||
|
Every 15 min, derive per-issue health (#222) and hand any persistent zombie
|
||||||
|
(a recovery loop, ``>=`` ``ZOMBIE_ESCALATE_MIN_RECOVERY`` recovery wakeups)
|
||||||
|
to the human via the loop-safe escalate primitive (#218) — closing the loop
|
||||||
|
that today burns budget until a human notices. Server-side on purpose: an
|
||||||
|
agent stuck in a recovery loop cannot reliably self-escalate. Idempotent —
|
||||||
|
an escalated issue is agent-unowned next sweep, so it is not re-escalated.
|
||||||
|
"""
|
||||||
|
from web.agent_health import zombie_escalation
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
await asyncio.sleep(900)
|
||||||
|
if not _ZOMBIE_AUTO_ESCALATE:
|
||||||
|
continue
|
||||||
|
health = await pc_get_agent_health()
|
||||||
|
for item in health.get("items", []):
|
||||||
|
decision = zombie_escalation(item)
|
||||||
|
if decision is None:
|
||||||
|
continue
|
||||||
|
severity, reason = decision
|
||||||
|
result = await pc_escalate_issue(item["issue_id"], severity, reason)
|
||||||
|
if result.get("ok"):
|
||||||
|
logger.warning(
|
||||||
|
"watchdog auto-escalated %s (%s, %d recovery wakeups) to chair [%s]",
|
||||||
|
item.get("identifier"), item.get("agent_name"),
|
||||||
|
item.get("recovery_wakeups", 0), severity,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
logger.warning(
|
||||||
|
"watchdog escalation failed for %s: %s",
|
||||||
|
item.get("identifier"), result.get("error"),
|
||||||
|
)
|
||||||
|
except asyncio.CancelledError:
|
||||||
|
raise
|
||||||
|
except Exception:
|
||||||
|
logger.warning("zombie escalation sweep failed", exc_info=True)
|
||||||
|
|
||||||
|
|
||||||
@asynccontextmanager
|
@asynccontextmanager
|
||||||
async def lifespan(app: FastAPI):
|
async def lifespan(app: FastAPI):
|
||||||
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
await db.init_schema()
|
await db.init_schema()
|
||||||
sync_task = asyncio.create_task(git_sync.sweep_loop())
|
sync_task = asyncio.create_task(git_sync.sweep_loop())
|
||||||
reaper_task = asyncio.create_task(_interaction_reaper_loop())
|
reaper_task = asyncio.create_task(_interaction_reaper_loop())
|
||||||
|
watchdog_task = asyncio.create_task(_zombie_escalation_loop())
|
||||||
try:
|
try:
|
||||||
yield
|
yield
|
||||||
finally:
|
finally:
|
||||||
sync_task.cancel()
|
sync_task.cancel()
|
||||||
reaper_task.cancel()
|
reaper_task.cancel()
|
||||||
for _t in (sync_task, reaper_task):
|
watchdog_task.cancel()
|
||||||
|
for _t in (sync_task, reaper_task, watchdog_task):
|
||||||
try:
|
try:
|
||||||
await _t
|
await _t
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
|
|||||||
@@ -19,7 +19,9 @@ from agent_health import ( # noqa: E402
|
|||||||
STALL_WAKEUP_THRESHOLD,
|
STALL_WAKEUP_THRESHOLD,
|
||||||
classify_issue_health,
|
classify_issue_health,
|
||||||
is_recovery_reason,
|
is_recovery_reason,
|
||||||
|
zombie_escalation,
|
||||||
)
|
)
|
||||||
|
from agent_health import ZOMBIE_ESCALATE_MIN_RECOVERY # noqa: E402
|
||||||
|
|
||||||
|
|
||||||
# ── pure classifier ────────────────────────────────────────────────────────
|
# ── pure classifier ────────────────────────────────────────────────────────
|
||||||
@@ -45,6 +47,34 @@ def test_quiet_open_issue_is_idle():
|
|||||||
assert classify_issue_health(has_live_run=False, recovery_wakeups=0, total_wakeups=1) == "idle"
|
assert classify_issue_health(has_live_run=False, recovery_wakeups=0, total_wakeups=1) == "idle"
|
||||||
|
|
||||||
|
|
||||||
|
# ── watchdog escalation policy (pure) ───────────────────────────────────────
|
||||||
|
def _item(health, recovery):
|
||||||
|
return {"health": health, "recovery_wakeups": recovery, "identifier": "CMP-141", "agent_name": "writer"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_zombie_escalation_medium_for_fresh_loop():
|
||||||
|
decision = zombie_escalation(_item("zombie", ZOMBIE_ESCALATE_MIN_RECOVERY))
|
||||||
|
assert decision is not None
|
||||||
|
severity, reason = decision
|
||||||
|
assert severity == "medium"
|
||||||
|
assert "CMP-141" in reason and "recovery" in reason.lower()
|
||||||
|
|
||||||
|
|
||||||
|
def test_zombie_escalation_high_for_long_loop():
|
||||||
|
severity, _ = zombie_escalation(_item("zombie", 5))
|
||||||
|
assert severity == "high"
|
||||||
|
|
||||||
|
|
||||||
|
def test_zombie_below_threshold_is_not_escalated():
|
||||||
|
assert zombie_escalation(_item("zombie", ZOMBIE_ESCALATE_MIN_RECOVERY - 1)) is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_non_zombie_never_escalates():
|
||||||
|
assert zombie_escalation(_item("stalled", 9)) is None
|
||||||
|
assert zombie_escalation(_item("idle", 0)) is None
|
||||||
|
assert zombie_escalation(_item("working", 9)) is None
|
||||||
|
|
||||||
|
|
||||||
def test_is_recovery_reason_markers():
|
def test_is_recovery_reason_markers():
|
||||||
assert is_recovery_reason("source_scoped_recovery_action")
|
assert is_recovery_reason("source_scoped_recovery_action")
|
||||||
assert is_recovery_reason("issue_continuation_needed")
|
assert is_recovery_reason("issue_continuation_needed")
|
||||||
|
|||||||
Reference in New Issue
Block a user