subagent-cache-rewrites.py

#!/usr/bin/env python3
"""Measure prompt-cache re-writes in Claude Code subagent transcripts.

A subagent that sits in a tool call longer than its cache TTL re-writes its
whole context on the next request. This reports how often that happens, how
many tokens it costs, which TTL the writes used, and what the agent was
blocked in.

    subagent-cache-rewrites.py                 # all history
    subagent-cache-rewrites.py --since 2026-09-19
"""

from __future__ import annotations

import argparse
import json
from collections import Counter, defaultdict
from collections.abc import Iterator
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path

TRANSCRIPTS = "*/*/subagents/agent-*.jsonl"
GAP_5M = 300
GAP_1H = 3600
MIN_CONTEXT = 20_000  # ignore re-writes of contexts too small to matter
REWRITE_SHARE = 0.5  # a re-write is a request that writes most of its context


@dataclass(frozen=True)
class Request:
    """One API request made by a subagent."""

    time: datetime
    gap: float  # seconds since the agent's previous transcript entry
    first: bool
    written: int
    written_5m: int
    written_1h: int
    context: int
    blocked_in: str | None  # tool call the agent was in before this request
    agent: str
    model: str
    project: str

    @property
    def is_rewrite(self) -> bool:
        return (
            not self.first
            and self.context > MIN_CONTEXT
            and self.written > REWRITE_SHARE * self.context
        )


def parse_time(stamp: str) -> datetime:
    return datetime.fromisoformat(stamp.replace("Z", "+00:00"))


def entries(path: Path) -> list[dict]:
    rows = []
    with path.open() as lines:
        for line in lines:
            try:
                row = json.loads(line)
            except json.JSONDecodeError:
                continue
            if row.get("type") in ("assistant", "user") and "timestamp" in row:
                rows.append(row)
    return sorted(rows, key=lambda row: row["timestamp"])


def tool_name(block: dict) -> str:
    background = block["name"] == "Bash" and (block.get("input") or {}).get("run_in_background")
    return "Bash(bg)" if background else block["name"]


def requests(path: Path, projects: Path) -> Iterator[Request]:
    project = path.relative_to(projects).parts[0]
    seen: set[str] = set()
    previous: datetime | None = None
    blocked_in: str | None = None
    for row in entries(path):
        time = parse_time(row["timestamp"])
        if row["type"] == "user":
            previous = previous or time
            continue
        message = row.get("message") or {}
        usage = message.get("usage")
        key = message.get("id") or row.get("requestId") or row["timestamp"]
        # Streaming repeats one request's usage on every content block.
        if usage and key not in seen:
            seen.add(key)
            written = usage.get("cache_creation_input_tokens") or 0
            split = usage.get("cache_creation") or {}
            yield Request(
                time=time,
                gap=(time - previous).total_seconds() if previous else 0.0,
                first=len(seen) == 1,
                written=written,
                written_5m=split.get("ephemeral_5m_input_tokens") or 0,
                written_1h=split.get("ephemeral_1h_input_tokens") or 0,
                context=written
                + (usage.get("cache_read_input_tokens") or 0)
                + (usage.get("input_tokens") or 0),
                blocked_in=blocked_in,
                agent=row.get("attributionAgent", "?"),
                model=message.get("model", "?"),
                project=project,
            )
        content = message.get("content")
        for block in content if isinstance(content, list) else []:
            if isinstance(block, dict) and block.get("type") == "tool_use":
                blocked_in = tool_name(block)
        previous = time


def millions(tokens: int) -> str:
    return f"{tokens / 1e6:.1f}M"


def share(part: int, whole: int) -> str:
    return f"{100 * part / whole:.1f}%" if whole else "n/a"


def report(found: list[Request], agents: int) -> None:
    written = sum(r.written for r in found)
    gap_rewrites = [r for r in found if r.is_rewrite and r.gap > GAP_5M]
    other_rewrites = [r for r in found if r.is_rewrite and r.gap <= GAP_5M]
    gap_tokens = sum(r.written for r in gap_rewrites)
    past_1h = [r for r in gap_rewrites if r.gap > GAP_1H]

    print(f"agents {agents}   requests {len(found)}   cache writes {millions(written)}")
    print(
        f"  at 5m TTL {share(sum(r.written_5m for r in found), written)}"
        f"   at 1h TTL {share(sum(r.written_1h for r in found), written)}"
    )
    print(
        f"re-writes after a gap over 5m: {len(gap_rewrites)} events, "
        f"{millions(gap_tokens)}, {share(gap_tokens, written)} of writes"
    )
    print(
        f"  gap over 1h (a 1h TTL does not help): {len(past_1h)} events, "
        f"{millions(sum(r.written for r in past_1h))}"
    )
    print(
        f"re-writes with no gap (not TTL expiry): {len(other_rewrites)} events, "
        f"{millions(sum(r.written for r in other_rewrites))}"
    )

    weeks: dict[str, Counter[str]] = defaultdict(Counter)
    for r in found:
        weeks[r.time.strftime("%G-W%V")]["written"] += r.written
    for r in gap_rewrites:
        weeks[r.time.strftime("%G-W%V")]["rewritten"] += r.written
        weeks[r.time.strftime("%G-W%V")]["events"] += 1
    print("\nweek       writes  gap re-writes   share  events")
    for week, c in sorted(weeks.items()):
        print(
            f"{week}  {millions(c['written']):>7}  {millions(c['rewritten']):>13}"
            f"  {share(c['rewritten'], c['written']):>6}  {c['events']:>6}"
        )

    for title, key in (("blocked in", "blocked_in"), ("agent type", "agent")):
        tokens: Counter[str] = Counter()
        events: Counter[str] = Counter()
        for r in gap_rewrites:
            tokens[str(getattr(r, key))] += r.written
            events[str(getattr(r, key))] += 1
        print(f"\ngap re-writes by {title}:")
        for name, total in tokens.most_common(8):
            print(f"  {name:<20} {events[name]:>4} events  {millions(total):>7}")

    print("\nlargest gap re-writes:")
    for r in sorted(gap_rewrites, key=lambda r: r.written, reverse=True)[:10]:
        print(
            f"  {r.written:>7} tok  {r.gap / 60:5.1f} min  {r.time.date()}"
            f"  {r.agent}  {r.model}  {r.project}"
        )


def main() -> None:
    parser = argparse.ArgumentParser(description="Measure prompt-cache re-writes in subagent transcripts.")
    parser.add_argument("--since", type=date.fromisoformat, help="only requests on or after YYYY-MM-DD")
    parser.add_argument("--projects", type=Path, default=Path.home() / ".claude" / "projects")
    args = parser.parse_args()

    per_agent = [list(requests(path, args.projects)) for path in args.projects.glob(TRANSCRIPTS)]
    if args.since:
        per_agent = [[r for r in agent if r.time.date() >= args.since] for agent in per_agent]
    per_agent = [agent for agent in per_agent if agent]
    report([r for agent in per_agent for r in agent], agents=len(per_agent))


if __name__ == "__main__":
    main()