From a936599160298ed88f651c4e7530fa8a7f071835 Mon Sep 17 00:00:00 2001
From: Pigbibi <20649888+Pigbibi@users.noreply.github.com>
Date: Mon, 5 Oct 2026 22:00:01 +0800
Subject: [PATCH] fix: make manual healthy receipts explicitly opt-in

---
 .../workflows/execution-report-heartbeat.yml  |  7 ++-
 tests/test_runtime_monitor_workflows.py       | 55 +++++++++++++++++++
 2 files changed, 61 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml
index 5f2d560..e6695ae 100644
--- a/.github/workflows/execution-report-heartbeat.yml
+++ b/.github/workflows/execution-report-heartbeat.yml
@@ -8,6 +8,11 @@ on:
         required: false
         type: string
         default: "36"
+      notify_on_success:
+        description: "Opt in to normal healthy-check notifications for this manual run."
+        required: false
+        type: boolean
+        default: false
       fail_workflow_on_alert:
         description: "Fail this workflow when an alert is emitted."
         required: false
@@ -44,7 +49,7 @@ jobs:
       RUNTIME_HEARTBEAT_ACCEPT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STAGES }}
       RUNTIME_HEARTBEAT_REJECT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STAGES }}
       RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }}
-      RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ github.event_name != 'schedule' && vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS || 'false' }}
+      RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ github.event_name == 'workflow_dispatch' && inputs.notify_on_success && 'true' || 'false' }}
       RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.FIRSTRADE_MARKET_CALENDAR }}
       RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.FIRSTRADE_MARKET_TIMEZONE }}
       RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }}
diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py
index 4a7b6a8..4a07fdf 100644
--- a/tests/test_runtime_monitor_workflows.py
+++ b/tests/test_runtime_monitor_workflows.py
@@ -140,3 +140,58 @@ def test_lifecycle_publishes_read_only_observation_for_exact_service() -> None:
     assert "CLOUD_RUN_SERVICES" not in publisher
     assert "gcloud scheduler jobs update" not in workflow
     assert "gcloud run deploy" not in workflow
+
+
+def _manual_input_block(workflow: str, name: str) -> str:
+    import re
+    match = re.search(rf"(?ms)^      {name}:\n(.*?)(?=^      [a-z_]+:|^  [a-z_]+:)", workflow)
+    assert match is not None, f"missing workflow_dispatch input {name}"
+    return match.group(1)
+
+
+def _evaluate_success_notify_expression(workflow: str, event: str, explicit_input, repository_value) -> str:
+    """Evaluate this bounded Actions boolean/string expression without running a workflow."""
+    import ast
+    import re
+    expression = re.search(r"RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: \$\{\{ (.*?) \}\}", workflow).group(1)
+    expression = expression.replace("github.event_name", repr(event))
+    expression = expression.replace("inputs.notify_on_success", repr(False if explicit_input is None else explicit_input))
+    expression = expression.replace("vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", repr(repository_value or ""))
+    expression = expression.replace("&&", " and ").replace("||", " or ")
+    parsed = ast.parse(expression, mode="eval")
+    assert all(isinstance(node, (ast.Expression, ast.BoolOp, ast.And, ast.Or, ast.Compare, ast.Eq, ast.NotEq, ast.Constant)) for node in ast.walk(parsed))
+    result = eval(compile(parsed, "<offline Actions expression>", "eval"), {"__builtins__": {}}, {})
+    return str(result).lower() if isinstance(result, bool) else str(result)
+
+
+def test_manual_healthy_notify_is_typed_and_explicitly_opt_in() -> None:
+    workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
+    block = _manual_input_block(workflow, "notify_on_success")
+    assert "type: boolean" in block
+    assert "default: false" in block
+    assert "required: false" in block
+    line = next(line for line in workflow.splitlines() if "RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS:" in line)
+    assert "inputs.notify_on_success" in line
+    assert "github.event.inputs" not in line
+    assert "vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS" not in line
+    assert "github.event_name == 'workflow_dispatch'" in line
+
+
+def test_schedule_and_manual_default_never_inherit_legacy_success_variable() -> None:
+    workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
+    for event in ("schedule", "workflow_dispatch", "repository_dispatch"):
+        for explicit_input in (None, False, True):
+            for repository_value in (None, "false", "true"):
+                actual = _evaluate_success_notify_expression(workflow, event, explicit_input, repository_value)
+                expected = "true" if event == "workflow_dispatch" and explicit_input is True else "false"
+                assert actual == expected, (event, explicit_input, repository_value, actual)
+
+
+def test_manual_quiet_control_preserves_existing_alert_and_report_steps() -> None:
+    workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
+    alert_block = _manual_input_block(workflow, "fail_workflow_on_alert")
+    assert 'default: "true"' in alert_block
+    assert "RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }}" in workflow
+    assert "uv run --no-sync python scripts/execution_report_heartbeat.py" in workflow
+    assert "Publish read-only runtime execution evidence" in workflow
+    assert 'cron: "20 22 * * *"' in workflow