Codex Cost Control skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Codex Cost Control"
  short_description: "Know tokens per task, then cut the waste"
  default_prompt: "$codex-cost-control I keep hitting my Codex usage limit by Wednesday, where is it going?"

#!/usr/bin/env python3
"""Sum token usage from `codex exec --json` event logs (JSONL).

Usage:
  python3 sum_usage.py run1.jsonl [run2.jsonl ...]
  python3 sum_usage.py runs/*.jsonl --rates 1.25,0.125,10 --budget 2000000
  codex exec --json "task" | tee run.jsonl | python3 sum_usage.py -

Reads every `turn.completed` event and totals its `usage` object per file.
`cached_input_tokens` is treated as a subset of `input_tokens`.
--rates IN,CACHED,OUT   price per 1M tokens (any unit: USD, credits) for an estimate.
--budget N              exit 2 if total tokens (input + output) exceed N.
Exit codes: 0 ok, 1 no usage found / bad input, 2 budget exceeded.
"""
import argparse
import json
import sys

FIELDS = ("input_tokens", "cached_input_tokens", "output_tokens", "reasoning_output_tokens")


def read_usage(stream, label):
    totals = dict.fromkeys(FIELDS, 0)
    turns = failed = bad = 0
    for line in stream:
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            bad += 1
            continue
        kind = event.get("type")
        if kind == "turn.completed":
            turns += 1
            usage = event.get("usage") or {}
            for key in FIELDS:
                value = usage.get(key) or 0
                if isinstance(value, (int, float)):
                    totals[key] += int(value)
        elif kind == "turn.failed":
            failed += 1
    return {"label": label, "turns": turns, "failed": failed, "bad_lines": bad, **totals}


def estimate(row, rates):
    if not rates:
        return None
    rate_in, rate_cached, rate_out = rates
    uncached = max(row["input_tokens"] - row["cached_input_tokens"], 0)
    return (uncached * rate_in + row["cached_input_tokens"] * rate_cached
            + row["output_tokens"] * rate_out) / 1_000_000


def parse_rates(text):
    parts = [float(p) for p in text.split(",")]
    if len(parts) != 3:
        raise argparse.ArgumentTypeError("--rates needs IN,CACHED,OUT")
    return parts


def main():
    parser = argparse.ArgumentParser(description="Total token usage from codex exec --json logs.")
    parser.add_argument("files", nargs="+", help="JSONL files, or - for stdin")
    parser.add_argument("--rates", type=parse_rates, help="IN,CACHED,OUT price per 1M tokens")
    parser.add_argument("--budget", type=int, help="fail (exit 2) above this many total tokens")
    args = parser.parse_args()

    rows = []
    for name in args.files:
        if name == "-":
            rows.append(read_usage(sys.stdin, "<stdin>"))
            continue
        try:
            with open(name, encoding="utf-8") as handle:
                rows.append(read_usage(handle, name))
        except OSError as err:
            print(f"error: cannot read {name}: {err}", file=sys.stderr)
            return 1

    header = f"{'file':40} {'turns':>5} {'input':>11} {'cached':>11} {'cache%':>6} {'output':>9} {'reason':>9}"
    if args.rates:
        header += f" {'est.cost':>10}"
    print(header)
    total = {"label": "TOTAL", "turns": 0, "failed": 0, "bad_lines": 0, **dict.fromkeys(FIELDS, 0)}
    for row in rows + [None]:
        if row is None:
            row = total
        else:
            for key in ("turns", "failed", "bad_lines", *FIELDS):
                total[key] += row[key]
        pct = 100 * row["cached_input_tokens"] / row["input_tokens"] if row["input_tokens"] else 0
        line = (f"{row['label'][-40:]:40} {row['turns']:>5} {row['input_tokens']:>11,} "
                f"{row['cached_input_tokens']:>11,} {pct:>5.0f}% {row['output_tokens']:>9,} "
                f"{row['reasoning_output_tokens']:>9,}")
        cost = estimate(row, args.rates)
        if cost is not None:
            line += f" {cost:>10.4f}"
        print(line)

    if total["failed"]:
        print(f"note: {total['failed']} failed turn(s) found", file=sys.stderr)
    if total["bad_lines"]:
        print(f"note: {total['bad_lines']} non-JSON line(s) skipped", file=sys.stderr)
    if total["turns"] == 0:
        print("error: no turn.completed events found (was the run started with --json?)", file=sys.stderr)
        return 1
    used = total["input_tokens"] + total["output_tokens"]
    if args.budget is not None and used > args.budget:
        print(f"BUDGET EXCEEDED: {used:,} tokens > {args.budget:,}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())