A hook for Claude Code and Codex that stops a live session when the agent repeats the same call, keeps failing, or spends past a dollar cap you set.
What you get
A sample. The numbers are invented; your install prints its own.
The install shows its plan first and writes nothing until you say yes:
**Runaway guard will stop Claude Code when the same call repeats 3 times, 5 calls fail in a row, or spend passes $10.00.**
Dry run: nothing is written yet. Run the same command with --write to apply this change.
Settings file ~/.claude/settings.json: adds one PreToolUse hook entry and leaves the 2 PreToolUse hooks already there unchanged.
Later, in the middle of a session, the guard steps in on its own. At the cap, the turn ends and you see a message such as:
Runaway guard: spend cap reached ($10.07 of $10.00); tool calls are blocked. To go on, raise
spend_cap_usd in /Users/you/.local/state/runaway-guard/runaway-guard.json, or start the count over
by running this in a terminal: python3 /Users/you/.claude/skills/runaway-guard/scripts/status.py
--session 5f0c3a1e-8d7b-4c2a-9e61-0123456789ab --reset
And the status report shows what it counted:
**Session `5f0c3a1e` has spent $10.07 of its $10.00 cap; the guard stepped in 3 times.**
| Trip wire | Limit | Now | Stepped in |
|---|---|---|---|
| Spend | $10.00 per session, warning at 80% | $10.07 (101%) | 1 call blocked |
| Loop | the same call 3 times with nothing changed | | 2 calls blocked |
| Failures | 5 failed calls in a row | 1 in a row | stepped in 0 times |
Latest events:
- 2026-09-28 14:02 UTC, loop: `Bash`, 3rd identical call
- 2026-09-28 14:31 UTC, warning: `Read`, $8.04 of the $10.00 cap
- 2026-09-28 14:40 UTC, spend: `Edit`, $10.07 of the $10.00 cap
Add --harness codex for Codex and --scope project to guard only the current project. Check a session with status.py, and start its counts over with status.py --reset.
To remove the hook, run install.py --uninstall, then again with --write. With no --harness or --scope, it checks Claude Code and Codex, user and project settings, and names each file it changes. Run it before you move, update, or remove the skill, then install again from the new place: the hook entry points at the skill's folder. If the folder disappears first, the hook does nothing and never blocks a call, and --uninstall from the new copy still removes the old entry.
How it works
Claude Code and Codex can run a command before every tool call, pass it the call as JSON, and let it allow, ask about, or block the call. That command is a hook. install.py adds one hook entry to your settings, next to any hooks you already have. Before each tool call, guard.py:
Spend: reads only the new lines of the session's transcript and its subagents' transcripts, counts each model response once, and prices it at API list prices. A model missing from the price table is priced like the most expensive known model of its family, so the cap still holds. It warns you at 80% of the cap. At the cap it blocks every tool call, and in Claude Code it ends the turn.
Loop: compares the call's tool name and input with the calls before it. The same call a third time, with only reads and searches in between, is blocked. An edit, another command, or a new prompt from you starts the count over, so re-running tests after a fix never trips it.
Failures: counts failed calls in a row from the transcript. After 5, Claude Code asks you whether the next call should run.
It keeps a small file per session with counts and hashes, answers in about 50 ms, and lets the call through if anything inside it goes wrong. How it decides covers every rule, with the numbers from replaying real sessions: across 14,663 tool calls from the 15 busiest sessions on one Mac, the loop rule never fired and the failure rule asked once.
Works with
Harness
Support
What differs
Claude Code
full; replayed on real sessions
all three trip wires; asks you after 5 failures; warns at 80%
Codex
loop, failures, and spend; transcript reading replayed on a real session
Codex hooks cannot ask or warn, so after 5 failures the call is blocked and the agent is told to ask you, and the 80% warning shows only in status.py; you trust the hook once in /hooks; not yet tested in a live Codex session
Gemini CLI
not supported
already detects loops on its own; no confirmed prices for a spend cap
OpenCode
not supported
already asks on repeated identical calls (doom_loop); extensions are JavaScript plugins
Cursor
not supported
its transcripts have no token counts; Cursor also runs Claude Code hooks, and the guard is built to answer them with no decision (not tested in Cursor)
It runs on macOS and Linux with Python 3.9 or newer and nothing else to install. Hook support per harness has the details and sources.
Limits
What the dollar figure is:
An estimate at API list prices. On a subscription plan it measures usage, not your bill. Batch discounts, fast mode, and regional pricing are left out.
A little behind. The check runs before each tool call, so a response that makes no tool call is counted at the next one, and each subagent's last response stays uncounted. The total runs about 1% to 3% low, more with many subagents.
High for a model the price table does not know yet, until you update the skill.
Counted from the guard's first check. A session that had already spent more than the cap when the guard first saw it is counted from that point, and the guard says so once.
What the trip wires leave to the spend cap:
Two different commands that alternate end each other's runs, so the loop rule misses them.
Commands that wait on purpose (sleep, wait) and calls that start subagents never count as loops.
A failure question can come one call late, because the newest results are read at the next call.
Where it runs:
In print mode (claude -p) no one can answer the failure question, so it counts as a no. Use --max-budget-usd there.
The first check of a long session reads the whole transcript once: about a quarter of a second for 5,000 lines.
Each check takes about 50 ms when Python can cache the compiled scripts next to them, as it does by default. With PYTHONDONTWRITEBYTECODE set, a check takes about 175 ms.
Windows outside WSL2 is untested.
Privacy
Read: the current session's transcript files and your settings files, on this machine. For Codex it also reads the first line of other recent rollout files to find this session's subagents, and keeps only their file names.
Kept: one small JSON file per session in ~/.local/state/runaway-guard/, with read positions, dollar totals, hashes of recent tool calls and responses (not their content), and the latest events (time, trip wire, tool name). Files older than 30 days are removed.
Written: your settings file and the cap file, only with --write. The install also keeps a private copy of your settings file as it was, so that --uninstall can put it back exactly; the uninstall deletes that copy.
Built-in parts of the same idea: Claude Code's print-mode --max-budget-usd (CLI reference), OpenCode's doom_loop (permissions), and Gemini CLI's loop detection.
---
name: runaway-guard
description: >-
Runaway guard: a hook that stops a live Claude Code or Codex session when
the agent loops on the same tool call, keeps failing, or exceeds a dollar
cap. Use when the user wants a spending cap, budget limit, or cost ceiling
for interactive agent sessions (the built-in --max-budget-usd works only in
print mode); wants to halt an agent that repeats the same command, is stuck
in a loop, or runs up a bill unattended overnight; wants a circuit breaker
for several failed tool calls in a row; or asks how much the current session
has spent against its cap, why a call was blocked, or how to raise or reset
the cap. Runs locally: reads the session transcripts before each tool call
and sends nothing.
license: MIT
compatibility: "Python 3.9+ on macOS or Linux, and Claude Code or Codex with hooks enabled. Makes no network calls."
metadata:
author: "Ryan Alberts"
version: "1.0.0"
source: "https://github.com/RyanAlberts/best-of-Agent-Harnesses"
---
# Runaway guard
An agent left alone can repeat a failing command, retry broken calls, or keep spending long after the
work stopped paying off, and Claude Code's `--max-budget-usd` cap works only in print mode. This skill
installs a **hook** (a command the harness runs before every tool call) that steps in at three trip
wires: the same call a third time with nothing changed, five failed calls in a row, and a dollar cap.
The hook reads the session's transcript files on this machine, keeps only counts and hashes, and sends
nothing anywhere.
## When to use
- The user wants a spend cap, budget, or cost ceiling for interactive Claude Code or Codex sessions.
- The user wants an agent stopped when it loops on one command or keeps failing, for example before
leaving it unattended.
- The user asks how much the current session has spent against its cap, why the guard blocked a call,
or how to raise the cap or start the count over.
- The user wants to remove the guard.
## When not to use
- Finding where past tokens and money went: use `session-waste-report`.
- Testing whether permission rules and hooks stop dangerous commands: use `guardrail-tester`.
- Turning a rule from AGENTS.md or CLAUDE.md into a hook: use `rules-to-guards`.
- A one-off print-mode run (`claude -p`): its own `--max-budget-usd` and `--max-turns` flags cap it.
- Gemini CLI, Cursor, and OpenCode: the guard does not install there. Gemini CLI and OpenCode already
detect loops; `references/harness-hooks.md` says what each lacks.
## Steps
`<skill-dir>` means the folder that holds this SKILL.md (Claude Code shows it as the skill's base
directory). Run every command through `python3` exactly as written, with the path in double quotes:
skill folders can sit under paths with spaces.
1. **Explain the three trip wires and their defaults** in plain words:
- **Loop**: the same tool call with the same input, a third time with nothing changed in between,
is blocked. Reads and searches in between change nothing; an edit, another command, or a new
user prompt does. Re-running tests after an edit never trips it, and neither do deliberate waits
or starting subagents.
- **Failures**: after 5 failed tool calls in a row, Claude Code asks the user before the next call
runs. Codex hooks cannot ask, so there the call is blocked with a message telling the agent to
ask the user.
- **Spend**: the running cost of the session and its subagents at API list prices. A warning at 80%
of the cap; at the cap every tool call is blocked (in Claude Code the turn ends) until the user
raises the cap or starts the count over. A model missing from the price table is priced like the
most expensive known model of its family, so the cap still holds.
Done when the user has heard all three wires and the $10 default cap.
2. **Ask for three choices**: the dollar cap, the harness (Claude Code or Codex), and the scope (user:
every session of this user; project: only sessions in the current project). Say that on a
subscription plan the dollar figure measures usage at API prices, not the bill. Done when you have
all three answers, or the user accepted the defaults: Claude Code, user scope, $10.
3. **Show the dry run** with the user's choices:
```bash
python3 "<skill-dir>/scripts/install.py" --harness claude-code --scope user --cap 10
```
It prints the headline, the settings file it would change, the diff, and notes, and writes nothing.
Add `--project <folder>` for project scope when the agent's working folder is not the project.
Done when the user has seen the headline and the diff and has said yes or no.
4. **Install only on a clear yes**: run the same command with `--write` added. Done when the output
starts with the headline and says "Done". Exit code 2 means a settings file could not be read as
JSON, has an unexpected shape, or could not be written; nothing was changed. Quote the message,
and leave the file for the user to fix.
5. **Relay the notes** printed under "Notes", in particular:
- Claude Code normally applies hook changes to sessions already running. A session the guard
already counts is blocked at its next tool call once it is over the cap; a session it sees for the
first time after spending more than the cap is counted from that point.
- Codex runs the hook only after the user trusts it in `/hooks`.
- Before the user moves, updates, or removes this skill, they run `--uninstall`, then install again
from the new place: the hook entry points at this folder.
Done when every note is passed on.
6. **Offer status**:
```bash
python3 "<skill-dir>/scripts/status.py"
```
Done when the user has the headline, or declined.
To remove the guard, run `python3 "<skill-dir>/scripts/install.py" --uninstall`, show the diff, and
add `--write` on a clear yes. With no `--harness`, `--scope`, or `--settings`, it checks all four places
the guard can be (Claude Code and Codex, user and project) and names each file it changes. It removes
only the guard's own entries, and puts a file back exactly as it was when nothing else changed it.
## When the guard stops a call
A blocked call comes back as a tool error that starts with "Runaway guard". Stop and pass the message
to the user in your own words, naming the trip wire.
- **Spend stop**: every tool call is blocked, `status.py` included, and in Claude Code the turn ends.
Answer in text only. Give the user the two ways on that the message names, for them to use in their
own terminal: the setting that holds the cap (the file, or the `RUNAWAY_GUARD_CAP_USD` variable), and
the `status.py ... --reset` command. To see the counts, they run `python3 "<skill-dir>/scripts/status.py"`
there too. The cap is theirs to lift.
- **Loop**: change the approach rather than the wording of the same call.
- **Failure question**: the user answers it in the permission prompt; continue with what they allow.
## Read the results
`install.py`:
- **Headline**: the limits the hook will enforce after the change, from `--cap`, the guard's settings
files, and any `RUNAWAY_GUARD_*` variable set in the shell.
- **Diff**: the one PreToolUse entry added or removed; the hooks already there stay unchanged.
- **Notes**: steps the user still takes, and settings that change the picture, such as
`disableAllHooks`.
- `--json` for an install: `headline`, `action`, `settings_path`, `config_path`, `command`, `cap_usd`,
`limits`, `changed`, `written`, `other_hooks`, `notes`. For `--uninstall`: `headline`, `action`,
`hook_entries_removed`, `changed`, `written`, `notes`, and `targets` (each file checked, with
`removed` and `restored_exactly`).
`status.py` (defaults to the session the guard checked most recently; `--session` takes an id or its
first characters, `--list` shows every session):
- **Headline**: the session's first 8 characters, dollars since the start or the last reset, the cap,
and how many times the guard stepped in.
- **Table**: each wire's limit, its current value, and how often it fired.
- **Latest events**: time, wire, tool name, detail. Warnings appear here too.
- **Estimate line**: the part of the dollar figure priced by estimate, for models missing from the
price table, and which models they are.
- `--json`: `session`, `spent_usd`, `total_usd`, `cap_usd`, `limits`, `trips`, `recent_trips`,
`failure_streak`, `transcripts`, `estimated_usd`, `estimated_tokens`, `estimated_models`, `resets`,
`errors_logged`.
`references/how-it-decides.md` has every rule, threshold, setting, and safeguard, with calibration on
real sessions; `references/harness-hooks.md` has what each harness's hooks allow, with sources.
## Report to the user
After an install or a dry run:
1. The headline, verbatim, in bold.
2. What changes: the settings file and the one entry, and the cap file when `--cap` was given.
3. The notes, as a short list.
4. One line: how to check status, and how to remove the guard.
For status:
1. The headline, verbatim, in bold.
2. The table.
3. When a wire has fired: the latest events and the next step. For a spend stop, the raise or reset
command from the block message, for the user to run in their own terminal.
## Files
- `scripts/guard.py`: the hook. Reads the hook input on stdin and prints a decision. Always exits 0,
and lets the call through on any internal error, logging it to `errors.log` in the state folder.
- `scripts/install.py`: prints the settings change and applies it with `--write`; `--uninstall`
removes only the guard's entries. Flags: `--harness`, `--scope`, `--project`, `--cap`, `--settings`,
`--json`, `--out`.
- `scripts/status.py`: spend and trip counts per session; `--list`, `--session`, `--reset`, `--json`,
`--out`.
- `scripts/transcripts.py`, `scripts/pricing.py`, `scripts/safe.py`: the shared transcript reader, price
table, and text cleaner that puts session ids, tool names, and model ids in the report inside inline
code, copied from this repository's shared code.
- `references/how-it-decides.md`: the trip wires, thresholds, settings, safeguards, and speed.
- `references/harness-hooks.md`: hook support in Claude Code, Codex, Gemini CLI, Cursor, and OpenCode.
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.
# Hook support per harness
What each coding agent's hooks allow, and what runaway guard does with it. Checked 2026-09-28 against each harness's documentation (linked per section) and, where noted, against real session files on one Mac.
## Claude Code: supported
- **Where the hook goes**: `~/.claude/settings.json` (user) or `.claude/settings.local.json` (project, not shared with the team). `$CLAUDE_CONFIG_DIR` moves the user folder. Source: https://code.claude.com/docs/en/hooks
- **Shape**: `{"hooks": {"PreToolUse": [{"hooks": [{"type": "command", "command": "...", "timeout": 10}]}]}}`. With no `matcher`, the hook runs for every tool. Hooks from all settings files run; the guard's entry sits beside the ones already there. The command is `python3 '<skill folder>/scripts/guard.py' || true`: without `|| true`, a moved or deleted skill folder would make `python3` exit 2, and exit 2 blocks the call.
- **Input the guard reads**: `session_id`, `transcript_path`, `cwd`, `prompt_id` (v2.1.196 and later), `tool_name`, `tool_input`, `tool_use_id`, and `agent_id` for calls made inside a subagent.
- **Answers the guard gives**: `permissionDecision: "deny"` (the reason is shown to Claude), `permissionDecision: "ask"` (the reason is shown to the user), and `systemMessage` (a message shown to the user). At the spend cap the block also carries `continue: false` and a `stopReason`, which end the turn. Exit code 2 would also block; the guard never uses it.
- **Hook changes**: Claude Code normally applies changes to hook settings in sessions that are already running.
- **Timeouts**: 600 seconds by default; the guard's entry sets 10. A PreToolUse hook that times out does not block the call.
- **Transcripts**: `~/.claude/projects/<project>/<session>.jsonl`, with subagents in `<session>/subagents/agent-<id>.jsonl`. One API response is written as several records that share a message id; the reader keeps the last one's usage. Source: https://code.claude.com/docs/en/sessions
- **Built-in caps**: `--max-budget-usd` and `--max-turns` work in print mode (`claude -p`) only. Source: https://code.claude.com/docs/en/cli-reference
- **Tested**: unit tests with transcripts built from the documented record shapes, and a replay of real local sessions through the hook (see [how-it-decides.md](how-it-decides.md)).
## Codex: supported, with two gaps
- **Where the hook goes**: `~/.codex/hooks.json` (user; `$CODEX_HOME` moves it) or `<repo>/.codex/hooks.json` (project). The shape matches Claude Code's. Source: https://learn.chatgpt.com/docs/hooks.md
- **Trust**: Codex runs a new or changed hook only after the user trusts it in `/hooks`. Project hooks run only in trusted projects.
- **Input the guard reads**: `session_id` (subagent hooks get the parent's id), `transcript_path` (may be null; then only the loop wire works), `turn_id`, `model`, `tool_name` (shell commands arrive as `Bash`, patches as `apply_patch`), `tool_input`, `tool_use_id`.
- **Answers**: `permissionDecision: "deny"` blocks. On PreToolUse, `ask`, `continue`, `stopReason`, and `suppressOutput` mark the hook as failed and the call proceeds, so the guard sends none of them to Codex.
- Gap 1: no question. After a run of failed calls the guard blocks the next call and tells the agent to ask the user.
- Gap 2: no warning. The early spend warning shows only in `status.py`.
- **Not seen by hooks**: hosted tools such as web search.
- **Transcripts**: `$CODEX_HOME/sessions/YYYY/MM/DD/rollout-*.jsonl`. Each subagent writes its own rollout, and its first line (`session_meta`) carries the root session id. The model comes from `turn_context` lines; the guard carries it from one read to the next. Source: https://github.com/openai/codex/blob/main/codex-rs/protocol/src/protocol.rs
- **Tested**: unit tests from the documented shapes, and a replay of one real Codex session with 36 subagent rollouts, which the guard found and priced within 0.6% of a full read. The hook input and answers follow the Codex documentation; they were not tested in a live Codex session.
## Gemini CLI: not supported
- Gemini CLI has `BeforeTool` hooks that can deny, with timeouts in milliseconds. Source: https://github.com/google-gemini/gemini-cli/blob/main/docs/hooks/reference.md
- It already detects loops, on by default (`model.disableLoopDetection` turns it off).
- A spend cap needs two things Gemini CLI does not give this guard: its chat files are an operation log that checkpoints and rewinds rewrite, which the shared reader does not read incrementally, and no Gemini price was confirmed on Google's official pricing page.
## Cursor: not supported
- Cursor's `preToolUse` hooks can deny; an `ask` answer is not enforced there. Source: https://cursor.com/docs/hooks.md
- Its transcripts hold no tool results, token usage, or timestamps, so neither the failure wire nor the spend wire could work.
- Cursor also runs the hooks in Claude Code settings files, a setting that is on by default, with its own tool names (`Shell`, not `Bash`). Source: https://cursor.com/docs/reference/third-party-hooks.md. The guard recognizes Cursor's input (it carries `cursor_version`) and answers `{}`, which makes no decision: Cursor reads invalid JSON from a permission hook as a block, and empty output may count as invalid. This has not been tested in Cursor.
## OpenCode: not supported
- OpenCode has no shell-command hooks; extensions are JavaScript plugins (`tool.execute.before`). Source: https://opencode.ai/docs/plugins
- It already ships the `doom_loop` permission, which asks when the same tool call repeats 3 times with identical input. Source: https://opencode.ai/docs/permissions
# How runaway guard decides
Before each tool call, the harness runs `guard.py` as a **PreToolUse hook**: a command the harness starts with the call's details as JSON on its input, before the call happens. The guard answers with nothing (the call goes on through the normal permission flow), a message, a question, or a block. The checks run in this order, and the first one that fires decides:
1. Spend cap reached: block (in Claude Code the turn also ends).
2. Loop: block.
3. Failure streak: ask the user (Codex: block, see below).
4. Messages: the spend warning, and the first model priced by estimate. The call goes on.
## Spend
**What it counts.** Every finished model response (one API call) in the session's transcript and in each subagent's transcript. The token counts come from the usage fields the harness writes. A response that the harness writes as several records counts once, with its final numbers. Each response is priced by `pricing.py` at API list prices checked on 2026-09-28: uncached input, cache reads, cache writes (5-minute and 1-hour), and output, per model. Model ids from Amazon Bedrock and Google Vertex AI get the model's list price.
**Each response counts once.** The guard keeps a short hash of every response it counted. A transcript read again from the start (a file that was replaced by a shorter one) adds nothing twice. A resumed session whose file begins with a copy of an earlier session's records skips the responses that the earlier session already counted: a small shared list in the state folder (`counted-ids.bin`, the latest 32,768 responses) records which session counted each one.
**Which transcripts.**
- Claude Code: the session file named in the hook input, plus every `agent-*.jsonl` under `<session>/subagents/` and `<session>/subagents/workflows/*/`.
- Codex: the rollout file named in the hook input, plus each subagent rollout whose first line (`session_meta`) carries the same session id. The guard looks in the main rollout's day folder and in today's and yesterday's folders. It reads each new file's first line, and marks the file as checked once that line is there.
**Lag.** The response still being written is counted at the next tool call. The last response of each subagent is never followed by another, so it is not counted. The total runs about 1% to 3% low, more with many subagents.
**Models missing from the price table.** A newer model, such as the next Opus version, is priced like the most expensive known model of its family (Claude Opus, Sonnet, Haiku, or Fable, or GPT; any Codex model counts as GPT), and a model of no known family like the most expensive known model. The estimate is too high rather than $0, so the cap still holds. Claude Code shows one message naming the first such model in a session; `status.py` shows how much of the total is estimated.
**The warning.** Once per cap, when spend is between `warn_at` (default 0.8) of the cap and the cap. Claude Code shows it to the user. Codex hooks have no way to show a message, so it appears only in `status.py`.
**The stop.** At the cap, every tool call is blocked, subagents' calls included. Four calls still pass, because they stop work or hand in a final answer: `TaskStop`, `KillShell`, `KillBash`, `StructuredOutput`. In Claude Code the block also ends the turn (`continue: false`), so each block does not start another model response. The block message tells the agent to stop and tell the user, and names the two ways to go on, for wherever the cap in force comes from:
- A project file: raise or remove `spend_cap_usd` in that file.
- `RUNAWAY_GUARD_CAP_USD`: raise or unset the variable where the agent runs.
- The user file: raise `spend_cap_usd` there (or set it, when no file sets a cap).
- In every case: run `status.py --session <id> --reset` in a terminal, which starts the count from $0 and keeps the session total in the record.
The user runs these in their own terminal: at the cap, the agent's own tool calls are blocked, `status.py` included, so the agent cannot lift its own cap.
**A session the guard sees for the first time.** Claude Code normally applies hook changes to sessions already running, so a new install can meet a session that has already spent more than the cap. At its first check, such a session is counted from that point: the guard sets the reset point there and says so once. A session the guard already counts is blocked at its next tool call once it is over the cap.
**Subscription plans.** The dollar figure is what the same tokens would cost at API list prices. On a subscription plan it measures usage, not the bill.
## Loop
**Fingerprint.** A call is its tool name plus its input. For shell commands, the `description` label is left out and runs of spaces count as one. Only a hash of the fingerprint is stored.
**The rule.** A call is blocked when the same fingerprint arrives for the third time (`loop_repeats`, default 3) with nothing changed in between. It stays blocked until something changes.
**What counts as a change.**
- Reads and searches change nothing: `Read`, `Grep`, `Glob`, `LS`, `NotebookRead`, `WebFetch`, `WebSearch`, `ToolSearch`. They neither end a run of repeats nor reset one.
- Any other call may change files or state: an edit, another shell command, a browser click, an MCP call. It ends every other call's run of repeats. A call's own repeats still count, so the same edit or the same command three times in a row is a loop.
- A new user prompt clears the counts: Claude Code's `prompt_id` and Codex's `turn_id` in the hook input. Older Claude Code versions send no `prompt_id`; there the typed prompt in the transcript clears them.
**Never counted** (these also end a run of repeats):
- Calls that start or message subagents: `Agent`, `Task`, `spawn_agent`, `send_message`, `followup_task`. Starting several subagents with the same prompt is a deliberate pattern.
- Calls that wait on purpose: `wait`, `wait_agent`, `sleep`, `TaskOutput`, `BashOutput`, `Monitor`, `ScheduleWakeup`, `list_agents`, and any shell command that contains `sleep` or `wait` (a poll, not a loop). The spend cap still covers a poll that never ends.
- Browser steps that repeat on purpose: an `action` of `scroll`, `key`, `press_key`, or `wait`, and tools whose name ends in `press_key`.
- Questions to the user: `AskUserQuestion`, `ExitPlanMode`, `request_user_input`.
**Per agent.** The main session and each subagent keep their own counts, so two subagents reading the same file are not a loop.
**Calibration on real data.** Over 19,625 tool calls in 400 local Claude Code transcripts, the rule would have fired twice, both in one session that ran the same browser script three and four times in a row. Replaying the 15 busiest sessions through the hook (14,663 calls, subagents included), it fired zero times. A looser rule that only an edit resets would have fired 71 times in 200 of those transcripts (9,559 calls), mostly on browser screenshots taken between clicks.
**What it misses.** Two different commands that alternate (A, B, A, B) end each other's runs, so that pattern never trips. The spend cap and the failure streak cover it.
## Failure streak
**Where results come from.** The transcript: each tool result is a success or a failure (an error result or a nonzero exit code).
- Not failures: a call denied by a permission rule, by another hook, or by the auto-mode reviewer. These neither count nor end the streak. Calls the guard itself blocked do not count either.
- Ends the streak: a success, a call the user rejected, an interrupt, a new typed prompt.
**At the limit** (`failure_streak`, default 5):
- Claude Code: the guard answers `ask`, so the user sees the reason and decides whether this call runs. The count then starts over.
- Codex: an `ask` answer makes Codex treat the hook as failed and run the call anyway, so the guard blocks the call instead, with a message that tells the agent to stop and ask the user. The count then starts over.
- Print mode (`claude -p`) has no one to answer, so a question there counts as a no. Use `--max-budget-usd` for print mode.
**Per agent**, like the loop counts.
**Lag.** The results of the newest response are read at the next call, so the question can come one call late.
**Calibration on real data.** Five failures in a row happened 5 times in 400 local transcripts. The replay of the 15 busiest sessions asked once in 14,663 calls.
## Settings
| Key | Default | Environment variable | Meaning |
|---|---|---|---|
| `spend_cap_usd` | 10 | `RUNAWAY_GUARD_CAP_USD` | Dollars per session; 0 turns the spend wire off |
| `warn_at` | 0.8 | `RUNAWAY_GUARD_WARN_AT` | Share of the cap that triggers the warning, above 0 and at most 1 |
| `loop_repeats` | 3 | `RUNAWAY_GUARD_LOOP_REPEATS` | Identical calls that count as a loop; 0 turns it off, otherwise 2 or more |
| `failure_streak` | 5 | `RUNAWAY_GUARD_FAILURE_STREAK` | Failed calls in a row before the question; 0 turns it off |
`RUNAWAY_GUARD_OFF=1` makes the hook do nothing.
**Where the values come from.** The user file is `runaway-guard.json` in the state folder, `${XDG_STATE_HOME:-~/.local/state}/runaway-guard/`. A project file is `<project>/.claude/runaway-guard.json` (or `.codex/`). An environment variable wins over the user file. A project file can make a limit stricter (a lower number that is not 0) but never looser, except for a value that neither the environment nor the user file sets. That way a repository you clone cannot switch off your cap. Invalid values are ignored. The hook reads the files on every call, so a change applies at the next call. When `install.py --cap` asks for a cap that a stricter setting overrides, the install says which setting keeps it.
## Never blocking work by mistake
- The hook always exits 0. Exit code 2 would block the tool call, so a bad flag, bad input, or an unreadable settings file lets the call through.
- The installed command ends in `|| true`. If the skill folder moves or is deleted, `python3` would exit 2 on the missing script and block every call; `|| true` turns that into a pass. Run `install.py --uninstall` before moving, updating, or removing the skill, then install again, so the guard keeps running.
- Any internal error lets the call through and adds one line to `errors.log` in the state folder: the time, the place in the code, the error type, and a short message with secrets masked.
- `install.py` sets a 10-second timeout. A PreToolUse hook that times out does not block: the harness goes on.
- Parallel calls of one session wait for each other through a lock file. A lock held longer than 2 seconds lets the call through unchecked.
- A transcript that cannot be read is skipped; the loop wire still works from the hook input alone.
- Cursor runs Claude Code hooks by default. The guard recognizes Cursor's input and answers `{}`, which makes no decision, because Cursor may read empty output as invalid JSON and block. This has not been tested in Cursor.
## Speed
Repeat calls read only the bytes added since the last call. On a 5,000-line transcript (4.6 MB), a repeat call took 54 ms at the median with Python 3.9.6 on a laptop, 22 ms of which is Python starting. The first call of a session reads the whole transcript once: 269 ms for the same file. Replaying a real 864-call session through the hook: 51 ms at the median, 57 ms at the 95th percentile. These times assume Python can cache the compiled scripts next to them, which it does by default. With `PYTHONDONTWRITEBYTECODE` set, or in a folder Python cannot write to, it compiles them on every call, and a check takes about 175 ms.
## State
One JSON file per session in the state folder: the transcripts' read offsets and sizes, the dollar total and the reset point, the estimated part and the models behind it, hashes of the responses counted, per-agent loop hashes and failure counts, the ids of calls the guard blocked, and the latest 50 events (time, trip wire, tool name, detail). The shared `counted-ids.bin` holds only hashes. Neither holds commands, prompts, or file contents. Session files not touched for 30 days are removed when a new session starts; a session's lock file stays as long as its session file is in use.
`install.py` also keeps, in `installs/`, a copy of each settings file as it was before the install, so that `--uninstall` can put it back byte for byte. That copy holds whatever the settings file held, such as other hook commands or `env` values, so it is created readable only by you, and the uninstall deletes it.
#!/usr/bin/env python3
"""Runaway guard: a PreToolUse hook for Claude Code and Codex.
Before each tool call it checks three trip wires and answers the harness:
loop the same tool call a third time with nothing changed in between: deny
failures five failed tool calls in a row: ask the user (Codex: deny, with a
reason that tells the agent to ask the user, because Codex hooks cannot ask)
spend running cost from the session transcripts: warn once at 80% of the
cap, deny every tool call at 100%
Settings: runaway-guard.json in the state folder (user) or in the project's
.claude/ or .codex/ folder, and RUNAWAY_GUARD_* environment variables. A
project file can lower a limit but never raise one set for the user.
The hook reads the hook JSON on stdin and prints a decision as JSON on stdout.
It always exits 0: exit code 2 would block the tool call, so a bug or a bad
flag in the guard must never produce it. On any internal error it allows the
call and appends one line to errors.log in the state folder.
Python 3.9+, standard library only. Nothing leaves the machine.
"""
from __future__ import annotations
import argparse
import datetime
import hashlib
import json
import os
import re
import shlex
import sys
import time
try:
import fcntl
except ImportError: # Windows has no fcntl: the guard then runs without a lock
fcntl = None
import pricing
import transcripts
STATE_VERSION = 2
CONFIG_NAME = "runaway-guard.json"
DEFAULTS = {"spend_cap_usd": 10.0, "warn_at": 0.8, "loop_repeats": 3, "failure_streak": 5}
ENV_NAMES = {"spend_cap_usd": "RUNAWAY_GUARD_CAP_USD", "warn_at": "RUNAWAY_GUARD_WARN_AT",
"loop_repeats": "RUNAWAY_GUARD_LOOP_REPEATS", "failure_streak": "RUNAWAY_GUARD_FAILURE_STREAK"}
LOCK_TIMEOUT = 2.0 # seconds to wait for a parallel call of the same session
MAX_COUNTS = 64 # loop fingerprints kept per agent
MAX_AGENTS = 100 # agents (main plus subagents) kept per session
MAX_TRIPS = 50
PRUNE_DAYS = 30 # session files older than this are removed
SHARED_IDS = "counted-ids.bin" # responses counted by any session: 8-byte id hash + 8-byte session hash
MAX_SHARED_IDS = 32768
MAX_SESSION_IDS = 4096
# Calls that change nothing, so repeats with only these in between still count.
READ_ONLY_TOOLS = {"Read", "Grep", "Glob", "LS", "NotebookRead", "WebFetch", "WebSearch", "ToolSearch"}
# Calls whose repeats are deliberate: they start or wait for subagents, wait for
# background work, or ask the user. They never count, and they end a run of repeats.
NEVER_LOOP = {"agent", "task", "spawn_agent", "followup_task", "send_message", "sendmessage", "wait",
"wait_agent", "sleep", "taskoutput", "bashoutput", "monitor", "schedulewakeup", "list_agents",
"listagents", "askuserquestion", "exitplanmode", "request_user_input",
"request_user_input_async"}
# Browser steps that repeat on purpose: scrolling, key presses, and waits.
NAV_ACTIONS = {"scroll", "key", "press_key", "wait"}
# Calls that stop work or hand in a final answer: they still pass at the spend cap.
STOP_TOOLS = {"TaskStop", "KillShell", "KillBash", "StructuredOutput"}
SHELL_TOOLS = {"Bash", "exec_command", "shell", "shell_command", "local_shell"}
# A model missing from the price table is priced like the dearest known model of its family.
FAMILIES = ("claude-opus", "claude-sonnet", "claude-haiku", "claude-fable", "gpt")
POLL_RE = re.compile(r"\b(?:sleep|wait)\b") # a command that waits on purpose is a poll, not a loop
_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------
def _now_iso() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def state_dir(env=None) -> str:
"""${XDG_STATE_HOME:-~/.local/state}/runaway-guard"""
env = os.environ if env is None else env
base = env.get("XDG_STATE_HOME") or ""
if not os.path.isabs(base):
home = env.get("HOME") or os.path.expanduser("~")
base = os.path.join(home, ".local", "state")
return os.path.join(base, "runaway-guard")
def session_stem(session_id) -> str:
"""A file name for a session id: the id itself when it is plain, else a hash."""
if _SAFE_ID_RE.match(session_id) and session_id != "runaway-guard":
return session_id
return "s-" + hashlib.sha256(session_id.encode("utf-8", "surrogatepass")).hexdigest()[:32]
def _read_json(path) -> dict:
try:
with open(path, encoding="utf-8") as fh:
value = json.load(fh)
except (OSError, ValueError):
return {}
return value if isinstance(value, dict) else {}
def _valid(key, value):
"""A setting as a number, or None when it is missing or out of range."""
if isinstance(value, bool) or value is None:
return None
try:
number = float(value)
except (TypeError, ValueError):
return None
if number != number or number in (float("inf"), float("-inf")):
return None
if key == "warn_at":
return number if 0 < number <= 1 else None
if key == "spend_cap_usd":
return number if number >= 0 else None
if number != int(number):
return None
number = int(number)
if key == "loop_repeats":
return number if number == 0 or number >= 2 else None
return number if number >= 0 else None # failure_streak
def _stricter(a, b) -> bool:
"""True when limit a is stricter than limit b (0 means off)."""
return a != 0 and (b == 0 or a < b)
def project_config_path(project_dir) -> str:
"""The project's runaway-guard.json (.claude/ first, then .codex/), or ''."""
for sub in (".claude", ".codex"):
path = os.path.join(project_dir, sub, CONFIG_NAME) if project_dir else ""
if path and os.path.isfile(path):
return path
return ""
def load_limits(env, project_dir):
"""(limits, sources) for a session whose project folder is `project_dir`."""
project = project_config_path(project_dir)
user = os.path.join(state_dir(env), CONFIG_NAME)
return merge_limits(env, _read_json(user), _read_json(project) if project else {}, user, project)
def merge_limits(env, user, project, user_path="", project_path=""):
"""Environment, then the user file; a project file can only make a limit stricter,
except where neither of the first two sets it. Returns (limits, sources), where
each source is ('environment', variable), ('user', file), ('project', file),
or ('default', '')."""
limits, sources = {}, {}
for key, default in DEFAULTS.items():
base, source = _valid(key, env.get(ENV_NAMES[key])), ("environment", ENV_NAMES[key])
if base is None:
base, source = _valid(key, user.get(key)), ("user", user_path)
local = _valid(key, project.get(key))
if base is None:
base, source = default, ("default", "")
if local is not None:
base, source = local, ("project", project_path)
elif local is not None and _stricter(local, base):
base, source = local, ("project", project_path)
limits[key], sources[key] = base, source
return limits, sources
def detect_harness(hook, flag=None) -> str:
if flag in ("claude-code", "codex"):
return flag
path = hook.get("transcript_path") or ""
if "turn_id" in hook or os.path.basename(str(path)).startswith("rollout-"):
return "codex"
return "claude-code"
# ---------------------------------------------------------------------------
# State: one JSON file per session, guarded by a lock file
# ---------------------------------------------------------------------------
def new_state(session_id, harness) -> dict:
now = _now_iso()
return {"version": STATE_VERSION, "session_id": session_id, "harness": harness, "created": now,
"created_ts": time.time(), "updated": now, "files": {}, "cost_usd": 0.0, "baseline_usd": 0.0,
"estimated_usd": 0.0, "estimated_tokens": 0, "estimated_models": [], "estimate_noted": False,
"warned_cap": None, "streaks": {}, "agents": {},
"denied_ids": [], "trips": [], "counts": {"loop": 0, "failures": 0, "spend": 0, "warning": 0},
"limits": {}}
def load_state(path):
state = _read_json(path)
return state if state.get("version") == STATE_VERSION else None
def save_state(path, state) -> None:
tmp = "%s.%d.tmp" % (path, os.getpid())
with open(tmp, "w", encoding="utf-8") as fh:
json.dump(state, fh, separators=(",", ":"))
os.replace(tmp, path)
def prune(folder, keep) -> None:
"""Remove session files not touched for PRUNE_DAYS (settings, the error log, and
the shared list of counted responses stay). A lock file stays while its session
file is fresh: a lock file's own time does not change while it is in use."""
cutoff = time.time() - PRUNE_DAYS * 86400
for e in _scandir(folder):
if (e.name in (CONFIG_NAME, "errors.log", SHARED_IDS) or e.name.startswith(keep + ".")
or e.name.startswith(SHARED_IDS + ".")):
continue
if e.name.endswith((".json", ".lock", ".tmp")):
try:
if e.name.endswith(".lock") and os.stat(e.path[:-len(".lock")] + ".json").st_mtime >= cutoff:
continue
except OSError:
pass
try:
if e.stat().st_mtime < cutoff:
os.remove(e.path)
except OSError:
pass
def _acquire(path):
"""An open, locked file descriptor; -1 when locks are not available; None on timeout."""
if fcntl is None:
return -1
fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600)
deadline = time.monotonic() + LOCK_TIMEOUT
while True:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return fd
except OSError:
if time.monotonic() >= deadline:
os.close(fd)
return None
time.sleep(0.005)
def _release(fd) -> None:
if fd is not None and fd >= 0:
try:
fcntl.flock(fd, fcntl.LOCK_UN)
finally:
os.close(fd)
def _record(state, wire, tool, detail, log=True) -> None:
"""Count one trip; `log` adds it to the list of recent trips as well."""
state["counts"][wire] = state["counts"].get(wire, 0) + 1
if log:
state["trips"] = (state["trips"] + [{"wire": wire, "at": _now_iso(), "tool": transcripts.safe_text(tool, 80),
"detail": detail}])[-MAX_TRIPS:]
# ---------------------------------------------------------------------------
# Loop wire
# ---------------------------------------------------------------------------
def fingerprint(tool, tool_input) -> str:
"""Tool name plus input. For shell tools the description label is left out and
runs of spaces in the command count as one."""
inp = tool_input
if tool in SHELL_TOOLS and isinstance(inp, dict):
inp = {k: (" ".join(v.split()) if k in ("command", "cmd") and isinstance(v, str) else v)
for k, v in inp.items() if k != "description"}
blob = json.dumps([tool, inp], sort_keys=True, ensure_ascii=False, default=str)
return hashlib.blake2b(blob.encode("utf-8", "surrogatepass"), digest_size=8).hexdigest()
def _shell_text(tool, tool_input) -> str:
if tool in SHELL_TOOLS and isinstance(tool_input, dict):
cmd = tool_input.get("command", tool_input.get("cmd"))
return cmd if isinstance(cmd, str) else ""
return ""
def loop_count(agent, tool, tool_input) -> int:
"""How many times this exact call has come with nothing changed in between
(this call included); 0 for calls that never count."""
short = tool.rsplit("__", 1)[-1].lower()
action = str(tool_input.get("action") or "").lower() if isinstance(tool_input, dict) else ""
if (short in NEVER_LOOP or short.endswith("press_key") or action in NAV_ACTIONS
or POLL_RE.search(_shell_text(tool, tool_input))):
agent["counts"] = []
return 0
fp = fingerprint(tool, tool_input)
counts = agent.get("counts") or []
n = next((c[1] for c in counts if c[0] == fp), 0) + 1
if tool in READ_ONLY_TOOLS:
agent["counts"] = ([c for c in counts if c[0] != fp] + [[fp, n]])[-MAX_COUNTS:]
else: # anything else may have changed files or state: earlier repeats no longer count
agent["counts"] = [[fp, n]]
return n
def _agent(state, key) -> dict:
agents = state["agents"]
agent = agents.pop(key, None) or {}
agents[key] = agent # most recent last
while len(agents) > MAX_AGENTS:
agents.pop(next(iter(agents)))
return agent
def agent_key(hook, harness) -> str:
if isinstance(hook.get("agent_id"), str) and hook["agent_id"]:
return hook["agent_id"]
if harness == "codex" and isinstance(hook.get("transcript_path"), str):
return "path:" + hook["transcript_path"]
return "main"
# ---------------------------------------------------------------------------
# Transcripts: the main file plus subagent files, each read from its own offset
# ---------------------------------------------------------------------------
def _scandir(folder) -> list:
try:
return list(os.scandir(folder))
except OSError:
return []
def _track(files, path, key) -> None:
if path not in files:
files[path] = {"offset": 0, "size": 0, "key": key}
def _track_files(state, hook, harness) -> dict:
"""Add this session's transcript files to state["files"]: path -> offset, size, key.
Keys match agent_key(): Claude Code 'main' and the subagent id; Codex 'path:<file>'."""
files = state.setdefault("files", {})
main = hook.get("transcript_path")
if not isinstance(main, str) or not main:
return files
if harness == "claude-code":
if os.path.basename(main).startswith("agent-"): # given a subagent's own file: use its session's
folder = os.path.dirname(main)
for _ in range(3):
if os.path.basename(folder) == "subagents":
main = os.path.dirname(folder) + ".jsonl"
break
folder = os.path.dirname(folder)
_track(files, main, "main")
# Subagents: <session>/subagents/agent-<id>.jsonl, workflow agents one level deeper.
sub = os.path.join(main[:-len(".jsonl")] if main.endswith(".jsonl") else main, "subagents")
folders = [sub] + [e.path for e in _scandir(os.path.join(sub, "workflows")) if e.is_dir()]
for folder in folders:
for e in _scandir(folder):
if e.name.startswith("agent-") and e.name.endswith(".jsonl"):
_track(files, e.path, e.name[len("agent-"):-len(".jsonl")])
else:
_track(files, main, "path:" + main)
_track_codex_children(state, files, main, hook.get("session_id"))
return files
def _codex_meta(path) -> dict:
"""The session_meta payload on the first line of a Codex rollout, or {}."""
try:
with open(path, "rb") as fh:
rec = json.loads(fh.readline())
except (OSError, ValueError):
return {}
return rec.get("payload") or {} if isinstance(rec, dict) and rec.get("type") == "session_meta" else {}
def _epoch(iso) -> float:
try:
return datetime.datetime.fromisoformat(str(iso).replace("Z", "+00:00")).timestamp()
except ValueError:
return 0.0
def _track_codex_children(state, files, main, session_id) -> None:
"""Codex writes each subagent to its own rollout, whose session_meta carries the
root session id. Look in the main rollout's day folder and in today's and
yesterday's, checking each rollout written since the session started once."""
if "codex_started" not in state:
state["codex_started"] = _epoch(_codex_meta(main).get("timestamp")) or state.get("created_ts", 0)
since = state["codex_started"] - 60
sessions = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(main))))
folders = {os.path.dirname(main)}
now = time.time()
for moment in (now, now - 86400):
for stamp in (time.gmtime(moment), time.localtime(moment)):
folders.add(os.path.join(sessions, *time.strftime("%Y %m %d", stamp).split()))
checked = state.setdefault("codex_checked", [])
seen = set(checked)
for folder in sorted(folders):
for e in _scandir(folder):
if (not e.name.startswith("rollout-") or not e.name.endswith(".jsonl") or e.path in files
or e.name in seen):
continue
try:
if e.stat().st_mtime < since:
continue
except OSError:
continue
meta = _codex_meta(e.path)
if not meta.get("id"): # still being written: look again at the next call
continue
seen.add(e.name)
checked.append(e.name)
if meta.get("session_id") == session_id and meta.get("id") != session_id:
_track(files, e.path, "path:" + e.path)
del checked[:-2000]
_TURN_RE = re.compile(rb'"type"\s*:\s*"turn_context"')
def _codex_model_in(path, start, end) -> str:
"""The model named by the last Codex turn_context line between two offsets, or ''.
Each read starts a fresh reader, which knows the model only from turn_context
lines inside the new bytes, so the guard carries it from read to read."""
if end <= start:
return ""
try:
with open(path, "rb") as fh:
fh.seek(start)
data = fh.read(end - start)
except OSError:
return ""
model = ""
if _TURN_RE.search(data):
for line in data.split(b"\n"):
if _TURN_RE.search(line):
try:
rec = json.loads(line)
except ValueError:
continue
payload = rec.get("payload") if isinstance(rec, dict) else None
if isinstance(payload, dict) and isinstance(payload.get("model"), str) and payload["model"]:
model = payload["model"]
return model
def estimate_usd(usage, model, harness) -> float:
"""The cost of usage from a model missing from the price table: the most it would
cost on any known model of the same family (Codex models count as GPT), or on any
known model when the family is unknown. Too high rather than $0, so a cap still holds."""
name = (model or "").lower()
family = next((f for f in FAMILIES if f in name), "gpt" if harness == "codex" else "")
rows = [k for k in pricing.PRICES if family and k.startswith(family)] or list(pricing.PRICES)
return max(pricing.cost_usd(usage, k) or 0.0 for k in rows)
def _add_cost(state, usage, model, harness) -> None:
cost = pricing.cost_usd(usage, model)
if cost is None:
cost = estimate_usd(usage, model, harness)
state["estimated_usd"] += cost
state["estimated_tokens"] += usage.total_input() + usage.output
name = transcripts.safe_text(model or "(no model id)", 60)
if name not in state["estimated_models"]:
state["estimated_models"] = (state["estimated_models"] + [name])[-10:]
state["cost_usd"] += cost
def _digest(text) -> bytes:
return hashlib.blake2b(text.encode("utf-8", "surrogatepass"), digest_size=8).digest()
def _shared_ids(folder) -> dict:
"""Responses any session has counted: id hash -> the hash of the session that counted it."""
try:
with open(os.path.join(folder, SHARED_IDS), "rb") as fh:
data = fh.read()
except OSError:
return {}
if len(data) % 16: # a cut-off write: start the list over rather than misread it
return {}
return {data[i:i + 8]: data[i + 8:i + 16] for i in range(0, len(data), 16)}
def _share_ids(folder, session, hashes) -> None:
"""Add this session's newly counted responses to the shared list, under its own lock."""
lock = _acquire(os.path.join(folder, SHARED_IDS + ".lock"))
if lock is None:
return
try:
path = os.path.join(folder, SHARED_IDS)
size = os.path.getsize(path) if os.path.exists(path) else 0
with open(path, "r+b" if size else "wb") as fh:
if size % 16:
fh.truncate(0)
fh.seek(0, os.SEEK_END)
fh.write(b"".join(h + session for h in hashes))
if os.path.getsize(path) > 2 * MAX_SHARED_IDS * 16:
with open(path, "rb") as fh:
fh.seek(-MAX_SHARED_IDS * 16, os.SEEK_END)
keep = fh.read()
with open(path + ".tmp", "wb") as fh:
fh.write(keep)
os.replace(path + ".tmp", path)
finally:
_release(lock)
def _read_transcripts(state, hook, harness, folder) -> None:
"""Read what each transcript gained since the last call and fold it into the state.
Each response counts once: a file read again from the start, or a resumed session
whose file begins with a copy of an earlier session's records, adds nothing twice."""
files = _track_files(state, hook, harness)
counted = state.get("counted") or ""
seen = {counted[i:i + 16] for i in range(0, len(counted), 16)}
me, shared, fresh = _digest(state["session_id"]), None, []
streaks = state.setdefault("streaks", {})
denied = set(state["denied_ids"])
no_prompt_id = not (hook.get("prompt_id") or hook.get("turn_id"))
hook_model = hook.get("model") if isinstance(hook.get("model"), str) else ""
for path, info in list(files.items()):
try:
size = os.path.getsize(path)
except OSError:
continue
if size == info["size"]:
continue
if size < info["offset"]: # the file was replaced: read it again from the start
info["offset"] = 0
start = info["offset"]
try:
events, offset = transcripts.read_new_events(harness, path, start)
except Exception as exc: # one unreadable file must not stop the others
_log_error(exc)
continue
info["offset"], info["size"] = offset, size
key = info["key"]
known_model = info.get("model") or hook_model
for ev in events:
if ev.kind == "assistant" and ev.usage is not None: # one event per API response
if ev.id:
h = _digest("%s:%s" % (ev.kind, ev.id))
if h.hex() in seen:
continue
if start == 0: # copies sit at the start of a file
shared = _shared_ids(folder) if shared is None else shared
if shared.get(h, me) != me:
continue
seen.add(h.hex())
fresh.append(h)
_add_cost(state, ev.usage, ev.model or known_model, harness)
elif ev.kind == "tool" and ev.tool is not None:
_count_result(streaks, key, ev.tool, denied)
elif ev.kind == "interrupt":
streaks[key] = 0
elif ev.kind == "user" and not ev.injected: # the user typed: they are in the loop
streaks[key] = 0
if no_prompt_id and key in state["agents"]:
state["agents"][key]["counts"] = []
if harness == "codex":
info["model"] = _codex_model_in(path, start, offset) or info.get("model", "")
if fresh:
state["counted"] = (counted + "".join(h.hex() for h in fresh))[-MAX_SESSION_IDS * 16:]
try:
_share_ids(folder, me, fresh)
except OSError as exc:
_log_error(exc)
def _count_result(streaks, key, call, denied) -> None:
if not call.has_result or call.id in denied:
return
if call.denied == "user-rejected" or call.interrupted:
streaks[key] = 0
elif call.denied: # a rule or another hook said no: not a failure of the call itself
return
elif call.is_error:
streaks[key] = streaks.get(key, 0) + 1
else:
streaks[key] = 0
# ---------------------------------------------------------------------------
# Decisions
# ---------------------------------------------------------------------------
def _decision(harness, kind, reason, user_message="") -> dict:
out = {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": kind,
"permissionDecisionReason": reason}}
if harness == "claude-code" and user_message:
out["systemMessage"] = user_message
return out
def _ordinal(n) -> str:
return "%d%s" % (n, "th" if 10 <= n % 100 <= 20 else {1: "st", 2: "nd", 3: "rd"}.get(n % 10, "th"))
def run_hook(hook, harness=None, env=None):
"""Decide one PreToolUse call. Returns the JSON object to print, or None to
let the normal permission flow run."""
env = os.environ if env is None else env
if not isinstance(hook, dict) or hook.get("hook_event_name") != "PreToolUse":
return None
if "cursor_version" in hook: # Cursor runs Claude Code hooks too; its transcripts lack usage and results
return {} # an empty answer could read as invalid JSON there, which blocks
if str(env.get("RUNAWAY_GUARD_OFF", "")).strip().lower() in ("1", "true", "yes", "on"):
return None
session_id = hook.get("session_id")
if not isinstance(session_id, str) or not session_id:
return None
harness = detect_harness(hook, harness)
limits, sources = load_limits(env, env.get("CLAUDE_PROJECT_DIR") or hook.get("cwd") or "")
folder = state_dir(env)
os.makedirs(folder, mode=0o700, exist_ok=True)
stem = session_stem(session_id)
lock = _acquire(os.path.join(folder, stem + ".lock"))
if lock is None: # another call of this session holds the state too long: allow
return None
try:
path = os.path.join(folder, stem + ".json")
first_check = not os.path.exists(path)
state = load_state(path)
if state is None: # a new session (or an unreadable state file): start fresh
state = new_state(session_id, harness)
prune(folder, stem)
state["limits"] = limits
out = _decide(state, hook, harness, limits, sources["spend_cap_usd"], env, first_check)
state["updated"] = _now_iso()
save_state(path, state)
return out
finally:
_release(lock)
def _deny(state, hook, harness, reason, user_message):
call_id = hook.get("tool_use_id")
if isinstance(call_id, str) and call_id: # its error result is the guard's doing, not a failure
state["denied_ids"] = (state["denied_ids"] + [call_id])[-200:]
return _decision(harness, "deny", reason, user_message)
def spent(state) -> float:
"""Dollars counted since the session started or was last reset."""
return max(0.0, state["cost_usd"] - state["baseline_usd"])
def _spend_help(state, env, source) -> str:
"""How to go on after a spend stop, for wherever the cap in force comes from."""
status = os.path.join(os.path.dirname(os.path.abspath(__file__)), "status.py")
kind, where = source
if kind == "environment":
change = "raise or unset %s where the agent runs" % where
elif kind == "project":
change = "raise or remove spend_cap_usd in %s" % transcripts.safe_text(where, 300)
elif kind == "user":
change = "raise spend_cap_usd in %s" % transcripts.safe_text(where, 300)
else:
change = "set a higher spend_cap_usd in %s" % transcripts.safe_text(
os.path.join(state_dir(env), CONFIG_NAME), 300)
return ("%s, or start the count over by running this in a terminal: python3 %s --session %s --reset" % (
change, transcripts.safe_text(shlex.quote(status), 300),
transcripts.safe_text(shlex.quote(state["session_id"]), 140)))
def _decide(state, hook, harness, limits, cap_source, env, first_check=False):
tool = hook.get("tool_name") if isinstance(hook.get("tool_name"), str) else ""
name = transcripts.safe_text(tool, 80)
key = agent_key(hook, harness)
agent = _agent(state, key)
_read_transcripts(state, hook, harness, state_dir(env))
prompt = hook.get("prompt_id") or hook.get("turn_id") or ""
if prompt and agent.get("prompt") != prompt: # a new user prompt: earlier repeats no longer count
agent["prompt"], agent["counts"] = prompt, []
n = loop_count(agent, tool, hook.get("tool_input"))
cap, used = limits["spend_cap_usd"], spent(state)
notes = []
if first_check and cap and used >= cap: # spent before the guard first saw it: count from here
state["baseline_usd"] = state["cost_usd"]
_record(state, "baseline", tool, "$%.2f spent before the first check" % used)
notes.append("Runaway guard started counting this session at $%.2f, which is already over its $%.2f "
"cap, so it counts from $0.00 from here." % (used, cap))
used = 0.0
if cap and used >= cap and tool not in STOP_TOOLS:
first = not state["trips"] or state["trips"][-1]["wire"] != "spend" # list one entry per stop
_record(state, "spend", tool, "$%.2f of the $%.2f cap" % (used, cap), log=first)
help_text = _spend_help(state, env, cap_source)
estimated = _estimate_note(state)
out = _deny(
state, hook, harness,
"Runaway guard: this session has spent about $%.2f, over its $%.2f cap, so tool calls are "
"blocked.%s Stop and tell the user. To go on, the user can %s" % (used, cap, estimated, help_text),
"Runaway guard: spend cap reached ($%.2f of $%.2f); tool calls are blocked.%s To go on, %s"
% (used, cap, estimated, help_text))
if harness == "claude-code": # end the turn, so each block does not start another response
out["continue"], out["stopReason"] = False, out["systemMessage"]
return out
if limits["loop_repeats"] and n >= limits["loop_repeats"]:
_record(state, "loop", tool, "%s identical call" % _ordinal(n))
return _deny(
state, hook, harness,
"Runaway guard blocked this %s call: it is the %s identical call with nothing changed in "
"between, so it would most likely give the same result. Change the approach, or stop and ask "
"the user." % (name, _ordinal(n)),
"Runaway guard blocked a repeated %s call (the same call %d times with nothing changed in "
"between)." % (name, n))
streak = state["streaks"].get(key, 0)
if limits["failure_streak"] and streak >= limits["failure_streak"]:
state["streaks"][key] = 0 # ask once; the count starts over
_record(state, "failures", tool, "%d failed calls in a row" % streak)
if harness == "claude-code":
return _decision(
harness, "ask",
"Runaway guard: the last %d tool calls failed in a row. Check that the agent is on track "
"before you allow this %s call." % (streak, name))
return _deny( # Codex hooks cannot ask, so the agent is told to ask
state, hook, harness,
"Runaway guard: the last %d tool calls failed in a row. Stop and ask the user how to go on "
"before you try again." % streak, "")
if cap and limits["warn_at"] * cap <= used < cap and state.get("warned_cap") != cap:
state["warned_cap"] = cap # warn once per cap
_record(state, "warning", tool, "$%.2f of the $%.2f cap" % (used, cap))
notes.append("Runaway guard: this session has spent about $%.2f of its $%.2f cap (%d%%). At the cap, "
"tool calls are blocked." % (used, cap, 100 * used / cap))
if state["estimated_models"] and not state["estimate_noted"]:
state["estimate_noted"] = True # name the first such model once per session
notes.append("Runaway guard: no price is known for %s, so its tokens are priced like the most "
"expensive model of the same family. Update the skill to get its list price."
% state["estimated_models"][0])
if notes and harness == "claude-code": # Codex hooks have no way to show a message
return {"systemMessage": " ".join(notes)}
return None
def _estimate_note(state) -> str:
if not state["estimated_usd"]:
return ""
return (" About $%.2f of it is an estimate for %s, which has no known price." % (
state["estimated_usd"], ", ".join(state["estimated_models"])))
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def _log_error(exc) -> None:
"""One line per internal error in <state folder>/errors.log; never raises."""
try:
import traceback
folder = state_dir()
os.makedirs(folder, mode=0o700, exist_ok=True)
path = os.path.join(folder, "errors.log")
if os.path.exists(path) and os.path.getsize(path) > 256 * 1024:
open(path, "w").close()
frames = traceback.extract_tb(exc.__traceback__) if exc.__traceback__ else []
where = "%s:%d" % (os.path.basename(frames[-1].filename), frames[-1].lineno) if frames else "?"
with open(path, "a", encoding="utf-8") as fh:
fh.write("%s %s %s: %s\n" % (_now_iso(), where, type(exc).__name__,
transcripts.safe_text(str(exc), 200)))
except Exception:
pass
def _parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description="Runaway guard hook: reads one PreToolUse hook JSON object on stdin and prints a "
"decision as JSON. Always exits 0, because exit code 2 would block the tool call. "
"install.py writes the settings entry that runs it.")
p.add_argument("--harness", choices=("claude-code", "codex"),
help="which harness calls the hook (default: read from the hook input)")
return p
def main(argv=None) -> int:
try:
# Extra words are ignored: run without a shell, the installed "|| true" arrives as two.
args, _extra = _parser().parse_known_args(argv)
except SystemExit as exc:
if exc.code in (0, None):
raise # --help
return 0 # a bad flag in the settings entry must not block tool calls
try:
raw = sys.stdin.read()
hook = json.loads(raw) if raw.strip() else None
out = run_hook(hook, args.harness)
if out is not None:
sys.stdout.write(json.dumps(out) + "\n")
sys.stdout.flush()
except Exception as exc: # a guard bug never blocks work
_log_error(exc)
return 0
if __name__ == "__main__":
sys.exit(main())
#!/usr/bin/env python3
"""Install or remove the runaway guard hook for Claude Code or Codex.
Without --write it only prints the planned change as a diff. With --write it
merges one PreToolUse entry into the settings file (the hooks already there
stay as they are) and, when --cap is given, sets spend_cap_usd in the guard's
own settings file. --uninstall removes only the guard's own entries; with no
--harness, --scope, or --settings it checks all four places the guard can be.
Where it writes:
Claude Code, user scope: ~/.claude/settings.json (or $CLAUDE_CONFIG_DIR/settings.json)
Claude Code, project scope: <project>/.claude/settings.local.json (not shared with the team)
Codex, user scope: ~/.codex/hooks.json (or $CODEX_HOME/hooks.json)
Codex, project scope: <project>/.codex/hooks.json
The cap, user scope: ${XDG_STATE_HOME:-~/.local/state}/runaway-guard/runaway-guard.json
The cap, project scope: <project>/.claude/ or <project>/.codex/runaway-guard.json
A copy of each settings file as it was before the install, so that
--uninstall can put it back exactly: <state folder>/installs/
Exit codes: 0 done (or dry run shown), 2 bad arguments, a settings file it
cannot merge safely, or a file it cannot write (then nothing is changed).
Python 3.9+, standard library only. Nothing leaves the machine.
"""
from __future__ import annotations
import argparse
import difflib
import hashlib
import json
import os
import re
import shlex
import sys
import guard
HERE = os.path.dirname(os.path.abspath(__file__))
TIMEOUT = 10 # seconds; a hook that times out lets the call through
OURS_RE = re.compile(r"runaway-guard[^/\\'\"]*[/\\]scripts[/\\]guard\.py")
HARNESS_NAMES = {"claude-code": "Claude Code", "codex": "Codex"}
safe = guard.transcripts.safe_text
class Refuse(Exception):
"""A file the installer will not touch, with the reason."""
class WriteFailed(Exception):
"""A file that could not be written; earlier writes of the same run were undone."""
# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------
def _home(env) -> str:
return env.get("HOME") or os.path.expanduser("~")
def settings_path(harness, scope, project, env) -> str:
if harness == "claude-code":
if scope == "user":
return os.path.join(env.get("CLAUDE_CONFIG_DIR") or os.path.join(_home(env), ".claude"),
"settings.json")
return os.path.join(project, ".claude", "settings.local.json")
if scope == "user":
return os.path.join(env.get("CODEX_HOME") or os.path.join(_home(env), ".codex"), "hooks.json")
return os.path.join(project, ".codex", "hooks.json")
def config_path(harness, scope, project, env) -> str:
if scope == "user":
return os.path.join(guard.state_dir(env), guard.CONFIG_NAME)
return os.path.join(project, ".claude" if harness == "claude-code" else ".codex", guard.CONFIG_NAME)
def hook_command(harness) -> str:
"""The command the harness runs: python3 plus the absolute path of guard.py.
'|| true' keeps a moved or deleted skill folder from blocking every tool call:
python3 exits 2 on a missing script, and exit 2 blocks the call. guard.py
itself always exits 0, so its decision still passes through."""
command = "python3 " + shlex.quote(os.path.join(HERE, "guard.py"))
return command + (" --harness codex" if harness == "codex" else "") + " || true"
def status_command() -> str:
return "python3 " + shlex.quote(os.path.join(HERE, "status.py"))
def shown(path, env) -> str:
"""A path for the report: the home folder as ~, made safe for markdown."""
home = _home(env).rstrip(os.sep)
if home and (path == home or path.startswith(home + os.sep)):
path = "~" + path[len(home):]
return safe(path, 300)
# ---------------------------------------------------------------------------
# Reading and merging
# ---------------------------------------------------------------------------
def _load(path, env, check_hooks=True):
"""(text, data) of a JSON settings file; ('', {}) when it does not exist."""
if not os.path.exists(path):
return "", {}
try:
with open(path, encoding="utf-8") as fh:
text = fh.read()
except (OSError, UnicodeDecodeError) as exc:
raise Refuse("cannot read %s (%s)" % (shown(path, env), type(exc).__name__))
if not text.strip():
return text, {}
try:
data = json.loads(text)
except ValueError as exc:
raise Refuse("%s is not valid JSON (line %d: %s); fix it, then run this again"
% (shown(path, env), exc.lineno, exc.msg))
if not isinstance(data, dict):
raise Refuse("%s does not hold a JSON object" % shown(path, env))
if check_hooks:
hooks = data.get("hooks")
if hooks is not None and not isinstance(hooks, dict):
raise Refuse("in %s, \"hooks\" is not an object" % shown(path, env))
if isinstance(hooks, dict) and "PreToolUse" in hooks and not isinstance(hooks["PreToolUse"], list):
raise Refuse("in %s, \"hooks.PreToolUse\" is not a list" % shown(path, env))
return text, data
def _is_ours(handler) -> bool:
return isinstance(handler, dict) and bool(OURS_RE.search(str(handler.get("command") or "")))
def without_ours(data):
"""A copy of the settings with the guard's handlers removed, and how many were removed.
A matcher group left empty by the removal goes too; so do the PreToolUse list and
the hooks object when the guard was all they held."""
new = json.loads(json.dumps(data))
hooks = new.get("hooks")
groups = hooks.get("PreToolUse") if isinstance(hooks, dict) else None
if not isinstance(groups, list):
return new, 0
kept, removed = [], 0
for group in groups:
handlers = group.get("hooks") if isinstance(group, dict) else None
if isinstance(handlers, list):
left = [h for h in handlers if not _is_ours(h)]
removed += len(handlers) - len(left)
if len(left) < len(handlers):
if not left:
continue
group["hooks"] = left
kept.append(group)
if removed:
if kept:
hooks["PreToolUse"] = kept
else:
del hooks["PreToolUse"]
if not hooks:
del new["hooks"]
return new, removed
def other_hooks(data) -> int:
"""PreToolUse handlers that are not the guard's."""
hooks = data.get("hooks")
groups = hooks.get("PreToolUse") if isinstance(hooks, dict) else None
return sum(1 for g in groups or [] if isinstance(g, dict)
for h in (g.get("hooks") if isinstance(g.get("hooks"), list) else []) if not _is_ours(h))
def with_ours(data, command):
new, _ = without_ours(data)
entry = {"hooks": [{"type": "command", "command": command, "timeout": TIMEOUT}]}
new.setdefault("hooks", {}).setdefault("PreToolUse", []).append(entry)
return new
def _indent(text):
"""The indentation the file already uses: a tab, a number of spaces, or 2."""
m = re.search(r"^\{[ \t]*\r?\n([ \t]+)\S", text or "")
if m:
return "\t" if m.group(1).startswith("\t") else len(m.group(1))
return 2
def dump(data, like) -> str:
return json.dumps(data, indent=_indent(like), ensure_ascii=False) + "\n"
def diff(old, new, path, env) -> str:
"""A unified diff for the report, with secrets masked and backticks replaced,
so a token in a settings file never reaches the screen."""
name = shown(path, env)
lines = difflib.unified_diff(old.splitlines(True), new.splitlines(True),
"%s (now)" % name, "%s (after)" % name)
return "".join(guard.transcripts.redact(line.rstrip("\n")).replace("`", "'") + "\n" for line in lines)
# ---------------------------------------------------------------------------
# Writing: one step per file, undone as a whole when any file fails
# ---------------------------------------------------------------------------
def _read_or_none(path):
try:
with open(path, encoding="utf-8") as fh:
return fh.read()
except FileNotFoundError:
return None
def write(path, text) -> None:
"""Replace the file in one step (None removes it). A symlinked file is written
through its link. The temp file is created with the final mode, so a private
file is never readable by others, even for a moment."""
real = os.path.realpath(path)
if text is None:
os.remove(real)
return
os.makedirs(os.path.dirname(real), exist_ok=True)
mode = os.stat(real).st_mode & 0o777 if os.path.exists(real) else 0o600
tmp = real + ".runaway-guard.tmp"
try:
os.remove(tmp) # a leftover from a crash could have looser permissions
except FileNotFoundError:
pass
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode)
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(text)
os.chmod(tmp, mode) # the umask may have removed bits the original file had
os.replace(tmp, real)
except OSError:
try:
os.remove(tmp)
except OSError:
pass
raise
def apply(changes) -> None:
"""Write each (path, text) in order; on a failure undo the earlier ones."""
done = []
for path, text in changes:
try:
before = _read_or_none(os.path.realpath(path))
write(path, text)
except OSError as exc:
for undo_path, undo_text in reversed(done):
try:
write(undo_path, undo_text)
except OSError:
pass
raise WriteFailed("cannot write %s (%s)" % (path, exc.strerror or exc))
done.append((path, before))
# A copy of each settings file as it was before the install, so that uninstall can
# put it back byte for byte when nothing else changed it in between.
def _backup_path(target, env) -> str:
key = hashlib.sha256(os.path.realpath(target).encode("utf-8", "surrogatepass")).hexdigest()[:16]
return os.path.join(guard.state_dir(env), "installs", key + ".json")
def _backup(target, env) -> dict:
record = guard._read_json(_backup_path(target, env))
return record if isinstance(record.get("written"), str) else {}
def remember_install(target, old_text, new_text, data, env) -> None:
"""Keep the file as it was before the guard was added (None when it did not exist)."""
record = _backup(target, env)
if record and record["written"] == old_text:
original = record.get("original") # a re-install: the first original stays
elif without_ours(data)[1] == 0:
original = old_text if os.path.exists(target) else None
else:
_forget_install(target, env) # installed before this record existed
return
write(_backup_path(target, env), json.dumps({"path": os.path.realpath(target), "original": original,
"written": new_text}))
def _forget_install(target, env) -> None:
try:
os.remove(_backup_path(target, env))
except OSError:
pass
# ---------------------------------------------------------------------------
# Report
# ---------------------------------------------------------------------------
def _hooks(n) -> str:
return "%d PreToolUse hook%s" % (n, "" if n == 1 else "s")
def _money(value) -> str:
return "$%.2f" % value
def headline(harness, limits) -> str:
parts = []
if limits["loop_repeats"]:
parts.append("the same call repeats %d times" % limits["loop_repeats"])
if limits["failure_streak"]:
parts.append("%d calls fail in a row" % limits["failure_streak"])
if limits["spend_cap_usd"]:
parts.append("spend passes %s" % _money(limits["spend_cap_usd"]))
name = HARNESS_NAMES[harness]
if not parts:
return "Runaway guard will run in %s with every trip wire turned off, so it will stop nothing." % name
if len(parts) == 1:
listed = parts[0]
else:
listed = ", ".join(parts[:-1]) + (", or " if len(parts) > 2 else " or ") + parts[-1]
return "Runaway guard will stop %s when %s." % (name, listed)
def _cap_source(source, env) -> str:
kind, where = source
if kind == "environment":
return "%s is set in this shell" % where
if kind == "project":
return "the project file %s sets a lower cap, and a project file can lower a limit but never raise one" \
% shown(where, env)
return "your user file %s sets a lower cap, and a project file can lower a limit but never raise one" \
% shown(where, env)
def notes_for(harness, scope, settings, limits, sources, cap, env) -> list:
notes = []
if cap is not None and limits["spend_cap_usd"] != cap:
notes.append("The cap stays %s because %s." % (_money(limits["spend_cap_usd"]),
_cap_source(sources["spend_cap_usd"], env)))
if settings.get("disableAllHooks") is True:
notes.append("This settings file sets disableAllHooks, which turns every hook off: the guard will "
"not run until that setting is removed.")
if harness == "claude-code":
notes.append("Claude Code normally applies hook changes to sessions that are already running. A "
"session the guard already counts is blocked at its next tool call once it is over the "
"cap; start its count over with %s --session <id> --reset. A session the guard checks "
"for the first time after it has spent more than the cap is counted from that point, "
"and the guard says so once." % safe(status_command(), 400))
if scope == "user":
notes.append("Cursor also runs the hooks in Claude Code settings files. The guard is built to "
"recognize Cursor and let its calls through; that has not been tested in Cursor.")
else:
notes.append("Codex runs a new or changed hook only after you trust it: start Codex, type /hooks, "
"and trust the runaway-guard entry.")
if scope == "project":
notes.append("Codex runs project hooks only in projects you have marked as trusted.")
if limits["failure_streak"] or limits["spend_cap_usd"]:
notes.append("Codex hooks cannot ask you a question or show a warning, so after a run of failed "
"calls the guard blocks the next call and tells the agent to ask you, and the "
"early spend warning shows only in status.py.")
for key, name in sorted(guard.ENV_NAMES.items()):
if env.get(name) and not (key == "spend_cap_usd" and cap is not None and limits[key] != cap):
notes.append("%s is set in this shell. Where the agent runs with it set, it wins over the "
"settings file." % name)
notes.append("Run --uninstall before you move, update, or remove this skill, then install again from "
"the new place: the hook entry points at this folder.")
if limits["spend_cap_usd"]:
notes.append("Dollar amounts are estimates from token counts at API list prices. On a subscription "
"plan they measure usage, not your bill.")
return notes
def emit(text, out_path) -> int:
"""Print the report, or write it to --out and say where."""
if not out_path:
sys.stdout.write(text)
return 0
try:
with open(out_path, "w", encoding="utf-8") as fh:
fh.write(text)
except OSError as exc:
print("error: cannot write the report to %s: %s" % (safe(out_path, 300), exc.strerror or exc),
file=sys.stderr)
return 2
print("Report written to %s" % safe(out_path, 300))
return 0
def _changed_nothing(reason) -> int:
print("Runaway guard changed nothing: %s." % safe(str(reason), 600), file=sys.stderr)
return 2
# ---------------------------------------------------------------------------
# Install
# ---------------------------------------------------------------------------
def install(args, env, project) -> int:
harness, scope = args.harness or "claude-code", args.scope or "user"
target = os.path.abspath(args.settings) if args.settings else settings_path(harness, scope, project, env)
cap_file = config_path(harness, scope, project, env)
command = hook_command(harness)
try:
old_text, data = _load(target, env)
cap_old, cap_data = _load(cap_file, env, check_hooks=False) if args.cap is not None else ("", {})
except Refuse as exc:
return _changed_nothing(exc)
new_data = with_ours(data, command)
new_text = old_text if old_text and new_data == data else dump(new_data, old_text)
cap_new = cap_old
if args.cap is not None and cap_data.get("spend_cap_usd") != args.cap:
cap_new = dump(dict(cap_data, spend_cap_usd=args.cap), cap_old)
# The limits the guard will use after this change.
user_path = os.path.join(guard.state_dir(env), guard.CONFIG_NAME)
user_cfg = guard._read_json(user_path)
project_file = guard.project_config_path(project)
project_cfg = guard._read_json(project_file) if project_file else {}
if args.cap is not None:
if scope == "user":
user_cfg = dict(user_cfg, spend_cap_usd=args.cap)
else:
project_cfg = dict(project_cfg, spend_cap_usd=args.cap)
project_file = project_file or cap_file
limits, sources = guard.merge_limits(env, user_cfg, project_cfg, user_path, project_file)
settings_changed, cap_changed = new_text != old_text, cap_new != cap_old
written = False
if args.write and (settings_changed or cap_changed):
changes = ([(cap_file, cap_new)] if cap_changed else []) + ([(target, new_text)] if settings_changed else [])
try:
apply(changes)
except WriteFailed as exc:
return _changed_nothing(exc)
if settings_changed:
try:
remember_install(target, old_text, new_text, data, env)
except OSError as exc:
guard._log_error(exc) # uninstall then removes the entry without restoring the old text
written = True
notes = notes_for(harness, scope, new_data, limits, sources, args.cap, env)
if args.json:
return emit(json.dumps({
"headline": headline(harness, limits), "action": "install", "harness": harness, "scope": scope,
"settings_path": target, "config_path": cap_file, "command": command,
"cap_usd": limits["spend_cap_usd"], "limits": limits, "changed": settings_changed or cap_changed,
"written": written, "other_hooks": other_hooks(data), "notes": notes,
}, indent=2) + "\n", args.out)
out = ["**%s**" % headline(harness, limits), ""]
if not (settings_changed or cap_changed):
out.append("Nothing to change: %s already matches." % shown(target, env))
elif written:
out.append("Done: the change below is written.")
else:
out.append("Dry run: nothing is written yet. Run the same command with --write to apply this change.")
if settings_changed:
others = other_hooks(data)
what = "adds one PreToolUse hook entry" + (
" and leaves the %s already there unchanged" % _hooks(others) if others else "")
out += ["", "Settings file %s: %s." % (shown(target, env), what), "", "```diff",
diff(old_text, new_text, target, env).rstrip("\n"), "```"]
if cap_changed:
out += ["", "Guard settings file %s: spend cap %s." % (shown(cap_file, env), _money(args.cap)), "",
"```diff", diff(cap_old, cap_new, cap_file, env).rstrip("\n"), "```"]
out += ["", "The hook command: `%s`" % safe(command, 400), "", "Notes:"] + ["- " + n for n in notes]
return emit("\n".join(out) + "\n", args.out)
# ---------------------------------------------------------------------------
# Uninstall
# ---------------------------------------------------------------------------
def _uninstall_plan(harness, scope, target, env) -> dict:
old_text, data = _load(target, env)
new_data, removed = without_ours(data)
plan = {"harness": harness, "scope": scope, "settings_path": target, "removed": removed,
"other_hooks": other_hooks(data), "old_text": old_text, "new_text": old_text,
"restored_exactly": False, "reformatted": False}
if removed:
record = _backup(target, env)
if record and record["written"] == old_text: # nothing else changed it: put the old file back
plan["new_text"], plan["restored_exactly"] = record.get("original"), True
else:
plan["new_text"] = dump(new_data, old_text)
plan["reformatted"] = dump(data, old_text) != old_text
return plan
def uninstall(args, env, project) -> int:
if args.settings:
places = [(args.harness or "claude-code", args.scope or "user", os.path.abspath(args.settings))]
else:
places = [(h, sc, settings_path(h, sc, project, env))
for h in ([args.harness] if args.harness else ["claude-code", "codex"])
for sc in ([args.scope] if args.scope else ["user", "project"])]
try:
plans = [_uninstall_plan(h, sc, path, env) for h, sc, path in places]
except Refuse as exc:
return _changed_nothing(exc)
found = [p for p in plans if p["removed"]]
written = False
if args.write and found:
try:
apply([(p["settings_path"], p["new_text"]) for p in found])
except WriteFailed as exc:
return _changed_nothing(exc)
for p in found:
_forget_install(p["settings_path"], env)
written = True
removed = sum(p["removed"] for p in found)
if not found:
title = "Runaway guard is not installed in any of the %d settings files checked." % len(plans)
elif len(found) == 1:
p = found[0]
title = "Runaway guard will be removed from %s%s." % (
shown(p["settings_path"], env),
", leaving the other %s there unchanged" % _hooks(p["other_hooks"]) if p["other_hooks"] else "")
else:
title = "Runaway guard will be removed from %d settings files: %s." % (
len(found), " and ".join(shown(p["settings_path"], env) for p in found))
notes = []
for p in found:
if p["new_text"] is None:
notes.append("%s held only the guard, so it is removed as before the install."
% shown(p["settings_path"], env))
elif p["reformatted"]:
notes.append("%s changed after the install, so only the guard's entry is removed, and the file is "
"written again as formatted JSON." % shown(p["settings_path"], env))
if found:
notes.append("The guard's own settings and session files stay in %s; delete that folder to remove "
"them too." % shown(guard.state_dir(env), env))
if args.json:
return emit(json.dumps({
"headline": title, "action": "uninstall", "hook_entries_removed": removed,
"changed": bool(found), "written": written, "notes": notes,
"targets": [{k: p[k] for k in ("harness", "scope", "settings_path", "removed", "other_hooks",
"restored_exactly")} for p in plans],
}, indent=2) + "\n", args.out)
out = ["**%s**" % title, ""]
if not found:
out += ["Checked:"] + ["- %s" % shown(p["settings_path"], env) for p in plans]
else:
out.append("Done: the change below is written." if written else
"Dry run: nothing is written yet. Run the same command with --write to apply this change.")
for p in found:
what = "removes %d runaway-guard hook entr%s" % (p["removed"], "y" if p["removed"] == 1 else "ies")
out += ["", "Settings file %s: %s." % (shown(p["settings_path"], env), what), "", "```diff",
diff(p["old_text"], p["new_text"] or "", p["settings_path"], env).rstrip("\n"), "```"]
if notes:
out += ["", "Notes:"] + ["- " + n for n in notes]
return emit("\n".join(out) + "\n", args.out)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def _cap(text):
try:
value = float(text)
except ValueError:
raise argparse.ArgumentTypeError("the cap is a dollar amount, such as 10 or 2.5")
if not value >= 0 or value == float("inf"):
raise argparse.ArgumentTypeError("the cap must be 0 (off) or more")
return int(value) if value == int(value) else value
def _parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
description="Install or remove the runaway guard hook. Prints the planned change as a diff; "
"--write applies it.")
p.add_argument("--harness", choices=("claude-code", "codex"),
help="which agent gets the hook (install default: claude-code; uninstall default: both)")
p.add_argument("--scope", choices=("user", "project"),
help="user: every session of this user; project: only sessions in --project "
"(install default: user; uninstall default: both)")
p.add_argument("--project", default=".", help="the project folder (default: the current folder)")
p.add_argument("--cap", type=_cap, metavar="USD",
help="spend cap in dollars per session; 0 turns the spend wire off "
"(default: keep the cap already set, else $10)")
p.add_argument("--settings", metavar="PATH", help="change this settings file instead of the usual one")
p.add_argument("--uninstall", action="store_true", help="remove the guard's hook entries")
p.add_argument("--write", action="store_true", help="apply the change (without it, nothing is written)")
p.add_argument("--json", action="store_true", help="print machine-readable JSON instead of markdown")
p.add_argument("--out", metavar="PATH", help="write the report to this file instead of printing it")
return p
def main(argv=None) -> int:
parser = _parser()
args = parser.parse_args(argv)
if args.uninstall and args.cap is not None:
parser.error("--cap does not go with --uninstall")
project = os.path.abspath(args.project)
return uninstall(args, os.environ, project) if args.uninstall else install(args, os.environ, project)
if __name__ == "__main__":
sys.exit(main())
# Copied from skills/evals/shared/pricing.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Token prices and cost math for the models that coding agents use.
Prices are USD per 1 million tokens, copied from the official pricing pages on
PRICES_CHECKED (see the source on each group below). Usage fields follow
Anthropic meanings, as transcripts.Usage does: `input` is uncached input,
`cache_read` and `cache_write` are separate, and `output` already includes
reasoning tokens, so reasoning is never charged twice.
Not covered, so a computed cost can differ from a bill:
- Batch discounts, fast mode, US-only inference (1.1x), and web search fees.
- GPT long-context rates (prompts over 272K input tokens cost 2x input and
cache, 1.5x output). Codex keeps its context window below that (258,400
tokens in local rollouts), so standard rates apply.
- Gemini models: no price was confirmed on Google's official pricing page, so
Gemini usage is reported as unpriced tokens.
- Prices on Amazon Bedrock and Google Vertex AI, which those platforms set.
Python 3.9+, standard library only.
"""
from __future__ import annotations
import re
from typing import Optional
PRICES_CHECKED = "2026-09-28"
def _anthropic(inp, cache_read, write_5m, write_1h, out) -> dict:
return {"input": inp, "output": out, "cache_read": cache_read,
"cache_write_5m": write_5m, "cache_write_1h": write_1h}
def _openai(inp, cached, write, out) -> dict:
# One cache-write rate; None where the pricing page lists none.
return {"input": inp, "output": out, "cache_read": cached,
"cache_write_5m": write, "cache_write_1h": write}
PRICES = {
# Anthropic API: https://platform.claude.com/docs/en/about-claude/pricing.md
# Cache reads are 0.1x input, except 0.025x on Fable 5.1 and Mythos 5.1 and 0.05x on Opus 5.5.
"claude-fable-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00),
"claude-mythos-5-1": _anthropic(10.00, 0.25, 12.50, 20.00, 50.00), # Fable 5.1 prices; id follows the naming pattern
"claude-fable-5": _anthropic(10.00, 1.00, 12.50, 20.00, 50.00),
"claude-opus-5-5": _anthropic(4.00, 0.20, 5.00, 8.00, 20.00),
"claude-opus-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-8": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-7": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-6": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-opus-4-5": _anthropic(5.00, 0.50, 6.25, 10.00, 25.00),
"claude-sonnet-5-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-5": _anthropic(2.00, 0.20, 2.50, 4.00, 10.00),
"claude-sonnet-4-6": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-sonnet-4-5": _anthropic(3.00, 0.30, 3.75, 6.00, 15.00),
"claude-haiku-4-5": _anthropic(1.00, 0.10, 1.25, 2.00, 5.00),
# OpenAI API, standard rates up to 272K input tokens: https://developers.openai.com/api/docs/pricing.md
# and the model pages https://developers.openai.com/api/docs/models/gpt-6-sol.md (also gpt-6-astra, gpt-6-luna).
"gpt-6-astra": _openai(10.00, 1.00, 12.50, 50.00),
"gpt-6-sol": _openai(2.00, 0.20, 2.50, 10.00),
"gpt-6-luna": _openai(0.10, 0.01, 0.125, 0.50),
"gpt-5.6-sol": _openai(4.00, 0.40, 5.00, 20.00),
"gpt-5.6-terra": _openai(2.00, 0.20, 2.50, 12.00),
"gpt-5.6-luna": _openai(0.20, 0.02, 0.25, 1.20),
"gpt-5.5": _openai(5.00, 0.50, None, 30.00),
"gpt-5.3-codex": _openai(1.75, 0.175, None, 14.00),
}
_CONTEXT_SUFFIX_RE = re.compile(r"\[[^\]]*\]$") # claude-opus-5-5[1m]
_DATE_SUFFIX_RE = re.compile(r"-(?:\d{8}|\d{4}-\d\d-\d\d)$") # -20251001 or -2026-09-01
# Cloud ids for the same models: Amazon Bedrock (us.anthropic.claude-opus-4-8-v1:0)
# and Google Vertex AI (claude-opus-4-8@20260101). They get the model's list price:
# the clouds set their own prices, which can differ a little (regional endpoints),
# but a list price keeps a spend cap working where "unknown" would count $0.
_PROVIDER_PREFIX_RE = re.compile(r"^(?:(?:us|eu|apac|jp|au|global|us-gov)\.)?(?:anthropic|openai)\.")
_PROVIDER_SUFFIX_RE = re.compile(r"(?:-v\d+(?::\d+)?|@\d{8}|@latest)$")
_SEPARATORS = "-_.@:"
_SIZE_WORDS = {"mini", "nano", "pro", "max", "lite", "turbo"} # another size is another price
def price_for(model) -> Optional[dict]:
"""Prices for a model id: the exact id, then the id without a cloud
provider prefix or suffix (Bedrock, Vertex AI), a context suffix, or a
date suffix, then the longest known id it starts with. A newer version
number or another size after the known id (claude-opus-5-6, gpt-5.5-pro)
does not count as a match. Returns None when the model is unknown."""
if not isinstance(model, str) or not model.strip():
return None
name = model.strip()
if name not in PRICES:
name = _PROVIDER_SUFFIX_RE.sub("", _CONTEXT_SUFFIX_RE.sub("", _PROVIDER_PREFIX_RE.sub("", name)))
name = _DATE_SUFFIX_RE.sub("", name)
if name not in PRICES:
best = ""
for known in PRICES:
if len(name) > len(known) and name.startswith(known) and name[len(known)] in _SEPARATORS:
next_word = re.split(r"[-_.@:]", name[len(known) + 1:])[0]
if not next_word.isdigit() and next_word not in _SIZE_WORDS and len(known) > len(best):
best = known
name = best
return dict(PRICES[name]) if name in PRICES else None
def _count(usage, key) -> int:
value = usage.get(key, 0) if isinstance(usage, dict) else getattr(usage, key, 0)
return value if isinstance(value, int) and not isinstance(value, bool) else 0
def cost_usd(usage, model) -> Optional[float]:
"""Dollar cost of one usage (a transcripts.Usage or a dict with the same
keys) on one model. None when the model is unknown: report those tokens as
unpriced. Cache writes without a known 1-hour part are priced at the
5-minute rate; a model with no listed cache-write price charges writes as input."""
p = price_for(model)
if p is None:
return None
write = _count(usage, "cache_write")
write_1h = min(_count(usage, "cache_write_1h"), write)
rate_5m = p["cache_write_5m"] if p["cache_write_5m"] is not None else p["input"]
rate_1h = p["cache_write_1h"] if p["cache_write_1h"] is not None else rate_5m
total = (_count(usage, "input") * p["input"] + _count(usage, "cache_read") * p["cache_read"]
+ (write - write_1h) * rate_5m + write_1h * rate_1h + _count(usage, "output") * p["output"])
return total / 1_000_000
# Copied from skills/evals/shared/safe.py by skills/evals/tools/sync_shared.py. Edit the source, then run the sync.
"""Make untrusted text safe to show in a Markdown report.
Text from a repository or a session transcript can hold secrets, carry
instructions for the agent that relays a report, or break a table. redact()
masks secrets; safe_text() also makes the text one inert line; code() puts
that line inside inline code.
Python 3.9+, standard library only.
"""
import re
import unicodedata
_MASK = "[REDACTED]"
# Word edges by ASCII rules. Python's \b counts a non-ASCII letter, such as an
# e with an acute accent, as part of the word, so a key right after one would
# stay unmasked.
_B = r"(?<![A-Za-z0-9_])"
_E = r"(?![A-Za-z0-9_])"
# Characters that draw nothing, besides the format characters (Unicode category
# Cf: zero-width spaces and joiners, direction marks, the soft hyphen): the
# combining grapheme joiner, Hangul fillers, Khmer inherent vowels, Mongolian
# and other variation selectors, and tag characters. Inside a key name they
# hide the key from the patterns below while the text still reads as the key.
_INVISIBLE_RE = re.compile(r"[\u034f\u115f\u1160\u17b4\u17b5\u180b-\u180f\u3164\ufe00-\ufe0f\uffa0"
r"\U000e0000-\U000e0fff]")
# Run time: hooks clean every tool call, and a tool input can be tens of
# kilobytes, so each pattern must take time in proportion to the text. A pattern
# that can start at many places inside one long run of characters, scan to the
# end of that run, and then fail takes time that grows with the square of the
# run (or worse). The patterns below limit or stop such scans; each one says how.
_SECRET_RES = [
# PEM private key blocks, including a block cut off before its END line.
re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----.*?(?:-----END [A-Z0-9 ]*PRIVATE KEY-----|\Z)", re.S),
re.compile(_B + r"sk-ant-[A-Za-z0-9_\-]{16,}"),
re.compile(_B + r"sk-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"gh[pousr]_[A-Za-z0-9]{20,}"),
re.compile(_B + r"github_pat_[A-Za-z0-9_]{20,}"),
re.compile(_B + r"glpat-[A-Za-z0-9_\-]{20,}"),
re.compile(_B + r"xox[abposr]-[A-Za-z0-9\-]{10,}"),
re.compile(_B + r"AIza[0-9A-Za-z_\-]{30,}"),
re.compile(_B + r"(?:AKIA|ASIA)[0-9A-Z]{16}" + _E),
re.compile(_B + r"[rsp]k_(?:live|test)_[0-9A-Za-z]{16,}"),
re.compile(_B + r"npm_[A-Za-z0-9]{30,}"),
re.compile(_B + r"hf_[A-Za-z0-9]{30,}"),
# A JSON web token. The first part stops before a later "-eyJ", where the
# next try starts, so no run is scanned again for each "-eyJ" in it.
re.compile(_B + r"eyJ(?:[A-Za-z0-9_]|-(?!eyJ)){8,}\.[A-Za-z0-9_\-]{8,}\.[A-Za-z0-9_\-]{8,}"),
]
# A mysql or mariadb program name, as one word.
_MYSQL = _B + r"(?:mysql|mysqldump|mysqladmin|mysqlsh|mariadb|mariadb-dump)" + _E
def _credential(before, unquoted, after="", flags=0):
"""A credential after a fixed prefix. Group 1, which redact() keeps, is
`before`, an optional quote (group 2), and `after`. The rest of the match
is masked: with a quote, everything up to the matching quote or the end of
the line, spaces included (a backslash escapes the next character, and the
closing quote stays); without one, `unquoted`."""
return re.compile(r"(%s([\"'])?%s)(?(2)(?:(?!\2)[^\\\n]|\\.)+|%s)" % (before, after, unquoted), flags)
# Keep group 1, mask the rest of the match.
_KEEP_PREFIX_RES = [
# Credentials passed as command flags: sshpass -p X, mysql -pX, curl -u user:X, --password X.
_credential(_B + r"sshpass\s+-p\s*", r"[^\s'\"]+"),
# The search for -p stops at the next program name, where the next try starts.
_credential(_MYSQL + r"(?:(?!" + _MYSQL + r")[^\n|;&])*?\s-p(?=[^\s-])", r"[^\s'\"]+", flags=re.I),
# One "=" or spaces after the flag: a user name can start with "=", so "[=\s]+"
# before it could split a run of "=" in every way.
_credential(r"(?:^|\s)(?:-u|--user)(?:=|\s+)", r"[^\s@'\"]+", after=r"[^\s:@'\"]+:"),
_credential(r"--pass(?:word|wd)?\s+", r"[^\s'\"-][^\s'\"]*", flags=re.I),
re.compile(r"(?i)(" + _B + r"(?:bearer|basic)\s+)[A-Za-z0-9._~+/\-]{16,}=*"),
# A URL scheme has at most 32 characters, so each try scans at most that far.
re.compile(r"(?i)(" + _B + r"[a-z][a-z0-9+.\-]{0,31}://[^/\s:@]+:)[^@\s/]+(?=@)"),
# A secret-like key and any non-empty value. The key is one whole name (a run
# of letters, digits, and _ . -) that holds a keyword anywhere. A try starts
# only where a name starts, and the lookahead stays inside that name, so each
# name is scanned once. "tokens" alone names a count (max_tokens: 1000), not a secret.
_credential(r"(?<![A-Za-z0-9_.\-])(?=[A-Za-z0-9_.\-]*?(?:api[_\-]?key|secret|token(?!s(?![A-Za-z0-9]))|"
r"passw(?:or)?d|pwd|credential|private[_\-]?key|access[_\-]?key))"
r"[A-Za-z0-9_.\-]+[\"']?\s*[:=]+\s*", r"[^\s\"',;}]+", flags=re.I),
]
_LONG_RUN_RE = re.compile(r"[A-Za-z0-9+=_\-]{40,}")
def _mask_long_run(m) -> str:
s = m.group(0)
if re.search(r"[0-9]", s) and re.search(r"[A-Za-z]", s):
return _MASK
return s
def _plain(text) -> str:
"""NFKC form (fullwidth and other compatibility letters become plain ones)
without the characters that draw nothing."""
if text.isascii():
return text
text = unicodedata.normalize("NFKC", text)
return _INVISIBLE_RE.sub("", "".join(ch for ch in text if unicodedata.category(ch) != "Cf"))
def redact(text) -> str:
"""Mask API keys, tokens, private keys, and long secret-like strings. The
text comes back in NFKC form without the characters that draw nothing, so a
fullwidth or split key name is masked like the plain one."""
if not text:
return ""
out = _plain(str(text))
for rx in _SECRET_RES:
out = rx.sub(_MASK, out)
for rx in _KEEP_PREFIX_RES:
out = rx.sub(lambda m: m.group(1) + _MASK, out)
return _LONG_RUN_RE.sub(_mask_long_run, out)
def safe_text(text, limit=160) -> str:
"""Make untrusted text (a command, a path, a claim, a tool description) safe
to show in a markdown report: secrets masked, control characters and line
breaks turned into spaces, backticks and table pipes replaced, runs of
spaces collapsed, and the result cut to `limit` characters. Text from a
repository or a transcript can carry instructions or break a table; this
keeps it one inert line."""
s = redact(text if isinstance(text, str) else str(text))
s = "".join(ch if ch.isprintable() else " " for ch in s)
s = " ".join(s.replace("`", "'").replace("|", "/").split())
return s if len(s) <= limit else s[: max(limit - 3, 0)] + "..."
def code(text, limit=160) -> str:
"""Untrusted text for a Markdown report, inside inline code so links, HTML,
and bare URLs stay literal; safe_text already replaces backticks and pipes.
Text that is empty after cleaning shows as `(empty)`, because a bare pair
of backticks renders as two stray backticks."""
return "`%s`" % (safe_text(text, limit) or "(empty)")
#!/usr/bin/env python3
"""Show what the runaway guard has counted for a session, and start the count over.
With no flags it reports the session the guard checked most recently: dollars
spent against the cap, how often each trip wire stepped in, and the latest
trips. --session picks another session (the full id or its first characters),
--list shows every session the guard knows, and --reset starts the counts of
one session over: spend counts from $0 again, and the loop and failure counts
clear. The session total stays in the record.
Exit codes: 0 done, 2 bad arguments or no session matching --session.
Python 3.9+, standard library only. Reads and writes only the guard's own
state folder: ${XDG_STATE_HOME:-~/.local/state}/runaway-guard.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import guard
from safe import code, safe_text
safe = safe_text
WIRES = ("loop", "failures", "spend")
# ---------------------------------------------------------------------------
# Loading
# ---------------------------------------------------------------------------
def sessions(folder) -> list:
"""(path, state) for every session file, most recently checked first."""
found = []
for e in guard._scandir(folder):
if e.name.endswith(".json") and e.name != guard.CONFIG_NAME:
state = guard.load_state(e.path)
if state is not None:
found.append((e.path, state))
found.sort(key=lambda item: (str(item[1].get("updated") or ""), os.path.getmtime(item[0])), reverse=True)
return found
def pick(found, wanted, parser):
if not found:
return None
if not wanted:
return found[0]
exact = [f for f in found if f[1].get("session_id") == wanted
or os.path.basename(f[0]) == wanted + ".json"]
if exact:
return exact[0]
prefixed = [f for f in found if str(f[1].get("session_id") or "").startswith(wanted)]
if len(prefixed) == 1:
return prefixed[0]
if prefixed:
parser.error("%d sessions start with %s; give more of the id" % (len(prefixed), safe(wanted, 80)))
parser.error("no session matches %s; run with --list to see the sessions the guard knows"
% safe(wanted, 80))
def errors_logged(folder) -> int:
try:
with open(os.path.join(folder, "errors.log"), encoding="utf-8", errors="replace") as fh:
return sum(1 for line in fh if line.strip())
except OSError:
return 0
# ---------------------------------------------------------------------------
# Summaries
# ---------------------------------------------------------------------------
def _number(value, default=0.0) -> float:
return value if isinstance(value, (int, float)) and not isinstance(value, bool) else default
def summary(path, state, folder) -> dict:
counts = state.get("counts") if isinstance(state.get("counts"), dict) else {}
limits = dict(guard.DEFAULTS)
if isinstance(state.get("limits"), dict):
limits.update({k: v for k, v in state["limits"].items() if k in guard.DEFAULTS})
total = _number(state.get("cost_usd"))
streaks = state.get("streaks") if isinstance(state.get("streaks"), dict) else {}
trips = [{"wire": safe(t.get("wire"), 20), "at": safe(t.get("at"), 30), "tool": safe(t.get("tool"), 80),
"detail": safe(t.get("detail"), 80)}
for t in state.get("trips") or [] if isinstance(t, dict)]
return {
"session": safe(state.get("session_id"), 140),
"harness": safe(state.get("harness"), 20),
"created": safe(state.get("created"), 30),
"updated": safe(state.get("updated"), 30),
"spent_usd": max(0.0, total - _number(state.get("baseline_usd"))),
"total_usd": total,
"cap_usd": limits["spend_cap_usd"],
"limits": limits,
"trips": {w: int(_number(counts.get(w), 0)) for w in WIRES + ("warning",)},
"recent_trips": trips[-10:],
"failure_streak": int(max([_number(v, 0) for v in streaks.values()] or [0])),
"transcripts": len(state.get("files") or {}),
"estimated_usd": _number(state.get("estimated_usd")),
"estimated_tokens": int(_number(state.get("estimated_tokens"), 0)),
"estimated_models": [safe(m, 60) for m in state.get("estimated_models") or []],
"resets": int(_number(state.get("resets"), 0)),
"errors_logged": errors_logged(folder),
"state_path": path,
}
def when(iso) -> str:
"""2026-09-28T10:02:05Z as 2026-09-28 10:02 UTC."""
return iso[:10] + " " + iso[11:16] + " UTC" if len(iso) >= 16 and iso[10:11] == "T" else iso
def _short(session) -> str:
return session[:8] if len(session) > 12 else session
def _stepped_in(s) -> int:
return sum(s["trips"][w] for w in WIRES)
def headline(s, markdown=True) -> str:
"""The one-line summary; the session id sits in inline code for the markdown report, plain for JSON."""
cap = s["cap_usd"]
money = ("has spent $%.2f of its $%.2f cap" % (s["spent_usd"], cap) if cap
else "has spent $%.2f (no spend cap is set)" % s["spent_usd"])
n = _stepped_in(s)
fired = "no trip wire has fired" if not n else "the guard stepped in %d time%s" % (n, "" if n == 1 else "s")
session = _short(s["session"])
return "Session %s %s; %s." % (code(session) if markdown else session, money, fired)
def report(s, reset_note="") -> str:
lim = s["limits"]
cap = s["cap_usd"]
rows = [
("Spend", "$%.2f per session, warning at %d%%" % (cap, round(100 * lim["warn_at"])) if cap else "off",
"$%.2f" % s["spent_usd"] + (" (%d%%)" % round(100 * s["spent_usd"] / cap) if cap else ""),
"%d call%s blocked" % (s["trips"]["spend"], "" if s["trips"]["spend"] == 1 else "s")),
("Loop", "the same call %d times with nothing changed" % lim["loop_repeats"] if lim["loop_repeats"]
else "off", "", "%d call%s blocked" % (s["trips"]["loop"], "" if s["trips"]["loop"] == 1 else "s")),
("Failures", "%d failed calls in a row" % lim["failure_streak"] if lim["failure_streak"] else "off",
"%d in a row" % s["failure_streak"],
"stepped in %d time%s" % (s["trips"]["failures"], "" if s["trips"]["failures"] == 1 else "s")),
]
out = ["**%s**" % headline(s), ""]
if reset_note:
out += [reset_note, ""]
out += ["| Trip wire | Limit | Now | Stepped in |", "|---|---|---|---|"]
out += ["| %s |" % " | ".join(row) for row in rows]
if s["recent_trips"]:
out += ["", "Latest events:"]
out += ["- %s, %s: %s, %s" % (when(t["at"]), t["wire"], code(t["tool"]) if t["tool"] else "a tool call",
t["detail"])
for t in s["recent_trips"][-5:]]
out.append("")
if s["trips"]["warning"]:
out.append("Spend warnings shown: %d." % s["trips"]["warning"])
if s["resets"]:
out.append("Counts started over %d time%s; the session total so far is $%.2f."
% (s["resets"], "" if s["resets"] == 1 else "s", s["total_usd"]))
if s["estimated_tokens"]:
out.append("$%.2f of the total is an estimate: no price is known for %s, so its %s tokens are priced "
"like the most expensive model of the same family." % (
s["estimated_usd"], ", ".join(code(m) for m in s["estimated_models"]) or "a model",
"{:,}".format(s["estimated_tokens"])))
out.append("Transcripts read: %d. Last check: %s. Limits as of that check." % (s["transcripts"], when(s["updated"])))
if s["errors_logged"]:
out.append("The guard logged %d internal error%s and let those calls through; see errors.log in the "
"state folder." % (s["errors_logged"], "" if s["errors_logged"] == 1 else "s"))
return "\n".join(out) + "\n"
def list_report(summaries) -> str:
out = ["**Runaway guard knows %d session%s.**" % (len(summaries), "" if len(summaries) == 1 else "s"), "",
"| Session | Last check | Spent | Cap | Stepped in |", "|---|---|---|---|---|"]
for s in summaries:
out.append("| %s | %s | $%.2f | %s | %d |" % (code(s["session"]), when(s["updated"]), s["spent_usd"],
"$%.2f" % s["cap_usd"] if s["cap_usd"] else "off",
_stepped_in(s)))
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# Reset
# ---------------------------------------------------------------------------
def reset(path, state) -> dict:
"""Start the counts over under the session lock; returns the state as saved."""
lock = guard._acquire(path[:-len(".json")] + ".lock")
if lock is None:
raise RuntimeError("the session is busy; try again in a moment")
try:
state = guard.load_state(path) or state
state["baseline_usd"] = _number(state.get("cost_usd"))
state["agents"], state["streaks"], state["trips"] = {}, {}, []
state["counts"] = {w: 0 for w in WIRES + ("warning",)}
state["warned_cap"] = None
state["resets"] = int(_number(state.get("resets"), 0)) + 1
state["updated"] = guard._now_iso()
guard.save_state(path, state)
return state
finally:
guard._release(lock)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(argv=None) -> int:
parser = argparse.ArgumentParser(
description="Show the runaway guard's spend and trip counts for a session, or start them over.")
parser.add_argument("--session", metavar="ID", help="the session id, or its first characters "
"(default: the session checked most recently)")
parser.add_argument("--list", action="store_true", help="list every session the guard knows")
parser.add_argument("--reset", action="store_true",
help="start the counts over: spend from $0, loop and failure counts cleared")
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)
if args.list and args.reset:
parser.error("--reset works on one session; leave out --list")
folder = guard.state_dir()
found = sessions(folder)
if args.list:
summaries = [summary(p, st, folder) for p, st in found]
text = (json.dumps({"state_folder": folder, "sessions": summaries}, indent=2) + "\n" if args.json
else list_report(summaries))
else:
chosen = pick(found, args.session, parser)
if chosen is None:
if args.reset:
parser.error("there is no session to reset yet")
head = "Runaway guard has not checked any session yet."
text = (json.dumps({"headline": head, "state_folder": folder, "session": None}, indent=2) + "\n"
if args.json else
"**%s**\n\nIt keeps its counts in %s. If the hook is installed, the first tool call of a "
"new session creates them.\n" % (head, safe(folder, 300)))
else:
path, state = chosen
note = ""
if args.reset:
try:
state = reset(path, state)
except RuntimeError as exc:
print("error: %s" % exc, file=sys.stderr)
return 2
note = ("Reset done: spend counts from $0.00 again, and the loop and failure counts are "
"cleared. The session total so far, $%.2f, stays in the record."
% _number(state.get("cost_usd")))
s = summary(path, state, folder)
if args.json:
text = json.dumps(dict(s, headline=headline(s, markdown=False)), indent=2) + "\n"
else:
text = report(s, note)
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(args.out, 300), exc.strerror or exc),
file=sys.stderr)
return 2
print("Report written to %s" % safe(args.out, 300))
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
sys.exit(main())
# 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