fix: Unify Firstrade operational error notifications

application/rebalance_service.py

@@ -275,13 +275,33 @@ def _publish_cycle_notification(
     settings: PlatformRuntimeSettings,
     notification_sender: Callable[[str], None] | None = None,
     log_message: Callable[[str], None] = print,
+    report: Mapping[str, Any] | None = None,
 ) -> bool:
     sender = notification_sender
     if sender is None:
         if not settings.tg_token or not settings.tg_chat_id:
             return False
         sender = build_sender(settings.tg_token, settings.tg_chat_id)
-    notification = render_cycle_notification(result, lang=settings.notify_lang)
+    from notifications.unified import alert_text, cycle_kind
+    kind = cycle_kind(result)
+    if kind in {"routine_cycle", "order_submitted", "fill"}:
+        notification = render_cycle_notification(result, lang=settings.notify_lang)
+    else:
+        from quant_platform_kit.notifications.events import RenderedNotification
+        from datetime import datetime
+        observed_at = None
+        if report:
+            try:
+                observed_at = datetime.fromisoformat(str(report.get("started_at", "")).replace("Z", "+00:00"))
+            except ValueError:
+                pass
+        skips = result.get("execution_blocking_skips") or result.get("skipped_orders") or ()
+        display_kind = "fractional_trading_disclosure_required" if any(
+            isinstance(item, Mapping) and item.get("reason") == "fractional_trading_disclosure_required"
+            for item in skips) else kind
+        text = alert_text(display_kind, settings.notify_lang, occurred_at=observed_at,
+                          run_url=(report or {}).get("workflow_run_url"))
+        notification = RenderedNotification(detailed_text=text, compact_text=text)
     def publish_log(text: str) -> None:
         try:
             log_message(text, flush=True)

main.py

@@ -196,16 +196,10 @@ def _telegram_notification_targets() -> tuple[tuple[str, str], ...]:
 
 
 def _runtime_error_notification_message(exc: Exception) -> str:
-    t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG"))
-    title = "runtime_probe_failure_title" if request.path == "/probe" else "runtime_failure_title"
-    strategy_name = os.getenv("STRATEGY_DISPLAY_NAME") or os.getenv("STRATEGY_PROFILE") or "Firstrade"
-    return "\n".join((
-        t(title),
-        t("strategy_label", name=strategy_name),
-        t("runtime_failure_context", context=os.getenv('ACCOUNT_REGION') or os.getenv('K_SERVICE') or 'Firstrade'),
-        t("runtime_failure_result"),
-        t("runtime_failure_action"),
-    ))
+    from notifications.unified import alert_text, coverage_locale, workflow_run_url
+    return alert_text("real_error", coverage_locale(), occurred_at=datetime.now(timezone.utc),
+                      source="heartbeat" if request.path == "/probe" else "runtime",
+                      run_url=workflow_run_url())
 
 
 def _notify_runtime_error(exc: Exception) -> bool:
@@ -550,7 +544,7 @@ def send_failure():
                     return result
                 result["report_persistence_error"] = "execution_report_write_unconfirmed"
             delivery = notify_cycle(report, result, settings=settings,
-                send=lambda: _publish_cycle_notification(result, settings=settings),
+                send=lambda: _publish_cycle_notification(result, settings=settings, report=report),
                 send_failure=send_failure)
             result["notification_consolidation"] = delivery
             result["notification_sent"] = delivery.get("transport_acknowledged") is True

notifications/unified.py

