fix: Preserve source evidence in attention notifications

docs/unified-daily-notification.md

@@ -175,3 +175,12 @@ unknown delivery and restart suppression remain authoritative. Run time and run
 links come from the existing run report; absent evidence stays explicitly
 unknown. This format change does not change execution, dependency pins or
 notification schedules and does not establish deployed adoption or delivery.
+
+Incident attention messages retain allowlisted completed block causes from
+`execution_blocked_reason` or `no_op_reason`; unknown provider text retains the
+fixed incident classification. The pinned shared compact renderer receives the
+report's existing `started_at` and `console_run_url`, just like newer renderers.
+It validates the supplied HTTPS link and removes query strings/fragments. A
+missing or unsafe source link remains unavailable; a monitor run or guessed
+console route is never substituted for a business run. These display fields do
+not change routing, incident identity, risk decisions or execution permission.

notifications/unified_adapter.py

@@ -469,7 +469,6 @@ def _kind(execution, *, dry_run=False):
 
 def _cycle_attention_notification(kind, execution, report, diagnostics=""):
     """Render incident facts with QPK; never forward a legacy heartbeat body."""
-    from zoneinfo import ZoneInfo
     from notifications.events import RenderedNotification
     from notifications.redaction import redact_notification_diagnostics
     from quant_platform_kit.notifications._redaction import redact_sensitive_text
@@ -480,30 +479,21 @@ def _cycle_attention_notification(kind, execution, report, diagnostics=""):
         os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG"))
     profile = report.get("strategy_profile") or (report.get("runtime_target") or {}).get("strategy_profile")
     profile = profile if isinstance(profile, str) and profile in STRATEGY_CATALOG.definitions else "unknown"
-    reason = ("empty_position_decision" if execution.get("execution_blocked_reason") == "empty_position_decision"
-              else kind)
+    # Only fixed producer codes may cross the private execution boundary.
+    known_blocks = {"empty_position_decision", "strategy_risk_rejected",
+                    "account_brake_source_or_submit_unavailable", "account_brake_submission_not_confirmed"}
+    reason = next((value for key in ("execution_blocked_reason", "no_op_reason")
+                   if isinstance(value := execution.get(key), str) and value in known_blocks), kind)
     decision = shared.AttentionDecision(
         level=shared.AttentionLevel.ACTION, reason_codes=("operational_uncertain",),
         dd_ratio=None, should_notify=True, next_step_key="open_console_confirm")
     arguments = dict(locale=locale, platform="ibkr", account_alias="unknown",
                      strategy_profile=profile, decision=decision, guard_reason_codes=(reason,))
-    renderer = getattr(shared, "render_attention_notification", None)
-    if renderer is not None:
-        text = renderer(**arguments, occurred_at=report.get("started_at"), run_url=report.get("console_run_url"))
-    else:
-        # The frozen consumer pin predates QPK #769. Keep its shared layout;
-        # do not construct a console URL or forward private diagnostic payloads.
-        text = shared.render_attention_compact(**arguments)
-        unknown = "未知" if locale == "zh" else "Unknown"
-        stamp = unknown
-        try:
-            value = datetime.fromisoformat(str(report.get("started_at")).replace("Z", "+00:00"))
-            if value.tzinfo is not None and value.utcoffset() is not None:
-                stamp = value.astimezone(ZoneInfo("Asia/Singapore")).strftime("%Y-%m-%d %H:%M:%S SGT")
-        except (TypeError, ValueError, OverflowError):
-            pass
-        text += (f"\n运行时间：{stamp}\n运行详情：当前格式版本不提供运行链接" if locale == "zh" else
-                 f"\nRun time: {stamp}\nRun details: Run link unavailable with the pinned formatter")
+    # The pinned compact renderer already supports these evidence fields and
+    # validates/strips URLs. Keep that shared boundary instead of dropping the
+    # link and appending a second run-time section in the compatibility path.
+    renderer = getattr(shared, "render_attention_notification", shared.render_attention_compact)
+    text = renderer(**arguments, occurred_at=report.get("started_at"), run_url=report.get("console_run_url"))
     # Carry only the caller's decision diagnostics, never its legacy dashboard/title.
     if diagnostics:
         label = "诊断" if locale == "zh" else "Diagnostics"

tests/test_attention_source_evidence.py

@@ -0,0 +1,49 @@
+"""Carry existing safe evidence through the pinned shared attention renderer."""
+from copy import deepcopy
+
+import pytest
+
+from notifications import unified_adapter as adapter
+
+
+@pytest.mark.parametrize("field,code", [
+    ("execution_blocked_reason", "empty_position_decision"),
+    ("no_op_reason", "strategy_risk_rejected"),
+    ("execution_blocked_reason", "account_brake_source_or_submit_unavailable"),
+])
+def test_blocked_attention_preserves_known_source_cause(monkeypatch, field, code):
+    monkeypatch.setenv("NOTIFY_LANG", "en")
+    execution = {"execution_status": "blocked", field: code}
+    original = deepcopy(execution)
+    body = adapter._cycle_attention_notification("blocked_run", execution, {}).detailed_text
+    assert code in body
+    assert execution == original
+
+
+@pytest.mark.parametrize("value", ["private account detail", "synthetic_private_code", None, [], {}])
+def test_unrecognized_source_reason_retains_fixed_fallback(monkeypatch, value):
+    monkeypatch.setenv("NOTIFY_LANG", "en")
+    body = adapter._cycle_attention_notification("blocked_run", {"no_op_reason": value}, {}).detailed_text
+    assert "blocked_run" in body
+    assert "private account detail" not in body and "synthetic_private_code" not in body
+
+
+def test_pinned_compact_renderer_keeps_provided_run_url_once(monkeypatch):
+    from quant_platform_kit.risk import attention
+    monkeypatch.delattr(attention, "render_attention_notification", raising=False)
+    monkeypatch.setenv("NOTIFY_LANG", "en")
+    report = {"started_at": "2026-11-05T20:45:01Z",
+              "console_run_url": "https://console.example.invalid/runs/synthetic?token=private"}
+    original = deepcopy(report)
+    body = adapter._cycle_attention_notification("blocked_run", {}, report).detailed_text
+    assert "https://console.example.invalid/runs/synthetic" in body
+    assert "private" not in body and "?token=" not in body
+    assert body.count("Run time:") == 1
+    assert report == original
+
+
+@pytest.mark.parametrize("url", [None, "javascript:synthetic", "https://user:pass@console.example.invalid/run"])
+def test_missing_or_unsafe_run_url_is_not_invented(monkeypatch, url):
+    monkeypatch.setenv("NOTIFY_LANG", "en")
+    body = adapter._cycle_attention_notification("blocked_run", {}, {"console_run_url": url}).detailed_text
+    assert "https://" not in body and "javascript:" not in body and "user:pass" not in body