deer-flow/backend/tests/test_llm_error_handling_middleware.py

from __future__ import annotations

import asyncio
import concurrent.futures
import json
import threading
import time
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
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
from deerflow.config.sandbox_config import SandboxConfig


@pytest.fixture(autouse=True)
def _reset_process_limiter() -> Iterator[None]:
    """Reset the module-global limiter + cap-resolved flag between tests.

    The limiter is a process singleton shared across all middleware instances,
    so cap / in-flight state from one test would otherwise bleed into the next.
    The cap is startup-only (frozen at first ``__init__``), so ``_CAP_RESOLVED``
    must be reset too - otherwise the first test to construct a middleware
    would freeze the cap for every later test regardless of what cap they ask
    for.
    """
    from deerflow.agents.middlewares import llm_error_handling_middleware as mod

    mod._PROCESS_LIMITER = None
    mod._CAP_RESOLVED = False
    yield
    mod._PROCESS_LIMITER = None
    mod._CAP_RESOLVED = False


@pytest.fixture(autouse=True)
def _guard_shared_exception_name() -> Iterator[None]:
    """Fail the test that renames the shared ``FakeError`` class.

    Several tests below drive classification by name (``type(exc).__name__``),
    so they set ``exc.__class__.__name__``. That mutates the *class*, not the
    instance: doing it to ``FakeError`` renames it for the rest of the session
    and every later test asserting ``error_type == "FakeError"`` fails. Use a
    dedicated stand-in (``_ReadError`` and friends) instead. Without this guard
    the damage only surfaces in whichever test happens to run next.
    """
    original_name = FakeError.__name__
    yield
    try:
        assert FakeError.__name__ == "FakeError", "this test renamed the shared FakeError class; rename a dedicated stand-in such as _ReadError instead of exc.__class__ on FakeError"
    finally:
        FakeError.__name__ = original_name


def _make_app_config() -> AppConfig:
    """Minimal AppConfig for middleware tests; circuit_breaker uses defaults."""
    return AppConfig(sandbox=SandboxConfig(use="test"))


class FakeError(Exception):
    def __init__(
        self,
        message: str,
        *,
        status_code: int | None = None,
        code: str | None = None,
        headers: dict[str, str] | None = None,
        body: dict | None = None,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.code = code
        self.body = body
        self.response = SimpleNamespace(status_code=status_code, headers=headers or {}) if status_code is not None or headers else None


# Middleware-level attribute -> ``LlmCallConfig`` field. ``llm_call`` knobs are
# routed through ``AppConfig`` so ``__init__`` resolves the cap on the process
# limiter (startup-only: the first construction freezes it). Circuit-breaker
# knobs are read per-call from ``self`` (not via the limiter), so setattr after
# ``__init__`` still works for them.
_LLM_CALL_ATTR_MAP: dict[str, str] = {
    "max_concurrent_llm_calls": "max_concurrent_calls",
    "retry_max_attempts": "retry_max_attempts",
    "retry_base_delay_ms": "retry_base_delay_ms",
    "retry_cap_delay_ms": "retry_cap_delay_ms",
    "burst_retry_base_delay_ms": "burst_retry_base_delay_ms",
}


def _build_middleware(**attrs: int) -> LLMErrorHandlingMiddleware:
    llm_call_fields = {_LLM_CALL_ATTR_MAP[key]: value for key, value in attrs.items() if key in _LLM_CALL_ATTR_MAP}
    app_config = AppConfig(
        sandbox=SandboxConfig(use="test"),
        llm_call=LlmCallConfig(**llm_call_fields),
    )
    middleware = LLMErrorHandlingMiddleware(app_config=app_config)
    for key, value in attrs.items():
        if key not in _LLM_CALL_ATTR_MAP:
            setattr(middleware, key, value)
    return middleware


def test_async_model_call_retries_busy_provider_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=25, retry_cap_delay_ms=25)
    attempts = 0
    waits: list[float] = []
    events: list[dict] = []
    dispatched_events: list[dict] = []

    async def fake_sleep(delay: float) -> None:
        waits.append(delay)

    def fake_writer():
        return events.append

    async def fake_emit_custom_event(payload, *, writer):
        writer(payload)
        dispatched_events.append(payload)

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise FakeError("当前服务集群负载较高，请稍后重试，感谢您的耐心等待。 (2064)")
        return AIMessage(content="ok")

    monkeypatch.setattr("asyncio.sleep", fake_sleep)
    monkeypatch.setattr(
        "langgraph.config.get_stream_writer",
        fake_writer,
    )
    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 == "ok"
    assert attempts == 3
    assert waits == [0.025, 0.025]
    assert [event["type"] for event in events] == ["llm_retry", "llm_retry"]
    assert dispatched_events == events


def test_async_model_call_returns_user_message_for_quota_errors() -> None:
    middleware = _build_middleware(retry_max_attempts=3)

    async def handler(_request) -> AIMessage:
        raise FakeError(
            "insufficient_quota: account balance is empty",
            status_code=429,
            code="insufficient_quota",
        )

    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))

    assert isinstance(result, AIMessage)
    assert "out of quota" in str(result.content)
    assert result.additional_kwargs["deerflow_error_fallback"] is True
    assert result.additional_kwargs["error_reason"] == "quota"
    assert result.additional_kwargs["error_type"] == "FakeError"


def test_async_model_call_marks_transient_retry_exhaustion_as_error_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=25, retry_cap_delay_ms=25)

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        raise FakeError("Connection error.", status_code=503)

    monkeypatch.setattr("asyncio.sleep", fake_sleep)

    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))

    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_reason"] == "transient"
    assert result.additional_kwargs["error_detail"] == "Connection error."


@pytest.mark.parametrize("stale_outcome", ["success", "failure", "non_retriable", "graph_bubble_up"])
def test_async_stale_completion_cannot_settle_an_owned_half_open_probe(stale_outcome: str) -> None:
    async def run() -> None:
        middleware = _build_middleware(retry_max_attempts=1)
        middleware.circuit_failure_threshold = 1
        middleware.circuit_recovery_timeout_sec = 0
        old_entered = asyncio.Event()
        release_old = asyncio.Event()
        probe_entered = asyncio.Event()
        release_probe = asyncio.Event()
        provider_calls: list[str] = []

        async def handler(request) -> AIMessage:
            name = request.name
            provider_calls.append(name)
            if name == "old":
                old_entered.set()
                await release_old.wait()
                if stale_outcome == "failure":
                    raise FakeError("old request failed", status_code=503)
                if stale_outcome == "non_retriable":
                    raise FakeError("insufficient_quota", status_code=429, code="insufficient_quota")
                if stale_outcome == "graph_bubble_up":
                    raise GraphBubbleUp()
            elif name == "open":
                raise FakeError("provider unavailable", status_code=503)
            elif name == "probe":
                probe_entered.set()
                await release_probe.wait()
            return AIMessage(content=f"{name} success")

        old_task = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(name="old"), handler))
        probe_task: asyncio.Task | None = None
        try:
            await asyncio.wait_for(old_entered.wait(), timeout=1)
            await middleware.awrap_model_call(SimpleNamespace(name="open"), handler)
            assert middleware._circuit_state == "open"

            probe_task = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(name="probe"), handler))
            await asyncio.wait_for(probe_entered.wait(), timeout=1)
            assert middleware._circuit_state == "half_open"
            assert middleware._circuit_probe_in_flight is True

            release_old.set()
            if stale_outcome == "graph_bubble_up":
                with pytest.raises(GraphBubbleUp):
                    await asyncio.wait_for(old_task, timeout=1)
            else:
                await asyncio.wait_for(old_task, timeout=1)

            assert middleware._circuit_state == "half_open"
            assert middleware._circuit_probe_in_flight is True
            calls_before_extra = list(provider_calls)
            extra = await middleware.awrap_model_call(SimpleNamespace(name="extra"), handler)
            assert provider_calls == calls_before_extra
            assert extra.additional_kwargs["error_type"] == "CircuitBreakerOpen"
        finally:
            release_old.set()
            release_probe.set()
            if not old_task.done():
                await old_task
            if probe_task is not None:
                await probe_task

    asyncio.run(run())


@pytest.mark.parametrize("stale_outcome", ["success", "failure", "non_retriable", "graph_bubble_up"])
def test_sync_stale_completion_cannot_settle_an_owned_half_open_probe(stale_outcome: str) -> None:
    middleware = _build_middleware(retry_max_attempts=1)
    middleware.circuit_failure_threshold = 1
    middleware.circuit_recovery_timeout_sec = 0
    old_entered = threading.Event()
    release_old = threading.Event()
    probe_entered = threading.Event()
    release_probe = threading.Event()
    provider_calls: list[str] = []
    provider_lock = threading.Lock()
    failures: list[BaseException] = []

    def handler(request) -> AIMessage:
        name = request.name
        with provider_lock:
            provider_calls.append(name)
        if name == "old":
            old_entered.set()
            assert release_old.wait(1)
            if stale_outcome == "failure":
                raise FakeError("old request failed", status_code=503)
            if stale_outcome == "non_retriable":
                raise FakeError("insufficient_quota", status_code=429, code="insufficient_quota")
            if stale_outcome == "graph_bubble_up":
                raise GraphBubbleUp()
        elif name == "open":
            raise FakeError("provider unavailable", status_code=503)
        elif name == "probe":
            probe_entered.set()
            assert release_probe.wait(1)
        return AIMessage(content=f"{name} success")

    def call(name: str) -> None:
        try:
            middleware.wrap_model_call(SimpleNamespace(name=name), handler)
        except BaseException as exc:
            failures.append(exc)

    old_thread = threading.Thread(target=call, args=("old",), daemon=True)
    probe_thread = threading.Thread(target=call, args=("probe",), daemon=True)
    try:
        old_thread.start()
        assert old_entered.wait(1)
        middleware.wrap_model_call(SimpleNamespace(name="open"), handler)
        assert middleware._circuit_state == "open"

        probe_thread.start()
        assert probe_entered.wait(1)
        assert middleware._circuit_state == "half_open"
        assert middleware._circuit_probe_in_flight is True

        release_old.set()
        old_thread.join(1)
        assert not old_thread.is_alive()
        if stale_outcome == "graph_bubble_up":
            assert len(failures) == 1
            assert isinstance(failures[0], GraphBubbleUp)
        else:
            assert not failures

        assert middleware._circuit_state == "half_open"
        assert middleware._circuit_probe_in_flight is True
        with provider_lock:
            calls_before_extra = list(provider_calls)
        extra = middleware.wrap_model_call(SimpleNamespace(name="extra"), handler)
        with provider_lock:
            assert provider_calls == calls_before_extra
        assert extra.additional_kwargs["error_type"] == "CircuitBreakerOpen"
    finally:
        release_old.set()
        release_probe.set()
        old_thread.join(1)
        probe_thread.join(1)
    if stale_outcome != "graph_bubble_up":
        assert not failures


def test_sync_model_call_uses_retry_after_header(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10)
    waits: list[float] = []
    events: list[dict] = []
    dispatched_events: list[dict] = []
    attempts = 0

    def fake_sleep(delay: float) -> None:
        waits.append(delay)

    def fake_emit_custom_event(payload, *, writer):
        writer(payload)
        dispatched_events.append(payload)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            raise FakeError(
                "server busy",
                status_code=503,
                headers={"Retry-After": "2"},
            )
        return AIMessage(content="ok")

    monkeypatch.setattr("time.sleep", fake_sleep)
    monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: events.append)
    monkeypatch.setattr(
        "deerflow.agents.middlewares.llm_error_handling_middleware.emit_custom_event",
        fake_emit_custom_event,
    )

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert result.content == "ok"
    assert waits == [2.0]
    assert dispatched_events == events
    assert [event["type"] for event in events] == ["llm_retry"]


def test_sync_empty_stop_retries_before_response_is_returned(monkeypatch: pytest.MonkeyPatch) -> None:
    """An empty stop is retried before the failed attempt reaches graph state."""
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            return AIMessage(content="", response_metadata={"finish_reason": "stop"})
        return AIMessage(content="recovered", response_metadata={"finish_reason": "stop"})

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert result.content == "recovered"
    assert attempts == 2


def test_persistent_empty_stop_returns_explicit_error_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert attempts == 2
    assert result.additional_kwargs["deerflow_error_fallback"] is True
    assert result.additional_kwargs["error_reason"] == "empty_response"
    assert result.additional_kwargs["error_type"] == "EmptyModelResponseError"
    assert "empty response" in str(result.content).lower()


