fix: explain sanitized runtime configuration failures

main.py

@@ -258,14 +258,34 @@ def t(key, **kwargs):
     return build_translator(NOTIFY_LANG)(key, **kwargs)
 
 
+def _runtime_failure_reason(exc: Exception) -> str:
+    # Match only fixed platform errors; never publish arbitrary provider text.
+    if isinstance(exc, RuntimeError) and str(exc) in {
+        "durable execution command producer is paper-only and cannot be enabled live",
+        "durable execution command consumer is paper-only and cannot be enabled live",
+        "strategy risk state persistence is paper-only and cannot be enabled live",
+    }:
+        return "paper_live_conflict"
+    return "unknown"
+
+
 def _compact_error_notification(
     exc: Exception,
     *,
     title: str | None = None,
     prefix: str = "",
+    phase: str = "strategy_cycle",
 ) -> str:
     error_text = type(exc).__name__
-    message = f"{title or t('error_title')}\n{prefix}{error_text}"
+    known_phases = {"market_hours", "runtime_configuration", "strategy_cycle", "reporting", "health_probe"}
+    safe_phase = phase if phase in known_phases else "strategy_cycle"
+    reason = _runtime_failure_reason(exc)
+    message = "\n".join((
+        title or t("error_title"),
+        t("failure_phase", phase=t(f"failure_phase_{safe_phase}")),
+        t(f"failure_reason_{reason}"),
+        f"{prefix}{error_text}",
+    ))
     if len(message) <= COMPACT_ERROR_NOTIFICATION_MAX_CHARS:
         return message
     return message[: COMPACT_ERROR_NOTIFICATION_MAX_CHARS - 1].rstrip() + "…"
@@ -662,6 +682,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
         signals=strategy_plugin_signals,
         error=strategy_plugin_error,
     )
