diff --git a/gateway/kanban_watchers.py b/gateway/kanban_watchers.py index 2f5137a8a84f..0584ac2376d0 100644 --- a/gateway/kanban_watchers.py +++ b/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): diff --git a/hermes_cli/kanban.py b/hermes_cli/kanban.py index a57728db6d48..16354514cdbf 100644 --- a/hermes_cli/kanban.py +++ b/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), diff --git a/hermes_cli/kanban_telemetry.py b/hermes_cli/kanban_telemetry.py index 8bac98d7f0f3..8ff60346fa75 100644 --- a/hermes_cli/kanban_telemetry.py +++ b/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"] diff --git a/tests/gateway/test_kanban_watchers_mixin.py b/tests/gateway/test_kanban_watchers_mixin.py index 8454b5fd33dc..7a67ac58dbdc 100644 --- a/tests/gateway/test_kanban_watchers_mixin.py +++ b/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 + + diff --git a/tests/hermes_cli/test_kanban_telemetry.py b/tests/hermes_cli/test_kanban_telemetry.py index 321ccdb0b401..5e08b1ebe964 100644 --- a/tests/hermes_cli/test_kanban_telemetry.py +++ b/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"