"""Outcome classifier for adversarial coverage experiments — Phase 2 §10. `uofa adversarial analyze` reads a batch manifest produced by :mod:`uofa_cli.adversarial.runner`, runs ``uofa rules`` per generated package to extract the rule firings, and classifies each package into one of seven outcome classes per spec §10.3: COV-HIT target rule fires; no unexpected rules COV-HIT-PLUS target fires; other rules also fire COV-MISS target does not fire; no other rules fire COV-WRONG target does not fire; different rules fire COV-CLEAN-CORRECT no rules fire (negative_control as desired) COV-CLEAN-WRONG rules fire on a negative_control (precision bug) GEN-INVALID package SHACL-failed during generation Output CSVs (per spec §10.3, §10.4, §11.2): outcomes.csv per-package row matrix.csv aggregated catalog × subtlety pivot summary.csv per-pattern aggregate (one row per shipped UofA core pattern; schema per UofA_Phase2_M4_Cleanup_Spec.md) Skeleton-mode generation: each synthetic package is a fresh standalone generation, not a delta against the underlying base COU. The classifier therefore reads observed Jena rule firings directly — no baseline subtraction, no inheritance assumption. ``base_cou_key`` is still recorded per row so summary.csv can bucket recall by COU (D1, v1.8 §10.4). Note on summary.csv: View-3 overall precision/recall metrics (catalog_recall, catalog_precision_1_minus_fpr, gap_probe_miss_rate) are emitted in the HTML report's View 3 only — they were previously also written to summary.csv but were moved out so summary.csv can carry the per-pattern aggregate that the D1 extension spec depends on. """ from __future__ import annotations import argparse import csv import json import os import re import socket import subprocess import time from collections import Counter, defaultdict from dataclasses import asdict, dataclass, field from pathlib import Path from uofa_cli.output import error, info, result_line, warn # Active UofA core patterns at v0.5.4. Source of truth: # packs/core/rules/uofa_weakener.rules (`uofa catalog` reports the # same 23 IDs at v0.5.4 HEAD; COMPOUND-02 is commented out and # excluded). Used as the row index for summary.csv per-pattern aggregate. _CORE_PATTERN_IDS: tuple[str, ...] = ( "W-AR-01", "W-AR-02", "W-AR-03", "W-AR-04", "W-AR-05", "W-EP-01", "W-EP-02", "W-EP-03", "W-EP-04", "W-AL-01", "W-AL-02", "W-ON-01", "W-ON-02", "W-SI-01", "W-SI-02", "W-CON-01", "W-CON-02", "W-CON-03", "W-CON-04", "W-CON-05", "W-PROV-01", "COMPOUND-01", "COMPOUND-03", ) SUMMARY_FIELDS: tuple[str, ...] = ( "pattern_id", "confirm_existing_count", "confirm_existing_hits", "recall", "negative_control_firings", "gap_probe_firings", "total_firings_across_battery", # D1 (v1.8) per-COU coverage delta columns "recall_morrison_cou1", "recall_morrison_cou2", "recall_nagaraja", "recall_min_per_cou", "recall_cou_disparity", "cou_dependent_flag", # M5-B (v0.5.7) per-rule precision columns. Address the "0 COV-HIT, # 100% HIT-PLUS" finding from Apr 27 M5 analyze: when a target rule # fires, other rules fire too, so package-level outcome class # conflates per-rule specificity with rule overlap. These columns # decompose: # nc_fpr — fraction of negative_control packages where THIS rule # fired. Direct FPR-driver signal: large values point at # over-permissive rule patterns deserving §13.3 audit. # precision_when_fires — across ALL rows where this rule fired # (any intent), fraction where the spec actually targeted # this rule (intent==confirm_existing or interaction with # target_weakener == this rule). Low values mean the rule # fires on the wrong things; high values mean it's a # specific, well-targeted detector. "nc_fpr", "precision_when_fires", ) # D1 (v1.8): _detect_baseline_key returns one of these for shipped base COUs. # These are the column-name suffixes used in summary.csv per-COU columns. _COU_KEY_TO_COLUMN: dict[str, str] = { "morrison/cou1": "recall_morrison_cou1", "morrison/cou2": "recall_morrison_cou2", "nagaraja/cou1": "recall_nagaraja", } # D1 (v1.8): rules with recall disparity >= this threshold across base COUs # are flagged as COU-dependent. Threshold per Phase 2 Spec v1.8 §10.4. COU_DEPENDENT_DISPARITY_THRESHOLD: float = 0.30 @dataclass class _OutcomeRow: spec_id: str variant_num: int target_weakener: str | None source_taxonomy: str | None coverage_intent: str subtlety: str outcome_class: str rules_fired: str target_rule_fired: bool section_6_7_candidate: str | None shacl_retries: int tokens: int cost_usd: float # D1 (v1.8): which base COU this row's spec was generated against. Values # are keys of _COU_KEY_TO_COLUMN ("morrison/cou1" etc.) or None for specs # whose base_cou doesn't match a shipped COU. Internal-only — not written # to outcomes.csv (v1.8 §10.3 only adds D2 timing columns). base_cou_key: str | None = None # D2 (v1.8) per-package timing capture; see Phase 2 Spec §5.4. total_eval_ms: int = 0 jena_load_ms: int = 0 jena_inference_ms: int = 0 output_serialize_ms: int = 0 eval_host_id: str = "" def _detect_baseline_key(base_cou: str | None) -> str | None: """Match ``packs/vv40/examples/morrison/cou1`` → ``morrison/cou1``. Used by D1 per-COU recall bucketing in summary.csv. The matched key is one of the entries in :data:`_COU_KEY_TO_COLUMN`. """ if not base_cou: return None s = str(base_cou) for key in _COU_KEY_TO_COLUMN: if key in s: return key return None # Pattern matching for `uofa rules` output (one annotation per line). _RULE_LINE = re.compile(r"^\s*[⚠⚡]\s+(W-[A-Z]+-\d+|COMPOUND-\d+)\s+\[") _HIT_COUNT = re.compile(r"—\s+(\d+)\s+hit\(s\)") def _parse_rule_firings_from_check(stdout: str) -> dict[str, int]: """Extract ``{pattern_id: hit_count}`` from `uofa check` / `uofa rules` stdout.""" firings: dict[str, int] = {} for line in stdout.splitlines(): m = _RULE_LINE.search(line) if not m: continue pattern = m.group(1) hit_match = _HIT_COUNT.search(line) firings[pattern] = int(hit_match.group(1)) if hit_match else 1 return firings def _resolve_eval_host_id() -> str: """D2 (v1.8) host id: env var override or hostname fallback.""" return os.environ.get("UOFA_EVAL_HOST_ID") or socket.gethostname() or "unknown" def _run_rules_on_package( package_path: Path, pack: str = "vv40" ) -> tuple[dict[str, int], dict[str, int]]: """Invoke `uofa rules` on a package and return (firings, timings). timings dict keys: ``total_eval_ms``, ``jena_load_ms``, ``jena_inference_ms``, ``output_serialize_ms``. The Jena split is best-effort — without Java-side instrumentation we cannot separate load from inference, so ``jena_load_ms`` is 0 and ``jena_inference_ms`` carries the lumped subprocess cost. ``output_serialize_ms`` is the Python-side parse time. See Phase 2 Spec v1.8 §5.4 fallback. Returns ({}, {timings}) on subprocess error so the classifier records GEN-INVALID without crashing. """ timings = { "total_eval_ms": 0, "jena_load_ms": 0, "jena_inference_ms": 0, "output_serialize_ms": 0, } t_start = time.perf_counter_ns() try: result = subprocess.run( ["python", "-m", "uofa_cli", "rules", "--pack", pack, str(package_path)], capture_output=True, text=True, timeout=120, ) except (subprocess.TimeoutExpired, OSError): timings["total_eval_ms"] = int((time.perf_counter_ns() - t_start) / 1_000_000) return {}, timings t_subprocess_end = time.perf_counter_ns() firings = _parse_rule_firings_from_check(result.stdout) t_parse_end = time.perf_counter_ns() subprocess_ms = int((t_subprocess_end - t_start) / 1_000_000) parse_ms = int((t_parse_end - t_subprocess_end) / 1_000_000) timings["total_eval_ms"] = subprocess_ms + parse_ms # Lump subprocess time into jena_inference_ms (see docstring fallback). timings["jena_inference_ms"] = subprocess_ms timings["output_serialize_ms"] = parse_ms return firings, timings def _classify( coverage_intent: str, target_weakener: str | None, firings: dict[str, int], package_exists: bool, ) -> tuple[str, bool]: """Return ``(outcome_class, target_rule_fired)``. *firings* is the raw ``{pattern_id: hit_count}`` dict observed by running the rule engine on the synthetic package. Skeleton-mode generation produces standalone packages, so observed firings ARE attributable to the package itself — no baseline subtraction. """ if not package_exists: return "GEN-INVALID", False fired = set(firings.keys()) target_fired = bool(target_weakener and target_weakener in fired) if coverage_intent == "negative_control": if not fired: return "COV-CLEAN-CORRECT", False return "COV-CLEAN-WRONG", False if coverage_intent == "gap_probe": if not fired: return "COV-MISS", False return "COV-WRONG", False if coverage_intent == "interaction": # Interaction expects multiple firings; we report HIT-PLUS if any # rule fires (the spec §13.4 acceptance for INT-1..4 checks # COMPOUND firings via the per-pattern matrix, not via this label). if not fired: return "COV-MISS", False return "COV-HIT-PLUS", target_fired # confirm_existing if target_fired: if len(fired) == 1: return "COV-HIT", True return "COV-HIT-PLUS", True if not fired: return "COV-MISS", False return "COV-WRONG", False @dataclass class _PackageWork: """One per-variant work item for the parallel Jena phase. Pre-computed metadata so the Jena phase only needs `package_path` + `package_exists`; the classification phase reuses everything else via field name. Held in original iteration order so the final rows list is identical to the sequential implementation. """ spec_id: str variant_num: int target_weakener: str | None source_taxonomy: str | None coverage_intent: str subtlety: str shacl_retries: int tokens: int cost_usd: float base_cou_key: str | None package_path: Path | None package_exists: bool # Filled in by the Jena phase: firings: dict[str, int] = field(default_factory=dict) timings: dict[str, int] = field( default_factory=lambda: { "total_eval_ms": 0, "jena_load_ms": 0, "jena_inference_ms": 0, "output_serialize_ms": 0, } ) def _run_rules_on_work(work: _PackageWork, pack: str) -> _PackageWork: """Thread worker: invoke Jena on *work*'s package and stash results. Returns the same _PackageWork instance with `firings` and `timings` populated. Returning the work item (rather than a tuple) makes the caller's `as_completed` reassembly trivial — the work item carries its own original-order index implicitly via list position. """ if work.package_exists and work.package_path is not None: work.firings, work.timings = _run_rules_on_package( work.package_path, pack=pack ) return work def _scan_outcomes( in_dir: Path, pack: str, parallel: int = 1 ) -> list[_OutcomeRow]: """Walk the batch_manifest.perSpecResults and produce per-package rows. *parallel* controls the Jena concurrency. Default 1 preserves the sequential behavior. Higher values use a ThreadPoolExecutor over the per-package `_run_rules_on_package` calls — each spawns its own JVM subprocess so thread-safety is essentially free. Phase 2 §11 metric semantics are unchanged: classification + accumulation happen sequentially in original order after the parallel Jena phase completes, so outcomes.csv row order is identical to the sequential implementation. """ manifest_path = in_dir / "batch_manifest.json" if not manifest_path.exists(): raise FileNotFoundError( f"batch_manifest.json not found in {in_dir}; " f"run `uofa adversarial run` first" ) batch = json.loads(manifest_path.read_text()) # D2 (v1.8): single host id per analyze invocation. eval_host_id = _resolve_eval_host_id() # Per-rule timing accumulator for rule_timing.csv (D2). Keyed by # (rule_id, package_path); value is rule_eval_ms. With Jena's native # per-rule timing unavailable, we record only that the rule fired # (rule_eval_ms = 0 placeholder) so the CSV is shape-correct for # downstream consumers. The fallback note in batch_manifest documents # the limitation. rule_timing_rows: list[dict] = [] rows: list[_OutcomeRow] = [] # Phase 1: build the work list in original (perSpecResults × variants) # order. Each item is a _PackageWork carrying everything Phase 2 (Jena) # and Phase 3 (classify+accumulate) need. Holding original order in # this list guarantees outcomes.csv row order is preserved when # parallel > 1. work_items: list[_PackageWork] = [] for per_spec in batch.get("perSpecResults", []): spec_id = per_spec["spec_id"] spec_out_dir = Path(per_spec["out_dir"]) coverage_intent = per_spec["coverage_intent"] target_weakener = per_spec.get("target_weakener") source_taxonomy = per_spec.get("source_taxonomy") per_spec_manifest_path = spec_out_dir / "manifest.json" if not per_spec_manifest_path.exists(): warn(f" (no per-spec manifest for {spec_id}; skipping)") continue per_spec_manifest = json.loads(per_spec_manifest_path.read_text()) # Detect base_cou key from spec_path → base_cou (best effort: read # spec). Used downstream for D1 per-COU recall bucketing in # summary.csv. ``base_cou_key`` is None when the spec's base_cou # doesn't match a shipped COU (e.g., NASA cross-pack specs). baseline_key = None try: spec_path = Path(per_spec["spec_path"]) from uofa_cli.adversarial.spec_loader import load_spec spec_obj = load_spec(spec_path) baseline_key = _detect_baseline_key(str(spec_obj.base_cou)) except Exception: pass for variant in per_spec_manifest.get("variants", []): variant_num = variant.get("variantNum") or variant.get("variant_num") package_path_str = ( variant.get("packagePath") or variant.get("package_path") ) tokens = variant.get("tokens", 0) shacl_retries = variant.get("shaclRetries") or variant.get("shacl_retries", 0) shacl_passed = ( variant.get("shaclPassed", False) or variant.get("shacl_passed", False) ) # Generator writes packagePath as a bare filename relative to # the per-spec output directory. Resolve it. package_path = None if package_path_str: p = Path(package_path_str) if not p.is_absolute(): p = spec_out_dir / p package_path = p package_exists = ( package_path is not None and package_path.exists() and shacl_passed ) work_items.append(_PackageWork( spec_id=spec_id, variant_num=variant_num or 0, target_weakener=target_weakener, source_taxonomy=source_taxonomy, coverage_intent=coverage_intent, subtlety=per_spec_manifest.get("subtlety", "high"), shacl_retries=shacl_retries, tokens=tokens, cost_usd=variant.get("estimatedCostUsd", 0.0), base_cou_key=baseline_key, package_path=package_path, package_exists=package_exists, )) # ----- Phase 2: parallel Jena ----- # Each worker calls _run_rules_on_package, which spawns its own JVM # subprocess. Thread-safety is free — no shared state between workers. # Default parallel=1 preserves sequential behavior; M5-scale batches # benefit from parallel=5+ (see m5_findings.md F7). parallel = max(1, int(parallel or 1)) if parallel == 1 or len(work_items) == 0: for w in work_items: _run_rules_on_work(w, pack=pack) else: from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=parallel) as pool: # Submit; we don't need as_completed because each worker # mutates the work item in place (it's the same object the # main thread will iterate sequentially in Phase 3). list(pool.map(lambda w: _run_rules_on_work(w, pack=pack), work_items)) # ----- Phase 3: sequential classification + accumulation ----- # Iterates work_items in original order, so outcomes.csv row order # is identical to the pre-F7 implementation. for w in work_items: firings = w.firings timings = w.timings # D2: collect per-rule timing rows. Jena native per-rule timing # is not exposed; record rule_fired only (rule_eval_ms 0). for pat in firings: rule_timing_rows.append({ "rule_id": pat, "package_path": str(w.package_path) if w.package_path else "", "rule_eval_ms": 0, "rule_fired": "True", }) outcome_class, target_fired = _classify( coverage_intent=w.coverage_intent, target_weakener=w.target_weakener, firings=firings, package_exists=w.package_exists, ) rules_fired_str = ",".join(sorted(firings.keys())) rows.append(_OutcomeRow( spec_id=w.spec_id, variant_num=w.variant_num, target_weakener=w.target_weakener, source_taxonomy=w.source_taxonomy, coverage_intent=w.coverage_intent, subtlety=w.subtlety, outcome_class=outcome_class, rules_fired=rules_fired_str, target_rule_fired=target_fired, section_6_7_candidate=None, shacl_retries=w.shacl_retries, tokens=w.tokens, cost_usd=w.cost_usd, base_cou_key=w.base_cou_key, total_eval_ms=timings["total_eval_ms"], jena_load_ms=timings["jena_load_ms"], jena_inference_ms=timings["jena_inference_ms"], output_serialize_ms=timings["output_serialize_ms"], eval_host_id=eval_host_id, )) # Stash rule_timing_rows on the function for run_analyze to pick up. # (Cleaner than rewiring the return tuple; classifier callers within # the codebase only consume rows.) _scan_outcomes._last_rule_timing_rows = rule_timing_rows # type: ignore[attr-defined] return rows # Internal-only _OutcomeRow fields that are NOT exported to outcomes.csv. # (D1 v1.8: base_cou_key is internal to the per-COU aggregator; v1.8 §10.3 # does not add a base_cou column to outcomes.csv.) _OUTCOMES_CSV_EXCLUDED_FIELDS: frozenset[str] = frozenset({"base_cou_key"}) def _write_outcomes_csv(rows: list[_OutcomeRow], out_path: Path) -> None: out_path.parent.mkdir(parents=True, exist_ok=True) fieldnames = [ f for f in _OutcomeRow.__dataclass_fields__ if f not in _OUTCOMES_CSV_EXCLUDED_FIELDS ] with open(out_path, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") w.writeheader() for row in rows: w.writerow(asdict(row)) def _build_matrix(rows: list[_OutcomeRow]) -> dict: """Pivot 1: catalog self-coverage = HIT+HIT-PLUS / total per (pattern × subtlety).""" pivot: dict[tuple[str, str], dict[str, int]] = defaultdict(lambda: {"hit": 0, "total": 0}) for r in rows: if r.coverage_intent != "confirm_existing" or not r.target_weakener: continue key = (r.target_weakener, r.subtlety) pivot[key]["total"] += 1 if r.outcome_class in ("COV-HIT", "COV-HIT-PLUS"): pivot[key]["hit"] += 1 return pivot def _write_matrix_csv(rows: list[_OutcomeRow], out_path: Path) -> None: pivot = _build_matrix(rows) out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", newline="") as f: w = csv.writer(f) w.writerow(["pattern", "subtlety", "hit_rate", "hits", "total"]) for (pattern, subtlety), counts in sorted(pivot.items()): total = counts["total"] rate = counts["hit"] / total if total else 0.0 w.writerow([pattern, subtlety, f"{rate:.3f}", counts["hit"], total]) def _split_rules_fired(rules_fired: str) -> set[str]: """Split a comma-separated ``rules_fired`` field into a set of pattern IDs.""" if not rules_fired: return set() return {p.strip() for p in rules_fired.split(",") if p.strip()} def _write_summary_csv(rows: list[_OutcomeRow], out_path: Path) -> None: """Per-pattern aggregate, one row per active UofA core pattern. Schema per `UofA_Phase2_M4_Cleanup_Spec.md` (closes v1.7 §13.1 gate #9 and unblocks the D1 extension spec, which appends per-COU breakdown columns to this same file): pattern_id, confirm_existing_count, confirm_existing_hits, recall, negative_control_firings, gap_probe_firings, total_firings_across_battery All counts derived from *rows* — no re-evaluation of packages. """ # Per-pattern accumulators, indexed by pattern_id from _CORE_PATTERN_IDS. confirm_count: Counter[str] = Counter() # evaluable rows only confirm_hits: Counter[str] = Counter() confirm_invalid: Counter[str] = Counter() # GEN-INVALID rows targeting nc_firings: Counter[str] = Counter() gp_firings: Counter[str] = Counter() total_firings: Counter[str] = Counter() # M5-B accumulators: # targeted_firings[R] = # rows where R fired AND the row's spec targeted R # (confirm_existing target=R, or interaction target=R) # The denominator for precision_when_fires is total_firings[R]. targeted_firings: Counter[str] = Counter() nc_total_evaluable = 0 # # NC rows excluding GEN-INVALID, used for nc_fpr # D1 (v1.8): per-(pattern, base_cou_key) accumulators for per-COU recall. per_cou_count: dict[tuple[str, str], int] = defaultdict(int) per_cou_hits: dict[tuple[str, str], int] = defaultdict(int) def _is_truthy(value) -> bool: return (value is True) or ( isinstance(value, str) and value.strip().lower() == "true" ) for r in rows: fired = _split_rules_fired(r.rules_fired) # Total firings across the battery — per pattern. for pat in fired: total_firings[pat] += 1 # confirm_existing: count attempts and hits per target pattern. # Exclude GEN-INVALID rows from the recall denominator — those are # generation failures, not coverage misses. But track them in # confirm_invalid so we can distinguish "no data for this pattern" # (empty recall) from "all data was unevaluable" (not_measurable). if r.coverage_intent == "confirm_existing" and r.target_weakener: if r.outcome_class == "GEN-INVALID": confirm_invalid[r.target_weakener] += 1 if ( r.coverage_intent == "confirm_existing" and r.target_weakener and r.outcome_class != "GEN-INVALID" ): confirm_count[r.target_weakener] += 1 # ``target_rule_fired`` arrives here as a Python bool from # _OutcomeRow but the same code path runs after CSV reads # produce strings; accept both. hit = _is_truthy(r.target_rule_fired) if hit: confirm_hits[r.target_weakener] += 1 # D1: per-COU bucketing (only when row carries a base_cou_key). cou_key = getattr(r, "base_cou_key", None) if cou_key in _COU_KEY_TO_COLUMN: per_cou_count[(r.target_weakener, cou_key)] += 1 if hit: per_cou_hits[(r.target_weakener, cou_key)] += 1 if r.coverage_intent == "negative_control": if r.outcome_class != "GEN-INVALID": nc_total_evaluable += 1 for pat in fired: nc_firings[pat] += 1 if r.coverage_intent == "gap_probe": for pat in fired: gp_firings[pat] += 1 # M5-B: a firing is "targeted" if the row's spec asked for that # rule. confirm_existing and interaction specs declare a target # weakener; gap_probe and negative_control never do (target=null # for gap_probe; sentinel for NC). if r.coverage_intent in ("confirm_existing", "interaction") and r.target_weakener: if r.target_weakener in fired: targeted_firings[r.target_weakener] += 1 out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=SUMMARY_FIELDS) w.writeheader() for pat in _CORE_PATTERN_IDS: n = confirm_count[pat] h = confirm_hits[pat] inv = confirm_invalid[pat] # Recall semantics (Phase 2 §11): # "" — no confirm_existing rows targeting this # pattern (pattern not in the experiment) # "" — n>0 evaluable rows; standard recall # "not_measurable" — n=0 evaluable but inv>0; e.g. weakener # semantics force gen-invalid (W-ON-01 omits # hasContextOfUse, W-SI-01 omits signature), # which the SHACL gate rejects → no # evaluable packages exist. Reporting "0%" # would imply rules failed to fire when in # fact rules never had a chance. if n > 0: recall_str = f"{(h / n):.3f}" elif inv > 0: recall_str = "not_measurable" else: recall_str = "" # D1: per-COU recall computation. per_cou_recall: dict[str, str] = { col: "" for col in _COU_KEY_TO_COLUMN.values() } non_empty_recalls: list[float] = [] for cou_key, col_name in _COU_KEY_TO_COLUMN.items(): count = per_cou_count.get((pat, cou_key), 0) if count == 0: continue hits = per_cou_hits.get((pat, cou_key), 0) value = hits / count per_cou_recall[col_name] = f"{value:.3f}" non_empty_recalls.append(value) if len(non_empty_recalls) >= 1: recall_min = f"{min(non_empty_recalls):.3f}" else: recall_min = "" if len(non_empty_recalls) >= 2: disparity = max(non_empty_recalls) - min(non_empty_recalls) disparity_str = f"{disparity:.3f}" cou_dependent = ( "True" if disparity >= COU_DEPENDENT_DISPARITY_THRESHOLD else "False" ) else: disparity_str = "" cou_dependent = "" # M5-B: per-rule precision computations. # nc_fpr semantics: "" when no NC data exists at all (unusual); # otherwise "0.000" through "1.000". "0.000" means rule never # fires on clean controls; "1.000" means it fires on every # clean control — direct §13.3 audit signal. if nc_total_evaluable > 0: nc_fpr_str = f"{(nc_firings[pat] / nc_total_evaluable):.3f}" else: nc_fpr_str = "" # precision_when_fires semantics: "" when this rule never fired # at all (so no precision to compute); otherwise the targeted- # firings ratio. 1.000 = rule only fires when targeted (specific); # 0.000 = rule never fires when targeted (always firing on # unrelated specs, signaling broken rule). if total_firings[pat] > 0: precision_str = f"{(targeted_firings[pat] / total_firings[pat]):.3f}" else: precision_str = "" w.writerow({ "pattern_id": pat, "confirm_existing_count": n, "confirm_existing_hits": h, "recall": recall_str, "negative_control_firings": nc_firings[pat], "gap_probe_firings": gp_firings[pat], "total_firings_across_battery": total_firings[pat], "recall_morrison_cou1": per_cou_recall["recall_morrison_cou1"], "recall_morrison_cou2": per_cou_recall["recall_morrison_cou2"], "recall_nagaraja": per_cou_recall["recall_nagaraja"], "recall_min_per_cou": recall_min, "recall_cou_disparity": disparity_str, "cou_dependent_flag": cou_dependent, "nc_fpr": nc_fpr_str, "precision_when_fires": precision_str, }) RULE_TIMING_FIELDS: tuple[str, ...] = ( "rule_id", "package_path", "rule_eval_ms", "rule_fired", ) #: Fallback note recorded in /batch_manifest.json (and in the #: companion rule_timing.csv.FALLBACK_NOTE.txt) when per-rule timing is #: unavailable from Jena natively. Per Phase 2 Spec v1.8 §10.5, under this #: fallback rule_timing.csv is OMITTED (an all-zeros rule_eval_ms file would #: be a misleading measurement). The lumped subprocess wall-clock instead #: appears in outcomes.csv jena_inference_ms / total_eval_ms. RULE_TIMING_FALLBACK_NOTE: str = ( "Per-rule wall-clock timing is not exposed by Jena's GenericRuleReasoner " "(FORWARD_RETE) without Java-side instrumentation. Per Phase 2 Spec v1.8 " "§10.5, rule_timing.csv is intentionally omitted under this fallback " "rather than emitted with rule_eval_ms=0 on every row (which would be a " "Jena-limitation marker, not a measurement). The lumped subprocess time " "appears in outcomes.csv jena_inference_ms / total_eval_ms, and this " "limitation is also recorded in batch_manifest.timing_fallback_note." ) def _write_rule_timing_csv(rule_timing_rows: list[dict], out_path: Path) -> None: """D2 (v1.8) §10.5: per-(rule, package) timing CSV. Reserved for the future non-fallback path (Java-side instrumentation exposing per-rule wall-clock). Under the current Jena GenericRuleReasoner fallback, ``run_analyze`` does NOT call this — see ``_write_rule_timing_fallback_note`` and RULE_TIMING_FALLBACK_NOTE. """ out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=RULE_TIMING_FIELDS) w.writeheader() for row in rule_timing_rows: w.writerow({k: row.get(k, "") for k in RULE_TIMING_FIELDS}) def _write_rule_timing_fallback_note(out_path: Path) -> None: """D2 (v1.8) §10.5 omit-with-note: write a companion text file alongside the place rule_timing.csv would have lived, explaining the omission. The note text is RULE_TIMING_FALLBACK_NOTE plus a header line so a human opening the file in the batch directory understands why rule_timing.csv is absent. """ out_path.parent.mkdir(parents=True, exist_ok=True) body = ( "rule_timing.csv intentionally omitted (Jena native per-rule timing " "fallback).\n\n" + RULE_TIMING_FALLBACK_NOTE + "\n" ) out_path.write_text(body) def _annotate_batch_manifest_with_timing_fallback(in_dir: Path) -> None: """Append D2 timing_fallback_note to the batch manifest if not already present. Idempotent — safe to call from analyze even when the runner didn't write it during generation.""" manifest_path = in_dir / "batch_manifest.json" if not manifest_path.exists(): return try: manifest = json.loads(manifest_path.read_text()) except (OSError, json.JSONDecodeError): return if manifest.get("timing_fallback_note"): return manifest["timing_fallback_note"] = RULE_TIMING_FALLBACK_NOTE manifest_path.write_text(json.dumps(manifest, indent=2)) def run_analyze(args) -> int: """Entry point for ``uofa adversarial analyze``.""" in_dir: Path = args.in_dir out_dir: Path = args.out pack = args.check_pack parallel = max(1, int(getattr(args, "parallel", 1) or 1)) info(f"analyzing batch at {in_dir}" + (f" (parallel={parallel})" if parallel > 1 else "")) try: rows = _scan_outcomes(in_dir, pack=pack, parallel=parallel) except FileNotFoundError as e: error(str(e)) return 2 if not rows: warn("no per-package rows produced; nothing to write") return 1 out_dir.mkdir(parents=True, exist_ok=True) outcomes_path = out_dir / "outcomes.csv" matrix_path = out_dir / "matrix.csv" summary_path = out_dir / "summary.csv" rule_timing_note_path = out_dir / "rule_timing.csv.FALLBACK_NOTE.txt" _write_outcomes_csv(rows, outcomes_path) _write_matrix_csv(rows, matrix_path) _write_summary_csv(rows, summary_path) # D2 (v1.8 §10.5): rule_timing.csv is intentionally OMITTED under the # Jena native fallback path. An all-zeros rule_eval_ms file would be a # misleading measurement; instead, drop a companion FALLBACK_NOTE # alongside the place where the file would have lived, and record the # same note in batch_manifest.timing_fallback_note. _write_rule_timing_fallback_note(rule_timing_note_path) _annotate_batch_manifest_with_timing_fallback(in_dir) # HTML report (delegated to reporter.py) from uofa_cli.adversarial.reporter import write_html_report html_path = out_dir / "index.html" write_html_report(rows, html_path) by_class = Counter(r.outcome_class for r in rows) info(f" outcomes by class: {dict(by_class)}") result_line("outcomes", True, str(outcomes_path)) result_line("matrix", True, str(matrix_path)) result_line("summary", True, str(summary_path)) result_line("rule_timing", False, "omitted — Jena native fallback " f"(see {rule_timing_note_path.name})") result_line("report", True, str(html_path)) # Phase 3 §2.1 bundle producer (opt-in via --emit-judge-bundle). # Without the flag, analyze behavior is byte-for-byte unchanged. if getattr(args, "emit_judge_bundle", False): from uofa_cli.adversarial.judge.bundle_writer import write_bundle, BundleWriteError bundle_path = out_dir / "judge_ready_bundle.tgz" try: write_result = write_bundle(in_dir, outcomes_path, bundle_path) except BundleWriteError as e: error(f"--emit-judge-bundle failed: {e}") return 3 info(f" judge bundle: {write_result.package_count} packages, " f"normalized class distribution {write_result.distribution}") result_line("judge_bundle", True, str(bundle_path)) for w in write_result.warnings: warn(w) return 0