scripts.trace_intel

Reference › scripts › trace_intel

scripts/trace_intel.py is the T1 traceability and use-case-narrative engine: a semantic Prolog layer over the structural code-intelligence knowledge base. It answers which code implements which requirement, which code is unreachable from any entrypoint, and which requirements nothing traces to. A human or agent runs it; derive_rtm imports it and tests/scripts/test_derive_rtm.py covers it. It uses orchestrata_ultimate.tools.ast_pipeline and orchestrata_ultimate.tools.codebase_tools. Its sibling for the time dimension is history_intel, and the graph-side counterpart is sync_traces_to_graph.

The extract subcommand regenerates traceability_facts.pl from plans/ast-ultimate.json and specs/. _traces_facts builds traces(QName, ReqId) facts from every @traces decorator (_iter_symbols, _extract_req_ids), and entrypoint facts are emitted for classes, CLI commands and other kinds (_entrypoint_kind, _emit_entrypoint, with class inheritance followed transitively). Query subcommands run against that base: report (all five finders with counts; --merged compares static with merged static-plus-runtime reachability), narrative <qname> (modules transitively reachable from one entrypoint), orphans (traced code unreachable from any entrypoint), untraced-reqs (requirements with zero @traces coverage), other limited listings (--limit, default 50), and why-not <req_id>, which names the first missing conjunct. Queries call ensure_kb first (honouring --no-refresh). The exit code has three values, matching the gate layer: 2 when the outcome is invalid (the fact base could not answer for the bytes on disk), 1 for an error and 0 otherwise.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

_WarmPrologWorker

Line 622

One persistent swipl process holding the STANDARD repo KB.

Functions

Source

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

#!/usr/bin/env python3
"""scripts/trace_intel.py — T1 traceability / use-case-narrative engine.

A semantic Prolog layer ON TOP OF the existing structural code-intel KB
(``src/orchestrata_ultimate/tools/codebase_db.pl`` + ``CodebaseTools``).
Answers questions the structural facts alone cannot: which entrypoint
reaches which module, which traced code is unreachable from any known
entrypoint, which spec requirements have zero implementing code, which
reachable code carries no ``@traces`` at all, and which specs never chain
to reachable code.

Facts (regenerated by the ``extract`` subcommand into
``.cache/trace_intel/traceability_facts.pl``):

  - ``traces(QName, ReqId)`` — one row per ``REQ-...`` id inside every
    ``@traces(spec=..., story=..., reqs=[...])`` decorator found in
    ``plans/ast-ultimate.json`` (QName is the symbol's qualified name,
    already prefix-stripped by the serializer — see ``_strip_prefix``
    for the defensive fallback).
  - ``spec_requirement(SpecId, ReqId, Desc)`` — via
    ``scripts/parse_spec_requirements.py`` (imported by path, REUSED not
    reimplemented) over every ``specs/*/requirements.md`` and
    ``specs/*/checklists/requirements.md``.
  - ``entrypoint(QName, Kind)`` — the real production entry surface, not
    just the MVP CLI seed: ``cli_main`` (a ``main``/``cli_main`` module-
    level function in a ``cli.py``/``cli_main.py``/``__main__.py``
    module), ``server_main`` (a ``main``/``serve`` module-level function
    in a ``*server.py``/``runtime_main.py`` module), ``volk_handler``
    (``.process`` on a concrete ``BaseAgent`` subclass -- the volk
    dispatch loop), ``tool_entry`` (``.execute`` on a concrete
    ``BaseTool`` subclass, or on ``ToolExecutor`` -- the tool-calling
    loop), ``gateway_handler`` (``.handle_*`` on a
    ``gateway.handlers.*`` class -- the gateway RPC table),
    ``cxp_handler`` (the CXP gRPC servicer method set, or ``.__call__``
    on a ``*PacketHandler`` class), and ``cli_command`` (a click
    ``@x.command()``/``@x.group()``-decorated function/method, or an
    argparse ``_cmd_*`` dispatch function in a ``*.cli`` module). See the
    detector-constants block above ``_entrypoint_facts`` for the full
    rationale and why test-exercised roots are deliberately excluded.

Rules: ``src/orchestrata_ultimate/tools/traceability_db.pl`` — consulted
alongside ``codebase_db.pl`` (for ``symbol/7``/``calls_resolved/3``).
Standard-KB queries run on a persistent warm swipl worker (consulted
once per process, mtime-keyed, SLG tables persisting across queries --
see ``_WarmPrologWorker``); non-standard KB lists and any worker
anomaly use the original one-swipl-subprocess-per-query path
(``_run_prolog_oneshot``). Large generated fact files load from
mtime-gated ``.qlf`` precompilations (``_qlf_for``).

The structural KB itself (``symbol/7``, ``calls_resolved/3``, ...) is
built via ``CodebaseTools(facts_file=...).reindex(plans/ast-ultimate.json)``
into a facts file OWNED by this script
(``.cache/trace_intel/codebase_facts.pl``) — never the default
``CodebaseTools`` facts file, so this tool can never race another
process's structural KB (see the module docstring of codebase_tools.py).

Usage:
    python scripts/trace_intel.py extract
    python scripts/trace_intel.py report [--json]
    python scripts/trace_intel.py narrative <entrypoint_qname> [--json]
    python scripts/trace_intel.py orphans [--limit N] [--json]
    python scripts/trace_intel.py untraced-reqs [--limit N] [--json]
    python scripts/trace_intel.py unspecced [--limit N] [--json]
    python scripts/trace_intel.py specless-usecases [--limit N] [--json]

Every query subcommand auto-runs ``extract`` (and reindexes the
structural KB) when ``traceability_facts.pl``/``codebase_facts.pl`` is
missing or older than ``plans/ast-ultimate.json`` — pass ``--no-refresh``
to skip that check and query whatever is already on disk.
"""

from __future__ import annotations

import argparse
import atexit
import contextlib
import importlib.util
import json
import os
import re
import subprocess
import sys
import tempfile
import threading
import time
import types
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import Any

from orchestrata_ultimate.tools.codebase_tools import CodebaseTools

