kotkan/claude-plugin-inference-arbitrage
3
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5bdc9846748bb9966dca98592b81f10a47c6ca98
claude-plugin-inference-arb…/bin/filing-plan
T
Oleks 5bdc984674 Fix markdown lint in the Phase 5 docs
2026-07-29 17:50:20 +03:00

451 lines
17 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""Decides, deterministically, what output path A does with each candidate:
open a new issue, comment on the one that already exists, or skip it — and
renders the exact body text for each (FR-6.1..FR-6.4, design/plan.md §10).
Everything here is a `pure-script` step by the plugin's own rubric: the verdicts
were already graded by bin/boundary-classify, the marker match is a literal
string comparison, and rendering a template is a total function of its inputs.
None of it may be done by inference — an audit that decides duplicates by
"looking at" the tracker is exactly the shape this plugin reports on, and
getting it wrong spams a real human's tracker.
This script never touches Gitea. It consumes the results of a search that
`anxious:issuer-agent` -> `cluster:gitea-agent` performed, and emits a plan that
the same chain executes. Two subcommands, in order:
filing-plan queries --classified c.json --judgments j.json [--target-path P]
filing-plan plan --classified c.json --judgments j.json --existing e.json ...
`queries` resolves the target repo (or reports that filing is unavailable) and
emits the marker searches to run. `plan` consumes their merged results.
"""
import argparse
import json
import os
import re
import subprocess
import sys
PLUGIN_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_TEMPLATE = os.path.join(PLUGIN_ROOT, "references", "issue-template.md")
DEFAULT_WORKSPACE = os.path.expanduser("~/projects/claude-plugins")
DEFAULT_GITEA_HOST = "git.oleks.space"
DEFAULT_WIKI_REPO = "kotkan/claude-plugin-inference-arbitrage"
OFFLOAD_LABEL = "token-offload"
TEMPLATE_SECTIONS = {"issue-title", "issue-body", "update-comment"}
# --------------------------------------------------------------------------
# template loading and rendering
# --------------------------------------------------------------------------
def load_templates(path):
"""Same contract as signals-catalog.md: one fenced block per recognized
`## <section>` heading, everything else prose."""
with open(path, encoding="utf-8") as fh:
text = fh.read()
out = {}
for m in re.finditer(r"^##\s+([a-z-]+)\s*$", text, re.MULTILINE):
name = m.group(1)
if name not in TEMPLATE_SECTIONS:
continue
block = re.search(r"```\S*\n(.*?)```", text[m.end():], re.DOTALL)
if block:
out[name] = block.group(1).strip("\n")
missing = TEMPLATE_SECTIONS - out.keys()
if missing:
sys.exit(f"error: {path} is missing section(s): {', '.join(sorted(missing))}")
return out
PLACEHOLDER = re.compile(r"\{\{([a-z_]+)\}\}")
def render(template, values):
"""Strict substitution. An unsupplied placeholder is an error, never an
empty string — a half-filled issue body is worse than a crashed audit."""
missing = sorted(
{k for k in PLACEHOLDER.findall(template) if values.get(k) in (None, "")}
)
if missing:
sys.exit(f"error: template placeholder(s) unsupplied: {', '.join(missing)}")
return PLACEHOLDER.sub(lambda m: str(values[m.group(1)]), template)
# --------------------------------------------------------------------------
# the marker (FR-6.3)
# --------------------------------------------------------------------------
def marker(candidate_id):
return f"<!-- ia-candidate: {candidate_id} -->"
def marker_re(candidate_id):
"""Tolerates whitespace drift inside the comment, nothing else. The id
itself must match exactly — a fuzzy id match would let two genuinely
different candidates collapse onto one issue."""
return re.compile(r"<!--\s*ia-candidate:\s*" + re.escape(candidate_id) + r"\s*-->")
def find_existing(candidate_id, issues):
"""Returns (issue_or_None, duplicates). Matches the body and every comment,
so an issue keeps its identity even if a human rewrites the body."""
pat = marker_re(candidate_id)
hits = []
for issue in issues:
haystacks = [issue.get("body") or ""]
haystacks += [c if isinstance(c, str) else (c.get("body") or "")
for c in issue.get("comments") or []]
if any(pat.search(h) for h in haystacks):
hits.append(issue)
if not hits:
return None, []
hits.sort(key=lambda i: i.get("number", 0))
return hits[0], hits[1:]
# --------------------------------------------------------------------------
# target repo resolution (FR-6.4)
# --------------------------------------------------------------------------
def git_remote(path):
if not os.path.isdir(os.path.join(path, ".git")):
return None
try:
out = subprocess.run(
["git", "-C", path, "remote", "get-url", "origin"],
capture_output=True, text=True, timeout=10,
)
except (OSError, subprocess.SubprocessError):
return None
return out.stdout.strip() if out.returncode == 0 else None
def parse_remote(url, host):
"""owner/repo if the remote points at our Gitea, else None."""
m = re.match(r"^(?:git@|ssh://git@|https://)([^/:]+)[:/](.+?)(?:\.git)?/?$", url)
if not m or m.group(1) != host:
return None
parts = m.group(2).split("/")
return "/".join(parts[-2:]) if len(parts) >= 2 else None
def resolve_repo(args, target_name):
"""(repo, filing, reason). `filing` is 'available' or 'unavailable'.
Order: explicit flag, the target checkout's own remote, then the workspace
sibling checkout — because an audit normally runs against
~/.claude/plugins/cache/... which carries no .git at all.
"""
if args.target_repo:
return args.target_repo, "available", None
tried = []
candidates = []
if args.target_path:
candidates.append(os.path.abspath(args.target_path))
if target_name and args.workspace:
sibling = os.path.join(args.workspace, target_name)
if sibling not in candidates:
candidates.append(sibling)
for path in candidates:
if not os.path.isdir(path):
continue
url = git_remote(path)
if url is None:
tried.append(f"{path}: no git remote")
continue
repo = parse_remote(url, args.gitea_host)
if repo:
return repo, "available", None
tried.append(f"{path}: remote {url} is not on {args.gitea_host}")
if not candidates:
detail = "no --target-repo, --target-path, or workspace checkout to resolve from"
else:
detail = "; ".join(tried) or "no checkout found at " + ", ".join(candidates)
return None, "unavailable", (
f"no writable repo for target {target_name!r} ({detail}). "
"Path A is skipped; the candidate is reported in the run summary only (FR-6.4)."
)
# --------------------------------------------------------------------------
# rendering values
# --------------------------------------------------------------------------
TICKS = {True: "✓", False: "✗"}
def fmt_tests(tests):
return " ".join(f"{t} {TICKS[tests.get(t) is True]}" for t in ("T1", "T2", "T3", "T4", "T5"))
def fmt_tokens(n):
if n >= 1_000_000:
return f"{n / 1_000_000:.1f}M"
if n >= 1_000:
return f"{n / 1_000:.1f}k"
return str(n)
def fmt_examples(examples):
return "\n".join(f"- {e}" for e in examples)
def fmt_evidence(strength, m):
inv, turns = m.get("invocations", 0), m.get("attributed_turns", 0)
if strength == "measured":
return f"`measured` — {inv} invocations, {turns} attributed turns."
if strength == "thin":
return (
f"`thin` — only {inv} invocation(s) and {turns} attributed turns in this "
"window. **The cost figure above is arithmetic, not evidence.** The "
"correctness case stands on the rubric; the value claim is unproven."
)
return (
"`unmeasured` — no transcript history for this target in the window. This "
"candidate rests on the plugin definition alone (FR-2.4)."
)
def fmt_cost(cand, cls, window):
"""Only states what was actually measured. Absent signals are omitted, never
guessed at — a fabricated MTR in a filed issue is unrecoverable."""
m = cand.get("measurement") or {}
bits = []
value = cls.get("offload_value") or 0
share = cls.get("share_of_audited_spend") or 0
inv = m.get("invocations", 0)
head = f"{fmt_tokens(value)} weighted tokens"
if inv:
head += f" over {inv} invocation{'s' if inv != 1 else ''}"
if window:
head += f" ({window})"
if share:
head += f", {share:.0%} of this plugin's audited spend"
bits.append(head + ".")
stats = []
if m.get("mtr") is not None:
stats.append(f"MTR {m['mtr']:.2f}")
if m.get("judgment_density") is not None:
stats.append(f"judgment density {m['judgment_density']:.2f}")
if stats:
bits.append(" ".join(stats) + ".")
ngram = m.get("ngram") or {}
if ngram.get("sig") and ngram.get("recurrences"):
seq = " -> ".join(ngram["sig"])
bits.append(f"Recurring sequence observed {ngram['recurrences']}x: {seq}.")
return " ".join(bits)
def headline(cand):
if cand.get("headline"):
return cand["headline"]
# Last path segment of the candidate id is its slug by construction
# (design/plan.md §6: "<plugin>/<skill>/<slug>").
return cand["candidate_id"].rsplit("/", 1)[-1].replace("-", " ")
# --------------------------------------------------------------------------
# the plan
# --------------------------------------------------------------------------
def index_judgments(doc):
return {c["candidate_id"]: c for c in doc.get("candidates", [])}
def build(args, cls_doc, jud_doc, existing, templates):
target = cls_doc.get("target") or jud_doc.get("target")
repo, filing, reason = resolve_repo(args, target)
judgments = index_judgments(jud_doc)
actions, warnings = [], []
for cls in cls_doc.get("candidates", []):
cid = cls["candidate_id"]
verdict = cls.get("verdict")
if verdict != "file":
actions.append({
"candidate_id": cid,
"action": "skip",
"reason": f"verdict {verdict!r} — path A files only 'file' verdicts "
"(FR-4.2 gate, enforced by boundary-classify)",
})
continue
if filing == "unavailable":
actions.append({
"candidate_id": cid,
"action": "skip",
"filing": "unavailable",
"reason": reason,
})
continue
cand = judgments.get(cid)
if cand is None:
sys.exit(f"error: candidate {cid!r} has verdict 'file' but no judgments entry")
f = cand.get("falsifiability") or {}
values = {
"candidate_id": cid,
"headline": headline(cand),
"skill": cand.get("skill") or cls.get("skill"),
"measured_cost": fmt_cost(cand, cls, args.window),
"evidence": fmt_evidence(cls.get("measurement_strength"), cand.get("measurement") or {}),
"position": cls.get("position"),
"boundary_confidence": cls.get("boundary_confidence"),
"determinism_tests": fmt_tests(cls.get("determinism_tests") or {}),
"signature": f.get("signature"),
"examples": fmt_examples(f.get("examples") or []),
"overrule_case": f.get("overrule_case"),
"escalation_path": cand.get("escalation_path"),
"digest_schema": cand.get("digest_schema") or "n/a — position 1, the whole step moves.",
"audit_page": args.audit_page,
"wiki_repo": args.wiki_repo,
"rubric_version": cls_doc.get("rubric_version", "1.0.0"),
"run_date": args.run_date,
"window": args.window,
"status": args.status,
}
found, dupes = find_existing(cid, existing)
if dupes:
numbers = ", ".join("#%s" % d.get("number") for d in [found] + dupes)
warnings.append(
f"{cid}: marker found on {1 + len(dupes)} issues ({numbers}) — "
"commenting on the lowest-numbered one; the rest want a human merge"
)
if found:
actions.append({
"candidate_id": cid,
"action": "comment",
"issue": f"{repo}#{found['number']}",
"issue_number": found["number"],
"issue_state": found.get("state", "unknown"),
"marker": marker(cid),
"body": render(templates["update-comment"], values),
"note": "existing issue is closed; issuer decides whether to reopen"
if found.get("state") == "closed" else None,
})
else:
actions.append({
"candidate_id": cid,
"action": "file",
"marker": marker(cid),
"proposed_title": render(templates["issue-title"], values),
"body": render(templates["issue-body"], values),
"required_labels": [OFFLOAD_LABEL],
"labels_note": "the four-axis anxious taxonomy is issuer's call; "
f"{OFFLOAD_LABEL} is required and issuer creates it if absent",
})
counts = {a: sum(1 for x in actions if x["action"] == a) for a in ("file", "comment", "skip")}
return {
"target": target,
"repo": repo,
"filing": filing,
"filing_unavailable_reason": reason,
"run_date": args.run_date,
"window": args.window,
"audit_page": args.audit_page,
"delegation": "offload-analyst -> Agent(anxious:issuer-agent) -> cluster:gitea-agent -> Gitea",
"summary": {
"to_file": counts["file"],
"to_comment": counts["comment"],
"skipped": counts["skip"],
"searched_issues": len(existing),
},
"warnings": warnings,
"actions": actions,
}
def build_queries(args, cls_doc, jud_doc):
target = cls_doc.get("target") or jud_doc.get("target")
repo, filing, reason = resolve_repo(args, target)
ids = [c["candidate_id"] for c in cls_doc.get("candidates", []) if c.get("verdict") == "file"]
doc = {
"target": target,
"repo": repo,
"filing": filing,
"filing_unavailable_reason": reason,
"candidates_to_search": ids,
}
if filing == "available" and ids:
doc["queries"] = {
"full_text": [
{"tool": "search_issues", "repo": repo, "state": "all",
"q": f"ia-candidate: {cid}"} for cid in ids
],
"label_scan": {"tool": "list_issues", "repo": repo, "state": "all",
"labels": [OFFLOAD_LABEL]},
}
doc["note"] = (
"Run BOTH and merge into one array of "
"{number, state, title, body, comments[]} for `filing-plan plan --existing`. "
"The label scan is authoritative; the full-text query is the fast path and "
"may miss on indexer tokenization. See references/issue-template.md."
)
return doc
def main():
ap = argparse.ArgumentParser(prog="filing-plan")
sub = ap.add_subparsers(dest="cmd", required=True)
def common(p):
p.add_argument("--classified", required=True, help="bin/boundary-classify output")
p.add_argument("--judgments", required=True, help="the analyst's judgments JSON")
p.add_argument("--target-repo", help="owner/repo, if already known")
p.add_argument("--target-path", help="the audited plugin's directory")
p.add_argument("--workspace", default=DEFAULT_WORKSPACE)
p.add_argument("--gitea-host", default=DEFAULT_GITEA_HOST)
p.add_argument("--json", action="store_true", help="compact single-line JSON")
q = sub.add_parser("queries", help="resolve the repo and emit the marker searches")
common(q)
p = sub.add_parser("plan", help="decide file/comment/skip per candidate")
common(p)
p.add_argument("--existing", required=True,
help="JSON array of issues from the marker search. Required even when "
"empty ([]): defaulting to 'nothing exists' would duplicate on re-run.")
p.add_argument("--audit-page", required=True, help="e.g. Audits/anxious/2026-07-29")
p.add_argument("--window", required=True, help="e.g. 2026-07-22 -> 2026-07-29")
p.add_argument("--run-date", required=True)
p.add_argument("--wiki-repo", default=DEFAULT_WIKI_REPO)
p.add_argument("--status", default="persisting",
help="snapshot-diff status for the update comment (audit-snapshot diff)")
p.add_argument("--template", default=DEFAULT_TEMPLATE)
args = ap.parse_args()
with open(args.classified) as fh:
cls_doc = json.load(fh)
with open(args.judgments) as fh:
jud_doc = json.load(fh)
if args.cmd == "queries":
out = build_queries(args, cls_doc, jud_doc)
else:
with open(args.existing) as fh:
existing = json.load(fh)
if not isinstance(existing, list):
sys.exit("error: --existing must be a JSON array of issues")
out = build(args, cls_doc, jud_doc, existing, load_templates(args.template))
json.dump(out, sys.stdout, indent=None if args.json else 2, ensure_ascii=False)
sys.stdout.write("\n")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink