feat(notifications): opt-in unified digest without trading dependency upgrades

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

@@ -286,6 +286,22 @@ jobs:
         if: ${{ !cancelled() && env.INSPECT_SG_PROBE_PERMISSIONS != 'true' && steps.account_history.outcome == 'failure' }}
         run: exit 1
 
+
+      - name: Publish existing reports as unified digest candidates
+        if: ${{ success() && vars.QSL_ROUTINE_NOTIFICATION_MODE == 'daily_digest' }}
+        env:
+          QSL_ROUTINE_NOTIFICATION_MODE: ${{ vars.QSL_ROUTINE_NOTIFICATION_MODE || 'immediate' }}
+          RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}
+          RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON || secrets.RUNTIME_TARGET_JSON }}
+          EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI || secrets.EXECUTION_REPORT_GCS_URI }}
+          DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI || secrets.DIGEST_CANDIDATES_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 }}
+          DIGEST_TARGET_BINDINGS_JSON: ${{ secrets.DIGEST_TARGET_BINDINGS_JSON }}
+          DIGEST_NOTIFICATION_SCHEDULE_JSON: ${{ vars.DIGEST_NOTIFICATION_SCHEDULE_JSON }}
+        run: uv run --no-sync python scripts/emit_unified_digest_candidates.py
+
   inspect-account-snapshot:
     needs: resolve-matrix
     if: ${{ github.event_name == 'workflow_dispatch' && ((inputs.inspect_sg_account_snapshot == true && inputs.target == 'sg') || (inputs.inspect_paper_account_snapshot == true && inputs.target == 'paper')) }}

.github/workflows/sync-cloud-run-env.yml

@@ -683,6 +683,8 @@ jobs:
       IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW: ${{ vars.IBIT_ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW }}
       RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}
       NOTIFY_LANG: ${{ vars.NOTIFY_LANG }}
+      QSL_ROUTINE_NOTIFICATION_MODE: ${{ vars.QSL_ROUTINE_NOTIFICATION_MODE }}
+      QSL_DIGEST_ROUTINE_ORDERS: ${{ vars.QSL_DIGEST_ROUTINE_ORDERS }}
       EXECUTION_REPORT_GCS_URI: ${{ vars.EXECUTION_REPORT_GCS_URI }}
       LIFECYCLE_PERFORMANCE_BUCKET: ${{ vars.LIFECYCLE_PERFORMANCE_BUCKET }}
       LONGBRIDGE_DRY_RUN_ONLY: ${{ vars.LONGBRIDGE_DRY_RUN_ONLY }}
@@ -1241,6 +1243,15 @@ jobs:
           }
 
           env_pairs=("${target_env_pairs[@]}")
