From ad39774c5029ae7bb98d237b407eb371bf00bd9a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:29:49 +0800 Subject: [PATCH 1/2] fix: preserve snapshot http failures and resolve digest binding Co-Authored-By: Codex --- .../workflows/execution-report-heartbeat.yml | 4 +- docs/unified-daily-notification.md | 22 ++++ scripts/sync_cached_account_facts.py | 19 ++- scripts/with_heartbeat_digest_binding.py | 106 +++++++++++++++ tests/test_heartbeat_digest_binding.py | 121 ++++++++++++++++++ tests/test_sync_cached_account_facts.py | 26 ++++ 6 files changed, 293 insertions(+), 5 deletions(-) create mode 100644 scripts/with_heartbeat_digest_binding.py create mode 100644 tests/test_heartbeat_digest_binding.py diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index 35d589a..aa0e6dc 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -146,7 +146,7 @@ jobs: - name: Check recent execution report env: RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH: ${{ runner.temp }}/firstrade-coverage-${{ github.run_id }}-${{ github.run_attempt }} - run: uv run --no-sync python scripts/execution_report_heartbeat.py + run: uv run --no-sync python scripts/with_heartbeat_digest_binding.py heartbeat - name: Publish read-only runtime execution evidence uses: QuantStrategyLab/QuantRuntimeSettings/actions/publish-runtime-execution-evidence@5922cb28afd4639efe55e2d5c7c03de8e4445f95 @@ -173,7 +173,7 @@ jobs: DIGEST_TARGET_ID: ${{ secrets.DIGEST_TARGET_ID }} DIGEST_TARGET_BINDINGS_JSON: ${{ secrets.DIGEST_TARGET_BINDINGS_JSON }} DIGEST_NOTIFICATION_SCHEDULE_JSON: ${{ vars.DIGEST_NOTIFICATION_SCHEDULE_JSON }} - run: uv run --no-sync python scripts/emit_unified_digest_candidates.py + run: uv run --no-sync python scripts/with_heartbeat_digest_binding.py candidates runtime_daily: name: Publish a fresh runtime daily projection diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index ca7af7c..7fc7e7a 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -56,3 +56,25 @@ provider audit entries remain in their original restricted logs and are not passed as Telegram technical details, including when the deployed formatter is pinned to an older shared package. Guard detection, deduplication and failure exit codes are unchanged. + +## Scheduled heartbeat identity resolution + +The heartbeat and completed-window projector use +`scripts/with_heartbeat_digest_binding.py` when running from the heartbeat workflow. +Existing complete workflow bindings and multi-target bindings retain their +current validation path. For an enabled single target with missing digest values, +the wrapper reads the sole 100% serving Cloud Run revision using the existing +workflow principal and explicit account. Its runtime identity must exactly match +the configured platform, profile, service, account selectors, scope and dry-run +mode. Only the existing plain-text digest UID, target, slot and destination are +consumed, in process memory; Secret Manager references are never dereferenced. +Configured values must agree with the serving binding. A changing traffic +observation is rejected. A missing schedule is taken only from the configured +runtime target's main schedule and timezone. + +Missing source values return `digest_identity_owner_input_required` or +`digest_schedule_owner_input_required`. The owner must supply the approved +private binding; do not generate a UID or infer an identity from an account label. +Metadata read failures and identity conflicts remain explicit failures. This +wrapper does not change GitHub settings, runtime flags, receiver bindings, +notification routes or cloud permissions. It does not establish delivery success. diff --git a/scripts/sync_cached_account_facts.py b/scripts/sync_cached_account_facts.py index f329a03..57ce2ec 100644 --- a/scripts/sync_cached_account_facts.py +++ b/scripts/sync_cached_account_facts.py @@ -15,8 +15,11 @@ class SnapshotSyncError(RuntimeError): def cloud(*args): + account = os.environ.get("GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT") + if not account: + raise SnapshotSyncError("snapshot_reader_account_missing") result = subprocess.run( - ["gcloud", *args, "--project", os.environ["GCP_PROJECT_ID"], "--format=json", "--quiet"], + ["gcloud", *args, "--account=" + account, "--project", os.environ["GCP_PROJECT_ID"], "--format=json", "--quiet"], capture_output=True, text=True, timeout=60, check=False, ) if result.returncode: @@ -46,6 +49,7 @@ def completed_snapshot(rows, job_name, service_url): """Require both Scheduler completion and the exact application HTTP 200.""" finished = [] app_completed = False + failures = set() for row in rows: payload = row.get("jsonPayload") or {} if ( @@ -58,9 +62,18 @@ def completed_snapshot(rows, job_name, service_url): if ( row.get("resource", {}).get("type") == "cloud_run_revision" and http.get("requestUrl") == service_url + "/account-facts-sync" - and http.get("status") == 200 ): - app_completed = True + status = http.get("status") + if status == 200: + app_completed = True + elif type(status) is int and status in {401, 403, 429, 500, 502, 503, 504}: + failures.add(status) + if failures: + # Preserve the observed HTTP class without exposing provider bodies or + # guessing that an unavailable service is an expired broker session. + if len(failures) == 1: + raise SnapshotSyncError("snapshot_request_http_" + str(next(iter(failures)))) + raise SnapshotSyncError("snapshot_request_multiple_http_failures") if any( event.get("status") not in (None, "OK") or "response code number = 200" not in event.get("debugInfo", "") diff --git a/scripts/with_heartbeat_digest_binding.py b/scripts/with_heartbeat_digest_binding.py new file mode 100644 index 0000000..30e43dd --- /dev/null +++ b/scripts/with_heartbeat_digest_binding.py @@ -0,0 +1,106 @@ +"""Run a heartbeat command with its existing serving digest binding in memory.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import re +import subprocess +import sys + +FIELDS = ("DIGEST_OPAQUE_ACCOUNT_UID", "DIGEST_TARGET_ID", "DIGEST_STORE_TARGET", "DIGEST_CANDIDATES_GCS_URI") +IDENTITY = ("platform_id", "strategy_profile", "service_name", "account_scope", "account_selector", "deployment_selector", "dry_run_only") +COMMANDS = {"heartbeat": "execution_report_heartbeat.py", "candidates": "emit_unified_digest_candidates.py"} + + +class BindingError(ValueError): + """Only fixed public codes may leave the metadata boundary.""" + + +def cloud_json(args, env): + account = env.get("GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT", "") + if not account: + raise BindingError("digest_reader_account_missing") + result = subprocess.run(["gcloud", *args, "--account=" + account, "--project=" + env["GCP_PROJECT_ID"], + "--region=" + env["CLOUD_RUN_REGION"], "--format=json", "--quiet"], + capture_output=True, text=True, timeout=60, check=False) + if result.returncode or len(result.stdout) > 256 * 1024: + raise BindingError("digest_metadata_unavailable") + return json.loads(result.stdout) + + +def resolve(env, read=cloud_json): + resolved = dict(env) + if env.get("RUNTIME_TARGET_ENABLED") != "true" or env.get("DIGEST_TARGET_BINDINGS_JSON"): + return resolved # Existing multi-target/disabled contracts retain ownership. + expected = json.loads(env.get("RUNTIME_TARGET_JSON") or "{}") + if not isinstance(expected, dict) or any(k not in expected for k in IDENTITY): + raise BindingError("digest_runtime_identity_missing") + missing = [key for key in FIELDS if not env.get(key)] + if missing: + service = env.get("CLOUD_RUN_SERVICE", "") + if not service or service != expected["service_name"]: + raise BindingError("digest_runtime_identity_mismatch") + metadata = read(["run", "services", "describe", service], env) + if metadata.get("metadata", {}).get("name") != service: + raise BindingError("digest_runtime_identity_mismatch") + traffic = [row for row in metadata.get("status", {}).get("traffic", []) if row.get("percent", 0) > 0] + if len(traffic) != 1 or type(traffic[0].get("percent")) is not int or traffic[0]["percent"] != 100: + raise BindingError("digest_serving_revision_ambiguous") + revision = traffic[0].get("revisionName", "") + if not re.fullmatch(re.escape(service) + r"-[a-z0-9-]+", revision): + raise BindingError("digest_serving_revision_ambiguous") + payload = read(["run", "revisions", "describe", revision], env) + if payload.get("metadata", {}).get("name") != revision: + raise BindingError("digest_runtime_identity_mismatch") + containers = payload.get("spec", {}).get("containers", []) + if len(containers) != 1: + raise BindingError("digest_metadata_invalid") + values = {} + for row in containers[0].get("env", []): + name = row.get("name") + if name in values: + raise BindingError("digest_metadata_invalid") + values[name] = row.get("value") # Never access Secret Manager references. + target = json.loads(values.get("RUNTIME_TARGET_JSON") or "{}") + if (not isinstance(target, dict) or any(target.get(k) != expected[k] for k in IDENTITY) + or values.get("RUNTIME_TARGET_ENABLED") != "true"): + raise BindingError("digest_runtime_identity_mismatch") + for key in FIELDS: + value = values.get(key) + if not isinstance(value, str) or not value or value != value.strip() or any(c in value for c in "\r\n\0"): + raise BindingError("digest_identity_owner_input_required") + if env.get(key) and env[key] != value: + raise BindingError("digest_configured_identity_mismatch") + resolved[key] = value + # A traffic change invalidates this observation instead of adopting another target. + after = read(["run", "services", "describe", service], env) + if after.get("status", {}).get("traffic") != metadata.get("status", {}).get("traffic"): + raise BindingError("digest_serving_revision_changed") + if not resolved.get("DIGEST_NOTIFICATION_SCHEDULE_JSON"): + scheduler = expected.get("scheduler") or {} + if not scheduler.get("main_time") or not scheduler.get("timezone"): + raise BindingError("digest_schedule_owner_input_required") + resolved["DIGEST_NOTIFICATION_SCHEDULE_JSON"] = json.dumps({"scheduler": { + "main_time": scheduler["main_time"], "timezone": scheduler["timezone"]}}) + return resolved + + +def main(): + if len(sys.argv) != 2 or sys.argv[1] not in COMMANDS: + return 2 + try: + env = resolve(os.environ) + except BindingError as error: + print(json.dumps({"status": "blocked", "reason": str(error)})) + return 1 + except (KeyError, TypeError, ValueError, OSError, subprocess.SubprocessError): + print(json.dumps({"status": "blocked", "reason": "digest_metadata_invalid"})) + return 1 + # Values are neither printed nor exported through GITHUB_ENV/output/artifacts. + command = Path(__file__).with_name(COMMANDS[sys.argv[1]]) + return subprocess.run([sys.executable, str(command)], env=env, check=False).returncode + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_heartbeat_digest_binding.py b/tests/test_heartbeat_digest_binding.py new file mode 100644 index 0000000..19db0bb --- /dev/null +++ b/tests/test_heartbeat_digest_binding.py @@ -0,0 +1,121 @@ +from copy import deepcopy +import json +from unittest.mock import Mock + +import pytest + +from scripts import with_heartbeat_digest_binding as binding + + +TARGET = dict(platform_id="synthetic", strategy_profile="core", service_name="synthetic-service", + account_scope="synthetic-scope", account_selector=["synthetic-account"], + deployment_selector="synthetic-deployment", dry_run_only=True, + scheduler={"main_time": "0 20 * * 1-5", "timezone": "UTC"}) +VALUES = dict(DIGEST_OPAQUE_ACCOUNT_UID="synthetic-uid", DIGEST_TARGET_ID="synthetic-target", + DIGEST_STORE_TARGET="synthetic-slot", DIGEST_CANDIDATES_GCS_URI="gs://synthetic-bucket/candidates") +REVISION = "synthetic-service-00001-test" + + +def environment(): + return dict(RUNTIME_TARGET_ENABLED="true", RUNTIME_TARGET_JSON=json.dumps(TARGET), + CLOUD_RUN_SERVICE=TARGET["service_name"], GCP_PROJECT_ID="synthetic-project", + CLOUD_RUN_REGION="synthetic-region", GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT="synthetic@example.invalid") + + +def metadata(): + service = dict(metadata={"name": TARGET["service_name"]}, status={ + "latestReadyRevisionName": "synthetic-service-00002-candidate", + "traffic": [{"revisionName": REVISION, "percent": 100}]}) + values = {**VALUES, "RUNTIME_TARGET_JSON": json.dumps(TARGET), "RUNTIME_TARGET_ENABLED": "true"} + revision = dict(metadata={"name": REVISION}, spec={"containers": [ + {"env": [{"name": k, "value": v} for k, v in values.items()]}]}) + return service, revision + + +def reader(): + service, revision = metadata() + return Mock(side_effect=[service, revision, deepcopy(service)]) + + +def test_missing_binding_uses_serving_revision_and_keeps_values_in_memory(capsys): + read = reader() + env = environment() + result = binding.resolve(env, read) + assert {k: result[k] for k in VALUES} == VALUES + assert result["RUNTIME_TARGET_JSON"] == env["RUNTIME_TARGET_JSON"] + assert read.call_args_list[1].args[0] == ["run", "revisions", "describe", REVISION] + assert json.loads(result["DIGEST_NOTIFICATION_SCHEDULE_JSON"])["scheduler"] == TARGET["scheduler"] + assert "DIGEST_TARGET_ID" not in env + assert capsys.readouterr().out == "" + + +@pytest.mark.parametrize("mutation,reason", [ + ("identity", "digest_runtime_identity_mismatch"), + ("missing", "digest_identity_owner_input_required"), + ("secret_reference", "digest_identity_owner_input_required"), + ("duplicate", "digest_metadata_invalid"), +]) +def test_unqualified_metadata_never_supplies_identity(mutation, reason): + service, revision = metadata() + rows = revision["spec"]["containers"][0]["env"] + if mutation == "identity": + changed = {**TARGET, "account_selector": ["another-synthetic-account"]} + next(r for r in rows if r["name"] == "RUNTIME_TARGET_JSON")["value"] = json.dumps(changed) + if mutation in {"missing", "secret_reference"}: + rows[:] = [r for r in rows if r["name"] != "DIGEST_OPAQUE_ACCOUNT_UID"] + if mutation == "secret_reference": + rows.append({"name": "DIGEST_OPAQUE_ACCOUNT_UID", "valueFrom": {"secretKeyRef": {"name": "synthetic-secret"}}}) + if mutation == "duplicate": + rows.append(rows[0]) + with pytest.raises(binding.BindingError, match=reason): + binding.resolve(environment(), Mock(side_effect=[service, revision])) + + +def test_partial_conflict_is_not_overwritten(): + with pytest.raises(binding.BindingError, match="digest_configured_identity_mismatch"): + binding.resolve({**environment(), "DIGEST_TARGET_ID": "another-target"}, reader()) + + +def test_split_or_changed_traffic_cannot_select_a_binding(): + service, revision = metadata() + split = deepcopy(service) + split["status"]["traffic"] = [{"revisionName": REVISION, "percent": 50}, + {"revisionName": "synthetic-service-00002-candidate", "percent": 50}] + with pytest.raises(binding.BindingError, match="digest_serving_revision_ambiguous"): + binding.resolve(environment(), Mock(return_value=split)) + with pytest.raises(binding.BindingError, match="digest_serving_revision_changed"): + binding.resolve(environment(), Mock(side_effect=[service, revision, split])) + + +def test_complete_disabled_and_multi_target_bindings_do_not_read_cloud(): + read = Mock(side_effect=AssertionError("unexpected cloud read")) + for env in ({**environment(), **VALUES}, {"RUNTIME_TARGET_ENABLED": "false"}, + {"DIGEST_TARGET_BINDINGS_JSON": "[]"}): + binding.resolve(env, read) + read.assert_not_called() + + +def test_schedule_is_not_invented(): + target = {**TARGET, "scheduler": {}} + with pytest.raises(binding.BindingError, match="digest_schedule_owner_input_required"): + binding.resolve({**environment(), **VALUES, "RUNTIME_TARGET_JSON": json.dumps(target)}) + + +def test_cloud_reader_has_explicit_account_and_suppresses_provider_errors(monkeypatch): + run = Mock(return_value=Mock(returncode=1, stdout="private-provider-output", stderr="private-error")) + monkeypatch.setattr(binding.subprocess, "run", run) + with pytest.raises(binding.BindingError, match="^digest_metadata_unavailable$"): + binding.cloud_json(["run", "services", "describe", TARGET["service_name"]], environment()) + args = run.call_args.args[0] + assert "--account=synthetic@example.invalid" in args + assert run.call_args.kwargs["capture_output"] is True + + +def test_blocked_resolution_never_runs_a_command_or_prints_private_data(monkeypatch, capsys): + monkeypatch.setattr(binding.sys, "argv", ["wrapper", "heartbeat"]) + monkeypatch.setattr(binding, "resolve", Mock(side_effect=binding.BindingError("digest_identity_owner_input_required"))) + run = Mock(side_effect=AssertionError("blocked command must not run")) + monkeypatch.setattr(binding.subprocess, "run", run) + assert binding.main() == 1 + assert json.loads(capsys.readouterr().out) == {"status": "blocked", "reason": "digest_identity_owner_input_required"} + run.assert_not_called() diff --git a/tests/test_sync_cached_account_facts.py b/tests/test_sync_cached_account_facts.py index 4ebe8a2..e307fcc 100644 --- a/tests/test_sync_cached_account_facts.py +++ b/tests/test_sync_cached_account_facts.py @@ -100,3 +100,29 @@ def test_unknown_dispatch_outcome_never_retries(monkeypatch, capsys): assert sync.main() == 1 assert run.call_count == 3 assert json.loads(capsys.readouterr().out)["retry"] is False + + +@pytest.mark.parametrize("status", [401, 403, 429, 500, 502, 503, 504]) +def test_exact_application_failure_is_not_collapsed_into_auth_or_provider(status): + row = {"resource": {"type": "cloud_run_revision"}, + "httpRequest": {"requestUrl": URL + "/account-facts-sync", "status": status}, + "textPayload": "private-provider-body"} + with pytest.raises(sync.SnapshotSyncError, match="^snapshot_request_http_" + str(status) + "$"): + sync.completed_snapshot([completion(), row], JOB, URL) + + +def test_failed_snapshot_is_not_erased_by_a_separate_success(): + rows = [{"resource": {"type": "cloud_run_revision"}, + "httpRequest": {"requestUrl": URL + "/account-facts-sync", "status": status}} + for status in [200, 503]] + with pytest.raises(sync.SnapshotSyncError, match="snapshot_request_http_503"): + sync.completed_snapshot([completion(), *rows], JOB, URL) + + +def test_cloud_commands_use_the_explicit_workflow_account(monkeypatch): + monkeypatch.setenv("GCP_PROJECT_ID", "synthetic") + monkeypatch.setenv("GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT", "synthetic@example.invalid") + run = Mock(return_value=Mock(returncode=0, stdout="{}")) + monkeypatch.setattr(sync.subprocess, "run", run) + sync.cloud("logging", "read", "synthetic-filter") + assert "--account=synthetic@example.invalid" in run.call_args.args[0] From b7715c56cac6c10b214656f3817abcee1d456335 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 02:38:54 +0800 Subject: [PATCH 2/2] test: validate wrapped heartbeat workflow commands Co-Authored-By: Codex --- tests/test_runtime_monitor_workflows.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py index 8b9e946..df8dafc 100644 --- a/tests/test_runtime_monitor_workflows.py +++ b/tests/test_runtime_monitor_workflows.py @@ -20,7 +20,7 @@ def test_execution_report_heartbeat_has_market_neutral_daily_schedule() -> None: def test_qpk_dependent_heartbeats_use_one_locked_uv_runtime_per_job() -> None: setup_uv = "astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9" qpk_scripts = { - "execution-report-heartbeat.yml": ("scripts/execution_report_heartbeat.py",), + "execution-report-heartbeat.yml": ("scripts/with_heartbeat_digest_binding.py heartbeat",), "runtime-target-lifecycle.yml": ( "scripts/cloud_run_runtime_guard.py", "scripts/execution_report_heartbeat.py", @@ -196,7 +196,7 @@ def test_manual_quiet_control_preserves_existing_alert_and_report_steps() -> Non 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 "uv run --no-sync python scripts/with_heartbeat_digest_binding.py heartbeat" in workflow assert "Publish read-only runtime execution evidence" in workflow assert 'cron: "20 22 * * *"' in workflow @@ -209,7 +209,7 @@ def test_scheduled_firstrade_candidate_projection_and_quiet_heartbeat_are_paired "(steps.gcp_auth_primary.outcome == 'success' || steps.gcp_auth_retry.outcome == 'success')") assert "id: gcp_auth_retry" in workflow assert " QSL_ROUTINE_NOTIFICATION_MODE: daily_digest\n" in candidate_step - assert "uv run --no-sync python scripts/emit_unified_digest_candidates.py" in candidate_step + assert "uv run --no-sync python scripts/with_heartbeat_digest_binding.py candidates" in candidate_step assert "RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}" in candidate_step assert "DIGEST_TARGET_BINDINGS_JSON: ${{ secrets.DIGEST_TARGET_BINDINGS_JSON }}" in candidate_step assert _evaluate_success_notify_expression(workflow, "schedule", None, "true") == "false"