five_point_harness.py

"""
Five-Point Harness for profile-level roundtrip validation.

Compares an original compiled blob against a roundtripped blob using five
structured checks that progressively degrade from byte-level to semantic
equivalence.  Each check provides a different lens on whether the roundtrip
preserved the profile's meaning.

Checks:
    1. **sbpl_eq**:       Canonical SBPL text comparison.
    2. **ir_eq**:         Per-operation subgraph hash comparison from
                          normalized Policy DAGs.
    3. **op_table_eq**:   Per-slot assignment_class comparison, ignoring
                          default and implicit slots.
    4. **graph_eq**:      For explicit slots, reachable subgraph structural
                          identity.
    5. **runtime_eq**:    Dual-oracle runtime comparison:
                          - PolicyWitness: sandbox_check based queries
                          - Sandboxed binary: actual file ops with secret marker
                          (None if unavailable).
"""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from integration.ir.profile.ir_builder import (
    build_profile_ir,
)
from pawl.contract.envelope import extract_subgraph_hashes
from pawl.forward import decode_policy_cd_from_compiled_blob
from integration.ir.profile.learn_contextual_implicits import (
    extract_explicit_ops_from_sbpl,
)
from integration.ir.profile.runtime_compare import (
    RuntimeCompareResult,
    compare_runtime,
    generate_mf_probe_plan,
    generate_probe_plan,
)
from runtime.oracle.profile import run_binary_oracle
from pawl.normalize.reversed import compare_source_and_reversed
from pawl.normalize.baseline.compiler_model import predict_transforms
from pawl.normalize.baseline.predicates import is_baseline_operation
from pawl.reverse.param_reconstruct import reconstruct_sbpl_params
from runtime.core.unavailability import EXECUTION_ERROR, ProbeUnavailable

# FIXME(2026-02-12): Temporary timeout wrapper to prevent hangs on certain fixtures.
# Remove once root cause is identified and fixed. See investigation notes in PLAN.md.
# Tracking: p1_regex_06_complex, p1_regex_12_named_capture_unsupported hang in FPH.
_FPH_TIMEOUT_SECONDS = 240  # keep per-casefile harness timeout conservative

_NON_EXPLICIT_ASSIGNMENT_CLASSES = frozenset({
    "default",
    "implicit_baseline",
    "implicit_nonterm_guard",
    "contextual_implicit",
    "contextual_implicit_regex",
    "contextual_implicit_filter",
    "contextual_implicit_mf",
    "bare_opposite_terminal",
    "shared_entrypoint",
    "unknown",
})


@dataclass
class ProfileFivePointResult:
    """Result of the profile five-point harness comparison.

    The harness publishes the comparison vector and distinguishing evidence
    directly. Any flat status view is a downstream derivation, not part of the
    harness contract.
    """

    checks: dict[str, bool | None]
    distinguishing: dict[str, Any]
    ir_a: dict
    ir_b: dict
    blob_hash_a: str
    blob_hash_b: str


@dataclass(frozen=True)
class _CompileDecodeSubgraphBridge:
    """Contract-level subgraph identity used by the profile harness `ir_eq` check."""

    subgraph_hashes: dict[str, str]


def _build_compile_decode_subgraph_bridge(
    blob_path: Path,
    *,
    source_path: Path,
    search_paths: list[Path] | None,
    repo_root: Path,
) -> _CompileDecodeSubgraphBridge:
    """Bridge one compiled blob into normalized per-operation subgraph hashes.

    `build_profile_ir()` owns slot classification for profile-harness checks.
    `ir_eq` is the one comparison that intentionally crosses into the normalized
    Policy DAG lane because subgraph identity is defined there, not in
    `pawl.profile.ir.v1`.
    """
    compile_decode = decode_policy_cd_from_compiled_blob(
        blob_path.read_bytes(),
        source_path,
        search_paths=search_paths or (),
        repo_root=repo_root,
    )
    return _CompileDecodeSubgraphBridge(
        subgraph_hashes=extract_subgraph_hashes(compile_decode.policy_normalized),
    )


def five_point_harness(
    source_sbpl: str,
    original_blob_path: Path,
    reversed_sbpl: str | None = None,
    roundtrip_blob_path: Path | None = None,
    *,
    repo_root: Path,
    disconnected_filters: list[dict] | None = None,
    param_bindings: dict[str, str] | None = None,
    search_paths: list[Path] | None = None,
    source_path: Path | None = None,
    timeout: int | None = None,
    node_sharing_sidecar: dict[str, Any] | None = None,
) -> ProfileFivePointResult:
    """Run the five-point harness comparing original and roundtripped profiles.

    Parameters
    ----------
    source_sbpl:
        Original SBPL source text (for extracting explicit ops).
    original_blob_path:
        Path to the original compiled blob.
    reversed_sbpl:
        SBPL text produced by the reverse engine. Optional for decode-only mode.
    roundtrip_blob_path:
        Path to the blob compiled from *reversed_sbpl*. Optional for decode-only
        mode (e.g., message-filter profiles that can't be recompiled).
    repo_root:
        Repository root.
    disconnected_filters:
        Disconnected filter information for predicate reconstruction.
    param_bindings:
        Mapping of param names to resolved values for source substitution.
    search_paths:
        Directories to search for imported profiles. When provided, import
        simulation filters out operations that came from imports (reducing
        false positives in canonical comparison).
    source_path:
        Path to the source file (for relative import resolution).
    timeout:
        Optional timeout in seconds. If None, uses _FPH_TIMEOUT_SECONDS default.

    Returns
    -------
    ProfileFivePointResult with the comparison checks and distinguishing
    evidence, or a decode-only/timeout-shaped result where the checks remain
    null and the distinguishing payload records why the comparison did not run.
    """
    # FIXME(2026-02-12): Timeout wrapper using threading - remove once hang is fixed
    # Signal-based timeout doesn't work reliably in all contexts.
    import threading
    import queue

    effective_timeout = timeout if timeout is not None else _FPH_TIMEOUT_SECONDS
    result_queue: queue.Queue = queue.Queue()

    def _run_impl():
        try:
            result = _five_point_harness_impl(
                source_sbpl=source_sbpl,
                original_blob_path=original_blob_path,
                reversed_sbpl=reversed_sbpl,
                roundtrip_blob_path=roundtrip_blob_path,
                repo_root=repo_root,
                disconnected_filters=disconnected_filters,
                param_bindings=param_bindings,
                search_paths=search_paths,
                source_path=source_path,
                node_sharing_sidecar=node_sharing_sidecar,
            )
            result_queue.put(("ok", result))
        except Exception as e:
            result_queue.put(("error", e))

    thread = threading.Thread(target=_run_impl, daemon=True)
    thread.start()
    thread.join(timeout=effective_timeout)

    if thread.is_alive():
        # Thread still running = timeout
        # Note: daemon thread will be killed when main process exits
        blob_hash = ""
        try:
            from tools import doctor
            blob_hash = doctor.sha256(original_blob_path)
        except Exception:
            pass
        return ProfileFivePointResult(
            checks={
                "sbpl_eq": None,
                "canonical_eq": None,
                "ir_eq": None,
                "op_table_eq": None,
                "graph_eq": None,
                "runtime_eq": None,
            },
            distinguishing={
                "timeout": True,
                "timeout_seconds": effective_timeout,
            },
            ir_a={},
            ir_b={},
            blob_hash_a=blob_hash,
            blob_hash_b="",
        )

    # Thread completed - get result
    try:
        status, result = result_queue.get_nowait()
        if status == "ok":
            return result
        else:
            raise result  # Re-raise exception
    except queue.Empty:
        # Shouldn't happen, but handle it
        return ProfileFivePointResult(
            checks={
                "sbpl_eq": None, "canonical_eq": None, "ir_eq": None,
                "op_table_eq": None, "graph_eq": None, "runtime_eq": None,
            },
            distinguishing={"error": "result queue empty"},
            ir_a={}, ir_b={},
            blob_hash_a="", blob_hash_b="",
        )


