diff --git a/src/providers/anthropic_provider.py b/src/providers/anthropic_provider.py index f3d1b90e..df47e6e1 100644 --- a/src/providers/anthropic_provider.py +++ b/src/providers/anthropic_provider.py @@ -161,46 +161,16 @@ def _api_timeout_seconds() -> float: return DEFAULT_API_TIMEOUT_S -#: Substrings identifying a TRANSPORT-level stream drop — the connection -#: died mid-response, with no HTTP status and no model output to salvage. -#: httpx raises these as ``RemoteProtocolError``/``ReadError``; the SDK -#: surfaces them as ``APIConnectionError``. Matched on the message as well -#: as the type because the SDK wraps and re-words some of them. -_TRANSIENT_STREAM_DROP_MARKERS = ( - "peer closed connection", - "incomplete chunked read", - "server disconnected", - "connection reset", - "connection aborted", +# Mid-stream drop classification is shared with the OpenAI-compatible +# wire (src/providers/stream_retry.py) — DeepSeek trials were dying on the +# exact error this path already survived. Re-exported under the original +# private names so existing call sites and tests keep resolving. +from .stream_retry import ( # noqa: E402 + TRANSIENT_STREAM_DROP_MARKERS as _TRANSIENT_STREAM_DROP_MARKERS, + is_transient_stream_drop as _is_transient_stream_drop, ) -def _is_transient_stream_drop(exc: BaseException) -> bool: - """True for a mid-stream disconnect that is safe to re-attempt. - - Deliberately narrow. A dropped connection carries no HTTP status and no - partial result worth keeping, so re-issuing the request is the same - decision the idle watchdog already makes — whereas a 4xx (auth, bad - request) must never be retried, and those arrive as ``APIStatusError`` - subclasses that this predicate rejects via the status-code check. - - Motivation: on terminal-bench 2.1 (regex-chess, 2026-07-25) a single - ``peer closed connection without sending complete message body - (incomplete chunked read)`` killed a 24-minute trial outright — the - agent had just finished a passing 1500-position fuzz run. Claude Code - survives the same class of blip; clawcodex scored 0. - """ - # Anything carrying an HTTP status is a server verdict, not a drop. - if getattr(exc, "status_code", None) is not None: - return False - name = type(exc).__name__ - if name in ("APIConnectionError", "RemoteProtocolError", "ReadError", - "ConnectError", "ChunkedEncodingError"): - return True - text = str(exc).lower() - return any(marker in text for marker in _TRANSIENT_STREAM_DROP_MARKERS) - - def _default_max_tokens(model: str | None) -> int: """Pick a sensible ``max_tokens`` ceiling for the given model. diff --git a/src/providers/openai_compatible.py b/src/providers/openai_compatible.py index 72b85eb1..e5e92e6e 100644 --- a/src/providers/openai_compatible.py +++ b/src/providers/openai_compatible.py @@ -801,7 +801,77 @@ def chat_stream_response( on_thinking_chunk: TextChunkCallback | None = None, **kwargs ) -> ChatResponse: - """Stream OpenAI-compatible chunks while rebuilding the final response. + """Stream a response, re-issuing once if the transport drops. + + A connection that dies mid-response carries no HTTP status and no + salvageable output, so re-issuing is the right call — the Anthropic + provider has done this since #747. This path did not, and paid for + it: on terminal-bench 2.1 with deepseek-v4-pro (2026-07-27), + ``peer closed connection without sending complete message body + (incomplete chunked read)`` ended 8 of 89 trials — 16% of every + failure — including one that had already run 391 seconds of + successful work. Same benchmark, same error, same harness; only the + wire differed. + + Retry is skipped once any text or reasoning has reached the caller, + because the transcript would otherwise show that prefix twice. In a + scored trial harbor runs ``--print --output-format stream-json`` + without ``--include-partial-messages``, so both callbacks are None + and the retry always fires. + + Only transport drops qualify (``is_transient_stream_drop``); a 4xx + is a server verdict and is re-raised untouched. On the final attempt + the ORIGINAL exception propagates, so callers see the real cause + rather than a retry wrapper. + """ + from .stream_retry import is_transient_stream_drop + + emitted = [False] + + def _mark_text(text: str) -> None: + emitted[0] = True + on_text_chunk(text) # type: ignore[misc] — only built when non-None + + def _mark_thinking(text: str) -> None: + emitted[0] = True + on_thinking_chunk(text) # type: ignore[misc] + + max_attempts = 2 + for attempt in range(1, max_attempts + 1): + emitted[0] = False + try: + return self._stream_attempt( + messages, + tools, + on_text_chunk=_mark_text if on_text_chunk else None, + abort_signal=abort_signal, + on_thinking_chunk=_mark_thinking if on_thinking_chunk else None, + **kwargs, + ) + except Exception as exc: + if ( + attempt >= max_attempts + or emitted[0] + or not is_transient_stream_drop(exc) + ): + raise + logger.warning( + "Stream dropped (%s); retrying once: %s", + type(exc).__name__, + str(exc)[:200], + ) + raise AssertionError("unreachable: loop returns or raises") + + def _stream_attempt( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]] = None, + on_text_chunk: TextChunkCallback | None = None, + abort_signal: Any = None, + on_thinking_chunk: TextChunkCallback | None = None, + **kwargs + ) -> ChatResponse: + """One streaming attempt. See ``chat_stream_response`` for retry. ESC-cancellation runs the SDK iteration on a daemon worker thread that pushes chunks into a ``queue.Queue``. The main diff --git a/src/providers/stream_retry.py b/src/providers/stream_retry.py new file mode 100644 index 00000000..8551ad76 --- /dev/null +++ b/src/providers/stream_retry.py @@ -0,0 +1,64 @@ +"""Shared classifier for mid-stream transport drops. + +Both wire families lose streams the same way and must make the same call +about re-issuing, so the predicate lives here rather than in either +provider. It was previously Anthropic-only, which is why DeepSeek trials +died on the identical error the Anthropic path had already learned to +survive. + +Kept dependency-free on purpose: ``openai_compatible`` importing +``anthropic_provider`` (or the reverse) would pull a heavy SDK into the +fast-path CLI's import graph and risk a cycle. +""" + +from __future__ import annotations + +#: Substrings of a connection that died mid-response, with no HTTP status +#: and no model output to salvage. httpx raises these as +#: ``RemoteProtocolError``/``ReadError``; the SDKs surface them as +#: ``APIConnectionError``. Matched on the message as well as the type +#: because the SDKs wrap and re-word some of them. +TRANSIENT_STREAM_DROP_MARKERS = ( + "peer closed connection", + "incomplete chunked read", + "server disconnected", + "connection reset", + "connection aborted", +) + +#: Exception type names that are always a transport drop regardless of text. +_TRANSIENT_STREAM_DROP_TYPES = frozenset({ + "APIConnectionError", + "RemoteProtocolError", + "ReadError", + "ConnectError", + "ChunkedEncodingError", +}) + + +def is_transient_stream_drop(exc: BaseException) -> bool: + """True for a mid-stream disconnect that is safe to re-attempt. + + Deliberately narrow. A dropped connection carries no HTTP status and no + partial result worth keeping, so re-issuing the request is the same + decision the idle watchdog already makes — whereas a 4xx (auth, bad + request) must never be retried, and those arrive as ``APIStatusError`` + subclasses that this predicate rejects via the status-code check. + + Motivation, one instance per wire: + + * Anthropic — terminal-bench 2.1 regex-chess (2026-07-25): a single + ``peer closed connection without sending complete message body + (incomplete chunked read)`` killed a 24-minute trial outright, just + after a passing 1500-position fuzz run. + * DeepSeek — the same benchmark (2026-07-27): the identical error ended + 8 of 89 trials, 16% of all failures, because this logic lived on the + Anthropic provider only. + """ + # Anything carrying an HTTP status is a server verdict, not a drop. + if getattr(exc, "status_code", None) is not None: + return False + if type(exc).__name__ in _TRANSIENT_STREAM_DROP_TYPES: + return True + text = str(exc).lower() + return any(marker in text for marker in TRANSIENT_STREAM_DROP_MARKERS) diff --git a/tests/test_openai_compat_stream_retry.py b/tests/test_openai_compat_stream_retry.py new file mode 100644 index 00000000..5b9b5d71 --- /dev/null +++ b/tests/test_openai_compat_stream_retry.py @@ -0,0 +1,153 @@ +"""Mid-stream transport drops must be retried on the OpenAI-compatible wire. + +The Anthropic provider has retried these since #747. This path did not, and +on terminal-bench 2.1 with deepseek-v4-pro (2026-07-27) that asymmetry ended +8 of 89 trials -- 16% of every failure -- on the identical error string the +Anthropic path already survived. +""" + +from __future__ import annotations + +import pytest + +from src.providers.base import ChatResponse +from src.providers.openai_compatible import OpenAICompatibleProvider +from src.providers.stream_retry import is_transient_stream_drop + + +DROP = "peer closed connection without sending complete message body (incomplete chunked read)" + + +class _Concrete(OpenAICompatibleProvider): + """Minimal concrete subclass — the ABC needs these two implemented.""" + + def _create_client(self): # pragma: no cover - never called + raise AssertionError("client must not be built in these tests") + + def get_available_models(self): + return ["deepseek-v4-pro"] + + +def _provider(monkeypatch, attempts: list, outcomes): + """A provider whose single streaming attempt yields ``outcomes`` in order.""" + p = _Concrete.__new__(_Concrete) + seq = iter(outcomes) + + def fake_attempt(self, *a, **kw): + attempts.append(kw) + nxt = next(seq) + if isinstance(nxt, BaseException): + raise nxt + return nxt + + monkeypatch.setattr(OpenAICompatibleProvider, "_stream_attempt", fake_attempt) + return p + + +def _ok(text="done"): + return ChatResponse( + content=text, + model="deepseek-v4-pro", + usage={"input_tokens": 1, "output_tokens": 1}, + finish_reason="stop", + ) + + +class TestTransientDropRetry: + def test_drop_is_retried_and_succeeds(self, monkeypatch): + attempts: list = [] + p = _provider(monkeypatch, attempts, [Exception(DROP), _ok()]) + out = p.chat_stream_response([{"role": "user", "content": "hi"}]) + assert out.content == "done" + assert len(attempts) == 2, "the drop should have been re-issued once" + + def test_retry_is_bounded_and_reraises_the_original(self, monkeypatch): + attempts: list = [] + original = Exception(DROP) + p = _provider(monkeypatch, attempts, [original, original]) + with pytest.raises(Exception) as ei: + p.chat_stream_response([{"role": "user", "content": "hi"}]) + assert ei.value is original, "callers must see the real cause, not a wrapper" + assert len(attempts) == 2, "must not retry unboundedly" + + def test_http_status_errors_are_never_retried(self, monkeypatch): + """A 400 is a server verdict — re-issuing it just burns a request. + + This is the guard that keeps the DeepSeek image_url 400 (a separate, + real bug) from being retried into a second identical rejection. + """ + class Status400(Exception): + status_code = 400 + + attempts: list = [] + err = Status400("unknown variant `image_url`, expected `text`") + p = _provider(monkeypatch, attempts, [err]) + with pytest.raises(Status400): + p.chat_stream_response([{"role": "user", "content": "hi"}]) + assert len(attempts) == 1 + + def test_no_retry_once_text_reached_the_caller(self, monkeypatch): + """Otherwise the retry replays a prefix the user already saw.""" + attempts: list = [] + seen: list[str] = [] + + def fake_attempt(self, *a, **kw): + attempts.append(kw) + cb = kw.get("on_text_chunk") + if cb: + cb("partial ") + raise Exception(DROP) + + monkeypatch.setattr(OpenAICompatibleProvider, "_stream_attempt", fake_attempt) + p = _Concrete.__new__(_Concrete) + with pytest.raises(Exception, match="peer closed"): + p.chat_stream_response( + [{"role": "user", "content": "hi"}], + on_text_chunk=seen.append, + ) + assert len(attempts) == 1, "streamed output must not be duplicated" + assert seen == ["partial "] + + def test_thinking_output_also_suppresses_retry(self, monkeypatch): + attempts: list = [] + + def fake_attempt(self, *a, **kw): + attempts.append(kw) + cb = kw.get("on_thinking_chunk") + if cb: + cb("reasoning ") + raise Exception(DROP) + + monkeypatch.setattr(OpenAICompatibleProvider, "_stream_attempt", fake_attempt) + p = _Concrete.__new__(_Concrete) + with pytest.raises(Exception, match="peer closed"): + p.chat_stream_response( + [{"role": "user", "content": "hi"}], + on_thinking_chunk=lambda _t: None, + ) + assert len(attempts) == 1 + + +class TestSharedPredicate: + """Both wires must classify identically — that was the whole defect.""" + + @pytest.mark.parametrize("msg", [ + DROP, + "Server disconnected without sending a response", + "Connection reset by peer", + "connection aborted", + ]) + def test_transport_drops_match(self, msg): + assert is_transient_stream_drop(Exception(msg)) + + def test_status_bearing_errors_do_not_match(self): + class E(Exception): + status_code = 429 + assert not is_transient_stream_drop(E("rate limited")) + + def test_unrelated_errors_do_not_match(self): + assert not is_transient_stream_drop(ValueError("bad tool schema")) + + def test_anthropic_alias_is_the_shared_function(self): + from src.providers.anthropic_provider import _is_transient_stream_drop + assert _is_transient_stream_drop is is_transient_stream_drop