claude-plugin-inference-arbitrage/bin/stability-classify

#!/usr/bin/env python3
"""The evidence-stability gate (FR-4.5, rubric §5b): refuses to file a candidate
whose evidence label is an artifact of which measurement window happened to be
audited.

WHY THIS IS A GATE AND NOT A CAVEAT
-----------------------------------
Phase 7 filed `anxious/agent-wip/release-policy-derivation` on a 30-day window
as `measurement_strength: measured`, 11.94% of audited spend — a confident,
concrete cost claim. The identical candidate, classified over a 7-day window,
came back `thin`, 0.00% of spend. Nothing about the plugin changed between the
two; only the date range did.

A human reading either issue body alone has no way to know the other exists.
That is the `rubric.md` §4a failure — fast, confident, and wrong, with nobody
watching — reached by the plugin's own headline number. It was caught once, by
a human running a second skeptical review, which is not a repeatable control:
the next candidate gets filed only if someone happens to be suspicious that day.

So stability becomes mechanical, in the same register as FR-4.2's overrule-case
gate: a candidate whose evidence flips across windows is downgraded to a
boundary question and never filed, and the flip is emitted as a finding against
this plugin's own repo (FR-8.2), because a confidence label that moves with the
calendar is a bug in the auditor, not a fact about the target.

WHAT COUNTS AS A FLIP
---------------------
Two things, either of which disqualifies:

  1. `measurement_strength` differs between the windows (`measured` vs `thin`
     vs `unmeasured`). This is the label the issue body speaks in, so a change
     here changes what a human is told.
  2. `share_of_audited_spend` lands on opposite sides of MIN_SHARE_OF_SPEND.
     That threshold is the filing gate itself (spec §5.2.7): if one window would
     have filed on value and the other would not, the value claim is not
     load-bearing evidence, it is a window artifact.

Ordinary movement within a band is NOT a flip. A candidate that measures 8% one
window and 5% the next is stable — same label, same side of the threshold — and
that variation is what `audit-snapshot diff` exists to report as a trend. This
gate fires only when the *description a human receives* changes.

A NARROW WINDOW IS NOT EVIDENCE OF INSTABILITY EITHER
-----------------------------------------------------
The gate as first written treated every flip as symmetric, and that produced a
false downgrade on the very candidate that motivated it
(kotkan/claude-plugin-inference-arbitrage#15). Re-measuring
`anxious/agent-wip/release-policy-derivation` across four real windows with the
same script showed a monotonic accumulation, not a flip:

    7 days   ->  0 invocations,  0.00% of spend  -> thin
    14 days  ->  2 invocations,  0.72%           -> thin
    27 days  -> 12 invocations, 11.73%           -> measured
    60 days  -> 15 invocations, 13.87%           -> measured

The candidate runs about once every two days. A 7-day window did not observe an
unstable candidate; it failed to observe a real one. The wider the window, the
higher the reading — the 60-day pass had every opportunity to dilute the share
back down and instead raised it. Calling that "the confidence label moves with
the calendar" gets the direction of the artifact backwards: the *narrow* window
was the artifact, and the gate punished the candidate for it.

So a weak reading only gets to overturn a `measured` one when it was taken over
a comparably wide window. Concretely, all three must hold:

  * the prior reading is `measured`,
  * the current reading is `thin` or `unmeasured`,
  * the current window is narrower than WINDOW_PARITY_RATIO of the prior
    window's observed width (`until - since`),

and then the comparison is not read as instability at all. The candidate takes
a third verdict, `insufficient-window`: it is neither filed nor downgraded, no
self-report is emitted, and it is carried forward — the wider prior snapshot
stays the standing comparison point, so the next audit run over an adequate
window compares against it rather than against this window's under-sample.

The parity is relative, not an absolute day count, on purpose. "At least 21
days" would be a guess about candidate frequency dressed up as a threshold —
this candidate needs ~27 days, a once-a-week step would need more, and a
once-an-hour step is fully measured in one. What the gate can know without
guessing is that a window materially narrower than the one that produced the
`measured` reading cannot fairly contradict it. WINDOW_PARITY_RATIO is 0.9
rather than 1.0 because window bounds are observed session spans, not requested
ranges, so two nominally identical windows differ by hours.

Three things stay downgrades, unchanged:

  1. the reverse direction (`thin` prior, `measured` current) — a wider or
     equally wide current window that now sees the candidate genuinely
     contradicts the earlier reading, and the earlier reading is the weak one;
  2. any flip between comparably wide windows;
  3. a threshold-side flip in which BOTH windows read `measured`. If the
     current window still reached `measured`, it observed the candidate often
     enough; a share that then crosses the filing threshold is a claim about
     relative spend, not a sampling failure, and narrowness does not excuse it.

Width is read off the snapshot's own `window.since`/`window.until`. When either
window does not carry both bounds the widths are not comparable, and the gate
does NOT quietly excuse the flip: it downgrades as before and says so in the
finding's `window_parity` block, so an unreadable window is visible rather than
a silent exemption.

NO PRIOR WINDOW MEANS UNCHECKED, NOT UNSTABLE
---------------------------------------------
A first-ever audit of a target, or a candidate seen for the first time, has
nothing to compare against. It files normally, marked
`stability: unchecked`. Manufacturing an instability finding out of an absence
of history would contradict FR-2.4 — a target with no history stays auditable on
its own terms — and would make the first audit of every plugin file nothing.

AN EMPTY LOCAL STORE IS NOT THE SAME CLAIM AS AN EMPTY WIKI
-------------------------------------------------------------
This script only ever reads the local filesystem mirror of the store — by
design, no `bin/` script holds wiki credentials (see `audit-snapshot`'s
docstring). That means an empty `--store` is ambiguous: it might be a genuine
first audit (FR-2.4), or it might just be a cold local cache sitting in front
of real wiki history nobody imported yet (kotkan/claude-plugin-inference-
arbitrage#17). The two look identical to `find_prior` but are not the same
fact about the target, and only the caller — who can read the wiki — knows
which one is true.

So when the store is empty, `--wiki-checked` is mandatory: `empty` asserts the
caller checked the wiki page `Data/<target>/snapshots.jsonl` and it genuinely
has no history (or the target truly is new), so `unchecked` is correct;
`imported` asserts the caller ran `audit-snapshot import` first — in which case
the store would no longer be empty and this path would not be taken at all.
Omitting the flag on an empty store is a hard error: it turns the SKILL.md
step 4b instruction to check the wiki first from a doc-only aside into
something the gate itself refuses to skip past.

The prior window is the most recent snapshot that contains this candidate by
FR-5.2 identity (issue > signature hash > slug), not merely the most recent
snapshot: a candidate that sat out one window is still comparable against the
window before it, and using the latest snapshot unconditionally would report
every such candidate as unmatched and let it through.

Runs after `boundary-classify` and before `filing-plan`; emits the same document
shape it consumes, with verdicts revised, so it drops into the pipeline in place
of `classified.json`.
"""

