Phase 3: dynamic pass (bin/offload-scan), synthetic transcript fixtures, tests
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
Executable
+762
@@ -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()
|
||||
@@ -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
|
||||
`design/plan.md` §3. `plugin-inventory` parses this file itself — a genuine
|
||||
`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)
|
||||
# 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.
|
||||
```
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -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"}
|
||||
@@ -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"}
|
||||
@@ -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}}}
|
||||
+242
@@ -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()
|
||||
Executable
+121
@@ -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"
|
||||
Reference in New Issue
Block a user