From 9f53a48b5f6ec64c3fb959c3fc61facce8f12c6f Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 26 Jul 2026 00:08:07 -0700 Subject: [PATCH] fix(harness): recover trials lost to empty turns, transport drops, and a headless-only tool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four defects found by diffing two full terminal-bench 2.1 runs at identical model/effort/dataset (claude-opus-5, xhigh): clawcodex 0.742 vs the official Claude Code harness 0.843. Of the 13 tasks Claude Code solved and clawcodex did not, 3 were crashes and 3 were trials that did nothing at all — none of it model capability. All four fixes are general robustness. Nothing here is conditional on an eval: no env checks, no branches, no wording aimed at a grader. ## An empty assistant turn ended the run as "completed" Three trials — break-filter-js-from-html, crack-7z-hash, vulnerable-secret — produced ONE output token in ~3 seconds, returned an empty result, reported `subtype: success`, and scored 0. Claude Code solved all three. No text and no tool calls is a degenerate response, not an answer. The continuation nudge could not catch it because it gates on the text being truthy, so an empty turn fell straight through to `Terminal(reason="completed")`. Empty turns are now re-prompted, bounded by the same MAX_CONTINUATION_NUDGES cap as every other nudge. Carve-out: a model that speaks through SendUserMessage (documented as the primary visible output channel) legitimately ends with empty assistant text, and agent_loop_compat already treats the outbox as the response text. The existing parity test caught that case. ## A transport blip killed a 24-minute trial regex-chess died on `peer closed connection without sending complete message body (incomplete chunked read)` seconds after a passing 1500-position fuzz run. Transport drops now take the same bounded retry the idle watchdog already takes, and the ORIGINAL exception is re-raised on exhaustion so the log names the real failure instead of an idle timeout. Gated two ways. Status-bearing errors (401/529/400) and aborts fail fast — retrying a revoked token just burns the budget on guaranteed 401s. And the retry is skipped once any text has been handed to the caller: a re-attempt replays the response from scratch, so retrying after partial output would emit those chunks twice, which is exactly what query.py:1635 forbids ("never after partial output"). That check sits above this layer and cannot see a retry taken here, so the gate has to live here too. ## AskUserQuestion was reachable in --print fix-git called it twice, then ended with "let me explain what I found" instead of finishing; the verifier found the files untouched. It is not in the initial toolset — the model reached it through ToolSearch. Nothing on the headless surface can answer it (StreamJsonReader parses only `user` messages; there is no can_use_tool protocol), so it is now unregistered rather than left registered with a stubbed responder. This is parity, not special-casing: the TS reference's AskUserQuestionTool already reports itself disabled when nobody is at the keyboard. Removal happens before the allow/deny filters so --allowed-tools cannot resurrect it, and it propagates to subagents, which share the registry instance. Test doubles gained `remove_tool` rather than the production call gaining a getattr guard — a silent no-op there is the bug class that left --allowed-tools dead for months (#739). ## The idle-timeout message was written for a grader It appended "Connection timed out" with a comment saying the phrasing was deliberate so eval harnesses would classify the failure as retryable. That optimizes for the scoreboard rather than the agent, and it was also untrue — an idle stream is not a connection timeout, so it pointed debuggers at the wrong subsystem. Removed; the test now asserts the honest wording and that the old clause stays gone. ## Diagnosed, deliberately NOT fixed Two trials died on `401 OAuth access token has been revoked` — not expiry (one died 4.7 minutes in against a 2h runway) but refresh rotation: containers hold a refresh-token-free copy, so anything rotating the host credential revokes the in-flight token. An in-adapter retry would hand clawcodex attempts the official harness would not give another agent, so this is documented in the adapter with a recommendation to use an API key for scored runs. Tests: 8841 passed. Every fix is revert-sensitive — each line was deleted and the intended test confirmed failing. Co-Authored-By: Claude Opus 5 --- eval/harbor/clawcodex_agent.py | 14 ++ src/entrypoints/headless.py | 12 ++ src/providers/anthropic_provider.py | 122 ++++++++++++-- src/query/continuation_nudge.py | 10 ++ src/query/query.py | 51 ++++++ tests/entrypoints/test_headless_goal.py | 5 + tests/test_dangerous_skip_permissions.py | 10 ++ tests/test_headless_cli.py | 5 + tests/test_headless_sigint.py | 7 + tests/test_headless_tool_filter.py | 35 ++++ tests/test_permission_ask_flow.py | 5 + tests/test_query_loop_wires_round3.py | 39 +++++ tests/test_stream_watchdog.py | 193 ++++++++++++++++++++++- 13 files changed, 489 insertions(+), 19 deletions(-) diff --git a/eval/harbor/clawcodex_agent.py b/eval/harbor/clawcodex_agent.py index b9f360ab0..ee25f1b68 100644 --- a/eval/harbor/clawcodex_agent.py +++ b/eval/harbor/clawcodex_agent.py @@ -59,6 +59,20 @@ ``ANTHROPIC_API_KEY`` is deliberately not forwarded so clawcodex takes the OAuth path (and combining with ``--ae ANTHROPIC_API_KEY`` is rejected). + + **Prefer an API key for any scored or submitted run.** The injected copy + deliberately carries no refresh token, so it cannot self-heal: when the + credential is rotated — by this adapter refreshing for a later trial, or + by ANY other process sharing ``anthropic-oauth.json`` (an interactive + clawcodex session, another job) — the provider revokes the token already + live inside running containers, and those trials die mid-run with + ``401 … OAuth access token has been revoked``. It is indistinguishable + from an agent crash in the results: Harbor records + ``NonZeroAgentExitCodeError`` and scores 0. Measured on the tb2.1 + 2026-07-25 run: 2 of 89 trials lost this way (one only 4.7 minutes in, + so runway was not the issue — rotation was), against 0 such failures for + the claude-code adapter in the comparable run. An API key has no + rotation and no expiry, so a scored run should not be exposed to it. """ import json diff --git a/src/entrypoints/headless.py b/src/entrypoints/headless.py index 6408c5fe1..4387a44c1 100644 --- a/src/entrypoints/headless.py +++ b/src/entrypoints/headless.py @@ -173,6 +173,18 @@ def run_headless(options: HeadlessOptions) -> int: session = Session.create(provider_name, getattr(provider, "model", model or "")) tool_registry = build_default_registry(provider=provider) + # AskUserQuestion cannot work here: headless has no terminal, and no + # surface routes questions back to a client (``ask_user`` below is + # unconditionally stubbed). Leaving it REGISTERED but stubbed meant the + # model could still find it — via ToolSearch even when deferred loading + # keeps it out of the initial set — spend turns on it, and then hand the + # task back to a user who does not exist. Observed live on terminal-bench + # 2.1 (fix-git, 2026-07-25): two AskUserQuestion calls, then a final + # "let me explain what I found" instead of finishing; the verifier found + # the files untouched. Unregistering matches what Claude Code exposes + # headlessly. ``_noop_ask_user`` stays as a backstop for any path that + # still reaches the context hook. + tool_registry.remove_tool("AskUserQuestion") # Canonicalize BOTH sets up front (before either filter runs) so an alias # form (e.g. --disallowed-tools KillShell) resolves while its tool is still # registered. diff --git a/src/providers/anthropic_provider.py b/src/providers/anthropic_provider.py index 289c6306a..f3d1b90ec 100644 --- a/src/providers/anthropic_provider.py +++ b/src/providers/anthropic_provider.py @@ -161,6 +161,46 @@ 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", +) + + +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. @@ -732,29 +772,77 @@ def chat_stream_response( max_attempts = stream_idle_max_attempts() def _idle_timeout_error() -> StreamIdleTimeout: - # "Connection timed out" phrasing is deliberate: eval harnesses - # (harbor) classify it as a retryable network error. + # Describe what happened, nothing more. This message previously + # carried a "Connection timed out" clause added specifically so + # eval harnesses would classify the failure as a retryable + # network error — i.e. wording chosen to earn retries from a + # grader rather than to inform a reader. That is optimizing for + # the scoreboard instead of the agent, and it made the message + # lie: an idle timeout is not a connection timeout. Removed. return StreamIdleTimeout( f"stream idle timeout: no stream events for " f"{stream_idle_timeout_seconds():.0f}s on {max_attempts} " - f"streaming attempt(s) (Connection timed out; " - f"x-client-request-id={request_id or 'n/a'})" + f"streaming attempt(s) " + f"(x-client-request-id={request_id or 'n/a'})" ) + # Track whether THIS attempt already handed text to the caller. A + # re-attempt replays the response from scratch, so retrying after + # partial output would emit those chunks twice — the exact thing + # ``query.py`` forbids ("never after partial output", query.py:1635, + # enforced there by ``_streamed_any``). That layer sits ABOVE this + # one and cannot see a retry taken here, so the gate has to live + # here too. Costs the eval nothing: harbor runs + # ``--print --output-format stream-json`` without + # ``--include-partial-messages``, so both callbacks are None in a + # scored trial and the retry always fires. + 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] + for attempt in range(1, max_attempts + 1): - response = self._stream_attempt( - client=client, - guard=guard, - model=model, - max_tokens=max_tokens, - anthropic_messages=anthropic_messages, - system=system, - extra_kwargs=extra_kwargs, - kwargs=kwargs, - request_id=request_id, - on_text_chunk=on_text_chunk, - on_thinking_chunk=on_thinking_chunk, - ) + emitted[0] = False + try: + response = self._stream_attempt( + client=client, + guard=guard, + model=model, + max_tokens=max_tokens, + anthropic_messages=anthropic_messages, + system=system, + extra_kwargs=extra_kwargs, + kwargs=kwargs, + request_id=request_id, + on_text_chunk=_mark_text if on_text_chunk else None, + on_thinking_chunk=_mark_thinking if on_thinking_chunk else None, + ) + except Exception as exc: + # A transport-level drop gets the SAME bounded retry the + # idle watchdog already gets — it's the identical decision + # (no status, no salvageable output, re-issue the request), + # and letting it propagate ends the whole agent run over a + # blip. On the last attempt the ORIGINAL exception is + # re-raised, not the idle-timeout one, so the failure keeps + # naming what actually happened. Aborts and every status- + # bearing error (auth/4xx) fail fast via the predicate. + if ( + attempt >= max_attempts + or emitted[0] + or not _is_transient_stream_drop(exc) + ): + raise + logger.warning( + "transient stream drop (attempt %d/%d, " + "x-client-request-id=%s): %s; retrying stream", + attempt, max_attempts, request_id or "n/a", exc, + ) + continue if response is not None: return response if attempt < max_attempts: diff --git a/src/query/continuation_nudge.py b/src/query/continuation_nudge.py index b2e1fbcfc..2a20069b6 100644 --- a/src/query/continuation_nudge.py +++ b/src/query/continuation_nudge.py @@ -16,6 +16,16 @@ NUDGE_MESSAGE = "Continue with the task. Use the appropriate tools to proceed." +#: Sent when an assistant turn comes back with no tool calls AND no text. +#: That is never a usable answer — it is a degenerate response that would +#: otherwise terminate the run as "completed" with nothing done. Kept short +#: and neutral so it reads as a retry rather than a correction. +EMPTY_TURN_NUDGE = ( + "Your last response was empty. Continue working on the task using the " + "available tools. If the task is genuinely already complete, say so " + "explicitly and state what you did." +) + EXHAUSTIVE_AUDIT_NUDGE = ( "Before finishing, audit the result against the user's exhaustive-result " "requirement. Actively search for additional valid results; do not assume " diff --git a/src/query/query.py b/src/query/query.py index 99a2cd87b..0ec9b3a62 100644 --- a/src/query/query.py +++ b/src/query/query.py @@ -31,6 +31,7 @@ from .config import QueryConfig, build_query_config from .continuation_nudge import ( + EMPTY_TURN_NUDGE, MAX_CONTINUATION_NUDGES, NUDGE_MESSAGE, detect_continuation_signal, @@ -2213,6 +2214,56 @@ def _marking_chunk_cb(text: str) -> None: transition=Transition(reason="continuation_nudge"), ) continue + # An assistant turn with NO tool calls and NO text is not a + # completion — it is a degenerate response, and accepting it + # ends the run with an empty answer while every caller + # (including eval harnesses) records a clean success. The + # nudge below can't catch it: it gates on ``last_text`` + # being truthy, so an empty turn falls straight through to + # ``Terminal(reason="completed")``. + # + # Measured on terminal-bench 2.1 (2026-07-25): 3 of 89 + # trials — break-filter-js-from-html, crack-7z-hash, + # vulnerable-secret — ended after ONE turn and ONE output + # token in ~3 seconds, each reported ``subtype: success`` + # with an empty result and scored 0. Claude Code solved all + # three. Re-prompting costs one round trip and is bounded by + # the same MAX_CONTINUATION_NUDGES cap as every other nudge. + # ...unless the model spoke through the user-visible outbox + # instead. SendUserMessage advertises itself as the primary + # visible output channel, so a model that obeys it + # legitimately ends with empty assistant text; + # agent_loop_compat already treats the outbox as the + # response text in that case. Nudging there would re-prompt + # a turn that actually delivered. + spoke_via_outbox = bool( + getattr(tool_use_context, "outbox", None) + ) + if not last_text.strip() and not spoke_via_outbox: + logger.warning( + "empty assistant turn (no text, no tool calls) — " + "re-prompting (%d/%d)", + state.continuation_nudge_count + 1, + MAX_CONTINUATION_NUDGES, + ) + state = QueryState( + messages=[ + *messages, + *assistant_messages, + UserMessage(content=EMPTY_TURN_NUDGE, isMeta=True), + ], + tool_use_context=tool_use_context, + auto_compact_tracking=state.auto_compact_tracking, + max_output_tokens_recovery_count=0, + has_attempted_reactive_compact=False, + max_output_tokens_override=None, + stop_hook_active=None, + turn_count=turn_count, + pending_tool_use_summary=None, + continuation_nudge_count=state.continuation_nudge_count + 1, + transition=Transition(reason="empty_turn_nudge"), + ) + continue if last_text and detect_continuation_signal(last_text): logger.debug( "Continuation nudge triggered (%d/%d)", diff --git a/tests/entrypoints/test_headless_goal.py b/tests/entrypoints/test_headless_goal.py index 2d7ee87f2..7999e566c 100644 --- a/tests/entrypoints/test_headless_goal.py +++ b/tests/entrypoints/test_headless_goal.py @@ -42,6 +42,11 @@ class _FakeRegistry: def list_tools(self): return [] + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (no user on + # this surface); real ToolRegistry returns a bool. + return False + class _Wiring: """Fixture handle: ``script`` feeds providers created after it's filled; diff --git a/tests/test_dangerous_skip_permissions.py b/tests/test_dangerous_skip_permissions.py index b953e86f6..3570bfda4 100644 --- a/tests/test_dangerous_skip_permissions.py +++ b/tests/test_dangerous_skip_permissions.py @@ -246,6 +246,11 @@ class _FakeRegistry: def list_tools(self): return [] + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (no user on + # this surface); real ToolRegistry returns a bool. + return False + monkeypatch.setattr( headless_mod, "get_provider_class", lambda n: _FakeProvider ) @@ -318,6 +323,11 @@ class _FakeRegistry: def list_tools(self): return [] + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (no user on + # this surface); real ToolRegistry returns a bool. + return False + monkeypatch.setattr( headless_mod, "get_provider_class", lambda n: _FakeProvider ) diff --git a/tests/test_headless_cli.py b/tests/test_headless_cli.py index 175e0c2fb..ab3455560 100644 --- a/tests/test_headless_cli.py +++ b/tests/test_headless_cli.py @@ -45,6 +45,11 @@ class _FakeRegistry: def list_tools(self): return [] + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (no user on + # this surface); real ToolRegistry returns a bool. + return False + @pytest.fixture def fake_wiring(monkeypatch): diff --git a/tests/test_headless_sigint.py b/tests/test_headless_sigint.py index 9f9344c6d..ccdf4b7d9 100644 --- a/tests/test_headless_sigint.py +++ b/tests/test_headless_sigint.py @@ -106,6 +106,13 @@ class _Registry: def list_tools(self): return list(self._tools) + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (nothing on this + # surface can answer it); real ToolRegistry returns a bool. + before = len(self._tools) + self._tools = [t for t in self._tools if getattr(t, "name", None) != name] + return len(self._tools) != before + def dispatch(self, call, context): return _fake_tool_call(call.input, context) diff --git a/tests/test_headless_tool_filter.py b/tests/test_headless_tool_filter.py index fa45c3f6d..5679ea45b 100644 --- a/tests/test_headless_tool_filter.py +++ b/tests/test_headless_tool_filter.py @@ -142,3 +142,38 @@ def test_canonicalize_skips_blanks_so_allowlist_cannot_wipe_all(): if allow: headless_mod._filter_registry(registry, keep=lambda n: n.lower() in allow) assert _names(registry) == before + + +def test_headless_unregisters_ask_user_question_source(): + """Headless must not merely stub AskUserQuestion — it must unregister it. + + Stubbing only ``tool_context.ask_user`` leaves the tool in the registry, + so the model can still reach it (via ToolSearch even when deferred + loading keeps it out of the initial set), spend turns on it, and hand + the task back to a user who cannot exist on this surface. Observed on + terminal-bench 2.1 (fix-git, 2026-07-25): two calls, then a "let me + explain what I found" ending with the work undone. + + Asserted against the source because ``run_headless`` needs a live + provider/session to reach the line. + """ + import inspect + + src = inspect.getsource(headless_mod.run_headless) + assert 'remove_tool("AskUserQuestion")' in src, ( + "headless must unregister AskUserQuestion, not just stub ask_user" + ) + # And the removal must precede the allow/deny filtering, so an explicit + # --allowed-tools list cannot resurrect it. + assert src.index('remove_tool("AskUserQuestion")') < src.index( + "_filter_registry" + ), "unregister AskUserQuestion before the allow/deny filters run" + + +def test_remove_tool_is_idempotent_for_ask_user_question(): + """The headless unregister must be safe when the tool is already absent + (e.g. a provider whose default registry never included it).""" + registry = build_default_registry(provider="anthropic") + assert registry.remove_tool("AskUserQuestion") is True + assert registry.remove_tool("AskUserQuestion") is False + assert "AskUserQuestion" not in _names(registry) diff --git a/tests/test_permission_ask_flow.py b/tests/test_permission_ask_flow.py index 92d312ff8..41e3da0d5 100644 --- a/tests/test_permission_ask_flow.py +++ b/tests/test_permission_ask_flow.py @@ -226,6 +226,11 @@ class _FakeRegistry: def list_tools(self): return [] + def remove_tool(self, name): + # run_headless unregisters AskUserQuestion (no user on + # this surface); real ToolRegistry returns a bool. + return False + with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) local = root / ".clawcodex" / "settings.local.json" diff --git a/tests/test_query_loop_wires_round3.py b/tests/test_query_loop_wires_round3.py index fa4e26e6e..3f4c39527 100644 --- a/tests/test_query_loop_wires_round3.py +++ b/tests/test_query_loop_wires_round3.py @@ -575,6 +575,45 @@ def test_nudges_then_caps(self): ) self.assertEqual(assistant_count, MAX_CONTINUATION_NUDGES + 1) + def test_empty_turn_is_reprompted_not_treated_as_completion(self): + """An empty assistant turn must not end the run as "completed". + + No text and no tool calls is a degenerate response, not an answer. + The continuation nudge can't catch it — that one gates on the text + being truthy — so an empty turn fell straight through to + Terminal(reason="completed") with an empty result, which every + caller records as a clean success. + + Measured on terminal-bench 2.1 (2026-07-25): 3 of 89 trials + (break-filter-js-from-html, crack-7z-hash, vulnerable-secret) ended + after ONE turn and ONE output token in ~3s, reported success, and + scored 0 on tasks Claude Code solved. + """ + from src.types.messages import AssistantMessage + + provider = _provider([_completion("")]) + msgs, terminal = _run(run_query(_params(self.workspace, provider))) + assistant_count = sum( + 1 for m in msgs if isinstance(m, AssistantMessage) + ) + self.assertGreater( + assistant_count, 1, + "an empty turn must be re-prompted, not accepted as the answer", + ) + # Bounded by the shared nudge cap — never an unbounded retry loop. + self.assertEqual(assistant_count, MAX_CONTINUATION_NUDGES + 1) + self.assertEqual(terminal.reason, "completed") + + def test_whitespace_only_turn_is_also_reprompted(self): + """Whitespace is as empty as empty — the check strips before testing.""" + from src.types.messages import AssistantMessage + + provider = _provider([_completion(" \n ")]) + msgs, _terminal = _run(run_query(_params(self.workspace, provider))) + self.assertGreater( + sum(1 for m in msgs if isinstance(m, AssistantMessage)), 1 + ) + def test_completion_text_no_nudge(self): provider = _provider([_completion("Done. Everything is complete.")]) msgs, terminal = _run(run_query(_params(self.workspace, provider))) diff --git a/tests/test_stream_watchdog.py b/tests/test_stream_watchdog.py index ef1a1523d..4fc92a569 100644 --- a/tests/test_stream_watchdog.py +++ b/tests/test_stream_watchdog.py @@ -13,6 +13,8 @@ import threading import time import unittest + +import pytest from unittest.mock import MagicMock from src.utils.stream_watchdog import ( @@ -330,8 +332,16 @@ def test_watchdog_exhaustion_raises_stream_idle_timeout(self): # Default budget: 3 total attempts, all streamed. self.assertEqual(fake_client.messages.stream.call_count, 3) mock_chat.assert_not_called() - # Harness-classifiable phrasing (harbor: NetworkConnectionError). - self.assertIn("Connection timed out", str(ctx.exception)) + # The message describes the actual failure and nothing else. It + # used to append "Connection timed out" purely so an eval harness + # would classify the trial as a retryable network error — wording + # aimed at a grader, not a reader, and inaccurate besides (an idle + # stream is not a connection timeout). Pin the honest phrasing so + # it does not creep back. + message = str(ctx.exception) + self.assertIn("stream idle timeout", message) + self.assertIn("no stream events for", message) + self.assertNotIn("Connection timed out", message) def test_retry_budget_env_override(self): from src.utils.stream_watchdog import stream_idle_max_attempts @@ -506,3 +516,182 @@ def blocked_reader(): if __name__ == "__main__": unittest.main() + + +class TestTransientStreamDropClassifier(unittest.TestCase): + """A mid-stream transport drop is retryable; a server verdict is not. + + Terminal-bench 2.1 (regex-chess, 2026-07-25): one ``peer closed + connection without sending complete message body (incomplete chunked + read)`` ended a 24-minute trial at reward 0, moments after a passing + 1500-position fuzz run. The idle watchdog already re-attempts the + stream for the equivalent condition; this classifier is what lets the + transport case take the same bounded path instead of propagating. + """ + + def _classify(self, exc): + from src.providers.anthropic_provider import _is_transient_stream_drop + + return _is_transient_stream_drop(exc) + + def test_retries_transport_drops(self): + class APIConnectionError(Exception): + pass + + for exc in ( + APIConnectionError("peer closed connection without sending " + "complete message body (incomplete chunked read)"), + Exception("Server disconnected without sending a response."), + Exception("Connection reset by peer"), + ): + self.assertTrue(self._classify(exc), exc) + + def test_never_retries_status_bearing_errors(self): + """Auth and overload carry an HTTP status — retrying an expired or + revoked token just burns the remaining budget on guaranteed 401s.""" + class APIStatusError(Exception): + def __init__(self, message, status_code): + super().__init__(message) + self.status_code = status_code + + for exc in ( + APIStatusError("OAuth access token has been revoked.", 401), + APIStatusError("overloaded_error", 529), + APIStatusError("invalid_request_error", 400), + ): + self.assertFalse(self._classify(exc), exc) + + def test_matches_the_real_sdk_and_httpx_classes(self): + """Pin against REAL exception objects, not local stand-ins. + + The local fakes validate name matching but cannot catch SDK + hierarchy drift — which is the actual risk, since the predicate is + subclass-blind by design. httpx.RemoteProtocolError is the concrete + class the regex-chess trial died on: the SDK does not wrap errors + raised while ITERATING a stream, so it arrives raw. + """ + httpx = pytest.importorskip("httpx") + anthropic = pytest.importorskip("anthropic") + + self.assertTrue(self._classify(httpx.RemoteProtocolError( + "peer closed connection without sending complete message body " + "(incomplete chunked read)"))) + req = httpx.Request("POST", "https://api.anthropic.com/v1/messages") + self.assertTrue(self._classify(httpx.ReadError("boom", request=req))) + self.assertTrue(self._classify( + anthropic.APIConnectionError(request=req))) + # Client-side protocol bugs are OUR fault — replaying just repeats it. + self.assertFalse(self._classify(httpx.LocalProtocolError("bad header"))) + + def test_never_retries_unrelated_failures(self): + for exc in (KeyboardInterrupt(), ValueError("bad input"), + Exception("prompt is too long")): + self.assertFalse(self._classify(exc), exc) + + def test_status_bearing_wins_over_message_match(self): + """A status-bearing error whose text happens to mention a drop is + still a server verdict — the status check must come first.""" + class Weird(Exception): + status_code = 400 + + self.assertFalse(self._classify(Weird("peer closed connection"))) + + +class TestTransientDropRetryLoop(unittest.TestCase): + """The classifier is only useful if the attempt loop acts on it.""" + + def _provider(self): + from src.providers.anthropic_provider import AnthropicProvider + + return AnthropicProvider(api_key="sk-test", model="claude-opus-5") + + def _call(self, provider, side_effect, on_text_chunk=None): + from unittest.mock import patch + + with patch.object( + type(provider), "_stream_attempt", side_effect=side_effect + ) as attempt, patch.object( + type(provider), "_client_for_request", return_value=MagicMock() + ), patch.object( + type(provider), "_merge_beta_headers" + ), patch.object( + type(provider), "_merge_request_id", return_value="req-test" + ): + try: + result = provider.chat_stream_response( + [{"role": "user", "content": "hi"}], + on_text_chunk=on_text_chunk, + ) + except Exception as exc: # noqa: BLE001 — the assertion subject + return attempt, exc + return attempt, result + + def test_retries_then_succeeds(self): + drop = Exception("peer closed connection without sending complete " + "message body (incomplete chunked read)") + sentinel = MagicMock(name="chat-response") + attempt, result = self._call(self._provider(), [drop, sentinel]) + self.assertIs(result, sentinel) + self.assertEqual(attempt.call_count, 2, "should re-attempt after a drop") + + def test_does_not_retry_after_text_was_already_emitted(self): + """A retry replays the response, so retrying after partial output + would emit those chunks TWICE — what query.py:1635 forbids + ("never after partial output"). That check lives one layer up and + cannot see a retry taken here, so this gate must hold locally. + + Harbor runs without --include-partial-messages, so on_text_chunk is + None in a scored trial and the rescue still fires; this only + protects the TUI/SDK surfaces that do stream chunks. + """ + seen = [] + drop = Exception("peer closed connection without sending complete " + "message body (incomplete chunked read)") + + def emit_then_drop(*a, **k): + cb = k.get("on_text_chunk") + if cb: + cb("partial ") + raise drop + + provider = self._provider() + attempt, exc = self._call( + provider, emit_then_drop, on_text_chunk=seen.append + ) + self.assertIn("peer closed connection", str(exc)) + self.assertEqual(attempt.call_count, 1, "must not replay after output") + self.assertEqual(seen, ["partial "], "no duplicated text") + + def test_retries_when_the_drop_happened_before_any_output(self): + """The eval case: nothing streamed yet, so replay is safe.""" + seen = [] + sentinel = MagicMock(name="chat-response") + drop = Exception("peer closed connection (incomplete chunked read)") + attempt, result = self._call( + self._provider(), [drop, sentinel], on_text_chunk=seen.append + ) + self.assertIs(result, sentinel) + self.assertEqual(attempt.call_count, 2) + self.assertEqual(seen, []) + + def test_last_attempt_reraises_the_real_error_not_idle_timeout(self): + """Exhausting attempts must surface the transport error itself — + reporting 'stream idle timeout' for a connection drop sends whoever + reads the log looking for the wrong bug.""" + drop = Exception("peer closed connection without sending complete " + "message body (incomplete chunked read)") + attempt, exc = self._call(self._provider(), [drop, drop, drop]) + self.assertIn("peer closed connection", str(exc)) + self.assertNotIn("idle timeout", str(exc).lower()) + self.assertEqual(attempt.call_count, 3, "must exhaust the attempts") + + def test_auth_error_fails_fast_without_retrying(self): + class APIStatusError(Exception): + def __init__(self, message, status_code): + super().__init__(message) + self.status_code = status_code + + err = APIStatusError("OAuth access token has been revoked.", 401) + attempt, exc = self._call(self._provider(), [err, err, err]) + self.assertIn("revoked", str(exc)) + self.assertEqual(attempt.call_count, 1, "auth failures must not retry")