Simplify model rate-limit run errors

Constraint: Aone 83566999 only needs model 429 output to terminate as AG-UI RUN_ERROR without hardcoded user-facing copy.\nRejected: fixed localized message or broad error framework | original provider error text is required and a small regex/status-code guard is enough.\nConfidence: high\nScope-risk: narrow\nDirective: Keep RUN_ERROR.message sourced from the original error text; do not introduce generic friendly copy.\nTested: uv run --extra server pytest targeted files -q; uv run ruff check targeted source/server-test files; git diff --check; uvicorn local harness for text 429, structured 429, and normal text.\nNot-tested: GitHub CI final status and DCO fix; earlier pushed commits still require sign-off remediation.
Signed-off-by: congxiao.wxx <congxiao.wxx@alibaba-inc.com>

Change-Id: I248d3f29cfd95560edc48f22d2bd014cc5762cef
Co-developed-by: Codex <noreply@openai.com>

Author: congxiao-wxx

agentrun/integration/langgraph/agent_converter.py

@@ -29,11 +29,11 @@
 import json
 from typing import Any, Dict, Iterator, List, Optional, Union
 
+from agentrun.server.model import AgentResult, EventType
 from agentrun.utils.error_utils import (
     build_error_event_data,
-    format_error_message,
+    is_rate_limited_error,
 )
-from agentrun.server.model import AgentResult, EventType
 from agentrun.utils.log import logger
 
 # 需要从工具输入中过滤掉的内部字段（LangGraph/MCP 注入的运行时对象）
