fix: localize concise runtime failure notifications

main.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from notifications.telegram import build_translator
+
 import os
 import re
 import traceback
@@ -139,40 +141,24 @@ def _telegram_notification_targets() -> tuple[tuple[str, str], ...]:
 
 
 def _runtime_error_notification_message(exc: Exception) -> str:
-    error_text = _safe_exception_text(exc, include_type=True)
-    if len(error_text) > 1200:
-        error_text = error_text[:1197] + "..."
-    is_health_check = request.path == "/probe"
-    if str(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG") or "").strip().lower().startswith("zh"):
-        return "\n".join(
-            (
-                "Firstrade 健康检查失败" if is_health_check else "Firstrade 策略运行失败",
-                f"服务: {os.getenv('K_SERVICE') or 'firstrade-quant-service'}",
-                f"版本: {os.getenv('K_REVISION') or '<unknown>'}",
-                f"路由: {request.method} {request.path}",
-                f"策略: {os.getenv('STRATEGY_PROFILE') or '<unset>'}",
-                f"账户范围: {os.getenv('ACCOUNT_REGION') or '<unset>'}",
-                f"错误: {error_text}",
-            )
-        )
-    return "\n".join(
-        (
-            "Firstrade health check failed" if is_health_check else "Firstrade strategy run failed",
-            f"service: {os.getenv('K_SERVICE') or 'firstrade-quant-service'}",
-            f"revision: {os.getenv('K_REVISION') or '<unknown>'}",
-            f"route: {request.method} {request.path}",
-            f"strategy: {os.getenv('STRATEGY_PROFILE') or '<unset>'}",
-            f"account_scope: {os.getenv('ACCOUNT_REGION') or '<unset>'}",
-            f"error: {error_text}",
-        )
-    )
+    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"),
+    ))
 
 
 def _notify_runtime_error(exc: Exception) -> bool:
+    t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG"))
     targets = _telegram_notification_targets()
     if not targets:
         print(
-            "Firstrade runtime error notification skipped: no Telegram target configured.",
+            t("runtime_notification_missing_target"),
             flush=True,
         )
         return False
@@ -182,10 +168,11 @@ def _notify_runtime_error(exc: Exception) -> bool:
     for token, chat_id in targets:
         attempted = True
         try:
-            build_sender(token, chat_id)(message)
-        except Exception as send_exc:  # pragma: no cover - build_sender normally handles this.
+            if build_sender(token, chat_id)(message) is False:
+                print(t("runtime_notification_delivery_failed"), flush=True)
+        except Exception:  # pragma: no cover - build_sender normally handles this.
             print(
-                f"Firstrade runtime error Telegram send failed: {redact_sensitive_text(send_exc)}",
+                t("runtime_notification_delivery_failed"),
                 flush=True,
             )
     return attempted
@@ -197,9 +184,8 @@ def _safe_exception_text(exc: Exception, *, include_type: bool = False) -> str:
 
 
 def _handle_strategy_run_exception(exc: Exception) -> bool:
-    print(f"Firstrade strategy run failed: {_safe_exception_text(exc, include_type=True)}", flush=True)
-    for line in traceback.format_exception(type(exc), exc, exc.__traceback__):
-        print(redact_sensitive_text(line.rstrip()), flush=True)
+    t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG"))
+    print(t("runtime_failure_log"), flush=True)
     return _notify_runtime_error(exc)
 
 

notifications/telegram.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale
+
 import re
 from collections.abc import Callable, Mapping
 from typing import Any