+    failure_phase = "market_hours"
     try:
         reporting_adapters.log_event(
             log_context,
@@ -747,6 +768,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             return True
         if not validation_only:
             publish_strategy_plugin_alerts(strategy_plugin_signals, report=report)
+        failure_phase = "runtime_configuration"
         try:
             rebalance_runtime = composer.build_rebalance_runtime(
                 silent_cycle_notifications=validation_only,
@@ -758,19 +780,19 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             rebalance_runtime = composer.build_rebalance_runtime(
                 silent_cycle_notifications=validation_only,
             )
-        cycle_result = run_rebalance_cycle(
-            runtime=rebalance_runtime,
-            config=composer.build_rebalance_config(
-                strategy_plugin_signals=strategy_plugin_signals,
-                strategy_plugin_error=strategy_plugin_error,
-                notification_title_key=(
-                    "dry_run_title"
-                    if validation_only and validation_label == "dry_run"
-                    else ""
-                ),
-                cash_only_execution=CASH_ONLY_EXECUTION,
+        rebalance_config = composer.build_rebalance_config(
+            strategy_plugin_signals=strategy_plugin_signals,
+            strategy_plugin_error=strategy_plugin_error,
+            notification_title_key=(
+                "dry_run_title"
+                if validation_only and validation_label == "dry_run"
+                else ""
             ),
+            cash_only_execution=CASH_ONLY_EXECUTION,
         )
+        failure_phase = "strategy_cycle"
+        cycle_result = run_rebalance_cycle(runtime=rebalance_runtime, config=rebalance_config)
+        failure_phase = "reporting"
         signal_snapshot = {}
         if cycle_result is not None:
             execution = dict(getattr(cycle_result, "execution", {}) or {})
@@ -822,6 +844,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
         return True
 
     except Exception as exc:
+        failure_reason = _runtime_failure_reason(exc)
         append_runtime_report_error(
             report,
             stage="strategy_cycle",
@@ -835,11 +858,13 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             severity="ERROR",
             error_type=type(exc).__name__,
             error_message="strategy_cycle_failed",
+            failure_phase=failure_phase,
+            failure_reason=failure_reason,
         )
         try:
             notification_adapters.publish_cycle_notification(
-                detailed_text=f"Strategy error: {type(exc).__name__}",
-                compact_text=_compact_error_notification(exc),
+                detailed_text=_compact_error_notification(exc, phase=failure_phase),
+                compact_text=_compact_error_notification(exc, phase=failure_phase),
             )
         except Exception as notification_exc:
             notification_delivery_events.append(
@@ -869,6 +894,7 @@ def run_strategy(*, force_run: bool = False, validation_only: bool = False, vali
             report,
             status="error",
             summary=error_summary or None,
+            diagnostics={"failure_phase": failure_phase, "failure_reason": failure_reason},
         )
         return False
     finally:
@@ -949,7 +975,7 @@ def run_probe(*, response_body: str = "Probe OK"):
                 error_message="health_probe_failed",
             )
         err = _compact_error_notification(
-            exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"),
+            exc, title=t("health_probe_title"), prefix=t("health_probe_error_prefix"), phase="health_probe",
         )
         if composer is not None:
             composer.build_notification_adapters().publish_cycle_notification(
@@ -958,6 +984,7 @@ def run_probe(*, response_body: str = "Probe OK"):
                     exc,
                     title=t("health_probe_title"),
                     prefix=t("health_probe_error_prefix"),
+                    phase="health_probe",
                 ),
             )
         else:

notifications/telegram.py

@@ -97,6 +97,14 @@ def _break_telegram_market_symbol_auto_links(value) -> str:
         "small_account_warning_reason_integer_shares_min_position_value_may_prevent_backtest_replication": "整数股和最小仓位限制可能导致实盘无法完全复现回测",
         "order_id_suffix": "（订单号: {order_id}）",
         "error_title": "🚨 【策略异常】",
+        "failure_phase": "失败环节：{phase}",
+        "failure_phase_market_hours": "交易时段检查",
+        "failure_phase_runtime_configuration": "运行配置检查",
+        "failure_phase_strategy_cycle": "策略执行",
+        "failure_phase_reporting": "执行报告整理",
+        "failure_phase_health_probe": "账户连接检查",
+        "failure_reason_paper_live_conflict": "模拟专用功能与正式运行模式冲突；请关闭误启用的模拟专用开关。",
+        "failure_reason_unknown": "本次运行未完成；请按失败环节检查最新执行报告。",
         "buy_skipped": "ℹ️ [买入跳过] {detail}",
         "sell_skipped": "ℹ️ [卖出跳过] {detail}",
         "sell_skip_whole_share_detail": "{symbol} 需减 ${diff}，整数股不足 1 股，无需下单",
@@ -296,6 +304,14 @@ def _break_telegram_market_symbol_auto_links(value) -> str:
         "small_account_warning_reason_integer_shares_min_position_value_may_prevent_backtest_replication": "integer-share minimum position sizing may prevent backtest replication",
         "order_id_suffix": "[order_id={order_id}]",
         "error_title": "🚨 【Strategy Error】",
+        "failure_phase": "Failed at: {phase}",
+        "failure_phase_market_hours": "market-hours check",
+        "failure_phase_runtime_configuration": "runtime configuration",
+        "failure_phase_strategy_cycle": "strategy execution",
+        "failure_phase_reporting": "execution reporting",
+        "failure_phase_health_probe": "account connectivity check",
+        "failure_reason_paper_live_conflict": "A paper-only feature conflicts with live mode; disable the unintended paper-only switch.",
+        "failure_reason_unknown": "The run did not finish; check the latest report for this stage.",
         "buy_skipped": "ℹ️ [Buy skipped] {detail}",
         "sell_skipped": "ℹ️ [Sell skipped] {detail}",
         "sell_skip_whole_share_detail": "{symbol} needs ${diff} trimmed; whole-share quantity rounds to 0; no order needed",

tests/test_request_handling.py

@@ -1249,6 +1249,52 @@ def test_notification_delivery_summary_keeps_failed_transport_receipt(self):
 class SanitizedRuntimeErrorTests(unittest.TestCase):
     marker = "SYNTHETIC_SECRET_DO_NOT_EMIT"
 
+    def test_paper_only_live_conflict_reports_configuration_phase_without_running_cycle(self):
+        from application.durable_execution_commands import resolve_paper_execution_command_producer_enabled
+
+        for language in ("zh-CN", "en"):
+            with self.subTest(language=language):
+                module = load_module(notify_lang=language)
+                report, events, notifications = {"status": "pending"}, [], []
+                reporting = types.SimpleNamespace(
+                    start_run=lambda: (types.SimpleNamespace(run_id="synthetic-run"), report),
+                    log_event=lambda *args, **kwargs: events.append(kwargs),
+                    persist_execution_report=lambda payload: "/tmp/synthetic-report.json",
+                )
+                composer = types.SimpleNamespace(
+                    build_reporting_adapters=lambda: reporting,
+                    build_notification_adapters=lambda **kwargs: types.SimpleNamespace(
+                        publish_cycle_notification=lambda **payload: notifications.append(payload)),
+                    load_strategy_plugin_signals=lambda *args: ((), None),
+                    attach_strategy_plugin_report=lambda *args, **kwargs: None,
+                    with_prefix=lambda text: text,
+                    build_rebalance_runtime=lambda **kwargs: types.SimpleNamespace(),
+                    build_rebalance_config=lambda **kwargs: resolve_paper_execution_command_producer_enabled(
+                        env_reader=lambda name, default="": "true", dry_run_only=False),
+                )
+                module.build_composer = lambda **kwargs: composer
+                module.is_market_open_now = lambda **kwargs: True
+                module.dca_execution_unsupported_reason = lambda *args: None
+                with patch.object(module, "run_rebalance_cycle") as cycle:
+                    self.assertFalse(module.run_strategy())
+                cycle.assert_not_called()
+                self.assertEqual(report["diagnostics"]["failure_phase"], "runtime_configuration")
+                self.assertEqual(report["diagnostics"]["failure_reason"], "paper_live_conflict")
+                failure = next(event for event in events if event.get("failure_reason"))
+                self.assertEqual(failure["failure_reason"], "paper_live_conflict")
+                text = notifications[0]["compact_text"]
+                self.assertIn("模拟专用" if language.startswith("zh") else "paper-only", text)
+                self.assertIn("配置检查" if language.startswith("zh") else "configuration", text)
+                self.assertNotIn("Traceback", text)
+
+    def test_unknown_failure_text_is_not_used_as_a_diagnostic_reason(self):
+        module = load_module(notify_lang="zh-CN")
+        exc = RuntimeError(self.marker)
+        self.assertEqual(module._runtime_failure_reason(exc), "unknown")
+        text = module._compact_error_notification(exc, phase="strategy_cycle")
+        self.assertIn("策略执行", text)
+        self._assert_sanitized(text)
+
     def _raise_sensitive(self, *args, **kwargs):
         raise RuntimeError(self.marker)