@pytest.mark.anyio
async def test_async_empty_stop_retries_once(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            return AIMessage(content="", response_metadata={"finish_reason": "stop"})
        return AIMessage(content="recovered asynchronously", response_metadata={"finish_reason": "stop"})

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert result.content == "recovered asynchronously"
    assert attempts == 2


@pytest.mark.parametrize(
    "message",
    [
        AIMessage(
            content="",
            tool_calls=[{"id": "call-1", "name": "bash", "args": {}}],
            response_metadata={"finish_reason": "tool_calls"},
        ),
        AIMessage(content="", response_metadata={"finish_reason": "length"}),
    ],
)
def test_nonempty_or_non_stop_response_is_not_classified_as_empty(message: AIMessage) -> None:
    middleware = _build_middleware(retry_max_attempts=3)
    attempts = 0

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return message

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert result is message
    assert attempts == 1


@pytest.mark.parametrize("block_type", [[], {}, ["text"], {"type": "text"}])
def test_unhashable_block_type_does_not_break_empty_response_classification(block_type: object) -> None:
    """A provider may emit ``type: []`` / ``type: {}``; the visible-text probe must skip, not raise."""
    message = AIMessage(
        content=[{"type": block_type, "text": "ignored"}, {"type": "text", "text": "visible answer"}],
        response_metadata={"finish_reason": "stop"},
    )
    middleware = _build_middleware(retry_max_attempts=3)
    attempts = 0

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return message

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert result is message
    assert attempts == 1


@pytest.mark.parametrize(
    "message",
    [
        AIMessage(content=" ", response_metadata={"finish_reason": "stop"}),
        AIMessage(content="", additional_kwargs={"reasoning_content": "thinking"}, response_metadata={"finish_reason": "stop"}),
        AIMessage(content=[{"type": "thinking", "thinking": "thinking"}], response_metadata={"finish_reason": "stop"}),
        AIMessage(content=[{"type": "reasoning", "reasoning": "thinking"}], response_metadata={"finish_reason": "stop"}),
    ],
)
def test_nonvisible_stop_response_retries_then_returns_marked_fallback(
    message: AIMessage,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return message

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2
    assert result.additional_kwargs["deerflow_error_fallback"] is True
    assert result.additional_kwargs["error_reason"] == "empty_response"


def test_empty_response_fallback_preserves_reasoning_payload(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    thinking_block = {"type": "thinking", "thinking": "internal reasoning"}
    message = AIMessage(
        content=[thinking_block],
        additional_kwargs={"reasoning_content": "provider reasoning"},
        response_metadata={"finish_reason": "stop"},
    )
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    result = middleware.wrap_model_call(SimpleNamespace(), lambda _request: message)

    assert result.content[0] == thinking_block
    assert result.content[-1]["type"] == "text"
    assert "returned an empty response" in result.content[-1]["text"]
    assert result.additional_kwargs["reasoning_content"] == "provider reasoning"
    assert result.response_metadata == message.response_metadata


def test_empty_model_response_container_retries_before_graph_state(monkeypatch: pytest.MonkeyPatch) -> None:
    """生产环境的 ModelResponse.result 结构也必须在模型边界完成判空。"""
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> ModelResponse:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            return ModelResponse(result=[AIMessage(content="", response_metadata={"finish_reason": "stop"})])
        return ModelResponse(result=[AIMessage(content="recovered", response_metadata={"finish_reason": "stop"})])

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, ModelResponse)
    assert result.result[-1].content == "recovered"
    assert attempts == 2


def test_empty_response_error_uses_one_retry_budget() -> None:
    middleware = _build_middleware(retry_max_attempts=3)

    assert middleware._max_attempts_for(EmptyModelResponseError()) == 2
    assert middleware.release_policy_parameters() == {
        "empty_response_retry_limit": 1,
        "empty_response_retry_scope": "run",
    }


def test_empty_response_retry_budget_is_shared_across_model_calls_in_one_run(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    request = SimpleNamespace(runtime=SimpleNamespace(context={"thread_id": "thread-1", "run_id": "run-1"}))
    first_attempts = 0
    second_attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def first_handler(_request) -> AIMessage:
        nonlocal first_attempts
        first_attempts += 1
        if first_attempts == 1:
            return AIMessage(content="", response_metadata={"finish_reason": "stop"})
        return AIMessage(
            content="",
            tool_calls=[{"id": "call-1", "name": "bash", "args": {}}],
            response_metadata={"finish_reason": "tool_calls"},
        )

    def second_handler(_request) -> AIMessage:
        nonlocal second_attempts
        second_attempts += 1
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    first_result = middleware.wrap_model_call(request, first_handler)
    second_result = middleware.wrap_model_call(request, second_handler)

    assert first_result.tool_calls
    assert first_attempts == 2
    assert second_attempts == 1
    assert second_result.additional_kwargs["error_reason"] == "empty_response"


def test_caller_cannot_preconsume_empty_response_retry_budget(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    request = SimpleNamespace(runtime=SimpleNamespace(context={"__empty_response_retry_consumed": True}))
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def empty_handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    result = middleware.wrap_model_call(request, empty_handler)

    assert attempts == 2
    assert result.additional_kwargs["error_reason"] == "empty_response"


@pytest.mark.anyio
async def test_async_empty_response_retry_budget_is_shared_across_model_calls_in_one_run(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    request = SimpleNamespace(runtime=SimpleNamespace(context={"thread_id": "thread-async", "run_id": "run-async"}))
    attempts = 0

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    first_result = await middleware.awrap_model_call(request, handler)
    second_result = await middleware.awrap_model_call(request, handler)

    assert attempts == 3
    assert first_result.additional_kwargs["error_reason"] == "empty_response"
    assert second_result.additional_kwargs["error_reason"] == "empty_response"


def test_empty_response_exhaustion_does_not_trip_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(
        circuit_failure_threshold=2,
        retry_max_attempts=2,
        retry_base_delay_ms=1,
        retry_cap_delay_ms=1,
    )
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def empty_handler(_request) -> AIMessage:
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    for index in range(3):
        request = SimpleNamespace(runtime=SimpleNamespace(context={"run_id": f"run-{index}"}))
        result = middleware.wrap_model_call(request, empty_handler)
        assert result.additional_kwargs["error_reason"] == "empty_response"

    healthy_calls = 0

    def healthy_handler(_request) -> AIMessage:
        nonlocal healthy_calls
        healthy_calls += 1
        return AIMessage(content="healthy")

    result = middleware.wrap_model_call(
        SimpleNamespace(runtime=SimpleNamespace(context={"run_id": "run-healthy"})),
        healthy_handler,
    )

    assert middleware._circuit_failure_count == 0
    assert middleware._circuit_state == "closed"
    assert healthy_calls == 1
    assert result.content == "healthy"


def test_empty_response_exhaustion_releases_half_open_probe(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    middleware._circuit_state = "half_open"
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def empty_handler(_request) -> AIMessage:
        return AIMessage(content="", response_metadata={"finish_reason": "stop"})

    result = middleware.wrap_model_call(
        SimpleNamespace(runtime=SimpleNamespace(context={"run_id": "half-open-run"})),
        empty_handler,
    )

    assert result.additional_kwargs["error_reason"] == "empty_response"
    assert middleware._circuit_state == "half_open"
    assert middleware._circuit_probe_in_flight is False


class _EmptyThenRecoveredGraphModel(BaseChatModel):
    call_count: int = 0

    @property
    def _llm_type(self) -> str:
        return "empty-then-recovered"

    def bind_tools(self, tools, **kwargs):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        self.call_count += 1
        content = "" if self.call_count == 1 else "recovered through graph stream"
        message = AIMessage(content=content, response_metadata={"finish_reason": "stop"})
        return ChatResult(generations=[ChatGeneration(message=message)])

    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
        return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)


@pytest.mark.anyio
async def test_empty_response_recovery_runs_through_create_agent_astream(monkeypatch: pytest.MonkeyPatch) -> None:
    """Exercise the same create_agent().astream() path used by Gateway."""
    model = _EmptyThenRecoveredGraphModel()
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1)

    async def fake_sleep(_delay: float) -> None:
        return None

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    agent = create_agent(model=model, tools=[], middleware=[middleware], context_schema=dict)
    states = [
        state
        async for state in agent.astream(
            {"messages": [HumanMessage(content="hello")]},
            stream_mode="values",
            context={"thread_id": "stream-thread", "run_id": "stream-run"},
        )
    ]

    assert model.call_count == 2
    assert states[-1]["messages"][-1].content == "recovered through graph stream"


class _EmptyRetryToolThenEmptyGraphModel(BaseChatModel):
    call_count: int = 0

    @property
    def _llm_type(self) -> str:
        return "empty-retry-tool-then-empty"

    def bind_tools(self, tools, **kwargs):
        return self

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        self.call_count += 1
        if self.call_count == 2:
            message = AIMessage(
                content="",
                tool_calls=[{"id": "call-probe", "name": "probe", "args": {}}],
                response_metadata={"finish_reason": "tool_calls"},
            )
        else:
            message = AIMessage(content="", response_metadata={"finish_reason": "stop"})
        return ChatResult(generations=[ChatGeneration(message=message)])

    async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs):
        return self._generate(messages, stop=stop, run_manager=run_manager, **kwargs)


@pytest.mark.anyio
async def test_empty_response_retry_budget_survives_real_agent_tool_loop(monkeypatch: pytest.MonkeyPatch) -> None:
    """真实工具循环中，整个 run 只能消费一次空响应重试。"""
    model = _EmptyRetryToolThenEmptyGraphModel()
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    tool_invocations: list[str] = []

    def probe() -> str:
        tool_invocations.append("probe")
        return "probe-result"

    async def fake_sleep(_delay: float) -> None:
        return None

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    tool = StructuredTool.from_function(probe, name="probe", description="Record one deterministic invocation")
    agent = create_agent(model=model, tools=[tool], middleware=[middleware], context_schema=dict)
    states = [
        state
        async for state in agent.astream(
            {"messages": [HumanMessage(content="use the probe")]},
            stream_mode="values",
            context={"thread_id": "tool-loop-thread", "run_id": "tool-loop-run"},
        )
    ]

    final_message = states[-1]["messages"][-1]
    assert model.call_count == 3
    assert tool_invocations == ["probe"]
    assert final_message.additional_kwargs["deerflow_error_fallback"] is True
    assert final_message.additional_kwargs["error_reason"] == "empty_response"


def test_sync_retry_event_preserves_langgraph_control_flow(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware()

    def interrupt_dispatch(*_args, **_kwargs):
        raise GraphBubbleUp

    monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: lambda _payload: None)
    monkeypatch.setattr(
        "deerflow.agents.middlewares.llm_error_handling_middleware.emit_custom_event",
        interrupt_dispatch,
    )

    with pytest.raises(GraphBubbleUp):
        middleware._emit_retry_event(1, 10, "busy", max_attempts=2)


@pytest.mark.anyio
async def test_async_retry_event_preserves_langgraph_control_flow(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware()

    async def interrupt_dispatch(*_args, **_kwargs):
        raise GraphBubbleUp

    monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: lambda _payload: None)
    monkeypatch.setattr(
        "deerflow.agents.middlewares.llm_error_handling_middleware.aemit_custom_event",
        interrupt_dispatch,
    )

    with pytest.raises(GraphBubbleUp):
        await middleware._aemit_retry_event(1, 10, "busy", max_attempts=2)


def test_sync_model_call_propagates_graph_bubble_up() -> None:
    middleware = _build_middleware()

    def handler(_request) -> AIMessage:
        raise GraphBubbleUp()

    with pytest.raises(GraphBubbleUp):
        middleware.wrap_model_call(SimpleNamespace(), handler)


def test_async_model_call_propagates_graph_bubble_up() -> None:
    middleware = _build_middleware()

    async def handler(_request) -> AIMessage:
        raise GraphBubbleUp()

    with pytest.raises(GraphBubbleUp):
        asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))


def test_sync_retry_event_graph_bubble_up_releases_half_open_probe(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=2)
    middleware._circuit_state = "half_open"

    def unavailable(_request) -> AIMessage:
        raise FakeError("Service unavailable", status_code=503)

    def interrupt_retry_event(*_args, **_kwargs) -> None:
        raise GraphBubbleUp()

    monkeypatch.setattr(middleware, "_emit_retry_event", interrupt_retry_event)

    with pytest.raises(GraphBubbleUp):
        middleware.wrap_model_call(SimpleNamespace(), unavailable)

    assert middleware._circuit_state == "half_open"
    assert middleware._circuit_probe_in_flight is False


@pytest.mark.anyio
async def test_async_retry_event_graph_bubble_up_releases_half_open_probe(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=2)
    middleware._circuit_state = "half_open"

    async def unavailable(_request) -> AIMessage:
        raise FakeError("Service unavailable", status_code=503)

    async def interrupt_retry_event(*_args, **_kwargs) -> None:
        raise GraphBubbleUp()

    monkeypatch.setattr(middleware, "_aemit_retry_event", interrupt_retry_event)

    with pytest.raises(GraphBubbleUp):
        await middleware.awrap_model_call(SimpleNamespace(), unavailable)

    assert middleware._circuit_state == "half_open"
    assert middleware._circuit_probe_in_flight is False


@pytest.mark.anyio
@pytest.mark.parametrize("cancel_during", ["provider", "backoff", "queue"])
async def test_cancelled_recovery_probe_allows_next_model_call(cancel_during: str, monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=1, circuit_failure_threshold=1, circuit_recovery_timeout_sec=0, max_concurrent_llm_calls=1)

    async def unavailable(_request) -> AIMessage:
        raise FakeError("Service unavailable", status_code=503)

    await middleware.awrap_model_call(SimpleNamespace(), unavailable)
    middleware.retry_max_attempts = 2
    entered = asyncio.Event()

    async def retry_sleep(_delay: float) -> None:
        entered.set()
        await asyncio.Event().wait()

    if cancel_during == "backoff":
        monkeypatch.setattr(asyncio, "sleep", retry_sleep)

    async def recovering(_request) -> AIMessage:
        assert cancel_during != "queue", "A queued request must not reach the provider"
        if cancel_during == "backoff":
            raise FakeError("Service unavailable", status_code=503)
        entered.set()
        await asyncio.Event().wait()
        return AIMessage(content="unreachable")

    occupied = asyncio.Event()
    release_slot = asyncio.Event()

    async def occupying_handler(_request) -> AIMessage:
        occupied.set()
        await release_slot.wait()
        return AIMessage(content="other call finished")

    async def run_probe():
        if cancel_during == "queue":
            entered.set()
        return await middleware.awrap_model_call(SimpleNamespace(), recovering)

    blocker = None
    probe = None
    try:
        if cancel_during == "queue":
            other_middleware = _build_middleware(max_concurrent_llm_calls=1)
            blocker = asyncio.create_task(other_middleware.awrap_model_call(SimpleNamespace(), occupying_handler))
            await asyncio.wait_for(occupied.wait(), timeout=1)
        probe = asyncio.create_task(run_probe())
        await asyncio.wait_for(entered.wait(), timeout=1)
        probe.cancel("user stopped recovery")
        with pytest.raises(asyncio.CancelledError, match="user stopped recovery"):
            await probe
    finally:
        if probe is not None:
            probe.cancel()
            await asyncio.gather(probe, return_exceptions=True)
        release_slot.set()
        if blocker is not None:
            await asyncio.wait_for(blocker, timeout=1)

    async def healthy(_request) -> AIMessage:
        return AIMessage(content="recovered")

    result = await asyncio.wait_for(middleware.awrap_model_call(SimpleNamespace(), healthy), timeout=1)
    assert result.content == "recovered"


@pytest.mark.anyio
async def test_cancelling_older_call_does_not_release_another_recovery_probe() -> None:
    middleware = _build_middleware(retry_max_attempts=1, circuit_failure_threshold=1, circuit_recovery_timeout_sec=0)
    old_entered = asyncio.Event()
    probe_entered = asyncio.Event()
    finish_probe = asyncio.Event()

    async def old_handler(_request) -> AIMessage:
        old_entered.set()
        await asyncio.Event().wait()
        return AIMessage(content="unreachable")

    async def unavailable(_request) -> AIMessage:
        raise FakeError("Service unavailable", status_code=503)

    async def recovering(_request) -> AIMessage:
        probe_entered.set()
        await finish_probe.wait()
        return AIMessage(content="recovered")

    async def extra_handler(_request) -> AIMessage:
        pytest.fail("Only the existing recovery probe may reach the provider")

    old_call = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), old_handler))
    probe = None
    try:
        await asyncio.wait_for(old_entered.wait(), timeout=1)
        await middleware.awrap_model_call(SimpleNamespace(), unavailable)
        probe = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), recovering))
        await asyncio.wait_for(probe_entered.wait(), timeout=1)

        old_call.cancel()
        with pytest.raises(asyncio.CancelledError):
            await old_call

        blocked = await middleware.awrap_model_call(SimpleNamespace(), extra_handler)
        assert blocked.additional_kwargs["error_type"] == "CircuitBreakerOpen"
        finish_probe.set()
        result = await probe
        assert result.content == "recovered"
    finally:
        tasks = [old_call] if probe is None else [old_call, probe]
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)


