fix: unify asset-management notification contract

docs/notification-routing-policy.md

@@ -46,3 +46,47 @@ per-run projection and bind observation counts to that same run identity.
 This source-level policy requires coordinated producer and digest consumer
 adoption, compatible pinned revisions, and runtime validation. Its presence in
 this repository does not establish production deployment or delivery.
+
+## Shared asset-management entrypoints
+
+`route_asset_management_event` is the P10 channel boundary. It accepts the
+existing `NotificationEvent` envelope and candidate writer. In every flag state,
+normal cycles, heartbeats, acknowledged orders, fills, recovery, ordinary
+maintenance reminders, research candidates and strategy signals require a
+confirmed candidate write before deferral. Unknown categories and unproven
+expected blocks retain the error path. An incident reason on a normal envelope
+is contradictory evidence and raises rather than being silently deferred,
+except on a producer-confirmed recovery that names the original incident.
+The producer must validate classification against its real source facts;
+this API does not infer health from message text, cash or a monitor check.
+
+`render_asset_management_error` provides the fixed Chinese (default) and English
+error body. It reuses operational time/reason localization and approved display
+names, omits private identities and raw diagnostics, and accepts only explicit
+HTTPS run links on caller-approved hosts. Missing links and counts stay unknown.
+It neither invents a run URL nor claims delivery. It never renders a normal fact
+as an error; normal facts belong in the existing central daily digest.
+
+For an asset-management plugin recipient, pass
+`asset_management_event_factory` and `persist_candidate` to
+`publish_strategy_plugin_telegram_alerts`. Bind each signal to its authentic run;
+archive the original signals in the existing producer report. Ordinary research
+candidates must use `research_candidate`, rather than an error envelope merely
+to retain an immediate push. Independent signal errors cannot be deferred as a
+normal candidate. Errors additionally require an `IncidentTracker` backed by a
+durable cross-process compare-and-set store via
+`asset_management_incident_tracker`; the legacy has/write marker is insufficient.
+A producer-confirmed recovery with the original incident reason clears that
+incident only after its candidate write succeeds, without sending a recovery
+push. An unbound recovery fact does not clear an arbitrary active incident.
+Reservation precedes transport. A boolean transport result has no receipt, so it
+remains pending/uncertain, including after restart, and is not automatically
+resent. State/marker failures propagate. Producer/archive/candidate failures
+still require the caller's operational error handling.
+
+The public transport, non-asset plugin uses and legacy routing helper defaults
+retain their behavior. This is an additive contract for subsequent QRS and
+broker notification PRs: consumers must pin the reviewed shared revision and
+explicitly bind their asset-management call sites. This library change alone
+does not establish consumer adoption, a deployment, or runtime P10 compliance.
+No daily service, schedule, database or delivery retry mechanism is added.

src/quant_platform_kit/notifications/_redaction.py

@@ -81,3 +81,30 @@ def redact_exception(exc: BaseException) -> str:
 def mask_identifier(value: object) -> str:
     """Fully mask a known identifier, including aliases and short numeric IDs."""
     return _REDACTED
+
+
+def notification_run_link(value: object, *, allowed_hosts: tuple[str, ...] = (),
+                          private_identifiers: tuple[str, ...] = ()) -> str | None:
+    """Accept an explicit approved HTTPS run link, dropping query and fragment.
+
+    Arbitrary diagnostic URLs are not evidence links. Never reconstruct a link
+    from a private identity or return credential/resource-bearing URLs.
+    """
+    from urllib.parse import unquote, urlsplit, urlunsplit
+
+    if not isinstance(value, str) or not value or len(value) > 800 or any(ord(c) < 33 for c in value):
+        return None
+    try:
+        parsed = urlsplit(value)
+        if (parsed.scheme != "https" or parsed.hostname not in allowed_hosts
+                or parsed.username or parsed.password or parsed.port not in (None, 443)):
+            return None
+        safe = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+        decoded = unquote(safe)
+        if any(identifier and identifier in decoded for identifier in private_identifiers):
+            return None
+        if redact_sensitive_text(decoded) != decoded:
+            return None
+        return safe
+    except ValueError:
+        return None

src/quant_platform_kit/notifications/renderer_base.py

