scripts.benchmark_prolog_codeintel

Reference › scripts › benchmark_prolog_codeintel

Benchmarks recursive code-intelligence queries in Prolog against the same queries in plain Python. It is a large script (about 1,400 lines) that a human runs to decide where recursive queries should execute. tests/scripts/test_benchmark_prolog_codeintel.py imports it, and it reuses the shipped codebase tools from codebase_tools.

Invocation: --ast-json (the serialized AST to start from), --runs (default 5), --workdir (default a temporary directory), --report (where the Markdown report goes) and --skip-cli. It exits 2 if the AST file is missing.

The stages are printed as they run. collect_environment records the host, generate_facts turns the AST JSON into a Prolog fact file, and measure_consult_overhead times loading it. dump_edges extracts the edge relations, and build_graph builds an in-memory adjacency structure (Graph) for the Python side, which implements closure, bfs_path, a depth-first path search and cycle enumeration with canonicalize_cycle. select_workloads picks real modules and symbols from the graph. Four workloads follow: run_w1 (call impact), run_w2 (dependencies in a direction), run_w3 (cycles) and run_w4 (dependency path). Each times a Prolog call per query (time_psub, psub_query), the warm in-process Janus path when available (janus_sanity_check), the Python version (time_warm), the shipped wrapper (time_shipped_wrapper) and the CLI (time_cli), and checks that the answers agree. The result is a WorkloadOutcome with summary and detail tables and a verdict line.

At a glance

Class diagram

Dependency diagram

API

Classes

QueryTiming (dataclass)

Decorators: dataclass · Line 76

WorkloadOutcome (dataclass)

Decorators: dataclass · Line 88

Graph (dataclass)

Decorators: dataclass · Line 248

Workloads (dataclass)

Decorators: dataclass · Line 440

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/benchmark_prolog_codeintel.py

