diff --git a/CHANGELOG.md b/CHANGELOG.md
index c89dfadde6..1342b9339a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1247,6 +1247,20 @@ This release closes that milestone with **621 merged pull requests**.
 
 ### Fixed
 
+- **llm:** A provider error that arrives inside a streaming response that has
+  already returned `200` is now retried like the same error sent as an HTTP
+  status. An OpenAI-compatible gateway that answers `200` early, to keep slow
+  requests alive with SSE keepalives, reports a later 504 as an SSE `error`
+  event; the openai SDK raises that as an `APIError` with no status, so
+  `LLMErrorHandlingMiddleware` classified it as generic and the run failed on
+  the first attempt. Such a status-less error now takes a 4xx or 5xx status
+  from a numeric `code` or `status` field, an explicit `[504]`, `HTTP 502` or
+  `status 503` in its message (not an index such as `messages[502]`), or
+  `type: server_error` (500); only when none yields one does its busy check
+  also match spellings such as `rate-limit` and `RATE_LIMIT_…`. Retry budget,
+  backoff and circuit breaker treat it as the equivalent HTTP error; errors
+  that carry a status are classified as before. (issue #6896, [#6897])
+
 - **channels:** Runs started from an IM channel no longer die with the Gateway
   instance that received the message. The manager holds `runs.wait` /
   `runs.stream` open for the whole run, and the Gateway cancels a run whose
@@ -11732,3 +11746,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6852]: https://github.com/bytedance/deer-flow/pull/6852
 [#6865]: https://github.com/bytedance/deer-flow/pull/6865
 [#6876]: https://github.com/bytedance/deer-flow/pull/6876
+[#6897]: https://github.com/bytedance/deer-flow/pull/6897
diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md
index 5ea8e271fa..e8591f6537 100644
--- a/CHANGELOG_zh.md
+++ b/CHANGELOG_zh.md
@@ -1002,6 +1002,16 @@
 
 ### 修复
 
+- **llm：** 流式响应已返回 `200` 之后才在流内报告的 provider 错误，现在会像以 HTTP
+  状态码返回的同一错误一样重试。为了用 SSE keepalive 维持慢请求而提前返回 `200` 的
+  OpenAI 兼容网关，会把之后发生的 504 作为 SSE `error` 事件发出；openai SDK 将其抛出为
+  不带状态码的 `APIError`，因此 `LLMErrorHandlingMiddleware` 此前把它归为普通错误，
+  运行在第一次尝试后即失败。现在这类不带状态码的错误会依次从数字型 `code` 或 `status`
+  字段、消息中明确写出的 `[504]`、`HTTP 502` 或 `status 503`（`messages[502]` 这类
+  数组下标不算）、以及 `type: server_error`（视为 500）中取得 4xx 或 5xx 状态码；都
+  取不到时，繁忙判断才会额外匹配 `rate-limit`、`RATE_LIMIT_…` 等写法。重试预算、退避
+  和熔断器按对应的 HTTP 错误处理；带状态码的错误分类不变。 (issue #6896, [#6897])
+
 - **渠道：** 从 IM 渠道发起的运行不再随接收消息的 Gateway 实例一起消亡。manager 在整个
   运行期间保持 `runs.wait` / `runs.stream` 请求打开，而 Gateway 会在创建请求断开时取消
   运行，除非该请求声明了 `on_disconnect: continue`；因此 leader 切换、滚动重启或接收
@@ -9358,3 +9368,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6852]: https://github.com/bytedance/deer-flow/pull/6852
 [#6865]: https://github.com/bytedance/deer-flow/pull/6865
 [#6876]: https://github.com/bytedance/deer-flow/pull/6876
+[#6897]: https://github.com/bytedance/deer-flow/pull/6897
diff --git a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
index 0335721e66..4ce11018ea 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
+++ b/backend/packages/harness/deerflow/agents/middlewares/AGENTS.md
@@ -81,7 +81,7 @@ No match: skip inspection, log ID. Tests: `test_read_mark_tool_call_correlation.
    their narrower discovery allowlists never rebuild the shared thread view or
    force eager sandbox acquisition.
 8. **DanglingToolCallMiddleware** - Injects placeholder ToolMessages for AIMessage tool_calls that lack responses (e.g., user interruption), preserving raw provider tool-call payloads in `additional_kwargs["tool_calls"]`; malformed tool-call names and arguments are sanitized in the model-bound request so strict OpenAI-compatible providers do not reject the next request
-9. **LLMErrorHandlingMiddleware** - Assistant error recovery. HTTP 529 is transient by status; retries re-enter admission. Honor numeric/date hints up to 24h via `utils.retry_after`, otherwise jitter. Only current circuit-generation owners settle/release sync/async half-open probes. Cancellation propagates without retry/failure accounting.
+9. **LLMErrorHandlingMiddleware** - Assistant error recovery. HTTP 529 is transient by status; retries re-enter admission. Honor numeric/date hints up to 24h via `utils.retry_after`, otherwise jitter. Only current circuit-generation owners settle/release sync/async half-open probes. Cancellation propagates without retry/failure accounting. A status-less (in-stream) error takes a 4xx/5xx status from a numeric `code`/`status`, a `[504]`/`HTTP 502` in its text, or `type: server_error`; else its busy match reads `_`/`-` as spaces.
 10. **ArtifactResolutionMiddleware** - See [tool artifacts](TOOL_ARTIFACTS.md).
 11. **Authorization / GuardrailMiddleware** - Up to two independent pre-tool-call gates run here. When `authorization.enabled`, the `AuthorizationProvider` instance already used for Layer 1 capability filtering is wrapped by `GuardrailAuthorizationAdapter` and reused for Layer 2 execution checks. A generated `tool_search` bypasses the adapter's second provider call only when the current build has a concrete deferred setup; its catalog was already filtered by Layer 1, and an ordinary same-named tool without that deferred setup receives no exemption. When `guardrails.enabled`, the explicitly configured `GuardrailProvider` is appended after authorization and still evaluates every call, including `tool_search`. Authorization therefore runs outermost and can deny before an external guardrail call; both use the existing middleware's fail-closed, audit, sync/async, and error-`ToolMessage` behavior. See the authorization RFC and [docs/GUARDRAILS.md](../../../../../docs/GUARDRAILS.md).
 
diff --git a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py
index 7602cab3a2..8d451d5293 100644
--- a/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py
+++ b/backend/packages/harness/deerflow/agents/middlewares/llm_error_handling_middleware.py
@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import random
+import re
 import threading
 import time
 from collections import deque
@@ -169,6 +170,22 @@ _REASON_RETRY_BUDGETS: dict[str, int] = {
 # streamed response (HTTP 200) yields no generation chunk at all.
 _EMPTY_STREAM_MARKER = "no generations found in stream"
 
+# An HTTP status written into the text of an error that carries no status:
+# ``[504]`` - but not an index such as ``messages[502]``, which the bracketed
+# form's lookbehind rejects; ``HTTP 502``, ``(HTTP 502)``, ``HTTP/1.1 502``;
+# ``status 503``, ``status: 503``, ``status_code=503``, ``"status": 503``. The
+# keyword is required and must be separated from the number - by spaces or tabs
+# after ``HTTP``, and also by ``:`` or ``=`` after ``status`` - so a bare number
+# (``130504 tokens``), one on the next line or one run into the keyword
+# (``status503``) never counts. Digits are ASCII only: ``\d`` and ``int()``
+# also accept other scripts' digits. The spaces before and after ``:``/``=``
+# are separate runs, which keeps matching linear in the length of the message.
+_IN_BAND_STATUS_RE = re.compile(r"(?<![\w\]])\[([0-9]{3})\]|\bhttp(?:/[0-9](?:\.[0-9])?)?[ \t]+([0-9]{3})\b|\bstatus(?:[ _]?code)?(?:[\"']?[ \t]*[:=][ \t]*|[\"']?[ \t]+)([0-9]{3})\b")
+
+# Runs of ``_``, ``-`` and whitespace. A status-less error's busy check reads each
+# run as one space, so ``rate-limit`` and ``RATE_LIMIT_...`` match ``rate limit``.
+_SEPARATOR_RUN_RE = re.compile(r"[\s_\-]+")
+
 # Exception class names that indicate the upstream stream-chunk watchdog
 # fired because the model stalled mid-flight. These deserve a more specific
 # user-facing message than the generic "temporarily unavailable" copy,
@@ -611,6 +628,13 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
         lowered = detail.lower()
         error_code = _extract_error_code(exc)
         status_code = _extract_status_code(exc)
+        # A provider that has already answered ``200`` reports a later failure
+        # inside the stream; the openai SDK raises it as a bare ``APIError`` with
+        # no status code and no response. Only such status-less errors take a
+        # status from their own fields and text (``_extract_in_band_status``).
+        # As with a mid-stream ``ReadError``, a retry re-runs the whole call:
+        # tokens the failed attempt already streamed are not taken back.
+        status_less = status_code is None
 
         if isinstance(exc, EmptyModelResponseError):
             return True, "empty_response"
@@ -659,10 +683,16 @@ class LLMErrorHandlingMiddleware(AgentMiddleware[AgentState]):
         # Same upstream-payload glitch, same retry/backoff path.
         if isinstance(exc, ValueError) and _EMPTY_STREAM_MARKER in lowered:
             return True, "transient"
+        if status_less:
+            status_code = _extract_in_band_status(exc)
+            if status_code is not None and 500 <= status_code <= 599:
+                return True, "transient"
         if status_code in _RETRIABLE_STATUS_CODES:
             return True, "transient"
         if _matches_any(lowered, _BUSY_PATTERNS):
             return True, "busy"
+        if status_code is None and (_matches_any(_normalize_separators(detail), _BUSY_PATTERNS) or _matches_any(_normalize_separators(str(error_code)), _BUSY_PATTERNS)):
+            return True, "busy"
 
         return False, "generic"
 
@@ -1078,6 +1108,57 @@ def _extract_status_code(exc: BaseException) -> int | None:
     return status if isinstance(status, int) else None
 
 
+def _extract_in_band_status(exc: BaseException) -> int | None:
+    """Return the HTTP status that a status-less error carries in its contents.
+
+    Three sources, in order; the first that yields a status decides:
+    a numeric ``code`` on the error, or a numeric ``code``/``status``/
+    ``status_code`` in its body or the body's ``error`` object; an explicit
+    status in the message (``_IN_BAND_STATUS_RE``, leftmost match in 400-599);
+    and ``type == "server_error"``, which OpenAI-compatible APIs use for 5xx
+    failures (500).
+    """
+    body = getattr(exc, "body", None)
+    body = body if isinstance(body, dict) else {}
+    error = body.get("error")
+    error = error if isinstance(error, dict) else {}
+    for value in (
+        getattr(exc, "code", None),
+        body.get("code"),
+        error.get("code"),
+        body.get("status"),
+        body.get("status_code"),
+        error.get("status"),
+        error.get("status_code"),
+    ):
+        status = _as_http_status(value)
+        if status is not None:
+            return status
+    for match in _IN_BAND_STATUS_RE.finditer(_extract_error_detail(exc).lower()):
+        status = int(next(group for group in match.groups() if group))
+        if 400 <= status <= 599:
+            return status
+    for value in (getattr(exc, "type", None), body.get("type"), error.get("type")):
+        if isinstance(value, str) and value.strip().lower() == "server_error":
+            return 500
+    return None
+
+
+def _as_http_status(value: Any) -> int | None:
+    """Return ``value`` as an HTTP status if it is an int or a three-digit string in 400-599."""
+    if isinstance(value, str):
+        text = value.strip()
+        value = int(text) if len(text) == 3 and text.isascii() and text.isdigit() else None
+    if isinstance(value, int) and not isinstance(value, bool) and 400 <= value <= 599:
+        return value
+    return None
+
+
+def _normalize_separators(text: str) -> str:
+    """Lower-case ``text`` and replace each run of ``_``, ``-`` and whitespace with one space."""
+    return _SEPARATOR_RUN_RE.sub(" ", text.lower())
+
+
 def _extract_retry_after_ms(exc: BaseException) -> int | None:
     response = getattr(exc, "response", None)
     headers = getattr(response, "headers", None)
diff --git a/backend/tests/test_llm_error_handling_middleware.py b/backend/tests/test_llm_error_handling_middleware.py
index b8d6986a0b..03191ff742 100644
--- a/backend/tests/test_llm_error_handling_middleware.py
+++ b/backend/tests/test_llm_error_handling_middleware.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import concurrent.futures
+import json
 import threading
 import time
 from collections.abc import Iterator
@@ -10,18 +11,23 @@ from email.utils import format_datetime
 from types import SimpleNamespace
 from typing import Any
 
+import httpx
+import openai
 import pytest
 from langchain.agents import create_agent
 from langchain.agents.middleware.types import ModelResponse
+from langchain_core.callbacks import AsyncCallbackHandler
 from langchain_core.language_models import BaseChatModel
 from langchain_core.messages import AIMessage, HumanMessage
 from langchain_core.outputs import ChatGeneration, ChatResult
 from langchain_core.tools import StructuredTool
+from langchain_openai import ChatOpenAI
 from langgraph.errors import GraphBubbleUp
 
 from deerflow.agents.middlewares.llm_error_handling_middleware import (
     EmptyModelResponseError,
     LLMErrorHandlingMiddleware,
+    _extract_in_band_status,
     _extract_retry_after_ms,
 )
 from deerflow.config.app_config import AppConfig, LlmCallConfig
@@ -3058,3 +3064,385 @@ def test_circuit_recovery_survives_a_backward_wall_clock_step(monkeypatch: pytes
     assert middleware._check_circuit() is False
     assert middleware._circuit_state == "half_open"
     assert middleware._circuit_probe_in_flight is True
+
+
+# ---------- Provider errors reported inside a 200 stream ----------
+#
+# A provider (or a gateway in front of one) that has already answered ``200``
+# reports a later failure as an SSE ``error`` event. The openai SDK raises it as
+# the base ``APIError``: ``body`` is the event's ``error`` object, ``code`` and
+# ``type`` are copied from it, and there is no status code and no response.
+# ``_in_stream_error`` builds exactly that object.
+
+# What OmniRoute (an OpenAI-compatible gateway) sent after committing to ``200``
+# with SSE keepalives: its combo's error, wrapping the target's own 504 frame.
+_OMNIROUTE_COMBO_504 = (
+    "openai-compatible-chat-conn:7b9b6797/glm-5.3: provider — data: "
+    '{"error":{"message":"[504]: Request exceeded OmniRoute\'s local rate-limit execution expiration '
+    "(legacy resilienceSettings.requestQueue.maxWaitMs=15000ms) for openai-compatible-chat-conn:7b9b6797/glm-5.3. "
+    "Bottleneck applies this deadline only after dispatch; it does not bound queue wait and is not an "
+    'upstream-generated timeout.","type":"server_error","code":"RATE_LIMIT_EXECUTION_TIMEOUT"}}'
+)
+_OMNIROUTE_TARGET_504 = (
+    "[504]: Request exceeded OmniRoute's local rate-limit execution expiration "
+    "(legacy resilienceSettings.requestQueue.maxWaitMs=15000ms) for openai-compatible-chat-conn:7b9b6797/glm-5.3. "
+    "Bottleneck applies this deadline only after dispatch; it does not bound queue wait and is not an "
+    "upstream-generated timeout."
+)
+_STREAM_REQUEST = httpx.Request("POST", "https://llm.invalid/v1/chat/completions")
+
+
+def _in_stream_error(message: str, **error_fields: Any) -> openai.APIError:
+    """Build the ``APIError`` the openai SDK raises for an SSE ``error`` event."""
+    return openai.APIError(message, _STREAM_REQUEST, body={"message": message, **error_fields})
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(_in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout"), id="omniroute-combo"),
+        pytest.param(_in_stream_error(_OMNIROUTE_TARGET_504, type="server_error", code="RATE_LIMIT_EXECUTION_TIMEOUT"), id="omniroute-target"),
+    ],
+)
+def test_omniroute_in_stream_504_is_transient(exc: openai.APIError) -> None:
+    assert _build_middleware()._classify_error(exc) == (True, "transient")
+
+
+@pytest.mark.parametrize(
+    "message",
+    [
+        pytest.param("[504]: upstream timed out", id="bracketed"),
+        pytest.param("upstream answered HTTP 502", id="http"),
+        pytest.param("upstream failed (HTTP 502)", id="http-in-parens"),
+        pytest.param("HTTP/1.1 502 Bad Gateway", id="http-1.1"),
+        pytest.param("HTTP/2 503 from upstream", id="http-2"),
+        pytest.param("upstream status 503", id="status-space"),
+        pytest.param("upstream status: 503", id="status-colon"),
+        pytest.param("upstream status=503", id="status-equals"),
+        pytest.param("upstream status code: 503", id="status-code"),
+        pytest.param("upstream status_code=503", id="status_code"),
+        pytest.param('upstream replied {"status": 503}', id="json-status"),
+        pytest.param('upstream replied {"status":503}', id="json-status-without-space"),
+        pytest.param('upstream replied {"statusCode": 503}', id="json-statusCode"),
+        pytest.param("gateway [999] relayed HTTP 503", id="out-of-range-match-skipped"),
+        pytest.param("error: [503] overloaded", id="bracketed-in-running-text"),
+        pytest.param("openai-compatible-chat-conn:7b9b6797/glm-5.3: provider — [504]: upstream timed out (HTTP 504)", id="omniroute-combo-text"),
+        pytest.param("upstream answered HTTP 200, then failed with [503]", id="status-200-skipped-for-later-match"),
+    ],
+)
+def test_in_stream_error_takes_its_status_from_the_message(message: str) -> None:
+    assert _build_middleware()._classify_error(_in_stream_error(message)) == (True, "transient")
+
+
+@pytest.mark.parametrize(
+    "error_fields",
+    [
+        pytest.param({"code": 503}, id="int-code"),
+        pytest.param({"code": "503"}, id="str-code"),
+        pytest.param({"code": " 503 "}, id="str-code-with-whitespace"),
+        pytest.param({"code": 503, "status": "UNAVAILABLE"}, id="google-style"),
+        pytest.param({"status": 502}, id="status"),
+        pytest.param({"status_code": 502}, id="status_code"),
+        pytest.param({"error": {"code": 503}}, id="nested-code"),
+        pytest.param({"error": {"status": 502}}, id="nested-status"),
+        pytest.param({"error": {"status_code": 502}}, id="nested-status_code"),
+        pytest.param({"type": "server_error"}, id="server-error-type"),
+        pytest.param({"type": "SERVER_ERROR"}, id="server-error-type-upper-case"),
+        pytest.param({"error": {"type": "server_error"}}, id="nested-server-error-type"),
+        pytest.param({"code": 200, "type": "server_error"}, id="code-200-skipped-for-server-error-type"),
+        pytest.param({"code": 200, "status": 503}, id="code-200-skipped-for-status"),
+        pytest.param({"code": 999, "status": 503}, id="code-999-skipped-for-status"),
+    ],
+)
+def test_in_stream_error_takes_its_status_from_structured_fields(error_fields: dict[str, Any]) -> None:
+    exc = _in_stream_error("Upstream request failed.", **error_fields)
+    assert _build_middleware()._classify_error(exc) == (True, "transient")
+
+
+def test_in_stream_non_error_status_does_not_hide_server_error_type() -> None:
+    exc = _in_stream_error("stream aborted after HTTP 200", type="server_error")
+    assert _build_middleware()._classify_error(exc) == (True, "transient")
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(_in_stream_error("stream aborted after HTTP 200"), id="message-http-200"),
+        pytest.param(_in_stream_error("Upstream request failed.", code=101), id="code-101"),
+        pytest.param(_in_stream_error("upstream redirected: status 302"), id="message-status-302"),
+    ],
+)
+def test_in_band_lookup_skips_non_error_statuses(exc: openai.APIError) -> None:
+    assert _extract_in_band_status(exc) is None
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(_in_stream_error("upstream rate-limit hit"), id="hyphenated-message"),
+        pytest.param(_in_stream_error("Upstream request failed.", code="rate_limit_exceeded"), id="snake-case-code"),
+        pytest.param(_in_stream_error("Request exceeded the local execution expiration.", code="RATE_LIMIT_EXECUTION_TIMEOUT"), id="upper-snake-case-code"),
+        pytest.param(RuntimeError("Codex API response.failed (rate_limit_exceeded): slow down"), id="codex-response-failed"),
+        pytest.param(_in_stream_error("rate-limit header malformed"), id="separator-busy-without-a-status"),
+    ],
+)
+def test_status_less_error_matches_busy_patterns_across_separators(exc: Exception) -> None:
+    assert _build_middleware()._classify_error(exc) == (True, "busy")
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(
+            _in_stream_error("The response was filtered due to the prompt triggering content management policy.", type="invalid_request_error", code="content_filter"),
+            id="content-filter",
+        ),
+        pytest.param(
+            _in_stream_error("This model's maximum context length is 131072 tokens. However, your messages resulted in 130504 tokens.", type="invalid_request_error", code="context_length_exceeded"),
+            id="context-length",
+        ),
+        pytest.param(_in_stream_error("Upstream request failed.", code=400), id="code-400"),
+        pytest.param(_in_stream_error("Upstream request failed.", code=400, type="server_error"), id="code-400-beats-server-error-type"),
+        pytest.param(_in_stream_error("upstream returned HTTP 400", type="server_error"), id="message-400-beats-server-error-type"),
+        pytest.param(_in_stream_error("upstream failed [999]"), id="out-of-range-status"),
+        pytest.param(_in_stream_error("Upstream request failed.", code=True), id="bool-code"),
+        pytest.param(_in_stream_error("upstream failed: [5040]"), id="bracketed-four-digits"),
+        pytest.param(_in_stream_error("upstream failed: [50]"), id="bracketed-two-digits"),
+        pytest.param(_in_stream_error("upstream failed: HTTP502"), id="http-without-space"),
+        pytest.param(_in_stream_error("upstream failed: https://host:5020"), id="url-port"),
+        pytest.param(_in_stream_error("upstream failed: status ok"), id="status-word"),
+        pytest.param(_in_stream_error("upstream failed: status 5030"), id="status-four-digits"),
+        pytest.param(_in_stream_error("upstream failed: status\n503"), id="status-number-on-next-line"),
+        pytest.param(_in_stream_error("upstream failed: status503"), id="status-without-separator"),
+        pytest.param(_in_stream_error("upstream failed: statuscode503"), id="statuscode-without-separator"),
+        pytest.param(_in_stream_error("upstream failed: status_code503"), id="status_code-without-separator"),
+        # Arabic-Indic and fullwidth digits for 503: int() accepts them, the status regex must not.
+        pytest.param(_in_stream_error("upstream answered HTTP \u0665\u0660\u0663"), id="http-arabic-indic-digits"),
+        pytest.param(_in_stream_error("upstream answered HTTP \uff15\uff10\uff13"), id="http-fullwidth-digits"),
+        pytest.param(_in_stream_error("upstream failed: [\u0665\u0660\u0663]"), id="bracketed-arabic-indic-digits"),
+        pytest.param(_in_stream_error("upstream status: \u0665\u0660\u0663"), id="status-arabic-indic-digits"),
+        pytest.param(_in_stream_error("upstream failed: error 504"), id="number-without-keyword"),
+        pytest.param(_in_stream_error("upstream said [503]", code=400), id="code-400-beats-message-503"),
+        pytest.param(_in_stream_error("upstream returned HTTP 400 after [503]"), id="leftmost-message-status-decides"),
+        pytest.param(_in_stream_error("invalid value at messages[502].content"), id="message-index"),
+        pytest.param(_in_stream_error("invalid tool call at tools[0][503].function"), id="nested-index"),
+        pytest.param(_in_stream_error("rate-limit header malformed", code=400), id="code-400-beats-separator-busy"),
+        pytest.param(_in_stream_error("HTTP 400: rate-limit header malformed"), id="message-400-beats-separator-busy"),
+    ],
+)
+def test_in_stream_error_without_a_retriable_signal_stays_generic(exc: openai.APIError) -> None:
+    assert _build_middleware()._classify_error(exc) == (False, "generic")
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body="upstream exploded"), id="string-body"),
+        pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body=["upstream", "exploded"]), id="list-body"),
+        pytest.param(openai.APIError("An error occurred during streaming", _STREAM_REQUEST, body=None), id="no-body"),
+        pytest.param(_in_stream_error("Upstream request failed.", error="upstream timeout"), id="string-error-field"),
+        pytest.param(_in_stream_error("Upstream request failed.", code={"value": 503}), id="dict-code"),
+        pytest.param(_in_stream_error("Upstream request failed.", status=["503"]), id="list-status"),
+        # Arabic-Indic digits for 503: int() accepts them, the status check must not.
+        pytest.param(_in_stream_error("Upstream request failed.", code="\u0665\u0660\u0663"), id="non-ascii-digit-code"),
+    ],
+)
+def test_in_stream_error_with_a_malformed_body_stays_generic(exc: openai.APIError) -> None:
+    assert _build_middleware()._classify_error(exc) == (False, "generic")
+
+
+def test_in_stream_status_parsing_stays_fast_on_a_long_message() -> None:
+    # Spaces before and after ``:``/``=`` must be separate runs in the status
+    # regex; two adjacent ``[ \t]*`` would backtrack quadratically here.
+    exc = _in_stream_error("status" + " " * 50_000 + "x")
+    started = time.perf_counter()
+    assert _build_middleware()._classify_error(exc) == (False, "generic")
+    assert time.perf_counter() - started < 1.0
+
+
+@pytest.mark.parametrize(
+    ("exc", "expected"),
+    [
+        pytest.param(
+            _in_stream_error("You exceeded your current quota, please check your plan and billing details.", type="server_error", code="insufficient_quota"),
+            (False, "quota"),
+            id="quota",
+        ),
+        pytest.param(_in_stream_error("Unauthorized", code=401), (False, "auth"), id="auth"),
+        pytest.param(_in_stream_error("Upstream request failed.", code="limit_burst_rate"), (True, "burst_rate"), id="burst-rate"),
+    ],
+)
+def test_in_stream_quota_auth_and_burst_rate_keep_precedence(exc: openai.APIError, expected: tuple[bool, str]) -> None:
+    assert _build_middleware()._classify_error(exc) == expected
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(_in_stream_error("rate limit exceeded", code=400), id="in-band-400"),
+        pytest.param(FakeError("rate limit exceeded", status_code=400), id="http-400"),
+    ],
+)
+def test_in_band_4xx_keeps_the_plain_busy_match(exc: Exception) -> None:
+    assert _build_middleware()._classify_error(exc) == (True, "busy")
+
+
+@pytest.mark.parametrize(
+    "exc",
+    [
+        pytest.param(FakeError("rate-limit exceeded", status_code=400), id="hyphenated-busy-text"),
+        pytest.param(FakeError("upstream said [503]", status_code=400), id="bracketed-status"),
+        pytest.param(FakeError("upstream said [503]", status_code=200), id="stream-status-200"),
+    ],
+)
+def test_error_with_a_status_ignores_in_band_signals(exc: FakeError) -> None:
+    assert _build_middleware()._classify_error(exc) == (False, "generic")
+
+
+def test_async_in_stream_504_retries_with_the_full_budget_then_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
+    attempts = 0
+    events: list[dict] = []
+
+    async def fake_sleep(_delay: float) -> None:
+        return None
+
+    async def fake_emit_custom_event(payload, *, writer):
+        writer(payload)
+
+    async def handler(_request) -> AIMessage:
+        nonlocal attempts
+        attempts += 1
+        if attempts < 3:
+            raise _in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout")
+        return AIMessage(content="recovered")
+
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+    monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: events.append)
+    monkeypatch.setattr("deerflow.agents.middlewares.llm_error_handling_middleware.aemit_custom_event", fake_emit_custom_event)
+
+    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))
+
+    assert isinstance(result, AIMessage)
+    assert result.content == "recovered"
+    assert attempts == 3
+    assert [(event["type"], event["reason"], event["max_attempts"]) for event in events] == [("llm_retry", "transient", 3)] * 2
+
+
+def test_sync_in_stream_504_exhausts_the_full_budget_into_the_transient_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
+    attempts = 0
+
+    def handler(_request) -> AIMessage:
+        nonlocal attempts
+        attempts += 1
+        raise _in_stream_error(_OMNIROUTE_COMBO_504, type="server_error", code="gateway_timeout")
+
+    monkeypatch.setattr("time.sleep", lambda _delay: None)
+
+    result = middleware.wrap_model_call(SimpleNamespace(), handler)
+
+    assert attempts == 3
+    assert isinstance(result, AIMessage)
+    assert "temporarily unavailable" in str(result.content)
+    assert result.additional_kwargs["deerflow_error_fallback"] is True
+    assert result.additional_kwargs["error_type"] == "APIError"
+    assert result.additional_kwargs["error_reason"] == "transient"
+    assert result.additional_kwargs["error_detail"] == _OMNIROUTE_COMBO_504
+
+
+def test_exhausted_in_stream_5xx_failures_open_the_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("time.sleep", lambda _delay: None)
+    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10, circuit_failure_threshold=2)
+    calls = 0
+
+    def handler(_request) -> AIMessage:
+        nonlocal calls
+        calls += 1
+        raise _in_stream_error("Upstream request failed.", type="server_error")
+
+    middleware.wrap_model_call(SimpleNamespace(), handler)
+    assert middleware._circuit_failure_count == 1
+    middleware.wrap_model_call(SimpleNamespace(), handler)
+    assert middleware._circuit_state == "open"
+
+    result = middleware.wrap_model_call(SimpleNamespace(), handler)
+
+    assert calls == 4
+    assert result.additional_kwargs["error_reason"] == "circuit_open"
+
+
+@pytest.mark.anyio
+async def test_in_stream_error_from_a_real_openai_stream_is_retried() -> None:
+    """End to end: the first stream answers ``200`` and then carries the error event."""
+    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
+    requests: list[httpx.Request] = []
+
+    def chunk(delta: dict[str, Any], finish_reason: str | None = None) -> dict[str, Any]:
+        return {"id": "chatcmpl-retry", "object": "chat.completion.chunk", "created": 0, "model": "glm-5.3", "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}]}
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        if len(requests) == 1:
+            error = {"error": {"message": _OMNIROUTE_COMBO_504, "type": "server_error", "code": "gateway_timeout"}}
+            body = f": keepalive\n\ndata: {json.dumps(error)}\n\n"
+        else:
+            body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "recovered"}), chunk({}, "stop"))) + "data: [DONE]\n\n"
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
+        model = ChatOpenAI(model="glm-5.3", api_key="offline-test-key", base_url="https://llm.invalid/v1", streaming=True, max_retries=0, http_async_client=client)
+
+        async def handler(_request) -> AIMessage:
+            return await model.ainvoke([HumanMessage(content="hello")])
+
+        result = await middleware.awrap_model_call(SimpleNamespace(), handler)
+
+    assert len(requests) == 2
+    assert isinstance(result, AIMessage)
+    assert result.content == "recovered"
+    assert "deerflow_error_fallback" not in result.additional_kwargs
+
+
+@pytest.mark.anyio
+async def test_in_stream_error_after_partial_output_is_retried_from_the_start() -> None:
+    """An ``error`` event after content chunks retries the whole call.
+
+    The result holds only the retried answer, but the tokens the first attempt
+    already streamed reached the callbacks (which feed a live stream) and are
+    not taken back. Mid-stream ``ReadError`` retries behave the same way; this
+    pins that accepted outcome.
+    """
+    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
+    requests: list[httpx.Request] = []
+    streamed: list[str] = []
+
+    class RecordTokens(AsyncCallbackHandler):
+        async def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
+            streamed.append(token)
+
+    def chunk(delta: dict[str, Any], finish_reason: str | None = None) -> dict[str, Any]:
+        return {"id": "chatcmpl-retry", "object": "chat.completion.chunk", "created": 0, "model": "glm-5.3", "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}]}
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        if len(requests) == 1:
+            error = {"error": {"message": "[504]: upstream timed out", "type": "server_error", "code": "gateway_timeout"}}
+            body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "partial answ"}), error))
+        else:
+            body = "".join(f"data: {json.dumps(event)}\n\n" for event in (chunk({"role": "assistant", "content": "recovered"}), chunk({}, "stop"))) + "data: [DONE]\n\n"
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
+
+    async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
+        model = ChatOpenAI(model="glm-5.3", api_key="offline-test-key", base_url="https://llm.invalid/v1", streaming=True, max_retries=0, http_async_client=client)
+
+        async def handler(_request) -> AIMessage:
+            return await model.ainvoke([HumanMessage(content="hello")], config={"callbacks": [RecordTokens()]})
+
+        result = await middleware.awrap_model_call(SimpleNamespace(), handler)
+
+    assert len(requests) == 2
+    assert isinstance(result, AIMessage)
+    assert result.content == "recovered"
+    assert "deerflow_error_fallback" not in result.additional_kwargs
+    assert "".join(streamed) == "partial answrecovered"