[notif-consolidate-1230] LongBridge: default-off target-bound notification consolidation

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

@@ -126,6 +126,14 @@ jobs:
       RUNTIME_HEARTBEAT_LOOKBACK_HOURS: ${{ inputs.lookback_hours || vars.RUNTIME_HEARTBEAT_LOOKBACK_HOURS || '36' }}
       RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }}
       RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS || 'false' }}
+      LONGBRIDGE_UNIFIED_NOTIFICATION_ENABLED: ${{ vars.LONGBRIDGE_UNIFIED_NOTIFICATION_ENABLED || 'false' }}
+      DIGEST_CANDIDATES_GCS_URI: ${{ vars.DIGEST_CANDIDATES_GCS_URI || secrets.DIGEST_CANDIDATES_GCS_URI }}
+      QSL_NOTIFICATION_MARKER_GCS_URI: ${{ vars.QSL_NOTIFICATION_MARKER_GCS_URI || secrets.QSL_NOTIFICATION_MARKER_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_MATRIX_TARGET_ID: ${{ matrix.target.id }}
+      DIGEST_MATRIX_ACCOUNT_SCOPE: ${{ matrix.target.label }}
       RUNTIME_HEARTBEAT_ACCEPT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STATUSES }}
       RUNTIME_HEARTBEAT_REJECT_STATUSES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STATUSES }}
       RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }}
@@ -288,7 +296,7 @@ jobs:
 
 
       - name: Publish existing reports as unified digest candidates
-        if: ${{ success() && vars.QSL_ROUTINE_NOTIFICATION_MODE == 'daily_digest' }}
+        if: ${{ (success() && vars.QSL_ROUTINE_NOTIFICATION_MODE == 'daily_digest' && vars.LONGBRIDGE_UNIFIED_NOTIFICATION_ENABLED != 'true') || (!cancelled() && env.INSPECT_SG_PROBE_PERMISSIONS != 'true' && vars.LONGBRIDGE_UNIFIED_NOTIFICATION_ENABLED == 'true') }}
         env:
           QSL_ROUTINE_NOTIFICATION_MODE: ${{ vars.QSL_ROUTINE_NOTIFICATION_MODE || 'immediate' }}
           RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}

application/account_new_risk_gate_support.py

@@ -386,8 +386,8 @@ def maybe_publish_attention_for_admission(
 ) -> Mapping[str, int]:
     """Publish ACTION/HALT attention when NEW_RISK / ops axes require a page.
 
-    Dedupes on transition keys within the process. Never grants live, raises RRL,
-    or invents daily-loss facts.
+    Legacy uses process-local transition keys; the opted-in notification adapter
+    uses durable claims. Never grants live, raises RRL, or invents daily-loss facts.
     """
 
     try:
@@ -417,6 +417,12 @@ def maybe_publish_attention_for_admission(
             mandate_dd_budget=resolve_mandate_dd_budget(profile),
         )
     )
+    from notifications.unified import enabled as unified_enabled, publish_attention
+    if unified_enabled():
+        return publish_attention(decision, telegram_sender=telegram_sender,
+            recover=(not prohibited and not operational_uncertain and snapshot is not None
+                     and str(snapshot.observation_status) == "COMPLETE"
+                     and str(snapshot.reconciliation_status) == "VERIFIED"))
     return publish_attention_telegram_transition(
         decision=decision,
         platform=_ATTENTION_PLATFORM,

application/rebalance_service.py

@@ -1237,7 +1237,13 @@ def submit_claimed_order(order_intent):
     if isinstance(position_currency_by_symbol, dict):
         execution["position_currency_by_symbol"] = dict(position_currency_by_symbol)
 
-    routine_to_digest = _defer_cycle_notification(execution_result, config=config)
+    from notifications.unified import enabled as unified_enabled, prepare_cycle
+    if unified_enabled():
+        routine_to_digest, notification_publisher = prepare_cycle(
+            execution_result, config=config, publisher=notification_publisher, notify_issue=runtime.notify_issue,
+        )
+    else:
+        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:

application/runtime_notification_adapters.py

@@ -12,6 +12,7 @@
     OrderLifecycleEvent,
     monitor_submitted_order_status,
     publish_order_lifecycle_event,
+    is_filled_status, is_partial_filled_status, is_terminal_error_status,
 )
 from notifications.telegram import build_issue_notifier
 from quant_platform_kit.common.quantity import format_quantity
@@ -89,23 +90,59 @@ def send_recorded_message(message: str) -> bool:
         log_message=log_message or (lambda message: print(with_prefix(message), flush=True)),
         send_message=send_recorded_message,
     )
