#!/usr/bin/env python3 """Dynamic pass over local Claude Code transcripts -> one JSON document. Usage: offload-scan --plugin [--days N | --since D --until D] [--root ~/.claude/projects] [--json] [--catalog PATH] Measures what a plugin's skills and agents ACTUALLY cost in real sessions, and how much of that cost bought no judgment: Mechanical Turn Ratio, recurring tool-call n-grams, read amplification, retry density, judgment density, fan-out multiplier, and the composite offload_value (design/spec.md §5.2). Single streaming stdlib process, offline. PRIVACY (FR-3.5): only normalized shapes and counts leave this process. Tool arguments are masked to //// before they ever enter a data structure that is emitted; Bash commands are reduced to their bare executable/subcommand words. No raw transcript content, file content, command argument, or user prose is ever printed. MONEY MATH (FR-3.2): the pricing table, tier weighting, and cache multipliers come from token-budget's bin/cc-tokens and are never reimplemented here. See resolve_cc_tokens() for the resolution order and the note there about why this imports the module rather than shelling out. """ import argparse import glob import importlib.machinery import importlib.util import json import math import os import re import sys from collections import Counter, defaultdict from datetime import datetime, timedelta, timezone PLUGIN_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DEFAULT_CATALOG = os.path.join(PLUGIN_ROOT, "references", "signals-catalog.md") DEFAULT_ROOT = os.path.expanduser("~/.claude/projects") CC_TOKENS_GLOB = os.path.expanduser( "~/.claude/plugins/cache/*/token-budget/*/bin/cc-tokens") # -------------------------------------------------------------------------- # cc-tokens dependency (plan.md §4.3) # -------------------------------------------------------------------------- def semver_key(text): m = re.match(r"^(\d+)\.(\d+)\.(\d+)", text) return tuple(int(g) for g in m.groups()) if m else (0, 0, 0) def resolve_cc_tokens(): """Locate cc-tokens: PATH first, then the plugin cache, else fail loudly. Imported as a module rather than shelled out with --json, because cc-tokens' subcommands aggregate by day/session/project/model and none of them can answer "weighted tokens for THIS set of (message.id, requestId) turns" — the attribution fields this pass filters on are not part of its CLI surface. Importing keeps the pricing table, the cache multipliers and the tier weighting in exactly one home (its Rec.cost / Rec.weighted are what compute every number here), which is the property plan.md §4.3 actually cares about. The clean long-term fix is still `cc-tokens attribute --by skill|plugin --json` upstream. """ path = None for d in os.environ.get("PATH", "").split(os.pathsep): cand = os.path.join(d, "cc-tokens") if os.path.isfile(cand) and os.access(cand, os.X_OK): path = cand break if path is None: cached = sorted(glob.glob(CC_TOKENS_GLOB), key=lambda p: semver_key(os.path.basename( os.path.dirname(os.path.dirname(p))))) if cached: path = cached[-1] if path is None: sys.exit( "error: cc-tokens not found on PATH and not in " "~/.claude/plugins/cache/*/token-budget/*/bin/cc-tokens.\n" " offload-scan takes all token and cost arithmetic from " "token-budget's cc-tokens and never vendors a copy.\n" " install: claude plugin install token-budget@oleks-local " "(or add token-budget/bin to PATH)") spec = importlib.util.spec_from_loader( "cc_tokens", importlib.machinery.SourceFileLoader("cc_tokens", path)) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) for attr in ("Rec", "price_for", "parse_day"): if not hasattr(module, attr): sys.exit(f"error: {path} has no {attr!r} — incompatible cc-tokens; " f"update token-budget or report this mismatch") return module, path # -------------------------------------------------------------------------- # signals-catalog.md loader (same fenced-block contract as plugin-inventory, # but tolerant of glob patterns like list_* in the tool sections) # -------------------------------------------------------------------------- SCAN_SECTIONS = {"mechanical-tool-calls", "read-tool-calls", "read-only-bash", "scan-thresholds"} def load_scan_catalog(path): with open(path, encoding="utf-8") as f: text = f.read() sections = {} for m in re.finditer(r"^##\s+([a-z-]+)\s*$", text, re.MULTILINE): name = m.group(1) if name not in SCAN_SECTIONS: continue block = re.search(r"```\S*\n(.*?)```", text[m.end():], re.DOTALL) if not block: continue body = block.group(1) if name == "scan-thresholds": values = {} for raw in body.split("\n"): line = raw.split("#", 1)[0].strip() if ":" in line: k, v = line.split(":", 1) values[k.strip()] = float(v.strip()) sections[name] = values else: items = set() for raw in body.split("\n"): for item in raw.split(","): item = item.strip() if re.match(r"^[A-Za-z_*][A-Za-z0-9_ *-]*$", item): items.add(item) sections[name] = items missing = SCAN_SECTIONS - sections.keys() if missing: sys.exit(f"error: {path} is missing section(s): {', '.join(sorted(missing))}") return sections def matches_any(name, patterns): for pat in patterns: if pat == name: return True if "*" in pat and re.fullmatch(re.escape(pat).replace(r"\*", ".*"), name): return True return False # -------------------------------------------------------------------------- # masking (FR-3.5) — nothing that has not been through here may be emitted # -------------------------------------------------------------------------- HEX_RE = re.compile(r"^[0-9a-fA-F]{7,}$") URL_RE = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE) PATH_RE = re.compile(r"^~?/|^\.{1,2}/|^[\w.-]+/[\w./-]+$") NUM_RE = re.compile(r"^-?\d+(\.\d+)?$") BARE_WORD_RE = re.compile(r"^[a-z][a-z0-9_-]*$") def mask_value(value): if isinstance(value, bool): return "" if isinstance(value, (int, float)): return "" if isinstance(value, dict): return "" if isinstance(value, list): return "" if value is None: return "" text = str(value).strip() if not text: return "" if URL_RE.match(text): return "" if NUM_RE.match(text): return "" if HEX_RE.match(text): return "" if PATH_RE.match(text) and " " not in text: return "" return "" def mask_bash(command): """Reduce a shell command to its executable/subcommand words only. Keeps bare lowercase words (`git`, `status`, `kubectl`, `get`) and masks everything else, so `cat /etc/secrets/token` becomes `cat ` and `curl https://…` becomes `curl `. Argument literals never survive. """ tokens = str(command).strip().split() kept = [] for tok in tokens[:3]: if BARE_WORD_RE.match(tok): kept.append(tok) else: kept.append(mask_value(tok)) break return " ".join(kept) if kept else "" def short_tool_name(name): """mcp__plugin_cluster_gitea-tools__issue_read -> issue_read.""" if name.startswith("mcp__") and "__" in name[5:]: return name.rsplit("__", 1)[-1] return name def tool_signature(block): name = short_tool_name(block.get("name") or "?") args = block.get("input") if not isinstance(args, dict) or not args: return f"{name}()" parts = [] for key in sorted(args): val = args[key] if name in ("Bash", "BashOutput") and key in ("command", "cmd"): parts.append(f"{key}={mask_bash(val)}") else: parts.append(f"{key}={mask_value(val)}") return f"{name}({', '.join(parts)})" # -------------------------------------------------------------------------- # turn model # -------------------------------------------------------------------------- class Turn: """One de-duplicated attributed assistant turn, plus its classification inputs. `rec` is a cc-tokens Rec — all token/cost arithmetic lives there.""" __slots__ = ("rec", "session", "ts", "skill", "agent", "plugin", "sidechain", "sigs", "tool_names", "text_words", "output_tokens", "after_error", "asks_user", "pulled_tokens", "referenced_tokens") def __init__(self, rec): self.rec = rec self.session = rec.session self.ts = rec.ts self.skill = None self.agent = None self.plugin = None self.sidechain = bool(rec.sidechain) self.sigs = [] self.tool_names = [] self.text_words = 0 self.output_tokens = rec.out self.after_error = False self.asks_user = False self.pulled_tokens = 0 self.referenced_tokens = 0 @property def group(self): return self.agent or self.skill or f"{self.plugin}:(unattributed)" @property def kind(self): return "agent" if self.agent else "skill" def est_tokens(text): """~4 chars per token. Only ever applied to material that stays local.""" return max(1, len(text) // 4) def content_text(blocks): out = [] if isinstance(blocks, str): return blocks if not isinstance(blocks, list): return "" for b in blocks: if isinstance(b, dict) and b.get("type") == "text": out.append(str(b.get("text") or "")) return "\n".join(out) def result_text(block): content = block.get("content") if isinstance(content, str): return content if isinstance(content, list): parts = [] for c in content: if isinstance(c, dict) and c.get("type") == "text": parts.append(str(c.get("text") or "")) return "\n".join(parts) return "" WORD_RE = re.compile(r"[A-Za-z0-9_./-]{3,}") def word_set(text, cap=4000): return set(WORD_RE.findall(text)[:cap]) # -------------------------------------------------------------------------- # streaming scan # -------------------------------------------------------------------------- def normalize_skill(value, plugin): if not value: return None return value if ":" in value else f"{plugin}:{value}" def scan(root, plugin, since, until, cc, catalog): """Streams every transcript once. Yields nothing; returns (turns, session_totals, sessions_with_target). Memory is O(turns attributed to the target + sessions), not O(all turns): unattributed turns are counted and dropped immediately. """ read_tools = catalog["read-tool-calls"] turns = [] session_turns = Counter() session_hit = set() session_span = {} floor = since.timestamp() - 86400 if since else None seen = set() for path in sorted(glob.iglob(os.path.join(root, "**", "*.jsonl"), recursive=True)): try: if floor is not None and os.stat(path).st_mtime < floor: continue except OSError: continue try: fh = open(path, "r", errors="replace") except OSError: continue pending = {} order = [] # Rolling state, valid only within one file (== one session's append # log, so file order is chronological). last_error = False pending_reads = [] tool_use_names = {} with fh: for line in fh: if '"type"' not in line: continue try: o = json.loads(line) except (ValueError, TypeError): continue kind = o.get("type") if kind == "user": blocks = (o.get("message") or {}).get("content") if not isinstance(blocks, list): continue for b in blocks: if not isinstance(b, dict) or b.get("type") != "tool_result": continue if b.get("is_error"): last_error = True name = tool_use_names.pop(b.get("tool_use_id"), None) if not name or not matches_any(name, read_tools): continue text = result_text(b) if text and len(pending_reads) < 32: pending_reads.append((est_tokens(text), word_set(text))) continue if kind != "assistant": continue msg = o.get("message") or {} usage = msg.get("usage") or {} if not usage or msg.get("model") == "": continue try: when = datetime.fromisoformat( str(o.get("timestamp")).replace("Z", "+00:00")) except (ValueError, TypeError): continue if since and when < since: continue if until and when > until: continue session = o.get("sessionId") or os.path.basename(path)[:-6] session_turns[session] += 1 span = session_span.get(session) if span is None: session_span[session] = [when, when] else: if when < span[0]: span[0] = when if when > span[1]: span[1] = when # Registered for every assistant turn, attributed or not: the # read is issued by turn N and consumed by turn N+1, and only # one of the two need belong to the target. if len(tool_use_names) > 256: tool_use_names.clear() for b in msg.get("content") or []: if isinstance(b, dict) and b.get("type") == "tool_use": tool_use_names[b.get("id")] = short_tool_name(b.get("name") or "?") attr_plugin = o.get("attributionPlugin") attr_agent = o.get("attributionAgent") attr_skill = o.get("attributionSkill") mine = (attr_plugin == plugin or (attr_agent or "").startswith(plugin + ":") or (attr_skill or "").startswith(plugin + ":")) if not mine: # Still consumed: a tool error or a read that precedes an # unattributed turn belongs to that turn, not to ours. last_error = False pending_reads = [] continue session_hit.add(session) cw = usage.get("cache_creation") or {} cw5 = int(cw.get("ephemeral_5m_input_tokens") or 0) cw1 = int(cw.get("ephemeral_1h_input_tokens") or 0) total_cw = int(usage.get("cache_creation_input_tokens") or 0) if not (cw5 or cw1) and total_cw: cw1 = total_cw rec = cc.Rec( ts=when, model=msg.get("model"), session=session, cwd=o.get("cwd") or "", sidechain=bool(o.get("isSidechain")), inp=int(usage.get("input_tokens") or 0), cw5=cw5, cw1=cw1, cr=int(usage.get("cache_read_input_tokens") or 0), out=int(usage.get("output_tokens") or 0)) key = (msg.get("id"), o.get("requestId")) if key == (None, None): key = ("_anon", o.get("uuid") or len(order)) turn = pending.get(key) if turn is None: turn = Turn(rec) turn.plugin = attr_plugin or plugin turn.skill = normalize_skill(attr_skill, turn.plugin) turn.agent = attr_agent turn.after_error = last_error pending[key] = turn order.append(key) else: # Same message, later streaming snapshot: max each field. # (The cache fields never vary; output_tokens grows.) for f in ("inp", "cw5", "cw1", "cr", "out"): setattr(turn.rec, f, max(getattr(turn.rec, f), getattr(rec, f))) turn.output_tokens = turn.rec.out for b in msg.get("content") or []: if not isinstance(b, dict): continue if b.get("type") == "tool_use": name = short_tool_name(b.get("name") or "?") turn.tool_names.append(name) turn.sigs.append(tool_signature(b)) if name == "AskUserQuestion": turn.asks_user = True elif b.get("type") == "text": turn.text_words += len(str(b.get("text") or "").split()) if pending_reads: words = word_set(content_text(msg.get("content"))) for pulled, result_words in pending_reads: turn.pulled_tokens += pulled overlap = words & result_words turn.referenced_tokens += est_tokens(" ".join(overlap)) pending_reads = [] last_error = False for key in order: if key in seen: continue seen.add(key) turns.append(pending[key]) return turns, session_turns, session_hit, session_span # -------------------------------------------------------------------------- # classification (spec §5.2.1, §5.2.4, §5.2.5) # -------------------------------------------------------------------------- def is_mechanical_tool(name, sig, catalog): if name in ("Bash", "BashOutput"): head = sig.split("cmd=", 1)[-1].split("command=", 1)[-1].rstrip(")") words = [w for w in head.split() if BARE_WORD_RE.match(w)] if not words: return False return (matches_any(words[0], catalog["read-only-bash"]) or matches_any(" ".join(words[:2]), catalog["read-only-bash"])) return matches_any(name, catalog["mechanical-tool-calls"]) def classify(turn, catalog): th = catalog["scan-thresholds"] if turn.after_error: return "retry" if turn.tool_names: if (turn.output_tokens < th["mechanical_output_tokens_max"] and all(is_mechanical_tool(n, s, catalog) for n, s in zip(turn.tool_names, turn.sigs))): return "mechanical" return "other" if turn.output_tokens >= th["judgment_output_tokens_min"] or turn.asks_user: return "judgment" return "other" # -------------------------------------------------------------------------- # invocation reconstruction (plan.md §4.1 step 4) # -------------------------------------------------------------------------- def reconstruct_invocations(turns, idle_gap_minutes): """Contiguous runs of attributed turns in one session and one group, split on an idle gap. Returns [(invocation_id, group, [turns])].""" by_session = defaultdict(list) for t in turns: by_session[t.session].append(t) invocations = [] gap = timedelta(minutes=idle_gap_minutes) for session, group_turns in sorted(by_session.items()): group_turns.sort(key=lambda t: t.ts) current = [] for t in group_turns: if current and (t.group != current[-1].group or t.ts - current[-1].ts > gap): invocations.append((f"{session}#{len(invocations)}", current[0].group, current)) current = [] current.append(t) if current: invocations.append((f"{session}#{len(invocations)}", current[0].group, current)) return invocations # -------------------------------------------------------------------------- # n-gram mining (spec §5.2.2) # -------------------------------------------------------------------------- def is_subsequence(short, long_): n = len(short) return any(long_[i:i + n] == short for i in range(len(long_) - n + 1)) def mine_ngrams(invocations, weighted_of, min_len, max_len, min_invocations, top=5): """Ordered signature sequences per invocation; n-grams of length >= min_len recurring in >= min_invocations distinct invocations. Only maximal n-grams are reported — a 3-gram fully inside a kept 5-gram with no extra recurrences is the same finding stated twice.""" seqs = [] for inv_id, _group, turns in invocations: seq = [] for t in turns: for sig in t.sigs: seq.append((sig, t)) if len(seq) >= min_len: seqs.append((inv_id, seq)) found = {} for n in range(min_len, max_len + 1): counts = defaultdict(lambda: [0, set(), set()]) for inv_id, seq in seqs: for i in range(len(seq) - n + 1): gram = tuple(s for s, _ in seq[i:i + n]) entry = counts[gram] entry[0] += 1 entry[1].add(inv_id) for _, t in seq[i:i + n]: entry[2].add(id(t)) for gram, (occurrences, invs, turn_ids) in counts.items(): if len(invs) >= min_invocations: found[gram] = (occurrences, len(invs), turn_ids) kept = [] for gram in sorted(found, key=lambda g: (-len(g), -found[g][0])): occurrences, invs, turn_ids = found[gram] if any(len(gram) < len(k) and is_subsequence(list(gram), list(k)) and found[k][1] >= invs for k in (kk for kk, _ in kept)): continue kept.append((gram, (occurrences, invs, turn_ids))) turn_weight = {id(t): weighted_of(t) for _i, _g, ts in invocations for t in ts} out = [] for gram, (occurrences, invs, turn_ids) in kept: waste = int(sum(turn_weight.get(tid, 0) for tid in turn_ids)) out.append({"sig": list(gram), "recurrences": occurrences, "invocations": invs, "waste": waste}) out.sort(key=lambda g: (-g["waste"], -g["recurrences"])) return out[:top] # -------------------------------------------------------------------------- # aggregation (spec §5.2.7) # -------------------------------------------------------------------------- def summarize_group(name, kind, turns, invocations, plugin_weighted, inline_mean_ctx, catalog): """fanout_multiplier is mean context per SIDECHAIN turn over mean context per inline turn of the same plugin — literally "a step in here costs N x the same step inline". It is 1.0 when the group has no sidechain turns, and it can legitimately come out BELOW 1: a subagent spawned with a tight brief carries less context than a marathon main session, which is the fan-out done right. Reported unclamped; it is a diagnostic, not an input to offload_value.""" th = catalog["scan-thresholds"] n = len(turns) weighted = sum(t.rec.weighted for t in turns) kinds = Counter(classify(t, catalog) for t in turns) mechanical = [t for t in turns if classify(t, catalog) == "mechanical"] offload_waste = sum(t.rec.weighted for t in mechanical) pulled = sum(t.pulled_tokens for t in turns) referenced = sum(t.referenced_tokens for t in turns) read_amp = round(pulled / referenced, 2) if referenced else ( float(pulled) if pulled else 0.0) sidechain = [t for t in turns if t.sidechain] if sidechain and inline_mean_ctx: mean_ctx = sum(t.rec.ctx for t in sidechain) / len(sidechain) fanout = round(mean_ctx / inline_mean_ctx, 2) else: fanout = 1.0 ngrams = mine_ngrams( invocations, lambda t: t.rec.weighted, int(th["ngram_min_len"]), int(th["ngram_max_len"]), int(th["ngram_min_invocations"])) judgment_density = round(kinds["judgment"] / n, 3) if n else 0.0 dominant = max((g["recurrences"] for g in ngrams), default=0) repetition_factor = 1 + math.log2(dominant) if dominant > 1 else 1.0 offload_value = int(offload_waste * (1 - judgment_density) * repetition_factor) return { ("agent" if kind == "agent" else "skill"): name, "invocations": len(invocations), "turns": n, "sidechain_turns": len(sidechain), "weighted_tokens": int(weighted), "cost_estimate_usd": round(sum(t.rec.cost for t in turns), 2), "share_of_plugin": round(weighted / plugin_weighted, 3) if plugin_weighted else 0.0, "mtr": round(len(mechanical) / n, 3) if n else 0.0, "judgment_density": judgment_density, "retry_density": round(kinds["retry"] / n, 3) if n else 0.0, "offload_waste": int(offload_waste), "read_amplification": read_amp, "fanout_multiplier": fanout, "repetition_factor": round(repetition_factor, 3), "ngrams": ngrams, "offload_value": offload_value, } # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def main(): ap = argparse.ArgumentParser(prog="offload-scan") ap.add_argument("--plugin", required=True, help="plugin name to attribute against") ap.add_argument("--days", type=int, help="window size in days back from now (default 30)") ap.add_argument("--since", help="inclusive lower bound YYYY-MM-DD") ap.add_argument("--until", help="inclusive upper bound YYYY-MM-DD") ap.add_argument("--root", default=DEFAULT_ROOT, help="transcripts root") ap.add_argument("--catalog", default=DEFAULT_CATALOG, help="path to signals-catalog.md") ap.add_argument("--json", action="store_true", help="compact single-line JSON") args = ap.parse_args() cc, cc_path = resolve_cc_tokens() catalog = load_scan_catalog(args.catalog) root = os.path.expanduser(args.root) if not os.path.isdir(root): sys.exit(f"error: no such directory: {root}") if args.since or args.until: since = cc.parse_day(args.since) if args.since else None until = cc.parse_day(args.until) if args.until else None if until: until += timedelta(days=1) - timedelta(microseconds=1) else: days = args.days if args.days else 30 since = datetime.now(timezone.utc) - timedelta(days=days) until = None turns, session_turns, session_hit, session_span = scan( root, args.plugin, since, until, cc, catalog) th = catalog["scan-thresholds"] invocations = reconstruct_invocations(turns, th["idle_gap_minutes"]) plugin_weighted = sum(t.rec.weighted for t in turns) inline = [t for t in turns if not t.sidechain] inline_mean_ctx = (sum(t.rec.ctx for t in inline) / len(inline)) if inline else 0.0 by_group = defaultdict(list) for t in turns: by_group[(t.group, t.kind)].append(t) invs_by_group = defaultdict(list) for inv in invocations: invs_by_group[inv[1]].append(inv) skills, agents = [], [] for (name, kind), group_turns in by_group.items(): row = summarize_group(name, kind, group_turns, invs_by_group[name], plugin_weighted, inline_mean_ctx, catalog) (agents if kind == "agent" else skills).append(row) skills.sort(key=lambda r: -r["offload_value"]) agents.sort(key=lambda r: -r["offload_value"]) candidate_turns = sum(c for s, c in session_turns.items() if s in session_hit) spans = [session_span[s] for s in session_hit if s in session_span] window_since = min((s[0] for s in spans), default=since) window_until = max((s[1] for s in spans), default=until) doc = { "window": { "since": window_since.isoformat() if window_since else None, "until": window_until.isoformat() if window_until else None, "requested_since": since.isoformat() if since else None, "requested_until": until.isoformat() if until else None, "sessions": len(session_hit), }, "plugin": args.plugin, "tool_versions": {"cc_tokens_path": cc_path}, "coverage": { "attributed_turns": len(turns), "candidate_turns": candidate_turns, "ratio": round(len(turns) / candidate_turns, 3) if candidate_turns else 0.0, "caveat": "hook-driven and unattributed agent turns excluded", }, "totals": { "weighted_tokens": int(plugin_weighted), "cost_estimate_usd": round(sum(t.rec.cost for t in turns), 2), "invocations": len(invocations), "mechanical_share": round( sum(1 for t in turns if classify(t, catalog) == "mechanical") / len(turns), 3) if turns else 0.0, # reconstruct_invocations() only sees already-attributed turns and # splits on a wall-clock idle gap; it has no signal for unrelated # activity between two genuinely separate same-skill calls, so # back-to-back invocations closer together than the gap collapse # into one reported invocation. Counts here are a LOWER BOUND on # the true invocation count, and per-invocation figures derived # from them (e.g. cost/invocation) are correspondingly inflated. "invocation_caveat": ( "invocation counts are a lower bound: two same-skill calls " f"within {int(th['idle_gap_minutes'])} minutes of each other on the wall " "clock are merged into one reported invocation"), }, "skills": skills, "agents": agents, } print(json.dumps(doc) if args.json else json.dumps(doc, indent=2)) if __name__ == "__main__": main()