#!/usr/bin/env python3 """Adjudicator for pipe's two highest-volume rules. Usage: python3 tool/adjudicate.py [--format json] Exit status: 0 when the controls pass, 1 when they fail or cannot discriminate (NO SIGNAL). A CI step must not read an uncertified run as a pass. Precision claims are only meaningful if the thing doing the judging can demonstrate it discriminates. So this script prints CONTROLS first: it must find a UI type test for every variant pipe calls *rendered*, and a construction for every event pipe calls *dispatched*. If a control is below 100%, the numbers after it are void. Claims under test ----------------- emitted-not-rendered : variant V of union U is emitted by owner B but is never type-tested/downcast in any non-logic (UI) file. dead-handler : event E is never constructed or dispatched in any file other than the one that declares E. It classifies every mention of every state/event name by regex on the source line, so it does not trust pipe's evidence judgements. It is only partly independent, though: the list of unions, variants, events and owners to check comes from pipe's own `state` output, so a variant pipe fails to enumerate is never checked here at all. It does deliberately NOT judge UX severity: whether an unrendered marker state matters is a judgement, and whether an unbound handler is worth deleting is a judgement. Only the factual claims are adjudicated. """ from __future__ import annotations import argparse import collections import json import re import subprocess import sys from pathlib import Path PIPE = ["dart", "run", "bin/pipe.dart"] def run_pipe(args: list[str], cwd: Path) -> dict: proc = subprocess.run( PIPE + args, cwd=cwd, capture_output=True, text=True ) if proc.returncode != 0: sys.exit(f"pipe failed: {' '.join(args)}\n{proc.stderr[:2000]}") try: return json.loads(proc.stdout) except json.JSONDecodeError as exc: sys.exit( f"pipe did not emit JSON for {' '.join(args)}: {exc}\n" f"first 500 chars: {proc.stdout[:500]!r}" ) def rg_lines(root: Path, names: list[str]) -> list[tuple[str, str]]: """One ripgrep pass; returns (path, line-content) for lines mentioning a name.""" if not names: return [] pattern = r"\b(" + "|".join(re.escape(n) for n in names) + r")\b" proc = subprocess.run( ["rg", "-n", "--with-filename", "-e", pattern, "lib"], cwd=root, capture_output=True, text=True, ) out = [] for line in proc.stdout.splitlines(): parts = line.split(":", 2) if len(parts) == 3: out.append((parts[0], parts[2])) return out def rg_files(root: Path, pattern: str) -> set[str]: proc = subprocess.run( ["rg", "-l", pattern, "lib"], cwd=root, capture_output=True, text=True ) return set(proc.stdout.split()) def collect_lines_without_comments(root: Path, names: list[str]) -> list[tuple[str, str]]: """Same as rg_lines but drops `//`-commented code, which is not a dispatch.""" kept = [] for path, content in rg_lines(root, names): stripped = content.strip() if stripped.startswith(("//", "*")): continue kept.append((path, content)) return kept def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("repo", help="path to the analyzed project") ap.add_argument("--format", choices=["text", "json"], default="text") ap.add_argument("--pipe-root", default=".", help="where to run pipe from") args = ap.parse_args() root = Path(args.repo).resolve() pipe_root = Path(args.pipe_root).resolve() check = run_pipe(["check", str(root), "-c", "8", "--format", "json"], pipe_root) state = run_pipe(["state", str(root), "-c", "8", "--format", "json"], pipe_root) r1 = [f for f in check["findings"] if f["rule"] == "emitted-not-rendered"] r6 = [f for f in check["findings"] if f["rule"] == "dead-handler"] vdecl = { (u["name"], v["name"]): (v.get("declaredIn") or u["declaredIn"]) for u in state["unions"] for v in u["variants"] } edecl = { v["name"]: (v.get("declaredIn") or u["declaredIn"]) for u in state["eventUnions"] for v in u["variants"] } state_names = sorted({v["name"] for u in state["unions"] for v in u["variants"]}) event_names = sorted({v["name"] for u in state["eventUnions"] for v in u["variants"]}) all_names = sorted(set(state_names) | set(event_names)) logic = rg_files(root, r"extends Bloc<|extends Cubit<") lines = collect_lines_without_comments(root, all_names) # Evidence that a variant reaches the UI. A switch can match a state in four # ways and all four must count, or the checker manufactures contradictions: # is X - type test # case X / case X v - object and declared-variable patterns with `case` # X v => - declared-variable pattern in a SWITCH EXPRESSION, # where Dart 3 drops the `case` keyword entirely # X(...) => - object pattern arm in a switch expression # `is!` is deliberately NOT evidence: it asserts the variant is absent. notype = {n: re.compile(r"\bis\s+" + re.escape(n) + r"\b") for n in all_names} casepat = {n: re.compile(r"\bcase\s+" + re.escape(n) + r"\b") for n in all_names} armvar = { n: re.compile(r"(?:^|[^\w.])" + re.escape(n) + r"\s+[A-Za-z_]\w*\s*=>") for n in all_names } armobj = { n: re.compile(re.escape(n) + r"\s*\([^)]*\)\s*=>") for n in all_names } ctor = {n: re.compile(r"\b" + re.escape(n) + r"\s*\(") for n in all_names} word = {n: re.compile(r"\b" + re.escape(n) + r"\b") for n in all_names} type_files: dict[str, set[str]] = collections.defaultdict(set) ctor_files: dict[str, set[str]] = collections.defaultdict(set) for path, content in lines: for name in all_names: if not word[name].search(content): continue if ( notype[name].search(content) or casepat[name].search(content) or armvar[name].search(content) or armobj[name].search(content) ): type_files[name].add(path) if ctor[name].search(content): ctor_files[name].add(path) # ---- evidence must be scoped to the variant's own union ---- # A variant name is only unique within its library: nuwayze declares # `ExportSuccess` in FOUR unions. `is ExportSuccess` inside a file that never # mentions ArtistStoreManagementState is evidence about some OTHER union's # variant, so crediting it here manufactured exactly the false contradiction # that pipe's own per-union scoping fixed. The rule is derived independently # (a file that does not name the union cannot testify about its variants) # rather than copied from the subject under test. uowner = {u["union"]: (u.get("owner") or "") for u in state["coverage"]} ev_union = {v["name"]: u["name"] for u in state["eventUnions"] for v in u["variants"]} ev_owner = {u["name"]: (u.get("owner") or "") for u in state["eventUnions"]} # Which union and owner names each file mentions. These are NOT in # `all_names` (that list holds variant names only), so they need their own # ripgrep pass - without one, `union in file_ids` can never be true and every # control silently reads 0 while the measurement looks perfect. scope_names = sorted( {n for n in uowner if n} | {o for o in uowner.values() if o} | {n for n in ev_owner if n} | {o for o in ev_owner.values() if o} ) file_ids: dict[str, set[str]] = collections.defaultdict(set) for path, content in rg_lines(root, scope_names): for name in scope_names: if re.search(r"\b" + re.escape(name) + r"\b", content): file_ids[path].add(name) # Scoping is only required for AMBIGUOUS names. A name declared by one union # is unambiguous, so a type test or constructor call anywhere is evidence # about that variant and demanding the union be named too would reject sound # evidence (event construction sites in particular need no type annotation). # A name declared by several unions, like nuwayze's `ExportSuccess` in four, # can only be attributed via a file that names the union or the owner. decl_count: collections.Counter[str] = collections.Counter() for union in (*state["unions"], *state["eventUnions"]): for variant in union["variants"]: decl_count[variant["name"]] += 1 # Base-class payload fields per union. pipe counts a variant as rendered when # the UI reads the union's payload THROUGH THE BASE CLASS, e.g. # `state.upcomingBookings` where `BookingsLoaded` passes those as # `required super.upcomingBookings`. That is legitimate - the payload really # does reach the screen - so a type-test-only control raises false failures # for exactly the unions that render this way. def base_fields_of(decl_file: str, union: str) -> set[str]: try: text = (root / decl_file).read_text(errors="replace") except OSError: return set() head = re.search( r"\b(?:abstract\s+|sealed\s+)?class\s+" + re.escape(union) + r"\b", text ) if not head: return set() tail = text[head.end() :] stop = re.search(r"\n\}", tail) body = tail[: stop.start() if stop else len(tail)] fields: set[str] = set() for pattern in ( r"^\s+(?:final|const|late\s+final)\s+[^;=]+?\s+([a-z_]\w*)\s*[;=]", r"^\s+[^;{]+?\s+get\s+([a-z_]\w*)", ): for found in re.finditer(pattern, body, re.M): fields.add(found.group(1)) return fields ubase = { u["name"]: base_fields_of(u.get("declaredIn") or "", u["name"]) for u in state["unions"] } bf_names = sorted({bf for s in ubase.values() for bf in s}) bf_files: dict[str, set[str]] = collections.defaultdict(set) for path, content in (rg_lines(root, bf_names) if bf_names else []): for bf in bf_names: if re.search(r"\b" + re.escape(bf) + r"\b", content): bf_files[path].add(bf) def scoped( files_with_evidence: set[str], union: str, owner: str, variant: str, exclude: set[str] | None = None, ) -> set[str]: skip = logic | set(exclude or ()) unambiguous = decl_count[variant] == 1 out = set() for f in files_with_evidence: if f in skip: continue if unambiguous: out.add(f) continue ids = file_ids.get(f, set()) if union in ids or (owner and owner in ids): out.add(f) return out # ---- controls: the adjudicator must find what pipe says is handled ---- rendered = sorted({v for u in state["coverage"] for v in u["rendered"]}) rendered_pairs = sorted( {(u["union"], v) for u in state["coverage"] for v in u["rendered"]} ) dispatched = sorted({d["event"] for d in state["dispatches"]} & set(event_names)) def classify( evidence: set[str], union: str, owner: str, name: str ) -> tuple[str, list[str]]: """Three-way, because a control that cannot discriminate must say so. found - attributable evidence exists unattributed - evidence exists only under a shared name, so a text-level checker cannot honestly assign it to THIS variant. That is a limit of the oracle, not a contradiction of pipe. missing - no evidence anywhere. The only real failure. """ if evidence: got = scoped(evidence, union, owner, name) if got: return "found", sorted(got)[:3] # Second, independent route: base-class payload reads. bf = ubase.get(union) or set() if bf: seen = file_ids.get("", set()) witnesses = sorted( f for f, fields in bf_files.items() if f not in logic and (fields & bf) and (union in file_ids.get(f, seen) or (owner and owner in file_ids.get(f, seen))) ) if witnesses: return "found", witnesses[:3] if evidence: return "unattributed", sorted(evidence)[:3] return "missing", [] c1 = [ (f"{u}.{v}", *classify(type_files.get(v, set()), u, uowner.get(u, ""), v)) for (u, v) in rendered_pairs ] c2 = [ ( e, *classify( ctor_files.get(e, set()), ev_union.get(e, ""), ev_owner.get(ev_union.get(e, ""), ""), e, ), ) for e in dispatched ] control1 = [k for k, status, _ in c1 if status == "found"] control2 = [k for k, status, _ in c2 if status == "found"] missing1 = [(k, w) for k, s, w in c1 if s == "missing"] missing2 = [(k, w) for k, s, w in c2 if s == "missing"] unattributed1 = [k for k, s, _ in c1 if s == "unattributed"] unattributed2 = [k for k, s, _ in c2 if s == "unattributed"] # Vacuous-pass guard. With no control material, "every control was found" is # `0 == 0` and says nothing - a corpus that exercises none of these claims # would otherwise print `controls pass: True` and a clean-looking run. That is # a check that cannot fail, which is the exact defect this script exists to # catch, so it is refused here instead of reported as a pass. controls_meaningful = bool(rendered) and bool(dispatched) controls_pass = controls_meaningful and not missing1 and not missing2 def ratio(hit: list, total: list) -> str: return f"{len(hit)}/{len(total)}" + ( f" = {100 * len(hit) / len(total):.1f}%" if total else "" ) # ---- the measurement ---- verified1, contradicted1 = [], [] for f in r1: u = f["data"]["union"] # R001 reports per root cause now: `variants` (a list) on grouped and # dormant findings, `variant` (a string) on the older per-variant shape. raw = f["data"].get("variants") candidates = raw if isinstance(raw, list) else [f["data"].get("variant")] for v in [name for name in candidates if name]: where = scoped( type_files.get(v, set()), u, uowner.get(u, ""), v, exclude={vdecl[(u, v)]} if (u, v) in vdecl else None, ) (contradicted1 if where else verified1).append( {"union": u, "variant": v, "where": sorted(where)[:3]} ) verified6, contradicted6 = [], [] for f in r6: e = f["data"]["event"] where = scoped( ctor_files.get(e, set()), ev_union.get(e, ""), ev_owner.get(ev_union.get(e, ""), ""), e, exclude={edecl[e]} if e in edecl else None, ) (contradicted6 if where else verified6).append( {"event": e, "where": sorted(where)[:3]} ) result = { "repo": str(root), "controls": { "rendered_variants_checker_finds": ratio(control1, rendered_pairs), "dispatched_events_checker_finds": ratio(control2, dispatched), "rendered_variant_evidence_unattributable": unattributed1, "dispatched_event_evidence_unattributable": unattributed2, "rendered_variants_with_no_evidence": missing1, "dispatched_events_with_no_evidence": missing2, "controls_meaningful": controls_meaningful, "controls_pass": controls_pass, }, "emitted-not-rendered": { # One finding can now carry several variants (grouped by root cause), # so precision is reported over the variant-level claims and both # denominators are stated. Dividing verified variants by findings # produced a nonsensical >100% precision. "findings": len(r1), "claims": len(verified1) + len(contradicted1), "verified": len(verified1), "contradicted": contradicted1, "precision": ratio(verified1, verified1 + contradicted1), }, "dead-handler": { "findings": len(r6), "claims": len(verified6) + len(contradicted6), "verified": len(verified6), "contradicted": contradicted6, "precision": ratio(verified6, verified6 + contradicted6), }, "caveat": ( "Adjudicates the factual claim only (never type-tested in UI / never " "constructed elsewhere). It does not judge UX severity, and it is " "measured on one repo, so it bounds rather than eliminates FP risk." ), } certified = result["controls"]["controls_meaningful"] and result["controls"]["controls_pass"] if args.format == "json": print(json.dumps(result, indent=2)) return 0 if certified else 1 c = result["controls"] print(f"repo {root}") print(f"CONTROL rendered variants the checker finds: {c['rendered_variants_checker_finds']}" f" (unattributable {len(c['rendered_variant_evidence_unattributable'])})") print(f"CONTROL dispatched events the checker finds: {c['dispatched_events_checker_finds']}" f" (unattributable {len(c['dispatched_event_evidence_unattributable'])})") if not c["controls_meaningful"]: print(" NO SIGNAL - this corpus renders no variants and/or dispatches no") print(" events, so the controls cannot discriminate at all. Any precision") print(" below is UNDEFINED, not perfect. Pick a corpus that exercises them.") elif not c["controls_pass"]: print(" ⚠ controls FAILED - evidence is MISSING for claims pipe makes, so the") print(" numbers below are not trustworthy:") for k, _w in (c["rendered_variants_with_no_evidence"] + c["dispatched_events_with_no_evidence"])[:10]: print(f" MISSING {k}") else: print(" controls pass") print() for rule in ("emitted-not-rendered", "dead-handler"): r = result[rule] precision = r["precision"] if r["claims"] else "undefined (no claims to check)" print(f"{rule:22s} findings={r['findings']:3d}" f" claims={r['claims']:3d} verified={r['verified']:3d}" f" precision={precision}") for bad in r["contradicted"][:10]: print(f" CONTRADICTED {bad}") print() print("caveat: " + result["caveat"]) return 0 if certified else 1 if __name__ == "__main__": raise SystemExit(main())