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/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/src/providers/openai_compatible.py b/src/providers/openai_compatible.py index 79a296c79..c9214efd0 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: @@ -1105,6 +1130,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 +1140,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/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/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/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" 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 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" 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())