+          for notification_key in QSL_ROUTINE_NOTIFICATION_MODE QSL_DIGEST_ROUTINE_ORDERS; do
+            if [ -n "${!notification_key:-}" ]; then
+              case "${notification_key}:${!notification_key}" in
+                QSL_ROUTINE_NOTIFICATION_MODE:immediate|QSL_ROUTINE_NOTIFICATION_MODE:daily_digest|QSL_DIGEST_ROUTINE_ORDERS:true|QSL_DIGEST_ROUTINE_ORDERS:false) ;;
+                *) echo "Invalid notification routing configuration" >&2; exit 2 ;;
+              esac
+              env_pairs+=("${notification_key}=${!notification_key}")
+            fi
+          done
           secret_pairs=()
           remove_env_vars=("${target_remove_env_vars[@]}")
           remove_env_vars+=(

application/rebalance_service.py

@@ -633,6 +633,16 @@ def _record_execution_marker(
 
 
 
+def _defer_cycle_notification(result, *, config):
+    from notifications.digest_snapshot.digest_routing import defer_routine_cycle_to_digest
+    evidence = {**dict(result.allocation or {}), **dict(result.execution or {})}
+    evidence.setdefault("execution_status", "completed" if result.action_done else "no_action")
+    evidence["orders_pending_count"] = len(tuple(getattr(result, "pending_orders", ()) or ()))
+    orders = tuple(getattr(result, "submitted_orders", ()) or ())
+    return not config.dry_run_only and defer_routine_cycle_to_digest(
+        evidence, submitted_order_count=len(orders), submitted_orders=orders)
+
+
 def run_strategy(
     *,
     runtime: LongBridgeRebalanceRuntime,
@@ -1227,7 +1237,10 @@ def submit_claimed_order(order_intent):
     if isinstance(position_currency_by_symbol, dict):
         execution["position_currency_by_symbol"] = dict(position_currency_by_symbol)
 
-    if pending_orders:
+    routine_to_digest = _defer_cycle_notification(execution_result, config=config)
+    if routine_to_digest:
+        print(config.with_prefix("notification_suppressed reason=daily_digest"), flush=True)
+    elif pending_orders:
         try:
             notification_publisher.publish(
                 notification_renderers.render_rebalance_notification(

docs/unified-daily-notification.md

@@ -0,0 +1,36 @@
+# Unified daily notification (code only)
+
+`QSL_ROUTINE_NOTIFICATION_MODE` defaults to `immediate`. Exact `daily_digest`
+opts into one central bilingual digest. `QSL_DIGEST_ROUTINE_ORDERS=false` retains
+immediate confirmed order notices. Failed, pending, uncertain, risk, persistence
+and authentication outcomes retain immediate alerts. Healthy idle cycles and
+routine plugin observations do not produce individual pushes.
+
+QPK source reference: `4cc5e55251269923771f90cd661bb62a5cfdf0cd` (notification modules only).
+
+The notification-only QPK source snapshot in `notifications/digest_snapshot/` avoids an
+unrelated trading dependency upgrade. Keep these three modules byte-identical
+to the reviewed QPK notification modules. Existing QPK/strategy pins, order
+execution, funding, risk, account enablement and durable report writes are unchanged.
+
+Before enabling routing, validate the exact active target's private archived
+report, candidate store and QRS receiver. Existing cloud monitor wiring uses
+`RUNTIME_TARGET_ENABLED=true`, exact protected `RUNTIME_TARGET_JSON`,
+`EXECUTION_REPORT_GCS_URI`, `DIGEST_CANDIDATES_GCS_URI`, `DIGEST_STORE_TARGET`,
+`DIGEST_OPAQUE_ACCOUNT_UID` and `DIGEST_TARGET_ID`. Missing configuration stops
+projection; disabled targets stay disabled. Do not infer private identities.
+The producer aggregates every unique cycle in the completed 06:00 SGT window;
+missing evidence is unknown, never zero. No new collector or scheduler is added.
+
+No deployment, variable application, IAM, secret or live message is authorized
+by this PR. Source adoption and delivery require separate approved release work.
+
+For multiple active targets, protected `DIGEST_TARGET_BINDINGS_JSON` supplies a
+list of per-target environment mappings. `DIGEST_NOTIFICATION_SCHEDULE_JSON`
+can reuse the existing heartbeat scheduler/calendar declaration to prove a
+window was not due. Missing due reports stay errors; closed configured sessions
+produce an `actually_ran=false` marker without inventing zero trading counts.
+
+Fill counts require individual execution identifiers. A completed order response
+proves an order completed, not how many fills it received. Missing fill counts
+remain unknown; confirmed routine orders still appear without a false diagnostic.

docs/unified-notification-trading-diff-review.zh-CN.md

@@ -0,0 +1,25 @@
+# 交易差异审查：LongBridgePlatform
+
+审查方式：本次作者自审，辅以 AST 调用参数比较；不声称独立人工审查。
+基线：`a0de99314ebd15fcfc7f3e99de3b54dbb924cb7c`。范围为当前 PR 相对基线的全部改动。
+
+结论：源码差异仅涉及通知路由、附加的通知报告事实、已有监测入口的候选投影、通知配置透传、测试和说明。交易依赖、锁文件、UES/QPK pin、信号、仓位、资金限额、订单构造、券商调用、状态 claim 和风险门禁保持原样。
+
+现有文件中的券商、订单、风险、交易状态和报告持久化调用，已对调用表达式及全部参数做 AST 比较：
+
+| 文件 | 敏感调用数量 | 参数完全相同 |
+|---|---:|---|
+| `application/rebalance_service.py` | 5 | 是 |
+| `main.py` | 12 | 是 |
+
+新增模块不调用券商。只有既有云监测器在 `daily_digest` 且目标明确启用时，才读已有私有报告并写候选；不启用账户、不修改订单或交易排程。关闭模式保持既有行为。失败、未决订单、风险、认证及持久化问题保留即时通知；已确认的普通订单可通过 `QSL_DIGEST_ROUTINE_ORDERS=false` 继续即时发送。
+
+通知代码取自 QPK `4cc5e55251269923771f90cd661bb62a5cfdf0cd`，有意不升级平台交易依赖。三个源文件逐字节相同：
+
+| 文件 | SHA256 |
+|---|---|
+| `digest_routing.py` | `60c3f258c4d1a03bc49c7d03b392c02c316541726dfacb0b1d4fae6dbd2e5017` |
+| `digest_candidates.py` | `6272c0ca16ee63eecd9bcb25870df7f575d8fac3f487d3ab1ac69cc3831042a0` |
+| `digest_archive.py` | `b9a92299adfddff42c62f7d493845e0cbde91df7e475a5cb6bf2114446045b49` |
+
+合并条件：相关验证通过、PR 当前精确 head 的 CI 全绿、当前远端主线的 push / workflow_run / 可复用 workflow 调用链无自动部署。否则保留草稿。此审查不批准从旧生产 revision 升级到整个当前主线；生产到候选的完整差异、UES 指纹与 baseline 仍须按后续发布方案单独审查。

main.py

@@ -426,6 +426,14 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict:
     account_identity = execution.get("account_identity")
     if isinstance(account_identity, dict):
         summary["account_identity"] = dict(account_identity)
+    from notifications.digest_snapshot.digest_routing import daily_digest_enabled
+    if daily_digest_enabled():
+        from notifications.digest_snapshot.digest_candidates import cycle_evidence
+        evidence = {**dict(getattr(cycle_result, "allocation", {}) or {}), **execution}
+        evidence.setdefault("execution_status", "completed" if broker_submission_done else "no_action")
+        evidence["orders_pending_count"] = len(pending_orders)
+        summary["notification_digest"] = cycle_evidence(evidence,
+            tuple(getattr(cycle_result, "submitted_orders", ()) or ()), dry_run=dry_run)
     return summary
 
 
@@ -716,8 +724,9 @@ def build_strategy_plugin_alert_context_label() -> str:
 
 
 def publish_strategy_plugin_alerts(signals, *, report=None):
+    from notifications.digest_snapshot.digest_routing import immediate_plugin_signals
     result = dispatch_strategy_plugin_alerts(
-        signals,
+        immediate_plugin_signals(signals),
         notification_settings=RUNTIME_SETTINGS,
         translator=t,
         strategy_label=STRATEGY_PROFILE,

notifications/digest_snapshot/__init__.py

@@ -0,0 +1 @@
+"""Notification-only QPK source snapshot; trading dependency pins stay unchanged."""

notifications/digest_snapshot/digest_archive.py

@@ -0,0 +1,127 @@
+"""Publish notification candidates from existing archives; disabled by default.
+
+Runs only inside an existing cloud monitor, with its existing identity. It never
+invokes a platform service or broker, reads secrets, or enables a runtime target.
+"""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from .digest_candidates import project_reports
+
+_SAFE = re.compile(r"[a-z0-9][a-z0-9_-]{0,40}\Z")
+
+
+def _read(args):
+    result = subprocess.run(["gcloud", "storage", *args], capture_output=True, text=True, check=False)
+    if result.returncode:
+        raise ValueError("candidate_archive_read_failed")
+    return json.loads(result.stdout)
+
+
+def emit(env, *, now=None, read=_read, upload=None):
+    if env.get("QSL_ROUTINE_NOTIFICATION_MODE", "immediate") != "daily_digest":
+        return {"status": "disabled", "uploaded": False}
+    bindings = env.get("DIGEST_TARGET_BINDINGS_JSON", "").strip()
+    if bindings:
+        values = json.loads(bindings)
+        if not isinstance(values, list) or not values or len(values) > 100 or not all(isinstance(v, dict) for v in values):
+            raise ValueError("candidate_bindings_invalid")
+        results = []
+        for value in values:
+            results.append(emit({**env, **value, "DIGEST_TARGET_BINDINGS_JSON": ""}, now=now, read=read, upload=upload))
+        return {"status": "projected", "uploaded": any(v["uploaded"] for v in results), "targets": len(results)}
+    if env.get("RUNTIME_TARGET_ENABLED") == "false":
+        return {"status": "target_disabled", "uploaded": False}
+    if env.get("RUNTIME_TARGET_ENABLED") != "true":
+        raise ValueError("candidate_target_enablement_unknown")
+    target = json.loads(env.get("RUNTIME_TARGET_JSON", ""))
+    platform = target.get("platform_id", "")
+    slot = env.get("DIGEST_STORE_TARGET", "")
+    if not _SAFE.fullmatch(platform) or not _SAFE.fullmatch(slot):
+        raise ValueError("candidate_store_identity_missing")
+    uid, tid = env.get("DIGEST_OPAQUE_ACCOUNT_UID", ""), env.get("DIGEST_TARGET_ID", "")
+    if not uid or not tid:
+        raise ValueError("candidate_identity_missing")
+    prefix = env.get("EXECUTION_REPORT_GCS_URI", "").rstrip("/")
+    destination = env.get("DIGEST_CANDIDATES_GCS_URI", "").rstrip("/")
+    if any(not re.fullmatch(r"gs://[a-z0-9][a-z0-9._-]+/[^\s?#*]+", value) for value in (prefix, destination)):
+        raise ValueError("candidate_store_configuration_missing")
+    current = now or datetime.now(timezone.utc)
+    if current.astimezone(ZoneInfo("Asia/Singapore")).hour < 6:
+        return {"status": "before_cutoff", "uploaded": False}
+    day = (current.astimezone(ZoneInfo("Asia/Singapore")) - timedelta(hours=6)).date() - timedelta(days=1)
+    # Monitor runs after the fixed 06:00 SGT cut-off. Reject early runs rather
+    # than overwriting the previous day's complete candidate with partial facts.
+    end = datetime(day.year, day.month, day.day, 6, tzinfo=ZoneInfo("Asia/Singapore")) + timedelta(days=1)
+    start = end - timedelta(days=1)
+    months = {start.astimezone(timezone.utc).strftime("%Y-%m"), end.astimezone(timezone.utc).strftime("%Y-%m")}
+    entries = []
+    for month in sorted(months):
+        entries.extend(read(["ls", "--json", f"{prefix}/{platform}/**/{month}/*.json"]))
+    if len(entries) > 10000:
+        raise ValueError("candidate_archive_budget_exceeded")
+    reports, uris = [], set()
+    for entry in entries:
+        uri = str(entry.get("url", "")).split("#", 1)[0]
+        if uri in uris:
+            continue
+        if not uri.startswith(f"{prefix}/{platform}/"):
+            raise ValueError("candidate_source_locator_mismatch")
+        uris.add(uri)
+        raw_stamp = (entry.get("metadata") or {}).get("updated") or (entry.get("metadata") or {}).get("timeCreated")
+        stamp = datetime.fromisoformat(str(raw_stamp).replace("Z", "+00:00"))
+        if stamp.tzinfo is None:
+            raise ValueError("candidate_archive_time_missing")
+        if stamp < start or stamp > current:
+            continue
+        source = read(["cat", uri])
+        source_target = source.get("runtime_target") or {}
+        keys = ("platform_id", "strategy_profile", "service_name", "account_scope", "account_selector", "deployment_selector", "dry_run_only")
+        if not isinstance(source_target, dict) or any(source_target.get(key) != target.get(key) for key in keys):
+            continue
+        source_stamp = datetime.fromisoformat(str(source.get("started_at", "")).replace("Z", "+00:00"))
+        if source_stamp.tzinfo is None:
+            raise ValueError("candidate_source_time_invalid")
+        if start <= source_stamp < end:
+            reports.append(source)
+    if not reports:
+        schedule_raw = env.get("DIGEST_NOTIFICATION_SCHEDULE_JSON", "").strip()
+        if not schedule_raw:
+            raise ValueError("candidate_source_absent")
+        from quant_platform_kit.common.runtime_heartbeat_policy import filter_due_targets
+        schedule = json.loads(schedule_raw)
+        due, evaluated = filter_due_targets([schedule], since=start, now=end - timedelta(microseconds=1),
+            publication_grace=timedelta(0), warning_logger=lambda message: None)
+        if not evaluated or due:
+            raise ValueError("candidate_due_source_absent")
+        payload = {"schema_version": "qsl.digest_candidates.v1", "runs": [{
+            "platform_id": platform, "strategy_profile": target["strategy_profile"],
+            "opaque_account_uid": uid, "target_id": tid, "business_day": day.isoformat(),
+            "actually_ran": False, "status": "ok", "scheduled_idle": True,
+            "evidence_provenance": "candidates",
+        }]}
+    else:
+        payload = project_reports(reports, expected_target=target, opaque_account_uid=uid, target_id=tid, now=current)
+    if upload is None:
+        from google.cloud import storage
+        def upload(uri, body):
+            storage.Blob.from_string(uri, client=storage.Client()).upload_from_string(
+                body, content_type="application/json", retry=None, timeout=30)
+    # Never retry an unknown write result. Public output contains no identities.
+    upload(f"{destination}/{platform}/{day.isoformat()}/{slot}.json", json.dumps(payload, allow_nan=False))
+    return {"status": "projected", "uploaded": True, "runs": len(reports)}
+
+
+def main():
+    try:
+        print(json.dumps(emit(os.environ), sort_keys=True))
+        return 0
+    except Exception:
+        print(json.dumps({"status": "blocked", "uploaded": False, "reason": "candidate_projection_failed"}))
+        return 1