def test_circuit_half_open_graph_bubble_up_resets_probe() -> None:
    """Verify that GraphBubbleUp in half_open state resets probe_in_flight."""
    middleware = _build_middleware()

    middleware._circuit_state = "half_open"
    middleware._circuit_probe_in_flight = False

    def handler(_request) -> AIMessage:
        raise GraphBubbleUp()

    with pytest.raises(GraphBubbleUp):
        middleware.wrap_model_call(SimpleNamespace(), handler)

    assert middleware._circuit_probe_in_flight is False
    assert middleware._circuit_state == "half_open"


@pytest.mark.anyio
async def test_async_circuit_half_open_graph_bubble_up_resets_probe() -> None:
    """Verify that GraphBubbleUp in half_open state resets probe_in_flight (async version)."""
    middleware = _build_middleware()

    middleware._circuit_state = "half_open"
    middleware._circuit_probe_in_flight = False

    async def handler(_request) -> AIMessage:
        raise GraphBubbleUp()

    with pytest.raises(GraphBubbleUp):
        await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert middleware._circuit_probe_in_flight is False
    assert middleware._circuit_state == "half_open"


def test_circuit_half_open_non_retriable_error_resets_probe() -> None:
    """A non-retriable error during a half-open probe must release the probe.

    Regression: the non-retriable branch neither recorded a failure (correct —
    business errors like quota/auth must not trip the breaker) nor reset
    ``_circuit_probe_in_flight``. So one non-retriable probe left the circuit
    stuck at half_open with probe_in_flight=True, and every subsequent call
    fast-failed forever because no later call could ever run the handler to
    reach ``_record_success`` / ``_record_failure``.
    """
    middleware = _build_middleware()

    middleware._circuit_state = "half_open"
    middleware._circuit_probe_in_flight = False

    def handler(_request) -> AIMessage:
        raise FakeError("insufficient_quota", status_code=429, code="insufficient_quota")

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    # Non-retriable errors still surface a graceful fallback (not a raise) and
    # must NOT trip the breaker.
    assert isinstance(result, AIMessage)
    assert middleware._circuit_state == "half_open"
    # The probe was released, so the real gate re-admits the next probe instead
    # of fast-failing forever.
    assert middleware._circuit_probe_in_flight is False
    assert middleware._check_circuit() is False
    assert middleware._circuit_probe_in_flight is True


@pytest.mark.anyio
async def test_async_circuit_half_open_non_retriable_error_resets_probe() -> None:
    """Async mirror: a non-retriable error during a half-open probe releases it."""
    middleware = _build_middleware()

    middleware._circuit_state = "half_open"
    middleware._circuit_probe_in_flight = False

    async def handler(_request) -> AIMessage:
        raise FakeError("insufficient_quota", status_code=429, code="insufficient_quota")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert middleware._circuit_state == "half_open"
    assert middleware._circuit_probe_in_flight is False
    assert middleware._check_circuit() is False
    assert middleware._circuit_probe_in_flight is True


# ---------- Circuit Breaker Tests ----------


def transient_failing_handler(request: Any) -> Any:
    raise FakeError("Server Error", status_code=502)  # Used for transient error


def quota_failing_handler(request: Any) -> Any:
    raise FakeError("Quota exceeded", body={"error": {"code": "insufficient_quota"}})  # Used for quota error


def success_handler(request: Any) -> Any:
    return AIMessage(content="Success")


def mock_classify_retriable(exc: BaseException) -> tuple[bool, str]:
    return True, "transient"


def mock_classify_non_retriable(exc: BaseException) -> tuple[bool, str]:
    return False, "quota"


def test_circuit_breaker_trips_and_recovers(monkeypatch: pytest.MonkeyPatch) -> None:
    """Verify that circuit breaker trips, fast fails, correctly transitions to Half-Open, and recovers or re-opens."""

    # Mock time.sleep to avoid slow tests during retry loops (Speed up from ~4s to 0.1s)
    waits: list[float] = []
    monkeypatch.setattr("time.sleep", lambda d: waits.append(d))

    # Drive the breaker's recovery timing through its injectable monotonic clock
    # so the test can time-travel deterministically. The breaker measures the
    # recovery window on this clock (not wall time), so advancing it is what
    # moves the open->half_open transition.
    current_time = 1000.0

    middleware = _build_middleware(circuit_failure_threshold=3, circuit_recovery_timeout_sec=10)
    middleware._circuit_clock = lambda: current_time
    monkeypatch.setattr(middleware, "_classify_error", mock_classify_retriable)

    request: Any = {"messages": []}

    # --- 0. Test initial state & Success ---
    # Success handler does not increase count. If it's already 0, it stays 0.
    middleware.wrap_model_call(request, success_handler)
    assert middleware._circuit_failure_count == 0
    assert middleware._check_circuit() is False

    # --- 1. Trip the circuit ---
    # Fails 3 overall calls. Threshold (3) is reached.
    middleware.wrap_model_call(request, transient_failing_handler)
    assert middleware._circuit_failure_count == 1
    middleware.wrap_model_call(request, transient_failing_handler)
    assert middleware._circuit_failure_count == 2
    middleware.wrap_model_call(request, transient_failing_handler)
    assert middleware._circuit_failure_count == 3
    assert middleware._check_circuit() is True  # Circuit is OPEN

    # --- 2. Fast Fail ---
    # 2nd call: fast fail immediately without calling handler.
    # Count should not increase during OPEN state.
    result = middleware.wrap_model_call(request, success_handler)
    assert result.content == middleware._build_circuit_breaker_message()
    assert middleware._circuit_failure_count == 3

    # --- 3. Half-Open -> Fail -> Re-Open ---
    # Time travel 11 seconds (timeout is 10s). Current time becomes 1011.0
    current_time += 11.0

    # Verify that the timeout was set EXACTLY relative to current_time + timeout_sec
    assert middleware._circuit_open_until == current_time - 11.0 + middleware.circuit_recovery_timeout_sec

    # Fails again! The request will go through the 3-attempt retry loop again.
    middleware.wrap_model_call(request, transient_failing_handler)
    assert middleware._circuit_failure_count == middleware.circuit_failure_threshold
    assert middleware._circuit_state == "open"  # Re-OPENed

    # --- 4. Half-Open -> Success -> Reset ---
    # Time travel another 11 seconds
    current_time += 11.0

    # Succeeds this time! Should completely reset.
    result = middleware.wrap_model_call(request, success_handler)
    assert result.content == "Success"
    assert middleware._circuit_failure_count == 0  # Fully RESET!
    assert middleware._check_circuit() is False


def test_circuit_breaker_does_not_trip_on_non_retriable_errors(monkeypatch: pytest.MonkeyPatch) -> None:
    """Verify that circuit breaker ignores business errors like Quota or Auth."""
    waits: list[float] = []
    monkeypatch.setattr("time.sleep", lambda d: waits.append(d))

    middleware = _build_middleware(circuit_failure_threshold=3)
    monkeypatch.setattr(middleware, "_classify_error", mock_classify_non_retriable)

    request: Any = {"messages": []}

    for _ in range(3):
        middleware.wrap_model_call(request, quota_failing_handler)

    assert middleware._circuit_failure_count == 0
    assert middleware._check_circuit() is False


# ---------- ReadError / RemoteProtocolError retriable classification ----------


class _ReadError(Exception):
    """Local stand-in for httpx.ReadError — same class name, no httpx dependency."""


class _RemoteProtocolError(Exception):
    """Local stand-in for httpx.RemoteProtocolError — same class name, no httpx dependency."""


_ReadError.__name__ = "ReadError"
_RemoteProtocolError.__name__ = "RemoteProtocolError"


def test_classify_error_read_error_is_retriable() -> None:
    middleware = _build_middleware()
    exc = _ReadError("Connection dropped mid-stream")
    exc.__class__.__name__ = "ReadError"
    retriable, reason = middleware._classify_error(exc)
    assert retriable is True
    assert reason == "transient"


def test_classify_error_remote_protocol_error_is_retriable() -> None:
    middleware = _build_middleware()
    exc = _RemoteProtocolError("Server closed connection unexpectedly")
    exc.__class__.__name__ = "RemoteProtocolError"
    retriable, reason = middleware._classify_error(exc)
    assert retriable is True
    assert reason == "transient"


