From 6c76bf2f66fce243ff1ae0d44842e06a26928543 Mon Sep 17 00:00:00 2001 From: "Claude (Opus 4.6)" Date: Tue, 24 Mar 2026 01:43:49 +0000 Subject: [PATCH] [claude] Integrate health snapshot into Daily Run pre-flight (#923) (#1280) --- tests/timmy_automations/test_orchestrator.py | 270 +++++++++++++++++++ timmy_automations/daily_run/orchestrator.py | 99 ++++++- 2 files changed, 368 insertions(+), 1 deletion(-) create mode 100644 tests/timmy_automations/test_orchestrator.py diff --git a/tests/timmy_automations/test_orchestrator.py b/tests/timmy_automations/test_orchestrator.py new file mode 100644 index 00000000..7355bd94 --- /dev/null +++ b/tests/timmy_automations/test_orchestrator.py @@ -0,0 +1,270 @@ +"""Tests for Daily Run orchestrator — health snapshot integration. + +Verifies that the orchestrator runs a pre-flight health snapshot before +any coding work begins, and aborts on red status unless --force is passed. + +Refs: #923 +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Add timmy_automations to path for imports +_TA_PATH = Path(__file__).resolve().parent.parent.parent / "timmy_automations" / "daily_run" +if str(_TA_PATH) not in sys.path: + sys.path.insert(0, str(_TA_PATH)) +# Also add utils path +_TA_UTILS = Path(__file__).resolve().parent.parent.parent / "timmy_automations" +if str(_TA_UTILS) not in sys.path: + sys.path.insert(0, str(_TA_UTILS)) + +import health_snapshot as hs +import orchestrator as orch + + +def _make_snapshot(overall_status: str) -> hs.HealthSnapshot: + """Build a minimal HealthSnapshot for testing.""" + return hs.HealthSnapshot( + timestamp="2026-01-01T00:00:00+00:00", + overall_status=overall_status, + ci=hs.CISignal(status="pass", message="CI passing"), + issues=hs.IssueSignal(count=0, p0_count=0, p1_count=0), + flakiness=hs.FlakinessSignal( + status="healthy", + recent_failures=0, + recent_cycles=10, + failure_rate=0.0, + message="All good", + ), + tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"), + ) + + +def _make_red_snapshot() -> hs.HealthSnapshot: + return hs.HealthSnapshot( + timestamp="2026-01-01T00:00:00+00:00", + overall_status="red", + ci=hs.CISignal(status="fail", message="CI failed"), + issues=hs.IssueSignal(count=1, p0_count=1, p1_count=0), + flakiness=hs.FlakinessSignal( + status="critical", + recent_failures=8, + recent_cycles=10, + failure_rate=0.8, + message="High flakiness", + ), + tokens=hs.TokenEconomySignal(status="unknown", message="No data"), + ) + + +def _default_args(**overrides) -> argparse.Namespace: + """Build an argparse Namespace with defaults matching the orchestrator flags.""" + defaults = { + "review": False, + "json": False, + "max_items": None, + "skip_health_check": False, + "force": False, + } + defaults.update(overrides) + return argparse.Namespace(**defaults) + + +class TestRunHealthSnapshot: + """Test run_health_snapshot() — the pre-flight check called by main().""" + + def test_green_returns_zero(self, capsys): + """Green snapshot returns 0 (proceed).""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("green")): + rc = orch.run_health_snapshot(args) + + assert rc == 0 + + def test_yellow_returns_zero(self, capsys): + """Yellow snapshot returns 0 (proceed with caution).""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("yellow")): + rc = orch.run_health_snapshot(args) + + assert rc == 0 + + def test_red_returns_one(self, capsys): + """Red snapshot returns 1 (abort).""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()): + rc = orch.run_health_snapshot(args) + + assert rc == 1 + + def test_red_with_force_returns_zero(self, capsys): + """Red snapshot with --force returns 0 (proceed anyway).""" + args = _default_args(force=True) + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()): + rc = orch.run_health_snapshot(args) + + assert rc == 0 + + def test_snapshot_exception_is_skipped(self, capsys): + """If health snapshot raises, it degrades gracefully and returns 0.""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", side_effect=RuntimeError("boom")): + rc = orch.run_health_snapshot(args) + + assert rc == 0 + captured = capsys.readouterr() + assert "warning" in captured.err.lower() or "skipping" in captured.err.lower() + + def test_snapshot_prints_summary(self, capsys): + """Health snapshot prints a pre-flight summary block.""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("green")): + orch.run_health_snapshot(args) + + captured = capsys.readouterr() + assert "PRE-FLIGHT HEALTH CHECK" in captured.out + assert "CI" in captured.out + + def test_red_prints_abort_message(self, capsys): + """Red snapshot prints an abort message to stderr.""" + args = _default_args() + + with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()): + orch.run_health_snapshot(args) + + captured = capsys.readouterr() + assert "RED" in captured.err or "aborting" in captured.err.lower() + + def test_p0_issues_shown_in_output(self, capsys): + """P0 issue count is shown in the pre-flight output.""" + args = _default_args() + snapshot = hs.HealthSnapshot( + timestamp="2026-01-01T00:00:00+00:00", + overall_status="red", + ci=hs.CISignal(status="pass", message="CI passing"), + issues=hs.IssueSignal(count=2, p0_count=2, p1_count=0), + flakiness=hs.FlakinessSignal( + status="healthy", + recent_failures=0, + recent_cycles=10, + failure_rate=0.0, + message="All good", + ), + tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"), + ) + + with patch.object(orch, "_generate_health_snapshot", return_value=snapshot): + orch.run_health_snapshot(args) + + captured = capsys.readouterr() + assert "P0" in captured.out + + +class TestMainHealthCheckIntegration: + """Test that main() runs health snapshot before any coding work.""" + + def _patch_gitea_unavailable(self): + return patch.object(orch.GiteaClient, "is_available", return_value=False) + + def test_main_runs_health_check_before_gitea(self): + """Health snapshot is called before Gitea client work.""" + call_order = [] + + def fake_snapshot(*_a, **_kw): + call_order.append("health") + return _make_snapshot("green") + + def fake_gitea_available(self): + call_order.append("gitea") + return False + + args = _default_args() + + with ( + patch.object(orch, "_generate_health_snapshot", side_effect=fake_snapshot), + patch.object(orch.GiteaClient, "is_available", fake_gitea_available), + patch("sys.argv", ["orchestrator"]), + ): + orch.main() + + assert call_order.index("health") < call_order.index("gitea") + + def test_main_aborts_on_red_before_gitea(self): + """main() aborts with non-zero exit code when health is red.""" + gitea_called = [] + + def fake_gitea_available(self): + gitea_called.append(True) + return True + + with ( + patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()), + patch.object(orch.GiteaClient, "is_available", fake_gitea_available), + patch("sys.argv", ["orchestrator"]), + ): + rc = orch.main() + + assert rc != 0 + assert not gitea_called, "Gitea should NOT be called when health is red" + + def test_main_skips_health_check_with_flag(self): + """--skip-health-check bypasses the pre-flight snapshot.""" + health_called = [] + + def fake_snapshot(*_a, **_kw): + health_called.append(True) + return _make_snapshot("green") + + with ( + patch.object(orch, "_generate_health_snapshot", side_effect=fake_snapshot), + patch.object(orch.GiteaClient, "is_available", return_value=False), + patch("sys.argv", ["orchestrator", "--skip-health-check"]), + ): + orch.main() + + assert not health_called, "Health snapshot should be skipped" + + def test_main_force_flag_continues_despite_red(self): + """--force allows Daily Run to continue even when health is red.""" + gitea_called = [] + + def fake_gitea_available(self): + gitea_called.append(True) + return False # Gitea unavailable → exits early but after health check + + with ( + patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()), + patch.object(orch.GiteaClient, "is_available", fake_gitea_available), + patch("sys.argv", ["orchestrator", "--force"]), + ): + orch.main() + + # Gitea was reached despite red status because --force was passed + assert gitea_called + + def test_main_json_output_on_red_includes_error(self, capsys): + """JSON output includes error key when health is red.""" + with ( + patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()), + patch.object(orch.GiteaClient, "is_available", return_value=True), + patch("sys.argv", ["orchestrator", "--json"]), + ): + rc = orch.main() + + assert rc != 0 + captured = capsys.readouterr() + data = json.loads(captured.out) + assert "error" in data diff --git a/timmy_automations/daily_run/orchestrator.py b/timmy_automations/daily_run/orchestrator.py index 1001704a..b454b870 100755 --- a/timmy_automations/daily_run/orchestrator.py +++ b/timmy_automations/daily_run/orchestrator.py @@ -4,10 +4,13 @@ Connects to local Gitea, fetches candidate issues, and produces a concise agenda plus a day summary (review mode). +The Daily Run begins with a Quick Health Snapshot (#710) to ensure mandatory +systems are green before burning cycles on work that cannot land. + Run: python3 timmy_automations/daily_run/orchestrator.py [--review] Env: See timmy_automations/config/daily_run.json for configuration -Refs: #703 +Refs: #703, #923 """ from __future__ import annotations @@ -30,6 +33,11 @@ sys.path.insert( ) from utils.token_rules import TokenRules, compute_token_reward +# Health snapshot lives in the same package +from health_snapshot import generate_snapshot as _generate_health_snapshot +from health_snapshot import get_token as _hs_get_token +from health_snapshot import load_config as _hs_load_config + # ── Configuration ───────────────────────────────────────────────────────── REPO_ROOT = Path(__file__).resolve().parent.parent.parent @@ -495,6 +503,16 @@ def parse_args() -> argparse.Namespace: default=None, help="Override max agenda items", ) + p.add_argument( + "--skip-health-check", + action="store_true", + help="Skip the pre-flight health snapshot (not recommended)", + ) + p.add_argument( + "--force", + action="store_true", + help="Continue even if health snapshot is red (overrides abort-on-red)", + ) return p.parse_args() @@ -535,6 +553,76 @@ def compute_daily_run_tokens(success: bool = True) -> dict[str, Any]: } +def run_health_snapshot(args: argparse.Namespace) -> int: + """Run pre-flight health snapshot and return 0 (ok) or 1 (abort). + + Prints a concise summary of CI, issues, flakiness, and token economy. + Returns 1 if the overall status is red AND --force was not passed. + Returns 0 for green/yellow or when --force is active. + On any import/runtime error the check is skipped with a warning. + """ + try: + hs_config = _hs_load_config() + hs_token = _hs_get_token(hs_config) + snapshot = _generate_health_snapshot(hs_config, hs_token) + except Exception as exc: # noqa: BLE001 + print(f"[health] Warning: health snapshot failed ({exc}) — skipping", file=sys.stderr) + return 0 + + # Print concise pre-flight header + status_emoji = {"green": "🟢", "yellow": "🟡", "red": "🔴"}.get( + snapshot.overall_status, "⚪" + ) + print("─" * 60) + print(f"PRE-FLIGHT HEALTH CHECK {status_emoji} {snapshot.overall_status.upper()}") + print("─" * 60) + + ci_emoji = {"pass": "✅", "fail": "❌", "unknown": "⚠️", "unavailable": "⚪"}.get( + snapshot.ci.status, "⚪" + ) + print(f" {ci_emoji} CI: {snapshot.ci.message}") + + if snapshot.issues.p0_count > 0: + issue_emoji = "🔴" + elif snapshot.issues.p1_count > 0: + issue_emoji = "🟡" + else: + issue_emoji = "✅" + critical_str = f"{snapshot.issues.count} critical" + if snapshot.issues.p0_count: + critical_str += f" (P0: {snapshot.issues.p0_count})" + if snapshot.issues.p1_count: + critical_str += f" (P1: {snapshot.issues.p1_count})" + print(f" {issue_emoji} Issues: {critical_str}") + + flak_emoji = {"healthy": "✅", "degraded": "🟡", "critical": "🔴", "unknown": "⚪"}.get( + snapshot.flakiness.status, "⚪" + ) + print(f" {flak_emoji} Flakiness: {snapshot.flakiness.message}") + + token_emoji = {"balanced": "✅", "inflationary": "🟡", "deflationary": "🔵", "unknown": "⚪"}.get( + snapshot.tokens.status, "⚪" + ) + print(f" {token_emoji} Tokens: {snapshot.tokens.message}") + print() + + if snapshot.overall_status == "red" and not args.force: + print( + "🛑 Health status is RED — aborting Daily Run to avoid burning cycles.", + file=sys.stderr, + ) + print( + " Fix the issues above or re-run with --force to override.", + file=sys.stderr, + ) + return 1 + + if snapshot.overall_status == "red": + print("⚠️ Health is RED but --force passed — proceeding anyway.", file=sys.stderr) + + return 0 + + def main() -> int: args = parse_args() config = load_config() @@ -542,6 +630,15 @@ def main() -> int: if args.max_items: config["max_agenda_items"] = args.max_items + # ── Step 0: Pre-flight health snapshot ────────────────────────────────── + if not args.skip_health_check: + health_rc = run_health_snapshot(args) + if health_rc != 0: + tokens = compute_daily_run_tokens(success=False) + if args.json: + print(json.dumps({"error": "health_check_failed", "tokens": tokens})) + return health_rc + token = get_token(config) client = GiteaClient(config, token)