cache_gap_check.py
| #!/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()) |
点赞
收藏
点踩
分享评论
?
参与讨论