src/uofa_cli/adversarial/classifier.py

"""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)
            #   "<float>"       — 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_dir>/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