Skip to content

Commit 1c27ff5

Browse files
congxiao-wxxcongxiao-wxx
committed
Constrain rate-limit normalization to explicit signals
Close the review gaps by moving the helper to a shared utils layer, preserving AG-UI encoder framing, limiting LangGraph normalization to LLM errors, and adding false-positive plus throttling regression coverage. Constraint: Aone 83566999 requires model rate limits to surface as stable AG-UI RUN_ERROR without leaking raw provider errors. Rejected: Matching generic HTTP/status/code 429 text | it misclassified explanatory non-rate-limit errors. Rejected: Hand-written RUN_ERROR SSE framing | it forked the AG-UI encoder contract. Confidence: high Scope-risk: narrow Directive: Keep future rate-limit text matching limited to explicit provider throttle/rate-limit semantics. Tested: uv run --extra server pytest tests/unittests/server/test_error_utils.py tests/unittests/server/test_invoker.py tests/unittests/server/test_agui_protocol.py tests/unittests/integration/test_langgraph_events.py tests/unittests/integration/test_langgraph_to_agent_event.py -q (143 passed, 1 warning); git diff --check; focused ruff; local uvicorn /ag-ui/agent harness for structured 429, throttling text, and false-positive code 429. Change-Id: Ie114d646380623dc8546f082cf2b0776035ccade Co-developed-by: Codex <noreply@openai.com> Not-tested: GitHub CI status due local gh auth/API limitations.
1 parent d75aca5 commit 1c27ff5

8 files changed

Lines changed: 142 additions & 23 deletions

File tree

‎agentrun/integration/langgraph/agent_converter.py‎

Lines changed: 9 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
import json
3030
from typing import Any, Dict, Iterator, List, Optional, Union
3131

