feat(plane): tie-break pass on disputed candidates; 2-of-3 agreement

The design revisit after plane-increment-1.json (14/20 agreed, 19/20 in either
pass): a third verify pass on the candidates passes a and b dispute, and agree
decides 2-of-3 where pass c asked (the a/b rule elsewhere; c alone never decides).

- verify: optional OUSAST_INPUT_ONLY restricts the hunt to the named candidates
  (agree's disputed.json); an empty list is done with 0 units and no model client.
- agree: optional OUSAST_INPUT_PASS_C; metrics keep every key and add tiebreak
  {asked, flipped_to_agreed, flipped_to_rejected, rule}; usd_total, hunts and
  declared-site metrics are on the final decision.
- generate: per case <case>-vc and <case>-final appended to validation-46; the
  first 60 entries stay byte-identical, so a rerun keeps their done state.
- Offline replay over the recorded outputs with an empty pass c reproduces the
  recorded agree summaries exactly (14/20 agreed, 23 agreed, 10 disputed).
- Spec: Requirement 6.4, design Measurement, task 8.1.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/plane/generate.py

@@ -233,12 +233,31 @@ def verify_budget(item: CaseInputs) -> dict[str, float | int]:
     return {"usd": math.ceil(HEADROOM * share * 100 - 1e-9) / 100, "calls": max(hunts, 1) * CALLS_PER_HUNT}
 
 
+def _tiebreak_entries(item: CaseInputs) -> list[dict[str, Any]]:
+    """The tie-break (Requirement 6.4): verify pass c on the candidates a and b disputed, then agree 2-of-3. Fresh
+    budget dicts, so YAML adds no anchor to the entries before them."""
+    case_id = item.case["id"]
+    return [
+        {
+            "name": f"{case_id}-vc", "task": f"verify-c-{case_id}", "dependsOn": [f"{case_id}-agree"],
+            "inputs": {"facts": f"{case_id}-facts/facts.json", "only": f"{case_id}-agree/disputed.json"},
+            "outputs": ["units.jsonl", "summary.json"], "budget": verify_budget(item),
+        },
+        {
+            "name": f"{case_id}-final", "task": f"agree-{case_id}",
+            "inputs": {pass_: f"{case_id}-v{pass_[-1]}/units.jsonl" for pass_ in ("pass_a", "pass_b", "pass_c")},
+            "outputs": ["agreed.json", "disputed.json"], "budget": {"usd": 0, "calls": 0},
+        },
+    ]  # fmt: skip
+
+
 def render(cases: Sequence[CaseInputs], templates: Mapping[str, Task], run_name: str, command: str) -> dict[str, str]:
     """Relative path under the plane root -> file text, for every generated manifest."""
     header = f"# generated by `{command}`; do not edit, regenerate.\n"
     out: dict[str, str] = {}
     tasks: list[dict[str, Any]] = []
     entries: list[dict[str, Any]] = []
+    tiebreak: list[dict[str, Any]] = []
     for item in cases:
         case_id = item.case["id"]
         pinned, inputs = _workspaces(item)
@@ -270,6 +289,11 @@ def render(cases: Sequence[CaseInputs], templates: Mapping[str, Task], run_name:
             "inputs": {"pass_a": f"{case_id}-va/units.jsonl", "pass_b": f"{case_id}-vb/units.jsonl"},
             "outputs": ["agreed.json", "disputed.json"], "budget": free,
         })  # fmt: skip
+        env = {**on_repo, "OUSAST_INPUT_CANDIDATES": f"{INPUTS_PATH}/candidates.json", "OUSAST_PASS": "c"}
+        tasks.append(_task(templates["verify"], f"verify-c-{case_id}", env, repo))
+        tiebreak.extend(_tiebreak_entries(item))
+    tasks.sort(key=lambda doc: 1 if doc["metadata"]["name"].startswith("verify-c-") else 0)  # stable: the old Tasks first
+    entries.extend(tiebreak)  # appended: the first Run's entries stay byte-identical, so a rerun keeps their state
     out[f"tasks/{run_name}.yaml"] = _dump(header, tasks)
     run = {"apiVersion": RUN_API_VERSION, "kind": "Run", "metadata": {"name": run_name}, "spec": {"tasks": entries}}
     total = sum(float(e["budget"]["usd"]) for e in entries)