import argparse
import datetime
import json
import sys

from ia_store import (DEFAULT_STORE, MIN_SHARE_OF_SPEND, build_index,
                      candidate_signature, find_match, load_snapshots,
                      signature_hash, tiered)

GATE = "FR-4.5"

UNCHECKED_NO_HISTORY = "unchecked — no prior window to compare"
UNCHECKED_NO_MATCH = "unchecked — no prior window recorded this candidate"

# The verdict a comparison takes when it is too lopsided to read either way.
# Distinct from `boundary-question` on purpose: a boundary question is something
# a human is asked to judge, and this is the auditor declining to make a claim.
INSUFFICIENT_WINDOW = "insufficient-window"

# Readings that mean "this window did not observe the candidate enough", as
# opposed to `measured`, which means it did.
WEAK_STRENGTHS = ("thin", "unmeasured")

# How wide the current window must be, as a fraction of the prior window that
# produced the `measured` reading, before a weak reading may overturn it. Not
# 1.0: window bounds are observed session spans, so two nominally equal windows
# differ by hours and an exact-parity rule would fire on clock noise.
WINDOW_PARITY_RATIO = 0.9


def keyed(cand, scan, explicit, issue_map):
    """The identity fields FR-5.2 matches on, computed for a candidate that is
    only classified and not yet snapshotted. Derived with the same
    `candidate_signature` the snapshot writer uses — a divergence here would
    make the gate silently match nothing and pass every unstable candidate."""
    sig, source = candidate_signature(cand, scan, explicit)
    return {
        "candidate_id": cand.get("candidate_id"),
        "issue": issue_map.get(cand.get("candidate_id")) or cand.get("issue"),
        "signature_hash": signature_hash(sig),
        "signature_source": source,
    }