REPO_ROOT = Path(__file__).resolve().parent.parent
AST_ULTIMATE_JSON = REPO_ROOT / "plans" / "ast-ultimate.json"
SPECS_ROOT = REPO_ROOT / "specs"
CODEBASE_DB = REPO_ROOT / "src" / "orchestrata_ultimate" / "tools" / "codebase_db.pl"
TRACEABILITY_DB = (
    REPO_ROOT / "src" / "orchestrata_ultimate" / "tools" / "traceability_db.pl"
)
# SPEC-114 FR-004: the why-not meta-interpreter (ADR-0031). Consulted into
# every session rather than only the `why-not` one, because a session is one
# swipl subprocess per query -- there is no session to add it to later, and
# loading a 262-line rulebase that defines no facts costs nothing measurable
# against the 14K-fact traceability file it sits beside.
EXPLAIN_PL = REPO_ROOT / "src" / "orchestrata_ultimate" / "prolog" / "explain.pl"
_PARSE_SPEC_REQ_PATH = Path(__file__).resolve().with_name("parse_spec_requirements.py")

# Never the default CodebaseTools() facts file (concurrency note in the
# brief): a fully private scratch dir, gitignored (.cache is in .gitignore).
SCRATCH_DIR = REPO_ROOT / ".cache" / "trace_intel"
STRUCTURAL_FACTS = SCRATCH_DIR / "codebase_facts.pl"
TRACEABILITY_FACTS = SCRATCH_DIR / "traceability_facts.pl"

_ORCH_PREFIX = "orchestrata_ultimate."
_CLI_BASENAMES = {"cli.py", "cli_main.py", "__main__.py"}
_CLI_FUNC_NAMES = {"main", "cli_main"}
_SERVER_BASENAMES = {"runtime_main.py"}
_SERVER_FUNC_NAMES = {"main", "serve"}

_REQS_LIST_RE = re.compile(r"reqs\s*=\s*\[([^\]]*)\]")
_REQ_ITEM_RE = re.compile(r"""['"]([^'"]+)['"]""")
_SPEC_DIR_RE = re.compile(r"(\d{3})-")


# ---------------------------------------------------------------------------
# extract — traceability_facts.pl generation
# ---------------------------------------------------------------------------


def _strip_prefix(qname: str) -> str:
    """Strip a leading ``orchestrata_ultimate.`` if present.

    Defensive fallback only — every ``qualified_name`` observed in
    ``plans/ast-ultimate.json`` is already prefix-stripped by the
    serializer (module ``path`` is relative to ``root_path``), matching
    the KB's own convention (codebase_db.pl's ``symbol/7`` stores exactly
    this form). Kept so extraction stays correct even if a future
    serializer run changes that convention.
    """
    return qname[len(_ORCH_PREFIX) :] if qname.startswith(_ORCH_PREFIX) else qname


def _pl_atom(value: str) -> str:
    """Quote a Python string as a Prolog atom (single-quote doubling)."""
    return "'" + value.replace("'", "''") + "'"


def _extract_req_ids(decorator_text: str) -> list[str]:
    """Pull every ``REQ-...`` id out of one ``traces(..., reqs=[...])``
    decorator source string (as captured verbatim by the AST serializer).
    """
    match = _REQS_LIST_RE.search(decorator_text)
    if not match:
        return []
    return _REQ_ITEM_RE.findall(match.group(1))


def _iter_symbols(modules: list[dict[str, Any]]) -> Iterator[tuple[str, list[Any]]]:
    """Yield (qualified_name, decorators) for every class, method, and
    top-level function symbol across all modules."""
    for mod in modules:
        for cls in mod.get("classes") or []:
            yield cls.get("qualified_name", ""), cls.get("decorators") or []
            for meth in cls.get("methods") or []:
                yield meth.get("qualified_name", ""), meth.get("decorators") or []
        for fn in mod.get("functions") or []:
            yield fn.get("qualified_name", ""), fn.get("decorators") or []


def _traces_facts(modules: list[dict[str, Any]]) -> list[str]:
    """Build ``traces(QName, ReqId).`` facts from every ``@traces(...)``
    decorator's ``reqs=[...]`` list."""
    facts: list[str] = []
    seen: set[tuple[str, str]] = set()
    for qname, decorators in _iter_symbols(modules):
        if not qname:
            continue
        q = _strip_prefix(qname)
        for dec in decorators:
            text = str(dec).strip()
            if not text.startswith("traces("):
                continue
            for req_id in _extract_req_ids(text):
                key = (q, req_id)
                if key in seen:
                    continue
                seen.add(key)
                facts.append(f"traces({_pl_atom(q)}, {_pl_atom(req_id)}).")
    return facts


def _entrypoint_kind(file_path: str) -> str | None:
    """Classify a module's file path as an entrypoint-hosting file, or
    None. cli_main/server_main only (see module docstring).

    ``server_main`` was called ``grpc_server`` until 2026-08-07. The label was
    always about SHAPE -- a ``main``/``serve`` module-level function in a
    server module -- never about gRPC, and keeping the old name after the
    grpc.aio transport was retired implied a coupling that never existed."""
    basename = Path(file_path).name
    if basename in _CLI_BASENAMES:
        return "cli_main"
    if basename in _SERVER_BASENAMES or basename.endswith("server.py"):
        return "server_main"
    return None


# ---------------------------------------------------------------------------
# entrypoint/2 detector constants -- PRODUCTION entries only.
#
# Every detector below answers one question: "can this symbol be reached
# by something a DEPLOYED orchestrata-ultimate process actually calls" --
# kernel boot, the gateway RPC table, a CLI parser, the CXP gRPC surface,
# the volk dispatch loop, the tool-calling loop. Deliberately EXCLUDED:
# anything only a test harness reaches (pytest fixtures/conftest wiring,
# `tests/generated_*` stubs, scenario runners under tests/). Seeding from
# test-exercised roots would make orphan_feature/unspecced_code/
# specless_usecase (traceability_db.pl) blind to code that is built but
# never actually wired into a running process -- exactly the failure mode
# this expansion exists to close (AGENTS.md Hard Rule 8: "built is not
# done -- WIRED is done"; the 2026-07-12 audit found a 63-module
# subsystem that shipped tested-but-unreachable because nothing but its
# own tests ever called it). A false-positive entrypoint here MASKS a
# real orphan, so every detector is intentionally narrow (exact
# method-name sets, module-path prefixes, base-class names) and skips
# ``is_protocol`` classes; when a genuine production entry doesn't fit
# these seams, that is itself a finding to report, never a reason to
# loosen a detector to force a number.
# ---------------------------------------------------------------------------

