feat(kanban): lifecycle-derivable hole rules (RULE_SET_VERSION 1.1.0)

hermes_cli/kanban_telemetry.py

@@ -11,7 +11,7 @@
 from pathlib import Path
 from typing import Any, Optional
 
-RULE_SET_VERSION = "1.0.0"
+RULE_SET_VERSION = "1.1.0"
 REVIEW_WINDOW_SECONDS = 48 * 60 * 60
 CADENCE_SECONDS = 12 * 60 * 60
 INTAKE_DECISION_SECONDS = 30 * 60
@@ -20,6 +20,24 @@
 FINDING_RETENTION_SECONDS = 7 * 24 * 60 * 60
 NOMINAL_BOUNDARY_MINUTE = 15
 
+# --- HEL-3113 follow-up (lifecycle-derivable holes, no HEL-3110 dependency) ---
+# A single protocol_violation is retried by the dispatcher's own bounded
+# budget (see kanban_db._PROTOCOL_VIOLATION_FAILURE_LIMIT); a SECOND one for
+# the same task/profile means the retry didn't fix it, so escalate.
+PROTOCOL_VIOLATION_REPEAT_THRESHOLD = 2
+# Two crash/timeout/spawn_failed events for the same task in-window is the
+# design's literal RETRY_THRASH trigger.
+RETRY_THRASH_FAILURE_THRESHOLD = 2
+# STALL.BLOCKED_AGED severity ladder (typed human blocks only).
+BLOCKED_AGED_MEDIUM_SECONDS = 12 * 60 * 60
+BLOCKED_AGED_HIGH_SECONDS = 24 * 60 * 60
+BLOCKED_AGED_CRITICAL_SECONDS = 48 * 60 * 60
+# Matches config default kanban.dispatch_interval_seconds; "two dispatcher
+# ticks" is the design's threshold for TODO_PROMOTABLE / READY_UNCLAIMED.
+DISPATCH_TICK_SECONDS = 60
+TODO_PROMOTABLE_THRESHOLD_SECONDS = 2 * DISPATCH_TICK_SECONDS
+READY_UNCLAIMED_THRESHOLD_SECONDS = 2 * DISPATCH_TICK_SECONDS
+
 _REQUIRED_FIELDS: dict[str, tuple[str, ...]] = {
     "linear_scoped": ("linear_issue_key", "sub_issue_keys", "cptc_estimates"),
     "intake_received": ("intake_id", "source_type", "source_ref_hash", "provenance_refs", "idempotency_key", "received_by"),
@@ -453,6 +471,129 @@ def run_review(
                 recommendation="Record a typed block kind, accountable owner, and next action at the block boundary.", evidence=[{"event_ids": [], "query_id": "Q-BLOCK-01", "fact": "Blocked task has no typed block owner."}], observed_at=end,
             ))
 