@@ -689,4 +689,62 @@ def resolve_i18n_key(key: str) -> str:
     "format_dashboard_from_sources",
     # i18n key aliases (Phase 2)
     "resolve_i18n_key",
+    "render_asset_management_error",
 ]
+
+
+def render_asset_management_error(event, *, locale: object = "zh-CN", run_url: str | None = None,
+                                  allowed_run_hosts: tuple[str, ...] = ()) -> str:
+    """Fixed P10 error body. Raw identity/diagnostics never become display text.
+
+    Counts are caller-proven single-run observations, not inferred from event
+    names or money. This renderer performs no IO and makes no delivery claim.
+    """
+    from quant_platform_kit.common.display_names import display_name
+    from quant_platform_kit.common.operational_notification_localization import (
+        format_sgt_time, localize_attention_reason_code, resolve_operational_notification_locale,
+    )
+    from ._redaction import notification_run_link, redact_sensitive_text
+    from .unified_contract import ASSET_MANAGEMENT_DIGEST_KINDS, NotificationEvent
+
+    if not isinstance(event, NotificationEvent) or event.event_kind in ASSET_MANAGEMENT_DIGEST_KINDS:
+        raise ValueError("notification_error_event_required")
+    lang = resolve_operational_notification_locale(locale)
+    zh = lang == "zh"
+    unknown = "未知" if zh else "unknown"
+    unavailable = "运行链接不可用" if zh else "Run link unavailable"
+    platforms = {"ibkr": "IBKR", "schwab": "Schwab", "charles_schwab": "Schwab",
+                 "firstrade": "Firstrade", "longbridge": "LongBridge", "binance": "Binance"}
+    platform = platforms.get(event.platform_id, unknown)
+    strategy = display_name("strategies", event.strategy_profile, lang, fallback=unknown)
+    private = tuple(value for value in (event.opaque_account_uid, event.target_id, event.source_run_id)
+                    if isinstance(value, str) and value)
+    if any(value in strategy for value in private):
+        strategy = unknown
+    strategy = redact_sensitive_text(strategy).replace("\n", " ").replace("\r", " ")[:120]
+    reason = event.incident_reason or event.event_kind
+    reasons = {
+        "empty_position_decision": ("策略决策不可用", "Strategy decision unavailable"),
+        "receipt_missing": ("缺少执行回执", "Execution receipt missing"),
+        "receipt_stale": ("执行回执已过期", "Execution receipt stale"),
+        "receipt_unacceptable": ("执行回执不符合验收要求", "Execution receipt unacceptable"),
+        "auth_failure": ("认证失败", "Authentication failed"),
+        "order_status_unknown": ("订单结果未知", "Order outcome unknown"),
+    }
+    if reason in reasons:
+        problem = reasons[reason][0 if zh else 1]
+    else:
+        translated = localize_attention_reason_code(lang, reason)
+        problem = translated if translated != reason else ("运行异常，具体原因未核实" if zh else "Operational error; reason unverified")
+    link = notification_run_link(run_url, allowed_hosts=allowed_run_hosts, private_identifiers=private) or unavailable
+    count = str(event.order_count) if event.coverage_status == "known" and event.order_count is not None else unknown
+    stamp = format_sgt_time(event.occurred_at, locale=lang)
+    if zh:
+        return "\n".join(("【资产管理｜运行错误】", f"时间：{stamp}", f"平台 / 策略：{platform} / {strategy}",
+            f"问题：{problem}", f"影响：是否提交订单：{count}；是否有待处理订单：未知",
+            f"证据：{link}", "处理：查看本次运行记录，核实输入；未恢复前保持执行保护",
+            "投递状态以回执为准"))
+    return "\n".join(("[Asset management | Runtime error]", f"Time: {stamp}", f"Platform / strategy: {platform} / {strategy}",
+        f"Problem: {problem}", f"Impact: Orders submitted: {count}; pending orders: unknown",
+        f"Evidence: {link}", "Action: inspect this run and verify inputs; keep execution protections active",
+        "Delivery status requires a receipt"))

src/quant_platform_kit/notifications/strategy_plugin_telegram.py

@@ -16,6 +16,10 @@
 from .alert_marker import CloudAlertMarkerStore, _clean_relative_key
 from ._redaction import redact_exception
 from .digest_routing import daily_digest_enabled
