diff --git a/devops_bench/agents/cli/claude_code/__init__.py b/devops_bench/agents/cli/claude_code/__init__.py new file mode 100644 index 00000000..efa1cfce --- /dev/null +++ b/devops_bench/agents/cli/claude_code/__init__.py @@ -0,0 +1,27 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Claude Code CLI agent harness, named to distinguish it from the Claude model. + +The harness driver lives in :mod:`.agent` and the stream-json parser in +:mod:`.parsing`. Importing this package self-registers the agent under the +``"claude"`` key via ``@AGENTS.register``. +""" + +from __future__ import annotations + +from devops_bench.agents.cli.claude_code.agent import ClaudeCodeAgent +from devops_bench.agents.cli.claude_code.parsing import parse_stream_json + +__all__ = ["ClaudeCodeAgent", "parse_stream_json"] diff --git a/devops_bench/agents/cli/claude_code/agent.py b/devops_bench/agents/cli/claude_code/agent.py new file mode 100644 index 00000000..ec556305 --- /dev/null +++ b/devops_bench/agents/cli/claude_code/agent.py @@ -0,0 +1,398 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Claude Code CLI agent harness driving the ``claude`` binary. + +Runs ``claude`` in headless mode (``-p --output-format stream-json --verbose``) +and extracts the canonical trajectory from the official event stream on stdout +(see :mod:`~devops_bench.agents.cli.claude_code.parsing`) — no session-file +reads off disk. + +Capability wiring uses Claude Code's native cwd-based channels, written into the +per-run working directory before invocation: + +* **Rules** — ``config.capabilities.rules.text`` → ``CLAUDE.md``, auto-loaded + from the cwd as the startup context. +* **Skills** — ``config.capabilities.skills.paths`` are materialized under + ``/.claude/skills//SKILL.md``, Claude Code's skill-discovery root. +* **MCP servers** — command-bearing bindings become a ``{"mcpServers": ...}`` + document at ``/.claude/mcp-config.json``, passed via ``--mcp-config``. + ``--strict-mcp-config`` is always set — including on a baseline arm with no + bindings — so any stray project ``.mcp.json`` is ignored and no trust prompt + fires. A bound run first checks the binary against + :data:`_MCP_WAIT_MIN_VERSION`. + +Auth is env-driven, matching the bench contract: ``config.api_key`` → +``ANTHROPIC_API_KEY`` for the direct API, or keyless Vertex / Bedrock via ADC / +AWS credentials. ``CLAUDE_CONFIG_DIR`` is redirected to a fresh per-run temp dir +so Claude Code's mutable global state never races across concurrent evals (see +:func:`_claude_config_dir` for the OAuth-debug escape hatch, which ``--parallel`` +overrides). +""" + +from __future__ import annotations + +import contextlib +import json +import os +import re +import tempfile +from collections.abc import Iterator +from functools import cache +from pathlib import Path + +from devops_bench.agents.base import AGENTS, AgentHarness +from devops_bench.agents.cli.claude_code.parsing import parse_stream_json +from devops_bench.agents.config import AgentConfig +from devops_bench.agents.result import AgentResult +from devops_bench.agents.shared.cli_capabilities import ( + agent_workdir, + build_mcp_servers, + materialize_skills, +) +from devops_bench.core import SubprocessError, get_logger +from devops_bench.core.config import get_bool +from devops_bench.core.model_providers import resolve_provider +from devops_bench.core.subprocess import run + +__all__ = ["ClaudeCodeAgent"] + +# Auto-loaded from the cwd as the operator brief (native system-prompt analog). +_CLAUDE_RULES_FILE = "CLAUDE.md" +# Workspace config dir/files Claude Code reads from its cwd: ``skills/`` is the +# skill-discovery root and ``mcp-config.json`` is passed via ``--mcp-config``. +_CLAUDE_CONFIG_DIRNAME = ".claude" +_CLAUDE_SKILLS_DIR = "skills" +_CLAUDE_MCP_FILE = "mcp-config.json" +# Relocates Claude Code's mutable global state (see _claude_config_dir). +_CONFIG_DIR_ENV = "CLAUDE_CONFIG_DIR" + +_log = get_logger("agents.cli.claude_code") + +# Child stderr is unbounded and reaches the persisted result record (both +# ``metadata`` and ``errors``), so every path clips it to the same tail. +_STDERR_TAIL_CHARS = 2000 + +# Under ``-p`` the CLI reads a piped prompt from stdin and waits out a 3s +# timeout before giving up when stdin stays open with nothing on it — which is +# what an inherited stdin looks like. Handing it an empty string closes the pipe +# immediately: ~3.7s saved per task, and no "no stdin data received" warning +# polluting ``metadata["stderr"]`` (and, on a non-zero exit, the error message, +# where it would displace the real cause). +_CLOSED_STDIN = "" + + +def _stderr_tail(stderr: str | None) -> str: + """Stripped last :data:`_STDERR_TAIL_CHARS` characters of ``stderr``.""" + return (stderr or "").strip()[-_STDERR_TAIL_CHARS:] + + +# ``--mcp-config`` under ``-p`` only waits for still-pending servers to connect +# before the first turn from this version on. An older binary accepts the flag +# and starts the turn anyway, so an MCP-augmented arm can run with none of its +# tools attached and still be scored as augmented — the same silent +# contamination ``--strict-mcp-config`` prevents from the other direction. +_MCP_WAIT_MIN_VERSION = (2, 1, 221) +_VERSION_RE = re.compile(r"(\d+)\.(\d+)\.(\d+)") + + +@cache +def _claude_version(target: str) -> tuple[int, int, int] | None: + """Parse ``claude --version``, or ``None`` when it cannot be determined. + + An unreadable version is not an error: ``config.target`` may be a wrapper + script with its own ``--version`` surface, and refusing to run on a probe + that merely failed to parse would be worse than the risk it guards. + + Cached per target: a matrix run drives one binary across every task, and the + version cannot change under a live process. An inconclusive probe caches too, + so a wrapper without a ``--version`` surface is not re-spawned per task. + """ + try: + completed = run([target, "--version"], check=False, timeout=30) + except (OSError, SubprocessError): + return None + match = _VERSION_RE.search(completed.stdout or "") + if completed.returncode != 0 or not match: + return None + return (int(match[1]), int(match[2]), int(match[3])) + + +def _build_argv( + target: str, + prompt: str, + *, + model: str | None, + max_turns: int | None, + mcp_config_path: str | None, +) -> list[str]: + """Build the ``claude`` headless invocation for ``prompt``. + + ``--verbose`` is mandatory (the CLI rejects ``stream-json`` under ``-p`` + without it) and ``--dangerously-skip-permissions`` keeps headless runs from + blocking on confirmation prompts. + + ``--strict-mcp-config`` is passed unconditionally, not just alongside + ``--mcp-config``: it pins the run to exactly the bound servers, and for a + baseline arm that set is empty. Without it, a stray ``.mcp.json`` in the task + workspace (the CLI's cwd) is loaded and silently grants MCP tools to the + un-augmented arm, contaminating the very comparison the bench measures. + + The prompt is the trailing positional after ``--``. The CLI's option parser + would otherwise read a prompt whose first token begins with ``-`` as a flag + and abort the run. + + Args: + target: Path to the ``claude`` binary (already user-expanded). + prompt: Task prompt, passed as an argv value (never through a shell). + model: Model id for ``--model``, or ``None`` to use the CLI default. + max_turns: Cap for ``--max-turns``; ``None`` or non-positive uses the + CLI default (the CLI rejects ``0``, so it is treated as unset). + mcp_config_path: Absolute path to the MCP config document, or ``None``. + + Returns: + The argv list ready to hand to ``core.subprocess.run``. + """ + argv = [ + target, + "-p", + "--output-format", + "stream-json", + "--verbose", + "--dangerously-skip-permissions", + "--strict-mcp-config", + ] + if model: + argv.extend(["--model", model]) + if max_turns is not None and max_turns > 0: + argv.extend(["--max-turns", str(max_turns)]) + if mcp_config_path: + argv.extend(["--mcp-config", mcp_config_path]) + argv.extend(["--", prompt]) + return argv + + +def _build_env(config: AgentConfig, *, config_dir: str | None) -> dict[str, str]: + """Build the env overlay that makes the Claude Code run model-agnostic. + + ``config.api_key`` routes onto the provider's key env var(s) via the shared + contract (default ``anthropic``); Vertex / Bedrock backends set their + ``CLAUDE_CODE_USE_*`` switch, with Vertex mapping the repo's ambient + ``GCP_PROJECT_ID`` / ``GCP_VERTEX_LOCATION`` onto the CLI's equivalents. The + model is never set here — it flows through the ``--model`` argv flag. + + Args: + config: Resolved :class:`AgentConfig` for this run. + config_dir: Per-run ``CLAUDE_CONFIG_DIR`` path, or ``None`` when the + operator exported their own (then the ambient value is left intact). + + Returns: + A mapping suitable for ``core.subprocess.run``'s ``extra_env``. + + Raises: + ConfigError: If ``config.provider`` is not a known provider. + """ + # Resolve unconditionally so an unknown provider fails loud even on a keyless + # (Vertex/Bedrock) run, not only when a key happens to be set. + spec = resolve_provider(config.provider, default="anthropic") + overlay: dict[str, str] = { + # Headless hygiene: no background telemetry/error traffic, no autoupdate. + "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1", + "DISABLE_AUTOUPDATER": "1", + } + # Guard on truthiness: run() overlays extra_env onto os.environ, so writing + # an empty key would clobber an ambient ANTHROPIC_API_KEY. + if config.api_key: + for var in spec.api_key_envs: + overlay[var] = config.api_key + if spec.backend == "vertex": + overlay["CLAUDE_CODE_USE_VERTEX"] = "1" + project = os.environ.get("GCP_PROJECT_ID") + if project: + overlay["ANTHROPIC_VERTEX_PROJECT_ID"] = project + # Prefer the repo var, then an operator-set native CLOUD_ML_REGION, so we + # only fall back to "global" when neither is set (never clobber it). + region = os.environ.get("GCP_VERTEX_LOCATION") or os.environ.get("CLOUD_ML_REGION") + overlay["CLOUD_ML_REGION"] = region or "global" + elif spec.backend == "bedrock": + overlay["CLAUDE_CODE_USE_BEDROCK"] = "1" + if config_dir is not None: + overlay[_CONFIG_DIR_ENV] = config_dir + # Operator-supplied extra_env is applied last and deliberately wins over the + # harness-set keys above (its escape hatch for backend/region overrides). + if config.extra_env: + overlay.update(config.extra_env) + return overlay + + +@contextlib.contextmanager +def _claude_config_dir() -> Iterator[str | None]: + """Yield a fresh per-run ``CLAUDE_CONFIG_DIR``, or ``None`` if operator-set. + + A non-empty ambient ``CLAUDE_CONFIG_DIR`` is the operator's escape hatch + (e.g. to reuse a cached OAuth login for local debugging); it is left + untouched and ``None`` is yielded so no per-run temp dir is created or + injected. An empty ambient value is ignored so per-run isolation still + applies (the CLI treats an empty var as unset and would race on ~/.claude). + + The hatch is refused under ``BENCH_PARALLEL``. Several benchmark processes + on one host would then share a single mutable config dir — the collision + class :mod:`devops_bench.core.run_env` exists to prevent — and the operator + has already declared concurrency, so isolation outranks the login cache. + """ + if os.environ.get(_CONFIG_DIR_ENV): + if not get_bool("BENCH_PARALLEL", False): + yield None + return + _log.warning( + "ignoring ambient %s under BENCH_PARALLEL: concurrent runs would share " + "one mutable Claude config dir; using a per-run dir instead", + _CONFIG_DIR_ENV, + ) + # ignore_cleanup_errors: Claude Code may leave straggler state/lock files or + # MCP-server children; a cleanup OSError must not turn a completed run into + # an errored one via the base safety net. + with tempfile.TemporaryDirectory(prefix="claude-config-", ignore_cleanup_errors=True) as tmpdir: + yield tmpdir + + +@AGENTS.register("claude") +class ClaudeCodeAgent(AgentHarness): + """Claude Code CLI agent harness driving the ``claude`` binary. + + The binary path is resolved from ``config.target``, falling back to + ``"claude"`` on ``$PATH``. Model / API key flow from ``config.model`` (via + ``--model``) / ``config.api_key`` (via the env overlay) — never hardcoded. + + ``__init__`` assigns ``self.mcp_servers``, ``self.skills`` and ``self.rules`` + from the granted config bindings, which is what makes + ``isinstance(agent, SupportsMcp / SupportsSkills / SupportsRules)`` return + ``True`` for orchestrator-side capability negotiation (the Protocols are + structural). + + """ + + def __init__(self, config: AgentConfig | None = None) -> None: + AgentHarness.__init__(self, config) + caps = self.config.capabilities + self.mcp_servers = caps.mcp_servers + self.skills = caps.skills + self.rules = caps.rules + + def _execute(self, prompt: str, workspace_path: Path | None = None) -> AgentResult: + """Build argv, run the CLI, and parse the stream-json output. + + Capabilities are laid down in the run directory first via the cwd + channels described in the module docstring. The temp working directory + (when ``workspace_path`` is ``None``) and the per-run + ``CLAUDE_CONFIG_DIR`` are cleaned up on return; a harness-supplied + ``workspace_path`` is left for the harness to collect. + """ + caps = self.config.capabilities + target = os.path.expanduser(self.config.target or "claude") + rules_text = caps.rules.text + + with agent_workdir(workspace_path, prefix="claude-run-") as workdir: + if rules_text: + (workdir / _CLAUDE_RULES_FILE).write_text(rules_text, encoding="utf-8") + + claude_dir = workdir / _CLAUDE_CONFIG_DIRNAME + materialize_skills(claude_dir / _CLAUDE_SKILLS_DIR, caps.skills.paths) + + mcp_config_path: str | None = None + servers = build_mcp_servers(caps.mcp_servers) + if servers: + claude_dir.mkdir(parents=True, exist_ok=True) + mcp_path = claude_dir / _CLAUDE_MCP_FILE + mcp_path.write_text(json.dumps({"mcpServers": servers}, indent=2), encoding="utf-8") + # A binding's argv can carry a server credential, and this file + # lands in the run workspace the harness later collects, so do + # not leave it at the umask default (0o644 on most machines). + mcp_path.chmod(0o600) + mcp_config_path = str(mcp_path) + version = _claude_version(target) + if version is not None and version < _MCP_WAIT_MIN_VERSION: + return AgentResult.errored( + "claude " + + ".".join(str(part) for part in version) + + " predates the --mcp-config startup wait (needs " + + ".".join(str(part) for part in _MCP_WAIT_MIN_VERSION) + + "); an MCP-bound arm would run without its servers. " + "Upgrade the binary or drop the mcp_servers binding." + ) + + argv = _build_argv( + target, + prompt, + model=self.config.model, + max_turns=self.config.max_turns, + mcp_config_path=mcp_config_path, + ) + with _claude_config_dir() as config_dir: + env_overlay = _build_env(self.config, config_dir=config_dir) + try: + completed = run( + argv, + extra_env=env_overlay, + cwd=workdir, + check=False, + timeout=self.config.timeout_sec, + input=_CLOSED_STDIN, + ) + except SubprocessError as exc: + # Under ``check=False`` the only way ``run`` raises is a + # timeout, so report it as one rather than as an exit -1 that + # reads like a crash. str(exc) embeds the child's full + # stderr, so rebuild the message from the clipped tail rather + # than interpolating it. The timeout carries the partial + # stream-json captured before the kill; fall through so the + # trajectory is recovered rather than dropped. + stderr = _stderr_tail(exc.stderr) + returncode = exc.returncode + stdout = exc.stdout or "" + reason = f"claude timed out after {self.config.timeout_sec}s" + if stderr: + reason += f": {stderr}" + except OSError as exc: + # Spawn failure core.subprocess.run does not wrap: usually a + # missing / non-executable binary, but also a vanished cwd. + return AgentResult.errored(f"failed to spawn claude: {exc}") + else: + stderr = _stderr_tail(completed.stderr) + returncode = completed.returncode + stdout = completed.stdout or "" + reason = ( + None + if returncode == 0 + else f"claude exited {returncode}: {stderr or ''}" + ) + + output, trajectory, tokens, parse_errors = parse_stream_json(stdout) + errors: list[str] = list(parse_errors) + metadata: dict = {} + if stderr: + # Keep stderr for diagnosis even on a clean exit — e.g. MCP startup + # warnings that leave the process returncode at 0. + metadata["stderr"] = stderr + if reason is not None: + errors.append(reason) + metadata["returncode"] = returncode + output = output or f"Error: {reason}" + return AgentResult( + output=output, + trajectory=trajectory, + tokens=tokens, + errors=errors, + metadata=metadata, + ) diff --git a/devops_bench/agents/cli/claude_code/parsing.py b/devops_bench/agents/cli/claude_code/parsing.py new file mode 100644 index 00000000..90043315 --- /dev/null +++ b/devops_bench/agents/cli/claude_code/parsing.py @@ -0,0 +1,378 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Parser for the Claude Code ``--output-format stream-json`` event stream. + +Folds the ``tool_use`` / ``tool_result`` content blocks into the canonical +:class:`ToolCall` list and pulls the final answer and token usage from the +terminal ``result`` event. ``thinking`` / ``redacted_thinking`` blocks are +dropped so the trajectory matches the tool-calls-only shape the other CLI +harnesses emit. +""" + +from __future__ import annotations + +import json +from collections.abc import Iterator + +from devops_bench.agents.result import ToolCall, empty_tokens + +__all__ = ["parse_stream_json"] + + +def _int_or_none(val: object) -> int | None: + """Return ``val`` if it is a real ``int`` (``bool`` rejected), else ``None``. + + Mirrors ``results/normalize._coerce_int``'s bool rejection so a stray JSON + ``true`` in a usage field never surfaces as a token count. + """ + return val if isinstance(val, int) and not isinstance(val, bool) else None + + +# Claude Code echoes the full body of every tool result into the stream, so an +# uncapped trajectory grows with the agent's file reads and command output. The +# whole trace is re-serialized into the LLM judge prompts downstream, where a +# multi-megabyte trace overruns the judge's context, so keep a head and tail +# slice — metric matching only needs the edges, not the middle. +_RESULT_HEAD_CHARS = 12000 +_RESULT_TAIL_CHARS = 4000 +_RESULT_MAX_CHARS = _RESULT_HEAD_CHARS + _RESULT_TAIL_CHARS + + +def _clip_result(text: str) -> str: + """Head+tail slice ``text`` to :data:`_RESULT_MAX_CHARS`, marking the elision.""" + if len(text) <= _RESULT_MAX_CHARS: + return text + elided = len(text) - _RESULT_MAX_CHARS + return ( + f"{text[:_RESULT_HEAD_CHARS]}\n... [{elided} chars elided] ...\n" + f"{text[-_RESULT_TAIL_CHARS:]}" + ) + + +def _block_text(content: object) -> str | None: + """Render a tool_result ``content`` payload to a string, or ``None``. + + Claude Code emits tool results either as a bare string or as a list of + content blocks. Text blocks contribute their text; any other block (e.g. an + ``image``) is JSON-encoded in place so nothing is dropped silently. The + rendered text is clipped (see :func:`_clip_result`). + """ + if content is None: + return None + if isinstance(content, str): + return _clip_result(content) + if isinstance(content, list): + if not content: + return None + parts = [ + block["text"] + if isinstance(block, dict) and isinstance(block.get("text"), str) + else json.dumps(block, default=str) + for block in content + ] + return _clip_result("".join(parts)) + return _clip_result(json.dumps(content, default=str)) + + +# Claude Code namespaces MCP tools as ``mcp____``. The rest of the +# pipeline uses the ``__`` convention (the metrics canonicalizer +# strips exactly one ``__`` segment to recover the bare tool name), so +# drop the literal ``mcp__`` client prefix to stay consistent with the other +# harnesses. Built-in tools (``Bash``, ``Read``, ...) carry no prefix. +_MCP_TOOL_PREFIX = "mcp__" + + +def _normalize_tool_name(name: str) -> str: + """Strip Claude Code's ``mcp__`` client prefix from an MCP tool name.""" + if name.startswith(_MCP_TOOL_PREFIX): + return name[len(_MCP_TOOL_PREFIX) :] + return name + + +# Terminal-failure MCP statuses in the ``init`` event. Transient ``pending`` / +# ``connecting`` states are excluded: a stdio server (e.g. gke-mcp) often reports +# them at init yet connects moments later, so flagging them would false-positive +# on a working run. +_MCP_FAILED_STATUSES = frozenset({"failed", "error", "disconnected", "needs-auth", "needs_auth"}) + + +def _iter_events(stdout: str) -> Iterator[tuple[object, str | None]]: + """Yield ``(event, error)`` pairs from the stream, one populated per item. + + The stream is normally newline-delimited JSON, but a rebuffered or truncated + pipe can concatenate several objects onto one physical line. Each line is + decoded with ``raw_decode`` in a loop so every object is recovered rather + than lost to a single ``Extra data`` error. A malformed remainder yields one + error and the rest of that line is abandoned. + + Splitting on ``"\\n"`` rather than :meth:`str.splitlines` is deliberate: the + CLI leaves U+0085 (NEL) unescaped inside JSON strings, and ``splitlines`` + would treat it as a line break, shredding that event into two undecodable + fragments. + """ + decoder = json.JSONDecoder() + for lineno, raw in enumerate(stdout.split("\n"), start=1): + line = raw.strip() + if not line: + continue + idx = 0 + while idx < len(line): + while idx < len(line) and line[idx].isspace(): + idx += 1 + if idx >= len(line): + break + try: + event, idx = decoder.raw_decode(line, idx) + except json.JSONDecodeError as exc: + yield None, f"stream-json line {lineno} parse error: {exc}" + break + yield event, None + + +def parse_stream_json(stdout: str) -> tuple[str, list[dict], dict, list[str]]: + """Parse a Claude Code ``--output-format stream-json`` stdout stream. + + The stream is newline-delimited JSON in the wrapped SDK form: each line is + an envelope with a top-level ``type`` (``system`` / ``assistant`` / ``user`` + / ``result``) carrying a nested Anthropic ``message`` object. The parser is + intentionally lenient (unknown event types are skipped) and surfaces both + per-line JSON decode errors and unmatched ``tool_result`` blocks on the + ``errors`` list rather than dropping them. + + | Event type | Handling | + |---------------|-----------------------------------------------------------| + | ``system`` | ``init`` MCP statuses checked; other metadata ignored | + | ``assistant`` | ``tool_use`` → pending ToolCalls; ``text`` → output; | + | | ``thinking`` / ``redacted_thinking`` dropped | + | ``user`` | ``tool_result`` blocks matched to pending ToolCalls | + | ``result`` | terminal: authoritative answer, token usage, failure flag | + + The accumulated assistant ``text`` doubles as a fallback answer when no + terminal ``result`` event arrives (a truncated pipe) or when it carries an + empty answer (error subtypes emit ``""``). Likewise token usage falls back to + the per-turn accumulator only when the terminal event reports no usage; + per-turn usage is deduped by message id, since Claude Code repeats the same + ``usage`` on every content-block envelope of one API message. That fallback + recovers the prompt-side buckets only (see :data:`_ACC_USAGE_KEYS`). + + Args: + stdout: Raw process stdout, possibly empty. + + Returns: + A ``(output, trajectory, tokens, errors)`` tuple. ``trajectory`` is a + list of ``ToolCall.to_dict()`` mappings ordered as emitted. + """ + text_parts: list[str] = [] + result_output: str | None = None + tokens: dict = empty_tokens() + result_usage_seen = False + acc_usage: dict = {} + seen_usage_ids: set[str] = set() + errors: list[str] = [] + # Each id maps to a FIFO queue of pending calls: distinct tool_use blocks can + # legitimately reuse an id, so results are matched in emission order rather + # than the second call silently overwriting the first. + pending: dict[str, list[ToolCall]] = {} + trajectory: list[ToolCall] = [] + + for event, error in _iter_events(stdout): + if error is not None: + errors.append(error) + continue + if not isinstance(event, dict): + continue + + etype = event.get("type") + if etype == "system": + # A failed MCP server leaves the run tool-less but still exits 0, so + # surface terminal-failure statuses (see ``_MCP_FAILED_STATUSES``) + # rather than scoring a silently-degraded run clean. + if event.get("subtype") == "init": + mcp_servers = event.get("mcp_servers") + if not isinstance(mcp_servers, list): + continue + for server in mcp_servers: + if not isinstance(server, dict): + continue + status = str(server.get("status", "")).lower() + if status in _MCP_FAILED_STATUSES: + name = server.get("name") or "" + errors.append(f"mcp server {name!r} failed to connect at init: {status}") + elif etype == "assistant": + message = event.get("message") + if not isinstance(message, dict): + continue + # Accumulate per-turn usage so a truncated stream (no terminal + # ``result`` event) still yields token counts. Claude Code emits one + # envelope per content block of a single API message, each repeating + # the identical ``usage``, so count each message id only once. + msg_id = message.get("id") + if not (isinstance(msg_id, str) and msg_id in seen_usage_ids): + if isinstance(msg_id, str): + seen_usage_ids.add(msg_id) + _add_usage(acc_usage, message.get("usage")) + content = message.get("content") + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict): + continue + btype = block.get("type") + if btype == "text": + if isinstance(block.get("text"), str): + text_parts.append(block["text"]) + elif btype in ("thinking", "redacted_thinking"): + continue # not part of the tool-calls-only trajectory + elif btype == "tool_use": + args = block.get("input") + raw_name = block.get("name") + call = ToolCall( + name=_normalize_tool_name(raw_name if isinstance(raw_name, str) else ""), + args=args if isinstance(args, dict) else {}, + status="called", + ) + trajectory.append(call) + call_id = block.get("id") + if call_id: + pending.setdefault(str(call_id), []).append(call) + elif etype == "user": + message = event.get("message") + content = message.get("content") if isinstance(message, dict) else None + if not isinstance(content, list): + # A user message with a bare-string content echoes the prompt. + continue + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + call_id = block.get("tool_use_id") or "" + queue = pending.get(str(call_id)) if call_id else None + target = queue.pop(0) if queue else None + if target is None: + errors.append( + f"stream-json tool_result without matching tool_use (id={call_id!r})" + ) + continue + target.result = _block_text(block.get("content")) + target.status = "error" if block.get("is_error") else "completed" + elif etype == "result": + # Terminal event: ``result`` is the authoritative answer, ``usage`` + # holds token accounting, and failure shows up as either an + # ``error_*`` subtype or an ``is_error`` flag. Guard against a later + # degenerate ``result`` (empty answer / no usage) clobbering an + # earlier good one. + tail = event.get("result") + if isinstance(tail, str) and not result_output: + result_output = tail + usage = event.get("usage") + if isinstance(usage, dict) and _has_usage(usage): + tokens = _usage_tokens(usage) + result_usage_seen = True + subtype = event.get("subtype") + if isinstance(subtype, str) and subtype.startswith("error_"): + errors.append(f"stream-json result error: {subtype}") + elif event.get("is_error"): + # A failed API call can still carry ``subtype: "success"`` and + # exit 0 (observed: a 404 model-not-found, whose ``result`` is + # the provider's error text). Without this the run would score + # as clean with an empty trajectory and zeroed usage. + status = event.get("api_error_status") + detail = f" (api status {status})" if status is not None else "" + errors.append(f"stream-json result flagged is_error{detail}") + + # ``result_output`` may be an empty string (error subtypes emit ``""``); fall + # back to the accumulated assistant text so a real partial answer survives. + output = result_output or "".join(text_parts) + # Only fall back to summed per-turn usage when the terminal event reported no + # recognized usage — a terminal event that reported genuine zeros is trusted. + if not result_usage_seen and acc_usage: + tokens = _usage_tokens(acc_usage) + return output, [call.to_dict() for call in trajectory], tokens, errors + + +_USAGE_KEYS = ( + "input_tokens", + "output_tokens", + "cache_read_input_tokens", + "cache_creation_input_tokens", +) + + +def _has_usage(usage: dict) -> bool: + """True if ``usage`` carries at least one recognized integer count. + + Distinguishes a terminal ``result`` that reported genuine (possibly zero) + counts from one that reported nothing, so the accumulator fallback only + fires in the latter case. + """ + return any(_int_or_none(usage.get(key)) is not None for key in _USAGE_KEYS) + + +# ``output_tokens`` is deliberately absent from the accumulator. The per-turn +# ``usage`` on an assistant envelope is the streaming ``message_start`` snapshot, +# whose ``output_tokens`` is a placeholder of a few tokens rather than the final +# count (observed: a summed 3 against a terminal 3069). The prompt-side fields do +# accumulate faithfully, so they are kept and ``output`` is left unreported — +# per the bucket contract, an absent number beats an invented one. +_ACC_USAGE_KEYS = tuple(key for key in _USAGE_KEYS if key != "output_tokens") + + +def _add_usage(acc: dict, usage: object) -> None: + """Fold an Anthropic per-turn ``usage`` block into a running accumulator. + + Callers dedupe by message id first, so each API message is added once. The + terminal ``result`` usage is cumulative and authoritative; this accumulator + is only a best-effort stand-in for a truncated stream that never emits it, + and covers :data:`_ACC_USAGE_KEYS` only. + """ + if not isinstance(usage, dict): + return + for key in _ACC_USAGE_KEYS: + val = _int_or_none(usage.get(key)) + if val is not None: + acc[key] = acc.get(key, 0) + val + + +def _usage_tokens(usage: dict) -> dict[str, int | None]: + """Normalize an Anthropic ``usage`` block onto :data:`TOKEN_BUCKETS`. + + ``input_tokens`` is already the uncached prompt; cache reads and writes stay + separate buckets (writes bill at a premium). Extended thinking is billed + *inside* ``output_tokens`` and counted again under + ``output_tokens_details.thinking_tokens``, so it is subtracted back out to + honour the contract that ``output`` excludes ``reasoning`` while ``total`` + stays exact. Without that details block ``reasoning`` stays ``None`` rather + than a fabricated ``0``. + + ``total`` is only filled in once ``output`` is known. The accumulator path + leaves that bucket unreported on purpose (see :data:`_ACC_USAGE_KEYS`), and + a prompt-side-only sum published under ``total`` reaches the dashboard row + verbatim — an absent total beats one that undercounts the whole output side. + """ + tokens = empty_tokens() + output = _int_or_none(usage.get("output_tokens")) + details = usage.get("output_tokens_details") + reasoning = _int_or_none(details.get("thinking_tokens")) if isinstance(details, dict) else None + if output is not None and reasoning is not None: + output = max(0, output - reasoning) + tokens.update( + input=_int_or_none(usage.get("input_tokens")), + cached=_int_or_none(usage.get("cache_read_input_tokens")), + cache_write=_int_or_none(usage.get("cache_creation_input_tokens")), + reasoning=reasoning, + output=output, + ) + if output is not None: + tokens["total"] = sum(v for k, v in tokens.items() if k != "total" and v is not None) + return tokens diff --git a/devops_bench/agents/config.py b/devops_bench/agents/config.py index 0d888143..e8d60278 100644 --- a/devops_bench/agents/config.py +++ b/devops_bench/agents/config.py @@ -96,8 +96,9 @@ class AgentConfig: only for tests / local debug) — :meth:`from_env` always falls back to the 600s default since ``AGENT_TIMEOUT_SEC`` has no sentinel for "disabled". - max_turns: Safety cap on the API agent's tool-use loop turns; flows from - ``AGENT_MAX_TURNS``. ``None`` uses the agent's built-in default. + max_turns: Safety cap on the API agent's tool-use loop turns and on the + Claude CLI's ``--max-turns``; flows from ``AGENT_MAX_TURNS``. + ``None`` uses the agent's built-in default. capabilities: Aggregate of the MCP / skills / rules bindings granted for the run. Constructed by the orchestrator from a benchmark catalog (no GKE-specific strings live in agent code). diff --git a/devops_bench/agents/result.py b/devops_bench/agents/result.py index ac345b93..cdff9210 100644 --- a/devops_bench/agents/result.py +++ b/devops_bench/agents/result.py @@ -125,7 +125,14 @@ def errored(cls, msg: str, *, latency: float = 0.0) -> AgentResult: latency: Elapsed seconds before the failure, when available. Returns: - An :class:`AgentResult` with empty trajectory and the message in - both ``output`` and ``errors``. + An :class:`AgentResult` with empty trajectory, the canonical + all-``None`` token shape, and the message in both ``output`` and + ``errors``. """ - return cls(output=f"Error: {msg}", trajectory=[], latency=latency, errors=[msg]) + return cls( + output=f"Error: {msg}", + trajectory=[], + tokens=empty_tokens(), + latency=latency, + errors=[msg], + ) diff --git a/devops_bench/evalharness/default.py b/devops_bench/evalharness/default.py index 35b9ccdb..26dc00e5 100644 --- a/devops_bench/evalharness/default.py +++ b/devops_bench/evalharness/default.py @@ -71,6 +71,7 @@ # registry, with no edit here. _BUILTIN_AGENT_MODULES: tuple[str, ...] = ( "devops_bench.agents.cli.gemini_cli", + "devops_bench.agents.cli.claude_code", "devops_bench.agents.cli.openclaw", "devops_bench.agents.cli.antigravity", "devops_bench.agents.api.agent", @@ -79,6 +80,7 @@ # Aliases normalized to canonical agent keys before registry lookup. _AGENT_TYPE_ALIASES: dict[str, str] = { "gemini-cli": "gemini", + "claude-code": "claude", } # Default agent type when neither --agent-type nor BENCH_AGENT_TYPE is set. @@ -123,6 +125,17 @@ def _ensure_builtin_agents_registered() -> None: _log.debug("optional agent module %s not importable: %s", module, exc) +def _canonical_agent_type(agent_type: str) -> str: + """Normalize an agent-type alias to its canonical registry key. + + The single source of truth for both registry lookup and result recording, + so an arm selected via a friendly alias (``claude-code`` / ``gemini-cli``) + aggregates under the same ``harness`` / ``setup_id`` as the canonical key + instead of splitting into a second dashboard setup. + """ + return _AGENT_TYPE_ALIASES.get(agent_type, agent_type) + + class DefaultEvalHarness(Harness): """Standard harness wiring every component into one pipeline. @@ -235,7 +248,7 @@ def resolve_agent(self, agent_type: str) -> Any: canonical key. """ _ensure_builtin_agents_registered() - key = _AGENT_TYPE_ALIASES.get(agent_type, agent_type) + key = _canonical_agent_type(agent_type) agent_cls = AGENTS.get(key) if agent_cls is None: raise NotRegisteredError(AGENTS.name, key, AGENTS.keys()) @@ -674,14 +687,18 @@ def _write_run_artifacts(self, run_dir: Path, detailed_results: list[dict[str, A augmentation = derive_augmentation( {"use_mcp": self.use_mcp, "skills": list(self._granted_skill_paths)} ) - model = self._agent_config.model or self._agent_config.provider or self.agent_type + # Record the canonical harness key so an arm selected via a friendly + # alias (e.g. ``claude-code`` / ``gemini-cli``) aggregates with the + # canonical key rather than splitting into a second dashboard setup. + harness = _canonical_agent_type(self.agent_type) + model = self._agent_config.model or self._agent_config.provider or harness manifest = Manifest( schema_version=SCHEMA_VERSION, run_id=run_dir.name, t=datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), - setup_id=results_setup_id(model, self.agent_type, augmentation), + setup_id=results_setup_id(model, harness, augmentation), model=model, - harness=self.agent_type, + harness=harness, augmentation=augmentation, ) rows = build_rows(detailed_results, manifest) diff --git a/docs/components/model_providers.md b/docs/components/model_providers.md index 65b7f654..a6f99a59 100644 --- a/docs/components/model_providers.md +++ b/docs/components/model_providers.md @@ -18,8 +18,8 @@ API-key variable, or keyless auth — those belong to the provider contract in ## Supported providers and models -Every harness — the `api` runner and the `gemini`/`openclaw` CLIs — resolves -`AGENT_PROVIDER` through one shared contract +Every harness — the `api` runner and the `gemini`/`claude`/`openclaw` CLIs — +resolves `AGENT_PROVIDER` through one shared contract (`devops_bench/core/model_providers.py`). A provider resolves to an *adapter family* (which `LLMClient` the `api` harness builds), a *backend* hint (genai/vertex/bedrock), the *openclaw wire-provider*, and the *API-key env @@ -96,7 +96,7 @@ final fallback. > [!NOTE] > **All harnesses share one provider contract.** The `api` runner and the -> `gemini`/`openclaw` CLIs all resolve `AGENT_PROVIDER` through +> `gemini`/`claude`/`openclaw` CLIs all resolve `AGENT_PROVIDER` through > `devops_bench/core/model_providers.py`. The `api` harness uses it to pick the > adapter family and backend for `get_model()`; the CLI harnesses use it to route > `AGENT_API_KEY` onto the binary's provider-specific env var(s) (e.g. `google` → diff --git a/docs/how-to/add-a-model-provider.md b/docs/how-to/add-a-model-provider.md index d3422ccd..94dfb0fb 100644 --- a/docs/how-to/add-a-model-provider.md +++ b/docs/how-to/add-a-model-provider.md @@ -74,9 +74,10 @@ absent. `AGENT_PROVIDER` resolves through the shared contract in `devops_bench/core/model_providers.py` — the one place every harness (the `api` -runner and the `gemini`/`openclaw` CLIs) reads. If your provider needs aliases, a -distinct backend, a non-default API-key env var, or keyless (ADC-style) auth, add -a `ProviderSpec` row to `_SPECS` and its alias(es) to `_ALIASES` there: +runner and the `gemini`/`claude`/`openclaw` CLIs) reads. If your provider needs +aliases, a distinct backend, a non-default API-key env var, or keyless +(ADC-style) auth, add a `ProviderSpec` row to `_SPECS` and its alias(es) to +`_ALIASES` there: ```python _SPECS["your-key"] = ProviderSpec( diff --git a/tests/unit/agents/test_agents_cli_claude_code.py b/tests/unit/agents/test_agents_cli_claude_code.py new file mode 100644 index 00000000..e2bae29d --- /dev/null +++ b/tests/unit/agents/test_agents_cli_claude_code.py @@ -0,0 +1,1285 @@ +# Copyright 2026 The Kubernetes Authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for devops_bench.agents.cli.claude_code.""" + +from __future__ import annotations + +import json +import os +import stat +import tempfile +from collections.abc import Iterator +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from devops_bench.agents import AGENTS, AgentConfig +from devops_bench.agents.capabilities import ( + AgentRules, + AllCapabilities, + McpBinding, + SkillBinding, + SupportsMcp, + SupportsRules, + SupportsSkills, +) +from devops_bench.agents.cli.claude_code import ClaudeCodeAgent, parse_stream_json +from devops_bench.agents.cli.claude_code import agent as claude_mod +from devops_bench.agents.cli.claude_code.agent import _build_argv, _build_env +from devops_bench.agents.result import TOKEN_BUCKETS, empty_tokens +from devops_bench.core.errors import ConfigError, SubprocessError +from devops_bench.results.normalize import normalize_tokens + + +@pytest.fixture(autouse=True) +def _reset_version_cache() -> Iterator[None]: + """``_claude_version`` is cached per target, and every test here drives the + same ``"claude"`` target through a different probe double.""" + claude_mod._claude_version.cache_clear() # noqa: SLF001 - cache under test control + yield + claude_mod._claude_version.cache_clear() # noqa: SLF001 - cache under test control + + +def _stream(*events: dict) -> str: + """Render a list of events as a stream-json stdout blob.""" + return "\n".join(json.dumps(event) for event in events) + "\n" + + +def _assistant(*blocks: dict, msg_id: str | None = None, usage: dict | None = None) -> dict: + message: dict = {"content": list(blocks)} + if msg_id is not None: + message["id"] = msg_id + if usage is not None: + message["usage"] = usage + return {"type": "assistant", "message": message} + + +def _user(*blocks: dict) -> dict: + return {"type": "user", "message": {"content": list(blocks)}} + + +SAMPLE_STREAM = _stream( + {"type": "system", "subtype": "init", "session_id": "abc-123", "model": "claude-opus-4-8"}, + _assistant( + { + "type": "tool_use", + "id": "call-1", + "name": "mcp__gke__list_clusters", + "input": {"project": "p1"}, + } + ), + _user({"type": "tool_result", "tool_use_id": "call-1", "content": "cluster-a, cluster-b"}), + _assistant( + {"type": "tool_use", "id": "call-2", "name": "mcp__gke__get_cluster", "input": {"c": "a"}} + ), + _user( + { + "type": "tool_result", + "tool_use_id": "call-2", + "content": [{"type": "text", "text": "v1.30"}], + "is_error": False, + } + ), + _assistant({"type": "text", "text": "Done."}), + { + "type": "result", + "subtype": "success", + "result": "Done.", + "usage": {"input_tokens": 10, "output_tokens": 20, "cache_read_input_tokens": 5}, + }, +) + + +# --------------------------------------------------------------------------- +# Parser +# --------------------------------------------------------------------------- + + +def _tok(**overrides: int | None) -> dict[str, int | None]: + """Canonical token dict with every bucket None, overridden per test. + + Built from the shared :func:`empty_tokens` rather than a local literal, so a + bucket added to :data:`TOKEN_BUCKETS` shows up in every token assertion here + instead of being silently absent. + """ + base = empty_tokens() + base.update(overrides) + return base + + +def test_parser_fills_the_shared_canonical_buckets() -> None: + """Guards against ``TOKEN_BUCKETS`` drifting away from what this parser emits.""" + _output, _trajectory, tokens, _errors = parse_stream_json(SAMPLE_STREAM) + + assert set(tokens) == set(TOKEN_BUCKETS) + + +def test_tokens_reach_the_result_row_unchanged() -> None: + """Cross-layer guard: the emitted keys are the ones ``normalize_tokens`` reads. + + A bucket named differently here would silently normalize to ``None`` on the + dashboard row rather than failing loudly. + """ + blob = _stream( + { + "type": "result", + "result": "ok", + "usage": { + "input_tokens": 2748, + "output_tokens": 11267, + "cache_read_input_tokens": 334987, + "cache_creation_input_tokens": 12000, + }, + } + ) + + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + normalized = normalize_tokens(tokens) + + assert normalized.input == 2748 + assert normalized.output == 11267 + assert normalized.cached == 334987 + assert normalized.cache_write == 12000 + assert normalized.reasoning is None + assert normalized.total == 2748 + 11267 + 334987 + 12000 + + +def test_thinking_tokens_split_out_of_output() -> None: + """Extended thinking is billed inside ``output_tokens`` and reported again + under ``output_tokens_details``. The canonical contract says ``output`` + excludes ``reasoning``, so it is subtracted back out — leaving ``total`` + (which sums every bucket) equal to the provider's own accounting.""" + blob = _stream( + { + "type": "result", + "result": "ok", + "usage": { + "input_tokens": 4, + "output_tokens": 3069, + "cache_read_input_tokens": 13445, + "cache_creation_input_tokens": 31257, + "output_tokens_details": {"thinking_tokens": 2778}, + }, + } + ) + + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + + assert tokens == _tok( + input=4, + cached=13445, + cache_write=31257, + reasoning=2778, + output=3069 - 2778, + total=4 + 13445 + 31257 + 3069, + ) + + +def test_thinking_tokens_absent_leaves_reasoning_unreported() -> None: + """No ``output_tokens_details`` means nothing to split: ``reasoning`` stays + ``None`` rather than a fabricated ``0``, and ``output`` is untouched.""" + blob = _stream( + {"type": "result", "result": "ok", "usage": {"input_tokens": 4, "output_tokens": 78}} + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=4, reasoning=None, output=78, total=82) + + +def test_thinking_tokens_exceeding_output_clamp_at_zero() -> None: + """A provider quirk must not produce a negative bucket.""" + blob = _stream( + { + "type": "result", + "result": "ok", + "usage": {"output_tokens": 5, "output_tokens_details": {"thinking_tokens": 9}}, + } + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens["output"] == 0 + assert tokens["reasoning"] == 9 + + +def test_parse_stream_json_emits_canonical_trajectory() -> None: + output, trajectory, tokens, errors = parse_stream_json(SAMPLE_STREAM) + assert output == "Done." + assert errors == [] + assert tokens == _tok(input=10, cached=5, output=20, total=35) + assert trajectory == [ + { + "name": "gke__list_clusters", + "args": {"project": "p1"}, + "result": "cluster-a, cluster-b", + "status": "completed", + }, + { + "name": "gke__get_cluster", + "args": {"c": "a"}, + "result": "v1.30", + "status": "completed", + }, + ] + + +def test_parse_stream_json_marks_failed_tool_results_as_error_status() -> None: + blob = _stream( + _assistant({"type": "tool_use", "id": "c", "name": "x", "input": {}}), + _user({"type": "tool_result", "tool_use_id": "c", "content": "oops", "is_error": True}), + ) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == [] + assert trajectory[0]["status"] == "error" + + +def test_parse_stream_json_records_unmatched_tool_results() -> None: + blob = _stream(_user({"type": "tool_result", "tool_use_id": "ghost", "content": "?"})) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + assert trajectory == [] + assert any("without matching tool_use" in msg for msg in errors) + + +def test_parse_stream_json_records_json_decode_errors_on_errors_list() -> None: + blob = "{not json}\n" + json.dumps({"type": "result", "result": "ok"}) + "\n" + output, trajectory, _tokens, errors = parse_stream_json(blob) + assert output == "ok" + assert trajectory == [] + assert len(errors) == 1 + assert "parse error" in errors[0] + + +def test_parse_stream_json_records_error_result_subtype() -> None: + blob = _stream({"type": "result", "subtype": "error_max_turns", "result": ""}) + _output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == ["stream-json result error: error_max_turns"] + + +def test_parse_stream_json_records_is_error_under_a_success_subtype() -> None: + """An API failure can exit 0 under ``subtype: "success"`` with ``is_error`` + set (observed: a 404 model-not-found), which must not score as a clean run.""" + blob = _stream( + { + "type": "result", + "subtype": "success", + "is_error": True, + "api_error_status": 404, + "result": "The model x is not available on your vertex deployment.", + } + ) + output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert output == "The model x is not available on your vertex deployment." + assert errors == ["stream-json result flagged is_error (api status 404)"] + + +def test_parse_stream_json_records_is_error_without_an_api_status() -> None: + blob = _stream({"type": "result", "subtype": "success", "is_error": True, "result": "nope"}) + _output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == ["stream-json result flagged is_error"] + + +def test_parse_stream_json_does_not_double_report_an_error_subtype() -> None: + """``error_*`` subtypes also carry ``is_error``; one error line, not two.""" + blob = _stream({"type": "result", "subtype": "error_max_turns", "is_error": True, "result": ""}) + _output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == ["stream-json result error: error_max_turns"] + + +def test_parse_stream_json_ignores_a_false_is_error() -> None: + blob = _stream({"type": "result", "subtype": "success", "is_error": False, "result": "ok"}) + _output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == [] + + +def test_parse_stream_json_empty_input_returns_empty() -> None: + assert parse_stream_json("") == ("", [], _tok(), []) + + +def test_parse_stream_json_flags_failed_mcp_server_at_init() -> None: + """A failed MCP server in the init event surfaces on ``errors`` — a + tool-less-but-exit-0 run must not look clean.""" + blob = _stream( + { + "type": "system", + "subtype": "init", + "mcp_servers": [ + {"name": "gke", "status": "connected"}, + {"name": "broken", "status": "failed"}, + ], + }, + {"type": "result", "subtype": "success", "result": "ok"}, + ) + output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert output == "ok" + assert any("broken" in e and "failed" in e for e in errors) + assert not any("gke" in e for e in errors) + + +def test_parse_stream_json_ignores_transient_pending_mcp_status_at_init() -> None: + """A ``pending`` MCP status at the init snapshot is transient — the server + (e.g. gke-mcp) connects moments later and serves tools normally — so it must + NOT surface an error, which would otherwise flip the run-level ``validated`` + gate to ``False`` on a fully working MCP run.""" + blob = _stream( + { + "type": "system", + "subtype": "init", + "mcp_servers": [{"name": "gke", "status": "pending"}], + }, + _assistant( + {"type": "tool_use", "id": "t1", "name": "mcp__gke__list_clusters", "input": {}} + ), + _user({"type": "tool_result", "tool_use_id": "t1", "content": "cluster-a"}), + {"type": "result", "subtype": "success", "result": "ok"}, + ) + output, trajectory, _tokens, errors = parse_stream_json(blob) + assert output == "ok" + assert errors == [] + assert trajectory[0]["status"] == "completed" + + +def test_parse_stream_json_strips_mcp_client_prefix_from_tool_names() -> None: + """Claude Code names MCP tools ``mcp____``; the parser drops the + literal ``mcp__`` prefix so names match the pipeline's ``__`` + convention (which the metrics canonicalizer reduces to the bare tool name). + Built-in tools keep their names.""" + blob = _stream( + _assistant( + {"type": "tool_use", "id": "a", "name": "mcp__default__generate_manifest", "input": {}}, + {"type": "tool_use", "id": "b", "name": "Bash", "input": {"command": "ls"}}, + ), + ) + _output, trajectory, _tokens, _errors = parse_stream_json(blob) + assert [t["name"] for t in trajectory] == ["default__generate_manifest", "Bash"] + + +def test_parse_stream_json_drops_thinking_blocks() -> None: + """``thinking`` / ``redacted_thinking`` blocks are dropped so the trajectory + is tool-calls-only, matching the shape the other CLI harnesses emit. Only the + tool call survives; the surrounding reasoning leaves no trajectory step.""" + blob = _stream( + _assistant( + {"type": "thinking", "thinking": "let me plan", "signature": "sig"}, + {"type": "tool_use", "id": "c1", "name": "Bash", "input": {"command": "ls"}}, + {"type": "redacted_thinking", "data": "enc"}, + ), + _user({"type": "tool_result", "tool_use_id": "c1", "content": "ok"}), + ) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == [] + assert trajectory == [ + {"name": "Bash", "args": {"command": "ls"}, "result": "ok", "status": "completed"}, + ] + + +def test_parse_stream_json_keeps_pending_tool_use_as_called() -> None: + """A tool_use with no matching tool_result (timeout-truncated stream) stays + in the trajectory with status ``called`` and a ``None`` result.""" + blob = _stream(_assistant({"type": "tool_use", "id": "c1", "name": "do", "input": {"k": "v"}})) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + assert trajectory == [{"name": "do", "args": {"k": "v"}, "result": None, "status": "called"}] + assert errors == [] + + +def test_parse_stream_json_falls_back_to_assistant_text_without_result_event() -> None: + """A truncated stream (no terminal ``result``) still yields the answer from + the accumulated assistant ``text`` blocks.""" + blob = _stream( + _assistant({"type": "text", "text": "partial "}), + _assistant({"type": "text", "text": "answer"}), + ) + output, _trajectory, tokens, errors = parse_stream_json(blob) + assert output == "partial answer" + assert tokens == _tok() + assert errors == [] + + +def test_parse_stream_json_falls_back_to_accumulated_usage_without_result_event() -> None: + """A truncated stream (no terminal ``result``) still yields prompt-side token + counts, summed from the per-turn assistant ``usage``. ``output`` stays + unreported — see the next test — so ``total`` stays unreported with it.""" + blob = _stream( + _assistant( + {"type": "text", "text": "a"}, + usage={"input_tokens": 10, "output_tokens": 5, "cache_read_input_tokens": 2}, + ), + _assistant( + {"type": "text", "text": "b"}, + usage={"input_tokens": 20, "output_tokens": 7, "cache_creation_input_tokens": 3}, + ), + ) + output, _trajectory, tokens, errors = parse_stream_json(blob) + assert output == "ab" + assert tokens == _tok(input=30, cached=2, cache_write=3) + assert errors == [] + + +def test_parse_stream_json_accumulator_leaves_output_unreported() -> None: + """Per-turn ``usage.output_tokens`` is the ``message_start`` placeholder — a + handful of tokens against a real terminal count in the thousands. Summing it + would persist an invented number, so the bucket is left ``None``. ``total`` + follows it: a prompt-side-only sum reaches the dashboard row as the run's + total, understating it by the whole output side.""" + blob = _stream( + _assistant( + {"type": "text", "text": "..."}, + msg_id="msg_1", + usage={"input_tokens": 4, "output_tokens": 3}, + ), + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens["output"] is None + assert tokens["total"] is None + assert tokens == _tok(input=4) + + +def test_parse_stream_json_result_usage_wins_over_accumulated() -> None: + """When the terminal ``result`` carries usage it is authoritative — the + accumulated per-turn usage is not added on top.""" + blob = _stream( + _assistant( + {"type": "text", "text": "x"}, usage={"input_tokens": 999, "output_tokens": 999} + ), + { + "type": "result", + "subtype": "success", + "result": "x", + "usage": {"input_tokens": 10, "output_tokens": 20}, + }, + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=10, output=20, total=30) + + +def test_parse_stream_json_falls_back_when_result_usage_degenerate() -> None: + """A terminal ``result`` whose ``usage`` carries no recognized counts must + not shadow the accumulated per-turn usage — the all-None result is treated + as absent so the summed per-turn counts survive.""" + blob = _stream( + _assistant({"type": "text", "text": "x"}, usage={"input_tokens": 15, "output_tokens": 4}), + {"type": "result", "subtype": "success", "result": "x", "usage": {}}, + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=15) + + +def test_parse_stream_json_result_string_is_authoritative_over_text() -> None: + """When a ``result`` event is present its string wins over assistant text + (which merely duplicates it) — no double-counting.""" + blob = _stream( + _assistant({"type": "text", "text": "Done."}), + {"type": "result", "subtype": "success", "result": "Done."}, + ) + output, _trajectory, _tokens, _errors = parse_stream_json(blob) + assert output == "Done." + + +def test_parse_stream_json_dedupes_accumulated_usage_by_message_id() -> None: + """Claude Code emits one envelope per content block of a single API message, + each repeating the identical ``usage``; a truncated stream must count that + message's usage once, not once per block.""" + usage = {"input_tokens": 100, "output_tokens": 40, "cache_read_input_tokens": 8} + blob = _stream( + _assistant({"type": "thinking"}, msg_id="msg_1", usage=usage), + _assistant({"type": "text", "text": "hi"}, msg_id="msg_1", usage=usage), + _assistant( + {"type": "tool_use", "id": "t", "name": "Bash", "input": {}}, + msg_id="msg_1", + usage=usage, + ), + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=100, cached=8) + + +def test_parse_stream_json_empty_result_falls_back_to_text() -> None: + """An error subtype emits ``result: ""``; the real partial answer in the + accumulated assistant text must survive rather than grading as empty.""" + blob = _stream( + _assistant({"type": "text", "text": "partial answer"}), + {"type": "result", "subtype": "error_max_turns", "result": ""}, + ) + output, _trajectory, _tokens, errors = parse_stream_json(blob) + assert output == "partial answer" + assert any("error_max_turns" in e for e in errors) + + +def test_parse_stream_json_all_zero_result_usage_is_authoritative() -> None: + """A terminal ``result`` reporting genuine zeros is trusted — it is not + conflated with 'no usage reported' and replaced by the accumulator.""" + blob = _stream( + _assistant({"type": "text", "text": "x"}, usage={"input_tokens": 50}), + { + "type": "result", + "subtype": "success", + "result": "x", + "usage": {"input_tokens": 0, "output_tokens": 0}, + }, + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=0, output=0, total=0) + + +def test_parse_stream_json_degenerate_second_result_does_not_clobber() -> None: + """A later degenerate ``result`` (empty answer, no usage) must not destroy a + good earlier one.""" + blob = _stream( + { + "type": "result", + "subtype": "success", + "result": "first", + "usage": {"input_tokens": 10, "output_tokens": 20}, + }, + {"type": "result", "subtype": "error_during_execution", "result": "", "usage": {}}, + ) + output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert output == "first" + assert tokens == _tok(input=10, output=20, total=30) + + +def test_parse_stream_json_recovers_concatenated_objects_on_one_line() -> None: + """A rebuffered stream that concatenates objects onto one physical line must + not lose the whole run to a single 'Extra data' error.""" + line = json.dumps(_assistant({"type": "text", "text": "hi"})) + json.dumps( + { + "type": "result", + "subtype": "success", + "result": "hi", + "usage": {"input_tokens": 3, "output_tokens": 1}, + } + ) + output, _trajectory, tokens, errors = parse_stream_json(line + "\n") + assert output == "hi" + assert tokens == _tok(input=3, output=1, total=4) + assert errors == [] + + +def test_parse_stream_json_survives_unescaped_unicode_line_breaks() -> None: + """The CLI leaves U+0085 (NEL) unescaped inside JSON strings — real command + output carries it when a log is mis-decoded as latin-1. Framing must key on + ``\\n`` alone; ``str.splitlines`` would shred that event into undecodable + fragments, losing the tool call and injecting bogus errors that flip the run + to unvalidated. + """ + body = "line\u0085next" + events = ( + _assistant({"type": "tool_use", "id": "t1", "name": "Bash", "input": {"cmd": "cat log"}}), + _user({"type": "tool_result", "tool_use_id": "t1", "content": body}), + {"type": "result", "subtype": "success", "result": "done"}, + ) + # ensure_ascii=False mirrors Node's JSON.stringify, which leaves U+0085 raw. + blob = "\n".join(json.dumps(event, ensure_ascii=False) for event in events) + "\n" + assert "\u0085" in blob # guard: the fixture must carry the raw character + + output, trajectory, _tokens, errors = parse_stream_json(blob) + assert errors == [] + assert output == "done" + assert len(trajectory) == 1 + assert trajectory[0]["status"] == "completed" + assert trajectory[0]["result"] == body + + +def test_parse_stream_json_clips_oversized_tool_results() -> None: + """Claude Code echoes the full body of every tool result. The trajectory is + re-serialized into the LLM judge prompts, so an uncapped result overruns the + judge's context — keep the head and tail with the middle marked elided.""" + payload = "A" * 20_000 + "TAIL" + blob = _stream( + _assistant({"type": "tool_use", "id": "t1", "name": "Read", "input": {}}), + _user({"type": "tool_result", "tool_use_id": "t1", "content": payload}), + ) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + result = trajectory[0]["result"] + assert errors == [] + assert len(result) < len(payload) + assert result.startswith("A" * 100) + assert result.endswith("TAIL") + assert "chars elided" in result + + +def test_parse_stream_json_keeps_tool_results_under_the_cap_verbatim() -> None: + """The clip must not touch an ordinary result.""" + payload = "B" * 500 + blob = _stream( + _assistant({"type": "tool_use", "id": "t1", "name": "Read", "input": {}}), + _user({"type": "tool_result", "tool_use_id": "t1", "content": payload}), + ) + _output, trajectory, _tokens, _errors = parse_stream_json(blob) + assert trajectory[0]["result"] == payload + + +def test_parse_stream_json_non_dict_message_does_not_crash() -> None: + """A malformed line whose ``message`` is a bare string is skipped, not fatal + — one bad envelope must not lose the whole otherwise-parseable run.""" + blob = _stream( + {"type": "assistant", "message": "Execution error"}, + {"type": "user", "message": "echo"}, + { + "type": "result", + "subtype": "success", + "result": "ok", + "usage": {"input_tokens": 5, "output_tokens": 2}, + }, + ) + output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert output == "ok" + assert tokens == _tok(input=5, output=2, total=7) + + +def test_parse_stream_json_rejects_bool_usage_values() -> None: + """A stray JSON ``true`` in a usage field is rejected, never summed as 1.""" + blob = _stream( + { + "type": "result", + "subtype": "success", + "result": "x", + "usage": {"input_tokens": True, "output_tokens": 20}, + }, + ) + _output, _trajectory, tokens, _errors = parse_stream_json(blob) + assert tokens == _tok(input=None, output=20, total=20) + + +def test_parse_stream_json_matches_duplicate_tool_use_ids_fifo() -> None: + """Distinct tool_use blocks reusing an id are matched in emission order, so + neither call is orphaned and no spurious 'unmatched' error is raised.""" + blob = _stream( + _assistant({"type": "tool_use", "id": "x", "name": "A", "input": {}}), + _assistant({"type": "tool_use", "id": "x", "name": "B", "input": {}}), + _user({"type": "tool_result", "tool_use_id": "x", "content": "r1"}), + _user({"type": "tool_result", "tool_use_id": "x", "content": "r2"}), + ) + _output, trajectory, _tokens, errors = parse_stream_json(blob) + assert [(t["name"], t["result"]) for t in trajectory] == [("A", "r1"), ("B", "r2")] + assert errors == [] + + +def test_block_text_preserves_non_text_blocks() -> None: + """A mixed tool_result content list must not silently drop non-text blocks + (e.g. an image) — they are JSON-encoded in place.""" + blob = _stream( + _assistant({"type": "tool_use", "id": "c", "name": "Read", "input": {}}), + _user( + { + "type": "tool_result", + "tool_use_id": "c", + "content": [{"type": "image", "source": {"x": 1}}, {"type": "text", "text": "ok"}], + } + ), + ) + _output, trajectory, _tokens, _errors = parse_stream_json(blob) + assert trajectory[0]["result"] == '{"type": "image", "source": {"x": 1}}ok' + + +# --------------------------------------------------------------------------- +# argv +# --------------------------------------------------------------------------- + + +def test_build_argv_base_flags_and_prompt_via_argv() -> None: + argv = _build_argv("/bin/claude", "hi", model=None, max_turns=None, mcp_config_path=None) + assert argv[0] == "/bin/claude" + # Prompt is the trailing argv value (never a shell string), behind ``--``. + assert argv[-2:] == ["--", "hi"] + assert "-p" in argv + assert "--output-format" in argv and "stream-json" in argv + assert "--verbose" in argv # required for stream-json under -p + assert "--dangerously-skip-permissions" in argv + # Optional flags absent when unset. + assert "--model" not in argv + assert "--max-turns" not in argv + assert "--mcp-config" not in argv + # This harness never emits an allowlist (bare names can't match mcp__ tools). + assert "--allowedTools" not in argv and "--allowed-tools" not in argv + + +def test_build_argv_shields_a_flag_like_prompt_behind_a_separator() -> None: + """A prompt whose first token looks like a flag must reach the CLI as the + prompt. Without ``--`` the option parser consumes it and aborts the run.""" + argv = _build_argv( + "/bin/claude", "--settings=/tmp/x.json", model=None, max_turns=None, mcp_config_path=None + ) + assert argv[-2:] == ["--", "--settings=/tmp/x.json"] + + +def test_build_argv_threads_model_and_max_turns_when_set() -> None: + argv = _build_argv( + "/bin/claude", "hi", model="claude-opus-4-8", max_turns=7, mcp_config_path=None + ) + assert argv[argv.index("--model") + 1] == "claude-opus-4-8" + assert argv[argv.index("--max-turns") + 1] == "7" + + +def test_build_argv_always_sets_strict_mcp_config() -> None: + """Strict mode is unconditional: on a baseline arm it pins the run to zero + servers, so a stray ``.mcp.json`` in the workspace cannot grant MCP tools.""" + bare = _build_argv("/bin/claude", "hi", model=None, max_turns=None, mcp_config_path=None) + assert "--strict-mcp-config" in bare + + bound = _build_argv( + "/bin/claude", "hi", model=None, max_turns=None, mcp_config_path="/w/.claude/mcp.json" + ) + assert bound[bound.index("--mcp-config") + 1] == "/w/.claude/mcp.json" + assert "--strict-mcp-config" in bound + + +# --------------------------------------------------------------------------- +# env +# --------------------------------------------------------------------------- + + +def test_build_env_threads_api_key_into_anthropic_var() -> None: + env = _build_env(AgentConfig(api_key="sk-abc"), config_dir=None) + assert env["ANTHROPIC_API_KEY"] == "sk-abc" + assert env["CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC"] == "1" + assert env["DISABLE_AUTOUPDATER"] == "1" + + +def test_build_env_keyless_vertex_sets_switch_and_maps_project_region( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("GCP_PROJECT_ID", "proj-42") + monkeypatch.setenv("GCP_VERTEX_LOCATION", "us-east5") + env = _build_env(AgentConfig(provider="anthropic-vertex"), config_dir=None) + assert env["CLAUDE_CODE_USE_VERTEX"] == "1" + assert env["ANTHROPIC_VERTEX_PROJECT_ID"] == "proj-42" + assert env["CLOUD_ML_REGION"] == "us-east5" + # Keyless: no API-key var written even absent an explicit api_key. + assert "ANTHROPIC_API_KEY" not in env + + +def test_build_env_vertex_region_defaults_to_global(monkeypatch: pytest.MonkeyPatch) -> None: + # Both inputs to the region chain must be cleared: an ambient CLOUD_ML_REGION + # on the developer's machine or the CI runner would shadow the fallback. + monkeypatch.delenv("GCP_VERTEX_LOCATION", raising=False) + monkeypatch.delenv("CLOUD_ML_REGION", raising=False) + monkeypatch.setenv("GCP_PROJECT_ID", "proj-42") + env = _build_env(AgentConfig(provider="anthropic-vertex"), config_dir=None) + assert env["CLOUD_ML_REGION"] == "global" + + +def test_build_env_keyless_bedrock_sets_switch() -> None: + env = _build_env(AgentConfig(provider="anthropic-bedrock"), config_dir=None) + assert env["CLAUDE_CODE_USE_BEDROCK"] == "1" + assert "ANTHROPIC_API_KEY" not in env + + +def test_build_env_unknown_provider_raises_even_when_keyless() -> None: + with pytest.raises(ConfigError): + _build_env(AgentConfig(provider="anthropi"), config_dir=None) + + +def test_build_env_extra_env_wins_over_computed_vars() -> None: + cfg = AgentConfig(api_key="sk-abc", extra_env={"ANTHROPIC_API_KEY": "override", "X": "y"}) + env = _build_env(cfg, config_dir=None) + assert env["ANTHROPIC_API_KEY"] == "override" + assert env["X"] == "y" + + +def test_build_env_sets_config_dir_when_provided() -> None: + env = _build_env(AgentConfig(), config_dir="/tmp/cfg") + assert env["CLAUDE_CONFIG_DIR"] == "/tmp/cfg" + + +def test_build_env_omits_config_dir_when_none() -> None: + env = _build_env(AgentConfig(), config_dir=None) + assert "CLAUDE_CONFIG_DIR" not in env + + +# --------------------------------------------------------------------------- +# Registry + capability protocols +# --------------------------------------------------------------------------- + + +def test_claude_agent_registered_under_canonical_key() -> None: + assert AGENTS.get("claude") is ClaudeCodeAgent + + +def test_claude_agent_satisfies_mcp_skills_and_rules_protocols() -> None: + agent = ClaudeCodeAgent(AgentConfig()) + assert isinstance(agent, SupportsMcp) + assert isinstance(agent, SupportsSkills) + assert isinstance(agent, SupportsRules) + + +def test_claude_agent_mirrors_capability_bindings_onto_mixin_attributes() -> None: + binding = McpBinding(name="x", command=(), tools=("t",)) + skills = SkillBinding(paths=("/some/skills",)) + caps = AllCapabilities( + mcp_servers=(binding,), + skills=skills, + rules=AgentRules(text="be a sre"), + ) + agent = ClaudeCodeAgent(AgentConfig(capabilities=caps)) + assert agent.mcp_servers == (binding,) + assert agent.skills == skills + assert agent.rules == AgentRules(text="be a sre") + + +# --------------------------------------------------------------------------- +# _execute: return shape, wiring, and error paths +# --------------------------------------------------------------------------- + + +def test_execute_returns_typed_result_with_trajectory(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["argv"] = argv + captured["timeout"] = kwargs.get("timeout") + return SimpleNamespace(stdout=SAMPLE_STREAM, stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude-x", timeout_sec=30.0)).run("ping") + assert result.output == "Done." + assert len(result.trajectory) == 2 + assert result.errors == [] + assert result.tokens == _tok(input=10, cached=5, output=20, total=35) + assert captured["timeout"] == 30.0 + assert captured["argv"][0].endswith("claude-x") + assert captured["argv"][-2:] == ["--", "ping"] + + +def test_execute_closes_child_stdin(monkeypatch: pytest.MonkeyPatch) -> None: + """An inherited stdin makes the CLI wait out its 3s piped-prompt timeout on + every run and warn on stderr; an empty ``input`` closes the pipe at once.""" + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["input"] = kwargs.get("input") + return SimpleNamespace(stdout=SAMPLE_STREAM, stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert captured["input"] == "" + + +def test_execute_wires_extra_env_into_subprocess_call(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["extra_env"] = kwargs.get("extra_env") + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude", api_key="sk-abc")).run("p") + env = captured["extra_env"] + assert env["ANTHROPIC_API_KEY"] == "sk-abc" + assert env["CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC"] == "1" + + +def test_execute_records_non_zero_exit(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + return SimpleNamespace(stdout="", stderr="boom", returncode=2) + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert result.has_errors() + assert any("exited 2" in e for e in result.errors) + assert result.metadata.get("returncode") == 2 + + +def test_execute_parses_stream_on_non_zero_exit(monkeypatch: pytest.MonkeyPatch) -> None: + """An ``error_max_turns`` run exits non-zero *after* emitting a full stream; + output and trajectory must still be parsed, with the exit + subtype recorded.""" + stream = _stream( + _assistant({"type": "tool_use", "id": "c1", "name": "do", "input": {}}), + {"type": "result", "subtype": "error_max_turns", "result": "hit the cap"}, + ) + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + return SimpleNamespace(stdout=stream, stderr="turn limit reached", returncode=1) + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert result.output == "hit the cap" + assert len(result.trajectory) == 1 + assert result.metadata["returncode"] == 1 + assert result.metadata["stderr"] == "turn limit reached" + assert any("error_max_turns" in e for e in result.errors) + assert any("exited 1" in e for e in result.errors) + + +def test_execute_captures_stderr_on_clean_exit(monkeypatch: pytest.MonkeyPatch) -> None: + """stderr is kept for diagnosis even when the process exits 0.""" + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + return SimpleNamespace(stdout=SAMPLE_STREAM, stderr="a warning", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert result.metadata["stderr"] == "a warning" + assert "returncode" not in result.metadata + assert not any("exited" in e for e in result.errors) + + +def test_execute_reports_a_timeout_as_a_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """``run`` is called with ``check=False``, so the only ``SubprocessError`` it + raises is a timeout. Naming it keeps the wall-clock budget out of the crash + bucket, where ``exit -1`` would have put it.""" + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + raise SubprocessError(argv, returncode=-1, stdout="", stderr="killed") + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude", timeout_sec=900)).run("p") + assert result.has_errors() + assert result.errors[0] == "claude timed out after 900s: killed" + assert result.metadata["returncode"] == -1 + assert result.trajectory == [] + + +def test_execute_recovers_partial_trajectory_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + """A timeout carries the partial stream-json captured before the kill; the + harness recovers the trajectory instead of discarding the run's work, while + still surfacing the error.""" + partial = _stream( + _assistant( + {"type": "tool_use", "id": "c1", "name": "mcp__gke__list_clusters", "input": {}} + ), + _user({"type": "tool_result", "tool_use_id": "c1", "content": "ok"}), + ) + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + raise SubprocessError(argv, returncode=-1, stdout=partial, stderr="killed after timeout") + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert result.has_errors() + assert any("timed out" in e for e in result.errors) + assert [step["name"] for step in result.trajectory] == ["gke__list_clusters"] + assert result.metadata["stderr"] == "killed after timeout" + + +def test_execute_handles_missing_binary(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + raise OSError("not found") + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert result.has_errors() + assert "failed to spawn claude" in result.errors[0] + assert result.tokens == _tok() + + +def test_execute_passes_timeout_to_subprocess(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured.update(kwargs) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude", timeout_sec=15.5)).run("p") + assert captured["timeout"] == 15.5 + + +# --------------------------------------------------------------------------- +# _execute side-effects: capabilities land in the cwd before the subprocess. +# Assertions run INSIDE the fake ``run`` because the temp cwd is cleaned up +# once ``_execute`` returns. +# --------------------------------------------------------------------------- + + +def test_execute_writes_claude_md_with_rules_text_before_subprocess( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + cwd = kwargs.get("cwd") + captured["cwd"] = cwd + claude_md = Path(cwd) / "CLAUDE.md" if cwd else None + captured["exists"] = bool(claude_md and claude_md.exists()) + captured["text"] = claude_md.read_text(encoding="utf-8") if captured["exists"] else None + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + caps = AllCapabilities(rules=AgentRules(text="you are a precise SRE")) + ClaudeCodeAgent(AgentConfig(target="claude", capabilities=caps)).run("p") + assert captured["exists"], "CLAUDE.md must exist in cwd before subprocess" + assert captured["text"] == "you are a precise SRE" + + +def test_execute_skips_writing_claude_md_when_rules_empty(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + cwd = kwargs.get("cwd") + captured["exists"] = bool(cwd and os.path.exists(os.path.join(cwd, "CLAUDE.md"))) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + assert captured["exists"] is False + + +def test_execute_writes_mcp_config_and_passes_flag(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + if "--version" in argv: + return SimpleNamespace(stdout="2.1.228 (Claude Code)\n", stderr="", returncode=0) + captured["argv"] = argv + mcp_path = os.path.join(kwargs["cwd"], ".claude", "mcp-config.json") + captured["exists"] = os.path.exists(mcp_path) + if captured["exists"]: + captured["mode"] = stat.S_IMODE(os.stat(mcp_path).st_mode) + with open(mcp_path) as f: + captured["payload"] = json.load(f) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + caps = AllCapabilities( + mcp_servers=(McpBinding(name="gke", command=("gke-mcp",), tools=("mcp__gke__x",)),), + ) + ClaudeCodeAgent(AgentConfig(target="claude", capabilities=caps)).run("p") + + assert captured["exists"], "mcp-config.json must exist in cwd before subprocess" + assert captured["payload"] == {"mcpServers": {"gke": {"command": "gke-mcp"}}} + # A binding's argv can carry a server credential, so the file must not be + # left at the umask default in a workspace the harness later collects. + assert captured["mode"] == 0o600 + argv = captured["argv"] + assert argv[argv.index("--mcp-config") + 1].endswith(os.path.join(".claude", "mcp-config.json")) + assert "--strict-mcp-config" in argv + + +def _mcp_caps() -> AllCapabilities: + return AllCapabilities( + mcp_servers=(McpBinding(name="gke", command=("gke-mcp",), tools=("mcp__gke__x",)),), + ) + + +def test_execute_refuses_mcp_run_on_a_binary_predating_the_startup_wait( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An older binary accepts ``--mcp-config`` but starts the first turn without + waiting for the servers, so the arm runs un-augmented yet is scored as + augmented. Fail loud instead of recording a contaminated result.""" + calls: list[list[str]] = [] + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + calls.append(list(argv)) + if "--version" in argv: + return SimpleNamespace(stdout="2.1.220 (Claude Code)\n", stderr="", returncode=0) + raise AssertionError("the agent turn must not run on an under-version binary") + + monkeypatch.setattr(claude_mod, "run", fake_run) + result = ClaudeCodeAgent(AgentConfig(target="claude", capabilities=_mcp_caps())).run("p") + + assert calls == [["claude", "--version"]] + assert result.tokens == empty_tokens() + assert "2.1.220" in result.errors[0] + assert "2.1.221" in result.errors[0] + + +def test_execute_probes_the_version_once_per_target(monkeypatch: pytest.MonkeyPatch) -> None: + """A matrix run drives one binary across every task, and the version cannot + change under a live process, so the probe is spawned once rather than once + per MCP-bound task.""" + probes = 0 + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + nonlocal probes + if "--version" in argv: + probes += 1 + return SimpleNamespace(stdout="2.1.228 (Claude Code)\n", stderr="", returncode=0) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + agent = ClaudeCodeAgent(AgentConfig(target="claude", capabilities=_mcp_caps())) + agent.run("first") + agent.run("second") + + assert probes == 1 + + +def test_execute_skips_the_version_probe_without_mcp_bindings( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The floor only bites on ``--mcp-config``; a baseline arm must not pay for + an extra subprocess on every run.""" + calls: list[list[str]] = [] + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + calls.append(list(argv)) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + + assert all("--version" not in argv for argv in calls) + + +@pytest.mark.parametrize( + "probe", + [ + SimpleNamespace(stdout="", stderr="not a claude binary", returncode=1), + SimpleNamespace(stdout="wrapper build abc\n", stderr="", returncode=0), + OSError("no such binary"), + ], + ids=["nonzero-exit", "unparseable", "spawn-failure"], +) +def test_execute_proceeds_when_the_version_probe_is_inconclusive( + monkeypatch: pytest.MonkeyPatch, probe: SimpleNamespace | OSError +) -> None: + """``config.target`` may be a wrapper with its own ``--version`` surface, so a + probe that fails to yield a number must not block the run.""" + ran: list[list[str]] = [] + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + if "--version" in argv: + if isinstance(probe, OSError): + raise probe + return probe + ran.append(list(argv)) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude", capabilities=_mcp_caps())).run("p") + + assert len(ran) == 1 + assert "--mcp-config" in ran[0] + + +def test_execute_writes_no_mcp_config_when_no_command(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["argv"] = argv + mcp_path = os.path.join(kwargs["cwd"], ".claude", "mcp-config.json") + captured["exists"] = os.path.exists(mcp_path) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + # Binding carries tools but no launch command → nothing to write. + caps = AllCapabilities( + mcp_servers=(McpBinding(name="builtin", command=(), tools=("alpha",)),), + ) + ClaudeCodeAgent(AgentConfig(target="claude", capabilities=caps)).run("p") + assert captured["exists"] is False + assert "--mcp-config" not in captured["argv"] + + +def test_execute_materializes_skills_into_workspace( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + src = tmp_path / "skills" / "my-skill" + src.mkdir(parents=True) + skill_text = "---\nname: my-skill\ndescription: do things\n---\nbody\n" + (src / "SKILL.md").write_text(skill_text) + + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + skill_path = os.path.join(kwargs["cwd"], ".claude", "skills", "my-skill", "SKILL.md") + captured["exists"] = os.path.exists(skill_path) + if captured["exists"]: + with open(skill_path) as f: + captured["text"] = f.read() + else: + captured["text"] = None + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + caps = AllCapabilities(skills=SkillBinding(paths=(str(tmp_path / "skills"),))) + ClaudeCodeAgent(AgentConfig(target="claude", capabilities=caps)).run("p") + assert captured["exists"], "skill must be materialized before subprocess" + assert captured["text"] == skill_text + + +# --------------------------------------------------------------------------- +# Parallel isolation + CLAUDE_CONFIG_DIR handling. +# --------------------------------------------------------------------------- + + +def test_execute_injects_per_run_config_dir_when_ambient_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["cwd"] = kwargs.get("cwd") + captured["config_dir"] = kwargs.get("extra_env", {}).get("CLAUDE_CONFIG_DIR") + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + + cfg_dir = captured["config_dir"] + assert cfg_dir is not None + assert os.path.basename(cfg_dir).startswith("claude-config-") + assert os.path.realpath(cfg_dir).startswith(os.path.realpath(tempfile.gettempdir())) + assert os.path.expanduser("~/.claude") not in os.path.realpath(cfg_dir) + # cwd is a fresh throwaway too. + assert os.path.basename(captured["cwd"]).startswith("claude-run-") + + +def test_execute_respects_operator_config_dir(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("CLAUDE_CONFIG_DIR", "/operator/claude") + monkeypatch.delenv("BENCH_PARALLEL", raising=False) + captured: dict = {} + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + captured["config_dir"] = kwargs.get("extra_env", {}).get("CLAUDE_CONFIG_DIR") + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + ClaudeCodeAgent(AgentConfig(target="claude")).run("p") + # The harness does not override an operator-exported value (it flows through + # os.environ, not the overlay), so no CLAUDE_CONFIG_DIR is added to extra_env. + assert captured["config_dir"] is None + + +def test_execute_overrides_operator_config_dir_under_parallel( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``--parallel`` means several benchmark processes share this host. Honouring + the operator's dir would hand them one mutable Claude config to race on, so + isolation outranks the login cache the hatch exists to preserve.""" + monkeypatch.setenv("CLAUDE_CONFIG_DIR", "/operator/claude") + monkeypatch.setenv("BENCH_PARALLEL", "true") + cfg_dirs: list[str | None] = [] + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + cfg_dirs.append(kwargs.get("extra_env", {}).get("CLAUDE_CONFIG_DIR")) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + agent = ClaudeCodeAgent(AgentConfig(target="claude")) + agent.run("p") + agent.run("p") + + assert all(d and d != "/operator/claude" for d in cfg_dirs), cfg_dirs + assert len(set(cfg_dirs)) == 2, f"config dirs must be unique per run, got {cfg_dirs}" + + +def test_execute_uses_distinct_cwd_and_config_dir_per_run(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + cwds: list[str] = [] + cfg_dirs: list[str] = [] + + def fake_run(argv: list[str], **kwargs: object) -> SimpleNamespace: + cwds.append(kwargs.get("cwd")) + cfg_dirs.append(kwargs.get("extra_env", {}).get("CLAUDE_CONFIG_DIR")) + return SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(claude_mod, "run", fake_run) + agent = ClaudeCodeAgent(AgentConfig(target="claude")) + agent.run("p") + agent.run("p") + + assert len(set(map(str, cwds))) == 2, f"cwds must be unique per run, got {cwds}" + assert len(set(cfg_dirs)) == 2, f"config dirs must be unique per run, got {cfg_dirs}" diff --git a/tests/unit/evalharness/test_registry_resolution.py b/tests/unit/evalharness/test_registry_resolution.py index 4c7d64eb..dd2f4677 100644 --- a/tests/unit/evalharness/test_registry_resolution.py +++ b/tests/unit/evalharness/test_registry_resolution.py @@ -23,6 +23,7 @@ from __future__ import annotations from collections.abc import Generator +from pathlib import Path import pytest from pytest_mock import MockerFixture @@ -87,9 +88,25 @@ def test_alias_normalizes_to_canonical_key() -> None: # ``gemini-cli`` is the friendly alias for the gemini agent; resolution must # not require a path table — the alias map normalizes to ``gemini`` and the # registry returns the registered class. - agent_cls = AGENTS.get("gemini") + # + # Resolve first: builtin harnesses self-register on the lazy import that + # ``resolve_agent`` performs, so reading ``AGENTS`` ahead of it would pass + # only when some earlier test happened to import the module. agent = harness.resolve_agent("gemini-cli") - assert isinstance(agent, agent_cls) + assert isinstance(agent, AGENTS.get("gemini")) + + +def test_claude_code_alias_normalizes_to_canonical_key() -> None: + """``claude-code`` resolves to the canonical ``claude`` agent. + + The ``AGENTS`` registry has no alias mechanism; the alias lives in + ``_AGENT_TYPE_ALIASES`` and is applied only by ``resolve_agent``, whose lazy + import is also what registers the builtin — hence resolving before reading. + """ + harness = DefaultEvalHarness(project_id="p", cluster_name="c") + + agent = harness.resolve_agent("claude-code") + assert isinstance(agent, AGENTS.get("claude")) def test_unknown_agent_type_raises_not_registered() -> None: @@ -150,3 +167,24 @@ def test_entry_point_agent_resolves_with_no_harness_edit( mock_eps.assert_called_once_with(group="devops_bench.agents") # The harness threaded its built config into the entry-point-loaded agent. assert isinstance(_DummyAgent.last_config, AgentConfig) + + +def test_manifest_records_the_canonical_harness_key(tmp_path: Path, mocker: MockerFixture) -> None: + """An arm selected via an alias records the canonical key, not the alias. + + ``harness`` is contractually the canonical key (``ResultRow.harness``), and + ``setup_id`` is derived from it, so recording the alias would split one arm + into two dashboard setups. Nothing else asserts what ``_write_run_artifacts`` + writes. + """ + harness = DefaultEvalHarness(project_id="p", cluster_name="c") + harness.agent_type = "claude-code" + harness._agent_config.model = "claude-opus-4-8" # noqa: SLF001 - arm identity under test + write_manifest = mocker.patch.object(harness.reporter, "write_manifest") + mocker.patch.object(harness.reporter, "write_rows") + + harness._write_run_artifacts(tmp_path, []) # noqa: SLF001 - the unit under test + + manifest = write_manifest.call_args.args[1] + assert manifest["harness"] == "claude" + assert manifest["setupId"].startswith("claude-opus-4-8-claude")