32-
from agentrun.server.error_utils import (
32+
from agentrun.utils.error_utils import (
3333
build_error_event_data,
3434
format_error_message,
3535
)
@@ -964,15 +964,14 @@ def _convert_astream_events_event(
964964

965965
yield AgentResult(
966966
event=EventType.ERROR,
967-
data=build_error_event_data(
968-
error,
969-
fallback_code="CHAIN_ERROR",
970-
fallback_message=(
967+
data={
968+
"message": (
971969
f"Chain '{chain_name}' error: {error_message}"
972970
if chain_name
973971
else error_message
974972
),
975-
),
973+
"code": "CHAIN_ERROR",
974+
},
976975
)
977976

978977
# 9. Retriever 错误
@@ -983,15 +982,14 @@ def _convert_astream_events_event(
983982

984983
yield AgentResult(
985984
event=EventType.ERROR,
986-
data=build_error_event_data(
987-
error,
988-
fallback_code="RETRIEVER_ERROR",
989-
fallback_message=(
985+
data={
986+
"message": (
990987
f"Retriever '{retriever_name}' error: {error_message}"
991988
if retriever_name
992989
else error_message
993990
),
994-
),
991+
"code": "RETRIEVER_ERROR",
992+
},
995993
)
996994

997995
# =========================================================================

‎agentrun/server/agui_protocol.py‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -748,17 +748,18 @@ def _process_event_with_boundaries(
748748
message=event.data.get("message", ""),
749749
code=event.data.get("code"),
750750
)
751-
event_dict = agui_event.model_dump(by_alias=True, exclude_none=True)
751+
extra_fields = {}
752752
for key in RUN_ERROR_EXTRA_FIELDS:
753753
value = event.data.get(key)
754754
if value is not None:
755-
event_dict[key] = value
755+
extra_fields[key] = value
756756
elif event.addition:
757757
value = event.addition.get(key)
758758
if value is not None:
759-
event_dict[key] = value
760-
json_str = json.dumps(event_dict, ensure_ascii=False)
761-
yield f"event: RUN_ERROR\ndata: {json_str}\n\n"
759+
extra_fields[key] = value
760+
if extra_fields:
761+
agui_event = agui_event.model_copy(update=extra_fields)
762+
yield self._encoder.encode(agui_event)
762763
return
763764

764765
# STATE 事件

‎agentrun/server/invoker.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424
)
2525
import uuid
2626

27-
from .error_utils import build_error_event_data
27+
from agentrun.utils.error_utils import build_error_event_data
2828
from .model import AgentEvent, AgentRequest, EventType
2929
from .protocol import (
3030
AsyncInvokeAgentHandler,

agentrun/server/error_utils.py renamed to agentrun/utils/error_utils.py

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
"""Error helpers for AgentRun server streams."""
1+
"""Error helpers for AgentRun event streams."""
22

33
import re
44
from typing import Any, Dict, Optional
@@ -25,9 +25,8 @@
2525
re.IGNORECASE,
2626
),
2727
re.compile(r"\bresource[-_\s]*throttled\b", re.IGNORECASE),
28-
re.compile(r"\bthrottling(?:exception| exception| error)\b", re.IGNORECASE),
2928
re.compile(
30-
r"\b(?:http|status|status code|code)\s*[:=]?\s*429\b",
29+
r"\b(?:throttling|throttlingexception|throttled)\b",
3130
re.IGNORECASE,
3231
),
3332
]

‎tests/unittests/integration/test_langgraph_events.py‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -890,6 +890,29 @@ def test_on_chain_error(self):
890890
assert "KeyError" in results[0].data["message"]
891891
assert results[0].data["code"] == "CHAIN_ERROR"
892892

893+
def test_on_chain_error_status_429_keeps_chain_error(self):
894+
"""测试 chain 429 不套用模型限流语义"""
895+
896+
class ChainStatusError(RuntimeError):
897+
status_code = 429
898+
899+
event = {
900+
"event": "on_chain_error",
901+
"name": "agent_chain",
902+
"run_id": "run_chain",
903+
"data": {
904+
"error": ChainStatusError("chain quota exceeded"),
905+
},
906+
}
907+
908+
results = list(AgentRunConverter().to_agui_events(event))
909+
910+
assert len(results) == 1
911+
assert results[0].event == EventType.ERROR
912+
assert results[0].data["code"] == "CHAIN_ERROR"
913+
assert "retryable" not in results[0].data
914+
assert "retryAfterMs" not in results[0].data
915+
893916
def test_on_retriever_error(self):
894917
"""测试 on_retriever_error 事件
895918
@@ -913,6 +936,29 @@ def test_on_retriever_error(self):
913936
assert "ConnectionError" in results[0].data["message"]
914937
assert results[0].data["code"] == "RETRIEVER_ERROR"
915938

939+
def test_on_retriever_error_status_429_keeps_retriever_error(self):
940+
"""测试 retriever 429 不套用模型限流语义"""
941+
942+
class RetrieverStatusError(RuntimeError):
943+
status_code = 429
944+
945+
event = {
946+
"event": "on_retriever_error",
947+
"name": "vector_store",
948+
"run_id": "run_retriever",
949+
"data": {
950+
"error": RetrieverStatusError("retriever quota exceeded"),
951+
},
952+
}
953+
954+
results = list(AgentRunConverter().to_agui_events(event))
955+
956+
assert len(results) == 1
957+
assert results[0].event == EventType.ERROR
958+
assert results[0].data["code"] == "RETRIEVER_ERROR"
959+
assert "retryable" not in results[0].data
960+
assert "retryAfterMs" not in results[0].data
961+
916962
def test_tool_error_in_complete_flow(self):
917963
"""测试完整流程中的工具错误
918964

‎tests/unittests/integration/test_langgraph_to_agent_event.py‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -888,6 +888,29 @@ def test_on_chain_error(self):
888888
assert "KeyError" in results[0].data["message"]
889889
assert results[0].data["code"] == "CHAIN_ERROR"
890890

891+
def test_on_chain_error_status_429_keeps_chain_error(self):
892+
"""测试 chain 429 不套用模型限流语义"""
893+
894+
class ChainStatusError(RuntimeError):
895+
status_code = 429
896+
897+
event = {
898+
"event": "on_chain_error",
899+
"name": "agent_chain",
900+
"run_id": "run_chain",
901+
"data": {
902+
"error": ChainStatusError("chain quota exceeded"),
903+
},
904+
}
905+
906+
results = list(AgentRunConverter().to_agui_events(event))
907+
908+
assert len(results) == 1
909+
assert results[0].event == EventType.ERROR
910+
assert results[0].data["code"] == "CHAIN_ERROR"
911+
assert "retryable" not in results[0].data
912+
assert "retryAfterMs" not in results[0].data
913+
891914
def test_on_retriever_error(self):
892915
"""测试 on_retriever_error 事件
893916
@@ -911,6 +934,29 @@ def test_on_retriever_error(self):
911934
assert "ConnectionError" in results[0].data["message"]
912935
assert results[0].data["code"] == "RETRIEVER_ERROR"
913936

937+
def test_on_retriever_error_status_429_keeps_retriever_error(self):
938+
"""测试 retriever 429 不套用模型限流语义"""
939+
940+
class RetrieverStatusError(RuntimeError):
941+
status_code = 429
942+
943+
event = {
944+
"event": "on_retriever_error",
945+
"name": "vector_store",
946+
"run_id": "run_retriever",
947+
"data": {
948+
"error": RetrieverStatusError("retriever quota exceeded"),
949+
},
950+
}
951+
952+
results = list(AgentRunConverter().to_agui_events(event))
953+
954+
assert len(results) == 1
955+
assert results[0].event == EventType.ERROR
956+
assert results[0].data["code"] == "RETRIEVER_ERROR"
957+
assert "retryable" not in results[0].data
958+
assert "retryAfterMs" not in results[0].data
959+
914960
def test_tool_error_in_complete_flow(self):
915961
"""测试完整流程中的工具错误
916962

‎tests/unittests/server/test_agui_protocol.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ def invoke_agent(request: AgentRequest):
128128
)
129129

130130
assert response.status_code == 200
131-
assert "event: RUN_ERROR" in response.text
131+
assert "event: RUN_ERROR" not in response.text
132132
events = _agui_sse_events(response)
133133
run_error = next(
134134
event for event in events if event.get("type") == "RUN_ERROR"
@@ -156,7 +156,7 @@ def invoke_agent(request: AgentRequest):
156156
)
157157

158158
assert response.status_code == 200
159-
assert "event: RUN_ERROR" in response.text
159+
assert "event: RUN_ERROR" not in response.text
160160
events = _agui_sse_events(response)
161161
types = [event.get("type") for event in events]
162162
run_error = next(

‎tests/unittests/server/test_error_utils.py‎

Lines changed: 30 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,38 @@
11
"""Tests for server error helpers."""
22

3-
from agentrun.server.error_utils import _get_header
3+
from agentrun.utils.error_utils import _get_header, is_rate_limited_error
44

55

66
def test_get_header_matches_name_case_insensitively():
77
headers = {"x-trace-id": "trace-123"}
88

99
assert _get_header(headers, "X-Trace-ID") == "trace-123"
10+
11+
12+
def test_explanatory_code_429_text_is_not_rate_limited():
13+
error = RuntimeError("validation failed for field code 429")
14+
15+
assert not is_rate_limited_error(error)
16+
17+
18+
def test_explanatory_http_429_text_is_not_rate_limited():
19+
error = RuntimeError(
20+
"docs mention HTTP 429 means rate limit; actual error is 401"
21+
)
22+
23+
assert not is_rate_limited_error(error)
24+
25+
26+
def test_explicit_throttling_text_is_rate_limited():
27+
assert is_rate_limited_error(RuntimeError("Throttling: model overloaded"))
28+
29+
30+
def test_explicit_throttled_text_is_rate_limited():
31+
assert is_rate_limited_error(RuntimeError("request throttled by provider"))
32+
33+
34+
def test_structured_status_429_is_rate_limited():
35+
class RateLimitError(RuntimeError):
36+
status_code = 429
37+
38+
assert is_rate_limited_error(RateLimitError("model overloaded"))

0 commit comments

Comments
 (0)