fix(providers): retry mid-stream transport drops on the OpenAI-compat wire

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.
 

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

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)

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