@@ -945,22 +945,40 @@ def _convert_astream_events_event(
         # 7. LLM 错误
         elif event_type == "on_llm_error":
             error = data.get("error")
-            error_message = format_error_message(error)
+            error_message = ""
+            if error is not None:
+                if isinstance(error, Exception):
+                    error_message = f"{type(error).__name__}: {str(error)}"
+                elif isinstance(error, str):
+                    error_message = error
+                else:
+                    error_message = str(error)
 
             yield AgentResult(
                 event=EventType.ERROR,
                 data=build_error_event_data(
                     error,
                     fallback_code="LLM_ERROR",
-                    fallback_message=f"LLM error: {error_message}",
+                    fallback_message=(
+                        error_message
+                        if is_rate_limited_error(error)
+                        else f"LLM error: {error_message}"
+                    ),
                 ),
             )
 
         # 8. Chain 错误
         elif event_type == "on_chain_error":
             error = data.get("error")
             chain_name = event_dict.get("name", "")
-            error_message = format_error_message(error)
+            error_message = ""
+            if error is not None:
+                if isinstance(error, Exception):
+                    error_message = f"{type(error).__name__}: {str(error)}"
+                elif isinstance(error, str):
+                    error_message = error
+                else:
+                    error_message = str(error)
 
             yield AgentResult(
                 event=EventType.ERROR,
@@ -978,7 +996,14 @@ def _convert_astream_events_event(
         elif event_type == "on_retriever_error":
             error = data.get("error")
             retriever_name = event_dict.get("name", "")
-            error_message = format_error_message(error)
+            error_message = ""
+            if error is not None:
+                if isinstance(error, Exception):
+                    error_message = f"{type(error).__name__}: {str(error)}"
+                elif isinstance(error, str):
+                    error_message = error
+                else:
+                    error_message = str(error)
 
             yield AgentResult(
                 event=EventType.ERROR,

agentrun/server/agui_protocol.py

@@ -748,15 +748,11 @@ def _process_event_with_boundaries(
                 message=event.data.get("message", ""),
                 code=event.data.get("code"),
             )
-            extra_fields = {}
-            for key in RUN_ERROR_EXTRA_FIELDS:
-                value = event.data.get(key)
-                if value is not None:
-                    extra_fields[key] = value
-                elif event.addition:
-                    value = event.addition.get(key)
-                    if value is not None:
-                        extra_fields[key] = value
+            extra_fields = {
+                key: event.data[key]
+                for key in RUN_ERROR_EXTRA_FIELDS
+                if key in event.data
+            }
             if extra_fields:
                 agui_event = agui_event.model_copy(update=extra_fields)
             yield self._encoder.encode(agui_event)

agentrun/server/invoker.py

@@ -24,7 +24,10 @@
 )
 import uuid
 
-from agentrun.utils.error_utils import build_error_event_data
+from agentrun.utils.error_utils import (
+    build_error_event_data,
+    is_rate_limited_error,
+)
 from .model import AgentEvent, AgentRequest, EventType
 from .protocol import (
     AsyncInvokeAgentHandler,
@@ -118,10 +121,7 @@ async def invoke_stream(
                     if isinstance(item, str):
                         if not item:  # 跳过空字符串
                             continue
-                        yield AgentEvent(
-                            event=EventType.TEXT,
-                            data={"delta": item},
-                        )
+                        yield self._wrap_text(item)
 
                     elif isinstance(item, AgentEvent):
                         # 处理用户返回的事件
@@ -232,12 +232,7 @@ def _wrap_non_stream(self, result: Any) -> List[AgentEvent]:
             return results
 
         if isinstance(result, str):
-            results.append(
-                AgentEvent(
-                    event=EventType.TEXT,
-                    data={"delta": result},
-                )
-            )
+            results.append(self._wrap_text(result))
 
         elif isinstance(result, AgentEvent):
             # 处理可能的 TOOL_CALL 展开
@@ -248,12 +243,7 @@ def _wrap_non_stream(self, result: Any) -> List[AgentEvent]:
                 if isinstance(item, AgentEvent):
                     results.extend(self._process_user_event(item))
                 elif isinstance(item, str) and item:
-                    results.append(
-                        AgentEvent(
-                            event=EventType.TEXT,
-                            data={"delta": item},
-                        )
-                    )
+                    results.append(self._wrap_text(item))
                 else:
                     results.extend(self._wrap_model_chunk(item))
 
@@ -280,10 +270,7 @@ async def _wrap_stream(
             if isinstance(item, str):
                 if not item:
                     continue
-                yield AgentEvent(
-                    event=EventType.TEXT,
-                    data={"delta": item},
-                )
+                yield self._wrap_text(item)
 
             elif isinstance(item, AgentEvent):
                 for processed_event in self._process_user_event(item):
@@ -351,15 +338,25 @@ def _wrap_model_chunk(self, item: Any) -> List[AgentEvent]:
 
         content = self._read_attr_or_key(item, "content")
         if isinstance(content, str) and content:
-            events.append(
-                AgentEvent(
-                    event=EventType.TEXT,
-                    data={"delta": content},
-                )
-            )
+            events.append(self._wrap_text(content))
 
         return events
 
+    def _wrap_text(self, text: str) -> AgentEvent:
+        if is_rate_limited_error(text):
+            return AgentEvent(
+                event=EventType.ERROR,
+                data=build_error_event_data(
+                    text,
+                    fallback_code=type(text).__name__,
+                    fallback_message=text,
+                ),
+            )
+        return AgentEvent(
+            event=EventType.TEXT,
+            data={"delta": text},
+        )
+
     def _read_attr_or_key(self, obj: Any, key: str) -> Any:
         if isinstance(obj, dict):
             return obj.get(key)

agentrun/utils/error_utils.py

@@ -1,52 +1,29 @@
-"""Error helpers for AgentRun event streams."""
+"""Small helpers for model rate-limit errors."""
 
 import re
 from typing import Any, Dict, Optional
 
 RATE_LIMITED_CODE = "RATE_LIMITED"
 RATE_LIMITED_RETRY_AFTER_MS = 2000
-
+_RATE_LIMIT_TEXT_RE = re.compile(
+    r"429|too[-_\s]*many[-_\s]*requests|rate[-_\s]*limit|throttl",
+    re.IGNORECASE,
+)
 _RATE_LIMIT_CODES = {
     "ratelimitexceeded",
     "ratelimited",
-    "resourcethrottled",
     "throttling",
-    "throttlingquota",
-    "throttlingratequota",
-    "throttlingexception",
     "toomanyrequests",
 }
 
-_RATE_LIMIT_TEXT_PATTERNS = [
-    re.compile(r"\btoo[-_\s]*many[-_\s]*requests\b", re.IGNORECASE),
-    re.compile(
-        r"\brate[-_\s]*limit(?:ed|[-_\s]*exceeded)\b",
-        re.IGNORECASE,
-    ),
-    re.compile(r"\bresource[-_\s]*throttled\b", re.IGNORECASE),
-    re.compile(
-        r"\b(?:throttling|throttlingexception|throttled)\b",
-        re.IGNORECASE,
-    ),
-]
-
-
-def format_error_message(error: Any) -> str:
-    """Format errors consistently with existing LangGraph conversion."""
-    if error is None:
-        return ""
-    if isinstance(error, Exception):
-        return f"{type(error).__name__}: {str(error)}"
-    return str(error)
-
 
 def build_error_event_data(
     error: Any,
     *,
     fallback_code: str,
     fallback_message: str,
 ) -> Dict[str, Any]:
-    """Build AgentEvent ERROR data, normalizing model rate limits."""
+    """Keep the original message; add rate-limit metadata only when matched."""
     if not is_rate_limited_error(error):
         return {"message": fallback_message, "code": fallback_code}
 
@@ -56,70 +33,42 @@ def build_error_event_data(
         "retryable": True,
         "retryAfterMs": RATE_LIMITED_RETRY_AFTER_MS,
     }
-    trace_id = _extract_trace_id(error)
+    trace_id = _get_value(error, "trace_id") or _get_value(error, "traceId")
     if trace_id:
         data["traceId"] = str(trace_id)
     return data
 
 
 def is_rate_limited_error(error: Any) -> bool:
-    """Return whether an error carries an explicit rate-limit signal."""
-    if _extract_status_code(error) == 429:
+    if error is None:
+        return False
+    if _status_code(error) == 429 or _status_code(_get_value(error, "response")) == 429:
         return True
-
-    if _has_rate_limit_code(error):
+    if _rate_limit_code(error) or _rate_limit_code(_get_value(error, "response")):
         return True
-
-    message = str(error or "")
-    return any(pattern.search(message) for pattern in _RATE_LIMIT_TEXT_PATTERNS)
+    return bool(_RATE_LIMIT_TEXT_RE.search(str(error)))
 
 
-def _extract_status_code(error: Any) -> Optional[int]:
-    fallback = None
-    for obj in (error, _get_value(error, "response")):
-        if obj is None:
+def _status_code(obj: Any) -> Optional[int]:
+    for name in ("status_code", "status", "http_status", "statusCode"):
+        value = _get_value(obj, name)
+        if value is None:
             continue
-        for name in ("status_code", "status", "http_status", "statusCode"):
-            status_code = _to_int(_get_value(obj, name))
-            if status_code == 429:
-                return status_code
-            if fallback is None and status_code is not None:
-                fallback = status_code
-    return fallback
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return None
+    return None
 
 
-def _has_rate_limit_code(error: Any) -> bool:
-    for obj in (error, _get_value(error, "response")):
-        if obj is None:
-            continue
-        for name in ("code", "error_code", "errorCode"):
-            error_code = _get_value(obj, name)
-            if (
-                error_code is not None
-                and _normalize_code(error_code) in _RATE_LIMIT_CODES
-            ):
-                return True
+def _rate_limit_code(obj: Any) -> bool:
+    for name in ("code", "error_code", "errorCode"):
+        code = _get_value(obj, name)
+        if code and _normalize_code(code) in _RATE_LIMIT_CODES:
+            return True
     return False
 
 
-def _extract_trace_id(error: Any) -> Optional[Any]:
-    for name in ("trace_id", "traceId", "request_id", "requestId"):
-        trace_id = _get_value(error, name)
-        if trace_id:
-            return trace_id
-
-    response = _get_value(error, "response")
-    headers = _get_value(response, "headers")
-    if not headers:
-        return None
-
-    for name in ("x-acs-request-id", "x-request-id", "x-trace-id", "trace-id"):
-        trace_id = _get_header(headers, name)
-        if trace_id:
-            return trace_id
-    return None
-
-
 def _get_value(obj: Any, name: str) -> Optional[Any]:
     if obj is None:
         return None
@@ -128,31 +77,9 @@ def _get_value(obj: Any, name: str) -> Optional[Any]:
     return getattr(obj, name, None)
 
 
-def _get_header(headers: Any, name: str) -> Optional[Any]:
-    target = str(name).lower()
-    if isinstance(headers, dict):
-        for key, value in headers.items():
-            if str(key).lower() == target:
-                return value
-        return None
-    get = getattr(headers, "get", None)
-    if callable(get):
-        return get(name)
-    return None
-
-
 def _normalize_code(code: Any) -> str:
-    normalized = re.sub(r"[^a-z0-9]", "", str(code).lower())
+    value = "".join(ch for ch in str(code).lower() if ch.isalnum())
     for suffix in ("exception", "error"):
-        if normalized.endswith(suffix):
-            return normalized[: -len(suffix)]
-    return normalized
-
-
-def _to_int(value: Any) -> Optional[int]:
-    if value is None:
-        return None
-    try:
-        return int(value)
-    except (TypeError, ValueError):
-        return None
+        if value.endswith(suffix):
+            return value[: -len(suffix)]
+    return value