diff --git a/.github/workflows/execution-report-heartbeat.yml b/.github/workflows/execution-report-heartbeat.yml index 5f3d748..fa0d77c 100644 --- a/.github/workflows/execution-report-heartbeat.yml +++ b/.github/workflows/execution-report-heartbeat.yml @@ -147,7 +147,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 @@ -174,7 +174,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 a407319..17cbc22 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -75,6 +75,28 @@ 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. + ## Publication before the central daily read GitHub schedule events can start late. A heartbeat workflow is therefore not a 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_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" 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]