#!/usr/bin/env python3
"""benchmark_prolog_codeintel.py — Prolog vs Python recursive code-intelligence benchmark.

Answers one question with numbers: is the Prolog codebase-intelligence layer
(``src/orchestrata_ultimate/tools/codebase_db.pl``) better/faster than plain
Python recursion for recursive code-intelligence queries (reverse-dependency
closure, transitive module deps, cycle enumeration, dependency paths)?

Four drivers are benchmarked over the IDENTICAL edge set (dumped once from
Prolog so resolver differences never pollute the recursion comparison):

  P-sub    — Prolog via subprocess swipl (cold: pays full consult every call).
  P-janus  — Prolog via in-process patched janus_swi (warm: consult once).
  Py-mem   — pure-Python recursion (BFS/DFS + memo) over the dumped edges.
  Py-cli   — end-to-end ``serialize_codebase_ast.py`` timings (context only).

Correctness parity (P-sub == P-janus == Py-mem, as frozensets) is asserted
BEFORE any timing number is trusted — see ``run_parity_and_benchmark``.

Usage:
    uv run python scripts/benchmark_prolog_codeintel.py \\
        [--ast-json plans/ast-ultimate.json] [--runs 5] [--workdir DIR] \\
        [--report reports/prolog-codeintel-benchmark.md] [--skip-cli]
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import statistics
import subprocess
import sys
import tempfile
import time
from collections import Counter, defaultdict, deque
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, cast

REPO_ROOT = Path(__file__).resolve().parent.parent
RULES_PL = REPO_ROOT / "src" / "orchestrata_ultimate" / "tools" / "codebase_db.pl"
DEFAULT_AST_JSON = REPO_ROOT / "plans" / "ast-ultimate.json"
DEFAULT_REPORT = REPO_ROOT / "reports" / "prolog-codeintel-benchmark.md"
NO_LIMIT = 1_000_000  # "limit" passed to tool_* predicates so nothing truncates
W4_MIN_HOPS = 3
W1_W2_COUNT = 10
SWIPL_TIMEOUT_S = 300


# =====================================================================
# Small utilities
# =====================================================================


def quote_atom(value: str) -> str:
    """Render *value* as a single-quoted Prolog atom literal (safe: no
    goal-string substitution — the value is embedded directly by position,
    never by search/replace, so it cannot corrupt other placeholders)."""
    escaped = value.replace("\\", "\\\\").replace("'", "\\'")
    return f"'{escaped}'"


def stats_of(times_s: list[float]) -> dict[str, float]:
    return {
        "median_ms": statistics.median(times_s) * 1000,
        "min_ms": min(times_s) * 1000,
        "max_ms": max(times_s) * 1000,
        "n": len(times_s),
    }


@dataclass
class QueryTiming:
    label: str
    result: Any
    times_s: list[float]
    first_call_s: float | None = None  # warm modes only: the discarded warmup

    @property
    def stats(self) -> dict[str, float]:
        return stats_of(self.times_s)


@dataclass
class WorkloadOutcome:
    name: str
    parity_ok: bool
    parity_notes: list[str] = field(default_factory=list)
    timings: dict[str, list[QueryTiming]] = field(default_factory=dict)


# =====================================================================
# Phase 0 — environment
# =====================================================================


def collect_environment() -> dict[str, str]:
    swipl_version = subprocess.run(
        ["swipl", "--version"], capture_output=True, text=True, timeout=30
    ).stdout.strip()
    gil_enabled = getattr(sys, "_is_gil_enabled", None)
    cpu_model = ""
    try:
        with open("/proc/cpuinfo", encoding="utf-8") as f:
            for line in f:
                if line.startswith("model name"):
                    cpu_model = line.split(":", 1)[1].strip()
                    break
    except OSError:
        pass
    return {
        "swipl_version": swipl_version,
        "python_version": sys.version.replace("\n", " "),
        "gil_enabled": str(gil_enabled()) if gil_enabled is not None else "unknown",
        "platform": platform.platform(),
        "cpu_model": cpu_model or platform.processor() or "unknown",
        "cpu_count": str(os.cpu_count()),
    }


# =====================================================================
# Phase 1 — facts generation (P-sub, cold, once)
# =====================================================================


def generate_facts(ast_json: Path, workdir: Path) -> tuple[Path, list[Any], float]:
    """Run load_codebase/2 + save_facts/1 via standalone swipl. Returns
    (facts_path, warnings, elapsed_seconds)."""
    facts_path = workdir / "facts.pl"
    query_path = workdir / "gen_facts.pl"
    warn_marker = "WARNCOUNT"
    query_path.write_text(
        f":- consult({quote_atom(str(RULES_PL))}).\n"
        "run :-\n"
        f"    load_codebase({quote_atom(str(ast_json))}, R),\n"
        "    get_dict(warnings, R, W),\n"
        f"    save_facts({quote_atom(str(facts_path))}),\n"
        f'    format("{warn_marker} ~q~n", [W]).\n',
        encoding="utf-8",
    )
    t0 = time.perf_counter()
    proc = subprocess.run(
        ["swipl", "-q", "-g", "run", "-g", "halt", str(query_path)],
        capture_output=True,
        text=True,
        timeout=SWIPL_TIMEOUT_S,
    )
    elapsed = time.perf_counter() - t0
    warnings: list[Any] = []
    for line in proc.stdout.splitlines():
        if line.startswith(warn_marker):
            raw = line[len(warn_marker) :].strip()
            warnings.append(raw)
    if not facts_path.exists():
        raise RuntimeError(
            f"facts generation failed: rc={proc.returncode}\n"
            f"stdout={proc.stdout[-2000:]}\nstderr={proc.stderr[-2000:]}"
        )
    return facts_path, warnings, elapsed


def measure_consult_overhead(facts_path: Path, runs: int) -> dict[str, float]:
    """P-sub consult overhead alone: consult rules+facts, goal = true, no query."""
    times: list[float] = []
    for _ in range(runs):
        query_path_fd, query_path_name = tempfile.mkstemp(
            suffix=".pl", dir=facts_path.parent
        )
        os.close(query_path_fd)
        query_path = Path(query_path_name)
        try:
            query_path.write_text(
                f":- consult({quote_atom(str(RULES_PL))}).\n"
                f":- consult({quote_atom(str(facts_path))}).\n"
                "run :- true.\n",
                encoding="utf-8",
            )
            t0 = time.perf_counter()
            subprocess.run(
                ["swipl", "-q", "-g", "run", "-g", "halt", str(query_path)],
                capture_output=True,
                text=True,
                timeout=SWIPL_TIMEOUT_S,
            )
            times.append(time.perf_counter() - t0)
        finally:
            query_path.unlink(missing_ok=True)
    return stats_of(times)


# =====================================================================
# Phase 2 — edge dump (Prolog -> JSON lines) + Py-mem graph build
# =====================================================================


def dump_edges(rules_pl: Path, facts_path: Path, workdir: Path) -> tuple[Path, float]:
    """Dump calls_resolved/3, inherits_resolved/2, dep_raw/2, module/3, symbol/7
    as JSON lines, via one swipl subprocess. Returns (edges_path, elapsed_s)."""
    edges_path = workdir / "edges.jsonl"
    query_path = workdir / "dump_edges.pl"
    query_path.write_text(
        ":- use_module(library(http/json)).\n"
        f":- consult({quote_atom(str(rules_pl))}).\n"
        f":- consult({quote_atom(str(facts_path))}).\n"
        "dump_line(Out, D) :- json_write(Out, D, [width(0)]), nl(Out).\n"
        "run :-\n"
        "    setup_call_cleanup(\n"
        f"        open({quote_atom(str(edges_path))}, write, Out, [encoding(utf8)]),\n"
        "        ( forall(calls_resolved(A,B,L),\n"
        "                 dump_line(Out, json{rel:calls_resolved, a:A, b:B, l:L})),\n"
        "          forall(inherits_resolved(A,B),\n"
        "                 dump_line(Out, json{rel:inherits_resolved, a:A, b:B})),\n"
        "          forall(dep_raw(A,B),\n"
        "                 dump_line(Out, json{rel:dep_raw, a:A, b:B})),\n"
        "          forall(module(M,F,L),\n"
        "                 dump_line(Out, json{rel:module, a:M, f:F, l:L})),\n"
        "          forall(symbol(Q,M,K,N,P,_,_),\n"
        "                 dump_line(Out, json{rel:symbol, q:Q, m:M, k:K, n:N, p:P}))\n"
        "        ),\n"
        "        close(Out)).\n",
        encoding="utf-8",
    )
    t0 = time.perf_counter()
    proc = subprocess.run(
        ["swipl", "-q", "-g", "run", "-g", "halt", str(query_path)],
        capture_output=True,
        text=True,
        timeout=SWIPL_TIMEOUT_S,
    )
    elapsed = time.perf_counter() - t0
    if not edges_path.exists() or not edges_path.stat().st_size:
        # A failed consult (e.g. missing facts file) is only a WARNING for
        # swipl: rc stays 0 and the forall dump still runs, producing an
        # EMPTY edges file. Any real codebase has module/symbol records, so
        # an empty dump always means the KB never loaded — treat it as the
        # failure it is (mirrors psub_query's empty-result check).
        raise RuntimeError(
            f"edge dump failed: rc={proc.returncode}\n"
            f"stdout={proc.stdout[-2000:]}\nstderr={proc.stderr[-2000:]}"
        )
    return edges_path, elapsed


@dataclass
class Graph:
    dep_edge_adj: dict[str, set[str]]
    dep_edge_radj: dict[str, set[str]]
    mod_edge_adj: dict[str, set[str]]
    mod_edge_radj: dict[str, set[str]]
    modules: set[str]
    symbols: dict[str, dict[str, str]]
    calls_in_degree: Counter[str]
    n_calls_resolved: int
    n_inherits_resolved: int
    n_dep_raw: int


def build_graph(edges_path: Path) -> tuple[Graph, float]:
    """Parse the JSON-lines edge dump and build the Py-mem adjacency
    structures. Returns (graph, elapsed_seconds) — the elapsed time IS the
    "Py-mem graph build" number reported separately from query time."""
    t0 = time.perf_counter()
    dep_edge_adj: dict[str, set[str]] = defaultdict(set)
    dep_edge_radj: dict[str, set[str]] = defaultdict(set)
    dep_raw_pairs: list[tuple[str, str]] = []
    modules: set[str] = set()
    symbols: dict[str, dict[str, str]] = {}
    calls_in_degree: Counter[str] = Counter()
    n_calls_resolved = 0
    n_inherits_resolved = 0

    with open(edges_path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            rec = json.loads(line)
            rel = rec["rel"]
            if rel == "calls_resolved":
                a, b = rec["a"], rec["b"]
                dep_edge_adj[a].add(b)
                dep_edge_radj[b].add(a)
                calls_in_degree[b] += 1
                n_calls_resolved += 1
            elif rel == "inherits_resolved":
                a, b = rec["a"], rec["b"]
                dep_edge_adj[a].add(b)
                dep_edge_radj[b].add(a)
                n_inherits_resolved += 1
            elif rel == "dep_raw":
                dep_raw_pairs.append((rec["a"], rec["b"]))
            elif rel == "module":
                modules.add(rec["a"])
            elif rel == "symbol":
                symbols[rec["q"]] = {
                    "mod": rec["m"],
                    "kind": rec["k"],
                    "name": rec["n"],
                    "parent": rec["p"],
                }

    mod_edge_adj: dict[str, set[str]] = defaultdict(set)
    mod_edge_radj: dict[str, set[str]] = defaultdict(set)
    for a, b in dep_raw_pairs:
        if b in modules:
            mod_edge_adj[a].add(b)
            mod_edge_radj[b].add(a)

    elapsed = time.perf_counter() - t0
    graph = Graph(
        dep_edge_adj=dict(dep_edge_adj),
        dep_edge_radj=dict(dep_edge_radj),
        mod_edge_adj=dict(mod_edge_adj),
        mod_edge_radj=dict(mod_edge_radj),
        modules=modules,
        symbols=symbols,
        calls_in_degree=calls_in_degree,
        n_calls_resolved=n_calls_resolved,
        n_inherits_resolved=n_inherits_resolved,
        n_dep_raw=len(dep_raw_pairs),
    )
    return graph, elapsed


# =====================================================================
# Py-mem recursive algorithms (mirror the tabled Prolog rules exactly)
# =====================================================================


def closure(start: str, adj: dict[str, set[str]]) -> set[str]:
    """Transitive closure of *start*'s direct successors in *adj* — mirrors
    affected/2 (when adj=dep_edge_radj) and mod_depends/2 (adj=mod_edge_adj
    or mod_edge_radj depending on direction). Iterative DFS with memo (a
    visited set doubling as the memo table)."""
    result: set[str] = set()
    stack = list(adj.get(start, ()))
    while stack:
        node = stack.pop()
        if node in result:
            continue
        result.add(node)
        stack.extend(adj.get(node, ()))
    return result


def bfs_path(a: str, b: str, adj: dict[str, set[str]]) -> list[str] | None:
    """Shortest simple path a -> b over adj, or None. A concrete, valid path
    — not necessarily identical to Prolog's DFS-found path (parity is on
    found/not-found, per protocol)."""
    if a == b:
        return [a]
    prev: dict[str, str | None] = {a: None}
    q: deque[str] = deque([a])
    while q:
        cur = q.popleft()
        if cur == b:
            break
        for nxt in adj.get(cur, ()):
            if nxt not in prev:
                prev[nxt] = cur
                q.append(nxt)
    if b not in prev:
        return None
    path = [b]
    while path[-1] != a:
        # Loop invariant: only the entry for `a` maps to None (the BFS seed);
        # every node reachable from it maps to its str predecessor, and this
        # branch only runs while path[-1] != a.
        pred = prev[path[-1]]
        assert pred is not None
        path.append(pred)
    path.reverse()
    return path


def _paths_to_target(
    start: str, target: str, adj: dict[str, set[str]]
) -> list[list[str]]:
    """All simple node-sequences start -> ... -> target over adj, mirroring
    walk_to/4's nondeterministic backtracking exactly (no repeated nodes
    except the closing return to target)."""
    results: list[list[str]] = []
    visited = {start}
    path = [start]

    def dfs(current: str) -> None:
        if current == target:
            results.append(list(path))
            return
        for nxt in adj.get(current, ()):
            if nxt == target:
                path.append(nxt)
                results.append(list(path))
                path.pop()
            elif nxt not in visited:
                visited.add(nxt)
                path.append(nxt)
                dfs(nxt)
                path.pop()
                visited.discard(nxt)

    dfs(start)
    return results


def enumerate_raw_cycles(adj: dict[str, set[str]]) -> list[list[str]]:
    """Mirrors mod_cycle/2: for every module M and every direct successor X,
    every simple path X -> ... -> M closes a cycle [M, X, ..., M]."""
    return [
        [m, *tail]
        for m in adj
        for x in adj.get(m, ())
        for tail in _paths_to_target(x, m, adj)
    ]


def canonicalize_cycle(cycle: list[str]) -> tuple[str, ...]:
    """Mirrors canonical_cycle/2: drop the closing duplicate, rotate to the
    (first occurrence of the) minimum element, re-append it as the closer."""
    mid = cycle[:-1]
    m = min(mid)
    i = mid.index(m)
    rot = mid[i:] + mid[:i]
    return tuple((*rot, m))


def py_mem_cycles(adj: dict[str, set[str]]) -> frozenset[tuple[str, ...]]:
    return frozenset(canonicalize_cycle(c) for c in enumerate_raw_cycles(adj))


# =====================================================================
# Workload selection (computed from the live graph)
# =====================================================================


@dataclass
class Workloads:
    w1_symbols: list[str]
    w2_pairs: list[tuple[str, str]]  # (module, direction)
    w4_pairs: list[tuple[str, str, int]]  # (from, to, hop_count)
    wrapper_symbol: str | None


def select_workloads(graph: Graph) -> Workloads:
    ranked_symbols = sorted(
        graph.calls_in_degree.items(), key=lambda kv: (-kv[1], kv[0])
    )
    w1_symbols = [q for q, _ in ranked_symbols[:W1_W2_COUNT]]

    fan_out = {m: len(graph.mod_edge_adj.get(m, ())) for m in graph.modules}
    fan_in = {m: len(graph.mod_edge_radj.get(m, ())) for m in graph.modules}
    top_fan_in = sorted(graph.modules, key=lambda m: (-fan_in[m], m))[:5]
    top_fan_out = sorted(graph.modules, key=lambda m: (-fan_out[m], m))[:5]
    w2_modules: list[str] = []
    for m in [*top_fan_in, *top_fan_out]:
        if m not in w2_modules:
            w2_modules.append(m)
    w2_pairs = [(m, d) for m in w2_modules for d in ("deps", "dependents")]

    candidates: list[tuple[str, str, int]] = []
    for a in sorted(graph.modules):
        dist: dict[str, int] = {a: 0}
        bfs_queue: deque[str] = deque([a])
        while bfs_queue:
            cur = bfs_queue.popleft()
            for nxt in graph.mod_edge_adj.get(cur, ()):
                if nxt not in dist:
                    dist[nxt] = dist[cur] + 1
                    bfs_queue.append(nxt)
        for b, d in dist.items():
            if d >= W4_MIN_HOPS:
                candidates.append((a, b, d))
    candidates.sort()
    w4_pairs: list[tuple[str, str, int]] = []
    seen_a: set[str] = set()
    for a, b, d in candidates:
        if a in seen_a:
            continue
        w4_pairs.append((a, b, d))
        seen_a.add(a)
        if len(w4_pairs) >= W1_W2_COUNT:
            break
    if len(w4_pairs) < W1_W2_COUNT:
        chosen_pairs = {(a, b) for a, b, _ in w4_pairs}
        for a, b, d in candidates:
            if (a, b) in chosen_pairs:
                continue
            w4_pairs.append((a, b, d))
            chosen_pairs.add((a, b))
            if len(w4_pairs) >= W1_W2_COUNT:
                break

    wrapper_symbol = None
    for sym_q, _ in ranked_symbols:
        if "L" not in sym_q:
            wrapper_symbol = sym_q
            break

    return Workloads(
        w1_symbols=w1_symbols,
        w2_pairs=w2_pairs,
        w4_pairs=w4_pairs,
        wrapper_symbol=wrapper_symbol,
    )


# =====================================================================
# P-sub driver (cold subprocess per query, JSON via atom_json_dict)
# =====================================================================


def psub_query(
    rules_pl: Path,
    facts_path: Path,
    workdir: Path,
    goal_template: str,
    out_var: str = "R",
) -> tuple[dict[str, Any], float]:
    """Run one Prolog goal via a fresh swipl subprocess; return (result_dict,
    elapsed_seconds) for the WHOLE subprocess (consult + query + write)."""
    fd, query_name = tempfile.mkstemp(suffix=".pl", dir=workdir)
    os.close(fd)
    fd2, result_name = tempfile.mkstemp(suffix=".json", dir=workdir)
    os.close(fd2)
    query_path = Path(query_name)
    result_path = Path(result_name)
    try:
        query_path.write_text(
            ":- use_module(library(http/json)).\n"
            f":- consult({quote_atom(str(rules_pl))}).\n"
            f":- consult({quote_atom(str(facts_path))}).\n"
            "run :-\n"
            f"    {goal_template},\n"
            f"    atom_json_dict(JsonAtom, {out_var}, [as(atom)]),\n"
            "    setup_call_cleanup(\n"
            f"        open({quote_atom(str(result_path))}, write, Out, [encoding(utf8)]),\n"
            '        format(Out, "~a", [JsonAtom]),\n'
            "        close(Out)).\n",
            encoding="utf-8",
        )
        t0 = time.perf_counter()
        proc = subprocess.run(
            ["swipl", "-q", "-g", "run", "-g", "halt", str(query_path)],
            capture_output=True,
            text=True,
            timeout=SWIPL_TIMEOUT_S,
        )
        elapsed = time.perf_counter() - t0
        if not result_path.exists() or not result_path.stat().st_size:
            raise RuntimeError(
                f"P-sub query produced no result: rc={proc.returncode}\n"
                f"goal={goal_template}\nstdout={proc.stdout[-1000:]}\nstderr={proc.stderr[-1000:]}"
            )
        result = json.loads(result_path.read_text(encoding="utf-8"))
        return result, elapsed
    finally:
        query_path.unlink(missing_ok=True)
        result_path.unlink(missing_ok=True)


def time_psub(
    rules_pl: Path, facts_path: Path, workdir: Path, goal_template: str, runs: int
) -> QueryTiming:
    result: dict[str, Any] | None = None
    times: list[float] = []
    for _ in range(runs):
        r, t = psub_query(rules_pl, facts_path, workdir, goal_template)
        if result is None:
            result = r
        times.append(t)
    return QueryTiming(label=goal_template, result=result, times_s=times)


# =====================================================================
# P-janus driver (in-process, warm) — sanity-checked in a SUBPROCESS first
# =====================================================================


def janus_sanity_check(rules_pl: Path, facts_path: Path) -> tuple[bool, str]:
    """Consult rules+facts and run one tabled query INSIDE A SUBPROCESS, so a
    segfault (if the patched janus_swi genuinely cannot handle this KB)
    cannot kill the main benchmark process."""
    script = (
        "import janus_swi as janus\n"
        f"janus.consult({str(rules_pl)!r})\n"
        f"janus.consult({str(facts_path)!r})\n"
        "r = janus.query_once('tool_stats(R)', {})\n"
        "assert r['truth'] is True, r\n"
        "print('JANUS_SANITY_OK')\n"
    )
    proc = subprocess.run(
        [sys.executable, "-c", script],
        capture_output=True,
        text=True,
        timeout=SWIPL_TIMEOUT_S,
    )
    ok = proc.returncode == 0 and "JANUS_SANITY_OK" in proc.stdout
    detail = f"rc={proc.returncode}\nstdout={proc.stdout[-2000:]}\nstderr={proc.stderr[-2000:]}"
    return ok, detail


def time_warm(
    fn: Callable[[], Any], runs: int, warmup: int = 1
) -> tuple[Any, list[float], float]:
    """Generic warm-mode timing harness: discard `warmup` calls (their
    result AND elapsed time of the LAST warmup call is kept — for tabled
    Prolog predicates this warmup call IS the "table build / first call"
    cost the plan asks to report separately), then time `runs` more calls."""
    result = None
    first_call_s = 0.0
    for _ in range(warmup):
        t0 = time.perf_counter()
        result = fn()
        first_call_s = time.perf_counter() - t0
    times: list[float] = []
    for _ in range(runs):
        t0 = time.perf_counter()
        result = fn()
        times.append(time.perf_counter() - t0)
    return result, times, first_call_s


# =====================================================================
# Py-cli driver (end-to-end serialize_codebase_ast.py, cold)
# =====================================================================


def cli_name_for_symbol(sym_info: dict[str, str]) -> str:
    if sym_info["kind"] == "method" and sym_info["parent"] != "none":
        class_name = sym_info["parent"].rsplit(".", 1)[-1]
        return f"{class_name}.{sym_info['name']}"
    return sym_info["name"]


def time_cli(args: list[str], runs: int) -> QueryTiming:
    result = None
    times: list[float] = []
    for _ in range(runs):
        t0 = time.perf_counter()
        proc = subprocess.run(
            [
                sys.executable,
                str(REPO_ROOT / "scripts" / "serialize_codebase_ast.py"),
                *args,
            ],
            capture_output=True,
            text=True,
            cwd=REPO_ROOT,
            timeout=SWIPL_TIMEOUT_S,
        )
        elapsed = time.perf_counter() - t0
        times.append(elapsed)
        if result is None:
            result = {"returncode": proc.returncode, "stdout": proc.stdout}
    return QueryTiming(label=" ".join(args), result=result, times_s=times)


# =====================================================================
# Production-wrapper row (post-remediation CodebaseTools: per-instance
# workdir + attach(); the pre-fix class-level /tmp constants are gone)
# =====================================================================


def time_shipped_wrapper(
    rules_pl: Path, facts_path: Path, workdir: Path, qname: str
) -> tuple[dict[str, Any], float]:
    sys.path.insert(0, str(REPO_ROOT / "src"))
    from orchestrata_ultimate.tools.codebase_tools import CodebaseTools  # noqa: PLC0415

    tools = CodebaseTools(prolog_file=str(rules_pl), workdir=str(workdir))
    tools.attach(str(facts_path))  # facts already generated; no reindex needed
    t0 = time.perf_counter()
    result = tools.impact(qname, limit=NO_LIMIT)
    elapsed = time.perf_counter() - t0
    return result, elapsed


# =====================================================================
# Parity + benchmark orchestration per workload
# =====================================================================


def qnames_of(dicts: list[dict[str, Any]], key: str) -> frozenset[str]:
    return frozenset(d[key] for d in dicts)


def run_w1(
    graph: Graph,
    workloads: Workloads,
    rules_pl: Path,
    facts_path: Path,
    workdir: Path,
    runs: int,
    janus_available: bool,
) -> WorkloadOutcome:
    outcome = WorkloadOutcome(name="W1 impact closure", parity_ok=True)
    outcome.timings["P-sub"] = []
    if janus_available:
        outcome.timings["P-janus"] = []
    outcome.timings["Py-mem"] = []

    for q in workloads.w1_symbols:
        goal = f"tool_impact({quote_atom(q)}, {NO_LIMIT}, R)"
        psub_t = time_psub(rules_pl, facts_path, workdir, goal, runs)
        psub_set = qnames_of(psub_t.result["affected"], "qname")
        outcome.timings["P-sub"].append(psub_t)

        pymem_set = closure(q, graph.dep_edge_radj)
        pymem_times: list[float] = []
        for _ in range(runs + 1):
            t0 = time.perf_counter()
            closure(q, graph.dep_edge_radj)
            pymem_times.append(time.perf_counter() - t0)
        outcome.timings["Py-mem"].append(
            QueryTiming(
                label=q,
                result=pymem_set,
                times_s=pymem_times[1:],
                first_call_s=pymem_times[0],
            )
        )

        if pymem_set != psub_set:
            outcome.parity_ok = False
            outcome.parity_notes.append(
                f"W1 {q}: P-sub={len(psub_set)} Py-mem={len(pymem_set)} "
                f"delta(P-sub-Py-mem)={sorted(psub_set - pymem_set)[:5]} "
                f"delta(Py-mem-P-sub)={sorted(pymem_set - psub_set)[:5]}"
            )

        if janus_available:
            import janus_swi as janus  # noqa: PLC0415

            def call_impact(q: str = q) -> dict[str, Any]:
                r = janus.query_once(
                    "tool_impact(Q, Limit, R)", {"Q": q, "Limit": NO_LIMIT}
                )
                assert r["truth"] is True, r
                return cast("dict[str, Any]", r["R"])

            jresult, jtimes, jfirst = time_warm(call_impact, runs)
            janus_set = qnames_of(jresult["affected"], "qname")
            outcome.timings["P-janus"].append(
                QueryTiming(
                    label=q, result=janus_set, times_s=jtimes, first_call_s=jfirst
                )
            )
            if janus_set != psub_set:
                outcome.parity_ok = False
                outcome.parity_notes.append(
                    f"W1 {q}: P-janus != P-sub "
                    f"delta={sorted(janus_set.symmetric_difference(psub_set))[:5]}"
                )
    return outcome


def run_w2(
    graph: Graph,
    workloads: Workloads,
    rules_pl: Path,
    facts_path: Path,
    workdir: Path,
    runs: int,
    janus_available: bool,
) -> WorkloadOutcome:
    outcome = WorkloadOutcome(name="W2 module deps", parity_ok=True)
    outcome.timings["P-sub"] = []
    if janus_available:
        outcome.timings["P-janus"] = []
    outcome.timings["Py-mem"] = []

    for mod, direction in workloads.w2_pairs:
        goal = f"tool_module_deps({quote_atom(mod)}, {direction}, {NO_LIMIT}, R)"
        psub_t = time_psub(rules_pl, facts_path, workdir, goal, runs)
        psub_set = qnames_of(psub_t.result["modules"], "module")
        outcome.timings["P-sub"].append(psub_t)

        adj = graph.mod_edge_adj if direction == "deps" else graph.mod_edge_radj
        pymem_times = []
        pymem_set = closure(mod, adj)
        for _ in range(runs + 1):
            t0 = time.perf_counter()
            closure(mod, adj)
            pymem_times.append(time.perf_counter() - t0)
        label = f"{mod}[{direction}]"
        outcome.timings["Py-mem"].append(
            QueryTiming(
                label=label,
                result=pymem_set,
                times_s=pymem_times[1:],
                first_call_s=pymem_times[0],
            )
        )

        if pymem_set != psub_set:
            outcome.parity_ok = False
            outcome.parity_notes.append(
                f"W2 {label}: P-sub={len(psub_set)} Py-mem={len(pymem_set)} "
                f"delta={sorted(psub_set.symmetric_difference(pymem_set))[:5]}"
            )

        if janus_available:
            import janus_swi as janus  # noqa: PLC0415

            def call_deps(mod: str = mod, direction: str = direction) -> dict[str, Any]:
                r = janus.query_once(
                    "tool_module_deps(M, D, Limit, R)",
                    {"M": mod, "D": direction, "Limit": NO_LIMIT},
                )
                assert r["truth"] is True, r
                return cast("dict[str, Any]", r["R"])

            jresult, jtimes, jfirst = time_warm(call_deps, runs)
            janus_set = qnames_of(jresult["modules"], "module")
            outcome.timings["P-janus"].append(
                QueryTiming(
                    label=label, result=janus_set, times_s=jtimes, first_call_s=jfirst
                )
            )
            if janus_set != psub_set:
                outcome.parity_ok = False
                outcome.parity_notes.append(
                    f"W2 {label}: P-janus != P-sub "
                    f"delta={sorted(janus_set.symmetric_difference(psub_set))[:5]}"
                )
    return outcome


def run_w3(
    graph: Graph,
    rules_pl: Path,
    facts_path: Path,
    workdir: Path,
    runs: int,
    janus_available: bool,
) -> WorkloadOutcome:
    outcome = WorkloadOutcome(name="W3 cycle enumeration", parity_ok=True)
    goal = f"tool_cycles({NO_LIMIT}, R)"
    psub_t = time_psub(rules_pl, facts_path, workdir, goal, runs)
    psub_cycles = frozenset(tuple(c) for c in psub_t.result["cycles"])
    outcome.timings["P-sub"] = [psub_t]

    pymem_times = []
    pymem_cycles = py_mem_cycles(graph.mod_edge_adj)
    for _ in range(runs + 1):
        t0 = time.perf_counter()
        py_mem_cycles(graph.mod_edge_adj)
        pymem_times.append(time.perf_counter() - t0)
    outcome.timings["Py-mem"] = [
        QueryTiming(
            label="tool_cycles",
            result=pymem_cycles,
            times_s=pymem_times[1:],
            first_call_s=pymem_times[0],
        )
    ]

    if pymem_cycles != psub_cycles:
        outcome.parity_ok = False
        outcome.parity_notes.append(
            f"W3: P-sub total={len(psub_cycles)} Py-mem total={len(pymem_cycles)} "
            f"delta(P-sub-Py-mem)={sorted(psub_cycles - pymem_cycles)[:5]} "
            f"delta(Py-mem-P-sub)={sorted(pymem_cycles - psub_cycles)[:5]}"
        )

    if janus_available:
        import janus_swi as janus  # noqa: PLC0415

        def call_cycles() -> dict[str, Any]:
            r = janus.query_once("tool_cycles(Limit, R)", {"Limit": NO_LIMIT})
            assert r["truth"] is True, r
            return cast("dict[str, Any]", r["R"])

        jresult, jtimes, jfirst = time_warm(call_cycles, runs)
        janus_cycles = frozenset(tuple(c) for c in jresult["cycles"])
        outcome.timings["P-janus"] = [
            QueryTiming(
                label="tool_cycles",
                result=janus_cycles,
                times_s=jtimes,
                first_call_s=jfirst,
            )
        ]
        if janus_cycles != psub_cycles:
            outcome.parity_ok = False
            outcome.parity_notes.append(
                f"W3: P-janus != P-sub delta={sorted(janus_cycles.symmetric_difference(psub_cycles))[:5]}"
            )
    return outcome


def run_w4(
    graph: Graph,
    workloads: Workloads,
    rules_pl: Path,
    facts_path: Path,
    workdir: Path,
    runs: int,
    janus_available: bool,
) -> WorkloadOutcome:
    outcome = WorkloadOutcome(name="W4 dep_path", parity_ok=True)
    outcome.timings["P-sub"] = []
    if janus_available:
        outcome.timings["P-janus"] = []
    outcome.timings["Py-mem"] = []

    for a, b, hops in workloads.w4_pairs:
        goal = f"tool_dep_path({quote_atom(a)}, {quote_atom(b)}, R)"
        psub_t = time_psub(rules_pl, facts_path, workdir, goal, runs)
        psub_found = bool(psub_t.result.get("found"))
        outcome.timings["P-sub"].append(psub_t)

        # Validate the Prolog-returned path is a real edge path.
        if psub_found:
            path = psub_t.result["path"]
            for x, y in zip(path, path[1:]):
                if y not in graph.mod_edge_adj.get(x, ()):
                    outcome.parity_ok = False
                    outcome.parity_notes.append(
                        f"W4 {a}->{b}: P-sub path has non-edge step {x}->{y}"
                    )

        pymem_times = []
        pymem_path = bfs_path(a, b, graph.mod_edge_adj)
        for _ in range(runs + 1):
            t0 = time.perf_counter()
            bfs_path(a, b, graph.mod_edge_adj)
            pymem_times.append(time.perf_counter() - t0)
        pymem_found = pymem_path is not None
        label = f"{a}->{b}(hops>={hops})"
        outcome.timings["Py-mem"].append(
            QueryTiming(
                label=label,
                result=pymem_found,
                times_s=pymem_times[1:],
                first_call_s=pymem_times[0],
            )
        )

        if pymem_found != psub_found:
            outcome.parity_ok = False
            outcome.parity_notes.append(
                f"W4 {label}: P-sub found={psub_found} Py-mem found={pymem_found}"
            )

        if janus_available:
            import janus_swi as janus  # noqa: PLC0415

            def call_dep_path(a: str = a, b: str = b) -> dict[str, Any]:
                r = janus.query_once("tool_dep_path(A, B, R)", {"A": a, "B": b})
                assert r["truth"] is True, r
                return cast("dict[str, Any]", r["R"])

            jresult, jtimes, jfirst = time_warm(call_dep_path, runs)
            janus_found = bool(jresult.get("found"))
            outcome.timings["P-janus"].append(
                QueryTiming(
                    label=label, result=janus_found, times_s=jtimes, first_call_s=jfirst
                )
            )
            if janus_found != psub_found:
                outcome.parity_ok = False
                outcome.parity_notes.append(
                    f"W4 {label}: P-janus found={janus_found} P-sub found={psub_found}"
                )
    return outcome


# =====================================================================
# Report generation
# =====================================================================


def fmt_ms(x: float) -> str:
    # Py-mem query times are frequently sub-millisecond; 2 decimals would
    # round them all to "0.00" and hide real (if tiny) differences.
    if 0 < x < 1:
        return f"{x:.4f}"
    return f"{x:.2f}"


def workload_summary_table(outcome: WorkloadOutcome) -> str:
    lines = [
        "| mode | median (ms) | min (ms) | max (ms) | n queries | n runs/query |",
        "|---|---:|---:|---:|---:|---:|",
    ]
    for mode, timings in outcome.timings.items():
        all_times = [t for qt in timings for t in qt.times_s]
        if not all_times:
            continue
        s = stats_of(all_times)
        lines.append(
            f"| {mode} | {fmt_ms(s['median_ms'])} | {fmt_ms(s['min_ms'])} | "
            f"{fmt_ms(s['max_ms'])} | {len(timings)} | {timings[0].stats['n']} |"
        )
    return "\n".join(lines)


def workload_detail_table(outcome: WorkloadOutcome) -> str:
    modes = list(outcome.timings.keys())
    lines = [
        "| query | "
        + " | ".join(f"{m} median (ms)" for m in modes)
        + " | P-janus first-call (ms) |",
        "|---" * (len(modes) + 2) + "|",
    ]
    n_queries = max((len(v) for v in outcome.timings.values()), default=0)
    for i in range(n_queries):
        row = []
        label = None
        janus_first = ""
        for m in modes:
            qts = outcome.timings[m]
            if i < len(qts):
                qt = qts[i]
                label = label or qt.label
                row.append(fmt_ms(qt.stats["median_ms"]))
                if m == "P-janus" and qt.first_call_s is not None:
                    janus_first = fmt_ms(qt.first_call_s * 1000)
            else:
                row.append("n/a")
        lines.append(f"| {label} | " + " | ".join(row) + f" | {janus_first or 'n/a'} |")
    return "\n".join(lines)


def verdict_line(outcome: WorkloadOutcome, mode: str, baseline: str) -> str | None:
    if mode not in outcome.timings or baseline not in outcome.timings:
        return None
    mode_times = [t for qt in outcome.timings[mode] for t in qt.times_s]
    base_times = [t for qt in outcome.timings[baseline] for t in qt.times_s]
    if not mode_times or not base_times:
        return None
    mode_med = statistics.median(mode_times)
    base_med = statistics.median(base_times)
    ratio = base_med / mode_med if mode_med else float("inf")
    verdict = "PASS" if ratio > 1.0 else "FAIL"
    return (
        f"- **{mode} vs {baseline}** ({outcome.name}): {verdict} — "
        f"{mode} median {fmt_ms(mode_med * 1000)} ms vs {baseline} median "
        f"{fmt_ms(base_med * 1000)} ms (ratio {ratio:.2f}x)"
    )


def build_report(
    env: dict[str, str],
    graph: Graph,
    workloads: Workloads,
    facts_elapsed: float,
    facts_warnings: list[Any],
    edge_dump_elapsed: float,
    graph_build_elapsed: float,
    consult_overhead: dict[str, float],
    janus_consult_s: float | None,
    janus_available: bool,
    janus_error: str,
    outcomes: dict[str, WorkloadOutcome],
    wrapper_row: tuple[str, dict[str, Any], float] | None,
    cli_outcomes: dict[str, QueryTiming],
    runs: int,
) -> str:
    lines: list[str] = []
    lines.append("# Prolog vs Python codebase-intelligence benchmark\n")
    lines.append(
        "Generated by `scripts/benchmark_prolog_codeintel.py`. Answers: is the Prolog "
        "codebase-intelligence layer (`codebase_db.pl`) better/faster than plain Python "
        "recursion for recursive code-intelligence queries?\n"
    )

    lines.append("## Environment\n")
    lines.extend(f"- **{k}**: {v}" for k, v in env.items())
    lines.append("")

    lines.append("## Graph stats\n")
    lines.append(f"- modules: {len(graph.modules)}")
    lines.append(f"- symbols: {len(graph.symbols)}")
    lines.append(f"- calls_resolved edges: {graph.n_calls_resolved}")
    lines.append(f"- inherits_resolved edges: {graph.n_inherits_resolved}")
    lines.append(f"- dep_raw edges: {graph.n_dep_raw}")
    lines.append(
        f"- facts-generation warnings: {facts_warnings if facts_warnings else '[]'}"
    )
    lines.append("")

    lines.append("## Setup timings (not part of any driver's query time)\n")
    lines.append("| step | elapsed (ms) |")
    lines.append("|---|---:|")
    lines.append(
        f"| facts generation (load_codebase + save_facts, cold) | {fmt_ms(facts_elapsed * 1000)} |"
    )
    lines.append(
        f"| P-sub consult overhead alone (empty goal, median of {runs}) | {fmt_ms(consult_overhead['median_ms'])} |"
    )
    lines.append(
        f"| P-sub consult overhead alone (empty goal, min) | {fmt_ms(consult_overhead['min_ms'])} |"
    )
    if janus_available and janus_consult_s is not None:
        lines.append(
            f"| janus consult (rules+facts, once) | {fmt_ms(janus_consult_s * 1000)} |"
        )
    lines.append(
        f"| edge dump (Prolog -> JSON lines, informational) | {fmt_ms(edge_dump_elapsed * 1000)} |"
    )
    lines.append(
        f"| Py-mem graph build (JSON lines -> adjacency dicts) | {fmt_ms(graph_build_elapsed * 1000)} |"
    )
    lines.append("")

    lines.append("## P-janus availability\n")
    if janus_available:
        lines.append(
            "Available. Sanity-checked in a subprocess before use in the main process.\n"
        )
    else:
        lines.append("**UNAVAILABLE.** Sanity-check subprocess failed. Exact error:\n")
        lines.append("```")
        lines.append(janus_error)
        lines.append("```\n")

    lines.append("## Workload selection\n")
    lines.append(
        f"- W1 symbols (top-{W1_W2_COUNT} by incoming calls_resolved edges): "
        f"{', '.join(workloads.w1_symbols)}"
    )
    lines.append(
        f"- W2 modules ({len(workloads.w2_pairs) // 2} modules x 2 directions = "
        f"{len(workloads.w2_pairs)} queries): "
        f"{', '.join(sorted({m for m, _ in workloads.w2_pairs}))}"
    )
    lines.append(
        f"- W4 pairs ({len(workloads.w4_pairs)}, hop_count >= {W4_MIN_HOPS}): "
        + "; ".join(f"{a}->{b} ({h} hops)" for a, b, h in workloads.w4_pairs)
    )
    lines.append(
        f"- as-shipped wrapper demo symbol (no uppercase 'L', dodges D2): "
        f"{workloads.wrapper_symbol}"
    )
    lines.append("")

    lines.append("## Parity results\n")
    lines.append("| workload | parity | notes |")
    lines.append("|---|---|---|")
    for outcome in outcomes.values():
        status = "PASS" if outcome.parity_ok else "FAIL"
        notes = (
            "; ".join(outcome.parity_notes)
            if outcome.parity_notes
            else "identical result sets"
        )
        lines.append(f"| {outcome.name} | {status} | {notes} |")
    lines.append("")

    lines.extend(
        item
        for outcome in outcomes.values()
        for item in (
            f"## {outcome.name}\n",
            "### Summary (all queries pooled)\n",
            workload_summary_table(outcome),
            "",
            "### Per-query median (ms)\n",
            workload_detail_table(outcome),
            "",
        )
    )

    if wrapper_row is not None:
        qname, result, elapsed = wrapper_row
        lines.append("## As-shipped wrapper (CodebaseTools, explicit KB path)\n")
        lines.append(
            f"`CodebaseTools(prolog_file=codebase_db.pl).impact({qname!r}, limit={NO_LIMIT})` "
            f"— D1 (broken default KB path) and D2 (goal-substitution corruption for symbols "
            f"containing uppercase 'L') dodged by construction; D4 (hardcoded /tmp paths) "
            f"dodged by pointing `FACTS_FILE`/`RESULT_FILE` at this run's workdir.\n"
        )
        lines.append(f"- elapsed: {fmt_ms(elapsed * 1000)} ms")
        lines.append(
            f"- total affected: {result.get('total', 'error: ' + str(result))}"
        )
        lines.append("")

    if cli_outcomes:
        lines.append("## Py-cli (end-to-end serialize_codebase_ast.py, context only)\n")
        lines.append("| command | median (ms) | min (ms) |")
        lines.append("|---|---:|---:|")
        for label, qt in cli_outcomes.items():
            s = qt.stats
            lines.append(
                f"| {label} | {fmt_ms(s['median_ms'])} | {fmt_ms(s['min_ms'])} |"
            )
        lines.append("")

    lines.append("## Verdict\n")
    lines.append(
        'Claim under test: "Prolog is better/faster than Python for recursive work." '
        "Assessed per deployment mode against the Py-mem (algorithmic) and Py-cli "
        "(end-to-end) baselines. Ratio = baseline_median / mode_median; ratio > 1.0 means "
        "the Prolog mode is faster (PASS), ratio <= 1.0 means it is not (FAIL).\n"
    )
    for outcome in outcomes.values():
        for mode in ("P-sub", "P-janus"):
            v = verdict_line(outcome, mode, "Py-mem")
            if v:
                lines.append(v)
    if not janus_available:
        lines.append("- **P-janus**: FAIL (unavailable) — see error above.")
    lines.append("")
    if cli_outcomes:
        w1_cli_times = [
            t
            for label, qt in cli_outcomes.items()
            if "call-graph" in label
            for t in qt.times_s
        ]
        w3_cli_label = next(
            (label for label in cli_outcomes if "find-cycles" in label), None
        )
        if w1_cli_times:
            cli_med = statistics.median(w1_cli_times)
            for mode in ("P-sub", "P-janus"):
                if (
                    "W1 impact closure" in outcomes
                    and mode in outcomes["W1 impact closure"].timings
                ):
                    mode_times = [
                        t
                        for qt in outcomes["W1 impact closure"].timings[mode]
                        for t in qt.times_s
                    ]
                    if mode_times:
                        mode_med = statistics.median(mode_times)
                        ratio = cli_med / mode_med if mode_med else float("inf")
                        verdict = "PASS" if ratio > 1.0 else "FAIL"
                        lines.append(
                            f"- **{mode} vs Py-cli** (W1, end-to-end): {verdict} — "
                            f"{mode} median {fmt_ms(mode_med * 1000)} ms vs Py-cli median "
                            f"{fmt_ms(cli_med * 1000)} ms (ratio {ratio:.2f}x)"
                        )
        if w3_cli_label:
            cli_med = statistics.median(cli_outcomes[w3_cli_label].times_s)
            for mode in ("P-sub", "P-janus"):
                if (
                    "W3 cycle enumeration" in outcomes
                    and mode in outcomes["W3 cycle enumeration"].timings
                ):
                    mode_times = [
                        t
                        for qt in outcomes["W3 cycle enumeration"].timings[mode]
                        for t in qt.times_s
                    ]
                    if mode_times:
                        mode_med = statistics.median(mode_times)
                        ratio = cli_med / mode_med if mode_med else float("inf")
                        verdict = "PASS" if ratio > 1.0 else "FAIL"
                        lines.append(
                            f"- **{mode} vs Py-cli** (W3, end-to-end): {verdict} — "
                            f"{mode} median {fmt_ms(mode_med * 1000)} ms vs Py-cli median "
                            f"{fmt_ms(cli_med * 1000)} ms (ratio {ratio:.2f}x)"
                        )
    lines.append("")

    return "\n".join(lines)


# =====================================================================
# main
# =====================================================================


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--ast-json", type=Path, default=DEFAULT_AST_JSON)
    parser.add_argument("--runs", type=int, default=5)
    parser.add_argument("--workdir", type=Path, default=None)
    parser.add_argument("--report", type=Path, default=DEFAULT_REPORT)
    parser.add_argument("--skip-cli", action="store_true")
    args = parser.parse_args()

    ast_json = (
        args.ast_json if args.ast_json.is_absolute() else REPO_ROOT / args.ast_json
    )
    if not ast_json.exists():
        print(f"error: AST json not found: {ast_json}", file=sys.stderr)
        return 2
    report_path = args.report if args.report.is_absolute() else REPO_ROOT / args.report
    workdir = args.workdir or Path(tempfile.mkdtemp(prefix="prolog_codeintel_bench_"))
    workdir.mkdir(parents=True, exist_ok=True)
    runs = args.runs

    print("[env]     collecting environment info")
    env = collect_environment()

    print(f"[facts]   generating facts from {ast_json} -> {workdir}")
    facts_path, facts_warnings, facts_elapsed = generate_facts(ast_json, workdir)
    print(f"[facts]   done in {facts_elapsed:.2f}s, warnings={facts_warnings}")

    print(f"[consult] measuring P-sub consult overhead ({runs} runs)")
    consult_overhead = measure_consult_overhead(facts_path, runs)

    print("[edges]   dumping edge relations")
    edges_path, edge_dump_elapsed = dump_edges(RULES_PL, facts_path, workdir)
    print(f"[edges]   done in {edge_dump_elapsed:.2f}s")

    print("[graph]   building Py-mem adjacency structures")
    graph, graph_build_elapsed = build_graph(edges_path)
    print(
        f"[graph]   done in {graph_build_elapsed:.2f}s ({len(graph.modules)} modules, "
        f"{len(graph.symbols)} symbols)"
    )

    print("[workload] selecting workloads from the live graph")
    workloads = select_workloads(graph)

    print("[janus]   sanity-checking janus_swi in a subprocess")
    janus_available, janus_error = janus_sanity_check(RULES_PL, facts_path)
    janus_consult_s: float | None = None
    if janus_available:
        print("[janus]   sanity OK; consulting rules+facts in-process")
        import janus_swi as janus  # noqa: PLC0415

        t0 = time.perf_counter()
        janus.consult(str(RULES_PL))
        janus.consult(str(facts_path))
        janus_consult_s = time.perf_counter() - t0
        print(f"[janus]   consult done in {janus_consult_s:.3f}s")
    else:
        print(f"[janus]   UNAVAILABLE: {janus_error}")

    outcomes: dict[str, WorkloadOutcome] = {}
    print(f"[W1]      impact closure ({len(workloads.w1_symbols)} symbols)")
    outcomes["W1 impact closure"] = run_w1(
        graph, workloads, RULES_PL, facts_path, workdir, runs, janus_available
    )
    print(f"[W2]      module deps ({len(workloads.w2_pairs)} queries)")
    outcomes["W2 module deps"] = run_w2(
        graph, workloads, RULES_PL, facts_path, workdir, runs, janus_available
    )
    print("[W3]      cycle enumeration")
    outcomes["W3 cycle enumeration"] = run_w3(
        graph, RULES_PL, facts_path, workdir, runs, janus_available
    )
    print(f"[W4]      dep_path ({len(workloads.w4_pairs)} pairs)")
    outcomes["W4 dep_path"] = run_w4(
        graph, workloads, RULES_PL, facts_path, workdir, runs, janus_available
    )

    wrapper_row = None
    if workloads.wrapper_symbol:
        print(f"[wrapper] as-shipped CodebaseTools demo on {workloads.wrapper_symbol}")
        wresult, welapsed = time_shipped_wrapper(
            RULES_PL, facts_path, workdir, workloads.wrapper_symbol
        )
        wrapper_row = (workloads.wrapper_symbol, wresult, welapsed)

    cli_outcomes: dict[str, QueryTiming] = {}
    if not args.skip_cli:
        print("[cli]     timing Py-cli call-graph + find-cycles")
        for q in workloads.w1_symbols:
            sym = graph.symbols.get(q)
            if sym is None:
                continue
            name = cli_name_for_symbol(sym)
            qt = time_cli(["call-graph", "--name", name], runs)
            cli_outcomes[f"call-graph --name {name}"] = qt
        cli_outcomes["find-cycles"] = time_cli(["find-cycles"], runs)

    all_parity_ok = all(o.parity_ok for o in outcomes.values())
    print(f"[parity]  overall: {'PASS' if all_parity_ok else 'FAIL'}")
    for o in outcomes.values():
        if not o.parity_ok:
            for note in o.parity_notes:
                print(f"[parity]  FAIL: {note}")

    report = build_report(
        env=env,
        graph=graph,
        workloads=workloads,
        facts_elapsed=facts_elapsed,
        facts_warnings=facts_warnings,
        edge_dump_elapsed=edge_dump_elapsed,
        graph_build_elapsed=graph_build_elapsed,
        consult_overhead=consult_overhead,
        janus_consult_s=janus_consult_s,
        janus_available=janus_available,
        janus_error=janus_error,
        outcomes=outcomes,
        wrapper_row=wrapper_row,
        cli_outcomes=cli_outcomes,
        runs=runs,
    )
    report_path.parent.mkdir(parents=True, exist_ok=True)
    report_path.write_text(report, encoding="utf-8")
    print(f"[report]  written to {report_path}")
    print(f"[workdir] {workdir}")
    return 0


if __name__ == "__main__":
    sys.exit(main())