def test_sync_read_error_triggers_retry_loop(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
    attempts = 0
    waits: list[float] = []
    monkeypatch.setattr("time.sleep", lambda d: waits.append(d))

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise _ReadError("Connection dropped mid-stream")

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    # ReadError is a generic connection drop, not a chunk-gap timeout, so
    # it must fall back to the legacy transient copy rather than the
    # specialized "split the work into smaller steps" guidance (#3195 CR).
    assert "temporarily unavailable" in result.content
    assert "streaming response was interrupted" not in result.content
    assert attempts == 3  # exhausted all retries
    assert len(waits) == 2  # slept between attempts 1→2 and 2→3


@pytest.mark.anyio
async def test_async_read_error_triggers_retry_loop(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
    attempts = 0
    waits: list[float] = []

    async def fake_sleep(d: float) -> None:
        waits.append(d)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise _ReadError("Connection dropped mid-stream")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    # ReadError is a generic connection drop, not a chunk-gap timeout, so
    # it must fall back to the legacy transient copy rather than the
    # specialized "split the work into smaller steps" guidance (#3195 CR).
    assert "temporarily unavailable" in result.content
    assert "streaming response was interrupted" not in result.content
    assert attempts == 3  # exhausted all retries
    assert len(waits) == 2  # slept between attempts 1→2 and 2→3


@pytest.mark.anyio
async def test_async_circuit_breaker_trips_and_recovers(monkeypatch: pytest.MonkeyPatch) -> None:
    """Verify async version of circuit breaker correctly handles state transitions."""
    waits: list[float] = []

    async def fake_sleep(d: float) -> None:
        waits.append(d)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    # Time-travel via the breaker's injectable monotonic clock (see the sync twin).
    current_time = 1000.0

    middleware = _build_middleware(circuit_failure_threshold=3, circuit_recovery_timeout_sec=10)
    middleware._circuit_clock = lambda: current_time
    monkeypatch.setattr(middleware, "_classify_error", mock_classify_retriable)

    async def async_failing_handler(request: Any) -> Any:
        raise FakeError("Server Error", status_code=502)

    request: Any = {"messages": []}

    # --- 1. Trip the circuit ---
    # Fails 3 overall calls. Threshold (3) is reached.
    await middleware.awrap_model_call(request, async_failing_handler)
    assert middleware._circuit_failure_count == 1
    await middleware.awrap_model_call(request, async_failing_handler)
    assert middleware._circuit_failure_count == 2
    await middleware.awrap_model_call(request, async_failing_handler)
    assert middleware._circuit_failure_count == 3
    assert middleware._check_circuit() is True

    # --- 2. Fast Fail ---
    # 2nd call: fast fail immediately without calling handler
    async def async_success_handler(request: Any) -> Any:
        return AIMessage(content="Success")

    result = await middleware.awrap_model_call(request, async_success_handler)
    assert result.content == middleware._build_circuit_breaker_message()
    assert middleware._circuit_failure_count == 3  # Unchanged

    # --- 3. Half-Open -> Fail -> Re-Open ---
    # Time travel 11 seconds
    current_time += 11.0

    # Verify timeout formula
    assert middleware._circuit_open_until == current_time - 11.0 + middleware.circuit_recovery_timeout_sec

    # Fails again! The request goes through the 3-attempt retry loop.
    await middleware.awrap_model_call(request, async_failing_handler)
    assert middleware._circuit_failure_count == middleware.circuit_failure_threshold
    assert middleware._circuit_state == "open"  # Re-OPENed

    # --- 4. Half-Open -> Success -> Reset ---
    # Time travel another 11 seconds
    current_time += 11.0

    result = await middleware.awrap_model_call(request, async_success_handler)
    assert result.content == "Success"
    assert middleware._circuit_failure_count == 0  # RESET
    assert middleware._check_circuit() is False


class _StreamChunkTimeoutError(Exception):
    """Local stand-in for langchain_openai's StreamChunkTimeoutError —
    matched by class name, no langchain-openai import needed (mirrors
    how this file already stubs httpx.ReadError / RemoteProtocolError).
    """


_StreamChunkTimeoutError.__name__ = "StreamChunkTimeoutError"


class _ReadTimeoutError(Exception):
    pass


_ReadTimeoutError.__name__ = "ReadTimeout"


def test_read_timeout_is_retried_and_exhaustion_returns_marked_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise _ReadTimeoutError("no bytes received before read deadline")

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2
    assert result.additional_kwargs["error_reason"] == "transient"
    assert result.additional_kwargs["error_type"] == "ReadTimeout"


def test_empty_model_result_is_retried_before_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=1, retry_cap_delay_ms=1)
    attempts = 0
    monkeypatch.setattr("time.sleep", lambda _delay: None)

    def handler(_request) -> ModelResponse:
        nonlocal attempts
        attempts += 1
        return ModelResponse(result=[])

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2
    assert result.additional_kwargs["error_reason"] == "empty_response"


def test_classify_error_stream_chunk_timeout_is_retriable() -> None:
    """StreamChunkTimeoutError must be classified as transient/retriable."""
    middleware = _build_middleware()
    exc = _StreamChunkTimeoutError("No streaming chunk received for 120.0s (model=mimo-v2.5, chunks_received=58).")
    exc.__class__.__name__ = "StreamChunkTimeoutError"
    retriable, reason = middleware._classify_error(exc)
    assert retriable is True
    assert reason == "transient"


def test_sync_stream_chunk_timeout_retries_once(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Sync handler raising StreamChunkTimeoutError is retried exactly once —
    the per-exception override caps it at 2 total attempts (1 first call + 1
    retry) even when retry_max_attempts=3.
    Same-payload retry on a chunk-gap timeout buffers the same way upstream;
    a full 3-attempt loop would stack 6-12 minutes of dead air before
    surfacing failure. We keep one cheap reconnect for genuine transient TCP
    blips, then surface the failure so the model can re-plan on its next turn.
    """
    middleware = _build_middleware(
        retry_max_attempts=3,
        retry_base_delay_ms=10,
        retry_cap_delay_ms=10,
    )
    attempts = 0
    waits: list[float] = []
    monkeypatch.setattr("time.sleep", lambda d: waits.append(d))

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise _StreamChunkTimeoutError("No streaming chunk received for 120.0s")

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert "streaming response was interrupted" in result.content
    # Override caps StreamChunkTimeoutError at 2 attempts (1 first call + 1 retry).
    assert attempts == 2
    # Exactly one sleep between the first attempt and the single retry.
    assert len(waits) == 1


@pytest.mark.anyio
async def test_async_stream_chunk_timeout_retries_once(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Async mirror of the sync test: StreamChunkTimeoutError is capped at
    2 attempts (1 first call + 1 retry) so we don't stack 6-12 minutes of
    dead air on a same-payload buffering failure.
    """
    middleware = _build_middleware(
        retry_max_attempts=3,
        retry_base_delay_ms=10,
        retry_cap_delay_ms=10,
    )
    attempts = 0
    waits: list[float] = []

    async def fake_sleep(d: float) -> None:
        waits.append(d)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise _StreamChunkTimeoutError("No streaming chunk received for 120.0s")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert isinstance(result, AIMessage)
    assert "streaming response was interrupted" in result.content
    assert attempts == 2
    # Exactly one sleep between the first attempt and the single retry.
    assert len(waits) == 1


def test_max_attempts_for_returns_override_for_stream_chunk_timeout() -> None:
    """StreamChunkTimeoutError must use the tightened budget (2 = "keep one retry"),
    not the default of 3."""
    middleware = _build_middleware(retry_max_attempts=3)
    exc = _StreamChunkTimeoutError("upstream stalled")
    exc.__class__.__name__ = "StreamChunkTimeoutError"

    assert middleware._max_attempts_for(exc) == 2


def test_max_attempts_for_falls_back_to_default_for_unlisted_exception() -> None:
    """ReadError / RemoteProtocolError keep the full retry budget — only
    StreamChunkTimeoutError pays for stalling upstream for `stream_chunk_timeout`
    seconds per attempt, so only it gets the tighter cap.
    """
    middleware = _build_middleware(retry_max_attempts=3)

    read_err = _ReadError("conn reset")
    read_err.__class__.__name__ = "ReadError"
    proto_err = _RemoteProtocolError("peer closed")
    proto_err.__class__.__name__ = "RemoteProtocolError"

    assert middleware._max_attempts_for(read_err) == 3
    assert middleware._max_attempts_for(proto_err) == 3
    assert middleware._max_attempts_for(FakeError("boom")) == 3


def test_max_attempts_for_override_never_exceeds_user_cap() -> None:
    """If the operator lowered retry_max_attempts below the override default,
    the user-configured cap wins — overrides only ever *tighten*, never loosen.
    """
    middleware = _build_middleware(retry_max_attempts=1)
    exc = _StreamChunkTimeoutError("upstream stalled")
    exc.__class__.__name__ = "StreamChunkTimeoutError"

    assert middleware._max_attempts_for(exc) == 1


def test_user_message_for_stream_chunk_timeout_mentions_split_or_shorten() -> None:
    """When the retry budget for StreamChunkTimeoutError is exhausted, the user
    message must guide the user toward splitting / shortening the request
    instead of suggesting a generic retry. This is the actionable advice
    Reviewer B asked for in the follow-up CR (issue #3189).
    """
    middleware = _build_middleware()
    exc = _StreamChunkTimeoutError("No streaming chunk received for 120.0s")
    exc.__class__.__name__ = "StreamChunkTimeoutError"

    message = middleware._build_user_message(exc, reason="transient")

    assert "streaming response was interrupted" in message
    assert "split" in message or "shorten" in message
    # The old generic "streaming response was interrupted" wording must NOT appear here,
    # otherwise the actionable guidance is buried.
    assert "temporarily unavailable" not in message


def test_user_message_for_remote_protocol_error_uses_generic_transient_copy() -> None:
    """RemoteProtocolError is a generic connection drop that can fire on
    transient network blips with perfectly normal payloads. The
    "split the work into smaller steps" guidance only applies when the
    upstream chunk-gap watchdog fires (StreamChunkTimeoutError), so
    RemoteProtocolError must fall back to the legacy transient copy.
    Regression guard for the #3195 CR feedback.
    """
    middleware = _build_middleware()
    exc = _RemoteProtocolError("Server closed connection unexpectedly")
    exc.__class__.__name__ = "RemoteProtocolError"

    message = middleware._build_user_message(exc, reason="transient")

    assert "temporarily unavailable" in message
    assert "streaming response was interrupted" not in message


def test_user_message_for_read_error_uses_generic_transient_copy() -> None:
    """httpx.ReadError is symmetric to RemoteProtocolError: a generic
    connection drop that must NOT receive the "split the work" guidance.
    Regression guard for the #3195 CR feedback.
    """
    middleware = _build_middleware()
    exc = _ReadError("connection dropped mid-stream")

    message = middleware._build_user_message(exc, reason="transient")

    assert "temporarily unavailable" in message
    assert "streaming response was interrupted" not in message


def test_user_message_for_generic_transient_keeps_legacy_copy() -> None:
    """Generic transient errors (HTTP 503, 'cluster busy', etc.) must keep
    the original 'streaming response was interrupted' message — only stream-drop
    exceptions get the new specialized copy. This prevents regression on
    callers who already rely on the legacy wording.
    """
    middleware = _build_middleware()
    exc = FakeError("server busy", status_code=503)

    message = middleware._build_user_message(exc, reason="transient")

    assert "temporarily unavailable" in message
    assert "streaming response was interrupted" not in message


def test_user_message_for_quota_unchanged() -> None:
    """Sanity check: the quota / auth branches must remain untouched by the
    stream-drop refactor.
    """
    middleware = _build_middleware()
    exc = FakeError("insufficient_quota", status_code=429, code="insufficient_quota")

    message = middleware._build_user_message(exc, reason="quota")

    assert "out of quota" in message
    assert "streaming response was interrupted" not in message


def test_classify_error_index_error_is_retriable_transient() -> None:
    """``langchain_core.language_models.chat_models.ainvoke`` crashes with
    ``IndexError: list index out of range`` when the upstream provider
    returns ``200 OK`` with ``generations == []`` (observed against the
    Volces "coding" endpoint at ark.cn-beijing.volces.com). That's an
    upstream-payload glitch we don't want killing the entire run, so it
    must classify as retriable/transient and go through the normal
    retry/backoff path.
    """
    middleware = _build_middleware()
    exc = IndexError("list index out of range")
    retriable, reason = middleware._classify_error(exc)
    assert retriable is True
    assert reason == "transient"


def test_async_index_error_retries_then_succeeds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Empty-``generations`` payloads from the upstream provider must not
    abort the run on the first failure. Confirm that the retry loop kicks
    in and the next attempt's successful AIMessage is returned to the
    caller instead of an error fallback.
    """
    middleware = _build_middleware(retry_max_attempts=3, retry_base_delay_ms=10, retry_cap_delay_ms=10)
    attempts = 0

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts < 2:
            raise IndexError("list index out of range")
        return AIMessage(content="ok")

    monkeypatch.setattr("asyncio.sleep", fake_sleep)

    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))

    assert isinstance(result, AIMessage)
    assert result.content == "ok"
    assert attempts == 2


def test_async_index_error_exhausted_returns_user_fallback(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """If every retry hits the same empty-``generations`` IndexError, the
    middleware must still produce a user-facing fallback AIMessage (with
    ``deerflow_error_fallback=True``) instead of letting the IndexError
    propagate out of the agent loop and ending the run in ``error``
    status with no GitHub-side reply.
    """
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10)

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        raise IndexError("list index out of range")

    monkeypatch.setattr("asyncio.sleep", fake_sleep)

    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))

    assert isinstance(result, AIMessage)
    assert result.additional_kwargs["deerflow_error_fallback"] is True
    assert result.additional_kwargs["error_reason"] == "transient"
    assert result.additional_kwargs["error_type"] == "IndexError"
    assert "temporarily unavailable" in str(result.content)


def test_classify_error_empty_stream_value_error_is_retriable_transient() -> None:
    """The streaming path reports an HTTP-200-but-empty response as
    ``ValueError("No generations found in stream.")`` (raised by
    ``langchain_core...generate_from_stream``). It is the same upstream glitch
    as the non-streaming empty-``generations`` IndexError and must be retried.
    """
    from langchain_core.language_models.chat_models import generate_from_stream

    middleware = _build_middleware()
    with pytest.raises(ValueError) as excinfo:
        generate_from_stream(iter(()))
    retriable, reason = middleware._classify_error(excinfo.value)
    assert retriable is True
    assert reason == "transient"


def test_classify_error_unrelated_value_error_stays_generic() -> None:
    middleware = _build_middleware()
    retriable, reason = middleware._classify_error(ValueError("invalid literal for int()"))
    assert retriable is False
    assert reason == "generic"


def test_async_empty_stream_retries_with_configured_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

    async def fake_sleep(_delay: float) -> None:
        return None

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise ValueError("No generations found in stream.")
        return AIMessage(content="ok")

    monkeypatch.setattr("asyncio.sleep", fake_sleep)

    result = asyncio.run(middleware.awrap_model_call(SimpleNamespace(), handler))

    assert isinstance(result, AIMessage)
    assert result.content == "ok"
    assert attempts == 3


def test_sync_empty_stream_exhausted_uses_configured_attempts(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=10, retry_cap_delay_ms=10)
    attempts = 0

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise ValueError("No generations found in stream.")

    monkeypatch.setattr("time.sleep", lambda _delay: None)

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2
    assert isinstance(result, AIMessage)
    assert result.additional_kwargs["deerflow_error_fallback"] is True
    assert result.additional_kwargs["error_reason"] == "transient"


# ---------- Process-wide concurrency limiter ----------


async def _run_concurrent(
    middleware: LLMErrorHandlingMiddleware,
    count: int,
    event: asyncio.Event,
) -> tuple[int, int]:
    """Fire ``count`` concurrent awrap_model_call tasks whose handlers park on
    ``event`` until the test releases them.

    Returns ``(max_in_flight, in_flight_at_steady_state)`` so callers can assert
    the concurrency cap. Handlers increment a shared counter on entry and
    decrement on exit, so the counter reflects only calls that got past the
    semaphore - parked-on-semaphore tasks do not count as in-flight.
    """
    in_flight = 0
    max_in_flight = 0

    async def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        in_flight += 1
        if in_flight > max_in_flight:
            max_in_flight = in_flight
        try:
            await event.wait()
        finally:
            in_flight -= 1
        return AIMessage(content="ok")

    tasks = [asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler)) for _ in range(count)]

    # Yield until the steady state is reached: the cap is hit (capped case), or
    # every task has been admitted (uncapped case).
    limit = middleware.max_concurrent_llm_calls
    target = count if limit <= 0 else min(count, limit)
    for _ in range(100):
        if max_in_flight >= target:
            break
        await asyncio.sleep(0)
    steady_in_flight = in_flight

    event.set()
    await asyncio.gather(*tasks)
    return max_in_flight, steady_in_flight


@pytest.mark.anyio
async def test_limiter_caps_concurrent_llm_calls() -> None:
    """With max_concurrent_llm_calls=2, five concurrent calls must never exceed
    two in-flight at once - the rest park on the process-global semaphore.
    """
    middleware = _build_middleware(max_concurrent_llm_calls=2)
    event = asyncio.Event()

    max_in_flight, steady_in_flight = await _run_concurrent(middleware, 5, event)

    assert max_in_flight == 2
    # At steady state exactly two handlers are parked on the event; the other
    # three are blocked on the semaphore and have not entered the handler.
    assert steady_in_flight == 2


@pytest.mark.anyio
async def test_limiter_disabled_by_default() -> None:
    """max_concurrent_llm_calls defaults to 0 (disabled): no cap, so all five
    concurrent calls run at once. Guards against the semaphore accidentally
    engaging for existing deployments that never opted in.
    """
    middleware = _build_middleware()  # default max_concurrent_llm_calls=0
    event = asyncio.Event()

    max_in_flight, _ = await _run_concurrent(middleware, 5, event)

    assert max_in_flight == 5


@pytest.mark.anyio
async def test_limiter_is_shared_across_instances() -> None:
    """The semaphore is process-global, not per-middleware-instance: two
    middlewares with the same limit share one cap, so four calls spread across
    them still never exceed two in-flight.
    """
    mw_a = _build_middleware(max_concurrent_llm_calls=2)
    mw_b = _build_middleware(max_concurrent_llm_calls=2)
    event = asyncio.Event()

    in_flight = 0
    max_in_flight = 0

    async def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        in_flight += 1
        if in_flight > max_in_flight:
            max_in_flight = in_flight
        try:
            await event.wait()
        finally:
            in_flight -= 1
        return AIMessage(content="ok")

    tasks = [
        asyncio.create_task(mw_a.awrap_model_call(SimpleNamespace(), handler)),
        asyncio.create_task(mw_b.awrap_model_call(SimpleNamespace(), handler)),
        asyncio.create_task(mw_a.awrap_model_call(SimpleNamespace(), handler)),
        asyncio.create_task(mw_b.awrap_model_call(SimpleNamespace(), handler)),
    ]
    for _ in range(100):
        if max_in_flight >= 2:
            break
        await asyncio.sleep(0)

    assert max_in_flight == 2  # shared global cap, not 2+2 per instance
    event.set()
    await asyncio.gather(*tasks)


@pytest.mark.anyio
async def test_limiter_releases_slot_during_backoff_sleep(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The semaphore wraps a single attempt only, not the retry loop. A call in
    its backoff sleep must release its slot so another caller can proceed -
    otherwise backoff would waste concurrency slots and worsen the burst we are
    trying to smooth.

    ``fake_sleep`` parks call A in backoff on ``backoff_gate`` so the test has a
    deterministic window in which to observe call B being admitted to the freed
    slot. ``asyncio.Event.wait`` / ``asyncio.wait_for`` do not route through
    ``asyncio.sleep``, so the monkeypatch does not disturb test orchestration.
    """
    middleware = _build_middleware(
        max_concurrent_llm_calls=1,
        retry_max_attempts=2,
        retry_base_delay_ms=10000,
        retry_cap_delay_ms=10000,
    )
    a_entered_backoff = asyncio.Event()
    backoff_gate = asyncio.Event()
    b_admitted = asyncio.Event()
    attempts_a = 0

    async def fake_sleep(_delay: float) -> None:
        a_entered_backoff.set()
        # Park call A in backoff until the test releases it.
        await backoff_gate.wait()

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    async def handler_a(_request) -> AIMessage:
        nonlocal attempts_a
        attempts_a += 1
        # Always fails -> call A enters backoff after attempt 1, exhausts after 2.
        raise FakeError("server busy", status_code=503)

    async def handler_b(_request) -> AIMessage:
        b_admitted.set()
        return AIMessage(content="b-ok")

    task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a))
    # Wait until call A has failed attempt 1 and parked in its backoff sleep -
    # at which point its semaphore slot has been released.
    await asyncio.wait_for(a_entered_backoff.wait(), timeout=2.0)

    task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b))
    # Call B must be admitted to the single slot while A is still parked in
    # backoff (A has not retried yet).
    await asyncio.wait_for(b_admitted.wait(), timeout=2.0)
    assert attempts_a == 1

    # Release A: it retries once more, fails again, and exhausts its budget.
    backoff_gate.set()
    result_a = await task_a
    result_b = await task_b
    assert result_b.content == "b-ok"
    assert result_a.additional_kwargs.get("deerflow_error_fallback") is True


