/agent-usage-counter

# agent-usage-counter

Count aggregate Claude Code and Codex usage from local JSONL logs with Python 3.10 or newer. The utility uses only the standard library and makes no network requests.

```bash
python3 -m unittest discover -s tests
python3 scripts/usage_counter.py \
  --claude ~/.claude/projects \
  --codex ~/.codex/sessions \
  --codex ~/.codex/archived_sessions
```

Repeat `--claude` or `--codex` for copied exports or another machine. Use `--json` for machine-readable aggregate output.

Supported Claude Code records contain `message.usage` and a top-level `requestId`. Duplicate content-block records with the same request ID count once. Claude input includes new input, cache reads, and cache creation. Only cache reads count toward cached input.

Supported Codex records have top-level type `event_msg`, payload type `token_count`, and `payload.info.last_token_usage`. The script ignores `total_token_usage`, which is cumulative within a session. Codex input already includes cached input.

Output contains totals, dates, and record diagnostics. It does not contain input paths, project names, prompts, responses, or session text. Incomplete usage records are skipped and reported. Invalid JSON lines are also reported.

Token traffic is not a price, quality, or efficiency comparison. Compare tools only when their dates, work, models, and settings are controlled.

Methodology: https://ivanmisic.net/blog/ai-tools/codex-vs-claude-code

---
name: agent-usage-counter
description: Count aggregate Claude Code and Codex usage from local JSONL session logs. Use when the user wants active days, input and output totals, cached input share, or input-to-output ratio. Keep all logs local and never expose prompts, paths, project names, or session text.
---

# Agent usage counter

Run `scripts/usage_counter.py` against paths the user names. Do not copy or upload their logs.

Use one or more of these options:

```bash
python3 scripts/usage_counter.py --claude PATH
python3 scripts/usage_counter.py --codex PATH
python3 scripts/usage_counter.py --claude PATH --codex PATH --json
```

Before reporting a comparison, state whether the tools cover the same dates and work. Unequal windows support separate usage descriptions, not a winner, price, quality, or efficiency claim.

#!/usr/bin/env python3
"""Aggregate local Claude Code and Codex usage without emitting log content."""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
from dataclasses import asdict, dataclass, field
from datetime import date
from pathlib import Path
from typing import Any, Iterable


@dataclass
class UsageTotals:
    tool: str
    input_tokens: int = 0
    output_tokens: int = 0
    cached_input_tokens: int = 0
    counted_requests: int = 0
    duplicate_records: int = 0
    missing_usage_records: int = 0
    invalid_json_lines: int = 0
    days: set[str] = field(default_factory=set)

    def add(self, timestamp: Any, input_tokens: int, output_tokens: int, cached_tokens: int) -> None:
        self.input_tokens += input_tokens
        self.output_tokens += output_tokens
        self.cached_input_tokens += cached_tokens
        self.counted_requests += 1
        parsed_day = iso_day(timestamp)
        if parsed_day is not None:
            self.days.add(parsed_day)

    def public(self) -> dict[str, Any]:
        cache_share = (
            round(self.cached_input_tokens * 100 / self.input_tokens, 2)
            if self.input_tokens > 0
            else None
        )
        ratio = (
            round(self.input_tokens / self.output_tokens, 2)
            if self.output_tokens > 0
            else None
        )
        values = asdict(self)
        values.pop("days")
        values.update(
            active_days=len(self.days),
            first_day=min(self.days) if self.days else None,
            last_day=max(self.days) if self.days else None,
            cache_share_percent=cache_share,
            input_to_output_ratio=ratio,
        )
        return values


def iso_day(value: Any) -> str | None:
    if not isinstance(value, str) or len(value) < 10:
        return None
    candidate = value[:10]
    try:
        date.fromisoformat(candidate)
    except ValueError:
        return None
    return candidate


def token(value: Any) -> int:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return 0
    return max(0, int(value))