src/openultrasast/plane/tasks/agree.py

@@ -17,7 +17,13 @@ divide by the candidates before triage when `case.json` gives that count, as the
 Env contract: `OUSAST_INPUT_PASS_A`, `OUSAST_INPUT_PASS_B` (a verify output directory or its `units.jsonl`),
 `OUSAST_INPUT_CANDIDATES`, `OUSAST_OUTPUT_DIR`, optional `OUSAST_INPUT_SITES` (the generator's `case.json`:
 `family`, `sites`, `ranges`, `sites_in_set`, `cost`; or a `population-v2.toml`-shaped TOML with `OUSAST_CASE_ID`
-naming the case). Writes `agreed.json`, `disputed.json`, `summary.json`. Exit 0 done, 2 failed, 3 unfinished.
+naming the case), optional `OUSAST_INPUT_PASS_C` (the tie-break pass: `verify` restricted to the candidates a and
+b disputed). Writes `agreed.json`, `disputed.json`, `summary.json`. Exit 0 done, 2 failed, 3 unfinished.
+
+With pass c the decision is 2 of 3 for every candidate pass c asked about and the a/b rule for the rest, so pass c
+alone never decides; `metrics.tiebreak` counts the candidates asked and the decisions that flipped against a/b,
+and names the rule. The declared-site metrics and `agreed`/`disputed` are on the final decision; `usd_total`,
+`hunts` and the per-hunt means include pass c. Without pass c every number is the two-pass one.
 """
 
 from __future__ import annotations
@@ -122,6 +128,28 @@ def _per_candidate(total: float | None, count: int) -> float | None:
     return None if total is None or not count else round(total / count, 6)
 
 
+RULE_A_B = "agreed when passes a and b both flagged the candidate (no pass c given)"
+RULE_2_OF_3 = (
+    "a candidate asked in pass c is agreed when at least 2 of passes a, b, c flagged it; a candidate not asked in "
+    "pass c is agreed when passes a and b both flagged it (pass c alone never decides)"
+)
+
+
+def _votes(candidate: str, flagged: Sequence[Mapping[str, Any]]) -> int:
+    return sum(candidate in f for f in flagged)
+
+
+def _pick(candidate: str, flagged: Sequence[Mapping[str, Mapping[str, Any]]]) -> dict[str, Any]:
+    return dict(next(f[candidate] for f in flagged if candidate in f))
+
+
+def _decided(candidate: str, asked_c: set[str], flagged: Sequence[Mapping[str, Any]]) -> bool:
+    """The final decision: 2 of 3 when pass c asked the candidate, else both of a and b."""
+    if candidate in asked_c:
+        return _votes(candidate, flagged) >= 2
+    return candidate in flagged[0] and candidate in flagged[1]
+
+
 def run(
     output_dir: Path,
     candidates: object,
@@ -129,15 +157,28 @@ def run(
     pass_b: Sequence[Mapping[str, Any]],
     *,
     sites: Mapping[str, Any] | None = None,
+    pass_c: Sequence[Mapping[str, Any]] | None = None,
 ) -> dict[str, Any]:
     output_dir.mkdir(parents=True, exist_ok=True)
     family, unique = load_candidates(candidates)
     names = [f"{p}::{fn}" for p, fn, _ in unique]
     lines = {f"{p}::{fn}": ln for p, fn, ln in unique}
-    asked = [{f"{r['path']}::{c[1]}" for r in rows for c in r.get("candidates", [])} for rows in (pass_a, pass_b)]
-    flagged = [{str(f["candidate"]): dict(f) for r in rows for f in r.get("flagged", [])} for rows in (pass_a, pass_b)]
-    agreed, disputed = agreement([list(flagged[0].values()), list(flagged[1].values())])
-    usd_by, turns_by = _per_candidate_cost([*pass_a, *pass_b])
+    passes = (pass_a, pass_b, pass_c or [])
+    asked = [{f"{r['path']}::{c[1]}" for r in rows for c in r.get("candidates", [])} for rows in passes]
+    flagged = [{str(f["candidate"]): dict(f) for r in rows for f in r.get("flagged", [])} for rows in passes]
+    ab_agreed, _ = agreement([list(flagged[0].values()), list(flagged[1].values())])
+    ab = {str(f["candidate"]) for f in ab_agreed}
+    final = {c for c in set(flagged[0]) | set(flagged[1]) | set(flagged[2]) if _decided(c, asked[2], flagged)}
+    agreed = [dict(f, passes=_votes(f["candidate"], flagged)) for f in ab_agreed if f["candidate"] in final]
+    agreed += [dict(_pick(c, flagged), passes=_votes(c, flagged)) for c in sorted(final - ab)]
+    agreed.sort(key=lambda f: str(f["candidate"]))
+    some = (set(flagged[0]) | set(flagged[1]) | set(flagged[2])) - final
+    disputed = [dict(_pick(c, flagged), passes=_votes(c, flagged)) for c in sorted(some)]
+    tiebreak = {
+        "asked": len(asked[2]), "flipped_to_agreed": len(final - ab), "flipped_to_rejected": len(ab - final),
+        "rule": RULE_2_OF_3 if pass_c is not None else RULE_A_B,
+    }  # fmt: skip
+    usd_by, turns_by = _per_candidate_cost([*pass_a, *pass_b, *(pass_c or [])])
     case = dict(sites) if sites else None
     if case is not None and "family" not in case and family:
         case["family"] = family
@@ -147,19 +188,20 @@ def run(
     rows: list[dict[str, Any]] = []
     for name in names:
         in_a, in_b = (name in flagged[0] if name in asked[0] else None), (name in flagged[1] if name in asked[1] else None)
-        finding = flagged[0].get(name) or flagged[1].get(name)
+        in_c = name in flagged[2] if name in asked[2] else None
+        finding = flagged[0].get(name) or flagged[1].get(name) or flagged[2].get(name)
         site = anchor_site(name, finding["site"]) if finding else f"{name.partition('::')[0]}:{lines[name]}:{name.partition('::')[2]}"
         site_match = None
         if assessed and case is not None and isinstance(ranges, Mapping):
             site_match = matches_declared({"site": site, "family": (finding or {}).get("family", family)}, case, ranges)
         cost = usd_by.get(name)
         rows.append({
-            "candidate": name, "site": site, "a": in_a, "b": in_b, "agreed": bool(in_a and in_b), "disputed": bool(in_a) != bool(in_b),
+            "candidate": name, "site": site, "a": in_a, "b": in_b, "c": in_c, "agreed": name in final, "disputed": bool(in_a) != bool(in_b),
             "site_match": site_match, "usd": (None if cost is None else round(cost, 6)), "turns": turns_by.get(name, 0), "finding": finding,
         })  # fmt: skip
     unasked = [r["candidate"] for r in rows if r["a"] is None or r["b"] is None]
     declared = [r for r in rows if r["site_match"]]
-    hunts = [*pass_a, *pass_b]
+    hunts = [*pass_a, *pass_b, *(pass_c or [])]
     usd_total = None if any(r.get("usd") is None for r in hunts) else round(sum(float(r["usd"]) for r in hunts), 6)
     reference: Mapping[str, Any] = (case or {}).get("cost") or {}
     before = int(reference.get("candidates_before_triage") or len(names))
@@ -177,6 +219,7 @@ def run(
         "usd_total_with_recorded_triage": with_triage, "cost_per_candidate_with_recorded_triage": _per_candidate(with_triage, before),
         "turns_per_hunt": _mean([float(r.get("turns") or 0) for r in hunts]),
         "calls_per_hunt": _mean([float((r.get("usage") or {}).get("calls") or 0) for r in hunts]),
+        "tiebreak": tiebreak,
     }  # fmt: skip
     (output_dir / "agreed.json").write_text(
         json.dumps({"family": family, "candidates": rows, "agreed": agreed, "disputed": disputed, "metrics": metrics}, indent=1) + "\n"
@@ -201,7 +244,8 @@ def main() -> int:
         candidates = json.loads(Path(os.environ["OUSAST_INPUT_CANDIDATES"]).read_text())
         sites_path = Path(os.environ["OUSAST_INPUT_SITES"]) if os.environ.get("OUSAST_INPUT_SITES") else None
         sites = load_sites(sites_path, os.environ.get("OUSAST_CASE_ID"))
-        summary = run(output_dir, candidates, pass_a, pass_b, sites=sites)
+        pass_c = load_pass(Path(os.environ["OUSAST_INPUT_PASS_C"])) if os.environ.get("OUSAST_INPUT_PASS_C") else None
+        summary = run(output_dir, candidates, pass_a, pass_b, sites=sites, pass_c=pass_c)
     except Exception:  # noqa: BLE001 -- a crash is `failed` with its traceback
         summary = {
             "status": "failed", "units_done": 0, "units_total": 0, "usd": 0, "calls": 0, "usage": dict.fromkeys(USAGE_FIELDS, 0),
@@ -214,7 +258,18 @@ def main() -> int:
     return EXIT_CODES.get(str(summary["status"]), 2)
 
 
-__all__ = ["LINE_WINDOW", "agreement", "anchor_site", "load_pass", "load_sites", "main", "matches_declared", "run"]
+__all__ = [
+    "LINE_WINDOW",
+    "RULE_2_OF_3",
+    "RULE_A_B",
+    "agreement",
+    "anchor_site",
+    "load_pass",
+    "load_sites",
+    "main",
+    "matches_declared",
+    "run",
+]
 
 
 if __name__ == "__main__":

src/openultrasast/plane/tasks/verify.py

@@ -10,8 +10,10 @@ completes, read back on start: Requirement 2.2). The chat client is wrapped by `
 last -- names the bound Model (Requirement 7.2) and is `done` only when every unit finished.
 
 Env contract: `OUSAST_WORKSPACE_DIR`, `OUSAST_OUTPUT_DIR`, `OUSAST_INPUT_CANDIDATES`, `OUSAST_INPUT_FACTS`
-(optional), `OUSAST_BUDGET_USD`, `OUSAST_BUDGET_CALLS`, `OUSAST_MODEL`, `OUSAST_MODEL_PARAMS`, `OUSAST_PASS`.
-Exit 0 done, 2 failed, 3 unfinished.
+(optional), `OUSAST_BUDGET_USD`, `OUSAST_BUDGET_CALLS`, `OUSAST_MODEL`, `OUSAST_MODEL_PARAMS`, `OUSAST_PASS`,
+`OUSAST_INPUT_ONLY` (optional: a JSON list of `path::function` names, `[path, function, ...]` rows or finding
+objects with a `candidate` -- `agree`'s `disputed.json` -- restricting the hunt to those candidates; the tie-break
+pass c. An empty list is a `done` task with 0 units and no model call). Exit 0 done, 2 failed, 3 unfinished.
 """
 
 from __future__ import annotations
