feat: route digest candidates by platform

docs/unified-daily-notification.md

@@ -110,6 +110,7 @@ Required protected configuration:
 | `QRS_UNIFIED_NOTIFICATION_ENABLED` | Exact `true` |
 | `DIGEST_EXPECTED_TARGETS_JSON` | Approved identity rows, each with its existing `candidate_slot` |
 | `DIGEST_CANDIDATES_GCS_URI` | Existing private candidates prefix |
+| `DIGEST_CANDIDATES_BY_PLATFORM_JSON` | Optional protected map from expected platform ID to its candidate root URI; default `{}` |
 | `DIGEST_DELIVERIES_GCS_URI` | Private prefix for daily delivery state |
 | `DIGEST_SCHEDULER_JOB_NAME` | Full approved Scheduler job resource name |
 | `DIGEST_RECIPIENT_SCOPE` | Stable recipient scope, preserved across releases/locales |
@@ -119,6 +120,32 @@ Required protected configuration:
 | `QSL_GLOBAL_TELEGRAM_CHAT_ID` | Existing approved recipient |
 | `QSL_NOTIFY_LANG` | `zh` or `en`, including normalized regional variants |
 
+### Separate candidate buckets
+
+Each platform reads exactly one candidate root. An entry in
+`DIGEST_CANDIDATES_BY_PLATFORM_JSON` overrides the default root for that platform;
+the object suffix remains `<platform>/<business-day>/<candidate-slot>.json`.
+Unmapped platforms keep `DIGEST_CANDIDATES_GCS_URI`. Invalid URIs, unknown
+platforms and malformed maps fail startup. A missing, denied or timed-out read
+from an explicit root never retries against the default root. Delivery receipts
+and producer notification markers retain their existing locations and keys.
+
+For a migration, first deploy support with the map empty. Outside publishing
+and digest windows, pause candidate writers, wait for in-flight writes, and
+copy the selected objects with generation-pinned reads and checksum verification.
+Switch all producer entries, including scheduled jobs and protected workflow
+configuration, while preserving explicit notification marker URIs. Validate
+creation, same-day update and independent readback before changing the central
+map. Use authenticated `/preview` to verify identity, business day and coverage
+without reserving or sending. Keep other platform sources and delivery settings
+unchanged. Observe at least two successful natural digest windows before removing
+superseded shared-bucket grants.
+
+After switching, retain the explicit new source during application rollback;
+use a compatible image and repair forward. Do not silently return to the old
+bucket or delete delivery reservations or notification markers. If validation
+fails before switching, leave the current authoritative source in place.
+
 ### Delivery reservation
 
 The key is `<recipient_scope>/<SGT-report-date>.json` under the deliveries

python/scripts/cloud_run_daily_digest.py

@@ -10,7 +10,7 @@
 import urllib.parse
 import urllib.request
 from copy import deepcopy
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import UTC, date, datetime, time, timedelta
 from pathlib import Path
 from uuid import uuid4
@@ -19,7 +19,7 @@
 import daily_digest_aggregator as aggregator
 import merge_digest_candidates as candidate_contract
 import notification_candidate_adapter as adapter
-from daily_digest_delivery_store import GCSDeliveryStore, StoreUnavailable
+from daily_digest_delivery_store import GCSDeliveryStore, StoreUnavailable, location
 from display_names import display_name
 from unified_daily_digest import _safe_display_text, render
 
@@ -36,9 +36,21 @@ class Configuration:
     cutover_date: str
     enabled: bool = False
     locale: str = 'zh'
+    candidate_uris: dict = field(default_factory=dict)
 
     def __post_init__(self):
         adapter.validate_expected(self.expected)
+        platforms = {row['platform_id'] for row in self.expected}
+        if not isinstance(self.candidate_uris, dict) or any(
+                platform not in platforms for platform in self.candidate_uris):
+            raise ValueError('digest_candidate_sources_invalid')
+        for uri in self.candidate_uris.values():
+            try:
+                if not isinstance(uri, str):
+                    raise TypeError()
+                location(uri)
+            except (TypeError, ValueError):
+                raise ValueError('digest_candidate_sources_invalid') from None
         identities, objects = set(), set()
         for row in self.expected:
             if not SLOT.fullmatch(row.get('candidate_slot', '')):
