fix(harness): recover trials lost to empty turns, transport drops, and a headless-only tool

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

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.

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:

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 "

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)",

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;

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
     )

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):

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)
 

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)

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"