def find_prior(key, history):
    """Most recent snapshot holding this candidate, newest first."""
    for snap in reversed(history):
        idx = build_index([c for c, _ in tiered(snap)])
        match, by = find_match(key, idx)
        if match:
            return snap, match, by
    return None, None, None


def crosses_threshold(a, b):
    return (a >= MIN_SHARE_OF_SPEND) != (b >= MIN_SHARE_OF_SPEND)


def flips(prior, curr_strength, curr_share):
    """Named, so the finding says exactly what moved rather than 'unstable'."""
    out = []
    prior_strength = prior.get("measurement_strength")
    prior_share = (prior.get("measurement") or {}).get("share_of_plugin", 0.0) or 0.0

    if prior_strength != curr_strength:
        out.append({
            "field": "measurement_strength",
            "prior": prior_strength,
            "current": curr_strength,
            "why": ("the evidence label the issue body speaks in changed with the window, "
                    "so the same candidate would be described differently to a human "
                    "depending only on when the audit ran"),
        })
    if crosses_threshold(prior_share, curr_share):
        out.append({
            "field": "share_of_audited_spend",
            "prior": prior_share,
            "current": curr_share,
            "threshold": MIN_SHARE_OF_SPEND,
            "why": (f"the windows land on opposite sides of the {MIN_SHARE_OF_SPEND:.0%} "
                    "filing threshold — one window would file on value and the other "
                    "would not, so the value claim is a window artifact"),
        })
    return out


def window_days(window):
    """Observed width of a window in days, or None when it cannot be read.

    None is a real answer, not a default: `window_parity` reports it as
    "not comparable" and lets the flip stand, rather than treating an
    unreadable window as narrow (which would excuse every flip) or as wide
    (which would hide that the check could not run)."""
    window = window or {}
    bounds = []
    for key in ("since", "until"):
        raw = window.get(key)
        if not isinstance(raw, str):
            return None
        try:
            bounds.append(datetime.datetime.fromisoformat(raw))
        except ValueError:
            return None
    since, until = bounds
    # One bound tz-aware and the other naive cannot be subtracted, and guessing
    # a zone for the naive one would invent a width.
    if (since.tzinfo is None) != (until.tzinfo is None):
        return None
    return (until - since).total_seconds() / 86400.0


def window_parity(prior_window, curr_window):
    """Whether the current window is wide enough to contradict the prior one.

    Always returned, whether or not it changes the outcome, so the finding
    records the widths it was decided on."""
    prior_days = window_days(prior_window)
    curr_days = window_days(curr_window)
    out = {
        "parity_ratio": WINDOW_PARITY_RATIO,
        "prior_days": round(prior_days, 2) if prior_days is not None else None,
        "current_days": round(curr_days, 2) if curr_days is not None else None,
        "required_days": (round(prior_days * WINDOW_PARITY_RATIO, 2)
                          if prior_days is not None else None),
    }
    if prior_days is None or curr_days is None:
        out["comparable"] = None
        out["why"] = ("window width is unreadable — one of the two windows does not carry "
                      "both `since` and `until` — so narrowness cannot be established and "
                      "the flip is not excused; this is recorded rather than defaulted")
        return out
    out["comparable"] = curr_days >= out["required_days"]
    out["why"] = (
        f"this window spans {out['current_days']}d against the prior window's "
        f"{out['prior_days']}d; a weak reading needs at least "
        f"{out['required_days']}d ({WINDOW_PARITY_RATIO:.0%} of the prior width) to "
        "contradict a `measured` one"
        if not out["comparable"] else
        f"this window spans {out['current_days']}d against the prior window's "
        f"{out['prior_days']}d, at or above the {out['required_days']}d parity floor, so "
        "the two readings are comparably sampled and the flip is real")
    return out


def deferrable(prior_strength, curr_strength):
    """The only direction narrowness can excuse: a wide `measured` reading that a
    narrow window failed to observe. The reverse (`thin` prior, `measured` now)
    is a genuine contradiction in which the *earlier* reading is the weak one,
    and a `measured` -> `measured` threshold crossing means both windows saw the
    candidate, so neither is excused here."""
    return prior_strength == "measured" and curr_strength in WEAK_STRENGTHS