# ---------- Decorrelated jitter ----------


def test_retry_delay_decorrelated_jitter_within_bounds() -> None:
    """First retry seeds from base: high = max(base, base*3) = base*3, so the
    delay lands in [base, base*3].
    """
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
    delay = middleware._build_retry_delay_ms(100, FakeError("server busy", status_code=503))
    assert 100 <= delay <= 300


def test_retry_delay_grows_from_previous_delay() -> None:
    """Decorrelated jitter grows off the previous delay, not a fixed schedule:
    prev=1000 -> high = max(100, 3000) = 3000 -> delay in [100, 3000].
    """
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
    delay = middleware._build_retry_delay_ms(1000, FakeError("server busy", status_code=503))
    assert 100 <= delay <= 3000


def test_retry_delay_respects_cap() -> None:
    """The cap always bounds the jittered delay, even when prev*3 would exceed it."""
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=200)
    delay = middleware._build_retry_delay_ms(1000, FakeError("server busy", status_code=503))
    assert delay <= 200


def test_retry_delay_base_equals_cap_is_deterministic() -> None:
    """When base == cap the jittered delay collapses to exactly cap regardless
    of the RNG draw - this is what keeps the fast/retry-budget tests stable.
    """
    middleware = _build_middleware(retry_base_delay_ms=25, retry_cap_delay_ms=25)
    for _ in range(20):
        delay = middleware._build_retry_delay_ms(25, FakeError("server busy", status_code=503))
        assert delay == 25


def test_retry_delay_honors_retry_after_without_jitter() -> None:
    """An explicit Retry-After is honored verbatim - the server said exactly
    when to come back, so jitter must not perturb it.
    """
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
    exc = FakeError("rate limited", status_code=429, headers={"retry-after-ms": "5000"})
    delay = middleware._build_retry_delay_ms(100, exc)
    assert delay == 5000


@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after", "Retry-After-Ms", "retry-after-ms"])
@pytest.mark.parametrize("header_value", ["1e999", "inf", "-inf", "nan", "1e15", "1e308", "99999999999999999999", "not-a-number"])
def test_unusable_retry_after_falls_back_to_jitter(header_name: str, header_value: str) -> None:
    """Unusable provider delays must fall back to bounded jitter."""
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
    exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
    assert 100 <= middleware._build_retry_delay_ms(100, exc) <= 300


@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after", "Retry-After-Ms", "retry-after-ms"])
@pytest.mark.parametrize("header_value", ["1e999", "inf", "-inf", "nan", "1e15", "1e308", "99999999999999999999", "not-a-number"])
def test_extract_unusable_retry_after_ms_returns_none(header_name: str, header_value: str) -> None:
    """Mirror the delay-level cases, including nan's existing safe fallback."""
    exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
    assert _extract_retry_after_ms(exc) is None


@pytest.mark.parametrize(
    ("header_name", "header_value", "expected_ms"),
    [
        ("Retry-After", "60", 60000),
        ("retry-after-ms", "60000", 60000),
        ("Retry-After", "2.5", 2500),
        ("retry-after-ms", "2.5", 2),
        ("Retry-After", "-2", 0),
        ("retry-after-ms", "-2", 0),
        ("Retry-After", "0", 0),
        ("retry-after-ms", "0", 0),
        ("Retry-After", "86400", 86400000),
        ("retry-after-ms", "86400000", 86400000),
        ("Retry-After", "86400.001", None),
        ("retry-after-ms", "86400001", None),
    ],
)
def test_retry_after_numeric_boundaries(header_name: str, header_value: str, expected_ms: int | None) -> None:
    exc = FakeError("rate limited", status_code=429, headers={header_name: header_value})
    assert _extract_retry_after_ms(exc) == expected_ms
    middleware = _build_middleware(retry_base_delay_ms=100, retry_cap_delay_ms=10000)
    delay = middleware._build_retry_delay_ms(100, exc)
    if expected_ms is None:
        assert 100 <= delay <= 300
    else:
        assert delay == expected_ms


@pytest.mark.parametrize(("delta_seconds", "expected_ms"), [(-60, 0), (60, 60000), (86400, 86400000), (86401, None)])
def test_retry_after_date_boundaries(monkeypatch: pytest.MonkeyPatch, delta_seconds: int, expected_ms: int | None) -> None:
    now = datetime(2026, 1, 1, tzinfo=UTC)
    monkeypatch.setattr(time, "time", lambda: now.timestamp())
    raw = format_datetime(now + timedelta(seconds=delta_seconds), usegmt=True)
    exc = FakeError("rate limited", status_code=429, headers={"Retry-After": raw})
    assert _extract_retry_after_ms(exc) == expected_ms


