Analytics for claude code / opencode / codex

session-tokens.py

#!/usr/bin/env python3

"""Report token usage from Claude Code transcripts, Codex CLI rollouts, or OpenCode SQLite."""

import argparse
import json
import math
import os
import re
import sqlite3
import sys
from collections import defaultdict
from datetime import datetime, time, timedelta
from pathlib import Path


MODEL_PRICES_PER_MILLION = {
    # input, output, cache read, cache write
    # Claude cache writes use the 1h-TTL rate (2x input, not the 5m 1.25x):
    # Claude Code caches with a 1-hour TTL.
    "claude-fable-5": (10.00, 50.00, 1.00, 20.00),
    "claude-opus-4-8": (5.00, 25.00, 0.50, 10.00),
    "claude-opus-4-7": (5.00, 25.00, 0.50, 10.00),
    "claude-opus-4-6": (5.00, 25.00, 0.50, 10.00),
    "claude-opus-4-5": (5.00, 25.00, 0.50, 10.00),
    "claude-sonnet-5": (2.00, 10.00, 0.20, 4.00),
    "claude-sonnet-4-6": (3.00, 15.00, 0.30, 6.00),
    "claude-sonnet-4-5": (3.00, 15.00, 0.30, 6.00),
    "claude-haiku-4-5": (1.00, 5.00, 0.10, 2.00),
    "openai/gpt-5.6-sol": (5.00, 30.00, 0.50, 6.25),
    "openai/gpt-5.6-terra": (2.50, 15.00, 0.25, 3.125),
    "openai/gpt-5.6-luna": (1.00, 6.00, 0.10, 1.25),
    "openai/gpt-5.5": (5.00, 30.00, 0.50, 0.00),
}


def parse_timestamp(value: str | None) -> float | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def parse_bound(value: str, end: bool = False) -> float:
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        raise ValueError(f"Invalid ISO date or timestamp: {value}") from None
    if end and len(value) == 10:
        parsed = datetime.combine(parsed.date(), time.max, parsed.tzinfo)
    return parsed.timestamp()


def parse_range_bound(value: str, end: bool = False) -> float:
    if len(value) >= 2 and value[:-1].isdigit() and value[-1].lower() in {"s", "m", "h", "d"}:
        return datetime.now().astimezone().timestamp() - parse_interval(value)
    return parse_bound(value, end=end)


def parse_interval(value: str) -> int:
    units = {"s": 1, "m": 60, "h": 3600, "d": 86400}
    if len(value) < 2 or not value[:-1].isdigit() or value[-1].lower() not in units:
        raise ValueError(f"Invalid interval: {value} (use values such as 10m, 1h, or 1d)")
    seconds = int(value[:-1]) * units[value[-1].lower()]
    if not seconds:
        raise ValueError("Interval must be greater than zero")
    return seconds


def default_interval(duration: float) -> tuple[int, str]:
    choices = (
        (20 * 60, "1m"),
        (100 * 60, "5m"),
        (200 * 60, "10m"),
        (10 * 3600, "30m"),
        (20 * 3600, "1h"),
        (40 * 3600, "2h"),
        (5 * 86400, "6h"),
        (10 * 86400, "12h"),
        (20 * 86400, "1d"),
        (40 * 86400, "2d"),
        (float("inf"), "5d"),
    )
    for maximum, label in choices:
        if duration <= maximum:
            return parse_interval(label), label
    raise AssertionError("unreachable")


def empty_time_stats() -> dict:
    return {"text": 0.0, "reasoning": 0.0, "tools": 0.0, "tool_calls": 0,
            "sub_calls": 0, "wall": 0.0, "per_model_think": defaultdict(float)}


def usage(calls: list[dict]) -> dict[str, int]:
    result = {"calls": len(calls), "input": 0, "cache_read": 0, "cache_write": 0, "output": 0}
    for call in calls:
        tokens = call["usage"]
        result["input"] += tokens.get("input_tokens", 0)
        result["cache_read"] += tokens.get("cache_read_input_tokens", 0)
        result["cache_write"] += tokens.get("cache_creation_input_tokens", 0)
        result["output"] += tokens.get("output_tokens", 0)
    result["total"] = sum(result[key] for key in ("input", "cache_read", "cache_write", "output"))
    return result


def add_usage(rows: list[dict[str, int]]) -> dict[str, int]:
    keys = ("calls", "input", "cache_read", "cache_write", "output", "total")
    return {key: sum(row[key] for row in rows) for key in keys}


def cache_hit_rate_rows(session_calls: dict) -> list[list[str]]:
    intervals = (
        (0, 60),
        (60, 300),
        (300, 600),
        (600, 1800),
        (1800, 3600),
        (3600, float("inf")),
    )
    models = {call["model"] for calls in session_calls.values() for call in calls}
    counts = {model: [[0, 0] for _ in intervals] for model in models}

    for calls in session_calls.values():
        ordered = sorted((call for call in calls if call.get("timestamp") is not None), key=lambda call: call["timestamp"])
        for previous, current in zip(ordered, ordered[1:]):
            if previous["model"] != current["model"]:
                continue
            gap = current["timestamp"] - previous["timestamp"]
            if gap < 0:
                continue
            for index, (lower, upper) in enumerate(intervals):
                if lower <= gap < upper:
                    counts[current["model"]][index][1] += 1
                    current_usage = current["usage"]
                    previous_usage = previous["usage"]
                    cache_read = current_usage.get("cache_read_input_tokens", 0)
                    previous_context = (
                        previous_usage.get("input_tokens", 0)
                        + previous_usage.get("cache_read_input_tokens", 0)
                        + previous_usage.get("cache_creation_input_tokens", 0)
                        + previous_usage.get("output_tokens", 0)
                    )
                    if previous_context and cache_read * 2 >= previous_context:
                        counts[current["model"]][index][0] += 1
                    break

    rows = []
    for model in sorted(models):
        cells = []
        for hits, total in counts[model]:
            cells.append(f"{hits / total * 100:.1f}%" if total else "-")
        rows.append([model, *cells])
    return rows


