orchestrata_ultimate.tools.codebase_tools

Reference › orchestrata_ultimate › tools › codebase_tools

This module is the LLM-facing layer over the Prolog codebase database, codebase_db.pl. CodebaseTools offers one public method for each tool predicate, and the docstrings double as agent-facing tool descriptions. The registry wrappers are in codebase_query_tools; the module also depends on core.janus_bootstrap. Tests are tests/tools/test_codebase_tools.py, tests/tools/test_codebase_query_tools.py and tests/scripts/test_trace_intel.py.

Two transports are chosen per instance. The default janus transport runs each query in-process through the free-threaded janus-swi bridge against one global SWI engine. Its constructor raises an actionable error if janus is unavailable or the GIL is enabled, and it does not fall back. _normalize_janus_value reconciles janus output with the JSON shape of the other transport. The subprocess transport runs standalone swipl per query with unique scratch files, and is only used when requested. Both build the same goal string and return the same dict shape.

Lifecycle: reindex rebuilds the knowledge base from a CodebaseAST JSON file, writing the facts file atomically, and attach reuses an already-indexed facts file. Queries cover stats, find_symbol, freshness, affected_symbols, affected_tests, literals, symbol_context, impact, callers and callees (static and merged), trace_path, module_deps, dep_path, cycles, externals, unused_symbols, hotspots, by_prop, signature_search and module_api.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

CodebaseTools

Line 168

LLM-facing wrapper over the Prolog codebase-intelligence database.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/tools/codebase_tools.py

"""codebase_tools.py — LLM-facing tool layer over the Prolog codebase database.

Every public method corresponds 1:1 to a tool predicate in codebase_db.pl.
Docstrings are written to double as agent-facing tool descriptions.

Two query transports, selected per-instance via ``transport=``:

- ``"janus"`` (the default) runs every query IN-PROCESS through the project's
  patched free-threaded janus-swi (SPEC-023) against a GLOBAL SWI engine --
  no subprocess/consult overhead per call (~1-2ms warm vs ~1.2s for a
  standalone-swipl round trip; see PROLOG_STATUS.md for measured numbers).
  Per project doctrine (the same fail-loud posture as the kernel-required
  gate), the constructor RAISES if janus_swi is unavailable or the process's
  GIL is enabled, rather than silently falling back -- the patched
  free-threaded janus is a declared dependency built by ``uv sync``, so its
  absence means the environment is broken, not a reason to degrade. See
  ``docs/free-threaded-prolog.md``'s "CodebaseTools wiring" section.
- ``"subprocess"`` is the original standalone-``swipl``-per-query path: after
  ``reindex()``, facts are saved to a per-instance facts file and each query
  gets its own scratch query/result file (via ``tempfile.mkstemp``) so
  concurrent instances never collide and a killed process can never leave a
  stale result behind. It is never selected implicitly -- callers opt in
  explicitly.

``reindex()`` dispatches on ``self.transport`` like every other method (see
its docstring). Historically it always shelled out to ``swipl`` regardless
of ``self.transport``, because ``library(json)`` (needed by
``load_codebase/2`` to parse the incoming AST JSON) failed to LOAD under
janus with a ``PL_release_stream`` ABI error -- see
``core/janus_bootstrap.py`` and ``docs/free-threaded-prolog.md``. That gap
is now closed by an ``RTLD_GLOBAL`` preload of libswipl performed before
``janus_swi`` is ever imported, so ``transport="janus"`` instances now
reindex in-process too.

Both transports build the IDENTICAL Prolog goal string per query (each
public method's ``goal = ...`` line is the single source of truth) and
return the identical dict shape -- see ``_normalize_janus_value`` for the
one place janus's native SWI-term rendering has to be reconciled with
``json_write_dict``'s (the atoms ``true``/``false``/``null`` render as the
Python strings ``'true'``/``'false'``/``'null'`` under janus, vs. real
``True``/``False``/``None`` after ``json.loads`` on the subprocess path).
"""