@pytest.mark.anyio
@pytest.mark.parametrize("mode", ["sync", "async"])
@pytest.mark.parametrize("header_name", ["Retry-After", "retry-after-ms"])
@pytest.mark.parametrize("header_value", ["1e999", "nan", "1e15", "99999999999999999999", "Fri, 31 Dec 9999 23:59:59 GMT"])
async def test_retry_loops_recover_from_unusable_hints(monkeypatch: pytest.MonkeyPatch, mode: str, header_name: str, header_value: str) -> None:
    middleware = _build_middleware(retry_max_attempts=2, retry_base_delay_ms=100, retry_cap_delay_ms=300)
    attempts = 0
    waits: list[float] = []

    def fake_sleep(delay: float) -> None:
        assert 0.1 <= delay <= 0.3
        waits.append(delay)

    async def fake_async_sleep(delay: float) -> None:
        fake_sleep(delay)

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts == 1:
            raise FakeError("rate limited", status_code=429, headers={header_name: header_value})
        return AIMessage(content="recovered")

    async def async_handler(request) -> AIMessage:
        return handler(request)

    monkeypatch.setattr(time, "sleep", fake_sleep)
    monkeypatch.setattr(asyncio, "sleep", fake_async_sleep)
    if mode == "sync":
        result = middleware.wrap_model_call(SimpleNamespace(), handler)
    else:
        result = await middleware.awrap_model_call(SimpleNamespace(), async_handler)
    assert result.content == "recovered"
    assert attempts == 2
    assert len(waits) == 1


@pytest.mark.anyio
async def test_async_retry_loop_emits_jittered_delays_within_bounds(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end: the async retry loop emits decorrelated-jitter delays, each
    within [base, cap], and prev_delay threads through so the second delay can
    grow off the first.
    """
    middleware = _build_middleware(
        retry_max_attempts=3,
        retry_base_delay_ms=100,
        retry_cap_delay_ms=10000,
    )
    waits: list[float] = []

    async def fake_sleep(delay: float) -> None:
        waits.append(delay)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    attempts = 0

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        if attempts < 3:
            raise FakeError("server busy", status_code=503)
        return AIMessage(content="ok")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert result.content == "ok"
    assert attempts == 3
    assert len(waits) == 2
    for w in waits:
        assert 0.1 <= w <= 10.0


# ---------- Config wiring ----------


def test_max_concurrent_llm_calls_defaults_to_disabled() -> None:
    """With no llm_call config, the cap defaults to 0 (disabled) - existing
    deployments see no behavior change.
    """
    middleware = LLMErrorHandlingMiddleware(app_config=_make_app_config())
    assert middleware.max_concurrent_llm_calls == 0


def test_max_concurrent_llm_calls_wired_from_config() -> None:
    """llm_call.max_concurrent_calls flows through AppConfig into the middleware
    attribute that ``__init__`` applies to the process limiter via
    ``_apply_configured_cap`` (the single generation-aware cap owner).
    """
    app_config = AppConfig(
        sandbox=SandboxConfig(use="test"),
        llm_call=LlmCallConfig(max_concurrent_calls=8),
    )
    middleware = LLMErrorHandlingMiddleware(app_config=app_config)
    assert middleware.max_concurrent_llm_calls == 8


@pytest.mark.anyio
async def test_configured_cap_bounds_concurrency_end_to_end() -> None:
    """A cap set via config.yaml (not via setattr) actually bounds in-flight
    calls end-to-end: AppConfig -> middleware -> process-wide limiter.
    """
    app_config = AppConfig(
        sandbox=SandboxConfig(use="test"),
        llm_call=LlmCallConfig(max_concurrent_calls=2),
    )
    middleware = LLMErrorHandlingMiddleware(app_config=app_config)
    event = asyncio.Event()

    max_in_flight, steady_in_flight = await _run_concurrent(middleware, 5, event)

    assert max_in_flight == 2
    assert steady_in_flight == 2


# ---------- Burst-rate (limit_burst_rate) classification ----------


def test_classify_error_limit_burst_rate_by_code() -> None:
    """A 429 with error code ``limit_burst_rate`` classifies as burst_rate,
    not generic transient - so it gets the tight budget + longer backoff.
    """
    middleware = _build_middleware()
    exc = FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate")
    assert middleware._classify_error(exc) == (True, "burst_rate")


def test_classify_error_limit_burst_rate_by_message() -> None:
    """Burst-rate is also detectable from the message alone (no code field),
    matching how Volcano Engine phrases the error.
    """
    middleware = _build_middleware()
    exc = FakeError("Request rate increased too quickly. To ensure system stability, please adjust your client logic.", status_code=429)
    assert middleware._classify_error(exc) == (True, "burst_rate")


def test_classify_error_normal_429_stays_transient() -> None:
    """A non-burst 429 (generic 'too many requests') must NOT be classified as
    burst_rate - it keeps the full retry budget and normal backoff.
    """
    middleware = _build_middleware()
    exc = FakeError("Too many requests", status_code=429)
    assert middleware._classify_error(exc) == (True, "transient")


def test_classify_error_burst_takes_precedence_over_transient_status() -> None:
    """Burst-rate detection runs before the generic 429->transient mapping."""
    middleware = _build_middleware()
    exc = FakeError("limit_burst_rate triggered", status_code=429, code="limit_burst_rate")
    assert middleware._classify_error(exc) == (True, "burst_rate")


def test_max_attempts_for_burst_rate_is_tight() -> None:
    """burst_rate gets a 2-attempt budget (1 + 1 retry) even when the global
    cap is higher - retrying into the burst adds demand to the throttled slope.
    """
    middleware = _build_middleware(retry_max_attempts=3)
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    assert middleware._max_attempts_for(exc, "burst_rate") == 2


def test_max_attempts_for_burst_rate_respects_user_cap() -> None:
    """If the operator lowered retry_max_attempts below the burst budget, the
    user cap wins - overrides only ever tighten, never loosen."""
    middleware = _build_middleware(retry_max_attempts=1)
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    assert middleware._max_attempts_for(exc, "burst_rate") == 1


def test_burst_rate_delay_uses_longer_base() -> None:
    """burst_rate backoff uses burst_retry_base_delay_ms (not the normal base),
    so the single retry lands after the throttle window subsides."""
    middleware = _build_middleware(
        retry_base_delay_ms=10,
        retry_cap_delay_ms=200,
        burst_retry_base_delay_ms=100,
    )
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    # prev = burst base (100) -> high = max(100, 300) = 300 -> delay in [100, 200] (capped)
    delay = middleware._build_retry_delay_ms(100, exc, reason="burst_rate")
    assert 100 <= delay <= 200


def test_normal_transient_delay_uses_normal_base_not_burst() -> None:
    """Non-burst transient errors keep using the normal base - the burst base
    only applies to reason='burst_rate'."""
    middleware = _build_middleware(
        retry_base_delay_ms=10,
        retry_cap_delay_ms=200,
        burst_retry_base_delay_ms=100,
    )
    exc = FakeError("server busy", status_code=503)
    # prev = 10 -> high = max(10, 30) = 30 -> delay in [10, 30]
    delay = middleware._build_retry_delay_ms(10, exc, reason="transient")
    assert 10 <= delay <= 30


def test_burst_rate_delay_prefers_retry_after() -> None:
    """An explicit Retry-After is honored verbatim for burst-rate errors - the
    server said exactly when to come back, so no jitter / longer base applies."""
    middleware = _build_middleware(
        retry_base_delay_ms=10,
        retry_cap_delay_ms=200,
        burst_retry_base_delay_ms=100,
    )
    exc = FakeError(
        "rate increased too quickly",
        status_code=429,
        code="limit_burst_rate",
        headers={"retry-after-ms": "5000"},
    )
    assert middleware._build_retry_delay_ms(100, exc, reason="burst_rate") == 5000


def test_burst_rate_exhausted_returns_distinct_message() -> None:
    """When the burst retry budget is exhausted, the user-facing message names
    the burst-rate throttle rather than the generic 'temporarily unavailable'."""
    middleware = _build_middleware()
    exc = FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate")
    message = middleware._build_user_message(exc, reason="burst_rate")
    assert "burst-rate" in message
    assert "temporarily unavailable" not in message


@pytest.mark.anyio
async def test_async_burst_rate_uses_tight_budget_and_longer_base(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end: a handler raising limit_burst_rate retries at most once
    (budget 2) with a delay drawn from the burst base, then surfaces a fallback.
    """
    middleware = _build_middleware(
        retry_max_attempts=3,
        retry_base_delay_ms=10,
        retry_cap_delay_ms=200,
        burst_retry_base_delay_ms=100,
    )
    waits: list[float] = []

    async def fake_sleep(delay: float) -> None:
        waits.append(delay)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    attempts = 0

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise FakeError("Request rate increased too quickly", status_code=429, code="limit_burst_rate")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2  # tight budget: 1 first attempt + 1 retry
    assert len(waits) == 1
    # Longer burst base: delay in [burst_base, cap] = [0.1s, 0.2s]
    assert 0.1 <= waits[0] <= 0.2
    assert result.additional_kwargs.get("deerflow_error_fallback") is True
    assert result.additional_kwargs.get("error_reason") == "burst_rate"


def test_burst_rate_exhaustion_does_not_trip_circuit_breaker(monkeypatch: pytest.MonkeyPatch) -> None:
    """P2: burst-rate (limit_burst_rate) is a transient provider slope-throttle,
    not "provider down". Exhausting its retry budget must NOT count toward the
    circuit breaker - otherwise N consecutive burst failures flip the CB open
    and fast-fail ALL calls for the recovery window, the exact self-inflicted
    outage #4290 is trying to prevent. The burst reason is already distinctively
    classified by this PR, so it is the natural place to exclude it.
    """
    monkeypatch.setattr("time.sleep", lambda _d: None)
    middleware = _build_middleware(circuit_failure_threshold=3, retry_max_attempts=3)

    def handler(_request) -> AIMessage:
        raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")

    # Exceed the failure threshold with burst-rate failures: the CB must stay
    # closed (burst-rate is transient-by-design, not a provider outage).
    for _ in range(5):
        result = middleware.wrap_model_call(SimpleNamespace(), handler)
        assert result.additional_kwargs.get("error_reason") == "burst_rate"

    assert middleware._circuit_failure_count == 0
    assert middleware._circuit_state == "closed"
    assert middleware._check_circuit() is False  # still admitting calls

    # A subsequent successful call still goes through (CB never opened).
    def ok_handler(_request) -> AIMessage:
        return AIMessage(content="ok")

    result = middleware.wrap_model_call(SimpleNamespace(), ok_handler)
    assert result.content == "ok"


@pytest.mark.anyio
async def test_async_burst_rate_exhaustion_does_not_trip_circuit_breaker(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Async mirror: burst-rate exhaustion on the async path also stays out of
    the circuit breaker (the gate lives in both ``wrap_model_call`` and
    ``awrap_model_call``).
    """

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(asyncio, "sleep", _noop_sleep)
    middleware = _build_middleware(circuit_failure_threshold=3, retry_max_attempts=3)

    async def handler(_request) -> AIMessage:
        raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")

    for _ in range(5):
        result = await middleware.awrap_model_call(SimpleNamespace(), handler)
        assert result.additional_kwargs.get("error_reason") == "burst_rate"

    assert middleware._circuit_failure_count == 0
    assert middleware._circuit_state == "closed"
    assert middleware._check_circuit() is False


# ---------- Effective retry budget in retry events (review P2) ----------


def _capture_retry_events(monkeypatch: pytest.MonkeyPatch) -> list[dict]:
    """Capture the ``llm_retry`` stream events emitted by ``_emit_retry_event``.

    ``_emit_retry_event`` lazily imports ``get_stream_writer`` from
    ``langgraph.config`` on each call, so monkeypatching the module attribute is
    enough to intercept the payloads without standing up a LangGraph run context
    (the real writer raises ``RuntimeError`` outside one - which the middleware
    swallows - so the patched writer must simply return a callable).
    """
    captured: list[dict] = []

    def _writer(payload: dict) -> None:
        captured.append(payload)

    monkeypatch.setattr("langgraph.config.get_stream_writer", lambda: _writer)
    return captured


def test_burst_rate_retry_event_reports_effective_budget_sync(monkeypatch: pytest.MonkeyPatch) -> None:
    """P2: a burst-rate (limit_burst_rate) call is capped at 2 attempts, so the
    emitted ``llm_retry`` event and its user-facing message must report the
    *effective* budget (``max_attempts=2``, message ``1/2``), not the configured
    ceiling (``retry_max_attempts=3``) - otherwise the frontend promises a retry
    that never happens. Exactly two handler attempts occur (1 first + 1 retry).
    """
    monkeypatch.setattr("time.sleep", lambda _d: None)
    events = _capture_retry_events(monkeypatch)
    middleware = _build_middleware(retry_max_attempts=3)  # default global budget

    attempts = 0

    def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")

    result = middleware.wrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2  # 1 first attempt + 1 retry; budget is 2, not 3
    assert result.additional_kwargs.get("error_reason") == "burst_rate"
    assert len(events) == 1  # exactly one retry event (attempt 1 -> 2)
    assert events[0]["max_attempts"] == 2  # effective budget, not the ceiling 3
    assert events[0]["attempt"] == 1
    assert "1/2" in events[0]["message"]


@pytest.mark.anyio
async def test_burst_rate_retry_event_reports_effective_budget_async(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Async mirror of the sync P2 test: the async burst-rate path also reports
    the effective budget (``max_attempts=2``, message ``1/2``) and runs exactly
    two handler attempts.
    """

    async def _noop_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr(asyncio, "sleep", _noop_sleep)
    events = _capture_retry_events(monkeypatch)
    middleware = _build_middleware(retry_max_attempts=3)

    attempts = 0

    async def handler(_request) -> AIMessage:
        nonlocal attempts
        attempts += 1
        raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)

    assert attempts == 2
    assert result.additional_kwargs.get("error_reason") == "burst_rate"
    assert len(events) == 1
    assert events[0]["max_attempts"] == 2
    assert events[0]["attempt"] == 1
    assert "1/2" in events[0]["message"]


# ---------- Retry params config wiring ----------


def test_retry_params_default_values() -> None:
    """With no llm_call config, retry params fall back to their documented
    defaults (matching the previous hard-coded class attributes)."""
    middleware = LLMErrorHandlingMiddleware(app_config=_make_app_config())
    assert middleware.retry_max_attempts == 3
    assert middleware.retry_base_delay_ms == 1000
    assert middleware.retry_cap_delay_ms == 8000
    assert middleware.burst_retry_base_delay_ms == 5000
    assert middleware.max_concurrent_llm_calls == 0


def test_retry_params_wired_from_config() -> None:
    """All retry/backoff knobs flow from config.yaml -> AppConfig -> middleware."""
    app_config = AppConfig(
        sandbox=SandboxConfig(use="test"),
        llm_call=LlmCallConfig(
            retry_max_attempts=7,
            retry_base_delay_ms=123,
            retry_cap_delay_ms=999,
            burst_retry_base_delay_ms=777,
            max_concurrent_calls=4,
        ),
    )
    middleware = LLMErrorHandlingMiddleware(app_config=app_config)
    assert middleware.retry_max_attempts == 7
    assert middleware.retry_base_delay_ms == 123
    assert middleware.retry_cap_delay_ms == 999
    assert middleware.burst_retry_base_delay_ms == 777
    assert middleware.max_concurrent_llm_calls == 4


def test_burst_retry_base_delay_is_configurable_end_to_end() -> None:
    """A burst base set via config actually drives the burst retry delay."""
    app_config = AppConfig(
        sandbox=SandboxConfig(use="test"),
        llm_call=LlmCallConfig(burst_retry_base_delay_ms=100, retry_cap_delay_ms=200),
    )
    middleware = LLMErrorHandlingMiddleware(app_config=app_config)
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    delay = middleware._build_retry_delay_ms(100, exc, reason="burst_rate")
    assert 100 <= delay <= 200


# ---------- Process-wide limiter across call paths (review P1) ----------


def _run_on_isolated_loop(coro_factory: Any) -> concurrent.futures.Future:
    """Run ``coro_factory()`` on a freshly-created event loop in a worker thread.

    Mirrors how ``subagents/executor.py`` runs subagent calls on an isolated
    persistent loop separate from the lead agent's loop. Returns a
    ``concurrent.futures.Future`` holding the result/exception.
    """
    done: concurrent.futures.Future = concurrent.futures.Future()

    def runner() -> None:
        loop = asyncio.new_event_loop()
        try:
            asyncio.set_event_loop(loop)
            done.set_result(loop.run_until_complete(coro_factory()))
        except Exception as exc:  # propagate to the awaiting test
            done.set_exception(exc)
        finally:
            loop.close()

    threading.Thread(target=runner, daemon=True).start()
    return done


@pytest.mark.anyio
async def test_limiter_is_process_wide_across_event_loops() -> None:
    """A lead-agent call (main loop) and a subagent call (isolated loop) share
    one process-wide cap - the limiter is NOT loop-bound (unlike
    asyncio.Semaphore). With cap=1, two concurrent calls on different loops must
    never both be in-flight. Regression for the per-loop cap the reviewer found.
    """
    middleware = _build_middleware(max_concurrent_llm_calls=1)
    gate = threading.Event()  # cross-loop-safe block
    in_flight = 0
    max_in_flight = 0
    lock = threading.Lock()

    async def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        with lock:
            in_flight += 1
            max_in_flight = max(max_in_flight, in_flight)
        try:
            await asyncio.to_thread(gate.wait)
        finally:
            with lock:
                in_flight -= 1
        return AIMessage(content="ok")

    # Call A on the test (main) loop; call B on a separate event loop / thread.
    task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler))
    fut_b = _run_on_isolated_loop(lambda: middleware.awrap_model_call(SimpleNamespace(), handler))

    # Let both loops run; with cap=1 only one handler may be in-flight at a time
    # across BOTH loops.
    for _ in range(100):
        await asyncio.sleep(0)
        with lock:
            if max_in_flight >= 1:
                break
    with lock:
        assert max_in_flight == 1

    gate.set()
    result_a = await task_a
    assert result_a.content == "ok"
    result_b = await asyncio.wait_for(asyncio.wrap_future(fut_b), timeout=5)
    assert result_b.content == "ok"
    with lock:
        assert in_flight == 0  # both calls completed; no handler left in-flight


