claude-plugin-inference-arbitrage/bin/candidate-digest

#!/usr/bin/env python3
"""Joins bin/plugin-inventory's PluginInventory and bin/offload-scan's
OffloadScan into a candidate table -- the reshaping the offload-analyst agent
used to do by hand on every single audit run (40.09% of this plugin's own
audited spend, kotkan/claude-plugin-inference-arbitrage#24).

This is a DIGEST, not an auto-filing script: position 3 of the rubric
(references/boundary-rubric.md), never position 1. It computes SHAPE -- which
rows cross the codified mechanical thresholds -- and leaves intent to a
human/the analyst. A row can cross every mechanical threshold (share, mtr,
read-amplification, a recurring n-gram) and still turn out to be this audit's
own necessary two-pass discipline (read the static inventory, then the
dynamic scan, then reshape both) rather than a genuinely wasteful loop. The
digest cannot see intent, only shape, so `flags` is advisory input to a
human's judgment call -- never a verdict, and never wired to filing.
`bin/boundary-classify` remains the only place a verdict is computed; this
script does not import it for grading, only for its two evidence-strength
constants (see below).

Usage: candidate-digest <inv.json|-> <scan.json> [--json]
"""
import argparse
import importlib.machinery
import importlib.util
import json
import os
import sys

BIN_DIR = os.path.dirname(os.path.abspath(__file__))

# ia_store.py has no hyphen, so a normal import works once bin/ is on
# sys.path -- which it already is (sys.path[0]) when this file is run as
# `bin/candidate-digest`, per ia_store.py's own docstring.
from ia_store import MIN_SHARE_OF_SPEND, scan_rows  # noqa: E402


def _load_bin_module(filename):
    """boundary-classify has a hyphenated filename and can't be `import`ed by
    name -- loaded the same way offload-scan loads token-budget's cc-tokens
    (resolve_cc_tokens there), so this plugin has exactly one pattern for
    "import a sibling bin/ script as a module"."""
    path = os.path.join(BIN_DIR, filename)
    mod_name = filename.replace("-", "_")
    spec = importlib.util.spec_from_loader(
        mod_name, importlib.machinery.SourceFileLoader(mod_name, path))
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


_boundary_classify = _load_bin_module("boundary-classify")
# Reused, not reinvented (kotkan/claude-plugin-inference-arbitrage#24 says to
# reuse these): the same bar boundary-classify already applies to call a
# candidate's evidence `measured` rather than `thin`.
MIN_INVOCATIONS = _boundary_classify.MIN_INVOCATIONS
MIN_ATTRIBUTED_TURNS = _boundary_classify.MIN_ATTRIBUTED_TURNS

# New in this script -- no existing catalog threshold covers "how much
# judgment_density is enough to call a recurring shape genuine work rather
# than waste". The issue's own worked example put a real non-flag case at
# judgment_density=0.55 and a real flag case at 0.034; 0.3 sits well clear of
# both and reads naturally as "under a third of this row's turns were
# judgment calls". This is the judgment_density BRAKE
# (agents/offload-analyst.md: "a brake, not an accelerator") -- it can only
# suppress a flag, never add one.
JUDGMENT_DENSITY_BRAKE = 0.3


def _has_bin_script(name, kind, inv):
    """True only when plugin-inventory's OWN static evidence shows this row's
    documented procedure already delegates to a bin/ script.

    Skills carry `command_blocks` (the fenced code in their SKILL.md); a
    block whose first line names a `bin/` path means the skill is *already*
    a thin wrapper around an existing script -- the
    "verb_ratio=0.71 backed by a 606-LOC script" example from the issue,
    where the conjunction (high verb ratio AND no script) fails and nothing
    should be surfaced.

    Agents carry no body or command-block text in plugin-inventory's JSON at
    all -- only skills do -- so this is conservatively False for every agent
    row. That is a real limit on what this script can see, not a judgment
    call it is dodging: an analyst who knows a given agent already delegates
    to bin/ overrides this field by hand, same as any other digest output.
    """
    if kind != "skill":
        return False
    for skill in inv.get("skills", []):
        if skill.get("name") == name:
            return any("bin/" in b.get("sig", "")
                       for b in skill.get("command_blocks", []))
    return False


def _dominant_ngram_recurrences(row):
    ngrams = row.get("ngrams") or []
    if not ngrams:
        return None
    return max(g.get("recurrences", 0) for g in ngrams)