-    notify_issue = build_issue_notifier(
+    legacy_notify_issue = build_issue_notifier(
         with_prefix_fn=with_prefix,
         send_tg_message_fn=send_recorded_message,
     )
+
+    def issue_with_ack(title, detail):
+        before = len(recorded_delivery_events)
+        legacy_notify_issue(title, detail)
+        events = recorded_delivery_events[before:]
+        return bool(events) and all(event.get("transport_acknowledged") is True for event in events)
+
+    def notify_issue(title, detail):
+        from notifications import unified
+        if not unified.enabled():
+            return legacy_notify_issue(title, detail)
+        # Existing callers provide fixed issue titles. Hash the category, never
+        # provider detail text, run ids or credentials into a marker identity.
+        reason = hashlib.sha256(str(title).encode()).hexdigest()
+        return unified.dispatch_immediate(lambda: issue_with_ack(title, detail),
+            kind="real_error", incident=f"runtime_issue:{reason}")
     order_event_publisher = NotificationPublisher(
         log_message=lambda _message: None,
         send_message=send_recorded_message,
     )
 
     def publish_order_event(event: OrderLifecycleEvent) -> None:
-        publish_order_lifecycle_event(
-            event,
-            translator=translator,
-            publisher=order_event_publisher,
-        )
+        from notifications import unified
+        def publish():
+            before = len(recorded_delivery_events)
+            outcome = publish_order_lifecycle_event(event, translator=translator, publisher=order_event_publisher)
+            if not unified.enabled():
+                return outcome
+            events = recorded_delivery_events[before:]
+            return bool(events) and all(item.get("transport_acknowledged") is True for item in events)
+        if unified.enabled() and not (is_filled_status(event.status) or is_partial_filled_status(event.status)):
+            from notifications.digest_snapshot.unified_contract import order_identity_digest
+            unified.dispatch_immediate(publish, kind="order_status_unknown",
+                incident=f"order_status:{order_identity_digest(event.order_id)}", state=event.status,
+                severity=2 if is_terminal_error_status(event.status) else 1)
+        else:
+            if unified.enabled() and is_filled_status(event.status):
+                unified.record_order_resolution(event)
+            publish()
 
     def post_submit_order(trade_context, order_intent, report) -> None:
+        from notifications import unified
+        issue_notifier = legacy_notify_issue
+        if unified.enabled():
+            from notifications.digest_snapshot.unified_contract import order_identity_digest
+            if report.broker_order_id:
+                incident = f"order_status_poll:{order_identity_digest(report.broker_order_id)}"
+                def issue_notifier(title, detail):
+                    return unified.dispatch_immediate(lambda: issue_with_ack(title, detail), kind="real_error", incident=incident)
         monitor_submitted_order_status(
             trade_context,
             str(order_intent.symbol),
@@ -116,7 +153,7 @@ def post_submit_order(trade_context, order_intent, report) -> None:
             order_poll_interval_sec=order_poll_interval_sec,
             order_poll_max_attempts=order_poll_max_attempts,
             publish_order_event=publish_order_event,
-            notify_issue=notify_issue,
+            notify_issue=issue_notifier,
             sleeper=sleeper,
         )
 

main.py

@@ -8,7 +8,7 @@
 import importlib
 import os
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 
 from flask import Flask, request
 
@@ -54,6 +54,7 @@
 from runtime_execution_policy import dca_execution_unsupported_reason
 from runtime_config_support import load_platform_runtime_settings
 from notifications.telegram import build_signal_text, build_strategy_display_name, build_translator
+from notifications import unified as unified_notifications
 from quant_platform_kit.common.runtime_reports import (
     append_runtime_report_error,
     build_runtime_report_base,
@@ -427,7 +428,9 @@ def _summarize_cycle_result_for_report(cycle_result, *, dry_run: bool) -> dict:
     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():
+    if unified_notifications.enabled():
+        _, summary["notification_digest"] = unified_notifications.cycle_facts(cycle_result, dry_run=dry_run)
+    elif 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")
@@ -659,7 +662,7 @@ def _runtime_error_notification_message(exc: Exception, *, route_label: str) ->
     ))
 
 
-def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
+def _send_runtime_error(exc: Exception, *, route_label: str) -> bool:
     targets = _runtime_error_notification_targets()
     if not targets:
         print(t("runtime_notification_missing_target"), flush=True)
@@ -689,6 +692,14 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
     return bool(outcomes) and all(outcomes)
 
 
+def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
+    return unified_notifications.dispatch_immediate(
+        lambda: _send_runtime_error(exc, route_label=route_label), kind="real_error",
+        incident=f"runtime_route:{route_label}:{_runtime_failure_reason(exc)}",
+        occurred_at=datetime.now(timezone.utc), run_id=build_run_id(),
+    ) if unified_notifications.enabled() else _send_runtime_error(exc, route_label=route_label)
+
+
 def _handle_route_runtime_error(exc: Exception, *, route_label: str):
     print(t("runtime_failure_log"), flush=True)
     _notify_runtime_error(exc, route_label=route_label)
@@ -697,7 +708,11 @@ def _handle_route_runtime_error(exc: Exception, *, route_label: str):
 
 def _route_with_runtime_error_fallback(handler, *, success_body: str, route_label: str, **kwargs):
     try:
