Oleks
88fc58b2fd
ci/woodpecker/push/test Pipeline was successful
The stability gate treated every evidence flip as symmetric, and that produced a false downgrade on the candidate it was written for (kotkan/claude-plugin-inference-arbitrage#15). Re-measuring anxious/agent-wip/release-policy-derivation over four real windows with the same script gave 0 -> 2 -> 12 -> 15 invocations at 7/14/27/60 days: monotonic accumulation of a step that runs about once every two days, not a label flipping about. The 7-day window had not caught an unstable candidate, it had failed to observe a real one. So a thin/unmeasured reading now only overturns a `measured` one when its window is at least 90% as wide (observed since->until span). Below that the comparison takes a third verdict, `insufficient-window`: not filed, not downgraded to a boundary question, no FR-6.5 self-report, and carried forward with the wider snapshot left as the standing comparison point. The parity is relative rather than an absolute day count because the adequate width is a property of the candidate's invocation rate, which the auditor does not know in advance. Unchanged, and tested: the reverse direction (thin prior -> measured now) still downgrades, a flip between comparably wide windows still downgrades, and a threshold crossing with `measured` on both sides is never excused by narrowness. An unreadable window is not an exemption either — the flip stands and window_parity records that the check could not run. - bin/stability-classify: window_days/window_parity/deferrable, the insufficient-window verdict, stability_deferrals, window_parity on findings too, insufficient_window + window_parity_ratio in the gate block - bin/audit-snapshot: record deferrals in the snapshot notes so the row is visible on the wiki page instead of silently dropped - design/spec.md: FR-4.5.1 as a testable requirement; S4 restated - design/rubric.md: the third boundary in §5b - skills/offload-audit/SKILL.md: do not file and do not self-report a deferral - tests: real 60-day anxious fixture; cases (f)/(f2) on real data, (j)-(n) synthetic — 7 test files pass Verified against the live store (~/.cache/inference-arbitrage/wiki): the real 7.0d-vs-26.49d comparison now reports insufficient-window with zero findings, while the real 33.4d-vs-26.49d one reports stable at 13.87% and files normally.
1133 lines
50 KiB
Python
Executable File
1133 lines
50 KiB
Python
Executable File
#!/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.
|
|
|
|
MEMORY (FR-9)
|
|
-------------
|
|
`memory-queries` and `memory-notes` are the same bridge pointed at mempalace
|
|
instead of the wiki: filesystem in, JSON out, no MCP call ever made here. The
|
|
caller runs `search` with the queries verbatim and `add_drawer` with the
|
|
payloads verbatim. No judgment lives in what text gets written — which is the
|
|
whole point, because a plugin that narrates its own bookkeeping at inference
|
|
time is the thing this plugin files issues about (FR-8.1).
|
|
|
|
The memory renderers are deliberately NOT `render_run`/`render_latest`. Those
|
|
emit `notes`, which is caller-supplied free text (`--note`), and
|
|
`falsifiability` prose is one field away. A wiki page lives in this plugin's
|
|
own repo; a drawer lands in a wing shared with unrelated projects. So the
|
|
memory renderers read from a fixed allowlist of structured snapshot fields —
|
|
ids, positions, booleans, counts, ratios — and no free-text field crosses into
|
|
a drawer at all (FR-3.5, FR-9.3). Allowlist, never blocklist: a blocklist is
|
|
one new snapshot field away from leaking.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
from ia_store import (DEFAULT_STORE, MIN_INVOCATIONS, MIN_SHARE_OF_SPEND,
|
|
audits_dir, build_index, candidate_signature, find_match,
|
|
load_snapshots, scan_rows, signature_hash, snapshots_path,
|
|
store_root, tiered)
|
|
|
|
# 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",
|
|
# the step ran in one window and not the other: no continuous
|
|
# series, so no direction (FR-5.4)
|
|
"unmeasured")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# store
|
|
# --------------------------------------------------------------------------
|
|
|
|
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
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 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 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 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),
|
|
# None, never 0, when the step did not run: a window in which a
|
|
# candidate was never invoked has no cost-per-invocation to report,
|
|
# and folding that to 0 makes "did not run" indistinguishable from
|
|
# "improved to zero" — the quiet-window artefact FR-5.4 exists to
|
|
# prevent, one level up from absolute tokens.
|
|
"cost_per_invocation": int(value / invocations) if invocations else None,
|
|
},
|
|
"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,
|
|
extra_notes=()):
|
|
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"]
|
|
# Derived notes first, caller-supplied after. The caller's are outcomes of
|
|
# the filing pass — `filing: unavailable` and why (FR-6.4) — which this
|
|
# script cannot see and which have to reach the wiki page, not just the
|
|
# user's terminal.
|
|
# `insufficient-window` is in this list, not in `candidates` or
|
|
# `boundary_questions`: an FR-4.5 deferral is neither a filed candidate nor
|
|
# a question put to a human, and the wider snapshot it was deferred against
|
|
# remains the standing comparison point. Recording it as a note keeps the
|
|
# deferral visible on the wiki page instead of silently dropping the row.
|
|
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", "insufficient-window")
|
|
] + list(extra_notes)
|
|
|
|
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),
|
|
"invocation_caveat": scan_total.get("invocation_caveat"),
|
|
},
|
|
"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 missing measurement dict means the candidate was absent from that
|
|
# snapshot entirely (new, or departed) and 0 is the right base. An explicit
|
|
# None means the candidate was present but not invoked — that is an
|
|
# absence of measurement and must not be folded to a number.
|
|
a = (prev_m or {}).get(key, 0)
|
|
b = (curr_m or {}).get(key, 0)
|
|
if a is None or b is None:
|
|
# At least one window has no measurement for this metric. There is no
|
|
# delta to take against an absence, and inventing one is how a window
|
|
# the step sat out becomes a reported improvement.
|
|
return {"from": a, "to": b, "delta": None, "rel": None, "unmeasured": True}
|
|
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."""
|
|
if primary.get("unmeasured"):
|
|
# The step was not invoked in one of the two windows, so the series is
|
|
# not continuous and no direction can be read from it.
|
|
return "unmeasured"
|
|
for m in (primary, secondary):
|
|
if m.get("unmeasured"):
|
|
continue
|
|
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 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 fmt_cpi(v):
|
|
"""`not run`, never 0 — the step had no invocations in that window."""
|
|
return "not run" if v is None else f"{v:,}"
|
|
|
|
|
|
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"{fmt_cpi(cpi['from'])} → {fmt_cpi(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')}",
|
|
]
|
|
invocation_caveat = snap["totals"].get("invocation_caveat")
|
|
if invocation_caveat:
|
|
lines.append(f"- Note: {invocation_caveat}, so per-invocation figures below are "
|
|
"correspondingly inflated")
|
|
lines += [
|
|
"",
|
|
"## 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 ({fmt_cpi(m['cost_per_invocation'])}/invocation, "
|
|
f"{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"{fmt_cpi(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)),
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# memory (FR-9)
|
|
# --------------------------------------------------------------------------
|
|
|
|
# Target-scoped: the audited plugin's own manifest name, mirroring FR-6.1's
|
|
# "the finding belongs on the target's surface, not this plugin's". The room is
|
|
# dedicated and namespaced by this plugin's name so that a wing collision — an
|
|
# audited plugin sharing a name with an unrelated wing, which really happens
|
|
# (`cluster`, `imagex`) — is a shelving overlap and never a content collision.
|
|
MEMORY_ROOM = "inference-arbitrage-audits"
|
|
|
|
# Cross-target: this plugin's own accumulated judgment, in the wing its own
|
|
# development history already lives in. Written every run regardless of target.
|
|
MEMORY_CROSS_WING = "claude-plugins"
|
|
MEMORY_CROSS_ROOM = "inference-arbitrage-lessons"
|
|
|
|
MEMORY_ADDED_BY = "inference-arbitrage"
|
|
|
|
# mempalace's search takes at most 250 characters of query.
|
|
QUERY_MAX = 250
|
|
|
|
REDACTION_NOTE = (
|
|
"Shapes and counts only — no transcript content, file contents, command "
|
|
"arguments, or user prose (FR-3.5). Rendered mechanically by "
|
|
"`audit-snapshot memory-notes`; nothing here was written at inference time."
|
|
)
|
|
|
|
ADVISORY_NOTE = (
|
|
"Results are advisory context for grading only. They never override "
|
|
"bin/boundary-classify (FR-4.2) or bin/stability-classify (FR-4.5), and a "
|
|
"prior drawer is never a reason to file, skip a marker search, or re-grade "
|
|
"a verdict (FR-9.2)."
|
|
)
|
|
|
|
|
|
def chunk_query(prefix, words):
|
|
"""Deterministic packing of names into queries that fit QUERY_MAX."""
|
|
out, cur = [], prefix
|
|
for w in words:
|
|
if len(cur) + 1 + len(w) > QUERY_MAX:
|
|
out.append(cur)
|
|
cur = prefix
|
|
cur = f"{cur} {w}" if cur else w
|
|
if cur != prefix:
|
|
out.append(cur)
|
|
return out
|
|
|
|
|
|
def surface_names(inventory, snap):
|
|
"""Skill and agent names, from the inventory if the caller has one (the
|
|
read step runs before any snapshot exists) and from the last snapshot
|
|
otherwise."""
|
|
if inventory:
|
|
names = {s.get("name") for s in inventory.get("skills", [])}
|
|
names |= {a.get("name") for a in inventory.get("agents", [])}
|
|
return sorted(n for n in names if n)
|
|
return sorted((snap or {}).get("target", {}).get("skills") or [])
|
|
|
|
|
|
def memory_queries(target, names):
|
|
q = [
|
|
{"wing": target, "room": MEMORY_ROOM,
|
|
"query": f"{target} offload audit declined candidates gotchas calibration"},
|
|
{"wing": MEMORY_CROSS_WING, "room": MEMORY_CROSS_ROOM,
|
|
"query": "offload audit calibration lessons false positives coverage dilution"},
|
|
{"wing": MEMORY_CROSS_WING, "room": MEMORY_CROSS_ROOM,
|
|
"query": f"{target} offload audit rubric verdict"},
|
|
]
|
|
for text in chunk_query(f"{target} offload candidates", names):
|
|
q.append({"wing": target, "room": MEMORY_ROOM, "query": text})
|
|
return q
|
|
|
|
|
|
def pct(x):
|
|
return f"{(x or 0):.1%}"
|
|
|
|
|
|
def share(x):
|
|
return f"{(x or 0):.2%}"
|
|
|
|
|
|
def tests_line(tests):
|
|
tests = tests or {}
|
|
return " ".join(f"{t} {'PASS' if tests.get(t) is True else 'FAIL'}"
|
|
for t in ("T1", "T2", "T3", "T4", "T5"))
|
|
|
|
|
|
def triple_line(f):
|
|
f = f or {}
|
|
return ", ".join(
|
|
f"{k.replace('_', ' ')} {'yes' if f.get(k) is True else 'no'}"
|
|
for k in ("signature", "examples", "overrule_case"))
|
|
|
|
|
|
def decline_reason(c, downgraded_ids=frozenset()):
|
|
"""Why the gate refused, read off the structured grade — never off the
|
|
graders' prose `reasons`, whose provenance is mixed.
|
|
|
|
Checks stay in FR-4.2's own order (overrule case, tests, confidence)
|
|
first — a candidate can be BOTH ungraded-for-filing AND coincidentally
|
|
named in a stability finding built from unrelated synthetic data, and
|
|
those real disqualifications must still win. Only the generic fallback at
|
|
the end — "did not clear the filing threshold" — is replaced when the
|
|
candidate is a genuine FR-4.5 downgrade, because a candidate that reaches
|
|
that fallback cleared every earlier check on its own grade: complete
|
|
triple, all determinism tests, `high` confidence. It cannot have failed
|
|
the filing threshold on its own merits and ALSO be one `stability-classify`
|
|
downgraded — real downgrades only ever touch candidates that already
|
|
cleared `boundary-classify`'s gate (verdict `file`), so reaching this line
|
|
while flagged in `downgraded_ids` means the fallback was always the wrong
|
|
reason for it and FR-4.5 is the real one. Fix for
|
|
kotkan/claude-plugin-inference-arbitrage#16, which contradicted the
|
|
correct FR-4.5 section rendered a few lines below it in the same drawer.
|
|
"""
|
|
f = c.get("falsifiability") or {}
|
|
tests = c.get("determinism_tests") or {}
|
|
if f.get("overrule_case") is not True:
|
|
return ("no overrule case — the falsifiability triple is incomplete, so the "
|
|
"hard FR-4.2 gate refused it")
|
|
failed = [t for t in ("T1", "T2", "T3", "T4", "T5") if tests.get(t) is not True]
|
|
if failed:
|
|
return (f"determinism test(s) {', '.join(failed)} did not hold — the cut is not "
|
|
"yet located")
|
|
if c.get("boundary_confidence") != "high":
|
|
return f"boundary confidence {c.get('boundary_confidence')} — reported, not filed"
|
|
if c.get("candidate_id") in downgraded_ids:
|
|
return ("cleared boundary-classify's gate on its own grade — downgraded "
|
|
"afterward by the evidence-stability gate (FR-4.5); see below")
|
|
return "did not clear the filing threshold in this window"
|
|
|
|
|
|
def cand_row(c):
|
|
m = c.get("measurement") or {}
|
|
return (f"| `{c.get('candidate_id')}` | `{c.get('skill')}` | {c.get('position')} | "
|
|
f"{c.get('boundary_confidence')} | {c.get('measurement_strength')} | "
|
|
f"{share(m.get('share_of_plugin'))} | {fmt_cpi(m.get('cost_per_invocation'))} | "
|
|
f"{c.get('issue') or '—'} |")
|
|
|
|
|
|
def render_memory_run(snap):
|
|
t = snap.get("target", {})
|
|
w = snap.get("window", {})
|
|
cov = snap.get("coverage", {})
|
|
tot = snap.get("totals", {})
|
|
lines = [
|
|
f"# inference-arbitrage audit — {t.get('name')} {t.get('version') or '?'} — run {snap.get('run_id')}",
|
|
"",
|
|
REDACTION_NOTE,
|
|
"",
|
|
f"- Target: `{t.get('name')}` {t.get('version') or '?'} ({t.get('source') or '?'})"
|
|
+ (f", repo `{t.get('repo')}`" if t.get("repo") else ""),
|
|
f"- Window: {w.get('since')} → {w.get('until')}",
|
|
f"- Attribution coverage: {pct(cov.get('ratio'))} of the turns in sessions where this "
|
|
f"plugin appeared ({cov.get('attributed_turns')} attributed turns) — a dilution "
|
|
"measure, not completeness.",
|
|
f"- Mechanical share of audited spend: {pct(tot.get('share_mechanical'))}",
|
|
f"- Audited volume: {tot.get('weighted_tokens', 0):,} weighted tokens over "
|
|
f"{tot.get('invocations', 0)} invocations",
|
|
f"- Rubric v{snap.get('rubric_version')}, inference-arbitrage "
|
|
f"v{snap.get('tool_versions', {}).get('inference-arbitrage')}",
|
|
"",
|
|
"## Filed candidates",
|
|
"",
|
|
]
|
|
header = ["| Candidate | Skill | Position | Confidence | Evidence | Share of spend "
|
|
"| Cost/invocation | Issue |", "|---|---|---|---|---|---|---|---|"]
|
|
filed = snap.get("candidates") or []
|
|
if filed:
|
|
lines += header + [cand_row(c) for c in filed]
|
|
else:
|
|
lines.append("— none — nothing cleared the filing gate in this window.")
|
|
lines += ["", "## Boundary questions — reported, never filed", ""]
|
|
questions = snap.get("boundary_questions") or []
|
|
if questions:
|
|
lines += header + [cand_row(c) for c in questions]
|
|
else:
|
|
lines.append("— none —")
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def render_memory_judgments(snap, findings):
|
|
t = snap.get("target", {})
|
|
questions = snap.get("boundary_questions") or []
|
|
if not questions and not findings:
|
|
return None
|
|
downgraded_ids = {f.get("candidate_id") for f in findings}
|
|
lines = [
|
|
f"# inference-arbitrage — judgment calls on {t.get('name')} (run {snap.get('run_id')})",
|
|
"",
|
|
"Why candidates were declined, so a later audit does not re-propose them without "
|
|
"new evidence. Every reason below is read off the structured grade, not off prose.",
|
|
"",
|
|
REDACTION_NOTE,
|
|
"",
|
|
]
|
|
if questions:
|
|
lines += ["## Declined — reported as boundary questions, never filed", ""]
|
|
for c in questions:
|
|
m = c.get("measurement") or {}
|
|
lines += [
|
|
f"- `{c.get('candidate_id')}` (`{c.get('skill')}`) — position "
|
|
f"{c.get('position')}, confidence {c.get('boundary_confidence')}, evidence "
|
|
f"{c.get('measurement_strength')} at {share(m.get('share_of_plugin'))} of "
|
|
f"audited spend over {m.get('invocations')} invocations.",
|
|
f" - Determinism tests: {tests_line(c.get('determinism_tests'))}",
|
|
f" - Falsifiability triple: {triple_line(c.get('falsifiability'))}",
|
|
f" - Declined because: {decline_reason(c, downgraded_ids)}",
|
|
]
|
|
lines.append("")
|
|
if findings:
|
|
lines += ["## Evidence-stability downgrades (FR-4.5)", "",
|
|
"The auditor described the same candidate two different ways depending on the "
|
|
"date range. That is a defect in this plugin, not a fact about the target.", ""]
|
|
for f in findings:
|
|
prior, curr = f.get("prior") or {}, f.get("current") or {}
|
|
lines += [
|
|
f"- `{f.get('candidate_id')}` (`{f.get('skill')}`), matched by "
|
|
f"{f.get('matched_by')}: {prior.get('measurement_strength')} at "
|
|
f"{share(prior.get('share_of_audited_spend'))} in run {prior.get('run_id')}, "
|
|
f"{curr.get('measurement_strength')} at "
|
|
f"{share(curr.get('share_of_audited_spend'))} in this window.",
|
|
f" - Self-reported on {(f.get('self_report') or {}).get('repo')}, marker "
|
|
f"`{(f.get('self_report') or {}).get('marker')}`",
|
|
]
|
|
lines.append("")
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def counted(items, key):
|
|
out = {}
|
|
for c in items:
|
|
out[c.get(key)] = out.get(c.get(key), 0) + 1
|
|
return ", ".join(f"{k} {v}" for k, v in sorted(out.items(), key=lambda kv: str(kv[0]))) or "none"
|
|
|
|
|
|
def render_memory_lessons(snap, findings):
|
|
t = snap.get("target", {})
|
|
cov = snap.get("coverage", {})
|
|
tot = snap.get("totals", {})
|
|
filed = snap.get("candidates") or []
|
|
questions = snap.get("boundary_questions") or []
|
|
every = filed + questions
|
|
|
|
lines = [
|
|
f"# inference-arbitrage — what the {t.get('name')} audit adds to the rubric's "
|
|
f"track record (run {snap.get('run_id')})",
|
|
"",
|
|
"Cross-target calibration record, written on every run regardless of outcome, so "
|
|
"this plugin's judgment accumulates independently of which plugin was audited.",
|
|
"",
|
|
REDACTION_NOTE,
|
|
"",
|
|
f"- Target: `{t.get('name')}` {t.get('version') or '?'} ({t.get('source') or '?'})",
|
|
f"- Rubric v{snap.get('rubric_version')}, inference-arbitrage "
|
|
f"v{snap.get('tool_versions', {}).get('inference-arbitrage')}",
|
|
f"- Coverage dilution {pct(cov.get('ratio'))} over {cov.get('attributed_turns')} "
|
|
f"attributed turns; mechanical share {pct(tot.get('share_mechanical'))}",
|
|
f"- Verdicts: {len(filed)} filed, {len(questions)} boundary questions",
|
|
f"- Positions seen: {counted(every, 'position')}",
|
|
f"- Evidence strengths: {counted(every, 'measurement_strength')}",
|
|
f"- Stability gate: {len(findings)} downgrade(s) this run",
|
|
"",
|
|
"## What this run says about the rubric",
|
|
"",
|
|
]
|
|
said = []
|
|
if not filed and (tot.get("share_mechanical") or 0) >= 0.2:
|
|
said.append(f"Nothing cleared the gate despite {pct(tot.get('share_mechanical'))} of "
|
|
"audited spend being mechanical — the conservative filing threshold held "
|
|
"against a target that looked like a candidate on the headline number.")
|
|
if not every:
|
|
said.append("No candidates at either tier: this target's boundary is already cut where "
|
|
"the rubric would put it.")
|
|
if findings:
|
|
said.append(f"{len(findings)} candidate(s) had window-dependent evidence, caught "
|
|
"mechanically by the stability gate rather than by human review.")
|
|
if len(questions) > len(filed):
|
|
said.append("More candidates were declined than filed at this rubric version — the "
|
|
"asymmetry (under-scripting costs money, over-scripting costs correctness) "
|
|
"is being applied, not just documented.")
|
|
if (cov.get("ratio") or 0) < 0.05:
|
|
said.append(f"Coverage dilution of {pct(cov.get('ratio'))} is normal for this "
|
|
"environment's long multi-topic sessions and is not a failed scan; the "
|
|
"honest completeness signal is the absolute count "
|
|
f"({cov.get('attributed_turns')} attributed turns).")
|
|
if not said:
|
|
said.append("Nothing anomalous: verdicts, evidence strengths and coverage all fell "
|
|
"inside the ranges the rubric was calibrated on.")
|
|
lines += [f"- {s}" for s in said]
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def memory_facts(snap, findings):
|
|
t = snap.get("target", {})
|
|
name = t.get("name")
|
|
day = str(snap.get("run_id"))[:10]
|
|
facts = [{"subject": name, "predicate": "audited_by", "object": "inference-arbitrage",
|
|
"valid_from": day}]
|
|
for c in snap.get("candidates") or []:
|
|
facts.append({"subject": c.get("candidate_id"), "predicate": "classified_as",
|
|
"object": c.get("position"), "valid_from": day})
|
|
if c.get("issue"):
|
|
facts.append({"subject": c.get("candidate_id"), "predicate": "filed_as",
|
|
"object": c.get("issue"), "valid_from": day})
|
|
for c in snap.get("boundary_questions") or []:
|
|
facts.append({"subject": c.get("candidate_id"), "predicate": "declined_as",
|
|
"object": "boundary-question", "valid_from": day})
|
|
for f in findings:
|
|
facts.append({"subject": f.get("candidate_id"), "predicate": "downgraded_by",
|
|
"object": "evidence-stability gate FR-4.5", "valid_from": day})
|
|
return facts
|
|
|
|
|
|
def memory_drawers(snap, findings):
|
|
t = snap.get("target", {})
|
|
name = t.get("name")
|
|
day = str(snap.get("run_id"))[:10]
|
|
src = f"inference-arbitrage/Audits/{name}/{day}.md"
|
|
out = [{
|
|
"kind": "run-summary",
|
|
"wing": name,
|
|
"room": MEMORY_ROOM,
|
|
"added_by": MEMORY_ADDED_BY,
|
|
"source_file": src,
|
|
"content": render_memory_run(snap),
|
|
}]
|
|
judgments = render_memory_judgments(snap, findings)
|
|
if judgments:
|
|
out.append({
|
|
"kind": "judgment-calls",
|
|
"wing": name,
|
|
"room": MEMORY_ROOM,
|
|
"added_by": MEMORY_ADDED_BY,
|
|
"source_file": src.replace(".md", ".judgments.md"),
|
|
"content": judgments,
|
|
})
|
|
out.append({
|
|
"kind": "cross-target-lessons",
|
|
"wing": MEMORY_CROSS_WING,
|
|
"room": MEMORY_CROSS_ROOM,
|
|
"added_by": MEMORY_ADDED_BY,
|
|
"source_file": src.replace(".md", ".lessons.md"),
|
|
"content": render_memory_lessons(snap, findings),
|
|
})
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 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, args.note)
|
|
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_memory_queries(args):
|
|
"""What to search mempalace for before grading (FR-9.1). Runs before any
|
|
snapshot exists, so it takes the inventory rather than the store."""
|
|
inventory = load_json(args.inventory) if args.inventory else None
|
|
history = load_snapshots(args.store, args.target)
|
|
names = surface_names(inventory, history[-1] if history else None)
|
|
return {
|
|
"target": args.target,
|
|
"wings": {"target": args.target, "cross_target": MEMORY_CROSS_WING},
|
|
"rooms": {"target": MEMORY_ROOM, "cross_target": MEMORY_CROSS_ROOM},
|
|
"queries": memory_queries(args.target, names),
|
|
"advisory": ADVISORY_NOTE,
|
|
}
|
|
|
|
|
|
def cmd_memory_notes(args):
|
|
"""What to write to mempalace after a run (FR-9.1). Same contract as
|
|
`pages`: the caller makes the MCP calls with these payloads verbatim."""
|
|
history = load_snapshots(args.store, args.target)
|
|
if not history:
|
|
sys.exit(f"error: no snapshots recorded for {args.target}")
|
|
snap = history[-1]
|
|
findings = (load_json(args.stable).get("stability_findings", []) if args.stable else [])
|
|
return {
|
|
"target": args.target,
|
|
"run_id": snap["run_id"],
|
|
"drawers": memory_drawers(snap, findings),
|
|
"facts": memory_facts(snap, findings),
|
|
"note": "Call add_drawer once per drawer and kg_add once per fact, with these "
|
|
"fields verbatim. Do not summarize, reword, or add commentary — the "
|
|
"content is already the exact text to store.",
|
|
}
|
|
|
|
|
|
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.add_argument("--note", action="append", default=[], metavar="TEXT",
|
|
help="add a note to the snapshot and its summary page; repeatable. "
|
|
"Carries filing outcomes the audit knows and this script does not, "
|
|
"notably FR-6.4's `filing: unavailable` and its reason")
|
|
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)
|
|
|
|
mq = sub.add_parser("memory-queries",
|
|
help="emit the mempalace searches to run before grading (FR-9)")
|
|
mq.add_argument("--target", required=True)
|
|
mq.add_argument("--inventory", help="bin/plugin-inventory output, for skill/agent names")
|
|
mq.set_defaults(func=cmd_memory_queries)
|
|
|
|
mn = sub.add_parser("memory-notes",
|
|
help="emit the mempalace drawers and facts to write, with content (FR-9)")
|
|
mn.add_argument("--target", required=True)
|
|
mn.add_argument("--stable", help="bin/stability-classify output, for stability_findings")
|
|
mn.set_defaults(func=cmd_memory_notes)
|
|
|
|
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()
|