fix: unify notification routing and preserve alert bodies

application/runtime_notification_adapters.py

@@ -11,10 +11,8 @@
 from notifications.order_alerts import (
     OrderLifecycleEvent,
     monitor_submitted_order_status,
-    publish_order_lifecycle_event,
-    is_filled_status, is_partial_filled_status, is_terminal_error_status,
+    is_filled_status, is_verified_partial_fill, is_terminal_error_status,
 )
-from notifications.telegram import build_issue_notifier
 from quant_platform_kit.common.quantity import format_quantity
 from quant_platform_kit.common.port_adapters import CallableNotificationPort
 from quant_platform_kit.common.ports import NotificationPort
@@ -90,10 +88,10 @@ 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,
     )
-    legacy_notify_issue = build_issue_notifier(
-        with_prefix_fn=with_prefix,
-        send_tg_message_fn=send_recorded_message,
-    )
+    def legacy_notify_issue(title, detail):
+        from notifications.renderers import render_runtime_error
+        # Raw broker detail belongs to controlled diagnostics, never the body.
+        return send_recorded_message(render_runtime_error(kind="real_error"))
 
     def issue_with_ack(title, detail):
         before = len(recorded_delivery_events)
@@ -119,20 +117,22 @@ def publish_order_event(event: OrderLifecycleEvent) -> None:
         from notifications import unified
         def publish():
             before = len(recorded_delivery_events)
-            outcome = publish_order_lifecycle_event(event, translator=translator, publisher=order_event_publisher)
+            from notifications.renderers import render_runtime_error
+            outcome = order_event_publisher.publish(RenderedNotification(
+                detailed_text="", compact_text=render_runtime_error(kind="order_status_unknown")))
             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()
+        if is_filled_status(event.status) or is_verified_partial_fill(event):
+            # P10 applies in either flag state. Persist before suppressing send.
+            if not unified.record_order_resolution(event):
+                notify_issue("Order notification evidence unavailable", "Candidate persistence is unconfirmed.")
+            return
+        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)
 
     def post_submit_order(trade_context, order_intent, report) -> None:
         from notifications import unified

notifications/order_alerts.py

@@ -5,6 +5,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 import traceback
+from decimal import Decimal, InvalidOperation
 
 from notifications.events import NotificationPublisher, RenderedNotification
 
@@ -19,6 +20,7 @@ class OrderLifecycleEvent:
     executed_qty: str = "0"
     executed_price: str = "0"
     reason: str = ""
+    status_verified: bool = False
 
 
 def is_filled_status(status):
@@ -29,6 +31,17 @@ def is_partial_filled_status(status):
     return "PartialFilled" in status
 
 
+def is_verified_partial_fill(event: OrderLifecycleEvent) -> bool:
+    """Only a broker observation with consistent quantities proves processing."""
+    if not event.status_verified or not is_partial_filled_status(event.status) or event.reason:
+        return False
+    try:
+        executed, requested, price = map(Decimal, (event.executed_qty, str(event.quantity), event.executed_price))
+        return all(value.is_finite() for value in (executed, requested, price)) and 0 < executed < requested and price > 0
+    except (InvalidOperation, ValueError, TypeError):
+        return False
+
+
 def is_terminal_error_status(status):
     return any(keyword in status for keyword in ["Rejected", "Canceled", "Expired"])
 