_VOLK_BASE_NAME = "BaseAgent"
_TOOL_BASE_NAME = "BaseTool"
_GATEWAY_HANDLERS_MODULE_SUBSTR = "gateway.handlers"
_GATEWAY_HANDLE_METHOD_RE = re.compile(r"^handle_[a-z]")
_CXP_SERVICER_METHOD_NAMES = frozenset(
    {
        "send_packet",
        "send_packet_stream",
        "health_check",
        "get_status",
        "receive_packet",
        "subscribe",
        "acknowledge",
    }
)
_CLICK_DECORATOR_RE = re.compile(r"^\w+\.(command|group)\(")
_CLI_CMD_FUNC_RE = re.compile(r"^_cmd_[a-z]")


def _build_class_index(
    modules: list[dict[str, Any]],
) -> dict[str, list[dict[str, Any]]]:
    """Simple-name -> [class dict, ...] index across all modules.

    ``bases`` in ``plans/ast-ultimate.json`` are bare names (no module
    qualification), so this bare-name index is the only resolution the
    data supports -- see ``_inherits_transitively``.
    """
    index: dict[str, list[dict[str, Any]]] = {}
    for mod in modules:
        for cls in mod.get("classes") or []:
            name = cls.get("name")
            if name:
                index.setdefault(name, []).append(cls)
    return index


def _inherits_transitively(
    cls: dict[str, Any],
    target_name: str,
    class_index: dict[str, list[dict[str, Any]]],
    _visited: set[str] | None = None,
) -> bool:
    """True if ``cls`` inherits from ``target_name``, directly or through
    any number of intermediate bases (e.g. a mixin chain), resolved via
    the bare-name ``class_index``.

    Bare-name resolution is inherently ambiguous when two classes share a
    name (see ``_build_class_index``) -- that ambiguity is accepted as
    the cost of the AST serializer not qualifying ``bases``, not papered
    over by guessing; it can only ever affect entrypoint/2 membership,
    never the traces/2 or spec_requirement/3 facts.
    """
    visited = _visited if _visited is not None else set()
    for base_name in cls.get("bases") or []:
        if base_name == target_name:
            return True
        if base_name in visited:
            continue
        visited.add(base_name)
        for base_cls in class_index.get(base_name, ()):
            if _inherits_transitively(base_cls, target_name, class_index, visited):
                return True
    return False


def _is_cli_command_decorator(text: str) -> bool:
    """True for a click ``@x.command(...)``/``@x.group(...)`` decorator
    source string (covers ``@click.command()``, ``@cli.command()``,
    ``@agents.group()``, and any decorator text merely containing
    ``click.command`` as a defensive fallback)."""
    stripped = text.strip()
    return bool(_CLICK_DECORATOR_RE.match(stripped)) or "click.command" in stripped


def _emit_entrypoint(
    facts: list[str], seen: set[tuple[str, str]], qname_raw: str, kind: str
) -> None:
    """Append ``entrypoint(QName, Kind).`` to ``facts`` unless ``qname_raw``
    is empty or ``(QName, Kind)`` was already emitted."""
    qname = _strip_prefix(qname_raw)
    if not qname:
        return
    key = (qname, kind)
    if key in seen:
        return
    seen.add(key)
    facts.append(f"entrypoint({_pl_atom(qname)}, {kind}).")


def _entrypoint_facts(modules: list[dict[str, Any]]) -> list[str]:
    """Build every ``entrypoint(QName, Kind).`` fact -- the production
    entry surface the traceability finders (orphan_feature/unspecced_code/
    specless_usecase) treat as reachable roots. See the detector-constants
    block above for what "production entry" means and why test-exercised
    roots are excluded.

    Kinds emitted (dedup'd on ``(QName, Kind)`` across the whole
    function, matching the original cli_main/server_main behaviour):
      - ``cli_main`` / ``server_main`` -- unchanged MVP detector, see
        ``_entrypoint_kind``.
      - ``volk_handler`` -- ``.process`` on a concrete (non-Protocol)
        class transitively inheriting ``BaseAgent``.
      - ``tool_entry`` -- ``.execute`` on a concrete (non-Protocol,
        non-abstract) class transitively inheriting ``BaseTool``, or on
        the ``ToolExecutor`` class (native-tool dispatch).
      - ``gateway_handler`` -- a ``handle_*`` method on a
        ``gateway.handlers.*`` class (the SPEC-014 RPC table).
      - ``cxp_handler`` -- the CXP servicer method set on a ``cxp.*``
        class, or ``.__call__`` on a ``*PacketHandler`` class. The
        ``*grpc_server*`` module-path alternative was dropped on 2026-08-07
        with the grpc transport: ``CXPServicer`` now lives in
        ``cxp.servicer``, which the ``cxp.`` prefix already covers.
      - ``cli_command`` -- a click ``@x.command()``/``@x.group()``-
        decorated function or method, or an argparse ``_cmd_*``
        module-level function in a ``*.cli`` module.
    """
    facts: list[str] = []
    seen: set[tuple[str, str]] = set()

    # -- cli_main / server_main (unchanged MVP detector) --------------------
    for mod in modules:
        kind = _entrypoint_kind(mod.get("file_path", ""))
        if kind is None:
            continue
        wanted_names = _CLI_FUNC_NAMES if kind == "cli_main" else _SERVER_FUNC_NAMES
        for fn in mod.get("functions") or []:
            if fn.get("name") in wanted_names:
                _emit_entrypoint(facts, seen, fn.get("qualified_name", ""), kind)

    class_index = _build_class_index(modules)

    for mod in modules:
        for cls in mod.get("classes") or []:
            if cls.get("is_protocol"):
                continue
            cls_name = cls.get("name") or ""
            module_path = cls.get("module_path") or ""
            is_abstract = bool(cls.get("is_abstract"))
            methods = cls.get("methods") or []

            # -- volk_handler -----------------------------------------------
            if _inherits_transitively(cls, _VOLK_BASE_NAME, class_index):
                for meth in methods:
                    if meth.get("name") == "process":
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "volk_handler"
                        )

            # -- tool_entry -------------------------------------------------
            if not is_abstract and _inherits_transitively(
                cls, _TOOL_BASE_NAME, class_index
            ):
                for meth in methods:
                    if meth.get("name") == "execute":
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "tool_entry"
                        )
            if cls_name == "ToolExecutor":
                for meth in methods:
                    if meth.get("name") == "execute":
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "tool_entry"
                        )

            # -- gateway_handler ----------------------------------------------
            if _GATEWAY_HANDLERS_MODULE_SUBSTR in module_path:
                for meth in methods:
                    if _GATEWAY_HANDLE_METHOD_RE.match(meth.get("name", "")):
                        _emit_entrypoint(
                            facts,
                            seen,
                            meth.get("qualified_name", ""),
                            "gateway_handler",
                        )

            # -- cxp_handler ----------------------------------------------------
            if module_path.startswith("cxp."):
                for meth in methods:
                    if meth.get("name") in _CXP_SERVICER_METHOD_NAMES:
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "cxp_handler"
                        )
            if cls_name.endswith("PacketHandler"):
                for meth in methods:
                    if meth.get("name") == "__call__":
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "cxp_handler"
                        )

            # -- cli_command (methods) ------------------------------------------
            for meth in methods:
                for dec in meth.get("decorators") or []:
                    if _is_cli_command_decorator(str(dec)):
                        _emit_entrypoint(
                            facts, seen, meth.get("qualified_name", ""), "cli_command"
                        )
                        break

    # -- cli_command (module-level functions) --------------------------------
    for mod in modules:
        for fn in mod.get("functions") or []:
            fn_module_path = fn.get("module_path") or ""
            for dec in fn.get("decorators") or []:
                if _is_cli_command_decorator(str(dec)):
                    _emit_entrypoint(
                        facts, seen, fn.get("qualified_name", ""), "cli_command"
                    )
                    break
            if "cli" in fn_module_path and _CLI_CMD_FUNC_RE.match(fn.get("name") or ""):
                _emit_entrypoint(
                    facts, seen, fn.get("qualified_name", ""), "cli_command"
                )

    return facts


