kotkan/claude-plugin-inference-arbitrage
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 3: dynamic pass (bin/offload-scan), synthetic transcript fixtures, tests

Browse Source 
Streams ~/.claude/projects/**/*.jsonl once, filters to turns attributed to the
target plugin (attributionPlugin / attributionSkill / attributionAgent),
de-duplicates by (message.id, requestId) taking max on every usage field, and
computes the spec.md §5.2 dynamic signals: MTR, offload_waste, recurring
tool-signature n-grams, read amplification, retry density, judgment density,
fan-out multiplier, and the composite offload_value — plus the FR-3.4
attribution-coverage headline.

Token and cost arithmetic is taken from token-budget's cc-tokens, never
reimplemented. It is imported as a module rather than shelled out with --json,
because cc-tokens' subcommands aggregate by day/session/project/model and none
can answer "weighted tokens for this set of turns" — the attribution fields are
not on its CLI surface. Its Rec.cost / Rec.weighted still compute every number
here, so the pricing table and cache multipliers keep exactly one home.

Two deviations from design/plan.md §4.1, both because the transcripts say
otherwise: agents attribute via attributionAgent on the subagent sidechain
files, not agentName (which is a separate `type: "agent-name"` line mapping a
session to a display label); and attributionSkill is sometimes bare rather than
plugin-qualified, so it is normalized.

Tool arguments are masked to <path>/<n>/<sha>/<url>/<str> before entering any
emitted structure, and Bash commands are reduced to bare executable words —
FR-3.5, asserted in the suite. Fixtures are synthetic by construction; no real
transcript content is in this repo.