def gate(doc, history, scan, explicit, issue_map):
    target = doc.get("target")
    window = scan.get("window") or {}
    findings = []
    deferrals = []
    checked = 0

    for cand in doc.get("candidates", []):
        if cand.get("verdict") != "file":
            continue

        if not history:
            cand["stability"] = UNCHECKED_NO_HISTORY
            continue

        key = keyed(cand, scan, explicit, issue_map)
        snap, prior, by = find_prior(key, history)
        if prior is None:
            cand["stability"] = UNCHECKED_NO_MATCH
            continue

        checked += 1
        curr_strength = cand.get("measurement_strength")
        curr_share = cand.get("share_of_audited_spend", 0.0) or 0.0
        moved = flips(prior, curr_strength, curr_share)

        if not moved:
            cand["stability"] = (
                f"stable against {snap.get('run_id')} (matched by {by}): "
                f"{curr_strength}, {curr_share:.2%} of audited spend either side")
            continue

        parity = window_parity(snap.get("window"), window)
        if deferrable(prior.get("measurement_strength"), curr_strength) \
                and parity["comparable"] is False:
            # Not filed, not downgraded, not self-reported: this comparison
            # cannot be made fairly, and saying so is the honest third answer.
            cand["verdict"] = INSUFFICIENT_WINDOW
            cand["stability"] = (
                f"insufficient-window against {snap.get('run_id')} (matched by {by}): "
                f"{parity['current_days']}d window read {curr_strength} against a "
                f"{parity['prior_days']}d window's measured "
                f"{((prior.get('measurement') or {}).get('share_of_plugin') or 0):.2%} — "
                "too narrow to compare fairly")
            cand.setdefault("reasons", []).append(
                f"DEFERRED ({GATE}): the current window ({parity['current_days']}d) is "
                f"narrower than the {parity['required_days']}d parity floor set by the "
                f"{parity['prior_days']}d window that produced the prior `measured` "
                "reading, so this window's weak reading cannot fairly contradict it. "
                "Neither filed nor downgraded to a boundary question this run; carried "
                "forward for a comparison over an adequately wide window.")
            deferrals.append({
                "gate": GATE,
                "candidate_id": cand.get("candidate_id"),
                "target": target,
                "skill": cand.get("skill"),
                "matched_by": by,
                "prior": {
                    "run_id": snap.get("run_id"),
                    "window": snap.get("window"),
                    "measurement_strength": prior.get("measurement_strength"),
                    "share_of_audited_spend": (prior.get("measurement") or {}).get("share_of_plugin"),
                },
                "current": {
                    "window": window,
                    "measurement_strength": curr_strength,
                    "share_of_audited_spend": curr_share,
                },
                # What WOULD have been reported as instability, kept so the
                # deferral is auditable rather than an unexplained absence.
                "suppressed_flips": moved,
                "window_parity": parity,
                "carry_forward": (
                    f"re-audit over a window of at least {parity['required_days']}d and "
                    f"compare against {snap.get('run_id')}, which remains the standing "
                    "comparison point — this window contributes no reading that could "
                    "overturn it"),
            })
            continue

        cand["verdict"] = "boundary-question"
        cand["stability"] = f"unstable against {snap.get('run_id')} (matched by {by})"
        cand.setdefault("reasons", []).append(
            f"HARD GATE ({GATE}): evidence is not stable across measurement windows — "
            + "; ".join(f"{f['field']} {f['prior']!r} -> {f['current']!r}" for f in moved)
            + ". Downgraded to a boundary question, never filed."
        )

        findings.append({
            "gate": GATE,
            "candidate_id": cand.get("candidate_id"),
            "target": target,
            "skill": cand.get("skill"),
            "matched_by": by,
            "prior": {
                "run_id": snap.get("run_id"),
                "window": snap.get("window"),
                "measurement_strength": prior.get("measurement_strength"),
                "share_of_audited_spend": (prior.get("measurement") or {}).get("share_of_plugin"),
            },
            "current": {
                "window": window,
                "measurement_strength": curr_strength,
                "share_of_audited_spend": curr_share,
            },
            "flips": moved,
            # Recorded on downgrades too, so a reader can see the widths this
            # was decided on and that the narrow-window carve-out was
            # considered and did not apply.
            "window_parity": parity,
            "headline": (
                f"{cand.get('candidate_id')} is {prior.get('measurement_strength')} at "
                f"{((prior.get('measurement') or {}).get('share_of_plugin') or 0):.2%} of spend in "
                f"{snap.get('run_id')} and {curr_strength} at {curr_share:.2%} in this window — "
                "the confidence label depends on the measurement window, not on the candidate"),
            # FR-8.2: the instability is a finding about THIS plugin, filed on
            # its own repo. The marker mirrors FR-6.3's `ia-candidate` so a
            # re-run comments instead of duplicating.
            "self_report": {
                "repo": "kotkan/claude-plugin-inference-arbitrage",
                "marker": f"<!-- ia-stability: {cand.get('candidate_id')} -->",
            },
        })

    rows = doc.get("candidates", [])
    summary = doc.setdefault("summary", {})
    summary["to_file"] = sum(1 for c in rows if c.get("verdict") == "file")
    summary["boundary_questions"] = sum(1 for c in rows if c.get("verdict") == "boundary-question")
    summary["stability_checked"] = checked
    summary["stability_downgraded"] = len(findings)
    summary["stability_insufficient_window"] = len(deferrals)

    doc["stability_gate"] = {
        "gate": GATE,
        "snapshots_available": len(history),
        "checked": checked,
        "downgraded": len(findings),
        "insufficient_window": len(deferrals),
        "window_parity_ratio": WINDOW_PARITY_RATIO,
    }
    doc["stability_findings"] = findings
    # Kept out of `stability_findings` deliberately: those are FR-6.5
    # self-reports the caller files against this plugin's repo, and a deferral
    # is not a defect to report — it is a comparison that was not made.
    doc["stability_deferrals"] = deferrals
    return doc