def print_cache_hit_rates(session_calls: dict) -> None:
    print("\nCache Hit Rate By Message Gap\n")
    print_table(
        ["Model", "<1m", "1-5m", "5-10m", "10-30m", "30-60m", "1h+"],
        cache_hit_rate_rows(session_calls),
        {1, 2, 3, 4, 5, 6},
    )


def short_id(value: str, limit: int = 10) -> str:
    return value if len(value) <= limit else value[:limit] + "…"


def short_title(value: str, limit: int = 45) -> str:
    return value if len(value) <= limit else value[: limit - 1] + "…"


def number(value: int) -> str:
    return f"{value:,}"


def humanize_seconds(value: float) -> str:
    seconds = int(round(value))
    if seconds < 60:
        return f"{seconds}s"
    if seconds < 3600:
        return f"{seconds // 60}m {seconds % 60:02d}s"
    if seconds < 86400:
        return f"{seconds // 3600}h {(seconds % 3600) // 60:02d}m {seconds % 60:02d}s"
    return f"{seconds // 86400}d {(seconds % 86400) // 3600:02d}h {(seconds % 3600) // 60:02d}m {seconds % 60:02d}s"


class Tee:
    def __init__(self, *streams):
        self.streams = streams

    def write(self, data: str) -> None:
        for stream in self.streams:
            stream.write(data)

    def flush(self) -> None:
        for stream in self.streams:
            stream.flush()


def print_table(headers: list[str], rows: list[list[str]], right_aligned: set[int]) -> None:
    widths = [max(len(headers[index]), *(len(row[index]) for row in rows)) for index in range(len(headers))]
    def render(row):
        return "  ".join(value.rjust(widths[index]) if index in right_aligned else value.ljust(widths[index])
                        for index, value in enumerate(row))
    print(render(headers))
    print(render(["-" * width for width in widths]))
    for row in rows:
        print(render(row))


