diff --git a/evalbench/generators/models/claude_code.py b/evalbench/generators/models/claude_code.py index 8c56294e..be6c6b53 100644 --- a/evalbench/generators/models/claude_code.py +++ b/evalbench/generators/models/claude_code.py @@ -8,6 +8,7 @@ import sys import re import shutil +import threading import time from typing import Optional, Union, Dict, List import uuid @@ -634,6 +635,142 @@ def _execute_cli_command( command, 1, "", f"An unexpected error occurred: {e}" ) + def _execute_cli_streaming( + self, command: list[str], env: dict[str, str] | None = None, + cwd: str | None = None, timeout_seconds: float | int | None = None, + ) -> tuple[subprocess.CompletedProcess, dict[str, int]]: + """Runs the CLI with line-streamed stdout, returning the usual + CompletedProcess plus a `{tool_use_id: duration_ms}` map. + + Claude's stream-json carries no timestamps, so stamping arrival times + here is the only path to per-tool latency. + """ + try: + proc = subprocess.Popen( + command, env=env, cwd=cwd or self.fake_home, text=True, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=1, # line-buffered so events arrive as Claude flushes them + ) + except FileNotFoundError: + return subprocess.CompletedProcess( + command, 127, "", f"Error: Command not found: {command[0]}" + ), {} + except Exception as e: + return subprocess.CompletedProcess( + command, 1, "", f"An unexpected error occurred: {e}" + ), {} + + stderr_chunks: list[str] = [] + + def _drain_stderr(): + try: + for line in proc.stderr: + stderr_chunks.append(line) + except Exception as e: + logging.debug(f"stderr drain failed: {e}") + + stderr_thread = threading.Thread(target=_drain_stderr, daemon=True) + stderr_thread.start() + + stdout_lines: list[str] = [] + started_at_ms: dict[str, float] = {} + tool_durations: dict[str, int] = {} + + def _drain_stdout(): + try: + for line in proc.stdout: + arrival_ms = time.monotonic() * 1000 + stdout_lines.append(line) + # Timing is a side channel; a bad line must not cost us + # the rest of stdout. + try: + self._stamp_tool_event( + line, arrival_ms, started_at_ms, tool_durations, + ) + except Exception as e: + logging.debug(f"tool timing skipped for a line: {e}") + except Exception as e: + logging.warning(f"stdout stream read failed: {e}") + + stdout_thread = threading.Thread(target=_drain_stdout, daemon=True) + stdout_thread.start() + + try: + proc.wait(timeout=timeout_seconds) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() + stdout_thread.join(timeout=5) + stderr_thread.join(timeout=5) + stderr_str = ( + f"TimeoutError: Command timed out after {timeout_seconds} seconds") + if stderr_chunks: + stderr_str = f"{stderr_str}\n{''.join(stderr_chunks)}" + return subprocess.CompletedProcess( + command, 124, "".join(stdout_lines), stderr_str, + ), tool_durations + + stdout_thread.join(timeout=5) + stderr_thread.join(timeout=5) + + completed = subprocess.CompletedProcess( + command, proc.returncode, + "".join(stdout_lines), "".join(stderr_chunks), + ) + return completed, tool_durations + + @staticmethod + def _stamp_tool_event( + line: str, arrival_ms: float, + started_at_ms: dict[str, float], tool_durations: dict[str, int], + ) -> None: + """Stamps a `tool_use` block's arrival and closes it on the matching + `tool_result`, which arrives either as its own event or nested in a + `user` message. + """ + line = line.strip() + if not line: + return + try: + event = json.loads(line) + except json.JSONDecodeError: + return + if not isinstance(event, dict): + return + + event_type = event.get("type") + message = event.get("message") + content = message.get("content") if isinstance(message, dict) else None + if not isinstance(content, list): + content = [] + + if event_type == "assistant": + for block in content: + if not isinstance(block, dict): + continue + if block.get("type") == "tool_use" and block.get("id"): + started_at_ms.setdefault(block["id"], arrival_ms) + return + + if event_type == "tool_result": + result_ids = [event.get("tool_use_id") or event.get("id", "")] + elif event_type == "user": + result_ids = [ + block.get("tool_use_id") or block.get("id", "") + for block in content + if isinstance(block, dict) + and block.get("type") == "tool_result" + ] + else: + return + + for tool_id in result_ids: + t0 = started_at_ms.pop(tool_id, None) if tool_id else None + if t0 is not None: + tool_durations[tool_id] = max(0, int(arrival_ms - t0)) + @staticmethod def _session_id_headers() -> str: """Builds a dynamic per-run ``ANTHROPIC_CUSTOM_HEADERS`` value carrying a unique session id. @@ -708,9 +845,11 @@ def _run_claude_code(self, cli_cmd: CLICommand, timeout_seconds=None): logging.info(f"Running Claude Code CLI: {' '.join(command)}") - result = self._execute_cli_command(command, env=env, cwd=cli_cmd.cwd, timeout_seconds=timeout_seconds) + result, tool_durations = self._execute_cli_streaming( + command, env=env, cwd=cli_cmd.cwd, timeout_seconds=timeout_seconds) if result.stdout: - result.stdout = self._parse_stream_json(result.stdout) + result.stdout = self._parse_stream_json( + result.stdout, tool_durations=tool_durations) return result @@ -739,9 +878,16 @@ def _record_tool_result(self, tool_calls_dict, tool_id, is_error, content): tool_calls_dict[tool_id]["response"] = self._stringify_tool_result( content) - def _parse_stream_json(self, stream_output: str) -> str: + def _parse_stream_json( + self, stream_output: str, + tool_durations: dict[str, int] | None = None, + ) -> str: """Parses Claude Code stream-json output into a normalized format - compatible with the eval pipeline.""" + compatible with the eval pipeline. + + ``tool_durations`` maps tool_use ids to milliseconds. + """ + tool_durations = tool_durations or {} from collections import OrderedDict @@ -964,6 +1110,10 @@ def _parse_stream_json(self, stream_output: str) -> str: tstat["decisions"]["accept"] += 1 tstat["decisions"]["auto_accept"] += 1 + duration = tool_durations.get(tc.get("tool_id"), 0) + tstat["durationMs"] += duration + tools_stats["totalDurationMs"] += duration + if tc.get("status") == "success": tstat["success"] += 1 elif tc.get("status") == "error": diff --git a/evalbench/test/claude_code_test.py b/evalbench/test/claude_code_test.py index fa4d6bc7..e7a117db 100644 --- a/evalbench/test/claude_code_test.py +++ b/evalbench/test/claude_code_test.py @@ -1,3 +1,4 @@ +import json import os import sys from unittest.mock import MagicMock, patch, ANY @@ -88,3 +89,143 @@ def test_install_plugin_runs_init_and_install(mock_open, mock_makedirs, monkeypa assert "plugins" in second_call_args assert "install" in second_call_args assert "my-plugin" in second_call_args + + +TOOL_USE_EVENT = { + "type": "assistant", + "message": { + "content": [ + { + "type": "tool_use", + "id": "toolu_01", + "name": "mcp__cloud-sql__list_instances", + "input": {"project": "p"}, + } + ] + }, +} + +TOOL_RESULT_EVENT = { + "type": "user", + "message": { + "content": [ + {"type": "tool_result", "tool_use_id": "toolu_01", "content": "ok"} + ] + }, +} + +RESULT_EVENT = {"type": "result", "session_id": "s1", "usage": {}} + + +def _stamp(events_at): + started_at_ms, tool_durations = {}, {} + for event, arrival_ms in events_at: + ClaudeCodeGenerator._stamp_tool_event( + json.dumps(event), arrival_ms, started_at_ms, tool_durations) + return tool_durations + + +def test_stamp_tool_event_measures_gap_between_use_and_result(): + assert _stamp([(TOOL_USE_EVENT, 1000.0)]) == {} + assert _stamp( + [(TOOL_USE_EVENT, 1000.0), (TOOL_RESULT_EVENT, 1250.0)] + ) == {"toolu_01": 250} + + +def test_stamp_tool_event_handles_top_level_tool_result(): + top_level = {"type": "tool_result", "tool_use_id": "toolu_01"} + assert _stamp( + [(TOOL_USE_EVENT, 1000.0), (top_level, 1100.0)] + ) == {"toolu_01": 100} + + +def test_stamp_tool_event_pairs_parallel_calls_by_id(): + parallel_use = { + "type": "assistant", + "message": { + "content": [ + {"type": "tool_use", "id": "toolu_01", "name": "a"}, + {"type": "tool_use", "id": "toolu_02", "name": "b"}, + ] + }, + } + + def result_for(tool_id): + return { + "type": "user", + "message": { + "content": [{"type": "tool_result", "tool_use_id": tool_id}] + }, + } + + assert _stamp([ + (parallel_use, 1000.0), + (result_for("toolu_02"), 1100.0), + (result_for("toolu_01"), 1300.0), + ]) == {"toolu_01": 300, "toolu_02": 100} + + +def test_stamp_tool_event_ignores_unpaired_and_malformed_lines(): + orphan_result = {"type": "user", "message": { + "content": [{"type": "tool_result", "tool_use_id": "unknown"}]}} + assert _stamp([(orphan_result, 1000.0)]) == {} + + started_at_ms, tool_durations = {}, {} + for line in ("", " ", "not json", "[]"): + ClaudeCodeGenerator._stamp_tool_event( + line, 1000.0, started_at_ms, tool_durations) + assert tool_durations == {} + + +def test_streaming_timeout_keeps_stderr_diagnostics(): + """The non-streaming path appends captured stderr to the timeout message; + dropping it here loses the CLI's own reason for hanging.""" + generator = object.__new__(ClaudeCodeGenerator) + generator.fake_home = os.getcwd() + + result, _ = ClaudeCodeGenerator._execute_cli_streaming( + generator, + ["sh", "-c", "echo 'rate limit reached' >&2; sleep 30"], + timeout_seconds=1, + ) + + assert result.returncode == 124 + assert "TimeoutError: Command timed out after 1 seconds" in result.stderr + assert "rate limit reached" in result.stderr + + +def test_stamp_tool_event_survives_null_message_and_content(): + """A null `message`/`content` defeats dict.get defaults, and the raised + AttributeError would drop the timing for every tool after it.""" + null_events = [ + {"type": "user", "message": None}, + {"type": "assistant", "message": None}, + {"type": "assistant", "message": {"content": None}}, + {"type": "user", "message": {"content": "plain text"}}, + ] + events_at = [(event, 1050.0) for event in null_events] + assert _stamp( + [(TOOL_USE_EVENT, 1000.0)] + events_at + [(TOOL_RESULT_EVENT, 1250.0)] + ) == {"toolu_01": 250} + + +@patch('generators.models.claude_code.os.makedirs') +@patch('generators.models.claude_code.open', create=True) +def test_parse_stream_json_accumulates_tool_durations( + mock_open, mock_makedirs, monkeypatch): + """durationMs was initialized and never accumulated, so tool_call_latency + scored 0 for every scenario.""" + monkeypatch.setenv("HOME", "/fake/real_home") + mock_open.return_value.__enter__.return_value.read.return_value = '{}' + + generator = ClaudeCodeGenerator({"model": "claude-opus-4-6"}) + stream = "\n".join( + json.dumps(e) + for e in (TOOL_USE_EVENT, TOOL_RESULT_EVENT, RESULT_EVENT)) + + parsed = json.loads(generator._parse_stream_json( + stream, tool_durations={"toolu_01": 250})) + + tools = parsed["stats"]["tools"] + assert tools["totalDurationMs"] == 250 + assert tools["byName"]["cloud-sql__list_instances"]["durationMs"] == 250