@@ -146,6 +148,14 @@ def format_small_account_whole_share_bootstrap_notes(
 
 I18N = {
     "zh": {
+        "runtime_failure_title": "⚠️ Firstrade 策略运行失败",
+        "runtime_probe_failure_title": "⚠️ Firstrade 健康检查失败",
+        "runtime_failure_context": "运行目标：{context}",
+        "runtime_failure_result": "本次运行未正常结束，请查看最新执行报告。",
+        "runtime_failure_action": "下一步：检查账户会话及最新执行报告。",
+        "runtime_notification_missing_target": "异常通知未发送：未配置 Telegram 接收目标（notification_target_missing）",
+        "runtime_notification_delivery_failed": "异常通知发送失败（notification_delivery_failed）",
+        "runtime_failure_log": "策略运行失败（runtime_setup_failed）",
         "rebalance_title": "🔔 【调仓指令】",
         "heartbeat_title": "💓 【心跳检测】",
         "strategy_label": "🧭 策略: {name}",
@@ -319,6 +329,14 @@ def format_small_account_whole_share_bootstrap_notes(
         "skip_symbols_reason": "{symbols}（{reason}）",
     },
     "en": {
+        "runtime_failure_title": "⚠️ Firstrade strategy run failed",
+        "runtime_probe_failure_title": "⚠️ Firstrade health check failed",
+        "runtime_failure_context": "Target: {context}",
+        "runtime_failure_result": "The run did not finish successfully; check the latest execution report.",
+        "runtime_failure_action": "Next: Check the account session and the latest execution report.",
+        "runtime_notification_missing_target": "Runtime alert not sent: Telegram target is not configured (notification_target_missing)",
+        "runtime_notification_delivery_failed": "Runtime alert delivery failed (notification_delivery_failed)",
+        "runtime_failure_log": "Strategy run failed (runtime_setup_failed)",
         "rebalance_title": "🔔 【Rebalance Instruction】",
         "heartbeat_title": "💓 【Heartbeat】",
         "strategy_label": "🧭 Strategy: {name}",
@@ -504,8 +522,7 @@ def format_small_account_whole_share_bootstrap_notes(
 
 
 def build_translator(lang: str | None) -> Callable[..., str]:
-    normalized = str(lang or "").lower()
-    active_lang = "zh" if normalized.startswith("zh") else "en"
+    active_lang = resolve_operational_notification_locale(lang)
 
     def translate(key: str, **kwargs) -> str:
         template = I18N[active_lang].get(key, I18N["en"].get(key, key))

tests/test_request_handling.py

@@ -455,7 +455,7 @@ def send(message):
     assert sent_messages[0][0] == "token-1"
     assert sent_messages[0][1] == "chat-1"
     assert "Firstrade health check failed" in sent_messages[0][2]
-    assert "RuntimeError: session denied" in sent_messages[0][2]
+    assert "RuntimeError: session denied" not in sent_messages[0][2]
 
 
 def test_run_endpoint_notifies_telegram_on_strategy_cycle_error(monkeypatch):
@@ -493,8 +493,8 @@ def send(message):
     assert sent_messages[0][0] == "token-1"
     assert sent_messages[0][1] == "chat-1"
     assert "Firstrade strategy run failed" in sent_messages[0][2]
-    assert "ValueError: snapshot denied" in sent_messages[0][2]
-    assert "strategy: russell_top50_leader_rotation" in sent_messages[0][2]
+    assert "ValueError: snapshot denied" not in sent_messages[0][2]
+    assert "Strategy: russell_top50_leader_rotation" in sent_messages[0][2]
 
 
 def test_run_endpoint_error_notification_uses_chinese_copy(monkeypatch):
@@ -526,7 +526,7 @@ def send(message):
     text = sent_messages[0][2]
     assert "Firstrade 策略运行失败" in text
     assert "策略: russell_top50_leader_rotation" in text
-    assert "错误: ValueError: snapshot denied" in text
+    assert "snapshot denied" not in text
 
 
 def test_run_endpoint_redacts_sensitive_error_text(monkeypatch):
@@ -559,7 +559,7 @@ def send(message):
     assert response.status_code == 500
     payload = response.get_json()
     assert "<redacted>" in payload["error"]
-    assert "<redacted>" in sent_messages[0][2]
+    assert "request failed" not in sent_messages[0][2]
     for raw_secret in ("supersecret123", "abcd1234efgh", "123456789:ABC", "key987654"):
         assert raw_secret not in payload["error"]
         assert raw_secret not in sent_messages[0][2]

tests/test_runtime_notification_i18n.py

@@ -0,0 +1,53 @@
+import ast
+import os
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from notifications.telegram import build_translator
+
+
+@pytest.mark.parametrize("locale", ["zh-CN", "zh_TW", "ZH-hans", " zh "])
+def test_chinese_locale_variants_use_chinese(locale):
+    assert build_translator(locale)("strategy_label", name="test") == build_translator("zh")("strategy_label", name="test")
+
+
+def runtime_function(name, **overrides):
+    tree = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text())
+    function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name)
+    namespace = dict(os=os, request=SimpleNamespace(method="GET", path="/main"),
+                     NOTIFY_LANG="zh-CN", SERVICE_NAME="Test service", SECRET_NAME="Test secret alias",
+                     STRATEGY_PROFILE="test_strategy", ACCOUNT_REGION="SG", ACCOUNT_GROUP="Test account",
+                     strategy_display_name="Test strategy", build_translator=build_translator,
+                     t=build_translator("zh-CN"))
+    namespace["_safe_exception_text"] = lambda exc, **kwargs: str(exc)
+    namespace.update(overrides)
+    exec(compile(ast.Module(body=[function], type_ignores=[]), "main.py", "exec"), namespace)
+    return namespace[name]
+
+
+@pytest.mark.parametrize("locale", ["zh-CN", "en"])
+def test_startup_alert_is_compact_localized_and_does_not_embed_provider_text(locale, monkeypatch):
+    monkeypatch.setenv("NOTIFY_LANG", locale)
+    monkeypatch.delenv("QSL_NOTIFY_LANG", raising=False)
+    monkeypatch.setenv("STRATEGY_PROFILE", "test_strategy")
+    fn = runtime_function("_runtime_error_notification_message", NOTIFY_LANG=locale, t=build_translator(locale))
+    message = fn(RuntimeError("PRIVATE_PROVIDER_SENTINEL"))
+    assert len(message.splitlines()) <= 5
+    assert "PRIVATE_PROVIDER_SENTINEL" not in message
+    assert "未提交订单" not in message
+    assert "no order" not in message.lower()
+    assert ("未正常结束" in message) is locale.startswith("zh")
+    assert ("did not finish successfully" in message) is (locale == "en")
+    assert "test_strategy" in message or "Test strategy" in message
+
+
+def test_failed_fallback_delivery_logs_failure_and_preserves_attempted_semantics(capsys):
+    fn = runtime_function(
+        "_notify_runtime_error", _telegram_notification_targets=lambda: (("test", "test"),),
+        _runtime_error_notification_message=lambda exc: "safe failure",
+        build_sender=lambda token, chat: lambda message: False,
+    )
+    assert fn(RuntimeError("private")) is True
+    assert "notification_delivery_failed" in capsys.readouterr().out