fix: preserve snapshot http failures and resolve digest binding

.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

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

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", "")

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())

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()

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"

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]