fix(llmcore): route OpenAI Responses SSE coded transient errors to _stream_with_retry (closes #753)

llmcore.py

@@ -142,10 +142,84 @@ def _parse_claude_json(data):
         elif b.get("type") == "thinking": yield ""
     return content_blocks
 
-def _raise_if_retryable_overload(emsg):
-    """HTTP 200 SSE/body overload → ConnectionError so _stream_with_retry can backoff."""
-    if emsg and re.search(r'concurrency|retry later|overloaded|rate.?limit', emsg, re.I):
-        raise requests.ConnectionError(emsg)
+# Coded transient errors from OpenAI Responses / Anthropic Messages / etc.
+# These come through SSE `error`/`response.failed` events and must trigger
+# _stream_with_retry's exponential-backoff loop. Keep the list explicit so a
+# permanent failure (e.g. invalid_request_error, context_length_exceeded) does
+# not retry and burn the budget.
+_RETRYABLE_STREAM_ERR_CODES = frozenset({
+    "rate_limit_error",
+    "rate_limit_exceeded",
+    "server_error",
+    "service_unavailable",
+    "api_error",
+    "overloaded",
+    "engine_overloaded",
+    "timeout",
+    "request_timeout",
+    "tokens_exceeded_retry",  # cohere-style transient
+    "upstream_error",
+    "temporary_error",
+    "too_many_requests",
+})
+_RETRYABLE_STREAM_ERR_RE = re.compile(
+    r"concurrency|retry later|overloaded|rate.?limit|server.{0,8}(?:error|busy|unavailable)"
+    r"|service.{0,8}unavailable|engine.{0,8}(?:overloaded|busy)|temporarily.{0,12}(?:unable|unavailable)"
+    r"|try again later|capacity",
+    re.I,
+)
+
+def _is_retryable_stream_err(err=None, emsg=None):
+    """True if a stream `error`/`response.failed` event should trigger _stream_with_retry.
+
+    Accepts either:
+      * err: dict with `code` / `type` / `message` (Responses API style), or
+      * emsg: bare error message string (legacy / prose path).
+    """
+    code = ""
+    message = emsg or ""
+    if isinstance(err, dict):
+        # OpenAI Responses nests error under {error: {code, type, message, ...}};
+        # some providers put the code/type/message at the top level instead.
+        for key in ("code", "type", "error_code"):
+            v = err.get(key)
+            if isinstance(v, str) and v:
+                code = v; break
+        sub = err.get("error")
+        if isinstance(sub, dict) and not code:
+            for key in ("code", "type"):
+                v = sub.get(key)
+                if isinstance(v, str) and v:
+                    code = v; break
+        # Surface message for prose matching if present.
+        m = err.get("message") or (sub.get("message") if isinstance(sub, dict) else None)
+        if isinstance(m, str) and m:
+            message = m
+    # Numeric error codes (rare in OpenAI, common in upstream-proxies): treat any
+    # 429 / 5xx that slipped into the SSE `error` payload as retryable.
+    if code and code.isdigit() and int(code) in {408, 409, 425, 429} | {500, 502, 503, 504, 520, 521, 522, 523, 524, 525, 526, 527, 529}:
+        return True
+    # Coded transient categories.
+    if code and code.lower() in _RETRYABLE_STREAM_ERR_CODES:
+        return True
+    # Prose fallback — widens net against future provider-side message rewording.
+    return bool(message) and bool(_RETRYABLE_STREAM_ERR_RE.search(message))
+
+def _raise_if_retryable_overload(emsg_or_err):
+    """HTTP 200 SSE/body overload → ConnectionError so _stream_with_retry can backoff.
+
+    Accepts either a string (legacy path) or a dict (SSE error payloads from
+    OpenAI Responses `error` / `response.failed` events, Anthropic `error` blocks,
+    and most third-party proxies).
+    """
+    if emsg_or_err and _is_retryable_stream_err(err=emsg_or_err if not isinstance(emsg_or_err, str) else None,
+                                                  emsg=emsg_or_err if isinstance(emsg_or_err, str) else None):
+        if isinstance(emsg_or_err, str):
+            raise requests.ConnectionError(emsg_or_err)
+        if isinstance(emsg_or_err, dict):
+            msg = emsg_or_err.get("message") or (emsg_or_err.get("error", {}) or {}).get("message") or "stream error"
+            raise requests.ConnectionError(msg)
+        raise requests.ConnectionError(str(emsg_or_err))
 
 def _parse_claude_sse(resp_lines):
     """Parse Anthropic SSE stream. Yields text chunks, returns list[content_block]."""
@@ -203,7 +277,7 @@ def _parse_claude_sse(resp_lines):
         elif evt_type == "error":
             err = evt.get("error", {})
             emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
-            _raise_if_retryable_overload(emsg)  # 走 _stream_with_retry，避免落到 ga 应用层
+            _raise_if_retryable_overload(err if isinstance(err, dict) else emsg)  # 走 _stream_with_retry，避免落到 ga 应用层
             warn = f"\n\n!!!Error: SSE {emsg}"; break
     if not warn:
         if not got_message_stop and not stop_reason: warn = "\n\n[!!! 流异常中断，未收到完整响应 !!!]"
@@ -279,7 +353,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"):
             elif etype == "error":
                 err = evt.get("error", {})
                 emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
-                _raise_if_retryable_overload(emsg)
+                _raise_if_retryable_overload(err if isinstance(err, dict) else emsg)
                 if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
                 break
             elif etype == "response.completed":
@@ -301,7 +375,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"):
                 _record_usage(usage, api_mode)
                 err = ((evt.get("response") or {}).get("error") or {})
                 emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
-                _raise_if_retryable_overload(emsg)
+                _raise_if_retryable_overload(err if isinstance(err, dict) else emsg)
                 if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}"
                 break
         blocks = []