import contextlib
import json
import os
import subprocess
import sys
import tempfile
import threading
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# janus transport -- process-global state.
#
# The SWI database (dynamic predicates, tables) consulted via janus is
# shared by every thread's engine in this process -- there is exactly ONE
# live (rules, facts) pair at a time. _JANUS_LOCK serializes every
# consult/load_facts/query_once so two CodebaseTools instances pointed at
# different facts can never interleave and silently answer each other's
# queries; _janus_loaded records which pair is currently live so a query
# against an already-loaded KB is a single query_once with no reconsult.
# ---------------------------------------------------------------------------

_JANUS_LOCK = threading.RLock()
# janus_swi ships no type stubs, so this is typed Any (matching the existing
# convention in core/prolog_adapter.py) rather than left implicitly None,
# which would make pyright reject every `.consult`/`.query_once` access.
_janus_module: Any = None
_janus_import_error: RuntimeError | None = None
# (rules_path, facts_path) currently consulted into the global janus engine,
# or None before the first janus query in this process.
_janus_loaded: tuple[str, str] | None = None


def _import_janus() -> Any:
    """Lazily import + cache ``janus_swi``, or raise an actionable RuntimeError.

    The patched free-threaded janus-swi is a declared dependency
    (``pyproject.toml``'s ``janus-swi`` entry, built via
    ``[tool.uv.sources]``'s ``no-build-isolation-package`` by ``uv sync``).
    Its absence means the environment is broken, not a reason to silently
    degrade to the subprocess transport -- the same doctrine as the
    kernel-required / no-false-green gates elsewhere in this project.
    """
    global _janus_module, _janus_import_error
    if _janus_module is not None:
        return _janus_module
    if _janus_import_error is not None:
        raise _janus_import_error
    try:
        from orchestrata_ultimate.core.janus_bootstrap import (  # noqa: PLC0415
            janus,
        )
    except ImportError as exc:
        _janus_import_error = RuntimeError(
            "transport='janus' requires the project's patched free-threaded "
            "janus-swi (a declared dependency -- see pyproject.toml's "
            "janus-swi entry and docs/free-threaded-prolog.md). It failed "
            f"to import ({exc.__class__.__name__}: {exc}). Run `uv sync` to "
            "(re)build it."
        )
        raise _janus_import_error from exc
    _janus_module = janus
    return janus


def _assert_free_threaded(gil_enabled: bool | None = None) -> None:
    """Raise if the process's GIL is enabled.

    The janus transport is only safe/meaningful under the project's
    genuinely GIL-off Python 3.14t build (SPEC-023) -- a GIL-on interpreter
    means an extension module silently re-enabled it on import somewhere in
    the process's import chain (stock janus-swi/grpcio doing exactly that is
    the historical regression ``docs/free-threaded-prolog.md`` documents and
    ``tests/conftest.py`` gates suite-wide). Fail loud rather than run
    "free-threaded" code that silently is not.

    ``gil_enabled`` is an injection seam: pass ``True``/``False`` explicitly
    to exercise both branches directly in tests without monkeypatching
    ``sys._is_gil_enabled`` (forbidden project-wide); the real call site
    (the default, ``None``) probes the actual interpreter.
    """
    if gil_enabled is None:
        gil_enabled = hasattr(sys, "_is_gil_enabled") and sys._is_gil_enabled()
    if gil_enabled:
        raise RuntimeError(
            "The GIL is enabled in this process -- transport='janus' "
            "requires genuine free-threading (SPEC-023). An extension "
            "module likely re-enabled it on import (grpc/msgpack are the "
            "known historical offenders -- see docs/free-threaded-prolog.md "
            "and tests/conftest.py's suite-wide probe). Rebuild the "
            "free-threaded toolchain and re-run."
        )


def _normalize_janus_value(value: Any) -> Any:
    """Recursively normalize a janus query result to match the subprocess
    transport's ``json.loads(json_write_dict(...))`` shape -- the parity
    contract between the two transports.

    janus converts SWI dicts/lists/atoms/strings/numbers to native Python
    dict/list/str/int/float automatically. The one place it diverges from
    ``json_write_dict``: every ``json{}`` field in codebase_db.pl that holds
    the Prolog atom ``true``/``false``/``null`` (e.g. every ``truncated``/
    ``found`` field) round-trips through ``json_write_dict`` + ``json.loads``
    as a real Python ``True``/``False``/``None``, but janus renders the SAME
    atoms as the plain strings ``'true'``/``'false'``/``'null'``. Symbol
    names can never collide with this (``true``/``false``/``null`` are not
    valid Python identifiers, so no real ``qname`` can equal them) -- the
    ambiguity this string comparison has is identical to json_write_dict's
    own (atom identity, not quoting, decides the rendering on either side).
    """
    if isinstance(value, dict):
        return {k: _normalize_janus_value(v) for k, v in value.items()}
    if isinstance(value, list):
        return [_normalize_janus_value(v) for v in value]
    if value == "true":
        return True
    if value == "false":
        return False
    if value == "null":
        return None
    return value


