fix: isolate candidate roots without moving notification markers

.github/workflows/execution-report-heartbeat.yml

@@ -102,6 +102,7 @@ jobs:
       TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}
       FIRSTRADE_UNIFIED_NOTIFICATION_ENABLED: ${{ vars.FIRSTRADE_UNIFIED_NOTIFICATION_ENABLED || 'false' }}
       DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI || secrets.DIGEST_CANDIDATES_GCS_URI }}
+      DIGEST_CANDIDATES_OVERRIDE_GCS_URI: ${{ vars.DIGEST_CANDIDATES_OVERRIDE_GCS_URI }}
       DIGEST_STORE_TARGET: ${{ vars.DIGEST_STORE_TARGET }}
       DIGEST_OPAQUE_ACCOUNT_UID: ${{ secrets.DIGEST_OPAQUE_ACCOUNT_UID }}
       DIGEST_TARGET_ID: ${{ secrets.DIGEST_TARGET_ID }}

docs/unified-daily-notification.md

@@ -38,6 +38,24 @@ and receipts have been migrated; changing the prefix alone would discard the
 history used to prevent duplicate alerts. Per-target bindings may set the
 variable independently.
 
+For a candidate-only bucket migration in the heartbeat workflow, set the
+protected repository variable `DIGEST_CANDIDATES_OVERRIDE_GCS_URI`. It overrides
+the candidate root after each private target binding is resolved, so an older
+`DIGEST_TARGET_BINDINGS_JSON` cannot silently select the previous bucket. The
+existing explicit marker URI is preserved; when absent, the binding's original
+candidate root remains its marker root. Target identity, archive source and
+enablement are unchanged. Missing or invalid candidate access never falls back
+to the previous bucket. An unset override preserves existing behavior.
+
+The standalone single-target publisher and runtime can instead update their
+explicit `DIGEST_CANDIDATES_GCS_URI` after freezing the existing marker location
+in `QSL_NOTIFICATION_MARKER_GCS_URI`. Pause all candidate writers outside the
+publication window, copy and verify the selected candidate objects, switch the
+writers, and then validate the central source through its no-send preview.
+Keep the selected new root during application rollback and retain every marker
+and central delivery reservation. Remove superseded permissions only after at
+least two successful natural digest windows.
+
 No deployment, variable application, IAM, secret or live message is authorized
 by this PR. Source adoption and delivery require separate approved release work.
 

notifications/unified.py

@@ -11,6 +11,7 @@
 import hashlib
 import json
 import os
+import re
 
 from notifications.digest_snapshot import digest_archive
 from notifications.digest_snapshot.digest_candidates import project_reports
@@ -115,7 +116,22 @@ def _bindings(env):
     values = json.loads(raw) if raw else [{}]
     if not isinstance(values, list) or not values or not all(isinstance(v, dict) for v in values):
         raise ValueError("candidate_bindings_invalid")
-    return [{**env, **v, "DIGEST_TARGET_BINDINGS_JSON": ""} for v in values]
+    bindings = [{**env, **v, "DIGEST_TARGET_BINDINGS_JSON": ""} for v in values]
+    override = env.get("DIGEST_CANDIDATES_OVERRIDE_GCS_URI", "")
+    if override:
+        override = override.rstrip("/") if isinstance(override, str) else override
+        if (not isinstance(override, str)
+                or not re.fullmatch(r"gs://[a-z0-9][a-z0-9._-]+/[^\s?#*]+", override)
+                or any(part in {"", ".", ".."} for part in override[5:].rstrip("/").split("/")[1:])):
+            raise ValueError("candidate_store_configuration_missing")
+        for binding in bindings:
+            marker = binding.get("QSL_NOTIFICATION_MARKER_GCS_URI") or binding.get("DIGEST_CANDIDATES_GCS_URI")
+            if not marker:
+                raise ValueError("candidate_store_configuration_missing")
+            binding["QSL_NOTIFICATION_MARKER_GCS_URI"] = marker
+            binding["DIGEST_CANDIDATES_GCS_URI"] = override
+            binding["DIGEST_CANDIDATES_OVERRIDE_GCS_URI"] = override
+    return bindings
 
 
 def _binding(env, report):

tests/test_notification_scoped_state.py

@@ -1,11 +1,64 @@
 """Scoped notification state keeps the shared candidate writer narrowly bound."""
 import re
+import json
+
+import pytest
 
 from notifications import unified
 from test_notification_consolidation import ENV, NOW, Store, notify, source
 
 
 STATE = "gs://synthetic/notification-markers/firstrade/synthetic-account"
+ISOLATED = "gs://synthetic-isolated/candidates"
+
+
+def test_candidate_override_wins_over_private_binding_and_preserves_marker_default():
+    old = "gs://synthetic-bound/candidates"
+    env = {**ENV, "DIGEST_CANDIDATES_OVERRIDE_GCS_URI": ISOLATED,
+           "DIGEST_TARGET_BINDINGS_JSON": json.dumps([{"DIGEST_CANDIDATES_GCS_URI": old}])}
+    binding, = unified._bindings(env)
+    assert binding["DIGEST_CANDIDATES_GCS_URI"] == ISOLATED
+    assert unified._state_prefix(binding) == old
+    assert unified._bindings(binding) == [binding]
+    assert json.loads(env["DIGEST_TARGET_BINDINGS_JSON"])[0]["DIGEST_CANDIDATES_GCS_URI"] == old
+
+
+def test_candidate_override_preserves_existing_incident_deduplication():
+    store, sent = Store(), []
+    report, result = source(status="blocked", execution_blocked=True)
+    env = {**ENV, "QSL_NOTIFICATION_MARKER_GCS_URI": STATE}
+    notify(store, report, result, sent, env=env)
+    migrated = {**env, "DIGEST_CANDIDATES_OVERRIDE_GCS_URI": ISOLATED}
+    repeated = notify(store, report, result, sent, env=migrated)
+    assert repeated["delivery_status"] == "duplicate"
+    assert len(sent) == 1
+    assert any(key.startswith(STATE + "/_notification_markers/") for key in store.data)
+    binding, = unified._bindings(migrated)
+    assert binding["DIGEST_CANDIDATES_GCS_URI"] == ISOLATED
+    assert unified._state_prefix(binding) == STATE
+
+
+def test_candidate_override_creates_and_updates_only_the_selected_root():
+    store, sent = Store(), []
+    env = {**ENV, "DIGEST_CANDIDATES_OVERRIDE_GCS_URI": ISOLATED}
+    for run_id in ("first-run", "second-run"):
+        report, result = source(run_id)
+        assert notify(store, report, result, sent, env=env)["route"] == "digest_candidate"
+    candidates = [(key, value) for key, value in store.data.items()
+                  if value[1].get("schema_version") == "qsl.digest_candidates.v1"]
+    assert len(candidates) == 1
+    key, (generation, body) = candidates[0]
+    assert key.startswith(ISOLATED + "/firstrade/")
+    assert generation == 2
+    assert {row["source_run_id"] for row in body["runs"]} == {"first-run", "second-run"}
+    assert not sent
+
+
+@pytest.mark.parametrize("uri", ["https://synthetic/path", "gs://synthetic", "gs://synthetic///", "gs://synthetic/a/../b",
+                                 "gs://synthetic/a?private=value", "gs://synthetic/a/*"])
+def test_invalid_candidate_override_fails_before_storage(uri):
+    with pytest.raises(ValueError, match="^candidate_store_configuration_missing$"):
+        unified._bindings({**ENV, "DIGEST_CANDIDATES_OVERRIDE_GCS_URI": uri})
 
 
 class ScopedStore(Store):