def _load_parse_spec_requirements() -> Any:
    """Load scripts/parse_spec_requirements.py by path (REUSED, not
    reimplemented — same pattern as scripts/ast_report.py's ``_load_tool``:
    scripts/ isn't an importable package member for direct ``import``)."""
    spec = importlib.util.spec_from_file_location(
        "parse_spec_requirements", _PARSE_SPEC_REQ_PATH
    )
    if spec is None or spec.loader is None:
        raise ImportError(f"cannot load {_PARSE_SPEC_REQ_PATH}")
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


def _spec_requirement_facts(specs_root: Path) -> list[str]:
    """Build ``spec_requirement(SpecId, ReqId, Desc).`` facts by running
    ``parse_spec_requirements.parse_requirements`` over every
    ``requirements.md`` checklist under ``specs_root`` (both the direct
    ``specs/NNN-.../requirements.md`` and the nested
    ``specs/NNN-.../checklists/requirements.md`` layouts are present in
    this repo)."""
    if not specs_root.is_dir():
        return []
    parser_mod = _load_parse_spec_requirements()
    facts: list[str] = []
    seen_paths: set[Path] = set()
    paths = sorted(specs_root.glob("*/requirements.md")) + sorted(
        specs_root.glob("*/checklists/requirements.md")
    )
    for path in paths:
        resolved = path.resolve()
        if resolved in seen_paths:
            continue
        seen_paths.add(resolved)
        spec_dirname = path.relative_to(specs_root).parts[0]
        match = _SPEC_DIR_RE.match(spec_dirname)
        if not match:
            continue
        spec_id = match.group(1)
        raw_facts = parser_mod.parse_requirements(spec_id, str(path))
        facts.extend(f"{fact}." for fact in raw_facts)
    return facts


def extract_facts(
    ast_json_path: Path = AST_ULTIMATE_JSON,
    specs_root: Path = SPECS_ROOT,
    output_path: Path = TRACEABILITY_FACTS,
) -> tuple[int, int, int]:
    """Regenerate ``output_path`` from ``ast_json_path`` + ``specs_root``.

    Returns ``(n_traces, n_spec_requirements, n_entrypoints)``.
    """
    with open(ast_json_path, encoding="utf-8") as fh:
        ast_data = json.load(fh)
    modules = ast_data.get("modules", [])

    traces_facts = _traces_facts(modules)
    entrypoint_facts = _entrypoint_facts(modules)
    specreq_facts = _spec_requirement_facts(specs_root)

    lines = [
        "% traceability_facts.pl -- GENERATED by scripts/trace_intel.py extract.",
        "% Do not hand-edit; regenerate via `python scripts/trace_intel.py extract`.",
        "",
        ":- dynamic traces/2, spec_requirement/3, entrypoint/2.",
        "",
        *traces_facts,
        "",
        *specreq_facts,
        "",
        *entrypoint_facts,
        "",
    ]

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    return len(traces_facts), len(specreq_facts), len(entrypoint_facts)


def cmd_extract(_args: argparse.Namespace) -> int:
    n_traces, n_specreqs, n_entrypoints = extract_facts()
    print(
        f"trace_intel: wrote {n_traces} traces/2, {n_specreqs} spec_requirement/3, "
        f"{n_entrypoints} entrypoint/2 facts to {TRACEABILITY_FACTS}",
        file=sys.stderr,
    )
    return 0


# ---------------------------------------------------------------------------
# Prolog session — codebase_db.pl + structural facts + traceability_db.pl
# + traceability_facts.pl, consulted together in one swipl subprocess.
# ---------------------------------------------------------------------------


def _kb_files() -> list[Path]:
    return [
        CODEBASE_DB,
        STRUCTURAL_FACTS,
        TRACEABILITY_DB,
        TRACEABILITY_FACTS,
        EXPLAIN_PL,
    ]


# The entrypoint-expansion (263 entrypoint/2 facts vs. the original 9)
# makes `report --merged` tabling `merged_reaches/2` (calls_resolved/3 UNION
# runtime_call/2, ~7K runtime edges) from every one of those 263 roots --
# measured at ~8GB of SLG table space and ~173s wall time on the real repo
# KB, against SWI's 1GB default `table_space` flag (immediate
# `private_table_space` resource error) and this module's prior 180s
# subprocess timeout (which the 173s real run would occasionally miss under
# load). Both bumped here so the pre-existing `report --merged` subcommand
# keeps working at the larger entrypoint scale; harmless for the cheaper
# non-merged queries since `--table-space` is a ceiling, not a reservation.
_SWIPL_TABLE_SPACE = "8G"
_PROLOG_TIMEOUT_SECONDS = 300

# Consulting the ~13MB generated structural-facts file dominates every
# swipl round trip (measured 1.70s of a 2.9s `report` call); the same
# clauses load from a precompiled .qlf in ~0.10s. Compilation is
# mtime-gated and BEST-EFFORT: on any failure the original .pl path is
# returned, so correctness never depends on QLF availability. The size
# floor keeps tiny fixture/rule files (where qcompile costs more than it
# saves) on the plain-consult path.
_QLF_MIN_BYTES = 1_000_000
_QLF_COMPILE_TIMEOUT_SECONDS = 120


def _qlf_for(pl_path: Path) -> Path:
    """Return a fresh precompiled ``.qlf`` twin for large ``.pl`` fact
    files, compiling it when missing/stale; fall back to ``pl_path``."""
    try:
        if pl_path.suffix != ".pl" or pl_path.stat().st_size < _QLF_MIN_BYTES:
            return pl_path
        qlf_path = pl_path.with_suffix(".qlf")
        if qlf_path.exists() and qlf_path.stat().st_mtime >= pl_path.stat().st_mtime:
            return qlf_path
        escaped = str(pl_path).replace("'", "''")
        proc = subprocess.run(
            ["swipl", "-q", "-g", f"qcompile('{escaped}')", "-t", "halt"],
            capture_output=True,
            text=True,
            timeout=_QLF_COMPILE_TIMEOUT_SECONDS,
        )
        if proc.returncode == 0 and qlf_path.exists():
            return qlf_path
    except OSError, subprocess.SubprocessError:
        pass
    return pl_path


_WORKER_BOOT_PL = """\
% GENERATED by scripts/trace_intel.py -- warm-worker bootstrap.
% ti_query/3 takes the goal as an ATOM (read at runtime) so the swipl
% toplevel's DWIM unknown-predicate scan can never abort the wrapper
% term itself; the done-file is written unconditionally so the Python
% side always unblocks, result file present or not.
:- use_module(library(http/json)).

ti_query(GoalAtom, ResFile, DoneFile) :-
    catch(ignore(( read_term_from_atom(GoalAtom, G, [variable_names(Vs)]),
                   memberchk('R'=R, Vs),
                   call(G),
                   setup_call_cleanup(
                       open(ResFile, write, Out, [encoding(utf8)]),
                       ( json_write_dict(Out, R, [width(0)]), nl(Out) ),
                       close(Out)) )),
          _, true),
    setup_call_cleanup(open(DoneFile, write, D, []), true, close(D)).
"""


class _WarmPrologWorker:
    """One persistent ``swipl`` process holding the STANDARD repo KB.

    The KB (codebase_db.pl + the two generated fact files +
    traceability_db.pl) is immutable between ``extract``/reindex runs,
    so consulting it once and streaming goals over stdin removes the
    per-query consult cost and lets SLG tables persist across queries.
    The worker is keyed on ``(path, mtime_ns, size)`` of every KB file:
    any change restarts it, so a stale KB can never be queried. Any
    anomaly (worker death, missing result, unreadable KB) makes
    ``query`` return ``None`` and the caller falls back to the one-shot
    subprocess path, whose semantics are unchanged.
    """

    def __init__(self) -> None:
        self._proc: subprocess.Popen[str] | None = None
        self._key: tuple[tuple[str, int, int], ...] | None = None
        self._lock = threading.Lock()

    @staticmethod
    def _kb_key(kb_files: list[Path]) -> tuple[tuple[str, int, int], ...]:
        return tuple((str(p), p.stat().st_mtime_ns, p.stat().st_size) for p in kb_files)

    def _start(self, kb_files: list[Path], key: tuple) -> None:
        self._stop_locked()
        SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
        boot_path = SCRATCH_DIR / "ti_worker_boot.pl"
        boot_path.write_text(_WORKER_BOOT_PL, encoding="utf-8")
        load_paths = [str(_qlf_for(p)) for p in kb_files]
        with open(SCRATCH_DIR / "ti_worker.log", "w", encoding="utf-8") as log:
            self._proc = subprocess.Popen(
                [
                    "swipl",
                    "-q",
                    f"--table-space={_SWIPL_TABLE_SPACE}",
                    str(boot_path),
                    *load_paths,
                ],
                stdin=subprocess.PIPE,
                stdout=log,
                stderr=subprocess.STDOUT,
                text=True,
            )
        self._key = key

    def _stop_locked(self) -> None:
        if self._proc is not None:
            with contextlib.suppress(OSError):
                self._proc.kill()
            with contextlib.suppress(Exception):
                self._proc.wait(timeout=10)
            self._proc = None
        self._key = None

    def shutdown(self) -> None:
        with self._lock:
            self._stop_locked()

    def query(self, goal: str, kb_files: list[Path]) -> dict[str, Any] | None:
        """Run ``goal`` on the warm worker; ``None`` => use the one-shot path."""
        with self._lock:
            try:
                key = self._kb_key(kb_files)
            except OSError:
                return None
            if self._proc is None or self._proc.poll() is not None or key != self._key:
                try:
                    self._start(kb_files, key)
                except OSError:
                    self._stop_locked()
                    return None
            return self._query_locked(goal)

    def _query_locked(self, goal: str) -> dict[str, Any] | None:
        if self._proc is None or self._proc.stdin is None:
            return None
        result_fd, result_path = tempfile.mkstemp(suffix=".json", dir=str(SCRATCH_DIR))
        os.close(result_fd)
        done_fd, done_path = tempfile.mkstemp(suffix=".done", dir=str(SCRATCH_DIR))
        os.close(done_fd)
        for path in (result_path, done_path):
            with contextlib.suppress(OSError):
                os.unlink(path)
        try:
            escaped_goal = goal.replace("\\", "\\\\").replace("'", "''")
            term = f"ti_query('{escaped_goal}', '{result_path}', '{done_path}').\n"
            self._proc.stdin.write(term)
            self._proc.stdin.flush()
            deadline = time.monotonic() + _PROLOG_TIMEOUT_SECONDS
            while not os.path.exists(done_path):
                if self._proc.poll() is not None:
                    self._stop_locked()
                    return None
                if time.monotonic() > deadline:
                    self._stop_locked()
                    return {"error": "timeout"}
                time.sleep(0.005)
            if os.path.exists(result_path):
                with open(result_path, encoding="utf-8") as fh:
                    content = fh.read().strip()
                if content:
                    parsed = json.loads(content)
                    if isinstance(parsed, dict):
                        return parsed
                    return {"error": "json_not_object", "raw": content[:500]}
            return None
        except OSError, ValueError:
            self._stop_locked()
            return None
        finally:
            for path in (result_path, done_path):
                with contextlib.suppress(OSError):
                    os.unlink(path)