@@ -402,7 +476,7 @@ def _parse_openai_json(data, api_mode="chat_completions"):
         if status == "failed":
             err = data.get("error") or {}
             emsg = err.get("message", str(err)) if isinstance(err, dict) else str(err)
-            _raise_if_retryable_overload(emsg)
+            _raise_if_retryable_overload(err if isinstance(err, dict) else emsg)
             if emsg: blocks.append({"type": "text", "text": f"!!!Error: {emsg}"}); yield f"!!!Error: {emsg}"
         elif status == "incomplete" and not any(b.get("type") == "text" for b in blocks):
             reason = ((data.get("incomplete_details") or {}).get("reason", "")) or "unknown"

tests/test_llmcore_sse_retry_codes.py

@@ -0,0 +1,210 @@
+"""Regression tests for issue #753 — OpenAI Responses SSE overload / transient
+errors must trigger _stream_with_retry instead of being returned to the caller
+as if the request succeeded.
+
+Covers:
+  * `_is_retryable_stream_err`: coded / numeric / prose detection on
+    dict-shaped and string-shaped payloads.
+  * `_raise_if_retryable_overload`: raises ConnectionError for retryable
+    payloads, never raises for permanent failures.
+  * `_parse_openai_sse(api_mode="responses")` `error` event with
+    `{code: "server_error"}` (no "overloaded" word anywhere) routes to retry.
+  * `_parse_openai_sse(api_mode="responses")` `response.failed` event with
+    `{code: "rate_limit_error"}` routes to retry.
+  * Permanent failures (`invalid_request_error`, `context_length_exceeded`)
+    do not retry — they remain visible as `!!!Error:` text in the stream.
+"""
+
+import os
+import sys
+import types
+import importlib
+
+import pytest
+
+
+def _import_llmcore():
+    """Import llmcore with `requests` and any missing deps stubbed in."""
+    sys.modules.pop("llmcore", None)
+    # llmcore.py imports requests at module scope — must already be present.
+    import llmcore  # noqa: F401
+    return sys.modules["llmcore"]
+
+
+def test_is_retryable_stream_err_codes():
+    lc = _import_llmcore()
+    is_retry = lc._is_retryable_stream_err
+    # OpenAI Responses / Anthropic / upstream-proxy transient codes.
+    assert is_retry(err={"code": "rate_limit_error"}) is True
+    assert is_retry(err={"type": "rate_limit_error"}) is True
+    assert is_retry(err={"code": "server_error"}) is True
+    assert is_retry(err={"code": "service_unavailable"}) is True
+    assert is_retry(err={"code": "overloaded"}) is True
+    assert is_retry(err={"code": "engine_overloaded"}) is True
+    assert is_retry(err={"code": "api_error"}) is True
+    assert is_retry(err={"code": "upstream_error"}) is True
+    assert is_retry(err={"code": "too_many_requests"}) is True
+    # Numeric codes slipped into SSE error payload.
+    assert is_retry(err={"code": "429"}) is True
+    assert is_retry(err={"code": "503"}) is True
+    assert is_retry(err={"code": "502"}) is True
+    # OpenAI Responses nested-shape error payload.
+    assert is_retry(err={"error": {"code": "rate_limit_error"}}) is True
+    assert is_retry(err={"error": {"type": "server_error", "message": "down"}}) is True
+    # Prose fallback — original behavior preserved for plain message strings.
+    assert is_retry(emsg="Our servers are currently overloaded. Please try again later.") is True
+    assert is_retry(emsg="concurrency limit hit") is True
+    assert is_retry(emsg="rate limit exceeded, please retry") is True
+    # New prose patterns the old regex missed.
+    assert is_retry(emsg="server is busy, retry") is True
+    assert is_retry(emsg="Service temporarily unavailable") is True
+    assert is_retry(emsg="engine overloaded, backoff") is True
+    assert is_retry(emsg="try again later") is True
+    assert is_retry(emsg="reached capacity, try later") is True
+
+
+def test_is_retryable_stream_err_permanent():
+    lc = _import_llmcore()
+    is_retry = lc._is_retryable_stream_err
+    # Permanent failures — must NOT retry, surface as `!!!Error:` text instead.
+    assert is_retry(err={"code": "invalid_request_error"}) is False
+    assert is_retry(err={"code": "context_length_exceeded"}) is False
+    assert is_retry(err={"code": "authentication_error"}) is False
+    assert is_retry(err={"code": "permission_denied"}) is False
+    assert is_retry(err={"code": "not_found"}) is False
+    assert is_retry(err={"code": "400"}) is False  # 400 is a permanent bad-request
+    assert is_retry(err={"code": "404"}) is False
+    # Plain text with no signal words.
+    assert is_retry(emsg="") is False
+    assert is_retry(emsg=None) is False
+    assert is_retry(emsg="something went wrong, please check your input") is False
+
+
+def test_raise_if_retryable_overload_dict():
+    lc = _import_llmcore()
+    rai = lc._raise_if_retryable_overload
+    # Coded transient → raise.
+    with pytest.raises(Exception) as ei:
+        rai({"code": "server_error", "message": "down"})
+    assert "down" in str(ei.value)
+    # String transient → raise.
+    with pytest.raises(Exception):
+        rai("server overloaded")
+
+
+def test_raise_if_retryable_overload_no_raise_for_permanent():
+    lc = _import_llmcore()
+    rai = lc._raise_if_retryable_overload
+    # Permanent: must return normally (no raise).
+    rai({"code": "invalid_request_error", "message": "bad"})
+    rai("unrelated message about syntax error")
+    rai("")
+    rai(None)
+
+
+def _sse(events):
+    """Build an iterable of byte/str lines that mimics `iter_lines()`."""
+    out = []
+    for e in events:
+        if isinstance(e, str):
+            out.append(e.encode("utf-8"))
+        else:
+            out.append(e)
+    return out
+
+
+def test_parse_openai_sse_responses_error_routes_to_retry():
+    """`_parse_openai_sse` with `error` event whose `code` is `server_error`
+    must raise ConnectionError so `_stream_with_retry` retries.
+    """
+    lc = _import_llmcore()
+    parse = lc._parse_openai_sse
+    lines = _sse([
+        'data: {"type": "response.output_text.delta", "delta": "hello"}',
+        # Coded transient — should NOT appear as assistant text, should raise.
+        'data: {"type": "error", "error": {"code": "server_error", "message": "Server is down"}}',
+    ])
+    gen = parse(iter(lines), api_mode="responses")
+    # First chunk streams fine.
+    first = next(gen)
+    assert first == "hello"
+    # The next call hits the error event — must raise ConnectionError.
+    with pytest.raises(Exception) as ei:
+        next(gen)
+    assert "Server is down" in str(ei.value)
+
+
+def test_parse_openai_sse_responses_response_failed_routes_to_retry():
+    lc = _import_llmcore()
+    parse = lc._parse_openai_sse
+    lines = _sse([
+        'data: {"type": "response.output_text.delta", "delta": "ok"}',
+        # response.failed path with rate_limit_error — must retry.
+        'data: {"type": "response.failed", "response": {"error": {"code": "rate_limit_error", "message": "Try again"}, "usage": {}}}',
+    ])
+    gen = parse(iter(lines), api_mode="responses")
+    assert next(gen) == "ok"
+    with pytest.raises(Exception) as ei:
+        next(gen)
+    assert "Try again" in str(ei.value)
+
+
+def test_parse_openai_sse_responses_permanent_failure_surfaces_as_text():
+    lc = _import_llmcore()
+    parse = lc._parse_openai_sse
+    lines = _sse([
+        'data: {"type": "response.output_text.delta", "delta": ""}',  # empty delta to get seen_delta path
+        'data: {"type": "response.failed", "response": {"error": {"code": "invalid_request_error", "message": "Bad model ID"}, "usage": {}}}',
+    ])
+    gen = parse(iter(lines), api_mode="responses")
+    # Must yield `!!!Error:` text, not raise.
+    chunks = []
+    for c in gen:
+        chunks.append(c)
+    assert any("Bad model ID" in c for c in chunks)
+
+
+def test_parse_claude_sse_error_routes_to_retry():
+    """Claude path's `error` event already used the helper, but verify it now
+    also honors the new prose patterns (the original regex didn't catch
+    'Server is busy, retry')."""
+    lc = _import_llmcore()
+    parse = lc._parse_claude_sse
+    lines = _sse([
+        'data: {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}',
+        'data: {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}}',
+        'data: {"type": "content_block_stop", "index": 0}',
+        'data: {"type": "error", "error": {"message": "Server is busy, retry"}}',
+    ])
+    gen = parse(iter(lines))
+    chunks = []
+    try:
+        for c in gen:
+            chunks.append(c)
+    except Exception as e:  # noqa: BLE001
+        # A ConnectionError is acceptable — the stream's terminal warn fires
+        # only AFTER the helper decides not to retry. For retryable errors it
+        # should raise and stop the generator before warn assignment.
+        assert "Server is busy" in str(e), f"unexpected: {type(e).__name__}: {e}"
+        return
+    pytest.fail("Expected retry raise; got chunks: %r" % (chunks,))
+
+
+def test_no_regression_on_legacy_message_only():
+    """Pre-#753 prose path: an error event with no `code`/`type` but text that
+    contains 'overloaded' must still retry (regression guard)."""
+    lc = _import_llmcore()
+    parse = lc._parse_openai_sse
+    lines = _sse([
+        'data: {"type": "response.output_text.delta", "delta": "x"}',
+        'data: {"type": "error", "error": {"message": "Our servers are currently overloaded."}}',
+    ])
+    gen = parse(iter(lines), api_mode="responses")
+    assert next(gen) == "x"
+    with pytest.raises(Exception) as ei:
+        next(gen)
+    assert "overloaded" in str(ei.value).lower()
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-vv"]))