682be14f97
The adjudicator printed "controls FAILED" or "NO SIGNAL" and then exited 0, so a CI step reading the exit status took an uncertified run as a pass. It now exits 1 unless the controls are meaningful and pass, in both text and json output. It was also described as independent. Its evidence is independent (its own regex over source), but the list of unions, variants, events and owners it checks comes from pipe's `state` output, so a variant pipe fails to enumerate is never checked. The docstring and README now say so instead of "independent adjudicator".
447 lines
19 KiB
Python
447 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Adjudicator for pipe's two highest-volume rules.
|
|
|
|
Usage:
|
|
python3 tool/adjudicate.py <repo-path> [--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())
|