-        result = handler(**kwargs)
+        if unified_notifications.enabled():
+            with unified_notifications.bind_route(route_label):
+                result = handler(**kwargs)
+        else:
+            result = handler(**kwargs)
     except Exception as exc:
         return _handle_route_runtime_error(exc, route_label=route_label)
     if result is False:
@@ -876,7 +891,11 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             live_execution_session_authorized=bool(market_open and not validation_only),
         )
         failure_phase = "strategy_cycle"
-        cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
+        if unified_notifications.enabled():
+            with unified_notifications.bind_report(report):
+                cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
+        else:
+            cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
         failure_phase = "reporting"
         signal_snapshot = {}
         if cycle_result is not None:
@@ -886,6 +905,12 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             cycle_result,
             dry_run=bool(report.get("dry_run")),
         )
+        if unified_notifications.enabled():
+            try:
+                unified_notifications.attach_cycle_event(execution_summary, cycle_result, report=report, config=rebalance_config)
+            except Exception:
+                print("notification_event_attachment_unconfirmed", flush=True)
+                unified_notifications.record_failure("notification_event_attachment_unconfirmed", report=report)
         notification_delivery_log = _build_notification_delivery_log_for_report(
             platform="longbridge",
             strategy_profile=STRATEGY_PROFILE,
@@ -896,6 +921,11 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
         )
         if notification_delivery_log:
             execution_summary["notification_delivery_log"] = notification_delivery_log
+        if unified_notifications.enabled():
+            notification_delivery_events.extend(
+                {"sink": "notification_store", "delivery_status": "failed", "transport_acknowledged": False,
+                 "error_type": code} for code in report.get("notification_unified_errors", ())
+            )
         notification_delivery_summary = _build_notification_delivery_summary(
             notification_delivery_events
         )
@@ -948,10 +978,13 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
         )
         if not (validation_only and _v7_application_candidate_bound()):
             try:
-                notification_adapters.publish_cycle_notification(
-                    detailed_text=_compact_error_notification(exc, phase=failure_phase),
-                    compact_text=_compact_error_notification(exc, phase=failure_phase),
-                )
+                def publish_failure():
+                    return notification_adapters.publish_cycle_notification(
+                        detailed_text=_compact_error_notification(exc, phase=failure_phase),
+                        compact_text=_compact_error_notification(exc, phase=failure_phase),
+                    )
+                unified_notifications.dispatch_immediate(publish_failure, kind="real_error",
+                    incident=f"runtime_failure:{failure_phase}:{failure_reason}", report=report, actually_ran=True)
             except Exception as notification_exc:
                 notification_delivery_events.append(
                     {
@@ -995,6 +1028,8 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             try:
                 report_path = reporting_adapters.persist_execution_report(report)
                 print(f"execution_report {report_path}", flush=True)
+                if unified_notifications.enabled():
+                    unified_notifications.recover_runtime(report=report)
             except Exception as persist_exc:
                 print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True)
 
@@ -1122,15 +1157,15 @@ def run_probe(*, response_body: str = "Probe OK"):
             exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe",
         )
         if composer is not None:
-            composer.build_notification_adapters().publish_cycle_notification(
-                detailed_text=err,
-                compact_text=_compact_error_notification(
-                    exc,
-                    title=t("health_probe_title"),
-                    prefix=t("health_probe_error_prefix"),
-                    phase="health_probe",
-                ),
-            )
+            def publish_probe_failure():
+                return composer.build_notification_adapters().publish_cycle_notification(
+                    detailed_text=err,
+                    compact_text=_compact_error_notification(
+                        exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe",
+                    ),
+                )
+            unified_notifications.dispatch_immediate(publish_probe_failure, kind="real_error",
+                incident="runtime_failure:health_probe", report=report)
         else:
             print(err, flush=True)
         return "Error", 500
@@ -1141,6 +1176,8 @@ def run_probe(*, response_body: str = "Probe OK"):
             if reporting_adapters is not None and report is not None:
                 report_path = reporting_adapters.persist_execution_report(report)
                 print(f"execution_report {report_path}", flush=True)
+                if unified_notifications.enabled():
+                    unified_notifications.recover_runtime(report=report, probe=True)
         except Exception as persist_exc:
             print(f"failed to persist execution report: {type(persist_exc).__name__}", flush=True)
 

notifications/digest_snapshot/__init__.py

@@ -1 +1,4 @@
-"""Notification-only QPK source snapshot; trading dependency pins stay unchanged."""
+"""Notification-only snapshot from merged QPK; trading pins stay unchanged."""
+
+UNIFIED_CONTRACT_QPK_SHA = "07e24c0c804a9b47316b6ec3c75bb63820d8f421"
+UNIFIED_CONTRACT_SHA256 = "7ba48d1d7c74e66be8cd3ff983c2117002776e82cda6675a084cfe0dc51aecc0"