@@ -95,6 +97,34 @@ def load_candidates(payload: object) -> tuple[str | None, list[Candidate]]:
     return family, list(unique.values())
 
 
+def only_names(payload: object) -> set[str]:
+    """The `path::function` names of an `OUSAST_INPUT_ONLY` list (names, rows, or `agree`'s finding objects)."""
+    rows = payload.get("disputed", []) if isinstance(payload, Mapping) else payload
+    if not isinstance(rows, list):
+        raise ValueError("OUSAST_INPUT_ONLY must be a JSON list (e.g. agree's disputed.json)")
+    names = set()
+    for row in rows:
+        if isinstance(row, Mapping):
+            names.add(str(row["candidate"]))
+        elif isinstance(row, list | tuple):
+            names.add(f"{row[0]}::{row[1]}")
+        else:
+            names.add(str(row))
+    return names
+
+
+def restrict(unique: Sequence[Candidate], only: set[str] | None) -> list[Candidate]:
+    """The candidates named by `only` (all of them when `only` is None)."""
+    return list(unique) if only is None else [c for c in unique if f"{c[0]}::{c[1]}" in only]
+
+
+class _NoModel:
+    """The client of a task with nothing to ask: any call is a defect, never a silent spend."""
+
+    def complete(self, **kw: Any) -> ChatResponse:
+        raise RuntimeError("no model call is expected: the candidate restriction is empty")
+
+
 def callers_of(facts: Mapping[str, Any] | None, path: str, function: str) -> list[dict[str, Any]]:
     """Known call sites of `path::function` from `facts.json`: the task-5 shape (`callers` keyed by
     `path::function`, rows `{"path", "line", "enclosing"}`) or the design's per-file shape (`{path: {"callers":
@@ -224,10 +254,12 @@ def run(
     model: str = "",
     pass_label: str = "a",
     family: str | None = None,
+    only: set[str] | None = None,
 ) -> dict[str, Any]:
     """Hunt every unfinished unit, append `units.jsonl` per hunt, write `summary.json` last; returns the summary.
 
-    `budget` is the meter to run under (built from `client` when absent); `client` is what the meter wraps.
+    `budget` is the meter to run under (built from `client` when absent); `client` is what the meter wraps;
+    `only` restricts the hunt to those `path::function` candidates.
     """
     output_dir.mkdir(parents=True, exist_ok=True)
     log, summary_path = output_dir / "units.jsonl", output_dir / "summary.json"
@@ -235,7 +267,7 @@ def run(
     model = model or getattr(metered, "model", "") or "unbound"
     recorded_family, unique = load_candidates(candidates)
     family = family or recorded_family
-    units = units_of(unique)
+    units = units_of(restrict(unique, only))
     rows = read_units(log)
     done = {unit_key(r["path"], r["candidates"]): r for r in rows if "error" not in r}
     status, reason = "done", ""
@@ -278,6 +310,8 @@ def run(
         "status": status, "units_done": len(done), "units_total": len(units), "usd": usd, "calls": calls, "usage": usage,
         "priced": metered.priced, "model": model, "pass": pass_label, "family": family,
     }  # fmt: skip
+    if only is not None:
+        summary["only"] = len(restrict(unique, only))
     if reason:
         summary["reason"] = reason
     summary_path.write_text(json.dumps(summary, indent=1) + "\n")
@@ -336,13 +370,16 @@ def main() -> int:
         params = json.loads(os.environ.get("OUSAST_MODEL_PARAMS", "") or "{}")
         candidates = json.loads(candidates_path.read_text())
         facts = json.loads(facts_path.read_text()) if facts_path is not None else None
-        client = build_client()
+        only_path = _env_path("OUSAST_INPUT_ONLY", required=False)
+        only = only_names(json.loads(only_path.read_text())) if only_path is not None else None
+        nothing = only is not None and not restrict(load_candidates(candidates)[1], only)
+        client: ChatClient = _NoModel() if nothing else build_client()
         metered = MeteredClient(
             client, prices=params, budget_usd=_env_number("OUSAST_BUDGET_USD", float), budget_calls=_env_number("OUSAST_BUDGET_CALLS", int)
         )
         summary = run(
             workspace, output_dir, candidates, facts, client=client, budget=metered, model=model, pass_label=pass_label,
-            family=os.environ.get("OUSAST_FAMILY") or None,
+            family=os.environ.get("OUSAST_FAMILY") or None, only=only,
         )  # fmt: skip
     except Exception:  # noqa: BLE001 -- a crash is `failed` with its traceback, never a silent zero
         summary = _failed(output_dir, traceback.format_exc()[-2000:], model, pass_label)
@@ -364,6 +401,8 @@ __all__ = [
     "hunt_prompt",
     "load_candidates",
     "main",
+    "only_names",
+    "restrict",
     "run",
     "units_of",
 ]