From 05fbe226185ef055abf0c0199f1825e44ebad25b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:08:01 +0800 Subject: [PATCH] fix: publish digest candidates independently before the daily read Co-Authored-By: Codex --- docs/unified-daily-notification.md | 72 +++++++ notifications/unified.py | 25 ++- scripts/emit_unified_digest_candidates.py | 7 +- scripts/publish_digest_before_read.py | 140 ++++++++++++ tests/test_publish_digest_before_read.py | 252 ++++++++++++++++++++++ 5 files changed, 492 insertions(+), 4 deletions(-) create mode 100644 scripts/publish_digest_before_read.py create mode 100644 tests/test_publish_digest_before_read.py diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index ca7af7c..c20a196 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -56,3 +56,75 @@ provider audit entries remain in their original restricted logs and are not 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. + +## Publication before the central daily read + +GitHub schedule events can start late. A heartbeat workflow is therefore not a +reliable deadline trigger for the 07:00 Asia/Shanghai read. Keep its existing +monitoring checks, but run the archive-only publisher independently in an +approved Cloud Run Job using the platform image and its frozen dependencies. + +Command: `/app/.venv/bin/python -m scripts.publish_digest_before_read`. +The command imports the existing producer, never the service entrypoint. It does +not run a drill, refresh a broker session, invoke a trading route, send Telegram, +or access Secret Manager. Archive reads use the Storage SDK and the job's +attached identity; the image does not need gcloud. Existing candidate writes, +merge semantics and identity checks remain authoritative. + +Use one job per existing central target, with its exact protected runtime JSON, +archive prefix, opaque UID, target ID, destination and slot. No identifiers can +be inferred from a display label. Supply the existing notification schedule so +an idle candidate is only produced when the completed window was not due. +Multi-target binding JSON is rejected by this entrypoint: one successful target +cannot mask another missing target. + +The job requires `DIGEST_PUBLISH_ENABLED=true`, +`QSL_ROUTINE_NOTIFICATION_MODE=daily_digest`, and the existing +`RUNTIME_TARGET_ENABLED=true`. These are publisher-job settings, not changes to +the trading service. Preserve the platform's existing unified-notification flag +and any notification marker prefix needed by its existing producer. Do not +inject broker credentials or Telegram token secrets into this job. + +The deployment proposal is: + +| Setting | Proposed value | +| --- | --- | +| Cloud Scheduler | `10 6 * * *`, timezone `Asia/Shanghai` | +| Scheduler target | OAuth-authenticated Cloud Run Jobs `:run` API for this job | +| Scheduler retries | 0; do not blindly retry an uncertain execution request | +| Job tasks / parallelism | 1 / 1 | +| Job retries | 0; an uncertain write is not proof of failure | +| Job task timeout | 600 seconds | +| Image | Reviewed immutable platform image containing this entrypoint | +| Publication window | Start at or after 06:00 and before 06:40 | +| In-process watchdog | 600 seconds, including projection and readback | +| Readback deadline | Before 06:55, leaving five minutes before the reader | + +The business day is the day before the job's Shanghai date, for the completed +[06:00, 06:00) window. A normal run starts at 06:10 and finishes by 06:20. A late +start, disabled/missing binding, due-but-absent report, unconfirmed write, wrong +business day or identity, failed readback, or expired deadline exits nonzero. +An interrupted write remains unknown and must be inspected before any retry. +The success record contains only the business day and a verified-readback flag. + +Deployment, Scheduler creation and IAM are separate operations requiring their +own authorization. The Scheduler identity needs permission to execute only the +job; the job identity needs the existing archive-read and candidate-store +permissions (including notification state used by the existing producer). +Do not substitute a project-wide role or copy the trading runtime's credentials. +A merged change or a successful unit test does not prove the job is deployed. + +After a separately authorized rollout, verify a natural job execution and the +exact candidate object generation created before 07:00, then check the central +reader's date, slot, identity and coverage. A Scheduler API acceptance alone is +not publication success. Keep missing/invalid coverage visible; do not substitute +old candidates, zeros, or an artificial `not_due` result. + +References: [GitHub schedule timing](https://docs.github.com/en/actions/reference/workflows-and-actions/events-that-trigger-workflows#schedule) +and [Cloud Run scheduled jobs](https://docs.cloud.google.com/run/docs/execute/jobs-on-schedule). + +Candidate persistence failures expose only a fixed `candidate_read` or +`candidate_write` stage and an allowlisted error category. Raw exception text, +resource paths and credential context stay out of command output. A permission +category identifies the failing storage operation; it does not establish that +an alerting Secret Manager denial caused the storage failure. diff --git a/notifications/unified.py b/notifications/unified.py index af52982..0652bfd 100644 --- a/notifications/unified.py +++ b/notifications/unified.py @@ -26,6 +26,20 @@ "account_selector", "deployment_selector", "dry_run_only") +class CandidateStorageError(RuntimeError): + """Fixed diagnostics only; never include a storage exception's message.""" + + def __init__(self, stage, error): + from google.api_core.exceptions import Forbidden, Unauthorized, PreconditionFailed + self.stage = stage + self.category = ("permission_denied" if isinstance(error, (Forbidden, Unauthorized)) else + "generation_conflict" if isinstance(error, PreconditionFailed) else + "timeout_unknown" if isinstance(error, TimeoutError) else + "serialization_invalid" if isinstance(error, (TypeError, ValueError)) else + "storage_unknown") + super().__init__("candidate_persistence_failed") + + def enabled(env=None): return unified_notification_enabled(os.environ if env is None else env, FLAG) @@ -191,7 +205,10 @@ def _aggregate(rows): def _merge(store, uri, rows, events, *, idle=None, archive=False): for _ in range(8): - version, previous = store.read(uri) + try: + version, previous = store.read(uri) + except Exception as error: + raise CandidateStorageError("candidate_read", error) from error # Read the former local ledger once for migration; new payloads store # all merge history in their per-run rows, as required by QRS. sources = dict((previous or {}).get("notification_source_runs") or {}) @@ -244,7 +261,11 @@ def _merge(store, uri, rows, events, *, idle=None, archive=False): "notification_events": by_run[run_id]} for run_id in sorted(sources)] payload = dict(schema_version="qsl.digest_candidates.v1", producer_status="projected", runs=run_rows if sources else [idle] if idle else []) - if store.compare_and_set(uri, version, payload): + try: + written = store.compare_and_set(uri, version, payload) + except Exception as error: + raise CandidateStorageError("candidate_write", error) from error + if written: return True raise RuntimeError("candidate_concurrent_update") diff --git a/scripts/emit_unified_digest_candidates.py b/scripts/emit_unified_digest_candidates.py index 7fc4ed5..178ed76 100644 --- a/scripts/emit_unified_digest_candidates.py +++ b/scripts/emit_unified_digest_candidates.py @@ -25,8 +25,11 @@ def main(): _check_binding_modes(os.environ) print(json.dumps(emit_candidates(os.environ), sort_keys=True)) return 0 - except Exception: - print(json.dumps({"status": "candidate_projection_failed", "uploaded": False})) + except Exception as error: + from notifications.unified import CandidateStorageError + diagnostic = ({"stage": error.stage, "error_category": error.category} + if isinstance(error, CandidateStorageError) else {}) + print(json.dumps({"status": "candidate_projection_failed", "uploaded": False, **diagnostic})) from notifications.unified import alert_text, coverage_locale from scripts.execution_report_heartbeat import _send_telegram marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH") diff --git a/scripts/publish_digest_before_read.py b/scripts/publish_digest_before_read.py new file mode 100644 index 0000000..c416209 --- /dev/null +++ b/scripts/publish_digest_before_read.py @@ -0,0 +1,140 @@ +"""Publish one completed-window candidate before the central read, without broker calls.""" +from __future__ import annotations + +import json +import os +import signal +from datetime import datetime, time, timedelta, timezone +from zoneinfo import ZoneInfo + +SHANGHAI = ZoneInfo("Asia/Shanghai") +PLATFORM = "firstrade" + + +class DeadlineExceeded(BaseException): + """Escape producer catch-all handlers; an interrupted write stays unknown.""" + + +def publication_day(now): + if now.tzinfo is None: + raise ValueError("candidate_timezone_required") + local = now.astimezone(SHANGHAI) + if not time(6) <= local.time() < time(6, 40): + raise ValueError("candidate_publication_window_closed") + return (local.date() - timedelta(days=1)).isoformat() + + +class ArchiveReader: + """Native Storage reads with the job identity; no gcloud or secret access.""" + + def __init__(self, client): + self.client = client + + def read(self, args): + uri = args[-1] + if not uri.startswith("gs://"): + raise ValueError("candidate_archive_uri_invalid") + bucket, separator, key = uri[5:].partition("/") + if not bucket or not separator or not key: + raise ValueError("candidate_archive_uri_invalid") + if args[0] == "ls": + blobs = self.client.list_blobs( + bucket, prefix=key.split("*", 1)[0], match_glob=key, + max_results=10001, retry=None, timeout=30, + ) + rows = [] + for blob in blobs: + rows.append({"url": f"gs://{bucket}/{blob.name}", + "metadata": {"updated": blob.updated.isoformat()}}) + if len(rows) > 10000: + raise ValueError("candidate_archive_budget_exceeded") + return rows + if args[0] != "cat": + raise ValueError("candidate_archive_operation_invalid") + blob = self.client.bucket(bucket).blob(key) + blob.reload(retry=None, timeout=30) + if blob.size is None or blob.size > 2 * 1024**2: + raise ValueError("candidate_archive_object_too_large") + return json.loads(blob.download_as_bytes( + if_generation_match=int(blob.generation), retry=None, timeout=30, + )) + + +def _emit(env, *, now, read): + from notifications.unified import emit_candidates + return emit_candidates(env, now=now, read=read) + + +def publish(env, *, now=None, client=None, emit=None): + clock = now or (lambda: datetime.now(timezone.utc)) + started = clock() + day = publication_day(started) + if (env.get("DIGEST_PUBLISH_ENABLED") != "true" + or env.get("QSL_ROUTINE_NOTIFICATION_MODE") != "daily_digest" + or env.get("RUNTIME_TARGET_ENABLED") != "true"): + raise ValueError("candidate_publication_not_enabled") + if str(env.get("DIGEST_TARGET_BINDINGS_JSON") or "").strip(): + raise ValueError("candidate_publication_single_target_required") + target = json.loads(env["RUNTIME_TARGET_JSON"]) + required = ("strategy_profile", "service_name", "account_scope", "deployment_selector") + if (not isinstance(target, dict) or target.get("platform_id") != PLATFORM + or any(not isinstance(target.get(key), str) or not target[key].strip() for key in required) + or type(target.get("dry_run_only")) is not bool + or not isinstance(target.get("account_selector"), list) or not target["account_selector"] + or any(not isinstance(value, str) or not value.strip() for value in target["account_selector"])): + raise ValueError("candidate_publication_target_invalid") + keys = ("DIGEST_OPAQUE_ACCOUNT_UID", "DIGEST_TARGET_ID", "DIGEST_STORE_TARGET", + "DIGEST_CANDIDATES_GCS_URI", "EXECUTION_REPORT_GCS_URI") + if any(not isinstance(env.get(key), str) or not env[key].strip() for key in keys): + raise ValueError("candidate_publication_binding_missing") + if client is None: + from google.cloud import storage + client = storage.Client() + reader = ArchiveReader(client) + result = (emit or _emit)(env, now=started, read=reader.read) + if result.get("status") != "projected" or result.get("uploaded") is not True: + raise ValueError("candidate_publication_unconfirmed") + identity = {"platform_id": target["platform_id"], + "strategy_profile": target["strategy_profile"], + "opaque_account_uid": env["DIGEST_OPAQUE_ACCOUNT_UID"], + "target_id": env["DIGEST_TARGET_ID"], "business_day": day} + uri = (f"{env['DIGEST_CANDIDATES_GCS_URI'].rstrip('/')}/{identity['platform_id']}/" + f"{day}/{env['DIGEST_STORE_TARGET']}.json") + body = reader.read(["cat", uri]) + rows = body.get("runs") + if (body.get("schema_version") != "qsl.digest_candidates.v1" + or not isinstance(rows, list) or not rows + or any(not isinstance(row, dict) + or any(row.get(key) != value for key, value in identity.items()) + or type(row.get("actually_ran")) is not bool for row in rows)): + raise ValueError("candidate_publication_readback_invalid") + finished = clock().astimezone(SHANGHAI) + if finished.date() != started.astimezone(SHANGHAI).date() or finished.time() >= time(6, 55): + raise ValueError("candidate_publication_deadline_missed") + return {"status": "published", "business_day": day, "readback_verified": True} + + +def main(): + def timeout(signum, frame): + raise DeadlineExceeded() + + previous = signal.signal(signal.SIGALRM, timeout) + signal.alarm(600) + try: + print(json.dumps(publish(os.environ), sort_keys=True)) + return 0 + except (Exception, DeadlineExceeded) as error: + # Never expose exception messages, private locators or credentials. + from notifications.unified import CandidateStorageError + diagnostic = ({"stage": error.stage, "error_category": error.category} + if isinstance(error, CandidateStorageError) else {}) + print(json.dumps({"status": "failed", "reason_code": "candidate_publication_unconfirmed", + **diagnostic})) + return 1 + finally: + signal.alarm(0) + signal.signal(signal.SIGALRM, previous) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_publish_digest_before_read.py b/tests/test_publish_digest_before_read.py new file mode 100644 index 0000000..9f224d8 --- /dev/null +++ b/tests/test_publish_digest_before_read.py @@ -0,0 +1,252 @@ +from datetime import datetime, timezone +import json +import fnmatch +from types import SimpleNamespace + +import pytest +from google.api_core.exceptions import Forbidden, NotFound, PreconditionFailed +from google.cloud import storage + +from scripts import publish_digest_before_read as publisher + +NOW = datetime(2026, 10, 31, 22, 10, tzinfo=timezone.utc) +PLATFORM = "firstrade" +ENV = { + "DIGEST_PUBLISH_ENABLED": "true", "RUNTIME_TARGET_ENABLED": "true", + "QSL_ROUTINE_NOTIFICATION_MODE": "daily_digest", + "RUNTIME_TARGET_JSON": json.dumps({"platform_id": PLATFORM, + "strategy_profile": "synthetic-strategy", "service_name": "synthetic-service", + "account_scope": "synthetic-scope", "account_selector": ["synthetic-account"], + "deployment_selector": "synthetic-deployment", "dry_run_only": False}), + "DIGEST_CANDIDATES_GCS_URI": "gs://synthetic-candidates/digest", + "EXECUTION_REPORT_GCS_URI": "gs://synthetic-archive/reports", + "DIGEST_STORE_TARGET": "synthetic-slot", "DIGEST_OPAQUE_ACCOUNT_UID": "synthetic-uid", + "DIGEST_TARGET_ID": "synthetic-target", + "DIGEST_NOTIFICATION_SCHEDULE_JSON": json.dumps({"scheduler": { + "main_time": "45 15 * * 1-5", "timezone": "America/New_York"}}), +} + + +class Blob: + def __init__(self, name): + self.name, self.body, self.generation, self.size = name, None, None, None + self.metadata = {} + self.updated = NOW + self.writes = 0 + + def reload(self, **kwargs): + assert kwargs == {"retry": None, "timeout": 30} + if self.body is None: + raise NotFound("synthetic missing object") + + def download_as_bytes(self, **kwargs): + self.reload(retry=None, timeout=30) + assert kwargs["retry"] is None + if "if_generation_match" in kwargs: + assert kwargs["if_generation_match"] == self.generation + return self.body.encode() + + def download_as_text(self, **kwargs): + return self.download_as_bytes(**kwargs).decode() + + def upload_from_string(self, body, **kwargs): + assert kwargs["retry"] is None + if "if_generation_match" in kwargs and kwargs["if_generation_match"] != (self.generation or 0): + raise PreconditionFailed("synthetic conflict") + self.body, self.size = body, len(body.encode()) + self.generation = (self.generation or 0) + 1 + self.writes += 1 + + +class Client: + def __init__(self): + self.blobs, self.lists = {}, [] + + def bucket(self, name): + return SimpleNamespace(blob=lambda key: self.blob(f"gs://{name}/{key}")) + + def blob(self, uri): + return self.blobs.setdefault(uri, Blob(uri)) + + def list_blobs(self, bucket, **kwargs): + self.lists.append((bucket, kwargs)) + assert kwargs["retry"] is None and kwargs["max_results"] == 10001 + prefix = f"gs://{bucket}/" + return [SimpleNamespace(name=uri[len(prefix):], updated=blob.updated) + for uri, blob in self.blobs.items() + if uri.startswith(prefix) and fnmatch.fnmatchcase(uri[len(prefix):], kwargs["match_glob"])] + + +@pytest.fixture +def client(monkeypatch): + value = Client() + monkeypatch.setattr(storage, "Client", lambda: value) + monkeypatch.setattr(storage.Blob, "from_string", lambda uri, **kwargs: value.blob(uri)) + return value + + +def test_real_projector_publishes_completed_weekend_then_reads_exact_object(client): + result = publisher.publish(ENV, now=lambda: NOW, client=client) + assert result == {"status": "published", "business_day": "2026-10-31", "readback_verified": True} + uri = f"gs://synthetic-candidates/digest/{PLATFORM}/2026-10-31/synthetic-slot.json" + rows = json.loads(client.blobs[uri].body)["runs"] + assert rows[0]["actually_ran"] is False + assert rows[0].get("order_count") is None and rows[0].get("fill_count") is None + assert client.lists + + +@pytest.mark.parametrize("hour,minute", [(21, 59), (22, 40), (23, 0), (1, 16)]) +def test_early_or_delayed_start_never_calls_emitter(hour, minute): + now = NOW.replace(hour=hour, minute=minute) + with pytest.raises(ValueError, match="publication_window_closed"): + publisher.publish(ENV, now=lambda: now, + emit=lambda *a, **kw: pytest.fail("producer must not run")) + + +@pytest.mark.parametrize("key,value", [ + ("DIGEST_PUBLISH_ENABLED", "false"), ("RUNTIME_TARGET_ENABLED", "false"), + ("QSL_ROUTINE_NOTIFICATION_MODE", "immediate"), + ("DIGEST_TARGET_BINDINGS_JSON", "[{}]"), ("DIGEST_STORE_TARGET", ""), +]) +def test_configuration_failure_prevents_io(key, value): + with pytest.raises(ValueError): + publisher.publish({**ENV, key: value}, now=lambda: NOW, + emit=lambda *a, **kw: pytest.fail("producer must not run")) + + +def test_due_but_absent_does_not_publish_idle(client): + env = {**ENV, "DIGEST_NOTIFICATION_SCHEDULE_JSON": json.dumps({"scheduler": { + "main_time": "45 15 * * *", "timezone": "America/New_York"}})} + with pytest.raises(ValueError, match="candidate_due_source_absent"): + publisher.publish(env, now=lambda: NOW, client=client) + assert not any(blob.writes for blob in client.blobs.values()) + + +@pytest.mark.parametrize("body", [ + {"runs": []}, {"runs": [{"actually_ran": False, "business_day": "2026-10-30"}]}, + {"runs": [{"actually_ran": False, "business_day": "2026-10-31", "target_id": "wrong"}]}, +]) +def test_success_without_valid_readback_fails(client, body): + def emit(*a, **kw): + uri = f"gs://synthetic-candidates/digest/{PLATFORM}/2026-10-31/synthetic-slot.json" + client.blob(uri).upload_from_string(json.dumps(body), retry=None) + return {"status": "projected", "uploaded": True} + with pytest.raises(ValueError, match="readback_invalid"): + publisher.publish(ENV, now=lambda: NOW, client=client, emit=emit) + + +def test_unconfirmed_upload_never_reports_success(client): + with pytest.raises(ValueError, match="publication_unconfirmed"): + publisher.publish(ENV, now=lambda: NOW, client=client, + emit=lambda *a, **kw: {"status": "projected", "uploaded": False}) + + +def test_unknown_write_is_not_retried(client): + calls = [] + def emit(*a, **kw): + calls.append(1) + raise TimeoutError("synthetic sensitive transport detail") + with pytest.raises(TimeoutError): + publisher.publish(ENV, now=lambda: NOW, client=client, emit=emit) + assert len(calls) == 1 + + +def test_read_permission_denial_is_not_missing(client, monkeypatch): + monkeypatch.setattr(client, "list_blobs", lambda *a, **kw: (_ for _ in ()).throw(Forbidden("private"))) + with pytest.raises(Forbidden): + publisher.publish(ENV, now=lambda: NOW, client=client) + assert not any(blob.writes for blob in client.blobs.values()) + + +def test_late_finish_cannot_be_reported_as_timely(client): + moments = iter([NOW, NOW.replace(hour=22, minute=55)]) + with pytest.raises(ValueError, match="deadline_missed"): + publisher.publish(ENV, now=lambda: next(moments), client=client) + + +def test_main_redacts_exception_and_clears_watchdog(monkeypatch, capsys): + alarms = [] + monkeypatch.setattr(publisher.signal, "signal", lambda *a: None) + monkeypatch.setattr(publisher.signal, "alarm", alarms.append) + monkeypatch.setattr(publisher, "publish", lambda *a: (_ for _ in ()).throw(RuntimeError("private-value"))) + assert publisher.main() == 1 + output = capsys.readouterr().out + assert "private-value" not in output + assert json.loads(output)["status"] == "failed" + assert alarms == [600, 0] + + +def test_watchdog_escapes_producer_exception_handlers(monkeypatch, capsys): + monkeypatch.setattr(publisher.signal, "signal", lambda *a: None) + monkeypatch.setattr(publisher.signal, "alarm", lambda *a: None) + def emit(*a): + try: + raise publisher.DeadlineExceeded() + except Exception: + pytest.fail("deadline must escape producer catch-all") + monkeypatch.setattr(publisher, "publish", emit) + assert publisher.main() == 1 + assert json.loads(capsys.readouterr().out)["status"] == "failed" + + +@pytest.mark.parametrize("stage", ["candidate_read", "candidate_write"]) +def test_persistence_diagnostic_identifies_operation_without_raw_error(stage, monkeypatch, capsys): + from notifications.unified import _merge, CandidateStorageError + calls = [] + class Store: + def read(self, uri): + calls.append("read") + if stage == "candidate_read": + raise Forbidden("private resource and credential context") + return None, None + def compare_and_set(self, *args): + calls.append("write") + raise Forbidden("private resource and credential context") + with pytest.raises(CandidateStorageError) as caught: + _merge(Store(), "gs://synthetic/path", {}, {}, idle={"actually_ran": False}) + assert caught.value.stage == stage + assert caught.value.category == "permission_denied" + assert len(calls) == (1 if stage == "candidate_read" else 2) + def fail(*a): + raise caught.value + monkeypatch.setattr(publisher, "publish", fail) + monkeypatch.setattr(publisher.signal, "signal", lambda *a: None) + monkeypatch.setattr(publisher.signal, "alarm", lambda *a: None) + assert publisher.main() == 1 + output = capsys.readouterr().out + assert "private resource" not in output + assert json.loads(output)["stage"] == stage + + +def test_existing_business_report_uses_native_archive_reader_and_verified_publication(client): + from notifications.digest_snapshot.digest_candidates import cycle_evidence + env = {**ENV, "FIRSTRADE_UNIFIED_NOTIFICATION_ENABLED": "true", + "RUNTIME_HEARTBEAT_REPORT_PLATFORM": "firstrade", + "IBKR_NOTIFICATION_MARKER_GCS_URI": "gs://synthetic-candidates/markers"} + target = json.loads(env["RUNTIME_TARGET_JSON"]) + report = {"schema_version": "runtime_report.v1", "platform": "firstrade", + "strategy_profile": target["strategy_profile"], "service_name": target["service_name"], + "runtime_target": target, "run_id": "synthetic-run", "dry_run": False, + "status": "ok", "started_at": "2026-10-31T20:00:00+00:00", + "summary": {"notification_digest": cycle_evidence( + {"execution_status": "no_action", "broker_submission_done": False}, [])}} + archive = client.blob("gs://synthetic-archive/reports/firstrade/synthetic/2026-10/run.json") + archive.upload_from_string(json.dumps(report), retry=None) + result = publisher.publish(env, now=lambda: NOW, client=client) + assert result["readback_verified"] is True + uri = f"gs://synthetic-candidates/digest/{PLATFORM}/2026-10-31/synthetic-slot.json" + rows = json.loads(client.blobs[uri].body)["runs"] + assert rows and all(row["actually_ran"] is True for row in rows) + assert all(row["business_day"] == "2026-10-31" for row in rows) + + +@pytest.mark.parametrize("key,value", [ + ("platform_id", "another-platform"), ("service_name", ""), + ("account_scope", None), ("dry_run_only", "false"), + ("account_selector", []), ("account_selector", "synthetic-account"), +]) +def test_invalid_runtime_identity_is_rejected_before_idle_projection(key, value): + target = {**json.loads(ENV["RUNTIME_TARGET_JSON"]), key: value} + with pytest.raises(ValueError, match="target_invalid"): + publisher.publish({**ENV, "RUNTIME_TARGET_JSON": json.dumps(target)}, now=lambda: NOW, + emit=lambda *a, **kw: pytest.fail("must not project an invalid target"))