def test_limiter_caps_concurrent_sync_calls() -> None:
    """The sync graph path now acquires the limiter too (previously bypassed it
    entirely). With cap=1, two concurrent sync calls must never both be
    in-flight. Regression for the sync-wrapper bypass the reviewer found.
    """
    middleware = _build_middleware(max_concurrent_llm_calls=1)
    gate = threading.Event()
    in_flight = 0
    max_in_flight = 0
    lock = threading.Lock()

    def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        with lock:
            in_flight += 1
            max_in_flight = max(max_in_flight, in_flight)
        gate.wait()
        with lock:
            in_flight -= 1
        return AIMessage(content="ok")

    results: list[AIMessage] = []

    def run_one() -> None:
        results.append(middleware.wrap_model_call(SimpleNamespace(), handler))

    t1 = threading.Thread(target=run_one)
    t2 = threading.Thread(target=run_one)
    t1.start()
    t2.start()
    # Wait until one handler is in-flight; the other blocks on the limiter.
    for _ in range(400):
        with lock:
            if max_in_flight >= 1:
                break
        time.sleep(0.005)
    with lock:
        assert max_in_flight == 1

    gate.set()
    t1.join(timeout=5)
    t2.join(timeout=5)
    assert not t1.is_alive() and not t2.is_alive()
    assert len(results) == 2


def test_cap_is_frozen_at_first_construction_and_unchanged_by_later_instances() -> None:
    """Startup-only cap: the first ``__init__`` resolves the cap and freezes it;
    later instances - whether they would raise it, lower it, or disable it - are
    no-ops (same limiter instance, cap unchanged, in-flight permits preserved).
    Replaces the prior generation-aware in-place update with a frozen-at-startup
    controller. There is no cap mutation at runtime, so the downscale race
    (review P1 Part A: a lowered cap handing excess permits to queued waiters,
    keeping ``in_flight`` pegged at the old cap) is structurally unreachable,
    and the construction-order / config-freshness race (Part B) has nothing to
    race on.
    """
    from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter

    first_mw = _build_middleware(max_concurrent_llm_calls=1)
    limiter = _get_process_limiter()
    assert limiter is not None
    assert limiter.limit == 1

    # A holder takes the single permit so we can observe whether later instances
    # disturb in-flight state (they must not).
    gate = threading.Event()
    in_flight = 0
    max_in_flight = 0
    lock = threading.Lock()

    def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        with lock:
            in_flight += 1
            max_in_flight = max(max_in_flight, in_flight)
        gate.wait()
        with lock:
            in_flight -= 1
        return AIMessage(content="ok")

    holder = threading.Thread(target=lambda: first_mw.wrap_model_call(SimpleNamespace(), handler))
    holder.start()
    try:
        for _ in range(400):
            with lock:
                if max_in_flight >= 1:
                    break
            time.sleep(0.005)
        with lock:
            assert max_in_flight == 1
        assert limiter.in_flight == 1  # holder holds the single permit

        # Later instances try to raise (3) and to disable (0): both ignored.
        _build_middleware(max_concurrent_llm_calls=3)
        assert _get_process_limiter() is limiter  # same instance, not recreated
        assert limiter.limit == 1  # frozen; raise attempt had no effect
        _build_middleware(max_concurrent_llm_calls=0)
        assert limiter.limit == 1  # frozen; disable attempt had no effect
        assert limiter.in_flight == 1  # holder's permit untouched
    finally:
        # Always release the holder so a failing assertion doesn't leak the thread.
        gate.set()
        holder.join(timeout=5)
    assert not holder.is_alive()
    assert limiter.in_flight == 0  # permit returned, not leaked


def test_first_constructed_cap_wins_over_later_config_snapshot() -> None:
    """Startup-only cap (review P1 Part B): the cap is frozen at the FIRST
    middleware construction, so a later instance holding an OLDER config
    snapshot (higher cap) cannot restore it - regardless of construction timing
    or config freshness. This is the startup-only replacement for the prior
    generation-aware guard: there is no "newer config wins" race because no
    instance can mutate the cap at all.

    Mirrors the reviewer's reverse-construction probe: the lower (newer) cap is
    constructed first, the higher (older/stale) cap second, and the older
    snapshot must NOT raise the live cap. Three overlapping calls through the
    older-cap instance are still bounded by the frozen cap of 1 - which also
    proves the Part A invariant (a sustained queue never admits callers above
    the frozen cap).
    """
    from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter

    # "Newer" config (cap 1) constructed FIRST -> freezes the cap at 1.
    _build_middleware(max_concurrent_llm_calls=1)
    # "Older/stale" config (cap 3) constructed AFTER -> must not raise the cap.
    older_mw = _build_middleware(max_concurrent_llm_calls=3)
    limiter = _get_process_limiter()
    assert limiter is not None
    assert limiter.limit == 1  # frozen at first construction; older snapshot did not raise

    gate = threading.Event()
    in_flight = 0
    max_in_flight = 0
    lock = threading.Lock()

    def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        with lock:
            in_flight += 1
            max_in_flight = max(max_in_flight, in_flight)
        gate.wait()
        with lock:
            in_flight -= 1
        return AIMessage(content="ok")

    # 3 overlapping calls through the OLDER-cap (cap 3) instance: the live cap is
    # frozen at 1, so only one may be in-flight - the older snapshot cannot
    # restore 3 AND callers cannot be admitted above the frozen cap.
    threads = [threading.Thread(target=lambda: older_mw.wrap_model_call(SimpleNamespace(), handler)) for _ in range(3)]
    for t in threads:
        t.start()
    try:
        for _ in range(400):
            with lock:
                if max_in_flight >= 1:
                    break
            time.sleep(0.005)
        # Give the queued calls a chance to (incorrectly) exceed the frozen cap.
        time.sleep(0.05)
        with lock:
            assert max_in_flight == 1  # never exceeded the frozen cap of 1
        assert limiter.limit == 1  # still 1; older snapshot did not restore 3
    finally:
        # Always release the holder threads so a failing assertion doesn't hang.
        gate.set()
        for t in threads:
            t.join(timeout=5)
    assert all(not t.is_alive() for t in threads)
    assert limiter.in_flight == 0