def token_buckets(calls, since, until, interval):
    count = max(1, math.ceil((until - since) / interval))
    buckets = [0] * count
    model_buckets = {}
    for call in calls:
        timestamp = call["timestamp"]
        if timestamp is None or not since <= timestamp <= until:
            continue
        index = min(int((timestamp - since) // interval), count - 1)
        tokens = usage([call])["total"]
        buckets[index] += tokens
        model_buckets.setdefault(call["model"], [0] * count)[index] += tokens
    return buckets, model_buckets


def model_symbols(model_buckets):
    palette = "█░◆●▲▼◇○■□◉"
    models = sorted(model_buckets, key=lambda model: (-sum(model_buckets[model]), model))
    if len(models) > len(palette):
        raise SystemExit(f"Too many models to chart: {len(models)}")
    return {model: palette[index] for index, model in enumerate(models)}


def print_legend(symbols):
    print("\nLegend")
    if not symbols:
        print("  (no model usage)")
    for model, symbol in symbols.items():
        print(f"  {symbol}  {model}")


def print_timeline(buckets, model_buckets, symbols, since, interval, label):
    maximum, width = max(buckets, default=0), 40
    print(f"\nToken Usage Timeline ({label} intervals)\n")
    for index, tokens in enumerate(buckets):
        bar, cumulative = "", 0
        for model, symbol in symbols.items():
            cumulative += model_buckets[model][index]
            target = max(1, round(cumulative / maximum * width)) if cumulative else 0
            bar += symbol * max(0, target - len(bar))
        start = datetime.fromtimestamp(since + index * interval).astimezone()
        print(f"{start:%Y-%m-%d %H:%M}  {bar:<{width}}  {number(tokens)}")
    print_legend(symbols)


def compact_number(value):
    for threshold, suffix in ((1_000_000_000, "B"), (1_000_000, "M"), (1_000, "K")):
        if value >= threshold:
            return f"{value / threshold:.1f}".rstrip("0").rstrip(".") + suffix
    return str(round(value))


def print_chart(buckets, model_buckets, symbols, since, interval, label):
    height, maximum = 10, max(buckets, default=0)
    scales = [maximum * level / height for level in range(height, 0, -1)]
    axis_width = max(len(compact_number(scale)) for scale in scales)
    columns = []
    for index in range(len(buckets)):
        column, cumulative = [], 0
        for model, symbol in symbols.items():
            cumulative += model_buckets[model][index]
            target = math.ceil(cumulative / maximum * height) if cumulative else 0
            column.extend(symbol for _ in range(max(0, target - len(column))))
        columns.append(column)
    print(f"\nToken Usage Chart ({label} intervals)\n")
    for level, scale in zip(range(height, 0, -1), scales):
        bars = "".join(f" {column[level - 1]}" if len(column) >= level else "  " for column in columns).rstrip()
        print(f"{compact_number(scale) if maximum else '':>{axis_width}} |{bars}")
    print(f"{'0':>{axis_width}} +{'--' * len(buckets)}")
    first = datetime.fromtimestamp(since).astimezone()
    last = datetime.fromtimestamp(since + (len(buckets) - 1) * interval).astimezone()
    print(f"{'':>{axis_width + 2}} {first:%m-%d %H:%M}  ...  {last:%m-%d %H:%M}")
    print_legend(symbols)


def limit_window_label(minutes: int) -> str:
    if not minutes:
        return "?"
    if minutes % 1440 == 0:
        return f"{minutes // 1440}d"
    if minutes % 60 == 0:
        return f"{minutes // 60}h"
    return f"{minutes}m"


def print_cost_approximation(calls, used, limits=None):
    cost, skipped = 0.0, set()
    for call in calls:
        prices = MODEL_PRICES_PER_MILLION.get(call["model"])
        if prices is None:
            skipped.add(call["model"])
            continue
        tokens = call["usage"]
        counts = (tokens.get("input_tokens", 0), tokens.get("output_tokens", 0),
                  tokens.get("cache_read_input_tokens", 0), tokens.get("cache_creation_input_tokens", 0))
        cost += sum(count * price for count, price in zip(counts, prices)) / 1_000_000
    print(f"\nCost Approximation\n\nSelected usage cost:  ${cost:,.2f}")
    if used != "auto":
        print(f"Reported usage:       {used:g}%\nEstimated 100% value: ${cost * 100 / used:,.2f}")
    elif limits:
        for name, slot in limits.items():
            first, last = slot["first"], slot["last"]
            consumed = slot["consumed"]
            label = limit_window_label(last.get("window_minutes"))
            if first is last:
                span = f"{last.get('used_percent') or 0:g}% used (single snapshot)"
            elif slot["resets"]:
                plural = "s" if slot["resets"] > 1 else ""
                span = f"{consumed:g}% used (across {slot['resets']} window reset{plural})"
            else:
                span = f"{consumed:g}% used"
            estimate = f", estimated 100% is ${cost * 100 / consumed:,.2f}" if consumed > 0 else ""
            print(f"{name.capitalize()} limit ({label} window): {span}{estimate}")
    else:
        print("No provider-reported usage found; showing cost without extrapolation")
    if skipped:
        print(f"Skipped as free:      {', '.join(sorted(skipped))}")


def projects_root(value: str | None) -> Path:
    if value:
        return Path(value).expanduser()
    return Path(os.environ.get("CLAUDE_CONFIG_DIR", Path.home() / ".claude")) / "projects"


def read_lines(path: Path) -> list[dict]:
    result = []
    with path.open(encoding="utf-8") as handle:
        for line in handle:
            try:
                if line.strip():
                    result.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return result


def claude_title(lines: list[dict]) -> str:
    for entry in reversed(lines):
        if entry.get("type") == "ai-title" and entry.get("aiTitle"):
            return entry["aiTitle"]
    for entry in lines:
        content = entry.get("message", {}).get("content") if entry.get("type") == "user" else None
        if isinstance(content, str) and content.strip():
            return " ".join(content.split())
    return "(untitled)"


def claude_recent_sessions(root: Path) -> list[dict]:
    files = [path for path in root.glob("*/*.jsonl") if path.stat().st_size]
    files.sort(key=lambda path: path.stat().st_mtime, reverse=True)
    return [{"path": path, "id": path.stem, "project": path.parent.name.split("-")[-1] or path.parent.name,
             "title": claude_title(read_lines(path)), "updated": path.stat().st_mtime} for path in files[:10]]


def print_claude_sessions(rows: list[dict], label: str = "Claude Code") -> None:
    if not rows:
        raise SystemExit(f"No {label} sessions found")
    print(f"\nRecent {label} sessions:\n")
    for index, row in enumerate(rows, 1):
        updated = datetime.fromtimestamp(row["updated"]).astimezone().strftime("%Y-%m-%d %H:%M")
        print(f"{index:>2}. {updated}  [{row['project']}] {short_title(row['title'])}  ({row['id']})")


def choose_claude_session(rows: list[dict], label: str = "Claude Code") -> Path:
    print_claude_sessions(rows, label)
    while True:
        try:
            selected = input("\nSelect session [1-10, q to quit]: ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            raise SystemExit(1)
        if selected.lower() in {"q", "quit"}:
            raise SystemExit(0)
        if selected.isdigit() and 1 <= int(selected) <= len(rows):
            return rows[int(selected) - 1]["path"]
        print(f"Enter a number from 1 to {len(rows)}.")


def find_claude_session(root: Path, value: str) -> Path:
    matches = [path for path in root.glob("*/*.jsonl") if path.stem.startswith(value)]
    if not matches:
        raise SystemExit(f"Session not found: {value}")
    if len(matches) > 1:
        raise SystemExit(f"Ambiguous session id {value}:\n" + "\n".join(f"  {path}" for path in matches))
    return matches[0]


def claude_subagent_title(path: Path) -> str:
    meta_path = path.parent / (path.stem + ".meta.json")
    if meta_path.exists():
        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
            return f"[{meta.get('agentType') or 'agent'}] {meta.get('description') or ''}".strip()
        except (json.JSONDecodeError, OSError):
            pass
    return "(subagent)"


def claude_tree(path: Path, since=None, until=None) -> list[dict]:
    result = []
    def visit(file: Path, depth: int, title=None):
        all_lines = read_lines(file)
        lines = all_lines
        if since is not None:
            lines = [entry for entry in all_lines
                     if (timestamp := parse_timestamp(entry.get("timestamp"))) is not None
                     and since <= timestamp <= until]
        result.append({"path": file, "id": file.stem, "depth": depth,
                       "title": title or claude_title(all_lines), "lines": lines})
        for child in sorted((file.parent / file.stem / "subagents").glob("agent-*.jsonl")):
            visit(child, depth + 1, claude_subagent_title(child))
    visit(path, 0)
    return result


def claude_sessions_in_range(root: Path, since: float, until: float) -> list[dict]:
    result = []
    for path in sorted(root.glob("*/*.jsonl")):
        sessions = claude_tree(path, since, until)
        if any(session["lines"] for session in sessions):
            result.extend(session for session in sessions if session["lines"] or session["depth"] == 0)
    return result


def claude_calls(lines: list[dict]) -> list[dict]:
    calls_by_id = {}
    for entry in lines:
        message = entry.get("message", {})
        usage_data = message.get("usage")
        model = message.get("model", "unknown")
        if entry.get("type") != "assistant" or not usage_data or model == "<synthetic>":
            continue
        key = entry.get("requestId") or message.get("id") or entry.get("uuid")
        calls_by_id[key] = {"id": key or "(unknown)", "model": model, "usage": usage_data,
                            "timestamp": parse_timestamp(entry.get("timestamp"))}
    return list(calls_by_id.values())


def block_type(entry: dict):
    content = entry.get("message", {}).get("content")
    if entry.get("type") == "assistant":
        if isinstance(content, list) and content:
            if content[0].get("type") == "thinking":
                return "reasoning"
            if content[0].get("type") == "tool_use":
                return "tool_use"
        return "text"
    if entry.get("type") == "user" and isinstance(content, list) and any(
            isinstance(item, dict) and item.get("type") == "tool_result" for item in content):
        return "tool_result"
    return None


def claude_time_stats(lines: list[dict]) -> dict:
    stats, timestamps, previous = empty_time_stats(), [], None
    for entry in lines:
        timestamp = parse_timestamp(entry.get("timestamp"))
        if timestamp is None:
            continue
        timestamps.append(timestamp)
        kind = block_type(entry)
        delta = max(0.0, timestamp - previous) if previous is not None else 0.0
        if kind in ("text", "reasoning", "tool_use"):
            stats["text" if kind == "tool_use" else kind] += delta
            model = entry.get("message", {}).get("model", "unknown")
            stats["per_model_think"][model] += delta
            if kind == "tool_use":
                content = entry.get("message", {}).get("content")
                name = content[0].get("name", "") if content else ""
                stats["sub_calls" if name in ("Task", "Agent") else "tool_calls"] += 1
        elif kind == "tool_result":
            stats["tools"] += delta
        previous = timestamp
    if timestamps:
        stats["wall"] = max(timestamps) - min(timestamps)
    return stats


def claude_bounds(sessions: list[dict]) -> tuple[float, float]:
    timestamps = [timestamp for session in sessions for entry in session["lines"]
                  if (timestamp := parse_timestamp(entry.get("timestamp"))) is not None]
    if not timestamps:
        raise SystemExit("Selected session has no timestamped messages")
    return min(timestamps), max(timestamps)


CODEX_ROLLOUT_STEM = re.compile(r"^rollout-\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}-(.+)$")


def codex_root(value: str | None) -> Path:
    if value:
        return Path(value).expanduser()
    return Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")) / "sessions"


def codex_session_id(path: Path) -> str:
    match = CODEX_ROLLOUT_STEM.match(path.stem)
    return match.group(1) if match else path.stem


def codex_files(root: Path) -> list[Path]:
    return [path for path in root.rglob("rollout-*.jsonl") if path.stat().st_size]


def codex_title(lines: list[dict]) -> str:
    for entry in lines:
        payload = entry.get("payload") or {}
        if entry.get("type") == "event_msg" and payload.get("type") == "user_message":
            message = payload.get("message") or ""
            if message.strip():
                return " ".join(message.split())
    return "(untitled)"


def codex_project(lines: list[dict]) -> str:
    for entry in lines:
        if entry.get("type") == "session_meta":
            cwd = (entry.get("payload") or {}).get("cwd") or ""
            return Path(cwd).name if cwd else ""
    return ""


def codex_recent_sessions(root: Path) -> list[dict]:
    files = codex_files(root)
    files.sort(key=lambda path: path.stat().st_mtime, reverse=True)
    result = []
    for path in files[:10]:
        lines = read_lines(path)
        result.append({"path": path, "id": codex_session_id(path), "project": codex_project(lines),
                       "title": codex_title(lines), "updated": path.stat().st_mtime})
    return result


def find_codex_session(root: Path, value: str) -> Path:
    matches = [path for path in codex_files(root) if codex_session_id(path).startswith(value)]
    if not matches:
        raise SystemExit(f"Session not found: {value}")
    if len(matches) > 1:
        raise SystemExit(f"Ambiguous session id {value}:\n" + "\n".join(f"  {path}" for path in matches))
    return matches[0]


def codex_tree(path: Path, since=None, until=None) -> list[dict]:
    all_lines = read_lines(path)
    lines = all_lines
    if since is not None:
        lines = [entry for entry in all_lines
                 if (timestamp := parse_timestamp(entry.get("timestamp"))) is not None
                 and since <= timestamp <= until]
    return [{"path": path, "id": codex_session_id(path), "depth": 0,
             "title": codex_title(all_lines), "lines": lines}]


def codex_sessions_in_range(root: Path, since: float, until: float) -> list[dict]:
    result = []
    for path in sorted(codex_files(root)):
        sessions = codex_tree(path, since, until)
        if any(session["lines"] for session in sessions):
            result.extend(sessions)
    return result


def codex_calls(lines: list[dict]) -> list[dict]:
    """Turn token_count events into per-call usage records.

    Each token_count event carrying an `info` payload corresponds to one API
    call; its `last_token_usage` values sum exactly to the session's final
    `total_token_usage`. Codex counts cached tokens inside `input_tokens`
    (OpenAI convention), so they are subtracted to match the Anthropic-style
    split used everywhere else in this script.
    """
    result, provider, model = [], "openai", "unknown"
    for entry in lines:
        payload = entry.get("payload") or {}
        kind = entry.get("type")
        if kind == "session_meta":
            provider = payload.get("model_provider") or provider
        elif kind == "turn_context":
            model = payload.get("model") or model
        elif kind == "event_msg" and payload.get("type") == "token_count":
            last = (payload.get("info") or {}).get("last_token_usage") or {}
            cached = last.get("cached_input_tokens", 0)
            usage_data = {"input_tokens": max(0, last.get("input_tokens", 0) - cached),
                          "cache_read_input_tokens": cached,
                          "cache_creation_input_tokens": 0,
                          "output_tokens": last.get("output_tokens", 0)}
            if not any(usage_data.values()):
                continue
            result.append({"id": f"call-{len(result) + 1}", "model": f"{provider}/{model}",
                           "usage": usage_data, "timestamp": parse_timestamp(entry.get("timestamp"))})
    return result


def codex_rate_limit_snapshots(lines: list[dict]) -> list[tuple]:
    """Collect provider-reported quota snapshots as (timestamp, name, window).

    Codex attaches account-wide rolling-window rate limits (primary and,
    when the plan has one, secondary) to token_count events. Each snapshot
    reflects cumulative usage of its window at that moment, so the percent
    consumed within a report's timeframe is the difference between the first
    and last snapshot in it.
    """
    result = []
    for entry in lines:
        payload = entry.get("payload") or {}
        if entry.get("type") != "event_msg" or payload.get("type") != "token_count":
            continue
        rate = payload.get("rate_limits") or {}
        timestamp = parse_timestamp(entry.get("timestamp")) or 0
        for name in ("primary", "secondary"):
            if rate.get(name):
                result.append((timestamp, name, rate[name]))
    return result


def codex_quota_usage(snapshots: list[tuple]) -> dict:
    """Reduce quota snapshots to the percent consumed per limit window.

    used_percent grows within a provider window and drops when the window
    resets, so consumption across the whole timeframe is the sum of each
    monotonic segment's growth. A segment after a reset counts from zero:
    whatever was consumed before its first snapshot still happened inside
    the selected range.
    """
    limits = {}
    for _, name, window in sorted(snapshots, key=lambda snapshot: snapshot[0]):
        percent = window.get("used_percent") or 0
        slot = limits.get(name)
        if slot is None:
            limits[name] = {"first": window, "last": window, "segment_start": percent,
                            "consumed": 0.0, "resets": 0}
            continue
        previous = slot["last"].get("used_percent") or 0
        if percent < previous:
            slot["consumed"] += previous - slot["segment_start"]
            slot["segment_start"] = 0.0
            slot["resets"] += 1
        slot["last"] = window
    for slot in limits.values():
        slot["consumed"] += (slot["last"].get("used_percent") or 0) - slot["segment_start"]
    return limits


def codex_block_type(entry: dict):
    if entry.get("type") != "response_item":
        return None
    payload = entry.get("payload") or {}
    kind = payload.get("type") or ""
    if kind == "reasoning":
        return "reasoning"
    if kind == "message":
        return "text" if payload.get("role") == "assistant" else None
    # Covers custom_tool_call, function_call, local_shell_call, and their outputs.
    if kind.endswith("_call_output"):
        return "tool_result"
    if kind.endswith("_call"):
        return "tool_use"
    return None


def codex_time_stats(lines: list[dict]) -> dict:
    stats, timestamps, previous = empty_time_stats(), [], None
    provider, model = "openai", "unknown"
    for entry in lines:
        payload = entry.get("payload") or {}
        if entry.get("type") == "session_meta":
            provider = payload.get("model_provider") or provider
        elif entry.get("type") == "turn_context":
            model = payload.get("model") or model
        timestamp = parse_timestamp(entry.get("timestamp"))
        if timestamp is None:
            continue
        timestamps.append(timestamp)
        kind = codex_block_type(entry)
        delta = max(0.0, timestamp - previous) if previous is not None else 0.0
        if kind in ("text", "reasoning", "tool_use"):
            stats["text" if kind == "tool_use" else kind] += delta
            stats["per_model_think"][f"{provider}/{model}"] += delta
            if kind == "tool_use":
                stats["tool_calls"] += 1
        elif kind == "tool_result":
            stats["tools"] += delta
        previous = timestamp
    if timestamps:
        stats["wall"] = max(timestamps) - min(timestamps)
    return stats


def database_path(value: str | None) -> Path:
    data = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")) / "opencode"
    configured = value or os.environ.get("OPENCODE_DB")
    if configured:
        path = Path(configured).expanduser()
        return path if path.is_absolute() else data / path
    default = data / "opencode.db"
    if default.exists():
        return default
    candidates = sorted(data.glob("opencode*.db"), key=lambda path: path.stat().st_mtime, reverse=True)
    return candidates[0] if candidates else default


def open_database(path: Path) -> sqlite3.Connection:
    if not path.is_file():
        raise SystemExit(f"OpenCode database not found: {path}")
    connection = sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
    connection.row_factory = sqlite3.Row
    return connection


def table_exists(connection: sqlite3.Connection, name: str) -> bool:
    return connection.execute(
        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (name,)
    ).fetchone() is not None


def session_rows(connection: sqlite3.Connection) -> list[dict]:
    return [dict(row) for row in connection.execute(
        "SELECT id, parent_id, title, directory, time_created, time_updated FROM session"
    ).fetchall()]


def recent_sessions(connection: sqlite3.Connection) -> list[dict]:
    rows = sorted((row for row in session_rows(connection) if row["parent_id"] is None),
                  key=lambda row: row["time_updated"], reverse=True)[:10]
    return [{"id": row["id"], "title": row["title"] or "(untitled)",
             "project": row["directory"] or "", "updated": row["time_updated"] / 1000}
            for row in rows]


def choose_session(rows: list[dict]) -> str:
    print_sessions(rows)
    while True:
        try:
            selected = input("\nSelect session [1-10, q to quit]: ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            raise SystemExit(1)
        if selected.lower() in {"q", "quit"}:
            raise SystemExit(0)
        if selected.isdigit() and 1 <= int(selected) <= len(rows):
            return rows[int(selected) - 1]["id"]
        print(f"Enter a number from 1 to {len(rows)}.")


def print_sessions(rows: list[dict]) -> None:
    if not rows:
        raise SystemExit("No OpenCode sessions found")
    print("\nRecent OpenCode sessions:\n")
    for index, row in enumerate(rows, 1):
        updated = datetime.fromtimestamp(row["updated"]).astimezone().strftime("%Y-%m-%d %H:%M")
        print(f"{index:>2}. {updated}  {row['title']}  ({row['id']})")


def find_session(connection: sqlite3.Connection, value: str) -> str:
    rows = connection.execute("SELECT id FROM session WHERE id LIKE ?", (value + "%",)).fetchall()
    if not rows:
        raise SystemExit(f"Session not found: {value}")
    if len(rows) > 1:
        raise SystemExit(f"Ambiguous session id {value}:\n" + "\n".join(f"  {row['id']}" for row in rows))
    return rows[0]["id"]


def tree(connection: sqlite3.Connection, root: str) -> list[dict]:
    rows = connection.execute("""
        WITH RECURSIVE tree(id, depth) AS (
          SELECT id, 0 FROM session WHERE id = ?
          UNION ALL
          SELECT session.id, tree.depth + 1 FROM session JOIN tree ON session.parent_id = tree.id
        )
        SELECT session.*, tree.depth FROM session JOIN tree ON session.id = tree.id
        ORDER BY tree.depth, session.time_created
    """, (root,)).fetchall()
    return [{"path": row["id"], "id": row["id"], "depth": row["depth"],
             "title": row["title"] or "(untitled)", "row": dict(row)} for row in rows]


def messages(connection: sqlite3.Connection, session_ids: list[str], since=None, until=None) -> list[dict]:
    if not session_ids or not table_exists(connection, "message"):
        return []
    marks = ",".join("?" for _ in session_ids)
    result = []
    for row in connection.execute(
            f"SELECT id, session_id, time_created, time_updated, data FROM message WHERE session_id IN ({marks})",
            session_ids):
        try:
            data = json.loads(row["data"])
        except (TypeError, json.JSONDecodeError):
            continue
        timestamp = row["time_created"] / 1000
        if data.get("role") != "assistant" or "tokens" not in data:
            continue
        if since is not None and not since <= timestamp <= until:
            continue
        result.append({"id": row["id"], "sessionID": row["session_id"],
                       "time_created": row["time_created"], "time_updated": row["time_updated"], **data})
    return result


def sessions_in_range(connection: sqlite3.Connection, since: float, until: float) -> list[dict]:
    rows = session_rows(connection)
    by_id = {row["id"]: row for row in rows}
    matching = {message["sessionID"] for message in messages(connection, list(by_id), since, until)}
    selected = set(matching)
    for session_id in matching:
        row = by_id.get(session_id)
        while row and row["parent_id"]:
            selected.add(row["parent_id"])
            row = by_id.get(row["parent_id"])
    result = []
    for row in sorted((by_id[value] for value in selected), key=lambda item: (item["time_created"], item["id"])):
        depth, parent = 0, row["parent_id"]
        while parent in selected:
            depth += 1
            parent = by_id[parent]["parent_id"]
        result.append({"path": row["id"], "id": row["id"], "depth": depth,
                       "title": row["title"] or "(untitled)", "row": row})
    return result


def model_name(message: dict) -> str:
    return f"{message.get('providerID', 'unknown')}/{message.get('modelID', 'unknown')}"


def calls(messages: list[dict]) -> list[dict]:
    result = []
    for message in messages:
        tokens = message["tokens"]
        cache = tokens.get("cache") or {}
        usage = {"input_tokens": tokens.get("input", 0),
                 "cache_read_input_tokens": cache.get("read", 0),
                 "cache_creation_input_tokens": cache.get("write", 0),
                 "output_tokens": tokens.get("output", 0) + tokens.get("reasoning", 0)}
        result.append({"id": message["id"], "model": model_name(message), "usage": usage,
                        "timestamp": message["time_created"] / 1000})
    return result


def opencode_time_stats(connection, assistant_messages: list[dict], since, until) -> dict:
    stats = {message["sessionID"]: empty_time_stats() for message in assistant_messages}
    if not assistant_messages or not table_exists(connection, "part"):
        return stats
    by_id = {message["id"]: message for message in assistant_messages}
    marks = ",".join("?" for _ in by_id)
    low = since * 1000 if since is not None else None
    high = until * 1000 if until is not None else None
    for row in connection.execute(
            f"SELECT message_id, time_created, time_updated, data FROM part WHERE message_id IN ({marks})",
            list(by_id)):
        message = by_id[row["message_id"]]
        start, end = row["time_created"], max(row["time_created"], row["time_updated"])
        if low is not None:
            start = max(start, low)
        if high is not None:
            end = min(end, high)
        if end <= start:
            continue
        try:
            data = json.loads(row["data"])
        except (TypeError, json.JSONDecodeError):
            continue
        current, duration = stats[message["sessionID"]], (end - start) / 1000
        if data.get("type") == "text":
            current["text"] += duration
            current["per_model_think"][model_name(message)] += duration
        elif data.get("type") == "reasoning":
            current["reasoning"] += duration
            current["per_model_think"][model_name(message)] += duration
        elif data.get("type") == "tool":
            current["tools"] += duration
            current["sub_calls" if data.get("tool") == "task" else "tool_calls"] += 1
        current.setdefault("intervals", []).append((start, end))
    for current in stats.values():
        intervals = current.pop("intervals", [])
        if intervals:
            current["wall"] = (max(end for _, end in intervals) - min(start for start, _ in intervals)) / 1000
    return stats


def opencode_bounds(connection: sqlite3.Connection, session_ids: list[str]) -> tuple[float, float]:
    marks = ",".join("?" for _ in session_ids)
    row = connection.execute(
        f"SELECT MIN(time_created), MAX(time_updated) FROM message WHERE session_id IN ({marks})",
        session_ids,
    ).fetchone()
    if not row or row[0] is None:
        raise SystemExit("Selected session has no timestamped messages")
    return row[0] / 1000, row[1] / 1000


def render(location, path, args, sessions, session_calls, session_times, all_calls, since, until, interval, limits=None):
    interval_report = args.timeline or args.chart
    grouped = defaultdict(list)
    for values in session_calls.values():
        for call in values:
            grouped[call["model"]].append(call)
    think = defaultdict(float)
    for stats in session_times.values():
        for model, seconds in stats["per_model_think"].items():
            think[model] += seconds
    model_rows, model_totals = [], []
    keys = ("calls", "input", "cache_read", "cache_write", "output", "total")
    for model, values in grouped.items():
        totals = usage(values)
        model_totals.append(totals)
        model_rows.append([model, humanize_seconds(think[model]), *[number(totals[key]) for key in keys]])
    model_rows.sort(key=lambda row: int(row[-1].replace(",", "")), reverse=True)
    totals = add_usage(model_totals)
    model_rows.append(["TOTAL", humanize_seconds(sum(think.values())), *[number(totals[key]) for key in keys]])
    per_session, session_rows, message_rows = [], [], []
    for session in sessions:
        values, totals, stats = session_calls[session["path"]], usage(session_calls[session["path"]]), session_times[session["path"]]
        per_session.append(totals)
        models = sorted({call["model"] for call in values})
        session_rows.append([f"{'  ' * session['depth']}{short_title(session['title'])} ({short_id(session['id'])})",
                             ", ".join(models) or "-", humanize_seconds(stats["text"] + stats["reasoning"]),
                             *[number(totals[key]) for key in keys]])
        for call in values:
            call_totals = usage([call])
            timestamp = call.get("timestamp")
            updated = datetime.fromtimestamp(timestamp).astimezone().strftime("%Y-%m-%d %H:%M:%S") if timestamp is not None else "-"
            message_rows.append((timestamp or 0, [
                updated,
                short_id(str(call.get("id", "(unknown)")), 18),
                f"{short_title(session['title'], 30)} ({short_id(session['id'])})",
                call["model"],
                *[number(call_totals[key]) for key in ("input", "cache_read", "cache_write", "output", "total")],
            ]))
    total = add_usage(per_session)
    session_rows.append(["TOTAL", "", humanize_seconds(sum(stats["text"] + stats["reasoning"] for stats in session_times.values())), *[number(total[key]) for key in keys]])
    print(f"\n{location}: {path}")
    if since is not None:
        roots = sum(session["depth"] == 0 for session in sessions)
        print(f"Range:     {args.since} through {args.until}")
        if not interval_report:
            print(f"Sessions:  {roots} roots, {len(sessions) - roots} subagents")
    else:
        print(f"Session:   {sessions[0]['title']}\nID:        {sessions[0]['id']}\nTree:      {len(sessions)} sessions ({len(sessions) - 1} descendants)")

    if interval_report:
        buckets, model_buckets = token_buckets(all_calls, since, until, interval)
        symbols = model_symbols(model_buckets)
        if args.timeline:
            print_timeline(buckets, model_buckets, symbols, since, interval, args.interval)
        if args.chart:
            print_chart(buckets, model_buckets, symbols, since, interval, args.interval)
    if args.models:
        print("\nUsage By Model\n")
        print_table(["Model", "Think", "Calls", "Input", "Cache read", "Cache write", "Output", "Total"], model_rows, {1, 2, 3, 4, 5, 6, 7})

    if args.messages:
        rows = [row for _, row in sorted(message_rows, key=lambda item: item[0])]
        rows.append(["TOTAL", "", "", "", *[number(total[key]) for key in ("input", "cache_read", "cache_write", "output", "total")]])
        print("\nUsage By Message\n")
        print_table(
            ["Time", "Message", "Session", "Model", "Input", "Cache read", "Cache write", "Output", "Total"],
            rows,
            {4, 5, 6, 7, 8},
        )
    if args.sessions:
        print("\nUsage By Session\n")
        print_table(["Session", "Model", "Think", "Calls", "Input", "Cache read", "Cache write", "Output", "Total"], session_rows, {2, 3, 4, 5, 6, 7, 8})
    if args.cache:
        print_cache_hit_rates(session_calls)

    if not interval_report:
        input_tokens = total["input"] + total["cache_read"] + total["cache_write"]
        print(f"\nTotals\n\nInput:             {number(total['input'])}\nCached input:      {number(total['cache_read'])}\nCache writes:      {number(total['cache_write'])}\nOutput:            {number(total['output'])}\nOverall tokens:    {number(total['total'])}\nInput cache rate:  {(total['cache_read'] / input_tokens * 100 if input_tokens else 0):.2f}%")
    if args.usage is not None:
        print_cost_approximation(all_calls, args.usage, limits)
    if args.time:
        rows, total_text, total_reason, total_tools = [], 0.0, 0.0, 0.0
        sub_calls = tool_calls = 0
        for session in sessions:
            stats = session_times[session["path"]]
            covered = stats["text"] + stats["reasoning"] + stats["tools"]
            total_text += stats["text"]
            total_reason += stats["reasoning"]
            total_tools += stats["tools"]
            sub_calls += stats["sub_calls"]
            tool_calls += stats["tool_calls"]
            rows.append([f"{'  ' * session['depth']}{short_title(session['title'])} ({short_id(session['id'])})", humanize_seconds(stats["text"]), humanize_seconds(stats["reasoning"]), f"{humanize_seconds(stats['tools'])} ({stats['tool_calls']})", str(stats["sub_calls"]), humanize_seconds(stats["wall"]), humanize_seconds(max(0, stats["wall"] - covered))])
        rows.append(["TOTAL", humanize_seconds(total_text), humanize_seconds(total_reason), f"{humanize_seconds(total_tools)} ({tool_calls})", str(sub_calls), "-", "-"])
        print("\nTime Breakdown By Session\n")
        print_table(["Session", "text", "reasoning", "tools", "subagents", "wall", "unaccounted"], rows, {1, 2, 3, 4, 5, 6})


def main():
    parser = argparse.ArgumentParser(description="Show token usage from Claude Code or OpenCode")
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument("--claude", action="store_true", help="Read Claude Code JSONL transcripts")
    source.add_argument("--codex", action="store_true", help="Read Codex CLI JSONL rollouts")
    source.add_argument("--opencode", action="store_true", help="Read the OpenCode SQLite database")
    parser.add_argument("--dir", help="Claude projects or Codex sessions directory")
    parser.add_argument("--db", help="OpenCode SQLite database path")
    parser.add_argument("-s", "--session", help="Session ID or unique prefix")
    parser.add_argument("--since", help="Start of range (ISO value or relative duration such as 1d)")
    parser.add_argument("--until", help="End of range (ISO value or duration ago; default: one hour from now)")
    parser.add_argument("--timeline", action="store_true", help="Show horizontal token bars by interval")
    parser.add_argument("--chart", action="store_true", help="Show a vertical token chart by interval")
    parser.add_argument("--interval", help="Override the automatic chart interval, such as 10m, 1h, or 1d")
    parser.add_argument("--usage", nargs="?", const="auto", metavar="PERCENT",
                        help="Estimate 100%% value from quota usage (omit the percent to use provider-reported usage)")
    parser.add_argument("--messages", action="store_true", help="Show usage per API message")
    parser.add_argument("--sessions", action="store_true", help="Show usage per session")
    parser.add_argument("--models", action="store_true", help="Show usage per model")
    parser.add_argument("--cache", action="store_true", help="Show cache hit rates by message gap")
    parser.add_argument("-t", "--time", action="store_true", help="Show time breakdown by session")
    parser.add_argument("-e", "--export", metavar="FILENAME", help="Also save the report to a file")
    parser.add_argument("-l", "--list", action="store_true", help="List recent sessions and exit")
    args = parser.parse_args()
    if args.usage not in (None, "auto"):
        try:
            args.usage = float(args.usage)
        except ValueError:
            parser.error(f"--usage expects a number, got: {args.usage}")
        if not 0 < args.usage <= 100:
            parser.error("--usage must be greater than 0 and no more than 100")
    if args.since and not args.until:
        args.until = (datetime.now().astimezone() + timedelta(hours=1)).isoformat(timespec="seconds")
    interval_report = args.timeline or args.chart
    if args.session and (args.since or args.until):
        parser.error("--session cannot be combined with --since/--until")
    if interval_report and not args.session and not any((args.since, args.until, args.interval)):
        now = datetime.now().astimezone()
        start = datetime.combine(now.date() - timedelta(days=1), time.min, now.tzinfo)
        args.since, args.until = start.isoformat(timespec="minutes"), now.isoformat(timespec="minutes")
    elif interval_report and not args.session and not (args.since and args.until):
        parser.error("timeline/chart mode requires --since (or --session)")
    elif not interval_report and bool(args.since) != bool(args.until):
        parser.error("--since and --until must be provided together")
    if args.interval and not interval_report:
        parser.error("--interval requires --timeline or --chart")
    try:
        since = parse_range_bound(args.since) if args.since else None
        until = parse_range_bound(args.until, end=True) if args.until else None
        interval = parse_interval(args.interval) if args.interval else None
    except ValueError as error:
        parser.error(str(error))
    if since is not None and since > until:
        parser.error("--since must not be later than --until")
    if interval_report and since is not None and interval is None:
        interval, args.interval = default_interval(until - since)

    if args.claude or args.codex:
        if args.db:
            parser.error("--db is only valid with --opencode")
        if args.claude:
            label, root = "Claude Code", projects_root(args.dir)
            recent_fn, find_fn, tree_fn = claude_recent_sessions, find_claude_session, claude_tree
            range_fn, calls_fn, times_fn = claude_sessions_in_range, claude_calls, claude_time_stats
        else:
            label, root = "Codex", codex_root(args.dir)
            recent_fn, find_fn, tree_fn = codex_recent_sessions, find_codex_session, codex_tree
            range_fn, calls_fn, times_fn = codex_sessions_in_range, codex_calls, codex_time_stats
        if not root.is_dir():
            raise SystemExit(f"{label} sessions directory not found: {root}")
        if args.list:
            print_claude_sessions(recent_fn(root), label)
            return
        selected = find_fn(root, args.session) if args.session else (
            None if since is not None else choose_claude_session(recent_fn(root), label))
        sessions = range_fn(root, since, until) if since is not None else tree_fn(selected)
        if interval_report and args.session:
            since, until = claude_bounds(sessions)
            args.since = datetime.fromtimestamp(since).astimezone().isoformat(timespec="seconds")
            args.until = datetime.fromtimestamp(until).astimezone().isoformat(timespec="seconds")
            if interval is None:
                interval, args.interval = default_interval(until - since)
        if not sessions and not interval_report:
            raise SystemExit(f"No {label} messages found in the specified range")
        export = None
        try:
            if args.export:
                export = open(Path(args.export).expanduser(), "w", encoding="utf-8")
                sys.stdout = Tee(sys.stdout, export)
            session_calls = {session["path"]: calls_fn(session["lines"]) for session in sessions}
            session_times = {session["path"]: times_fn(session["lines"]) for session in sessions}
            all_calls = [call for values in session_calls.values() for call in values]
            limits = None
            if args.codex and args.usage == "auto":
                limits = codex_quota_usage([snapshot for session in sessions
                                            for snapshot in codex_rate_limit_snapshots(session["lines"])])
            report_path = root if selected is None else selected.parent
            render("Directory", report_path, args, sessions, session_calls, session_times,
                   all_calls, since, until, interval, limits)
        finally:
            if export is not None:
                sys.stdout = sys.__stdout__
                export.close()
        return

    if args.dir:
        parser.error("--dir is only valid with --claude or --codex")
    path = database_path(args.db)
    connection = open_database(path)
    export = None
    try:
        if args.list:
            print_sessions(recent_sessions(connection))
            return
        root = find_session(connection, args.session) if args.session else (None if since is not None else choose_session(recent_sessions(connection)))
        sessions = sessions_in_range(connection, since, until) if since is not None else tree(connection, root)
        ids = [session["id"] for session in sessions]
        if interval_report and args.session:
            since, until = opencode_bounds(connection, ids)
            args.since = datetime.fromtimestamp(since).astimezone().isoformat(timespec="seconds")
            args.until = datetime.fromtimestamp(until).astimezone().isoformat(timespec="seconds")
            if interval is None:
                interval, args.interval = default_interval(until - since)
        assistant_messages = messages(connection, ids, since, until)
        if not sessions and not interval_report:
            raise SystemExit("No OpenCode messages found in the specified range")
        if args.export:
            export = open(Path(args.export).expanduser(), "w", encoding="utf-8")
            sys.stdout = Tee(sys.stdout, export)
        by_session = defaultdict(list)
        for message in assistant_messages:
            by_session[message["sessionID"]].append(message)
        session_calls = {session["path"]: calls(by_session[session["id"]]) for session in sessions}
        raw_times = opencode_time_stats(connection, assistant_messages, since, until)
        session_times = {session["path"]: raw_times.get(session["id"], empty_time_stats()) for session in sessions}
        all_calls = [call for values in session_calls.values() for call in values]
        render("Database", path, args, sessions, session_calls, session_times, all_calls, since, until, interval)
    finally:
        if export is not None:
            sys.stdout = sys.__stdout__
            export.close()
        connection.close()


if __name__ == "__main__":
    main()