fix: localize concise runtime failure notifications

main.py

@@ -549,36 +549,20 @@ def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]:
 
 
 def _runtime_error_notification_message(exc: Exception, *, route_label: str) -> str:
-    error_text = type(exc).__name__
-    if str(NOTIFY_LANG or "").strip().lower().startswith("zh"):
-        return "\n".join(
-            (
-                "LongBridge 策略运行失败",
-                f"服务: {os.getenv('K_SERVICE') or SECRET_NAME or 'longbridge-platform'}",
-                f"版本: {os.getenv('K_REVISION') or '<unknown>'}",
-                f"路由: {route_label}",
-                f"策略: {STRATEGY_PROFILE}",
-                f"账户范围: {ACCOUNT_REGION}",
-                f"错误: {error_text}",
-            )
-        )
-    return "\n".join(
-        (
-            "LongBridge strategy run failed",
-            f"service: {os.getenv('K_SERVICE') or SECRET_NAME or 'longbridge-platform'}",
-            f"revision: {os.getenv('K_REVISION') or '<unknown>'}",
-            f"route: {route_label}",
-            f"strategy: {STRATEGY_PROFILE}",
-            f"account_scope: {ACCOUNT_REGION}",
-            f"error: {error_text}",
-        )
-    )
+    title = "runtime_failure_title"
+    return "\n".join((
+        t(title),
+        t("strategy_label", name=strategy_display_name or STRATEGY_PROFILE),
+        t("runtime_failure_context", context=ACCOUNT_REGION),
+        t("runtime_failure_result"),
+        t("runtime_failure_action"),
+    ))
 
 
 def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
     targets = _runtime_error_notification_targets()
     if not targets:
-        print("LongBridge runtime error notification skipped: no Telegram target configured.", flush=True)
+        print(t("runtime_notification_missing_target"), flush=True)
         return False
     message = _runtime_error_notification_message(exc, route_label=route_label)
     outcomes = []
@@ -596,17 +580,17 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool:
             if isinstance(payload, dict) and payload.get("ok") is False:
                 acknowledged = False
             outcomes.append(acknowledged)
-        except Exception as send_exc:
+        except Exception:
             print(
-                f"LongBridge runtime error Telegram send failed: {type(send_exc).__name__}",
+                t("runtime_notification_delivery_failed"),
                 flush=True,
             )
             outcomes.append(False)
     return bool(outcomes) and all(outcomes)
 
 
 def _handle_route_runtime_error(exc: Exception, *, route_label: str):
-    print(f"LongBridge route failed before strategy-cycle handling: {type(exc).__name__}", flush=True)
+    print(t("runtime_failure_log"), flush=True)
     _notify_runtime_error(exc, route_label=route_label)
     return "Error", 500
 

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 Mapping
 from typing import Any
@@ -41,6 +43,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str:
 
 I18N = {
     "zh": {
+        "runtime_failure_title": "⚠️ LongBridge 策略运行失败",
+        "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": "🔔 【调仓指令】",
         "pending_order_title": "⏳ 【订单待券商最终确认】",
         "dry_run_banner": "🧪 模拟运行模式，本次不会真实下单",
@@ -233,6 +242,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str:
         "strategy_plugin_guidance_taco_rebound_shadow_taco_rebound_notify_manual_review": "TACO 仅提示可能的反弹窗口；可考虑小仓位、分批、预设止损/失效条件的人工博弈，不建议一次性满仓。",
     },
     "en": {
+        "runtime_failure_title": "⚠️ LongBridge strategy run 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 account authorization, connectivity 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": "🔔 【Trade Execution Report】",
         "pending_order_title": "⏳ 【Broker Order Pending Confirmation】",
         "dry_run_banner": "🧪 Dry run mode, no real orders will be submitted",
@@ -437,8 +453,8 @@ def _break_telegram_market_symbol_auto_links(value) -> str:
 
 
 def build_translator(lang):
+    active_lang = resolve_operational_notification_locale(lang)
     def translate(key, **kwargs):
-        active_lang = lang if lang in I18N else "en"
         template = I18N[active_lang].get(key, key)
         return template.format(**kwargs) if kwargs else template
 

tests/test_request_handling.py

@@ -361,7 +361,7 @@ def fake_post(_url, *, json, timeout):
         self.assertEqual(len(observed["payloads"]), 1)
         self.assertEqual(observed["payloads"][0][0]["chat_id"], "chat-1")
         self.assertIn("LongBridge strategy run failed", observed["payloads"][0][0]["text"])
-        self.assertIn("RuntimeError", observed["payloads"][0][0]["text"])
+        self.assertIn("did not finish successfully", observed["payloads"][0][0]["text"])
         self.assertNotIn("boom", observed["payloads"][0][0]["text"])
 
     def test_handle_trigger_runtime_error_fallback_uses_chinese_copy(self):
@@ -387,8 +387,8 @@ def fake_post(_url, *, json, timeout):
         self.assertEqual(body, "Error")
         text = observed["payloads"][0][0]["text"]
         self.assertIn("LongBridge 策略运行失败", text)
-        self.assertIn("服务:", text)
-        self.assertIn("错误: RuntimeError", text)
+        self.assertIn("运行目标：", text)
+        self.assertIn("未正常结束", text)
         self.assertNotIn("boom", text)
 
     def test_handle_trigger_rejects_get_without_running_strategy(self):
@@ -1255,7 +1255,7 @@ def _raise_sensitive(self, *args, **kwargs):
     def _assert_sanitized(self, output):
         self.assertNotIn(self.marker, output)
         self.assertNotIn("Traceback", output)
-        self.assertIn("RuntimeError", output)
+        self.assertTrue("RuntimeError" in output or "runtime_setup_failed" in output)
 
     def test_compact_and_route_failure_omit_exception_body(self):
         import io

tests/test_runtime_notification_i18n.py

@@ -0,0 +1,42 @@
+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.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"), route_label="/main")
+    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