Find the update where your coding agent started working differently
Splits your own Claude Code or Codex sessions by version, model, or week, measures how the agent works in each, and places the change at the update, or within the span of versions, where the numbers moved, along with anything else that changed then.
What you get
A sample report. Every number and version in it is invented; your agent prints its own.
**After Claude Code `2.1.270`, your agent reads 41% less before it edits and gets interrupted twice as often.**
Claude Code, last 90 days, split by version: 2,412 turns in 318 sessions across 21 versions, with 14 updates
tested. Each session counts once: the test compares one value per session, and a change is flagged when p is
under 0.01 after adjusting for the number of tests and the change is at least 20%.
## Flagged changes
| Where | What changed | Before (per session) | After (per session) | Change | Usually means | Compared | Sessions (before, after) | Adjusted p |
|-----------------|----------------------------------------------|----------------------|---------------------|--------|---------------|------------------------|--------------------------|------------|
| after `2.1.270` | Reads before the first edit (session median) | 5 | 3 | -41% | worse | `2.1.268` vs `2.1.270` | 34, 29 | 0.0021 |
| after `2.1.270` | Interrupts per 100 turns (session mean) | 4.1 | 8.3 | +102% | worse | `2.1.268` vs `2.1.270` | 34, 29 | 0.0068 |
## What else changed at the same point
These numbers show what changed, not why: the kind of work may have changed too.
- after `2.1.270`: One project, `~/code/api`, holds 64% of the turns after. Run again with
`--project '~/code/api'` to check the change within that project alone.
## By version
| Version | Days | Sessions | Turns | Reads before edit | Reads per edit | Interrupts/100 | Cost/turn |
|-----------|--------------------------|----------|-------|-------------------|----------------|----------------|-----------|
| `2.1.262` | 2026-08-30 to 2026-09-06 | 19 | 141 | 5 | 4.2 | 3.5 | $0.184 |
| `2.1.268` | 2026-09-06 to 2026-09-12 | 15 | 118 | 5 | 4.0 | 4.2 | $0.191 |
| `2.1.270` | 2026-09-12 to 2026-09-19 | 29 | 244 | 3 | 2.6 | 8.3 | $0.163 |
## Notes
- Left out `2.1.199`: it ran 2026-09-01 to 2026-09-18, after newer versions, which usually means a second
install such as the desktop app or an SDK script. Run with --by week to see it in time order.
- Not tested: `2.1.265` (4 sessions): each side of an update needs 20 sessions and 30 turns, even with
neighbors added.
The real report has a column for each of the eleven numbers (reads before the first edit, reads per edit, edits to files not read earlier, edits per turn, failed tool calls, interrupts, corrections, output tokens, reasoning share, tool calls, and cost per turn). When the test had to borrow neighboring versions, "Where" reads "between 2.1.260 and 2.1.270": the change is placed within that span. With --svg it also draws one small line chart per flagged number, and with --json it prints the same result for a program.
Manual: copy skills/regression-finder/ 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: "Did Claude Code get worse after the last update? Check my sessions."
The agent runs the check, leads with the headline, and reruns it to rule out what else changed at the same point, such as a new model or different projects. To run the script yourself:
Add --harness codex for Codex, --by model or --by week to split another way, --project <folder> to compare like with like, --since 180d for a longer window, --svg chart.svg for the chart, and --fail-on worse to exit 1 when a flagged change usually means worse (useful in a scheduled job).
How it works
Turns. Each prompt you typed, plus everything the agent did until your next prompt, is one turn. Work done by subagents counts in the turn that started it.
Eleven numbers, two of them measured as in issue #42796 (reads per edit, and edits to files not read earlier in the session) and the rest this skill's own, such as reads before the first edit, interrupts, and cost. The metrics reference defines each one.
Versions in time order. Turns are grouped by the harness version recorded with each prompt (Claude Code) or session (Codex). A version that ran after newer ones, which usually means a second install such as the desktop app or an SDK script, is left out and named in the notes.
One test per update. The sessions just before an update are compared with the sessions just after it. An update with few sessions borrows its neighbors until each side has 20 sessions and 30 turns, and then the change is placed within the span of versions it covers.
Each session counts once. Turns from one session resemble each other, so the test (Mann-Whitney U, with Fisher's exact test for rare events) runs on one value per session, and the sizes shown come from the same values. In our simulations of histories with no real change, testing turns one by one raised a false alarm in 147 of 150 of them; the full method raised one in about 1 of 100, also when a second install ran an old version the whole time.
Flagged means: p under 0.01 after adjusting for the number of tests (Benjamini-Hochberg, over tests that could reach significance, with overlapping windows counted once), a change of at least 20% per session, and both pointing the same way. The statistics reference has the details and the limits.
Confounders named. For each flagged update the report checks whether the model, the projects, the length of sessions, or the share of scripted runs changed at the same point, or whether both sides ran at the same time. It says so in the headline and suggests the rerun that separates them.
Works with
Harness
Split by version
Split by model or week
Support
Claude Code
yes, from the version on each prompt record
yes
full; smoke-tested on macOS
Codex
yes, from cli_version (once per session file)
yes
full; smoke-tested on macOS
Gemini CLI
no: its transcripts record no version
yes
tested on synthetic files and a small real history
OpenCode
yes, from the session's version
yes
not verified on a real install
Cursor
no
no
keeps no usable transcripts
It needs Python 3.9 or newer and nothing else to install. It has been run on macOS.
Limits
It shows what changed, not why. A new model, different projects, or scripted runs can move the numbers as much as an update; the report names the ones it can see, and it cannot see a change in how hard your tasks were.
Small changes need many sessions to show. Even a 50% drop in reads before editing is rarely found with a few sessions a day. With fewer than 20 sessions on each side of an update, it reports that the history is too short instead of guessing.
A change found by borrowing neighboring versions is placed within a span of versions, not at one version.
A second install on an old version, or a rollback to an old version after an update, is left out of the version split and named in the notes; --by week shows those sessions in time order.
It counts edits made through edit tools. Files changed by shell commands (sed -i, cat > file) are left out of the edit counts.
Corrections come from a short list of phrases, so many corrections go uncounted; the same list applies to every version.
Claude Code deletes transcripts after 30 days by default (cleanupPeriodDays), except sessions from Claude Desktop, so a 90-day window may hold 30 days. The report says so when it happens.
Codex records its version once per session file, so a Codex session resumed after an update keeps its first version.
Costs use list prices checked on 2026-09-28; turns with a model that has no known price are left out of the cost number.
Privacy
Read: session files under ~/.claude/projects/, ~/.codex/sessions/, ~/.gemini/tmp/, or the OpenCode database, only those changed within the window. Prompt text is read on your machine only to spot corrections.
Printed: counts, rates, token totals, costs, version numbers, model names, and project folders (your home folder shown as ~). Never prompt text, commands, or file contents.
Sent: nothing. It makes no network calls.
Written: nothing, unless you pass --out or --svg, which write the one file you name.
harness-test-drive: compare harnesses on tasks from your own git history.
The method comes from anthropics/claude-code#42796 by Stella Laurenzo, a one-off analysis of 6,852 session files that measured the same kinds of numbers. claude-session-analyzer repeats that analysis for Claude Code, split into two date halves.
Statistics: Mann and Whitney (1947) for the rank test, Fisher (1935) for the exact test, Tarone (1990) for setting aside tests that cannot pass, and Benjamini and Hochberg (1995) for the adjustment across tests.
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: regression-finder
description: >-
Regression check for coding agents: shows how the agent behaved before and
after each harness update, model switch, or week in the user's own Claude
Code or Codex history, and finds the point where it changed. Use when the
user says the agent's quality dropped or it got worse, dumber, lazier, or
degraded since an update or since they upgraded; asks whether a new Claude
Code or Codex version or model made it worse than the old one; wants to know
which release, version, or week it regressed in; or wants numbers to report a
regression, such as reads before edits, interruptions, and corrections per
version. Runs locally and reads transcripts only; nothing goes over the
network.
license: MIT
metadata:
author: "Ryan Alberts"
version: "1.0.0"
source: "https://github.com/RyanAlberts/best-of-Agent-Harnesses"
---
# Regression finder
When a coding agent seems worse after an update, the user's own session history can show whether it changed and when. This skill splits the user's Claude Code or Codex sessions by harness version, model, or week, measures the same behavior in each (reads before the first edit, reads per edit, edits to files not read first, interrupts, corrections, failed tool calls, output, cost), and places the change at the update, or within the span of versions, where the numbers moved, along with anything else that changed at the same point. It reads local transcripts, prints counts and rates, never prints prompt text, and sends nothing anywhere.
## When to use
- The user says the agent got worse, dumber, or lazier after an update, or asks "is it just me?"
- The user asks whether a new Claude Code or Codex version, or a new model, changed how the agent works.
- The user wants to know which release or week a regression started.
- The user wants evidence for a bug report about a regression, like [anthropics/claude-code#42796](https://github.com/anthropics/claude-code/issues/42796).
## When not to use
- Where tokens and money go in general (re-reads, cache rebuilds, oversized results): use `session-waste-report`.
- Stopping a session that loops or overspends right now: use `runaway-guard`.
- Comparing two harnesses on the same tasks: use `harness-test-drive`.
- Checking whether "tests pass" claims were true: use `claim-check`.
- Finding which instruction-file rules the agent breaks: use `rules-to-guards`.
- Cursor: it keeps no usable transcripts, so there is nothing to measure.
## What it reads
Tell the user this when they ask what the check looks at:
- **Read**: the session files each harness keeps (Claude Code under `~/.claude/projects/`, Codex under `~/.codex/sessions/`), only those changed within the window. Prompt text is read on this machine only to spot corrections such as "no, that's wrong".
- **Printed**: counts, rates, token totals, versions, model names, and project folders. Never prompt text, commands, or file contents.
- **Written**: nothing, unless the user asks for `--out` or `--svg`, which write the one file named.
- **Sent**: nothing. It makes no network calls.
## Steps
`<skill-dir>` means the folder that holds this SKILL.md (Claude Code shows it as the skill's base directory). Keep the quotes around the script path in every command: skill folders can sit under paths with spaces.
1. **Run the default check.** Pick the harness the user asks about (`claude-code` by default, or `codex`) and run:
```bash
python3 "<skill-dir>/scripts/regress.py" --harness claude-code
```
It reads the last 90 days and splits by harness version. Match the flags to the question:
| The user says | Add |
|---|---|
| "since the last Codex update" | `--harness codex` |
| "since I switched to the new model" | `--by model` |
| "worse these past few weeks" | `--by week` |
| "in my api repo" | `--project <that folder>` |
| "since the spring" | `--since 180d` |
It takes a few seconds per thousand sessions and exits 0 even when it finds nothing. Done when the output starts with a bold headline, or you have told the user the exact error.
2. **Answer the user's own question first.** If the user named an update or a time such as last week, find it in the By version table first. If it was not tested, lead with that (for example: 2.1.280 had 7 sessions, and the test needs 20 on each side), then give the headline as background, not as the answer. The notes list every update that was not tested, with its sessions.
Then read the headline. It is one of these kinds:
- A flagged change: "After Claude Code `2.1.270`, your agent reads 41% less before it edits...", or "After an update between Claude Code `2.1.260` and `2.1.270`, ..." when the test had to borrow neighboring versions. The change lies somewhere in that span; say the span, not one version. Go to step 3.
- "No behavior change passed the test across ...": the numbers wobble but nothing passed. Say so plainly and give the session counts; go to step 5.
- "No lasting behavior change passed the test ...; 1 version stands out": one version differs from the versions on both sides of it. Report the "Stands out" line as it is.
- "Not enough history to test an update yet": fewer than 20 sessions on each side of every update. Offer, in this order, a longer window (`--since 180d`, when the history goes back that far), a split by time (`--by week`), and last a lower bar (`--min-sessions 12`, the floor). A lower bar tests more updates, but each test can only catch larger changes.
- "No ... found", "records no version", "ran on one ...", or "Not enough history to compare models": nothing to compare yet. Say so, and name what the notes list as left out.
Done when the user's own update or time is answered, or you know which kind of headline it is.
3. **Follow each confounder.** A confounder is anything else that changed at the same update and could explain the numbers. The section "What else changed at the same point" lists them, and the headline ends with "but ... changed at the same point" or "both sides were in use at the same time" when they exist. For each line:
- The model changed: run again with `--by model` and see whether the change follows the model instead.
- The harness version changed (in a model or week split): run again with `--by version`.
- The work moved between projects, or one project holds most turns: run again with the `--project` argument the line prints. Copy the --project argument exactly as printed, quotes included.
- Sessions changed shape, or the share of scripted runs changed: tell the user the kind of work changed (for example scripted runs against long conversations), so the update may not be the cause.
- Both sides were in use at the same time, or both models were: tell the user two installs or two models ran side by side, so the difference may come from what each was used for.
Done when each confounder line has a rerun result or a one-sentence explanation for the user.
4. **Offer the chart** when something is flagged and the user wants to see it or share it:
```bash
python3 "<skill-dir>/scripts/regress.py" --harness claude-code --svg regression.svg
```
It writes one small line chart per flagged number to the path given, and nothing else. Done when you have given the user the path, or the user declined.
5. **Report** in the shape below.
## Read the results
- **Headline**: the update with the most flagged changes, its two most telling changes, and the confounders at that update.
- **Flagged changes**: one row per change. "Where" is "after X" when the test compared X with the version just before it, or "between X and Y" when it borrowed neighbors: the change is placed within that span, not at one version. Before and after are per-session values, the same values the test ranks (the median or mean of one value per session, as the row says). The sessions on each side are the sample size to quote.
- **A change is flagged** when p is under 0.01 after adjusting for the number of tests, the per-session value moved at least 20%, and both point the same way. Windows that overlap and show the same change count as one test and one row.
- **Stands out**: a version (or week) that differs from the ones on both sides of it. It is reported once and is not a lasting change.
- **By version (or model, or week)**: every number for every slice, turn by turn, including slices too small to test. "n/a" means that number cannot be measured there, for example reasoning tokens that Claude Code did not record.
- **Notes**: what was left out and why: versions that ran after newer ones (a second install, such as the desktop app or an SDK script), updates not tested and their sessions, comparisons that could not reach significance, rare events seen too seldom to test, models with too little data, turns older than the window, and a history shorter than the window.
- `--json` prints the same result for a program: `headline`, `totals`, `thresholds`, `slices`, `tests` (every comparison, flagged or not), `flagged`, `stand_outs`, `confounders`, `untested`, `left_out`, and `notes`.
Open `references/metrics.md` to explain what a number measures and why it matters, and `references/statistics.md` when the user asks how sure the result is or why a visible change was not flagged.
## Report to the user
1. The headline, verbatim, in bold.
2. The flagged changes as a short table: where, what changed, before and after (per session), change, sessions before and after. Six rows at most, worse changes first.
3. The confounders, one line each, with what the rerun in step 3 showed.
4. Two or three next actions that fit the result:
- Read what changed in the releases the change is placed in: the [Claude Code changelog](https://github.com/anthropics/claude-code/blob/main/CHANGELOG.md) or the [Codex releases](https://github.com/openai/codex/releases).
- Narrow the check with `--project <path>` or `--by model`.
- If it holds after the reruns, file it upstream with the numbers: `--json` for the data and `--svg` for the chart.
- For tokens and money lost to habits rather than updates, run `session-waste-report`.
An example of the shape (the numbers are invented):
> **After Claude Code `2.1.270`, your agent reads 41% less before it edits and gets interrupted twice as often.**
>
> | Where | What changed | Before | After | Change | Sessions |
> |---|---|---|---|---|---|
> | after `2.1.270` | Reads before the first edit (session median) | 5 | 3 | -41% | 34, 29 |
> | after `2.1.270` | Interrupts per 100 turns (session mean) | 4.1 | 8.3 | +102% | 34, 29 |
>
> Nothing else changed at that update: same model, same projects. Next: read the `2.1.270` entry in the Claude Code changelog, and if it matches, file it with `--json` and `--svg`.
When nothing was flagged, say what was compared (versions, sessions, turns) and that small histories cannot show small changes; quote no percentages as findings.
Quote versions, models, and paths exactly as the report prints them, inside inline code: it shows the home folder as `~`, and it has already made any text taken from transcripts safe to display.
## Files
- `scripts/regress.py`: the check. Flags: `--harness`, `--by version|model|week`, `--since 90d`, `--project`, `--min-turns 30`, `--min-sessions 20` (at least 12), `--svg <path>`, `--json`, `--out <path>`, `--fail-on worse|any` (exit 1 when a flagged change usually means worse, or when anything is flagged).
- `scripts/transcripts.py`, `scripts/pricing.py`, `scripts/safe.py`: the shared reader for session files, the price table, and the text cleaner that puts transcript text in the report inside inline code; several skills in this repository use them.
- `references/metrics.md`: each number's definition and why it matters, next to the method of issue #42796.
- `references/statistics.md`: the test, the thresholds, the minimum samples, how changes are placed, and the limits.
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.
# What each number measures
Every number starts from **turns**: one prompt the user typed, plus everything the agent did until the next typed prompt. Work done by subagents (helper agents the main agent starts) counts in the turn that started them. Messages the harness adds on its own, such as context files, reminders, and hook output, do not start a turn.
The numbers follow the analysis in [anthropics/claude-code#42796](https://github.com/anthropics/claude-code/issues/42796) (opened 2026-04-02, 3,287 reactions when checked on 2026-09-28). Its author mined 6,852 Claude Code session files and 234,760 tool calls from January 30 to April 1, 2026, and tied a quality drop to a change in how the agent worked: it read less before editing and was interrupted more. Two of the numbers below are measured the way that issue measured them; the others are this skill's own, and each section says which.
Each number is shown two ways:
- **Per session**, in the flagged changes: one value per session (the mean of its turns, or for ratios and shares the session's own total ratio), then the median or the mean of those session values. The test ranks the same per-session values, so the size of a change and its p value come from the same data.
- **Turn by turn**, in the By version table: every turn of the slice pooled together.
| Number | Per session | Usually a bad sign when |
|---|---|---|
| Reads before the first edit | median of sessions | it falls |
| Reads per edit | median of sessions | it falls |
| Edits to files not read earlier in the session (%) | mean of sessions | it rises |
| Edits per turn | median of sessions | no clear direction |
| Failed tool calls (%) | mean of sessions | it rises |
| Interrupts per 100 turns | mean of sessions | it rises |
| Corrections per 100 prompts | mean of sessions | it rises |
| Output tokens per turn | median of sessions | no clear direction |
| Share of output spent reasoning (%) | mean of sessions | it falls |
| Tool calls per turn | median of sessions | no clear direction |
| Cost per turn | median of sessions | it rises |
## Reads and edits
A **read** is a call that looks at files without changing them:
- File-reading and search tools: Claude Code `Read`, `Grep`, `Glob`, and `LS`; Codex `view_image`; Gemini CLI `read_file`, `glob`, and its search tools; OpenCode `read`, `grep`, `glob`, and `list`.
- Shell commands that only inspect, such as `cat`, `head`, `tail`, `nl`, `wc`, `ls`, `tree`, `find` (without `-delete` or `-exec`), `fd`, `rg`, `grep`, `sed` without `-i`, `awk`, `jq`, `diff`, and `git log`, `show`, `diff`, `status`, `blame`, or `grep`. A command counts only when every part of it inspects: `cd src && rg parse` is a read, `rg parse > out.txt` and `rg parse; npm test` are not.
- Codex writes most shell work as a short script (code mode). The script counts as a read when every command it passes to `exec_command` is a read.
An **edit** is a call to a tool that edits or writes a file: Claude Code `Edit`, `MultiEdit`, `Write`, and `NotebookEdit`, Codex `apply_patch`, and the Gemini CLI and OpenCode equivalents. Files changed through shell commands (for example `sed -i`) are not counted as edits, so a harness that moves editing into the shell will look like it edits less.
### Reads per edit (as in #42796)
All reads in a session divided by all its edits, for sessions with at least one edit. #42796 measured this ratio over all tool calls in a period and saw it fall from 6.6 reads per edit to 2.0.
### Edits to files not read earlier in the session (as in #42796)
The share of edits made to a file the session had not read before. A file counts as known once any read in the session names it (a `Read` call, or a shell read such as `cat src/a.py` or `sed -n 1,80p src/a.py`) or once the session has edited it. Files are matched on the last two parts of their path, so `/work/app/src/a.py` and `src/a.py` match. Whole-file writes (`Write`), which often create new files, are left out. In #42796 this share rose from 6.2% to 33.7% of edits, and those edits broke surrounding code.
### Reads before the first edit (this skill's own measure)
The reads a turn makes before its first edit, for turns that edit. It is a proxy for "does the agent look before it changes something" at the level of one request; #42796 did not measure it this way.
## Edits per turn
How many edit calls a turn makes. More edits can mean bigger tasks or trial and error on the same file (#42796 lists repeated edits to one file as a symptom); fewer can mean smaller tasks or an agent that stops early. Read it together with the other numbers.
## Failed tool calls
A call fails when the tool reports an error or a shell command exits with a nonzero code. Calls blocked by a person, a permission rule, a hook, or an automatic reviewer do not count as failures, and neither do calls the user interrupted: those are choices, not failures.
## Interrupts per 100 turns
An interrupt is the user stopping the agent: Claude Code records `[Request interrupted by user]` (Escape), and Codex records an aborted turn. This skill counts them per 100 turns; #42796 counted them per 1,000 tool calls, where they rose from 0.9 to 11.4. The two rates are not comparable number for number.
## Corrections per 100 prompts
A correction is a typed prompt that tells the agent it got something wrong. The script matches a short list of phrases chosen for precision over recall, so it misses many corrections but rarely counts a prompt that is not one. The same list applies to every group, so missed phrases affect every group alike.
- At the start of a prompt: "no," "nope," "wrong," "that's wrong," "that's not what I asked," "undo that," "revert that," "why did you," "you didn't," "you forgot," "you broke," "I said," "I told you," "that didn't work," "it still doesn't work," "still failing," "try again," "stop," "wait," "don't do that."
- Anywhere in a prompt: "not what I asked," "I already told you," "you're not done," "read the file first," "you didn't read."
In #42796 the share of prompts with frustration words rose from 5.8% to 9.8%.
## Output tokens and the share spent reasoning
Output tokens include the model's reasoning (thinking) tokens and the output of subagents. The reasoning share is reasoning tokens divided by output tokens. Codex records reasoning tokens for every response. Claude Code records them only for some models and versions, so a slice where most turns record none shows "n/a", and its turns are left out of both sides of every reasoning comparison instead of counting as 0%. #42796 estimated that thinking depth had fallen by about two thirds by late February 2026, before the quality drop was reported in March.
## Tool calls per turn
All tool calls in the turn, subagents included. More calls can mean more thorough work or more thrashing; read it with the reading numbers and the failure rate.
## Cost per turn
Tokens times the list prices in `scripts/pricing.py` (checked 2026-09-28): uncached input, cache reads, cache writes, and output, each at its own rate. A turn that used a model without a known price is left out of this number, and the report names those models. On a subscription plan you do not pay per token, but the number still tracks how much work a turn takes.
## Where the numbers come from
`scripts/transcripts.py` reads the session files: Claude Code `~/.claude/projects/`, Codex `~/.codex/sessions/`, Gemini CLI `~/.gemini/tmp/`, and the OpenCode database (not checked on a real install). It counts each API response once, drops copies that a forked or resumed session repeats, and skips events older than the window.
The harness version and the way Claude Code was started (the entrypoint: the command line, the desktop app, or an SDK script) come from each Claude Code prompt record, so a session resumed after an update counts its later turns under the newer version. Codex records its version once per session file (`cli_version`), so a Codex session keeps the version it started with. Gemini CLI records no version: split its sessions by model or week instead.
# How a change gets flagged
The goal is to name real shifts and stay quiet when the numbers only wobble. Every rule below serves one of the two.
## Each session counts once
Turns from one session resemble each other: a long refactoring session reads a lot in every turn, a quick question session reads little. On one developer's history we checked, the intraclass correlation (how much of the spread comes from which session a turn belongs to) was about 0.2 to 0.3 for reads before the first edit, tool calls, and output tokens. Testing turns as if each were independent treats 40 turns of one unusual session as 40 pieces of evidence: in our simulations of histories with no real change, a turn-by-turn test flagged a change in 147 of 150 of them.
So the unit is the session. For each number, each session gets one value: the mean of its turns' values, or for ratios and shares (reads per edit, edits to unread files, failed calls, reasoning) the session's own total ratio. A session that runs across an update counts once on each side. The size of a change, its 20% rule, and its p value all come from these per-session values: the median of sessions for counts and tokens, the mean of sessions for rates and shares.
## The test
Two-sided Mann-Whitney U test (Mann and Whitney, 1947) on the per-session values, with the normal approximation, the correction for ties, and a continuity correction of 0.5. These are the defaults of R's `wilcox.test(exact = FALSE)` and of the asymptotic method in `scipy.stats.mannwhitneyu`; the tests in `skills/evals/regression-finder` check the implementation against published values from both.
Rare events need more care. When one value (usually 0) holds more than half of the sessions, the normal approximation gives p values far too small: 2 sessions with an interrupt out of 20, against none out of 200, gives 0.0000078, while the exact answer is 0.0079. There the script also runs Fisher's exact test (Fisher, 1935) on the sessions away from that value against the sessions at it, and keeps the larger p. It tests such a number only when the busier side has at least 5 sessions away from the common value.
## Versions in time order
Versions sort by number (semantic versioning: `2.1.9` before `2.1.10`, a pre-release such as `0.155.0-alpha.16` before `0.155.0`). A second install can break that order: a desktop app or an SDK script still on an old version runs next to the updated command line, and its version number sorts among versions that ran months earlier. The script keeps the longest run of versions whose median turn time rises with the version number, counted in versions first and sessions second, and leaves the rest out with a note that names them and the dates they ran. Counting sessions first would keep a busy second install and drop the versions of the install that was actually being updated. `--by week` shows every session in time order, second installs included.
## The windows around each update
For each update, the script takes the version just before it and the version just after it, and adds the next nearest versions on each side until a side has at least 30 turns from at least 20 sessions (`--min-turns`, `--min-sessions`, which cannot go below 12). An update without enough on both sides is not tested, and the report lists it with its sessions. Weeks are ISO weeks in UTC and work the same way.
Models run side by side more often than they replace each other, so a model is compared with the model used before it only within two weeks of its first use: the earlier model's sessions from the two weeks before, against the new model's sessions from the two weeks after. When the earlier model still ran 30% or more of those later turns, the report says both were in use at the same time.
Within a window, a number is tested only when each side has at least 12 sessions with a value for it.
## Which tests count
1. **Tests that could never pass are set aside** (Tarone, 1990). With few sessions or rare events, even the most one-sided arrangement of the values cannot give a small p. The script finds, for each test, the smallest p its values could give, and keeps only the tests that could reach significance given how many are kept. The rest are counted in a note and do not dilute the adjustment.
2. **Overlapping windows are one test.** Windows that share versions and point the same way on one number see the same change. They form a group, the group's smallest p stands for it, and the adjustment counts groups, not windows.
3. **The adjustment**: Benjamini-Hochberg (Benjamini and Hochberg, 1995) over the groups. A change is flagged when the adjusted p is under 0.01, the per-session value moved at least 20% (a change from zero counts), and the test and the value point the same way.
## Where a change is placed
A window that borrowed neighbors cannot say which of its versions brought the change, so the report places it within a span: "between X and Y", from the second version of the before side to the last version of the after side. A window that compared one version with the one before it says "after X". Among the windows of a flagged group, the report names the one whose own boundary versions hold the most sessions, then the one with the smaller p, and it never names a version with fewer than 3 sessions on its own. In our simulations of a real change next to a version with 2 sessions, naming the smallest p put the change at the wrong version in 10 of 117 flags; with spans, 0 of 131 fell outside the span shown.
When a flag at one update is followed, within its after side, by the opposite flag on the same number, the versions between differ from both neighbors. The report says that version stands out from the versions around it, once, and headlines neither flag.
## What else changed
For each flagged update the report checks, on the same two sides, whether the model mix changed, whether the work moved between projects or one project holds most of it, whether sessions changed shape (the median length, or the share of one-turn sessions by 25 points or more), whether the share of scripted SDK runs changed by 25 points or more, and whether the two sides were in use at the same time. It names each one it finds in the headline and in its own section, with the rerun that separates it.
## How well it works
In our simulations (histories with 1-turn to 20-turn sessions, each session with its own habits, and edits in about 60% of turns):
- **No real change**: 1 run in 100 flagged anything, with 6 or with 20 sessions per version, and the same with a second install on an old version running scripts the whole time (before the time-order rule, such an install produced a flag in nearly every run).
- **Rare events, 200 sessions against 24**: 1 run in 200 flagged, from a 1-in-1,700 draw; the normal approximation alone flagged 3.
- **A real drop in reads before the first edit**: a 70% drop was found, within the span shown, in about 55 of 100 runs; a 50% drop in about 8 of 100. None was placed outside its span.
Smaller changes need more sessions than most people have. "No change passed the test" often means "not enough evidence yet", not "nothing changed".
## What the numbers cannot tell you
- **Why.** A flag says the numbers moved in a span of updates, not that an update moved them. The kind of work may have changed too; the report names the changes it can see, and it cannot see how hard the tasks were.
- **Small histories.** With few sessions the test cannot reach p under 0.01, whatever the change: 5 sessions against 5 cannot.
- **Retention.** Claude Code deletes transcripts older than `cleanupPeriodDays` (30 days by default), except sessions started or continued in Claude Desktop. Gemini CLI keeps 30 days by default. The report says when the history is shorter than the window you asked for.
- **Codex versions** are recorded once per session file, so a Codex session keeps the version it started with.
## Sources
- Mann, H. B., and Whitney, D. R. (1947). On a test of whether one of two random variables is stochastically larger than the other. *Annals of Mathematical Statistics* 18(1), 50-60.
- Fisher, R. A. (1935). *The Design of Experiments*. Oliver and Boyd. (The exact test for a two-by-two table.)
- Tarone, R. E. (1990). A modified Bonferroni method for discrete data. *Biometrics* 46(2), 515-522.
- Benjamini, Y., and Hochberg, Y. (1995). Controlling the false discovery rate: a practical and powerful approach to multiple testing. *Journal of the Royal Statistical Society, Series B* 57(1), 289-300.
- The analysis this skill repeats on your own history: [anthropics/claude-code#42796](https://github.com/anthropics/claude-code/issues/42796).
# 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
#!/usr/bin/env python3
"""Find the harness version, model, or week where a coding agent's behavior changed.
Reads the user's own session transcripts (through transcripts.py), cuts them
into user turns, measures the same behavior numbers for every turn, splits the
turns by harness version, model, or ISO week, and tests each update: the
sessions just before it against the sessions just after it, each session
counting once. Read-only; nothing leaves the machine. Python 3.9+, standard
library only.
"""
from __future__ import annotations
import argparse
import calendar
import datetime
import json
import math
import os
import re
import shlex
import statistics
import sys
import time
from dataclasses import dataclass, field
from typing import Optional
from xml.sax.saxutils import escape
sys.dont_write_bytecode = True # leave no cache files in the skill folder
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pricing # noqa: E402
import transcripts # noqa: E402
from safe import code, safe_text # noqa: E402
# ---------------------------------------------------------------------------
# User corrections (references/metrics.md). High precision over recall: the
# same list applies to every group, so a missed phrase shifts no comparison.
# ---------------------------------------------------------------------------
_CORRECTION_START = re.compile(r"""^(?:(?:please|hey|ok|okay|hmm|um|so)[\s,]+)?(?:
no+\s*[,.!;:]|no+\s*$|nope\b|nah\b
| (?:that|this|it)(?:'s|\s+is|\s+was)\s+(?:wrong|incorrect|not\s+(?:right|correct|what\s+i))
| wrong\s*[,.!;:]|wrong\s+(?:file|one|place|branch|function|folder|directory|approach|way|command|path)\b
| not\s+what\s+i\b
| (?:undo|revert)\s+(?:that|this|it|your|those|these|what\s+you)\b
| why\s+(?:did|would|are|do|have)\s+you\b
| you\s+(?:didn't|did\s+not|forgot|missed|broke|ignored|skipped|deleted|removed|keep|still|never
|haven't|have\s+not)\b
| i\s+(?:already\s+)?(?:said|told\s+you|asked(?:\s+you)?)\b
| (?:that|this|it)\s+(?:still\s+)?(?:didn't|did\s+not|doesn't|does\s+not|isn't|is\s+not)\s+work
| still\s+(?:broken|failing|fails|wrong|not|the\s+same|doesn't|does\s+not|isn't|getting|seeing)\b
| (?:try|do\s+it)\s+again\b
| stop\s*[,.!;:]|stop\s+(?:doing|that|it|changing|editing|adding|making|guessing|asking)\b
| wait\s*[,.!;:]
| (?:don't|do\s+not)\s+do\s+that\b
)""", re.I | re.X)
_CORRECTION_ANYWHERE = re.compile(
r"\b(?:not\s+what\s+i\s+(?:asked|wanted|meant|said)|i\s+(?:already|just)\s+told\s+you"
r"|you(?:'re|\s+are)\s+not\s+done|read\s+the\s+(?:file|code|docs?)\s+first"
r"|you\s+(?:didn't|did\s+not)\s+(?:read|check|test|run|look))\b", re.I)
def is_correction(text) -> bool:
"""True when a typed prompt tells the agent it got something wrong."""
s = str(text or "").replace("\u2019", "'").strip()
return bool(s) and bool(_CORRECTION_START.match(s) or _CORRECTION_ANYWHERE.search(s))
# ---------------------------------------------------------------------------
# Tool-call classes (references/metrics.md): "research" reads or searches
# files, "edit" changes a file through an edit tool, "other" is the rest.
# ---------------------------------------------------------------------------
# Programs that only look at files, including the filters a pipeline ends in.
_READERS = {"cat", "head", "tail", "less", "more", "bat", "nl", "wc", "file", "stat", "ls", "tree",
"find", "fd", "rg", "grep", "egrep", "fgrep", "ag", "ack", "jq", "yq", "diff", "cmp", "du",
"readlink", "realpath", "sed", "awk", "git", "sort", "uniq", "cut", "tr", "column"}
_NEUTRAL = {"cd", "pushd", "popd", "echo", "printf", "pwd", "true", ":", "export", "set", "unset"}
_WRAPPERS = {"time", "command", "builtin", "nohup", "env"}
_GIT_READS = {"log", "show", "diff", "status", "blame", "grep", "ls-files", "ls-tree", "rev-parse",
"describe", "shortlog", "reflog", "cat-file"}
_FIND_ACTIONS = {"-delete", "-exec", "-execdir", "-ok", "-okdir", "-fprint", "-fprintf", "-fls"}
_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")
def _split_shell(command):
"""Split a command at unquoted ;, &&, ||, |, & and line breaks.
Returns (segments, writes): `writes` is True when output is redirected to
a file other than /dev/null. Returns None when the quotes do not balance."""
segments, cur, quote, writes, i, n = [], [], None, False, 0, len(command)
while i < n:
ch = command[i]
if quote:
cur.append(ch)
if ch == "\\" and quote == '"' and i + 1 < n:
cur.append(command[i + 1])
i += 1
elif ch == quote:
quote = None
elif ch == "\\" and i + 1 < n:
cur.extend((ch, command[i + 1]))
i += 1
elif ch in "'\"":
quote = ch
cur.append(ch)
elif ch == "#" and (not cur or cur[-1].isspace()):
while i + 1 < n and command[i + 1] != "\n":
i += 1
elif ch == ">":
cur.append(ch)
j = i + 1
if j < n and command[j] == ">":
j += 1
if j < n and command[j] == "&": # 2>&1, >&2: a copy of another output, not a file
i = j
cur.append("&")
else:
while j < n and command[j] in " \t":
j += 1
k = j
while k < n and not command[k].isspace() and command[k] not in ";|&<>()":
k += 1
if command[j:k] != "/dev/null":
writes = True
elif ch == "&" and ((cur and cur[-1] == ">") or command[i + 1:i + 2] == ">"):
cur.append(ch) # &> and >& are redirections, not separators
elif ch in ";|&\n":
segments.append("".join(cur))
cur = []
if ch in "&|" and command[i + 1:i + 2] == ch:
i += 1
else:
cur.append(ch)
i += 1
if quote:
return None
segments.append("".join(cur))
return [s.strip() for s in segments if s.strip()], writes
def _segment_class(words) -> str:
"""'read', 'neutral', or 'other' for one simple command, split into words."""
while words and (_ASSIGNMENT_RE.match(words[0]) or words[0] in _WRAPPERS):
words = words[1:]
if not words:
return "neutral"
program, args = os.path.basename(words[0]), words[1:]
if program in _NEUTRAL:
return "neutral"
if program not in _READERS:
return "other"
if program == "sed" and any(a == "--in-place" or (a.startswith("-") and not a.startswith("--") and "i" in a)
for a in args):
return "other"
if program == "awk" and "inplace" in args:
return "other"
if program == "find" and _FIND_ACTIONS.intersection(args):
return "other"
if program == "fd" and ({"-x", "-X", "--exec", "--exec-batch"} & set(args)):
return "other"
if program == "git":
rest = list(args)
while rest and rest[0].startswith("-"):
rest = rest[2:] if rest[0] in ("-C", "-c") else rest[1:]
return "read" if rest and rest[0] in _GIT_READS else "other"
return "read"
def shell_is_read(command) -> bool:
"""True when a shell command only inspects files: every part is a reader
(cat, rg, sed -n, git log, ...) or a no-op (cd, echo), at least one part
reads, and nothing is written to a file."""
split = _split_shell(str(command or ""))
if split is None or split[1]:
return False
classes = []
for segment in split[0]:
try:
words = shlex.split(segment)
except ValueError:
return False
classes.append(_segment_class(words))
return "read" in classes and "other" not in classes
# Codex code mode: the model writes one script that calls tools.<name>(...).
_JS_STRING = r'"(?:\\.|[^"\\])*"|\'(?:\\.|[^\'\\])*\'|`(?:\\.|[^`\\])*`'
_CODE_CMD_RE = re.compile(r"exec_command\s*\(\s*\{[^{}]*?(?:\"cmd\"|'cmd'|\bcmd)\s*:\s*(" + _JS_STRING + ")", re.S)
_CODE_TOOL_RE = re.compile(r"\btools\.([A-Za-z_][A-Za-z0-9_]*)\s*\(")
_JS_ESCAPES = {"n": "\n", "t": "\t", "r": "\r"}
def _js_unquote(literal) -> str:
if literal.startswith('"'):
try:
return json.loads(literal)
except ValueError:
pass
return re.sub(r"\\(.)", lambda m: _JS_ESCAPES.get(m.group(1), m.group(1)), literal[1:-1], flags=re.S)
def script_commands(script) -> list:
"""The shell commands a Codex code-mode script runs through tools.exec_command."""
return [_js_unquote(m.group(1)) for m in _CODE_CMD_RE.finditer(str(script or ""))]
def classify_call(call) -> str:
"""'research', 'edit', or 'other' for a transcripts.ToolCall."""
if call.kind in ("edit", "write"):
return "edit"
if call.kind in ("read", "search"):
return "research"
if call.kind != "shell":
return "other"
if call.name != "exec": # a plain shell command
return "research" if shell_is_read(call.command) else "other"
tools = set(_CODE_TOOL_RE.findall(call.command)) # a Codex code-mode script
if "apply_patch" in tools:
return "edit"
commands = script_commands(call.command)
runs = len(re.findall(r"\bexec_command\s*\(", call.command))
if (tools and tools <= {"exec_command", "view_image"} and len(commands) == runs
and all(map(shell_is_read, commands))):
return "research"
return "other"
# Files a read-only command names, for "edits to files not read earlier".
_PATH_READERS = {"cat", "head", "tail", "nl", "less", "more", "bat", "wc", "sed", "awk"}
_VALUE_OPTIONS = {"head": {"-n", "-c", "--lines", "--bytes"}, "tail": {"-n", "-c", "--lines", "--bytes"},
"sed": {"-e", "-f", "--expression", "--file"}, "awk": {"-F", "-v", "-f"}}
_SCRIPT_OPTIONS = {"-e", "-f", "--expression", "--file"} # sed or awk read their script from this, not a plain argument
_PATCH_FILE_RE = re.compile(r"\*\*\* (?:Update|Add|Delete) File: ([^\n\\\"'`]+)")
def shell_read_paths(command) -> list:
"""The files a read-only shell command names: `cat a b`, `sed -n 1,9p f`, `head -n 20 f`."""
split = _split_shell(str(command or ""))
if split is None or split[1]:
return []
out = []
for segment in split[0]:
try:
words = shlex.split(segment)
except ValueError:
return []
while words and (_ASSIGNMENT_RE.match(words[0]) or words[0] in _WRAPPERS):
words = words[1:]
program = os.path.basename(words[0]) if words else ""
if program not in _PATH_READERS:
continue
takes, plain, has_script, args, j = _VALUE_OPTIONS.get(program, set()), [], False, words[1:], 0
while j < len(args):
a = args[j]
if a in takes:
has_script = has_script or a in _SCRIPT_OPTIONS
j += 2
continue
if not (a.startswith("-") and a != "-"):
plain.append(a)
j += 1
if program in ("sed", "awk") and not has_script:
plain = plain[1:] # the first plain argument is the script
out.extend(p for p in plain if not p.isdigit())
return out
def path_key(path) -> str:
"""The last two parts of a file path, so /work/app/src/a.py and src/a.py match."""
parts = [p for p in str(path or "").replace("\\", "/").split("/") if p and p != "."]
return "/".join(parts[-2:])
def _call_paths(call, act) -> tuple:
"""Keys of the files a read or edit call touched, when they can be known."""
if act not in ("research", "edit"):
return ()
paths = list(call.paths)
if not paths and call.kind == "shell":
if call.name != "exec":
paths = shell_read_paths(call.command)
elif act == "edit":
paths = [p.strip() for p in _PATCH_FILE_RE.findall(call.command)]
else:
paths = [p for c in script_commands(call.command) for p in shell_read_paths(c)]
return tuple(k for k in (path_key(p) for p in paths) if k)
# ---------------------------------------------------------------------------
# Turns: one typed prompt and everything the agent did until the next one
# ---------------------------------------------------------------------------
_TS_RE = re.compile(r"^(\d{4})-(\d\d)-(\d\d)[T ](\d\d):(\d\d):(\d\d)(?:\.(\d+))?(Z|[+-]\d\d:?\d\d)?$")
def epoch(ts) -> Optional[float]:
"""Seconds since 1970 for an ISO 8601 time (UTC when it has no zone); None when it does not parse."""
m = _TS_RE.match(ts.strip()) if isinstance(ts, str) else None
if not m:
return None
value = float(calendar.timegm(tuple(int(g) for g in m.group(1, 2, 3, 4, 5, 6)) + (0, 0, 0)))
if m.group(7):
value += int(m.group(7)[:6]) / 10.0 ** len(m.group(7)[:6])
zone = m.group(8)
if zone and zone != "Z":
value -= (1 if zone[0] == "+" else -1) * (int(zone[1:3]) * 3600 + int(zone[-2:]) * 60)
return value
@dataclass
class Turn:
"""One typed prompt and what the agent did until the next one, subagents included."""
harness: str
session: str
project: str
t: float # when the prompt was sent: seconds since 1970, UTC
version: str = ""
model: str = ""
entrypoint: str = "" # how Claude Code was started: cli, claude-desktop, sdk-ts, ...
correction: bool = False # the prompt corrects the agent
interrupts: int = 0
calls: int = 0
reads: int = 0 # calls that read or search files
edits: int = 0
path_edits: int = 0 # edits (not whole-file writes) to a known file
blind_edits: int = 0 # of those, edits to a file not read earlier in the session
errors: int = 0 # failed calls, not counting denials and interruptions
reads_before_edit: Optional[int] = None # None when the turn made no edit
responses: int = 0 # API responses with token counts
output: int = 0
reasoning: int = 0
cost: Optional[float] = 0.0 # None when a model in the turn has no known price
@dataclass
class _Ev:
"""The parts of a transcripts.Event this script keeps; the rest is dropped to save memory."""
kind: str # user | assistant | tool | interrupt
id: str
t: float
ts: str = "" # the event's own ISO time, for transcripts.unique_events(since=...)
version: str = "" # user: the Claude Code record's version
entrypoint: str = "" # user: the Claude Code record's entrypoint
correction: bool = False # user
act: str = "" # tool: research | edit | other
failed: bool = False # tool
paths: tuple = () # tool: keys of the files read or edited
whole_file: bool = False # tool: a whole-file write rather than an edit
model: str = "" # assistant
usage: object = None # assistant: a transcripts.Usage or None
class _Draft:
def __init__(self, session, prompt, fallback_model):
self.session, self.prompt, self.fallback_model = session, prompt, fallback_model
self.model, self.interrupts, self.acts, self.responses = "", 0, [], []
def _prompt_records(path) -> dict:
"""Record uuid -> (version, entrypoint) for Claude Code user records. Session.version
is the file's first version; a session resumed after an update records later
turns under the newer version, and each record says which."""
found = {}
try:
with open(path, "rb") as fh:
for raw in fh:
if b'"user"' not in raw or b'"tool_use_id"' in raw:
continue
try:
rec = json.loads(raw)
except ValueError:
continue
if (isinstance(rec, dict) and rec.get("type") == "user" and isinstance(rec.get("uuid"), str)
and isinstance(rec.get("version"), str)):
entry = rec.get("entrypoint")
found[rec["uuid"]] = (rec["version"], entry if isinstance(entry, str) else "")
except OSError:
pass
return found
def _compact(session, records) -> tuple:
"""(compact events, prompts left out because they have no readable time)."""
out, dropped, last_t = [], 0, epoch(session.started)
for e in session.events:
t = epoch(e.ts)
if t is None:
if last_t is None: # nothing in the session has a time yet: the event cannot be placed
dropped += int(e.kind == "user" and not e.injected)
continue
t = last_t
last_t = t
if e.kind == "user" and not e.injected:
version, entry = records.get(e.id, ("", ""))
out.append(_Ev("user", e.id, t, e.ts, version=version, entrypoint=entry, correction=is_correction(e.text)))
elif e.kind == "tool" and e.tool is not None:
c = e.tool
act = classify_call(c)
out.append(_Ev("tool", e.id, t, e.ts, act=act, failed=c.is_error and not c.denied and not c.interrupted,
paths=_call_paths(c, act), whole_file=c.kind == "write"))
elif e.kind == "assistant":
model = e.model if e.model and not e.model.startswith("<") else "" # skip "<synthetic>"
out.append(_Ev("assistant", e.id, t, e.ts, model=model, usage=e.usage))
elif e.kind == "interrupt":
out.append(_Ev("interrupt", e.id, t, e.ts))
return out, dropped
def _drafts(session, events) -> list:
drafts, cur, last_model = [], None, ""
for e in events:
if e.kind == "user":
cur = _Draft(session, e, last_model)
drafts.append(cur)
continue
if e.kind == "assistant" and e.model:
last_model = e.model
if cur is None:
continue
if e.kind == "tool":
cur.acts.append((e.t, e.act, e.failed, e.paths, e.whole_file))
elif e.kind == "assistant":
cur.model = cur.model or e.model
if e.usage is not None:
cur.responses.append((e.model, e.usage))
elif e.kind == "interrupt":
cur.interrupts += 1
return drafts
def _finish(draft, known) -> tuple:
"""The Turn for a draft. `known` holds the files the session has read or
edited so far; it grows as the session's turns are finished in order."""
s, p = draft.session, draft.prompt
turn = Turn(harness=s.harness, session=s.id, project=s.cwd, t=p.t, version=p.version or s.version,
model=draft.model or draft.fallback_model, entrypoint=p.entrypoint, correction=p.correction,
interrupts=draft.interrupts)
acts = sorted(draft.acts, key=lambda a: a[0]) # subagent calls fall in between by time
turn.calls = len(acts)
turn.reads = sum(1 for a in acts if a[1] == "research")
turn.edits = sum(1 for a in acts if a[1] == "edit")
turn.errors = sum(1 for a in acts if a[2])
first_edit = next((i for i, a in enumerate(acts) if a[1] == "edit"), None)
if first_edit is not None:
turn.reads_before_edit = sum(1 for a in acts[:first_edit] if a[1] == "research")
for _t, act, _failed, paths, whole_file in acts:
if act == "edit" and paths and not whole_file:
turn.path_edits += 1
turn.blind_edits += int(not any(k in known for k in paths))
if act in ("research", "edit"):
known.update(paths)
unpriced = set()
for model, usage in draft.responses:
turn.responses += 1
turn.output += usage.output
turn.reasoning += usage.reasoning
cost = pricing.cost_usd(usage, model)
if cost is not None:
turn.cost = None if turn.cost is None else turn.cost + cost
elif usage.total_input() + usage.output:
turn.cost = None
unpriced.add(model or "(no model id)")
return turn, unpriced
def collect_turns(harness, since_days=90, project=None, home=None) -> tuple:
"""Turns from the last `since_days` days, oldest first, and counts for the report's notes.
Files are picked by modified time, and a recently modified file can hold
months-old records, so events older than the window are dropped too."""
info = {"files": 0, "subagent_files": 0, "subagents_attached": 0, "turns_before_window": 0,
"warnings": 0, "unpriced_models": {}, "prompts_without_time": 0}
sessions = []
for h, path in transcripts.find_sessions(harness=harness, since_days=since_days, project=project, home=home):
s = transcripts.load_session(h, path)
info["files"] += 1
info["subagent_files"] += int(s.is_subagent)
info["warnings"] += len(s.warnings)
s.events, dropped = _compact(s, _prompt_records(path) if h == "claude-code" and not s.is_subagent else {})
info["prompts_without_time"] += dropped
sessions.append(s)
since = transcripts.cutoff(since_days) if since_days else None
by_session = {}
for s, e in transcripts.unique_events(sessions, since=since): # a forked session's copy counts once
by_session.setdefault(id(s), (s, []))[1].append(e)
if since:
info["turns_before_window"] = sum(1 for _s, e in transcripts.unique_events(sessions)
if e.kind == "user" and e.ts and e.ts[:19] < since[:19])
drafts, subagents = {}, []
for s, events in by_session.values():
if s.is_subagent:
subagents.append((s, events))
else:
drafts[(s.harness, s.id)] = _drafts(s, events)
for s, events in subagents: # subagent work belongs to the turn that was running when it started
parent = drafts.get((s.harness, s.parent_id))
if not parent or not events:
continue
target = parent[0]
for d in parent:
if d.prompt.t > events[0].t:
break
target = d
info["subagents_attached"] += 1
for e in events:
if e.kind == "tool":
target.acts.append((e.t, e.act, e.failed, e.paths, e.whole_file))
elif e.kind == "assistant" and e.usage is not None:
target.responses.append((e.model, e.usage))
turns = []
for session_drafts in drafts.values():
known = set()
for d in session_drafts:
turn, unpriced = _finish(d, known)
turns.append(turn)
for model in unpriced:
info["unpriced_models"][model] = info["unpriced_models"].get(model, 0) + 1
turns.sort(key=lambda t: t.t)
return turns, info
# ---------------------------------------------------------------------------
# Statistics (references/statistics.md)
# ---------------------------------------------------------------------------
P_LIMIT = 0.01 # adjusted p value a change must stay under
MIN_CHANGE_PCT = 20.0 # smallest change worth flagging, in the per-session values
MIN_TURNS = 30 # turns each side of an update needs
MIN_SESSIONS = 20 # sessions each side of an update needs
MIN_SESSION_VALUES = 12 # sessions with a value for one metric, each side, before it is tested
MIN_EVENT_SESSIONS = 5 # for a rare event: sessions with the event on the busier side
TWO_WEEKS = 14 * 86400.0 # how far from a model's first use its comparison reaches
def _ranks(values) -> tuple:
"""(average ranks in input order, the tie term: sum of t^3 - t over runs of ties)."""
order = sorted(range(len(values)), key=lambda k: values[k])
ranks, tie_term, i = [0.0] * len(values), 0.0, 0
while i < len(order):
j = i
while j + 1 < len(order) and values[order[j + 1]] == values[order[i]]:
j += 1
for k in range(i, j + 1):
ranks[order[k]] = (i + j) / 2.0 + 1 # a run of ties shares the average rank
tie_term += (j - i + 1) ** 3 - (j - i + 1)
i = j + 1
return ranks, tie_term
def mann_whitney(a, b) -> tuple:
"""Two-sided Mann-Whitney U test with the normal approximation, the tie
correction, and a continuity correction of 0.5 (the defaults of R's
wilcox.test with exact=FALSE and of scipy's asymptotic method).
Returns (U of `a`, z, p). z is negative when `a` tends to be smaller."""
n1, n2 = len(a), len(b)
if not n1 or not n2:
return 0.0, 0.0, 1.0
ranks, tie_term = _ranks(list(a) + list(b))
u = sum(ranks[:n1]) - n1 * (n1 + 1) / 2.0
n = n1 + n2
mean = n1 * n2 / 2.0
var = n1 * n2 / 12.0 * ((n + 1) - tie_term / (n * (n - 1)))
if var <= 0:
return u, 0.0, 1.0
z = max(abs(u - mean) - 0.5, 0.0) / math.sqrt(var)
return u, math.copysign(z, u - mean), min(1.0, math.erfc(z / math.sqrt(2)))
def fisher_exact(k1, n1, k2, n2) -> float:
"""Two-sided Fisher exact test for k1 events in n1 against k2 in n2: with
the margins fixed, the chance of a table no more likely than the one seen.
Exact integer arithmetic (R's fisher.test gives the same values)."""
n, k = n1 + n2, k1 + k2
low, high = max(0, n1 - (n - k)), min(k, n1)
weights, w = [], math.comb(k, low) * math.comb(n - k, n1 - low)
for x in range(low, high + 1): # C(k, x) * C(n - k, n1 - x), one step at a time
weights.append(w)
if x < high:
w = w * (k - x) * (n1 - x) // ((x + 1) * (n - k - n1 + x + 1))
seen = weights[k1 - low]
return min(1.0, sum(v for v in weights if v <= seen) / math.comb(n, n1))
def _mode(values) -> tuple:
counts = {}
for v in values:
counts[v] = counts.get(v, 0) + 1
value = max(counts, key=lambda v: (counts[v], -v))
return value, counts[value]
def _p_value(a, b) -> tuple:
"""(z, p): Mann-Whitney, and when one value holds more than half the
sessions, the larger of that p and Fisher's exact test on the sessions away
from that value against the sessions at it (the normal approximation is too
small there). For a rare event the common value is 0, so "away" means "had it"."""
_u, z, p = mann_whitney(a, b)
mode, count = _mode(list(a) + list(b))
if 2 * count > len(a) + len(b):
away_a, away_b = sum(1 for v in a if v != mode), sum(1 for v in b if v != mode)
p = max(p, fisher_exact(away_a, len(a), away_b, len(b)))
return z, p
def session_p(a, b) -> tuple:
"""(z, p, best possible p for these values and side sizes) for per-session values."""
z, p = _p_value(a, b)
pooled, n = sorted(list(a) + list(b)), len(a)
best = min(_p_value(pooled[:n], pooled[n:])[1], _p_value(pooled[-n:], pooled[:-n])[1])
return z, p, best
def session_test(a, b) -> Optional[tuple]:
"""session_p, or None when one value holds more than half the sessions and
fewer than MIN_EVENT_SESSIONS sessions on the busier side are away from it
(a rare event seen too seldom to test)."""
mode, count = _mode(list(a) + list(b))
if 2 * count > len(a) + len(b):
if max(sum(1 for v in a if v != mode), sum(1 for v in b if v != mode)) < MIN_EVENT_SESSIONS:
return None
return session_p(a, b)
def bh_adjust(pvalues) -> list:
"""Benjamini-Hochberg adjusted p values (R's p.adjust(p, "BH")), in input order."""
m = len(pvalues)
order = sorted(range(m), key=lambda k: pvalues[k])
adjusted, running = [1.0] * m, 1.0
for rank in range(m, 0, -1):
k = order[rank - 1]
running = min(running, pvalues[k] * m / rank)
adjusted[k] = running
return adjusted
def tarone(best_ps, alpha=P_LIMIT) -> list:
"""Indices of the tests that can reach significance at all (Tarone, 1990):
for the smallest K with at most K tests whose best possible p is under
alpha / K, those tests. The others cannot pass and only dilute the adjustment."""
for k in range(1, len(best_ps) + 1):
keep = [i for i, best in enumerate(best_ps) if best < alpha / k]
if len(keep) <= k:
return keep
return []
_SEMVER_RE = re.compile(r"^v?(\d+)(?:\.(\d+))?(?:\.(\d+))?(?:-([0-9A-Za-z.-]+))?(?:\+[0-9A-Za-z.-]*)?$")
def version_key(version) -> tuple:
"""Sort key: semantic-version order (a pre-release sorts before its
release; numeric parts compare as numbers), other strings after, by name."""
m = _SEMVER_RE.match(str(version).strip())
if not m:
return (1, str(version))
core = tuple(int(g or 0) for g in m.group(1, 2, 3))
if m.group(4) is None:
return (0, core, (1,))
parts = tuple((0, int(p), "") if p.isdigit() else (1, 0, p) for p in m.group(4).split("."))
return (0, core, (0, parts))
# ---------------------------------------------------------------------------
# Metrics (references/metrics.md)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Metric:
key: str
label: str
turn_summary: str # the By version table: median | mean | per100 | share (a percent) | ratio
session_summary: str # the test and the flagged changes: median or mean of the per-session values
scale: float # 100 for rates and shares shown as per 100 or percent
bad: str # the direction that usually means worse: up, down, or "" for neither
phrases: tuple # less, more, times as much, dropped to zero, rose from zero
METRICS = (
Metric("reads_before_edit", "Reads before the first edit", "median", "median", 1, "down",
("reads {pct}% less before it edits", "reads {pct}% more before it edits",
"reads {times} as much before it edits", "edits without reading first in most sessions",
"reads before it edits in most sessions, where before it mostly did not")),
Metric("reads_per_edit", "Reads per edit", "ratio", "median", 1, "down",
("reads {pct}% less per edit", "reads {pct}% more per edit", "reads {times} as much per edit",
"edits without reading in most sessions",
"reads before editing in most sessions, where before it mostly did not")),
Metric("blind_edits", "Edits to files not read earlier in the session (%)", "share", "mean", 100, "up",
("makes {pct}% fewer edits to files it had not read", "makes {pct}% more edits to files it had not read",
"makes {times} as many edits to files it had not read", "no longer edits files it had not read",
"now edits files it had not read, where before it did not")),
Metric("edits_per_turn", "Edits per turn", "mean", "median", 1, "",
("makes {pct}% fewer edits per turn", "makes {pct}% more edits per turn",
"makes {times} as many edits per turn", "makes no edits in most sessions",
"makes edits in most sessions, where before it mostly did not")),
Metric("tool_error_rate", "Failed tool calls (%)", "share", "mean", 100, "up",
("hits {pct}% fewer tool errors", "hits {pct}% more tool errors", "hits tool errors {times} as often",
"no longer hits tool errors", "now hits tool errors, where before it had none")),
Metric("interrupts", "Interrupts per 100 turns", "per100", "mean", 100, "up",
("gets interrupted {pct}% less often", "gets interrupted {pct}% more often",
"gets interrupted {times} as often", "no longer gets interrupted",
"now gets interrupted, where before it never was")),
Metric("corrections", "Corrections per 100 prompts", "per100", "mean", 100, "up",
("needs correcting {pct}% less often", "needs correcting {pct}% more often",
"needs correcting {times} as often", "no longer needs correcting",
"now needs correcting, where before it never did")),
Metric("output_tokens", "Output tokens per turn", "median", "median", 1, "",
("writes {pct}% less output per turn", "writes {pct}% more output per turn",
"writes {times} as much output per turn", "writes no output in most sessions",
"writes output in most sessions, where before it mostly did not")),
Metric("reasoning_share", "Share of output spent reasoning (%)", "share", "mean", 100, "down",
("spends {pct}% less of its output on reasoning", "spends {pct}% more of its output on reasoning",
"spends {times} as much of its output on reasoning", "no longer spends output on reasoning",
"now spends output on reasoning, where before it did not")),
Metric("tool_calls", "Tool calls per turn", "median", "median", 1, "",
("makes {pct}% fewer tool calls per turn", "makes {pct}% more tool calls per turn",
"makes {times} as many tool calls per turn", "makes no tool calls in most sessions",
"makes tool calls in most sessions, where before it mostly did not")),
Metric("cost", "Cost per turn (USD)", "median", "median", 1, "up",
("costs {pct}% less per turn", "costs {pct}% more per turn", "costs {times} as much per turn",
"costs nothing in most sessions", "costs money in most sessions, where before it mostly did not")),
)
METRIC = {m.key: m for m in METRICS}
_TOTALS = {"reads_per_edit": ("reads", "edits"), "blind_edits": ("blind_edits", "path_edits"),
"tool_error_rate": ("errors", "calls"), "reasoning_share": ("reasoning", "output")}
def _sample(key, t):
"""The turn's value for a per-turn metric, or None when the turn does not count for it."""
if key == "reads_before_edit":
return t.reads_before_edit
if key == "edits_per_turn":
return t.edits
if key == "interrupts":
return t.interrupts
if key == "corrections":
return 1 if t.correction else 0
if key == "output_tokens":
return t.output if t.responses else None
if key == "tool_calls":
return t.calls
if key == "cost":
return t.cost if t.responses and t.cost is not None else None
num, den = _TOTALS[key]
return getattr(t, num) / float(getattr(t, den)) if getattr(t, den) else None
def samples(key, turns) -> list:
return [v for v in (_sample(key, t) for t in turns) if v is not None]
def _session_value(key, items):
if key in _TOTALS: # a session's own total ratio, not an average of its turns
num, den = _TOTALS[key]
total = sum(getattr(t, den) for t in items)
return sum(getattr(t, num) for t in items) / float(total) if total else None
values = samples(key, items)
return sum(values) / float(len(values)) if values else None
def session_values(key, turns) -> list:
"""One value per session, so each session counts once in the test: the mean
of its turns' values, or for ratios and shares the session's own total ratio."""
by_session = {}
for t in turns:
by_session.setdefault(t.session, []).append(t)
return [v for v in (_session_value(key, items) for items in by_session.values()) if v is not None]
def session_summary(key, values):
"""The median or mean of per-session values, on the scale the report shows."""
if not values:
return None
m = METRIC[key]
value = statistics.median(values) if m.session_summary == "median" else sum(values) / float(len(values))
return value * m.scale
def _reasoning_recorded(turns) -> bool:
"""Claude Code records thinking tokens only for some models and versions: a
slice counts as recording reasoning when most of its turns with output do."""
with_output = [t for t in turns if t.output]
return bool(with_output) and 2 * sum(1 for t in with_output if t.reasoning) >= len(with_output)
def metric_value(key, turns) -> tuple:
"""Turn-level value for the By version table: (value, turns it rests on), or (None, 0)."""
if key == "reasoning_share" and not _reasoning_recorded(turns):
return None, 0
kind = METRIC[key].turn_summary
if kind in ("ratio", "share"):
num, den = _TOTALS[key]
total = sum(getattr(t, den) for t in turns)
if not total:
return None, 0
value = sum(getattr(t, num) for t in turns) / float(total)
return value * (100.0 if kind == "share" else 1.0), sum(1 for t in turns if getattr(t, den))
values = samples(key, turns)
if not values:
return None, 0
if kind == "median":
return statistics.median(values), len(values)
if kind == "mean":
return sum(values) / len(values), len(values)
return 100.0 * sum(values) / len(values), len(values) # per100
def group_values(turns) -> dict:
return {m.key: metric_value(m.key, turns) for m in METRICS}
# ---------------------------------------------------------------------------
# Slices and the windows around each update
# ---------------------------------------------------------------------------
@dataclass
class Slice:
key: str # a version, a model id, or an ISO week such as 2026-W38
turns: list = field(default_factory=list)
@dataclass
class Window:
i: int # the update: the first slice of the after side
b0: int # the before side comes from slices[b0:i]
a1: int # the after side from slices[i:a1]
before: list # [(slice index, turns)]
after: list
def _week(t) -> str:
year, week, _day = datetime.datetime.fromtimestamp(t, tz=datetime.timezone.utc).isocalendar()
return "%d-W%02d" % (year, week)
def week_start(key) -> str:
"""The Monday of an ISO week key such as 2026-W38."""
year, week = key.split("-W")
return datetime.date.fromisocalendar(int(year), int(week), 1).isoformat()
def _sessions(turns) -> int:
return len({t.session for t in turns})
def _enough(turns, min_turns, min_sessions) -> bool:
return len(turns) >= min_turns and _sessions(turns) >= min_sessions
def _time_chain(slices) -> list:
"""Indices of the versions to keep: the longest run, in version order, whose
median turn time never falls, counted in versions first and sessions second.
A version outside it ran later than newer versions: a second install."""
med = [statistics.median(t.t for t in s.turns) for s in slices]
size = [_sessions(s.turns) for s in slices]
best, prev = [(1, size[i]) for i in range(len(slices))], [-1] * len(slices)
for i in range(len(slices)):
for j in range(i):
candidate = (best[j][0] + 1, best[j][1] + size[i])
if med[j] <= med[i] and candidate > best[i]:
best[i], prev[i] = candidate, j
i, chain = max(range(len(slices)), key=lambda k: (best[k], k)), []
while i != -1:
chain.append(i)
i = prev[i]
return chain[::-1]
def make_slices(turns, by, min_turns=MIN_TURNS, min_sessions=MIN_SESSIONS) -> tuple:
"""Slices in order (versions by number, weeks by date, models by first use)
and notes: turns with no version or model, versions left out because they
ran after newer ones, and models left out for having too little data."""
keyed, first_seen = {}, {}
notes = {"no_key": 0, "left_out": [], "out_of_order": []}
for t in turns:
key = t.version if by == "version" else t.model if by == "model" else _week(t.t)
if not key:
notes["no_key"] += 1
continue
keyed.setdefault(key, []).append(t)
first_seen[key] = min(first_seen.get(key, t.t), t.t)
if by == "version":
order = sorted(keyed, key=version_key)
elif by == "model":
order = sorted(keyed, key=lambda k: (first_seen[k], k))
else:
order = sorted(keyed)
slices = [Slice(k, sorted(keyed[k], key=lambda t: t.t)) for k in order]
if by == "version" and len(slices) > 1:
keep = set(_time_chain(slices))
notes["out_of_order"] = [(s.key, s.turns[0].t, s.turns[-1].t, _sessions(s.turns))
for i, s in enumerate(slices) if i not in keep]
slices = [s for i, s in enumerate(slices) if i in keep]
if by == "model":
notes["left_out"] = [(s.key, len(s.turns), _sessions(s.turns)) for s in slices
if not _enough(s.turns, min_turns, min_sessions)]
slices = [s for s in slices if _enough(s.turns, min_turns, min_sessions)]
return slices, notes
def windows(slices, by, min_turns=MIN_TURNS, min_sessions=MIN_SESSIONS) -> list:
"""A Window for each update with enough data around it. Versions and weeks
borrow their nearest neighbors until a side has min_turns turns from
min_sessions sessions. A model is compared with the model before it, using
only the two weeks before and after its first use."""
out = []
for i in range(1, len(slices)):
if by == "model":
start = slices[i].turns[0].t
before = [t for t in slices[i - 1].turns if start - TWO_WEEKS <= t.t < start]
after = [t for t in slices[i].turns if t.t < start + TWO_WEEKS]
if _enough(before, min_turns, min_sessions) and _enough(after, min_turns, min_sessions):
out.append(Window(i, i - 1, i + 1, [(i - 1, before)], [(i, after)]))
continue
before, pooled_b, b0 = [], [], i
while b0 > 0 and not _enough(pooled_b, min_turns, min_sessions):
b0 -= 1
before.append((b0, slices[b0].turns))
pooled_b += slices[b0].turns
after, pooled_a, a1 = [], [], i
while a1 < len(slices) and not _enough(pooled_a, min_turns, min_sessions):
after.append((a1, slices[a1].turns))
pooled_a += slices[a1].turns
a1 += 1
if _enough(pooled_b, min_turns, min_sessions) and _enough(pooled_a, min_turns, min_sessions):
out.append(Window(i, b0, a1, before[::-1], after))
return out
# ---------------------------------------------------------------------------
# Comparisons (references/statistics.md)
# ---------------------------------------------------------------------------
@dataclass
class Comparison:
metric: str
update: int # index of the first slice after the update
b0: int # the before side is slices[b0:update]
a1: int # the after side is slices[update:a1]
before_value: float # the per-session summary each side, on the scale shown
after_value: float
n_before: int # turns on each side
n_after: int
sessions_before: int # sessions with a value: the units of the test
sessions_after: int
z: float
p: float
best_p: float = 0.0 # the smallest p these values could give
in_adjustment: bool = True # False: it could not reach significance (Tarone)
group: int = -1 # windows of one metric and direction that overlap share a group
representative: bool = False # the group's smallest p, the one the adjustment counts
p_adjusted: float = 1.0 # the group's Benjamini-Hochberg adjusted p
passes: bool = False # passes every rule
same_as: Optional[int] = None # the update named for this change instead
flagged: bool = False
stand_out: bool = False # part of a slice that differs from both neighbors
@property
def change(self) -> tuple:
"""(percent change, from_zero). The percent is None when the change starts from zero."""
return _change(self.before_value, self.after_value)
def _change(before, after) -> tuple:
if before == 0:
return (0.0, False) if after == 0 else (None, True)
return (after - before) / abs(before) * 100.0, False
def _rising(before, after) -> bool:
return after > before
def _way(c) -> str:
return "up" if c.z < 0 else "down" if c.z > 0 else ""
def _passes(c) -> bool:
"""A change of MIN_CHANGE_PCT or more, p under P_LIMIT, and the test
pointing the same way as the change (z < 0: the after side is higher)."""
pct, from_zero = c.change
big = from_zero or abs(pct) >= MIN_CHANGE_PCT
return c.p < P_LIMIT and big and c.z != 0 and (c.z < 0) == _rising(c.before_value, c.after_value)
def _group(pool) -> list:
"""Overlapping windows of one metric pointing the same way test one change:
each group is the smallest-p window left plus every window whose update
lies inside its span."""
groups, left = [], sorted(pool, key=lambda c: (c.p, c.update))
while left:
seed = left.pop(0)
members = [seed] + [c for c in left if _way(seed) and c.metric == seed.metric and _way(c) == _way(seed)
and seed.b0 < c.update < seed.a1]
taken = {id(c) for c in members}
left = [c for c in left if id(c) not in taken]
groups.append(members)
return groups
def compare(slices, wins) -> tuple:
"""Test every metric at every window, each session counting once. Returns
(comparisons, counts for the notes)."""
recorded = [_reasoning_recorded(s.turns) for s in slices]
size = [_sessions(s.turns) for s in slices]
out, sparse = [], 0
for w in wins:
for m in METRICS:
before, after = w.before, w.after
if m.key == "reasoning_share": # only slices that record reasoning, or the rest count as 0%
before = [(j, ts) for j, ts in before if recorded[j]]
after = [(j, ts) for j, ts in after if recorded[j]]
tb = [t for _j, ts in before for t in ts]
ta = [t for _j, ts in after for t in ts]
sb, sa = session_values(m.key, tb), session_values(m.key, ta)
if len(sb) < MIN_SESSION_VALUES or len(sa) < MIN_SESSION_VALUES or len(set(sb + sa)) == 1:
continue # too few sessions, or every session has the same value
tested = session_test(sb, sa)
if tested is None:
sparse += 1
continue
z, p, best = tested
out.append(Comparison(m.key, w.i, w.b0, w.a1, session_summary(m.key, sb), session_summary(m.key, sa),
len(tb), len(ta), len(sb), len(sa), z, p, best))
testable = set(tarone([c.best_p for c in out]))
for k, c in enumerate(out):
c.in_adjustment = k in testable
groups = _group([c for c in out if c.in_adjustment])
for gid, (members, q) in enumerate(zip(groups, bh_adjust([g[0].p for g in groups]))):
members[0].representative = True
for c in members:
c.group, c.p_adjusted = gid, q
passing = [c for c in members if _passes(c)] if q < P_LIMIT else []
if passing: # name the window whose own boundary is best supported, then the smallest p
named = max(passing, key=lambda c: (min(size[c.update - 1], size[c.update]), -c.p, -c.update))
named.flagged = True
for c in passing:
c.passes = True
c.same_as = None if c is named else named.update
return out, {"sparse": sparse, "untestable": len(out) - len(testable)}
def _stand_outs(by, slices, comparisons) -> list:
"""A flag at one update followed, within its after side, by the opposite
flag on the same metric: the slices between differ from both neighbors.
Report them once as standing out, and headline neither flag."""
flags = sorted((c for c in comparisons if c.flagged), key=lambda c: c.update)
out = []
for f in flags:
for g in flags:
if (not f.stand_out and not g.stand_out and g.metric == f.metric and f.update < g.update <= f.a1
and _way(g) != _way(f)):
f.stand_out = g.stand_out = True
f.flagged = g.flagged = False
keys = [clean_key(s.key) for s in slices[f.update:g.update]]
span = keys[0] if len(keys) == 1 else "%s to %s" % (keys[0], keys[-1])
one, many = UNITS[by]
if by == "week":
who = "The week of %s stands" % week_start(slices[f.update].key) if len(keys) == 1 else \
"The weeks %s stand" % span
else:
who = "%s %s stands" % (one.capitalize(), _code_keys(span)) if len(keys) == 1 else \
"%s %s stand" % (many.capitalize(), _code_keys(span))
around = "the %s around %s" % (many, "it" if len(keys) == 1 else "them")
out.append({"metric": f.metric, "versions": span, "text": "%s out from %s: %s %s before, %s there, "
"%s after." % (who, around, METRIC[f.metric].label.lower(), _fmt(f.metric, f.before_value),
_fmt(f.metric, f.after_value), _fmt(g.metric, g.after_value))})
return out
# ---------------------------------------------------------------------------
# The result: headline, slices, comparisons
# ---------------------------------------------------------------------------
HARNESS_NAMES = {"claude-code": "Claude Code", "codex": "Codex", "gemini-cli": "Gemini CLI", "opencode": "OpenCode"}
UNITS = {"version": ("version", "versions"), "model": ("model", "models"), "week": ("week", "weeks")}
_KEY_RE = re.compile(r"[^A-Za-z0-9._:@/+-]")
def clean_key(text) -> str:
"""A version or model id as plain characters only, secrets masked first:
nothing in it can carry markup, a table break, or an instruction."""
return _KEY_RE.sub("", transcripts.redact(str(text)))[:80] or "(unnamed)"
def _k(text) -> str:
"""A version or model id for markdown text: plain characters inside inline
code, so a key such as a bare URL does not render as a link."""
return code(clean_key(text))
_JOINS = ("after", "between", "and", "to", "vs")
def _code_keys(text) -> str:
"""A place or span made of keys ("between 2.1.260 and 2.1.270") for
markdown: each key in inline code, the joining words plain. Keys hold no
spaces, so a split on spaces finds them."""
return " ".join(w if w in _JOINS else code(w) for w in text.split(" "))
def _safe(text, limit=80) -> str:
"""Untrusted text such as a folder name as one inert line."""
return safe_text(text, limit=limit)
def _home(path) -> str:
home = os.path.expanduser("~")
if path and (path == home or path.startswith(home + os.sep)):
return "~" + path[len(home):]
return path
def _folder(path) -> str:
"""A folder for the markdown report: shortened, made inert, inside inline code."""
return code(_home(path), 160)
def _num(value):
if value is None or isinstance(value, int):
return value
return float("%.6g" % value)
def _day(t) -> str:
return datetime.datetime.fromtimestamp(t, tz=datetime.timezone.utc).strftime("%Y-%m-%d")
def _span(slices, first, stop) -> str:
keys = [clean_key(s.key) for s in slices[first:stop]]
return keys[0] if len(keys) == 1 else "%s to %s" % (keys[0], keys[-1])
def _pct_int(pct) -> int:
"""One whole percent, rounded half away from zero, for the phrase and the table alike."""
return int(math.copysign(math.floor(abs(pct) + 0.5), pct))
def _times(ratio) -> str:
if 1.95 <= ratio < 2.05:
return "twice"
return ("%.1f" % ratio).rstrip("0").rstrip(".") + " times"
def phrase(c) -> str:
"""What the agent does differently, as a verb phrase ("reads 41% less before it edits")."""
less, more, times, to_zero, from_zero = METRIC[c.metric].phrases
pct, starts_at_zero = c.change
if starts_at_zero:
return from_zero
if c.after_value == 0:
return to_zero
if pct >= 100:
return times.format(times=_times(c.after_value / c.before_value))
return (less if pct < 0 else more).format(pct=abs(_pct_int(pct)))
def _usually(c) -> str:
bad = METRIC[c.metric].bad
if not bad:
return "no clear direction"
return "worse" if ("up" if _rising(c.before_value, c.after_value) else "down") == bad else "better"
def _where(by, slices, c) -> str:
"""Where the change sits: after one slice, or between two when the window
borrowed neighbors (or its own slice is too small to name alone)."""
key = clean_key(slices[c.update].key)
if by == "model":
return "%s vs %s" % (key, clean_key(slices[c.update - 1].key))
borrowed = c.update - c.b0 > 1 or c.a1 - c.update > 1
if (borrowed or _sessions(slices[c.update].turns) < 3) and c.a1 - 1 > c.b0 + 1:
return "between %s and %s" % (clean_key(slices[c.b0 + 1].key), clean_key(slices[c.a1 - 1].key))
return "after %s" % key
def _comparison_dict(c, slices, by="version") -> dict:
pct, from_zero = c.change
return {
"metric": c.metric, "label": METRIC[c.metric].label, "update": clean_key(slices[c.update].key),
"where": _where(by, slices, c), "before": _span(slices, c.b0, c.update),
"after": _span(slices, c.update, c.a1), "measured": "%s per session" % METRIC[c.metric].session_summary,
"before_value": _num(c.before_value), "after_value": _num(c.after_value),
"change_pct": None if pct is None else _pct_int(pct), "from_zero": from_zero,
"direction": "up" if _rising(c.before_value, c.after_value) else "down", "usually": _usually(c),
"phrase": phrase(c), "p": _num(c.p), "best_p": _num(c.best_p), "in_adjustment": c.in_adjustment,
"group": c.group, "representative": c.representative, "p_adjusted": _num(c.p_adjusted),
"n_before": c.n_before, "n_after": c.n_after,
"sessions_before": c.sessions_before, "sessions_after": c.sessions_after,
"passes_test": c.passes, "flagged": c.flagged, "stand_out": c.stand_out,
"same_change_as": None if c.same_as is None else clean_key(slices[c.same_as].key),
}
def _lead(by, name, slices, c) -> str:
where = _where(by, slices, c)
if by == "model":
return "On %s, compared with %s" % (_k(slices[c.update].key), _k(slices[c.update - 1].key))
if by == "version":
if where.startswith("between"):
return "After an update %s" % _code_keys(where).replace("between ", "between %s " % name, 1)
return "After %s %s" % (name, _k(slices[c.update].key))
if where.startswith("between"):
return "After a change between the weeks of %s and %s" % (week_start(slices[c.b0 + 1].key),
week_start(slices[c.a1 - 1].key))
return "Since the week of %s" % week_start(slices[c.update].key)
def _window_phrase(since_days) -> str:
return " in the last %d days" % since_days if since_days else ""
_FAMILY = {"reads_before_edit": "reading", "reads_per_edit": "reading"}
_ALSO_CHANGED = (("model", "the model"), ("version", "the harness version"), ("project_mix", "the projects"),
("work", "the kind of work"), ("entrypoint", "the share of scripted runs"))
def _and(words) -> str:
return words[0] if len(words) == 1 else "%s and %s" % (", ".join(words[:-1]), words[-1])
def _headline(by, harness, turns, slices, wins, flagged, stand_outs, info, since_days, min_sessions, confounders,
slice_notes, project) -> str:
name = HARNESS_NAMES.get(harness, harness)
one, many = UNITS[by]
where = " in %s" % _folder(project) if project else ""
if not turns:
if info and info.get("files"):
return "No typed prompts found in {:,} {} session files{}{}.".format(
info["files"], name, where, _window_phrase(since_days))
text = "No %s turns found%s%s." % (name, where, _window_phrase(since_days))
if project:
text += (" Check the --project path: it must be the folder the sessions ran in, or a folder that "
"contains it.")
return text
if slice_notes["no_key"] == len(turns):
others = [u for u in ("version", "model", "week") if u != by]
return "%s records no %s, so there is nothing to split by %s. Run again with --by %s or --by %s." % (
name, one, one, others[0], others[1])
if by == "model" and len(slices) < 2 and slice_notes["left_out"]:
return ("Not enough history to compare models yet: {:,} {} turns from {:,} sessions, and each model needs "
"{} sessions.".format(len(turns), name, _sessions(turns), min_sessions))
if len(slices) == 1:
place = "in one week" if by == "week" else "on one %s" % one
return "All {:,} {} turns{} ran {} ({}), so there is no update to test yet.".format(
len(turns), name, _window_phrase(since_days), place, _k(slices[0].key))
if not wins and by == "model":
return ("Not enough history to compare models yet: {:,} {} turns from {:,} sessions across {} models, and "
"each model needs {} sessions within two weeks of the switch.".format(
len(turns), name, _sessions(turns), len(slices), min_sessions))
if not wins:
return ("Not enough history to test an update yet: {:,} {} turns from {:,} sessions across {} {}, and each "
"side of an update needs {} sessions.".format(len(turns), name, _sessions(turns), len(slices),
many, min_sessions))
if not flagged:
if stand_outs:
return ("No lasting behavior change passed the test across {} {} {} and {:,} turns; {} {} out from "
"the ones around {}.".format(len(slices), name, many, len(turns),
_count(len(stand_outs), one),
"stands" if len(stand_outs) == 1 else "stand",
"it" if len(stand_outs) == 1 else "them"))
return "No behavior change passed the test across {} {} {} and {:,} turns.".format(
len(slices), name, many, len(turns))
counts = {}
for c in flagged:
counts[c.update] = counts.get(c.update, 0) + 1
update = max(counts, key=lambda i: (counts[i], i))
order = [m.key for m in METRICS] # the metrics in order of how much they say about quality
here, families = [], set()
for c in sorted((c for c in flagged if c.update == update),
key=lambda c: (_usually(c) != "worse", order.index(c.metric))):
family = _FAMILY.get(c.metric, c.metric) # two reading measures would say the same thing twice
if family not in families:
families.add(family)
here.append(c)
text = "%s, your agent %s" % (_lead(by, name, slices, here[0]), " and ".join(phrase(c) for c in here[:2]))
kinds = {c["kind"] for c in confounders if c["update"] == clean_key(slices[update].key)}
extra = []
also = [words for kind, words in _ALSO_CHANGED if kind in kinds]
if also:
extra.append("%s changed at the same point" % _and(also))
if "overlap" in kinds:
extra.append("both sides were in use at the same time")
if "concurrent" in kinds:
extra.append("both models were in use at the same time")
text += ", but %s." % " and ".join(extra) if extra else "."
rest = len(flagged) - min(2, len(here))
if rest:
text += " %d more flagged change%s listed below." % (rest, "s are" if rest > 1 else " is")
return text
# ---------------------------------------------------------------------------
# Confounders and notes: what else changed, and what was left out
# ---------------------------------------------------------------------------
def _pct(share) -> str:
return "%d%%" % int(math.floor(100 * share + 0.5))
def _shares(turns, attr) -> list:
"""[(value, share of turns)], largest first."""
counts = {}
for t in turns:
key = getattr(t, attr) or "(unknown)"
counts[key] = counts.get(key, 0) + 1
return sorted(((k, n / float(len(turns))) for k, n in counts.items()), key=lambda x: (-x[1], x[0]))
def _count(n, word) -> str:
return "{:,} {}{}".format(n, word, "" if n == 1 else "s")
def _session_lengths(turns) -> list:
counts = {}
for t in turns:
counts[t.session] = counts.get(t.session, 0) + 1
return list(counts.values())
def _turns_word(m) -> str:
m = int(m) if m == int(m) else round(m, 1)
return "%s turn%s" % (m, "" if m == 1 else "s")
def _confounders(by, slices, flagged, wins, filtered_by_project) -> list:
"""For each update with a flagged change: the other things that changed at the same point."""
out, by_update = [], {w.i: w for w in wins}
mixes = {"version": [("model", "model", "--by model")],
"model": [("version", "harness version", "--by version")],
"week": [("model", "model", "--by model"), ("version", "harness version", "--by version")]}[by]
for update in sorted({c.update for c in flagged}):
c = next(c for c in flagged if c.update == update)
w = by_update[update]
before = [t for _j, ts in w.before for t in ts]
after = [t for _j, ts in w.after for t in ts]
key, where = clean_key(slices[update].key), _where(by, slices, c)
def add(kind, text, **extra):
out.append(dict({"update": key, "where": where, "kind": kind, "text": text}, **extra))
for attr, word, flag in mixes:
sb, sa = _shares(before, attr), _shares(after, attr)
if sb[0][0] != sa[0][0] or abs(sa[0][1] - dict(sb).get(sa[0][0], 0.0)) >= 0.3:
add(attr, "The %s mix changed at the same point: before, %s ran %s of turns; after, %s ran %s. Run "
"again with %s to separate the two." % (word, _k(sb[0][0]), _pct(sb[0][1]),
_k(sa[0][0]), _pct(sa[0][1]), flag))
if not filtered_by_project:
pb, pa = dict(_shares(before, "project")), dict(_shares(after, "project"))
common = set(pb) & set(pa)
shared = min((sum(v for k, v in pa.items() if k in common), "after"),
(sum(v for k, v in pb.items() if k in common), "before"))
if shared[0] < 0.5 and common:
best = max(common, key=lambda k: (min(pb[k], pa[k]), k))
add("project_mix", "The work moved between projects: only %s of the turns %s come from projects that "
"appear on both sides. To compare like with like, run again with `--project %s`." % (
_pct(shared[0]), shared[1], shlex.quote(_safe(_home(best), 160))),
project=_safe(_home(best), 500))
elif shared[0] < 0.5:
add("project_mix", "The two sides share no project, so the change may come from the work, not the "
"update.")
elif len(set(pb) | set(pa)) > 1:
top, side = max((max(pb.items(), key=lambda x: x[1]), "before"),
(max(pa.items(), key=lambda x: x[1]), "after"), key=lambda x: x[0][1])
if top[1] >= 0.6:
add("one_project", "One project, %s, holds %s of the turns %s. Run again with `--project %s` to "
"check the change within that project alone." % (_folder(top[0]), _pct(top[1]), side,
shlex.quote(_safe(_home(top[0]), 160))),
project=_safe(_home(top[0]), 500))
lb, la = _session_lengths(before), _session_lengths(after)
mb, ma = statistics.median(lb), statistics.median(la)
ob, oa = sum(1 for n in lb if n == 1) / float(len(lb)), sum(1 for n in la if n == 1) / float(len(la))
if max(mb, ma) >= 2 * min(mb, ma) or abs(oa - ob) >= 0.25:
add("work", "Sessions changed shape: the median session had %s before and %s after, and %s of sessions "
"had one turn before against %s after. That often means a different kind of work, such as scripted "
"runs instead of long conversations." % (_turns_word(mb), _turns_word(ma), _pct(ob), _pct(oa)))
if max(t.t for t in before) > min(t.t for t in after) + 86400:
add("overlap", "Both sides were in use at the same time: the before side ran until %s, after the after "
"side began on %s. That usually means two installs, so the difference may come from what each was "
"used for." % (_day(max(t.t for t in before)), _day(min(t.t for t in after))))
if any(t.entrypoint for t in before + after):
sdk_b = sum(1 for t in before if t.entrypoint.startswith("sdk-")) / float(len(before))
sdk_a = sum(1 for t in after if t.entrypoint.startswith("sdk-")) / float(len(after))
if abs(sdk_a - sdk_b) >= 0.25:
add("entrypoint", "The share of scripted runs changed: %s of turns before came from the SDK and %s "
"after." % (_pct(sdk_b), _pct(sdk_a)))
if by == "model":
start = slices[update].turns[0].t
old = sum(1 for t in slices[update - 1].turns if start <= t.t < start + TWO_WEEKS)
if old and old >= 0.3 * (old + len(after)):
add("concurrent", "Both models were in use at the same time, so the difference may come from the "
"tasks each was used for. In the two weeks after %s first ran, %s still ran %s of the turns of "
"the two." % (_k(key), _k(slices[update - 1].key), _pct(old / float(old + len(after)))))
return out
_RETENTION = {
"claude-code": "Claude Code deletes transcripts older than cleanupPeriodDays (30 days by default); set it "
"higher in ~/.claude/settings.json to keep more history for this check.",
"gemini-cli": "Gemini CLI deletes sessions older than general.sessionRetention.maxAge (30 days by default).",
}
def _names(keys) -> str:
return _and([_k(k) for k in keys])
def _notes(by, harness, turns, slices, slice_notes, wins, comparisons, counts, info, since_days, min_turns,
min_sessions, untested) -> list:
notes = []
if info:
text = "Read %s" % _count(info.get("files", 0), "session file")
if info.get("subagent_files"):
text += "; %d of the %d subagent files are counted in the turn that started them" % (
info.get("subagents_attached", 0), info["subagent_files"])
notes.append(text + ".")
if info.get("turns_before_window"):
notes.append("Left out %s older than the %d-day window." % (
_count(info["turns_before_window"], "turn"), since_days or 0))
if info.get("prompts_without_time"):
notes.append("Left out %s with no readable time." % _count(info["prompts_without_time"], "prompt"))
if info.get("unpriced_models"):
notes.append("Cost per turn leaves out %s that used models without a known price: %s." % (
_count(sum(info["unpriced_models"].values()), "turn"), _names(sorted(info["unpriced_models"]))))
if info.get("warnings"):
notes.append("Skipped %s." % _count(info["warnings"], "unreadable line"))
if slice_notes["no_key"] and slice_notes["no_key"] < len(turns):
notes.append("Left out %s with no recorded %s." % (_count(slice_notes["no_key"], "turn"), UNITS[by][0]))
rows = slice_notes["out_of_order"]
if rows:
one = len(rows) == 1
notes.append("Left out %s: %s ran %s to %s, after newer versions, which usually means a second install "
"such as the desktop app or an SDK script. Run with --by week to see %s in time order." % (
_names([r[0] for r in rows]), "it" if one else "they", _day(min(r[1] for r in rows)),
_day(max(r[2] for r in rows)), "it" if one else "them"))
if slice_notes["left_out"]:
notes.append("Left out models with too little data (each needs %d turns from %d sessions): %s." % (
min_turns, min_sessions, ", ".join("%s (%s, %s)" % (_k(k), _count(n, "turn"), _count(s, "session"))
for k, n, s in slice_notes["left_out"])))
if untested and wins:
notes.append("Not tested: %s: each side of %s needs %d sessions and %d turns, even with neighbors "
"added." % (", ".join("%s (%s)" % (code(u["update"]), _count(u["sessions"], "session"))
for u in untested),
"a model change" if by == "model" else "an update", min_sessions, min_turns))
if counts["untestable"]:
notes.append("%s could not reach p under %g with this many sessions and %s left out of the adjustment "
"for the number of tests." % (_count(counts["untestable"], "comparison"), P_LIMIT,
"was" if counts["untestable"] == 1 else "were"))
if counts["sparse"]:
notes.append("%s of rare events %s not tested: the busier side had fewer than %d sessions with the "
"event." % (_count(counts["sparse"], "comparison"), "was" if counts["sparse"] == 1 else "were",
MIN_EVENT_SESSIONS))
same = sum(1 for c in comparisons if c.passes and not c.flagged and not c.stand_out)
if same:
notes.append("%s showed the same changes from a neighboring update; each change is listed once." % (
_count(same, "more window")))
silent = [s.key for s in slices if not _reasoning_recorded(s.turns)]
if silent and len(silent) == len(slices):
notes.append("No reasoning tokens are recorded here, so the reasoning share is left out.")
elif silent:
notes.append("The reasoning share is left out for %s: most of their turns record no reasoning tokens." % (
_names(silent[:8]) + (" and %d more" % (len(silent) - 8) if len(silent) > 8 else "")))
if turns and since_days:
days_back = (time.time() - min(t.t for t in turns)) / 86400.0
if days_back < since_days - 7:
text = "The oldest turn found is %d days old, though the window is %d days." % (int(days_back), since_days)
if days_back <= 35 and harness in _RETENTION: # a history that fits a 30-day cleanup
text += " " + _RETENTION[harness]
notes.append(text)
return notes
def evaluate(turns, by="version", min_turns=MIN_TURNS, min_sessions=MIN_SESSIONS, harness=None, info=None,
since_days=None, project=None) -> dict:
"""Slice the turns, test each update window, and build the result (the --json shape)."""
harness = harness or (turns[0].harness if turns else "claude-code")
slices, slice_notes = make_slices(turns, by, min_turns, min_sessions)
wins = windows(slices, by, min_turns, min_sessions)
comparisons, counts = compare(slices, wins)
stand_outs = _stand_outs(by, slices, comparisons)
flagged = [c for c in comparisons if c.flagged]
confounders = _confounders(by, slices, flagged, wins, bool(project))
tested = {w.i for w in wins}
untested = [{"update": clean_key(slices[i].key), "sessions": _sessions(slices[i].turns)}
for i in range(1, len(slices)) if i not in tested]
recorded = [_reasoning_recorded(s.turns) for s in slices]
result = {
"harness": harness, "by": by, "since_days": since_days,
"project": _safe(_home(project), 200) if project else None,
"headline": _headline(by, harness, turns, slices, wins, flagged, stand_outs, info, since_days, min_sessions,
confounders, slice_notes, project),
"totals": {"turns": len(turns), "sessions": _sessions(turns), "slices": len(slices),
"updates_tested": len(wins)},
"thresholds": {"p_adjusted_below": P_LIMIT, "min_change_pct": MIN_CHANGE_PCT, "min_turns": min_turns,
"min_sessions": min_sessions, "min_session_values": MIN_SESSION_VALUES,
"min_event_sessions": MIN_EVENT_SESSIONS},
"slices": [], "tests": [_comparison_dict(c, slices, by) for c in comparisons],
"flagged": [_comparison_dict(c, slices, by) for c in flagged], "stand_outs": stand_outs,
"confounders": confounders, "untested": untested,
"left_out": {"versions": [{"version": clean_key(k), "first_day": _day(a), "last_day": _day(b), "sessions": n}
for k, a, b, n in slice_notes["out_of_order"]],
"models": [{"model": clean_key(k), "turns": n, "sessions": s}
for k, n, s in slice_notes["left_out"]]},
"notes": _notes(by, harness, turns, slices, slice_notes, wins, comparisons, counts, info, since_days,
min_turns, min_sessions, untested),
"svg": None,
}
for j, sl in enumerate(slices):
result["slices"].append({
"key": clean_key(sl.key), "first_day": _day(sl.turns[0].t), "last_day": _day(sl.turns[-1].t),
"sessions": _sessions(sl.turns), "turns": len(sl.turns),
"values": {k: {"value": _num(v), "n": n} for k, (v, n) in group_values(sl.turns).items()},
"per_session": {m.key: None if m.key == "reasoning_share" and not recorded[j]
else _num(session_summary(m.key, session_values(m.key, sl.turns))) for m in METRICS},
})
return result
def analyze(harness="claude-code", by="version", since_days=90, project=None, min_turns=MIN_TURNS,
min_sessions=MIN_SESSIONS, home=None) -> dict:
"""Read the sessions and evaluate them: the whole run behind the report."""
turns, info = collect_turns(harness, since_days=since_days, project=project, home=home)
return evaluate(turns, by, min_turns, min_sessions, harness=harness, info=info, since_days=since_days,
project=project)
# ---------------------------------------------------------------------------
# Markdown report
# ---------------------------------------------------------------------------
SHORT = {"reads_before_edit": "Reads before edit", "reads_per_edit": "Reads per edit",
"blind_edits": "Edits to unread files %", "edits_per_turn": "Edits/turn", "tool_error_rate": "Tool errors %",
"interrupts": "Interrupts/100", "corrections": "Corrections/100", "output_tokens": "Output tokens",
"reasoning_share": "Reasoning %", "tool_calls": "Tool calls", "cost": "Cost/turn"}
CHANGES = {"version": "update", "model": "model change", "week": "week-to-week change"}
def _fmt(key, value) -> str:
if value is None:
return "n/a"
if key == "cost":
return "$%.3f" % value if value >= 0.001 else "$%.4f" % value
if key == "output_tokens":
return "{:,}".format(int(round(value)))
if METRIC[key].scale == 100 or key == "edits_per_turn":
return "%.1f" % value if METRIC[key].scale == 100 else "%.2f" % value
return "%d" % value if value == int(value) else "%.1f" % value
def _fmt_change(c) -> str:
return "from 0" if c["from_zero"] else "%+d%%" % c["change_pct"]
def _fmt_p(p) -> str:
return "< 0.0001" if p < 0.0001 else "%.4f" % p
def _days(first, last) -> str:
return first if first == last else "%s to %s" % (first, last)
def render_markdown(result) -> str:
by, name = result["by"], HARNESS_NAMES.get(result["harness"], result["harness"])
lines = ["**%s**" % result["headline"], ""]
totals = result["totals"]
if totals["turns"]:
where = " in %s" % code(result["project"], 200) if result["project"] else ""
window = ", last %d days" % result["since_days"] if result["since_days"] else ""
text = "%s%s%s, split by %s: %s in %s across %d %s, with %s tested." % (
name, where, window, by, _count(totals["turns"], "turn"), _count(totals["sessions"], "session"),
totals["slices"], UNITS[by][0] if totals["slices"] == 1 else UNITS[by][1],
_count(totals["updates_tested"], CHANGES[by]))
if result["tests"]:
text += (" Each session counts once: the test compares one value per session, and a change is flagged "
"when p is under %g after adjusting for the number of tests and the change is at least %d%%."
% (P_LIMIT, int(MIN_CHANGE_PCT)))
lines += [text, ""]
if result["flagged"]:
lines += ["## Flagged changes", "",
"| Where | What changed | Before (per session) | After (per session) | Change | Usually means "
"| Compared | Sessions (before, after) | Adjusted p |",
"|---|---|---|---|---|---|---|---|---|"]
for c in result["flagged"]:
lines.append("| %s | %s (session %s) | %s | %s | %s | %s | %s vs %s | %d, %d | %s |" % (
_code_keys(c["where"]), c["label"], c["measured"].split()[0], _fmt(c["metric"], c["before_value"]),
_fmt(c["metric"], c["after_value"]), _fmt_change(c), c["usually"], _code_keys(c["before"]),
_code_keys(c["after"]),
c["sessions_before"], c["sessions_after"], _fmt_p(c["p_adjusted"])))
lines.append("")
if result["stand_outs"]:
lines += ["## Stands out", ""] + ["- %s" % s["text"] for s in result["stand_outs"]] + [""]
if result["confounders"]:
lines += ["## What else changed at the same point", "",
"These numbers show what changed, not why: the kind of work may have changed too.", ""]
lines += ["- %s: %s" % (_code_keys(c["where"]), c["text"]) for c in result["confounders"]]
lines.append("")
if result["slices"]:
first = UNITS[by][0].capitalize()
lines += ["## By %s" % by, "",
"| %s | Days | Sessions | Turns | %s |" % (first, " | ".join(SHORT[m.key] for m in METRICS)),
"|---|---|---|---|%s" % ("---|" * len(METRICS))]
for sl in result["slices"]:
lines.append("| %s | %s | %d | %d | %s |" % (
code(sl["key"]), _days(sl["first_day"], sl["last_day"]), sl["sessions"], sl["turns"],
" | ".join(_fmt(m.key, sl["values"][m.key]["value"]) for m in METRICS)))
lines.append("")
if result["notes"]:
lines += ["## Notes", ""] + ["- " + n for n in result["notes"]] + [""]
return "\n".join(lines)
# ---------------------------------------------------------------------------
# SVG: one small line chart per flagged metric, no dependencies
# ---------------------------------------------------------------------------
_INK, _LINE, _FLAG = "#767676", "#4e79a7", "#d9730d" # mid tones that read on light and dark backgrounds
def _attr(text) -> str:
return escape(str(text), {'"': """})
def render_svg(result) -> Optional[str]:
"""The chart as SVG text, or None when nothing is flagged. It plots the same
per-session values the flagged changes are measured in."""
if not result["flagged"]:
return None
order = [m.key for m in METRICS]
metrics = sorted({c["metric"] for c in result["flagged"]}, key=order.index)
slices = result["slices"]
width, height, top = 560, 170, 36
left, right = 64, width - 24
out = ['<svg xmlns="http://www.w3.org/2000/svg" width="%d" height="%d" viewBox="0 0 %d %d" role="img" '
'font-family="system-ui, -apple-system, Segoe UI, sans-serif" font-size="11">' % (
width, top + height * len(metrics), width, top + height * len(metrics)),
"<title>%s</title>" % escape(result["headline"].replace("`", "")), # no inline-code marks in a chart
'<text x="12" y="22" fill="%s" font-size="13">%s by %s, one value per session</text>' % (
_INK, escape(HARNESS_NAMES.get(result["harness"], result["harness"])), escape(result["by"]))]
n = len(slices)
def x_at(i):
return left + (right - left) * (i / float(n - 1) if n > 1 else 0.5)
for k, key in enumerate(metrics):
y0 = top + k * height
plot_top, plot_bottom = y0 + 26, y0 + height - 34
points = [(i, sl["per_session"][key]) for i, sl in enumerate(slices) if sl["per_session"][key] is not None]
low, high = min(v for _i, v in points), max(v for _i, v in points)
if high == low:
low, high = low - 1, high + 1
def y_at(v):
return plot_bottom - (v - low) / float(high - low) * (plot_bottom - plot_top)
out.append('<g class="chart" data-metric="%s">' % _attr(key))
out.append('<text x="12" y="%d" fill="%s" font-size="12">%s, %s per session</text>' % (
y0 + 14, _INK, escape(METRIC[key].label), METRIC[key].session_summary))
out.append('<line x1="%d" y1="%.1f" x2="%d" y2="%.1f" stroke="%s" stroke-width="1"/>' % (
left, plot_bottom, right, plot_bottom, _INK))
for value in (high, low):
out.append('<text x="%d" y="%.1f" fill="%s" text-anchor="end">%s</text>' % (
left - 6, y_at(value) + 4, _INK, escape(_fmt(key, value))))
out.append('<polyline fill="none" stroke="%s" stroke-width="2" points="%s"/>' % (
_LINE, " ".join("%.1f,%.1f" % (x_at(i), y_at(v)) for i, v in points)))
for i, v in points:
out.append('<circle cx="%.1f" cy="%.1f" r="3" fill="%s"/>' % (x_at(i), y_at(v), _LINE))
keys = [sl["key"] for sl in slices]
for c in (c for c in result["flagged"] if c["metric"] == key and c["update"] in keys):
u = keys.index(c["update"])
x = (x_at(u - 1) + x_at(u)) / 2.0 if u > 0 else x_at(u)
out.append('<line x1="%.1f" y1="%d" x2="%.1f" y2="%.1f" stroke="%s" stroke-width="1.5" '
'stroke-dasharray="4 3"/>' % (x, plot_top - 6, x, plot_bottom, _FLAG))
out.append('<text x="%.1f" y="%d" fill="%s" text-anchor="middle">%s: %s</text>' % (
x, plot_top - 9, _FLAG, escape(c["where"]), escape(_fmt_change(c))))
out.append('<text x="%d" y="%.1f" fill="%s">%s</text>' % (left, plot_bottom + 16, _INK, escape(keys[0])))
out.append('<text x="%d" y="%.1f" fill="%s" text-anchor="end">%s</text>' % (
right, plot_bottom + 16, _INK, escape(keys[-1])))
out.append("</g>")
out.append("</svg>")
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# Command line
# ---------------------------------------------------------------------------
def parse_since(text) -> int:
m = re.match(r"^\s*(\d+)\s*d?\s*$", str(text or ""))
if not m or int(m.group(1)) < 1:
raise ValueError("expected a number of days, such as 90d")
return int(m.group(1))
def _write(path, text) -> Optional[str]:
try:
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
except OSError as exc:
return "error: cannot write %s: %s" % (path, exc.strerror or exc)
return None
def main(argv=None) -> int:
parser = argparse.ArgumentParser(
prog="regress.py", description=__doc__.strip().split("\n\n")[0],
epilog="Exit codes: 0 done, 1 a flagged change matches --fail-on, 2 usage error.")
parser.add_argument("--since", default="90d", help="how far back to read, in days, such as 30d (default 90d)")
parser.add_argument("--harness", default="claude-code", choices=transcripts.HARNESSES,
help="which agent's sessions to read (default claude-code); Cursor keeps no usable transcripts")
parser.add_argument("--by", default="version", choices=("version", "model", "week"),
help="split by harness version (default), model, or ISO week")
parser.add_argument("--project", metavar="PATH", help="only sessions whose working folder is PATH or inside it")
parser.add_argument("--min-turns", type=int, default=MIN_TURNS,
help="turns each side of an update needs before it is tested (default %d)" % MIN_TURNS)
parser.add_argument("--min-sessions", type=int, default=MIN_SESSIONS,
help="sessions each side of an update needs (default %d, at least %d)" % (
MIN_SESSIONS, MIN_SESSION_VALUES))
parser.add_argument("--svg", metavar="PATH", help="also write a line chart of each flagged change to PATH")
parser.add_argument("--json", action="store_true", help="print machine-readable JSON")
parser.add_argument("--out", metavar="PATH", help="write the report to this file instead of printing it")
parser.add_argument("--fail-on", choices=("worse", "any"),
help="exit 1 when a flagged change usually means worse (worse) or when anything is flagged (any)")
args = parser.parse_args(argv)
try:
since = parse_since(args.since)
except ValueError:
parser.error("--since takes a number of days, such as 90d")
if args.min_sessions < MIN_SESSION_VALUES:
parser.error("--min-sessions must be at least %d" % MIN_SESSION_VALUES)
if args.min_turns < 1:
parser.error("--min-turns must be at least 1")
result = analyze(args.harness, args.by, since, args.project, args.min_turns, args.min_sessions)
chart_line = ""
if args.svg:
svg = render_svg(result)
if svg is None:
chart_line = "No chart written: nothing was flagged."
else:
error = _write(args.svg, svg)
if error:
print(error, file=sys.stderr)
return 2
result["svg"] = args.svg
chart_line = "Chart written to %s" % args.svg
if args.json:
text = json.dumps(result, indent=2) + "\n"
else:
text = render_markdown(result) + ("\n" + chart_line + "\n" if chart_line else "")
if args.out:
error = _write(args.out, text)
if error:
print(error, file=sys.stderr)
return 2
print("Report written to %s" % args.out)
if chart_line and args.json:
print(chart_line)
else:
sys.stdout.write(text)
if args.fail_on and any(c["usually"] == "worse" or args.fail_on == "any" for c in result["flagged"]):
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
# 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)")
# Copied from skills/evals/shared/transcripts.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Read coding-agent session transcripts into one simple shape.
Supported: Claude Code, Codex CLI, Gemini CLI, and OpenCode. OpenCode support
is best effort: it follows the documented database schema and is not verified
on a real install. Cursor keeps no usable transcripts (no tool results, token
usage, or timestamps), so it is skipped.
Read-only, Python 3.9+, standard library only. Every format rule comes from
the harness facts file (research-harness-facts.md, section Q1) plus read-only
checks of real files. The formats are internal to each harness and change
between versions, so unknown records are counted in Session.warnings and never
stop the parse.
Token counts use Anthropic meanings for every harness: `input` is uncached
input, cache reads and writes are separate, and `output` includes reasoning.
"""
from __future__ import annotations
import datetime
import hashlib
import json
import os
import pathlib
import re
import shlex
import sqlite3
import time
from dataclasses import dataclass, field, fields
from typing import Optional
# safe.py sits next to this file (sync_shared.py copies both). Skills reach these
# names through this module too, as transcripts.safe_text and transcripts.redact.
from safe import code, redact, safe_text # noqa: F401
HARNESSES = ("claude-code", "codex", "gemini-cli", "opencode")
MAX_OUTPUT_CHARS = 20000
@dataclass
class Usage:
input: int = 0 # uncached input tokens
cache_read: int = 0
cache_write: int = 0 # all cache-write tokens (5-minute plus 1-hour)
cache_write_1h: int = 0 # the 1-hour part of cache_write, when known
output: int = 0 # includes reasoning or thinking tokens
reasoning: int = 0 # the reasoning part of output, when known
def total_input(self) -> int:
return self.input + self.cache_read + self.cache_write
def __add__(self, other: "Usage") -> "Usage":
return Usage(**{f.name: getattr(self, f.name) + getattr(other, f.name) for f in fields(self)})
@dataclass
class ToolCall:
id: str
name: str # harness-native name: Bash, exec_command, run_shell_command, mcp__x__y, ...
kind: str # "shell" | "read" | "edit" | "write" | "search" | "web" | "agent" | "mcp" | "other"
input: dict # parsed input; free-form input becomes {"raw": "<text>"}
command: str = "" # the shell command or script text when kind == "shell"
paths: list = field(default_factory=list) # files read or changed, when known
ts: str = ""
output: str = "" # result text, cut to MAX_OUTPUT_CHARS
output_chars: int = 0 # full result length before the cut
has_result: bool = False
is_error: bool = False
exit_code: Optional[int] = None # only when the harness records it (or, for Claude Code, implies 0)
denied: str = "" # "" | "user-rejected" | "permission-rule" | "hook" | "auto-reviewer" | "other"
interrupted: bool = False
@dataclass
class Event:
kind: str # "user" | "assistant" | "tool" | "interrupt" | "compaction" | "api_error"
ts: str = "" # ISO 8601 UTC
text: str = "" # user or assistant text; compaction summary; error text
injected: bool = False # user-role content the harness added (context files, reminders, hook output)
tool: Optional[ToolCall] = None # kind == "tool"
usage: Optional[Usage] = None # kind == "assistant": one Event with usage per API response
model: str = ""
sidechain: bool = False
id: str = "" # the harness id behind the event (API response id, record id, call id), when known.
# A forked or resumed Claude Code session can begin with a copy of the earlier
# session's records, ids included: when totaling several sessions, count each id once.
mode: str = "" # Claude Code permission mode in effect for a tool call (the latest
# user record's permissionMode: default, plan, acceptEdits, auto, bypassPermissions); "" if unknown.
@dataclass
class Session:
harness: str
id: str
path: str # the file; for OpenCode '<database file>#<session id>'
cwd: str = ""
version: str = "" # Claude Code record "version", Codex cli_version, OpenCode version; else ""
models: list = field(default_factory=list)
started: str = ""
ended: str = ""
is_subagent: bool = False
parent_id: str = ""
events: list = field(default_factory=list)
warnings: list = field(default_factory=list) # skipped lines and unknown records: no content
def tool_calls(self) -> list:
return [e.tool for e in self.events if e.kind == "tool" and e.tool is not None]
def usage_total(self) -> Usage:
total = Usage()
for e in self.events:
if e.usage is not None:
total = total + e.usage
return total
def usage_by_model(self) -> dict:
out = {}
for e in self.events:
if e.usage is not None:
out[e.model] = out.get(e.model, Usage()) + e.usage
return out
def user_prompts(self) -> list:
return [e for e in self.events if e.kind == "user" and not e.injected]
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
_SAFE_NAME_RE = re.compile(r"^[A-Za-z0-9_.:\-]{1,40}$")
def _warn(session, lineno, what) -> None:
session.warnings.append("line %d: %s" % (lineno, what) if lineno else what)
def _name(value) -> str:
"""A record type name that is safe to put in a warning (no content)."""
return repr(value) if isinstance(value, str) and _SAFE_NAME_RE.match(value) else "(unnamed)"
def _add(session, event) -> Event:
session.events.append(event)
return event
def _set_output(call, text) -> None:
text = text if isinstance(text, str) else ""
call.output_chars = len(text)
call.output = text[:MAX_OUTPUT_CHARS]
def _block_text(content) -> str:
"""Text of a string or of a list of content blocks (non-text blocks skipped)."""
if isinstance(content, str):
return content
if isinstance(content, list):
return "\n".join(b.get("text") or "" for b in content
if isinstance(b, dict) and isinstance(b.get("text"), str))
return ""
def _as_dict(value) -> dict:
return value if isinstance(value, dict) else {}
def _int(value) -> int:
return value if isinstance(value, int) and not isinstance(value, bool) else 0
def _track_time(session, ts) -> None:
if isinstance(ts, str) and ts:
if not session.started or ts < session.started:
session.started = ts
if ts > session.ended:
session.ended = ts
def _raw_lines(path, start=0, complete_only=False):
"""Yield (lineno, offset, raw bytes) per line. With complete_only, a last line
that has no newline yet (a file still being written) is not read."""
with open(path, "rb") as fh:
fh.seek(start)
offset = start
for lineno, raw in enumerate(fh, 1):
if complete_only and not raw.endswith(b"\n"):
break
yield lineno, offset, raw
offset += len(raw)
def _parse(session, lineno, raw):
if not raw.strip():
return None
try:
rec = json.loads(raw)
except ValueError:
_warn(session, lineno, "malformed JSON line")
return None
if not isinstance(rec, dict):
_warn(session, lineno, "line is not a JSON object")
return None
return rec
def _records(path, session):
"""Yield (lineno, record) for each JSON line; bad lines become warnings."""
try:
for lineno, _offset, raw in _raw_lines(path):
rec = _parse(session, lineno, raw)
if rec is not None:
yield lineno, rec
except OSError as exc:
_warn(session, 0, "cannot read file: %s" % type(exc).__name__)
# ---------------------------------------------------------------------------
# Claude Code (facts Q1.1)
# ---------------------------------------------------------------------------
_CC_KINDS = {
"Bash": "shell", "Monitor": "shell", "Read": "read", "NotebookRead": "read", "Edit": "edit", "MultiEdit": "edit",
"NotebookEdit": "edit", "Write": "write", "Grep": "search", "Glob": "search", "LS": "search",
"WebFetch": "web", "WebSearch": "web", "Agent": "agent", "Task": "agent",
}
_CC_METADATA = {
"last-prompt", "custom-title", "ai-title", "agent-name", "queue-operation", "bridge-session",
"atis-latch", "mode", "file-history-snapshot", "file-history-delta", "cost-state", "summary",
"pr-link", "frame-link", "artifact-comment-monitor", "artifact-autoreact-ledger", "relocated",
}
_CC_INTERRUPTS = ("[Request interrupted by user]", "[Request interrupted by user for tool use]")
_CC_INJECTED_TAGS = ("<system-reminder>", "<task-notification>", "<local-command-stdout>",
"<local-command-stderr>", "<local-command-caveat>", "<bash-stdout>",
"<bash-stderr>")
_CC_DENIALS = {"user-rejected": "user-rejected", "permission-rule": "permission-rule",
"automode-blocked": "auto-reviewer"}
_CC_EXIT_RE = re.compile(r"Exit code (-?\d+)")
def _cc_kind(name) -> str:
if name.startswith("mcp__"):
return "mcp"
return _CC_KINDS.get(name, "other")
def _cc_usage(u) -> Optional[Usage]:
if not isinstance(u, dict):
return None
write = _int(u.get("cache_creation_input_tokens"))
return Usage(
input=_int(u.get("input_tokens")),
cache_read=_int(u.get("cache_read_input_tokens")),
cache_write=write,
cache_write_1h=min(write, _int(_as_dict(u.get("cache_creation")).get("ephemeral_1h_input_tokens"))),
output=_int(u.get("output_tokens")),
reasoning=_int(_as_dict(u.get("output_tokens_details")).get("thinking_tokens")),
)
class _Claude:
"""Turns Claude Code records into Events, in file order."""
def __init__(self, session):
self.s = session
self.responses = {} # message.id (or requestId) -> assistant Event
self.calls = {} # tool_use id -> ToolCall
self.hook_denied = set()
self.last_compaction = None
self.uuid = "" # id of the record being read
self.mode = "" # latest permissionMode seen on a user record
def feed(self, lineno, rec) -> None:
kind = rec.get("type")
if kind in ("user", "assistant", "system", "attachment"):
_track_time(self.s, rec.get("timestamp"))
if not self.s.cwd and isinstance(rec.get("cwd"), str):
self.s.cwd = rec["cwd"]
if not self.s.version and isinstance(rec.get("version"), str):
self.s.version = rec["version"]
self.uuid = rec.get("uuid") if isinstance(rec.get("uuid"), str) else ""
getattr(self, "_" + kind)(lineno, rec, rec.get("timestamp") or "", bool(rec.get("isSidechain")))
elif kind is None:
_warn(self.s, lineno, "record without a type")
elif kind not in _CC_METADATA:
_warn(self.s, lineno, "unknown record type %s" % _name(kind))
def _assistant(self, lineno, rec, ts, side) -> None:
msg = _as_dict(rec.get("message"))
model = msg.get("model") if isinstance(msg.get("model"), str) else ""
blocks = msg.get("content")
if isinstance(blocks, str):
blocks = [{"type": "text", "text": blocks}]
blocks = [b for b in blocks if isinstance(b, dict)] if isinstance(blocks, list) else []
texts = [b["text"] for b in blocks if b.get("type") == "text" and isinstance(b.get("text"), str)]
if model == "<synthetic>": # harness-made message, not an API response: no usage
kind = "api_error" if rec.get("isApiErrorMessage") else "assistant"
_add(self.s, Event(kind, ts, "\n".join(texts), model="" if kind == "api_error" else model,
sidechain=side, id=self.uuid))
return
key = msg.get("id") or rec.get("requestId") or "line-%d" % lineno
usage = _cc_usage(msg.get("usage"))
ev = self.responses.get(key)
if ev is None: # first record of this API response
ev = self.responses[key] = _add(self.s, Event("assistant", ts, usage=usage, model=model,
sidechain=side, id=str(key)))
if model and model not in self.s.models:
self.s.models.append(model)
elif usage is not None: # later records repeat the input side and grow the output
ev.usage = usage
if texts:
ev.text = "\n".join([ev.text] + texts) if ev.text else "\n".join(texts)
for b in blocks:
if b.get("type") == "tool_use":
call = self._call(b, ts)
_add(self.s, Event("tool", ts, tool=call, model=model, sidechain=side, id=call.id,
mode=self.mode))
def _call(self, b, ts) -> ToolCall:
name = b.get("name") if isinstance(b.get("name"), str) else ""
inp = b.get("input")
inp = inp if isinstance(inp, dict) else ({} if inp is None else {"raw": str(inp)})
call = ToolCall(id=str(b.get("id") or ""), name=name, kind=_cc_kind(name), input=inp, ts=ts)
if call.kind == "shell":
call.command = str(inp.get("command") or "")
elif call.kind in ("read", "edit", "write"):
path = inp.get("file_path") or inp.get("notebook_path")
if isinstance(path, str) and path:
call.paths = [path]
self.calls[call.id] = call
return call
def _user(self, lineno, rec, ts, side) -> None:
if isinstance(rec.get("permissionMode"), str) and rec["permissionMode"]:
self.mode = rec["permissionMode"]
content = _as_dict(rec.get("message")).get("content")
blocks = [{"type": "text", "text": content}] if isinstance(content, str) else content
blocks = [b for b in blocks if isinstance(b, dict)] if isinstance(blocks, list) else []
if rec.get("isCompactSummary"):
summary = _block_text(blocks)
if self.last_compaction is not None and not self.last_compaction.text:
self.last_compaction.text = summary
else:
self.last_compaction = _add(self.s, Event("compaction", ts, summary, sidechain=side, id=self.uuid))
return
origin = rec.get("origin")
origin_kind = origin.get("kind") if isinstance(origin, dict) else None
harness_made = bool(rec.get("isMeta")) or origin_kind not in (None, "human")
run, run_injected = [], False
for b in blocks:
if b.get("type") == "tool_result":
self._result(lineno, b, rec)
continue
if b.get("type") != "text" or not isinstance(b.get("text"), str):
continue
txt = b["text"]
if txt.strip() in _CC_INTERRUPTS:
self._flush(run, run_injected, ts, side)
run = []
_add(self.s, Event("interrupt", ts, txt.strip(), sidechain=side, id=self.uuid))
continue
injected = harness_made or txt.lstrip().startswith(_CC_INJECTED_TAGS)
if run and injected != run_injected:
self._flush(run, run_injected, ts, side)
run = []
run.append(txt)
run_injected = injected
self._flush(run, run_injected, ts, side)
def _flush(self, texts, injected, ts, side) -> None:
if texts:
_add(self.s, Event("user", ts, "\n".join(texts), injected=injected, sidechain=side, id=self.uuid))
def _result(self, lineno, b, rec) -> None:
call = self.calls.get(b.get("tool_use_id"))
if call is None:
_warn(self.s, lineno, "tool result without a matching call")
return
txt = _block_text(b.get("content"))
call.has_result = True
call.is_error = bool(b.get("is_error"))
_set_output(call, txt)
extra = rec.get("toolUseResult")
extra = extra if isinstance(extra, dict) else None
if extra is not None and extra.get("interrupted") is True:
call.interrupted = True
denial = rec.get("toolDenialKind")
if isinstance(denial, str) and denial:
call.denied = _CC_DENIALS.get(denial, "other")
call.is_error = True
if call.kind == "shell":
m = _CC_EXIT_RE.match(txt) if call.is_error else None
if m:
call.exit_code = int(m.group(1))
elif (not call.is_error and not call.interrupted and extra is not None
and not extra.get("returnCodeInterpretation") and not extra.get("backgroundTaskId")):
call.exit_code = 0 # the Bash tool flags every nonzero exit it does not explain
def _system(self, lineno, rec, ts, side) -> None:
sub = rec.get("subtype")
if sub == "compact_boundary":
self.last_compaction = _add(self.s, Event("compaction", ts, sidechain=side, id=self.uuid))
elif sub == "api_error":
err = _as_dict(rec.get("error"))
words = [str(err[k]) for k in ("status", "message") if err.get(k) not in (None, "")]
_add(self.s, Event("api_error", ts, " ".join(words), sidechain=side, id=self.uuid))
def _attachment(self, lineno, rec, ts, side) -> None:
att = _as_dict(rec.get("attachment"))
kind = att.get("type")
if (kind == "hook_blocking_error" and att.get("hookEvent") in ("PreToolUse", "PermissionRequest")
and att.get("toolUseID")):
self.hook_denied.add(att["toolUseID"])
elif kind == "hook_additional_context":
content = att.get("content")
txt = "\n".join(str(x) for x in content) if isinstance(content, list) else str(content or "")
if txt:
_add(self.s, Event("user", ts, txt, injected=True, sidechain=side, id=self.uuid))
def finish(self) -> None:
for tool_id in self.hook_denied:
call = self.calls.get(tool_id)
if call is not None and not call.denied:
call.denied = "hook"
call.is_error = True
def _cc_session(path) -> Session:
stem = os.path.basename(path)[:-len(".jsonl")] if path.endswith(".jsonl") else os.path.basename(path)
# Subagents live in <session>/subagents/, Workflow agents in <session>/subagents/workflows/<id>/.
folder = os.path.dirname(path)
for _ in range(3):
if os.path.basename(folder) == "subagents" and stem.startswith("agent-"):
return Session("claude-code", stem[len("agent-"):], path, is_subagent=True,
parent_id=os.path.basename(os.path.dirname(folder)))
folder = os.path.dirname(folder)
return Session("claude-code", stem, path)
def _load_claude(path) -> Session:
s = _cc_session(path)
reader = _Claude(s)
for lineno, rec in _records(path, s):
reader.feed(lineno, rec)
reader.finish()
return s
def _cc_newest_response(rows):
"""Index of the first record of the newest API response, or None."""
first_seen = {}
for i, (_lineno, rec) in enumerate(rows):
if rec.get("type") == "assistant":
msg = _as_dict(rec.get("message"))
if msg.get("model") != "<synthetic>":
first_seen.setdefault(msg.get("id") or rec.get("requestId") or i, i)
return max(first_seen.values()) if first_seen else None
# ---------------------------------------------------------------------------
# Codex CLI (facts Q1.2)
# ---------------------------------------------------------------------------
# Response items the model produces; the first one after a usage line starts a response.
_CX_MODEL_ITEMS = {"reasoning", "function_call", "custom_tool_call", "local_shell_call", "web_search_call",
"tool_search_call", "image_generation_call"}
_CX_OUTPUTS = {"function_call_output", "custom_tool_call_output", "tool_search_output",
"local_shell_call_output"}
_CX_QUIET_ITEMS = {"compaction", "agent_message", "ghost_snapshot"}
_CX_QUIET_LINES = {"world_state", "inter_agent_communication", "inter_agent_communication_metadata",
"retained_context", "security_risk_score", "realtime_item"}
_CX_SHELL = {"shell", "exec_command", "shell_command", "container.exec", "local_shell"}
_CX_DENIALS = (("exec command rejected by user", "user-rejected"), ("patch rejected by user", "user-rejected"),
("rejected by configuration", "hook"),
("automatic approval review denied the action", "auto-reviewer"))
_CX_EXIT_RES = (re.compile(r"^Exit code: (-?\d+)", re.M), re.compile(r"^Process exited with code (-?\d+)", re.M))
_CX_PATCH_RE = re.compile(r"^\*\*\* (?:Update|Add|Delete) File: (.+)$|^\*\*\* Move to: (.+)$", re.M)
# Context the harness adds as a user message: one whole <tag>...</tag> block, or AGENTS.md text.
_CX_CONTEXT_RE = re.compile(r"^\s*(?:<([A-Za-z_][A-Za-z0-9_]*)[\s>].*</\1>\s*$|# AGENTS\.md instructions)", re.S)
# Matches the record type key only: inside text values the quotes are escaped.
_CX_USAGE_MARK = re.compile(rb'"type"\s*:\s*"token_usage_record"')
_CX_NAME_RE = re.compile(r"rollout-\d{4}-\d\d-\d\dT\d\d-\d\d-\d\d-(.+?)\.jsonl")
def _cx_kind(name, namespace) -> str:
if namespace.startswith("mcp__") or name.startswith("mcp__"):
return "mcp"
if not namespace and name in _CX_SHELL:
return "shell"
if name == "apply_patch":
return "edit"
if name == "view_image":
return "read"
return {"collaboration": "agent", "web": "web"}.get(namespace, "other")
def _cx_argv(argv) -> str:
"""The script text of a [shell, -lc, script] command, else the joined argv."""
if isinstance(argv, str):
return argv
if not isinstance(argv, list) or not all(isinstance(a, str) for a in argv):
return ""
if len(argv) >= 3 and argv[-2] in ("-c", "-lc", "-ic", "-lic"):
return argv[-1]
return shlex.join(argv)
def _cx_usage(u) -> Optional[Usage]:
"""OpenAI meanings (cached is part of input, reasoning part of output) to Usage."""
if not isinstance(u, dict):
return None
cached = _int(u.get("cached_input_tokens"))
return Usage(input=max(0, _int(u.get("input_tokens")) - cached), cache_read=cached,
cache_write=_int(u.get("cache_write_input_tokens")), output=_int(u.get("output_tokens")),
reasoning=_int(u.get("reasoning_output_tokens")))
def _cx_text(output) -> str:
if isinstance(output, str):
return output
if isinstance(output, list):
return "\n".join(x["text"] for x in output if isinstance(x, dict) and isinstance(x.get("text"), str))
return ""
def _cx_denial(text) -> str:
"""Denial kind from the start of a tool output (where Codex puts the message);
output that merely quotes the words further down is not a denial."""
head = text[:200]
for marker, kind in _CX_DENIALS:
if marker in head:
return kind
return ""
class _Codex:
"""Turns Codex rollout lines into Events, in file order.
Usage comes from token_usage_record lines (one per response id) when the file
has them, otherwise from token_count events with repeats removed. In code
mode the model calls one `exec` script that runs the real operations; each
operation is logged as an item_completed event, so those become the tool
calls and the script call itself is kept only when it ran none."""
def __init__(self, session, use_records):
self.s = session
self.use_records = use_records
self.model = ""
self.meta_seen = False
self.all_injected = False # review subagents: every user message is harness-made
self.pending = None # assistant Event of the response being read
self.pending_at = None # where that Event goes once it has text or usage
self.calls = {} # call_id -> ToolCall
self.open_scripts = [] # exec call ids still waiting for their output
self.script_events = {} # exec call id -> [Event, number of operations it ran]
self.dropped = set()
self.response_ids = set()
self.last_total = None
# -- helpers ------------------------------------------------------------
def _assistant(self, ts) -> Event:
if self.pending is None:
self.pending = Event("assistant", ts, model=self.model)
at = len(self.s.events) if self.pending_at is None else self.pending_at
self.s.events.insert(at, self.pending)
self.pending_at = None
return self.pending
def _close(self) -> None:
self.pending, self.pending_at = None, None
def _set_model(self, model) -> None:
if isinstance(model, str) and model:
self.model = model
if model not in self.s.models:
self.s.models.append(model)
def _usage(self, ts, usage, response_id="") -> None:
if usage is None:
return
ev = self._assistant(ts)
ev.usage, ev.id = usage, response_id
ev.model = ev.model or self.model
self._close()
def _tool(self, ts, call) -> Event:
return _add(self.s, Event("tool", ts, tool=call, model=self.model, id=call.id))
# -- dispatch -----------------------------------------------------------
def feed(self, lineno, rec) -> None:
kind, ts = rec.get("type"), rec.get("timestamp") or ""
_track_time(self.s, ts)
p = _as_dict(rec.get("payload"))
if kind == "session_meta":
self._meta(p)
elif kind == "turn_context":
self._close()
self._set_model(p.get("model"))
elif kind == "response_item":
self._item(lineno, ts, p)
elif kind == "event_msg":
self._event(ts, p)
elif kind == "token_usage_record":
if self.use_records and p.get("response_id") not in self.response_ids:
self.response_ids.add(p.get("response_id"))
self._usage(ts, _cx_usage(p.get("usage")), str(p.get("response_id") or ""))
elif kind == "compacted":
_add(self.s, Event("compaction", ts, str(p.get("message") or "")))
elif kind is None:
_warn(self.s, lineno, "record without a type")
elif kind not in _CX_QUIET_LINES:
_warn(self.s, lineno, "unknown record type %s" % _name(kind))
def _meta(self, p) -> None:
if self.meta_seen: # a later session_meta describes the thread this one forked from
return
self.meta_seen = True
s = self.s
s.id = str(p.get("id") or s.id)
s.cwd = p.get("cwd") if isinstance(p.get("cwd"), str) else ""
s.version = p.get("cli_version") if isinstance(p.get("cli_version"), str) else ""
source = p.get("source")
sub = source.get("subagent") if isinstance(source, dict) else None
if sub is not None:
spawn = sub.get("thread_spawn") if isinstance(sub, dict) else None
s.is_subagent = True
s.parent_id = str(p.get("parent_thread_id") or _as_dict(spawn).get("parent_thread_id") or "")
self.all_injected = not isinstance(spawn, dict)
def _event(self, ts, p) -> None:
kind = p.get("type")
if kind == "token_count" and not self.use_records:
info = p.get("info")
if isinstance(info, dict):
total = _as_dict(info.get("total_token_usage")).get("total_tokens")
if total != self.last_total: # the same total repeated is the same response
self.last_total = total
self._usage(ts, _cx_usage(info.get("last_token_usage")))
elif kind == "turn_aborted":
self._close()
_add(self.s, Event("interrupt", ts, str(p.get("reason") or "")))
elif kind == "thread_settings_applied":
self._set_model(_as_dict(p.get("thread_settings")).get("model"))
elif kind == "item_completed":
self._operation(ts, _as_dict(p.get("item")))
def _item(self, lineno, ts, p) -> None:
kind = p.get("type")
if kind == "message":
role = p.get("role")
if role == "assistant":
ev = self._assistant(ts)
txt = _cx_text(p.get("content"))
ev.text = ev.text + "\n" + txt if ev.text and txt else ev.text or txt
else:
self._close()
self._message(ts, role, p.get("content"))
elif kind in _CX_MODEL_ITEMS:
if self.pending is None and self.pending_at is None:
self.pending_at = len(self.s.events)
if kind != "reasoning":
self._call(ts, p)
elif kind in _CX_OUTPUTS:
self._close()
self._result(lineno, p)
elif kind not in _CX_QUIET_ITEMS:
_warn(self.s, lineno, "unknown response item %s" % _name(kind))
def _message(self, ts, role, content) -> None:
typed, context = [], []
for x in content if isinstance(content, list) else []:
if isinstance(x, dict) and isinstance(x.get("text"), str):
harness_made = role != "user" or self.all_injected or bool(_CX_CONTEXT_RE.match(x["text"]))
(context if harness_made else typed).append(x["text"])
if context:
_add(self.s, Event("user", ts, "\n".join(context), injected=True))
if typed:
_add(self.s, Event("user", ts, "\n".join(typed)))
def _call(self, ts, p) -> None:
kind = p.get("type")
name = p.get("name") if isinstance(p.get("name"), str) else kind
namespace = p.get("namespace") if isinstance(p.get("namespace"), str) else ""
call_id = str(p.get("call_id") or p.get("id") or "")
if kind == "function_call":
raw = p.get("arguments")
try:
args = json.loads(raw) if isinstance(raw, str) else raw
except ValueError:
args = None
inp = args if isinstance(args, dict) else {"raw": str(raw)}
elif kind == "custom_tool_call":
inp = {"raw": str(p.get("input") or "")}
else:
inp = _as_dict(p.get("action"))
full = namespace + "__" + name if namespace and not name.startswith(namespace) else name
call = ToolCall(id=call_id, name=full, kind=_cx_kind(name, namespace), input=inp, ts=ts)
if kind == "local_shell_call":
call.kind, call.command = "shell", _cx_argv(inp.get("command"))
elif kind == "web_search_call":
call.kind, call.has_result = "web", p.get("status") == "completed"
elif name == "exec" and kind == "custom_tool_call":
call.kind, call.command = "shell", inp["raw"]
elif call.kind == "shell":
call.command = _cx_argv(inp.get("cmd") or inp.get("command"))
elif name == "apply_patch":
call.paths = [a or b for a, b in _CX_PATCH_RE.findall(inp.get("raw") or inp.get("input") or "")]
if call_id:
self.calls[call_id] = call
ev = self._tool(ts, call)
if name == "exec" and kind == "custom_tool_call":
self.open_scripts.append(call_id)
self.script_events[call_id] = [ev, 0]
def _result(self, lineno, p) -> None:
call = self.calls.get(p.get("call_id"))
if call is None:
_warn(self.s, lineno, "tool output without a matching call")
return
txt = _cx_text(p.get("output"))
exit_code = None
if txt.startswith("{"): # older shell tools: {"output": ..., "metadata": {"exit_code": N}}
try:
legacy = json.loads(txt)
except ValueError:
legacy = None
if isinstance(legacy, dict) and isinstance(legacy.get("metadata"), dict) and "output" in legacy:
txt = str(legacy.get("output") or "")
exit_code = legacy["metadata"].get("exit_code")
if exit_code is None and call.kind in ("shell", "edit"):
for rx in _CX_EXIT_RES:
m = rx.search(txt[:2000])
if m:
exit_code = int(m.group(1))
break
call.has_result = True
_set_output(call, txt)
call.exit_code = exit_code if isinstance(exit_code, int) else None
call.is_error = call.is_error or (call.exit_code not in (None, 0)) or txt.startswith("Script failed")
if txt.strip() == "aborted" or txt.startswith("aborted by user"):
call.interrupted = True
call.denied = _cx_denial(txt)
if call.denied:
call.is_error = True
script = self.script_events.pop(call.id, None)
if script is not None:
if call.id in self.open_scripts:
self.open_scripts.remove(call.id)
if script[1] and not call.denied: # its operations stand in for it
self.dropped.add(id(script[0]))
def _operation(self, ts, item) -> None:
kind = item.get("type")
if kind not in ("CommandExecution", "FileChange", "McpToolCall", "Extension", "ImageView"):
return
item_id = str(item.get("id") or "")
direct = self.calls.get(item_id)
if direct is not None: # the log entry of a call the model made directly
if item.get("status") == "failed":
direct.is_error = True
return
status = item.get("status")
call = ToolCall(id=item_id, name="", kind="other", input={}, ts=ts,
has_result=status in (None, "completed", "failed"), is_error=status == "failed")
if kind == "CommandExecution":
call.name, call.kind = "exec_command", "shell"
call.command = _cx_argv(item.get("command"))
call.input = {"command": item.get("command"), "cwd": item.get("cwd")}
call.paths = [c["path"] for c in item.get("parsed_cmd") or []
if isinstance(c, dict) and c.get("type") == "read" and isinstance(c.get("path"), str)]
code = item.get("exit_code")
call.exit_code = code if isinstance(code, int) and not isinstance(code, bool) else None
call.is_error = call.is_error or call.exit_code not in (None, 0)
out = item.get("aggregated_output")
if not isinstance(out, str):
out = "\n".join(x for x in (item.get("stdout"), item.get("stderr")) if isinstance(x, str) and x)
_set_output(call, out)
elif kind == "FileChange":
changes = _as_dict(item.get("changes"))
call.name, call.kind = "apply_patch", "edit"
call.input = {"changes": {k: _as_dict(v).get("type") for k, v in changes.items()}}
call.paths = list(changes) + [v["move_path"] for v in changes.values()
if isinstance(v, dict) and isinstance(v.get("move_path"), str)]
_set_output(call, "\n".join(x for x in (item.get("stdout"), item.get("stderr")) if isinstance(x, str) and x))
elif kind == "McpToolCall":
call.name, call.kind = "mcp__%s__%s" % (item.get("server"), item.get("tool")), "mcp"
args = item.get("arguments")
call.input = args if isinstance(args, dict) else {"raw": str(args)}
result = item.get("result")
_set_output(call, _cx_text(_as_dict(result).get("content")) if isinstance(result, dict) else _cx_text(result))
elif kind == "Extension":
name = item.get("kind") if isinstance(item.get("kind"), str) else "extension"
call.name, call.kind, call.input = name, "web" if name.startswith("web") else "other", _as_dict(item.get("action"))
else:
path = item.get("path") if isinstance(item.get("path"), str) else ""
call.name, call.kind, call.input, call.paths = "view_image", "read", {"path": path}, [path] if path else []
call.denied = _cx_denial(call.output)
if call.denied:
call.is_error = True
self._tool(ts, call)
if self.open_scripts:
self.script_events[self.open_scripts[-1]][1] += 1
def finish(self) -> None:
for ev, ran in self.script_events.values():
if ran:
self.dropped.add(id(ev))
if self.dropped:
self.s.events = [e for e in self.s.events if id(e) not in self.dropped]
def _cx_session(path) -> Session:
m = _CX_NAME_RE.search(os.path.basename(path))
return Session("codex", m.group(1) if m else os.path.basename(path), path)
def _cx_newest_response(rows):
"""Index of the first item of the newest response, or None."""
newest, after_usage = None, True
for i, (_lineno, rec) in enumerate(rows):
kind, p = rec.get("type"), _as_dict(rec.get("payload"))
if kind == "token_usage_record" or (kind == "event_msg" and p.get("type") == "token_count"):
after_usage = True
elif kind == "response_item" and after_usage and (
p.get("type") in _CX_MODEL_ITEMS or (p.get("type") == "message" and p.get("role") == "assistant")):
newest, after_usage = i, False
return newest
def _file_matches(path, pattern) -> bool:
"""Search a file for a bytes pattern, 1 MB at a time, without parsing it."""
try:
with open(path, "rb") as fh:
tail = b""
while True:
chunk = fh.read(1 << 20)
if not chunk:
return False
if pattern.search(tail + chunk):
return True
tail = chunk[-256:]
except OSError:
return False
def _cx_total_before(path, offset) -> Optional[int]:
"""total_tokens of the last token_count event in the 4 MB before `offset`."""
start = max(0, offset - (4 << 20))
try:
with open(path, "rb") as fh:
fh.seek(start)
data = fh.read(offset - start)
except OSError:
return None
for line in reversed(data.split(b"\n")[1 if start else 0:]):
if b"token_count" in line:
try:
rec = json.loads(line)
except ValueError:
continue
p = _as_dict(_as_dict(rec).get("payload"))
if _as_dict(rec).get("type") == "event_msg" and p.get("type") == "token_count" and isinstance(p.get("info"), dict):
return _as_dict(p["info"].get("total_token_usage")).get("total_tokens")
return None
def _load_codex(path) -> Session:
s = _cx_session(path)
if path.endswith(".zst"):
_warn(s, 0, "compressed rollout (.jsonl.zst) skipped: reading zstd needs Python 3.14")
return s
reader = _Codex(s, _file_matches(path, _CX_USAGE_MARK))
for lineno, rec in _records(path, s):
reader.feed(lineno, rec)
reader.finish()
return s
# ---------------------------------------------------------------------------
# Gemini CLI (facts Q1.3)
# ---------------------------------------------------------------------------
_GM_KINDS = {
"run_shell_command": "shell", "read_file": "read", "read_many_files": "read", "write_file": "write",
"replace": "edit", "edit": "edit", "glob": "search", "search_file_content": "search",
"grep_search": "search", "grep": "search", "list_directory": "search", "ls": "search",
"google_web_search": "web", "web_search": "web", "web_fetch": "web",
}
_GM_EXIT_RE = re.compile(r"^Exit Code: (-?\d+)", re.M)
def _gm_kind(name) -> str:
if name in _GM_KINDS:
return _GM_KINDS[name]
return "mcp" if name.startswith("mcp_") or "__" in name else "other"
def _gm_usage(t) -> Optional[Usage]:
"""Gemini meanings (cached is part of input; thoughts are not in output) to Usage."""
if not isinstance(t, dict):
return None
cached, thoughts = _int(t.get("cached")), _int(t.get("thoughts"))
return Usage(input=max(0, _int(t.get("input")) - cached), cache_read=cached,
output=_int(t.get("output")) + thoughts, reasoning=thoughts)
def _gm_result(result) -> str:
parts = []
for part in result if isinstance(result, list) else []:
response = _as_dict(_as_dict(part).get("functionResponse")).get("response")
if isinstance(response, dict):
value = response.get("output", response.get("error"))
parts.append(value if isinstance(value, str) else json.dumps(response))
elif isinstance(_as_dict(part).get("text"), str):
parts.append(part["text"])
return "\n".join(parts)
def _gm_call(tc, ts) -> ToolCall:
name = tc.get("name") if isinstance(tc.get("name"), str) else ""
args = tc.get("args") if isinstance(tc.get("args"), dict) else {}
status = tc.get("status")
call = ToolCall(id=str(tc.get("id") or ""), name=name, kind=_gm_kind(name), input=args,
ts=tc.get("timestamp") if isinstance(tc.get("timestamp"), str) else ts,
has_result=status in ("success", "error", "cancelled"), is_error=status == "error",
denied="other" if status == "cancelled" else "")
if call.kind == "shell":
call.command = str(args.get("command") or "")
elif call.kind in ("read", "write", "edit"):
many = args.get("paths")
one = args.get("absolute_path") or args.get("file_path") or args.get("path")
call.paths = [p for p in many if isinstance(p, str)] if isinstance(many, list) else (
[one] if isinstance(one, str) and one else [])
text = _gm_result(tc.get("result"))
if not text and isinstance(tc.get("resultDisplay"), str):
text = tc["resultDisplay"]
_set_output(call, text)
if call.kind == "shell":
m = _GM_EXIT_RE.search(text)
if m:
call.exit_code = int(m.group(1))
call.is_error = call.is_error or call.exit_code != 0
return call
def _gm_project_dir(path) -> str:
"""The <project temp dir> that holds this chat file."""
chats = os.path.dirname(path)
if os.path.basename(chats) != "chats": # a subagent chat: chats/<parent session id>/<id>.jsonl
chats = os.path.dirname(chats)
return os.path.dirname(chats)
def _gm_project_root(folder) -> str:
try:
with open(os.path.join(folder, ".project_root"), encoding="utf-8") as fh:
return fh.read().strip()
except OSError:
pass
try:
with open(os.path.join(os.path.dirname(os.path.dirname(folder)), "projects.json"), encoding="utf-8") as fh:
projects = _as_dict(_as_dict(json.load(fh)).get("projects"))
except (OSError, ValueError):
return ""
name = os.path.basename(folder) # a slug, or sha256(root) for folders made by older versions
return next((root for root, slug in projects.items()
if name in (slug, hashlib.sha256(root.encode("utf-8", "surrogateescape")).hexdigest())), "")
def _load_gemini(path) -> Session:
stem = os.path.splitext(os.path.basename(path))[0]
s = Session("gemini-cli", stem, path)
meta, order, messages = {}, [], {}
def keep(m):
if isinstance(m, dict) and isinstance(m.get("id"), str):
if m["id"] not in messages:
order.append(m["id"])
messages[m["id"]] = m # a later line with the same id replaces the earlier one
if path.endswith(".json"): # legacy: one JSON document
try:
with open(path, encoding="utf-8") as fh:
doc = json.load(fh)
except (OSError, ValueError) as exc:
_warn(s, 0, "cannot read file: %s" % type(exc).__name__)
return s
meta = _as_dict(doc)
for m in meta.get("messages") or []:
keep(m)
else: # the current format is an operation log: replay it in order
for lineno, rec in _records(path, s):
if "$set" in rec:
update = _as_dict(rec["$set"])
if isinstance(update.get("messages"), list): # a checkpoint replaces all messages
order, messages = [], {}
for m in update["messages"]:
keep(m)
meta.update((k, v) for k, v in update.items() if k != "messages")
elif "$rewindTo" in rec: # drop that message and everything after it
target = rec["$rewindTo"]
cut = order.index(target) if target in messages else 0
for mid in order[cut:]:
del messages[mid]
del order[cut:]
elif isinstance(rec.get("id"), str):
keep(rec)
elif isinstance(rec.get("sessionId"), str):
meta.update(rec)
else:
_warn(s, lineno, "unknown line shape")
s.id = meta.get("sessionId") if isinstance(meta.get("sessionId"), str) else stem
s.cwd = _gm_project_root(_gm_project_dir(path))
parent_dir = os.path.basename(os.path.dirname(path))
if meta.get("kind") == "subagent" or parent_dir != "chats":
s.is_subagent = True
s.parent_id = parent_dir if parent_dir != "chats" else ""
for key in ("startTime", "lastUpdated"):
_track_time(s, meta.get(key))
for mid in order:
m = messages[mid]
kind, ts = m.get("type"), m.get("timestamp") if isinstance(m.get("timestamp"), str) else ""
_track_time(s, ts)
text = _block_text(m.get("content"))
if kind == "user":
_add(s, Event("user", ts, text, id=mid))
elif kind == "gemini":
model = m.get("model") if isinstance(m.get("model"), str) else ""
if model and model not in s.models:
s.models.append(model)
_add(s, Event("assistant", ts, text, usage=_gm_usage(m.get("tokens")), model=model, id=mid))
for tc in m.get("toolCalls") or []:
if isinstance(tc, dict):
call = _gm_call(tc, ts)
_add(s, Event("tool", call.ts, tool=call, model=model, id=call.id))
elif kind == "error":
_add(s, Event("api_error", ts, text, id=mid))
elif kind not in ("info", "warning"):
_warn(s, 0, "unknown message type %s" % _name(kind))
return s
# ---------------------------------------------------------------------------
# OpenCode (facts Q1.5). Built from the documented database schema; not
# verified on a real install. Session paths are '<database file>#<session id>'.
# ---------------------------------------------------------------------------
_OC_KINDS = {"bash": "shell", "read": "read", "write": "write", "edit": "edit", "multiedit": "edit",
"patch": "edit", "grep": "search", "glob": "search", "list": "search", "webfetch": "web",
"websearch": "web", "task": "agent"}
# Error texts for permission denials (unverified wording from the OpenCode source).
_OC_DENIALS = (("rejected permission", "user-rejected"), ("specified a rule which prevents", "permission-rule"))
def _iso_ms(ms) -> str:
if not isinstance(ms, (int, float)) or isinstance(ms, bool):
return ""
stamp = datetime.datetime.fromtimestamp(ms / 1000.0, tz=datetime.timezone.utc)
return stamp.strftime("%Y-%m-%dT%H:%M:%S.") + "%03dZ" % (int(ms) % 1000)
def _oc_connect(db):
"""Open the database read-only; it is never written."""
uri = pathlib.Path(os.path.abspath(db)).as_uri() + "?mode=ro"
return sqlite3.connect(uri, uri=True)
def _oc_usage(t) -> Optional[Usage]:
# Recorded values are kept as they are: input without cache, output as the
# provider reported it, reasoning as its own count (unverified on a real install).
if not isinstance(t, dict):
return None
cache = _as_dict(t.get("cache"))
return Usage(input=_int(t.get("input")), cache_read=_int(cache.get("read")),
cache_write=_int(cache.get("write")), output=_int(t.get("output")),
reasoning=_int(t.get("reasoning")))
def _oc_error_text(err) -> str:
err = _as_dict(err)
msg = _as_dict(err.get("data")).get("message")
return msg if isinstance(msg, str) else str(err.get("name") or "")
def _oc_call(part, ts) -> ToolCall:
name = part.get("tool") if isinstance(part.get("tool"), str) else ""
state = _as_dict(part.get("state"))
inp = state.get("input") if isinstance(state.get("input"), dict) else {}
status = state.get("status")
start = _as_dict(state.get("time")).get("start")
call = ToolCall(id=str(part.get("callID") or ""), name=name, kind=_OC_KINDS.get(name, "other"), input=inp,
ts=_iso_ms(start) or ts, has_result=status in ("completed", "error"), is_error=status == "error")
if call.kind == "shell":
call.command = str(inp.get("command") or "")
elif name == "patch":
call.paths = [a or b for a, b in _CX_PATCH_RE.findall(str(inp.get("patchText") or ""))]
elif isinstance(inp.get("filePath"), str) and call.kind in ("read", "write", "edit"):
call.paths = [inp["filePath"]]
text = state.get("output") if status == "completed" else state.get("error")
_set_output(call, text if isinstance(text, str) else "")
exit_code = _as_dict(state.get("metadata")).get("exit")
if call.kind == "shell" and isinstance(exit_code, int) and not isinstance(exit_code, bool):
call.exit_code = exit_code
call.is_error = call.is_error or exit_code != 0
if status == "error":
lowered = call.output.lower()
call.denied = next((kind for marker, kind in _OC_DENIALS if marker in lowered), "")
call.interrupted = "aborted" in lowered
return call
def _oc_assistant(s, message_id, data, parts) -> None:
ts = _iso_ms(_as_dict(data.get("time")).get("created"))
model = data.get("modelID") if isinstance(data.get("modelID"), str) else ""
if model and model not in s.models:
s.models.append(model)
step, stepped, steps = None, False, 0 # one Event per model step: a step-finish part holds that call's tokens
def current():
nonlocal step, steps
if step is None:
steps += 1
step = _add(s, Event("assistant", ts, model=model, id="%s:%d" % (message_id, steps)))
return step
for part in parts:
kind = part.get("type")
if kind == "step-start":
step = None
current()
elif kind == "text" and isinstance(part.get("text"), str):
ev = current()
ev.text = ev.text + "\n" + part["text"] if ev.text else part["text"]
elif kind == "tool":
call = _oc_call(part, ts)
current()
_add(s, Event("tool", call.ts, tool=call, model=model, id=call.id))
elif kind == "step-finish":
current().usage = _oc_usage(part.get("tokens"))
step, stepped = None, True
elif kind == "retry":
_add(s, Event("api_error", ts, _oc_error_text(part.get("error")), id=message_id))
elif kind == "compaction":
_add(s, Event("compaction", ts, id=message_id))
if not stepped: # no step parts: the message holds the tokens of its one call
usage = _oc_usage(data.get("tokens"))
if step is not None or (usage is not None and usage != Usage()):
current().usage = usage
err = data.get("error")
if isinstance(err, dict):
aborted = "abort" in str(err.get("name") or "").lower()
_add(s, Event("interrupt" if aborted else "api_error", ts, _oc_error_text(err), id=message_id))
def _oc_user(s, message_id, data, parts) -> None:
ts = _iso_ms(_as_dict(data.get("time")).get("created"))
for part in parts:
if part.get("type") == "text" and isinstance(part.get("text"), str):
_add(s, Event("user", ts, part["text"], injected=bool(part.get("synthetic")), id=message_id))
elif part.get("type") == "compaction":
_add(s, Event("compaction", ts, id=message_id))
def _oc_json(s, raw) -> dict:
try:
value = json.loads(raw)
except (TypeError, ValueError):
_warn(s, 0, "malformed JSON in the database")
return {}
return _as_dict(value)
def _load_opencode(path) -> Session:
db, _, sid = path.rpartition("#")
s = Session("opencode", sid, path)
try:
con = _oc_connect(db)
try:
con.row_factory = sqlite3.Row
row = con.execute("SELECT * FROM session WHERE id = ?", (sid,)).fetchone()
messages = con.execute("SELECT id, data FROM message WHERE session_id = ? ORDER BY id", (sid,)).fetchall()
parts = con.execute("SELECT message_id, data FROM part WHERE session_id = ? ORDER BY id", (sid,)).fetchall()
finally:
con.close()
except sqlite3.Error as exc:
_warn(s, 0, "cannot read database: %s" % type(exc).__name__)
return s
if row is not None:
info = dict(row)
s.version = str(info.get("version") or "")
s.parent_id = str(info.get("parent_id") or "")
s.is_subagent = bool(s.parent_id)
for key in ("time_created", "time_updated"):
_track_time(s, _iso_ms(info.get(key)))
by_message = {}
for message_id, raw in parts:
by_message.setdefault(message_id, []).append(_oc_json(s, raw))
for message_id, raw in messages:
data = _oc_json(s, raw)
own = by_message.get(message_id, [])
if data.get("role") == "assistant":
if not s.cwd and isinstance(_as_dict(data.get("path")).get("cwd"), str):
s.cwd = data["path"]["cwd"]
_oc_assistant(s, message_id, data, own)
elif data.get("role") == "user":
_oc_user(s, message_id, data, own)
else:
_warn(s, 0, "unknown message role %s" % _name(data.get("role")))
for e in s.events:
_track_time(s, e.ts)
return s
# ---------------------------------------------------------------------------
# Where each harness keeps its files
# ---------------------------------------------------------------------------
def _roots(home=None) -> dict:
"""Root folders per harness. `home` stands in for the home folder (tests);
when it is given, CLAUDE_CONFIG_DIR, CODEX_HOME, XDG_DATA_HOME, and
OPENCODE_DB are ignored so that nothing outside `home` is read."""
if home is None:
env, base = os.environ, os.path.expanduser("~")
else:
env, base = {}, str(home)
data = env.get("XDG_DATA_HOME") or os.path.join(base, ".local", "share")
return {
"claude-code": env.get("CLAUDE_CONFIG_DIR") or os.path.join(base, ".claude"),
"codex": env.get("CODEX_HOME") or os.path.join(base, ".codex"),
"gemini-cli": os.path.join(base, ".gemini"),
"opencode": os.path.join(data, "opencode"),
"opencode-db": env.get("OPENCODE_DB") or "",
}
def _scan(folder):
try:
return list(os.scandir(folder))
except OSError:
return []
def _mtime(entry) -> float:
try:
return entry.stat().st_mtime
except OSError:
return 0.0
def _norm(path) -> str:
return os.path.normpath(os.path.abspath(os.path.expanduser(str(path))))
def _under(cwd, project) -> bool:
if not cwd:
return False
cwd = os.path.normpath(cwd)
return cwd == project or cwd.startswith(project.rstrip(os.sep) + os.sep)
def _cc_encode(path) -> str:
return re.sub(r"[^A-Za-z0-9]", "-", path)
def _cc_first_cwd(path) -> str:
probe = Session("claude-code", "", path)
for lineno, rec in _records(path, probe):
if isinstance(rec.get("cwd"), str) and rec["cwd"]:
return rec["cwd"]
if lineno >= 50:
break
return ""
def _find_claude(roots, cutoff, project, include_subagents) -> list:
found = []
enc = _cc_encode(project) if project else ""
for folder in _scan(os.path.join(roots["claude-code"], "projects")):
if not folder.is_dir():
continue
# The folder name is the lossy encoded cwd (cut to 200 characters plus a
# hash when long), so it only preselects; the recorded cwd decides.
if project and not (folder.name == enc or folder.name.startswith(enc + "-")
or (len(enc) > 200 and folder.name.startswith(enc[:200]))):
continue
candidates = []
for entry in _scan(folder.path):
if entry.is_file():
candidates.append(entry)
elif include_subagents and entry.is_dir(): # agent-<id>.jsonl only: workflows also keep a journal.jsonl
subagents = os.path.join(entry.path, "subagents")
folders = [subagents] + [f.path for f in _scan(os.path.join(subagents, "workflows"))]
candidates.extend(e for f in folders for e in _scan(f) if e.name.startswith("agent-") and e.is_file())
for entry in candidates:
name = entry.name
if not name.endswith(".jsonl") or ".orphaned-" in name or ".superseded-" in name:
continue
mtime = _mtime(entry)
if cutoff is not None and mtime < cutoff:
continue
if project:
cwd = _cc_first_cwd(entry.path)
if not (_under(cwd, project) if cwd else folder.name == enc):
continue
found.append((mtime, "claude-code", entry.path))
return found
def _cx_meta(path) -> dict:
"""The session_meta payload on the first line of a rollout, or {}."""
try:
with open(path, "rb") as fh:
rec = json.loads(fh.readline())
except (OSError, ValueError):
return {}
return _as_dict(rec.get("payload")) if isinstance(rec, dict) and rec.get("type") == "session_meta" else {}
def _find_codex(roots, cutoff, project, include_subagents) -> list:
entries = []
for year in _scan(os.path.join(roots["codex"], "sessions")):
for month in _scan(year.path):
for day in _scan(month.path):
entries.extend(_scan(day.path))
entries.extend(_scan(os.path.join(roots["codex"], "archived_sessions")))
found = []
for entry in entries:
name = entry.name
if not name.startswith("rollout-") or not (name.endswith(".jsonl") or name.endswith(".jsonl.zst")):
continue
mtime = _mtime(entry)
if cutoff is not None and mtime < cutoff:
continue
if name.endswith(".zst"):
if project: # cannot read the working folder of a compressed file
continue
elif project or not include_subagents:
meta = _cx_meta(entry.path)
source = meta.get("source")
if not include_subagents and isinstance(source, dict) and "subagent" in source:
continue
if project and not _under(meta.get("cwd") if isinstance(meta.get("cwd"), str) else "", project):
continue
found.append((mtime, "codex", entry.path))
return found
def _find_gemini(roots, cutoff, project, include_subagents) -> list:
found = []
legacy_name = hashlib.sha256(project.encode("utf-8", "surrogateescape")).hexdigest() if project else ""
for folder in _scan(os.path.join(roots["gemini-cli"], "tmp")):
if not folder.is_dir():
continue
if project and folder.name != legacy_name and not _under(_gm_project_root(folder.path), project):
continue
for entry in _scan(os.path.join(folder.path, "chats")): # stat only; nothing is opened here
if entry.is_dir():
if include_subagents:
found.extend((_mtime(e), "gemini-cli", e.path) for e in _scan(entry.path)
if e.name.endswith(".jsonl"))
elif entry.name.startswith("session-") and entry.name.endswith((".jsonl", ".json")):
found.append((_mtime(entry), "gemini-cli", entry.path))
return [f for f in found if cutoff is None or f[0] >= cutoff]
def _oc_databases(roots) -> list:
if roots["opencode-db"]:
return [roots["opencode-db"]] if os.path.isfile(roots["opencode-db"]) else []
return sorted(e.path for e in _scan(roots["opencode"])
if e.is_file() and e.name.startswith("opencode") and e.name.endswith(".db"))
def _find_opencode(roots, cutoff, project, include_subagents) -> list:
found = []
for db in _oc_databases(roots):
changed = max(os.path.getmtime(p) for p in (db, db + "-wal") if os.path.exists(p))
if cutoff is not None and changed < cutoff:
continue
try:
con = _oc_connect(db)
try:
columns = [r[1] for r in con.execute("PRAGMA table_info(session)")]
stamp = next((c for c in ("time_updated", "time_created") if c in columns), None)
rows = con.execute("SELECT id, parent_id, %s FROM session" % (stamp or "NULL")).fetchall()
cwds = {}
if project:
for sid, raw in con.execute("SELECT session_id, data FROM message ORDER BY id"):
if sid not in cwds and '"cwd"' in (raw or ""):
cwd = _as_dict(_oc_json(Session("opencode", sid, db), raw).get("path")).get("cwd")
if isinstance(cwd, str):
cwds[sid] = cwd
finally:
con.close()
except sqlite3.Error:
continue
for sid, parent, updated in rows:
when = updated / 1000.0 if isinstance(updated, (int, float)) else changed
if (cutoff is not None and when < cutoff) or (parent and not include_subagents):
continue
if project and not _under(cwds.get(sid, ""), project):
continue
found.append((when, "opencode", "%s#%s" % (db, sid)))
return found
def detect_harnesses(home=None) -> dict:
"""Map each installed harness to its root folder (only harnesses with data folders)."""
roots = _roots(home)
checks = {
"claude-code": [os.path.join(roots["claude-code"], "projects")],
"codex": [os.path.join(roots["codex"], "sessions"), os.path.join(roots["codex"], "archived_sessions")],
"gemini-cli": [os.path.join(roots["gemini-cli"], "tmp")],
}
found = {h: roots[h] for h in HARNESSES if any(os.path.isdir(p) for p in checks.get(h, []))}
databases = _oc_databases(roots)
if databases:
found["opencode"] = os.path.dirname(databases[0])
return found
# ---------------------------------------------------------------------------
# Public entry points
# ---------------------------------------------------------------------------
def load_session(harness, path) -> Session:
"""Parse one session file (OpenCode: '<database>#<session id>')."""
loaders = {"claude-code": _load_claude, "codex": _load_codex, "gemini-cli": _load_gemini,
"opencode": _load_opencode}
if harness not in loaders:
raise ValueError("unknown harness %r; expected one of %s" % (harness, ", ".join(HARNESSES)))
return loaders[harness](str(path))
_FINDERS = {"claude-code": _find_claude, "codex": _find_codex, "gemini-cli": _find_gemini,
"opencode": _find_opencode}
def find_sessions(harness=None, since_days=30, project=None, home=None,
include_subagents=True) -> list:
"""List (harness, path) pairs, newest first. Files are filtered by modified
time before any is opened. `project` keeps sessions whose working folder is
that folder or inside it. `since_days=None` means no time limit."""
if harness in (None, "all"):
wanted = HARNESSES
elif harness == "cursor":
return [] # Cursor transcripts have no results, usage, or timestamps
elif harness in HARNESSES:
wanted = (harness,)
else:
raise ValueError("unknown harness %r; expected one of %s" % (harness, ", ".join(HARNESSES)))
cutoff = None if not since_days else time.time() - float(since_days) * 86400
roots = _roots(home)
project = _norm(project) if project else None
found = []
for h in wanted:
if h in _FINDERS:
found.extend(_FINDERS[h](roots, cutoff, project, include_subagents))
found.sort(key=lambda item: item[0], reverse=True)
return [(h, p) for _mtime_, h, p in found]
def iter_sessions(**kwargs):
"""Yield each Session that find_sessions(**kwargs) lists."""
for harness, path in find_sessions(**kwargs):
yield load_session(harness, path)
def cutoff(days) -> str:
"""The UTC time `days` days ago, in the form event timestamps use, for
unique_events(since=...)."""
stamp = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=days)
return stamp.strftime("%Y-%m-%dT%H:%M:%S")
def unique_events(sessions, since=None):
"""Yield (session, event) for every event across `sessions`, oldest session
first, skipping an event whose (kind, id) an earlier session already had.
A forked or resumed Claude Code session starts with a copy of the earlier
session's records, ids included; totals must count that copy once.
With `since` (an ISO 8601 UTC time, see cutoff()), events older than it
are skipped too: find_sessions picks files by modified time, and a
recently modified file can hold records months old. Events without a
timestamp are kept."""
seen = set()
floor = since[:19] if since else ""
for s in sorted(sessions, key=lambda s: s.started or s.ended or ""):
for e in s.events:
if floor and e.ts and e.ts[:19] < floor:
continue
if e.id:
if (e.kind, e.id) in seen:
continue
seen.add((e.kind, e.id))
yield s, e
def _incremental_reader(harness, session, path, offset):
if harness == "claude-code":
return _Claude(session)
reader = _Codex(session, _file_matches(path, _CX_USAGE_MARK)) # the source is chosen per file
if not reader.use_records:
reader.last_total = _cx_total_before(path, offset)
return reader
_INCREMENTAL = {"claude-code": (_cc_session, _cc_newest_response), "codex": (_cx_session, _cx_newest_response)}
def read_new_events(harness, path, offset) -> tuple:
"""Read the lines added since `offset` and return (events, new_offset).
For hooks that re-read a growing transcript. Only whole lines are read. The
newest API response, and everything after it, is held back until the next
response starts, because its records are still being written; so every
response comes back exactly once, with its final usage and with every tool
result linked to its call. Pass new_offset to the next call. Supported for
claude-code and codex (JSONL files that only grow). Subagent files are
separate files: read each one with its own offset."""
if harness not in _INCREMENTAL:
raise ValueError("read_new_events supports %s, not %r" % (", ".join(sorted(_INCREMENTAL)), harness))
make_session, newest = _INCREMENTAL[harness]
session = make_session(str(path))
rows, starts = [], []
end = offset
try:
for lineno, start, raw in _raw_lines(str(path), offset, complete_only=True):
end = start + len(raw)
rec = _parse(session, lineno, raw)
if rec is not None:
rows.append((lineno, rec))
starts.append(start)
except OSError:
return [], offset
cut = newest(rows)
if cut is not None:
end = starts[cut]
rows = rows[:cut]
reader = _incremental_reader(harness, session, str(path), offset)
for lineno, rec in rows:
reader.feed(lineno, rec)
reader.finish()
return session.events, end