@@ -66,10 +78,14 @@ def from_environment(cls):
         flag = os.environ.get('DIGEST_DELIVERY_ENABLED', 'false')
         if flag not in {'true', 'false'}:
             raise ValueError('digest_delivery_flag_invalid')
+        try:
+            candidate_uris = json.loads(os.environ.get('DIGEST_CANDIDATES_BY_PLATFORM_JSON', '{}'))
+        except ValueError:
+            raise ValueError('digest_candidate_sources_invalid') from None
         return cls(json.loads(os.environ['DIGEST_EXPECTED_TARGETS_JSON']),
                    os.environ['DIGEST_SCHEDULER_JOB_NAME'], os.environ['DIGEST_RECIPIENT_SCOPE'],
                    os.environ['DIGEST_CUTOVER_REPORT_DATE'], flag == 'true',
-                   os.environ.get('QSL_NOTIFY_LANG', 'zh'))
+                   os.environ.get('QSL_NOTIFY_LANG', 'zh'), candidate_uris)
 
 
 def utc_now():
@@ -284,7 +300,8 @@ def send(text):
 def main():
     from http.server import BaseHTTPRequestHandler, HTTPServer
     configuration = Configuration.from_environment()
-    store = GCSDeliveryStore(os.environ['DIGEST_CANDIDATES_GCS_URI'], os.environ['DIGEST_DELIVERIES_GCS_URI'])
+    store = GCSDeliveryStore(os.environ['DIGEST_CANDIDATES_GCS_URI'], os.environ['DIGEST_DELIVERIES_GCS_URI'],
+                             candidates_by_platform=configuration.candidate_uris)
     send = telegram_sender(os.environ['TELEGRAM_TOKEN'], os.environ['QSL_GLOBAL_TELEGRAM_CHAT_ID'])
 
     class Handler(BaseHTTPRequestHandler):

python/scripts/daily_digest_delivery_store.py

@@ -20,18 +20,23 @@ def location(uri):
 
 
 class GCSDeliveryStore:
-    def __init__(self, candidates_uri, deliveries_uri, *, client=None):
+    def __init__(self, candidates_uri, deliveries_uri, *, candidates_by_platform=None, client=None):
         from google.cloud import storage
         self.client = client or storage.Client()
         self.candidates_bucket, self.candidates_prefix = location(candidates_uri)
+        self.candidates_by_platform = {
+            platform: location(uri) for platform, uri in (candidates_by_platform or {}).items()
+        }
         self.deliveries_bucket, self.deliveries_prefix = location(deliveries_uri)
 
     def _blob(self, bucket, prefix, key):
         return self.client.bucket(bucket).blob(prefix + '/' + key)
 
     def candidate(self, key, timeout):
         from google.api_core.exceptions import NotFound
-        blob = self._blob(self.candidates_bucket, self.candidates_prefix, key)
+        bucket, prefix = self.candidates_by_platform.get(
+            key.split('/', 1)[0], (self.candidates_bucket, self.candidates_prefix))
+        blob = self._blob(bucket, prefix, key)
         try:
             body = blob.download_as_bytes(retry=None, timeout=timeout, if_generation_match=None)
             if len(body) > 2_000_000:

python/tests/test_cloud_run_daily_digest.py

@@ -98,6 +98,35 @@ def test_locale_change_does_not_create_a_second_daily_delivery(self):
         self.assertNotRegex(self.sent[0], r"[\u4e00-\u9fff]")
         self.assertEqual(list(self.store.records), ["synthetic/2026-10-16.json"])
 