def jsonl_files(paths: Iterable[str]) -> list[Path]:
    found: dict[str, Path] = {}
    for raw in paths:
        path = Path(raw).expanduser()
        candidates = [path] if path.is_file() else path.rglob("*.jsonl") if path.is_dir() else []
        for candidate in candidates:
            try:
                resolved = candidate.resolve(strict=True)
            except OSError:
                continue
            if resolved.is_file():
                found[str(resolved)] = resolved
    return [found[key] for key in sorted(found)]


def records(path: Path, totals: UsageTotals) -> Iterable[dict[str, Any]]:
    try:
        handle = path.open(encoding="utf-8", errors="replace")
    except OSError:
        return
    with handle:
        for line in handle:
            try:
                value = json.loads(line)
            except (json.JSONDecodeError, TypeError):
                totals.invalid_json_lines += 1
                continue
            if isinstance(value, dict):
                yield value


def stable_hash(value: Any) -> str:
    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def count_claude(paths: Iterable[str]) -> UsageTotals:
    totals = UsageTotals("Claude Code")
    seen: set[str] = set()
    for path in jsonl_files(paths):
        for record in records(path, totals):
            message = record.get("message")
            if not isinstance(message, dict) or message.get("role") != "assistant":
                continue
            usage = message.get("usage")
            if not isinstance(usage, dict):
                totals.missing_usage_records += 1
                continue
            identity = record.get("requestId") or record.get("request_id") or message.get("id")
            key = "id:" + str(identity) if identity else "record:" + stable_hash(record)
            if key in seen:
                totals.duplicate_records += 1
                continue
            seen.add(key)
            fresh = token(usage.get("input_tokens"))
            cached = token(usage.get("cache_read_input_tokens"))
            cache_creation = token(usage.get("cache_creation_input_tokens"))
            output = token(usage.get("output_tokens"))
            if fresh + cached + cache_creation + output == 0:
                totals.missing_usage_records += 1
                continue
            totals.add(record.get("timestamp"), fresh + cached + cache_creation, output, cached)
    return totals


def count_codex(paths: Iterable[str]) -> UsageTotals:
    totals = UsageTotals("Codex")
    seen: set[str] = set()
    for path in jsonl_files(paths):
        session_id = ""
        for record in records(path, totals):
            payload = record.get("payload")
            if record.get("type") == "session_meta" and isinstance(payload, dict):
                session_id = str(payload.get("id") or payload.get("session_id") or "")
                continue
            if record.get("type") != "event_msg" or not isinstance(payload, dict):
                continue
            if payload.get("type") != "token_count":
                continue
            info = payload.get("info")
            usage = info.get("last_token_usage") if isinstance(info, dict) else None
            if not isinstance(usage, dict):
                totals.missing_usage_records += 1
                continue
            input_tokens = token(usage.get("input_tokens"))
            cached = token(usage.get("cached_input_tokens"))
            output = token(usage.get("output_tokens"))
            if input_tokens + cached + output == 0:
                totals.missing_usage_records += 1
                continue
            identity = {
                "session": session_id,
                "timestamp": record.get("timestamp"),
                "ordinal": record.get("ordinal"),
                "input": input_tokens,
                "cached": cached,
                "output": output,
                "total": token(usage.get("total_tokens")),
            }
            key = stable_hash(identity)
            if key in seen:
                totals.duplicate_records += 1
                continue
            seen.add(key)
            totals.add(record.get("timestamp"), input_tokens, output, cached)
    return totals


