From f84427682a8bc631a6be2930d5fc4a28678f0514 Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 2 Aug 2026 17:58:33 -0700 Subject: [PATCH 1/5] fix(bash): stop refusing standard flags, and honor bypass mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent defects in the pre-spawn safety guard, both found by a terminal-bench 2.1 differential (2026-08-02). `\breboot\b` matches inside `-no-reboot` — a standard QEMU flag — because `-` is a non-word character. That refused `qemu-system-x86_64 … -no-reboot` as "potentially dangerous": 7 blocked Bash calls across qemu-startup and qemu-alpine-ssh, both of which then scored 0 (the opus baseline solved qemu-alpine-ssh). Replaced the `\b` boundaries on the command-NAME patterns with `(? --- src/tool_system/tools/bash/bash_tool.py | 95 ++++++++++++++--- src/tool_system/tools/monitor.py | 2 +- tests/tools/test_bash_safety_guard.py | 131 ++++++++++++++++++++++++ 3 files changed, 214 insertions(+), 14 deletions(-) create mode 100644 tests/tools/test_bash_safety_guard.py diff --git a/src/tool_system/tools/bash/bash_tool.py b/src/tool_system/tools/bash/bash_tool.py index 0a6915e1c..28ea34b96 100644 --- a/src/tool_system/tools/bash/bash_tool.py +++ b/src/tool_system/tools/bash/bash_tool.py @@ -173,14 +173,30 @@ def _captured(value: Any) -> str: timed_out=timed_out, ) +# ``\b`` is the WRONG boundary for a command NAME: ``-`` is a non-word +# character, so ``\breboot\b`` matches inside ``-no-reboot`` — a standard QEMU +# flag — and ``qemu-system-x86_64 … -no-reboot`` was refused as "potentially +# dangerous". Measured on terminal-bench 2.1 (2026-08-02): 7 refusals across +# ``qemu-startup`` and ``qemu-alpine-ssh``, both of which then scored 0. +# +# ``(? str: + return rf"(? None: +def _is_bypass_permissions(context: Any) -> bool: + """True when the session was DELIBERATELY put in a bypass posture. + + Requires ``is_bypass_permissions_mode_available`` as well as the mode, and + that second condition is load-bearing — it is not belt-and-braces. + ``ToolContext.permission_context`` defaults to + ``ToolPermissionContext(mode="bypassPermissions")`` (context.py:69-71), an + inversion of the TS default that ``entrypoints/mcp_serve.py:11-23`` already + documents and works around. So ``mode`` ALONE is true for every + default-constructed context — including ones built by callers that never + thought about permissions at all — and keying only on it would silently + drop the blocklist far outside the sessions the user opted in for. + + ``is_bypass_permissions_mode_available`` is not part of that default + (``types.py:367`` → ``False``); it is set only where a real session + resolves the posture: ``--dangerously-skip-permissions`` on the headless + path sets both (headless.py:164-172), as do + ``agent_server_cli.py:273`` and ``tui_launcher.py:168``. Requiring both + distinguishes "the user passed the flag" from "nobody set a mode". + + Fails CLOSED in every uncertain case: no context, no permission context, or + a mode without availability all keep the blocklist armed. Deliberately + stricter than the ``_bash_permissions`` passthrough above, which keys on + mode alone — that one governs whether to ASK, this one governs a hard + refusal, so it wants the stronger signal. + """ + perm_ctx = getattr(context, "permission_context", None) + if not getattr(perm_ctx, "is_bypass_permissions_mode_available", False): + return False + return getattr(perm_ctx, "mode", None) in ("bypassPermissions", "plan") + + +def bash_command_safety_guard(command: str, context: Any = None) -> None: """Pre-spawn safety for any shell command run through the bash machinery: the hardcoded-dangerous-pattern block + the C8 sandbox hard-gate. @@ -489,10 +537,31 @@ def bash_command_safety_guard(command: str) -> None: can't be a way around these guards (critic C5-P2). Raises ``ToolPermissionError`` to refuse; the sandbox check is best-effort (a settings problem must not crash the tool), but a hard-gate refusal always - propagates.""" - for pat in _HARDCODED_DANGEROUS_PATTERNS: - if pat.search(command): - raise ToolPermissionError("refusing to run potentially dangerous command") + propagates. + + ``context`` is optional and defaults to None = KEEP GUARDING, so any caller + that doesn't pass one behaves exactly as before. + """ + # The hardcoded blocklist is a port-added defense-in-depth layer with no TS + # counterpart: real Claude Code has no un-bypassable command blocklist, and + # ``--dangerously-skip-permissions`` genuinely skips permission checks + # there. Keeping it unconditional meant a bypass-mode session — the + # explicit "I accept the risk" mode, and the mode every sandboxed eval + # container runs in — still could not run ``sudo`` or a real ``reboot``. + # Same parity direction as PR #673 (removing the un-grantable class safety + # screen). Every non-bypass mode (default / acceptEdits / plan-without- + # bypass) is unaffected. + if not _is_bypass_permissions(context): + for pat in _HARDCODED_DANGEROUS_PATTERNS: + if pat.search(command): + raise ToolPermissionError( + "refusing to run potentially dangerous command" + ) + + # NB the sandbox hard-gate below is deliberately NOT bypassed. It comes + # from MANAGED settings (an admin policy: "never silently run + # unsandboxed"), and a user-supplied CLI flag must not override an + # administrator's control — unlike the blocklist above, which is ours. # Sandbox guard (C8): the port has no sandbox ENFORCEMENT, so a # ``sandbox.enabled`` setting maps onto TS's documented sandbox-unavailable @@ -531,7 +600,7 @@ def _bash_call(tool_input: dict[str, Any], context: ToolContext) -> ToolResult: # sandbox hard-gate. Shared with the Monitor tool (which spawns via # spawn_background_bash directly, bypassing this function) — so the guards # can't drift and Monitor can't be a hole around them. - bash_command_safety_guard(command) + bash_command_safety_guard(command, context) explicit_cwd = tool_input.get("cwd") if explicit_cwd is not None: diff --git a/src/tool_system/tools/monitor.py b/src/tool_system/tools/monitor.py index ba583d9e4..f8c5655a3 100644 --- a/src/tool_system/tools/monitor.py +++ b/src/tool_system/tools/monitor.py @@ -217,7 +217,7 @@ def _monitor_call(tool_input: dict[str, Any], context: ToolContext) -> ToolResul # hard safety gates that live below the permission layer. from .bash.bash_tool import bash_command_safety_guard - bash_command_safety_guard(command) + bash_command_safety_guard(command, context) from .bash.background import spawn_background_bash diff --git a/tests/tools/test_bash_safety_guard.py b/tests/tools/test_bash_safety_guard.py new file mode 100644 index 000000000..223160d30 --- /dev/null +++ b/tests/tools/test_bash_safety_guard.py @@ -0,0 +1,131 @@ +"""Pre-spawn bash safety guard: token boundaries + bypass-mode gating. + +Both behaviours were found by a terminal-bench 2.1 differential (2026-08-02): +``qemu-system-x86_64 … -no-reboot`` was refused as "potentially dangerous" +because ``\\breboot\\b`` matches inside the flag, and the refusal stood even +under ``--dangerously-skip-permissions``. +""" + +from __future__ import annotations + +import pytest + +from src.tool_system.errors import ToolPermissionError +from src.tool_system.tools.bash.bash_tool import bash_command_safety_guard + + +class _PermCtx: + def __init__(self, mode: str, bypass_available: bool = False) -> None: + self.mode = mode + self.is_bypass_permissions_mode_available = bypass_available + + +class _Ctx: + def __init__(self, mode: str, bypass_available: bool = False) -> None: + self.permission_context = _PermCtx(mode, bypass_available) + + +def _bypass() -> _Ctx: + """What ``--dangerously-skip-permissions`` actually resolves to.""" + return _Ctx("bypassPermissions", bypass_available=True) + + +# --- the flags that must stop being refused ------------------------------- + +# Real commands the agent legitimately needs. Each previously matched a +# hardcoded pattern through a hyphen. +ALLOWED = [ + # the exact command terminal-bench's qemu-startup was refused on + "qemu-system-x86_64 -name alpine -m 1024 -cdrom /app/alpine.iso " + "-boot d -display none -no-reboot -daemonize", + "qemu-system-x86_64 -no-reboot -no-shutdown", + "qemu-system-x86_64 --no-reboot", + "systemctl list-units --no-reboot-check", + "./configure --enable-reboot-tests", + "echo reboot-helper", +] + + +@pytest.mark.parametrize("command", ALLOWED) +def test_hyphenated_neighbour_is_not_a_bare_command(command: str) -> None: + bash_command_safety_guard(command) + + +# --- what must STILL be refused ------------------------------------------- + +REFUSED = [ + "reboot", + "reboot -f", + "/sbin/reboot", + "sudo rm -f /etc/passwd", + "shutdown -h now", + "mkfs.ext4 /dev/sda1", + "dd if=/dev/zero of=/dev/sda", + "rm -rf / ", + "rm -rf /", + ":(){ :|:& };:", + # still caught when it is one clause of a compound command + "cd /tmp && sudo make install", +] + + +@pytest.mark.parametrize("command", REFUSED) +def test_real_dangerous_commands_still_refused(command: str) -> None: + with pytest.raises(ToolPermissionError): + bash_command_safety_guard(command) + + +def test_uppercase_still_refused() -> None: + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("SUDO rm -rf /etc") + + +# --- bypass-mode gating --------------------------------------------------- + + +def test_bypass_permissions_skips_the_blocklist() -> None: + bash_command_safety_guard("sudo apt-get install -y qemu", _bypass()) + + +def test_plan_mode_with_bypass_available_skips_the_blocklist() -> None: + bash_command_safety_guard("reboot", _Ctx("plan", bypass_available=True)) + + +@pytest.mark.parametrize("mode", ["default", "acceptEdits", "plan"]) +def test_non_bypass_modes_still_refuse(mode: str) -> None: + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("sudo rm -rf /etc", _Ctx(mode)) + + +def test_bypass_mode_without_availability_still_refuses() -> None: + """The load-bearing half of the predicate. + + ``ToolContext.permission_context`` DEFAULTS to + ``ToolPermissionContext(mode="bypassPermissions")`` while leaving + ``is_bypass_permissions_mode_available`` False. Keying the skip on mode + alone would therefore disarm the blocklist for every default-constructed + context — mcp_serve, tests, any caller that never considered permissions. + Only a session that actually resolved the posture sets both. + """ + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("sudo rm -rf /etc", _Ctx("bypassPermissions")) + + +def test_real_default_tool_context_still_refuses() -> None: + """Pin the above against the REAL dataclass, not a stub.""" + from pathlib import Path + + from src.tool_system.context import ToolContext + + ctx = ToolContext(workspace_root=Path(".")) + assert ctx.permission_context.mode == "bypassPermissions" + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("sudo rm -rf /etc", ctx) + + +def test_missing_context_fails_closed() -> None: + """No context at all must keep guarding — the pre-fix behaviour.""" + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("sudo rm -rf /etc") + with pytest.raises(ToolPermissionError): + bash_command_safety_guard("sudo rm -rf /etc", object()) From 4dbc78b6348bdaf0e6860a3df50f162263a4cf1d Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 2 Aug 2026 18:13:44 -0700 Subject: [PATCH 2/5] fix(providers): bound a stalled stream on the OpenAI-compatible wires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Anthropic wire has had an idle watchdog since WI-5.2. The OpenAI-compatible consumers had no elapsed-time bound of any kind: their poll loops checked `guard.aborted` and nothing else, so a connection that was accepted and then produced nothing hung until something above killed the process. On terminal-bench 2.1 (2026-08-02) model-extraction-relu-logits sat through 880 SECONDS of total silence after a single tool call, then died at harbor's 900 s ceiling having written 2 KB of log. The httpx `read` timeout does not cover this. `read` bounds the gap between BYTES, and a stalled provider is usually still sending — keepalive comments, empty delta frames — so it re-arms indefinitely. That 880 s stall exceeding 120 s x (1 + 1 retry) is the proof it never fired. Byte liveness cannot answer "is this stream still producing anything?". Adds `ContentProgressDeadline`, the semantic sibling of `StreamWatchdog`: progress is recorded for content, reasoning and tool-call deltas and the terminal finish_reason — deliberately NOT for bare frame arrival, since a provider emitting empty frames forever is the failure being caught. Wired into both consumers that had the gap (Chat Completions and the ChatGPT subscription Responses loop), driven from their existing poll ticks, so no timer thread is introduced. Checked on every loop iteration rather than only when the chunk queue is empty. Gating it on Empty looks natural and is wrong for exactly the case this exists for: a stalled-but-chatty stream keeps the queue non-empty, so the check never runs. Both new tests hang against that variant. Threshold is the FIRST-EVENT grace (300 s), not the 90 s inter-event idle, and stays flat instead of tightening after the first delta. Prompt processing on a large context legitimately runs minutes before the first token, and a model doing hidden internal reasoning can legitimately go quiet mid-response; no healthy provider goes five minutes between deltas. Erring long costs one stalled request, erring short would truncate healthy generations on slow providers. A test pins that a slow but productive stream is never interrupted. Recovery is one re-issue through the existing retry wrapper, reusing its `emitted` guard so a stream that already reached the caller is never replayed. StreamIdleTimeout stays a plain Exception and stays out of `is_transport_error`, so the two retry budgets still refuse to compose — see the note in services/api/errors.py. Reuses CLAUDE_STREAM_FIRST_EVENT_TIMEOUT_MS / CLAUDE_STREAM_IDLE_TIMEOUT_MS; no new env surface. Co-Authored-By: Claude Opus 5 --- src/providers/openai_compatible.py | 43 ++++- src/providers/openai_provider.py | 25 ++- src/utils/stream_watchdog.py | 93 +++++++++++ tests/test_openai_compat_stream_idle.py | 208 ++++++++++++++++++++++++ tests/test_openai_subscription.py | 59 +++++++ 5 files changed, 422 insertions(+), 6 deletions(-) create mode 100644 tests/test_openai_compat_stream_idle.py diff --git a/src/providers/openai_compatible.py b/src/providers/openai_compatible.py index 79a296c79..d67af982d 100644 --- a/src/providers/openai_compatible.py +++ b/src/providers/openai_compatible.py @@ -899,8 +899,18 @@ def chat_stream_response( the ORIGINAL exception propagates, so callers see the real cause rather than a retry wrapper. """ + from src.utils.stream_watchdog import StreamIdleTimeout + from .stream_retry import is_transient_stream_drop + def _is_retryable(exc: BaseException) -> bool: + # A content-idle timeout is the same decision as a dropped + # connection: nothing was salvaged, so re-issuing is sound. The + # Anthropic wire has retried its own idle timeouts since WI-5.2 + # (and notes the retry usually starts fast, because attempt 1 + # warmed the prompt cache); this wire now matches. + return isinstance(exc, StreamIdleTimeout) or is_transient_stream_drop(exc) + emitted = [False] def _mark_text(text: str) -> None: @@ -924,11 +934,7 @@ def _mark_thinking(text: str) -> None: **kwargs, ) except Exception as exc: - if ( - attempt >= max_attempts - or emitted[0] - or not is_transient_stream_drop(exc) - ): + if attempt >= max_attempts or emitted[0] or not _is_retryable(exc): raise logger.warning( "Stream dropped (%s); retrying once: %s", @@ -1038,6 +1044,18 @@ def _stream_attempt( import queue as _queue import threading as _threading + from src.utils.stream_watchdog import ContentProgressDeadline + + # This wire had no elapsed-time bound of any kind, so a stream that + # accepted the connection and then produced nothing hung until + # something above killed the process — 880 s of silence on + # terminal-bench's model-extraction-relu-logits (2026-08-02). The + # httpx ``read`` timeout does not cover it (that bounds the gap + # between BYTES, and keepalives are bytes); see the class docstring + # for why content progress is the right signal here and byte liveness + # is the right one on the Anthropic wire. + deadline = ContentProgressDeadline(stream=stream, label=f"model={model}") + _DONE = object() # Bounded so an orphaned worker (abort fired but the SDK # iterator never honors ``response.close()`` — the LiteLLM @@ -1066,6 +1084,13 @@ def _drain_stream() -> None: with guard.attach(stream): worker.start() while True: + # Checked on EVERY iteration, not just when the queue is + # empty. A stalled provider is usually still SENDING — + # keepalive comments, empty delta frames — so the queue is + # never empty and an Empty-only check never runs. That is the + # exact production shape this guards, and gating it on Empty + # reproduced the hang. + deadline.check() try: item = chunk_queue.get(timeout=0.1) except _queue.Empty: @@ -1074,6 +1099,7 @@ def _drain_stream() -> None: # waits between pressing ESC and the prompt # returning, regardless of how slow / blocked the # underlying SDK iteration is. + # if guard.aborted: # Use ``raise_if_post_aborted`` so the abort # reason from the controller is preserved @@ -1105,6 +1131,9 @@ def _drain_stream() -> None: choice = choices[0] if getattr(choice, "finish_reason", None): finish_reason = choice.finish_reason + # The stream is concluding — that is progress even + # though it carries no delta. + deadline.note_progress() delta = getattr(choice, "delta", None) if delta is not None: @@ -1112,16 +1141,20 @@ def _drain_stream() -> None: if content_piece: piece = str(content_piece) content_parts.append(piece) + deadline.note_progress() if on_text_chunk is not None: on_text_chunk(piece) reasoning_piece = _extract_reasoning(delta) if reasoning_piece: reasoning_parts.append(str(reasoning_piece)) + deadline.note_progress() if on_thinking_chunk is not None: on_thinking_chunk(str(reasoning_piece)) tool_call_deltas = getattr(delta, "tool_calls", None) or [] + if tool_call_deltas: + deadline.note_progress() for tc in tool_call_deltas: idx = getattr(tc, "index", 0) entry = tool_calls_by_index.setdefault(idx, {"id": "", "name": "", "arguments": ""}) diff --git a/src/providers/openai_provider.py b/src/providers/openai_provider.py index cd8eb4328..31e83bd5a 100644 --- a/src/providers/openai_provider.py +++ b/src/providers/openai_provider.py @@ -49,6 +49,8 @@ except ModuleNotFoundError: # pragma: no cover OpenAI = None +from src.utils.stream_watchdog import ContentProgressDeadline + from .base import BaseProvider, ChatResponse, MessageInput, TextChunkCallback from .openai_compatible import ( _CHUNK_QUEUE_MAXSIZE, @@ -714,9 +716,25 @@ def _drain() -> None: target=_drain, daemon=True, name=f"openai-subscription-{id(response)}" ) - with guard.attach(_HttpxStreamHolder(response)): + # Same content-progress deadline as the Chat Completions consumer in + # ``openai_compatible.py`` — this loop has the identical shape and had + # the identical gap: nothing bounded a connection that was accepted and + # then produced no events. See ``ContentProgressDeadline`` for why the + # signal is semantic deltas rather than byte liveness. The holder is + # what ``force_close_response`` needs (it exposes ``.response``), so a + # fire also unblocks the worker parked in the socket read. + holder = _HttpxStreamHolder(response) + deadline = ContentProgressDeadline( + stream=holder, label=f"model={request_model}" + ) + + with guard.attach(holder): worker.start() while True: + # Every iteration, not just on Empty — a stalled backend is + # usually still sending keepalive frames, so the queue never + # empties and an Empty-only check would never run. + deadline.check() try: item = line_queue.get(timeout=0.1) except queue.Empty: @@ -740,15 +758,18 @@ def _drain() -> None: delta = str(event.get("delta", "") or "") if delta: content_parts.append(delta) + deadline.note_progress() if on_text_chunk is not None: on_text_chunk(delta) elif etype == "response.reasoning_summary_text.delta": delta = str(event.get("delta", "") or "") if delta: reasoning_parts.append(delta) + deadline.note_progress() if on_thinking_chunk is not None: on_thinking_chunk(delta) elif etype == "response.output_item.done": + deadline.note_progress() raw_item = event.get("item") if isinstance(raw_item, dict): stripped = strip_item_for_replay(raw_item) @@ -762,12 +783,14 @@ def _drain() -> None: ), }) elif etype == "response.completed": + deadline.note_progress() payload = event.get("response") or {} usage = build_usage_dict( payload.get("usage"), subscription=subscription ) response_model = str(payload.get("model") or response_model) elif etype == "response.incomplete": + deadline.note_progress() payload = event.get("response") or {} details = payload.get("incomplete_details") or {} if "max_output_tokens" in str(details.get("reason", "")): diff --git a/src/utils/stream_watchdog.py b/src/utils/stream_watchdog.py index 7cd0f5610..713fa81a1 100644 --- a/src/utils/stream_watchdog.py +++ b/src/utils/stream_watchdog.py @@ -65,6 +65,7 @@ import os import socket as _socket import threading +import time as _time from typing import Any, Callable logger = logging.getLogger(__name__) @@ -73,6 +74,7 @@ "DEFAULT_STREAM_IDLE_TIMEOUT_S", "DEFAULT_STREAM_FIRST_EVENT_TIMEOUT_S", "DEFAULT_STREAM_IDLE_MAX_ATTEMPTS", + "ContentProgressDeadline", "StreamIdleTimeout", "stream_idle_timeout_seconds", "stream_first_event_timeout_seconds", @@ -228,6 +230,97 @@ class StreamIdleTimeout(Exception): """ +class ContentProgressDeadline: + """Bound how long a stream may deliver nothing MEANINGFUL. + + The sibling of :class:`StreamWatchdog`, for the OpenAI-compatible wires. + Where the watchdog asks "are bytes still arriving?" — the right question on + the Anthropic SDK, whose typed iterator hides the ``ping`` keepalives — this + asks "is the stream still producing semantic deltas?". + + That difference is the whole point. The OpenAI-compatible wires already + bound dead air through httpx (``_apply_client_timeout``: ``read`` = the max + gap between BYTES, 120 s). What nothing bounded was a stream that keeps + *bytes* flowing — SSE keepalives, empty delta frames — while never + producing content. Byte liveness re-arms forever in that state, so a + byte-aware watchdog is exactly the wrong instrument for it. Measured on + terminal-bench 2.1 (2026-08-02): 880 s of total silence on + ``model-extraction-relu-logits`` before the harness killed the trial. + + Used from the consumer's existing poll tick — no timer thread, and it can + only fire while the chunk queue is empty, so a stream that is actually + delivering is never interrupted by it. + + The default threshold is the FIRST-EVENT grace (300 s), not the tighter + inter-event idle (90 s), and it stays flat for the whole stream instead of + tightening after the first delta. Prompt processing on a large context + legitimately runs minutes before the first token, and a model doing hidden + internal reasoning can legitimately go quiet mid-response — but no healthy + provider goes five minutes between deltas. Erring long here costs one + stalled request; erring short would truncate healthy generations on slow + providers, which is the strictly worse failure. + + Usage:: + + deadline = ContentProgressDeadline(stream=stream, label=model) + while True: + try: + item = q.get(timeout=0.1) + except Empty: + deadline.check() # raises StreamIdleTimeout when expired + continue + ... + deadline.note_progress() # a real delta arrived + """ + + def __init__( + self, + *, + stream: Any = None, + timeout_s: float | None = None, + label: str = "", + ) -> None: + self._stream = stream + self._timeout_s = ( + timeout_s + if timeout_s is not None + else stream_first_event_timeout_seconds() + ) + self._label = label + self._last = _time.monotonic() + + @property + def timeout_s(self) -> float: + return self._timeout_s + + def note_progress(self) -> None: + """Record that a semantic delta arrived. Call for content, reasoning, + tool-call deltas and the terminal finish_reason — NOT for bare chunk + arrival, which is precisely the signal a stalled-but-chatty stream + keeps producing.""" + self._last = _time.monotonic() + + def expired(self) -> bool: + return _time.monotonic() - self._last > self._timeout_s + + def check(self) -> None: + """Raise :class:`StreamIdleTimeout` if the deadline has lapsed. + + Closes the stream first when one was supplied: the worker thread is + parked in a blocking socket read, and a bare ``close()`` from another + thread does not wake it (see :func:`force_close_response`). Without + that the daemon thread leaks for the life of the connection. + """ + if not self.expired(): + return + if self._stream is not None: + force_close_response(self._stream) + suffix = f" ({self._label})" if self._label else "" + raise StreamIdleTimeout( + f"stream produced no content for {self._timeout_s:.0f}s{suffix}" + ) + + class StreamWatchdog: """Manage a per-stream idle deadline. diff --git a/tests/test_openai_compat_stream_idle.py b/tests/test_openai_compat_stream_idle.py new file mode 100644 index 000000000..374e81646 --- /dev/null +++ b/tests/test_openai_compat_stream_idle.py @@ -0,0 +1,208 @@ +"""A stalled stream on the OpenAI-compatible wire must not hang forever. + +The Anthropic wire has had an idle watchdog since WI-5.2. This wire had no +elapsed-time bound at all, and on terminal-bench 2.1 (2026-08-02) +``model-extraction-relu-logits`` sat for 880 SECONDS of total silence after +one tool call before the harness killed the trial at its 900 s ceiling. + +The httpx ``read`` timeout does not cover it: ``read`` is the gap between +BYTES, and SSE keepalives are bytes. So the deadline here is measured on +SEMANTIC deltas, and the tests below pin both directions — + + * a stream that keeps delivering frames but no deltas MUST fire, and + * a slow-but-productive stream MUST NOT (the "don't hurt other models" + guard; that one is the whole reason the threshold is the 300 s + first-event grace rather than the 90 s inter-event idle). +""" + +from __future__ import annotations + +import threading +import time +import types + +import pytest + +from src.providers.openai_compatible import OpenAICompatibleProvider +from src.utils.stream_watchdog import StreamIdleTimeout + + +# Deadline used by every test here. Real default is 300 s; these override the +# env so the suite stays fast. BOTH vars are needed — +# ``stream_first_event_timeout_seconds`` floors its result at the inter-event +# idle, so setting only the first-event var leaves it pinned at 90 s. +_DEADLINE_MS = 300 +_DEADLINE_S = _DEADLINE_MS / 1000.0 + + +@pytest.fixture(autouse=True) +def _fast_deadline(monkeypatch): + monkeypatch.setenv("CLAUDE_STREAM_FIRST_EVENT_TIMEOUT_MS", str(_DEADLINE_MS)) + monkeypatch.setenv("CLAUDE_STREAM_IDLE_TIMEOUT_MS", str(_DEADLINE_MS)) + + +def _chunk(*, content=None, reasoning=None, finish=None): + """One streaming chunk in the shape the consumer duck-types.""" + delta = types.SimpleNamespace( + content=content, reasoning_content=reasoning, tool_calls=None + ) + choice = types.SimpleNamespace(delta=delta, finish_reason=finish) + return types.SimpleNamespace(choices=[choice], model="m", usage=None) + + +class _FakeStream: + """Iterable stream driven by a script of ``(delay_s, chunk_or_None)``. + + A ``None`` chunk models a frame that carries no semantic delta — the SSE + keepalive case. Iteration stops when the script runs out OR when + ``stop`` is set, so a test that raises mid-stream never leaks the daemon + worker thread. + """ + + def __init__(self, script, *, loop_last=False): + self._script = list(script) + self._loop_last = loop_last + self.stop = threading.Event() + self.frames_yielded = 0 + + def __iter__(self): + for delay, chunk in self._script: + if self.stop.wait(delay): + return + self.frames_yielded += 1 + if chunk is not None: + yield chunk + while self._loop_last and not self.stop.wait(0.02): + # Keepalive frames forever. These must be really YIELDED, not + # merely counted: the consumer's queue then never empties, which + # is what made an Empty-gated deadline check never run and let + # the hang through. Yielding an empty-delta chunk is the shape a + # real stalled provider produces. + self.frames_yielded += 1 + yield _chunk() + + +class _Provider(OpenAICompatibleProvider): + def __init__(self, stream): + super().__init__(api_key="k", model="test-model") + self._stream = stream + self.attempts = 0 + + def _create_client(self): + provider = self + + def create(**_kwargs): + provider.attempts += 1 + return provider._stream + + return types.SimpleNamespace( + chat=types.SimpleNamespace(completions=types.SimpleNamespace(create=create)) + ) + + def get_available_models(self): + return ["test-model"] + + +def _run(provider): + return provider.chat_stream_response([{"role": "user", "content": "hi"}]) + + +# --- must fire ------------------------------------------------------------ + + +def test_keepalive_frames_without_deltas_still_time_out(): + """The actual bug: frames keep arriving, no content ever does. + + Counting bare chunk arrival as progress would make this hang forever, + which is why progress is tracked on semantic deltas only. + """ + stream = _FakeStream([], loop_last=True) + provider = _Provider(stream) + try: + started = time.monotonic() + with pytest.raises(StreamIdleTimeout): + _run(provider) + elapsed = time.monotonic() - started + finally: + stream.stop.set() + + assert stream.frames_yielded > 0, "the stream must really have been delivering" + assert provider.attempts == 2, "an idle timeout should be re-issued once" + assert elapsed < _DEADLINE_S * 6, f"took {elapsed:.2f}s — deadline not enforced" + + +def test_total_silence_times_out(): + stream = _FakeStream([(30.0, None)]) + provider = _Provider(stream) + try: + with pytest.raises(StreamIdleTimeout): + _run(provider) + finally: + stream.stop.set() + assert provider.attempts == 2 + + +def test_stall_after_partial_content_is_not_retried(): + """Retry is skipped once output reached the caller — replaying it would + duplicate the prefix. Same invariant the transport-drop lane enforces.""" + stream = _FakeStream([(0.0, _chunk(content="partial"))], loop_last=True) + provider = _Provider(stream) + seen: list[str] = [] + try: + with pytest.raises(StreamIdleTimeout): + provider.chat_stream_response( + [{"role": "user", "content": "hi"}], on_text_chunk=seen.append + ) + finally: + stream.stop.set() + assert seen == ["partial"] + assert provider.attempts == 1, "must not replay a stream that already emitted" + + +# --- must NOT fire -------------------------------------------------------- + + +def test_slow_but_productive_stream_is_never_interrupted(): + """The regression guard for "don't hurt other models". + + Deltas arrive with gaps just under the deadline, for longer in total + than the deadline itself. A wall-clock cap would kill this; a + content-progress deadline must not. + """ + gap = _DEADLINE_S * 0.6 + stream = _FakeStream([ + (gap, _chunk(content="a")), + (gap, _chunk(content="b")), + (gap, _chunk(content="c")), + (gap, _chunk(finish="stop")), + ]) + provider = _Provider(stream) + out = _run(provider) + assert out.content == "abc" + assert provider.attempts == 1, "a healthy stream must not be re-issued" + + +def test_reasoning_deltas_count_as_progress(): + """A model that thinks for a long time before speaking stays alive.""" + gap = _DEADLINE_S * 0.6 + stream = _FakeStream([ + (gap, _chunk(reasoning="thinking...")), + (gap, _chunk(reasoning="still thinking...")), + (gap, _chunk(content="answer")), + (0.0, _chunk(finish="stop")), + ]) + provider = _Provider(stream) + out = _run(provider) + assert out.content == "answer" + assert out.reasoning_content == "thinking...still thinking..." + assert provider.attempts == 1 + + +def test_short_empty_response_completes_without_firing(): + """An empty completion that ENDS is a model quirk, not a hang — the + query loop's empty-turn nudge owns that case, not this deadline.""" + stream = _FakeStream([(0.0, _chunk(finish="stop"))]) + provider = _Provider(stream) + out = _run(provider) + assert out.content == "" + assert provider.attempts == 1 diff --git a/tests/test_openai_subscription.py b/tests/test_openai_subscription.py index 6e4485c6d..425fc21ba 100644 --- a/tests/test_openai_subscription.py +++ b/tests/test_openai_subscription.py @@ -3,13 +3,17 @@ import base64 import hashlib import json +import threading import time import urllib.parse from pathlib import Path from unittest.mock import patch +import pytest + from src.auth import openai_subscription as auth from src.providers.openai_provider import OpenAIProvider +from src.utils.stream_watchdog import StreamIdleTimeout from src.providers.openai_responses import ( RESPONSES_ITEM_BLOCK_TYPE, build_usage_dict, @@ -660,3 +664,58 @@ def test_provider_validation_accepts_subscription(tmp_path: Path, monkeypatch) - assert provider_has_credentials("anthropic", "") anth.remove_credentials() assert not provider_has_credentials("anthropic", "") + + +# --- content-progress deadline (Responses wire) ----------------------------- + + +class _StallingStreamResponse(_FakeStreamResponse): + """A connection that stays open and keeps emitting frames carrying no + semantic delta — the SSE-keepalive shape a byte-liveness check can never + catch. Terminates on ``stop`` so the daemon worker never leaks.""" + + def __init__(self) -> None: + super().__init__([]) + self.stop = threading.Event() + self.frames = 0 + + def iter_lines(self): + while not self.stop.wait(0.02): + self.frames += 1 + yield ": keep-alive" + + +def test_responses_stream_that_never_produces_content_times_out(monkeypatch) -> None: + """The Responses wire had the same unbounded loop as Chat Completions. + + Without the deadline this test does not fail — it hangs forever, which is + exactly what happened in production. + """ + monkeypatch.setenv("CLAUDE_STREAM_FIRST_EVENT_TIMEOUT_MS", "300") + monkeypatch.setenv("CLAUDE_STREAM_IDLE_TIMEOUT_MS", "300") + provider = _subscription_provider(monkeypatch) + fake_response = _StallingStreamResponse() + + class _FakeClient: + def __init__(self, **kwargs): + pass + + def build_request(self, method, url, headers=None, json=None): + return "request" + + def send(self, request, stream=False): + return fake_response + + def close(self): + pass + + try: + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch("httpx.Client", _FakeClient), pytest.raises(StreamIdleTimeout): + provider.chat_stream_response([{"role": "user", "content": "hi"}]) + finally: + fake_response.stop.set() + + assert fake_response.frames > 0, "the stream must really have been delivering" From f532227297dd1f97c37f37250771322e402d8aab Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 2 Aug 2026 18:15:49 -0700 Subject: [PATCH 3/5] fix(fusion): retry an empty vision response before caching it as failed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_describe` raised on an empty 200 and `_substitute` cached that as a permanent `_Failure`, so one transient empty completion lost the image for the rest of the session. Observed on terminal-bench gcode-to-text (2026-08-02): `openai:gpt-5.6-luna` returned no text for image 2 of 5 and the task scored 0 against a baseline that solved it. The empty case now gets exactly one retry. Nothing else does, and that asymmetry is the point. A transport failure means the provider is unreachable, and caching it permanently is deliberate — the docstring's arithmetic (8 images x 60 s x 2 attempts on every turn, forever, uninterruptible) is why it exists. An empty completion is the opposite situation: the provider is up, answered fast, and just produced nothing. Scoping the retry to it keeps the outage cost at one call per image. Typed as `_EmptyVisionResponse` rather than matched on the message, and the retry is charged to the request's call budget so a provider stuck returning empty 200s cannot double the fan-out that cap bounds. A retry is also skipped outright once the budget's call/time limit is reached. Mutants covered by the new tests: no retry, retrying every failure, and a retry that does not charge the budget. Co-Authored-By: Claude Opus 5 --- src/providers/fusion_provider.py | 54 +++++++++++++++- tests/providers/test_fusion_provider.py | 83 +++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 3 deletions(-) diff --git a/src/providers/fusion_provider.py b/src/providers/fusion_provider.py index 2dad653a0..b488b060b 100644 --- a/src/providers/fusion_provider.py +++ b/src/providers/fusion_provider.py @@ -124,6 +124,15 @@ def expired(self) -> bool: return time.monotonic() >= self.expires_at +class _EmptyVisionResponse(RuntimeError): + """The vision model answered 200 with no usable text. + + Distinguished from every other vision failure because it is the one that + is worth a retry: the provider is reachable and fast, it just produced + nothing this turn. See :meth:`FusionProvider._describe`. + """ + + class _Budget: """Per-request limits on the rewrite: call count, wall clock, and abort. @@ -452,12 +461,48 @@ def _vision_timeout(self, provider: Any) -> Any: except Exception: # noqa: BLE001 — the float still satisfies the SDK return seconds - def _describe(self, source: dict[str, Any]) -> str: + def _describe(self, source: dict[str, Any], budget: "_Budget | None" = None) -> str: """Ask the vision model about one image. Returns its description. Raises on failure; :meth:`_substitute` converts that into a text note so invariant 1 (no image block survives) always holds. + + An EMPTY 200 gets one retry; nothing else does. The distinction is + what keeps the negative cache honest. A transport failure means the + vision provider is unreachable, and ``_substitute`` caches that + permanently on purpose — the docstring's arithmetic (8 images x 60 s + x 2 attempts on every turn, forever, uninterruptible) is why. An empty + completion is the opposite situation: the provider is up, answered + fast, and just produced nothing that turn. Caching THAT permanently + loses the image for the rest of the session over a transient quirk — + observed on terminal-bench gcode-to-text (2026-08-02), where + ``openai:gpt-5.6-luna`` returned no text for image 2 of 5 and the task + scored 0 against a baseline that solved it. + + Bounded at one extra round trip per distinct image, and only while the + request's own call/time budget still allows it, so the outage + arithmetic above is unchanged. """ + try: + return self._describe_once(source) + except _EmptyVisionResponse: + blocked = budget.exhausted() if budget is not None else None + if blocked is not None: + logger.warning( + "[fusion] vision returned no text and %s; not retrying", blocked + ) + raise + if budget is not None: + # A retry is a real network call, so charge it. Otherwise a + # provider stuck returning empty 200s would double the + # fan-out this cap exists to bound. + budget.calls -= 1 + logger.info("[fusion] vision returned no text; retrying once") + return self._describe_once(source) + + def _describe_once(self, source: dict[str, Any]) -> str: + """One vision call. Raises :class:`_EmptyVisionResponse` for an empty + 200 and lets every other failure propagate as-is.""" prompt = self._prompt() provider = self._vision() # The image is handed over in ANTHROPIC block shape and the target @@ -502,7 +547,10 @@ def _describe(self, source: dict[str, Any]) -> str: # response shape drops). Treat it as a failure so the caller # emits an explicit note rather than an empty text block that # reads to the base model as "the image was blank". - raise RuntimeError( + # + # Typed, not a bare RuntimeError: ``_describe`` retries THIS and + # only this, and a string match would be a fragile way to say so. + raise _EmptyVisionResponse( f"vision model {self._fusion.vision.selector} returned no text" ) if len(text) > _MAX_DESCRIPTION_CHARS: @@ -551,7 +599,7 @@ def _substitute(self, block: Any, budget: "_Budget") -> Any: budget.calls -= 1 try: - description = self._describe(source) + description = self._describe(source, budget) except Exception as exc: # noqa: BLE001 — invariant 1: never re-raise # Degrading to a note keeps the turn alive. Re-raising, or # leaving the image in place, reproduces the exact 400 this diff --git a/tests/providers/test_fusion_provider.py b/tests/providers/test_fusion_provider.py index ca3d3b024..6e4158388 100644 --- a/tests/providers/test_fusion_provider.py +++ b/tests/providers/test_fusion_provider.py @@ -647,3 +647,86 @@ def worker(n: int): assert errors == [] # 3 distinct images; the cache may race so allow one extra call each. assert 3 <= len(vision.calls) <= 12 + + +# --- empty-200 retry -------------------------------------------------------- +# +# A vision model that answers 200 with no text is up, fast, and just produced +# nothing that turn. Caching that permanently loses the image for the rest of +# the session over a transient quirk -- terminal-bench gcode-to-text +# (2026-08-02), where the vision leg returned no text for image 2 of 5 and the +# task scored 0 against a baseline that solved it. Transport failures keep the +# old single-attempt + negative-cache behaviour, which is what keeps a real +# outage from costing 2x the calls on every turn forever. + + +class FlakyVision: + """Empty text for the first ``empty_first`` calls, then a description.""" + + def __init__(self, empty_first: int = 1, text: str = "A gcode preview.") -> None: + self.empty_first = empty_first + self.text = text + self.calls = 0 + + def chat(self, messages, **kwargs): + self.calls += 1 + if self.calls <= self.empty_first: + return FakeResponse("") + return FakeResponse(self.text) + + +def test_empty_vision_response_is_retried_once_and_succeeds(): + provider, inner, vision = build(vision=FlakyVision()) + provider.chat([{"role": "user", "content": [image()]}]) + + assert vision.calls == 2, "the empty 200 should have been re-issued once" + assert count_images(inner.seen) == 0 + text = inner.seen[0]["content"][0]["text"] + assert "A gcode preview." in text + assert "could not be described" not in text + + +def test_retry_is_bounded_at_one_extra_call(): + provider, inner, vision = build(vision=FlakyVision(empty_first=99)) + provider.chat([{"role": "user", "content": [image()]}]) + + assert vision.calls == 2, "exactly one retry, not a loop" + assert "could not be described" in inner.seen[0]["content"][0]["text"] + + +def test_transport_failure_is_not_retried(): + """The load-bearing asymmetry: an outage must still cost ONE call. + + Retrying here would restore the 8 images x 60s x 2 attempts every turn + that the negative cache exists to prevent. + """ + provider, inner, vision = build(vision=ExplodingVision()) + provider.chat([{"role": "user", "content": [image()]}]) + + assert vision.calls == 1 + assert "vision is down" in inner.seen[0]["content"][0]["text"] + + +def test_retry_charges_the_call_budget(): + """Two images, a cap of 2, and both need a retry: the cap must count the + retries, so the second image is refused rather than silently doubling + the fan-out the cap exists to bound.""" + provider, inner, vision = build(vision=FlakyVision(), max_images=2) + provider.chat([{"role": "user", "content": [image("A"), image("B")]}]) + + assert vision.calls == 2, "image A's two calls should exhaust a cap of 2" + blocks = inner.seen[0]["content"] + assert "A gcode preview." in blocks[0]["text"] + assert "image limit" in blocks[1]["text"] + assert count_images(inner.seen) == 0 + + +def test_a_retried_failure_is_still_cached_negatively(): + """After the retry also comes back empty, the failure caches as before — + a replayed history must not re-pay for it every turn.""" + provider, inner, vision = build(vision=FlakyVision(empty_first=99)) + msg = [{"role": "user", "content": [image()]}] + provider.chat(copy.deepcopy(msg)) + provider.chat(copy.deepcopy(msg)) + + assert vision.calls == 2, "the second turn must replay the cached failure" From a421fb2ed6d5b84a47a13cb1b0705db02d610c64 Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 2 Aug 2026 18:21:56 -0700 Subject: [PATCH 4/5] fix(headless): emit incremental usage so a killed run stays measurable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ResultEvent` carries the authoritative usage but only exists if the run REACHES the end. A run that is killed — an eval harness hitting its per-task ceiling, a SIGKILL, a dropped connection — emitted nothing, so everything it had spent became unmeasurable. On terminal-bench 2.1 (2026-08-02) every trial missing token metrics was a killed one: 21 of 46 in the fusion job, 17 of 89 in the opus job. Those are the LONGEST and most expensive trials, since length is what makes them time out. Harbor sums job totals from per-trial values, so the headline cost was a floor biased low by exactly the trials that cost the most, and the two jobs' timeout rates differ 2.4x — which makes any cost comparison between them meaningless. Adds a `usage` stream-json event carrying cumulative totals, emitted per assistant message (one model round trip). That granularity is the point: a turn killed before it completes never reaches the per-turn accounting, but it has already produced assistant messages. Verified live — the first usage event lands before the first tool_use, which is exactly where model-extraction-relu-logits died. Accumulated in its own `live_usage` dict rather than the existing `usage_total`: that one folds `result.usage` once per completed turn, and feeding both from either source would double-count. The ResultEvent is untouched. Additive by design — a new event `type`, so consumers that switch on the types they know are unaffected, and `ResultEvent.usage` keeps its exact meaning. The harbor adapter gains it as a third and last-resort lane behind session totals and the result event, never as a replacement. The adapter's arithmetic is now shared through `_usage_columns` so the two lanes cannot drift on counting cached tokens — the bug #786 fixed. Adapter tests skip in the repo's 3.11 venv (the adapter needs 3.12+ `typing.override` and harbor); verified under `uv run --python 3.13 --with harbor`, 7 passed. Co-Authored-By: Claude Opus 5 --- eval/harbor/clawcodex_agent.py | 102 ++++++++++++++---- src/cli_core/__init__.py | 2 + src/cli_core/structured_io.py | 27 +++++ src/entrypoints/headless.py | 36 +++++++ tests/test_headless_usage_events.py | 156 ++++++++++++++++++++++++++++ 5 files changed, 301 insertions(+), 22 deletions(-) create mode 100644 tests/test_headless_usage_events.py diff --git a/eval/harbor/clawcodex_agent.py b/eval/harbor/clawcodex_agent.py index d68f8cfbd..b5cec68ec 100644 --- a/eval/harbor/clawcodex_agent.py +++ b/eval/harbor/clawcodex_agent.py @@ -723,6 +723,7 @@ def populate_context_post_run(self, context: AgentContext) -> None: """ events = self._parse_stream_events() result_event = self._last_result_event(events) + usage_event = self._last_usage_event(events) # Surface an EARLY STOP where a human reading the trial will see it. # clawcodex now marks a run the agent loop cut short with a non-success @@ -752,7 +753,9 @@ def populate_context_post_run(self, context: AgentContext) -> None: } try: - trajectory, totals = self._build_trajectory(events, result_event) + trajectory, totals = self._build_trajectory( + events, result_event, usage_event + ) except Exception as exc: # noqa: BLE001 — never fail a trial over this self.logger.debug(f"clawcodex trajectory build failed: {exc}") trajectory, totals = None, None @@ -772,15 +775,19 @@ def populate_context_post_run(self, context: AgentContext) -> None: context.n_output_tokens = totals["completion"] if totals["cost"] is not None: context.cost_usd = totals["cost"] - elif result_event: - usage = result_event.get("usage") - if isinstance(usage, dict): - input_tokens = usage.get("input_tokens") or 0 - cache_read = usage.get("cache_read_input_tokens") or 0 - cache_creation = usage.get("cache_creation_input_tokens") or 0 - context.n_input_tokens = input_tokens + cache_read + cache_creation - context.n_cache_tokens = cache_read - context.n_output_tokens = usage.get("output_tokens") or 0 + elif result_event and isinstance(result_event.get("usage"), dict): + prompt, cached, completion = self._usage_columns(result_event["usage"]) + context.n_input_tokens = prompt + context.n_cache_tokens = cached + context.n_output_tokens = completion + elif usage_event is not None: + # Killed trial: no copy-back, no result event. The stream log is + # a live bind mount so its last cumulative usage line survived, + # and it is the only measurement of what this run spent. + prompt, cached, completion = self._usage_columns(usage_event["usage"]) + context.n_input_tokens = prompt + context.n_cache_tokens = cached + context.n_output_tokens = completion if trajectory is None: return @@ -819,17 +826,65 @@ def _last_result_event(events: list[dict[str, Any]]) -> dict[str, Any] | None: return event return None + @staticmethod + def _last_usage_event(events: list[dict[str, Any]]) -> dict[str, Any] | None: + """The most recent incremental ``usage`` event, or None. + + The metrics lane of last resort, for a trial that was KILLED. Harbor + cancels the exec on timeout, so the container copy-back never runs + (no session data) and clawcodex never reaches its terminal ``result`` + event (no result usage) — the two sources below. Every killed trial + therefore reported no tokens and no cost at all: 21 of 46 in one + terminal-bench job (2026-08-02), and precisely the longest, most + expensive ones, because those are what time out. Job totals are summed + from per-trial values, so the headline cost was a floor biased low by + exactly the trials that cost the most. + + The stream log survives a kill (it is a live bind mount of the host + trial dir), so the last cumulative usage line in it is a real + measurement of everything the run had spent up to the kill. + """ + for event in reversed(events): + if event.get("type") == "usage" and isinstance(event.get("usage"), dict): + return event + return None + + @staticmethod + def _usage_columns(usage: dict[str, Any]) -> tuple[int, int, int]: + """``(prompt, cached, completion)`` from a stream-json usage dict. + + ``input_tokens`` is only the NON-cached part of the prompt, so a total + built from input+output alone silently omits every cached token — the + bug #786 fixed. One helper so the result-event and usage-event lanes + cannot drift apart on that arithmetic again. + """ + cache_read = usage.get("cache_read_input_tokens") or 0 + cache_creation = usage.get("cache_creation_input_tokens") or 0 + prompt = (usage.get("input_tokens") or 0) + cache_read + cache_creation + return prompt, cache_read, (usage.get("output_tokens") or 0) + def _final_metrics( self, result_event: dict[str, Any] | None, total_steps: int, totals: dict[str, Any] | None, + usage_event: dict[str, Any] | None = None, ) -> FinalMetrics | None: - """``totals`` = authoritative session billing totals (preferred). - Falls back to the stream-json usage when no session cost block was - synced; that fallback now carries cumulative cache counters, so it is - no longer short by the cached portion — though it still omits - subagent and compaction tokens, which only the cost block sees.""" + """Three lanes, in descending order of authority: + + 1. ``totals`` — the session's BILLING totals. Includes subagent and + compaction tokens, which nothing else sees. + 2. the terminal ``result`` event's usage — main loop only, but + carries cumulative cache counters so it is not short by the + cached portion. + 3. the last incremental ``usage`` event — the only one of the three + that survives a KILLED run, where the copy-back never ran and no + result event was ever emitted. See ``_last_usage_event``. + + Cost is unavailable in lane 3 (the incremental event carries tokens + only); harbor prices those itself, and a token count is what the + killed trials were missing entirely. + """ usage = (result_event or {}).get("usage") if not isinstance(usage, dict): usage = {} @@ -839,17 +894,17 @@ def _final_metrics( cached = totals["cached"] cost = totals["cost"] elif result_event: - input_tokens = usage.get("input_tokens") or 0 - cache_read = usage.get("cache_read_input_tokens") or 0 - cache_creation = usage.get("cache_creation_input_tokens") or 0 - prompt = input_tokens + cache_read + cache_creation - completion = usage.get("output_tokens") or 0 - cached = cache_read + prompt, cached, completion = self._usage_columns(usage) cost = result_event.get("total_cost_usd") + elif usage_event: + prompt, cached, completion = self._usage_columns(usage_event["usage"]) + cost = None else: return None extra: dict[str, Any] = {} num_turns = (result_event or {}).get("num_turns") + if not isinstance(num_turns, int) and usage_event: + num_turns = usage_event.get("num_turns") if isinstance(num_turns, int): extra["num_turns"] = num_turns duration_ms = (result_event or {}).get("duration_ms") @@ -992,6 +1047,7 @@ def _build_trajectory( self, events: list[dict[str, Any]], result_event: dict[str, Any] | None, + usage_event: dict[str, Any] | None = None, ) -> tuple[Trajectory | None, dict[str, Any] | None]: """Returns ``(trajectory, billing_totals)`` — the totals are surfaced so the caller can also set the leaderboard token/cost columns.""" @@ -1023,7 +1079,9 @@ def _build_trajectory( agent=agent, steps=steps, notes=notes, - final_metrics=self._final_metrics(result_event, len(steps), totals), + final_metrics=self._final_metrics( + result_event, len(steps), totals, usage_event + ), ) return trajectory, totals diff --git a/src/cli_core/__init__.py b/src/cli_core/__init__.py index acbdb37f5..672de3549 100644 --- a/src/cli_core/__init__.py +++ b/src/cli_core/__init__.py @@ -18,6 +18,7 @@ StreamJsonWriter, SystemEvent, ToolResultEvent, + UsageEvent, ToolUseEvent, UserInputMessage, ) @@ -34,6 +35,7 @@ "StreamJsonWriter", "SystemEvent", "ToolResultEvent", + "UsageEvent", "ToolUseEvent", "UserInputMessage", ] diff --git a/src/cli_core/structured_io.py b/src/cli_core/structured_io.py index f4c362eb6..3f3ccafde 100644 --- a/src/cli_core/structured_io.py +++ b/src/cli_core/structured_io.py @@ -82,6 +82,33 @@ class ToolResultEvent(HeadlessEvent): is_error: bool = False +@dataclass +class UsageEvent(HeadlessEvent): + """Running token totals, emitted as the run proceeds. + + ``ResultEvent`` carries the authoritative usage, but only at the very end. + A run that is killed — an eval harness hitting its per-task ceiling, a + SIGKILL, a lost connection — never emits one, so everything it spent + became unmeasurable. Measured on terminal-bench 2.1 (2026-08-02): 21 of + 46 trials in one job reported no tokens and no cost at all, and they were + exactly the killed ones, i.e. the longest and most expensive. Job totals + are summed from per-trial values, so the headline cost was a floor biased + low by precisely the trials that cost the most. + + Emitted per assistant message (one model round trip), which is the + granularity that survives a kill mid-turn. Cumulative, not per-message, so + a consumer only needs the LAST one it saw and can ignore the rest. + + Additive: a new event ``type``, so existing consumers that switch on the + types they know are unaffected. It does not replace or change + ``ResultEvent.usage``. + """ + + type: str = "usage" + usage: dict[str, Any] = field(default_factory=dict) + num_turns: int = 0 + + @dataclass class ResultEvent(HeadlessEvent): type: str = "result" diff --git a/src/entrypoints/headless.py b/src/entrypoints/headless.py index fcb32dec6..92d27b359 100644 --- a/src/entrypoints/headless.py +++ b/src/entrypoints/headless.py @@ -44,6 +44,7 @@ SystemEvent, ToolResultEvent, ToolUseEvent, + UsageEvent, UserInputMessage, cli_error, ndjson_safe_dumps, @@ -617,6 +618,13 @@ def _with_goal_continuations( aggregate_tool_events: list[dict] = [] num_turns_total = 0 usage_total: dict[str, int] = {} + # Running totals for the incremental UsageEvent lane. Deliberately a + # SEPARATE accumulator from ``usage_total`` above: that one folds + # ``result.usage`` once per completed turn, this one folds each assistant + # message's usage as it lands. Feeding both from either source would + # double-count, and this lane exists precisely for runs that never reach + # the per-turn accounting. + live_usage: dict[str, int] = {} exit_code = 0 # Terminal reason of the LAST agent-loop turn, when it stopped the run # early rather than the model finishing (``tool_failure_loop``, @@ -710,6 +718,34 @@ def _persist(msg: Any) -> None: msg.content, usage=getattr(msg, "usage", None), ) + # Publish the running total as we go. The + # ResultEvent below is authoritative but only + # exists if the run REACHES the end; a killed + # run (eval ceiling, SIGKILL, dropped + # connection) emitted nothing, so everything + # it spent was unmeasurable. Per assistant + # message is the granularity that survives a + # kill mid-turn — a turn that never completes + # never reaches the per-turn accounting below. + # + # Tracked SEPARATELY from ``usage_total``, + # which folds ``result.usage`` once per turn: + # accumulating both from one source would + # double-count. This lane feeds the stream + # only; the ResultEvent is untouched. + _msg_usage = getattr(msg, "usage", None) + if ( + writer is not None + and msg.role == "assistant" + and _msg_usage + ): + _accumulate_usage(live_usage, _msg_usage) + writer.write( + UsageEvent( + usage=dict(live_usage), + num_turns=num_turns_total, + ) + ) # Claude Code's stream-json exposes signed # thinking blocks as assistant content events. # Emit only the private blocks here; the visible diff --git a/tests/test_headless_usage_events.py b/tests/test_headless_usage_events.py new file mode 100644 index 000000000..546393d79 --- /dev/null +++ b/tests/test_headless_usage_events.py @@ -0,0 +1,156 @@ +"""Incremental usage events keep a KILLED run measurable. + +``ResultEvent`` carries the authoritative usage but only exists if the run +reaches the end. A run killed by an eval ceiling / SIGKILL / dropped +connection emitted nothing, so everything it spent was unmeasurable — +21 of 46 trials in one terminal-bench job (2026-08-02), and precisely the +longest and most expensive ones, since those are what time out. +""" + +from __future__ import annotations + +import json + +from src.cli_core import ResultEvent, StreamJsonWriter, UsageEvent + + +class _Sink: + def __init__(self) -> None: + self.lines: list[str] = [] + + def write(self, text: str) -> None: + self.lines.append(text) + + def flush(self) -> None: + pass + + def events(self) -> list[dict]: + out = [] + for line in "".join(self.lines).splitlines(): + line = line.strip() + if line.startswith("{"): + out.append(json.loads(line)) + return out + + +def test_usage_event_serializes_with_a_distinct_type(): + sink = _Sink() + StreamJsonWriter(sink).write( + UsageEvent(usage={"input_tokens": 10, "output_tokens": 3}, num_turns=2) + ) + (event,) = sink.events() + assert event["type"] == "usage" + assert event["usage"] == {"input_tokens": 10, "output_tokens": 3} + assert event["num_turns"] == 2 + + +def test_usage_event_does_not_collide_with_the_result_event(): + """Additive: the terminal result event is unchanged, so a consumer that + only knows ``result`` keeps working exactly as before.""" + sink = _Sink() + writer = StreamJsonWriter(sink) + writer.write(UsageEvent(usage={"input_tokens": 1})) + writer.write(ResultEvent(session_id="s", num_turns=1, result="done")) + types = [e["type"] for e in sink.events()] + assert types == ["usage", "result"] + assert sink.events()[1]["subtype"] == "success" + + +# --- the harbor adapter's fallback chain ---------------------------------- + + +def _adapter(): + """The adapter class without running Harbor's ``__init__``. + + Skipped where it cannot be imported. The adapter is eval-only tooling that + runs under Harbor's own uv-managed CPython 3.13 with harbor installed; it + uses ``typing.override`` (3.12+) and imports ``harbor.*``, so the repo's + 3.11 test venv cannot load it. Run these under that interpreter: + + uv run --python 3.13 --with harbor -m pytest \\ + tests/test_headless_usage_events.py + """ + import sys + from pathlib import Path + + import pytest + + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "eval" / "harbor")) + try: + from clawcodex_agent import Clawcodex + except Exception as exc: # noqa: BLE001 — environment, not a failure + pytest.skip(f"harbor adapter not importable here: {exc}") + + return Clawcodex + + +def test_last_usage_event_is_the_cumulative_one(): + Clawcodex = _adapter() + events = [ + {"type": "usage", "usage": {"input_tokens": 5}}, + {"type": "tool_use", "name": "Bash"}, + {"type": "usage", "usage": {"input_tokens": 40}}, + {"type": "tool_use", "name": "Read"}, + ] + found = Clawcodex._last_usage_event(events) + assert found["usage"]["input_tokens"] == 40, "must take the LAST, not the first" + + assert Clawcodex._last_usage_event([{"type": "tool_use"}]) is None + # A malformed usage payload must not be mistaken for a real one. + assert Clawcodex._last_usage_event([{"type": "usage", "usage": "nope"}]) is None + + +def test_usage_columns_counts_cached_tokens(): + """``input_tokens`` is only the NON-cached part — a total built from + input+output alone silently omits every cached token (the #786 bug).""" + Clawcodex = _adapter() + prompt, cached, completion = Clawcodex._usage_columns({ + "input_tokens": 100, + "cache_read_input_tokens": 900, + "cache_creation_input_tokens": 50, + "output_tokens": 7, + }) + assert prompt == 1050 + assert cached == 900 + assert completion == 7 + + +def test_final_metrics_falls_back_to_the_usage_event(): + """The killed-trial lane: no session totals, no result event.""" + Clawcodex = _adapter() + adapter = Clawcodex.__new__(Clawcodex) + metrics = Clawcodex._final_metrics( + adapter, + None, # result_event — a killed run never emits one + 12, # total_steps + None, # totals — copy-back never ran + {"type": "usage", "num_turns": 4, + "usage": {"input_tokens": 100, "cache_read_input_tokens": 900, + "output_tokens": 7}}, + ) + assert metrics is not None, "a killed trial must still report tokens" + assert metrics.total_prompt_tokens == 1000 + assert metrics.total_completion_tokens == 7 + + +def test_final_metrics_prefers_the_result_event_over_the_usage_event(): + """Lane order matters: the usage event is a LAST resort, never a + replacement for the authoritative terminal numbers.""" + Clawcodex = _adapter() + adapter = Clawcodex.__new__(Clawcodex) + metrics = Clawcodex._final_metrics( + adapter, + {"type": "result", "usage": {"input_tokens": 999, "output_tokens": 99}, + "num_turns": 9}, + 3, + None, + {"type": "usage", "usage": {"input_tokens": 1, "output_tokens": 1}}, + ) + assert metrics.total_prompt_tokens == 999 + assert metrics.total_completion_tokens == 99 + + +def test_final_metrics_still_returns_none_with_no_source_at_all(): + Clawcodex = _adapter() + adapter = Clawcodex.__new__(Clawcodex) + assert Clawcodex._final_metrics(adapter, None, 0, None, None) is None From 61b1eaeb418cd0bf4332ee8d47629a8b5baec777 Mon Sep 17 00:00:00 2001 From: Eric Lee Date: Sun, 2 Aug 2026 18:34:28 -0700 Subject: [PATCH 5/5] style(providers): drop a stray comment marker Co-Authored-By: Claude Opus 5 --- src/providers/openai_compatible.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/providers/openai_compatible.py b/src/providers/openai_compatible.py index d67af982d..c9214efd0 100644 --- a/src/providers/openai_compatible.py +++ b/src/providers/openai_compatible.py @@ -1099,7 +1099,6 @@ def _drain_stream() -> None: # waits between pressing ESC and the prompt # returning, regardless of how slow / blocked the # underlying SDK iteration is. - # if guard.aborted: # Use ``raise_if_post_aborted`` so the abort # reason from the controller is preserved