+    def test_candidate_source_map_is_optional_and_validated_at_startup(self):
+        env = {'DIGEST_EXPECTED_TARGETS_JSON': json.dumps(EXPECTED),
+               'DIGEST_SCHEDULER_JOB_NAME': JOB, 'DIGEST_RECIPIENT_SCOPE': 'synthetic',
+               'DIGEST_CUTOVER_REPORT_DATE': '2026-10-16'}
+        with patch.dict(os.environ, env):
+            self.assertEqual(app.Configuration.from_environment().candidate_uris, {})
+            source = {ROW['platform_id']: 'gs://synthetic-isolated/candidates'}
+            with patch.dict(os.environ, DIGEST_CANDIDATES_BY_PLATFORM_JSON=json.dumps(source)):
+                self.assertEqual(app.Configuration.from_environment().candidate_uris, source)
+            for raw in ('', '{private', 'null', '[]', 'true',
+                        '{"unknown": "gs://synthetic/candidates"}',
+                        json.dumps({ROW['platform_id']: None}),
+                        json.dumps({ROW['platform_id']: 'gs://synthetic'}),
+                        json.dumps({ROW['platform_id']: 'gs://synthetic/a/../b'})):
+                with (self.subTest(raw=raw), patch.dict(os.environ, DIGEST_CANDIDATES_BY_PLATFORM_JSON=raw),
+                      self.assertRaisesRegex(ValueError, '^digest_candidate_sources_invalid$')):
+                    app.Configuration.from_environment()
+
+    def test_candidate_source_change_preserves_preview_and_delivery_reservation(self):
+        from dataclasses import replace
+        self.run_request(configuration=self.enable())
+        isolated = replace(self.enable(), candidate_uris={ROW['platform_id']: 'gs://synthetic-isolated/candidates'})
+        _, preview = self.run_request(preview=True, configuration=isolated)
+        self.assertEqual(preview['delivery'], 'preview')
+        _, repeated = self.run_request(configuration=isolated)
+        self.assertEqual(repeated['delivery'], 'duplicate')
+        self.assertEqual(len(self.sent), 1)
+        self.assertEqual(list(self.store.records), ['synthetic/2026-10-16.json'])
+
     def test_preview_never_reserves_or_sends_and_returns_no_financial_body(self):
         self.store.rows[next(iter(self.store.rows))]['runs'][0] = dict(ROW, equity_usd=123456)
         status, result = self.run_request(preview=True)

python/tests/test_daily_digest_delivery_store.py

@@ -60,6 +60,41 @@ def test_missing_is_not_conflated_with_read_failure(self):
         with self.assertRaises(StoreUnavailable):
             self.store.candidate('key', 5)
 
+    def test_platform_root_changes_only_selected_candidate_reads(self):
+        store = GCSDeliveryStore('gs://synthetic/candidates', 'gs://synthetic/deliveries',
+                                 candidates_by_platform={'firstrade': 'gs://synthetic-isolated/intake'},
+                                 client=self.client)
+        self.blob.download_as_bytes.return_value = b'{"runs": []}'
+        for platform, bucket, prefix in (
+                ('firstrade', 'synthetic-isolated', 'intake'), ('ibkr', 'synthetic', 'candidates')):
+            key = f'{platform}/2026-10-15/slot.json'
+            self.assertEqual(store.candidate(key, 5), {'runs': []})
+            self.client.bucket.assert_called_with(bucket)
+            self.client.bucket.return_value.blob.assert_called_with(prefix + '/' + key)
+            self.assertEqual(self.blob.download_as_bytes.call_args.kwargs['timeout'], 5)
+            self.assertIsNone(self.blob.download_as_bytes.call_args.kwargs['retry'])
+        store.reserve('scope/2026-10-16.json', {'delivery': 'sending'})
+        self.client.bucket.assert_called_with('synthetic')
+        self.client.bucket.return_value.blob.assert_called_with('deliveries/scope/2026-10-16.json')
+        self.assertEqual(self.blob.upload_from_string.call_args.kwargs['if_generation_match'], 0)
+
+    def test_selected_root_never_falls_back_after_missing_denied_or_timeout(self):
+        store = GCSDeliveryStore('gs://synthetic/candidates', 'gs://synthetic/deliveries',
+                                 candidates_by_platform={'firstrade': 'gs://synthetic-isolated/intake'},
+                                 client=self.client)
+        for error in (Missing(), PermissionError('private-path'), TimeoutError('private-path')):
+            with self.subTest(error=type(error).__name__):
+                self.client.reset_mock()
+                self.blob.download_as_bytes.side_effect = error
+                if isinstance(error, Missing):
+                    self.assertIsNone(store.candidate('firstrade/day/slot.json', 5))
+                else:
+                    with self.assertRaises(StoreUnavailable) as caught:
+                        store.candidate('firstrade/day/slot.json', 5)
+                    self.assertNotIn('private-path', str(caught.exception))
+                self.client.bucket.assert_called_once_with('synthetic-isolated')
+                self.blob.download_as_bytes.assert_called_once()
+
     def test_unreadable_existing_reservation_does_not_become_absent(self):
         self.blob.upload_from_string.side_effect = Conflict()
         self.blob.download_as_bytes.return_value = b'not-json'