Julian lechnerandClaude Opus 5.5
89bee27dd2
Run tasks with and without the installed setup and compare pass rate, rework, duration, tokens, cost, tool calls, and workflow checks; exclude runs that end with a client or API error. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
271 lines
16 KiB
Python
271 lines
16 KiB
Python
"""Run the benchmark tasks with and without the installed setup and compare the results.
|
|
|
|
Each run copies a task fixture into a fresh Git repository, gives the client the task prompt,
|
|
then runs hidden checks the agent never saw. A failed check gets one rework turn with the check
|
|
output. Results land in <out>/results.jsonl and <out>/summary.md.
|
|
|
|
Arms:
|
|
setup the installed configuration (CLAUDE.md, settings, plugins, skills, MCP servers).
|
|
baseline Claude with --setting-sources "" --strict-mcp-config (no user settings, plugins,
|
|
user skills, CLAUDE.md, or MCP servers); Codex with CODEX_HOME set to --codex-baseline-home,
|
|
an empty home where you ran `codex login` once.
|
|
"""
|
|
import argparse
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import queue
|
|
import re
|
|
import shutil
|
|
import statistics
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
|
|
TASKS = Path(__file__).resolve().parent / 'tasks'
|
|
CHECK = [sys.executable, '-m', 'unittest', 'discover', '-s', 'bench_hidden', '-t', '.']
|
|
CLAUDE_TOOLS = ['Read', 'Edit', 'Write', 'Glob', 'Grep', 'Agent', 'Skill', 'TodoWrite', 'Bash(python *)', 'Bash(git *)',
|
|
'Bash(ls *)', 'PowerShell(python *)', 'PowerShell(git *)', 'PowerShell(Get-ChildItem *)']
|
|
|
|
|
|
def clean_env(**values):
|
|
env = {key: value for key, value in os.environ.items() if not key.startswith(('BUN_INSPECT', 'CLAUDECODE', 'CLAUDE_CODE_ENTRYPOINT'))}
|
|
env.update(values)
|
|
return env
|
|
|
|
|
|
def kill_tree(process):
|
|
if os.name == 'nt':
|
|
subprocess.run(['taskkill', '/T', '/F', '/PID', str(process.pid)], capture_output=True, timeout=30)
|
|
else:
|
|
process.kill()
|
|
process.wait(timeout=30)
|
|
|
|
|
|
def run_jsonl(command, cwd, env, timeout):
|
|
"""Run a client, return (events, seconds, timed_out); output is read on a thread so the deadline holds."""
|
|
start = time.monotonic()
|
|
process = subprocess.Popen(command, cwd=cwd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, text=True, encoding='utf-8', errors='replace')
|
|
lines = queue.Queue()
|
|
threading.Thread(target=lambda: ([lines.put(line) for line in process.stdout], lines.put(None)), daemon=True).start()
|
|
events, timed_out = [], False
|
|
while True:
|
|
try:
|
|
line = lines.get(timeout=max(1, timeout - (time.monotonic() - start)))
|
|
except queue.Empty:
|
|
timed_out = True
|
|
break
|
|
if line is None:
|
|
break
|
|
try:
|
|
events.append(json.loads(line))
|
|
except ValueError:
|
|
pass
|
|
kill_tree(process) if process.poll() is None else process.wait()
|
|
return events, time.monotonic() - start, timed_out
|
|
|
|
|
|
def claude_turn(prompt, cwd, arm, model, session, timeout):
|
|
command = [shutil.which('claude'), '-p', prompt, '--output-format', 'stream-json', '--verbose', '--model', model,
|
|
'--permission-mode', 'auto', '--allowedTools', *CLAUDE_TOOLS]
|
|
if arm == 'baseline':
|
|
command += ['--setting-sources', '', '--strict-mcp-config']
|
|
if session:
|
|
command += ['--resume', session]
|
|
events, seconds, timed_out = run_jsonl(command, cwd, clean_env(), timeout)
|
|
result = next((event for event in reversed(events) if event.get('type') == 'result'), {})
|
|
usage = result.get('usage', {})
|
|
calls = [block for event in events if event.get('type') == 'assistant'
|
|
for block in event.get('message', {}).get('content', []) if block.get('type') == 'tool_use']
|
|
commands = [str(call.get('input', {}).get('command', '')) for call in calls if call.get('name') in ('Bash', 'PowerShell')]
|
|
return {
|
|
'session': result.get('session_id') or session, 'seconds': seconds, 'timed_out': timed_out,
|
|
'input_tokens': usage.get('input_tokens', 0) + usage.get('cache_read_input_tokens', 0) + usage.get('cache_creation_input_tokens', 0),
|
|
'output_tokens': usage.get('output_tokens', 0), 'cost_usd': result.get('total_cost_usd') or 0,
|
|
'tool_calls': len(calls), 'agents': sum(call.get('name') == 'Agent' for call in calls),
|
|
'skills': [str(call.get('input', {}).get('skill', '')) for call in calls if call.get('name') == 'Skill'],
|
|
'commands': commands, 'final': str(result.get('result', '')),
|
|
'error': str(result.get('result', '') or 'no result event')[:300] if result.get('is_error') or not result else '',
|
|
}
|
|
|
|
|
|
def codex_turn(prompt, cwd, arm, model, session, timeout, baseline_home):
|
|
command = [shutil.which('codex'), 'exec']
|
|
if session:
|
|
command += ['resume', session]
|
|
command += ['--json', '--skip-git-repo-check', '--sandbox', 'workspace-write']
|
|
if model:
|
|
command += ['-m', model]
|
|
command += ['-C', str(cwd)] if not session else []
|
|
command.append(prompt)
|
|
env = clean_env(CODEX_HOME=str(baseline_home)) if arm == 'baseline' else clean_env()
|
|
events, seconds, timed_out = run_jsonl(command, cwd, env, timeout)
|
|
items = [event.get('item', {}) for event in events if event.get('type') == 'item.completed']
|
|
tools = [item for item in items if item.get('type') not in ('agent_message', 'reasoning', 'error')]
|
|
usage = [event.get('usage', {}) for event in events if event.get('type') == 'turn.completed']
|
|
thread = next((event.get('thread_id') for event in events if event.get('type') == 'thread.started'), session)
|
|
return {
|
|
'session': thread, 'seconds': seconds, 'timed_out': timed_out,
|
|
'input_tokens': sum(entry.get('input_tokens', 0) for entry in usage),
|
|
'output_tokens': sum(entry.get('output_tokens', 0) for entry in usage), 'cost_usd': 0,
|
|
'tool_calls': len(tools), 'agents': sum(item.get('type') == 'collab_tool_call' for item in tools),
|
|
'skills': [], 'commands': [str(item.get('command', '')) for item in tools if item.get('type') == 'command_execution'],
|
|
'final': next((str(item.get('text', '')) for item in reversed(items) if item.get('type') == 'agent_message'), ''),
|
|
'error': next((str(event.get('message', event.get('error', '')))[:300] for event in events if event.get('type') in ('error', 'turn.failed')
|
|
and not str(event.get('message', '')).startswith('Reconnecting')), '') if not usage else '',
|
|
}
|
|
|
|
|
|
def check(work, task):
|
|
hidden = work / 'bench_hidden'
|
|
shutil.rmtree(hidden, ignore_errors=True)
|
|
shutil.copytree(task / 'hidden', hidden)
|
|
result = subprocess.run(CHECK, cwd=work, capture_output=True, text=True, timeout=300)
|
|
shutil.rmtree(hidden, ignore_errors=True)
|
|
return result.returncode == 0, (result.stdout + result.stderr)[-3000:]
|
|
|
|
|
|
def workflow_violations(spec, turns):
|
|
skills = [skill for turn in turns for skill in turn['skills']]
|
|
agents = sum(turn['agents'] for turn in turns)
|
|
calls = sum(turn['tool_calls'] for turn in turns)
|
|
commands = [command for turn in turns for command in turn['commands']]
|
|
problems = [f'used planning skill {skill}' for skill in skills if skill in spec.get('forbid_skills', [])]
|
|
if agents > spec.get('max_agents', 99):
|
|
problems.append(f'{agents} agents > {spec["max_agents"]}')
|
|
if agents < spec.get('min_agents', 0):
|
|
problems.append(f'{agents} agents < {spec["min_agents"]}')
|
|
if calls > spec.get('max_tool_calls', 10 ** 6):
|
|
problems.append(f'{calls} tool calls > {spec["max_tool_calls"]}')
|
|
if spec.get('require_command') and not any(re.search(spec['require_command'], command) for command in commands):
|
|
problems.append(f'no command matching {spec["require_command"]}')
|
|
return problems
|
|
|
|
|
|
def run_one(job, args):
|
|
client, arm, task, repetition = job
|
|
spec = json.loads((task / 'task.json').read_text(encoding='utf-8'))
|
|
with tempfile.TemporaryDirectory(prefix=f'ai-bench-{task.name[:2]}-', ignore_cleanup_errors=True) as directory:
|
|
work = Path(directory)
|
|
shutil.copytree(task / 'repo', work, dirs_exist_ok=True)
|
|
for git in (['init', '-q'], ['add', '-A'], ['-c', 'user.name=bench', '-c', 'user.email=bench@example.invalid', 'commit', '-qm', 'fixture']):
|
|
subprocess.run(['git', '-c', 'core.autocrlf=false', *git], cwd=work, check=True, timeout=60)
|
|
turns, prompt, session = [], spec['prompt'], None
|
|
for round_number in range(1 + args.rework):
|
|
if client == 'claude':
|
|
turn = claude_turn(prompt, work, arm, args.claude_model, session, args.timeout)
|
|
else:
|
|
turn = codex_turn(prompt, work, arm, args.codex_model, session, args.timeout, args.codex_baseline_home)
|
|
turns.append(turn)
|
|
session = turn['session']
|
|
if turn['error']:
|
|
passed, output = False, ''
|
|
break
|
|
passed, output = check(work, task)
|
|
if passed or turn['timed_out'] or not session:
|
|
break
|
|
prompt = f'The acceptance checks failed:\n\n{output}\n\nFix the code so they pass.'
|
|
if not passed and args.keep_failed:
|
|
shutil.copytree(work, args.out / 'failed' / f'{client}-{arm}-{task.name}-{repetition}', ignore=shutil.ignore_patterns('.git'))
|
|
return {
|
|
'client': client, 'arm': arm, 'task': task.name, 'repetition': repetition, 'passed': passed,
|
|
'error': next((turn['error'] for turn in turns if turn['error']), ''),
|
|
'first_try': passed and len(turns) == 1, 'rework_turns': len(turns) - 1,
|
|
'seconds': round(sum(turn['seconds'] for turn in turns), 1),
|
|
'input_tokens': sum(turn['input_tokens'] for turn in turns), 'output_tokens': sum(turn['output_tokens'] for turn in turns),
|
|
'cost_usd': round(sum(turn['cost_usd'] for turn in turns), 4), 'tool_calls': sum(turn['tool_calls'] for turn in turns),
|
|
'agents': sum(turn['agents'] for turn in turns), 'skills': [skill for turn in turns for skill in turn['skills']],
|
|
'commands': [command for turn in turns for command in turn['commands']],
|
|
'timed_out': any(turn['timed_out'] for turn in turns), 'workflow_kind': spec['workflow'].get('kind'),
|
|
'workflow_violations': workflow_violations(spec['workflow'], turns), 'check_output': '' if passed else output,
|
|
}
|
|
|
|
|
|
def summarize(results):
|
|
errors = [result for result in results if result.get('error')]
|
|
results = [result for result in results if not result.get('error')]
|
|
lines = [f'Excluded {len(errors)} runs that ended with a client or API error (for example a usage limit).', '','| client | arm | runs | pass | first try | rework turns | median s | median input tok | median output tok | median tool calls | cost USD | workflow ok |',
|
|
'|---|---|---|---|---|---|---|---|---|---|---|---|']
|
|
groups = {}
|
|
for result in results:
|
|
groups.setdefault((result['client'], result['arm']), []).append(result)
|
|
for (client, arm), runs in sorted(groups.items()):
|
|
median = lambda key: statistics.median(run[key] for run in runs)
|
|
lines.append(f'| {client} | {arm} | {len(runs)} | {sum(run["passed"] for run in runs) / len(runs):.0%} | '
|
|
f'{sum(run["first_try"] for run in runs) / len(runs):.0%} | {sum(run["rework_turns"] for run in runs)} | '
|
|
f'{median("seconds"):.0f} | {median("input_tokens"):.0f} | {median("output_tokens"):.0f} | {median("tool_calls"):.0f} | '
|
|
f'{sum(run["cost_usd"] for run in runs):.2f} | {sum(not run["workflow_violations"] for run in runs) / len(runs):.0%} |')
|
|
lines += ['', '## Workflow checks', '', '| client | arm | task | violations |', '|---|---|---|---|']
|
|
for result in sorted(results, key=lambda run: (run['client'], run['arm'], run['task'], run['repetition'])):
|
|
if result['workflow_violations']:
|
|
lines.append(f'| {result["client"]} | {result["arm"]} | {result["task"]} #{result["repetition"]} | {"; ".join(result["workflow_violations"])} |')
|
|
return '\n'.join(lines) + '\n'
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0], formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
|
|
parser.add_argument('--clients', nargs='+', choices=('claude', 'codex'), default=['claude', 'codex'])
|
|
parser.add_argument('--arms', nargs='+', choices=('setup', 'baseline'), default=['setup', 'baseline'])
|
|
parser.add_argument('--tasks', nargs='+', help='task directory names (default: all)')
|
|
parser.add_argument('--repetitions', type=int, default=3)
|
|
parser.add_argument('--first-repetition', type=int, default=1, help='number of the first repetition, to rerun selected repetitions')
|
|
parser.add_argument('--rework', type=int, default=1, help='rework turns after a failed check')
|
|
parser.add_argument('--jobs', type=int, default=2)
|
|
parser.add_argument('--timeout', type=int, default=900, help='seconds per client turn')
|
|
parser.add_argument('--claude-model', default='sonnet')
|
|
parser.add_argument('--codex-model', help='default: the CLI default, identical for both arms')
|
|
parser.add_argument('--codex-baseline-home', type=Path, default=Path.home() / '.my-ai-configuration/benchmark/codex-home')
|
|
parser.add_argument('--out', type=Path, default=Path(tempfile.gettempdir()) / f'ai-config-ab-{time.strftime("%Y%m%d-%H%M%S")}')
|
|
parser.add_argument('--dry-run', action='store_true', help='list the runs without starting clients')
|
|
parser.add_argument('--keep-failed', action='store_true', help='copy failed work trees to <out>/failed')
|
|
parser.add_argument('--summarize', nargs='+', type=Path, metavar='RESULTS', help='only summarize results.jsonl files; a later file replaces runs of an earlier one')
|
|
args = parser.parse_args()
|
|
if args.summarize:
|
|
merged = {}
|
|
for path in args.summarize:
|
|
for line in path.read_text(encoding='utf-8').splitlines():
|
|
result = json.loads(line)
|
|
key = (result['client'], result['arm'], result['task'], result['repetition'])
|
|
if not result.get('error') or key not in merged:
|
|
merged[key] = result
|
|
print(summarize(list(merged.values())))
|
|
return 0
|
|
tasks = [TASKS / name for name in args.tasks] if args.tasks else sorted(path for path in TASKS.iterdir() if path.is_dir())
|
|
for client in args.clients:
|
|
if not shutil.which(client):
|
|
parser.error(f'{client} CLI unavailable')
|
|
if 'codex' in args.clients and 'baseline' in args.arms and not (args.codex_baseline_home / 'auth.json').is_file():
|
|
parser.error(f'Codex baseline needs a login: run CODEX_HOME="{args.codex_baseline_home}" codex login')
|
|
jobs = [(client, arm, task, repetition) for repetition in range(args.first_repetition, args.first_repetition + args.repetitions)
|
|
for task in tasks for client in args.clients for arm in args.arms]
|
|
print(f'{len(jobs)} runs -> {args.out}')
|
|
if args.dry_run:
|
|
for client, arm, task, repetition in jobs:
|
|
print(f' {client} {arm} {task.name} #{repetition}')
|
|
return 0
|
|
args.out.mkdir(parents=True, exist_ok=True)
|
|
results, lock = [], threading.Lock()
|
|
|
|
def record(job):
|
|
result = run_one(job, args)
|
|
with lock:
|
|
results.append(result)
|
|
with open(args.out / 'results.jsonl', 'a', encoding='utf-8') as stream:
|
|
stream.write(json.dumps(result) + '\n')
|
|
print(f'[{len(results)}/{len(jobs)}] {result["client"]} {result["arm"]} {result["task"]} #{result["repetition"]}: '
|
|
f'{"PASS" if result["passed"] else "FAIL"} {result["seconds"]}s {result["tool_calls"]} tools', flush=True)
|
|
|
|
with ThreadPoolExecutor(max_workers=args.jobs) as pool:
|
|
list(pool.map(record, jobs))
|
|
(args.out / 'summary.md').write_text(summarize(results), encoding='utf-8')
|
|
print(summarize(results))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|