def _flags(share, invocations, turns, judgment_density, dominant_ngram,
          has_bin_script):
    """The escalation path documented in kotkan/claude-plugin-inference-
    arbitrage#24: flag a row when it crosses the same thresholds
    boundary-classify already uses for evidence strength and filing share
    (MIN_INVOCATIONS, MIN_ATTRIBUTED_TURNS, MIN_SHARE_OF_SPEND), shows a
    dominant recurring n-gram, AND judgment_density sits below the brake.

    `judgment_density` is a brake, not an accelerator: a row that crosses
    every other threshold but carries real judgment density is the healthy
    case (this audit's own two-pass read-then-reshape discipline, not a
    wasteful loop) and MUST come back with flags=[] -- that is the low_judgment
    check below, and it is the whole reason this is a digest and not an
    auto-filing script.

    A row already backed by a bin/ script (has_bin_script) is never flagged
    at all: whatever mechanical share it carries is already offloaded, so
    there is nothing left here to propose.
    """
    if has_bin_script:
        return []
    measured = invocations >= MIN_INVOCATIONS and turns >= MIN_ATTRIBUTED_TURNS
    significant_share = share >= MIN_SHARE_OF_SPEND
    recurring = dominant_ngram is not None and dominant_ngram > 1
    low_judgment = judgment_density < JUDGMENT_DENSITY_BRAKE
    flags = []
    if measured and significant_share and recurring and low_judgment:
        flags.append("high-share-low-judgment-recurring-ngram")
    return flags


def candidate_digest(inv, scan):
    """PluginInventory, OffloadScan -> list[CandidateRow]
    (kotkan/claude-plugin-inference-arbitrage#24).

    Pure function of its two already-computed JSON documents -- no
    filesystem access beyond what the caller already read, no network, no
    re-derivation of anything plugin-inventory or offload-scan already
    computed. This is the entire join the offload-analyst agent used to
    reshape by hand every audit run.

    CandidateRow = {row_kind: 'skill'|'agent', name: str,
    share_of_spend: float, mtr: float, judgment_density: float,
    read_amplification: float, retry_density: float,
    fanout_multiplier: float, dominant_ngram_recurrences: int|None,
    has_bin_script: bool, flags: list[str]}
    """
    rows = []
    for name, row in scan_rows(scan):
        # scan_rows() (ia_store.py) yields every scan["skills"] row keyed by
        # row["skill"] and every scan["agents"] row keyed by row["agent"];
        # the row itself still carries whichever of those two keys it came
        # from, which is the only place kind is recoverable from here.
        kind = "agent" if "agent" in row else "skill"
        has_bin = _has_bin_script(name, kind, inv)
        dominant = _dominant_ngram_recurrences(row)
        share = row.get("share_of_plugin", 0.0)
        judgment_density = row.get("judgment_density", 0.0)
        rows.append({
            "row_kind": kind,
            "name": name,
            "share_of_spend": share,
            "mtr": row.get("mtr", 0.0),
            "judgment_density": judgment_density,
            "read_amplification": row.get("read_amplification", 0.0),
            "retry_density": row.get("retry_density", 0.0),
            "fanout_multiplier": row.get("fanout_multiplier", 1.0),
            "dominant_ngram_recurrences": dominant,
            "has_bin_script": has_bin,
            "flags": _flags(share, row.get("invocations", 0),
                            row.get("turns", 0), judgment_density,
                            dominant, has_bin),
        })
    rows.sort(key=lambda r: -r["share_of_spend"])
    return rows


def main():
    ap = argparse.ArgumentParser(prog="candidate-digest")
    ap.add_argument("inventory",
                    help="path to plugin-inventory --json output, or - for stdin")
    ap.add_argument("scan", help="path to offload-scan --json output")
    ap.add_argument("--json", action="store_true", help="compact single-line JSON")
    args = ap.parse_args()

    with (sys.stdin if args.inventory == "-" else open(args.inventory)) as fh:
        inv = json.load(fh)
    with open(args.scan) as fh:
        scan = json.load(fh)

    rows = candidate_digest(inv, scan)
    json.dump(rows, sys.stdout, indent=None if args.json else 2)
    sys.stdout.write("\n")


if __name__ == "__main__":
    main()