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

Phase 5: snapshot persistence and volume-normalized trend diffing

Browse Source 
bin/audit-snapshot write|list|diff, plus two subcommands that carry the wiki
bridge: `pages` emits what to publish and `import` seeds a cold cache from a
wiki-fetched snapshots.jsonl. The script is offline and filesystem-only — it
mirrors the wiki tree into a local store and never opens a socket, so the
plugin still holds no Gitea credentials and the push stays an explicit caller
action. Resolves the tension in plan §5, where "cloned and pushed" and "never
writes Gitea directly" cannot both be true of one process.

Identity resolves issue -> signature hash -> slug (FR-5.2); a hash change under
a stable identity is reported as signature-drifted rather than a silent
disappearance. Diff verdicts come from cost-per-invocation and share-of-spend
with a 10% noise band (FR-5.4) so a quiet window cannot read as progress.
`resolved` is the strict FR-5.5 conjunction — issue closed AND remeasured under
the same 2%/3-invocation bar boundary-classify uses to file; everything else
that merely stopped being reported is claimed-fixed-unconfirmed, naming the
missing leg.

Two additions to the plan §6 schema, both because a requirement is otherwise
uncomputable: target.skills (stale needs to tell "deleted" from "got cheap")
and measurements_by_signature (resolution needs a later window's cost for a
candidate that is gone).
This commit is contained in:
Oleks
2026-07-29 17:49:44 +03:00
parent 8da16add54
commit ff997d3162
6 changed files with 1304 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude-plugin/plugin.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "inference-arbitrage",
"version": "0.2.0",
"version": "0.3.0",
"description": "Audits a Claude Code plugin's definitions and usage transcripts to find steps that should be a deterministic script instead of raw LLM inference, and files the well-evidenced ones as issues on the target repo.",
"author": {
"name": "oleks"
bin/audit-snapshot
Executable
+775
View File
@@ -0,0 +1,775 @@
#!/usr/bin/env python3
"""Persists one audit run as a snapshot record and diffs runs over time
(spec FR-5, plan §5-6).
It takes already-computed data — `plugin-inventory`, `offload-scan`,
`boundary-classify` — and does nothing but record, match, and subtract. Every
step here passes all five determinism tests, so none of it belongs in a model
(FR-8.1).
STORAGE AND THE PUSH BOUNDARY
-----------------------------
plan §5 says the store is this plugin's own Gitea wiki, "cloned to a local
cache and pushed via cluster:gitea-agent (this plugin never writes Gitea
directly)" — which is in tension with itself, because a git push to the wiki
repo IS a write to Gitea.
Resolved as: **this script is offline and filesystem-only.** It reads and
writes a local store directory laid out exactly like the wiki tree:
<store>/Data/<target>/snapshots.jsonl
<store>/Audits/<target>/<YYYY-MM-DD>.md
<store>/Audits/<target>/Latest.md
It never opens a socket — no git push, no Gitea API — which keeps plan §7's
"offline, only the two output paths touch the network" true, and keeps the
wiki credentials entirely outside this plugin. Publishing is a separate,
explicit action by the caller (the `offload-trend` / `offload-audit` skill, or
a human): `audit-snapshot pages` emits exactly which wiki pages to write with
their full content, and the caller performs the writes with the Gitea wiki
tools. `audit-snapshot import` is the other half of that bridge — it seeds the
local store from a `snapshots.jsonl` the caller fetched off the wiki, so a
machine with a cold cache still diffs against real history.
The append-only JSONL lives on the wiki as a page whose name is the file path;
Gitea wiki pages are files in a git repo, so the wiki IS the git-backed store
plan §5 asks for, and `git log` on it is the audit trail. Appending is
read-current-content, add one line, write back — which is why `pages` emits
the whole file, not a delta.
"""
import argparse
import hashlib
import json
import os
import sys
from datetime import datetime, timezone
DEFAULT_STORE = "~/.cache/inference-arbitrage/wiki"
# Both mirror bin/boundary-classify deliberately: a candidate must clear the
# same bar to be called fixed that it had to clear to be filed. Changing one
# without the other would make "resolved" mean something the filing gate never
# meant.
MIN_SHARE_OF_SPEND = 0.02
MIN_INVOCATIONS = 3
# Relative change in the volume-normalized metrics below which a candidate is
# `persisting` rather than grown/shrunk. Sampling noise across two windows of
# different length is easily a few percent; 10% is the smallest band that does
# not turn every run into a trend.
NOISE_BAND = 0.10
STATUSES = ("new", "persisting", "grown", "shrunk", "resolved",
"signature-drifted", "stale", "claimed-fixed-unconfirmed")
# --------------------------------------------------------------------------
# store
# --------------------------------------------------------------------------
def store_root(store):
return os.path.expanduser(store)
def snapshots_path(store, target):
return os.path.join(store_root(store), "Data", target, "snapshots.jsonl")
def audits_dir(store, target):
return os.path.join(store_root(store), "Audits", target)
def load_snapshots(store, target):
path = snapshots_path(store, target)
if not os.path.exists(path):
return []
out = []
with open(path) as fh:
for lineno, line in enumerate(fh, 1):
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except json.JSONDecodeError as exc:
sys.exit(f"error: {path}:{lineno}: corrupt snapshot record: {exc}")
out.sort(key=lambda s: s.get("run_id", ""))
return out
def append_snapshot(store, target, snap):
path = snapshots_path(store, target)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "a") as fh:
fh.write(json.dumps(snap, sort_keys=True) + "\n")
return path
def write_text(path, text):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as fh:
fh.write(text)
return path
# --------------------------------------------------------------------------
# identity (FR-5.2)
# --------------------------------------------------------------------------
def signature_hash(sig):
"""Hash of a normalized tool-sequence signature. `sig` is the list of
masked signatures offload-scan already emits, so normalization happened
upstream; joining them is all that is left."""
return hashlib.sha256("\n".join(sig).encode()).hexdigest()[:16]
def identity(cand):
"""FR-5.2 resolution order. Issue number first: it is the only identifier
a human also uses, and it survives a refactor that moves the code."""
if cand.get("issue"):
return ("issue", cand["issue"])
if cand.get("signature_hash"):
return ("signature", cand["signature_hash"])
return ("slug", cand.get("candidate_id"))
def build_index(cands):
idx = {"issue": {}, "signature": {}, "slug": {}}
for c in cands:
if c.get("issue"):
idx["issue"].setdefault(c["issue"], c)
if c.get("signature_hash"):
idx["signature"].setdefault(c["signature_hash"], c)
idx["slug"].setdefault(c.get("candidate_id"), c)
return idx
def find_match(cand, idx):
for kind, key in (("issue", cand.get("issue")),
("signature", cand.get("signature_hash")),
("slug", cand.get("candidate_id"))):
if key and key in idx[kind]:
return idx[kind][key], kind
return None, None
# --------------------------------------------------------------------------
# write
# --------------------------------------------------------------------------
def tool_version():
manifest = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
".claude-plugin", "plugin.json")
try:
with open(manifest) as fh:
return json.load(fh).get("version")
except OSError:
return None
def scan_rows(scan):
for row in scan.get("skills", []):
yield row.get("skill"), row
for row in scan.get("agents", []):
yield row.get("agent"), row
def measurements_by_signature(scan):
"""Every recurring signature the scan saw, keyed by hash — including ones
that no longer classify as candidates. FR-5.5 needs exactly this: a later
window's measured cost for a signature whose candidate is gone."""
total = scan.get("totals", {}).get("weighted_tokens") or 0
out = {}
for _name, row in scan_rows(scan):
for ng in row.get("ngrams", []):
h = signature_hash(ng["sig"])
invocations = ng.get("invocations", 0)
waste = ng.get("waste", 0)
out[h] = {
"invocations": invocations,
"offload_value": waste,
"share_of_plugin": round(waste / total, 4) if total else 0.0,
"cost_per_invocation": int(waste / invocations) if invocations else 0,
}
return out
def candidate_signature(cand, scan, explicit):
"""Signature resolution, most trustworthy first: an explicitly supplied
tool sequence, then the dominant recurring n-gram of the candidate's own
skill, then the slug. The source is recorded because a slug-derived
identity cannot detect drift — it IS the thing drift is measured against."""
cid = cand.get("candidate_id")
if explicit.get(cid):
return list(explicit[cid]), "explicit"
for name, row in scan_rows(scan):
if name == cand.get("skill") and row.get("ngrams"):
return list(row["ngrams"][0]["sig"]), "scan-ngram"
return [f"slug:{cid}"], "slug"
def snapshot_candidate(cand, scan, explicit, issue_map):
sig, source = candidate_signature(cand, scan, explicit)
row = next((r for n, r in scan_rows(scan) if n == cand.get("skill")), {})
invocations = row.get("invocations", 0)
value = cand.get("offload_value", 0)
return {
"candidate_id": cand.get("candidate_id"),
"signature": sig,
"signature_source": source,
"signature_hash": signature_hash(sig),
"skill": cand.get("skill"),
"position": cand.get("position"),
"boundary_confidence": cand.get("boundary_confidence"),
"determinism_tests": cand.get("determinism_tests"),
"falsifiability": cand.get("falsifiability_triple"),
"measurement": {
"invocations": invocations,
"offload_waste": row.get("offload_waste", 0),
"offload_value": value,
"share_of_plugin": cand.get("share_of_audited_spend", 0.0),
"cost_per_invocation": int(value / invocations) if invocations else 0,
},
"measurement_strength": cand.get("measurement_strength"),
"digest_schema": cand.get("digest_schema"),
"escalation_path": cand.get("escalation_path"),
"issue": issue_map.get(cand.get("candidate_id")) or cand.get("issue"),
"verdict": cand.get("verdict"),
}
def build_snapshot(target, inventory, scan, classified, issue_map, explicit, run_id, repo):
plugin = inventory.get("plugin", {})
scan_total = scan.get("totals", {})
rows = classified.get("candidates", [])
filed = [c for c in rows if c.get("verdict") == "file"]
questions = [c for c in rows if c.get("verdict") == "boundary-question"]
notes = [
f"{c['candidate_id']}: {c['verdict']} — {c['reasons'][-1] if c.get('reasons') else ''}"
for c in rows if c.get("verdict") in ("no-offload", "drift-note")
]
return {
"run_id": run_id,
"rubric_version": classified.get("rubric_version"),
"tool_versions": {
"inference-arbitrage": tool_version(),
"cc-tokens": scan.get("tool_versions", {}).get("cc_tokens_path"),
},
"target": {
"name": target,
"version": plugin.get("version"),
"source": plugin.get("source"),
"repo": repo,
# Not in plan §6, but FR-5.3's `stale` is uncomputable without it:
# "the skill no longer exists" needs the skill list of the run it
# no longer exists in.
"skills": sorted({s.get("name") for s in inventory.get("skills", []) if s.get("name")}
| {a.get("name") for a in inventory.get("agents", []) if a.get("name")}),
},
"window": scan.get("window", {}),
"coverage": {
"ratio": scan.get("coverage", {}).get("ratio"),
"attributed_turns": scan.get("coverage", {}).get("attributed_turns"),
},
"totals": {
"weighted_tokens": scan_total.get("weighted_tokens", 0),
"invocations": scan_total.get("invocations", 0),
"share_mechanical": scan_total.get("mechanical_share", 0.0),
},
"candidates": [snapshot_candidate(c, scan, explicit, issue_map) for c in filed],
"boundary_questions": [snapshot_candidate(c, scan, explicit, issue_map) for c in questions],
"measurements_by_signature": measurements_by_signature(scan),
"notes": notes,
}
# --------------------------------------------------------------------------
# diff (FR-5.3, FR-5.4, FR-5.5)
# --------------------------------------------------------------------------
def rel(before, after):
"""None when the base is zero — a ratio against nothing is not a ratio.
Callers fall back to the sign of the absolute delta in that case."""
if not before:
return None
return round((after - before) / before, 4)
def metric_delta(prev_m, curr_m, key):
a = (prev_m or {}).get(key, 0)
b = (curr_m or {}).get(key, 0)
return {"from": a, "to": b, "delta": b - a, "rel": rel(a, b)}
def direction(primary, secondary):
"""Volume-normalized verdict (FR-5.4). cost-per-invocation leads: it is the
metric a quiet week cannot move. share-of-spend only breaks the tie, and
absolute tokens never vote."""
for m in (primary, secondary):
r = m["rel"]
if r is None:
# Base of zero: the ratio is undefined, so the sign of the move is
# all there is.
if m["delta"] > 0:
return "grown"
continue
if r > NOISE_BAND:
return "grown"
if r < -NOISE_BAND:
return "shrunk"
return "persisting"
def remeasured(prev_cand, curr_snap):
return (curr_snap.get("measurements_by_signature") or {}).get(
prev_cand.get("signature_hash"))
def departed_status(prev_cand, curr_snap, issue_state):
"""A candidate that is no longer on the list. `resolved` is the strict
FR-5.5 conjunction — issue closed AND remeasured below threshold with
enough invocations. Everything else that merely stopped being reported is
`claimed-fixed-unconfirmed`, with the missing leg named: committed is not
deployed, and closed is not fixed."""
skills = set((curr_snap.get("target") or {}).get("skills") or [])
if skills and prev_cand.get("skill") not in skills:
return "stale", (f"skill {prev_cand.get('skill')} is absent from "
f"{curr_snap['target'].get('name')} "
f"{curr_snap['target'].get('version')} — the candidate no longer exists")
issue = prev_cand.get("issue")
state = issue_state.get(issue) if issue else None
m = remeasured(prev_cand, curr_snap)
below = bool(m and m["invocations"] >= MIN_INVOCATIONS
and m["share_of_plugin"] < MIN_SHARE_OF_SPEND)
if state == "closed" and below:
return "resolved", (f"{issue} closed and remeasured at "
f"{m['share_of_plugin']:.2%} of spend over "
f"{m['invocations']} invocations")
if state == "closed":
if m:
return "claimed-fixed-unconfirmed", (
f"{issue} closed but remeasurement is {m['invocations']} invocation(s) at "
f"{m['share_of_plugin']:.2%} — needs >= {MIN_INVOCATIONS} invocations below "
f"{MIN_SHARE_OF_SPEND:.0%}")
return "claimed-fixed-unconfirmed", (
f"{issue} closed but the signature was not remeasured in this window")
if issue:
return "claimed-fixed-unconfirmed", (
f"dropped off the candidate list while {issue} is still {state or 'unknown'}")
return "claimed-fixed-unconfirmed", (
"dropped off the candidate list with no linked issue and no remeasurement below threshold")
def row_of(cand, tier, status, reason, matched_by, prev_cand=None):
prev_m = (prev_cand or {}).get("measurement")
curr_m = cand.get("measurement") if cand else None
return {
"candidate_id": (cand or prev_cand).get("candidate_id"),
"skill": (cand or prev_cand).get("skill"),
"tier": tier,
"status": status,
"matched_by": matched_by,
"issue": (cand or prev_cand).get("issue"),
"primary": {
"cost_per_invocation": metric_delta(prev_m, curr_m, "cost_per_invocation"),
"share_of_plugin": metric_delta(prev_m, curr_m, "share_of_plugin"),
},
"context_only": {
"offload_value": metric_delta(prev_m, curr_m, "offload_value"),
"invocations": metric_delta(prev_m, curr_m, "invocations"),
},
"reason": reason,
}
def tiered(snap):
for c in snap.get("candidates", []):
yield c, "candidate"
for c in snap.get("boundary_questions", []):
yield c, "boundary-question"
def diff_snapshots(prev, curr, issue_state):
prev_all = list(tiered(prev))
idx = build_index([c for c, _ in prev_all])
prev_tier = {c["candidate_id"]: t for c, t in prev_all}
seen = set()
rows = []
for cand, tier in tiered(curr):
match, by = find_match(cand, idx)
if not match:
rows.append(row_of(cand, tier, "new", "not present in the previous snapshot", None))
continue
seen.add(match["candidate_id"])
if match.get("signature_hash") != cand.get("signature_hash"):
reason = (f"tool-sequence signature changed "
f"{match.get('signature_hash')} -> {cand.get('signature_hash')} "
f"(matched by {by}); the step was refactored, so the cost series is not continuous")
rows.append(row_of(cand, tier, "signature-drifted", reason, by, match))
continue
row = row_of(cand, tier, "persisting", None, by, match)
row["status"] = direction(row["primary"]["cost_per_invocation"],
row["primary"]["share_of_plugin"])
row["reason"] = (f"cost/invocation {row['primary']['cost_per_invocation']['rel']}, "
f"share {row['primary']['share_of_plugin']['rel']} "
f"(band +/-{NOISE_BAND:.0%})")
rows.append(row)
for cand, tier in prev_all:
if cand["candidate_id"] in seen:
continue
status, reason = departed_status(cand, curr, issue_state)
rows.append(row_of(None, tier, status, reason, None, cand))
summary = {s: sum(1 for r in rows if r["status"] == s) for s in STATUSES}
unaddressed = [
r for r in rows
if r["tier"] == "candidate" and r["status"] in ("persisting", "grown", "signature-drifted")
]
promotions = [
r for r in rows
if r["tier"] == "boundary-question" and r["status"] == "grown"
and prev_tier.get(r["candidate_id"]) == "boundary-question"
]
prev_vol = prev.get("totals", {}).get("weighted_tokens", 0)
curr_vol = curr.get("totals", {}).get("weighted_tokens", 0)
return {
"target": curr.get("target", {}).get("name"),
"from": {"run_id": prev.get("run_id"), "window": prev.get("window"),
"rubric_version": prev.get("rubric_version"),
"target_version": prev.get("target", {}).get("version")},
"to": {"run_id": curr.get("run_id"), "window": curr.get("window"),
"rubric_version": curr.get("rubric_version"),
"target_version": curr.get("target", {}).get("version")},
"rubric_comparable": prev.get("rubric_version") == curr.get("rubric_version"),
"volume": {
"weighted_tokens": {"from": prev_vol, "to": curr_vol, "rel": rel(prev_vol, curr_vol)},
"note": ("absolute totals are context only — a quieter window lowers every absolute "
"number without anything improving. Per-candidate verdicts come from "
"cost-per-invocation and share-of-spend."),
},
"candidates": rows,
"summary": summary,
"unaddressed": [r["candidate_id"] for r in unaddressed],
"unaddressed_issues": sorted({r["issue"] for r in unaddressed if r["issue"]}),
"promotions": [r["candidate_id"] for r in promotions],
}
# --------------------------------------------------------------------------
# rendering
# --------------------------------------------------------------------------
def fmt_rel(r):
if r is None:
return "n/a"
return f"{r:+.0%}"
def candidate_table(rows):
if not rows:
return "— none —\n"
out = ["| Candidate | Status | Cost/invocation | Share of spend | Issue |",
"|---|---|---|---|---|"]
for r in rows:
cpi = r["primary"]["cost_per_invocation"]
share = r["primary"]["share_of_plugin"]
out.append(
f"| `{r['candidate_id']}` | {r['status']} | "
f"{cpi['from']:,} → {cpi['to']:,} ({fmt_rel(cpi['rel'])}) | "
f"{share['from']:.2%} → {share['to']:.2%} ({fmt_rel(share['rel'])}) | "
f"{r['issue'] or '—'} |")
return "\n".join(out) + "\n"
def render_run(snap, diffdoc):
t = snap["target"]
w = snap.get("window", {})
lines = [
f"# {t['name']} — audit {snap['run_id']}",
"",
f"- Target: `{t['name']}` {t.get('version') or '?'} ({t.get('source') or '?'})"
+ (f", repo `{t['repo']}`" if t.get("repo") else ""),
f"- Window: {w.get('since')} → {w.get('until')}",
f"- Attribution coverage: {(snap['coverage'].get('ratio') or 0):.1%} "
f"({snap['coverage'].get('attributed_turns')} attributed turns) — "
"hook-driven and unattributed agent turns are excluded, so every figure below is a lower bound.",
f"- Mechanical share of audited spend: {(snap['totals'].get('share_mechanical') or 0):.1%}",
f"- Rubric v{snap.get('rubric_version')}, inference-arbitrage "
f"v{snap.get('tool_versions', {}).get('inference-arbitrage')}",
"",
"## Candidates",
"",
]
for c in snap.get("candidates", []):
m = c["measurement"]
lines += [
f"### `{c['candidate_id']}`",
"",
f"- Skill: `{c['skill']}` — position {c['position']}, confidence {c['boundary_confidence']}",
f"- {m['offload_value']:,} weighted tokens of offload value over {m['invocations']} "
f"invocations ({m['cost_per_invocation']:,}/invocation, {m['share_of_plugin']:.2%} of audited spend)",
f"- Evidence: {c.get('measurement_strength')}; signature `{c['signature_hash']}` "
f"({c.get('signature_source')})",
f"- Issue: {c.get('issue') or 'not yet filed'}",
"",
]
if not snap.get("candidates"):
lines += ["Nothing to offload here.", ""]
if snap.get("boundary_questions"):
lines += ["## Boundary questions", "",
"Reported, not filed — the rubric could not reach `high` confidence with an "
"overrule case.", ""]
for c in snap["boundary_questions"]:
lines.append(f"- `{c['candidate_id']}` (`{c['skill']}`) — {c['boundary_confidence']}, "
f"{c['measurement']['share_of_plugin']:.2%} of spend")
lines.append("")
if snap.get("notes"):
lines += ["## Notes", ""] + [f"- {n}" for n in snap["notes"]] + [""]
if diffdoc:
lines += ["## Change since " + str(diffdoc["from"]["run_id"]), "",
candidate_table(diffdoc["candidates"]), ""]
if not diffdoc["rubric_comparable"]:
lines.append(f"**Rubric version changed** "
f"({diffdoc['from']['rubric_version']} → {diffdoc['to']['rubric_version']}): "
"classifications either side of the change were not graded by the same "
"table and must not be compared naively.\n")
return "\n".join(lines).rstrip() + "\n"
def render_latest(snap, diffdoc, history):
t = snap["target"]
lines = [
f"# {t['name']} — latest audit",
"",
f"Regenerated from `Data/{t['name']}/snapshots.jsonl` — do not edit by hand.",
"",
f"- Latest run: `{snap['run_id']}` against {t['name']} {t.get('version') or '?'}",
f"- Runs recorded: {len(history)} (first `{history[0]['run_id']}`)",
f"- Open candidates: {len(snap.get('candidates', []))}; "
f"boundary questions: {len(snap.get('boundary_questions', []))}",
"",
]
if diffdoc:
s = diffdoc["summary"]
lines += [
"## Trend",
"",
f"Against `{diffdoc['from']['run_id']}`. Audited volume moved "
f"{fmt_rel(diffdoc['volume']['weighted_tokens']['rel'])} — context only; the verdicts "
"below are per-invocation and share-of-spend.",
"",
"| Status | Count |", "|---|---|",
] + [f"| {k} | {v} |" for k, v in s.items() if v] + [""]
if diffdoc["unaddressed"]:
refs = ", ".join(diffdoc["unaddressed_issues"]) or "none filed"
lines += [f"**{len(diffdoc['unaddressed'])} prior recommendation(s) still unaddressed** "
f"— {refs}", ""]
if diffdoc["promotions"]:
lines += ["**Boundary questions recurring with growing cost** (candidates for promotion): "
+ ", ".join(f"`{c}`" for c in diffdoc["promotions"]), ""]
lines += [candidate_table(diffdoc["candidates"]), ""]
lines += ["## Open candidates", "",
"| Candidate | Skill | Share of spend | Cost/invocation | Issue |",
"|---|---|---|---|---|"]
for c in snap.get("candidates", []):
m = c["measurement"]
lines.append(f"| `{c['candidate_id']}` | `{c['skill']}` | {m['share_of_plugin']:.2%} | "
f"{m['cost_per_invocation']:,} | {c.get('issue') or '—'} |")
if not snap.get("candidates"):
lines.append("| — none — | | | | |")
lines += ["", "## Run history", "", "| Run | Target version | Coverage | Candidates |",
"|---|---|---|---|"]
for h in history[-10:]:
lines.append(f"| `{h['run_id']}` | {h['target'].get('version') or '?'} | "
f"{(h.get('coverage', {}).get('ratio') or 0):.1%} | "
f"{len(h.get('candidates', []))} |")
return "\n".join(lines).rstrip() + "\n"
def regenerate(store, target, history):
"""Dated page appended, Latest.md regenerated (plan §5)."""
snap = history[-1]
diffdoc = diff_snapshots(history[-2], snap, {}) if len(history) > 1 else None
day = str(snap["run_id"])[:10]
return [
write_text(os.path.join(audits_dir(store, target), f"{day}.md"), render_run(snap, diffdoc)),
write_text(os.path.join(audits_dir(store, target), "Latest.md"),
render_latest(snap, diffdoc, history)),
]
# --------------------------------------------------------------------------
# commands
# --------------------------------------------------------------------------
def load_json(path):
with open(path) as fh:
return json.load(fh)
def cmd_write(args):
inventory = load_json(args.from_files[0])
scan = load_json(args.from_files[1])
classified = load_json(args.classified)
issue_map = load_json(args.issue_map) if args.issue_map else {}
explicit = load_json(args.signatures) if args.signatures else {}
run_id = args.run_id or datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
history = load_snapshots(args.store, args.target)
if any(s.get("run_id") == run_id for s in history):
sys.exit(f"error: run_id {run_id} already recorded for {args.target}")
snap = build_snapshot(args.target, inventory, scan, classified,
issue_map, explicit, run_id, args.repo)
path = append_snapshot(args.store, args.target, snap)
history.append(snap)
pages = regenerate(args.store, args.target, history)
return {"run_id": run_id, "snapshots": path, "pages": pages,
"candidates": len(snap["candidates"]),
"boundary_questions": len(snap["boundary_questions"])}
def cmd_list(args):
history = load_snapshots(args.store, args.target)
return {
"target": args.target,
"runs": [{
"run_id": s.get("run_id"),
"target_version": s.get("target", {}).get("version"),
"rubric_version": s.get("rubric_version"),
"window": s.get("window", {}),
"coverage": s.get("coverage", {}).get("ratio"),
"weighted_tokens": s.get("totals", {}).get("weighted_tokens"),
"candidates": len(s.get("candidates", [])),
"boundary_questions": len(s.get("boundary_questions", [])),
} for s in history],
}
def cmd_diff(args):
history = load_snapshots(args.store, args.target)
if len(history) < 2:
sys.exit(f"error: {args.target} has {len(history)} snapshot(s); diff needs 2")
curr = history[-1]
if args.against:
prev = next((s for s in history if s.get("run_id") == args.against), None)
if prev is None:
sys.exit(f"error: no snapshot {args.against} for {args.target}")
else:
prev = history[-2]
issue_state = load_json(args.issue_state) if args.issue_state else {}
return diff_snapshots(prev, curr, issue_state)
def cmd_pages(args):
"""What to publish to the wiki, and with what content. The caller performs
the writes — see the module docstring on why this script never does."""
history = load_snapshots(args.store, args.target)
if not history:
sys.exit(f"error: no snapshots recorded for {args.target}")
snap = history[-1]
day = str(snap["run_id"])[:10]
root = store_root(args.store)
pages = [{
"page": f"Data/{args.target}/snapshots.jsonl",
"title": f"Data/{args.target}/snapshots.jsonl",
"append_only": True,
"local": snapshots_path(args.store, args.target),
}, {
"page": f"Audits/{args.target}/{day}",
"title": f"Audits/{args.target}/{day}",
"append_only": False,
"local": os.path.join(audits_dir(args.store, args.target), f"{day}.md"),
}, {
"page": f"Audits/{args.target}/Latest",
"title": f"Audits/{args.target}/Latest",
"append_only": False,
"local": os.path.join(audits_dir(args.store, args.target), "Latest.md"),
}]
for p in pages:
with open(p["local"]) as fh:
p["content"] = fh.read()
p["local"] = os.path.relpath(p["local"], root)
return {"target": args.target, "store": root, "run_id": snap["run_id"], "pages": pages}
def cmd_import(args):
"""Seed the local store from a snapshots.jsonl fetched off the wiki, so a
cold cache still diffs against real history."""
with open(args.snapshots) as fh:
incoming = [json.loads(line) for line in fh if line.strip()]
# The only place a snapshot arrives from outside this script: a wiki page a
# human can edit. Validate here rather than defensively everywhere after.
for i, snap in enumerate(incoming, 1):
missing = [k for k in ("run_id", "target", "window", "coverage", "totals",
"candidates", "boundary_questions") if k not in snap]
if missing:
sys.exit(f"error: {args.snapshots}:{i}: not a snapshot record, missing {missing}")
existing = {s.get("run_id") for s in load_snapshots(args.store, args.target)}
added = [s for s in incoming if s.get("run_id") not in existing]
for snap in added:
append_snapshot(args.store, args.target, snap)
history = load_snapshots(args.store, args.target)
pages = regenerate(args.store, args.target, history) if history else []
return {"target": args.target, "imported": len(added), "skipped": len(incoming) - len(added),
"runs": len(history), "pages": pages}
def main():
ap = argparse.ArgumentParser(prog="audit-snapshot")
ap.add_argument("--store", default=DEFAULT_STORE,
help=f"local mirror of the wiki store (default {DEFAULT_STORE})")
ap.add_argument("--json", action="store_true", help="compact single-line JSON")
sub = ap.add_subparsers(dest="cmd", required=True)
w = sub.add_parser("write", help="append one snapshot and regenerate the audit pages")
w.add_argument("--target", required=True)
w.add_argument("--from", dest="from_files", nargs=2, required=True,
metavar=("INVENTORY", "SCAN"))
w.add_argument("--classified", required=True, help="bin/boundary-classify output")
w.add_argument("--issue-map", help='JSON {candidate_id: "owner/repo#N"}')
w.add_argument("--signatures", help="JSON {candidate_id: [tool signature, ...]}")
w.add_argument("--repo", help="target's repo, owner/repo")
w.add_argument("--run-id", help="override the run id (tests; default is now, UTC)")
w.set_defaults(func=cmd_write)
ls = sub.add_parser("list", help="list recorded runs")
ls.add_argument("--target", required=True)
ls.set_defaults(func=cmd_list)
d = sub.add_parser("diff", help="diff the latest run against a prior one")
d.add_argument("--target", required=True)
d.add_argument("--against", help="run_id to compare against (default: previous)")
d.add_argument("--issue-state", help='JSON {"owner/repo#N": "open"|"closed"}')
d.set_defaults(func=cmd_diff)
p = sub.add_parser("pages", help="emit the wiki pages to publish, with content")
p.add_argument("--target", required=True)
p.set_defaults(func=cmd_pages)
i = sub.add_parser("import", help="seed the store from a wiki-fetched snapshots.jsonl")
i.add_argument("--target", required=True)
i.add_argument("--snapshots", required=True)
i.set_defaults(func=cmd_import)
args = ap.parse_args()
result = args.func(args)
json.dump(result, sys.stdout, indent=None if args.json else 2)
sys.stdout.write("\n")
if __name__ == "__main__":
main()
design/tasks.md
+12 -8
View File
@@ -105,16 +105,20 @@ These block design, not just code. Answers go into `Methodology/Calibration.md`.
## Phase 5 — Snapshots and trend
- [ ] **5.1** `bin/audit-snapshot write|list|diff`; schema per PLAN §6.
- [ ] **5.2** Candidate identity resolution (issue → signature → slug) and
- [x] **5.1** `bin/audit-snapshot write|list|diff`; schema per PLAN §6.
- [x] **5.2** Candidate identity resolution (issue → signature → slug) and
signature-drift detection.
- [ ] **5.3** Volume-normalized diffing: cost-per-invocation and
- [x] **5.3** Volume-normalized diffing: cost-per-invocation and
share-of-spend primary, absolute tokens as context only (FR-5.4).
- [ ] **5.4** `resolved` vs `claimed-fixed-unconfirmed` logic (FR-5.5).
- [ ] **5.5** Wiki read/write through `cluster:gitea-agent`; `Latest.md`
regeneration.
- [ ] **5.6** `skills/offload-trend/SKILL.md`.
- [ ] **5.7** `tests/snapshot.test.sh` — two synthetic snapshots, assert the diff.
- [x] **5.4** `resolved` vs `claimed-fixed-unconfirmed` logic (FR-5.5).
- [x] **5.5** Wiki read/write; `Latest.md` regeneration. Resolved as: the
script is offline and filesystem-only, mirroring the wiki tree into a
local store; `audit-snapshot pages` emits what to publish and the caller
(skill, or `cluster:gitea-agent`) performs the wiki writes, and
`audit-snapshot import` seeds a cold cache from a wiki-fetched
`snapshots.jsonl`. See the `bin/audit-snapshot` docstring.
- [x] **5.6** `skills/offload-trend/SKILL.md`.
- [x] **5.7** `tests/snapshot.test.sh` — two synthetic snapshots, assert the diff.
## Phase 6 — Output paths
references/snapshot-schema.md
+71
View File
@@ -0,0 +1,71 @@
# Snapshot schema
One JSON object per audit run, one line per object, appended to
`Data/<target>/snapshots.jsonl`. Written by `bin/audit-snapshot write`;
plan §6 is the source, with the two additions noted at the bottom.
```jsonc
{
"run_id": "2026-07-29T14:02:11Z", // UTC, and the primary key
"rubric_version": "1.0.0", // from bin/boundary-classify
"tool_versions": {"inference-arbitrage": "0.3.0", "cc-tokens": "<path>"},
"target": {
"name": "anxious", "version": "0.31.0", "source": "oleks-local",
"repo": "oleks/claude-plugin-anxious",
"skills": ["anxious:steward-agent", "..."] // ADDITION, see below
},
"window": {"since": "...", "until": "...", "days": 7},
"coverage": {"ratio": 0.735, "attributed_turns": 812},
"totals": {"weighted_tokens": 41200000, "invocations": 60, "share_mechanical": 0.61},
"candidates": [{
"candidate_id": "anxious/steward/label-derivation",
"signature": ["list_issues(<str>)", "issue_read(<n>)"],
"signature_source": "explicit | scan-ngram | slug",
"signature_hash": "b41f...", // sha256 of the joined signature, 16 hex
"skill": "anxious:steward-agent",
"position": "llm-over-script-digest",
"boundary_confidence": "high",
"determinism_tests": {"T1": true, "...": true},
"falsifiability": {"signature": true, "examples": true, "overrule_case": true},
"measurement": {"invocations": 22, "offload_waste": 6900000, "offload_value": 8400000,
"share_of_plugin": 0.22, "cost_per_invocation": 381818},
"measurement_strength": "measured | thin | unmeasured",
"digest_schema": "...", "escalation_path": "...",
"issue": "oleks/claude-plugin-anxious#41",
"verdict": "file"
}],
"boundary_questions": [ /* same shape, verdict boundary-question */ ],
"measurements_by_signature": {"b41f...": { /* measurement */ }}, // ADDITION
"notes": ["<candidate>: no-offload — ..."]
}
```
`status` is **not** stored on a candidate. Status is a property of a *pair* of
snapshots, not of one, so it is computed by `diff` and only ever rendered.
Storing it would let a stale status outlive the comparison that produced it.
## Two additions to plan §6
**`target.skills`.** FR-5.3's `stale` means "the target's version changed such
that the skill no longer exists". That is uncomputable from a candidate list
alone — a skill absent from the list because it got cheap looks identical to
one absent because it was deleted. The skill inventory of each run is what
tells them apart.
**`measurements_by_signature`.** FR-5.5 requires a *later window's measured
cost* for a signature whose candidate is gone. Once a candidate drops off the
list nothing else in the snapshot carries its cost, so the resolution test
would be unfalsifiable. This records every recurring signature the scan saw,
candidate or not, keyed by the same hash used for identity.
## Identity and drift
`identity()` resolves in FR-5.2 order — issue, then signature hash, then slug —
and `diff` matches on all three in that order, recording which one hit in
`matched_by`. Matching on issue or slug while the hashes differ is
`signature-drifted`: the same tracked concern, a different tool sequence. The
cost series either side of a drift is not continuous.
A candidate whose `signature_source` is `slug` cannot drift by construction —
its hash is derived from its identity. Treat drift-silence on those as absence
of evidence.
skills/offload-trend/SKILL.md
+167
View File
@@ -0,0 +1,167 @@
---
name: offload-trend
description: Compare this audit against previous ones — did the plugin get worse, are prior recommendations being acted on, which boundary questions keep coming back with growing cost. Reports the volume-normalized trend, never absolute tokens as the headline. Trigger on "has this plugin got worse", "offload trend", "are my recommendations being acted on", "diff the last audit", "what did we say last time", "did anything get fixed", "still unaddressed".
allowed-tools: Read, Bash
---
# Offload trend — what changed since last time
Answers three questions about a target that has been audited before:
1. **Did it get worse?** — per-invocation cost and share of spend, not tokens.
2. **Is anything being acted on?** — which filed recommendations are still live.
3. **What did we downgrade that we should not have?** — boundary questions
recurring with growing cost.
This skill **does not run an audit.** It reads snapshots that `offload-audit`
already wrote. If the target has fewer than two snapshots, say so and stop —
"one audit is not a trend" is the correct answer, not a reason to go measure
something new.
---
## Step 1 — make sure the local store has the history
The store is a local mirror of this plugin's own wiki
(`kotkan/claude-plugin-inference-arbitrage`), default
`~/.cache/inference-arbitrage/wiki`. On a machine that has run audits it is
already populated. On a cold cache it is empty, and the history lives only on
the wiki.
```bash
${CLAUDE_PLUGIN_ROOT}/bin/audit-snapshot list --target <name>
```
If that returns no runs, fetch `Data/<target>/snapshots.jsonl` from the wiki
(`wiki_read`, or delegate to `cluster:gitea-agent`), write it to a file, and
seed the store:
```bash
${CLAUDE_PLUGIN_ROOT}/bin/audit-snapshot import --target <name> --snapshots <file>
```
`import` is idempotent — re-importing skips run ids already present.
## Step 2 — resolve the issue states
`resolved` is a conjunction, and one half of it is the tracker's opinion. Read
the state of every issue referenced by the previous snapshot's candidates —
`issue_read` on each, or delegate to `cluster:gitea-agent` — and write a map:
```json
{"oleks/claude-plugin-anxious#41": "closed",
"oleks/claude-plugin-anxious#43": "open"}
```
Skipping this is not neutral: without it every departed candidate reports
`claimed-fixed-unconfirmed`, which understates real progress.
## Step 3 — diff
```bash
${CLAUDE_PLUGIN_ROOT}/bin/audit-snapshot diff --target <name> \
[--against <run_id>] --issue-state <file>
```
Do not compute any of this by hand. The statuses, the deltas, the promotion
list and the unaddressed list are all in the output.
---
## Reading the output
### The statuses
| Status | Means |
|---|---|
| `new` | Not present in the compared snapshot. |
| `persisting` | Both volume-normalized metrics inside the ±10% noise band. |
| `grown` / `shrunk` | Cost per invocation moved beyond the band; share of spend breaks ties. |
| `signature-drifted` | The tool sequence changed. The step was refactored — the cost series either side is **not** continuous and must not be presented as one trend line. |
| `stale` | The skill is gone from the target's current version. Not a fix; nothing was measured. |
| `resolved` | Issue closed **and** remeasured under 2% of spend with ≥3 invocations. |
| `claimed-fixed-unconfirmed` | It stopped being reported, but the conjunction above does not hold. |
`claimed-fixed-unconfirmed` is the important one and the easiest to soften by
accident. It is this environment's `nixos-deploy-pending` distinction applied to
audits: **closed is not fixed**, the way committed is not deployed. Report it as
its own line, never folded into `resolved`, and quote the `reason` field —
it names exactly which leg is missing.
### The headline must be volume-normalized
Lead with **cost per invocation** and **share of audited spend**. Absolute
weighted tokens go in a parenthetical at most, always next to the volume
change from `volume.weighted_tokens.rel`.
The failure mode this exists to prevent: a week with half the usage halves
every absolute number, and a report led by absolutes reads as a 50%
improvement when nothing improved. If you are about to write "down 1.2M
tokens", check whether the window shrank first.
### Recommendations still unaddressed
`unaddressed` lists candidates still `persisting`, `grown` or
`signature-drifted`; `unaddressed_issues` lists their trackers. Report them in
full `owner/repo#num` form — a bare `#41` is unusable without knowing the repo.
State the count plainly: "3 of 5 prior recommendations are still unaddressed."
### Promotions
`promotions` lists boundary questions that recurred with growing cost. A
question that keeps coming back and keeps getting more expensive is evidence
the earlier `medium` was too cautious. Say so, and say what would settle it —
which determinism test was in doubt and what evidence would resolve it. Do
**not** file it from this skill; a promotion is a recommendation to re-run
`boundary-rubric` on it with the new evidence, and filing still goes through
the `offload-audit` path with a full overrule case.
### Rubric version
If `rubric_comparable` is false the two snapshots were graded by different
tables. Say it out loud and do not compare confidences across the boundary —
positions and measurements are still comparable, gradings are not.
---
## Step 4 — publish, if anything changed
`diff` is read-only. If you ran a fresh audit before this and want the wiki to
reflect it:
```bash
${CLAUDE_PLUGIN_ROOT}/bin/audit-snapshot pages --target <name>
```
That emits each wiki page's name and full content. **The script never writes
Gitea** — it has no credentials and opens no socket. Perform the writes with
`wiki_write` (or delegate to `cluster:gitea-agent`) on
`kotkan/claude-plugin-inference-arbitrage`:
- `Data/<target>/snapshots.jsonl` — append-only; the emitted content is the
whole file, so this is an `update` with the full body, not a patch.
- `Audits/<target>/<YYYY-MM-DD>` — the dated run page, created once.
- `Audits/<target>/Latest` — regenerated every run.
Two Gitea wiki quirks, both verified against this repo's wiki:
- Pass the page **title** (`Data/anxious/snapshots.jsonl`) to `create` and
`update`. Slashes and the `.jsonl` extension are both accepted.
- `delete` does **not** take the title — it takes the escaped `sub_url` the
write returned, URL-decoded (`Data/anxious/snapshots.jsonl.-`). Deleting by
title 404s. Read the page first and use its `sub_url` if you ever need to
remove one.
---
## What not to do
- **Do not re-derive numbers.** If the trend you want is not in the diff
output, that is a gap in `bin/audit-snapshot`, not an invitation to compute
it in prose. A plugin about offloading arithmetic to scripts doing its own
arithmetic by hand is the anti-pattern it reports on.
- **Do not treat a disappearance as a win.** Both `stale` and
`claimed-fixed-unconfirmed` mean "we stopped seeing it", which is not the
same as "it stopped costing".
- **"Nothing changed" is a complete answer.** A flat trend across two windows
is information; padding it with movement inside the noise band is not.
tests/snapshot.test.sh
Executable
+278
View File
@@ -0,0 +1,278 @@
#!/usr/bin/env bash
# bin/audit-snapshot: write / list / diff over two synthetic snapshots of one
# fake target, with hand-computable numbers. Every expectation below is
# arithmetic done by hand in the comments, not a golden file — a golden file
# would pass just as happily on a wrong-but-stable answer.
set -euo pipefail
cd "$(dirname "${BASH_SOURCE[0]}")"
BIN=../bin/audit-snapshot
tmp=$(mktemp -d)
trap 'rm -rf "$tmp"' EXIT
store=$tmp/store
fail=0
check() {
local what=$1 got=$2 want=$3
if [ "$got" = "$want" ]; then
echo " ok $what = $want"
else
echo " FAIL $what = $got (want $want)" >&2
fail=1
fi
}
# status <diff.json> <candidate_id>
status() {
python3 -c "
import json,sys
d=json.load(open(sys.argv[1]))
r=[c for c in d['candidates'] if c['candidate_id']==sys.argv[2]]
print(r[0]['status'] if r else 'ABSENT')" "$@"
}
metric() {
python3 -c "
import json,sys
d=json.load(open(sys.argv[1]))
r=[c for c in d['candidates'] if c['candidate_id']==sys.argv[2]][0]
print(r[sys.argv[3]][sys.argv[4]][sys.argv[5]])" "$@"
}
jq_path() {
python3 -c "
import json,sys
d=json.load(open(sys.argv[1]))
for k in sys.argv[2].split('.'):
d=d[int(k)] if k.isdigit() else d[k]
print(d)" "$@"
}
# --------------------------------------------------------------------------
# Fixtures. Two runs of a fake plugin 'synth'.
#
# Run 1 (2026-07-01), audited spend 10,000,000 weighted tokens:
# steady 20 invocations, offload_value 2,000,000 -> 100,000/inv, 20% share
# growing 10 invocations, offload_value 400,000 -> 40,000/inv, 4% share
# fixed 10 invocations, offload_value 1,000,000 -> 100,000/inv, 10% share
# gone 10 invocations, offload_value 500,000 -> 50,000/inv, 5% share
# drifter 10 invocations, offload_value 600,000 -> 60,000/inv, 6% share
# quiet (boundary question) 10 inv, 300,000 -> 30,000/inv, 3% share
#
# Run 2 (2026-07-08) is a QUIETER window: audited spend 5,000,000, half of run
# 1. Every absolute number falls. Volume-normalized, only `growing` grew.
# --------------------------------------------------------------------------
# inventory <file> <version> <skill...>
inventory() {
local out=$1 version=$2
shift 2
python3 - "$out" "$version" "$@" <<'EOF'
import json, sys
out, version, *skills = sys.argv[1:]
json.dump({"plugin": {"name": "synth", "version": version, "source": "oleks-local"},
"skills": [{"name": s} for s in skills], "agents": []}, open(out, "w"))
EOF
}
# scan <file> <total> then rows on stdin: skill invocations offload_value sig
scan() {
local out=$1 total=$2
shift 2
python3 - "$out" "$total" "$@" <<'EOF'
import json, sys
out, total, *rows = sys.argv[1:]
skills = []
for row in rows:
skill, inv, value, sig = row.split(":")
skills.append({"skill": skill, "invocations": int(inv),
"offload_waste": int(value), "offload_value": int(value),
"ngrams": [{"sig": sig.split(","), "recurrences": int(inv),
"invocations": int(inv), "waste": int(value)}]})
json.dump({"window": {"since": "2026-07-01", "until": "2026-07-08", "days": 7},
"plugin": "synth", "tool_versions": {"cc_tokens_path": "/synthetic/cc-tokens"},
"coverage": {"attributed_turns": 800, "candidate_turns": 1000, "ratio": 0.8},
"totals": {"weighted_tokens": int(total), "invocations": 60,
"mechanical_share": 0.61},
"skills": skills, "agents": []}, open(out, "w"))
EOF
}
# classified <file> <total> then rows: candidate_id:skill:value:share:verdict
classified() {
local out=$1
shift
python3 - "$out" "$@" <<'EOF'
import json, sys
out, *rows = sys.argv[1:]
cands = []
for row in rows:
cid, skill, value, share, verdict = row.split(":")
cands.append({"candidate_id": cid, "skill": skill, "position": "llm-over-script-digest",
"determinism_tests": {t: True for t in ("T1", "T2", "T3", "T4", "T5")},
"falsifiability_triple": {"signature": True, "examples": True,
"overrule_case": True},
"boundary_confidence": "high" if verdict == "file" else "medium",
"measurement_strength": "measured",
"share_of_audited_spend": float(share), "offload_value": int(value),
"escalation_path": "escalate", "digest_schema": "typed rows",
"verdict": verdict, "reasons": ["synthetic"]})
json.dump({"rubric_version": "1.0.0", "target": "synth", "candidates": cands}, open(out, "w"))
EOF
}
echo "== run 1: write =="
inventory "$tmp/inv1.json" 1.0.0 steady growing fixed gone drifter quiet
scan "$tmp/scan1.json" 10000000 \
"steady:20:2000000:list_issues,issue_read" \
"growing:10:400000:glob,read" \
"fixed:10:1000000:grep,read,read" \
"gone:10:500000:bash,bash" \
"drifter:10:600000:label_read,issue_read" \
"quiet:10:300000:read,read"
classified "$tmp/cls1.json" \
"synth/steady/a:steady:2000000:0.20:file" \
"synth/growing/a:growing:400000:0.04:file" \
"synth/fixed/a:fixed:1000000:0.10:file" \
"synth/gone/a:gone:500000:0.05:file" \
"synth/drifter/a:drifter:600000:0.06:file" \
"synth/quiet/a:quiet:300000:0.03:boundary-question"
cat >"$tmp/issues1.json" <<'EOF'
{"synth/fixed/a": "kotkan/claude-plugin-synth#1",
"synth/steady/a": "kotkan/claude-plugin-synth#2",
"synth/gone/a": "kotkan/claude-plugin-synth#3"}
EOF
$BIN --store "$store" write --target synth --run-id 2026-07-01T00:00:00Z \
--from "$tmp/inv1.json" "$tmp/scan1.json" --classified "$tmp/cls1.json" \
--issue-map "$tmp/issues1.json" --repo kotkan/claude-plugin-synth >"$tmp/w1.json"
check "run 1 candidates" "$(jq_path "$tmp/w1.json" candidates)" "5"
check "run 1 boundary questions" "$(jq_path "$tmp/w1.json" boundary_questions)" "1"
echo "== list after one run =="
$BIN --store "$store" list --target synth >"$tmp/l1.json"
check "runs" "$(jq_path "$tmp/l1.json" runs.0.run_id)" "2026-07-01T00:00:00Z"
check "weighted tokens" "$(jq_path "$tmp/l1.json" runs.0.weighted_tokens)" "10000000"
echo "== diff with one snapshot refuses =="
if $BIN --store "$store" diff --target synth >/dev/null 2>&1; then
echo " FAIL diff succeeded with a single snapshot" >&2
fail=1
else
echo " ok diff needs two snapshots"
fi
echo "== run 2: a quieter window =="
# `gone`'s skill is dropped from the target entirely -> stale.
# `drifter` keeps its candidate id but its tool sequence changes -> signature-drifted.
inventory "$tmp/inv2.json" 1.1.0 steady growing fixed drifter quiet newbie
scan "$tmp/scan2.json" 5000000 \
"steady:10:1000000:list_issues,issue_read" \
"growing:5:400000:glob,read" \
"fixed:5:50000:grep,read,read" \
"drifter:10:600000:label_read,issue_read,issue_write" \
"quiet:5:250000:read,read" \
"newbie:5:300000:web_fetch,read"
# steady : 1,000,000/10 = 100,000/inv (unchanged), share 20% -> persisting
# growing: 400,000/ 5 = 80,000/inv (+100%), share 8% -> grown
# fixed : dropped off the candidate list; issue #1 closed; remeasured 50,000
# over 5 invocations = 1% of 5,000,000, under the 2% threshold with
# >= 3 invocations -> resolved
# gone : skill absent from inventory 1.1.0 -> stale
# drifter: signature hash changes -> signature-drifted
# quiet : boundary question, 250,000/5 = 50,000/inv vs 30,000 (+67%) -> grown,
# so it is promoted
classified "$tmp/cls2.json" \
"synth/steady/a:steady:1000000:0.20:file" \
"synth/growing/a:growing:400000:0.08:file" \
"synth/drifter/a:drifter:600000:0.12:file" \
"synth/newbie/a:newbie:300000:0.06:file" \
"synth/quiet/a:quiet:250000:0.05:boundary-question"
cat >"$tmp/issues2.json" <<'EOF'
{"synth/steady/a": "kotkan/claude-plugin-synth#2",
"synth/drifter/a": "kotkan/claude-plugin-synth#4"}
EOF
$BIN --store "$store" write --target synth --run-id 2026-07-08T00:00:00Z \
--from "$tmp/inv2.json" "$tmp/scan2.json" --classified "$tmp/cls2.json" \
--issue-map "$tmp/issues2.json" --repo kotkan/claude-plugin-synth >/dev/null
cat >"$tmp/state.json" <<'EOF'
{"kotkan/claude-plugin-synth#1": "closed",
"kotkan/claude-plugin-synth#2": "open",
"kotkan/claude-plugin-synth#3": "closed"}
EOF
$BIN --store "$store" diff --target synth --issue-state "$tmp/state.json" >"$tmp/diff.json"
echo "== per-candidate statuses =="
check "steady" "$(status "$tmp/diff.json" synth/steady/a)" "persisting"
check "growing" "$(status "$tmp/diff.json" synth/growing/a)" "grown"
check "fixed" "$(status "$tmp/diff.json" synth/fixed/a)" "resolved"
check "gone" "$(status "$tmp/diff.json" synth/gone/a)" "stale"
check "drifter" "$(status "$tmp/diff.json" synth/drifter/a)" "signature-drifted"
check "newbie" "$(status "$tmp/diff.json" synth/newbie/a)" "new"
check "quiet" "$(status "$tmp/diff.json" synth/quiet/a)" "grown"
echo "== volume normalization: the window halved, nothing may read as progress =="
check "volume rel" "$(jq_path "$tmp/diff.json" volume.weighted_tokens.rel)" "-0.5"
# steady's absolute offload_value halved (2,000,000 -> 1,000,000) while its
# cost per invocation and share are flat. Absolute must not drive the verdict.
check "steady abs delta" "$(metric "$tmp/diff.json" synth/steady/a context_only offload_value delta)" "-1000000"
check "steady cost/inv rel" "$(metric "$tmp/diff.json" synth/steady/a primary cost_per_invocation rel)" "0.0"
check "growing cost/inv rel" "$(metric "$tmp/diff.json" synth/growing/a primary cost_per_invocation rel)" "1.0"
check "growing share rel" "$(metric "$tmp/diff.json" synth/growing/a primary share_of_plugin rel)" "1.0"
echo "== follow-through =="
check "unaddressed count" "$(python3 -c "import json;print(len(json.load(open('$tmp/diff.json'))['unaddressed']))")" "3"
check "unaddressed issues" "$(jq_path "$tmp/diff.json" unaddressed_issues.0)" "kotkan/claude-plugin-synth#2"
check "promotions" "$(jq_path "$tmp/diff.json" promotions.0)" "synth/quiet/a"
check "rubric comparable" "$(jq_path "$tmp/diff.json" rubric_comparable)" "True"
echo "== claimed-fixed-unconfirmed: closed issue, no remeasurement =="
# Same diff, but #1 closed with `fixed`'s signature absent from run 2's
# measurements. Rebuild run 2 without the `fixed` skill's scan row.
store2=$tmp/store2
scan "$tmp/scan2b.json" 5000000 \
"steady:10:1000000:list_issues,issue_read" \
"growing:5:400000:glob,read"
inventory "$tmp/inv2b.json" 1.1.0 steady growing fixed
classified "$tmp/cls2b.json" \
"synth/steady/a:steady:1000000:0.20:file" \
"synth/growing/a:growing:400000:0.08:file"
$BIN --store "$store2" write --target synth --run-id 2026-07-01T00:00:00Z \
--from "$tmp/inv1.json" "$tmp/scan1.json" --classified "$tmp/cls1.json" \
--issue-map "$tmp/issues1.json" >/dev/null
$BIN --store "$store2" write --target synth --run-id 2026-07-08T00:00:00Z \
--from "$tmp/inv2b.json" "$tmp/scan2b.json" --classified "$tmp/cls2b.json" >/dev/null
$BIN --store "$store2" diff --target synth --issue-state "$tmp/state.json" >"$tmp/diff2.json"
check "closed but unmeasured" "$(status "$tmp/diff2.json" synth/fixed/a)" "claimed-fixed-unconfirmed"
echo "== rendered pages =="
$BIN --store "$store" pages --target synth >"$tmp/pages.json"
check "jsonl page" "$(jq_path "$tmp/pages.json" pages.0.page)" "Data/synth/snapshots.jsonl"
check "dated page" "$(jq_path "$tmp/pages.json" pages.1.page)" "Audits/synth/2026-07-08"
check "latest page" "$(jq_path "$tmp/pages.json" pages.2.page)" "Audits/synth/Latest"
check "jsonl lines" "$(wc -l <"$store/Data/synth/snapshots.jsonl")" "2"
for want in "still unaddressed" "growing cost" "Open candidates" "Run history"; do
if grep -qF "$want" "$store/Audits/synth/Latest.md"; then
echo " ok Latest.md carries \"$want\""
else
echo " FAIL Latest.md missing \"$want\"" >&2
fail=1
fi
done
echo "== import is idempotent =="
store3=$tmp/store3
$BIN --store "$store3" import --target synth --snapshots "$store/Data/synth/snapshots.jsonl" >"$tmp/i1.json"
check "imported" "$(jq_path "$tmp/i1.json" imported)" "2"
$BIN --store "$store3" import --target synth --snapshots "$store/Data/synth/snapshots.jsonl" >"$tmp/i2.json"
check "re-import skipped" "$(jq_path "$tmp/i2.json" skipped)" "2"
check "still two runs" "$(wc -l <"$store3/Data/synth/snapshots.jsonl")" "2"
echo "== a re-run of the same run_id is refused =="
if $BIN --store "$store" write --target synth --run-id 2026-07-08T00:00:00Z \
--from "$tmp/inv2.json" "$tmp/scan2.json" --classified "$tmp/cls2.json" >/dev/null 2>&1; then
echo " FAIL duplicate run_id accepted" >&2
fail=1
else
echo " ok duplicate run_id refused"
fi
exit "$fail"