def load_json(path):
    with open(path) as fh:
        return json.load(fh)


def main():
    ap = argparse.ArgumentParser(prog="stability-classify")
    ap.add_argument("--classified", required=True, help="bin/boundary-classify output")
    ap.add_argument("--target", required=True, help="target plugin name, as snapshotted")
    ap.add_argument("--store", default=DEFAULT_STORE,
                    help=f"local mirror of the wiki store (default {DEFAULT_STORE})")
    ap.add_argument("--scan", help="bin/offload-scan output, for signature identity")
    ap.add_argument("--signatures", help="JSON {candidate_id: [tool signature, ...]}")
    ap.add_argument("--issue-map", help='JSON {candidate_id: "owner/repo#N"}')
    ap.add_argument("--exclude-run", help="run_id to treat as not-yet-recorded (tests, re-runs)")
    ap.add_argument("--wiki-checked", choices=["empty", "imported"],
                    help="required when the local store is empty for this target: 'empty' "
                         "asserts the wiki was checked and genuinely has no history; "
                         "'imported' asserts `audit-snapshot import` already ran. Without "
                         "this, an empty store is ambiguous between a genuine first audit "
                         "(FR-2.4) and a merely cold local cache in front of real wiki "
                         "history (kotkan/claude-plugin-inference-arbitrage#17).")
    ap.add_argument("--json", action="store_true", help="compact single-line JSON")
    args = ap.parse_args()

    doc = load_json(args.classified)
    history = [s for s in load_snapshots(args.store, args.target)
               if s.get("run_id") != args.exclude_run]

    if not history and args.wiki_checked is None:
        sys.exit(
            "error: the store has no history for this target and --wiki-checked was not "
            "given. This is ambiguous: it may be a genuine first audit (FR-2.4), or it may "
            "just be a cold local cache in front of real wiki history "
            "(kotkan/claude-plugin-inference-arbitrage#17). Check the wiki page "
            f"Data/{args.target}/snapshots.jsonl first, then either pass "
            "--wiki-checked=empty (the wiki genuinely has no history for this target), or "
            "run `audit-snapshot import` to seed the store and re-run with "
            "--wiki-checked=imported.")

    result = gate(doc, history,
                  load_json(args.scan) if args.scan else {},
                  load_json(args.signatures) if args.signatures else {},
                  load_json(args.issue_map) if args.issue_map else {})

    json.dump(result, sys.stdout, indent=None if args.json else 2)
    sys.stdout.write("\n")


if __name__ == "__main__":
    main()