def format_table(rows: list[dict[str, Any]]) -> str:
    headers = ["Tool", "Days", "Input", "Output", "Cached", "Cache %", "Input/output"]
    values = []
    for row in rows:
        values.append([
            row["tool"],
            str(row["active_days"]),
            f'{row["input_tokens"]:,}',
            f'{row["output_tokens"]:,}',
            f'{row["cached_input_tokens"]:,}',
            "n/a" if row["cache_share_percent"] is None else f'{row["cache_share_percent"]:.2f}',
            "n/a" if row["input_to_output_ratio"] is None else f'{row["input_to_output_ratio"]:.2f}:1',
        ])
    widths = [max(len(headers[i]), *(len(row[i]) for row in values)) for i in range(len(headers))]
    lines = ["  ".join(headers[i].ljust(widths[i]) for i in range(len(headers)))]
    lines.append("  ".join("-" * width for width in widths))
    lines.extend("  ".join(row[i].ljust(widths[i]) for i in range(len(headers))) for row in values)
    for row in rows:
        lines.append(
            f'{row["tool"]}: {row["counted_requests"]} counted, '
            f'{row["duplicate_records"]} duplicates, '
            f'{row["missing_usage_records"]} missing usage, '
            f'{row["invalid_json_lines"]} invalid JSON lines; '
            f'window {row["first_day"] or "n/a"} to {row["last_day"] or "n/a"}.'
        )
    return "\n".join(lines)


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--claude", action="append", default=[], metavar="PATH")
    parser.add_argument("--codex", action="append", default=[], metavar="PATH")
    parser.add_argument("--json", action="store_true", help="print aggregate JSON")
    args = parser.parse_args(argv)
    if not args.claude and not args.codex:
        parser.error("add at least one --claude or --codex path")
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(sys.argv[1:] if argv is None else argv)
    totals = []
    if args.claude:
        totals.append(count_claude(args.claude).public())
    if args.codex:
        totals.append(count_codex(args.codex).public())
    if args.json:
        print(json.dumps({"results": totals}, indent=2))
    else:
        print(format_table(totals))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import importlib.util
import json
import subprocess
import sys
import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "usage_counter.py"
SPEC = importlib.util.spec_from_file_location("usage_counter", SCRIPT)
assert SPEC is not None and SPEC.loader is not None
MODULE = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = MODULE
SPEC.loader.exec_module(MODULE)


class UsageCounterTest(unittest.TestCase):
    def test_claude_deduplicates_request_ids_and_overlapping_exports(self) -> None:
        totals = MODULE.count_claude([
            str(ROOT / "tests/fixtures/claude/export-a"),
            str(ROOT / "tests/fixtures/claude/export-b"),
        ]).public()

        self.assertEqual(210, totals["input_tokens"])
        self.assertEqual(15, totals["output_tokens"])
        self.assertEqual(170, totals["cached_input_tokens"])
        self.assertEqual(2, totals["counted_requests"])
        self.assertEqual(3, totals["duplicate_records"])
        self.assertEqual(1, totals["missing_usage_records"])
        self.assertEqual(2, totals["active_days"])

    def test_codex_uses_per_turn_usage_and_deduplicates_exports(self) -> None:
        totals = MODULE.count_codex([
            str(ROOT / "tests/fixtures/codex/export-a"),
            str(ROOT / "tests/fixtures/codex/export-b"),
        ]).public()

        self.assertEqual(300, totals["input_tokens"])
        self.assertEqual(30, totals["output_tokens"])
        self.assertEqual(180, totals["cached_input_tokens"])
        self.assertEqual(2, totals["counted_requests"])
        self.assertEqual(2, totals["duplicate_records"])
        self.assertEqual(1, totals["missing_usage_records"])
        self.assertEqual(2, totals["active_days"])

    def test_json_output_excludes_private_fixture_content(self) -> None:
        completed = subprocess.run(
            [
                sys.executable,
                str(SCRIPT),
                "--claude",
                str(ROOT / "tests/fixtures/claude"),
                "--codex",
                str(ROOT / "tests/fixtures/codex"),
                "--json",
            ],
            check=True,
            capture_output=True,
            text=True,
        )
        result = json.loads(completed.stdout)
        self.assertEqual(2, len(result["results"]))
        for private_value in ("secret-project", "private prompt", "/workspace/private"):
            self.assertNotIn(private_value, completed.stdout)


if __name__ == "__main__":
    unittest.main()