Test-drive coding agents on tasks from your own repo
Runs Claude Code, Codex, and Gemini CLI on tasks taken from your own git history, scores each one with your repository's tests, and reports how many tasks each passed, the dollars per pass, the minutes, and the size of each change.
What you get
A sample report. The numbers and commits are invented; your run prints its own.
**On 8 tasks from your git history, Claude Code passed 7 at $0.84 each and Codex passed 6 at $0.47 each.**
| Harness | Version | Passed | Pass rate | Median minutes | Cost per pass | Total cost | Median lines changed |
|---|---|---|---|---|---|---|---|
| Claude Code | `2.1.300 (Claude Code)` | 7 of 8 | 88% | 4.2 | $0.84 | $5.88 | 14 |
| Codex | `codex-cli 0.160.0` | 6 of 8 | 75% | 6.8 | $0.47 | $2.82 | 9 |
Per task (the size of the original fix in changed lines, then each harness):
| Task | Change | Original fix lines | Claude Code | Codex |
|---|---|---|---|---|
| 3f2a91c | `Fix off-by-one on the pager's last page` | 6 | passed, 3.1 min, $0.48 | passed, 5.2 min, $0.31 |
| 9b07d4e | `Retry uploads after a 429 response` | 41 | passed, 7.9 min, $1.62 | failed, 9.4 min, $0.66 |
| c41e8a2 | `Keep time zones when parsing due dates` | 12 | failed, 6.3 min, $0.97 | passed, 4.4 min, $0.28 |
- Dollar figures are API list prices: what the harness reported, or its tokens times the price
table checked 2026-09-28. On a subscription plan, runs count against the plan's limits instead.
Counted toward the spend cap: $8.70.
When a harness cannot run at all, the headline says so instead of scoring it: "On 8 tasks from your git history, Codex passed 6 at $0.47 each; Claude Code could not run (Failed to authenticate: OAuth session expired)." The run prints the fix ("sign in: run claude once") and retries those tasks next time at no charge.
Manual: copy skills/harness-test-drive/ into your agent's skills folder: ~/.claude/skills/ (Claude Code), ~/.agents/skills/ (Codex, Gemini CLI, Cursor, and OpenCode all read it).
Use it
Ask your agent: "Which coding agent works best on this repo? Test-drive Claude Code and Codex on tasks from my git history."
The agent finds your test command, keeps the past commits whose tests fail before the change and pass after it, shows how many runs that means and a cost range, and asks for your spending cap before it runs anything. To run the steps yourself, from the root of your repository:
The setup command installs into each fresh copy: npm ci, pnpm install --frozen-lockfile, or python3 -m venv .venv && .venv/bin/pip install -e . with the test command .venv/bin/python -m pytest. Add --model codex=gpt-6-sol to pin a model, and --allow-unpriced gemini-cli to include Gemini CLI, whose spend cannot be counted. A stopped run resumes with the same command.
How it works
Mine.mine_tasks.py reads your git history for commits that changed both code and tests. The commit message becomes the task; the commit's test changes become hidden tests the agent never sees.
Check. With --validate, each task runs twice in a fresh copy of your repository: the tests must fail at the commit before the change (with the hidden tests added) and pass with the change. Tasks that fail this check would score an agent that does nothing, and tasks whose result changes between runs are flaky; both are dropped.
Run.drive.py run gives every harness the same prompt in its own fresh copy at the commit before the change, with the most restrictive settings that still allow edits and the test command. The copy is a new git repository whose only commit is that starting point, so the real fix is not in its history (an agent that searched your disk could still find your repository).
Score. The agent's saved diff, minus test files, goes into a clean copy with the hidden tests, the setup runs again, and your test command decides: passed or failed. A harness that exits with an error and changes nothing could not run, and that run is not scored. The scoreboard compares harnesses only on tasks all of them finished.
Your spending cap is required. The script stops starting runs once the counted spend reaches it, gives Claude Code the budget left as its own limit, and counts earlier runs when it resumes. Stopping it (Ctrl-C, a closed terminal, a stopped background task) ends the agent, records the run with its spend, and deletes the copy. The method reference covers every rule, and the command reference lists each harness's flags with sources.
Works with
Harness
Run as
Cost measured from
Per-run limit
Support
Claude Code
claude -p, edits allowed; in the shell, the test command, plus your own allow rules; no MCP servers
its own report
budget left, plus the timeout
full
Codex
codex exec, workspace-write sandbox
tokens times the price of its model
the timeout
needs a known model: its config's, or --model codex=<id>
Gemini CLI
gemini -p, edits allowed; in the shell, the test command, plus your own allow rules
not measured
the timeout
runs and scores with --allow-unpriced gemini-cli
OpenCode, Cursor CLI
not supported
OpenCode's JSON events are not documented, and Cursor reports no tokens
Python 3.9 or newer and git, on macOS or Linux. Nothing else to install besides the harnesses you compare.
Limits
A handful of tasks gives a rough answer: with five or ten tasks, one or two passes of difference can be noise.
Tests check behavior, not code quality. Every run's diff is saved in .harness-test-drive/logs/ for a human review.
Test files and test-runner settings are put back before scoring, but a change to a file that mixes other settings in, such as package.json or pyproject.toml, can still alter how tests run. The diff shows it.
For a public repository, a model may have seen the real change during training.
Only commits that changed tests become tasks, so well-tested code is overrepresented, and a terse commit message makes a hard prompt for every harness alike.
Each harness still loads your personal settings: its instruction file in your home folder, hooks, and (except Claude Code) MCP servers.
The copies leave out untracked files, submodules, and Git LFS content; tasks that need them are dropped by the check.
The cap limits counted spend. Claude Code stops itself at the budget left. A Codex run has no limit except --timeout, and a run cut off there is counted as an estimate, so the bill can pass the cap by more than one run. Runs allowed with --allow-unpriced count $0.
Privacy
Read: your git history, the files of the commits it copies, and the model line of Codex's config.
Sent: nothing, by the scripts. Each harness sends its prompt (your commit message) and the code its agent reads to its model provider, as in normal use, and that costs money.
Passed on: each harness and test command gets your environment, minus the variables that point back to the agent session that started the script.
Written: .harness-test-drive/ in your repository (tasks, results, and logs: the HEAD check's output, and per run the harness output, diff, and test output), a folder git ignores. Each run's copy lives in the system temporary folder and is deleted afterwards unless you pass --keep.
Untrusted code: the check runs your test and setup commands, and agents edit files and run the test command in each copy. For a repository you did not write, run the skill inside a container or VM.
Related
How to test-drive a harness: the manual method this skill automates, including the human review it cannot do.
The fail-before, pass-after check comes from SWE-bench. Harbor runs agents on benchmark tasks at scale, RepoTrials mines tasks from git history, and sample-agent-cost-bench measures cost across agents. Harness-Bench found that harness rankings barely carry over between models.
Files in this skill
SKILL.md is the part your agent reads. The scripts do the checking, and the references hold the detail the agent opens only when it needs it.
---
name: harness-test-drive
description: >-
Test-drives coding agents (Claude Code, Codex, Gemini CLI) on tasks mined
from the user's own git history: each agent gets a past commit message in a
fresh copy of the repo, and the repo's own tests score how many tasks it
completes, dollars per task, minutes, and change size. Use when the user asks
which coding agent or harness works best on their codebase; wants to compare
or benchmark agents on their own repository instead of trusting leaderboards
such as SWE-bench; wants to trial one agent against another before
switching; or asks what each agent costs per fixed bug. Spends money: the
harness CLIs call their model providers, so a dollar cap is required. The
scripts make no network calls.
license: MIT
compatibility: "Python 3.9+ and git on macOS or Linux, plus the harness CLIs to compare (claude, codex, gemini). The scripts make no network calls themselves; each harness CLI sends the prompt and the code it reads to its model provider over the network, and those runs cost money. The repository's test command must run on this machine."
metadata:
author: "Ryan Alberts"
version: "1.0.0"
source: "https://github.com/RyanAlberts/best-of-Agent-Harnesses"
---
# Harness test drive
Public leaderboards measure someone else's code, and harness rankings barely carry over from one
repository to the next. This skill runs coding agents on tasks taken from the user's own git history:
past commits whose tests failed before the change and passed after it. Each agent gets the commit
message as its prompt in a fresh copy of the repository, and the repository's own tests decide
whether it passed. The result is a scoreboard of passes, dollars per pass, minutes, and change size.
The scripts read the repository and send nothing anywhere; each harness sends the prompt and the
code it reads to its model provider, as it does in normal use, and that costs money.
## When to use
- The user asks which coding agent or harness is best for their own code.
- The user wants to compare Claude Code, Codex, or Gemini CLI on real tasks before choosing or
switching.
- The user asks what an agent costs per fixed bug, or how long it takes, on their repository.
## When not to use
- Totals of past spend or wasted tokens: use `session-waste-report`.
- Whether an agent got worse after an update: use `regression-finder`.
- Stopping a live session that loops or overspends: use `runaway-guard`.
- Checking whether an agent's "tests pass" claims were true: use `claim-check`.
- A repository with no test command that passes, or no commits that change tests: explain that
the scores come from the tests, and point to the manual method in
[How to test-drive a harness](https://github.com/RyanAlberts/best-of-Agent-Harnesses/blob/main/comparisons/how-to-test-drive-a-harness.md).
## Steps
`<skill-dir>` means the folder that holds this SKILL.md (Claude Code shows it as the skill's base
directory). Run every command from the root of the user's repository, with the skill path in quotes
as shown. Inside the repository the scripts write only into `.harness-test-drive/`, a folder that
ignores itself in git. Every check and every agent run works in its own temporary copy, deleted
afterwards.
**Before step 2, for a repository the user did not write, recommend a container or VM.** Step 2
runs the repository's test and setup commands at up to 31 commits on this computer. In step 5 each
harness loads the repository's own agent settings (Gemini CLI with the copy trusted), edits files,
runs the test command, and reads commit messages as prompts.
1. **Find the test command and the candidate tasks.** This reads git history and runs nothing:
```bash
python3 "<skill-dir>/scripts/mine_tasks.py" --repo .
```
It prints the test command it detected, or exits 2 asking for one. Confirm the command with the
user. Each check runs in a fresh copy with nothing installed, so when the tests need
dependencies, add a `--setup-cmd` that installs inside the copy: `npm ci`,
`pnpm install --frozen-lockfile`, or `python3 -m venv .venv && .venv/bin/pip install -e .` with
the test command `.venv/bin/python -m pytest`. Never install into the user's own environment.
Prefer a test command that runs offline and writes only inside the copy, so Codex's sandbox can
run it too. Done when the user has confirmed a test command and the report shows at least one
candidate, or you have told the user why there is none.
2. **Check the tasks.** For each candidate, in a fresh copy of the repository, the test command must
fail at the commit before the change (with its tests added) and pass with the whole change, twice:
```bash
python3 "<skill-dir>/scripts/mine_tasks.py" --repo . --test-cmd "<test-cmd>" --validate
```
Add the `--setup-cmd` from step 1 when there is one. It first checks that the tests pass at HEAD,
then keeps up to 10 tasks (`--max`). Expect two to four test-suite runs per candidate and up to 30
candidates, so run it in the background when the suite takes more than a minute. Done when the
headline says how many tasks were kept. Exit 2 with "fails at HEAD" means the command or the
setup is wrong. The message ends with the test output. A missing module or command means the
clean copy needs `--setup-cmd`. Fix it with the user and rerun. Fewer than five tasks makes a
weak comparison; say so, and offer `--since 730d` for more history.
3. **Choose the harnesses and show the estimate:**
```bash
python3 "<skill-dir>/scripts/drive.py" estimate --tasks .harness-test-drive/tasks.json
```
It lists which harnesses are on this computer, the number of runs, and a dollar range per
harness. Ask the user to pick two or three. Done when the user has picked the harnesses and seen
the range for them (rerun with `--harness` to show only those).
4. **Get an explicit dollar cap.** This is the one skill in the set that spends money. Ask: "What is
the most you want to spend in total?" and wait for a number; a number the user already gave
counts, so confirm it next to the estimate. Never choose the cap yourself. In the same message,
say:
- The cap limits counted spend. Claude Code stops itself at the budget left. A Codex run has no
limit except `--timeout`, and a run cut off there is counted as an estimate, so the bill can
pass the cap by more than one run.
- On a subscription plan, runs count against the plan's limits; the cap still counts API prices.
- Gemini CLI's spend cannot be counted (it reports no cost, and the price table has no Gemini
prices), so it runs only with `--allow-unpriced gemini-cli`, after the user agrees to that.
- Codex runs are priced with the model named in Codex's config, which the estimate shows. When
the config names none, pin one with `--model codex=<id>`; Codex never runs unpriced.
- For someone else's repository, run inside a container or VM (see above).
Done when the user has given a dollar amount.
5. **Run:**
```bash
python3 "<skill-dir>/scripts/drive.py" run --tasks .harness-test-drive/tasks.json --harness claude-code,codex --max-usd <cap>
```
Add `--allow-unpriced gemini-cli` only when the user agreed to it in step 4, and `--model` only
for a model the user pinned. Each run can take up to `--timeout` seconds (900 by default), so
start it in the background and relay the progress lines. A harness that cannot run (an expired
login, an old version) prints a fix, such as "sign in: run claude once" or "upgrade Codex", and
its other runs are skipped at no cost. Every run is appended to
`.harness-test-drive/results.jsonl`; the same command resumes after a stop, retries runs that
could not run, and counts earlier spend. Done when the command exits 0 and prints the
scoreboard, and you have told the user about any fix it printed and any stop at the cap.
6. **Report** in the shape below. To print the scoreboard again at any time:
```bash
python3 "<skill-dir>/scripts/drive.py" report
```
## Read the results
- **Headline**: "On 6 tasks from your git history, Claude Code passed 5 at $0.91 each and Codex
passed 4 at $0.42 each." Harnesses appear in rank order: most passes, then the lowest cost per
pass. The count covers the tasks every harness finished, so a stop at the cap leaves a fair
comparison. Dollars per pass is all the spend on those tasks, failed runs included, divided by
the passes. A harness with no finished run appears as "could not run", with its own error.
- **Per-harness table**: passed, pass rate, median minutes per run, cost per pass, total cost, and
median lines changed. Compare lines changed with "Original fix lines" in the per-task table: far
more lines than the original often means a messier change.
- **Per-task table**: `passed`, `failed`, `(time limit)` when the run was stopped at `--timeout`
and scored on the changes the agent had made, `could not run` (not scored; see the notes),
`interrupted`, or `not run`.
- **Scoring**: the agent is graded only by the repository's tests. Its saved diff, minus test
files, is applied to a clean copy at the starting commit, the change's own tests are added, the
setup runs again, and the test command decides. The whole suite must pass.
- **Notes**: tasks left out of the comparison, time limits, runs that could not run and their fix,
costs that were not measured, and how often Codex's own test runs failed inside its sandbox.
- `references/method.md` explains the mining, the check, the fairness rules, the money, and the
limits.
## Report to the user
1. The headline, verbatim, in bold.
2. The per-harness table, then at most five rows of the per-task table where the harnesses
disagree.
3. Two or three next steps, each with its command or link:
- Read the diffs where harnesses disagree: each run's changes are saved as
`.harness-test-drive/logs/<task>-<harness>.diff`. Tests check behavior, not code quality.
- For a firmer answer, add tasks (`--max 20`, or `--since 730d`); under ten tasks is a small
sample.
- For the human half of a trial (a second person accepts or rejects each diff, interventions,
setup friction), follow
[How to test-drive a harness](https://github.com/RyanAlberts/best-of-Agent-Harnesses/blob/main/comparisons/how-to-test-drive-a-harness.md).
4. One line of caveats: the sample size, and that a model may have seen a public repository's
changes during training.
Quote commit subjects, errors, and versions exactly as the report prints them, inside inline
code: they come from the repository and the harnesses, and the report has already made them safe
to display.
## Files
- `scripts/mine_tasks.py`: finds candidate commits, detects the test command, and checks tasks.
- `scripts/drive.py`: `estimate`, `run`, and `report`.
- `scripts/harnesses.py`: the headless command and output parser for each harness.
- `scripts/common.py`: fresh copies of the repository, commands and test runs.
- `scripts/pricing.py`: token prices, shared with other skills in this repository.
- `scripts/safe.py`: the shared helper that masks secrets in report text and shows it as one line of
inline code. A synced copy; do not edit it here.
- `references/method.md`: task mining, the fail-to-pass check, fairness rules, money, and limits.
- `references/harness-commands.md`: each harness's command and flags, with sources and the date
checked.
MIT License
Copyright (c) 2026 Ryan Alberts
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
# Harness commands
How `scripts/harnesses.py` runs each coding agent without a person at the keyboard (headless), and
how it reads the result. Checked 2026-09-28 against each harness's docs and source, linked below.
Each command uses the most restrictive settings that still let the agent edit files in its copy of
the repository and run the test command. The prompt is the same for every harness (see
`method.md`). Every harness runs with the user's environment minus the variables that lead back to
the agent session that started the script.
## Claude Code
```
claude -p "<prompt>" --output-format stream-json --verbose --permission-mode acceptEdits
--allowedTools "Bash(<test command>),Bash(<test command> *)" --no-session-persistence
--strict-mcp-config --max-budget-usd <budget left> [--model <id>]
```
- `-p` runs one prompt and exits. It cannot answer permission prompts, so any call not allowed up
front is denied.
- `--permission-mode acceptEdits` approves file edits in the working folder. `--allowedTools`
approves the test command, with and without extra arguments; a compound test command such as
`npm ci && npm test` gets one rule per part, because a rule must match every part. The user's own
allow rules from their settings still apply on top.
- `--strict-mcp-config` with no `--mcp-config` loads no MCP server, so the agent works with its
built-in tools only.
- `--max-budget-usd` stops the run when its spend reaches the budget left under `--max-usd`
(subagent spend counts). The script rounds the budget down to the cent.
- `--no-session-persistence` keeps these runs out of the user's session history.
- `stream-json` with `--verbose` prints one JSON object per line as the run goes. The last line is
the result: `total_cost_usd`, `num_turns`, `modelUsage` (tokens and cost per model, subagents
included), `permission_denials`, and `is_error`. When `is_error` is true, the error shown is the
result's own message (`result`, else the first of `errors`, else `terminal_reason`), never the
`subtype`, which reads `success` on a failed login. When a run is stopped before the result, the
cost is rebuilt from the `usage` of each assistant message (the last copy of each message id,
leaving out `<synthetic>` placeholder replies), priced with `pricing.py`. Any assistant message
with usage counts as model work.
- Not used: `--dangerously-skip-permissions` and `--permission-mode bypassPermissions`.
Sources: https://code.claude.com/docs/en/headless , https://code.claude.com/docs/en/cli-reference ,
https://code.claude.com/docs/en/permissions
## Codex
```
codex exec --json --sandbox workspace-write --ephemeral [--model <id>] "<prompt>"
```
- `codex exec` runs one prompt without a person. It needs a git repository, which each fresh copy
is.
- `--sandbox workspace-write` lets commands write inside the working folder and blocks the network
unless the user's Codex config allows it; `.git` stays read-only. A test command that needs the
network, or writes outside the folder (for example a package manager's cache), fails inside the
sandbox, so Codex cannot check its own work with it. The score does not depend on that, because
the script runs the tests itself, outside any sandbox. Still, prefer a setup that installs into
the copy and a test command that runs offline. The script counts Codex's own runs of the test
command (`command_execution` items whose command contains it) and how many failed, and the report
notes the failures.
- `--ephemeral` writes no session file, so these runs stay out of the user's history.
- `--json` prints events as JSON lines: `thread.started`, `turn.completed` with `usage`
(`input_tokens`, `cached_input_tokens`, `cache_write_input_tokens`, `output_tokens`,
`reasoning_output_tokens`), `turn.failed`, `error`, and `item.completed` items such as a
`command_execution` whose status is `declined`. The error shown is the `error` or `turn.failed`
message. Any `item.completed` counts as model work.
- Cost: Codex reports neither a cost nor its model. The script takes the model from `--model` or
the top-level `model` key in `$CODEX_HOME/config.toml` (`~/.codex/config.toml` by default), and
prices the tokens with `pricing.py`. Cached input is part of `input_tokens` and reasoning is part
of `output_tokens`, so the script subtracts the first and never adds the second again. With no
known model Codex never runs, even with `--allow-unpriced`: pin one with `--model codex=<id>`.
- Codex has no per-run spend or turn limit, so the run's timeout is its only limit.
- Codex skips the project's AGENTS.md in folders it does not trust. Whether `codex exec` trusts a
new temporary folder is not documented.
- Not used: `--dangerously-bypass-approvals-and-sandbox`, `--sandbox danger-full-access`, and the
deprecated `--full-auto`.
Sources: https://learn.chatgpt.com/docs/non-interactive-mode.md ,
https://github.com/openai/codex/blob/main/codex-rs/exec/src/exec_events.rs ,
https://github.com/openai/codex/blob/main/codex-rs/exec/src/cli.rs ,
https://learn.chatgpt.com/docs/agent-approvals-security.md ,
https://learn.chatgpt.com/docs/agent-configuration/agents-md.md
## Gemini CLI
```
gemini --output-format json --approval-mode auto_edit --skip-trust
--allowed-tools "run_shell_command(<test command>)" -p "<prompt>"
```
- `--approval-mode auto_edit` approves file edits. `--allowed-tools` approves shell commands that
start with the test command, one entry per part of a compound command; the user's own policy
rules still apply. In headless mode any other call that would ask the user is denied. The docs
mark `--allowed-tools` deprecated in favor of policy files, which cannot be passed on the command
line.
- `--skip-trust`: a new folder is untrusted, and Gemini CLI turns off automatic approval in
untrusted folders, so without it the agent could not edit. With it, the copy's own
`.gemini/settings.json` and MCP servers load too.
- `--output-format json` prints one object: `response`, `stats`, and `error` when the run failed.
Tokens come from `stats.models.<model>.tokens`: `prompt` includes `cached`, and `thoughts` is
not part of `candidates`, so the script counts input as `prompt` minus `cached` and output as
`candidates` plus `thoughts`. Turns are the sum of `api.totalRequests`; denials are
`stats.tools.totalDecisions.reject`. The error shown is `error.message`.
- Cost: Gemini CLI reports no cost, and `pricing.py` has no Gemini prices, because none were
confirmed on Google's official pricing page. Its runs need `--allow-unpriced gemini-cli` and
count $0 toward the cap. With a Google account's free tier that is the real cost; with a paid API
key it is not.
- The headless docs list no model flag, so `--model` is refused for Gemini CLI. Set the model in
its settings.
- Not used: `--approval-mode yolo` and `--yolo`.
Sources: https://github.com/google-gemini/gemini-cli/blob/main/docs/cli/headless.md ,
https://github.com/google-gemini/gemini-cli/blob/main/docs/cli/cli-reference.md ,
https://github.com/google-gemini/gemini-cli/blob/main/docs/cli/trusted-folders.md ,
https://github.com/google-gemini/gemini-cli/blob/main/docs/reference/configuration.md
## Errors that stop a harness
A run that did not time out, exited with an error, and changed nothing could not run. When its
message shows an error that will repeat on every task, the harness's other runs are skipped and the
fix is printed: a login problem ("authenticate", "OAuth", "not logged in", "invalid API key") gives
"sign in: run `<program>` once"; "requires a newer version" gives "upgrade `<harness>`"; a missing
program gives "install `<harness>`, or put `<program>` on PATH".
## Left out
- OpenCode: `opencode run --format json` exists, but its JSON event format is not documented, so
its output cannot be parsed with confidence. Source: https://opencode.ai/docs/cli
- Cursor CLI (`agent -p`): its JSON result documents no token or cost fields, so its spend cannot
be counted. Source: https://cursor.com/docs/cli/reference/output-format.md
## Prices
`scripts/pricing.py` holds the token prices, copied from the official pages on 2026-09-28:
https://platform.claude.com/docs/en/about-claude/pricing.md and
https://developers.openai.com/api/docs/pricing.md . Batch discounts, fast mode, and long-context
rates are not modeled.
# How the test drive works
The manual method is in
[How to test-drive a harness](https://github.com/RyanAlberts/best-of-Agent-Harnesses/blob/main/comparisons/how-to-test-drive-a-harness.md).
This skill automates its measurable half: the same tasks for every harness, taken from your own
history, scored by your own tests, with cost and time. Checked 2026-09-28.
## Mining tasks (`mine_tasks.py`)
A **task** is a past commit on the current branch that changed both source files and test files.
Tasks include bug fixes, features, and refactors: anything whose tests failed before the commit and
passed after it.
- Commits come from `git log HEAD` since `--since` (365 days by default). Merge commits are left
out. So are the first commit (no parent to start from), commits that change more than 30 files
or more than 1,000 lines ("too large"), commits with no source changes, and commits with no test
changes.
- A **test file** sits in a folder named `test`, `tests`, `__tests__`, `spec`, `specs`, `testdata`,
`__snapshots__`, or `__mocks__`, or has a test name: `test_x.py`, `x_test.py`, `conftest.py`,
`x_test.go`, `x.test.ts` and `x.spec.js` (any JavaScript or TypeScript extension), `x_spec.rb`,
`x_test.exs`, or `XTest.java` (also Kotlin, Scala, Swift, PHP, C#). Test-runner settings count as
test files too (`pytest.ini`, `tox.ini`, `noxfile.py`, `phpunit.xml`, `.mocharc`, `.nycrc`,
`.rspec`, `.rspec-local`, and the jest, vitest, karma, ava, and playwright config files), because
editing them can switch tests off. A folder named `testing` is not a test folder, because
libraries such as `numpy.testing` ship code there.
- A **doc** is Markdown, reStructuredText, text, or AsciiDoc, anything under `doc/` or `docs/`, and
files such as LICENSE and CHANGELOG. A **source file** has a code extension (Python, JavaScript,
TypeScript, Go, Rust, Java, Ruby, C, C#, Swift, shell, and more). Everything else, such as
configuration and data, is "other": it can be part of a change, but a commit needs a source change
to count.
- The **task text** is the commit subject and body. A last paragraph made only of trailers
(`Co-Authored-By:`, `Signed-off-by:`) is removed.
- The **hidden tests** are the commit's test-file changes; the agent never sees them. The **gold
patch** is the rest of the commit. "Original fix lines" counts the added and deleted lines in
its source and other files.
- The **test command** is `--test-cmd`, or the first match of: the `test` script in package.json
(run with pnpm, yarn, bun, or npm after the lockfile; npm's placeholder script is skipped), a
pytest setup (`pytest.ini`, `[tool.pytest` in pyproject.toml, `[tool:pytest]` in setup.cfg,
`[pytest]` in tox.ini, a conftest.py at the top or in `tests/` or `test/`, or `test_*.py` files
in `tests/`; `uv run pytest -q` when uv.lock exists), go.mod (`go test ./...`), Cargo.toml
(`cargo test`), or a Makefile `test` target. When nothing matches, the command CI runs is usually
in `.github/workflows` or the README.
## Checking tasks (`--validate`)
A task is kept only when its tests prove the change: they **fail before it and pass after it,
twice**.
1. In a fresh copy of HEAD, the test command must pass. If it fails, the script stops with exit 2
and ends the message with the last 15 lines of the output. The full output is in
`.harness-test-drive/logs/head-check.log` (and `head-setup.log` for the setup command). A
missing module or command means the clean copy needs `--setup-cmd`; a time-out means the suite
needs a longer `--test-timeout`.
2. For each candidate, newest first, up to three times `--max`: a fresh copy of the parent commit,
plus the hidden tests. The test command must fail.
3. The same copy plus the gold patch (now equal to the commit): the test command must pass.
4. Back to the parent's files, then the commit's again: the tests must fail and pass once more.
Rejected candidates are counted by reason:
- "tests pass before the fix": the tests would not notice the missing change, so every agent,
including one that does nothing, would score a pass.
- "tests fail with the fix": the commit does not pass its own tests here, often because of a
missing dependency, a flaky test, or a commit that was broken when made.
- "flaky": the second round gave a different result, so the tests cannot score an agent.
- "tests time out with the fix" (after `--test-timeout`, 600 seconds by default), "setup command
failed", and "could not copy the commit". A time-out before the fix counts as a failure, because
a hang can be the bug.
`--setup-cmd` runs in every fresh copy before the tests and before the agent, and again whenever
the files just written include a dependency manifest or lock file (package.json, a lockfile,
pyproject.toml, requirements*.txt, go.mod, Cargo.toml, Gemfile, and others). It must install inside
the copy, for example `npm ci`, `pnpm install --frozen-lockfile`, or
`python3 -m venv .venv && .venv/bin/pip install -e .` with the test command
`.venv/bin/python -m pytest`. A setup that installs into the user's own environment changes that
environment 30 times over. No test run reads Python bytecode from an earlier run: Python trusts a
cached file when the source has the same size and the same modified second, so a test file written
right after an earlier run could otherwise run as its old version. Before each run the copy's
`__pycache__` folders are deleted, and `PYTHONPYCACHEPREFIX` points the run at a new empty folder,
which also covers interpreters that keep bytecode outside the copy, such as Apple's
`/usr/bin/python3`.
## The fresh copy
Every check and every agent run gets a new temporary folder holding the files of one commit,
committed once as a brand-new git repository. It is not a git worktree, for two reasons:
- A worktree shares the repository's history, so `git log --all` inside it shows the commit that
made the change. In the fresh copy, later commits do not exist. (An agent that searched the disk
could still find the user's repository.)
- A worktree shares branches, the stash, hooks, and config with the user's repository, so an agent
could change them. The fresh copy shares nothing.
The user's repository is only read, with `git log`, `git ls-tree`, `git cat-file`, and
`git rev-parse`. The copy leaves out untracked files (such as `.env`), submodules, and Git LFS
content (only the pointer files are copied). Tests that need them fail the check, so those tasks are
dropped. A partial clone (`git clone --filter`) keeps old file contents on the server; the skill
fetches nothing, so it stops and asks for the full history first. Restored files are written only
inside the copy: a link the agent planted where a folder belongs is removed first. When counting the
agent's changes, the copy ignores common tool caches (`__pycache__`, `node_modules`, `.venv`,
`dist`, `build`, `target`, `*.egg-info`, and others).
Every command the scripts start (setup, tests, and each harness) gets this process's environment
minus the variables that lead back to the agent session that started it (`CLAUDECODE`, the
`CLAUDE_CODE_*` session, messaging, and bridge variables, `CLAUDE_PID`, `AI_AGENT`, and
`CODEX_SANDBOX*`), so an agent running repository code cannot reach the user's live session.
## Running (`drive.py run`)
Every harness gets the same prompt:
```
Make the change described below in this repository.
Change request:
<commit subject>
<commit body>
When you finish, the repository's own tests will check your work. Test files you edit are put back first, so change the code, not the tests. You can run the tests with this command: <test command>
Do not commit. No one will answer questions, so make reasonable choices and complete the change.
```
(The third paragraph is one line in the prompt; the body paragraph appears only when the commit
message has one.)
Runs go task by task: every chosen harness runs a task before the next task starts. Each run has
`--timeout` seconds (900 by default). When a run ends, times out, or is stopped, every process it
started is stopped too: its process group, and every group below it, found by walking the process
tree (Claude Code runs its shell commands in their own sessions). The flags per harness are in
`harness-commands.md`.
- **One run per results file.** `<results>.lock` holds the running script's process id; a second
`run` on the same file stops with an error while that process is alive.
- **Stops.** SIGINT (Ctrl-C), SIGTERM (a stopped background task), and SIGHUP (a closed terminal)
end the agent, record the run as interrupted with its spend, delete the copy, and exit 130. The
next `run` starts that run again.
- **SIGKILL.** Before each agent starts, `.harness-test-drive/inflight.json` records the task, the
harness, the copy, and a high cost estimate; it is deleted once the run is recorded. When the
next `run` finds it, it records the lost run as interrupted with that estimate, deletes the copy,
and runs the task again.
- **Logs.** For each run, `.harness-test-drive/logs/` keeps `<task>-<harness>.out` (the harness
output), `.err`, `.diff` (the agent's changes), and `.tests` (the scoring test run).
## Scoring
1. The agent's changes are counted against the starting commit (lines added plus deleted, and
files touched, new files and deletions included) and saved as the run's diff.
2. When the run did not time out, the harness exited with an error, and it changed nothing, the run
**could not run**: it is not scored and stays out of pass rates and cost per pass. Its error is
the harness's own message. An error that will repeat on every task (a login, an old version, a
missing program) skips that harness's remaining runs and prints the fix, such as "sign in: run
claude once" or "upgrade Codex". A run that could not run and cost nothing is tried again by the
next `run`.
3. Otherwise the copy is put back to the starting commit, every untracked and ignored file is
deleted, the saved diff minus test files is applied, the change's own tests are written, and the
setup runs again. What a person reviews in the diff is then exactly what was scored: an edit
inside `node_modules` or another ignored folder does not count.
4. The test command runs. Exit 0 means **passed**. The whole suite must pass, so a change that
breaks another test fails.
The scoreboard compares harnesses only on the tasks that every one of them finished, so a stop at
the cap or an error on one side does not skew the numbers; a harness with no finished run is listed
as could not run. Harnesses appear in rank order: most passes, then the lowest cost per pass. Cost
per pass is all the spend on those tasks, failed runs included, divided by the passes: the guide's
dollars per accepted task.
## Fairness rules from the guide
| Rule in the guide | What the skill does |
|---|---|
| Hold the model constant where you can | Not possible across these three providers, so a result compares whole stacks (harness plus model). `--model` pins a model per harness for repeatable runs; every result records the model. |
| Port your instructions once | Each copy holds the repository's own instruction files as committed at that point. Each harness still loads the user's personal instructions and hooks from the home folder (and, except Claude Code, its MCP servers); keep those comparable, or write the difference down. |
| Pin versions and write them down | Each result records the harness's `--version`. |
| Fresh git worktree per candidate per task | A fresh copy per harness per task (safer than a worktree, as above). |
| Define "pass" before running anything | Pass means the commit's own tests pass, fixed when the task was mined. |
| Never let the agent grade itself | The tests are hidden, and only the agent's non-test changes are scored. |
Every harness also gets the same prompt, the same timeout, and the same test command.
## Money
- `--max-usd` is required. It covers every run in `results.jsonl`, including runs from earlier
commands, so a rerun resumes without resetting the count.
- The cap limits counted spend. Claude Code stops itself at the budget left. A Codex run has no
limit except `--timeout`, and a run cut off there is counted as an estimate, so the bill can pass
the cap by more than one run. Runs allowed with `--allow-unpriced` count $0.
- Before each run the script compares the counted spend with the cap and starts no run once the cap
is reached.
- What counts: Claude Code's reported `total_cost_usd` (an estimate at API prices; on a
subscription plan nothing is billed per token). Codex tokens times the price of its model, which
comes from `--model` or the top-level `model` in Codex's config; with neither, Codex does not run.
Gemini CLI reports no cost and runs only with `--allow-unpriced gemini-cli`.
- When a priced run's cost cannot be read: a run cut off at the time limit counts a high estimate
for every 900 seconds it ran; a run whose output shows model work (for Codex any completed item,
for Claude Code any reply with usage) counts one high estimate; a run that never reached a model
counts $0. A Claude Code run stopped before its result counts the replies it sent.
- The estimate assumes a short run of 194,000 tokens (20,000 input, 150,000 cached reads, 20,000
cache writes, 4,000 output) on the cheaper model, up to a long run of 3,350,000 tokens (150,000
input, 3,000,000 cached reads, 150,000 cache writes, 50,000 output) on the pricier one: Claude
Sonnet 5.5 to Claude Opus 5.5 for Claude Code, and the configured model for Codex. These are
assumptions, not measurements.
## Limits
- **Sample size.** Five or ten tasks give a rough answer. A difference of one or two passes can be
noise.
- **Tests check behavior, not quality.** A passing change can still be one you would reject in
review. The saved diffs are there for the human half of the guide: a second person accepts or
rejects each one.
- **Training data.** For a public repository, a model may have seen the real change during
training.
- **Flaky tests** can still flip a result. The check runs each task twice, but a test that fails
only now and then can slip through.
- **Skew.** Tasks come from commits that changed tests, so well-tested parts of the code are
overrepresented. A terse commit message makes a hard prompt, the same for every harness.
- **Mixed settings files.** Test-runner settings files are put back, but an agent could still change
how tests run through a file that holds other things too, such as `package.json`,
`pyproject.toml`, or a Makefile. The saved diffs show it.
- **Untrusted repositories.** The check runs the repository's test and setup commands at up to 31
commits; each harness loads the repository's own agent settings, edits files, runs the test
command, and reads commit messages as prompts. For code you did not write, run the skill inside a
container or VM.
- macOS and Linux only: stopping a run relies on POSIX process groups and `ps`.
"""Helpers shared by mine_tasks.py and drive.py.
- safe_text() and code(): from safe.py, the shared helpers that make untrusted
text (commit messages, file names, harness output) safe to show in a
markdown report.
- classify(): sorts a repository path into test, doc, source, or other.
- make_workspace(): a fresh folder holding one commit's files as a new git
repository with a single commit, so an agent working there cannot see later
commits (the real fix) and cannot touch the real repository's branches.
- write_files(), run_command(), remove_workspace().
Only read-only git commands touch the user's repository (ls-tree, cat-file,
log, rev-parse). Python 3.9+, standard library only.
"""
from __future__ import annotations
import os
import re
import shutil
import signal
import subprocess
import tempfile
import time
from safe import code, safe_text # noqa: F401 (also kept for callers that import them from here)
# --- File classes ------------------------------------------------------------------
_TEST_DIRS = {"test", "tests", "__tests__", "spec", "specs", "testdata", "__snapshots__", "__mocks__"}
_TEST_NAME_RE = re.compile(
r"(^test_.*\.py$|_test\.py$|^conftest\.py$|_test\.go$|_spec\.rb$|_test\.exs$"
r"|\.(test|spec)\.[cm]?[jt]sx?$"
r"|(Test|Tests)\.(java|kt|scala|swift|php|cs)$"
# test-runner settings: an agent that edits them could switch tests off
r"|^(pytest\.ini|tox\.ini|noxfile\.py|phpunit\.xml(\.dist)?|\.mocharc(\.\w+)?|\.nycrc(\.\w+)?|\.rspec(-local)?)$"
r"|^(jest|vitest|karma)\.conf(ig)?\.[cm]?[jt]s(on)?$|^(ava|playwright)\.config\.[cm]?[jt]s$)")
_DOC_DIRS = {"doc", "docs"}
_DOC_EXT = {".md", ".markdown", ".rst", ".txt", ".adoc"}
_DOC_NAME_RE = re.compile(r"^(LICENSE|LICENCE|COPYING|CHANGELOG|CHANGES|AUTHORS|NOTICE|CONTRIBUTORS)(\..*)?$", re.I)
_SOURCE_EXT = {
".py", ".pyi", ".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx", ".mts", ".cts", ".go", ".rs",
".java", ".kt", ".kts", ".scala", ".groovy", ".rb", ".php", ".c", ".h", ".cc", ".cpp", ".cxx",
".hpp", ".hh", ".cs", ".fs", ".swift", ".m", ".mm", ".dart", ".ex", ".exs", ".erl", ".hs",
".ml", ".clj", ".lua", ".pl", ".pm", ".r", ".jl", ".sh", ".bash", ".zsh", ".sql", ".vue",
".svelte", ".zig", ".nim", ".ps1",
}
_MANIFEST_RE = re.compile(
r"^(package(-lock)?\.json|npm-shrinkwrap\.json|pnpm-lock\.yaml|yarn\.lock|bun\.lockb?|pyproject\.toml"
r"|setup\.(py|cfg)|requirements[\w.-]*\.txt|Pipfile(\.lock)?|poetry\.lock|uv\.lock|go\.(mod|sum)"
r"|Cargo\.(toml|lock)|Gemfile(\.lock)?|composer\.(json|lock))$")
def touches_manifest(paths) -> bool:
"""True when a path is a dependency manifest or lock file, so the setup
command must run again after it changes."""
return any(_MANIFEST_RE.match(p.rsplit("/", 1)[-1]) for p in paths)
def classify(path) -> str:
"""'test', 'doc', 'source', or 'other' for a repository path (forward slashes)."""
parts = path.split("/")
name = parts[-1]
if _TEST_NAME_RE.search(name) or any(p in _TEST_DIRS for p in parts[:-1]):
return "test"
ext = os.path.splitext(name)[1].lower()
if any(p.lower() in _DOC_DIRS for p in parts[:-1]) or ext in _DOC_EXT or _DOC_NAME_RE.match(name):
return "doc"
return "source" if ext in _SOURCE_EXT else "other"
# --- Git and workspaces --------------------------------------------------------------
# Files that tools create while an agent works or tests run; the diff ignores them.
_EXCLUDE = ["__pycache__/", "*.pyc", ".pytest_cache/", ".mypy_cache/", ".ruff_cache/", ".tox/",
".nox/", ".venv/", "venv/", "node_modules/", ".coverage", "coverage/", ".next/",
".turbo/", "target/", "dist/", "build/", "*.egg-info/", ".eggs/"]
def _read_env() -> dict:
# No lock files, no password prompts, paths taken literally, and in a
# partial clone no fetching of missing objects over the network.
return dict(os.environ, GIT_OPTIONAL_LOCKS="0", GIT_TERMINAL_PROMPT="0", GIT_LITERAL_PATHSPECS="1",
GIT_NO_LAZY_FETCH="1")
def git(repo, *args) -> bytes:
"""Run a read-only git command in the user's repository; return stdout."""
res = subprocess.run(["git", "-C", repo] + list(args), env=_read_env(), stdin=subprocess.DEVNULL,
capture_output=True)
if res.returncode != 0:
raise RuntimeError("git %s failed: %s" % (args[0], code(res.stderr.decode("utf-8", "replace"))))
return res.stdout
def _workspace_git(ws, *args, input=None) -> bytes:
"""git inside a workspace, ignoring the user's global config (hooks, signing)."""
env = dict(os.environ, GIT_CONFIG_GLOBAL=os.devnull, GIT_CONFIG_NOSYSTEM="1", GIT_TERMINAL_PROMPT="0",
GIT_LITERAL_PATHSPECS="1")
feed = {"input": input} if input is not None else {"stdin": subprocess.DEVNULL}
res = subprocess.run(["git", "-c", "user.name=harness-test-drive", "-c", "user.email=harness-test-drive@localhost"]
+ list(args), cwd=ws, env=env, capture_output=True, **feed)
if res.returncode != 0:
raise RuntimeError("git %s failed in the workspace: %s"
% (args[0], code(res.stderr.decode("utf-8", "replace"))))
return res.stdout
def _tree(repo, commit, paths=None) -> dict:
"""{path: (mode, type, object id)} for a commit, limited to `paths` when given."""
args = ["ls-tree", "-r", "-z", "--full-tree", commit]
if paths is not None:
args += ["--"] + list(paths)
entries = {}
for rec in git(repo, *args).split(b"\0"):
if rec:
meta, _, path = rec.partition(b"\t")
mode, kind, oid = meta.decode().split()
entries[os.fsdecode(path)] = (mode, kind, oid)
return entries
def _inside(path) -> bool:
"""True for a relative path that stays inside the workspace and out of .git."""
parts = path.split("/")
return not path.startswith("/") and ".." not in parts and ".git" not in (p.lower() for p in parts)
def _remove(target) -> None:
if os.path.islink(target) or os.path.isfile(target):
os.unlink(target)
elif os.path.isdir(target):
shutil.rmtree(target)
def _clear_path(dest, path) -> str:
"""Remove whatever stands at `path` inside `dest`, and any link or file where
one of its parent folders belongs, so a write never follows a link an agent
planted out of the workspace. Returns the full path."""
current = dest
for part in path.split("/")[:-1]:
current = os.path.join(current, part)
if os.path.islink(current) or (os.path.lexists(current) and not os.path.isdir(current)):
os.unlink(current)
target = os.path.join(dest, path)
_remove(target)
return target
def _write_entries(repo, entries, dest) -> None:
blobs = [(p, mode, oid) for p, (mode, kind, oid) in sorted(entries.items()) if kind == "blob" and _inside(p)]
if not blobs:
return
proc = subprocess.Popen(["git", "-C", repo, "cat-file", "--batch"], stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=_read_env())
try:
for path, mode, oid in blobs:
proc.stdin.write(oid.encode() + b"\n")
proc.stdin.flush()
header = proc.stdout.readline().split()
if len(header) != 3: # "<id> missing": a partial or damaged clone
raise RuntimeError("git object %s is not in this clone" % oid)
size = int(header[2])
data = proc.stdout.read(size)
proc.stdout.read(1) # the newline after each object
target = _clear_path(dest, path)
os.makedirs(os.path.dirname(target), exist_ok=True)
if mode == "120000":
os.symlink(os.fsdecode(data), target)
else:
with open(target, "wb") as fh:
fh.write(data)
if mode == "100755":
os.chmod(target, 0o755)
finally:
proc.stdin.close()
proc.stdout.close()
proc.wait()
def write_files(repo, commit, paths, dest) -> None:
"""Set each path in `dest` to its state at `commit`: written when the commit
has it, removed when it does not."""
paths = [p for p in paths if _inside(p)]
entries = _tree(repo, commit, paths)
for path in paths:
if path not in entries:
_clear_path(dest, path)
_write_entries(repo, entries, dest)
def make_workspace(repo, commit, root=None) -> str:
"""A new temporary folder with the files of `commit`, committed once as a
fresh git repository. Later commits never reach it."""
ws = tempfile.mkdtemp(prefix="harness-test-drive-", dir=root)
try:
_write_entries(repo, _tree(repo, commit), ws)
_workspace_git(ws, "init", "-q")
with open(os.path.join(ws, ".git", "info", "exclude"), "a") as fh:
fh.write("\n".join(_EXCLUDE) + "\n")
_workspace_git(ws, "add", "-A", "-f")
_workspace_git(ws, "commit", "-q", "--no-verify", "--allow-empty", "-m", "Starting point for the task")
except BaseException:
remove_workspace(ws)
raise
return ws
def workspace_head(ws) -> str:
return _workspace_git(ws, "rev-parse", "HEAD").decode().strip()
def changed_files(ws, base, patch_path=None) -> tuple:
"""(lines added plus deleted, paths) for everything in the workspace that
differs from commit `base`: edits, new files, deletions, and commits the
agent made. Tool caches listed in _EXCLUDE are left out. With `patch_path`,
the full diff is saved there for review."""
_workspace_git(ws, "add", "-A")
if patch_path:
with open(patch_path, "wb") as fh:
fh.write(_workspace_git(ws, "diff", "--cached", "--binary", "--no-renames", base))
lines, paths = 0, []
for rec in _workspace_git(ws, "diff", "--cached", "--numstat", "-z", "--no-renames", base).split(b"\0"):
if rec:
added, deleted, path = rec.split(b"\t", 2)
lines += sum(int(n) for n in (added, deleted) if n.isdigit()) # "-" for binary files
paths.append(os.fsdecode(path))
return lines, paths
def score_patch(ws, base, paths) -> bytes:
"""The staged diff from `base` of `paths` only; changed_files() staged it."""
return _workspace_git(ws, "diff", "--cached", "--binary", "--no-renames", base, "--", *paths) if paths else b""
def reset_workspace(ws, base) -> None:
"""Put the copy back to commit `base`, deleting every untracked and ignored
file, such as an edit inside node_modules."""
_workspace_git(ws, "reset", "-q", "--hard", base)
_workspace_git(ws, "clean", "-q", "-ffdx")
def apply_patch(ws, patch) -> None:
if patch:
_workspace_git(ws, "apply", "--binary", "--whitespace=nowarn", "-", input=patch)
def remove_workspace(path) -> None:
shutil.rmtree(path, ignore_errors=True)
if os.path.exists(path): # an agent made a folder read-only
for dirpath, _, _ in os.walk(path):
os.chmod(dirpath, 0o700)
shutil.rmtree(path, ignore_errors=True)
# --- Running commands ------------------------------------------------------------------
# Handles to the session that started this script. A harness, or repository code
# it runs, could use them to reach the user's live agent session.
_SESSION_VARS = {
"CLAUDECODE", "AI_AGENT", "CLAUDE_PID", "CLAUDE_EFFORT", "CLAUDE_CODE_SESSION_ID", "CLAUDE_CODE_CHILD_SESSION",
"CLAUDE_CODE_BRIDGE_SESSION_ID", "CLAUDE_CODE_ENTRYPOINT", "CLAUDE_CODE_HOST_SESSION_ID", "CLAUDE_CODE_EXECPATH",
"CLAUDE_CODE_MESSAGING_SOCKET", "CLAUDE_CODE_MESSAGING_TOKEN", "CLAUDE_AGENT_SDK_VERSION",
"CLAUDE_CODE_SDK_HAS_HOST_AUTH_REFRESH", "CLAUDE_CODE_SESSION_ATTENDED"}
def clean_env() -> dict:
"""This process's environment without the calling session's handles."""
return {k: os.environ[k] for k in os.environ if k not in _SESSION_VARS and not k.startswith("CODEX_SANDBOX")}
def _raise_interrupt(signum, frame):
raise KeyboardInterrupt
def install_stop_handlers() -> dict:
"""Make SIGINT, SIGTERM, and SIGHUP (a stopped background task, a closed
terminal) raise KeyboardInterrupt, so every stop takes the path that ends
the agent, records the run, and deletes the copy. Returns the old handlers."""
old = {}
for sig in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
old[sig] = signal.signal(sig, _raise_interrupt)
return old
def restore_handlers(old) -> None:
for sig, handler in old.items():
signal.signal(sig, handler)
def _groups_under(pid) -> set:
"""Process groups of `pid` and every process below it. A child can leave
the group (Claude Code runs its shells in their own sessions), so killpg on
the first group is not enough."""
try:
listing = subprocess.run(["ps", "-A", "-o", "pid=,ppid=,pgid="], capture_output=True, text=True,
timeout=10, stdin=subprocess.DEVNULL).stdout
except (OSError, subprocess.TimeoutExpired):
return {pid}
children, group = {}, {}
for line in listing.splitlines():
parts = line.split()
if len(parts) == 3 and all(p.isdigit() for p in parts):
child, parent, pgid = (int(p) for p in parts)
children.setdefault(parent, []).append(child)
group[child] = pgid
found, stack, seen = {pid}, [pid], set()
while stack:
current = stack.pop()
if current not in seen:
seen.add(current)
found.add(group.get(current, pid))
stack.extend(children.get(current, []))
return {g for g in found if g > 1 and g != os.getpgrp()}
def _signal_groups(groups, sig) -> None:
for g in groups:
try:
os.killpg(g, sig)
except (ProcessLookupError, PermissionError):
pass
def _stop_group(proc) -> None:
"""End every process the command started: its process group, every group
below it, and background children left behind after a normal exit."""
groups = _groups_under(proc.pid) if proc.poll() is None else {proc.pid}
_signal_groups(groups, signal.SIGTERM)
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
pass
_signal_groups(groups, signal.SIGKILL) # whatever ignored SIGTERM
proc.wait()
def run_command(cmd, cwd, timeout, out_path=None, err_path=None, env=None) -> dict:
"""Run a shell command (a string) or a program (a list) in `cwd` with no
input and, unless `env` is given, clean_env(). Output goes to `out_path` and
`err_path` when given, else nowhere. On timeout every process it started is
stopped. Returns exit, seconds, timed_out."""
out = open(out_path, "wb") if out_path else subprocess.DEVNULL
err = open(err_path, "wb") if err_path else (subprocess.STDOUT if out_path else subprocess.DEVNULL)
started = time.monotonic()
timed_out = False
try:
proc = subprocess.Popen(cmd, shell=isinstance(cmd, str), cwd=cwd, stdin=subprocess.DEVNULL,
stdout=out, stderr=err, start_new_session=True,
env=clean_env() if env is None else env)
try:
proc.wait(timeout=timeout)
except subprocess.TimeoutExpired:
timed_out = True
except BaseException: # Ctrl-C: never leave an agent running on its own
_stop_group(proc)
raise
_stop_group(proc)
finally:
for fh in (out, err):
if hasattr(fh, "close"):
fh.close()
return {"exit": None if timed_out else proc.returncode,
"seconds": round(time.monotonic() - started, 2), "timed_out": timed_out}
def duration(seconds) -> str:
"""'45 s', '3 min 5 s', '2 h 4 min'."""
seconds = int(round(seconds or 0))
if seconds < 60:
return "%d s" % seconds
if seconds < 3600:
return "%d min %d s" % (seconds // 60, seconds % 60)
return "%d h %d min" % (seconds // 3600, seconds % 3600 // 60)
def run_tests(cmd, cwd, timeout, out_path=None) -> dict:
"""run_command for the test command with no Python bytecode from earlier
runs: Python trusts a cached file when the source has the same size and
modified second, so a restored test or gold file written right after an
earlier run could otherwise run as its old version. The workspace's
__pycache__ folders are deleted, and PYTHONPYCACHEPREFIX points Python 3.8
and later at a new empty folder, also on interpreters that keep bytecode
elsewhere (Apple's /usr/bin/python3 uses ~/Library/Caches). The folder is
deleted after the run."""
for dirpath, dirnames, _ in os.walk(cwd):
dirnames[:] = [d for d in dirnames if d not in (".git", "node_modules", ".venv", "venv")]
if "__pycache__" in dirnames:
dirnames.remove("__pycache__")
shutil.rmtree(os.path.join(dirpath, "__pycache__"), ignore_errors=True)
prefix = tempfile.mkdtemp(prefix="harness-test-drive-pycache-")
try:
return run_command(cmd, cwd, timeout, out_path, env=dict(clean_env(), PYTHONPYCACHEPREFIX=prefix))
finally:
shutil.rmtree(prefix, ignore_errors=True)
"""Run coding agents on mined tasks, score them with the repository's tests,
and report the scoreboard.
python3 drive.py estimate --tasks tasks.json [--harness claude-code,codex,gemini-cli]
python3 drive.py run --tasks tasks.json --harness claude-code,codex --max-usd 5
[--timeout 900] [--keep] [--allow-unpriced gemini-cli] [--model claude-code=<id>]
python3 drive.py report [--results results.jsonl] [--json] [--out <path>]
Each run gets a fresh copy of the repository at the commit before the fix (a new
git repository with one commit, so the fix is not in its history), the commit
message as the prompt, and the harness's most restrictive settings that still
allow edits and the test command. Afterwards the agent's diff, minus test
files, is applied to a clean copy with the fix commit's tests, and the test
command decides: passed or failed. A harness that exits with an error and
changes nothing "could not run"; that run is not scored.
Money: `run` needs --max-usd. It stops starting runs once the counted spend
reaches the cap, passes the budget left to harnesses that take a per-run cap
(Claude Code --max-budget-usd), and appends every run to results.jsonl, so a
rerun resumes and counts earlier spend. SIGINT, SIGTERM, and SIGHUP stop the
agent, record the run with its spend, and delete the copy. After a SIGKILL,
inflight.json lets the next run count the lost run. Exit codes: 0 success,
2 usage or input error, 130 stopped. Python 3.9+, standard library only.
"""
from __future__ import annotations
import argparse
import contextlib
import datetime
import json
import math
import os
import sys
import harnesses
from common import (apply_patch, changed_files, classify, duration, install_stop_handlers, make_workspace,
remove_workspace, reset_workspace, restore_handlers, run_command, run_tests, score_patch,
workspace_head, write_files)
from harnesses import HIGH_RUN, LOW_RUN, fatal_hint
from pricing import PRICES_CHECKED
from safe import code, safe_text
DEFAULT_TASKS = os.path.join(".harness-test-drive", "tasks.json")
DEFAULT_RESULTS = os.path.join(".harness-test-drive", "results.jsonl")
SCORED = ("passed", "failed")
PROMPT = """Make the change described below in this repository.
Change request:
{task}
When you finish, the repository's own tests will check your work. Test files you edit are put back first, so \
change the code, not the tests. You can run the tests with this command: {test_cmd}
Do not commit. No one will answer questions, so make reasonable choices and complete the change.
"""
class UsageError(Exception):
"""Bad input: the message says what to change."""
def _now() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def _stderr(line) -> None:
print(line, file=sys.stderr, flush=True)
def build_prompt(task, test_cmd) -> str:
text = task["subject"] + ("\n\n" + task["body"] if task.get("body") else "")
return PROMPT.format(task=text, test_cmd=test_cmd)
def load_results(path) -> list:
"""Every run recorded in results.jsonl; a missing file is an empty list."""
records = []
try:
with open(path, encoding="utf-8") as fh:
for line in fh:
try:
rec = json.loads(line)
except ValueError:
continue # a line cut short when a run was interrupted
if isinstance(rec, dict) and rec.get("task") and rec.get("harness"):
records.append(rec)
except OSError:
pass
return records
def _append(path, rec) -> None:
with open(path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(rec) + "\n")
def _read(path) -> str:
try:
with open(path, encoding="utf-8", errors="replace") as fh:
return fh.read()
except OSError:
return ""
def _record(task, harness, label, version="") -> dict:
return {"schema": "harness-test-drive/result/1", "task": task.get("id", ""), "commit": task.get("commit", ""),
"subject": safe_text(task.get("subject") or "", 100), "gold_lines": task.get("gold_lines"),
"harness": harness, "label": label, "version": version, "started": _now(),
"status": "error", "timed_out": False, "exit_code": None, "error": "", "hint": "",
"agent_seconds": None, "test_seconds": None, "cost_usd": None, "cost_source": None, "charged_usd": 0.0,
"tokens": None, "model": "", "turns": None, "denials": 0, "test_runs": None, "test_failures": None,
"lines_changed": None, "files_touched": None, "log": "", "diff": ""}
def _charge(rec, adapter, out_path, err_path, test_cmd) -> None:
"""Read the harness output into `rec`, and set what the spend cap counts:
the measured cost; else, for a priced harness, a high estimate when the run
was cut off at the time limit (one per 900 seconds) or its output shows model
work; else nothing (an unpriced harness, or one that never reached a model)."""
stdout, stderr = _read(out_path), _read(err_path)
try:
parsed = adapter.parse_output(stdout, test_cmd=test_cmd)
except Exception: # unexpected output must not lose the run or its spend
parsed = {"cost_usd": None, "cost_source": None, "tokens": None, "model": "", "turns": None,
"error": "output could not be read", "denials": 0, "worked": bool(stdout.strip()),
"test_runs": None, "test_failures": None}
lines = [ln for ln in stderr.splitlines() if ln.strip()]
rec.update(cost_usd=parsed["cost_usd"], tokens=parsed["tokens"], model=safe_text(parsed["model"], 60),
turns=parsed["turns"], denials=parsed["denials"], test_runs=parsed.get("test_runs"),
test_failures=parsed.get("test_failures"),
error=safe_text(rec["error"] or parsed["error"] or (lines[-1] if rec["exit_code"] and lines else "")))
if parsed["cost_usd"] is not None:
rec.update(cost_source=parsed["cost_source"], charged_usd=parsed["cost_usd"])
elif not adapter.priced():
rec.update(cost_source="unpriced", charged_usd=0.0)
elif rec["timed_out"]:
rec.update(cost_source="unmeasured",
charged_usd=adapter.fallback_cost() * max(1.0, (rec["agent_seconds"] or 0) / 900.0))
elif parsed.get("worked"):
rec.update(cost_source="unmeasured", charged_usd=adapter.fallback_cost())
else:
rec.update(cost_source=None, charged_usd=0.0)
def run_one(doc, task, adapter, budget_left, timeout, keep, work_root, logs_dir, version, inflight) -> dict:
"""One agent on one task: copy, agent, diff, then the score in a clean copy.
Ctrl-C (or SIGTERM, SIGHUP) returns the record with status "interrupted"."""
rec = _record(task, adapter.name, adapter.label, version)
name = "%s-%s" % (task["id"], adapter.name)
out_path, err_path = os.path.join(logs_dir, name + ".out"), os.path.join(logs_dir, name + ".err")
setup_cmd, test_timeout = doc.get("setup_cmd"), doc.get("test_timeout", 600)
ws, started, charged = None, False, False
try:
ws = make_workspace(doc["repo"], task["parent"], root=work_root)
base = workspace_head(ws)
if setup_cmd and run_command(setup_cmd, ws, test_timeout)["exit"] != 0:
rec["error"] = "the setup command failed"
return rec
argv = adapter.build_command(build_prompt(task, doc["test_cmd"]), ws,
{"test_cmd": doc["test_cmd"], "max_usd": budget_left})
with open(inflight, "w", encoding="utf-8") as fh: # what a SIGKILL would lose
json.dump({"task": task["id"], "harness": adapter.name, "label": adapter.label, "started": rec["started"],
"charged_usd": adapter.fallback_cost() if adapter.priced() else 0.0, "workspace": ws}, fh)
rec["log"], started = out_path, True
agent = run_command(argv, ws, timeout, out_path, err_path)
rec.update(timed_out=agent["timed_out"], exit_code=agent["exit"], agent_seconds=agent["seconds"])
_charge(rec, adapter, out_path, err_path, doc["test_cmd"])
charged = True
rec["diff"] = os.path.join(logs_dir, name + ".diff")
rec["lines_changed"], touched = changed_files(ws, base, rec["diff"])
rec["files_touched"] = len(touched)
if not rec["timed_out"] and rec["exit_code"] != 0 and not touched: # could not run: nothing to score
rec["error"] = rec["error"] or "exited with code %s and changed nothing" % rec["exit_code"]
rec["hint"] = fatal_hint(adapter.label, getattr(adapter, "binary", adapter.name), rec["error"])
return rec
# Score exactly the saved diff, minus test files, in a clean copy with the fix commit's tests.
patch = score_patch(ws, base, [p for p in touched if classify(p) != "test"])
reset_workspace(ws, base)
apply_patch(ws, patch)
write_files(doc["repo"], task["commit"], task["hidden_tests"], ws)
if setup_cmd and run_command(setup_cmd, ws, test_timeout)["exit"] != 0:
rec.update(status="failed", error="the setup command failed after the agent's changes")
return rec
tests = run_tests(doc["test_cmd"], ws, test_timeout, os.path.join(logs_dir, name + ".tests"))
rec.update(status="passed" if tests["exit"] == 0 else "failed", test_seconds=tests["seconds"])
except KeyboardInterrupt:
rec.update(status="interrupted", error="stopped by the user")
except (OSError, RuntimeError, ValueError) as err:
rec["error"] = safe_text(err)
rec["hint"] = fatal_hint(adapter.label, getattr(adapter, "binary", adapter.name), rec["error"])
finally:
try:
if started and not charged:
_charge(rec, adapter, out_path, err_path, doc["test_cmd"])
finally:
if ws and keep:
rec["workspace"] = ws
elif ws:
remove_workspace(ws)
return rec
def _alive(pid) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
@contextlib.contextmanager
def _lock(results_path):
"""One run per results file: <results>.lock holds the pid of the run."""
path = results_path + ".lock"
for _ in range(2):
try:
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
except FileExistsError:
try:
pid = int(_read(path).strip() or 0)
except ValueError:
pid = 0
if pid > 0 and _alive(pid):
raise UsageError("Another drive.py run (process %d) is using %s. Let it finish, or stop it first."
% (pid, code(results_path, 300)))
with contextlib.suppress(FileNotFoundError):
os.unlink(path) # left by a run that is gone
continue
with os.fdopen(fd, "w") as fh:
fh.write(str(os.getpid()))
break
else:
raise UsageError("Could not lock %s." % code(path, 300))
try:
yield
finally:
with contextlib.suppress(OSError):
os.unlink(path)
def _recover(inflight, results_path, doc) -> None:
"""Turn a run lost to SIGKILL (inflight.json left behind) into an
'interrupted' record that counts its estimated spend, and delete its copy."""
try:
with open(inflight, encoding="utf-8") as fh:
left = json.load(fh)
except FileNotFoundError:
return
except (OSError, ValueError):
left = None
if isinstance(left, dict) and left.get("task") and left.get("harness"):
recorded = any(r["task"] == left["task"] and r["harness"] == left["harness"]
and r.get("started") == left.get("started") for r in load_results(results_path))
if not recorded:
task = next((t for t in doc.get("tasks", []) if t.get("id") == left["task"]), {"id": left["task"]})
rec = _record(task, str(left["harness"]), safe_text(left.get("label") or left["harness"], 40))
try:
charge = max(0.0, float(left.get("charged_usd") or 0.0))
except (TypeError, ValueError):
charge = 0.0
rec.update(status="interrupted", started=safe_text(left.get("started") or "", 40) or rec["started"],
error="the run was killed before it finished; its cost is an estimate",
cost_source="unmeasured", charged_usd=charge)
_append(results_path, rec)
ws = left.get("workspace")
if (isinstance(ws, str) and os.path.basename(ws).startswith("harness-test-drive-")
and os.path.isdir(os.path.join(ws, ".git"))):
remove_workspace(ws)
with contextlib.suppress(FileNotFoundError):
os.unlink(inflight)
def _done(rec) -> bool:
"""A recorded run that needs no retry. An interrupted run, or an error that
cost nothing (a login to renew, a program to install), runs again."""
return rec.get("status") != "interrupted" and not (rec.get("status") == "error" and not rec.get("charged_usd"))
def run(doc, results_path, adapters, max_usd, timeout=900, keep=False, allow_unpriced=(), work_root=None,
progress=_stderr) -> dict:
"""Run every (task, harness) pair not yet done in results_path, until the spend cap."""
tasks = [t for t in doc.get("tasks", []) if t.get("validated")]
if not tasks:
raise UsageError("The tasks file has no validated tasks. Run mine_tasks.py with --validate first.")
if not (isinstance(max_usd, (int, float)) and math.isfinite(max_usd) and max_usd > 0):
raise UsageError("--max-usd must be a dollar amount above zero.")
if timeout < 1:
raise UsageError("--timeout must be at least 1 second.")
if not os.path.isdir(doc.get("repo") or ""):
raise UsageError("The repository in the tasks file is gone: %s" % code(doc.get("repo") or "", 300))
installed = [a for a in adapters if a.available()]
allowed = set(allow_unpriced or ())
for a in installed:
if not a.priced() and not (a.can_run_unpriced and a.name in allowed):
raise UsageError("%s cannot be priced: %s. %s." % (a.label, a.unpriced_reason(), a.unpriced_fix()))
summary = {"results_file": safe_text(results_path, 300), "max_usd": max_usd, "runs": 0, "already_done": 0,
"stopped_at_cap": False, "not_started": 0, "spent_usd": 0.0, "skipped": {},
"not_installed": [a.name for a in adapters if not a.available()]}
folder = os.path.dirname(os.path.abspath(results_path))
os.makedirs(folder, exist_ok=True)
inflight = os.path.join(folder, "inflight.json")
with _lock(results_path):
_recover(inflight, results_path, doc)
earlier = load_results(results_path)
done = {(r["task"], r["harness"]) for r in earlier if _done(r)}
spent = sum(r.get("charged_usd") or 0.0 for r in earlier) # every run costs, retried ones too
pending = [(t, a) for t in tasks for a in installed if (t["id"], a.name) not in done]
summary["already_done"] = len(tasks) * len(installed) - len(pending)
logs_dir = os.path.join(folder, "logs")
os.makedirs(logs_dir, exist_ok=True)
versions = {a.name: a.version() for a in installed} if pending else {}
for i, (task, adapter) in enumerate(pending):
if adapter.name in summary["skipped"]:
continue
if spent >= max_usd:
summary.update(stopped_at_cap=True, not_started=sum(
1 for _, a in pending[i:] if a.name not in summary["skipped"]))
break
progress("[%d/%d] %s on %s: %s" % (i + 1, len(pending), adapter.label, task["id"][:7],
code(task["subject"], 60)))
rec = run_one(doc, task, adapter, max_usd - spent, timeout, keep, work_root, logs_dir,
versions[adapter.name], inflight)
_append(results_path, rec)
with contextlib.suppress(FileNotFoundError):
os.unlink(inflight)
spent += rec["charged_usd"]
summary["runs"] += 1
if rec["status"] == "interrupted":
raise KeyboardInterrupt
if rec["status"] == "error":
progress(" could not run (%s), $%.2f counted" % (code(rec["error"], 120), rec["charged_usd"]))
if rec["hint"]:
summary["skipped"][adapter.name] = rec["hint"]
progress(" Fix: %s. The other %s runs are skipped." % (rec["hint"], adapter.label))
continue
progress(" %s in %s, $%.2f counted, $%.2f of $%.2f spent" % (
rec["status"], duration(rec["agent_seconds"]), rec["charged_usd"], spent, max_usd))
summary["spent_usd"] = round(spent, 4)
return summary
# --- Scoreboard ---------------------------------------------------------------------------
def _median(values):
vals = sorted(v for v in values if isinstance(v, (int, float)))
if not vals:
return None
mid = len(vals) // 2
return vals[mid] if len(vals) % 2 else (vals[mid - 1] + vals[mid]) / 2
def _and(items) -> str:
items = list(items)
if len(items) < 3:
return " and ".join(items)
return ", ".join(items[:-1]) + ", and " + items[-1]
def _money(value) -> str:
return "$%.2f" % value
def _plural(n, word) -> str:
return "%d %s%s" % (n, word, "" if n == 1 else "s")
def _shown(x) -> str:
"""A harness's name in markdown: the label from the harness table, or, for a
harness the table does not know, the recorded label in inline code."""
adapter = harnesses.ADAPTERS.get(x["harness"])
return adapter.label if adapter else code(x["label"], 40)
def _fix(r) -> str:
"""A run's fix in markdown: rebuilt from the harness table and the run's error,
or else the recorded hint in inline code."""
adapter = harnesses.ADAPTERS.get(r["harness"])
rebuilt = fatal_hint(adapter.label, adapter.binary, r.get("error") or "") if adapter else ""
return rebuilt or code(r["hint"], 80)
def summarize(records) -> dict:
"""The scoreboard. Harnesses are compared only on the tasks every one of them
finished; a harness with no finished run is listed as could not run. The
headline and notes are markdown sentences, so the error text in them sits in
inline code, and harness names and fixes come from the harness table (or sit
in inline code); the other text fields are plain, with secrets masked."""
rep = {"headline": "No runs recorded yet.", "tasks_compared": 0, "harnesses": [], "tasks": [],
"spent_usd": 0.0, "runs": 0, "notes": []}
latest, names, labels, order, info = {}, [], {}, [], {}
for r in records:
latest[(r["task"], r["harness"])] = r # a rerun of the same pair replaces the older record
if r["harness"] not in labels:
names.append(r["harness"])
adapter = harnesses.ADAPTERS.get(r["harness"])
labels[r["harness"]] = adapter.label if adapter else safe_text(r.get("label") or r["harness"], 40)
if r["task"] not in info:
order.append(r["task"])
info[r["task"]] = r
if not records:
return rep
finished = {key: r for key, r in latest.items() if r.get("status") in SCORED}
ran = [h for h in names if any(n == h for _, n in finished)]
common = [t for t in order if ran and all((t, h) in finished for h in ran)]
for h in names:
runs = [finished[(t, h)] for t in common] if h in ran else []
passed = sum(r["status"] == "passed" for r in runs)
costs = [r.get("cost_usd") for r in runs]
unmeasured = sum(c is None for c in costs)
total = round(sum(c for c in costs if c is not None), 4)
minutes = _median(r.get("agent_seconds") for r in runs)
errors = [r for (t, n), r in latest.items() if n == h and (t, n) not in finished]
rep["harnesses"].append({
"harness": h, "label": labels[h],
"versions": sorted({safe_text(r.get("version") or "", 60)
for (t, n), r in latest.items() if n == h} - {""}),
"runs": len(runs), "passed": passed, "pass_rate": round(passed / len(runs), 3) if runs else None,
"median_minutes": round(minutes / 60, 1) if minutes is not None else None,
"cost_per_pass": round(total / passed, 4) if passed and not unmeasured else None,
"total_cost": total if runs and unmeasured < len(runs) else None, "unmeasured_runs": unmeasured,
"median_lines_changed": _median(r.get("lines_changed") for r in runs),
"timeouts": sum(bool(r.get("timed_out")) for r in runs), "errors": len(errors),
"could_not_run": None if h in ran else (safe_text(errors[0].get("error") or "", 100)
if errors else "no result")})
compared = sorted([x for x in rep["harnesses"] if x["could_not_run"] is None],
key=lambda x: (-x["passed"], x["cost_per_pass"] is None, x["cost_per_pass"] or 0))
parts = []
for x in compared:
part = "%s passed %d" % (_shown(x), x["passed"])
if x["passed"] and x["cost_per_pass"] is not None:
part += " at %s each" % _money(x["cost_per_pass"])
elif x["passed"]:
part += " (cost not measured)"
parts.append(part)
cannot = ["%s could not run (%s)" % (_shown(x), code(x["could_not_run"], 100))
for x in rep["harnesses"] if x["could_not_run"]]
if common:
rep["headline"] = "On %s from your git history, %s%s." % (
_plural(len(common), "task"), _and(parts), "; " + "; ".join(cannot) if cannot else "")
elif cannot and not compared:
rep["headline"] = "No harness could run: %s." % "; ".join(cannot)
else:
rep["headline"] = "No task has a result from every harness yet."
rep["tasks_compared"] = len(common)
rep["runs"] = len(latest)
rep["spent_usd"] = round(sum(r.get("charged_usd") or 0.0 for r in records), 4) # every run, reruns too
for t in order:
rep["tasks"].append({"task": t, "commit": info[t].get("commit", ""),
"subject": safe_text(info[t].get("subject") or "", 100),
"gold_lines": info[t].get("gold_lines"),
"results": {h: {"status": latest[(t, h)].get("status"),
"timed_out": bool(latest[(t, h)].get("timed_out")),
"minutes": round((latest[(t, h)].get("agent_seconds") or 0) / 60, 1),
"cost_usd": latest[(t, h)].get("cost_usd"),
"error": safe_text(latest[(t, h)].get("error") or "", 120)}
for h in names if (t, h) in latest}})
_notes(rep, latest, finished, order, common)
return rep
def _notes(rep, latest, finished, order, common) -> None:
notes = rep["notes"]
left_out = len(order) - len(common)
if left_out and common:
notes.append("%s left out of the comparison because not every harness finished %s; the per-task "
"table shows them." % (_plural(left_out, "task") + (" is" if left_out == 1 else " are"),
"it" if left_out == 1 else "them"))
timeouts = sum(bool(r.get("timed_out")) for r in latest.values())
if timeouts:
notes.append("%s hit the time limit and %s scored on the changes the agent had made."
% (_plural(timeouts, "run"), "was" if timeouts == 1 else "were"))
for x in rep["harnesses"]:
mine = [r for (t, n), r in latest.items() if n == x["harness"] and (t, n) not in finished]
stopped = [r for r in mine if r.get("status") == "interrupted"]
failed = [r for r in mine if r.get("status") != "interrupted"]
hinted = [r for r in failed if r.get("hint")]
if stopped:
notes.append("%s: %s stopped before %s finished; the next run starts %s again." % (
_shown(x), _plural(len(stopped), "run") + (" was" if len(stopped) == 1 else " were"),
"it" if len(stopped) == 1 else "they", "it" if len(stopped) == 1 else "them"))
if failed:
notes.append("%s: %s could not run, for example: %s%s" % (
_shown(x), _plural(len(failed), "run"),
code(failed[0]["error"], 120) if failed[0].get("error") else "unknown",
". Fix: %s." % _fix(hinted[0]) if hinted else "."))
counted = [r for (t, n), r in latest.items() if n == x["harness"] and isinstance(r.get("test_runs"), int)]
test_runs = sum(r["test_runs"] for r in counted)
test_failures = sum(r.get("test_failures") or 0 for r in counted)
if test_failures:
notes.append("%s's own runs of the test command failed %d of %d times. Its sandbox blocks the network "
"and writes outside the copy, so a test command that needs either fails there; a setup "
"that installs into the copy and a test command that runs offline let it check its work."
% (_shown(x), test_failures, test_runs))
unmeasured = [r for r in latest.values() if r.get("cost_source") in ("unmeasured", "unpriced")]
if unmeasured:
notes.append("Cost not measured for %s. The spend cap counted a high estimate for priced harnesses "
"and $0 for unpriced ones." % _plural(len(unmeasured), "run"))
if any(r.get("cost_source") == "partial" for r in latest.values()):
notes.append("Some runs were stopped before the harness reported its cost; their cost counts only the "
"replies sent before the stop.")
notes.append("Dollar figures are API list prices: what the harness reported, or its tokens times the price table "
"checked %s. On a subscription plan, runs count against the plan's limits instead." % PRICES_CHECKED)
def _cell(res) -> str:
if res is None:
return "not run"
if res["status"] == "error":
return "could not run"
if res["status"] not in SCORED: # "interrupted" is the skill's own; any other status is recorded text
return res["status"] if res["status"] == "interrupted" else code(res["status"], 40)
cell = res["status"] + (" (time limit)" if res["timed_out"] else "") + ", %.1f min" % res["minutes"]
return cell + (", " + _money(res["cost_usd"]) if res["cost_usd"] is not None else "")
def render_report(rep) -> str:
lines = ["**%s**" % rep["headline"], ""]
if not rep["harnesses"]:
return "\n".join(lines)
if rep["tasks_compared"]:
lines += ["| Harness | Version | Passed | Pass rate | Median minutes | Cost per pass | Total cost | "
"Median lines changed |", "|---|---|---|---|---|---|---|---|"]
for x in rep["harnesses"]:
version = ", ".join(code(v, 60) for v in x["versions"]) or "unknown"
if x["could_not_run"]:
lines.append("| %s | %s | could not run | n/a | n/a | n/a | n/a | n/a |" % (_shown(x), version))
continue
if x["total_cost"] is None:
total = "not measured"
else:
total = _money(x["total_cost"]) + (" + %d not measured" % x["unmeasured_runs"]
if x["unmeasured_runs"] else "")
lines.append("| %s | %s | %d of %d | %d%% | %s | %s | %s | %s |" % (
_shown(x), version, x["passed"], x["runs"], round(100 * x["pass_rate"]),
"%.1f" % x["median_minutes"] if x["median_minutes"] is not None else "n/a",
_money(x["cost_per_pass"]) if x["cost_per_pass"] is not None else "n/a", total,
"%g" % x["median_lines_changed"] if x["median_lines_changed"] is not None else "n/a"))
lines.append("")
labels = [(x["harness"], _shown(x)) for x in rep["harnesses"]]
lines += ["Per task (the size of the original fix in changed lines, then each harness):", "",
"| Task | Change | Original fix lines | " + " | ".join(label for _, label in labels) + " |",
"|---|---|---|" + "---|" * len(labels)]
for t in rep["tasks"]:
lines.append("| %s | %s | %s | %s |" % (
code(safe_text(t["task"], 40)[:7]), code(t["subject"], 100),
"?" if t["gold_lines"] is None else t["gold_lines"] if isinstance(t["gold_lines"], int)
else code(t["gold_lines"], 20),
" | ".join(_cell(t["results"].get(h)) for h, _ in labels)))
lines += [""] + ["- " + note for note in rep["notes"]]
lines += ["", "Counted toward the spend cap: %s." % _money(rep["spent_usd"])]
return "\n".join(lines) + "\n"
# --- Estimate -------------------------------------------------------------------------------
def estimate(doc, adapters) -> dict:
n = len([t for t in doc.get("tasks", []) if t.get("validated")])
rows = []
for a in adapters:
rng = a.cost_range()
rows.append({"harness": a.name, "label": a.label, "installed": a.available(), "runs": n,
"per_run": [round(rng[0], 4), round(rng[1], 4)] if rng else None,
"total": [round(rng[0] * n, 4), round(rng[1] * n, 4)] if rng else None,
"models": [safe_text(m, 60) for m in a.estimate_models()],
"unpriced_reason": None if rng else a.unpriced_reason()})
priced = [r for r in rows if r["total"]]
return {"tasks": n, "runs": n * len(rows), "harnesses": rows,
"total": [round(sum(r["total"][0] for r in priced), 4), round(sum(r["total"][1] for r in priced), 4)]
if priced else None,
"unpriced": [r["label"] for r in rows if not r["total"]], "prices_checked": PRICES_CHECKED,
"tokens_per_run": [sum(LOW_RUN.values()), sum(HIGH_RUN.values())]}
def render_estimate(est) -> str:
k = len(est["harnesses"])
head = "%s: %s on %d harness%s." % (_plural(est["runs"], "run"), _plural(est["tasks"], "task"),
k, "" if k == 1 else "es")
if est["total"]:
head += " Estimated cost at API prices: %s to %s" % (_money(est["total"][0]), _money(est["total"][1]))
head += (", plus %s, which cannot be priced." % _and(est["unpriced"])) if est["unpriced"] else "."
elif est["unpriced"]:
head += " %s cannot be priced." % _and(est["unpriced"])
lines = ["**%s**" % head, "", "| Harness | Found on this computer | Runs | Per run | Total | Priced as |",
"|---|---|---|---|---|---|"]
for r in est["harnesses"]:
if r["per_run"]:
per = "%s to %s" % (_money(r["per_run"][0]), _money(r["per_run"][1]))
total = "%s to %s" % (_money(r["total"][0]), _money(r["total"][1]))
low, high = (code(m, 60) for m in r["models"])
basis = low if low == high else "%s to %s" % (low, high)
else:
per = total = "n/a"
basis = "cannot be priced: %s" % r["unpriced_reason"]
lines.append("| %s | %s | %d | %s | %s | %s |" % (r["label"], "yes" if r["installed"] else "no",
r["runs"], per, total, basis))
lines += ["", "- The real cost depends on the model each harness uses and on your plan. On a subscription plan "
"(such as Claude Max or ChatGPT), runs count against the plan's limits instead of a bill.",
"- Each run is assumed to use %s tokens (a short fix) to %s tokens (a long one), mostly cached "
"reads. Prices checked %s." % ("{:,}".format(est["tokens_per_run"][0]),
"{:,}".format(est["tokens_per_run"][1]), est["prices_checked"]),
"- The cap limits counted spend. Claude Code stops itself at the budget left. A Codex run has no "
"limit except --timeout, and a run cut off there is counted as an estimate, so the bill can pass the "
"cap by more than one run. Runs allowed with --allow-unpriced count $0."]
missing = [r["label"] for r in est["harnesses"] if not r["installed"]]
if missing:
lines.append("- Not found on this computer: %s. Install it or leave it out." % _and(missing))
if not est["tasks"]:
lines.append("- The tasks file has no validated tasks. Run mine_tasks.py with --validate first.")
return "\n".join(lines) + "\n"
# --- Command line ---------------------------------------------------------------------------------
def _load_tasks(path) -> dict:
try:
with open(path, encoding="utf-8") as fh:
doc = json.load(fh)
except (OSError, ValueError):
raise UsageError("Cannot read the tasks file %s. Run mine_tasks.py first." % code(path, 300))
if not isinstance(doc, dict) or "tasks" not in doc:
raise UsageError("%s is not a tasks file from mine_tasks.py." % code(path, 300))
return doc
def _names(text) -> list:
return list(dict.fromkeys(n.strip() for n in (text or "").split(",") if n.strip()))
def _adapters(names, models, registry) -> list:
wanted = _names(names)
unknown = [n for n in wanted if n not in registry]
if unknown or not wanted:
raise UsageError("Unknown harness %s. Known: %s." % (safe_text(", ".join(unknown) or names, 80),
", ".join(sorted(registry))))
pinned = {}
for pair in _names(models):
name, sep, model = pair.partition("=")
if not sep or name.strip() not in wanted or not model.strip():
raise UsageError("--model takes harness=model pairs for the chosen harnesses, such as "
"claude-code=claude-sonnet-5-5.")
pinned[name.strip()] = model.strip()
try:
return [registry[n](model=pinned.get(n)) for n in wanted]
except ValueError as err:
raise UsageError(str(err))
def _emit(markdown, data, args) -> None:
"""Print the markdown report, or JSON with --json; --out always gets the markdown."""
sys.stdout.write(json.dumps(data, indent=2) + "\n" if args.json else markdown)
if args.out:
with open(args.out, "w", encoding="utf-8") as fh:
fh.write(markdown)
def main(argv=None, registry=None) -> int:
registry = registry or harnesses.ADAPTERS
parser = argparse.ArgumentParser(description="Test-drive coding agents on tasks mined from your git history.")
sub = parser.add_subparsers(dest="command", required=True)
p_run = sub.add_parser("run", help="run the agents and score them")
p_est = sub.add_parser("estimate", help="count the runs and estimate the cost")
p_rep = sub.add_parser("report", help="print the scoreboard from results.jsonl")
for p in (p_run, p_est):
p.add_argument("--tasks", default=DEFAULT_TASKS, help="tasks.json from mine_tasks.py (default %s)" % DEFAULT_TASKS)
p_run.add_argument("--harness", required=True, help="comma-separated: %s" % ", ".join(sorted(registry)))
p_est.add_argument("--harness", default=",".join(sorted(registry)), help="comma-separated (default: all)")
p_run.add_argument("--max-usd", type=float, required=True, help="total spend cap in dollars, required")
p_run.add_argument("--timeout", type=int, default=900, help="seconds per agent run (default 900)")
p_run.add_argument("--keep", action="store_true", help="keep each run's workspace folder")
p_run.add_argument("--allow-unpriced", metavar="HARNESSES",
help="comma-separated harnesses to run although their cost cannot be measured (counted as $0)")
p_run.add_argument("--model", help="pin models: harness=model pairs, comma-separated")
p_run.add_argument("--results", help="results.jsonl (default: next to the tasks file)")
p_rep.add_argument("--results", default=DEFAULT_RESULTS, help="results.jsonl (default %s)" % DEFAULT_RESULTS)
for p in (p_run, p_est, p_rep):
p.add_argument("--json", action="store_true", help="print JSON instead of markdown")
p.add_argument("--out", help="also write the markdown to this file")
args = parser.parse_args(argv)
handlers = install_stop_handlers()
try:
return _main(args, registry)
except KeyboardInterrupt:
_stderr("Stopped. The run in progress is recorded with its spend and its copy is deleted; run the same "
"command to continue.")
return 130
finally:
restore_handlers(handlers)
def _main(args, registry) -> int:
try:
if args.command == "report":
rep = summarize(load_results(args.results))
_emit(render_report(rep), rep, args)
return 0
doc = _load_tasks(args.tasks)
adapters = _adapters(args.harness, getattr(args, "model", None), registry)
if args.command == "estimate":
est = estimate(doc, adapters)
_emit(render_estimate(est), est, args)
return 0
chosen = {a.name for a in adapters}
allowed = _names(args.allow_unpriced)
for name in allowed:
if name not in chosen:
raise UsageError("%s is not among the chosen harnesses (--harness)." % safe_text(name, 40))
results = args.results or os.path.join(os.path.dirname(os.path.abspath(args.tasks)), "results.jsonl")
summary = run(doc, results, adapters, args.max_usd, args.timeout, args.keep, allowed)
except UsageError as err:
print("error: %s" % err, file=sys.stderr)
return 2
if summary["not_installed"]:
_stderr("Not found on this computer, so not run: %s." % ", ".join(summary["not_installed"]))
for name, hint in summary["skipped"].items():
_stderr("%s could not run. Fix: %s. Its other runs were skipped; run the same command again after that."
% (name, hint))
if summary["stopped_at_cap"]:
_stderr("Stopped at the %s cap with %s not started. Finished runs are kept: rerun with a higher "
"--max-usd to continue." % (_money(args.max_usd), _plural(summary["not_started"], "run")))
_stderr("Results: %s" % code(results, 300))
rep = summarize(load_results(results))
_emit(render_report(rep), dict(rep, run=summary), args)
return 0
if __name__ == "__main__":
sys.exit(main())
"""Harness adapters: how to run each coding agent headless and read its output.
Each adapter is a small class with the same methods:
- available(): the program is on PATH.
- version(): "<program> --version", first line.
- build_command(prompt, workdir, caps): the argument list for one run. caps has
"test_cmd" (the repository's test command) and "max_usd" (the budget left).
- parse_output(stdout, test_cmd=""): {cost_usd, cost_source, tokens, model, turns,
error, denials, worked, test_runs, test_failures}. cost_source is "reported"
(the harness said), "tokens" (tokens times the price table), "partial" (a
stopped run, priced from what it reported), or None. worked is True when the
output shows the model did work. error is the harness's own message.
- priced(): whether a run's dollar cost can be measured; unpriced_reason() and
unpriced_fix() say why not and what to do; can_run_unpriced says whether
--allow-unpriced may run it anyway.
- cost_range(): (low, high) dollars for one run, for estimates.
Every command uses the most restrictive settings that still let the agent edit
files in its workspace and run the test command. Flags come from the headless
docs of each harness, checked 2026-09-28 (see references/harness-commands.md).
OpenCode is left out: the JSON event format of `opencode run --format json` is
not documented. Python 3.9+, standard library only.
"""
from __future__ import annotations
import json
import math
import os
import re
import shutil
import subprocess
from common import clean_env
from pricing import cost_usd, price_for
from safe import code, safe_text
# Token use assumed for one run when estimating cost (Anthropic meanings:
# input is uncached input). A short fix and a long one; both are guesses.
LOW_RUN = {"input": 20_000, "cache_read": 150_000, "cache_write": 20_000, "output": 4_000}
HIGH_RUN = {"input": 150_000, "cache_read": 3_000_000, "cache_write": 150_000, "output": 50_000}
def command_parts(test_cmd) -> list:
"""The simple commands inside a compound test command, such as
"npm ci && npm test". Permission rules match each part on its own."""
return [p.strip() for p in re.split(r"&&|\|\||;|\|", test_cmd) if p.strip()]
def _int(value) -> int:
return value if isinstance(value, int) and not isinstance(value, bool) else 0
def _json(text):
try:
return json.loads(text)
except ValueError:
return None
def _records(stdout) -> list:
"""Every JSON object printed one per line."""
return [r for r in (_json(line) for line in stdout.splitlines() if line.startswith("{")) if isinstance(r, dict)]
_AUTH_RE = re.compile(r"(?i)authenticat|oauth|not logged in|please (log|sign) ?in|invalid api key|unauthorized")
_VERSION_RE = re.compile(r"(?i)newer version|please upgrade|update required")
_MISSING_RE = re.compile(r"(?i)no such file or directory|command not found")
def fatal_hint(label, binary, error) -> str:
"""The fix for an error that will repeat on every task (a login, an old
version, a missing program), or "" for an error of one run."""
if _AUTH_RE.search(error):
return "sign in: run %s once" % binary
if _VERSION_RE.search(error):
return "upgrade %s" % label
if _MISSING_RE.search(error):
return "install %s, or put %s on PATH" % (label, binary)
return ""
def _add(total, part) -> dict:
return {k: (total or {}).get(k, 0) + part.get(k, 0) for k in ("input", "cache_read", "cache_write", "output")}
class Harness:
name = ""
label = ""
binary = ""
low_model = high_model = None
can_run_unpriced = True
def __init__(self, model=None):
self.model = model
def available(self) -> bool:
return shutil.which(self.binary) is not None
def version(self) -> str:
try:
res = subprocess.run([self.binary, "--version"], capture_output=True, text=True, timeout=30,
stdin=subprocess.DEVNULL, env=clean_env())
except (OSError, subprocess.TimeoutExpired):
return ""
lines = [line for line in res.stdout.splitlines() if line.strip()]
return safe_text(lines[0], 60) if lines else ""
def priced(self) -> bool:
return self.model is not None and price_for(self.model) is not None
def unpriced_reason(self) -> str:
return "the price table has no price for %s" % code(self.model, 60)
def unpriced_fix(self) -> str:
return "Pass --allow-unpriced %s to run it anyway; its runs count $0" % self.name
def estimate_models(self) -> tuple:
"""(model for the low estimate, model for the high one), or () when unpriced."""
if self.priced():
return (self.model, self.model)
return (self.low_model, self.high_model) if self.low_model else ()
def cost_range(self):
models = self.estimate_models()
return (cost_usd(LOW_RUN, models[0]), cost_usd(HIGH_RUN, models[1])) if models else None
def fallback_cost(self) -> float:
"""What the spend cap counts for a run whose cost could not be read."""
rng = self.cost_range()
return rng[1] if rng else 0.0
class ClaudeCode(Harness):
name, label, binary = "claude-code", "Claude Code", "claude"
low_model, high_model = "claude-sonnet-5-5", "claude-opus-5-5"
def priced(self) -> bool:
return True # the result carries total_cost_usd
def estimate_models(self) -> tuple:
if self.model and price_for(self.model):
return (self.model, self.model)
return (self.low_model, self.high_model)
def build_command(self, prompt, workdir, caps) -> list:
# -p cannot answer permission prompts, so anything not allowed here is denied.
rules = []
for part in command_parts(caps["test_cmd"]):
rules += ["Bash(%s)" % part, "Bash(%s *)" % part]
argv = [self.binary, "-p", prompt, "--output-format", "stream-json", "--verbose",
"--permission-mode", "acceptEdits", "--allowedTools", ",".join(rules),
"--no-session-persistence", "--strict-mcp-config"] # no --mcp-config: no MCP server loads
if caps.get("max_usd") is not None:
cents = math.floor(caps["max_usd"] * 100 + 1e-9)
argv += ["--max-budget-usd", "%.2f" % (max(cents, 1) / 100)]
if self.model:
argv += ["--model", self.model]
return argv
def parse_output(self, stdout, test_cmd="") -> dict:
result, messages = None, {}
for rec in _records(stdout):
if rec.get("type") == "result":
result = rec
elif rec.get("type") == "assistant":
msg = rec.get("message") or {}
if msg.get("id") and isinstance(msg.get("usage"), dict) and msg.get("model") != "<synthetic>":
messages[msg["id"]] = (msg.get("model") or "", msg["usage"]) # split records: keep the last
if result is None and "{" in stdout:
whole = _json(stdout[stdout.find("{"):])
result = whole if isinstance(whole, dict) and whole.get("type") == "result" else None
if result is not None:
by_model = result.get("modelUsage") or {}
tokens = None
for usage in by_model.values():
tokens = _add(tokens, {"input": _int(usage.get("inputTokens")),
"cache_read": _int(usage.get("cacheReadInputTokens")),
"cache_write": _int(usage.get("cacheCreationInputTokens")),
"output": _int(usage.get("outputTokens"))})
cost = result.get("total_cost_usd")
cost = float(cost) if isinstance(cost, (int, float)) and not isinstance(cost, bool) else None
error = ""
if result.get("is_error"): # the message, never the subtype ("success" on a failed login)
errors = result.get("errors") if isinstance(result.get("errors"), list) else []
error = safe_text(result.get("result") or (errors[0] if errors else "")
or result.get("terminal_reason") or "error")
return {"cost_usd": cost, "cost_source": "reported" if cost is not None else None, "tokens": tokens,
"model": max(by_model, key=lambda m: by_model[m].get("costUSD") or 0) if by_model else "",
"turns": result.get("num_turns"), "error": error,
"denials": len(result.get("permission_denials") or []), "worked": bool(messages),
"test_runs": None, "test_failures": None}
tokens, cost, output_by_model = None, 0.0, {}
for model, usage in messages.values():
part = {"input": _int(usage.get("input_tokens")), "cache_read": _int(usage.get("cache_read_input_tokens")),
"cache_write": _int(usage.get("cache_creation_input_tokens")), "output": _int(usage.get("output_tokens")),
"cache_write_1h": _int((usage.get("cache_creation") or {}).get("ephemeral_1h_input_tokens"))}
tokens = _add(tokens, part)
price = cost_usd(part, model)
cost = None if cost is None or price is None else cost + price
output_by_model[model] = output_by_model.get(model, 0) + part["output"]
if tokens is None:
cost = None
return {"cost_usd": cost, "cost_source": "partial" if cost is not None else None, "tokens": tokens,
"model": max(output_by_model, key=output_by_model.get) if output_by_model else "", "turns": None,
"error": "no final result: the run was stopped" if messages else "no output", "denials": 0,
"worked": bool(messages), "test_runs": None, "test_failures": None}
def _codex_config_model():
"""The top-level `model` in Codex's config.toml, or None."""
home = os.environ.get("CODEX_HOME") or os.path.join(os.path.expanduser("~"), ".codex")
try:
with open(os.path.join(home, "config.toml"), encoding="utf-8") as fh:
lines = fh.read().splitlines()
except OSError:
return None
for line in lines:
if line.strip().startswith("["):
return None # later keys belong to tables such as [profiles.x]
m = re.match(r"""\s*model\s*=\s*["']([^"']+)["']""", line)
if m:
return m.group(1)
return None
class Codex(Harness):
name, label, binary = "codex", "Codex", "codex"
can_run_unpriced = False # without a model there is no price, and no cap
def __init__(self, model=None):
# `codex exec --json` reports tokens but neither model nor cost, so the
# model comes from --model or config.toml, and the price from pricing.py.
super().__init__(model or _codex_config_model())
def unpriced_reason(self) -> str:
return "Codex's config names no model" if self.model is None else super().unpriced_reason()
def unpriced_fix(self) -> str:
return "Pin one with --model codex=<id>"
def build_command(self, prompt, workdir, caps) -> list:
argv = [self.binary, "exec", "--json", "--sandbox", "workspace-write", "--ephemeral"]
if self.model:
argv += ["--model", self.model]
return argv + [prompt]
def parse_output(self, stdout, test_cmd="") -> dict:
tokens, error, denials, worked, runs, failures = None, "", 0, False, 0, 0
parts = command_parts(test_cmd) if test_cmd else []
for rec in _records(stdout):
kind = rec.get("type")
if kind == "turn.completed":
u = rec.get("usage") or {}
tokens = _add(tokens, {"input": _int(u.get("input_tokens")) - _int(u.get("cached_input_tokens")),
"cache_read": _int(u.get("cached_input_tokens")),
"cache_write": _int(u.get("cache_write_input_tokens")),
"output": _int(u.get("output_tokens"))}) # includes reasoning
elif kind == "turn.failed":
error = (rec.get("error") or {}).get("message") or "turn failed"
elif kind == "error":
error = rec.get("message") or "error"
elif kind == "item.completed":
worked = True
item = rec.get("item") or {}
if item.get("type") == "command_execution":
denials += item.get("status") == "declined"
if any(part in str(item.get("command") or "") for part in parts): # a run of the test command
runs += 1
code = item.get("exit_code")
failures += item.get("status") == "failed" or (isinstance(code, int) and code != 0)
cost = cost_usd(tokens, self.model) if tokens is not None and self.model else None
return {"cost_usd": cost, "cost_source": "tokens" if cost is not None else None, "tokens": tokens,
"model": self.model or "", "turns": None, "error": safe_text(error), "denials": int(denials),
"worked": worked, "test_runs": runs, "test_failures": int(failures)}
class GeminiCli(Harness):
name, label, binary = "gemini-cli", "Gemini CLI", "gemini"
def __init__(self, model=None):
if model:
raise ValueError("Gemini CLI takes no --model here: its headless docs list no model flag. "
"Set the model in Gemini CLI's settings instead.")
super().__init__()
def priced(self) -> bool:
return False # no cost field, and pricing.py has no confirmed Gemini prices
def unpriced_reason(self) -> str:
return "it reports no cost, and the price table has no Gemini prices"
def build_command(self, prompt, workdir, caps) -> list:
# --skip-trust: a new folder is untrusted, and untrusted folders turn off auto_edit.
argv = [self.binary, "--output-format", "json", "--approval-mode", "auto_edit", "--skip-trust",
"--allowed-tools"]
argv += ["run_shell_command(%s)" % part for part in command_parts(caps["test_cmd"])]
return argv + ["-p", prompt]
def parse_output(self, stdout, test_cmd="") -> dict:
doc = _json(stdout[stdout.find("{"):]) if "{" in stdout else None
if not isinstance(doc, dict):
return {"cost_usd": None, "cost_source": None, "tokens": None, "model": "", "turns": None,
"error": "no JSON result", "denials": 0, "worked": False, "test_runs": None,
"test_failures": None}
stats = doc.get("stats") or {}
models = stats.get("models") or {}
tokens, cost, turns, size = None, 0.0 if models else None, 0, {}
for name, entry in models.items():
t = entry.get("tokens") or {}
cached = _int(t.get("cached"))
part = {"input": _int(t["input"]) if "input" in t else _int(t.get("prompt")) - cached,
"cache_read": cached, "cache_write": 0,
"output": _int(t.get("candidates")) + _int(t.get("thoughts"))}
tokens = _add(tokens, part)
price = cost_usd(part, name)
cost = None if cost is None or price is None else cost + price
turns += _int((entry.get("api") or {}).get("totalRequests"))
size[name] = sum(part.values())
error = doc.get("error")
return {"cost_usd": cost, "cost_source": "tokens" if cost is not None else None, "tokens": tokens,
"model": max(size, key=size.get) if size else "", "turns": turns if models else None,
"error": safe_text(error.get("message") or "error") if isinstance(error, dict) else "",
"denials": _int(((stats.get("tools") or {}).get("totalDecisions") or {}).get("reject")),
"worked": turns > 0, "test_runs": None, "test_failures": None}
ADAPTERS = {cls.name: cls for cls in (ClaudeCode, Codex, GeminiCli)}
"""Mine test-drive tasks from a repository's own git history.
A task is a past commit that changed both source files and test files. The
task text is the commit message, the hidden tests are the commit's test-file
changes, and the gold patch is the rest of the change. With --validate each
task is checked in a fresh temporary copy of the repository: at the parent
commit plus the hidden tests the test command must fail, and with the whole
commit it must pass, twice (a changing result is flaky). Only tasks that pass
that check are kept. The HEAD check's output goes to <tasks folder>/logs/.
Usage:
python3 mine_tasks.py --repo . [--test-cmd "<cmd>"] [--validate] [--max 10] [--since 365d]
[--setup-cmd "<cmd>"] [--test-timeout 600] [--tasks <path>] [--json] [--out <path>]
Exit codes: 0 success (including "no tasks found"), 2 usage or input error
(not a git repository, no test command, or a test command that fails at HEAD),
130 stopped (SIGINT, SIGTERM, or SIGHUP; the temporary copies are deleted).
The user's repository is only read: tasks.json is the one file written.
Python 3.9+, standard library only.
"""
from __future__ import annotations
import argparse
import datetime
import json
import os
import re
import sys
import time
from common import (classify, duration, git, install_stop_handlers, make_workspace, remove_workspace,
restore_handlers, run_command, run_tests, touches_manifest, write_files)
from safe import code, safe_text
MAX_FILES = 30 # commits that change more files are too large for a task
MAX_LINES = 1000 # same for added plus deleted lines
_TRAILER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9-]*: \S")
_NO_TEST_SCRIPT = "no test specified" # npm's placeholder "test" script
MAX_CHECKS_PER_TASK = 3 # check at most this many candidates per wanted task
DEFAULT_FOLDER = ".harness-test-drive"
class UsageError(Exception):
"""Bad input: the message says what to change."""
def detect_test_cmd(repo) -> tuple:
"""(command, what it came from) from the project files, or (None, "")."""
def path(name):
return os.path.join(repo, name)
def read(name):
try:
with open(path(name), encoding="utf-8", errors="replace") as fh:
return fh.read()
except OSError:
return ""
try:
script = json.loads(read("package.json") or "{}").get("scripts", {}).get("test")
except (ValueError, AttributeError):
script = None
if isinstance(script, str) and script.strip() and _NO_TEST_SCRIPT not in script:
runner = ("pnpm test" if os.path.exists(path("pnpm-lock.yaml")) else
"yarn test" if os.path.exists(path("yarn.lock")) else
"bun run test" if os.path.exists(path("bun.lock")) or os.path.exists(path("bun.lockb")) else
"npm test")
return runner, "package.json scripts.test"
pytest = "uv run pytest -q" if os.path.exists(path("uv.lock")) else "python3 -m pytest -q"
for name, marker in (("pytest.ini", ""), ("pyproject.toml", "[tool.pytest"), ("setup.cfg", "[tool:pytest]"),
("tox.ini", "[pytest]"), ("conftest.py", ""), ("tests/conftest.py", ""),
("test/conftest.py", "")):
if os.path.isfile(path(name)) and marker in read(name):
return pytest, name
if os.path.isdir(path("tests")) and any(n.startswith("test_") and n.endswith(".py") for n in os.listdir(path("tests"))):
return pytest, "test files in tests/"
if os.path.isfile(path("go.mod")):
return "go test ./...", "go.mod"
if os.path.isfile(path("Cargo.toml")):
return "cargo test", "Cargo.toml"
if re.search(r"^test\s*:", read("Makefile"), re.M):
return "make test", "Makefile test target"
return None, ""
def parse_since(text) -> str:
"""'365d' becomes '365 days ago'; a YYYY-MM-DD date passes through."""
m = re.fullmatch(r"(\d+)d", text or "")
if m:
return "%d days ago" % int(m.group(1))
if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text or ""):
datetime.date.fromisoformat(text) # raises ValueError for a bad date
return text
raise ValueError("--since takes a number of days such as 365d, or a date such as 2026-01-15")
def _lines(count) -> int:
return int(count) if count.isdigit() else 0 # "-" for binary files
def find_candidates(repo, since="365d", max_files=MAX_FILES, max_lines=MAX_LINES) -> tuple:
"""Commits on HEAD since `since`, newest first, that change both source and
test files. Returns (candidates, counts)."""
try:
out = git(repo, "log", "-z", "--no-merges", "--no-renames", "--since=" + parse_since(since),
"--format=%x1e%H%x1f%P", "--numstat", "HEAD")
except RuntimeError as err:
raise UsageError("Could not read the history (%s). A partial clone keeps old file contents on the "
"server, and this skill fetches nothing: fetch the full history first "
"(git fetch --refetch), then run again." % err)
commits = []
for tok in out.split(b"\0"):
tok = tok.lstrip(b"\n")
if tok.startswith(b"\x1e"):
sha, _, parents = tok[1:].decode().partition("\x1f")
commits.append((sha, parents.split(), []))
elif tok and commits:
added, deleted, name = tok.split(b"\t", 2)
commits[-1][2].append((_lines(added.decode()) + _lines(deleted.decode()), os.fsdecode(name)))
candidates, skipped = [], {}
for sha, parents, files in commits:
kinds = {name: classify(name) for _, name in files}
tests = [name for _, name in files if kinds[name] == "test"]
if not parents:
reason = "first commit"
elif len(files) > max_files or sum(n for n, _ in files) > max_lines:
reason = "too large"
elif "source" not in kinds.values():
reason = "no source changes"
elif not tests:
reason = "no test changes"
else:
candidates.append({
"commit": sha, "parent": parents[0], "hidden_tests": tests,
"gold_files": [name for _, name in files if kinds[name] != "test"],
"gold_lines": sum(n for n, name in files if kinds[name] in ("source", "other"))})
continue
skipped[reason] = skipped.get(reason, 0) + 1
return candidates, {"commits": len(commits), "candidates": len(candidates), "skipped": skipped}
def task_text(repo, sha) -> tuple:
"""(subject, body) of a commit message, with trailers such as
Co-Authored-By removed from the body."""
message = git(repo, "log", "-1", "--format=%B", sha).decode("utf-8", "replace").strip()
subject, _, body = message.partition("\n")
paragraphs = [p for p in body.strip().split("\n\n") if p.strip()]
if paragraphs and all(_TRAILER_RE.match(line) for line in paragraphs[-1].splitlines() if line.strip()):
paragraphs.pop()
return subject.strip(), "\n\n".join(paragraphs).strip()
def _check(repo, commit, test_cmd, setup_cmd, timeout, work_root, hidden=(), gold=(), logs=None) -> tuple:
"""Run the tests in a fresh copy of `commit`. With no `gold`, this is the
HEAD check: the tests must pass as they are. Otherwise the tests must fail
with the hidden tests written, pass with the gold files written too, and do
both once more. Returns (reason the check failed or "", the runs)."""
try:
ws = make_workspace(repo, commit, root=work_root)
except (OSError, RuntimeError): # a damaged clone or a full disk
return "could not copy the commit", {}
runs, logs = {}, logs or {}
def setup():
return not setup_cmd or run_command(setup_cmd, ws, timeout, logs.get("setup"))["exit"] == 0
def switch(at, paths): # write paths as they are at `at`; a changed manifest needs the setup again
write_files(repo, at, paths, ws)
return not touches_manifest(paths) or setup()
try:
if not setup():
return "setup command failed", runs
if not gold:
runs["after"] = run_tests(test_cmd, ws, timeout, logs.get("tests"))
return ("" if runs["after"]["exit"] == 0 else "tests fail"), runs
fix, gold_files = gold
if not switch(fix, hidden):
return "setup command failed", runs
runs["before"] = run_tests(test_cmd, ws, timeout)
if runs["before"]["exit"] == 0:
return "tests pass before the fix", runs
if not switch(fix, gold_files):
return "setup command failed", runs
runs["after"] = run_tests(test_cmd, ws, timeout)
if runs["after"]["exit"] != 0:
return ("tests time out with the fix" if runs["after"]["timed_out"] else "tests fail with the fix"), runs
# Once more each way: tests whose result changes between runs cannot score an agent.
if not switch(commit, gold_files):
return "setup command failed", runs
runs["before_again"] = run_tests(test_cmd, ws, timeout)
if not switch(fix, gold_files):
return "setup command failed", runs
runs["after_again"] = run_tests(test_cmd, ws, timeout)
if runs["before_again"]["exit"] == 0 or runs["after_again"]["exit"] != 0:
return "flaky", runs
return "", runs
finally:
remove_workspace(ws)
def _output_tail(path, lines=15) -> str:
"""The last lines of a saved log, made safe, and where the full log is."""
try:
with open(path, encoding="utf-8", errors="replace") as fh:
text = [code(line, 200) for line in fh.read().splitlines() if line.strip()]
except (OSError, TypeError):
return ""
shown = "\n".join(" " + line for line in text[-lines:]) or " (no output)"
return "\nLast lines of its output:\n%s\nFull output: %s" % (shown, code(path, 300))
def _task(repo, cand, validated, runs=None) -> dict:
subject, body = task_text(repo, cand["commit"])
return dict({"id": cand["commit"][:12], "subject": subject, "body": body}, **cand,
validated=validated, check=runs or {})
def mine(repo, test_cmd, test_cmd_source="--test-cmd", since="365d", max_tasks=10, validate=False,
setup_cmd="", test_timeout=600, work_root=None, logs_dir=None) -> dict:
"""Find tasks, check them when `validate` is set, and return the tasks
document. With `logs_dir`, the HEAD check saves its output there."""
top = git(repo, "rev-parse", "--show-toplevel").decode().strip()
head = git(top, "rev-parse", "HEAD").decode().strip()
candidates, counts = find_candidates(top, since)
counts.update(checked=0, validated=0, rejected={})
doc = {"schema": "harness-test-drive/tasks/1", "repo": top, "head": head, "since": since,
"created": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"test_cmd": test_cmd, "test_cmd_source": test_cmd_source, "setup_cmd": setup_cmd,
"test_timeout": test_timeout, "head_check": None, "counts": counts, "validation_seconds": None,
"tasks": []}
if not validate:
doc["tasks"] = [_task(top, c, False) for c in candidates[:max_tasks]]
return doc
started = time.monotonic()
logs = {}
if logs_dir:
os.makedirs(logs_dir, exist_ok=True)
logs = {"setup": os.path.join(logs_dir, "head-setup.log"), "tests": os.path.join(logs_dir, "head-check.log")}
reason, runs = _check(top, head, test_cmd, setup_cmd, test_timeout, work_root, logs=logs)
doc["head_check"] = runs.get("after")
if reason == "could not copy the commit":
raise UsageError("Could not copy the files of HEAD into a fresh folder (a partial or damaged clone?).")
if reason == "setup command failed":
raise UsageError("The setup command fails at HEAD in a fresh copy of the repository. Fix --setup-cmd."
+ _output_tail(logs.get("setup")))
if reason:
failed = runs["after"]
what = ("it timed out after %s; raise --test-timeout if the tests need longer" % duration(failed["seconds"])
if failed["timed_out"] else "exit %s after %s" % (failed["exit"], duration(failed["seconds"])))
raise UsageError("The test command fails at HEAD in a fresh copy of the repository (%s). Pass a working "
"--test-cmd, or --setup-cmd to install what the tests need in the copy.%s"
% (what, _output_tail(logs.get("tests"))))
for cand in candidates[:MAX_CHECKS_PER_TASK * max_tasks]:
if len(doc["tasks"]) >= max_tasks:
break
counts["checked"] += 1
reason, runs = _check(top, cand["parent"], test_cmd, setup_cmd, test_timeout, work_root,
hidden=cand["hidden_tests"], gold=(cand["commit"], cand["gold_files"]))
if reason:
counts["rejected"][reason] = counts["rejected"].get(reason, 0) + 1
else:
doc["tasks"].append(_task(top, cand, True, runs))
counts["validated"] = len(doc["tasks"])
doc["validation_seconds"] = round(time.monotonic() - started, 1)
return doc
def _counted(counts) -> str:
return ", ".join("%d %s" % (n, reason) for reason, n in sorted(counts.items(), key=lambda kv: -kv[1]))
def render_report(doc, tasks_path) -> str:
counts, tasks = doc["counts"], doc["tasks"]
since = safe_text(parse_since(doc["since"]))
if tasks and counts["checked"]:
headline = ("Found %d task%s in your git history: %d of %d checked commits fail their tests before "
"the change and pass after it, twice." % (len(tasks), "" if len(tasks) == 1 else "s",
counts["validated"], counts["checked"]))
elif tasks:
headline = ("Found %d candidate task%s in your git history. None is checked yet: run again with "
"--validate to keep only tasks whose tests fail before the change and pass after it."
% (len(tasks), "" if len(tasks) == 1 else "s"))
elif counts["checked"]:
headline = ("Found no usable tasks: none of %d checked commits fails its tests before the change and "
"passes after it, twice." % counts["checked"])
else:
headline = "Found no candidate tasks in your git history since %s." % since
lines = ["**%s**" % headline, ""]
cmd = "Test command: %s (%s)." % (code(doc["test_cmd"], 200), safe_text(doc["test_cmd_source"]))
if doc["head_check"]:
cmd += " It passes at HEAD in %s." % duration(doc["head_check"]["seconds"])
lines += [cmd, ""]
if tasks:
lines += ["| # | Task | Change | Hidden test files | Lines in the original fix | Checked |",
"|---|---|---|---|---|---|"]
for i, t in enumerate(tasks, 1):
lines.append("| %d | %s | %s | %d | %d | %s |" % (
i, t["id"][:7], code(t["subject"], 70), len(t["hidden_tests"]), t["gold_lines"],
"yes" if t["validated"] else "no"))
lines.append("")
history = "History since %s: %d commits, %d change both source and test files." % (
since, counts["commits"], counts["candidates"])
if counts["skipped"]:
history += " Skipped: %s." % _counted(counts["skipped"])
lines.append(history)
if counts["checked"]:
checked = "Checked %d in fresh copies of the repository: %d kept" % (counts["checked"], counts["validated"])
if counts["rejected"]:
checked += ", %s" % _counted(counts["rejected"])
lines.append("%s. Checking took %s." % (checked, duration(doc["validation_seconds"])))
lines += ["", "Tasks file: %s" % code(tasks_path, 300)]
if tasks and counts["checked"]:
lines.append("Next: %s shows the number of runs and a cost range."
% code("drive.py estimate --tasks " + tasks_path, 330))
return "\n".join(lines) + "\n"
def summary_json(doc, tasks_path) -> dict:
"""The --json output: stable keys, untrusted text made safe."""
return {
"tasks_file": safe_text(tasks_path, 300), "repo": safe_text(doc["repo"], 300), "head": doc["head"],
"test_cmd": safe_text(doc["test_cmd"], 300), "test_cmd_source": doc["test_cmd_source"],
"head_check": doc["head_check"], "counts": doc["counts"], "validation_seconds": doc["validation_seconds"],
"tasks": [{"id": t["id"], "commit": t["commit"], "subject": safe_text(t["subject"], 120),
"hidden_tests": [safe_text(p, 200) for p in t["hidden_tests"]],
"gold_files": [safe_text(p, 200) for p in t["gold_files"]],
"gold_lines": t["gold_lines"], "validated": t["validated"]} for t in doc["tasks"]]}
def main(argv=None) -> int:
parser = argparse.ArgumentParser(description="Mine test-drive tasks from a repository's git history.")
parser.add_argument("--repo", default=".", help="the git repository (default: the current folder)")
parser.add_argument("--test-cmd", help="shell command that runs the tests (default: detected)")
parser.add_argument("--validate", action="store_true",
help="keep only tasks whose tests fail before the change and pass after it, twice")
parser.add_argument("--max", type=int, default=10, help="most tasks to keep (default 10)")
parser.add_argument("--since", default="365d", help="how far back: 365d, or a date such as 2026-01-15")
parser.add_argument("--setup-cmd", default="", help="shell command run in each fresh copy before the tests")
parser.add_argument("--test-timeout", type=int, default=600, help="seconds per test run (default 600)")
parser.add_argument("--tasks", help="where to write tasks.json (default: <repo>/%s/tasks.json)" % DEFAULT_FOLDER)
parser.add_argument("--json", action="store_true", help="print JSON instead of the markdown report")
parser.add_argument("--out", help="also write the report to this file")
args = parser.parse_args(argv)
handlers = install_stop_handlers()
try:
return _main(args)
except KeyboardInterrupt:
print("Stopped. The temporary copies are deleted, and no tasks file was written.", file=sys.stderr)
return 130
finally:
restore_handlers(handlers)
def _main(args) -> int:
try:
if args.max < 1 or args.test_timeout < 1:
raise UsageError("--max and --test-timeout must be at least 1.")
parse_since(args.since)
try:
top = git(args.repo, "rev-parse", "--show-toplevel").decode().strip()
except (RuntimeError, OSError):
raise UsageError("%s is not a git repository." % code(args.repo, 300))
test_cmd, source = (args.test_cmd, "--test-cmd") if args.test_cmd else detect_test_cmd(top)
if not test_cmd:
raise UsageError("No test command found (no package.json test script, pytest config or tests folder, "
"go.mod, Cargo.toml, or Makefile test target). Pass --test-cmd. Look in "
".github/workflows or the README for the command CI runs.")
tasks_path = args.tasks or os.path.join(top, DEFAULT_FOLDER, "tasks.json")
folder = os.path.dirname(os.path.abspath(tasks_path))
os.makedirs(folder, exist_ok=True)
if not args.tasks and not os.path.exists(os.path.join(folder, ".gitignore")):
with open(os.path.join(folder, ".gitignore"), "w") as fh:
fh.write("# Written by harness-test-drive. Keeps this folder out of git.\n*\n")
print("Test command: %s (%s)" % (code(test_cmd, 200), source), file=sys.stderr)
doc = mine(top, test_cmd, source, args.since, args.max, args.validate, args.setup_cmd, args.test_timeout,
logs_dir=os.path.join(folder, "logs"))
except (UsageError, ValueError, RuntimeError) as err: # RuntimeError: a git read failed, such as no commits
print("error: %s" % err, file=sys.stderr)
return 2
with open(tasks_path, "w", encoding="utf-8") as fh:
json.dump(doc, fh, indent=2)
fh.write("\n")
text = (json.dumps(summary_json(doc, tasks_path), indent=2) + "\n" if args.json
else render_report(doc, tasks_path))
sys.stdout.write(text)
if args.out:
with open(args.out, "w", encoding="utf-8") as fh:
fh.write(render_report(doc, tasks_path))
return 0
if __name__ == "__main__":
sys.exit(main())
# Copied from skills/evals/shared/pricing.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Token prices and cost math for the models that coding agents use.
Prices are USD per 1 million tokens, copied from the official pricing pages on
PRICES_CHECKED (see the source on each group below). Usage fields follow
Anthropic meanings, as transcripts.Usage does: `input` is uncached input,
`cache_read` and `cache_write` are separate, and `output` already includes
reasoning tokens, so reasoning is never charged twice.
Not covered, so a computed cost can differ from a bill:
- Batch discounts, fast mode, US-only inference (1.1x), and web search fees.
- GPT long-context rates (prompts over 272K input tokens cost 2x input and
cache, 1.5x output). Codex keeps its context window below that (258,400
tokens in local rollouts), so standard rates apply.
- Gemini models: no price was confirmed on Google's official pricing page, so
Gemini usage is reported as unpriced tokens.
- Prices on Amazon Bedrock and Google Vertex AI, which those platforms set.
Python 3.9+, standard library only.
"""
from __future__ import annotations
import re
from typing import Optional
PRICES_CHECKED = "2026-09-28"
def _anthropic(inp, cache_read, write_5m, write_1h, out) -> dict:
return {"input": inp, "output": out, "cache_read": cache_read,
"cache_write_5m": write_5m, "cache_write_1h": write_1h}
def _openai(inp, cached, write, out) -> dict:
# One cache-write rate; None where the pricing page lists none.
return {"input": inp, "output": out, "cache_read": cached,
"cache_write_5m": write, "cache_write_1h": write}
PRICES = {
# Anthropic API: https://platform.claude.com/docs/en/about-claude/pricing.md
# Cache reads are 0.1x input, except 0.025x on Fable 5.1 and Mythos 5.1 and 0.05x on Opus 5.5.
"claude-fable-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00),
"claude-mythos-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00), # Fable 5.1 prices; id follows the naming pattern
"claude-fable-5": _anthropic(10.00, 1.00, 12.50, 20.00, 50.00),
"claude-opus-5-5": _anthropic(4.00, 0.20, 5.00, 8.00, 20.00),
"claude-opus-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-8": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-7": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-6": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-sonnet-5-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-4-6": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-sonnet-4-5": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-haiku-4-5": _anthropic(1.00, 0.10, 1.25, 2.00, 5.00),
# OpenAI API, standard rates up to 272K input tokens: https://developers.openai.com/api/docs/pricing.md
# and the model pages https://developers.openai.com/api/docs/models/gpt-6-sol.md (also gpt-6-astra, gpt-6-luna).
"gpt-6-astra": _openai(10.00, 1.00, 12.50, 50.00),
"gpt-6-sol": _openai(2.00, 0.20, 2.50, 10.00),
"gpt-6-luna": _openai(0.10, 0.01, 0.125, 0.50),
"gpt-5.6-sol": _openai(4.00, 0.40, 5.00, 20.00),
"gpt-5.6-terra": _openai(2.00, 0.20, 2.50, 12.00),
"gpt-5.6-luna": _openai(0.20, 0.02, 0.25, 1.20),
"gpt-5.5": _openai(5.00, 0.50, None, 30.00),
"gpt-5.3-codex": _openai(1.75, 0.175, None, 14.00),
}
_CONTEXT_SUFFIX_RE = re.compile(r"\[[^\]]*\]$") # claude-opus-5-5[1m]
_DATE_SUFFIX_RE = re.compile(r"-(?:\d{8}|\d{4}-\d\d-\d\d)$") # -20251001 or -2026-09-01
# Cloud ids for the same models: Amazon Bedrock (us.anthropic.claude-opus-4-8-v1:0)
# and Google Vertex AI (claude-opus-4-8@20260101). They get the model's list price:
# the clouds set their own prices, which can differ a little (regional endpoints),
# but a list price keeps a spend cap working where "unknown" would count $0.
_PROVIDER_PREFIX_RE = re.compile(r"^(?:(?:us|eu|apac|jp|au|global|us-gov)\.)?(?:anthropic|openai)\.")
_PROVIDER_SUFFIX_RE = re.compile(r"(?:-v\d+(?::\d+)?|@\d{8}|@latest)$")
_SEPARATORS = "-_.@:"
_SIZE_WORDS = {"mini", "nano", "pro", "max", "lite", "turbo"} # another size is another price
def price_for(model) -> Optional[dict]:
"""Prices for a model id: the exact id, then the id without a cloud
provider prefix or suffix (Bedrock, Vertex AI), a context suffix, or a
date suffix, then the longest known id it starts with. A newer version
number or another size after the known id (claude-opus-5-6, gpt-5.5-pro)
does not count as a match. Returns None when the model is unknown."""
if not isinstance(model, str) or not model.strip():
return None
name = model.strip()
if name not in PRICES:
name = _PROVIDER_SUFFIX_RE.sub("", _CONTEXT_SUFFIX_RE.sub("", _PROVIDER_PREFIX_RE.sub("", name)))
name = _DATE_SUFFIX_RE.sub("", name)
if name not in PRICES:
best = ""
for known in PRICES:
if len(name) > len(known) and name.startswith(known) and name[len(known)] in _SEPARATORS:
next_word = re.split(r"[-_.@:]", name[len(known) + 1:])[0]
if not next_word.isdigit() and next_word not in _SIZE_WORDS and len(known) > len(best):
best = known
name = best
return dict(PRICES[name]) if name in PRICES else None
def _count(usage, key) -> int:
value = usage.get(key, 0) if isinstance(usage, dict) else getattr(usage, key, 0)
return value if isinstance(value, int) and not isinstance(value, bool) else 0
def cost_usd(usage, model) -> Optional[float]:
"""Dollar cost of one usage (a transcripts.Usage or a dict with the same
keys) on one model. None when the model is unknown: report those tokens as
unpriced. Cache writes without a known 1-hour part are priced at the
5-minute rate; a model with no listed cache-write price charges writes as input."""
p = price_for(model)
if p is None:
return None
write = _count(usage, "cache_write")
write_1h = min(_count(usage, "cache_write_1h"), write)
rate_5m = p["cache_write_5m"] if p["cache_write_5m"] is not None else p["input"]
rate_1h = p["cache_write_1h"] if p["cache_write_1h"] is not None else rate_5m
total = (_count(usage, "input") * p["input"] + _count(usage, "cache_read") * p["cache_read"]
+ (write - write_1h) * rate_5m + write_1h * rate_1h + _count(usage, "output") * p["output"])
return total / 1_000_000
# Copied from skills/evals/shared/safe.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Make untrusted text safe to show in a Markdown report.
Text from a repository or a session transcript can hold secrets, carry
instructions for the agent that relays a report, or break a table. redact()
masks secrets; safe_text() also makes the text one inert line; code() puts
that line inside inline code.
Python 3.9+, standard library only.
"""
import re
import unicodedata
_MASK = "[REDACTED]"
# Word edges by ASCII rules. Python's \b counts a non-ASCII letter, such as an
# e with an acute accent, as part of the word, so a key right after one would
# stay unmasked.
_B = r"(?<![A-Za-z0-9_])"
_E = r"(?![A-Za-z0-9_])"
# Characters that draw nothing, besides the format characters (Unicode category
# Cf: zero-width spaces and joiners, direction marks, the soft hyphen): the
# combining grapheme joiner, Hangul fillers, Khmer inherent vowels, Mongolian
# and other variation selectors, and tag characters. Inside a key name they
# hide the key from the patterns below while the text still reads as the key.
_INVISIBLE_RE = re.compile(r"[\u034f\u115f\u1160\u17b4\u17b5\u180b-\u180f\u3164\ufe00-\ufe0f\uffa0"
r"\U000e0000-\U000e0fff]")
# Run time: hooks clean every tool call, and a tool input can be tens of
# kilobytes, so each pattern must take time in proportion to the text. A pattern
# that can start at many places inside one long run of characters, scan to the
# end of that run, and then fail takes time that grows with the square of the
# run (or worse). The patterns below limit or stop such scans; each one says how.
_SECRET_RES = [
# PEM private key blocks, including a block cut off before its END line.
re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----.*?(?:-----END [A-Z0-9 ]*PRIVATE KEY-----|\Z)", re.S),
re.compile(_B + r"sk-ant-[A-Za-z0-9_\-]{16,}"),
re.compile(_B + r"sk-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"gh[pousr]_[A-Za-z0-9]{20,}"),
re.compile(_B + r"github_pat_[A-Za-z0-9_]{20,}"),
re.compile(_B + r"glpat-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"xox[abposr]-[A-Za-z0-9\-]{10,}"),
re.compile(_B + r"AIza[0-9A-Za-z_\-]{30,}"),
re.compile(_B + r"(?:AKIA|ASIA)[0-9A-Z]{16}" + _E),
re.compile(_B + r"[rsp]k_(?:live|test)_[0-9A-Za-z]{16,}"),
re.compile(_B + r"npm_[A-Za-z0-9]{30,}"),
re.compile(_B + r"hf_[A-Za-z0-9]{30,}"),
# A JSON web token. The first part stops before a later "-eyJ", where the
# next try starts, so no run is scanned again for each "-eyJ" in it.
re.compile(_B + r"eyJ(?:[A-Za-z0-9_]|-(?!eyJ)){8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}"),
]
# A mysql or mariadb program name, as one word.
_MYSQL = _B + r"(?:mysql|mysqldump|mysqladmin|mysqlsh|mariadb|mariadb-dump)" + _E
def _credential(before, unquoted, after="", flags=0):
"""A credential after a fixed prefix. Group 1, which redact() keeps, is
`before`, an optional quote (group 2), and `after`. The rest of the match
is masked: with a quote, everything up to the matching quote or the end of
the line, spaces included (a backslash escapes the next character, and the
closing quote stays); without one, `unquoted`."""
return re.compile(r"(%s([\"'])?%s)(?(2)(?:(?!\2)[^\\\n]|\\.)+|%s)" % (before, after, unquoted), flags)
# Keep group 1, mask the rest of the match.
_KEEP_PREFIX_RES = [
# Credentials passed as command flags: sshpass -p X, mysql -pX, curl -u user:X, --password X.
_credential(_B + r"sshpass\s+-p\s*", r"[^\s'\"]+"),
# The search for -p stops at the next program name, where the next try starts.
_credential(_MYSQL + r"(?:(?!" + _MYSQL + r")[^\n|;&])*?\s-p(?=[^\s-])", r"[^\s'\"]+", flags=re.I),
# One "=" or spaces after the flag: a user name can start with "=", so "[=\s]+"
# before it could split a run of "=" in every way.
_credential(r"(?:^|\s)(?:-u|--user)(?:=|\s+)", r"[^\s@'\"]+", after=r"[^\s:@'\"]+:"),
_credential(r"--pass(?:word|wd)?\s+", r"[^\s'\"-][^\s'\"]*", flags=re.I),
re.compile(r"(?i)(" + _B + r"(?:bearer|basic)\s+)[A-Za-z0-9._~+/\-]{16,}=*"),
# A URL scheme has at most 32 characters, so each try scans at most that far.
re.compile(r"(?i)(" + _B + r"[a-z][a-z0-9+.\-]{0,31}://[^/\s:@]+:)[^@\s/]+(?=@)"),
# A secret-like key and any non-empty value. The key is one whole name (a run
# of letters, digits, and _ . -) that holds a keyword anywhere. A try starts
# only where a name starts, and the lookahead stays inside that name, so each
# name is scanned once. "tokens" alone names a count (max_tokens: 1000), not a secret.
_credential(r"(?<![A-Za-z0-9_.\-])(?=[A-Za-z0-9_.\-]*?(?:api[_\-]?key|secret|token(?!s(?![A-Za-z0-9]))|"
r"passw(?:or)?d|pwd|credential|private[_\-]?key|access[_\-]?key))"
r"[A-Za-z0-9_.\-]+[\"']?\s*[:=]+\s*", r"[^\s\"',;}]+", flags=re.I),
]
_LONG_RUN_RE = re.compile(r"[A-Za-z0-9+=_\-]{40,}")
def _mask_long_run(m) -> str:
s = m.group(0)
if re.search(r"[0-9]", s) and re.search(r"[A-Za-z]", s):
return _MASK
return s
def _plain(text) -> str:
"""NFKC form (fullwidth and other compatibility letters become plain ones)
without the characters that draw nothing."""
if text.isascii():
return text
text = unicodedata.normalize("NFKC", text)
return _INVISIBLE_RE.sub("", "".join(ch for ch in text if unicodedata.category(ch) != "Cf"))
def redact(text) -> str:
"""Mask API keys, tokens, private keys, and long secret-like strings. The
text comes back in NFKC form without the characters that draw nothing, so a
fullwidth or split key name is masked like the plain one."""
if not text:
return ""
out = _plain(str(text))
for rx in _SECRET_RES:
out = rx.sub(_MASK, out)
for rx in _KEEP_PREFIX_RES:
out = rx.sub(lambda m: m.group(1) + _MASK, out)
return _LONG_RUN_RE.sub(_mask_long_run, out)
def safe_text(text, limit=160) -> str:
"""Make untrusted text (a command, a path, a claim, a tool description) safe
to show in a markdown report: secrets masked, control characters and line
breaks turned into spaces, backticks and table pipes replaced, runs of
spaces collapsed, and the result cut to `limit` characters. Text from a
repository or a transcript can carry instructions or break a table; this
keeps it one inert line."""
s = redact(text if isinstance(text, str) else str(text))
s = "".join(ch if ch.isprintable() else " " for ch in s)
s = " ".join(s.replace("`", "'").replace("|", "/").split())
return s if len(s) <= limit else s[: max(limit - 3, 0)] + "..."
def code(text, limit=160) -> str:
"""Untrusted text for a Markdown report, inside inline code so links, HTML,
and bare URLs stay literal; safe_text already replaces backticks and pipes.
Text that is empty after cleaning shows as `(empty)`, because a bare pair
of backticks renders as two stray backticks."""
return "`%s`" % (safe_text(text, limit) or "(empty)")