Notify on Firstrade session check failures (#97)

Author: Pigbibi

main.py

@@ -69,10 +69,11 @@ def _runtime_error_notification_message(exc: Exception) -> str:
     error_text = f"{type(exc).__name__}: {exc}"
     if len(error_text) > 1200:
         error_text = error_text[:1197] + "..."
+    is_health_check = request.path in {"/session-check", "/probe"}
     if str(os.getenv("NOTIFY_LANG") or "").strip().lower().startswith("zh"):
         return "\n".join(
             (
-                "Firstrade 策略运行失败",
+                "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}",
@@ -83,7 +84,7 @@ def _runtime_error_notification_message(exc: Exception) -> str:
         )
     return "\n".join(
         (
-            "Firstrade strategy run failed",
+            "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}",
@@ -366,9 +367,29 @@ def session_check():
     try:
         return jsonify(run_session_check())
     except FirstradePlatformError as exc:
-        return jsonify({"ok": False, "error": str(exc)}), 500
+        notification_attempted = _notify_runtime_error(exc)
+        return (
+            jsonify(
+                {
+                    "ok": False,
+                    "error": str(exc),
+                    "runtime_error_notification_attempted": notification_attempted,
+                }
+            ),
+            500,
+        )
     except Exception as exc:
-        return jsonify({"ok": False, "error": f"{type(exc).__name__}: {exc}"}), 500
+        notification_attempted = _notify_runtime_error(exc)
+        return (
+            jsonify(
+                {
+                    "ok": False,
+                    "error": f"{type(exc).__name__}: {exc}",
+                    "runtime_error_notification_attempted": notification_attempted,
+                }
+            ),
+            500,
+        )
 
 
 @app.post("/")

tests/test_request_handling.py

@@ -79,11 +79,13 @@ def test_session_check_endpoint_is_disabled_without_explicit_http_gate(monkeypat
 
 def test_session_check_endpoint_calls_service_when_gate_enabled(monkeypatch):
     monkeypatch.setenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", "true")
+    sent_messages = []
     monkeypatch.setattr(
         main,
         "run_session_check",
         lambda: {"ok": True, "session_reused": True, "snapshot_persisted": True},
     )
+    monkeypatch.setattr(main, "build_sender", lambda *_args, **_kwargs: sent_messages.append)
     client = main.app.test_client()
 
     response = client.post("/session-check")
@@ -94,6 +96,40 @@ def test_session_check_endpoint_calls_service_when_gate_enabled(monkeypatch):
         "session_reused": True,
         "snapshot_persisted": True,
     }
+    assert sent_messages == []
+
+
+def test_session_check_endpoint_notifies_only_on_error(monkeypatch):
+    monkeypatch.setenv("FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP", "true")
+    monkeypatch.setenv("TELEGRAM_TOKEN", "token-1")
+    monkeypatch.setenv("GLOBAL_TELEGRAM_CHAT_ID", "chat-1")
+    monkeypatch.setattr(
+        main,
+        "run_session_check",
+        lambda: (_ for _ in ()).throw(RuntimeError("session denied")),
+    )
+    sent_messages = []
+
+    def fake_build_sender(token, chat_id):
+        def send(message):
+            sent_messages.append((token, chat_id, message))
+
+        return send
+
+    monkeypatch.setattr(main, "build_sender", fake_build_sender)
+    client = main.app.test_client()
+
+    response = client.post("/session-check")
+
+    assert response.status_code == 500
+    payload = response.get_json()
+    assert payload["ok"] is False
+    assert payload["runtime_error_notification_attempted"] is True
+    assert len(sent_messages) == 1
+    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]
 
 
 def test_root_post_calls_strategy_cycle_when_gate_enabled(monkeypatch):