# The tests load this script FRESH per test via importlib, so a plain
# module global would silently fork one worker (and one KB consult) per
# test; anchoring the singleton in sys.modules keeps ONE warm engine per
# process. Benign creation race accepted: construction spawns nothing
# (the swipl process starts lazily on first query).
_WORKER_REGISTRY_MODULE = "trace_intel__warm_worker"


def _get_worker() -> _WarmPrologWorker:
    registry = sys.modules.setdefault(
        _WORKER_REGISTRY_MODULE, types.ModuleType(_WORKER_REGISTRY_MODULE)
    )
    worker = getattr(registry, "worker", None)
    if worker is None:
        worker = _WarmPrologWorker()
        setattr(registry, "worker", worker)  # noqa: B010 -- dynamic module attr
        atexit.register(worker.shutdown)
    return worker


def _run_prolog(goal: str, kb_files: list[Path]) -> dict[str, Any]:
    """Run ``goal`` against ``kb_files``: the standard repo KB goes to
    the warm persistent worker (falling back to one-shot on any
    anomaly); every non-standard KB list (fixtures, empty) always uses
    the one-shot subprocess with unchanged semantics."""
    if kb_files == _kb_files() and all(p.exists() for p in kb_files):
        result = _get_worker().query(goal, kb_files)
        if result is not None:
            return result
    return _run_prolog_oneshot(goal, kb_files)


def _run_prolog_oneshot(goal: str, kb_files: list[Path]) -> dict[str, Any]:
    """Consult every file in ``kb_files`` (in order) then run ``goal`` in
    one standalone ``swipl`` subprocess, returning its ``R`` binding
    (a ``json{}`` dict written via ``json_write_dict``) as a Python dict.

    Mirrors ``CodebaseTools._run_subprocess``'s query/result scratch-file
    protocol (own tempfiles, unlinked result file so a stale prior run can
    never be read back) but generalized to consult N files, not just
    (rules, facts).
    """
    SCRATCH_DIR.mkdir(parents=True, exist_ok=True)
    query_fd, query_path = tempfile.mkstemp(suffix=".pl", dir=str(SCRATCH_DIR))
    os.close(query_fd)
    result_fd, result_path = tempfile.mkstemp(suffix=".json", dir=str(SCRATCH_DIR))
    os.close(result_fd)
    with contextlib.suppress(OSError):
        os.unlink(result_path)

    try:
        with open(query_path, "w", encoding="utf-8") as fh:
            # json_write_dict lives in library(http/json); the warm worker
            # declares it explicitly (_WORKER_BOOT_PL) and the one-shot file
            # must too — autoload does not resolve it on every SWI build
            # (guix swipl 9.3.8: "Unknown procedure: json_write_dict/3").
            fh.write(":- use_module(library(http/json)).\n")
            for kb_file in kb_files:
                escaped = str(_qlf_for(kb_file)).replace("'", "''")
                fh.write(f":- consult('{escaped}').\n")
            escaped_result = result_path.replace("'", "''")
            fh.write(f"run :- {goal}, ")
            fh.write("setup_call_cleanup(")
            fh.write(f"open('{escaped_result}', write, Out, [encoding(utf8)]),")
            fh.write("(json_write_dict(Out, R, [width(0)]), nl(Out)),")
            fh.write("close(Out)).\n")

        cmd = [
            "swipl",
            "-q",
            f"--table-space={_SWIPL_TABLE_SPACE}",
            "-g",
            "run",
            "-g",
            "halt",
            query_path,
        ]
        proc = subprocess.run(
            cmd, capture_output=True, text=True, timeout=_PROLOG_TIMEOUT_SECONDS
        )

        if os.path.exists(result_path):
            with open(result_path, encoding="utf-8") as fh:
                content = fh.read().strip()
            if content:
                parsed = json.loads(content)
                if isinstance(parsed, dict):
                    return parsed
                return {"error": "json_not_object", "raw": content[:500]}

        return {
            "error": "prolog_failed",
            "returncode": proc.returncode,
            "stderr": proc.stderr[:1000],
            "stdout": proc.stdout[:1000],
        }
    except subprocess.TimeoutExpired:
        return {"error": "timeout"}
    finally:
        with contextlib.suppress(OSError):
            os.unlink(query_path)
        with contextlib.suppress(OSError):
            os.unlink(result_path)


def _needs_refresh(artifact: Path, *sources: Path) -> bool:
    """True when ``artifact`` is missing or older than ANY of ``sources``."""
    if not artifact.exists():
        return True
    stamp = artifact.stat().st_mtime
    return any(s.exists() and stamp < s.stat().st_mtime for s in sources)


def _traceability_sources() -> list[Path]:
    """Every input ``extract_facts`` reads, for the staleness gate.

    ``traceability_facts.pl`` used to be gated on ``plans/ast-ultimate.json``
    alone, but ``extract_facts`` also reads the requirement checklists under
    ``specs/`` and the parser module that reads them. Editing a checklist --
    or the parser -- therefore left the cached facts on disk and every
    consumer (``derive_rtm``, ``trace_intel`` queries) silently answered from
    the pre-edit KB. That is a false-number generator: a measurement taken
    right after such an edit reports the state BEFORE it, which is
    indistinguishable from "the edit had no effect".
    """
    sources = [AST_ULTIMATE_JSON, _PARSE_SPEC_REQ_PATH]
    if SPECS_ROOT.is_dir():
        sources += sorted(SPECS_ROOT.glob("*/requirements.md"))
        sources += sorted(SPECS_ROOT.glob("*/checklists/requirements.md"))
    return sources