+    for task_id, task in tasks.items():
+        task_events = by_task.get(task_id, [])
+        violations = [e for e in task_events if e["kind"] == "protocol_violation"]
+        if not violations:
+            continue
+        latest = violations[-1]
+        severity = (
+            "CRITICAL"
+            if len(violations) >= PROTOCOL_VIOLATION_REPEAT_THRESHOLD
+            else "HIGH"
+        )
+        holes.append(_finding(
+            "FAILURE.PROTOCOL_VIOLATION", board_slug,
+            _subject(task_ids=[task_id], profiles=[task["assignee"]]), severity=severity,
+            evidence_state="MEASURED", title="Worker exited without kanban_complete/kanban_block",
+            owner=task["assignee"] or "OWNER.UNRESOLVED",
+            recommendation="Repair the worker/session so it always ends with a terminal kanban call; investigate repeat causes for this profile.",
+            evidence=[{"event_ids": [e["id"] for e in violations], "query_id": "Q-FAILURE-01", "fact": f"{len(violations)} protocol_violation event(s) in window for this task."}],
+            observed_at=latest["created_at"], next_expected_event="kanban_complete",
+        ))
+
+    for task_id, task in tasks.items():
+        task_events = by_task.get(task_id, [])
+        thrash_kinds = {"crashed", "timed_out", "spawn_failed"}
+        thrash_events = [e for e in task_events if e["kind"] in thrash_kinds]
+        breaker_tripped = any(e["kind"] == "gave_up" for e in task_events)
+        if len(thrash_events) < RETRY_THRASH_FAILURE_THRESHOLD and not breaker_tripped:
+            continue
+        latest = (thrash_events or task_events)[-1]
+        severity = "CRITICAL" if breaker_tripped else "HIGH"
+        holes.append(_finding(
+            "FAILURE.RETRY_THRASH", board_slug,
+            _subject(task_ids=[task_id], profiles=[task["assignee"]]), severity=severity,
+            evidence_state="MEASURED", title="Task is thrashing on repeated crash/timeout/spawn failures",
+            owner=task["assignee"] or "OWNER.UNRESOLVED",
+            recommendation="Diagnose the shared failure cause (worker, environment, or task scope) before further respawns.",
+            evidence=[{"event_ids": [e["id"] for e in thrash_events], "query_id": "Q-FAILURE-02", "fact": f"{len(thrash_events)} crash/timeout/spawn_failed event(s) in window; breaker_tripped={breaker_tripped}."}],
+            observed_at=latest["created_at"],
+        ))
+
+    for task_id, task in tasks.items():
+        if task["status"] != "blocked" or not task["block_kind"]:
+            continue
+        block_events = [e for e in by_task.get(task_id, []) if e["kind"] in {"blocked", "block_loop_detected"}]
+        if not block_events:
+            continue
+        blocked_since = block_events[-1]["created_at"]
+        age = end - blocked_since
+        if age < BLOCKED_AGED_MEDIUM_SECONDS:
+            continue
+        if age >= BLOCKED_AGED_CRITICAL_SECONDS:
+            severity = "CRITICAL"
+        elif age >= BLOCKED_AGED_HIGH_SECONDS:
+            severity = "HIGH"
+        else:
+            severity = "MEDIUM"
+        holes.append(_finding(
+            "STALL.BLOCKED_AGED", board_slug,
+            _subject(task_ids=[task_id], profiles=[task["assignee"]]), severity=severity,
+            evidence_state="MEASURED", title="Typed human block has aged without a decision",
+            owner=task["assignee"] or "OWNER.UNRESOLVED",
+            recommendation="Record a decision or unblock event for this typed block.",
+            evidence=[{"event_ids": [block_events[-1]["id"]], "query_id": "Q-STALL-01", "fact": f"Blocked for {age} seconds since {_utc(blocked_since)}."}],
+            observed_at=blocked_since,
+        ))
+
+    for task_id, task in tasks.items():
+        if task["status"] != "todo":
+            continue
+        parent_rows = conn.execute(
+            "SELECT t.status, t.completed_at FROM tasks t "
+            "JOIN task_links l ON l.parent_id = t.id WHERE l.child_id = ?",
+            (task_id,),
+        ).fetchall()
+        if not parent_rows or not all(p["status"] in ("done", "archived") for p in parent_rows):
+            continue
+        completions = [p["completed_at"] for p in parent_rows if p["completed_at"]]
+        eligible_since = max(completions) if completions else task["created_at"]
+        age = end - eligible_since
+        if age < TODO_PROMOTABLE_THRESHOLD_SECONDS:
+            continue
+        promoted_after = any(
+            e["kind"] == "promoted" and e["created_at"] > eligible_since
+            for e in by_task.get(task_id, [])
+        )
+        if promoted_after:
+            continue
+        holes.append(_finding(
+            "STALL.TODO_PROMOTABLE", board_slug,
+            _subject(task_ids=[task_id], profiles=[task["assignee"]]), severity="HIGH",
+            evidence_state="MEASURED", title="All parents terminal but task was not promoted to ready",
+            owner="OWNER.UNRESOLVED",
+            recommendation="Repair recompute_ready so parent-terminal todo tasks promote within two dispatcher ticks.",
+            evidence=[{"event_ids": [], "query_id": "Q-STALL-02", "fact": f"All parents terminal since {_utc(eligible_since)}; no promoted event followed within {TODO_PROMOTABLE_THRESHOLD_SECONDS}s."}],
+            observed_at=eligible_since, next_expected_event="promoted",
+        ))
+
+    _ready_predecessor_kinds = {"promoted", "created", "reclaimed", "crashed", "rate_limited", "timed_out", "spawn_failed"}
+    for task_id, task in tasks.items():
+        if task["status"] != "ready":
+            continue
+        task_events = by_task.get(task_id, [])
+        ready_candidates = [e["created_at"] for e in task_events if e["kind"] in _ready_predecessor_kinds]
+        ready_since = max(ready_candidates) if ready_candidates else task["created_at"]
+        age = end - ready_since
+        if age < READY_UNCLAIMED_THRESHOLD_SECONDS:
+            continue
+        claimed_after = any(
+            e["kind"] == "claimed" and e["created_at"] > ready_since
+            for e in task_events
+        )
+        if claimed_after:
+            continue
+        holes.append(_finding(
+            "STALL.READY_UNCLAIMED", board_slug,
+            _subject(task_ids=[task_id], profiles=[task["assignee"]]), severity="HIGH",
+            evidence_state="MEASURED", title="Ready task was not claimed within two dispatcher ticks",
+            owner="OWNER.UNRESOLVED",
+            recommendation="Verify dispatcher health and profile spawnability for this assignee; this measurement does not yet apply full idle-agent capacity safeguards.",
+            evidence=[{"event_ids": [], "query_id": "Q-STALL-03", "fact": f"Ready since {_utc(ready_since)}; no claimed event followed within {READY_UNCLAIMED_THRESHOLD_SECONDS}s."}],
+            observed_at=ready_since, next_expected_event="claimed",
+        ))
+
     holes = _apply_prior_states(conn, holes, board_slug=board_slug, observed_at=end)
     severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
     holes.sort(key=lambda hole: (0 if hole["state"] != "RESOLVED" else 1, severity_order.get(hole["severity"], 99), hole["rule_id"], hole["finding_key"]))