@@ -606,13 +606,69 @@ def notify_heartbeat(*, kind, send, now, env=None, store=None, recovery_reports=
         return {"candidate_persisted": False, "failed": True}
 
 
-def alert_text(kind, locale):
+def alert_text(kind, locale=None, *, occurred_at=None, source="runtime", run_url=None):
+    """Plan P10 presentation contract; never interpolate private runtime payloads.
+
+    Monitor observations are not business runs. Missing source time/link stays
+    unverified. Replace this renderer with the step-1 shared formatter once its
+    reviewed dependency pin is available; no vendor snapshot is edited here.
+    """
+    from quant_platform_kit.notifications._redaction import redact_sensitive_text
+    from zoneinfo import ZoneInfo
+
+    zh = not str(locale or "zh").strip().lower().startswith("en")
     messages = {
-        "receipt_missing": ("执行回执缺失。", "Execution receipt is missing."),
-        "receipt_unacceptable": ("执行回执需要检查。", "Execution receipt needs review."),
-        "real_error": ("运行发生异常。", "The runtime encountered an error."),
-        "candidate_write_failure": ("通知候选未能确认写入。", "Notification candidate storage could not be confirmed."),
+        "receipt_missing": ("执行回执缺失", "Execution receipt is missing"),
+        "receipt_unacceptable": ("执行回执无法核实", "Execution receipt cannot be verified"),
+        "receipt_identity_conflict": ("执行回执身份冲突", "Execution receipt identity conflict"),
+        "receipt_stale": ("执行回执早于应运行时间", "Execution receipt predates the required run"),
+        "real_error": ("运行发生异常", "Runtime error"),
+        "auth_failure": ("认证失败", "Authentication failed"),
+        "blocked_run": ("执行被阻断", "Execution is blocked"),
+        "risk_escalation": ("风险检查需要处理", "Risk check requires attention"),
+        "order_status_unknown": ("订单结果无法核实", "Order outcome is unknown"),
+        "candidate_write_failure": ("通知候选未能确认写入", "Notification candidate storage could not be confirmed"),
+        "runtime_guard": ("运行监控发现异常", "Runtime monitor detected an error"),
     }
-    zh = str(locale or "").strip().lower().startswith("zh")
-    title = messages.get(kind, messages["receipt_unacceptable"])[0 if zh else 1]
-    return title + ("\n请查看本次运行记录。" if zh else "\nReview the current run record.")
+    problem = messages.get(kind, messages["receipt_unacceptable"])[0 if zh else 1]
+    if kind == "fractional_trading_disclosure_required":
+        from notifications.telegram import build_translator
+        problem = build_translator("zh" if zh else "en")("skip_reason_fractional_trading_disclosure_required")
+    stamp = "未核实" if zh else "unverified"
+    if isinstance(occurred_at, datetime) and occurred_at.tzinfo is not None and occurred_at.utcoffset() is not None:
+        stamp = occurred_at.astimezone(ZoneInfo("Asia/Singapore")).strftime("%Y-%m-%d %H:%M:%S")
+        if not zh:
+            stamp += " SGT"
+        else:
+            stamp += "（新加坡时间）"
+    producers = {
+        "runtime": ("策略运行", "strategy runtime"),
+        "heartbeat": ("回执检查", "receipt monitor"),
+        "runtime_guard": ("运行监控", "runtime monitor"),
+    }
+    producer = producers.get(source, producers["runtime"])[0 if zh else 1]
+    # Only authentic first-party Actions references, never arbitrary hostnames,
+    # query strings, account selectors or cloud resource paths.
+    link = "运行链接不可用" if zh else "run link unavailable"
+    if isinstance(run_url, str) and re.fullmatch(
+        r"https://github\.com/QuantStrategyLab/FirstradePlatform/actions/runs/[0-9]+", run_url
+    ):
+        link = redact_sensitive_text(run_url)
+    if zh:
+        lines = ["【资产管理｜运行错误】", f"时间：{stamp}", "平台 / 策略：第一证券 / 未核实",
+            f"问题：{problem}", "影响：是否提交订单：未知；是否有待处理订单：未知",
+            f"证据：{producer}；{link}", "处理：查看本次运行记录，核实输入；未恢复前保持执行保护",
+            "事件：未提供安全事件引用｜投递状态以回执为准"]
+    else:
+        lines = ["[Asset management | Runtime error]", f"Time: {stamp}", "Platform / Strategy: Firstrade / unverified",
+            f"Problem: {problem}", "Impact: order submission unknown; pending orders unknown",
+            f"Evidence: {producer}; {link}", "Action: review this run and its inputs; retain execution protection until recovery",
+            "Event: safe reference unavailable | Delivery status follows the receipt"]
+    return redact_sensitive_text("\n".join(lines))
+
+
+def workflow_run_url(env=None):
+    """Return the real monitor workflow reference, never a synthesized business run."""
+    env = os.environ if env is None else env
+    return (f"{env.get('GITHUB_SERVER_URL', '')}/{env.get('GITHUB_REPOSITORY', '')}"
+            f"/actions/runs/{env.get('GITHUB_RUN_ID', '')}")

scripts/cloud_run_runtime_guard.py

@@ -72,7 +72,7 @@ def _env_bool(name: str, default: bool = False) -> bool:
 
 
 def _notification_locale() -> str:
-    return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG"))
+    return resolve_operational_notification_locale(os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG"))
 
 
 def _notice(key: str, /, **values: object) -> str:
@@ -792,16 +792,9 @@ def main() -> int:
             f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}"
             f"/actions/runs/{os.environ['GITHUB_RUN_ID']}"
         )
-    message = format_operational_alert(
-        locale=_notification_locale(),
-        alert_type="runtime_guard",
-        name=name,
-        context={"project": project, "lookback_minutes": lookback_minutes},
-        issues=issues,
-        # Keep provider audit payloads in their source logs, including on pinned QPK.
-        technical_details=(),
-        workflow_url=run_url,
-    )
+    from notifications.unified import alert_text
+    message = alert_text("runtime_guard", _notification_locale(), occurred_at=dt.datetime.now(dt.timezone.utc),
+                         source="runtime_guard", run_url=run_url)
     print(message)
     decision = alert_dedupe.decide(name=name, issues=issues, event_ids=alert_event_ids)
     print(decision.log_line())