+from .renderer_base import render_asset_management_error
+from .unified_contract import (
+    ASSET_MANAGEMENT_DIGEST_KINDS, CandidatePersistenceError, IncidentTracker, NotificationEvent, route_asset_management_event,
+)
 from .telegram import (
     DEFAULT_TELEGRAM_BOT_API_BASE_URL,
     parse_telegram_chat_ids,
@@ -185,6 +189,9 @@ def publish_strategy_plugin_telegram_alerts(
     alert_store: StrategyPluginTelegramAlertMarkerStore | object | None = None,
     send_notification: Callable[..., bool] = send_strategy_plugin_telegram,
     log_message: Callable[..., Any] = print,
+    asset_management_event_factory: Callable[[object], NotificationEvent] | None = None,
+    persist_candidate: Callable[[dict[str, object]], bool] | None = None,
+    asset_management_incident_tracker: IncidentTracker | None = None,
 ) -> StrategyPluginTelegramAlertPublishResult:
     """Use the same builder eligibility as email, preserving Telegram exclusions.
 
@@ -194,6 +201,11 @@ def publish_strategy_plugin_telegram_alerts(
     per-channel dedupe and delivery failures are independent of eligibility.
     """
     settings = StrategyPluginTelegramSettings.from_object(telegram_settings)
+    if asset_management_event_factory is not None:
+        return _publish_asset_management_signals(
+            signals, event_factory=asset_management_event_factory, persist_candidate=persist_candidate,
+            tracker=asset_management_incident_tracker, settings=settings, send_notification=send_notification,
+        )
     del context_label  # The plugin Telegram bot is a unified manual-review channel.
     telegram_signals = tuple(
         signal for signal in signals if not _is_auto_consumable_signal(signal)
@@ -245,6 +257,69 @@ def publish_strategy_plugin_telegram_alerts(
     return result
 
 
+def _publish_asset_management_signals(signals, *, event_factory, persist_candidate, tracker,
+                                      settings, send_notification):
+    """Explicit asset-management use only; other plugin channels keep parity.
+
+    A durable incident tracker is mandatory for errors. Pending/uncertain sends
+    survive restart and are never retried merely because no ACK was returned.
+    The producer archives its original signals; the factory binds each fact to
+    the same real source run. Candidate persistence is not delivery.
+    """
+    if persist_candidate is None:
+        raise CandidatePersistenceError("notification_candidate_writer_required")
+    deliveries = []
+    for signal in signals:
+        event = event_factory(signal)
+        if not isinstance(event, NotificationEvent):
+            raise ValueError("notification_structured_event_required")
+        if event.event_kind == "recovery" and event.incident_reason and tracker is None:
+            raise RuntimeError("notification_durable_incident_tracker_required")
+        if event.event_kind in ASSET_MANAGEMENT_DIGEST_KINDS and any(
+            getattr(signal, key, None) for key in ("error", "errors", "risk_halt", "auth_expired")
+        ):
+            raise ValueError("notification_routine_incident_conflict")
+
+        def publish_error(fact):
+            if tracker is None:
+                raise RuntimeError("notification_durable_incident_tracker_required")
+            incident = fact if fact.incident_reason else replace(fact, incident_reason="unclassified_event")
+            # Check transport before reserving: missing configuration is a
+            # failure, not a successful delivery or an ambiguous send attempt.
+            if settings.missing_fields():
+                raise RuntimeError("notification_transport_configuration_missing")
+            decision = tracker.observe(incident, state=incident.event_kind)
+            if decision.action == "coverage_unknown":
+                raise RuntimeError("notification_incident_identity_unknown")
+            key = decision.transition_key or ""
+            if decision.action != "send":
+                return StrategyPluginTelegramAlertDelivery(key, "", "skipped", reason="duplicate_alert")
+            message = StrategyPluginAlertMessage(subject="", body=render_asset_management_error(fact), alert_key=key)
+            sent, error = _send_message(send_notification, message, settings)
+            # A transport boolean is not a receipt. Both false and exceptions
+            # are ambiguous; preserve the reservation and never blindly retry.
+            tracker.record_delivery(decision)
+            return StrategyPluginTelegramAlertDelivery(key, "", "pending" if sent else "failed",
+                reason=None if sent else "send_failed",
+                error="notification_transport_unconfirmed" if error else None,
+                metadata={"delivery_status": "uncertain", "transport_acknowledged": False,
+                          "transport_attempted": True})
+
+        result = route_asset_management_event(event, persist_candidate=persist_candidate, publish_error=publish_error)
+        if result.route == "digest_candidate":
+            if event.event_kind == "recovery" and event.incident_reason:
+                decision = tracker.observe(event, state="recovered", severity=0, recovered=True)
+                if decision.action == "coverage_unknown":
+                    raise RuntimeError("notification_incident_identity_unknown")
+            deliveries.append(StrategyPluginTelegramAlertDelivery("", "", "deferred_digest",
+                metadata={"candidate_persisted": True, "transport_acknowledged": False}))
+        else:
+            if not result.candidate_persisted:
+                raise CandidatePersistenceError("notification_candidate_write_unconfirmed")
+            deliveries.append(result.result)
+    return StrategyPluginTelegramAlertPublishResult(tuple(deliveries))
+
+
 def _routine_observation_to_digest(signal: object) -> bool:
     """Routine observations remain in the persisted signal report, without a push."""
     if not daily_digest_enabled():

src/quant_platform_kit/notifications/unified_contract.py

@@ -23,6 +23,12 @@
 })
 ROUTINE_EVENT_KINDS = frozenset({"routine_cycle", "healthy_heartbeat"})
 ORDER_EVENT_KINDS = frozenset({"order_submitted", "fill"})
+# Additional P10 facts are opt-in for the asset-management channel. Legacy
+# users of the public transport and router keep their existing semantics.
+ASSET_MANAGEMENT_DIGEST_KINDS = ROUTINE_EVENT_KINDS | ORDER_EVENT_KINDS | frozenset({
+    "recovery", "maintenance_reminder", "research_candidate", "strategy_signal",
+})
+EVENT_KINDS = EVENT_KINDS | ASSET_MANAGEMENT_DIGEST_KINDS
 SGT = ZoneInfo("Asia/Singapore")
 
 
@@ -157,13 +163,21 @@ def route_notification_event(
     event: NotificationEvent, *, env: Mapping[str, object], flag_key: str,
     persist_candidate: Callable[[dict[str, object]], bool],
     legacy: Callable[[], object], immediate: Callable[[], object],
+    asset_management: bool = False,
 ) -> RoutingResult:
     """Persist before deferral. Alert delivery survives candidate-write failure.
 
     The immediate callback owns durable incident reservation and receipt ACK.
     Callers must inspect candidate_persisted; deferral is never delivery.
     """
-    route = select_notification_route(event_kind=event.event_kind, env=env, flag_key=flag_key)
+    if asset_management:
+        if not isinstance(event, NotificationEvent):
+            raise ValueError("notification_structured_event_required")
+        route = "digest_candidate" if event.event_kind in ASSET_MANAGEMENT_DIGEST_KINDS else "immediate"
+        if route == "digest_candidate" and event.incident_reason and event.event_kind != "recovery":
+            raise ValueError("notification_routine_incident_conflict")
+    else:
+        route = select_notification_route(event_kind=event.event_kind, env=env, flag_key=flag_key)
     if route == "legacy":
         return RoutingResult(route, False, legacy())
     if route == "digest_candidate":
@@ -184,6 +198,23 @@ def route_notification_event(
     return RoutingResult(route, persisted, immediate())
 
 
+def route_asset_management_event(
+    event: NotificationEvent, *,
+    persist_candidate: Callable[[dict[str, object]], bool],
+    publish_error: Callable[[NotificationEvent], object],
+) -> RoutingResult:
+    """P10 channel boundary: only structured errors reach the incident owner.
+
+    The producer must establish classification before calling. Expected blocks
+    still need their existing producer proof; a bare evidence string is not
+    sufficient to defer them. Unknown kinds remain errors. publish_error owns
+    durable incident reservation/receipts, exactly as the existing router does.
+    No release flag can turn normal asset-management facts into an instant send.
+    """
+    return route_notification_event(event, env={}, flag_key="asset_management", asset_management=True,
+        persist_candidate=persist_candidate, legacy=lambda: None, immediate=lambda: publish_error(event))
+
+
 class IncidentMarkerStore(Protocol):
     """Durable JSON state, with atomic compare-and-set across workers.