From f1ab913a08ba3b02cf98e43b8bd02a83b23acc2f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:15:46 +0800 Subject: [PATCH 1/2] feat: route digest candidates by platform Co-Authored-By: Codex --- docs/unified-daily-notification.md | 27 ++++++++++++++ python/scripts/cloud_run_daily_digest.py | 25 ++++++++++--- python/scripts/daily_digest_delivery_store.py | 9 +++-- python/tests/test_cloud_run_daily_digest.py | 29 +++++++++++++++ .../tests/test_daily_digest_delivery_store.py | 35 +++++++++++++++++++ 5 files changed, 119 insertions(+), 6 deletions(-) diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index a07c35dd..50777f9a 100644 --- a/docs/unified-daily-notification.md +++ b/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 `//.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 `/.json` under the deliveries diff --git a/python/scripts/cloud_run_daily_digest.py b/python/scripts/cloud_run_daily_digest.py index c6f2fc23..3d7a40aa 100644 --- a/python/scripts/cloud_run_daily_digest.py +++ b/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 ValueError() + location(uri) + except 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): diff --git a/python/scripts/daily_digest_delivery_store.py b/python/scripts/daily_digest_delivery_store.py index f07567d8..2c644f58 100644 --- a/python/scripts/daily_digest_delivery_store.py +++ b/python/scripts/daily_digest_delivery_store.py @@ -20,10 +20,13 @@ 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): @@ -31,7 +34,9 @@ def _blob(self, bucket, 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: diff --git a/python/tests/test_cloud_run_daily_digest.py b/python/tests/test_cloud_run_daily_digest.py index 962d34b8..b46c6ff3 100644 --- a/python/tests/test_cloud_run_daily_digest.py +++ b/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 = dict(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): + with 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) diff --git a/python/tests/test_daily_digest_delivery_store.py b/python/tests/test_daily_digest_delivery_store.py index 17187c7c..28c0cff2 100644 --- a/python/tests/test_daily_digest_delivery_store.py +++ b/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' From 6d7546b68928e9f012573d5f0a8646a34812d895 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 16:16:06 +0800 Subject: [PATCH 2/2] fix: normalize candidate map validation errors Co-Authored-By: Codex --- python/scripts/cloud_run_daily_digest.py | 4 ++-- python/tests/test_cloud_run_daily_digest.py | 12 ++++++------ 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/python/scripts/cloud_run_daily_digest.py b/python/scripts/cloud_run_daily_digest.py index 3d7a40aa..602553d6 100644 --- a/python/scripts/cloud_run_daily_digest.py +++ b/python/scripts/cloud_run_daily_digest.py @@ -47,9 +47,9 @@ def __post_init__(self): for uri in self.candidate_uris.values(): try: if not isinstance(uri, str): - raise ValueError() + raise TypeError() location(uri) - except ValueError: + except (TypeError, ValueError): raise ValueError('digest_candidate_sources_invalid') from None identities, objects = set(), set() for row in self.expected: diff --git a/python/tests/test_cloud_run_daily_digest.py b/python/tests/test_cloud_run_daily_digest.py index b46c6ff3..2d2f9573 100644 --- a/python/tests/test_cloud_run_daily_digest.py +++ b/python/tests/test_cloud_run_daily_digest.py @@ -99,9 +99,9 @@ def test_locale_change_does_not_create_a_second_daily_delivery(self): self.assertEqual(list(self.store.records), ["synthetic/2026-10-16.json"]) def test_candidate_source_map_is_optional_and_validated_at_startup(self): - env = dict(DIGEST_EXPECTED_TARGETS_JSON=json.dumps(EXPECTED), - DIGEST_SCHEDULER_JOB_NAME=JOB, DIGEST_RECIPIENT_SCOPE='synthetic', - DIGEST_CUTOVER_REPORT_DATE='2026-10-16') + 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'} @@ -112,9 +112,9 @@ def test_candidate_source_map_is_optional_and_validated_at_startup(self): 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): - with self.assertRaisesRegex(ValueError, '^digest_candidate_sources_invalid$'): - app.Configuration.from_environment() + 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