HEL-3113: supervise rolling telemetry-hole review cadence

gateway/kanban_watchers.py

@@ -57,6 +57,18 @@ def _resolve_auto_decompose_settings(
     return enabled, per_tick
 
 
+def _telemetry_review_due(
+    *,
+    now: int,
+    last_boundary: int | None,
+) -> tuple[bool, int]:
+    """Return whether the current nominal review boundary needs a run."""
+    from hermes_cli import kanban_telemetry
+
+    boundary = kanban_telemetry.nominal_window_end(now)
+    return boundary != last_boundary, boundary
+
+
 def _acquire_singleton_lock(lock_path) -> "tuple[Optional[object], str]":
     """Take an exclusive, non-blocking advisory lock for the sole dispatcher.
 
@@ -1143,6 +1155,7 @@ async def _kanban_dispatcher_watcher(self) -> None:
         HEALTH_WINDOW = 6
         bad_ticks = 0
         last_warn_at = 0
+        last_telemetry_review_boundary: int | None = None
         # Avoid hot-looping corrupt-looking board DBs, but do not suppress
         # same-fingerprint retries forever: transient WAL/open races can
         # surface as "database disk image is malformed" for one tick.
@@ -1315,6 +1328,21 @@ def _ready_nonempty() -> bool:
                             pass
             return False
 
+        def _telemetry_review_tick() -> None:
+            """Run one persisted review cycle for every active board."""
+            from hermes_cli import kanban_telemetry as _telemetry
+
+            results = _telemetry.run_scheduled_reviews(now=int(time.time()))
+            for slug, report, json_path, markdown_path in results:
+                logger.info(
+                    "kanban telemetry review [%s]: status=%s holes=%d json=%s markdown=%s",
+                    slug,
+                    report["review"]["status"],
+                    len(report["holes"]),
+                    json_path,
+                    markdown_path,
+                )
+
         # Auto-decompose: turn fresh triage tasks into ready workgraphs
         # before the dispatcher fans out workers. Gated by
         # ``kanban.auto_decompose`` (default True). Capped by
@@ -1436,6 +1464,14 @@ def _auto_decompose_tick(auto_decompose_per_tick: int) -> int:
                 if _ad_enabled:
                     await asyncio.to_thread(_auto_decompose_tick, _ad_per_tick)
                 results = await asyncio.to_thread(_tick_once)
+                due, boundary = _telemetry_review_due(
+                    now=int(time.time()),
+                    last_boundary=last_telemetry_review_boundary,
+                )
+                healthy_board_exists = any(result is not None for _, result in (results or []))
+                if due and healthy_board_exists:
+                    await asyncio.to_thread(_telemetry_review_tick)
+                    last_telemetry_review_boundary = boundary
                 any_spawned = False
                 for slug, res in (results or []):
                     if res is not None and getattr(res, "spawned", None):

hermes_cli/kanban.py

@@ -1595,21 +1595,16 @@ def _cmd_intake(args: argparse.Namespace) -> int:
 def _cmd_telemetry_review(args: argparse.Namespace) -> int:
     from hermes_cli import kanban_telemetry
 
-    with kb.connect_closing() as conn:
-        report = kanban_telemetry.run_review(
-            conn,
-            board_slug=kb.get_current_board(),
-            db_path=kb.kanban_db_path(),
-            window_end=args.window_end,
-        )
     output_dir = (
         Path(args.output_dir).expanduser()
         if args.output_dir
-        else Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "kanban" / "reviews"
+        else None
+    )
+    report, json_path, md_path = kanban_telemetry.run_scheduled_review(
+        board_slug=kb.get_current_board(),
+        window_end=args.window_end,
+        output_dir=output_dir,
     )
-    json_path, md_path = kanban_telemetry.write_artifacts(report, output_dir)
-    with kb.connect_closing() as conn:
-        kanban_telemetry.persist_review(conn, report, json_path, md_path)
     payload = {
         "status": report["review"]["status"],
         "json": str(json_path),

hermes_cli/kanban_telemetry.py

@@ -7,7 +7,7 @@
 import sqlite3
 import subprocess
 import time
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Optional
 
@@ -18,6 +18,7 @@
 RUNNING_INACTIVITY_SECONDS = 60 * 60
 BLOCKED_AGING_SECONDS = 12 * 60 * 60
 FINDING_RETENTION_SECONDS = 7 * 24 * 60 * 60
+NOMINAL_BOUNDARY_MINUTE = 15
 
 _REQUIRED_FIELDS: dict[str, tuple[str, ...]] = {
     "linear_scoped": ("linear_issue_key", "sub_issue_keys", "cptc_estimates"),
@@ -76,6 +77,26 @@ def _utc(epoch: int) -> str:
     return datetime.fromtimestamp(int(epoch), timezone.utc).isoformat().replace("+00:00", "Z")
 
 
+def nominal_window_end(now: Optional[int] = None) -> int:
+    """Return the latest 00:15/12:15 America/Phoenix cadence boundary."""
+    from zoneinfo import ZoneInfo
+
+    current = datetime.fromtimestamp(
+        int(now if now is not None else time.time()),
+        ZoneInfo("America/Phoenix"),
+    )
+    boundary_hour = 12 if (current.hour, current.minute) >= (12, NOMINAL_BOUNDARY_MINUTE) else 0
+    candidate = current.replace(
+        hour=boundary_hour,
+        minute=NOMINAL_BOUNDARY_MINUTE,
+        second=0,
+        microsecond=0,
+    )
+    if current < candidate:
+        candidate -= timedelta(hours=12)
+    return int(candidate.timestamp())
+
+
 def _implementation_sha() -> str:
     try:
         result = subprocess.run(
@@ -231,8 +252,9 @@ def _capture_snapshot(
             placeholders = ",".join("?" for _ in tasks)
             all_rows = conn.execute(
                 f"SELECT id, task_id, kind, payload, created_at, run_id FROM task_events "
-                f"WHERE task_id IN ({placeholders}) AND id <= ? ORDER BY id",
-                (*sorted(tasks), high),
+                f"WHERE task_id IN ({placeholders}) AND id <= ? "
+                f"AND created_at < ? ORDER BY id",
+                (*sorted(tasks), high, end),
             ).fetchall()
             all_events = [
                 {
@@ -323,6 +345,23 @@ def run_review(
         by_kind.setdefault(event["kind"], []).append(event)
 
     holes: list[dict[str, Any]] = []
+    previous_run = conn.execute(
+        "SELECT window_end FROM telemetry_review_runs "
+        "WHERE board_slug = ? AND status = 'COMPLETE' AND window_end < ? "
+        "ORDER BY window_end DESC LIMIT 1",
+        (board_slug, end),
+    ).fetchone()
+    if previous_run is not None:
+        expected_previous = end - CADENCE_SECONDS
+        previous_end = int(previous_run["window_end"])
+        if previous_end < expected_previous:
+            holes.append(_finding(
+                "REVIEW.MISSED_RUN", board_slug, _subject(), severity="CRITICAL",
+                evidence_state="MEASURED", title="Scheduled telemetry review boundary was missed",
+                owner="rhea-ramos", recommendation="Repair supervision so every nominal 12-hour boundary persists a complete artifact.",
+                evidence=[{"event_ids": [], "query_id": "Q-REVIEW-01", "fact": f"Latest prior complete boundary was {_utc(previous_end)}; expected {_utc(expected_previous)}."}],
+                observed_at=expected_previous, next_expected_event="telemetry_review_completed", due_by=end,
+            ))
     included_ids = sorted(tasks)
     if included_ids:
         holes.append(_finding(
@@ -453,6 +492,57 @@ def run_review(
     }
 
 
+def run_scheduled_review(
+    *,
+    board_slug: str,
+    window_end: Optional[int] = None,
+    generated_at: Optional[int] = None,
+    output_dir: Optional[Path] = None,
+) -> tuple[dict[str, Any], Path, Path]:
+    """Run and persist one nominal review cycle for a board."""
+    from hermes_cli import kanban_db as kb
+
+    end = int(window_end if window_end is not None else nominal_window_end(generated_at))
+    db_path = kb.kanban_db_path(board_slug)
+    with kb.connect_closing(board=board_slug) as conn:
+        report = run_review(
+            conn,
+            board_slug=board_slug,
+            db_path=db_path,
+            window_end=end,
+            generated_at=generated_at,
+        )
+    destination = output_dir or kb.board_dir(board_slug) / "reviews"
+    json_path, markdown_path = write_artifacts(report, destination)
+    with kb.connect_closing(board=board_slug) as conn:
+        persist_review(conn, report, json_path, markdown_path)
+    return report, json_path, markdown_path
+
+
+def run_scheduled_reviews(
+    *,
+    now: Optional[int] = None,
+) -> list[tuple[str, dict[str, Any], Path, Path]]:
+    """Run one deterministic cycle for every active board."""
+    from hermes_cli import kanban_db as kb
+
+    end = nominal_window_end(now)
+    results: list[tuple[str, dict[str, Any], Path, Path]] = []
+    try:
+        boards = kb.list_boards(include_archived=False)
+    except Exception:
+        boards = [kb.read_board_metadata(kb.DEFAULT_BOARD)]
+    for board in boards:
+        slug = board.get("slug") or kb.DEFAULT_BOARD
+        report, json_path, markdown_path = run_scheduled_review(
+            board_slug=slug,
+            window_end=end,
+            generated_at=now,
+        )
+        results.append((slug, report, json_path, markdown_path))
+    return results
+
+
 def render_markdown(report: dict[str, Any]) -> str:
     lines = ["# TRC 48-Hour Telemetry Hole Review", "", f"Verdict: {report['council']['verdict']}", ""]
     active = [hole for hole in report["holes"] if hole["state"] != "RESOLVED"]

tests/gateway/test_kanban_watchers_mixin.py

@@ -8,8 +8,10 @@
 from __future__ import annotations
 
 import inspect
+from datetime import datetime
+from zoneinfo import ZoneInfo
 
-from gateway.kanban_watchers import GatewayKanbanWatchersMixin
+from gateway.kanban_watchers import GatewayKanbanWatchersMixin, _telemetry_review_due
 
 KANBAN_METHODS = [
     "_kanban_notifier_watcher",
@@ -26,3 +28,18 @@ def test_mixin_defines_kanban_methods():
         assert hasattr(GatewayKanbanWatchersMixin, m), f"mixin missing {m}"
 
 
+def test_telemetry_review_runs_once_per_nominal_boundary():
+    phoenix = ZoneInfo("America/Phoenix")
+    now = int(datetime(2026, 7, 30, 12, 30, tzinfo=phoenix).timestamp())
+
+    due, boundary = _telemetry_review_due(now=now, last_boundary=None)
+    duplicate_due, duplicate_boundary = _telemetry_review_due(
+        now=now,
+        last_boundary=boundary,
+    )
+
+    assert due is True
+    assert duplicate_due is False
+    assert duplicate_boundary == boundary
+
+

tests/hermes_cli/test_kanban_telemetry.py

@@ -2,7 +2,9 @@
 from __future__ import annotations
 
 import json
+from datetime import datetime
 from pathlib import Path
+from zoneinfo import ZoneInfo
 
 import pytest
 
@@ -120,3 +122,69 @@ def test_governed_event_validation_fails_closed(board):
     scoped = [event for event in events if event.kind == "linear_scoped"]
     assert len(scoped) == 1
     assert scoped[0].payload["schema_version"] == 1
+
+
+@pytest.mark.parametrize(
+    ("local_now", "expected_local"),
+    [
+        ("2026-07-30T00:14:59", "2026-07-29T12:15:00"),
+        ("2026-07-30T00:15:00", "2026-07-30T00:15:00"),
+        ("2026-07-30T12:14:59", "2026-07-30T00:15:00"),
+        ("2026-07-30T12:15:00", "2026-07-30T12:15:00"),
+    ],
+)
+def test_nominal_window_end_follows_phoenix_boundaries(local_now, expected_local):
+    phoenix = ZoneInfo("America/Phoenix")
+    now = int(datetime.fromisoformat(local_now).replace(tzinfo=phoenix).timestamp())
+    expected = int(datetime.fromisoformat(expected_local).replace(tzinfo=phoenix).timestamp())
+
+    assert telemetry.nominal_window_end(now) == expected
+
+
+def test_scheduled_review_writes_and_persists_both_artifacts(board, tmp_path):
+    end = 1_800_000_000
+    with kb.connect_closing() as conn:
+        kb.create_task(conn, title="scheduled", triage=True)
+
+    report, json_path, markdown_path = telemetry.run_scheduled_review(
+        board_slug="default",
+        window_end=end,
+        generated_at=end,
+        output_dir=tmp_path / "reviews",
+    )
+
+    assert json_path.is_file()
+    assert markdown_path.is_file()
+    with kb.connect_closing() as conn:
+        stored = conn.execute(
+            "SELECT json_path, markdown_path, status FROM telemetry_review_runs "
+            "WHERE review_id = ?",
+            (report["review"]["review_id"],),
+        ).fetchone()
+    assert stored["json_path"] == str(json_path)
+    assert stored["markdown_path"] == str(markdown_path)
+    assert stored["status"] == "COMPLETE"
+
+
+def test_missing_nominal_boundary_emits_critical_review_health_hole(board):
+    end = 1_800_000_000
+    with kb.connect_closing() as conn:
+        conn.execute(
+            "INSERT INTO telemetry_review_runs "
+            "(review_id, board_slug, window_end, event_id_low_exclusive, "
+            "event_id_high_inclusive, generated_at, status) "
+            "VALUES (?, ?, ?, 0, 0, ?, 'COMPLETE')",
+            ("old", "default", end - (2 * telemetry.CADENCE_SECONDS), end),
+        )
+        report = telemetry.run_review(
+            conn,
+            board_slug="default",
+            db_path=kb.kanban_db_path(),
+            window_end=end,
+            generated_at=end,
+        )
+
+    missed = [hole for hole in report["holes"] if hole["rule_id"] == "REVIEW.MISSED_RUN"]
+    assert len(missed) == 1
+    assert missed[0]["severity"] == "CRITICAL"
+    assert missed[0]["owner"] == "rhea-ramos"