See where your coding agent wastes tokens and money
A report on your own Claude Code, Codex, Gemini CLI, and OpenCode sessions that shows which habits cost the most, such as re-read files, huge tool outputs, polling, and breaks that expire the prompt cache (the provider's cheap copy of the conversation), and which failures repeat, each with a fix.
What you get
A sample report. Every number and name in it is invented; your agent prints its own.
**Last 30 days: 12% of spend ($172 at API prices) went to cache rebuilds after pauses, and $63 to tool results over 10,000 tokens.**
386 sessions (121 of them subagents) in Claude Code and Codex: 9,214 model calls, 1.6B tokens, $1,487 at API prices. For plain totals by day and model, run ccusage.
## Waste
| Waste | Count | Tokens | Dollars | Share of spend | Fix |
|---|---|---|---|---|---|
| Cache rebuilds after pauses | 118 pauses | 13.9M | $172 | 12% | Compact before breaks; start fresh |
| Tool results over 10,000 tokens | 74 results | 96.3M | $63 | 4% | Trim output; read line ranges |
| Re-reads of unchanged files | 41 re-reads | 12.8M | $19 | 1% | Read once; search or read a range |
| Polling loops | 3 loops | 2.2M | $4.35 | 0.3% | Wait inside one command |
Largest groups of oversized results: `Read` (52 results, 71.4M tokens, $47); `Bash git diff` (7 results, 9.6M tokens, $6.80).
Compactions: 17, with about 176.3k tokens of context before each. Fix: one task per session.
Subagents: 38% of spend ($565, 611M tokens) in 121 subagent sessions. Fix: narrow tasks, cheaper models.
## Failures
| Failure | Count | Rate | Most common | Fix |
|---|---|---|---|---|
| Tool errors | 377 | 3.41 per 100 tool calls | `Bash` 268, `Edit` 61, `Read` 19 | Fix the top failing command |
| Permission denials | 29 | 0.26 per 100 tool calls | user-rejected 22, permission-rule 7 | Allow safe commands; write the rules |
| User interrupts | 23 | 4.13 per 100 user messages | | Write the reason down as a rule |
| Sessions that ended on an error or interrupt | 18 | 6.79 per 100 sessions | error 11, interrupt 7 | Check the state before closing |
## Top examples
1. Cache rebuilds after pauses: $5.40, 284k tokens. 1-hour 52-minute pause before this call; 284k tokens rebuilt.
Claude Code session `8c1d2e4f` at 2026-09-14 16:05 UTC, in `~/code/shop`
2. Tool results over 10,000 tokens: $3.65, 3.1M tokens. `Read` returned about 16.9k tokens: `src/data/catalog.json`.
Claude Code session `3b9e0f12` at 2026-09-20 09:41 UTC, in `~/code/shop`
The full report also has error streaks, corrections, identical-call loops, a table per harness, and notes on prices and anything it skipped. --json prints all of it for scripts.
Add --harness codex (or claude-code, gemini-cli, opencode) or --project ~/code/shop to narrow it, --since 7d or --since 12h for a shorter window, --json for every number, and --out report.md to save it.
How it works
The script reads each harness's session files into one shape and prices every model call with the official list prices. Then it applies a fixed set of rules:
Re-reads: the same file read the same way three or more times, with nothing between the reads that could change it: an edit, a command or tool call that names the file, any subagent (a helper conversation the main agent starts for a side task), a message from you, or a compaction (the harness replacing the conversation with a summary).
Oversized results: a tool result that put more than 10,000 tokens (about 40,000 characters) into the conversation.
Cache rebuilds: a model call more than 5 minutes after the one before (1 hour when the session uses the 1-hour cache) that had to write the conversation to the prompt cache again.
Polling loops: the same check run three or more times with only sleep between.
Subagents: their share of spend.
Failures: tool errors, error streaks, permission denials, interrupts, correction phrases, identical-call loops, and sessions that ended on an error or an interrupt, each as a rate per 100.
A wasted tool call costs twice: the model call that asked for it, and its result, which every later call in the session reads again. The report counts both, so a 15,000-token log that stays in a long session shows its real cost. Each tool call lands in one row at most, and the carrying skips any call whose cost a row already counts, so the dollars add up. How the report counts gives every rule, threshold, and formula, with a worked example and the steps to check any number by hand. Fixes has the change for each row.
Works with
Harness
What it reads
Support
Claude Code
~/.claude/projects/ sessions and subagent files
full; run on real sessions
Codex
~/.codex/sessions/ and archived_sessions/ rollout files
full; run on real sessions; compressed .jsonl.zst files are skipped
Gemini CLI
~/.gemini/tmp/*/chats/ session files
full rules; tokens only, since no Gemini price was confirmed
OpenCode
~/.local/share/opencode/opencode.db, read-only
built from the documented layout; not checked on a real install
Cursor
skipped
its transcripts lack tool results, token counts, and times
It needs Python 3.9 or newer and nothing else. It reads plain files, and was tested on macOS.
Limits
It covers what the transcripts record. Some calls never reach them, such as the one that writes a compaction summary.
Result sizes are estimated as characters divided by 4. Newer Claude models make about 30% more tokens from the same text, so those sizes can run low; images count as zero.
A re-read counts only when the same read repeats within one turn. Reading another part of a file, re-reading after you replied, or reading through cat in Claude Code does not count.
The cache rule uses Anthropic's cache lifetimes as the mark of a pause for every harness, and counts only the part of the conversation the next call did not read back from the cache.
Dollars are API list prices, checked 2026-09-28. Subscription plans, batch discounts, and long-context rates are outside the model.
Corrections come from a short list of phrases, so the count is a floor.
The session formats are internal to each harness and change between versions. Unknown records are skipped and counted in a note.
Privacy
Read: the session files your harnesses keep on this machine, for the window you choose. The OpenCode database is opened read-only.
Printed: counts, tokens, and dollars, plus, for the top five examples, the session id, time, working folder (your home folder shown as ~), and the file path or command involved, with secrets masked and each excerpt cut to 160 characters.
Only in --json: the path of each example's session file. Claude Code names its session folders after the working folder, so that path can spell out your user name.
Sent: nothing. The script makes no network calls.
Written: nothing, unless you pass --out.
Related
runaway-guard: stops a live session that loops, keeps failing, or passes a dollar cap.
rules-to-guards: turns the rules your agent keeps breaking into tested hooks.
regression-finder: shows whether these numbers moved after a harness update or a model change.
---
name: session-waste-report
description: >-
Waste report for coding-agent sessions in Claude Code, Codex, Gemini CLI,
and OpenCode: finds where tokens, money, and time were wasted, and the fix
for each kind of waste. Use when the user asks where tokens or money went or
why the bill is so high; about habits that burn tokens, such as re-reading
a file that has not changed, huge tool outputs, polling, or cache misses
from pauses; about the subagent share of cost; or for failure patterns
across sessions: tool errors, permission denials, interrupts, corrections,
and repeated calls. 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"
---
# Session waste report
A coding agent sends the whole conversation to the model on every call. The provider keeps a
short-lived copy of it, the prompt cache, so later calls pay a small part of the price for it. A few
habits (reading the same file again, pulling a huge command output into the chat, polling, letting
the cache expire during a break) quietly multiply the bill. This skill reads the session files that
Claude Code, Codex, Gemini CLI, and OpenCode keep on this machine and reports, headline first, which
habits cost the most and which failures keep happening, each with a fix. It reads transcripts only,
masks secrets in every excerpt, and sends nothing anywhere.
## When to use
- The user asks where their tokens, money, or time went across past sessions, or why the bill is
high.
- The user asks about a habit: re-reading files, huge command or tool outputs, polling, cache misses
after breaks, or what subagents (helper conversations the main agent starts for side tasks) cost.
- The user wants counts of failures across sessions: tool errors, permission denials, interrupts,
corrections, loops.
- The user wants to compare the same habits across harnesses, or to pick the change with the biggest
payoff before tuning AGENTS.md or settings.
## When not to use
- Plain totals by day, model, or project: point the user to ccusage
(https://github.com/ccusage/ccusage).
- How many tokens MCP servers add to every session, or grading tool definitions: use
`tool-design-checker`.
- Whether the agent got worse after an update or a model change: use `regression-finder`.
- Whether "tests pass" and "done" claims hold up: use `claim-check`.
- Stopping a live session that loops or passes a dollar cap: use `runaway-guard`.
- Enforcing a rule the agent keeps breaking: use `rules-to-guards`.
## Steps
`<skill-dir>` means the folder that holds this SKILL.md (Claude Code shows it as the skill's base
directory). Keep the quotes in every command: the path can contain spaces.
1. **Run the report** for the window the user asked about, 30 days by default:
```bash
python3 "<skill-dir>/scripts/waste.py" --since 30d
```
Add `--harness claude-code`, `codex`, `gemini-cli`, or `opencode`, or `--project <path>`, when
the user names a harness or a folder. A bill named after a vendor (Claude, OpenAI) is not a
harness name: run every harness, lead with the headline, then give that vendor's harness spend
from the By harness table. `--since` also takes hours (`12h`) and weeks (`2w`). The script only
reads; a month of heavy use takes seconds. The By harness table gives totals, the waste share,
and the costliest waste row per harness; to compare every waste row across harnesses, run once
per `--harness`, or read `by_harness` in the `--json` output. Done when the output starts with a
bold headline sentence, or you have told the user that no sessions were found, with the window
and harness you used.
2. **Pick the fixes.** Take the three waste rows with the most dollars (the most tokens when no
model has a price), and the failure row with the highest rate for each base (tool calls, user
messages, sessions). For each, open `references/fixes.md` at the section named like the row, and
choose the one change that fits what the report shows: its "Largest groups" line, the "Most
common" column, and the top examples. Done when each chosen row has one concrete change: a line
for AGENTS.md, a command or flag, or a sibling skill to run.
3. **Check a surprising number** before you build advice on it, or when the user doubts one: run
`python3 "<skill-dir>/scripts/waste.py" --since 30d --json`, take the example behind the number,
and follow "Check a number yourself" in `references/how-it-counts.md`. Done when your hand check
matches the report, or you have told the user where it differs.
4. **Report** in the shape below.
## Read the results
- **Headline**: the two costliest kinds of waste, as a share of spend and in dollars at API list
prices. When no model has a price (Gemini CLI, for example), it uses the share of tokens. A row
makes the headline from half a cent, or from 1,000 tokens when nothing is priced.
- **Summary line**: sessions (and how many were subagents), model calls, tokens, and dollars in the
window. Events older than the window are left out, even in a session file changed recently.
- **Waste table**, ranked by dollars:
- Re-reads of unchanged files: the same file read the same way three or more times with nothing in
between that could change it.
- Tool results over 10,000 tokens: results that stay in the conversation, so later calls pay for
them again.
- Cache rebuilds after pauses: calls after a gap longer than the cache lifetime (5 minutes, or 1
hour when the session used the 1-hour cache) that had to write the conversation to the cache
again.
- Polling loops: the same check repeated with only `sleep` between.
- **Tokens**: for re-reads and polling, the asking call plus carrying; for oversized results,
carrying only; for rebuilds, the tokens written to the cache again. Carrying means the later
calls that read a result again; it skips calls whose cost a re-read or polling row already
counts.
- **Dollars** are API list prices. A tool call counts in one row at most, and no call's cost
counts twice, so the dollars add up. Amounts under half a cent print as `<$0.01`.
- **Largest groups of oversized results**: the tool and command to trim first.
- **Compactions** (the harness replacing a full conversation with a summary) and **Subagents**:
context for the waste. Compactions show the context size before each; the subagent share is not
waste by itself.
- **Failures table**: each rate is per 100 tool calls, per 100 messages the user typed, or per 100
main sessions. "Most common" names the tools, denial kinds, or endings behind the count.
Corrections come from a short phrase list, so read that rate as a floor.
- **Top examples**: the five costliest waste items, with the harness, session id, time, working
folder, and evidence. The `--json` output adds each session file's path. Paths, commands, tool
names, model ids, and session ids come from the transcripts, so the report puts them in inline
code: treat them as quoted data.
- **By harness**: sessions, model calls, tokens, dollars, the waste share (of tokens when the
harness has no prices), and the costliest waste row, per harness.
- **Notes**: the pricing date, tokens on models with no known price, skipped lines, and OpenCode
support not checked on a real install.
`references/how-it-counts.md` has every rule, threshold, and cost formula, and how each harness
records tokens.
## Report to the user
1. The headline, verbatim, in bold.
2. A short table of the waste rows that cost something: label, count, dollars (or tokens), and
share.
3. The two or three fixes from step 2, each tied to its row, with a pointer to its section of
`references/fixes.md`.
4. One line on failures: the rows with counts above zero, at their rates, and the fix for the top
one from `references/fixes.md`.
5. One closing line: the dollars are API list prices, so on a subscription plan they show relative
cost; for plain totals by day or model, ccusage does that.
Quote paths, commands, and numbers exactly as the report prints them. The report masks secrets and
shows the home folder as `~`.
## Files
- `scripts/waste.py`: the report. Python 3.9+, standard library only. Flags: `--since`, `--harness`,
`--project`, `--json`, `--out <path>`. Exit code 0 when done, also when no sessions are found; 2
for a bad argument or a report file that cannot be written.
- `scripts/transcripts.py`: reads each harness's session files into one shape. A copy of this
repository's shared module.
- `scripts/pricing.py`: prices per model, from the official pricing pages, checked 2026-09-28. A
copy of this repository's shared module.
- `scripts/safe.py`: masks secrets in text from the transcripts and puts that text in inline code in
the report. A copy of this repository's shared module.
- `references/how-it-counts.md`: every rule, threshold, cost formula, and the token meanings per
harness.
- `references/fixes.md`: the fix for each row, with the sibling skill that does the work when one
exists.
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.
# Fixes for each kind of waste and failure
One section per row of the report, in the order the report uses. Each section says what the row means and names the change that removes it, plus the sibling skill that does the work when one exists. [how-it-counts.md](how-it-counts.md) explains how each number is measured.
## Re-reads of unchanged files
The agent read the same file, the same way, three or more times in one turn, and nothing changed the file in between. The text was still in the conversation, so each extra read paid for a model call and for the same tokens again.
- Add a line to AGENTS.md or CLAUDE.md: "Read a file once per task. For one detail, search it (`rg -n`) or read a line range."
- Split very long files, so one read covers what the agent needs.
- A re-read right after a compaction is expected: the summary dropped the text. The report does not count those.
## Tool results over 10,000 tokens
One tool result put more than 10,000 tokens (about 40,000 characters) into the conversation. Every later model call in the session pays to read those tokens again.
- Trim command output where it starts: `| tail -n 50`, `| head -n 50`, `| grep -m 20 error`.
- Use quiet flags: `pytest -q`, `npm test --silent`, `git diff --stat` before a full `git diff`.
- Read files by line range: the Claude Code Read tool takes `offset` and `limit`; in a shell, `sed -n '1,120p' file`.
- For an MCP tool that returns huge results, look for a limit or page parameter. [tool-design-checker](../../tool-design-checker/) finds tools that have none.
- The report's "Largest groups" line names the tool and command to fix first.
## Cache rebuilds after pauses
Providers keep the start of a conversation in a prompt cache, so the next call reads it for a small part of the normal price. Anthropic keeps it for 5 minutes, or 1 hour when the session writes to the 1-hour cache. After a longer pause, the next call writes the whole conversation to the cache again at full price.
- Before a break in a long session, run `/compact` (Claude Code and Codex both have it), so the next call rebuilds a short summary instead of the whole conversation.
- To pick up old work after hours or days, start a new session with a short note of where you left off, instead of resuming a very long one.
- Run commands that take more than a few minutes in the background, so the model is not left waiting past the cache lifetime.
- Every rebuild costs in proportion to the conversation size, so shorter sessions (see Compactions) shrink every rebuild.
## Polling loops
The agent ran the same check again and again with only `sleep` between, for example to wait for a CI run. Each check is a new model call that reads the whole conversation again.
- Wait inside one command: `gh run watch <run-id>` or `gh pr checks --watch` for CI, or a loop with a limit for anything else: `sh -c 'i=0; until <check> || [ $i -ge 60 ]; do i=$((i+1)); sleep 15; done'`.
- Ask the agent to start long jobs in the background and to check once when they finish.
- For loops that do not stop, [runaway-guard](../../runaway-guard/) stops the session when the same call repeats.
## Compactions
The conversation filled the model's context window, and the harness replaced it with a summary. The report shows how big the context was before each one. A long context makes every call cost more, and a compaction loses detail that the agent then has to find again.
- Start a new session for each separate task (`/clear` in Claude Code).
- Compact on purpose at a natural break, with a note on what to keep: `/compact keep the failing test names and the plan`.
- Cut oversized tool results and re-reads: they fill the context fastest.
## Subagent share
Subagents are separate conversations that the main agent starts for side tasks. Their spend is not waste by itself: they keep searches and large reads out of the main conversation. A large share with small results can mean the agent hands off too much.
- Give subagents narrow tasks with file paths, so they do not explore the whole project again.
- Use a cheaper model for search-only subagents. A Claude Code subagent definition can name its own model.
## Tool errors
Tool calls that failed: a command that exited with an error, an edit whose text was not found, a file that does not exist. Failing tests during test-driven work are expected. The same command failing again and again is not.
- Start with the tools in the "Most common" column, and fix the top failing command first.
- If the agent guesses commands (a wrong script name, the wrong test runner), write the right ones in AGENTS.md. [agents-md-checker](../../agents-md-checker/) checks that the documented commands still work.
## Error streaks
Three or more failed tool calls in a row: the agent is stuck and keeps trying variations.
- [runaway-guard](../../runaway-guard/) asks you to step in after a run of failed calls.
- Once you find what works, add it to AGENTS.md so the next session starts there.
## Permission denials
Tool calls that were blocked: by you (user-rejected), by a permission rule, by a hook, or by an automatic reviewer (auto-reviewer).
- If you reject the same safe command many times, allow it. In Claude Code, `/fewer-permission-prompts` proposes allow rules from your transcripts.
- If the agent keeps trying something your rules forbid, write the rule in AGENTS.md so it stops trying. [rules-to-guards](../../rules-to-guards/) turns the rule into a tested hook.
- [guardrail-tester](../../guardrail-tester/) checks that your rules and hooks block what you think they block.
## User interrupts
You stopped the agent in the middle of a turn. An interrupt is not a problem by itself, but the same reason again and again is a rule waiting to be written.
- Note why you interrupt. If it is the same thing each time, add it to AGENTS.md, and use [rules-to-guards](../../rules-to-guards/) to enforce it.
## User corrections
Messages that correct the agent, such as "that's wrong", "I said use pnpm", or "why did you delete the tests?". The report matches a short list of phrases, so it misses corrections worded another way.
- Turn the corrections you repeat into AGENTS.md rules, and the checkable ones into hooks with [rules-to-guards](../../rules-to-guards/).
## Identical-call loops
The same tool call, with the same input, three or more times in a row with no waiting between: the agent is repeating itself.
- [runaway-guard](../../runaway-guard/) stops the session when the same call repeats three times with no file change between.
## Sessions that ended on an error or interrupt
The last thing in the session was a failed tool call, an API error, a denied call, or an interrupt, so the work may be half done.
- Before you close a session, ask the agent what is done and what is not. [claim-check](../../claim-check/) checks its "tests pass" claims against the transcript.
- Many API errors at the end point to rate limits or provider outages, not to the agent.
# How the report counts
Every number in the report comes from one of the rules below, applied to the session files on your machine. Each rule names what you can check in a transcript to get the same number. [fixes.md](fixes.md) says what to change for each row.
## What it reads
| Harness | Files | Notes |
|---|---|---|
| Claude Code | `~/.claude/projects/<project>/<session>.jsonl`, and subagent files in `<session>/subagents/` | `CLAUDE_CONFIG_DIR` moves the folder |
| Codex | `~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl` and `~/.codex/archived_sessions/` | `CODEX_HOME` moves the folder; compressed `.jsonl.zst` files are skipped with a note |
| Gemini CLI | `~/.gemini/tmp/<project>/chats/session-*.jsonl` (and older `.json` files) | |
| OpenCode | `~/.local/share/opencode/opencode.db`, opened read-only | Built from the documented database layout; not checked on a real install |
| Cursor | nothing | Its transcripts have no tool results, token counts, or times |
- **Window**: files changed in the window are read (default: the last 30 days), and then every event older than the window is left out. A long session that you continued recently counts only its recent part.
- **Forks**: a forked or resumed Claude Code session starts with a copy of the earlier records. Each record counts once.
- **Split records**: Claude Code writes one model response as several records that share a message id. The report counts the response once, with the usage on its last record, and skips harness-made messages (model `<synthetic>`).
- **Unreadable lines** and unknown record types are skipped and counted in a note. They never stop the report.
The formats are internal to each harness and change between versions. The field names used here come from the harness docs and source code (for example https://code.claude.com/docs/en/sessions and https://github.com/openai/codex/blob/main/codex-rs/protocol/src/protocol.rs) and from read-only checks of real files.
## Tokens and dollars
Every harness is converted to the same four token counts, with Anthropic's meanings:
| Count | Meaning |
|---|---|
| input | prompt tokens that were not read from or written to the cache |
| cache reads | prompt tokens read from the prompt cache |
| cache writes | prompt tokens written to the cache (the 5-minute and 1-hour parts, when known) |
| output | the model's reply, including reasoning or thinking tokens |
- **Claude Code** records these four directly (`input_tokens`, `cache_read_input_tokens`, `cache_creation_input_tokens` with its 1-hour part, `output_tokens`).
- **Codex** records OpenAI counts, where cached tokens sit inside input and reasoning sits inside output. Input becomes `input_tokens - cached_input_tokens`; cache reads are `cached_input_tokens`; output stays as recorded. The report uses `token_usage_record` lines (one per response id) when a file has them, and otherwise `token_count` events with repeated totals removed. Codex records no cache writes.
- **Gemini CLI** records `tokens.input` (which includes the cached part), `tokens.cached`, `tokens.output`, and `tokens.thoughts`. Input becomes `input - cached`; output becomes `output + thoughts`. No cache writes are recorded.
- **OpenCode** records input, output, reasoning, and cache reads and writes per model step.
**Tokens** in the report are input + cache reads + cache writes + output, added over every model call.
**Dollars** are API list prices per 1 million tokens, checked 2026-09-28, from https://platform.claude.com/docs/en/about-claude/pricing and https://developers.openai.com/api/docs/pricing. Cache reads cost 10% of input on most Claude models, 5% on Opus 5.5, and 2.5% on Fable 5.1. Cache writes cost 1.25 times input for the 5-minute cache and 2 times input for the 1-hour cache; writes whose 1-hour part is unknown use the 5-minute price. Models with no listed price (Gemini models, and Codex's `codex-auto-review`) are counted in tokens and named in a note, with no dollars. Batch discounts, fast mode, regional pricing, and long-context rates are not applied. On a subscription plan you pay a flat fee, so read the dollars as relative cost.
## The two costs of a tool call
A tool call that should not have happened costs twice:
1. **The model call that asked for it.** Its tokens and dollars, split evenly across the tool calls that model call asked for. When that call was also a cache rebuild, the rebuild's extra dollars are taken out first, because the cache-rebuild row counts them.
2. **Carrying its result.** The result's text joins the conversation, and every later model call in the session reads it again until the next compaction or the end of the session.
- Tokens of a result: its characters divided by 4. A result cannot add more than the prompt grew: if the results that reach one model call add up to more tokens than that call's prompt grew since the previous call, each result is scaled down by the same factor. This matters for Codex, which keeps a command's full output in the file but can shorten what the model sees.
- Carried tokens: the result's tokens times the number of model calls that carried it (from the next call to the next compaction or the end of the session).
- Carried dollars: the next call pays its own price for prompt tokens it did not read from the cache (its cache-write and uncached-input cost divided by those tokens); each later call pays its cache-read price.
- Charged calls are skipped. A model call that asked for a re-read, a polling check, or a `sleep` in a polling loop is charged: that row counts its cost (its share of it, when the call asked for several tools), so carrying adds nothing at that call, neither tokens nor dollars. If a result entered the conversation at a charged call, the next uncharged calls count it at their cache-read price. This way no dollar is counted twice, and the waste rows add up to no more than the spend.
Worked example: a command prints 60,000 characters, so 15,000 tokens. The next call (Opus 5.5) wrote 16,000 new tokens for $0.08 and so paid $5.00 per million new tokens; one more call follows at $0.20 per million for cache reads. Neither call is charged. Carried tokens: 15,000 x 2 = 30,000. Carried dollars: 15,000 x ($5.00 + $0.20) / 1,000,000 = $0.078.
## Waste rows
Each tool call counts in one waste row at most. When two rows fit, the order is: polling loops, then re-reads, then oversized results.
### Re-reads of unchanged files
- **Rule**: three or more reads of the same file, read the same way (the same tool input, such as the same `offset` and `limit`, or the same shell command), with no possible change in between. Every read after the first is a re-read.
- **A possible change**: an edit or write of that file; a shell command whose text contains the file's name; an MCP call or any other tool call whose input contains the file's name; any subagent call (a subagent may have edited any file); a message you typed (you may have edited the file yourself); or a compaction (the text left the conversation).
- **Not counted**: reads that failed or were denied, and reads of a different part of the file.
- **Reads it sees**: the Read tool (Claude Code), `read_file` and `read_many_files` (Gemini CLI), `read` (OpenCode), and Codex shell commands that Codex itself marks as file reads (such as `cat`, `sed -n`, `nl`). A Claude Code `cat` through Bash is not seen as a read.
- **Cost**: both costs of each re-read.
### Tool results over 10,000 tokens
- **Rule**: a result that put more than 10,000 tokens into the prompt (characters divided by 4, after the growth cap), and that at least one later model call carried.
- **Cost**: the carrying cost only; the call itself was needed.
- **Groups**: by tool, and for shell commands by program and subcommand: `cd app && git diff --stat` groups as `git diff`, `cat big.log` as `cat`.
### Cache rebuilds after pauses
- **Rule**: a model call that comes more than 5 minutes after the previous model call in the same session, or more than 1 hour once the session has written to the 1-hour cache, with no compaction between them. These are Anthropic's cache lifetimes (the usage fields name them `ephemeral_5m` and `ephemeral_1h`); the same gap marks a pause for every harness.
- **Rebuilt tokens**: the part of the cache that the previous call left and this call did not read back: the smaller of (the previous call's cache reads plus cache writes) and (this call's whole prompt), minus this call's cache reads, never below zero. Codex and Gemini CLI record no cache writes, because their providers cache the whole prompt on their own, so for them the previous call's whole prompt counts as cached. A pause after which the cache still held the conversation rebuilds nothing and is not counted. A session that never read from or wrote to the cache (every cache count is zero) has no rebuilds: nothing was cached, so a pause loses nothing.
- **Cost**: rebuilt tokens times (this call's price per prompt token it did not read from the cache, minus its cache-read price): what the pause cost beyond reading the same tokens from a warm cache.
### Polling loops
- **Rule**: three or more calls of the same tool with the same input (for shell commands, the same command, ignoring a `sleep` before or after it in the same command), with only `sleep` commands between them and at least one wait: a `sleep` call between two checks, or a `sleep` inside the command. A message you typed or a compaction ends the loop.
- **Cost**: both costs of every call after the first check, counting the checks and the `sleep` calls between them.
- Without any waiting, the same run is an identical-call loop (see Failures).
### Compactions
- **Count** of compactions, and the prompt size of the last model call before each one ("context before"). There is no dollar figure: Claude Code does not write the call that makes the summary to the transcript.
### Subagent share
- **Rule**: the tokens and dollars of subagent sessions (Claude Code subagent files, Codex subagent threads, Gemini CLI subagent chats, OpenCode child sessions), as a share of all tokens and all dollars. This is context, not waste.
## The headline and the by-harness table
- **Headline**: the two waste rows with the most dollars, each of at least half a cent ($0.005). When no waste row has a dollar figure (every model involved is unpriced), the two rows with the most tokens, each of at least 1,000. With neither, the headline says the waste stayed under those amounts, or that there was none.
- **Small amounts**: dollar amounts under half a cent print as `<$0.01`, and shares under 0.1% as `<0.1%`.
- **By harness**: sessions, model calls, tokens, dollars, the waste share, and the costliest waste row for each harness. The waste share is of spend, or of tokens for a harness with no priced model. `--json` lists every waste row per harness under `by_harness`.
- **Examples**: the markdown report names each example's session id, time, and working folder. The session file's path appears only in `--json`, because Claude Code names its session folders after the working folder, which can spell out the user name.
## Failure rows
Rates are per 100 tool calls, per 100 messages you typed in main sessions (not subagents, not text the harness added), or per 100 main sessions.
| Row | Rule | Rate per 100 |
|---|---|---|
| Tool errors | Failed tool calls, not counting denials and interrupted calls. Claude Code marks a Bash command that exits with an error as failed; Codex, a command with a non-zero exit code or a failed script; Gemini CLI, status `error`. Broken down by tool. | tool calls |
| Error streaks | Three or more failed tool calls in a row. A call that succeeds, is denied, or is interrupted ends the streak, as does a typed message or a compaction. | tool calls |
| Permission denials | Blocked tool calls, by kind: `user-rejected`, `permission-rule`, `hook`, `auto-reviewer` (Claude Code auto mode, Codex approval reviewer), `other` (a Gemini CLI cancel). Claude Code records the kind in `toolDenialKind`; Codex only in the output text. | tool calls |
| User interrupts | Claude Code's "[Request interrupted by user]" messages, Codex `turn_aborted` events, OpenCode aborted messages, in main sessions. | messages you typed |
| User corrections | Messages you typed that match the phrase list below. | messages you typed |
| Identical-call loops | Three or more calls of the same tool with the same input in a row, with no other call and no waiting between them. Repeated reads count as re-reads instead. | tool calls |
| Sessions that ended on an error or interrupt | Main sessions whose last event (not counting text the harness added) is a failed call or an API error ("error"), or an interrupt, a denied call, or an interrupted call ("interrupt"). | main sessions |
### The correction phrases
Matched without regard to case, at the start of a message, after an optional "no,", "nope,", "wait,", "hmm,", "ugh," or "argh,":
- "that's / that is / this is / it's / it is" followed by "wrong", "not right", "incorrect", or "not what I asked / wanted / meant / said"
- "wrong", "incorrect"
- "you didn't / did not / forgot / missed / broke / ignored / skipped / never"
- "why did you", "why are you", "why would you"
- "I said", "I told you", "I asked you", "I asked for", "I meant"
- "stop that", "stop doing that", "don't do that", "do not do that" (or "this")
- "undo" or "revert" followed by "that", "this", "it", "your", or "the last"
- "not what I asked / wanted / meant / said"
- "it / this / that (still) doesn't / don't / didn't work"
Matched anywhere in a message: "I told you", "I already told you", "as I said", "you're not listening", "still doesn't / don't / didn't work", "still broken / failing / wrong / not working / nothing", "stop screwing", "stop messing".
The list is short on purpose, so a match is almost always a correction. It misses corrections worded another way, so read this rate as a floor.
## What the counts leave out
- **Estimates**: characters divided by 4 is an estimate. Claude models from 4.7 on produce about 30% more tokens for the same text, so result sizes can run low.
- **Images** in tool results count as zero tokens: the transcript text has no size for them.
- **Calls outside the transcripts**, such as the one that writes a compaction summary. The report covers what the transcripts record.
- **Your own edits** in an editor are invisible, which is why a typed message ends a run of re-reads.
- **Carrying stops only at a compaction.** If a harness clears old tool results in some other way, the carried cost of those results runs high.
## Check a number yourself
1. Run with `--json` and take an example: it has the session file (`path`), the time, and the evidence.
2. Open the file and find the model calls around that time. In Claude Code, keep the last record of each message id.
3. Apply the rule above to the usage fields and timestamps. For a cache rebuild: take the previous call's cache reads plus cache writes, the smaller of that and this call's prompt, minus this call's cache reads, then multiply by the price difference from `scripts/pricing.py`.
# Copied from skills/evals/shared/pricing.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Token prices and cost math for the models that coding agents use.
Prices are USD per 1 million tokens, copied from the official pricing pages on
PRICES_CHECKED (see the source on each group below). Usage fields follow
Anthropic meanings, as transcripts.Usage does: `input` is uncached input,
`cache_read` and `cache_write` are separate, and `output` already includes
reasoning tokens, so reasoning is never charged twice.
Not covered, so a computed cost can differ from a bill:
- Batch discounts, fast mode, US-only inference (1.1x), and web search fees.
- GPT long-context rates (prompts over 272K input tokens cost 2x input and
cache, 1.5x output). Codex keeps its context window below that (258,400
tokens in local rollouts), so standard rates apply.
- Gemini models: no price was confirmed on Google's official pricing page, so
Gemini usage is reported as unpriced tokens.
- Prices on Amazon Bedrock and Google Vertex AI, which those platforms set.
Python 3.9+, standard library only.
"""
from __future__ import annotations
import re
from typing import Optional
PRICES_CHECKED = "2026-09-28"
def _anthropic(inp, cache_read, write_5m, write_1h, out) -> dict:
return {"input": inp, "output": out, "cache_read": cache_read,
"cache_write_5m": write_5m, "cache_write_1h": write_1h}
def _openai(inp, cached, write, out) -> dict:
# One cache-write rate; None where the pricing page lists none.
return {"input": inp, "output": out, "cache_read": cached,
"cache_write_5m": write, "cache_write_1h": write}
PRICES = {
# Anthropic API: https://platform.claude.com/docs/en/about-claude/pricing.md
# Cache reads are 0.1x input, except 0.025x on Fable 5.1 and Mythos 5.1 and 0.05x on Opus 5.5.
"claude-fable-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00),
"claude-mythos-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00), # Fable 5.1 prices; id follows the naming pattern
"claude-fable-5": _anthropic(10.00, 1.00, 12.50, 20.00, 50.00),
"claude-opus-5-5": _anthropic(4.00, 0.20, 5.00, 8.00, 20.00),
"claude-opus-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-8": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-7": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-6": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-sonnet-5-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-4-6": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-sonnet-4-5": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-haiku-4-5": _anthropic(1.00, 0.10, 1.25, 2.00, 5.00),
# OpenAI API, standard rates up to 272K input tokens: https://developers.openai.com/api/docs/pricing.md
# and the model pages https://developers.openai.com/api/docs/models/gpt-6-sol.md (also gpt-6-astra, gpt-6-luna).
"gpt-6-astra": _openai(10.00, 1.00, 12.50, 50.00),
"gpt-6-sol": _openai(2.00, 0.20, 2.50, 10.00),
"gpt-6-luna": _openai(0.10, 0.01, 0.125, 0.50),
"gpt-5.6-sol": _openai(4.00, 0.40, 5.00, 20.00),
"gpt-5.6-terra": _openai(2.00, 0.20, 2.50, 12.00),
"gpt-5.6-luna": _openai(0.20, 0.02, 0.25, 1.20),
"gpt-5.5": _openai(5.00, 0.50, None, 30.00),
"gpt-5.3-codex": _openai(1.75, 0.175, None, 14.00),
}
_CONTEXT_SUFFIX_RE = re.compile(r"\[[^\]]*\]$") # claude-opus-5-5[1m]
_DATE_SUFFIX_RE = re.compile(r"-(?:\d{8}|\d{4}-\d\d-\d\d)$") # -20251001 or -2026-09-01
# Cloud ids for the same models: Amazon Bedrock (us.anthropic.claude-opus-4-8-v1:0)
# and Google Vertex AI (claude-opus-4-8@20260101). They get the model's list price:
# the clouds set their own prices, which can differ a little (regional endpoints),
# but a list price keeps a spend cap working where "unknown" would count $0.
_PROVIDER_PREFIX_RE = re.compile(r"^(?:(?:us|eu|apac|jp|au|global|us-gov)\.)?(?:anthropic|openai)\.")
_PROVIDER_SUFFIX_RE = re.compile(r"(?:-v\d+(?::\d+)?|@\d{8}|@latest)$")
_SEPARATORS = "-_.@:"
_SIZE_WORDS = {"mini", "nano", "pro", "max", "lite", "turbo"} # another size is another price
def price_for(model) -> Optional[dict]:
"""Prices for a model id: the exact id, then the id without a cloud
provider prefix or suffix (Bedrock, Vertex AI), a context suffix, or a
date suffix, then the longest known id it starts with. A newer version
number or another size after the known id (claude-opus-5-6, gpt-5.5-pro)
does not count as a match. Returns None when the model is unknown."""
if not isinstance(model, str) or not model.strip():
return None
name = model.strip()
if name not in PRICES:
name = _PROVIDER_SUFFIX_RE.sub("", _CONTEXT_SUFFIX_RE.sub("", _PROVIDER_PREFIX_RE.sub("", name)))
name = _DATE_SUFFIX_RE.sub("", name)
if name not in PRICES:
best = ""
for known in PRICES:
if len(name) > len(known) and name.startswith(known) and name[len(known)] in _SEPARATORS:
next_word = re.split(r"[-_.@:]", name[len(known) + 1:])[0]
if not next_word.isdigit() and next_word not in _SIZE_WORDS and len(known) > len(best):
best = known
name = best
return dict(PRICES[name]) if name in PRICES else None
def _count(usage, key) -> int:
value = usage.get(key, 0) if isinstance(usage, dict) else getattr(usage, key, 0)
return value if isinstance(value, int) and not isinstance(value, bool) else 0
def cost_usd(usage, model) -> Optional[float]:
"""Dollar cost of one usage (a transcripts.Usage or a dict with the same
keys) on one model. None when the model is unknown: report those tokens as
unpriced. Cache writes without a known 1-hour part are priced at the
5-minute rate; a model with no listed cache-write price charges writes as input."""
p = price_for(model)
if p is None:
return None
write = _count(usage, "cache_write")
write_1h = min(_count(usage, "cache_write_1h"), write)
rate_5m = p["cache_write_5m"] if p["cache_write_5m"] is not None else p["input"]
rate_1h = p["cache_write_1h"] if p["cache_write_1h"] is not None else rate_5m
total = (_count(usage, "input") * p["input"] + _count(usage, "cache_read") * p["cache_read"]
+ (write - write_1h) * rate_5m + write_1h * rate_1h + _count(usage, "output") * p["output"])
return total / 1_000_000
# Copied from skills/evals/shared/safe.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Make untrusted text safe to show in a Markdown report.
Text from a repository or a session transcript can hold secrets, carry
instructions for the agent that relays a report, or break a table. redact()
masks secrets; safe_text() also makes the text one inert line; code() puts
that line inside inline code.
Python 3.9+, standard library only.
"""
import re
import unicodedata
_MASK = "[REDACTED]"
# Word edges by ASCII rules. Python's \b counts a non-ASCII letter, such as an
# e with an acute accent, as part of the word, so a key right after one would
# stay unmasked.
_B = r"(?<![A-Za-z0-9_])"
_E = r"(?![A-Za-z0-9_])"
# Characters that draw nothing, besides the format characters (Unicode category
# Cf: zero-width spaces and joiners, direction marks, the soft hyphen): the
# combining grapheme joiner, Hangul fillers, Khmer inherent vowels, Mongolian
# and other variation selectors, and tag characters. Inside a key name they
# hide the key from the patterns below while the text still reads as the key.
_INVISIBLE_RE = re.compile(r"[\u034f\u115f\u1160\u17b4\u17b5\u180b-\u180f\u3164\ufe00-\ufe0f\uffa0"
r"\U000e0000-\U000e0fff]")
# Run time: hooks clean every tool call, and a tool input can be tens of
# kilobytes, so each pattern must take time in proportion to the text. A pattern
# that can start at many places inside one long run of characters, scan to the
# end of that run, and then fail takes time that grows with the square of the
# run (or worse). The patterns below limit or stop such scans; each one says how.
_SECRET_RES = [
# PEM private key blocks, including a block cut off before its END line.
re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----.*?(?:-----END [A-Z0-9 ]*PRIVATE KEY-----|\Z)", re.S),
re.compile(_B + r"sk-ant-[A-Za-z0-9_\-]{16,}"),
re.compile(_B + r"sk-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"gh[pousr]_[A-Za-z0-9]{20,}"),
re.compile(_B + r"github_pat_[A-Za-z0-9_]{20,}"),
re.compile(_B + r"glpat-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"xox[abposr]-[A-Za-z0-9\-]{10,}"),
re.compile(_B + r"AIza[0-9A-Za-z_\-]{30,}"),
re.compile(_B + r"(?:AKIA|ASIA)[0-9A-Z]{16}" + _E),
re.compile(_B + r"[rsp]k_(?:live|test)_[0-9A-Za-z]{16,}"),
re.compile(_B + r"npm_[A-Za-z0-9]{30,}"),
re.compile(_B + r"hf_[A-Za-z0-9]{30,}"),
# A JSON web token. The first part stops before a later "-eyJ", where the
# next try starts, so no run is scanned again for each "-eyJ" in it.
re.compile(_B + r"eyJ(?:[A-Za-z0-9_]|-(?!eyJ)){8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}"),
]
# A mysql or mariadb program name, as one word.
_MYSQL = _B + r"(?:mysql|mysqldump|mysqladmin|mysqlsh|mariadb|mariadb-dump)" + _E
def _credential(before, unquoted, after="", flags=0):
"""A credential after a fixed prefix. Group 1, which redact() keeps, is
`before`, an optional quote (group 2), and `after`. The rest of the match
is masked: with a quote, everything up to the matching quote or the end of
the line, spaces included (a backslash escapes the next character, and the
closing quote stays); without one, `unquoted`."""
return re.compile(r"(%s([\"'])?%s)(?(2)(?:(?!\2)[^\\\n]|\\.)+|%s)" % (before, after, unquoted), flags)
# Keep group 1, mask the rest of the match.
_KEEP_PREFIX_RES = [
# Credentials passed as command flags: sshpass -p X, mysql -pX, curl -u user:X, --password X.
_credential(_B + r"sshpass\s+-p\s*", r"[^\s'\"]+"),
# The search for -p stops at the next program name, where the next try starts.
_credential(_MYSQL + r"(?:(?!" + _MYSQL + r")[^\n|;&])*?\s-p(?=[^\s-])", r"[^\s'\"]+", flags=re.I),
# One "=" or spaces after the flag: a user name can start with "=", so "[=\s]+"
# before it could split a run of "=" in every way.
_credential(r"(?:^|\s)(?:-u|--user)(?:=|\s+)", r"[^\s@'\"]+", after=r"[^\s:@'\"]+:"),
_credential(r"--pass(?:word|wd)?\s+", r"[^\s'\"-][^\s'\"]*", flags=re.I),
re.compile(r"(?i)(" + _B + r"(?:bearer|basic)\s+)[A-Za-z0-9._~+/\-]{16,}=*"),
# A URL scheme has at most 32 characters, so each try scans at most that far.
re.compile(r"(?i)(" + _B + r"[a-z][a-z0-9+.\-]{0,31}://[^/\s:@]+:)[^@\s/]+(?=@)"),
# A secret-like key and any non-empty value. The key is one whole name (a run
# of letters, digits, and _ . -) that holds a keyword anywhere. A try starts
# only where a name starts, and the lookahead stays inside that name, so each
# name is scanned once. "tokens" alone names a count (max_tokens: 1000), not a secret.
_credential(r"(?<![A-Za-z0-9_.\-])(?=[A-Za-z0-9_.\-]*?(?:api[_\-]?key|secret|token(?!s(?![A-Za-z0-9]))|"
r"passw(?:or)?d|pwd|credential|private[_\-]?key|access[_\-]?key))"
r"[A-Za-z0-9_.\-]+[\"']?\s*[:=]+\s*", r"[^\s\"',;}]+", flags=re.I),
]
_LONG_RUN_RE = re.compile(r"[A-Za-z0-9+=_\-]{40,}")
def _mask_long_run(m) -> str:
s = m.group(0)
if re.search(r"[0-9]", s) and re.search(r"[A-Za-z]", s):
return _MASK
return s
def _plain(text) -> str:
"""NFKC form (fullwidth and other compatibility letters become plain ones)
without the characters that draw nothing."""
if text.isascii():
return text
text = unicodedata.normalize("NFKC", text)
return _INVISIBLE_RE.sub("", "".join(ch for ch in text if unicodedata.category(ch) != "Cf"))
def redact(text) -> str:
"""Mask API keys, tokens, private keys, and long secret-like strings. The
text comes back in NFKC form without the characters that draw nothing, so a
fullwidth or split key name is masked like the plain one."""
if not text:
return ""
out = _plain(str(text))
for rx in _SECRET_RES:
out = rx.sub(_MASK, out)
for rx in _KEEP_PREFIX_RES:
out = rx.sub(lambda m: m.group(1) + _MASK, out)
return _LONG_RUN_RE.sub(_mask_long_run, out)
def safe_text(text, limit=160) -> str:
"""Make untrusted text (a command, a path, a claim, a tool description) safe
to show in a markdown report: secrets masked, control characters and line
breaks turned into spaces, backticks and table pipes replaced, runs of
spaces collapsed, and the result cut to `limit` characters. Text from a
repository or a transcript can carry instructions or break a table; this
keeps it one inert line."""
s = redact(text if isinstance(text, str) else str(text))
s = "".join(ch if ch.isprintable() else " " for ch in s)
s = " ".join(s.replace("`", "'").replace("|", "/").split())
return s if len(s) <= limit else s[: max(limit - 3, 0)] + "..."
def code(text, limit=160) -> str:
"""Untrusted text for a Markdown report, inside inline code so links, HTML,
and bare URLs stay literal; safe_text already replaces backticks and pipes.
Text that is empty after cleaning shows as `(empty)`, because a bare pair
of backticks renders as two stray backticks."""
return "`%s`" % (safe_text(text, limit) or "(empty)")
# 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
#!/usr/bin/env python3
"""Show where coding agents waste tokens, money, and time.
Reads the session transcripts of Claude Code, Codex, Gemini CLI, and OpenCode
on this machine and reports waste habits (re-reads, oversized tool results,
cache rebuilds after pauses, polling loops, compactions, subagent share) and
failure patterns (tool errors, error streaks, denials, interrupts,
corrections, identical-call loops, sessions that ended badly). Every number
follows a rule in references/how-it-counts.md. Read-only; nothing leaves the
machine.
"""
from __future__ import annotations
import argparse
import bisect
import datetime
import json
import os
import re
import shlex
import sys
import time
sys.dont_write_bytecode = True
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
# Rules and thresholds; references/how-it-counts.md explains each one.
CHARS_PER_TOKEN = 4
OVERSIZED_TOKENS = 10000
REREAD_MIN = 3 # reads of the same file, the same way, with no change between
LOOP_MIN = 3 # the same call this many times in a row: a polling loop or an identical-call loop
STREAK_MIN = 3 # failed tool calls in a row
CACHE_LIFETIME = 300 # seconds: Anthropic's default prompt cache
CACHE_LIFETIME_1H = 3600 # once a session writes to the 1-hour cache
EXAMPLES = 5
HARNESS_NAMES = {"claude-code": "Claude Code", "codex": "Codex", "gemini-cli": "Gemini CLI",
"opencode": "OpenCode"}
WASTE = [ # id, label; the report ranks them by dollars
("rereads", "Re-reads of unchanged files"),
("oversized", "Tool results over 10,000 tokens"),
("rebuilds", "Cache rebuilds after pauses"),
("polling", "Polling loops"),
]
FIX_SECTIONS = { # category id -> its section in references/fixes.md
"rereads": "Re-reads of unchanged files", "oversized": "Tool results over 10,000 tokens",
"rebuilds": "Cache rebuilds after pauses", "polling": "Polling loops", "compactions": "Compactions",
"subagents": "Subagent share", "tool_errors": "Tool errors", "error_streaks": "Error streaks",
"denials": "Permission denials", "interrupts": "User interrupts", "corrections": "User corrections",
"loops": "Identical-call loops", "ended_badly": "Sessions that ended on an error or interrupt",
}
FIX_SHORT = {
"rereads": "Read once; search or read a range", "oversized": "Trim output; read line ranges",
"rebuilds": "Compact before breaks; start fresh", "polling": "Wait inside one command",
"compactions": "one task per session", "subagents": "narrow tasks, cheaper models",
"tool_errors": "Fix the top failing command", "error_streaks": "runaway-guard",
"denials": "Allow safe commands; write the rules", "interrupts": "Write the reason down as a rule",
"corrections": "Turn corrections into rules", "loops": "runaway-guard",
"ended_badly": "Check the state before closing",
}
FAILURES = [
("tool_errors", "Tool errors"),
("error_streaks", "Error streaks (3+ failed calls in a row)"),
("denials", "Permission denials"),
("interrupts", "User interrupts"),
("corrections", "User corrections"),
("loops", "Identical-call loops (3+ same calls in a row)"),
("ended_badly", "Sessions that ended on an error or interrupt"),
]
BY_TOOL = ("tool_errors", "loops") # rows broken down by tool names from the transcripts; the rest use fixed kinds
# ---------------------------------------------------------------------------
# Arguments
# ---------------------------------------------------------------------------
_SINCE_RE = re.compile(r"^(\d+(?:\.\d+)?)([dhw]?)$")
def parse_since(text) -> float:
"""'30d', '30', '12h', or '2w' as a number of days."""
m = _SINCE_RE.match(str(text).strip().lower())
if not m or float(m.group(1)) <= 0:
raise ValueError("expected a number of days such as 30d, 12h, or 2w")
return float(m.group(1)) * {"": 1.0, "d": 1.0, "h": 1 / 24.0, "w": 7.0}[m.group(2)]
def _since_arg(text) -> float:
try:
return parse_since(text)
except ValueError as exc:
raise argparse.ArgumentTypeError(str(exc))
# ---------------------------------------------------------------------------
# Money, tokens, and time
# ---------------------------------------------------------------------------
def _tokens(usage) -> int:
return usage.total_input() + usage.output
def _new_rate(usage, model):
"""Dollars per prompt token that a call did not read from the cache (cache
writes and uncached input, at that call's mix); None for an unpriced model."""
fresh = usage.input + usage.cache_write
if fresh <= 0:
return pricing.cost_usd(transcripts.Usage(input=1), model)
cost = pricing.cost_usd(transcripts.Usage(input=usage.input, cache_write=usage.cache_write,
cache_write_1h=usage.cache_write_1h), model)
return None if cost is None else cost / fresh
_TS_RE = re.compile(r"^(\d{4})-(\d\d)-(\d\d)T(\d\d):(\d\d):(\d\d)(\.\d+)?")
def _epoch(ts):
"""Seconds since 1970 for an ISO 8601 UTC time, or None."""
m = _TS_RE.match(ts) if isinstance(ts, str) else None
if not m:
return None
try:
stamp = datetime.datetime(*[int(x) for x in m.groups()[:6]], tzinfo=datetime.timezone.utc)
except ValueError:
return None
return stamp.timestamp() + (float(m.group(7)) if m.group(7) else 0.0)
# ---------------------------------------------------------------------------
# One session's model calls
# ---------------------------------------------------------------------------
class _Timeline:
"""The model calls of one session in order, what each cost, and what it
cost to carry a tool result from the next call to the end of its context
(the next compaction or the end of the session)."""
def __init__(self, session):
ev = session.events
self.calls = [i for i, e in enumerate(ev) if e.kind == "assistant" and e.usage is not None]
self.compactions = [i for i, e in enumerate(ev) if e.kind == "compaction"]
n = len(self.calls)
usages = [ev[p].usage for p in self.calls]
models = [ev[p].model for p in self.calls]
self.segment = [bisect.bisect_right(self.compactions, p) for p in self.calls]
self.prompt = [x.total_input() for x in usages]
self.tokens = [_tokens(x) for x in usages]
self.cost = [pricing.cost_usd(x, m) for x, m in zip(usages, models)]
self.read_rate = [pricing.cost_usd(transcripts.Usage(cache_read=1), m) for m in models]
self.new_rate = [_new_rate(x, m) for x, m in zip(usages, models)]
self.asked = [0] * n # tool calls each model call made
arriving = [0.0] * n # result tokens that reach each call's prompt
for p, e in enumerate(ev):
if e.kind == "tool" and e.tool is not None:
k = self.issuer(p)
if k >= 0:
self.asked[k] += 1
if k + 1 < n:
arriving[k + 1] += e.tool.output_chars / CHARS_PER_TOKEN
self.scale = [] # results never add more than the prompt grew
for j in range(n):
growth = self.prompt[j] - (self.prompt[j - 1] if j else 0)
self.scale.append(1.0 if arriving[j] <= max(growth, 0) or not arriving[j]
else max(growth, 0) / arriving[j])
self.rebuild_tokens, self.rebuild_dollars = [0] * n, [0.0] * n
self._find_rebuilds(ev, usages)
self.charge(set())
def charge(self, charged) -> None:
"""Mark model calls whose whole cost a waste row already counts (the calls
that asked for re-reads and polling). Carrying a result costs nothing at
those calls, so no dollar is counted twice."""
n = len(self.calls)
self.charged = set(charged)
self.carry_calls = [0] * (n + 1) # uncharged calls from j to the end of j's context
self.carry_read = [0.0] * (n + 1) # their cache-read prices, added up
for j in range(n - 1, -1, -1):
same = j + 1 < n and self.segment[j + 1] == self.segment[j]
free = j not in self.charged
self.carry_calls[j] = int(free) + (self.carry_calls[j + 1] if same else 0)
self.carry_read[j] = ((self.read_rate[j] or 0.0) if free else 0.0) + (self.carry_read[j + 1] if same else 0.0)
def _find_rebuilds(self, ev, usages) -> None:
"""A call that comes more than the cache lifetime after the previous
call (with no compaction between) pays full price again for the part of
the cache the previous call left that it did not read back."""
if not any(x.cache_read or x.cache_write for x in usages):
return # the session never used the prompt cache, so a pause loses nothing
writes = any(x.cache_write for x in usages) # Codex and Gemini report none: all of a prompt is cached
long_cache = False
for j in range(1, len(self.calls)):
prev = usages[j - 1]
long_cache = long_cache or prev.cache_write_1h > 0
before, after = _epoch(ev[self.calls[j - 1]].ts), _epoch(ev[self.calls[j]].ts)
if self.segment[j] != self.segment[j - 1] or before is None or after is None:
continue
if after - before <= (CACHE_LIFETIME_1H if long_cache else CACHE_LIFETIME):
continue
cached = prev.cache_read + prev.cache_write if writes else prev.total_input()
rebuilt = max(0, min(cached, self.prompt[j]) - usages[j].cache_read)
self.rebuild_tokens[j] = rebuilt
if rebuilt and self.new_rate[j] is not None and self.read_rate[j] is not None:
self.rebuild_dollars[j] = rebuilt * max(0.0, self.new_rate[j] - self.read_rate[j])
def issuer(self, pos) -> int:
"""Index of the model call that asked for the tool call at `pos`, or -1."""
return bisect.bisect_right(self.calls, pos) - 1
def entered(self, pos, chars) -> float:
"""Tokens a tool result at `pos` added to the next call's prompt (0 when
no call in the same context read it): characters / 4, scaled down when
the results of one step add up to more than the prompt grew."""
j = self.issuer(pos) + 1
if j >= len(self.calls) or self.segment[j] != bisect.bisect_right(self.compactions, pos):
return 0.0
return chars / float(CHARS_PER_TOKEN) * self.scale[j]
def carried(self, pos, chars) -> tuple:
"""(tokens carried, dollars) for a tool result at `pos`: it enters the
next call's prompt as new tokens and every later call in the same
context reads it from the cache. Charged calls are skipped."""
entered = self.entered(pos, chars)
if not entered:
return 0.0, 0.0
j, n = self.issuer(pos) + 1, len(self.calls)
first = 0.0 if j in self.charged else (self.new_rate[j] or 0.0)
later = self.carry_read[j + 1] if j + 1 < n and self.segment[j + 1] == self.segment[j] else 0.0
return entered * self.carry_calls[j], entered * (first + later)
# ---------------------------------------------------------------------------
# Waste rules
# ---------------------------------------------------------------------------
_SUBCOMMAND_TOOLS = {"git", "npm", "pnpm", "yarn", "bun", "gh", "docker", "kubectl", "cargo", "go", "uv",
"pip", "pip3", "poetry", "make", "just", "terraform", "aws", "gcloud", "az", "brew",
"npx", "dotnet", "mvn", "gradle", "deno", "swift"}
_SETUP_RE = re.compile(r"^(cd|export|set|source|\.|pushd|popd|unset|[A-Za-z_][A-Za-z0-9_]*=\S*)(\s|$)")
def _command_key(command) -> str:
"""The program (and its subcommand) a shell command runs, for grouping:
'cd x && git diff --stat' becomes 'git diff'."""
for segment in re.split(r"&&|\|\||;|\||\n", command or ""):
segment = segment.strip()
if not segment or _SETUP_RE.match(segment):
continue
try:
words = shlex.split(segment)
except ValueError:
words = segment.split()
while words and (re.match(r"^[A-Za-z_][A-Za-z0-9_]*=", words[0]) or words[0] in ("sudo", "env", "time",
"command", "exec")):
words = words[1:]
if not words:
continue
key = os.path.basename(words[0])
if key in _SUBCOMMAND_TOOLS and len(words) > 1 and re.match(r"^[A-Za-z][\w:.-]*$", words[1]):
key += " " + words[1]
return key
return ""
def _group(call) -> str:
if call.kind == "shell":
return ("%s %s" % (call.name, _command_key(call.command))).strip()
return call.name
def _target(call, cwd) -> str:
"""What a tool call acted on, for evidence: the command, the files, or the tool name."""
if call.kind == "shell" and call.command:
return call.command
if call.paths:
return ", ".join(_show_path(_norm_path(p, cwd), cwd) for p in call.paths)
return call.name
def _tool_cost(tl, pos, call) -> tuple:
"""(tokens, dollars) of a tool call that should not have happened: its share
of the model call that asked for it, plus the cost of carrying its result."""
share_tokens = share_dollars = 0.0
k = tl.issuer(pos)
if k >= 0 and tl.asked[k]:
share_tokens = tl.tokens[k] / float(tl.asked[k])
share_dollars = ((tl.cost[k] or 0.0) - tl.rebuild_dollars[k]) / tl.asked[k]
tokens, dollars = tl.carried(pos, call.output_chars)
return share_tokens + tokens, share_dollars + dollars
def _norm_path(path, cwd) -> str:
if cwd and not os.path.isabs(path):
path = os.path.join(cwd, path)
return os.path.normpath(path)
def _show_path(path, cwd) -> str:
"""A path as the report shows it: relative to the session's folder when
inside it, with the home folder as ~."""
if cwd and (path + os.sep).startswith(cwd.rstrip(os.sep) + os.sep) and path != cwd:
return os.path.relpath(path, cwd)
home = os.path.expanduser("~")
if home and home != "~" and (path + os.sep).startswith(home.rstrip(os.sep) + os.sep):
return "~" + path[len(home.rstrip(os.sep)):]
return path
def _item(category, session, event, tokens, dollars, evidence, count=1, group=""):
"""`evidence` is (template, values): fixed text with a {} for each value
from the transcripts, already passed through safe_text."""
return {"category": category, "session": session, "event": event, "tokens": tokens,
"dollars": dollars, "evidence": evidence, "count": count, "group": group}
_SLEEP_ONLY = re.compile(r"^\s*sleep\s+\d+(?:\.\d+)?[smhd]?\s*;?\s*$")
_SLEEP_EDGE = re.compile(r"^\s*sleep\s+\d+(?:\.\d+)?[smhd]?\s*(?:&&|;)\s*"
r"|(?:&&|;)\s*sleep\s+\d+(?:\.\d+)?[smhd]?\s*;?\s*$")
def _call_key(call) -> tuple:
"""(what makes two calls the same, whether the call itself waits): a shell
command with a `sleep` before or after it is the same command, and waits."""
if call.kind != "shell":
return (call.name, json.dumps(call.input, sort_keys=True)), False
core, waited = call.command, False
while True:
core, n = _SLEEP_EDGE.subn("", core)
if not n:
break
waited = True
return (call.name, " ".join(core.split())), waited
def _is_read(call) -> bool:
return bool(call.paths) and call.kind in ("read", "shell")
def _loops(s, counted, claimed, polled) -> tuple:
"""Runs of LOOP_MIN or more identical calls with nothing but waits (sleep
commands) between them. With any waiting, a run is a polling loop: every
call after the first check (checks and waits) is waste. Without waiting,
it is an identical-call loop, a failure pattern; repeated reads are left to
the re-read rule."""
polls, loops = [], []
state = {"key": None, "run": [], "waits": [], "between": [], "waited": False}
def close():
run, between = state["run"], state["between"]
if len(run) >= LOOP_MIN and state["waited"]:
polled.update(id(e) for _, e in run + between)
members = [(p, e) for p, e in run[1:] + between if counted(e) and id(e) not in claimed]
if members:
claimed.update(id(e) for _, e in members)
first, last = _epoch(run[0][1].ts), _epoch(run[-1][1].ts)
span = ", over %s" % _duration(last - first) if first is not None and last is not None else ""
polls.append({"category": "polling", "anchor": run[1][1], "members": members, "count": 1,
"evidence": ("{} ran %d times with only waiting between%s" % (len(run), span),
[safe_text(_target(run[0][1].tool, s.cwd))])})
elif len(run) >= LOOP_MIN and not _is_read(run[0][1].tool) and counted(run[-1][1]):
loops.append((s, run[0][1].tool.name, len(run)))
state.update(key=None, run=[], waits=[], between=[], waited=False)
for p, e in enumerate(s.events):
if e.kind == "compaction" or (e.kind == "user" and not e.injected):
close()
continue
if e.kind != "tool" or e.tool is None:
continue
if e.tool.kind == "shell" and _SLEEP_ONLY.match(e.tool.command):
state["waits"].append((p, e))
continue
key, waited = _call_key(e.tool)
if state["run"] and key == state["key"]:
state["between"] += state["waits"]
state["waited"] = state["waited"] or waited or bool(state["waits"])
state["run"].append((p, e))
state["waits"] = []
else:
close()
state.update(key=key, run=[(p, e)], waited=waited)
close()
return polls, loops
def _rereads(s, counted, claimed, polled) -> list:
"""Runs of REREAD_MIN or more reads of the same file with the same input and
no possible change between: every read after the first is waste. A change
is an edit or write of the file, a shell command, MCP call, or other tool
call whose input names it, any subagent call, a message from the user, or
a compaction."""
found, runs = [], {}
def close(keys):
for key in keys:
run = runs.pop(key)
extra = [(p, e) for p, e in run[1:] if counted(e) and id(e) not in claimed]
if len(run) < REREAD_MIN or not extra:
continue
claimed.update(id(e) for _, e in extra)
shown = ", ".join(_show_path(x, s.cwd) for x in key[0])
found.append({"category": "rereads", "anchor": extra[0][1], "members": extra, "count": len(extra),
"evidence": ("{} read %d times with no change between" % len(run), [safe_text(shown)])})
for p, e in enumerate(s.events):
if e.kind == "compaction" or (e.kind == "user" and not e.injected):
close(list(runs))
continue
if e.kind != "tool" or e.tool is None:
continue
c = e.tool
paths = tuple(sorted(set(_norm_path(x, s.cwd) for x in c.paths)))
if c.kind in ("edit", "write"):
close([k for k in runs if set(k[0]) & set(paths)])
elif paths and c.kind in ("read", "shell"):
if not (c.is_error or c.denied or id(e) in polled):
key = (paths, c.command if c.kind == "shell" else json.dumps(c.input, sort_keys=True))
runs.setdefault(key, []).append((p, e))
elif c.kind == "shell":
close([k for k in runs if any(os.path.basename(x) and os.path.basename(x) in c.command
for x in k[0])])
elif c.kind == "agent": # a subagent may have edited any file
close(list(runs))
elif c.kind in ("mcp", "other"):
text = json.dumps(c.input, ensure_ascii=False)
close([k for k in runs if any(os.path.basename(x) and os.path.basename(x) in text for x in k[0])])
close(list(runs))
return found
def _duration(seconds) -> str:
if seconds < 120:
return "%d seconds" % round(seconds)
if seconds < 7200:
return "%d minutes" % round(seconds / 60.0)
return "%.1f hours" % (seconds / 3600.0)
def _pause(seconds) -> str:
minutes = int(round(seconds / 60.0))
if minutes < 120:
return "%d-minute" % minutes
if minutes < 48 * 60:
hours, minutes = divmod(minutes, 60)
return "%d-hour" % hours + (" %d-minute" % minutes if minutes else "")
days, minutes = divmod(minutes, 1440)
return "%d-day" % days + (" %d-hour" % (minutes // 60) if minutes >= 60 else "")
def _rebuilds(s, tl, counted) -> list:
items = []
for j, pos in enumerate(tl.calls):
e = s.events[pos]
if tl.rebuild_tokens[j] and counted(e):
gap = _epoch(e.ts) - _epoch(s.events[tl.calls[j - 1]].ts)
items.append(_item("rebuilds", s, e, tl.rebuild_tokens[j], tl.rebuild_dollars[j],
("%s pause before this call; %s tokens rebuilt" % (
_pause(gap), _fmt_tokens(tl.rebuild_tokens[j])), [])))
return items
def _oversized(s, tl, counted, claimed) -> list:
"""Results that added more than OVERSIZED_TOKENS to the prompt."""
found = []
for p, e in enumerate(s.events):
if e.kind != "tool" or e.tool is None or id(e) in claimed or not counted(e):
continue
entered = tl.entered(p, e.tool.output_chars)
if entered > OVERSIZED_TOKENS:
found.append({"category": "oversized", "anchor": e, "members": [(p, e)], "count": 1,
"carry_only": True, "group": _group(e.tool),
"evidence": ("{} returned about %s tokens: {}" % _fmt_tokens(entered),
[safe_text(e.tool.name, 80), safe_text(_target(e.tool, s.cwd))])})
return found
def _costed(s, tl, found) -> dict:
"""An item with the tokens and dollars of what a detector found: both costs
of each wasted tool call, or only the carrying of an oversized result."""
tokens = dollars = 0.0
for p, e in found["members"]:
t, d = tl.carried(p, e.tool.output_chars) if found.get("carry_only") else _tool_cost(tl, p, e.tool)
tokens, dollars = tokens + t, dollars + d
return _item(found["category"], s, found["anchor"], tokens, dollars, found["evidence"],
count=found["count"], group=found.get("group", ""))
# ---------------------------------------------------------------------------
# Analysis
# ---------------------------------------------------------------------------
def _fmt_tokens(n) -> str:
n = float(n)
for size, unit in ((1e9, "B"), (1e6, "M"), (1e3, "k")):
if abs(n) >= size:
return ("%.1f" % (n / size)).rstrip("0").rstrip(".") + unit
return "%d" % round(n)
def _show_time(ts) -> str:
stamp = _epoch(ts)
if stamp is None:
return ""
return datetime.datetime.fromtimestamp(stamp, tz=datetime.timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
def _summarize(items, totals) -> dict:
tokens = sum(i["tokens"] for i in items)
dollars = sum(i["dollars"] for i in items)
return {"count": sum(i["count"] for i in items), "tokens": int(round(tokens)), "dollars": round(dollars, 6),
"share_of_spend": round(dollars / totals["dollars"], 6) if totals["dollars"] else 0.0,
"share_of_tokens": round(tokens / totals["tokens"], 6) if totals["tokens"] else 0.0}
# User corrections: a short list of phrases, matched at the start of a message
# (after an optional "no," or "wait,") or, for the second list, anywhere.
_CORRECTION_LEAD = r"^\s*(?:(?:no|nope|wait|hmm|ugh|argh)\b[,.!]*\s*)?"
_CORRECTION_STARTS = [
r"(?:that'?s|that is|this is|it'?s|it is)\s+(?:wrong|not right|incorrect|not what i (?:asked|wanted|meant|said))\b",
r"(?:wrong|incorrect)\b",
r"you\s+(?:didn'?t|did not|forgot|missed|broke|ignored|skipped|never)\b",
r"why\s+(?:did|are|would)\s+you\b",
r"i\s+(?:said|told you|asked you|asked for|meant)\b",
r"(?:stop(?:\s+doing)?|don'?t\s+do|do not\s+do)\s+(?:that|this)\b",
r"(?:undo|revert)\s+(?:that|this|it|your|the last)\b",
r"not what i (?:asked|wanted|meant|said)\b",
r"(?:it|this|that)\s+(?:still\s+)?(?:do(?:es)?n'?t|do(?:es)? not|didn'?t|did not)\s+work\b",
]
_CORRECTION_ANYWHERE = [r"\bi (?:already )?told you\b", r"\bas i said\b", r"\byou(?:'re| are) not listening\b",
r"\bstill\s+(?:do(?:es)?n'?t|do(?:es)? not|didn'?t|did not)\s+work\b",
r"\bstill\s+(?:broken|failing|wrong|not working|nothing)\b",
r"\b(?:stop|quit)\s+(?:screwing|messing)\b"]
_CORRECTION_RE = re.compile(_CORRECTION_LEAD + "(?:" + "|".join(_CORRECTION_STARTS) + ")|"
+ "|".join(_CORRECTION_ANYWHERE), re.I)
def is_correction(text) -> bool:
return bool(_CORRECTION_RE.search((text or "").replace("\u2019", "'")))
def _failed(call) -> bool:
return call.is_error and not call.denied and not call.interrupted
def _ending(s):
"""How a main session ended: 'error', 'interrupt', or '' for a normal end."""
for e in reversed(s.events):
if (e.kind == "user" and e.injected) or e.kind == "compaction" or (
e.kind == "assistant" and not e.text and e.usage is None):
continue
if e.kind == "interrupt" or (e.kind == "tool" and e.tool is not None and (e.tool.denied or e.tool.interrupted)):
return "interrupt", e
if e.kind == "api_error" or (e.kind == "tool" and e.tool is not None and _failed(e.tool)):
return "error", e
return "", e
return "", None
def _session_failures(s, counted, tally) -> None:
"""Add one session's failure counts to `tally`."""
streak = []
def close():
if len(streak) >= STREAK_MIN and counted(streak[0]):
tally["error_streaks"] += 1
del streak[:]
for e in s.events:
if e.kind == "compaction" or (e.kind == "user" and not e.injected):
close()
if e.kind == "tool" and e.tool is not None:
c = e.tool
if _failed(c):
streak.append(e)
else:
close()
if counted(e) and _failed(c):
tally["tool_errors"][c.name] = tally["tool_errors"].get(c.name, 0) + 1
if counted(e) and c.denied:
tally["denials"][c.denied] = tally["denials"].get(c.denied, 0) + 1
elif counted(e) and not s.is_subagent:
if e.kind == "interrupt":
tally["interrupts"] += 1
elif e.kind == "user" and not e.injected and is_correction(e.text):
tally["corrections"] += 1
close()
if not s.is_subagent:
how, last = _ending(s)
if how and counted(last):
tally["ended_badly"][how] = tally["ended_badly"].get(how, 0) + 1
def _rate(count, base) -> float:
return round(100.0 * count / base, 2) if base else 0.0
def _failures(totals, tally, loops, main_sessions) -> list:
by_tool = {}
for _s, name, _n in loops:
by_tool[name] = by_tool.get(name, 0) + 1
counts = {
"tool_errors": (tally["tool_errors"], "tool calls"),
"error_streaks": (tally["error_streaks"], "tool calls"),
"denials": (tally["denials"], "tool calls"),
"interrupts": (tally["interrupts"], "user messages"),
"corrections": (tally["corrections"], "user messages"),
"loops": (by_tool, "tool calls"),
"ended_badly": (tally["ended_badly"], "sessions"),
}
bases = {"tool calls": totals["tool_calls"], "user messages": totals["user_messages"], "sessions": main_sessions}
rows = []
for fid, label in FAILURES:
value, per = counts[fid]
breakdown = {}
if isinstance(value, dict):
for k, v in sorted(value.items(), key=lambda kv: (-kv[1], kv[0])):
k = safe_text(k)
breakdown[k] = breakdown.get(k, 0) + v
value = sum(breakdown.values())
rows.append({"id": fid, "label": label, "count": value, "per_100": _rate(value, bases[per]), "per": per,
"breakdown": breakdown, "fix": _fix(fid)})
return rows
def _empty_counts() -> dict:
return {"sessions": 0, "subagent_sessions": 0, "model_calls": 0, "tool_calls": 0, "user_messages": 0,
"tokens": 0, "dollars": 0.0, "unpriced_tokens": 0}
def _anchor(heading) -> str:
"""The link anchor GitHub gives a markdown heading."""
return "-".join("".join(ch for ch in heading.lower() if ch.isalnum() or ch in " -").split())
def _fix(cid) -> str:
return "references/fixes.md#" + _anchor(FIX_SECTIONS[cid])
def analyze(sessions, since_days=30.0, now=None, harness="all", project=None) -> dict:
"""The whole report as a dict (the --json output)."""
sessions = list(sessions)
now = time.time() if now is None else now
cutoff = now - float(since_days) * 86400
totals, by_harness = _empty_counts(), {}
unpriced, unique, active = set(), set(), set()
sub = {"sessions": set(), "tokens": 0, "dollars": 0.0}
for s, e in transcripts.unique_events(sessions):
stamp = _epoch(e.ts)
if stamp is not None and stamp < cutoff:
continue
unique.add(id(e))
buckets = (totals, by_harness.setdefault(s.harness, _empty_counts()))
if id(s) not in active:
active.add(id(s))
for b in buckets:
b["sessions"] += 1
b["subagent_sessions"] += int(s.is_subagent)
if e.usage is not None:
tokens, cost = _tokens(e.usage), pricing.cost_usd(e.usage, e.model)
for b in buckets:
b["model_calls"] += 1
b["tokens"] += tokens
b["unpriced_tokens" if cost is None else "dollars"] += tokens if cost is None else cost
if cost is None:
unpriced.add(e.model or "(no model id)")
if s.is_subagent:
sub["sessions"].add(id(s))
sub["tokens"] += tokens
sub["dollars"] += cost or 0.0
elif e.kind == "tool":
for b in buckets:
b["tool_calls"] += 1
elif e.kind == "user" and not e.injected and not s.is_subagent:
for b in buckets:
b["user_messages"] += 1
def counted(e):
return id(e) in unique
items, loops = [], []
tally = {"tool_errors": {}, "error_streaks": 0, "denials": {}, "interrupts": 0, "corrections": 0,
"ended_badly": {}}
compactions = {"count": 0, "tokens_before": 0}
for s in sessions:
tl = _Timeline(s)
_session_failures(s, counted, tally)
for pos in tl.compactions:
if counted(s.events[pos]):
k = tl.issuer(pos)
compactions["count"] += 1
compactions["tokens_before"] += tl.prompt[k] if k >= 0 else 0
claimed, polled = set(), set() # a tool call counts in one waste row at most
polls, session_loops = _loops(s, counted, claimed, polled)
rereads = _rereads(s, counted, claimed, polled)
found = polls + rereads + _oversized(s, tl, counted, claimed)
tl.charge(set(tl.issuer(p) for f in polls + rereads for p, _e in f["members"]) - {-1})
items += [_costed(s, tl, f) for f in found] + _rebuilds(s, tl, counted)
loops += session_loops
waste = []
for cid, label in WASTE:
row = {"id": cid, "label": label}
row.update(_summarize([i for i in items if i["category"] == cid], totals))
row["fix"] = _fix(cid)
waste.append(row)
waste.sort(key=lambda r: (-r["dollars"], -r["tokens"]))
groups = {}
for i in items:
if i["category"] == "oversized":
groups.setdefault(i["group"], []).append(i)
oversized_groups = []
for name, members in groups.items():
row = {"group": safe_text(name)}
row.update(_summarize(members, totals))
oversized_groups.append(row)
oversized_groups.sort(key=lambda r: (-r["dollars"], -r["tokens"]))
labels = dict(WASTE)
examples = []
for i in sorted(items, key=lambda i: (-i["dollars"], -i["tokens"]))[:EXAMPLES]:
(s, e), (template, values) = (i["session"], i["event"]), i["evidence"]
examples.append({"category": i["category"], "label": labels[i["category"]],
"tokens": int(round(i["tokens"])), "dollars": round(i["dollars"], 6),
"harness": s.harness, "session": safe_text(s.id[:8]),
"folder": safe_text(_show_path(s.cwd, ""), 400) if s.cwd else "",
"path": safe_text(_show_path(s.path, ""), 400),
"time": _show_time(e.ts), "evidence": template.format(*values),
"evidence_template": template, "evidence_values": values})
for h, counts in by_harness.items():
spent = [i for i in items if i["session"].harness == h]
counts["dollars"] = round(counts["dollars"], 6)
counts["waste"] = {}
for cid, _label in WASTE:
mine = [i for i in spent if i["category"] == cid]
counts["waste"][cid] = {"count": sum(i["count"] for i in mine),
"tokens": int(round(sum(i["tokens"] for i in mine))),
"dollars": round(sum(i["dollars"] for i in mine), 6)}
counts["waste_dollars"] = round(sum(i["dollars"] for i in spent), 6)
counts["waste_tokens"] = int(round(sum(i["tokens"] for i in spent)))
if counts["dollars"]:
counts["waste_share"], counts["waste_share_of"] = round(counts["waste_dollars"] / counts["dollars"], 6), "spend"
else:
counts["waste_share"] = round(counts["waste_tokens"] / float(counts["tokens"]), 6) if counts["tokens"] else 0.0
counts["waste_share_of"] = "tokens"
main_sessions = sum(1 for x in sessions if id(x) in active and not x.is_subagent)
warned = [x for x in sessions if x.warnings]
result = {
"window_days": since_days, "harness": harness,
"project": safe_text(project, 400) if project else None,
"prices_checked": pricing.PRICES_CHECKED, "headline": "",
"totals": dict(totals, dollars=round(totals["dollars"], 6),
unpriced_models=sorted(safe_text(m) for m in unpriced)),
"by_harness": {h: by_harness[h] for h in sorted(by_harness)},
"waste": waste, "oversized_groups": oversized_groups,
"compactions": {"count": compactions["count"], "tokens_before": compactions["tokens_before"],
"average_tokens_before": int(round(compactions["tokens_before"] / float(compactions["count"])))
if compactions["count"] else 0, "fix": _fix("compactions")},
"subagents": {"sessions": len(sub["sessions"]), "tokens": sub["tokens"], "dollars": round(sub["dollars"], 6),
"share_of_spend": round(sub["dollars"] / totals["dollars"], 6) if totals["dollars"] else 0.0,
"share_of_tokens": round(sub["tokens"] / float(totals["tokens"]), 6) if totals["tokens"] else 0.0,
"fix": _fix("subagents")},
"failures": _failures(totals, tally, loops, main_sessions),
"examples": examples, "notes": [],
"warnings": {"sessions": len(warned), "lines": sum(len(x.warnings) for x in warned)},
}
result["headline"] = _headline(result)
result["notes"] = _notes(result, safe_text)
return result
# ---------------------------------------------------------------------------
# Report
# ---------------------------------------------------------------------------
HEADLINE_PHRASES = {"rereads": "re-reading files that had not changed",
"oversized": "tool results over 10,000 tokens",
"rebuilds": "cache rebuilds after pauses", "polling": "polling loops"}
UNITS = {"rereads": "re-read", "oversized": "result", "rebuilds": "pause", "polling": "loop"}
def _plural(n, word) -> str:
return "%s %s%s" % (format(n, ","), word, "" if n == 1 else "s")
def _money(d) -> str:
if d >= 10:
return "$" + format(int(round(d)), ",")
if d >= 0.005:
return "$%.2f" % d
return "<$0.01" if d > 0 else "$0"
def _pct(share) -> str:
x = share * 100.0
if x >= 1:
return "%d%%" % round(x)
if x >= 0.1:
return "%.1f%%" % x
return "<0.1%" if x > 0 else "0%"
def _window_words(days) -> str:
if days < 1:
return _plural(int(round(days * 24)), "hour")
return "%s day%s" % (("%g" % days), "" if days == 1 else "s")
HEADLINE_DOLLARS = 0.005 # a row makes the headline from half a cent, or 1,000 tokens when nothing is priced
HEADLINE_TOKENS = 1000
def _ranked(waste) -> tuple:
"""(rows for the headline, whether they are ranked by dollars): by dollars
when any row has a price, otherwise by tokens."""
if any(c["dollars"] > 0 for c in waste):
return sorted([c for c in waste if c["dollars"] >= HEADLINE_DOLLARS], key=lambda c: -c["dollars"]), True
return sorted([c for c in waste if c["tokens"] >= HEADLINE_TOKENS], key=lambda c: -c["tokens"]), False
def _headline(r) -> str:
t, window = r["totals"], _window_words(r["window_days"])
if not t["sessions"]:
return "No sessions found in the last %s." % window
ranked, priced = _ranked(r["waste"])
if not ranked:
cost = ", %s at API prices" % _money(t["dollars"]) if t["dollars"] else ""
if not any(c["dollars"] or c["tokens"] for c in r["waste"]):
what = "no waste found"
else:
what = "waste under $0.01" if priced else "waste under 1,000 tokens"
return "Last %s: %s in %s (%s tokens%s)." % (
window, what, _plural(t["sessions"], "session"), _fmt_tokens(t["tokens"]), cost)
first = ranked[0]
if priced:
text = "Last %s: %s of spend (%s at API prices) went to %s" % (
window, _pct(first["share_of_spend"]), _money(first["dollars"]), HEADLINE_PHRASES[first["id"]])
if len(ranked) > 1:
text += ", and %s to %s" % (_money(ranked[1]["dollars"]), HEADLINE_PHRASES[ranked[1]["id"]])
else:
text = "Last %s: %s of tokens went to %s" % (
window, _pct(first["share_of_tokens"]), HEADLINE_PHRASES[first["id"]])
if len(ranked) > 1:
text += ", and %s tokens to %s" % (_fmt_tokens(ranked[1]["tokens"]), HEADLINE_PHRASES[ranked[1]["id"]])
return text + "."
def _notes(r, show) -> list:
"""The notes, with `show` applied to model ids: safe_text for JSON, code for markdown."""
t, notes = r["totals"], []
if t["sessions"]:
notes.append("Dollars are API list prices checked %s. On a subscription plan, read them as relative cost."
% r["prices_checked"])
notes.append("This report covers what the transcripts record. Some calls never reach them, such as the "
"one that writes a compaction summary.")
if t["unpriced_tokens"]:
notes.append("%s tokens on models with no known price (%s) are left out of the dollar figures." % (
_fmt_tokens(t["unpriced_tokens"]), ", ".join(map(show, t["unpriced_models"]))))
w = r["warnings"]
if w["lines"]:
notes.append("%s in %s could not be read or had an unknown record type, and %s skipped." % (
_plural(w["lines"], "line"), _plural(w["sessions"], "session"), "was" if w["lines"] == 1 else "were"))
if "opencode" in r["by_harness"]:
notes.append("OpenCode support follows its documented database layout and has not been checked on a "
"real install.")
if r["harness"] == "cursor":
notes.append("Cursor keeps no transcripts with tool results, token usage, or times, "
"so there is nothing to measure.")
return notes
def _num(x) -> str:
return ("%.2f" % x).rstrip("0").rstrip(".")
def _top(breakdown, show=str, n=3) -> str:
return ", ".join("%s %s" % (show(k), format(v, ",")) for k, v in list(breakdown.items())[:n])
def render_markdown(r) -> str:
t = r["totals"]
out = ["**%s**" % r["headline"], ""]
if t["sessions"]:
names = [HARNESS_NAMES.get(h, h) for h in r["by_harness"]]
where = " and ".join([", ".join(names[:-1]), names[-1]]) if len(names) > 1 else names[0]
out.append("%s%s in %s: %s, %s tokens, %s at API prices. For plain totals by day and model, run ccusage."
% (_plural(t["sessions"], "session"),
" (%s of them subagents)" % format(t["subagent_sessions"], ",") if t["subagent_sessions"] else "",
where, _plural(t["model_calls"], "model call"), _fmt_tokens(t["tokens"]), _money(t["dollars"])))
share, share_key = ("Share of spend", "share_of_spend") if t["dollars"] else ("Share of tokens", "share_of_tokens")
out += ["", "## Waste", "", "| Waste | Count | Tokens | Dollars | %s | Fix |" % share, "|---|---|---|---|---|---|"]
for c in r["waste"]:
out.append("| %s | %s | %s | %s | %s | %s |" % (c["label"], _plural(c["count"], UNITS[c["id"]]),
_fmt_tokens(c["tokens"]), _money(c["dollars"]),
_pct(c[share_key]), FIX_SHORT[c["id"]]))
out.append("")
if r["oversized_groups"]:
out.append("Largest groups of oversized results: %s." % "; ".join(
"%s (%s, %s tokens, %s)" % (code(g["group"]), _plural(g["count"], "result"), _fmt_tokens(g["tokens"]),
_money(g["dollars"])) for g in r["oversized_groups"][:5]))
comp, sub = r["compactions"], r["subagents"]
if comp["count"]:
out.append("Compactions: %s, with about %s tokens of context before each. Fix: %s." % (
format(comp["count"], ","), _fmt_tokens(comp["average_tokens_before"]), FIX_SHORT["compactions"]))
else:
out.append("Compactions: none.")
if sub["sessions"]:
out.append("Subagents: %s of %s (%s, %s tokens) in %s. Fix: %s." % (
_pct(sub[share_key]), share.split()[-1], _money(sub["dollars"]), _fmt_tokens(sub["tokens"]),
_plural(sub["sessions"], "subagent session"), FIX_SHORT["subagents"]))
else:
out.append("Subagents: none.")
out += ["", "## Failures", "", "| Failure | Count | Rate | Most common | Fix |", "|---|---|---|---|---|"]
for f in r["failures"]:
out.append("| %s | %s | %s per 100 %s | %s | %s |" % (f["label"], format(f["count"], ","),
_num(f["per_100"]), f["per"],
_top(f["breakdown"], code if f["id"] in BY_TOOL else str),
FIX_SHORT[f["id"]]))
out += ["", "## Top examples", ""]
if not r["examples"]:
out.append("No waste found.")
for n, x in enumerate(r["examples"], 1):
out.append("%d. %s: %s, %s tokens. %s." % (n, x["label"], _money(x["dollars"]), _fmt_tokens(x["tokens"]),
x["evidence_template"].format(*map(code, x["evidence_values"]))))
out.append(" %s session %s%s%s" % (HARNESS_NAMES.get(x["harness"], x["harness"]), code(x["session"]),
" at " + x["time"] if x["time"] else "",
", in %s" % code(x["folder"], 400) if x["folder"] else ""))
out += ["", "## By harness", "",
"| Harness | Sessions | Model calls | Tokens | Dollars | Waste share | Top waste |",
"|---|---|---|---|---|---|---|"]
labels = dict(WASTE)
for h, c in r["by_harness"].items():
rows = [dict(v, id=k) for k, v in c["waste"].items()]
key = "dollars" if any(x["dollars"] > 0 for x in rows) else "tokens"
top = max(rows, key=lambda x: x[key])
top_text = "none" if not top[key] else "%s %s" % (labels[top["id"]], _money(top["dollars"]) if
key == "dollars" else _fmt_tokens(top["tokens"]) + " tokens")
share = _pct(c["waste_share"]) + ("" if c["waste_share_of"] == "spend" else " of tokens")
out.append("| %s | %s | %s | %s | %s | %s | %s |" % (
HARNESS_NAMES.get(h, h), format(c["sessions"], ","), format(c["model_calls"], ","),
_fmt_tokens(c["tokens"]), _money(c["dollars"]), share, top_text))
out.append("")
out.append("Each fix is explained in references/fixes.md, in the section named like the row.")
notes = _notes(r, code)
if notes:
out += ["", "## Notes", ""] + ["- " + n for n in notes]
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# Command line
# ---------------------------------------------------------------------------
def load_sessions(harness="all", since_days=30.0, project=None, home=None) -> list:
"""Sessions oldest file first, so a fork's copied records count in the original."""
found = transcripts.find_sessions(harness=harness, since_days=since_days, project=project, home=home)
sessions = []
for h, p in reversed(found):
s = transcripts.load_session(h, p)
for e in s.events:
if e.tool is not None:
e.tool.output = "" # only the size (output_chars) is used; this keeps memory low
sessions.append(s)
return sessions
def main(argv=None, home=None) -> int:
parser = argparse.ArgumentParser(
prog="waste.py", description=__doc__.strip().split("\n\n")[0],
epilog="Exit codes: 0 done (also when no sessions are found), 2 usage error or a report file that "
"cannot be written.")
parser.add_argument("--since", type=_since_arg, default=30.0, metavar="30d",
help="how far back to look: days (30d or 30), hours (12h), or weeks (2w); default 30d")
parser.add_argument("--harness", default="all", choices=("all",) + transcripts.HARNESSES + ("cursor",),
help="which harness to read (default: all)")
parser.add_argument("--project", metavar="PATH",
help="only sessions whose working folder is this folder or inside it")
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")
args = parser.parse_args(argv)
try:
(args.project or "").encode("utf-8")
except UnicodeEncodeError:
print("error: --project %s is not a UTF-8 path, so it cannot match a session's folder"
% safe_text(args.project, 400), file=sys.stderr)
return 2
sessions = [] if args.harness == "cursor" else load_sessions(args.harness, args.since, args.project, home)
result = analyze(sessions, since_days=args.since, harness=args.harness, project=args.project)
text = json.dumps(result, indent=2, ensure_ascii=False) + "\n" if args.json else render_markdown(result)
if args.out:
try:
with open(args.out, "w", encoding="utf-8") as fh:
fh.write(text)
except OSError as exc:
print("error: cannot write the report to %s: %s" % (safe_text(args.out, 400),
exc.strerror or type(exc).__name__), file=sys.stderr)
return 2
print("Report written to %s" % args.out)
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
sys.exit(main())