@pytest.mark.anyio
async def test_frozen_cap_binds_calls_across_isolated_loop() -> None:
    """Startup-only cap (cross-loop): the cap frozen at first construction binds
    async calls across the lead loop AND an isolated subagent loop - a later
    instance (higher cap) cannot raise it, so cross-loop in-flight calls never
    exceed the frozen cap. Replaces the prior generation-aware cross-loop test.
    """
    from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter

    # First construction (cap 1) freezes the cap; a later cap=3 instance can't raise it.
    _build_middleware(max_concurrent_llm_calls=1)
    older_mw = _build_middleware(max_concurrent_llm_calls=3)
    limiter = _get_process_limiter()
    assert limiter is not None
    assert limiter.limit == 1  # frozen at 1; later cap=3 instance did not raise

    gate = threading.Event()
    counter_lock = threading.Lock()
    in_flight = 0
    max_in_flight = 0

    async def handler(_request) -> AIMessage:
        nonlocal in_flight, max_in_flight
        with counter_lock:
            in_flight += 1
            max_in_flight = max(max_in_flight, in_flight)
        try:
            await asyncio.to_thread(gate.wait)
        finally:
            with counter_lock:
                in_flight -= 1
        return AIMessage(content="ok")

    # Lead-loop call + isolated-loop call, both on the cap-3 (older) instance:
    # the frozen cap is 1, so only one may be in-flight across BOTH loops.
    task_lead = asyncio.create_task(older_mw.awrap_model_call(SimpleNamespace(), handler))
    fut_iso = _run_on_isolated_loop(lambda: older_mw.awrap_model_call(SimpleNamespace(), handler))
    try:
        for _ in range(100):
            await asyncio.sleep(0)
            with counter_lock:
                if max_in_flight >= 1:
                    break
        await asyncio.sleep(0.05)  # let the second call queue on the limiter
        with counter_lock:
            assert max_in_flight == 1  # never exceeded the frozen cap across loops
        assert limiter.limit == 1  # still 1; later cap=3 instance did not raise
    finally:
        # Release the gate so parked handlers unblock and complete, then drain
        # both tasks. return_exceptions surfaces any late failure as a value
        # rather than masking the assertion above with a raise; gate.set()
        # guarantees both complete, so no timeout/empty-except is needed.
        gate.set()
        await asyncio.gather(task_lead, asyncio.wrap_future(fut_iso), return_exceptions=True)
    with counter_lock:
        assert in_flight == 0


@pytest.mark.anyio
async def test_limiter_cancellation_does_not_leak_capacity() -> None:
    """A caller cancelled while waiting on the limiter must not leak a permit:
    after it's cancelled and the in-flight call releases, a fresh call is still
    admitted to the same cap. Regression for the cancellation-capacity-leak
    concern the reviewer raised.
    """
    middleware = _build_middleware(max_concurrent_llm_calls=1)
    gate = threading.Event()
    a_started = asyncio.Event()

    async def handler_a(_request) -> AIMessage:
        a_started.set()
        await asyncio.to_thread(gate.wait)
        return AIMessage(content="a-ok")

    async def handler_b(_request) -> AIMessage:
        return AIMessage(content="b-ok")

    # A acquires the single permit and blocks.
    task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a))
    await asyncio.wait_for(a_started.wait(), timeout=2)

    # B queues on the limiter (no permit available).
    task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b))
    await asyncio.sleep(0.05)
    assert not task_b.done()  # B is waiting, not admitted

    # Cancel B while it waits on the limiter.
    task_b.cancel()
    with pytest.raises(asyncio.CancelledError):
        await asyncio.gather(task_b)

    # Release A; its permit returns. B's cancellation must not have consumed it.
    gate.set()
    result_a = await task_a
    assert result_a.content == "a-ok"

    # A fresh call C must be admitted immediately - proving B's cancellation
    # didn't leak capacity (in_flight back to 0, not stuck at 1).
    task_c = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_b))
    result_c = await asyncio.wait_for(task_c, timeout=2)
    assert result_c.content == "b-ok"


@pytest.mark.anyio
async def test_limiter_cancellation_after_dequeue_hands_off_to_next_waiter(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """P1 #1 regression: cap=1, A holds the permit; B and C queue. A releases
    (reserving/granting B); B is cancelled in the post-dequeue / pre-reacquire
    handoff window; C must still complete WITHOUT another release - the reserved
    permit is handed off to C rather than stranded with capacity idle.

    ``test_limiter_cancellation_does_not_leak_capacity`` cancels B while it is
    still purely queued (before A releases); this test cancels B *after* it has
    been dequeued+granted but *before* it wakes, which is the window the prior
    limiter stranded the next waiter in.
    """
    from deerflow.agents.middlewares.llm_error_handling_middleware import _get_process_limiter

    middleware = _build_middleware(max_concurrent_llm_calls=1)
    a_started = asyncio.Event()
    gate = threading.Event()

    async def handler_a(_request) -> AIMessage:
        a_started.set()
        await asyncio.to_thread(gate.wait)
        return AIMessage(content="a-ok")

    async def handler_ok(_request) -> AIMessage:
        return AIMessage(content="ok")

    task_a = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_a))
    await asyncio.wait_for(a_started.wait(), timeout=2)

    # B and C queue on the limiter (no permit available).
    task_b = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_ok))
    task_c = asyncio.create_task(middleware.awrap_model_call(SimpleNamespace(), handler_ok))
    await asyncio.sleep(0.05)  # let B and C register as waiters
    assert not task_b.done() and not task_c.done()

    limiter = _get_process_limiter()
    assert limiter is not None

    # Patch the wake so the FIRST granted waiter (B) does NOT wake yet - this
    # is the post-dequeue / pre-reacquire handoff window. Later wakes (C, handed
    # off from B's cancellation) proceed via the real path. ``_wake_locked`` is
    # called under the limiter lock and must not block, so we only toggle which
    # branch runs; B's event is simply never set, leaving B parked until cancel.
    real_wake = type(limiter)._wake_locked
    state = {"first_done": False}

    def patched_wake(waiter: Any) -> bool:
        if not state["first_done"]:
            state["first_done"] = True
            return True  # pretend the loop is alive; do NOT set B's event
        return real_wake(limiter, waiter)

    monkeypatch.setattr(limiter, "_wake_locked", patched_wake)

    # Release A: reserves the permit for B (dequeues B, granted=True) but B does
    # not wake (patched). B sits in the handoff window.
    gate.set()
    result_a = await task_a
    assert result_a.content == "a-ok"
    await asyncio.sleep(0.02)  # let A's release grant B
    assert state["first_done"]  # B was dequeued+granted, never woken

    # Cancel B while it is granted-but-not-yet-awake. Its cancellation must hand
    # the reserved permit to C (the next waiter), waking C via the real path.
    task_b.cancel()
    # gather(return_exceptions=True) surfaces B's cancellation as a value so we
    # can assert on it directly, instead of a bare ``await task_b`` inside
    # pytest.raises (whose discarded result trips the "statement has no effect"
    # analyzer without adding any assertion strength here).
    b_outcomes = await asyncio.gather(task_b, return_exceptions=True)
    assert isinstance(b_outcomes[0], asyncio.CancelledError)

    # C completes WITHOUT another release - no stranded permit.
    result_c = await asyncio.wait_for(task_c, timeout=2)
    assert result_c.content == "ok"

    # No permit leaked: in_flight is back to 0 after C completes and releases.
    assert limiter.in_flight == 0


# ---------- Burst-rate first-retry jitter (review P1) ----------


def test_burst_first_retry_is_non_degenerate_with_default_config() -> None:
    """With shipped defaults (burst_base=5000, cap=8000, normal_base=1000), the
    first burst retry must be drawn from a NON-degenerate window [5000, 8000],
    not fixed at 5000ms. Regression for the deterministic-herd bug (prev was
    seeded from the 1000ms normal base, collapsing the window to a point).
    """
    middleware = _build_middleware()  # defaults
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    # prev_delay_ms=None simulates the first retry (loops now init it to None).
    delays = {middleware._build_retry_delay_ms(None, exc, reason="burst_rate") for _ in range(20)}
    assert all(5000 <= d <= 8000 for d in delays)
    assert len(delays) > 1  # non-degenerate: more than one distinct value observed


def test_burst_first_retry_uses_jitter_not_fixed_value(monkeypatch: pytest.MonkeyPatch) -> None:
    """Controlled RNG: forcing randint to an upper-window value yields that
    value (capped), proving the first burst retry is jittered rather than a
    constant 5000ms."""
    middleware = _build_middleware()  # defaults
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    monkeypatch.setattr(
        "deerflow.agents.middlewares.llm_error_handling_middleware.random.randint",
        lambda lo, hi: 7000,
    )
    assert middleware._build_retry_delay_ms(None, exc, reason="burst_rate") == 7000


@pytest.mark.anyio
async def test_async_burst_first_retry_non_degenerate_default_config(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """End-to-end on the real async wrapper with default burst config: the
    first (and only) burst retry lands at a jittered value, not fixed at 5s."""
    middleware = _build_middleware()  # defaults: burst_base=5000, cap=8000
    waits: list[float] = []

    async def fake_sleep(delay: float) -> None:
        waits.append(delay)

    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
    monkeypatch.setattr(
        "deerflow.agents.middlewares.llm_error_handling_middleware.random.randint",
        lambda lo, hi: 7000,
    )

    async def handler(_request) -> AIMessage:
        raise FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")

    result = await middleware.awrap_model_call(SimpleNamespace(), handler)
    assert len(waits) == 1
    assert waits[0] == 7.0  # 7000ms, jittered - not the fixed 5.0s
    assert result.additional_kwargs.get("error_reason") == "burst_rate"


def test_concurrent_burst_failures_get_distinct_jittered_delays() -> None:
    """De-synchronization invariant: concurrent burst-rate failures get distinct
    retry delays, so a fleet that failed together does not realign on one tick.

    Uses a seeded real RNG (not a monkeypatch) so the window computation still
    runs - on the old code the window collapsed to ``randint(5000, 5000)`` and
    every delay was 5000ms (``len(set) == 1``); the jittered window yields many
    distinct values.
    """
    import random as _random

    middleware = _build_middleware()  # defaults
    exc = FakeError("rate increased too quickly", status_code=429, code="limit_burst_rate")
    saved_state = _random.getstate()
    _random.seed(42)
    try:
        delays = [middleware._build_retry_delay_ms(None, exc, reason="burst_rate") for _ in range(50)]
    finally:
        _random.setstate(saved_state)
    assert all(5000 <= d <= 8000 for d in delays)
    assert len(set(delays)) > 1  # de-synchronized, not a single 5000ms tick


class _FakeMonotonic:
    """A manually advanced monotonic clock for deterministic recovery timing."""

    def __init__(self, start: float = 1000.0) -> None:
        self._now = start

    def __call__(self) -> float:
        return self._now

    def advance(self, seconds: float) -> None:
        self._now += seconds


def _trip_circuit_open(middleware: LLMErrorHandlingMiddleware) -> None:
    """Drive the breaker to OPEN with a configurable recovery window."""
    middleware.circuit_failure_threshold = 1
    middleware.circuit_recovery_timeout_sec = 60
    middleware._record_failure()
    assert middleware._circuit_state == "open"
    assert middleware._check_circuit() is True  # fast-fail while open


def test_circuit_recovery_is_measured_on_the_injected_monotonic_clock() -> None:
    """The open->half_open transition tracks monotonic *elapsed* time.

    The breaker stores ``_circuit_open_until = clock() + recovery_timeout_sec``
    and fast-fails while ``clock() < _circuit_open_until``. Reading the recovery
    deadline from the injected monotonic clock means it recovers after exactly
    ``recovery_timeout_sec`` of elapsed time, independent of any wall-clock move.
    """
    clock = _FakeMonotonic()
    middleware = _build_middleware(retry_max_attempts=1)
    middleware._circuit_clock = clock
    _trip_circuit_open(middleware)

    # Just before the window elapses: still open.
    clock.advance(59.9)
    assert middleware._check_circuit() is True
    assert middleware._circuit_state == "open"

    # Once the monotonic window elapses: a single probe is admitted (half_open).
    clock.advance(0.1)
    assert middleware._check_circuit() is False
    assert middleware._circuit_state == "half_open"
    assert middleware._circuit_probe_in_flight is True


def test_circuit_recovery_survives_a_backward_wall_clock_step(monkeypatch: pytest.MonkeyPatch) -> None:
    """A backward wall-clock step must not extend the fast-fail window.

    Regression for the breaker measuring recovery with wall-clock ``time.time()``:
    an NTP correction / VM resume / manual clock change that steps the wall clock
    backward while the circuit is open would hold ``now < _circuit_open_until``
    true far past ``recovery_timeout_sec``, shedding every LLM call until the wall
    clock caught back up. Monotonic timing is immune: the recovery window advances
    with real elapsed time regardless of what the wall clock does.
    """
    import deerflow.agents.middlewares.llm_error_handling_middleware as mod

    clock = _FakeMonotonic()
    middleware = _build_middleware(retry_max_attempts=1)
    middleware._circuit_clock = clock
    _trip_circuit_open(middleware)

    # Simulate a large backward wall-clock step (e.g. the host's clock is pulled
    # back by an hour) after the circuit opened. Monotonic elapsed time still
    # reaches the recovery window, so the breaker must probe on schedule.
    monkeypatch.setattr(mod.time, "time", lambda: 1_000_000.0 - 3600.0)
    clock.advance(60.0)

    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"