def _five_point_harness_impl(
    source_sbpl: str,
    original_blob_path: Path,
    reversed_sbpl: str | None = None,
    roundtrip_blob_path: Path | None = None,
    *,
    repo_root: Path,
    disconnected_filters: list[dict] | None = None,
    param_bindings: dict[str, str] | None = None,
    search_paths: list[Path] | None = None,
    source_path: Path | None = None,
    node_sharing_sidecar: dict[str, Any] | None = None,
) -> ProfileFivePointResult:
    """Implementation of five_point_harness (wrapped with timeout above)."""
    # Load sb_ops vocabulary for wildcard expansion.
    ops_vocab_path = (
        repo_root
        / "integration"
        / "carton"
        / "contract"
        / "bundle"
        / "relationships"
        / "mappings"
        / "vocab"
        / "ops.json"
    )
    sb_ops: list[str] | None = None
    if ops_vocab_path.exists():
        sb_ops = [e["name"] for e in json.loads(ops_vocab_path.read_text())["ops"]]

    source_explicit_ops = extract_explicit_ops_from_sbpl(source_sbpl, sb_ops=sb_ops)

    ir_a = build_profile_ir(
        original_blob_path,
        repo_root=repo_root,
        source_explicit_ops=source_explicit_ops,
    )

    # Decode-only mode: MF profiles (0x4000) cannot be roundtripped because
    # sandbox_apply requires the private entitlement com.apple.private.security.message-filter.
    # Also triggered if roundtrip_blob_path is not provided.
    is_mf_profile = bool(ir_a.get("profile_type_flags", 0) & 0x4000)
    decode_only = is_mf_profile or roundtrip_blob_path is None

    if decode_only:
        # Return decode-only result: IR is valid but no roundtrip comparison possible.
        decode_reason = "mf_profile_requires_private_entitlement" if is_mf_profile else "no_roundtrip_blob"
        return ProfileFivePointResult(
            checks={
                "sbpl_eq": None,
                "canonical_eq": None,
                "ir_eq": None,
                "op_table_eq": None,
                "graph_eq": None,
                "runtime_eq": None,
            },
            distinguishing={
                "decode_only": True,
                "decode_reason": decode_reason,
                "mf_profile": is_mf_profile,
                "profile_type_flags": ir_a.get("profile_type_flags", 0),
            },
            ir_a=ir_a,
            ir_b={},  # No roundtrip IR
            blob_hash_a=ir_a["blob_sha256"],
            blob_hash_b="",  # No roundtrip blob
        )

    # Build roundtrip IR for comparison
    ir_b = build_profile_ir(
        roundtrip_blob_path,
        repo_root=repo_root,
        source_explicit_ops=source_explicit_ops,
    )

    # Ensure reversed_sbpl is available for comparison
    if reversed_sbpl is None:
        reversed_sbpl = ""

    checks: dict[str, bool | None] = {}
    distinguishing: dict[str, Any] = {}

    promotion_remap = _detect_promotion_remap(ir_a, ir_b, source_explicit_ops)

    # ── Check 1: SBPL text equality ──────────────────────────────────────
    #    Canonical form, IR-informed filtering.
    #    Filter both texts to remove rules for ops the IR classifies as
    #    profile-type baseline.  This avoids false negatives from the renderer
    #    emitting baseline operations that weren't in the source.
    #
    #    If param_bindings are provided, reconstruct param references in the
    #    reversed SBPL before comparison. This handles cases where the compiler
    #    resolved (param "X") to a literal value at compile time.
    # Assignment classes that are NOT source-declared.  Operations with these
    # classes are excluded from sbpl_eq text comparison entirely — the renderer
    # may emit rules for them, but they have no source-side counterpart to match.
    _implicit_classes = {
        "implicit_baseline", "implicit_nonterm_guard",
        "contextual_implicit", "contextual_implicit_regex",
        "contextual_implicit_filter", "contextual_implicit_mf",
        "bare_opposite_terminal", "default", "shared_entrypoint",
    }
    _implicit_ops_a = {
        s["op_name"] for s in ir_a["op_table"]["slots"]
        if s["assignment_class"] in _implicit_classes
    }
    # Shared-entrypoint ops: the compiler wired these to the same decision-graph
    # node as an explicit sibling, so the reverser emits the sibling's predicates
    # on them.  They're already excluded from sbpl_eq above (implicit class), but
    # canonical_eq/semantic_eq need the list for predicate-level filtering —
    # the normalize layer doesn't have the IR, so we pass the list explicitly.
    _shared_entrypoint_ops_a = {
        s["op_name"] for s in ir_a["op_table"]["slots"]
        if s.get("assignment_class") == "shared_entrypoint"
    }
    _implicit_ops_a |= {
        s["op_name"] for s in ir_a["op_table"]["slots"]
        if s["assignment_class"] == "unknown"
        and is_baseline_operation(s["op_name"])
    }
    if promotion_remap:
        # Promoted parents are the semantic representative of source-declared
        # child operations; keep them in the sbpl_eq comparison.
        _implicit_ops_a.difference_update(promotion_remap.values())

    normalized_source = _normalize_sbpl(source_sbpl)
    if promotion_remap:
        normalized_source = _remap_sbpl_operations(normalized_source, promotion_remap)

    # Reconstruct param references in reversed SBPL if bindings provided
    reversed_sbpl_for_compare = reversed_sbpl
    if param_bindings:
        reversed_sbpl_for_compare = reconstruct_sbpl_params(reversed_sbpl, param_bindings)

    # Source-aware reconciliation (NORMALIZE-COVERAGE-PLAN Phase 1)
    from integration.ir.profile.sbpl_reconcile import reconcile_sbpl_for_compare
    reversed_sbpl_for_compare, _reconcile_info = reconcile_sbpl_for_compare(
        reversed_sbpl_for_compare,
        source_sbpl,
        param_bindings=param_bindings,
        search_paths=search_paths,
    )

    normalized_reversed = _normalize_sbpl(reversed_sbpl_for_compare)
    filtered_source = _filter_sbpl_by_ir(normalized_source, _implicit_ops_a)
    filtered_reversed = _filter_sbpl_by_ir(normalized_reversed, _implicit_ops_a)
    checks["sbpl_eq"] = filtered_source == filtered_reversed

    checks["sbpl_eq"], sbpl_eq_import_filtered = _maybe_recompute_sbpl_eq_with_import_filter(
        current_eq=checks["sbpl_eq"],
        normalized_source=normalized_source,
        normalized_reversed=normalized_reversed,
        implicit_ops=_implicit_ops_a,
        source_sbpl=source_sbpl,
        search_paths=search_paths,
        source_path=source_path,
    )
    if sbpl_eq_import_filtered:
        distinguishing["sbpl_eq_import_filtered"] = True

    # ── Check 2: Per-operation subgraph hash equality ────────────────────
    #    Decode both blobs through the forward pipeline to get normalized
    #    Policy DAGs, extract per-operation subgraph hashes, and compare.
    if source_path is not None:
        try:
            bridge_a = _build_compile_decode_subgraph_bridge(
                original_blob_path,
                source_path=source_path,
                search_paths=search_paths,
                repo_root=repo_root,
            )
            bridge_b = _build_compile_decode_subgraph_bridge(
                roundtrip_blob_path,
                source_path=source_path,
                search_paths=search_paths,
                repo_root=repo_root,
            )
            sg_a = bridge_a.subgraph_hashes
            sg_b = bridge_b.subgraph_hashes
            shared_ops = set(sg_a) & set(sg_b)
            divergent_ops = [op for op in sorted(shared_ops) if sg_a[op] != sg_b[op]]
            checks["ir_eq"] = len(divergent_ops) == 0
            if not checks["ir_eq"]:
                distinguishing["ir_divergent_ops"] = divergent_ops[:10]
                distinguishing["ir_op_count_a"] = len(sg_a)
                distinguishing["ir_op_count_b"] = len(sg_b)
        except Exception as exc:
            checks["ir_eq"] = False
            distinguishing["ir_eq_error"] = str(exc)
    else:
        checks["ir_eq"] = None
        distinguishing["ir_eq_skipped"] = "no_source_path"

    # ── Check 3: Op-table per-slot comparison (explicit + unknown only) ──
    op_diff = _compare_op_tables(ir_a, ir_b, promotion_remap=promotion_remap)
    checks["op_table_eq"] = len(op_diff) == 0
    if op_diff:
        distinguishing["op_table_diffs"] = op_diff[:10]

    # Surface unknown-slot counts for progress tracking.
    unknown_a = sum(1 for s in ir_a["op_table"]["slots"] if s["assignment_class"] == "unknown")
    unknown_b = sum(1 for s in ir_b["op_table"]["slots"] if s["assignment_class"] == "unknown")
    if unknown_a or unknown_b:
        distinguishing["unknown_count_a"] = unknown_a
        distinguishing["unknown_count_b"] = unknown_b

    # ── Check 4: Graph equality for explicit and unknown slots ───────────
    #    Use semantic comparison (decision paths) rather than structural comparison
    #    to tolerate filter reordering and tree restructuring.
    #    Exclusion set is computed across BOTH IRs so that ops non-explicit
    #    in either IR are excluded from both sides (cross-IR alignment).
    _graph_excl = _graph_exclude_ops(ir_a, ir_b)
    _has_imports = search_paths is not None
    graph_ok, graph_diverging = _compare_explicit_graphs_semantic(
        _normalize_ir_for_graph_compare(ir_a, promotion_remap=promotion_remap, source_side=True, exclude_ops=_graph_excl),
        _normalize_ir_for_graph_compare(ir_b, promotion_remap=promotion_remap, source_side=False, exclude_ops=_graph_excl),
    )
    checks["graph_eq"] = graph_ok

    # ── Check 5: Canonical policy equality ───────────────────────────────
    #    Deferred to after checks 2-4 because the REDUNDANT (C09) normalize
    #    pass needs structural gate results (ir_eq, op_table_eq, graph_eq)
    #    as context for its cluster-B guardrail.
    #
    #    Uses the pawl.normalize module to:
    #    - Normalize predicates (character class ordering, whitespace)
    #    - Filter message-filter-only predicates from wrong scope
    #    - Reconstruct predicates from disconnected filter nodes
    #    - Substitute param references with bound values
    redundant_context = {
        "source_explicit_ops": sorted(source_explicit_ops),
        "source_slots": ir_a["op_table"]["slots"],
        "reversed_slots": ir_b["op_table"]["slots"],
        "blob_equal": ir_a["blob_sha256"] == ir_b["blob_sha256"],
        "ir_eq": checks.get("ir_eq") is True,
        "op_table_eq": checks.get("op_table_eq") is True,
        "graph_eq": checks.get("graph_eq") is True,
    }
    canonical_result = compare_source_and_reversed(
        source_sbpl,
        reversed_sbpl,
        disconnected_filters=disconnected_filters or [],
        param_bindings=param_bindings,
        semantic=True,  # Also compute semantic equivalence
        filter_imports=search_paths is not None,
        filter_baseline=True,  # Filter known baseline predicates from imports
        search_paths=search_paths,
        source_path=source_path,
        augment_entitlements=True,  # Inject entitlement blocks from source into reversed
        filter_shared_predicates=True,  # Filter compiler-shared predicates (see _shared_entrypoint_ops_a)
        shared_operations=sorted(_shared_entrypoint_ops_a),  # IR-derived; normalize layer has no IR access
        filter_redundant_rules=True,  # C09 REDUNDANT filtering lives in normalize, not FPH.
        redundant_context=redundant_context,
        node_sharing_sidecar=node_sharing_sidecar,
    )
    checks["canonical_eq"] = canonical_result["equivalent"]
    # ── Check 6: Semantic equivalence ────────────────────────────────────
    #    Ignores require-any/require-all grouping.  Passes when source and
    #    reversed have the same leaf predicates even if grouping differs.
    checks["semantic_eq"] = canonical_result.get("semantic_equivalent", False)
    if _should_promote_graph_eq_for_import_context(
        has_imports=_has_imports,
        graph_ok=graph_ok,
        canonical_eq=checks["canonical_eq"],
        semantic_eq=checks["semantic_eq"],
    ):
        checks["graph_eq"] = True
        distinguishing["graph_eq_promoted_from_normalize"] = {
            "reason": "import_context_normalized_equivalent",
            "diverging_ops": graph_diverging[:10],
        }
    elif not graph_ok:
        distinguishing["graph_diverging_ops"] = graph_diverging[:10]
    if not checks["canonical_eq"]:
        distinguishing["canonical_diff"] = canonical_result.get("diff")
        # Include semantic diff if structural failed but semantic might have passed
        if checks["semantic_eq"] and canonical_result.get("semantic_diff"):
            distinguishing["semantic_diff"] = canonical_result.get("semantic_diff")
    if canonical_result.get("has_reconstruction"):
        distinguishing["has_reconstruction"] = True
        distinguishing["predicted_transforms"] = [
            t.name for t in predict_transforms(source_sbpl)
        ]
    # Include import filtering metadata if available
    if canonical_result.get("import_filtered_count"):
        distinguishing["import_filtered_count"] = canonical_result["import_filtered_count"]
    if canonical_result.get("unresolved_imports"):
        distinguishing["unresolved_imports"] = canonical_result["unresolved_imports"]
    # ── Compiler-loss adjusted eq ────────────────────────────────────────
    #    When the node-sharing sidecar is provided, compiler_loss_adjusted_eq
    #    is True when ALL remaining diffs are explained by compiler optimization.
    #    This reclassifies sub-group A failures from "reverser defect" to
    #    "compiler loss" (documented and attributed, not unknown).
    if "compiler_loss_adjusted_eq" in canonical_result:
        checks["compiler_loss_adjusted_eq"] = canonical_result["compiler_loss_adjusted_eq"]
    if canonical_result.get("node_sharing_info"):
        distinguishing["node_sharing_info"] = canonical_result["node_sharing_info"]
    if canonical_result.get("shared_predicate_info"):
        distinguishing["shared_predicate_info"] = canonical_result["shared_predicate_info"]
    if canonical_result.get("redundant_info"):
        distinguishing["redundant_info"] = canonical_result["redundant_info"]

        # The REDUNDANT normalize pass can designate a bounded set of source ops
        # to omit from sbpl_eq text comparison. This keeps sbpl_eq aligned with
        # canonical/semantic reconciliation without hardcoding category rules in FPH.
        redundant_ops = set(
            canonical_result["redundant_info"].get("sbpl_filter_operations", [])
        )
        if redundant_ops:
            checks["sbpl_eq"] = (
                _filter_sbpl_by_ir(normalized_source, _implicit_ops_a | redundant_ops)
                == _filter_sbpl_by_ir(normalized_reversed, _implicit_ops_a | redundant_ops)
            )
            distinguishing["sbpl_eq_redundant_filtered"] = sorted(redundant_ops)

    # ── Check 7: Runtime equality ────────────────────────────────────────
    #    Dual-oracle comparison.
    #    Oracle 1: PolicyWitness (sandbox_check based)
    #    Oracle 2: Sandboxed binary (actual file ops with secret marker)
    #
    #    Both must pass for runtime_eq=True. If either fails or mismatches,
    #    runtime_eq=False. If both are unavailable, runtime_eq=None.
    runtime_result: dict[str, Any] = {
        "policy_witness": None,
        "sandboxed_binary": None,
        "combined": None,
    }
    is_mf = bool(ir_a.get("profile_type_flags", 0) & 0x4000)

    # Oracle 1: PolicyWitness
    pw_result: RuntimeCompareResult | None = None
    try:
        if is_mf:
            probe_plan = generate_mf_probe_plan(ir_a)
        else:
            probe_plan = generate_probe_plan(ir_a)

        runner_requirements: dict[str, Any] | None = None
        if is_mf:
            runner_requirements = {"requires_xpc": True}

        pw_result = compare_runtime(
            source_sbpl, reversed_sbpl, probe_plan,
            runner_requirements=runner_requirements,
            param_bindings=param_bindings,
        )
        runtime_result["policy_witness"] = {
            "eq": pw_result.equal,
            "status": pw_result.status,
        }
        if pw_result.diagnostics:
            runtime_result["policy_witness"]["diagnostics"] = pw_result.diagnostics
        if pw_result.unavailable:
            runtime_result["policy_witness"]["unavailable"] = pw_result.unavailable
        if pw_result.equal is not True and pw_result.first_divergence:
            runtime_result["policy_witness"]["first_divergence"] = pw_result.first_divergence
    except Exception as exc:
        runtime_result["policy_witness"] = {"eq": None, "error": str(exc)}

    # Oracle 2: Sandboxed binary (for file operations)
    #   Requires SBPL files on disk - create temp files from SBPL text.
    sb_result: dict[str, Any] | None = None
    try:
        import tempfile
        import shutil

        # Create temp directory for SBPL files and workspace
        workspace = Path(tempfile.mkdtemp(prefix="pawl_dual_oracle_"))
        try:
            source_sbpl_path = workspace / "source.sb"
            roundtrip_sbpl_path = workspace / "roundtrip.sb"
            # Pass raw SBPL with (param ...) references intact; the runner
            # uses sandbox_init_with_parameters to resolve them natively,
            # which also resolves params inside (import ...) targets.
            # For the roundtrip, param_reconstruct may have converted literal
            # paths back to (param ...) — same bindings resolve them.
            source_sbpl_path.write_text(source_sbpl)
            roundtrip_sbpl_path.write_text(reversed_sbpl)

            sb_result = run_binary_oracle(
                source_sbpl_path,
                roundtrip_sbpl_path,
                ir_a,
                workspace=workspace / "probes",
                param_bindings=param_bindings,
            )
            runtime_result["sandboxed_binary"] = {
                "eq": sb_result.get("eq"),
                "probes": sb_result.get("probes", 0),
            }
            if sb_result.get("skipped"):
                runtime_result["sandboxed_binary"]["skipped"] = sb_result["skipped"]
            if sb_result.get("mismatches"):
                runtime_result["sandboxed_binary"]["mismatches"] = sb_result["mismatches"][:5]
        finally:
            shutil.rmtree(workspace, ignore_errors=True)
    except Exception as exc:
        reason = f"sandboxed_binary_error: {exc}"
        runtime_result["sandboxed_binary"] = {
            "eq": None,
            "error": str(exc),
            "unavailable": [
                ProbeUnavailable(
                    category=EXECUTION_ERROR,
                    reason=reason,
                    layer="execution",
                ).to_json()
            ],
        }

    # Combine results: both oracles must pass (or be skipped/unavailable)
    pw_eq = runtime_result["policy_witness"].get("eq") if runtime_result["policy_witness"] else None
    sb_eq = runtime_result["sandboxed_binary"].get("eq") if runtime_result["sandboxed_binary"] else None
    sb_skipped = runtime_result["sandboxed_binary"].get("skipped") if runtime_result["sandboxed_binary"] else None

    if pw_eq is True and (sb_eq is True or sb_skipped):
        runtime_result["combined"] = True
        checks["runtime_eq"] = True
        distinguishing["runtime_eq"] = runtime_result
    elif pw_eq is False or sb_eq is False:
        runtime_result["combined"] = False
        checks["runtime_eq"] = False
        distinguishing["runtime_eq"] = runtime_result
    elif pw_eq is None and sb_eq is None:
        runtime_result["combined"] = None
        checks["runtime_eq"] = None
    else:
        # One oracle passed/skipped, other unavailable - partial success
        runtime_result["combined"] = pw_eq if pw_eq is not None else sb_eq
        checks["runtime_eq"] = runtime_result["combined"]
        if runtime_result["combined"] is not True:
            distinguishing["runtime_eq"] = runtime_result

    if (
        isinstance(runtime_result.get("policy_witness"), dict)
        and runtime_result["policy_witness"].get("unavailable")
    ):
        distinguishing["runtime_eq"] = runtime_result

    return ProfileFivePointResult(
        checks=checks,
        distinguishing=distinguishing,
        ir_a=ir_a,
        ir_b=ir_b,
        blob_hash_a=ir_a["blob_sha256"],
        blob_hash_b=ir_b["blob_sha256"],
    )


