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

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)