From 86101d5c3ff26ef3c0ad8bbf7c380017d8fa215a Mon Sep 17 00:00:00 2001
From: Pigbibi <20649888+Pigbibi@users.noreply.github.com>
Date: Fri, 9 Oct 2026 18:06:58 +0800
Subject: [PATCH] fix(ops): skip Runtime heartbeat alert when main.yml is
 disabled

When GitHub reports the watched Runtime workflow as disabled_manually
or disabled_inactivity, treat the heartbeat as not_applicable instead of
alerting on an empty lookback. Matches run 37911924433 (0 strategy runs
while Runtime was disabled).
---
 scripts/runtime_workflow_heartbeat.py    | 36 ++++++++++++++++++++++++
 tests/test_runtime_workflow_heartbeat.py | 24 ++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/scripts/runtime_workflow_heartbeat.py b/scripts/runtime_workflow_heartbeat.py
index b4cecb44..fbad18b1 100644
--- a/scripts/runtime_workflow_heartbeat.py
+++ b/scripts/runtime_workflow_heartbeat.py
@@ -301,6 +301,19 @@ def _list_runtime_runs(
     return [run for run in runs if run.get("display_title") == "Runtime · strategy"]
 
 
+
+def _get_workflow_state(*, repository: str, workflow: str, token: str) -> str | None:
+    """Return GitHub Actions workflow state (active/disabled_manually/...), or None if unknown."""
+    url = f"https://api.github.com/repos/{repository}/actions/workflows/{workflow}"
+    try:
+        payload = _github_request(url, token)
+    except Exception as exc:  # noqa: BLE001
+        print(f"Workflow state lookup skipped: {exc}", file=sys.stderr)
+        return None
+    state = payload.get("state")
+    return state if isinstance(state, str) and state else None
+
+
 def _run_summary(run: dict[str, Any] | None) -> dict[str, Any] | None:
     if run is None:
         return None
@@ -470,6 +483,29 @@ def main() -> int:
     if not token:
         raise SystemExit("GITHUB_TOKEN is required")
 
+    workflow_state = _get_workflow_state(repository=repository, workflow=workflow, token=token)
+    if workflow_state in {"disabled_manually", "disabled_inactivity"}:
+        assessment = {
+            "schema": _ASSESSMENT_SCHEMA,
+            "observed_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
+            "status": "not_applicable",
+            "reason": "runtime_workflow_disabled",
+            "query": {
+                "workflow": workflow,
+                "branch": branch,
+                "workflow_state": workflow_state,
+                "runs_returned": 0,
+            },
+        }
+        _write_assessment(assessment)
+        print(json.dumps(assessment, sort_keys=True))
+        print(
+            "Runtime workflow heartbeat skipped: "
+            f"workflow={workflow} state={workflow_state} "
+            "(no strategy runs expected while disabled)"
+        )
+        return 0
+
     now = dt.datetime.now(dt.timezone.utc)
     since = now - dt.timedelta(hours=max(lookback_hours, expected_interval_hours * (max_consecutive_misses + 1)))
     try:
diff --git a/tests/test_runtime_workflow_heartbeat.py b/tests/test_runtime_workflow_heartbeat.py
index 820f866e..bff5b748 100644
--- a/tests/test_runtime_workflow_heartbeat.py
+++ b/tests/test_runtime_workflow_heartbeat.py
@@ -188,6 +188,28 @@ def test_disabled_target_skips_github_api_lookup_and_writes_assessment(self) ->
 
         request.assert_not_called()
 
+    def test_disabled_runtime_workflow_skips_alert_without_failing(self) -> None:
+        with patch.dict(
+            os.environ,
+            {
+                "GITHUB_TOKEN": "token-1",
+                "RUNTIME_TARGET_ENABLED": "true",
+                "RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT": "true",
+            },
+            clear=True,
+        ):
+            with patch.object(
+                heartbeat,
+                "_get_workflow_state",
+                return_value="disabled_manually",
+            ) as state:
+                with patch.object(heartbeat, "_list_runtime_runs") as list_runs:
+                    with patch.object(heartbeat, "_send_telegram") as send_telegram:
+                        self.assertEqual(heartbeat.main(), 0)
+        state.assert_called_once()
+        list_runs.assert_not_called()
+        send_telegram.assert_not_called()
+
     def test_github_api_outage_is_reported_as_unavailable_when_non_failing(self) -> None:
         with patch.dict(
             os.environ,
@@ -352,6 +374,8 @@ def test_repository_runs_fallback_finds_recent_runtime_success(self) -> None:
         def fake_github_request(url: str, token: str) -> dict[str, object]:
             requested_urls.append(url)
             self.assertEqual(token, "token-1")
+            if url.endswith("/actions/workflows/main.yml"):
+                return {"state": "active", "path": ".github/workflows/main.yml"}
             if "/actions/workflows/main.yml/runs?" in url:
                 return {"workflow_runs": []}
             if "/actions/runs?" in url: