#!/usr/bin/env python3 """challenge_lib.py -- shared harness for the LintQ challenge example set. This module runs every example in ``challenges.json`` through *both* LintQ engines (Engine A: ``engine/lintq.dl`` over the extractor EDB schema; Engine B: ``rules/lintq.dl`` over the compact ``Op/ActsOn/Succ`` schema) and classifies each detector's behaviour against the ground-truth labels. It is the single source of truth for the acceptance logic, shared by: * ``verify_challenges.py`` -- a human-readable CI gate (exits non-zero on any mismatch). * ``test_challenges.py`` -- a ``unittest`` suite that turns each (example, detector) pair into a real test case, so the examples double as unit tests. Ground-truth label vocabulary ----------------------------- Each example carries a ``ground_truth`` map keyed by detector id (``A_OpAfterMeas``, ``B_MissingReset``, ...). The expected verdict for a label is: FIRE -- the bug class is present; this detector MUST emit a warning. SILENT -- the bug class is intentionally ABSENT; the detector MUST stay silent (negative example). FALSE_POS -- a *known, documented* engine limitation: a reset-blind (or otherwise limited) detector will erroneously fire here. The engine firing is the EXPECTED manifestation of that limitation, so it is reported as FALSE_POS(limitation) and is NOT a regression (an upgraded, limitation-free detector would instead produce SILENT(limitation)). Kept distinct from a plain FALSE_POS so the report is honest about *why* a fire happened. N/A -- this example does not exercise this detector (e.g. Engine A has no MissingReset rule, or the example targets a different bug class). N/A is **not asserted**: a capable detector firing on an out-of-scope circuit is allowed and is reported truthfully but never treated as a failure. This is what makes the harness regression-catching rather than regression-hiding: only explicit FIRE / SILENT / FALSE_POS expectations can fail. A detector is "capable" when its owning engine actually implements it. That is a static property of the engine, NOT inferred from whether a warning file happened to be produced -- inferring capability from output was the bug in the original harness (it silently relabelled a missing FIRE as N/A instead of a FALSE_NEG). """ from __future__ import annotations import json import os import subprocess import tempfile HERE = os.path.dirname(os.path.abspath(__file__)) REPO = os.path.dirname(HERE) ENGINE_A_DL = os.path.join(REPO, "engine", "lintq.dl") ENGINE_B_DL = os.path.join(REPO, "rules", "lintq.dl") CHALLENGES_PATH = os.path.join(HERE, "challenges.json") # Which detectors each engine implements (static capability map). ENGINE_DETECTORS = { "A": {"OpAfterMeas", "DoubleMeas", "GhostCompose"}, "B": {"OpAfterMeas", "DoubleMeas", "MissingReset"}, } # detector id -> (owning engine, bug class) DETECTOR_BUG = { "A_OpAfterMeas": ("A", "OpAfterMeas"), "A_DoubleMeas": ("A", "DoubleMeas"), "A_GhostCompose": ("A", "GhostCompose"), "B_OpAfterMeas": ("B", "OpAfterMeas"), "B_DoubleMeas": ("B", "DoubleMeas"), "B_MissingReset": ("B", "MissingReset"), } # detector id -> Souffle output relation that carries its warnings DETECTOR_TO_WARN = { "A_OpAfterMeas": "WarnOpAfterMeas", "A_DoubleMeas": "WarnDoubleMeas", "A_GhostCompose": "WarnGhostCompose", "B_OpAfterMeas": "WarnOpAfterMeas", "B_DoubleMeas": "WarnDoubleMeas", "B_MissingReset": "WarnMissingReset", } ALL_WARNINGS = ( "WarnOpAfterMeas", "WarnDoubleMeas", "WarnGhostCompose", "WarnMissingReset", ) def detector_capable(det: str) -> bool: """True iff the engine that owns ``det`` implements it.""" engine, bug = DETECTOR_BUG[det] return bug in ENGINE_DETECTORS[engine] def run_engine(dl: str, facts_dir: str, out_dir: str) -> set[str]: """Run Souffle over ``facts_dir`` and return the set of warning relations that were emitted **non-empty**.""" os.makedirs(out_dir, exist_ok=True) subprocess.run( ["souffle", "-F", facts_dir, "-D", out_dir, dl], check=True, capture_output=True, text=True) fired = set() for name in ALL_WARNINGS: path = os.path.join(out_dir, name + ".csv") if os.path.exists(path) and os.path.getsize(path) > 0: fired.add(name) return fired def extractor_facts(source: str, facts_dir: str): """Parse ``source`` with the AST extractor and write Engine-A EDB facts. Returns the extractor instance (used by the compact-fact converter).""" import sys sys.path.insert(0, os.path.join(REPO, "extractor")) import ast_to_facts # noqa: E402 ext = ast_to_facts.QiskitFactExtractor() ext.run(source, facts_dir) return ext def write_compact_facts(ext, facts_dir: str) -> None: """Synthesise the Engine B (Op/ActsOn/Succ) schema from the extractor facts. We walk statements in ascending id order, appending one Op row per gate/measure fact (reset is modelled as kind="reset"), and chain CFG successor edges into Succ over those op ids. The same op id is used for Op, ActsOn and Succ so they line up exactly. Engine B's reset-aware OpAfterMeas then correctly sees the `reset` op between a measure and a later gate (e.g. ex02, ex07, ex10).""" ops, acts, succ = [], [], [] oid = 0 s2o: dict[str, list[int]] = {} stmts = sorted(int(s[0]) for s in ext.stmts) for sid in stmts: sid = str(sid) local: list[int] = [] for (gsid, circuit, gate, q) in ext.gate_ops: if str(gsid) == sid: oid += 1 kind = "reset" if gate == "reset" else "gate" ops.append((oid, kind)) acts.append((oid, q)) local.append(oid) for (msid, circuit, q, c) in ext.measure_ops: if str(msid) == sid: oid += 1 ops.append((oid, "meas")) acts.append((oid, q)) local.append(oid) s2o[sid] = local prev = None for sid in stmts: cur = s2o[str(sid)] if prev and cur: for p in prev: for c in cur: succ.append((p, c)) prev = cur if cur else prev def dump(fname, rows): with open(os.path.join(facts_dir, fname), "w") as fh: for r in rows: fh.write("\t".join(str(x) for x in r) + "\n") dump("Op.facts", ops) dump("ActsOn.facts", acts) dump("Succ.facts", succ) def classify_verdict(expected: str, did_fire: bool) -> str: """Map (expected label, actual firing) to a verdict for a *capable* detector.""" if expected == "FIRE": return "FIRE" if did_fire else "FALSE_NEG" if expected == "SILENT": return "SILENT" if not did_fire else "FALSE_POS" if expected == "FALSE_POS": return "SILENT(limitation)" if not did_fire else "FALSE_POS(limitation)" # expected == "N/A": out of scope for this example; never asserted. return "N/A" def is_mismatch(verdict: str) -> bool: """A result is a failure (regression) only when an explicit expectation was violated: a missing FIRE (FALSE_NEG) or an *unwanted* fire on a SILENT negative example (FALSE_POS). ``FALSE_POS(limitation)`` is the *expected* manifestation of a recorded engine limitation (the engine firing is documented), so it is NOT a failure; ``N/A`` and the correct positive/negative verdicts are likewise not failures.""" return verdict in ("FALSE_NEG", "FALSE_POS") class ChallengeResult: """One (example, detector) evaluation.""" __slots__ = ("ex_id", "detector", "expected", "did_fire", "capable", "verdict") def __init__(self, ex_id, detector, expected, did_fire, capable, verdict): self.ex_id = ex_id self.detector = detector self.expected = expected self.did_fire = did_fire self.capable = capable self.verdict = verdict @property def failed(self) -> bool: return is_mismatch(self.verdict) def evaluate_example(ex: dict, tmp_root: str) -> list[ChallengeResult]: """Run one challenge example through both engines and return its results.""" import sys sys.path.insert(0, os.path.join(REPO, "extractor")) import ast_to_facts # noqa: E402 (kept local; re-import safe) ex_id = ex["id"] labels = ex["ground_truth"] src = os.path.join(HERE, ex["file"]) with open(src) as fh: source = fh.read() # ---- Engine A ---- a_facts = os.path.join(tmp_root, ex_id, "A") ext = ast_to_facts.QiskitFactExtractor() ext.run(source, a_facts) a_out = os.path.join(tmp_root, ex_id, "A_out") a_warn = run_engine(ENGINE_A_DL, a_facts, a_out) # ---- Engine B (compact facts from extractor) ---- b_facts = os.path.join(tmp_root, ex_id, "B") os.makedirs(b_facts, exist_ok=True) write_compact_facts(ext, b_facts) b_out = os.path.join(tmp_root, ex_id, "B_out") b_warn = run_engine(ENGINE_B_DL, b_facts, b_out) results: list[ChallengeResult] = [] for det, expected in labels.items(): warn = DETECTOR_TO_WARN[det] engine = det[0] # "A" or "B" did_fire = (warn in a_warn) if engine == "A" else (warn in b_warn) capable = detector_capable(det) if not capable: verdict = "N/A" else: verdict = classify_verdict(expected, did_fire) results.append(ChallengeResult( ex_id, det, expected, did_fire, capable, verdict)) return results def load_challenges() -> dict: with open(CHALLENGES_PATH) as fh: return json.load(fh) def run_all() -> list[ChallengeResult]: """Evaluate every example; return the flattened list of results.""" data = load_challenges() tmp_root = tempfile.mkdtemp(prefix="lintq_challenges_") all_results: list[ChallengeResult] = [] for ex in data["examples"]: all_results.extend(evaluate_example(ex, tmp_root)) return all_results def count_mismatches(results: list[ChallengeResult]) -> int: return sum(1 for r in results if r.failed)