Real-history check: 30-day window over a 1.7 GB history, 94s wall, 29 MB peak
RSS, one process.
This commit is contained in:
Oleks
2026-07-29 16:58:00 +03:00
parent 07b56376fc
commit bdb62da898
8 changed files with 1240 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/offload-scan
Executable
+762
View File
@@ -0,0 +1,762 @@
#!/usr/bin/env python3
"""Dynamic pass over local Claude Code transcripts -> one JSON document.
Usage: offload-scan --plugin <name> [--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 <path>/<n>/<sha>/<url>/<str> 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 "<bool>"
if isinstance(value, (int, float)):
return "<n>"
if isinstance(value, dict):
return "<obj>"
if isinstance(value, list):
return "<list>"
if value is None:
return "<null>"
text = str(value).strip()
if not text:
return "<empty>"
if URL_RE.match(text):
return "<url>"
if NUM_RE.match(text):
return "<n>"
if HEX_RE.match(text):
return "<sha>"
if PATH_RE.match(text) and " " not in text:
return "<path>"
return "<str>"
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 <path>` and
`curl https://…` becomes `curl <url>`. 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 "<empty>"
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") == "<synthetic>":
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,
},
"skills": skills,
"agents": agents,
}
print(json.dumps(doc) if args.json else json.dumps(doc, indent=2))
if __name__ == "__main__":
main()
references/signals-catalog.md
+88 -2
View File
@@ -1,6 +1,8 @@
# signals-catalog.md — static-pass lexicons and thresholds # signals-catalog.md — lexicons and thresholds
Loaded at runtime by `bin/plugin-inventory`. Kept out of the script body so a Loaded at runtime by `bin/plugin-inventory` (static pass, §5.1 signals) and
`bin/offload-scan` (dynamic pass, §5.2 signals). Each reads only the sections
it knows and ignores the rest. Kept out of the script body so a
lexicon or threshold change is a reviewable diff, not a code change — per lexicon or threshold change is a reviewable diff, not a code change — per
`design/plan.md` §3. `plugin-inventory` parses this file itself — a genuine `design/plan.md` §3. `plugin-inventory` parses this file itself — a genuine
`pure-script` read (T1–T5 all pass: closed format, one right parse, `pure-script` read (T1–T5 all pass: closed format, one right parse,
@@ -92,3 +94,87 @@ hook_prose_drift_min_shared_tokens: 2
# English words, which overlap constantly between topically-related prose) # English words, which overlap constantly between topically-related prose)
# counts as the hook's logic being restated in prose — two sources of truth. # counts as the hook's logic being restated in prose — two sources of truth.
``` ```
---
The sections below are read by `bin/offload-scan` only. Same fenced-block
contract, except that the tool sections also accept `*` glob patterns —
MCP tool names are matched on their last `__`-separated segment, so
`mcp__plugin_cluster_gitea-tools__issue_read` is matched as `issue_read`.
## mechanical-tool-calls
A turn counts as mechanical only if EVERY tool it calls matches one of these
(spec.md §5.2.1). Read-only `Bash` is handled separately, by `read-only-bash`
below, because the tool name alone says nothing about whether the shell
command mutates.
```text
Read, Grep, Glob, ToolSearch, BashOutput, LS, NotebookRead, TaskGet, TaskList,
WebFetch, list_*, get_*, *_read, search_*, *_list, *_stats
```
Deliberately absent: `Edit`, `Write`, `NotebookEdit`, `Bash` (unqualified),
`Agent`, `AskUserQuestion`, and every `*_write` / `create_*` / `delete_*` MCP
tool. A turn containing any of those is not mechanical no matter how small its
output, because a mistake there is durable.
## read-tool-calls
Tools whose results count as material pulled into context for the read
amplification ratio (spec.md §5.2.3).
```text
Read, Grep, Glob, WebFetch, NotebookRead, get_*, *_read, list_*, search_*
```
## read-only-bash
Leading command words that make a `Bash` call read-only. Matched against the
first bare word of the masked command, and against the first two words joined
(so `git log` matches while `git push` does not).
```text
ls, cat, head, tail, wc, grep, rg, find, file, stat, du, df, ps, env,
printenv, which, whereis, echo, date, uname, hostname, jq, yq, sort, uniq,
cut, tree, realpath, basename, dirname, readlink, sha256sum, md5sum,
git status, git log, git diff, git show, git branch, git remote, git config,
git ls-files, git worktree, git rev-parse, git describe, git blame,
kubectl get, kubectl describe, kubectl logs, kubectl top, helm list,
helm status, docker ps, docker images, systemctl status, journalctl,
nix eval, nix show-derivation, nix path-info
```
Conspicuously absent: `sed`, `awk`, `curl`. Masking strips a command's flags,
so `sed -i` and `sed -n` are indistinguishable by the time this list is
consulted, and `curl` is a `POST` away from being a mutation. Counting a
mutating turn as mechanical would recommend scripting away a step that has a
blast radius, which is the expensive direction to be wrong in — so the
ambiguous cases are excluded and their turns fall through to `other`.
## scan-thresholds
```text
mechanical_output_tokens_max: 400
# spec.md §5.2.1 — above this the model emitted more than a tool call, so the
# turn is not mechanical even if every tool in it is read-only.
judgment_output_tokens_min: 400
# spec.md §5.2.5 — a no-tool turn at or above this is where the model was
# actually thinking. Same number as above by construction: the two definitions
# partition the same axis and must not leave a gap.
idle_gap_minutes: 10
# wall-clock gap between attributed turns in one session that ends an
# invocation. Chosen because a skill's own steps run back to back in seconds;
# ten minutes means a human went away and came back.
ngram_min_len: 3
ngram_max_len: 6
# spec.md §5.2.2 — sequences shorter than 3 are coincidence; longer than 6 are
# already covered by their maximal prefix and only inflate the report.
ngram_min_invocations: 3
# a sequence must recur across this many DISTINCT invocations. Repetition
# inside one invocation is a loop, not an algorithm worth extracting.
```
tests/fixtures/transcripts/fake-project/sess-a.jsonl
Vendored
+19
View File
File diff suppressed because one or more lines are too long
tests/fixtures/transcripts/fake-project/sess-b.jsonl
Vendored
+4
View File
@@ -0,0 +1,4 @@
{"type": "assistant", "uuid": "u-11", "requestId": "req-11", "sessionId": "sess-b", "isSidechain": false, "cwd": "/fake/project", "timestamp": "2026-07-01T12:00:00Z", "message": {"id": "msg-11", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "text", "text": "a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation a long deliberation "}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 100000, "output_tokens": 2000}}, "attributionPlugin": "fixture-plugin", "attributionSkill": "fixture-plugin:think-hard"}
{"type": "assistant", "uuid": "u-12", "requestId": "req-12", "sessionId": "sess-b", "isSidechain": false, "cwd": "/fake/project", "timestamp": "2026-07-01T12:01:00Z", "message": {"id": "msg-12", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "tool_use", "id": "tb1", "name": "Write", "input": {"file_path": "/fake/out.md", "content": "x"}}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 100000, "output_tokens": 100}}, "attributionPlugin": "fixture-plugin", "attributionSkill": "fixture-plugin:think-hard"}
{"type": "user", "uuid": "r-tb1", "sessionId": "sess-b", "isSidechain": false, "timestamp": "2026-07-01T12:01:00Z", "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tb1", "content": "permission denied", "is_error": true}]}}
{"type": "assistant", "uuid": "u-13", "requestId": "req-13", "sessionId": "sess-b", "isSidechain": false, "cwd": "/fake/project", "timestamp": "2026-07-01T12:02:00Z", "message": {"id": "msg-13", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "tool_use", "id": "tb2", "name": "Write", "input": {"file_path": "/fake/out.md", "content": "x"}}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 100000, "output_tokens": 100}}, "attributionPlugin": "fixture-plugin", "attributionSkill": "fixture-plugin:think-hard"}
tests/fixtures/transcripts/fake-project/sess-c/subagents/agent-fetcher.jsonl
Vendored
+2
View File
@@ -0,0 +1,2 @@
{"type": "assistant", "uuid": "u-14", "requestId": "req-14", "sessionId": "sess-c", "isSidechain": true, "cwd": "/fake/project", "timestamp": "2026-07-01T12:00:00Z", "message": {"id": "msg-14", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "tool_use", "id": "tc1", "name": "Glob", "input": {"pattern": "*.md"}}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 200000, "output_tokens": 100}}, "attributionPlugin": "fixture-plugin", "attributionAgent": "fixture-plugin:fetcher-agent"}
{"type": "assistant", "uuid": "u-15", "requestId": "req-15", "sessionId": "sess-c", "isSidechain": true, "cwd": "/fake/project", "timestamp": "2026-07-01T12:01:00Z", "message": {"id": "msg-15", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "tool_use", "id": "tc2", "name": "Read", "input": {"file_path": "/fake/z.md"}}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 200000, "output_tokens": 100}}, "attributionPlugin": "fixture-plugin", "attributionAgent": "fixture-plugin:fetcher-agent"}
tests/fixtures/transcripts/fake-project/sess-d.jsonl
Vendored
+2
View File
@@ -0,0 +1,2 @@
{"type": "assistant", "uuid": "u-16", "requestId": "req-16", "sessionId": "sess-d", "isSidechain": false, "cwd": "/fake/project", "timestamp": "2026-07-01T12:00:00Z", "message": {"id": "msg-16", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "text", "text": "someone else's work"}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 100000, "output_tokens": 300}}}
{"type": "assistant", "uuid": "u-17", "requestId": "req-17", "sessionId": "sess-d", "isSidechain": false, "cwd": "/fake/project", "timestamp": "2026-07-01T12:01:00Z", "message": {"id": "msg-17", "model": "claude-sonnet-5", "role": "assistant", "content": [{"type": "text", "text": "still not ours"}], "usage": {"input_tokens": 0, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 100000, "output_tokens": 300}}}
tests/fixtures/transcripts/generate.py
Vendored
+242
View File
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""Regenerates the synthetic fixture transcripts in this directory.
Every value here is invented. NOTHING in these fixtures comes from a real
session — real transcripts contain secrets and private work and must never be
committed (design/spec.md FR-3.5). The committed .jsonl files are the test
input; this generator exists so that a threshold change can be reflected in
them without hand-editing JSON.
Shape mimics ~/.claude/projects/**/*.jsonl: one JSON object per line, assistant
lines carrying message.usage + attributionPlugin/attributionSkill (main
sessions) or attributionAgent (subagent sidechain files), user lines carrying
tool_result blocks.
The numbers are chosen so every metric is hand-computable — see
tests/scan.test.sh for the arithmetic each assertion checks.
"""
import json
import os
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
PROJECT = os.path.join(HERE, "fake-project")
MODEL = "claude-sonnet-5"
T0 = datetime(2026, 7, 1, 12, 0, 0, tzinfo=timezone.utc)
# cache_read 100k + output 100, tier 0.4 (sonnet-5 input $2/Mtok / 5)
# weighted = 0.4 * (100000*0.1 + 100*5) = 4200 per inline turn
CTX = 100_000
SIDECHAIN_CTX = 200_000
_counter = [0]
def usage(ctx, out):
return {
"input_tokens": 0,
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": ctx,
"output_tokens": out,
}
def assistant(session, minute, content, out, ctx=CTX, sidechain=False, **attrib):
_counter[0] += 1
line = {
"type": "assistant",
"uuid": f"u-{_counter[0]}",
"requestId": f"req-{_counter[0]}",
"sessionId": session,
"isSidechain": sidechain,
"cwd": "/fake/project",
"timestamp": (T0 + timedelta(minutes=minute))
.isoformat()
.replace("+00:00", "Z"),
"message": {
"id": f"msg-{_counter[0]}",
"model": MODEL,
"role": "assistant",
"content": content,
"usage": usage(ctx, out),
},
}
line.update(attrib)
return line
def tool_use(tid, name, args):
return {"type": "tool_use", "id": tid, "name": name, "input": args}
def text(body):
return {"type": "text", "text": body}
def tool_result(session, minute, tid, body, is_error=False):
return {
"type": "user",
"uuid": f"r-{tid}",
"sessionId": session,
"isSidechain": False,
"timestamp": (T0 + timedelta(minutes=minute))
.isoformat()
.replace("+00:00", "Z"),
"message": {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tid,
"content": body,
"is_error": is_error,
}
],
},
}
# Deliberately verbose so read amplification is > 1: the model quotes back only
# one word of a long result.
HAYSTACK = " ".join(f"haystack_line_{i} filler filler filler" for i in range(200))
SKILL = {
"attributionPlugin": "fixture-plugin",
"attributionSkill": "fixture-plugin:mine-me",
}
THINK = {
"attributionPlugin": "fixture-plugin",
"attributionSkill": "fixture-plugin:think-hard",
}
AGENT = {
"attributionPlugin": "fixture-plugin",
"attributionAgent": "fixture-plugin:fetcher-agent",
}
def session_a():
"""3 invocations of the same 3-tool mechanical sequence, 30 min apart."""
lines = []
for inv in range(3):
base = inv * 30
lines.append(
assistant(
"sess-a",
base,
[
tool_use(
f"t{inv}a", "Grep", {"pattern": "needle", "path": "/fake/x"}
)
],
100,
**SKILL,
)
)
lines.append(tool_result("sess-a", base, f"t{inv}a", HAYSTACK))
lines.append(
assistant(
"sess-a",
base + 1,
[
text("needle"),
tool_use(f"t{inv}b", "Read", {"file_path": "/fake/y.md"}),
],
100,
**SKILL,
)
)
lines.append(tool_result("sess-a", base + 1, f"t{inv}b", "short result"))
lines.append(
assistant(
"sess-a",
base + 2,
[
tool_use(
f"t{inv}c", "mcp__plugin_tracker__issue_read", {"index": 41}
)
],
100,
**SKILL,
)
)
lines.append(tool_result("sess-a", base + 2, f"t{inv}c", "ok"))
# One unattributed turn in the same session: counts toward candidate_turns
# (coverage denominator) but never toward the plugin's spend.
lines.append(assistant("sess-a", 95, [text("unrelated work")], 500))
return lines
def session_b():
"""One invocation exercising judgment, a mutating tool, and a retry."""
lines = [
assistant("sess-b", 0, [text("a long deliberation " * 50)], 2000, **THINK),
assistant(
"sess-b",
1,
[tool_use("tb1", "Write", {"file_path": "/fake/out.md", "content": "x"})],
100,
**THINK,
),
tool_result("sess-b", 1, "tb1", "permission denied", is_error=True),
assistant(
"sess-b",
2,
[tool_use("tb2", "Write", {"file_path": "/fake/out.md", "content": "x"})],
100,
**THINK,
),
]
return lines
def session_c():
"""Sidechain file: agent-attributed, double the inline context."""
return [
assistant(
"sess-c",
0,
[tool_use("tc1", "Glob", {"pattern": "*.md"})],
100,
ctx=SIDECHAIN_CTX,
sidechain=True,
**AGENT,
),
assistant(
"sess-c",
1,
[tool_use("tc2", "Read", {"file_path": "/fake/z.md"})],
100,
ctx=SIDECHAIN_CTX,
sidechain=True,
**AGENT,
),
]
def session_d():
"""A session the target never touched: must not reach the denominator."""
return [
assistant("sess-d", 0, [text("someone else's work")], 300),
assistant("sess-d", 1, [text("still not ours")], 300),
]
def write(path, lines):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
for line in lines:
f.write(json.dumps(line) + "\n")
def main():
write(os.path.join(PROJECT, "sess-a.jsonl"), session_a())
write(os.path.join(PROJECT, "sess-b.jsonl"), session_b())
write(
os.path.join(PROJECT, "sess-c", "subagents", "agent-fetcher.jsonl"), session_c()
)
write(os.path.join(PROJECT, "sess-d.jsonl"), session_d())
if __name__ == "__main__":
main()
tests/scan.test.sh
Executable
+121
View File
@@ -0,0 +1,121 @@
#!/usr/bin/env bash
# Asserts bin/offload-scan against tests/fixtures/transcripts, which are
# SYNTHETIC by construction (see fixtures/transcripts/generate.py — no real
# transcript content may ever enter this repo, per design/spec.md FR-3.5).
#
# The arithmetic each assertion checks, once, here:
# model claude-sonnet-5 -> tier 0.4 (input $2/Mtok / 5)
# inline turn = 0.4 * (100000 cache_read * 0.1 + out * 5)
# = 4200 at out=100, 8000 at out=2000
# sidechain turn = 0.4 * (200000 * 0.1 + 100 * 5) = 8200
# mine-me 9 turns @4200 = 37800 weighted, all mechanical
# think-hard 8000 + 4200 + 4200 = 16400 weighted, 0 mechanical
# fetcher 2 @8200 = 16400 weighted, both sidechain
# offload_value(mine-me) = 37800 * (1 - 0) * (1 + log2(3)) = 97711
set -euo pipefail
cd "$(dirname "${BASH_SOURCE[0]}")/.."
OUT=$(python3 bin/offload-scan --plugin fixture-plugin \
--root tests/fixtures/transcripts --since 2026-06-01 --until 2026-08-01 --json)
fail=0
check() {
local desc="$1" expr="$2"
if ! python3 -c "
import json, sys
d = json.loads(sys.stdin.read())
assert ($expr), 'FAILED: $desc'
" <<<"$OUT"; then
echo "FAIL: $desc" >&2
fail=1
else
echo "ok: $desc"
fi
}
skill() { echo "[s for s in d['skills'] if s['skill'] == 'fixture-plugin:$1'][0]"; }
# --- 3.1 attribution filter and de-dup ------------------------------------
check "only target-plugin turns are attributed" "d['coverage']['attributed_turns'] == 14"
check "two skills and one agent found" "len(d['skills']) == 2 and len(d['agents']) == 1"
check "agent attributed via attributionAgent" "d['agents'][0]['agent'] == 'fixture-plugin:fetcher-agent'"
# --- 3.2 invocation reconstruction ----------------------------------------
check "idle gap splits mine-me into 3 invocations" "$(skill mine-me)['invocations'] == 3"
check "think-hard is one contiguous invocation" "$(skill think-hard)['invocations'] == 1"
# --- 3.3 turn classification ----------------------------------------------
check "mine-me is wholly mechanical" "$(skill mine-me)['mtr'] == 1.0"
check "a Write turn is never mechanical" "$(skill think-hard)['mtr'] == 0.0"
check "a no-tool 2000-token turn is judgment" "$(skill think-hard)['judgment_density'] == 0.333"
check "the turn after a tool error is a retry" "$(skill think-hard)['retry_density'] == 0.333"
# --- 3.4 signature normalization and masking ------------------------------
check "signatures carry shapes, not literals" "
$(skill mine-me)['ngrams'][0]['sig'] == ['Grep(path=<path>, pattern=<str>)',
'Read(file_path=<path>)',
'issue_read(index=<n>)']
"
check "MCP tool names are shortened to their last segment" "
all('mcp__' not in s for s in $(skill mine-me)['ngrams'][0]['sig'])
"
for literal in needle haystack_line "/fake/y.md" "out.md" "deliberation"; do
if grep -qF -- "$literal" <<<"$OUT"; then
echo "FAIL: raw fixture literal '$literal' leaked into output (FR-3.5)" >&2
fail=1
else
echo "ok: '$literal' does not appear in output"
fi
done
# --- 3.5 n-gram mining ------------------------------------------------------
check "the recurring 3-gram is found once, maximal" "len($(skill mine-me)['ngrams']) == 1"
check "3-gram recurs in 3 distinct invocations" "
$(skill mine-me)['ngrams'][0]['recurrences'] == 3
and $(skill mine-me)['ngrams'][0]['invocations'] == 3
"
check "a one-invocation skill mines no n-grams" "$(skill think-hard)['ngrams'] == []"
# --- 3.6 metrics -------------------------------------------------------------
check "weighted tokens come from the cc-tokens tier weighting" "
$(skill mine-me)['weighted_tokens'] == 37800
and $(skill think-hard)['weighted_tokens'] == 16400
and d['agents'][0]['weighted_tokens'] == 16400
"
check "plugin total is the sum of its groups" "d['totals']['weighted_tokens'] == 70600"
check "offload_waste counts only mechanical turns" "
$(skill mine-me)['offload_waste'] == 37800 and $(skill think-hard)['offload_waste'] == 0
"
check "offload_value = waste * (1-judgment) * (1+log2(recurrences))" "
$(skill mine-me)['offload_value'] == 97711
"
check "read amplification is high when a haystack yields one word" "
$(skill mine-me)['read_amplification'] > 100
"
check "fanout multiplier is the sidechain context multiple" "d['agents'][0]['fanout_multiplier'] == 2.0"
check "no sidechain turns means no fanout" "$(skill mine-me)['fanout_multiplier'] == 1.0"
# --- 3.7 attribution coverage -------------------------------------------------
check "coverage denominator includes the unattributed turn in an active session" "
d['coverage']['candidate_turns'] == 15 and d['coverage']['ratio'] == 0.933
"
check "a session the target never touched is excluded entirely" "d['window']['sessions'] == 3"
# --- failure modes ------------------------------------------------------------
if python3 bin/offload-scan --plugin fixture-plugin --root /nonexistent-root >/dev/null 2>&1; then
echo "FAIL: missing transcript root should exit non-zero" >&2
fail=1
else
echo "ok: missing transcript root exits non-zero"
fi
EMPTY=$(python3 bin/offload-scan --plugin no-such-plugin \
--root tests/fixtures/transcripts --since 2026-06-01 --until 2026-08-01 --json)
if [ "$(python3 -c "import json,sys; d=json.load(sys.stdin); print(d['coverage']['attributed_turns'], len(d['skills']))" <<<"$EMPTY")" = "0 0" ]; then
echo "ok: a plugin with no history yields an empty, non-crashing report"
else
echo "FAIL: unknown plugin should yield an empty report" >&2
fail=1
fi
exit "$fail"