claude-plugin-inference-arbitrage/bin/ia_store.py

"""Snapshot store access and candidate-identity resolution (FR-5.2), shared by
`bin/audit-snapshot` (which writes and diffs snapshots) and
`bin/stability-classify` (which reads them to gate filing, FR-4.5).

It exists so that identity is computed exactly once. The gate's whole job is to
find the *same* candidate in an earlier window; if it hashed a signature even
slightly differently from the writer, it would silently find nothing and pass
every unstable candidate through — a gate that fails open and looks like it
worked. Copy-pasting `signature_hash` between the two scripts is therefore not
a style problem, it is the failure mode.

Imported by name from `bin/`, which is `sys.path[0]` for a script run as
`bin/<tool>`.
"""

import hashlib
import json
import os
import sys

DEFAULT_STORE = "~/.cache/inference-arbitrage/wiki"

# spec §5.2.7's filing threshold — single definition (kotkan/claude-plugin-
# inference-arbitrage#14). bin/boundary-classify imports this rather than
# keeping its own copy, so a candidate must clear the same bar to be called
# fixed, or stable, that it had to clear to be filed. Changing it here changes
# it everywhere; the two can no longer drift apart silently.
MIN_SHARE_OF_SPEND = 0.02
MIN_INVOCATIONS = 3


# --------------------------------------------------------------------------
# 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


# --------------------------------------------------------------------------
# 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


def tiered(snap):
    """Both tiers of a snapshot. A candidate that was a boundary question last
    window and a filing candidate this one is the same candidate, so identity
    matching must see both piles."""
    for c in snap.get("candidates", []):
        yield c, "candidate"
    for c in snap.get("boundary_questions", []):
        yield c, "boundary-question"


# --------------------------------------------------------------------------
# signatures derived from a scan
# --------------------------------------------------------------------------


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 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"