class CodebaseTools:
    """LLM-facing wrapper over the Prolog codebase-intelligence database.

    Each public method assembles one Prolog goal against ``codebase_db.pl``
    (consulted together with this instance's facts file) and runs it via
    ``self.transport`` ("janus" in-process, or "subprocess" via a standalone
    ``swipl`` invocation), returning the result as a dict.
    """

    def __init__(
        self,
        prolog_file: str | None = None,
        workdir: str | None = None,
        facts_file: str | None = None,
        transport: str = "janus",
    ) -> None:
        """Initialize the tool layer.

        Args:
            prolog_file: path to codebase_db.pl. Defaults to the file next
                to this module.
            workdir: directory to hold this instance's facts file and
                scratch query/result files (subprocess transport only).
                Defaults to a fresh temp directory (``codebase_kb_*``).
            facts_file: path to an already-indexed facts file to attach to,
                skipping reindex(). Defaults to ``<workdir>/facts.pl``; if
                that path already exists on disk it is auto-attached.
            transport: ``"janus"`` (default, in-process, fast -- see the
                module docstring) or ``"subprocess"`` (explicit opt-in only;
                ``reindex()`` always uses it internally regardless of this
                setting, because ``library(http/json)`` cannot load under
                janus).

        Raises:
            ValueError: ``transport`` is neither ``"janus"`` nor
                ``"subprocess"``.
            RuntimeError: ``transport="janus"`` and janus_swi is
                unavailable, or the process's GIL is enabled.
        """
        if transport not in ("janus", "subprocess"):
            raise ValueError(
                f"invalid transport {transport!r} -- must be 'janus' or 'subprocess'"
            )
        if transport == "janus":
            _import_janus()
            _assert_free_threaded()
        self.transport = transport

        if prolog_file is None:
            prolog_file = str(Path(__file__).with_name("codebase_db.pl"))
        self.prolog_file = prolog_file

        self._workdir = (
            Path(workdir) if workdir else Path(tempfile.mkdtemp(prefix="codebase_kb_"))
        )
        self._workdir.mkdir(parents=True, exist_ok=True)

        self._facts_file = (
            Path(facts_file) if facts_file else self._workdir / "facts.pl"
        )
        self._loaded = self._facts_file.exists()

    @property
    def facts_file(self) -> str:
        """Path to this instance's Prolog facts file (populated by reindex())."""
        return str(self._facts_file)

    def attach(self, facts_file: str) -> None:
        """Attach to an already-indexed facts file, skipping reindex().

        Lets a fresh ``CodebaseTools`` instance reuse a knowledge base that
        another instance (or a previous process) already built.
        """
        self._facts_file = Path(facts_file)
        self._loaded = self._facts_file.exists()

    def _atomic_facts_tmp(self) -> Path:
        """Allocate a scratch path in the SAME directory as this instance's
        facts file, for the tmp-then-``os.replace`` write pattern that
        ``_reindex_janus``/``_reindex_subprocess`` point ``save_facts/1`` at
        instead of the live path.

        ``save_facts/1`` (codebase_db.pl) opens its target with a plain
        ``open(File, write, ...)`` -- a truncate-in-place write. Callers
        impose short timeouts (``CODE_INTEL_TIMEOUT = 25`` in
        ``~/.claude/tools/module_context.py``, ``MODULE_TIMEOUT = 30`` in
        the dispatch-time context-injection hook), so a kill mid-write can
        leave ``facts.pl`` truncated -- readable enough to half-parse, wrong
        enough to answer queries with a partial KB. Same directory means
        same filesystem, so the eventual ``os.replace`` in
        ``_finalize_facts_write`` is a rename, not a copy.
        """
        self._facts_file.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp_name = tempfile.mkstemp(
            dir=str(self._facts_file.parent),
            prefix=f".{self._facts_file.name}.",
            suffix=".tmp",
        )
        os.close(fd)
        return Path(tmp_name)

    def _finalize_facts_write(self, tmp_path: Path) -> None:
        """fsync the freshly-written temp facts file, then atomically
        replace the live facts file with it.

        A crash between ``save_facts/1`` finishing and this call leaves the
        PREVIOUS facts file exactly as it was -- never a spliced/truncated
        one -- because ``os.replace`` either fully lands or doesn't happen
        at all. Mirrors ``core.atomic_json.dump_json_atomic``'s pattern
        (mode preservation included) for the same reason it exists there:
        a partial write must never be indistinguishable from a good one.
        """
        try:
            mode = self._facts_file.stat().st_mode & 0o777
        except FileNotFoundError:
            mode = 0o644
        fd = os.open(str(tmp_path), os.O_RDONLY)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)
        os.chmod(tmp_path, mode)
        os.replace(tmp_path, self._facts_file)

    def _scratch_files(self) -> tuple[str, str]:
        """Allocate a unique (query_file, result_file) pair inside workdir.

        The result file is unlinked immediately after allocation so that
        success can only be proven by THIS run's own write — no stale
        content from a prior run can ever be read back.
        """
        fd, query_file = tempfile.mkstemp(suffix=".pl", dir=str(self._workdir))
        os.close(fd)
        fd, result_file = tempfile.mkstemp(suffix=".json", dir=str(self._workdir))
        os.close(fd)
        with contextlib.suppress(OSError):
            os.unlink(result_file)
        return query_file, result_file

    def _run(self, goal: str) -> dict:
        """Run a fully-assembled Prolog goal and return the result as a dict.

        Dispatches on ``self.transport``. Every public tool method builds
        the SAME goal string regardless of transport (see each method's
        ``goal = ...`` line) — that assembly is the one source of truth both
        transports share, so they are provably answering the identical
        query.
        """
        if self.transport == "janus":
            return self._run_janus(goal)
        return self._run_subprocess(goal)

    def _ensure_janus_loaded(self, janus: Any) -> None:
        """Make the global janus engine's live (rules, facts) match this
        instance's, swapping via ``load_facts/1`` (codebase_db.pl) if not.

        Caller MUST hold ``_JANUS_LOCK`` — the SWI database is process-global
        (shared by every thread's engine), so an unswapped state here would
        silently answer one instance's query with another instance's facts.
        """
        global _janus_loaded
        target = (self.prolog_file, str(self._facts_file))
        if _janus_loaded == target:
            return
        if _janus_loaded is None or _janus_loaded[0] != self.prolog_file:
            janus.consult(self.prolog_file)
        janus.query_once(f"load_facts({self._prolog_val(str(self._facts_file))})")
        _janus_loaded = target

    def _invalidate_janus_cache(self) -> None:
        """Drop the module-level janus loaded-state record if it currently
        points at THIS instance's facts file — called after reindex()
        rewrites it, so the next janus query re-consults instead of serving
        stale (pre-reindex) facts. reindex() always runs via subprocess
        regardless of ``self.transport`` (see reindex()'s docstring), so
        this is the only hook that needs to know about janus staleness.
        Called unconditionally (regardless of ``self.transport``): the janus
        cache is process-global and another instance may share this exact
        facts_file via ``attach()``.
        """
        global _janus_loaded
        with _JANUS_LOCK:
            if _janus_loaded is not None and _janus_loaded[1] == str(self._facts_file):
                _janus_loaded = None

    def _run_janus(self, goal: str) -> dict:
        """Run ``goal`` in-process via janus against the global SWI engine.

        Swaps in this instance's (rules, facts) first if needed (see
        ``_ensure_janus_loaded``), all under ``_JANUS_LOCK`` so the swap and
        the query are atomic from every other thread's perspective. Never
        raises to the caller — a janus failure (bad goal, corrupt facts,
        engine error) is caught and returned in the same structured-error
        shape ``_run_subprocess`` uses, so error handling cannot tell which
        transport is behind an instance.
        """
        janus = _import_janus()
        try:
            with _JANUS_LOCK:
                self._ensure_janus_loaded(janus)
                result = janus.query_once(goal)
        except Exception as e:
            return {"error": "exception", "details": str(e)}

        if result is None or not result.get("truth"):
            return {"error": "prolog_failed", "details": "goal did not succeed"}
        r = result.get("R")
        if not isinstance(r, dict):
            return {"error": "json_not_object", "raw": str(r)[:500]}
        normalized: dict = _normalize_janus_value(r)
        return normalized

    def _run_subprocess(self, goal: str) -> dict:
        """Run a fully-assembled Prolog goal via standalone swipl and return JSON."""
        escaped_rules = self.prolog_file.replace("'", "''")
        query_file, result_file = self._scratch_files()
        escaped_result = result_file.replace("'", "''")

        try:
            with open(query_file, "w", encoding="utf-8") as f:
                f.write(f":- consult('{escaped_rules}').\n")
                if self._facts_file.exists():
                    escaped_facts = str(self._facts_file).replace("'", "''")
                    f.write(f":- consult('{escaped_facts}').\n")
                f.write(f"run :- {goal}, ")
                # write_json_result/2 (codebase_db.pl) owns the json-library
                # load + json_write_dict marshaling -- a standalone swipl on
                # the query path never passes through load_codebase/2, and
                # autoload of json_write_dict/3 is not guaranteed (Guix 9.3.8
                # does not index library/ext/http).
                f.write(f"write_json_result('{escaped_result}', R).\n")

            cmd = ["swipl", "-q", "-g", "run", "-g", "halt", query_file]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)

            # Read the result file if THIS run wrote one
            # (swipl may exit 1 even on success if choicepoints remain).
            parsed = self._read_result(result_file)
            if parsed is not None:
                return parsed

            return {
                "error": "prolog_failed",
                "returncode": result.returncode,
                "stderr": result.stderr[:500],
                "stdout": result.stdout[:500],
            }

        except subprocess.TimeoutExpired:
            return {"error": "timeout"}
        except OSError as e:
            return {"error": "exception", "details": str(e)}
        finally:
            with contextlib.suppress(OSError):
                os.unlink(query_file)
            with contextlib.suppress(OSError):
                os.unlink(result_file)

    def _read_result(self, result_file: str) -> dict | None:
        """Read and parse THIS run's result file, if the goal wrote one."""
        if not os.path.exists(result_file):
            return None
        with open(result_file, encoding="utf-8") as f:
            content = f.read().strip()
        if not content:
            return None
        return self._parse_swipl_json(content)

    def _parse_swipl_json(self, content: str) -> dict:
        """Parse SWI-Prolog json_write_dict output (valid JSON) into a dict."""
        try:
            parsed = json.loads(content)
        except json.JSONDecodeError as e:
            return {
                "error": "json_parse_failed",
                "details": str(e),
                "raw": content[:500],
            }
        if isinstance(parsed, dict):
            return parsed
        return {"error": "json_not_object", "raw": content[:500]}

    @staticmethod
    def _prolog_val(v: object) -> str:
        """Convert a Python scalar to a Prolog literal for goal assembly."""
        if isinstance(v, str):
            escaped = v.replace("'", "''")
            return f"'{escaped}'"
        if isinstance(v, bool):
            return "true" if v else "false"
        if v is None:
            return "null"
        return str(v)

    # ------------------------------------------------------------------
    # lifecycle
    # ------------------------------------------------------------------
    def reindex(self, ast_json_path: str) -> dict:
        """Rebuild the knowledge base from a CodebaseAST JSON file.

        Call after serialize_codebase_ast.py regenerates the AST, or whenever
        source files have changed. Returns {"warnings": [...]} — non-empty warnings
        mean the loader's symbol counts disagree with the serializer totals.

        Dispatches on ``self.transport``, same as every query method. Until
        SPEC-023's ``core/janus_bootstrap.py`` closed the
        ``PL_release_stream`` ABI gap (see ``docs/free-threaded-prolog.md``),
        ``load_codebase/2`` could not run under janus at all -- this method
        always shelled out to a standalone ``swipl`` regardless of
        ``self.transport``. With the RTLD_GLOBAL preload in place,
        ``library(json)`` loads and runs in-process, so ``transport="janus"``
        instances now reindex in-process too (verified: in-process and
        subprocess reindex of the same AST JSON produce byte-identical
        ``stats()`` dicts). ``transport="subprocess"`` instances are
        unaffected -- see ``_reindex_subprocess``.
        """
        if self.transport == "janus":
            return self._reindex_janus(ast_json_path)
        return self._reindex_subprocess(ast_json_path)

    def _reindex_janus(self, ast_json_path: str) -> dict:
        """In-process reindex via janus -- ``load_codebase/2`` populates the
        process-global SWI database directly, then ``save_facts/1`` persists
        it to this instance's facts file (so ``attach()``/the subprocess
        transport can still pick it up later). Runs under ``_JANUS_LOCK`` so
        no other instance's query can interleave with a half-loaded KB.

        On success, the module-level janus-loaded cache is set directly to
        ``(self.prolog_file, self._facts_file)`` rather than invalidated --
        the facts are already live in the engine from ``load_codebase/2``,
        so the next query does not need to re-consult them from disk.

        ``save_facts/1`` is pointed at a scratch tmp path (same directory as
        the live facts file), not the live path itself -- a killed goal
        (``MODULE_TIMEOUT``/``CODE_INTEL_TIMEOUT`` callers) then leaves the
        tmp file torn and the live facts file untouched, instead of the
        other way around. See ``_atomic_facts_tmp``/``_finalize_facts_write``.

        ``load_codebase/2`` (codebase_db.pl) calls ``retract_facts`` as its
        FIRST action, unconditionally -- whether or not the load that
        follows then succeeds. So the moment this instance owns the live
        janus cache, the goal below is about to empty the live engine
        regardless of outcome; ``_invalidate_janus_cache()`` drops that
        ownership BEFORE the goal runs (harmless if it doesn't own it) so a
        FAILED reindex leaves the cache correctly pointing at nothing,
        rather than still claiming the (untouched, on-disk) facts file
        matches an engine that was just emptied. Re-established on success
        below once the finalized file agrees with the freshly-loaded engine.
        """
        global _janus_loaded
        janus = _import_janus()
        escaped_path = ast_json_path.replace("'", "''")
        tmp_path = self._atomic_facts_tmp()
        escaped_facts = str(tmp_path).replace("'", "''")
        goal = f"load_codebase('{escaped_path}', R), save_facts('{escaped_facts}')"
        try:
            with _JANUS_LOCK:
                if _janus_loaded is None or _janus_loaded[0] != self.prolog_file:
                    janus.consult(self.prolog_file)
                self._invalidate_janus_cache()
                result = janus.query_once(goal)
                if result is None or not result.get("truth"):
                    return {
                        "error": "reindex_failed",
                        "details": "goal did not succeed",
                    }
                r = result.get("R")
                if not isinstance(r, dict):
                    return {"error": "json_not_object", "raw": str(r)[:500]}
                self._finalize_facts_write(tmp_path)
                self._loaded = self._facts_file.exists()
                _janus_loaded = (self.prolog_file, str(self._facts_file))
                normalized: dict = _normalize_janus_value(r)
                return normalized
        except Exception as e:  # janus_swi.PrologError has no stable public type
            return {"error": "reindex_exception", "details": str(e)}
        finally:
            with contextlib.suppress(OSError):
                tmp_path.unlink(missing_ok=True)

    def _reindex_subprocess(self, ast_json_path: str) -> dict:
        """Rebuild the knowledge base via a standalone ``swipl`` subprocess.

        Reached from ``reindex()`` for ``transport="subprocess"`` instances,
        or directly by a caller that wants the subprocess path explicitly.
        On success the module-level janus-loaded cache is invalidated
        (unconditionally -- the janus cache is process-global and another
        instance may share this exact facts_file via ``attach()``) so the
        next janus query re-consults instead of serving stale facts.

        Same atomic-write hardening as ``_reindex_janus``: ``save_facts/1``
        targets a scratch tmp path (same directory as the live facts file),
        finalized via ``os.replace`` only once ``swipl`` has fully written
        the result file -- which the goal's conjunction cannot reach until
        ``save_facts`` itself has finished and closed the tmp file. A
        ``subprocess.run`` timeout killing ``swipl`` mid-write leaves the
        tmp file torn (cleaned up below) and the live facts file untouched.
        """
        escaped_path = ast_json_path.replace("'", "''")
        escaped_rules = self.prolog_file.replace("'", "''")
        tmp_path = self._atomic_facts_tmp()
        escaped_facts = str(tmp_path).replace("'", "''")
        query_file, result_file = self._scratch_files()
        escaped_result = result_file.replace("'", "''")

        try:
            with open(query_file, "w", encoding="utf-8") as f:
                f.write(f":- consult('{escaped_rules}').\n")
                f.write(f"run :- load_codebase('{escaped_path}', R), ")
                f.write(f"save_facts('{escaped_facts}'), ")
                f.write(f"write_json_result('{escaped_result}', R).\n")

            cmd = ["swipl", "-q", "-g", "run", "-g", "halt", query_file]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)

            parsed = self._read_result(result_file)
            if parsed is not None:
                self._finalize_facts_write(tmp_path)
                self._loaded = self._facts_file.exists()
                self._invalidate_janus_cache()
                return parsed

            return {
                "error": "reindex_failed",
                "returncode": result.returncode,
                "stderr": result.stderr[:500],
                "stdout": result.stdout[:500],
            }
        except subprocess.TimeoutExpired:
            return {"error": "timeout"}
        except OSError as e:
            return {"error": "reindex_exception", "details": str(e)}
        finally:
            with contextlib.suppress(OSError):
                os.unlink(query_file)
            with contextlib.suppress(OSError):
                os.unlink(result_file)
            with contextlib.suppress(OSError):
                tmp_path.unlink(missing_ok=True)

    # ------------------------------------------------------------------
    # orientation
    # ------------------------------------------------------------------
    def stats(self) -> dict:
        """Overview of the indexed codebase."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        return self._run("tool_stats(R)")

    def find_symbol(self, text: str, limit: int = 20) -> dict:
        """Find a class, function, or method by exact or partial name."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_find_symbol({self._prolog_val(text)}, {int(limit)}, R)"
        return self._run(goal)

    def freshness(self, limit: int = 50) -> dict:
        """Has the code moved under the KB? Answered by INFERENCE.

        Three-valued by design: ``fresh`` (checked, current), ``stale``
        (checked, N modules differ) and ``unknown`` (the snapshot carries
        no ``file_meta``, so nothing could be checked). ``unknown`` must
        never be read as ``fresh`` -- "I cannot tell" and "I checked" are
        different answers.
        """
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        return self._run(f"kb_freshness({int(limit)}, R)")

    def affected_symbols(self, qname: str, limit: int = 100) -> dict:
        """Blast radius at CLASS/METHOD granularity, not module granularity.

        Every symbol transitively reachable from ``qname`` through
        RESOLVED calls and inheritance -- i.e. what a change to it can
        break. Per-kind counts are computed before truncation.
        """
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_affected_symbols({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def affected_tests(self, qname: str, limit: int = 100) -> dict:
        """Test modules to run for a change, selected by SYMBOL.

        A test module qualifies because one of its own symbols is in the
        affected closure of ``qname`` -- finer than module-import-level
        selection. UNDER-SELECTS wherever call resolution is incomplete
        (function-local imports produce no resolved edge), so the result
        is a LOWER BOUND on what to run, never proof that the omitted
        tests are unaffected.
        """
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_affected_tests({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def literals(self, value: str, limit: int = 50) -> dict:
        """Find every DECLARED site whose literal equals ``value``.

        Covers parameter defaults, module-level variables and class
        variables -- the declared constants the AST snapshot carries. A
        literal written inside a function body (``sleep(60)``) is NOT in
        the snapshot, so an empty result means "not declared anywhere",
        never "absent from the source": callers must not read absence as
        proof. The returned ``total`` is the pre-truncation count so a
        prefix is always distinguishable from a complete answer.
        """
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_find_literal({self._prolog_val(value)}, {int(limit)}, R)"
        return self._run(goal)

    def symbol_context(self, qname: str) -> dict:
        """Full orientation card for one symbol."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_context({self._prolog_val(qname)}, R)"
        return self._run(goal)

    # ------------------------------------------------------------------
    # dependency / impact
    # ------------------------------------------------------------------
    def impact(self, qname: str, limit: int = 25) -> dict:
        """Everything that transitively depends on a symbol."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_impact({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def callers(self, qname: str, limit: int = 50) -> dict:
        """Direct callers of a symbol."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_callers({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def callees(self, qname: str, limit: int = 50) -> dict:
        """Symbols directly called by this symbol."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_callees({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def callers_merged(self, qname: str, limit: int = 50) -> dict:
        """Direct callers over the MERGED graph (static ``calls_resolved``
        union runtime-captured ``runtime_call`` edges — see
        ``scripts/build_runtime_callgraph.py``). Each result carries a
        ``via: "static"|"runtime"`` tag."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_callers_merged({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def callees_merged(self, qname: str, limit: int = 50) -> dict:
        """Symbols directly called over the MERGED graph (static ∪
        runtime). Each result carries a ``via: "static"|"runtime"`` tag."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_callees_merged({self._prolog_val(qname)}, {int(limit)}, R)"
        return self._run(goal)

    def trace_path(
        self, from_qname: str, to_qname: str, static_only: bool = False
    ) -> dict:
        """Shortest end-to-end call path between two qnames.

        By default walks the MERGED graph (static ``calls_resolved`` union
        runtime-captured ``runtime_call`` edges) — this is what lets a trace
        cross DI/dispatch/CXP-routed edges pure static analysis cannot see.
        Pass ``static_only=True`` to walk only ``calls_resolved`` (the same
        graph ``callers``/``callees`` use by default). Each hop in the
        returned path is labeled ``via: "static"|"runtime"``.
        """
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        mode = "static" if static_only else "merged"
        goal = (
            f"tool_trace_path({self._prolog_val(from_qname)}, "
            f"{self._prolog_val(to_qname)}, {self._prolog_val(mode)}, R)"
        )
        return self._run(goal)

    def module_deps(
        self, module: str, direction: str = "deps", limit: int = 50
    ) -> dict:
        """Transitive internal module dependencies."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = (
            f"tool_module_deps({self._prolog_val(module)}, "
            f"{self._prolog_val(direction)}, {int(limit)}, R)"
        )
        return self._run(goal)

    def dep_path(self, from_module: str, to_module: str) -> dict:
        """One concrete dependency chain between modules."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = (
            f"tool_dep_path({self._prolog_val(from_module)}, "
            f"{self._prolog_val(to_module)}, R)"
        )
        return self._run(goal)

    def cycles(self, limit: int = 20) -> dict:
        """All import cycles among internal modules."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_cycles({int(limit)}, R)"
        return self._run(goal)

    def externals(self, module: str) -> dict:
        """A module's third-party/stdlib surface."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_externals({self._prolog_val(module)}, R)"
        return self._run(goal)

    def unused_symbols(self, limit: int = 50) -> dict:
        """Dead-code candidates."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_unused({int(limit)}, R)"
        return self._run(goal)

    # ------------------------------------------------------------------
    # metrics / properties
    # ------------------------------------------------------------------
    def hotspots(self, limit: int = 20) -> dict:
        """Functions ranked by cyclomatic complexity."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_hotspots({int(limit)}, R)"
        return self._run(goal)

    def by_prop(self, prop: str, limit: int = 50) -> dict:
        """All symbols carrying a boolean property."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_by_prop({self._prolog_val(prop)}, {int(limit)}, R)"
        return self._run(goal)

    def signature_search(self, text: str, limit: int = 25) -> dict:
        """Substring search over rendered signatures."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_signature_search({self._prolog_val(text)}, {int(limit)}, R)"
        return self._run(goal)

    def module_api(self, module: str) -> dict:
        """The public API of a module."""
        if not self._loaded:
            return {"error": "not_loaded", "hint": "call reindex() first"}
        goal = f"tool_module_api({self._prolog_val(module)}, R)"
        return self._run(goal)