tests/hermes_cli/test_kanban_telemetry.py

@@ -190,6 +190,114 @@ def test_missing_nominal_boundary_emits_critical_review_health_hole(board):
     assert missed[0]["owner"] == "rhea-ramos"
 
 
+def test_protocol_violation_first_is_high_repeat_is_critical(board):
+    end = 1_800_000_000
+    with kb.connect_closing() as conn:
+        task = kb.create_task(conn, title="worker", assignee="felix-steele")
+        _insert_event(conn, task, "protocol_violation", {"pid": 1, "claimer": "host:1", "exit_code": 0}, end - 3000)
+        first = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+        _insert_event(conn, task, "protocol_violation", {"pid": 2, "claimer": "host:1", "exit_code": 0}, end - 1000)
+        second = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    first_hole = next(h for h in first["holes"] if h["rule_id"] == "FAILURE.PROTOCOL_VIOLATION")
+    second_hole = next(h for h in second["holes"] if h["rule_id"] == "FAILURE.PROTOCOL_VIOLATION")
+    assert first_hole["severity"] == "HIGH"
+    assert second_hole["severity"] == "CRITICAL"
+
+
+def test_retry_thrash_flags_two_failures_and_escalates_on_breaker(board):
+    end = 1_800_000_000
+    with kb.connect_closing() as conn:
+        task = kb.create_task(conn, title="thrashing", assignee="felix-steele")
+        _insert_event(conn, task, "crashed", {"pid": 1}, end - 5000)
+        _insert_event(conn, task, "timed_out", {}, end - 3000)
+        no_breaker = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+        _insert_event(conn, task, "gave_up", {"failures": 3}, end - 1000)
+        with_breaker = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    hole_before = next(h for h in no_breaker["holes"] if h["rule_id"] == "FAILURE.RETRY_THRASH")
+    hole_after = next(h for h in with_breaker["holes"] if h["rule_id"] == "FAILURE.RETRY_THRASH")
+    assert hole_before["severity"] == "HIGH"
+    assert hole_after["severity"] == "CRITICAL"
+
+
+@pytest.mark.parametrize(
+    ("age_seconds", "expected_severity"),
+    [
+        (telemetry.BLOCKED_AGED_MEDIUM_SECONDS, "MEDIUM"),
+        (telemetry.BLOCKED_AGED_HIGH_SECONDS, "HIGH"),
+        (telemetry.BLOCKED_AGED_CRITICAL_SECONDS, "CRITICAL"),
+    ],
+)
+def test_blocked_aged_severity_ladder(board, age_seconds, expected_severity):
+    end = 1_800_000_000
+    blocked_at = end - age_seconds
+    with kb.connect_closing() as conn:
+        task = kb.create_task(conn, title="blocked", assignee="felix-steele", initial_status="blocked")
+        conn.execute(
+            "UPDATE tasks SET block_kind = 'needs_input' WHERE id = ?", (task,),
+        )
+        _insert_event(conn, task, "blocked", {"reason": "needs decision", "kind": "needs_input", "recurrences": 1}, blocked_at)
+        report = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    hole = next(h for h in report["holes"] if h["rule_id"] == "STALL.BLOCKED_AGED")
+    assert hole["severity"] == expected_severity
+
+
+def test_todo_promotable_flags_stuck_task_after_two_ticks(board):
+    end = 1_800_000_000
+    eligible_since = end - telemetry.TODO_PROMOTABLE_THRESHOLD_SECONDS - 10
+    with kb.connect_closing() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="felix-steele")
+        child = kb.create_task(conn, title="child", parents=[parent], assignee="felix-steele")
+        conn.execute(
+            "UPDATE tasks SET status = 'done', completed_at = ? WHERE id = ?",
+            (eligible_since, parent),
+        )
+        report = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    hole = next(h for h in report["holes"] if h["rule_id"] == "STALL.TODO_PROMOTABLE")
+    assert hole["severity"] == "HIGH"
+    assert child in hole["subject"]["task_ids"]
+
+
+def test_todo_promotable_suppressed_after_promotion(board):
+    end = 1_800_000_000
+    eligible_since = end - telemetry.TODO_PROMOTABLE_THRESHOLD_SECONDS - 10
+    with kb.connect_closing() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="felix-steele")
+        child = kb.create_task(conn, title="child", parents=[parent], assignee="felix-steele")
+        conn.execute(
+            "UPDATE tasks SET status = 'ready', completed_at = ? WHERE id IN (?, ?)",
+            (eligible_since, parent, child),
+        )
+        conn.execute("UPDATE tasks SET status = 'done' WHERE id = ?", (parent,))
+        _insert_event(conn, child, "promoted", {"from_status": "todo", "to_status": "ready"}, eligible_since + 5)
+        report = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    rule_ids = {h["rule_id"] for h in report["holes"]}
+    assert "STALL.TODO_PROMOTABLE" not in rule_ids
+
+
+def test_ready_unclaimed_flags_task_after_two_ticks(board):
+    end = 1_800_000_000
+    ready_since = end - telemetry.READY_UNCLAIMED_THRESHOLD_SECONDS - 10
+    with kb.connect_closing() as conn:
+        task = kb.create_task(conn, title="unclaimed", assignee="felix-steele")
+        _insert_event(conn, task, "created", {}, ready_since)
+        report = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    hole = next(h for h in report["holes"] if h["rule_id"] == "STALL.READY_UNCLAIMED")
+    assert hole["severity"] == "HIGH"
+    assert task in hole["subject"]["task_ids"]
+
+
+def test_ready_unclaimed_suppressed_after_claim(board):
+    end = 1_800_000_000
+    ready_since = end - telemetry.READY_UNCLAIMED_THRESHOLD_SECONDS - 10
+    with kb.connect_closing() as conn:
+        task = kb.create_task(conn, title="claimed", assignee="felix-steele")
+        _insert_event(conn, task, "created", {}, ready_since)
+        _insert_event(conn, task, "claimed", {"lock": "host:1"}, ready_since + 5)
+        report = telemetry.run_review(conn, board_slug="default", db_path=kb.kanban_db_path(), window_end=end, generated_at=end)
+    rule_ids = {h["rule_id"] for h in report["holes"]}
+    assert "STALL.READY_UNCLAIMED" not in rule_ids
+
+
 def test_persist_review_replay_preserves_dispositioned_verified_ledger(board, tmp_path):
     """AGA P1 #1: replaying a telemetry source must never reset the ledger.