_PROMOTION_EQUIVALENTS: dict[str, str] = {
    # 3c-B: mach promotion.
    "mach-lookup": "mach-kernel-endpoint",
    # 3c-E: file/ipc child -> wildcard promotions.
    "file-read-data": "file-read*",
    "ipc-posix-shm-read-data": "ipc-posix-shm*",
    "ipc-posix-shm-write-data": "ipc-posix-shm*",
    # 3c-C partial: iokit user-client widening.
    "iokit-open-user-client": "iokit-open*",
}


def _slot_lookup(ir: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """Build op_name -> slot record lookup from a profile IR."""
    slots = ir.get("op_table", {}).get("slots", [])
    return {
        str(s["op_name"]): s
        for s in slots
        if isinstance(s, dict) and "op_name" in s
    }


def _detect_promotion_remap(
    ir_a: dict[str, Any],
    ir_b: dict[str, Any],
    source_explicit_ops: set[str],
) -> dict[str, str]:
    """Detect source child-ops that should compare as promoted parents.

    The compiler may promote a source child operation (e.g.,
    ``file-read-data``) to its wildcard parent (``file-read*``), routing
    the child's predicates through the parent's decision node.  When this
    happens, the reversed SBPL uses the parent operation name and the text
    comparison needs to remap the source to match.

    Detection conditions:
    - Source explicitly declares the child, not the parent.
    - The original IR routes the parent through an active node (class is
      not default/explicit/None — typically ``unknown``).
    - The roundtrip IR has the parent actively wired (``explicit`` or
      ``shared_entrypoint`` — both indicate an active decision node).
    - Node offsets do not need to match across blobs.
    - If the roundtrip child is still ``explicit``, it must share the
      parent node; otherwise it represents an independent declaration.
    """
    slots_a = _slot_lookup(ir_a)
    slots_b = _slot_lookup(ir_b)
    remap: dict[str, str] = {}

    # Classes that indicate the roundtrip parent is actively wired.
    _active_parent_classes = {"explicit", "shared_entrypoint"}

    for child, parent in _PROMOTION_EQUIVALENTS.items():
        if child not in source_explicit_ops or parent in source_explicit_ops:
            continue

        sa_parent = slots_a.get(parent)
        sb_parent = slots_b.get(parent)
        if not sa_parent or not sb_parent:
            continue
        if sa_parent.get("assignment_class") in (None, "default", "explicit"):
            continue
        if sb_parent.get("assignment_class") not in _active_parent_classes:
            continue
        # Note: node_offset equality is NOT required.  Original and
        # roundtrip blobs have different layouts, so the same decision
        # content lives at different offsets.  Semantic content is
        # validated by decision_path_eq, not by offset identity.

        # When the roundtrip child shares the same node as the parent,
        # its explicit classification comes from wildcard coverage, not
        # from an independent declaration.  Allow the remap.
        sb_child = slots_b.get(child)
        if sb_child and sb_child.get("assignment_class") == "explicit":
            child_offset = sb_child.get("node_offset")
            parent_offset = sb_parent.get("node_offset")
            if child_offset != parent_offset:
                continue  # child has its own node — not promoted

        remap[child] = parent

    return remap


def _remap_sbpl_operations(normalized: str, remap: dict[str, str]) -> str:
    """Rewrite allow/deny operation names in normalized SBPL using *remap*."""
    if not remap:
        return normalized
    rewritten: list[str] = []
    for line in normalized.splitlines():
        stripped = line.strip()
        if not stripped.startswith("(") or not stripped.endswith(")"):
            rewritten.append(line)
            continue
        children = _extract_sexp_children(stripped[1:-1])
        if len(children) < 2 or children[0] not in {"allow", "deny"}:
            rewritten.append(line)
            continue
        op = children[1]
        replacement = remap.get(op)
        if replacement:
            children[1] = replacement
            rewritten.append("(" + " ".join(children) + ")")
            continue
        rewritten.append(line)
    return "\n".join(rewritten)


def _normalized_slots_for_compare(
    ir: dict[str, Any],
    *,
    promotion_remap: dict[str, str] | None,
    source_side: bool,
) -> dict[str, dict[str, Any]]:
    """Return op_name->slot map normalized for promotion-aware comparison."""
    slots = {k: dict(v) for k, v in _slot_lookup(ir).items()}
    if not promotion_remap:
        return slots

    for child, parent in promotion_remap.items():
        child_slot = slots.get(child)
        parent_slot = slots.get(parent)
        if source_side:
            if child_slot is None:
                continue
            merged = dict(parent_slot) if parent_slot is not None else dict(child_slot)
            merged["op_name"] = parent
            merged["assignment_class"] = "explicit"
            child_class = child_slot.get("assignment_class")
            parent_class = parent_slot.get("assignment_class") if parent_slot else None
            if child_class != "default":
                merged["node_offset"] = child_slot.get("node_offset")
            elif parent_slot is not None and parent_class != "default":
                merged["node_offset"] = parent_slot.get("node_offset")
            else:
                merged["node_offset"] = child_slot.get("node_offset")
            slots[parent] = merged
            slots.pop(child, None)
            continue

        # Roundtrip side: pop the child when its parent is actively wired
        # (explicit or shared_entrypoint).  The child's explicitness in
        # the roundtrip comes from the parent's wildcard coverage, not
        # from an independent source declaration.
        _active = {"explicit", "shared_entrypoint"}
        if (
            child_slot is not None
            and parent_slot is not None
            and parent_slot.get("assignment_class") in _active
        ):
            slots.pop(child, None)

    return slots


def _normalize_ir_for_graph_compare(
    ir: dict[str, Any],
    *,
    promotion_remap: dict[str, str] | None,
    source_side: bool,
    exclude_ops: frozenset[str] | None = None,
) -> dict[str, Any]:
    """Return an IR view filtered to explicit ops for graph comparison.

    Non-explicit ops (default, implicit_*, contextual_implicit_*,
    bare_opposite_terminal, shared_entrypoint, unknown) are excluded
    because they represent compiler-managed operations whose decision
    graphs diverge structurally between original and roundtrip blobs
    without semantic significance.

    *exclude_ops*, when provided, is an additional set of op names to
    exclude.  This is used to align exclusion sets across original and
    roundtrip IRs when an op has different assignment classes in each
    (e.g. shared_entrypoint in original but explicit in roundtrip).
    """
    raw_slots = _normalized_slots_for_compare(
        ir,
        promotion_remap=promotion_remap,
        source_side=source_side,
    )
    # Keep only explicitly-declared operations for graph comparison.
    filtered = {
        name: slot for name, slot in raw_slots.items()
        if slot.get("assignment_class") not in _NON_EXPLICIT_ASSIGNMENT_CLASSES
        and (exclude_ops is None or name not in exclude_ops)
    }
    slots = sorted(filtered.values(), key=lambda s: int(s.get("slot", 0)))
    out = dict(ir)
    op_table = dict(ir.get("op_table", {}))
    op_table["slots"] = slots
    out["op_table"] = op_table
    return out


def _graph_exclude_ops(ir_a: dict, ir_b: dict) -> frozenset[str]:
    """Compute ops to exclude from graph comparison across both IRs.

    An op should be excluded if it is non-explicit in EITHER IR.
    This prevents false divergences from cross-IR classification
    asymmetry (e.g. shared_entrypoint in original, explicit in roundtrip).
    """
    exclude: set[str] = set()
    for ir in (ir_a, ir_b):
        for s in ir.get("op_table", {}).get("slots", []):
            if s.get("assignment_class") in _NON_EXPLICIT_ASSIGNMENT_CLASSES:
                exclude.add(s["op_name"])
    return frozenset(exclude)


def _should_promote_graph_eq_for_import_context(
    *,
    has_imports: bool,
    graph_ok: bool,
    canonical_eq: bool,
    semantic_eq: bool,
) -> bool:
    """Decide whether import context justifies promoting `graph_eq`.

    `graph_eq` is an IR-level surface. When import context is present, the
    compiled original can contain import-inherited predicates that the
    roundtrip graph does not reproduce structurally. We only promote the
    surface after the stronger normalize-layer checks already proved both
    structural and leaf-predicate equivalence.
    """
    return has_imports and not graph_ok and canonical_eq and semantic_eq


def _maybe_recompute_sbpl_eq_with_import_filter(
    *,
    current_eq: bool,
    normalized_source: str,
    normalized_reversed: str,
    implicit_ops: set[str],
    source_sbpl: str,
    search_paths: list[Path] | None,
    source_path: Path | None,
) -> tuple[bool, bool]:
    """Retry `sbpl_eq` with imported operations filtered out.

    Returns `(sbpl_eq, import_filtered)` where `import_filtered` is true only
    when the import-aware retry resolved the mismatch.
    """
    if current_eq or search_paths is None:
        return current_eq, False

    from pawl.normalize.baseline.imports import compute_import_expansion
    from pawl.normalize.operations import get_equivalent_operations

    import_expansion = compute_import_expansion(
        source_sbpl,
        search_paths=search_paths,
        source_path=source_path,
    )
    if not import_expansion.imported_operations:
        return current_eq, False

    imported_equivs: set[str] = set()
    for imported_op in import_expansion.imported_operations:
        imported_equivs.update(get_equivalent_operations(imported_op))

    filtered_source = _filter_sbpl_by_ir(
        normalized_source, implicit_ops | imported_equivs,
    )
    filtered_reversed = _filter_sbpl_by_ir(
        normalized_reversed, implicit_ops | imported_equivs,
    )
    recomputed_eq = filtered_source == filtered_reversed
    return recomputed_eq, recomputed_eq


def _normalize_sbpl(text: str) -> str:
    """Normalize SBPL to canonical form for order-insensitive comparison.

    - Strip comments (lines starting with ``;``)
    - Normalize ``(version 1)`` header
    - Collapse multi-line rules to single canonical lines
    - Lowercase quoted string literals (compiler lowercases them)
    - Sort ``require-all`` / ``require-any`` children (compiler may reorder)
    - Sort top-level rules alphabetically
    """
    # Strip comments and blank lines.
    lines = []
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped or stripped.startswith(";"):
            continue
        lines.append(stripped)

    # Join into one string, then split into top-level S-expressions.
    joined = " ".join(lines)

    # Extract top-level parenthesized forms.
    rules: list[str] = []
    depth = 0
    start = None
    for i, ch in enumerate(joined):
        if ch == "(":
            if depth == 0:
                start = i
            depth += 1
        elif ch == ")":
            depth -= 1
            if depth == 0 and start is not None:
                rule = joined[start : i + 1]
                # Collapse internal whitespace to single spaces.
                rule = " ".join(rule.split())
                rules.append(rule)
                start = None

    # Normalize each rule.
    normalized: list[str] = []
    for rule in rules:
        # Expand with-filter to equivalent allow/deny forms:
        # (with-filter (pred) (allow op)) → (allow op (pred))
        rule = _expand_with_filter(rule)
        # Lowercase quoted strings: "Foo" → "foo"
        rule = _lowercase_quoted_strings(rule)
        # Strip trailing slashes from subpath predicates so that
        # (subpath "/X/") and (subpath "/X") compare as identical.
        rule = _normalize_subpath_trailing_slash(rule)
        # Strip (entitlement-value #t) — the compiler always drops it
        # because #t is the default.  Both forms compile identically.
        rule = _strip_default_entitlement_value(rule)
        # Sort children of require-all / require-any.
        rule = _sort_require_children(rule)
        # Sort filter predicates within allow/deny rules:
        # (allow op (pred-b ...) (pred-a ...)) → (allow op (pred-a ...) (pred-b ...))
        rule = _sort_rule_predicates(rule)
        normalized.append(rule)

    # Separate header ((version 1)) from body rules.
    header = []
    body = []
    for rule in normalized:
        if rule.startswith("(version"):
            header.append("(version 1)")
        else:
            body.append(rule)

    # Sort body rules alphabetically for order-insensitive comparison.
    body.sort()

    return "\n".join(header + body)


def _expand_with_filter(rule: str) -> str:
    """Expand ``(with-filter (pred) (allow op ...))`` to ``(allow op ... (pred))``.

    The ``with-filter`` directive is syntactic sugar that attaches a predicate
    to the inner rule.  The compiler emits the predicate as part of the rule's
    filter chain.  Converting to the expanded form enables comparison with
    reversed output which uses the direct form.
    """
    prefix = "(with-filter "
    if not rule.startswith(prefix):
        return rule
    # Parse: (with-filter <pred-sexp> <body-sexp>)
    inner = rule[len(prefix):-1]  # strip outer parens
    children = _extract_sexp_children(inner)
    if len(children) != 2:
        return rule  # unexpected structure, don't transform
    pred, body = children
    # body should be (allow/deny op ...)
    body_children = _extract_sexp_children(body[1:-1])
    if len(body_children) < 2:
        return rule
    action = body_children[0]
    if action not in ("allow", "deny"):
        return rule
    op = body_children[1]
    existing_preds = body_children[2:]
    return "(" + " ".join([action, op] + existing_preds + [pred]) + ")"


import re as _re

_QUOTED_STRING_RE = _re.compile(r'"([^"]*)"')
_SUBPATH_TRAILING_SLASH_RE = _re.compile(r'\(subpath "([^"]+)/"\)')


def _normalize_subpath_trailing_slash(rule: str) -> str:
    """Strip trailing slashes from ``(subpath "…/")`` predicates.

    The compiler may generate both ``(subpath "/X/")`` and ``(subpath "/X")``
    for the same path.  Normalizing to the slash-free form allows the dedup
    step in ``_sort_require_children`` to collapse the duplicates.
    """
    def _strip(m: _re.Match) -> str:
        path = m.group(1)
        return f'(subpath "{path}")'
    return _SUBPATH_TRAILING_SLASH_RE.sub(_strip, rule)


_ENTITLEMENT_VALUE_TRUE_RE = _re.compile(r'\s*\(entitlement-value #t\)')


def _strip_default_entitlement_value(rule: str) -> str:
    """Strip ``(entitlement-value #t)`` from ``require-entitlement`` predicates.

    ``#t`` is the default entitlement value check.  The compiler always drops
    it, so ``(require-entitlement "key" (entitlement-value #t))`` and
    ``(require-entitlement "key")`` compile to the same blob.
    """
    return _ENTITLEMENT_VALUE_TRUE_RE.sub("", rule)


def _lowercase_quoted_strings(rule: str) -> str:
    """Lowercase contents of double-quoted strings in an SBPL rule.

    The compiler lowercases string literals, so we normalize to match.
    Regex patterns (prefixed with ``#``) are left alone.
    """
    def _lower(m: _re.Match) -> str:
        # Don't lowercase if preceded by # (regex literal).
        start = m.start()
        if start > 0 and rule[start - 1] == "#":
            return m.group(0)
        return f'"{m.group(1).lower()}"'
    return _QUOTED_STRING_RE.sub(_lower, rule)


def _sort_require_children(rule: str) -> str:
    """Sort children of ``(require-all ...)`` and ``(require-any ...)`` forms.

    The compiler may reorder filter predicates within conjunctions/disjunctions.
    Sorting them produces a canonical form for comparison.
    """
    # Find (require-all ...) or (require-any ...) and sort their children.
    result = []
    i = 0
    while i < len(rule):
        # Look for (require-all or (require-any
        for keyword in ("(require-all ", "(require-any "):
            if rule[i:].startswith(keyword):
                # Find the matching close paren.
                depth = 0
                start = i
                for j in range(i, len(rule)):
                    if rule[j] == "(":
                        depth += 1
                    elif rule[j] == ")":
                        depth -= 1
                        if depth == 0:
                            # Extract the full form.
                            form = rule[start:j + 1]
                            inner = form[len(keyword):-1]
                            # Parse children (sub-expressions).
                            children = _extract_sexp_children(inner)
                            # Recursively normalize children, then sort.
                            children = [_sort_require_children(c) for c in children]
                            # Flatten nested same-type operators:
                            # (require-any (require-any A B) C) → (require-any A B C)
                            flattened: list[str] = []
                            for child in children:
                                if child.startswith(keyword.rstrip()):
                                    inner_children = _extract_sexp_children(
                                        child[len(keyword):-1],
                                    )
                                    flattened.extend(inner_children)
                                else:
                                    flattened.append(child)
                            children = flattened
                            children.sort()
                            # Dedup children (e.g., after trailing-slash
                            # normalization produces identical atoms).
                            children = list(dict.fromkeys(children))
                            # Unwrap singleton: (require-any X) → X
                            if len(children) == 1:
                                result.append(children[0])
                            else:
                                result.append(keyword + " ".join(children) + ")")
                            i = j + 1
                            break
                else:
                    result.append(rule[i])
                    i += 1
                break
        else:
            result.append(rule[i])
            i += 1
    return "".join(result)


def _extract_sexp_children(text: str) -> list[str]:
    """Extract top-level S-expression children from a string."""
    children: list[str] = []
    depth = 0
    start = None
    i = 0
    text = text.strip()
    while i < len(text):
        if text[i] == "(":
            if depth == 0:
                start = i
            depth += 1
        elif text[i] == ")":
            depth -= 1
            if depth == 0 and start is not None:
                children.append(text[start:i + 1])
                start = None
        elif depth == 0 and not text[i].isspace():
            # Bare token (e.g. a symbol).
            end = i
            while end < len(text) and not text[end].isspace() and text[end] not in "()":
                end += 1
            children.append(text[i:end])
            i = end
            continue
        i += 1
    return children


def _sort_rule_predicates(rule: str) -> str:
    """Sort filter predicates within an allow/deny rule.

    Given ``(allow op (pred-b ...) (pred-a ...))``, reorder the predicates
    (children after the op name) alphabetically.  The ``apply-message-filter``
    predicate is always placed last since it's structurally special.
    """
    children = _extract_sexp_children(rule[1:-1])  # strip outer parens
    if len(children) < 3:
        return rule  # (action op) or (action op pred) — nothing to sort
    action = children[0]
    if action not in ("allow", "deny"):
        return rule
    op = children[1]
    predicates = children[2:]
    # Keep apply-message-filter last.
    mf = [p for p in predicates if p.startswith("(apply-message-filter")]
    rest = [p for p in predicates if not p.startswith("(apply-message-filter")]
    rest.sort()
    return "(" + " ".join([action, op] + rest + mf) + ")"


def _filter_sbpl_by_ir(normalized: str, implicit_ops: set[str]) -> str:
    """Remove rules targeting ops in *implicit_ops* from normalized SBPL.

    Each line of *normalized* is a single top-level S-expression produced by
    ``_normalize_sbpl``.  The op name is the second token inside the outermost
    parens, e.g. ``(allow mach-task-name)`` → ``mach-task-name``.

    Rules whose op name matches *implicit_ops* are dropped.  Directives that
    don't survive the compile→reverse roundtrip (``import``,
    ``require-entitlement``, ``entitlement-value``) are also dropped so that
    only actionable allow/deny rules remain for comparison.
    """
    # Directives that exist in source SBPL but cannot be reconstructed from
    # a compiled blob.  Their *effects* (promoted ops, entitlement checks)
    # appear as explicit rules in the reversed output.
    _NONROUNDTRIP_DIRECTIVES = {
        "import", "require-entitlement", "entitlement-value",
        "sandbox-array-entitlement", "let", "cond",
        "define", "when", "unless",
    }
    # SBPL forms that appear as top-level actions or directives.
    # Anything whose first token is not in this set is a bare function
    # call (e.g. ``(corefoundation)``), which cannot survive roundtrip.
    _KNOWN_TOPLEVEL = _NONROUNDTRIP_DIRECTIVES | {
        "version", "allow", "deny", "with-filter",
    }
    kept: list[str] = []
    for line in normalized.splitlines():
        tokens = line.strip("() ").split()
        if len(tokens) < 2:
            # Single-token forms: either ``(version 1)`` with unusual
            # spacing or bare function calls like ``(corefoundation)``.
            if tokens and tokens[0] not in _KNOWN_TOPLEVEL:
                continue  # bare function call — not roundtrippable
            kept.append(line)
            continue
        directive = tokens[0]
        if directive == "version":
            kept.append(line)
            continue
        if directive in _NONROUNDTRIP_DIRECTIVES:
            continue
        # Bare function calls whose first token isn't a known directive.
        if directive not in _KNOWN_TOPLEVEL:
            continue
        # Strip bare default-action declarations — the reversed output always
        # emits them but source may omit them.
        if directive in ("allow", "deny") and tokens[1] == "default":
            continue
        # For allow/deny rules, check if the op is classified as implicit.
        op_name = tokens[1]
        if implicit_ops and op_name in implicit_ops:
            continue
        kept.append(line)
    return "\n".join(kept)


def _compare_op_tables(
    ir_a: dict,
    ir_b: dict,
    *,
    promotion_remap: dict[str, str] | None = None,
) -> list[dict]:
    """Compare op-table slots, returning diffs for non-default/non-implicit slots.

    Classification tolerance: certain class transitions are tolerated because
    they represent equivalent semantic states:

    - default ↔ implicit_* ↔ contextual_implicit: All represent "compiler-managed"
      operations that aren't explicitly declared in source SBPL.
    - unknown ↔ any implicit class: "unknown" means the classifier couldn't match
      a known pattern, but the operation may still be semantically implicit.
    - bare_opposite_terminal: Compiler-promoted opposite-action terminals.

    We only flag mismatches between "explicit" and non-explicit classes, as those
    represent true semantic differences (source-declared vs compiler-managed).
    """
    slots_a = _normalized_slots_for_compare(
        ir_a, promotion_remap=promotion_remap, source_side=True
    )
    slots_b = _normalized_slots_for_compare(
        ir_b, promotion_remap=promotion_remap, source_side=False
    )

    diffs = []
    # Classes that represent "compiler-managed" operations (not source-declared).
    for op_name in sorted(set(slots_a) | set(slots_b)):
        sa = slots_a.get(op_name)
        sb = slots_b.get(op_name)
        if sa is None or sb is None:
            diffs.append({"op": op_name, "a": sa, "b": sb, "reason": "missing"})
            continue

        class_a = sa["assignment_class"]
        class_b = sb["assignment_class"]

        # Skip if both are in the implicit family - these are tolerable transitions.
        if (
            class_a in _NON_EXPLICIT_ASSIGNMENT_CLASSES
            and class_b in _NON_EXPLICIT_ASSIGNMENT_CLASSES
        ):
            continue

        # Wildcard-coverage tolerance: shared_entrypoint means the child
        # shares a decision-graph entry point with a wildcard parent.
        # The child IS semantically explicit, just routed differently by
        # each compiler's wildcard expansion.
        if {class_a, class_b} == {"explicit", "shared_entrypoint"}:
            continue

        # Cross-IR classification asymmetry: an op classified as ``default``
        # in the original (compiler-managed, no source declaration) may
        # become ``explicit`` in the roundtrip because the reverser renders
        # import-inherited rules explicitly. This is expected and not a
        # semantic difference.
        if {class_a, class_b} == {"default", "explicit"}:
            continue

        # Flag explicit ↔ non-explicit transitions as real mismatches.
        if class_a != class_b:
            diffs.append({
                "op": op_name,
                "class_a": class_a,
                "class_b": class_b,
                "reason": "class_mismatch",
            })
        # Note: node_offset differences between explicit ops are expected
        # in roundtrip blobs (different compilation produces different
        # graph layout). Semantic content is captured by decision_path_eq
        # (graph_eq). Do not flag offset_mismatch as a failure.

    return diffs


def _compare_explicit_graphs(ir_a: dict, ir_b: dict) -> tuple[bool, list[str]]:
    """Compare reachable subgraphs for explicit and unknown slots.

    For each slot classified as ``explicit`` or ``unknown`` in both IRs
    (at the same classification), compare the reachable node structure
    (kind + action at each step).

    Returns ``(all_match, list_of_diverging_op_names)``.
    """
    slots_a = {s["op_name"]: s for s in ir_a["op_table"]["slots"]}
    slots_b = {s["op_name"]: s for s in ir_b["op_table"]["slots"]}
    nodes_a = {n["offset"]: n for n in ir_a["node_graph"]["nodes"]}
    nodes_b = {n["offset"]: n for n in ir_b["node_graph"]["nodes"]}

    compare_classes = {"explicit", "unknown"}
    diverging: list[str] = []

    for op_name in sorted(set(slots_a) & set(slots_b)):
        sa = slots_a[op_name]
        sb = slots_b[op_name]
        if sa["assignment_class"] not in compare_classes:
            continue
        if sa["assignment_class"] != sb["assignment_class"]:
            continue

        sig_a = _subgraph_signature(sa["node_offset"], nodes_a)
        sig_b = _subgraph_signature(sb["node_offset"], nodes_b)
        if sig_a != sig_b:
            diverging.append(op_name)

    return (len(diverging) == 0, diverging)


def _subgraph_signature(
    start_offset: int,
    nodes: dict[int, dict],
    _visited: set[int] | None = None,
) -> str:
    """Compute a canonical signature of the reachable subgraph from start_offset.

    NOTE: This is the STRUCTURAL signature. For semantic comparison that
    tolerates filter reordering, use graph_signature from semantic_engine.
    """
    if _visited is None:
        _visited = set()
    if start_offset in _visited:
        return "cycle"
    _visited.add(start_offset)

    node = nodes.get(start_offset)
    if node is None:
        return "missing"

    if node["kind"] == "terminal":
        return f"T:{node.get('action')}"
    elif node["kind"] == "non_terminal":
        m = node.get("match_offset")
        u = node.get("unmatch_offset")
        m_sig = _subgraph_signature(m, nodes, _visited) if m is not None else "null"
        u_sig = _subgraph_signature(u, nodes, _visited) if u is not None else "null"
        fid = node.get("filter_id")
        return f"NT(f={fid},m={m_sig},u={u_sig})"
    else:
        return f"UNK:{start_offset}"


# ---------------------------------------------------------------------------
# Semantic signature functions — imported from shared engine
# ---------------------------------------------------------------------------
from pawl.reverse.analysis.semantic_engine import (
    PredicateId,
    compare_graphs_semantic as _compare_explicit_graphs_semantic,
    graph_signature as _semantic_graph_signature,
    predicate_id as _semantic_predicate_id,
)