#: The UNTRACKED live snapshot. ``plans/ast-ultimate.json`` is a committed
#: artifact and nothing on a query path may write it -- an always-on writer
#: kept it permanently dirty and churned it into unrelated commits
#: (DESIGN_WAIVER 2026-08-09, backlog #284), and one refresh was measured
#: modifying two tracked files again on 2026-08-22. So the query path keeps
#: its own copy under ``.cache/`` (gitignored), seeded from the committed one.
LIVE_SNAPSHOT_DIR = REPO_ROOT / ".cache" / "ast"
LIVE_SNAPSHOT = LIVE_SNAPSHOT_DIR / "ast-ultimate.json"
_AST_SYNC = REPO_ROOT / ".venv" / "bin" / "ast-sync"
_RUNTIME_CALLGRAPH = REPO_ROOT / "plans" / "runtime-callgraph.json"


def _active_snapshot() -> Path:
    """The snapshot the structural KB is reindexed from.

    The live one whenever it exists; the committed one is the cold-start seed
    and the CI artifact, and is only ever as fresh as the last gate run.
    """
    return LIVE_SNAPSHOT if LIVE_SNAPSHOT.exists() else AST_ULTIMATE_JSON


def _newest_source_mtime() -> float:
    """Newest mtime across the trees the snapshot is serialized from.

    Imported from ``ast_pipeline`` rather than restated: that module is the
    one place that knows which roots are in the snapshot (2977 modules across
    src/, scripts/, tests/ and claude/), and a second copy of that list here
    would drift and silently stop refreshing whichever tree it forgot.
    """
    from orchestrata_ultimate.tools.ast_pipeline import source_trees

    newest = 0.0
    for tree in source_trees():
        if not tree.exists():
            continue
        for path in tree.rglob("*.py"):
            mtime = path.stat().st_mtime
            if mtime > newest:
                newest = mtime
    return newest


def _refresh_snapshot() -> None:
    """Bring the LIVE snapshot up to date with the working tree.

    This is the leg that was missing. ``ensure_kb`` refreshed the fact files
    FROM the snapshot, but nothing refreshed the snapshot FROM disk, so an
    edit left the entire chain internally consistent with bytes that no
    longer existed -- and re-running the query could not fix it. The sidekick
    could only refuse and name a command for a human to run, which is exactly
    the manual step this removes.

    Cheap check, expensive fix. Stat-walking the source trees costs
    milliseconds; the refresh itself was measured at ~12s warm (reusing 2976
    of 2977 modules), which is fine on demand and unaffordable per query.

    ``runtime_calls`` is re-applied afterwards, always. ``serialize`` knows
    nothing about that field, so a bare re-serialize silently strips the
    merged static+runtime call graph that ``derive_rtm``, ``code_intel``'s
    merged queries and ``trace-path`` depend on -- measured twice, and the
    reason gates.py pairs the two steps in a single function.

    Failure is deliberately NOT fatal: an unrefreshed snapshot stays stale,
    ``kb_module_changed/2`` keeps reporting it, and the sidekick still answers
    INVALID-with-names (SPEC-114 FR4-003). Raising here would replace a
    precise refusal with a crash, and swallowing it would answer over stale
    bytes -- the very thing FR4-003 exists to prevent.
    """
    if not _AST_SYNC.exists():
        return
    snapshot = _active_snapshot()
    if snapshot.exists() and snapshot.stat().st_mtime >= _newest_source_mtime():
        return
    print(
        "trace_intel: source tree is newer than the AST snapshot -- "
        "refreshing incrementally",
        file=sys.stderr,
    )
    synced = subprocess.run(
        [str(_AST_SYNC), "--output-dir", str(LIVE_SNAPSHOT_DIR)],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
    )
    if synced.returncode != 0:
        print(
            f"trace_intel: snapshot refresh FAILED ({synced.stderr.strip()[-400:]}) "
            "-- answering from the previous snapshot, which stays reportable as "
            "stale",
            file=sys.stderr,
        )
        return
    if _RUNTIME_CALLGRAPH.exists():
        subprocess.run(
            [
                sys.executable,
                str(REPO_ROOT / "scripts" / "build_runtime_callgraph.py"),
                "reaugment",
                "--ast",
                str(LIVE_SNAPSHOT),
                "--from",
                str(_RUNTIME_CALLGRAPH),
            ],
            cwd=REPO_ROOT,
            capture_output=True,
            text=True,
        )


def ensure_kb(no_refresh: bool) -> None:
    """Auto-run ``extract`` + reindex the structural KB when either
    artifact is missing or older than any of its inputs.

    ``--no-refresh`` skips this entirely and queries whatever is already
    on disk under ``.cache/trace_intel/`` (raising if nothing is there).
    """
    if not no_refresh and _needs_refresh(TRACEABILITY_FACTS, *_traceability_sources()):
        print(
            f"trace_intel: {TRACEABILITY_FACTS} missing/stale -- running extract",
            file=sys.stderr,
        )
        extract_facts()
    if not TRACEABILITY_FACTS.exists():
        raise SystemExit(
            f"trace_intel: {TRACEABILITY_FACTS} does not exist -- run "
            "`extract` first, or drop --no-refresh"
        )

    if not no_refresh:
        _refresh_snapshot()
    snapshot = _active_snapshot()
    if not no_refresh and _needs_refresh(STRUCTURAL_FACTS, snapshot):
        print(
            f"trace_intel: {STRUCTURAL_FACTS} missing/stale -- reindexing structural KB",
            file=sys.stderr,
        )
        tools = CodebaseTools(facts_file=str(STRUCTURAL_FACTS), transport="subprocess")
        result = tools.reindex(str(snapshot))
        if "error" in result:
            raise SystemExit(f"trace_intel: reindex failed: {result}")
    if not STRUCTURAL_FACTS.exists():
        raise SystemExit(
            f"trace_intel: {STRUCTURAL_FACTS} does not exist -- reindex failed "
            "silently, or --no-refresh was passed before a first run"
        )


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _print_result(result: dict[str, Any], json_mode: bool) -> None:
    if json_mode:
        print(json.dumps(result, indent=2, sort_keys=True))
        return
    if "error" in result:
        print(f"error: {result}", file=sys.stderr)
        return
    for key, value in result.items():
        if isinstance(value, list):
            print(f"{key}: ({len(value)})")
            for item in value:
                print(f"  - {item}")
        else:
            print(f"{key}: {value}")