@@ -47,6 +60,7 @@ def build_order_lifecycle_event(
     executed_qty="0",
     executed_price="0",
     reason="",
+    status_verified=False,
 ):
     return OrderLifecycleEvent(
         symbol=str(symbol),
@@ -57,6 +71,7 @@ def build_order_lifecycle_event(
         executed_qty=str(executed_qty or "0"),
         executed_price=str(executed_price or "0"),
         reason=str(reason or ""),
+        status_verified=status_verified,
     )
 
 
@@ -163,7 +178,6 @@ def monitor_submitted_order_status(
         return
 
     try:
-        published_any_status = False
         for _ in range(order_poll_max_attempts):
             sleeper(order_poll_interval_sec)
             order_status = fetch_order_status(trade_context, order_id)
@@ -183,6 +197,7 @@ def monitor_submitted_order_status(
                 executed_qty=executed_qty,
                 executed_price=executed_price,
                 reason=reject_message,
+                status_verified=True,
             )
 
             if is_filled_status(status):
@@ -191,22 +206,21 @@ def monitor_submitted_order_status(
 
             if is_partial_filled_status(status):
                 publish_order_event(event)
-                published_any_status = True
 
             if is_terminal_error_status(status):
                 publish_order_event(event)
                 return
-        if not published_any_status:
-            publish_order_event(
-                build_order_lifecycle_event(
-                    symbol,
-                    side_text,
-                    quantity,
-                    order_id,
-                    "StatusCheckTimeout",
-                    reason="poll_timeout",
-                )
+        # A partial observation does not prove the eventual terminal outcome.
+        publish_order_event(
+            build_order_lifecycle_event(
+                symbol,
+                side_text,
+                quantity,
+                order_id,
+                "StatusCheckTimeout",
+                reason="poll_timeout",
             )
+        )
     except Exception:
         notify_issue(
             "Order status poll failed",

notifications/renderers.py

@@ -536,3 +536,22 @@ def _heartbeat_outcome_key(execution, *, skip_logs, note_logs) -> str:
     if bool(execution.get("direct_live_routing_blocked")):
         return "heartbeat_execution_blocked"
     return "no_executable_orders" if (skip_logs or note_logs) else "no_trades"
+
+
+def render_runtime_error(*, kind="real_error", occurred_at=None, report=None, locale=None):
+    """Adapt runtime facts to the step-1 shared formatter; never render diagnostics.
+
+    The QPK step-1 revision must be pinned before deployment. Monitor time is
+    not a business-run observation; missing producer links/counts stay unknown.
+    """
+    import os
+    from datetime import datetime, timezone
+    from quant_platform_kit.notifications.renderer_base import render_asset_management_error
+    from quant_platform_kit.notifications.unified_contract import NotificationEvent
+
+    report = report or {}
+    stamp = occurred_at or datetime.now(timezone.utc)
+    event = NotificationEvent(kind, stamp, platform_id="longbridge",
+        strategy_profile=report.get("strategy_profile"),
+        incident_reason=kind)
+    return render_asset_management_error(event, locale=locale or os.environ.get("NOTIFY_LANG") or "zh-CN")

notifications/unified.py

@@ -541,11 +541,17 @@ def record_order_resolution(order_event):
     try:
         digest = order_identity_digest(order_event.order_id)
         fact = make_event("fill", order_execution_digest=digest)
-        persist_candidate(fact.to_candidate(), actually_ran=False)
+        if persist_candidate(fact.to_candidate(), actually_ran=False) is not True:
+            raise CandidatePersistenceError("notification_candidate_write_unconfirmed")
+        from .order_alerts import is_filled_status
+        if not is_filled_status(order_event.status):
+            return True
         for stream in ("order_status", "order_status_poll"):
             observe_incident(replace(fact, incident_reason=f"{stream}:{digest}"), state="filled", recovered=True, severity=0)
+        return True
     except Exception:
         record_failure("order_notification_evidence_unconfirmed")
+        return False
 
 
 def recover_runtime(*, report=None, route_label=None, occurred_at=None, run_id=None, probe=False):

scripts/execution_report_heartbeat.py

@@ -1186,13 +1186,8 @@ def _notify_normal_heartbeat(name: str, detail: str, *, observed_at=None, checke
 def _notify_heartbeat_alert(message, *, now, kind, inspected_count=0):
     if not unified_notifications.enabled():
         return _send_telegram(message)
-    locale = _notification_locale()
-    issue = (_notice("heartbeat_no_recent_report", lookback_hours=os.environ.get("RUNTIME_HEARTBEAT_LOOKBACK_HOURS") or "36")
-             if kind == "receipt_missing" else _notice("heartbeat_no_acceptable_report", count=inspected_count))
-    # Keep raw diagnostics in the existing logs; use the shared localized
-    # template without private target names, archive URIs or mixed raw reasons.
-    compact = format_operational_alert(locale=locale, alert_type="execution_report_heartbeat",
-        name="Execution receipts" if locale == "en" else "执行回执", context={}, issues=[issue])
+    from notifications.renderers import render_runtime_error
+    compact = render_runtime_error(kind=kind, occurred_at=now, locale=_notification_locale())
     return unified_notifications.dispatch_immediate(lambda: _send_telegram(compact[:3900]),
         kind=kind, incident="execution_receipt", occurred_at=now,
         run_id=os.environ.get("GITHUB_RUN_ID"), actually_ran=False)

scripts/paper_observation/core.py

@@ -470,9 +470,24 @@ def __init__(self, store, send, calendar, start, activation, verification=False)
         if verification:
             self.send = lambda message: send("[verification] " + message)
 
+    def record_routine(self, root, kind, incident, now):
+        # Observation facts are not business cycles or delivery receipts. Keep
+        # the existing append-only store, explicitly pending central ingestion.
+        record = {"event_kind": kind, "occurred_at": now.isoformat(),
+                  "incident_reason": incident, "actually_ran": False,
+                  "cycle_count": 0, "order_count": None, "fill_count": None,
+                  "delivery_status": "pending_central"}
+        key = f"{root}/digest/{kind}.json"
+        if not self.store.create(key, record) and not self.store.get(key):
+            raise ObservationError("storage_unavailable")
+
     def notify(self, incident, active, now):
         prefix = "verification/" if self.verification else ""
         root = f"alerts/{prefix}{incident}"
+        if incident in {f"calendar/coverage_{days}" for days in (90, 30, 7)}:
+            if active:
+                self.record_routine(root, "maintenance_reminder", incident, now)
+            return
         if active and not self.store.get(f"{root}/ack.json"):
             claim = f"{root}/claims/{slot(now)}.json"
             if self.store.create(claim, {"at": now.isoformat()}) and self.send(
@@ -491,10 +506,8 @@ def notify(self, incident, active, now):
             and self.store.get(f"{root}/ack.json")
             and not self.store.get(f"{root}/recovered.json")
         ):
-            if self.store.create(
-                f"{root}/recover-claims/{slot(now)}.json", {"at": now.isoformat()}
-            ) and self.send(f"PAPER observation recovery: {incident}"):
-                self.store.create(f"{root}/recovered.json", {"at": now.isoformat()})
+            self.record_routine(root, "recovery", incident, now)
+            self.store.create(f"{root}/recovered.json", {"at": now.isoformat()})
 
     def run(self, now):
         if now < self.activation:
@@ -544,8 +557,7 @@ def run(self, now):
                     pass
             # Storage failure must not depend on storage-based deduplication.
             code = "storage_unavailable"
-            self.send(
-                f"PAPER observation alert: {now.date()}/{code}; persistent dedup unavailable"
-            )
+            from notifications.renderers import render_runtime_error
+            self.send(render_runtime_error(kind="real_error", occurred_at=now))
             return code
         return "checked"

tests/test_order_alerts.py

@@ -68,7 +68,7 @@ def test_monitor_submitted_order_status_emits_lifecycle_events_via_callback():
         sleeper=lambda _seconds: None,
     )
 
-    assert len(events) == 1
+    assert [event.status for event in events] == ["PartialFilled", "StatusCheckTimeout"]
     assert events[0].symbol == "SOXL.US"
     assert events[0].status == "PartialFilled"
     assert events[0].executed_qty == "4"
@@ -91,7 +91,7 @@ def test_monitor_submitted_order_status_emits_unconfirmed_event_after_poll_windo
         sleeper=lambda _seconds: None,
     )
 
-    assert len(events) == 1
+    assert [event.status for event in events] == ["StatusCheckTimeout"]
     assert events[0].symbol == "SOXL.US"
     assert events[0].status == "StatusCheckTimeout"
     assert events[0].reason == "poll_timeout"