agent-skills/skills/nightshift/usage-window.py

#!/usr/bin/env python3
"""Estimate token usage inside the rolling 5-hour limit window.

The real quota is server-side and nothing local states it. What IS local is
every assistant message Claude Code has written, with a timestamp and a usage
block, in ~/.claude/projects/**/*.jsonl. Summing those over a rolling window
gives a usable proxy: not the quota, but a consistent measure of how hard the
window has been worked, plus when the oldest tokens in it age out.

SCOPE — read this before trusting the number:

  Counted:     every Claude Code session on THIS machine, all projects, not
               just the session asking. Parallel sessions in other repos land
               in the same window and this picks them up.
  NOT counted: Claude Code on any other machine, claude.ai web usage, direct
               API calls, and anything else that never writes a transcript here.

So the figure is a floor, never a ceiling. Real usage is this or higher, and
the gap is however much the account is being used elsewhere. Size your margin
accordingly: if the user works on several machines, treat a "comfortable"
reading with suspicion.

Treat the number as a trend, not a truth. It is here so a long autonomous run
can back off *before* an agent dies mid-task, rather than discovering the limit
by being killed by it. The reactive backstop still matters: an agent that dies
on a limit error is telling you the truth this script only estimates.

Usage:
    usage-window.py                 # human summary
    usage-window.py --json          # machine readable
    usage-window.py --hours 5       # window size (default 5)
"""

from __future__ import annotations

import argparse
import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path

PROJECTS = Path.home() / ".claude" / "projects"


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


def billable(usage: dict) -> int:
    """Tokens that plausibly count against a quota.

    Cache reads are excluded: they are the cheap path and counting them would
    make a long cached session look far heavier than it is. Cache *creation* is
    included, because writing the cache is real work.
    """
    return (
        int(usage.get("input_tokens") or 0)
        + int(usage.get("cache_creation_input_tokens") or 0)
        + int(usage.get("output_tokens") or 0)
    )


def collect(hours: float) -> dict:
    now = datetime.now(timezone.utc)
    cutoff = now - timedelta(hours=hours)

    total = 0
    output = 0
    messages = 0
    oldest: datetime | None = None
    newest: datetime | None = None
    per_project: dict[str, int] = {}

    if not PROJECTS.is_dir():
        return {
            "error": f"no transcript directory at {PROJECTS}",
            "windowHours": hours,
        }

    for path in PROJECTS.glob("*/*.jsonl"):
        # Cheap skip: a file untouched since the cutoff cannot contribute.
        try:
            if datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) < cutoff:
                continue
        except OSError:
            continue

        project = path.parent.name
        try:
            with path.open(encoding="utf-8", errors="replace") as handle:
                for line in handle:
                    if '"usage"' not in line:
                        continue
                    try:
                        entry = json.loads(line)
                    except json.JSONDecodeError:
                        continue
                    stamp = parse_ts(entry.get("timestamp"))
                    if stamp is None or stamp < cutoff:
                        continue
                    usage = (entry.get("message") or {}).get("usage")
                    if not isinstance(usage, dict):
                        continue

                    tokens = billable(usage)
                    if tokens <= 0:
                        continue

                    total += tokens
                    output += int(usage.get("output_tokens") or 0)
                    messages += 1
                    per_project[project] = per_project.get(project, 0) + tokens
                    oldest = stamp if oldest is None or stamp < oldest else oldest
                    newest = stamp if newest is None or stamp > newest else newest
        except OSError:
            continue

    # When the oldest tokens in the window age out. This is the soonest the
    # window measurably loosens; it is NOT a quota reset time, which only the
    # server knows.
    ages_out_in = None
    if oldest is not None:
        ages_out_in = max(0, int(((oldest + timedelta(hours=hours)) - now).total_seconds()))

    return {
        "windowHours": hours,
        "tokens": total,
        "outputTokens": output,
        "messages": messages,
        "projects": dict(sorted(per_project.items(), key=lambda kv: -kv[1])),
        "oldestInWindow": oldest.isoformat() if oldest else None,
        "newestInWindow": newest.isoformat() if newest else None,
        "oldestAgesOutInSeconds": ages_out_in,
        "now": now.isoformat(),
    }


def human(report: dict) -> str:
    if "error" in report:
        return f"usage: unavailable ({report['error']})"

    lines = [
        f"rolling {report['windowHours']}h window — all local Claude Code sessions",
        f"  tokens   {report['tokens']:,} ({report['outputTokens']:,} output) "
        f"over {report['messages']:,} messages",
    ]
    if report["oldestAgesOutInSeconds"] is not None:
        secs = report["oldestAgesOutInSeconds"]
        lines.append(f"  oldest entry ages out in {secs // 60}m{secs % 60:02d}s")
    top = list(report["projects"].items())[:3]
    if top:
        lines.append("  busiest: " + ", ".join(f"{k} {v:,}" for k, v in top))
    lines.append(f"  sessions: {len(report['projects'])} project(s) contributing")
    lines.append(
        "  FLOOR, not a quota: misses other machines, claude.ai and direct API use"
    )
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--hours", type=float, default=5.0)
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    report = collect(args.hours)
    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print(human(report))


if __name__ == "__main__":
    main()