scripts/execution_report_heartbeat.py

@@ -55,7 +55,7 @@ def _env_bool(name: str, default: bool = False) -> bool:
 
 
 def _notification_locale() -> str:
-    return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG"))
+    return resolve_operational_notification_locale(os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG"))
 
 
 def _notice(key: str, /, **values: object) -> str:
@@ -760,52 +760,45 @@ def _send_telegram(message: str) -> bool:
 
 
 def _deliver_alert(message: str, *, kind: str, now: dt.datetime, affected_keys=None) -> bool:
-    from notifications.unified import alert_text, enabled, notify_heartbeat
+    from notifications.unified import alert_text, enabled, notify_heartbeat, workflow_run_url
+    display_kind = "receipt_identity_conflict" if "conflicting report identity" in message else kind
+    message = alert_text(display_kind, _notification_locale(), occurred_at=now,
+                         source="heartbeat", run_url=workflow_run_url())
     if not enabled():
         return _send_telegram(message)
-    result = notify_heartbeat(kind=kind, send=lambda: _send_telegram(alert_text(kind, _notification_locale())),
+    result = notify_heartbeat(kind=kind, send=lambda: _send_telegram(message),
                               now=now, affected_keys=affected_keys)
     return any(item.get("transport_acknowledged") for item in result.get("deliveries", []))
 
 
 def _send_normal_heartbeat(name: str, detail: str, *, quiet_eligible: bool = True,
                            now: dt.datetime | None = None, recovery_reports=(), affected_keys=None) -> int:
-    from notifications.unified import alert_text, coverage_locale, enabled, notify_heartbeat
-    if quiet_eligible or enabled():
-        locale = coverage_locale() if quiet_eligible else _notification_locale()
-        message = alert_text("candidate_write_failure" if quiet_eligible else "receipt_unacceptable", locale)
-        def send_attention():
-            marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH")
-            if quiet_eligible and marker:
-                try:
-                    Path(marker).write_text("attempted\n")
-                except OSError:
-                    pass
-            acknowledged = _send_telegram(message[:3900])
-            if quiet_eligible and acknowledged is True and marker:
-                try:
-                    Path(marker).write_text("acknowledged\n")
-                except OSError:
-                    pass
-            return acknowledged
-        result = notify_heartbeat(kind="healthy_heartbeat" if quiet_eligible else "receipt_unacceptable",
-            send=send_attention, now=now or dt.datetime.now(dt.timezone.utc),
-            recovery_reports=recovery_reports, affected_keys=affected_keys)
-        return 1 if result.get("failed") or not result.get("candidate_persisted") else 0
-    if quiet_eligible and not _env_bool("RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", False):
-        return 0
-    if quiet_eligible:
-        message = format_operational_heartbeat_status(
-            locale=_notification_locale(), name=name, detail=detail,
-        )
-    else:
-        message = format_operational_alert(
-            locale=_notification_locale(), alert_type="execution_report_heartbeat", name=name,
-            issues=["运行检查证据待核对" if _notification_locale() == "zh" else "Runtime check evidence needs review"],
-            technical_details=[detail],
-        )
-    print(message)
-    return 0 if _send_telegram(message[:3900]) else 1
+    from notifications.unified import alert_text, coverage_locale, enabled, notify_heartbeat, workflow_run_url
+    locale = coverage_locale() if quiet_eligible else _notification_locale()
+    kind = "candidate_write_failure" if quiet_eligible else (
+        "receipt_stale" if "predates latest due" in detail else "receipt_unacceptable")
+    message = alert_text(kind, locale,
+                         occurred_at=now, source="heartbeat", run_url=workflow_run_url())
+    if not quiet_eligible and not enabled():
+        return 0 if _send_telegram(message) else 1
+    def send_attention():
+        marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH")
+        if quiet_eligible and marker:
+            try:
+                Path(marker).write_text("attempted\n")
+            except OSError:
+                pass
+        acknowledged = _send_telegram(message[:3900])
+        if quiet_eligible and acknowledged is True and marker:
+            try:
+                Path(marker).write_text("acknowledged\n")
+            except OSError:
+                pass
+        return acknowledged
+    result = notify_heartbeat(kind="healthy_heartbeat" if quiet_eligible else "receipt_unacceptable",
+        send=send_attention, now=now or dt.datetime.now(dt.timezone.utc),
+        recovery_reports=recovery_reports, affected_keys=affected_keys)
+    return 1 if result.get("failed") or not result.get("candidate_persisted") else 0
 
 
 def main(now: dt.datetime | None = None) -> int: