Add the reference adversary and record its first run (evidence only; no score changes)

docs/audit.md

@@ -65,3 +65,133 @@ genuine model errors (`llm_error`) should survive to a scored snapshot —
 
 Re-running the script is safe: a case is skipped once it has a verdict, so
 interrupted runs resume and failures can be re-attempted by re-invoking.
+
+# Reference adversary
+
+The diagnosis judge above is built to explain misses, not to doubt references:
+its prompt says to treat the reference and its derivation as correct, labels
+engine facts authoritative, and allows doubt only through `reference_suspect`
+with a concrete contradiction. On 2026-10-05 that design missed reference issues
+a reading of the data found: Louisiana's 2026 standard deduction was
+PolicyEngine's own CPI computation rather than a published amount; the
+`payroll_tax` reference counted paid-leave employee shares that the employer
+may, but need not, pass on, though the output asks for "mandatory employee state
+payroll taxes"; and the income tax references left out a dependent's own return.
+The reference adversary is a separate pass whose only job is to attack the
+reference. Its verdicts never change a score.
+
+## Pipeline
+
+```bash
+# 1. Consensus trigger: scored cells where a cluster of models shares a wrong
+#    answer (>= 15 of the models, or >= 3 of the top 5).
+uv run policybench consensus-flags \
+  --payload paper/snapshot/20260501/runs/<run>/data.json.gz \
+  --output <dir>/consensus_flags.json
+
+# 2. One two-stage case per flagged cell, minus cells another audit owns.
+uv run policybench adversary-prepare \
+  --payload paper/snapshot/20260501/runs/<run>/data.json.gz \
+  --flags <dir>/consensus_flags.json \
+  --annotations-dir annotations/<run> \
+  --skip-cells <dir>/covered_elsewhere.json \
+  --adversary-dir <adv>
+
+# 3. Judge inside a Subfleet lane (subscription billing, never an API key).
+scripts/run_reference_adversary_claude.sh <adv>   # Claude lane
+scripts/run_reference_adversary_codex.sh <adv>    # Codex lane
+
+# 4. Verdict table and adjudication queue (one or more judges).
+uv run policybench adversary-collect \
+  --adversary-dir claude=<adv-claude> --adversary-dir codex=<adv-codex> \
+  --output-dir <out>
+```
+
+`consensus-flags` records its parameters in the report. `--prototype`
+reproduces the 2026-10-05 prototype (answers truncated to whole dollars,
+eligibility outputs never flagged): 41 of the frozen run's 1,928 scored cells.
+The defaults round answers to the nearest dollar, which merges answers such as
+13,387.65 and 13,388, and compare eligibility outputs by mismatch: 61 cells. A
+model whose own answer is within the tolerance of an amount reference counts as
+exact and never toward a wrong cluster, even when its rounded key misses.
+
+## Two stages
+
+- **Stage 1, law first.** The judge sees the household prompt, the output's
+  definition from `benchmark_specs.json`, the consensus answer with each
+  member's model id, answer and explanation, and the reference value. It does
+  not see the engine derivation. It works the answer from primary law
+  (statutes, regulations, agency publications, forms and instructions), cites
+  each rule with its publication date and whether it predates the 2026-07-03
+  reference freeze, and says which answer the law supports.
+  - The Claude runner gives it only web tools (no file access) and denies
+    WebFetch on the PolicyBench, PolicyEngine and GitHub domains
+    (`BLOCKED_DOMAINS`). WebSearch has no deny rule, so its results reach the
+    judge. The prompt asks the judge to pass those domains as
+    `blocked_domains` on every search. The transcript audit rejects an output
+    whose search results list a blocked URL or name PolicyEngine or
+    PolicyBench.
+  - The Codex runner works from an empty directory and rejects a stage 1
+    whose log shows a read of derivation-bearing paths. Codex also adds text
+    to a session that produces no event, so the runner keeps the lane's
+    configuration from the judge: each call skips the lane's `config.toml`
+    (`--ignore-user-config`, which drops `developer_instructions`,
+    `model_instructions_file` and configured MCP servers), turns the
+    memories, hooks, plugins and apps features off, and gets a fresh, empty
+    `HOME`, so no user skills reach it. The runner refuses a Codex home
+    holding `AGENTS.override.md` or `AGENTS.md`, or any skills entry but the
+    bundled `.system` directory. It does not control what Codex bundles, an
+    administrator's `/etc/codex`, or a feature those flags leave on; calling
+    an app or MCP server is an MCP tool event, which the audit rejects.
+    Codex's event log records a search's query but not its results, so this
+    runner cannot check what a search returned; the prompt's request is its
+    only guard there.
+- **Stage 2, reconcile.** A fresh call gets the frozen stage-1 JSON (bound by
+  its sha256) and only now the engine derivation. It returns a verdict:
+  `reference_holds`, `reference_wrong`, `definition_mismatch` or
+  `prompt_ambiguous`, with citations and a suggested adjudication.
+
+## Where verdicts go
+
+`adversary-collect` writes each judge's verdict table and
+`adversary_adjudication_queue.csv`, in the case-notes schema with
+`reference_suspect=true`. The queue holds every case with a verdict other than
+`reference_holds`. It also holds every case whose verdict is inconsistent, such
+as a holding verdict that drops stage 1's finding for the consensus without
+naming a stage-1 error. Its rules:
+
+- A verdict counts only with a sidecar bound to the current stage 1.
+- A case without one is listed as missing, and the command then exits
+  non-zero unless `--allow-missing` is given.
+- Each `--adversary-dir` needs its own label (`LABEL=DIR`), which names its
+  files (`adversary_<label>_<table>.csv`). A repeated label is refused,
+  because one judge's verdicts would overwrite another's; so are two labels
+  that differ only in case, which share files on a case-insensitive file
+  system, and the label `merged`, which the merged table uses. The label is
+  letters, digits, `.`, `_` or `-`, and the directory is everything after
+  the first `=`, so a directory may contain `=`; an empty directory is
+  refused. A bare `DIR` is labeled with its last path component. A relative
+  bare `DIR` whose text before an `=` would pass as a label (`adv=2`) reads
+  as `LABEL=DIR`, so pass it as `./adv=2`.
+- A directory without `cases.jsonl` is refused, so a mistyped path cannot
+  pass as a judge with no cases.
+
+`policybench.reference_adversary.apply_adversary_flags` sets those flags on a
+release's case notes, so `scripts/freeze_snapshot.py` refuses to freeze until
+each carries a developer `reference_verdict` (`policybench.adjudications`):
+the existing adjudication path. A change to a published reference or an
+exclusion still needs a release and the maintainer's ruling.
+
+## Definition and publication checks
+
+Two engine-side checks run with the reference venv (policyengine-us 2.15.17
+plus the reference conventions) and write under
+`reference_audit/2026-10-05-reference-adversary/verification/`:
+
+- `definition_conformance` lists the policyengine-us variables each output's
+  reference sums and tests them against the qualifiers in the output's
+  definition ("employee-side", "mandatory", "household", "excluding the ACA
+  PTC", "excluding local income and payroll taxes").
+- `publication_sources` lists every 2026 parameter value a scored reference
+  reads and reports whether its metadata cites a government publication dated
+  before the freeze, flagging computed or indexed values.

policybench/cli.py

@@ -3,6 +3,7 @@
 import argparse
 import hashlib
 import json
+import re
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
@@ -582,6 +583,86 @@ def main():
         "(default: refuse and list the hedged cases for re-judging)",
     )
 
+    # Reference adversary: consensus trigger -> law-first judge -> adjudication
+    consensus_parser = subparsers.add_parser(
+        "consensus-flags",
+        help="Flag scored cells where a cluster of models shares a wrong answer",
+    )
+    consensus_parser.add_argument(
+        "--payload",
+        required=True,
+        help="Frozen run payload (data.json.gz) or a release dashboard-data.json",
+    )
+    consensus_parser.add_argument("--output", required=True, help="Report JSON path")
+    consensus_parser.add_argument(
+        "--prototype",
+        action="store_true",
+        help="Use the 2026-10-05 prototype's parameters (answers truncated to "
+        "whole dollars, eligibility outputs never flagged); explicit flags "
+        "still override",
+    )
+    consensus_parser.add_argument("--min-models", type=int, default=None)
+    consensus_parser.add_argument("--top-k", type=int, default=None)
+    consensus_parser.add_argument("--min-top", type=int, default=None)
+    consensus_parser.add_argument("--tolerance", type=float, default=None)
+    consensus_parser.add_argument(
+        "--answer-rounding",
+        choices=["nearest", "truncate", "cents"],
+        default=None,
+    )
+    consensus_parser.add_argument("--zero-cluster-min-models", type=int, default=None)
+    consensus_parser.add_argument(
+        "--binary-outputs",
+        choices=["mismatch", "skip"],
+        default=None,
+        help="Eligibility outputs: wrong when the flag differs (mismatch, the "
+        "default) or judged by the dollar tolerance and so never flagged (skip)",
+    )
+
+    adversary_prepare_parser = subparsers.add_parser(
+        "adversary-prepare",
+        help="Write two-stage reference-adversary prompts for consensus-flagged cells",
+    )
+    adversary_prepare_parser.add_argument("--payload", required=True)
+    adversary_prepare_parser.add_argument(
+        "--flags", required=True, help="consensus-flags report JSON"
+    )
+    adversary_prepare_parser.add_argument("--adversary-dir", required=True)
+    adversary_prepare_parser.add_argument(
+        "--annotations-dir",
+        default=None,
+        help="Frozen annotations holding us_case_reference_explanations.csv "
+        "(the stage-2 derivations); default: the payload's referenceExplanation",
+    )
+    adversary_prepare_parser.add_argument(
+        "--skip-cells",
+        default=None,
+        help="JSON list of {scenario_id, variable, covered_by} cells another "
+        "audit already covers; they get no adversary case",
+    )
+
+    adversary_collect_parser = subparsers.add_parser(
+        "adversary-collect",
+        help="Fold reference-adversary verdicts into a verdict table and an "
+        "adjudication queue (changes no score)",
+    )
+    adversary_collect_parser.add_argument(
+        "--adversary-dir",
+        action="append",
+        required=True,
+        metavar="LABEL=DIR",
+        help="One judge's adversary directory, optionally labeled LABEL=DIR, "
+        "where LABEL is letters, digits, '.', '_' or '-' and not 'merged' "
+        "(repeatable; pass a bare directory containing '=' as ./NAME=...)",
+    )
+    adversary_collect_parser.add_argument("--output-dir", required=True)
+    adversary_collect_parser.add_argument(
+        "--allow-missing",
+        action="store_true",
+        help="Exit 0 even when a case has no usable verdict (by default such "
+        "a case fails the command after the tables are written)",
+    )
+
     # Population weights
     weight_parser = subparsers.add_parser(
         "population-weights",
@@ -1335,6 +1416,164 @@ def main():
             f"{n_suspect} cases flag the PolicyEngine reference as suspect."
         )
 
+    elif args.command == "consensus-flags":
+        from dataclasses import replace
+
+        from policybench.consensus import (
+            PROTOTYPE_PARAMS,
+            ConsensusParams,
+            consensus_report,
+            file_sha256,
+            load_us_payload,
+        )
+
+        params = PROTOTYPE_PARAMS if args.prototype else ConsensusParams()
+        overrides = {
+            name: getattr(args, name)
+            for name in (
+                "min_models",
+                "top_k",
+                "min_top",
+                "tolerance",
+                "answer_rounding",
+                "zero_cluster_min_models",
+                "binary_outputs",
+            )
+            if getattr(args, name) is not None
+        }
+        params = replace(params, **overrides)
+        payload_path = Path(args.payload)
+        report = consensus_report(
+            load_us_payload(payload_path),
+            params,
+            source=str(payload_path),
+            source_sha256=file_sha256(payload_path),
+        )
+        _ensure_parent_dir(args.output)
+        Path(args.output).write_text(json.dumps(report, indent=2) + "\n")
+        print(
+            f"Flagged {report['flagged_cells']} of {report['scored_cells']} scored "
+            f"cells ({report['models']} models) -> {args.output}"
+        )
+
+    elif args.command == "adversary-prepare":
+        from policybench.consensus import load_us_payload
+        from policybench.reference_adversary import (
+            build_adversary_cases,
+            load_derivations,
+            prepare_adversary,
+        )
+
+        flags = json.loads(Path(args.flags).read_text())["flags"]
+        skipped: dict[tuple[str, str], str] = {}
+        if args.skip_cells:
+            for cell in json.loads(Path(args.skip_cells).read_text()):
+                key = (str(cell["scenario_id"]), str(cell["variable"]))
+                skipped[key] = str(cell.get("covered_by", ""))
+        kept = [f for f in flags if (f["scenario_id"], f["variable"]) not in skipped]
+        derivations = (
+            load_derivations(Path(args.annotations_dir))
+            if args.annotations_dir
+            else None
+        )
+        cases = build_adversary_cases(
+            load_us_payload(Path(args.payload)), kept, derivations=derivations
+        )
+        prepare_adversary(Path(args.adversary_dir), cases)
+        print(
+            f"Prepared {len(cases)} adversary cases under {args.adversary_dir} "
+            f"({len(flags) - len(kept)} flagged cells skipped as covered elsewhere). "
+            "Run scripts/run_reference_adversary_claude.sh or "
+            "scripts/run_reference_adversary_codex.sh inside a Subfleet lane."
+        )
+
+    elif args.command == "adversary-collect":
+        import pandas as pd
+
+        from policybench.reference_adversary import (
+            adjudication_queue,
+            collect_adversary,
+            merge_judges,
+        )
+
+        specs = []
+        for spec in args.adversary_dir:
+            # LABEL=DIR splits at the first "=", so a directory may contain
+            # "="; text before it that is not a label is part of a bare DIR.
+            label, separator, directory = spec.partition("=")
+            if not (separator and re.fullmatch(r"[A-Za-z0-9._-]+", label)):
+                label, directory = Path(spec).name, spec
+            if not directory:
+                raise SystemExit(
+                    f"adversary-collect: {spec!r} names no directory after '='"
+                )
+            specs.append((label, directory))
+        absent = [
+            directory
+            for _, directory in specs
+            if not (Path(directory) / "cases.jsonl").is_file()
+        ]
+        if absent:
+            # A mistyped directory would otherwise collect as an empty judge.
+            raise SystemExit(
+                f"adversary-collect: no cases.jsonl in {', '.join(absent)}; "
+                "not a prepared adversary directory"
+            )
+        # Each label names its judge's files, adversary_<label>_<table>.csv. A
+        # repeated label would overwrite one judge's files with another's, and
+        # so would two that differ only in case on a case-insensitive file
+        # system (macOS by default); "merged" would collide with the merged
+        # table.
+        folded = [label.casefold() for label, _ in specs]
+        repeated = sorted(
+            {label for label, _ in specs if folded.count(label.casefold()) > 1}
+        )
+        if repeated:
+            raise SystemExit(
+                f"adversary-collect: judge label(s) {', '.join(repeated)} given "
+                "more than once (labels ignore case); label each directory "
+                "(LABEL=DIR) uniquely"
+            )
+        if "merged" in folded:
+            raise SystemExit(
+                "adversary-collect: the judge label 'merged' is reserved for "
+                "the merged table; label that directory otherwise (LABEL=DIR)"
+            )
+        output_dir = Path(args.output_dir)
+        output_dir.mkdir(parents=True, exist_ok=True)
+        by_judge = {}
+        inconsistent = []
+        missing = 0
+        for label, directory in specs:
+            out = collect_adversary(Path(directory))
+            by_judge[label] = out["verdicts"]
+            inconsistent.append(out["inconsistent"].assign(judge=label))
+            missing += len(out["missing"])
+            for key in ("verdicts", "missing", "inconsistent"):
+                out[key].to_csv(
+                    output_dir / f"adversary_{label}_{key}.csv", index=False
+                )
+            print(
+                f"{label}: {len(out['verdicts'])} verdicts, "
+                f"{len(out['missing'])} missing, "
+                f"{len(out['inconsistent'])} inconsistent"
+            )
+        merged = merge_judges(by_judge)
+        merged.to_csv(output_dir / "adversary_merged_verdicts.csv", index=False)
+        queue = adjudication_queue(merged, pd.concat(inconsistent, ignore_index=True))
+        queue.to_csv(output_dir / "adversary_adjudication_queue.csv", index=False)
+        print(
+            f"{len(queue)} of {len(merged)} cases go to developer adjudication "
+            f"(a verdict other than reference_holds, or an inconsistent one) "
+            f"-> {output_dir}"
+        )
+        if missing and not args.allow_missing:
+            raise SystemExit(
+                f"adversary-collect: {missing} case(s) have no usable verdict "
+                "(see adversary_<label>_missing.csv); re-run the judge, or pass "
+                "--allow-missing to accept an incomplete collection"
+            )
+
     elif args.command == "population-weights":
         from policybench.config import TAX_YEAR
         from policybench.population_weights import write_population_weight_payload