cache_gap_check.py

A developer published cache_gap_check.py, a read-only Python 3 stdlib script that scans Claude Code transcripts (~/.claude/projects/**/*.jsonl) for subagent sidechain requests, measures idle gaps between each subagent's consecutive requests, and estimates prompt-cache cost under the current 5-minute TTL versus a hypothetical 1-hour TTL in multiples of base input price. The tool flags gaps between 5 and 60 minutes as the window where the longer TTL could help and reports a verdict when estimated savings exceed 5%. It makes no network calls and never writes files.

| | /usr/bin/env python3 | | | """cache gap check.py - READ-ONLY. Is raising subagentPromptCacheTtl 5m - 1h likely worth it? | | | Scans Claude Code transcripts ~/.claude/projects/ / .jsonl for subagent sidechain | | | requests, measures idle gaps between each subagent's consecutive requests, and estimates | | | cache cost under the current 5m TTL vs a hypothetical 1h TTL, in multiples of base input | | | price. Python 3 stdlib only. Never writes files and makes no network calls. | | | """ | | | import argparse, json, os, sys | | | from collections import defaultdict | | | from datetime import datetime, timedelta, timezone | | | W5, W1H = 1.25, 2.0 cache write multipliers Anthropic prompt-caching docs | | | GAP LO, GAP HI = 300.0, 3600.0 5 min <= gap < 60 min is the window where 1h can help | | | VERDICT PCT = 5.0 5% estimated saving = "likely worth it"; within +-5% = marginal | | | def parse ts s : | | | try: | | | return datetime.fromisoformat str s .replace "Z", "+00:00" | | | except Exception: | | | return None | | | def find files path : | | | if os.path.isfile path : | | | return path | | | out = | | | for root, , files in os.walk path : | | | out += os.path.join root, f for f in files if f.endswith ".jsonl" | | | return sorted out | | | def load requests files, since : | | | """Return {agent key: request dicts } for sidechain/subagent assistant messages with usage.""" | | | agents = defaultdict dict key - {request id: req} dedupes streamed duplicate lines | | | for f in files: | | | in sub dir = "subagents" in f.replace "\\", "/" .split "/" | | | try: | | | fh = open f, "r", encoding="utf-8", errors="replace" | | | except OSError: | | | continue | | | with fh: | | | for n, line in enumerate fh : | | | try: | | | o = json.loads line | | | except Exception: | | | continue | | | if not isinstance o, dict : | | | continue | | | msg = o.get "message" | | | if not isinstance msg, dict or not isinstance msg.get "usage" , dict : | | | continue | | | if not o.get "isSidechain" is True or o.get "agentId" or in sub dir : | | | continue main-thread request: not a subagent | | | ts = parse ts o.get "timestamp" | | | if ts is None: | | | continue | | | if ts.tzinfo is None: | | | ts = ts.replace tzinfo=timezone.utc | | | if ts < since: | | | continue | | | u = msg "usage" | | | cc = u.get "cache creation" if isinstance u.get "cache creation" , dict else {} | | | create = int u.get "cache creation input tokens" or 0 | | | w1h = int cc.get "ephemeral 1h input tokens" or 0 | | | w5 = int cc.get "ephemeral 5m input tokens" or 0 | | | if w1h + w5 < create: no/partial breakdown: treat remainder as 5m | | | w5 = create - w1h | | | req = dict ts=ts, create=create, read=int u.get "cache read input tokens" or 0 , | | | w5=w5, w1h=w1h | | | key = o.get "agentId" or f | | | rid = msg.get "id" or o.get "requestId" or o.get "uuid" or "%s:%d" % f, n | | | prev = agents key .get rid | | | if prev is None or ts < prev "ts" : | | | keep ts = ts | | | else: | | | keep ts = prev "ts" | | | req "ts" = keep ts earliest line of a streamed message; usage from last line | | | agents key rid = req | | | return {k: sorted v.values , key=lambda r: r "ts" for k, v in agents.items } | | | def analyze agents, read mult : | | | s = dict agents=len agents , requests=0, with prev=0, lt5m=0, win=0, ge60m=0, | | | win rewritten=0, already 1h tokens=0, cost 5m=0.0, cost 1h=0.0 | | | for reqs in agents.values : | | | prev = None | | | for r in reqs: | | | total = r "create" + r "read" cached-prefix tokens this request used | | | s "requests" += 1 | | | s "already 1h tokens" += r "w1h" | | | base = W5 r "w5" + W1H r "w1h" + read mult r "read" | | | hyp read = r "read" | | | gap = None | | | if prev is not None: | | | gap = r "ts" - prev "ts" .total seconds | | | s "with prev" += 1 | | | if gap < GAP LO: | | | s "lt5m" += 1 | | | elif gap < GAP HI: | | | s "win" += 1 | | | s "win rewritten" += r "create" | | | with 1h, the previous request's cached prefix would still be alive | | | hyp read = max r "read" , min prev "create" + prev "read" , total | | | else: | | | s "ge60m" += 1 | | | hyp write = total - hyp read | | | hyp = W1H hyp write + read mult hyp read | | | s "cost 5m" += base | | | s "cost 1h" += hyp | | | prev = r | | | return s | | | def verdict s : | | | if s "requests" == 0: | | | return "no-data", "No subagent requests found in range - nothing to conclude." | | | if s "cost 5m" <= 0: | | | return "no-cache", "Subagents show no cache activity - TTL change would not matter." | | | d = s "cost 1h" - s "cost 5m" / s "cost 5m" 100.0 | | | if d < -VERDICT PCT: | | | return "worth-it", "LIKELY WORTH IT: est. %.1f%% lower cache cost with 1h." % -d | | | if d VERDICT PCT: | | | return "not-worth-it", "NOT WORTH IT: est. %.1f%% higher cache cost with 1h." % d | | | return "marginal", "MARGINAL %+.1f%% : within +-%d%%, leave at 5m unless latency matters." % d, VERDICT PCT | | | def main : | | | ap = argparse.ArgumentParser description="Estimate whether subagentPromptCacheTtl 5m- 1h is worth it read-only ." | | | ap.add argument "--days", type=float, default=14, help="look-back window in days default 14 " | | | ap.add argument "--path", default=os.path.expanduser "~/.claude/projects" , | | | help="transcript dir or file default ~/.claude/projects " | | | ap.add argument "--json", action="store true", help="machine-readable output" | | | ap.add argument "--read-mult", type=float, default=0.1, | | | help="cache-read multiplier default 0.1; docs list 0.05/0.025 for some newer models " | | | a = ap.parse args | | | since = datetime.now timezone.utc - timedelta days=a.days | | | files = find files a.path | | | agents = load requests files, since | | | s = analyze agents, a.read mult | | | code, text = verdict s | | | req = s "requests" | | | pct = lambda n: 100.0 n / req if req else 0.0 | | | delta = s "cost 1h" - s "cost 5m" / s "cost 5m" 100.0 if s "cost 5m" else 0.0 | | | if a.json: | | | print json.dumps dict path=a.path, days=a.days, files scanned=len files , subagents=s "agents" , | | | requests=req, requests with prev=s "with prev" , gap lt 5m=s "lt5m" , | | | gap 5m to 60m=s "win" , gap 5m to 60m pct=round pct s "win" , 1 , gap ge 60m=s "ge60m" , | | | tokens rewritten after 5m 60m gaps=s "win rewritten" , | | | tokens already written as 1h=s "already 1h tokens" , | | | cost now 5m base units=round s "cost 5m" , 1 , cost 1h base units=round s "cost 1h" , 1 , | | | est change pct=round delta, 1 , verdict=code, verdict text=text, | | | multipliers=dict write 5m=W5, write 1h=W1H, read=a.read mult , indent=2 | | | return 0 | | | print "Subagent cache-gap check \| last %g d \| %s" % a.days, a.path | | | print "Scanned %d files - %d subagents, %d subagent requests" % len files , s "agents" , req | | | if req: | | | print "Gap 5m-60m 1h could help : %d %.1f%% of requests \| <5m: %d \| =60m: %d \| first-in-agent: %d" | | | % s "win" , pct s "win" , s "lt5m" , s "ge60m" , req - s "with prev" | | | print "Cache tokens re-written after those gaps: %s" % format s "win rewritten" , "," | | | print "Est. cost now 5m : %s base-input-token units" % format round s "cost 5m" , "," | | | print "Est. cost with 1h : %s base-input-token units %+.1f%% " % format round s "cost 1h" , "," , delta | | | if s "already 1h tokens" : | | | print "Note: %s written tokens were already 1h in the data priced 2x in 'now' ." % format s "already 1h tokens" , "," | | | print "Verdict: " + text | | | print "Rule: 1h write=2x vs 5m=1.25x +0.75x every write ; each avoided expired re-write saves 1.25-%.2g=%.3gx." % a.read mult, W5 - a.read mult | | | return 0 | | | if name == " main ": | | | sys.exit main |