# static-key, merged-key, display-label triples for the --merged comparison
# rows -- tool_report_merged/1's json{} carries exactly these six counters
# plus runtime_edges (see traceability_db.pl).
_MERGED_ROWS: tuple[tuple[str, str, str], ...] = (
    ("orphans_static", "orphans_merged", "orphans"),
    ("unspecced_static", "unspecced_merged", "unspecced"),
    ("specless_static", "specless_merged", "specless_usecases"),
)


def _print_merged_result(result: dict[str, Any]) -> None:
    """Text-mode renderer for ``report --merged``: a static-vs-merged
    comparison (each row's static count, merged count, and delta) instead
    of ``_print_result``'s flat key/value dump -- the default (no-flag)
    ``report`` output is untouched."""
    if "error" in result:
        print(f"error: {result}", file=sys.stderr)
        return
    print(f"runtime_edges: {result.get('runtime_edges', 0)}")
    for static_key, merged_key, label in _MERGED_ROWS:
        static_v = result.get(static_key, 0)
        merged_v = result.get(merged_key, 0)
        delta = merged_v - static_v
        sign = "+" if delta >= 0 else ""
        print(f"{label}: static={static_v} merged={merged_v} delta={sign}{delta}")


def _cmd_report(args: argparse.Namespace) -> dict[str, Any]:
    goal = (
        "tool_report_merged(R)" if getattr(args, "merged", False) else "tool_report(R)"
    )
    return _run_prolog(goal, _kb_files())


def _cmd_narrative(args: argparse.Namespace) -> dict[str, Any]:
    goal = f"tool_narrative({_pl_atom(args.qname)}, R)"
    return _run_prolog(goal, _kb_files())


def _cmd_orphans(args: argparse.Namespace) -> dict[str, Any]:
    return _run_prolog(f"tool_orphans({int(args.limit)}, R)", _kb_files())


def _cmd_untraced_reqs(args: argparse.Namespace) -> dict[str, Any]:
    return _run_prolog(f"tool_untraced_reqs({int(args.limit)}, R)", _kb_files())


def _cmd_unspecced(args: argparse.Namespace) -> dict[str, Any]:
    return _run_prolog(f"tool_unspecced({int(args.limit)}, R)", _kb_files())


def _cmd_specless_usecases(args: argparse.Namespace) -> dict[str, Any]:
    return _run_prolog(f"tool_specless_usecases({int(args.limit)}, R)", _kb_files())


def _cmd_why_not(args: argparse.Namespace) -> dict[str, Any]:
    """SPEC-114 FR-004 — why isn't this requirement implemented?"""
    goal = f"tool_why_not_req({_pl_atom(args.req_id)}, R)"
    return _run_prolog(goal, _kb_files())


_QUERY_DISPATCH: dict[str, Callable[[argparse.Namespace], dict[str, Any]]] = {
    "report": _cmd_report,
    "narrative": _cmd_narrative,
    "orphans": _cmd_orphans,
    "untraced-reqs": _cmd_untraced_reqs,
    "unspecced": _cmd_unspecced,
    "specless-usecases": _cmd_specless_usecases,
    "why-not": _cmd_why_not,
}


def _add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--json", action="store_true", help="Print raw JSON, not text.")
    parser.add_argument(
        "--no-refresh",
        action="store_true",
        help="Skip the auto extract/reindex freshness check; query the KB as-is.",
    )


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="trace_intel",
        description=(
            "T1 traceability / use-case-narrative engine -- semantic Prolog "
            "layer over the structural code-intel KB."
        ),
    )
    sub = parser.add_subparsers(dest="command", required=True)

    sub.add_parser(
        "extract",
        help="Regenerate traceability_facts.pl from plans/ast-ultimate.json + specs/.",
    )

    report_p = sub.add_parser(
        "report", help="Run all five finders and print their counts."
    )
    report_p.add_argument(
        "--merged",
        action="store_true",
        help=(
            "Compare static vs merged (static+runtime) reachability via "
            "tool_report_merged/1 instead of the default tool_report/1."
        ),
    )
    _add_common_args(report_p)

    narrative_p = sub.add_parser(
        "narrative", help="Modules transitively reachable from one entrypoint."
    )
    narrative_p.add_argument(
        "qname", help="Entrypoint qualified name (see `report`'s entrypoints count)."
    )
    _add_common_args(narrative_p)

    limited = [
        ("orphans", "Traced code unreachable from any known entrypoint."),
        ("untraced-reqs", "Spec requirements with zero @traces coverage."),
        ("unspecced", "Code reachable from an entrypoint but carrying no @traces."),
        (
            "specless-usecases",
            "Specs whose requirements never chain to reachable code.",
        ),
    ]
    for name, help_text in limited:
        p = sub.add_parser(name, help=help_text)
        p.add_argument(
            "--limit", type=int, default=50, help="Max items to list (default 50)."
        )
        _add_common_args(p)

    why_not_p = sub.add_parser(
        "why-not",
        help="Why isn't requirement X implemented? Names the first missing conjunct.",
    )
    why_not_p.add_argument(
        "req_id", help="Requirement id, e.g. REQ-000-004 (see `untraced-reqs`)."
    )
    _add_common_args(why_not_p)

    return parser


def main(argv: list[str] | None = None) -> int:
    """Parse ``argv`` and dispatch to the matching subcommand handler."""
    parser = _build_parser()
    args = parser.parse_args(argv)

    if args.command == "extract":
        return cmd_extract(args)

    ensure_kb(no_refresh=args.no_refresh)
    result = _QUERY_DISPATCH[args.command](args)
    if args.command == "report" and getattr(args, "merged", False) and not args.json:
        _print_merged_result(result)
    else:
        _print_result(result, json_mode=args.json)

    # REQ-114-FR4-003: three-valued, matching the gate layer's own contract
    # (`gate run` exits 0 pass / 1 fail / 2 invalid). `invalid` is NOT an
    # error in the query — the query was well-formed and the requirement may
    # be perfectly fine; the FACT BASE could not answer for the bytes on
    # disk. Collapsing it into 0 is the false-green freshness.pl was written
    # against; collapsing it into 1 would read as "this requirement failed".
    if result.get("outcome") == "invalid":
        return 2
    return 1 if "error" in result else 0


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