diff --git a/docs/notification-routing-policy.md b/docs/notification-routing-policy.md
index 923083c6..a39023f4 100644
--- a/docs/notification-routing-policy.md
+++ b/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.
diff --git a/src/quant_platform_kit/notifications/_redaction.py b/src/quant_platform_kit/notifications/_redaction.py
index d04067c3..8d2ed8ae 100644
--- a/src/quant_platform_kit/notifications/_redaction.py
+++ b/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
diff --git a/src/quant_platform_kit/notifications/renderer_base.py b/src/quant_platform_kit/notifications/renderer_base.py
index 21b5fc6a..acab54cc 100644
--- a/src/quant_platform_kit/notifications/renderer_base.py
+++ b/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"))
diff --git a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py b/src/quant_platform_kit/notifications/strategy_plugin_telegram.py
index fcc90170..7c1dbc1d 100644
--- a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py
+++ b/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():
diff --git a/src/quant_platform_kit/notifications/unified_contract.py b/src/quant_platform_kit/notifications/unified_contract.py
index bbb2cdbe..0d195f2c 100644
--- a/src/quant_platform_kit/notifications/unified_contract.py
+++ b/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.
 
diff --git a/tests/test_asset_management_notifications.py b/tests/test_asset_management_notifications.py
new file mode 100644
index 00000000..537f8e7c
--- /dev/null
+++ b/tests/test_asset_management_notifications.py
@@ -0,0 +1,318 @@
+"""P10 shared entrypoints; fake-only producers, storage and transport."""
+from dataclasses import replace
+from datetime import datetime
+from types import SimpleNamespace
+
+import pytest
+
+from quant_platform_kit.notifications import renderer_base, unified_contract as contract
+from quant_platform_kit.notifications.strategy_plugin_telegram import (
+    StrategyPluginTelegramSettings, publish_strategy_plugin_telegram_alerts,
+)
+
+STAMP = datetime.fromisoformat("2026-11-04T23:00:00+08:00")
+
+
+def event(kind="real_error", **fields):
+    return replace(contract.NotificationEvent(kind, STAMP, "ibkr", "fake-opaque", "fake-target",
+                                            "fake-strategy", "fake-run"), **fields)
+
+
+@pytest.mark.parametrize("kind", ["routine_cycle", "healthy_heartbeat", "order_submitted", "fill",
+                                  "recovery", "maintenance_reminder", "research_candidate", "strategy_signal"])
+def test_asset_non_errors_persist_without_any_send(kind):
+    rows, calls = [], []
+    fact = event(kind, order_execution_digest="a" * 64)
+    result = contract.route_asset_management_event(fact,
+        persist_candidate=lambda row: rows.append(row) or True,
+        publish_error=lambda _: calls.append("error"))
+    assert result.route == "digest_candidate" and result.candidate_persisted
+    assert rows[0]["source_run_id"] == "fake-run" and not calls
+
+
+@pytest.mark.parametrize("failure", [False, None, "raise"])
+def test_asset_deferral_never_swallows_write_failures(failure):
+    def persist(_):
+        if failure == "raise":
+            raise OSError("fake storage outage")
+        return failure
+    with pytest.raises(contract.CandidatePersistenceError):
+        contract.route_asset_management_event(event("recovery"), persist_candidate=persist,
+            publish_error=lambda _: pytest.fail("normal message fallback"))
+
+
+def test_asset_error_still_publishes_when_candidate_write_fails():
+    calls = []
+    fact = event(incident_reason="empty_position_decision")
+    result = contract.route_asset_management_event(fact, persist_candidate=lambda _: False,
+        publish_error=lambda item: calls.append(item) or "receipt-pending")
+    assert calls == [fact] and result.route == "immediate" and not result.candidate_persisted
+    assert result.result == "receipt-pending"
+
+
+def test_asset_unknown_classification_is_an_error_not_a_healthy_label():
+    calls = []
+    contract.route_asset_management_event(event("future_unknown"), persist_candidate=lambda _: True,
+                                         publish_error=lambda item: calls.append(item))
+    assert len(calls) == 1
+
+
+@pytest.mark.parametrize("locale", ["zh-CN", "en"])
+def test_error_format_omits_private_identity_raw_reason_and_unknown_counts(locale):
+    fact = event(incident_reason="fake-account secret=FAKE_SECRET")
+    text = renderer_base.render_asset_management_error(fact, locale=locale,
+        run_url="https://example.test/runs/1?token=FAKE_SECRET", allowed_run_hosts=("example.test",))
+    assert "2026-11-04 23:00:00" in text
+    assert "https://example.test/runs/1" in text
+    assert all(value not in text for value in ("fake-opaque", "fake-target", "fake-strategy", "fake-run",
+                                               "fake-account", "FAKE_SECRET", "token="))
+    assert ("是否提交订单：未知" if locale == "zh-CN" else "Orders submitted: unknown") in text
+
+
+@pytest.mark.parametrize("url", ["https://fake-private.test/runs/1", "https://example.test/fake-opaque",
+                                  "https://user:pass@example.test/run", "gs://fake-bucket/run"])
+def test_unapproved_or_identity_bearing_run_link_is_unavailable(url):
+    text = renderer_base.render_asset_management_error(event(), run_url=url,
+                                                       allowed_run_hosts=("example.test",))
+    assert url not in text and "运行链接不可用" in text
+
+
+def test_error_formatter_rejects_healthy_envelope_and_keeps_empty_decision_error():
+    with pytest.raises(ValueError):
+        renderer_base.render_asset_management_error(event("recovery"))
+    text = renderer_base.render_asset_management_error(event(incident_reason="empty_position_decision"))
+    assert "策略决策不可用" in text and "无需调仓" not in text
+    assert "是否提交订单：未知" in text
+
+
+def signal():
+    return SimpleNamespace(strategy="fake-strategy", plugin="market_regime_control", effective_mode="shadow",
+        as_of="2026-11-04", canonical_route="opportunity_watch", suggested_action="notify_manual_review",
+        would_trade_if_enabled=False, execution_controls={"notification_profile": "shadow_only"})
+
+
+@pytest.mark.parametrize("mode", ["immediate", "daily_digest", "invalid"])
+def test_asset_plugin_entrypoint_uses_structured_route_in_every_mode(monkeypatch, mode):
+    monkeypatch.setenv("QSL_ROUTINE_NOTIFICATION_MODE", mode)
+    calls, rows = [], []
+    result = publish_strategy_plugin_telegram_alerts([signal()],
+        telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+        asset_management_event_factory=lambda _: event("research_candidate"),
+        persist_candidate=lambda row: rows.append(row) or True,
+        send_notification=lambda **kwargs: calls.append(kwargs) or True, log_message=lambda *a, **k: None)
+    assert not calls and len(rows) == 1 and result.sent_count == 0
+
+
+def test_asset_plugin_error_uses_fixed_body_and_existing_marker_dedupe():
+    calls = []
+    store = Store()
+    kwargs = dict(telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+        asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+        persist_candidate=lambda _: True, asset_management_incident_tracker=contract.IncidentTracker(store),
+        send_notification=lambda **kw: calls.append(kw) or True, log_message=lambda *a, **k: None)
+    publish_strategy_plugin_telegram_alerts([signal()], **kwargs)
+    publish_strategy_plugin_telegram_alerts([signal()], **kwargs)
+    assert len(calls) == 1 and calls[0]["title"] == ""
+    assert calls[0]["body"].startswith("【资产管理｜运行错误】")
+    assert "无需调仓" not in calls[0]["body"] and "fake-strategy" not in calls[0]["body"]
+
+
+def test_asset_plugin_requires_persistence_before_sending():
+    with pytest.raises(contract.CandidatePersistenceError):
+        publish_strategy_plugin_telegram_alerts([signal()], telegram_settings={},
+            asset_management_event_factory=lambda _: event("research_candidate"),
+            send_notification=lambda **_: pytest.fail("must not send"))
+
+
+class Store:
+    def __init__(self):
+        self.rows = {}
+
+    def read(self, key):
+        return self.rows.get(key, (None, None))
+
+    def compare_and_set(self, key, version, value):
+        current, _ = self.read(key)
+        if current != version:
+            return False
+        self.rows[key] = ((current or 0) + 1, dict(value))
+        return True
+
+
+def test_asset_error_requires_durable_tracker_before_transport():
+    with pytest.raises(RuntimeError, match="durable_incident_tracker_required"):
+        publish_strategy_plugin_telegram_alerts([signal()],
+            telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+            asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+            persist_candidate=lambda _: True,
+            send_notification=lambda **_: pytest.fail("unreserved send"))
+
+
+@pytest.mark.parametrize("send_result", [False, True, "raise"])
+def test_asset_error_restart_never_retries_unknown_transport(send_result):
+    calls, store = [], Store()
+    def send(**_):
+        calls.append("send")
+        if send_result == "raise":
+            raise TimeoutError("fake timeout")
+        return send_result
+    for _ in range(2):
+        publish_strategy_plugin_telegram_alerts([signal()],
+            telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+            asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+            persist_candidate=lambda _: True,
+            asset_management_incident_tracker=contract.IncidentTracker(store), send_notification=send)
+    assert calls == ["send"]
+    assert next(iter(store.rows.values()))[1]["delivery_status"] == "uncertain"
+
+
+def test_current_incident_transition_is_not_overwritten_after_sender_await():
+    store = Store()
+    fact = event(incident_reason="empty_position_decision")
+    def send(**_):
+        contract.IncidentTracker(store).observe(fact, state="new_failure", severity=2)
+        return True
+    with pytest.raises(RuntimeError, match="transition_changed"):
+        publish_strategy_plugin_telegram_alerts([signal()],
+            telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+            asset_management_event_factory=lambda _: fact, persist_candidate=lambda _: True,
+            asset_management_incident_tracker=contract.IncidentTracker(store), send_notification=send)
+    row = next(iter(store.rows.values()))[1]
+    assert row["state"] == "new_failure" and row["delivery_status"] == "pending"
+
+
+def test_conflicting_plugin_error_cannot_be_hidden_in_research_candidate():
+    item = signal()
+    item.error = "fake input failure"
+    with pytest.raises(ValueError, match="routine_incident_conflict"):
+        publish_strategy_plugin_telegram_alerts([item], telegram_settings={},
+            asset_management_event_factory=lambda _: event("research_candidate"),
+            persist_candidate=lambda _: pytest.fail("must not archive as healthy"))
+
+
+def test_overlong_link_does_not_exceed_single_message_budget():
+    text = renderer_base.render_asset_management_error(event(),
+        run_url="https://example.test/" + "x" * 5000, allowed_run_hosts=("example.test",))
+    assert len(text.encode("utf-16-le")) // 2 <= 3900
+    assert "运行链接不可用" in text
+
+
+def test_asset_candidate_unknown_identity_and_conflicting_reason_fail_closed():
+    for fact in (event("recovery", source_run_id=None), event("research_candidate", incident_reason="auth_failure")):
+        with pytest.raises((contract.CandidatePersistenceError, ValueError)):
+            contract.route_asset_management_event(fact,
+                persist_candidate=lambda _: pytest.fail("invalid fact must not be persisted"),
+                publish_error=lambda _: pytest.fail("invalid fact must not silently route"))
+
+
+class SQLiteStore:
+    """Test-only durable CAS implementation shared by independent processes."""
+    def __init__(self, path):
+        self.path = path
+
+    def read(self, key):
+        import json
+        import sqlite3
+        with sqlite3.connect(self.path) as db:
+            row = db.execute("SELECT version, value FROM markers WHERE key = ?", (key,)).fetchone()
+        return (row[0], json.loads(row[1])) if row else (None, None)
+
+    def compare_and_set(self, key, version, value):
+        import json
+        import sqlite3
+        with sqlite3.connect(self.path) as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute("SELECT version FROM markers WHERE key = ?", (key,)).fetchone()
+            current = row[0] if row else None
+            if current != version:
+                return False
+            db.execute("INSERT OR REPLACE INTO markers VALUES (?, ?, ?)",
+                       (key, (current or 0) + 1, json.dumps(value)))
+        return True
+
+
+def _asset_worker(path, barrier, queue):
+    barrier.wait(timeout=10)
+    result = publish_strategy_plugin_telegram_alerts([signal()],
+        telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+        asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+        persist_candidate=lambda _: True,
+        asset_management_incident_tracker=contract.IncidentTracker(SQLiteStore(path)),
+        send_notification=lambda **_: True)
+    queue.put(sum(item.metadata.get("transport_attempted", False) for item in result.deliveries))
+
+
+def test_asset_error_cross_process_reservation_and_restart(tmp_path):
+    import multiprocessing
+    import sqlite3
+    path = str(tmp_path / "fake-markers.sqlite")
+    with sqlite3.connect(path) as db:
+        db.execute("CREATE TABLE markers (key TEXT PRIMARY KEY, version INTEGER, value TEXT)")
+    ctx = multiprocessing.get_context("spawn")
+    barrier, queue = ctx.Barrier(4), ctx.Queue()
+    processes = [ctx.Process(target=_asset_worker, args=(path, barrier, queue)) for _ in range(4)]
+    for process in processes:
+        process.start()
+    for process in processes:
+        process.join(timeout=20)
+        assert process.exitcode == 0
+    assert sum(queue.get(timeout=2) for _ in processes) == 1
+    # A fresh process after all earlier writers exited cannot resend uncertainty.
+    barrier = ctx.Barrier(1)
+    process = ctx.Process(target=_asset_worker, args=(path, barrier, queue))
+    process.start()
+    process.join(timeout=20)
+    assert process.exitcode == 0 and queue.get(timeout=2) == 0
+    queue.close()
+
+
+def test_asset_transport_success_without_receipt_is_pending_and_errors_are_private():
+    for failure in (False, True):
+        def send(**_):
+            if failure:
+                raise OSError("fake-opaque fake-account fake-project token=FAKE_SECRET")
+            return True
+        result = publish_strategy_plugin_telegram_alerts([signal()],
+            telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+            asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+            persist_candidate=lambda _: True, asset_management_incident_tracker=contract.IncidentTracker(Store()),
+            send_notification=send)
+        assert result.sent_count == 0
+        assert result.deliveries[0].status == ("failed" if failure else "pending")
+        assert all(value not in repr(result) for value in ("fake-opaque", "fake-account", "fake-project", "FAKE_SECRET"))
+
+
+def test_confirmed_recovery_clears_incident_without_push_and_next_error_is_new():
+    store, calls = Store(), []
+    for kind in ("real_error", "recovery", "real_error"):
+        result = publish_strategy_plugin_telegram_alerts([signal()],
+            telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+            asset_management_event_factory=lambda _, kind=kind: event(kind, incident_reason="empty_position_decision"),
+            persist_candidate=lambda _: True, asset_management_incident_tracker=contract.IncidentTracker(store),
+            send_notification=lambda **_: calls.append(kind) or True)
+        assert result.sent_count == 0
+    assert calls == ["real_error", "real_error"]
+
+
+def test_failed_recovery_persistence_does_not_clear_active_incident():
+    store = Store()
+    tracker = contract.IncidentTracker(store)
+    tracker.observe(event(incident_reason="empty_position_decision"), state="real_error")
+    with pytest.raises(contract.CandidatePersistenceError):
+        publish_strategy_plugin_telegram_alerts([signal()], telegram_settings={},
+            asset_management_event_factory=lambda _: event("recovery", incident_reason="empty_position_decision"),
+            persist_candidate=lambda _: False, asset_management_incident_tracker=tracker,
+            send_notification=lambda **_: pytest.fail("recovery must not push"))
+    assert next(iter(store.rows.values()))[1]["active"] is True
+
+
+def test_asset_error_candidate_failure_alerts_once_then_propagates_failure():
+    calls, store = [], Store()
+    for _ in range(2):
+        with pytest.raises(contract.CandidatePersistenceError):
+            publish_strategy_plugin_telegram_alerts([signal()],
+                telegram_settings=StrategyPluginTelegramSettings(chat_ids=("fake-chat",), bot_token="fake-credential"),
+                asset_management_event_factory=lambda _: event(incident_reason="empty_position_decision"),
+                persist_candidate=lambda _: False, asset_management_incident_tracker=contract.IncidentTracker(store),
+                send_notification=lambda **_: calls.append("send") or True)
+    assert calls == ["send"]