fix: expose Router upstream errors

gfa_core/mas/agents/router_runtime.py

@@ -1,5 +1,7 @@
+import asyncio
 import json
 import hashlib
+import logging
 import os
 import re
 from collections import Counter
@@ -18,6 +20,63 @@ from gfa_core.mas.workflow_catalog import (
 from gfa_core.mas.workflow_models import GoalLedger
 
 
+logger = logging.getLogger(__name__)
+
+
+_ROUTER_ERROR_TEXT_LIMIT = 1000
+_ROUTER_CREDENTIAL_PATTERN = re.compile(
+    r"(?i)\b(api[_-]?key|authorization|(?:access|refresh)[_-]?token|token|secret)"
+    r"\b\s*[:=]\s*[^,;}\]\r\n]+"
+)
+
+
+def _sanitize_router_error_text(value: Any) -> str:
+    text = str(value or "").strip()
+    if not text:
+        return ""
+    return _ROUTER_CREDENTIAL_PATTERN.sub(r"\1=<REDACTED>", text)[
+        :_ROUTER_ERROR_TEXT_LIMIT
+    ]
+
+
+class RouterModelCallError(RuntimeError):
+    """Allow-listed details from a failed Router model request."""
+
+    def __init__(self, cause: Exception, *, attempts: int, model: str) -> None:
+        self.status_code = getattr(cause, "status_code", None)
+        self.attempts = attempts
+        self.model = _sanitize_router_error_text(model) or "unknown"
+        self.request_id = _sanitize_router_error_text(
+            getattr(cause, "request_id", "")
+        )
+
+        body = getattr(cause, "body", None)
+        error: Mapping[str, Any] = {}
+        if isinstance(body, Mapping):
+            nested_error = body.get("error")
+            error = nested_error if isinstance(nested_error, Mapping) else body
+        self.error_code = _sanitize_router_error_text(error.get("code"))
+        self.upstream_message = _sanitize_router_error_text(error.get("message"))
+        if not self.upstream_message:
+            self.upstream_message = _sanitize_router_error_text(str(cause))
+        if not self.upstream_message:
+            self.upstream_message = type(cause).__name__
+
+        super().__init__(self.user_message())
+
+    def user_message(self) -> str:
+        details = []
+        if self.status_code is not None:
+            details.append(f"HTTP {self.status_code}")
+        details.append(f"模型 {self.model}")
+        details.append(f"已尝试 {self.attempts} 次")
+        if self.error_code:
+            details.append(f"错误码 {self.error_code}")
+        if self.request_id:
+            details.append(f"Request ID {self.request_id}")
+        return f"Router 调用 DeepSeek 失败（{'，'.join(details)}）：{self.upstream_message}"
+
+
 InteractionKind = Literal[
     "supplement",
     "correction",
@@ -400,6 +459,7 @@ class RouterRuntime:
     ) -> None:
         self.llm_manager = llm_manager
         self.tools = tools
+        self._last_model_call_error: RouterModelCallError | None = None
         if max_iterations is None:
             raw_budget = os.environ.get("PLANTAI_MAIN_AGENT_MAX_TOOL_CALLS", "10")
             try:
@@ -416,6 +476,54 @@ class RouterRuntime:
         else:
             self.max_iterations = min(max(0, max_iterations), 5)
 
+    async def _call_router_model(
+        self,
+        messages: list[LLMMessage],
+        **kwargs: Any,
+    ) -> Any:
+        """Call the Router model with bounded retries for transient failures."""
+        max_attempts = 3
+        # This OpenAI-compatible upstream can emit a transient 402 from its
+        # billing gateway even when the same credential succeeds immediately
+        # afterwards. Keep retries bounded so a real balance problem still
+        # fails quickly and remains visible in the logs.
+        retryable_statuses = {402, 408, 409, 429, 500, 502, 503, 504}
+        for attempt in range(1, max_attempts + 1):
+            try:
+                return await self.llm_manager.acall(
+                    messages,
+                    provider="deepseek",
+                    allow_fallback=False,
+                    **kwargs,
+                )
+            except Exception as exc:
+                status = getattr(exc, "status_code", None)
+                retryable = status is None or status in retryable_statuses
+                logger.warning(
+                    "Router model call failed: attempt=%s/%s exception=%s status=%s retryable=%s",
+                    attempt,
+                    max_attempts,
+                    type(exc).__name__,
+                    status,
+                    retryable,
+                )
+                if not retryable or attempt >= max_attempts:
+                    failure = RouterModelCallError(
+                        exc,
+                        attempts=attempt,
+                        model=str(
+                            getattr(
+                                self.llm_manager,
+                                "_default_model",
+                                "deepseek-v4-flash",
+                            )
+                            or "deepseek-v4-flash"
+                        ),
+                    )
+                    logger.error("Router model unavailable: %s", failure.user_message())
+                    raise failure from exc
+                await asyncio.sleep(0.5 * (2 ** (attempt - 1)))
+
     @staticmethod
     def _action_for_route(
         decision: RouteDecision,
@@ -603,14 +711,15 @@ class RouterRuntime:
         calls = 0
         for attempt in range(min(2, self.max_iterations)):
             try:
-                response = await self.llm_manager.acall(
+                response = await self._call_router_model(
                     route_messages,
-                    provider="deepseek",
-                    allow_fallback=False,
                     tools=[],
                     tool_choice="none",
                     temperature=0,
                 )
+            except RouterModelCallError as exc:
+                self._last_model_call_error = exc
+                return None, calls + 1, []
             except Exception:
                 return None, calls + 1, []
             calls += 1
@@ -651,6 +760,7 @@ class RouterRuntime:
         job_id: str,
         goal_ledger: GoalLedger | None = None,
     ) -> RouterOutcome:
+        self._last_model_call_error = None
         ui_action = self._config_dialog_ui_action(query)
         if ui_action is not None:
             return RouterOutcome(
@@ -732,7 +842,11 @@ class RouterRuntime:
             if not decision_trace:
                 return RouterOutcome(
                     status="agent_unavailable",
-                    message="The Router model is currently unavailable.",
+                    message=(
+                        self._last_model_call_error.user_message()
+                        if self._last_model_call_error is not None
+                        else "The Router model is currently unavailable."
+                    ),
                 )
             if decision_calls < 2:
                 return RouterOutcome(
@@ -865,16 +979,20 @@ class RouterRuntime:
             ]
             force_final = not remaining_needs or recognized_intent is not None
             try:
-                response = await self.llm_manager.acall(
+                response = await self._call_router_model(
                     messages,
-                    provider="deepseek",
-                    allow_fallback=False,
                     tools=[] if force_final else available_definitions,
                     tool_choice=(
                         "none" if force_final else "required"
                     ),
                     temperature=0,
                 )
+            except RouterModelCallError as exc:
+                return RouterOutcome(
+                    status="agent_unavailable",
+                    message=exc.user_message(),
+                    tool_trace=tool_trace,
+                )
             except Exception:
                 return RouterOutcome(
                     status="agent_unavailable",

tests/unit/test_router_analysis_handoff_boundary.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+from types import SimpleNamespace
 from typing import Any
 
 import pytest
@@ -99,3 +100,169 @@ def test_new_analysis_decision_discards_main_agent_workflow_and_clarification()
     assert normalized.workflow_id is None
     assert normalized.clarification_question is None
     assert normalized.analysis_request_kind == "execution"
+
+
+def test_route_decision_retries_transient_model_failures(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class TransientPaymentGatewayFailure(RuntimeError):
+        status_code = 402
+
+    class TransientLLM:
+        calls = 0
+
+        async def acall(self, *_: Any, **__: Any) -> Any:
+            self.calls += 1
+            if self.calls < 3:
+                raise TransientPaymentGatewayFailure("temporary billing gateway failure")
+            return SimpleNamespace(
+                content=(
+                    '{"goal":"new_analysis","needs":["analysis_intent"],'
+                    '"subject":"执行数据准备","analysis_request_kind":"execution"}'
+                )
+            )
+
+    async def no_sleep(_: float) -> None:
+        return None
+
+    llm = TransientLLM()
+    runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5)  # type: ignore[arg-type]
+    monkeypatch.setattr(asyncio, "sleep", no_sleep)
+
+    decision, calls, trace = asyncio.run(
+        runtime._decide_route(
+            query="执行数据准备",
+            history=[],
+            session_id="retry-session",
+            job_id="job-retry",
+            goal_ledger=None,
+        )
+    )
+
+    assert llm.calls == 3
+    assert calls == 1
+    assert trace == []
+    assert decision is not None
+    assert decision.goal == "new_analysis"
+
+
+def test_route_decision_does_not_retry_authentication_failure() -> None:
+    class AuthenticationFailure(RuntimeError):
+        status_code = 401
+
+    class UnauthorizedLLM:
+        calls = 0
+
+        async def acall(self, *_: Any, **__: Any) -> Any:
+            self.calls += 1
+            raise AuthenticationFailure("unauthorized")
+
+    llm = UnauthorizedLLM()
+    runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5)  # type: ignore[arg-type]
+
+    decision, calls, trace = asyncio.run(
+        runtime._decide_route(
+            query="执行数据准备",
+            history=[],
+            session_id="auth-session",
+            job_id="job-auth",
+            goal_ledger=None,
+        )
+    )
+
+    assert llm.calls == 1
+    assert decision is None
+    assert calls == 1
+    assert trace == []
+
+
+def test_route_failure_exposes_sanitized_upstream_reason(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class StructuredPaymentGatewayFailure(RuntimeError):
+        status_code = 402
+        request_id = "req-router-402"
+        body = {
+            "error": {
+                "code": "request_policy_rejected",
+                "message": (
+                    "the concrete upstream reason; "
+                    "Authorization: Bearer must-not-leak"
+                ),
+                "api_key": "must-not-leak",
+            }
+        }
+
+    class FailingLLM:
+        _default_model = "deepseek-v4-flash"
+        calls = 0
+
+        async def acall(self, *_: Any, **__: Any) -> Any:
+            self.calls += 1
+            raise StructuredPaymentGatewayFailure("generic SDK error")
+
+    async def no_sleep(_: float) -> None:
+        return None
+
+    llm = FailingLLM()
+    runtime = RouterRuntime(llm, _PassThroughTools(), max_iterations=5)  # type: ignore[arg-type]
+    monkeypatch.setattr(asyncio, "sleep", no_sleep)
+
+    outcome = asyncio.run(
+        runtime.run(
+            query="执行数据准备",
+            history=[],
+            session_id="failure-detail-session",
+            job_id="job-failure-detail",
+            goal_ledger=None,
+        )
+    )
+
+    assert llm.calls == 3
+    assert outcome.status == "agent_unavailable"
+    assert "HTTP 402" in outcome.message
+    assert "request_policy_rejected" in outcome.message
+    assert "the concrete upstream reason" in outcome.message
+    assert "deepseek-v4-flash" in outcome.message
+    assert "3 次" in outcome.message
+    assert "req-router-402" in outcome.message
+    assert "must-not-leak" not in outcome.message
+
+
+def test_route_failure_reads_flat_openai_error_body(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class FlatOpenAIError(RuntimeError):
+        status_code = 402
+        body = {
+            "code": "invalid_request_error",
+            "message": "Insufficient Balance",
+            "type": "unknown_error",
+        }
+
+    class FailingLLM:
+        _default_model = "deepseek-v4-flash"
+
+        async def acall(self, *_: Any, **__: Any) -> Any:
+            raise FlatOpenAIError("serialized SDK fallback")
+
+    async def no_sleep(_: float) -> None:
+        return None
+
+    runtime = RouterRuntime(FailingLLM(), _PassThroughTools(), max_iterations=5)  # type: ignore[arg-type]
+    monkeypatch.setattr(asyncio, "sleep", no_sleep)
+
+    outcome = asyncio.run(
+        runtime.run(
+            query="执行数据准备",
+            history=[],
+            session_id="flat-openai-error-session",
+            job_id="job-flat-openai-error",
+            goal_ledger=None,
+        )
+    )
+
+    assert outcome.status == "agent_unavailable"
+    assert "错误码 invalid_request_error" in outcome.message
+    assert outcome.message.endswith("Insufficient Balance")
+    assert "serialized SDK fallback" not in outcome.message