fix: publish digest candidates before the central daily read

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.

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

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

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