scripts.code_intel

Reference › scripts › code_intel

A command-line wrapper over the Prolog code-intelligence knowledge base, with a freshness check. It builds a CodebaseTools object from codebase_tools, loads the code snapshot into it and answers structural questions about the code. A human or an agent runs it, and tests/scripts/test_code_intel.py imports it. build_runtime_callgraph feeds the snapshot it reads.

Invocation: main dispatches on a subcommand: reindex, stats, find-symbol, literals, freshness, affected-symbols, affected-tests, impact, callers, callees, dep-path, trace-path, cycles, unused, module-api and hotspots. Results print as human-readable text, or as JSON when requested (_emit, _print_human). An unknown command prints help and exits 1; an exception prints an error and exits 1; otherwise the exit code comes from the handler (_result_exit_code).

Before a query, _prepare_query_kb checks that the snapshot is current: _ensure_fresh compares it with the source and rebuilds the knowledge base when it is stale. _resolve_paths takes the snapshot and cache locations from the environment. Queries that take a qualified name go through _run_qname_query, which strips a leading package prefix (_strip_prefix) and reports when the name is unknown.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

#!/usr/bin/env python3
"""CLI + freshness wrapper over the Prolog code-intelligence KB (CodebaseTools).

Thin argparse front end over
:class:`orchestrata_ultimate.tools.codebase_tools.CodebaseTools` -- the
janus-backed Prolog codebase-intelligence layer that, until now, was only
reachable through the ``codebase_query_tools.py`` volk-tool wiring (and only
four of its twelve query methods at that). This CLI exposes the full query
surface directly so the orchestrator (or a human) can run it from a shell.

Subcommands
-----------
reindex [ast_json_path]    Rebuild the KB from a CodebaseAST JSON snapshot
                            (default: ``plans/ast-ultimate.json``).
stats                       Overview of the indexed codebase.
find-symbol TEXT            Find a class/function/method by name.
literals VALUE              Every site whose literal equals VALUE -- the
                            constant-origin query, which names the assignment
                            a grep for the bare number cannot rank.
freshness                   Three-valued staleness of the KB against the live
                            tree: fresh / stale / unknown, plus which modules
                            changed since indexing and why.
affected-symbols QNAME      Blast radius as CLASSES and METHODS, not merely
                            modules -- the symbol-level closure over
                            calls_resolved plus inherits_resolved.
affected-tests QNAME        The runnable half of blast radius: every test
                            module holding an affected symbol, with its path,
                            so the answer can be handed to the runner.
impact QNAME                Everything that transitively depends on a symbol.
callers QNAME                Direct callers of a symbol (pass --runtime to
                            include runtime-captured edges too).
callees QNAME                 Symbols directly called by this symbol (pass
                            --runtime to include runtime-captured edges too).
dep-path FROM TO            One concrete dependency chain between modules.
trace-path FROM TO          Shortest end-to-end call path between two qnames
                            (merged static+runtime graph by default; pass
                            --static-only to walk calls_resolved alone).
cycles                       All import cycles among internal modules.
unused                       Dead-code candidates.
module-api MODULE            The public API of a module.
hotspots                     Functions ranked by cyclomatic complexity.

Freshness
---------
Every subcommand except ``reindex`` first checks this instance's facts file
(``CodebaseTools.facts_file``, under a stable workdir resolved the same way
``codebase_query_tools.py``'s ``_build_production_kb`` does --
``ORCHESTRATA_CODEBASE_KB_DIR``/``ORCHESTRATA_AST_SNAPSHOT`` env vars,
defaulting to ``.claude/run/codebase_kb`` / ``plans/ast-ultimate.json`` under
the project root). A missing or stale facts file (mtime older than the AST
snapshot) triggers an automatic ``reindex()`` first, with a one-line note on
stderr -- pass ``--no-reindex`` to skip this and query whatever is already
loaded (or fail with ``not_loaded``/stale results). Sharing the KB dir with
the volk-tool wiring means ``code_intel.py reindex`` also refreshes what
those tools see.

qname ergonomics
-----------------
``impact``/``callers``/``callees``/``module-api`` accept a qname/module with
or without the leading ``orchestrata_ultimate.`` prefix (the KB's keys are
always prefix-stripped -- see ``codebase_tools.py``'s module docstring). If
the (normalized) query still comes back ``unknown_symbol``, this CLI falls
back to ``find_symbol()`` on the original text and prints the top matches as
suggestions, exiting non-zero.

Examples
--------
    uv run python scripts/code_intel.py cycles --json
    uv run python scripts/code_intel.py impact orchestrata_ultimate.core.lru_cache.LRUCache.get --json
    uv run python scripts/code_intel.py reindex
    uv run python scripts/code_intel.py find-symbol LRUCache --limit 5
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

# Ensure repo src is on path when run directly (mirrors scripts/kg_schema.py /
# scripts/graph_query.py; also lets static analysis resolve the import
# without the editable-install .pth machinery).
_repo_src = Path(__file__).resolve().parent.parent / "src"
if str(_repo_src) not in sys.path:
    sys.path.insert(0, str(_repo_src))

from orchestrata_ultimate.tools.codebase_tools import CodebaseTools  # noqa: E402

# Same convention (and, by default, the same physical directory) as
# tools/codebase_query_tools.py's _build_production_kb -- reindexing via this
# CLI also refreshes the KB the volk registry's codebase_* tools see.
_DEFAULT_KB_DIR = ".claude/run/codebase_kb"
_DEFAULT_AST_SNAPSHOT = "plans/ast-ultimate.json"
_QNAME_PREFIX = "orchestrata_ultimate."

# Deliberately NOT orchestrata_ultimate.core.paths.get_project_root(): that
# helper walks up from core/paths.py looking for the first ancestor with a
# prolog/ subdir, and src/orchestrata_ultimate/ itself ships one (the
# package's bundled Prolog rules) -- absent an ORCHESTRATA_ROOT override (not
# set in this repo's .env), it resolves to src/orchestrata_ultimate, NOT the
# repo root (tests/core/test_paths.py's own
# test_is_the_packaged_rules_dir_not_the_root_marker calls this "the stale
# root-discovery marker"). code_intel.py already knows its own true repo
# root unambiguously (two parents up from this file, same computation
# _repo_src above uses), so it anchors there instead of inheriting that
# defect. See this build's report for the upstream fix this exposes.
_REPO_ROOT = Path(__file__).resolve().parent.parent


def _resolve_paths(environ: dict[str, str] | None) -> tuple[Path, Path]:
    """Resolve (kb_dir, ast_snapshot_path) from env, defaulting under the repo root."""
    env = environ if environ is not None else os.environ
    root = env.get("ORCHESTRATA_ROOT", "")
    # The .env template ships ``ORCHESTRATA_ROOT=CHANGEME_ORCHESTRATA_ROOT`` and
    # the placeholder is only substituted on provisioned deployments; on an
    # un-provisioned workspace it leaks through and silently poisons every path
    # below. Treat the ``CHANGEME_`` sentinel as "not configured" so the
    # repo-root fallback wins (2026-09-17).
    if root.startswith("CHANGEME_"):
        root = ""
    project_root = Path(root) if root else _REPO_ROOT
    kb_dir = Path(
        env.get("ORCHESTRATA_CODEBASE_KB_DIR", str(project_root / _DEFAULT_KB_DIR))
    )
    snapshot_path = Path(
        env.get("ORCHESTRATA_AST_SNAPSHOT", str(project_root / _DEFAULT_AST_SNAPSHOT))
    )
    return kb_dir, snapshot_path


def _build_kb(environ: dict[str, str] | None = None) -> CodebaseTools:
    kb_dir, _ = _resolve_paths(environ)
    return CodebaseTools(workdir=str(kb_dir))


def _ensure_fresh(kb: CodebaseTools, snapshot_path: Path) -> dict[str, Any] | None:
    """Reindex ``kb`` from ``snapshot_path`` if its facts file is missing/stale.

    Returns the ``reindex()`` result dict if a reindex ran, else ``None``.
    Raises ``RuntimeError`` if a reindex is needed but ``snapshot_path``
    doesn't exist -- the same fail-loud posture as
    ``codebase_query_tools.py``'s ``_build_production_kb``. Reads
    ``sys.stderr`` at call time (not as a default-argument value, which
    would freeze in whatever object held that name at *function-definition*
    time -- e.g. before a test's capsys fixture swaps it in).
    """
    facts_file = Path(kb.facts_file)
    stale = not facts_file.exists() or (
        snapshot_path.exists()
        and facts_file.stat().st_mtime < snapshot_path.stat().st_mtime
    )
    if not stale:
        return None
    if not snapshot_path.exists():
        raise RuntimeError(
            f"codebase KB facts are missing/stale and no AST snapshot exists "
            f"at {snapshot_path} -- generate one with `uv run python "
            "scripts/serialize_codebase_ast.py serialize --source "
            "src/orchestrata_ultimate --name orchestrata-ultimate --output "
            f"{snapshot_path}`, or pass an explicit path to `reindex`."
        )
    result = kb.reindex(str(snapshot_path))
    print(f"[code_intel] reindexed from {snapshot_path}", file=sys.stderr)
    return result


def _prepare_query_kb(
    args: argparse.Namespace, environ: dict[str, str] | None
) -> CodebaseTools:
    kb = _build_kb(environ)
    if not args.no_reindex:
        _, snapshot_path = _resolve_paths(environ)
        _ensure_fresh(kb, snapshot_path)
    return kb


# ---------------------------------------------------------------------------
# qname ergonomics
# ---------------------------------------------------------------------------


def _strip_prefix(qname: str) -> str:
    return qname[len(_QNAME_PREFIX) :] if qname.startswith(_QNAME_PREFIX) else qname


# codebase_db.pl's "not found" error differs by predicate: qname-taking
# predicates (tool_impact/tool_callers/tool_callees) fail with
# error:unknown_symbol; the module-taking predicate (tool_module_api) fails
# with error:unknown_module. Both mean the same thing from this CLI's POV
# ("that identifier isn't in the KB") so both trigger the suggestion fallback.
_NOT_FOUND_ERRORS = ("unknown_symbol", "unknown_module")


def _run_qname_query(
    kb: CodebaseTools, original: str, query: Any
) -> tuple[dict[str, Any], int]:
    """Run a normalized-qname query, falling back to find_symbol() suggestions.

    ``query`` is a one-arg callable taking the normalized qname/module and
    returning the raw CodebaseTools result dict. On an unknown-symbol/-module
    error the ORIGINAL (unnormalized) text is handed to ``find_symbol`` -- a
    caller who typed the wrong prefix by mistake still gets useful
    suggestions.
    """
    normalized = _strip_prefix(original)
    result = query(normalized)
    if isinstance(result, dict) and result.get("error") in _NOT_FOUND_ERRORS:
        suggestions = kb.find_symbol(original)
        return (
            {
                "error": result.get("error"),
                "queried": original,
                "normalized": normalized,
                "suggestions": suggestions,
            },
            1,
        )
    return result, 0


# ---------------------------------------------------------------------------
# output
# ---------------------------------------------------------------------------


def _result_exit_code(result: Any) -> int:
    return 1 if isinstance(result, dict) and "error" in result else 0


def _print_human(result: Any, indent: int = 0) -> None:
    pad = "  " * indent
    if isinstance(result, dict):
        for key, value in result.items():
            if (
                isinstance(value, list)
                and value
                and all(isinstance(v, dict) for v in value)
            ):
                print(f"{pad}{key}: ({len(value)})")
                for item in value:
                    parts = ", ".join(f"{k}={v}" for k, v in item.items())
                    print(f"{pad}  - {parts}")
            elif isinstance(value, dict):
                print(f"{pad}{key}:")
                _print_human(value, indent + 1)
            elif isinstance(value, list):
                if value:
                    print(f"{pad}{key}: {', '.join(str(v) for v in value)}")
                else:
                    print(f"{pad}{key}: []")
            else:
                print(f"{pad}{key}: {value}")
    else:
        print(f"{pad}{result}")


def _emit(result: Any, as_json: bool) -> None:
    if as_json:
        print(json.dumps(result, indent=2))
    else:
        _print_human(result)


# ---------------------------------------------------------------------------
# subcommand handlers
# ---------------------------------------------------------------------------


def _cmd_reindex(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _build_kb(environ)
    _, default_snapshot = _resolve_paths(environ)
    snapshot = Path(args.ast_json_path) if args.ast_json_path else default_snapshot
    result = kb.reindex(str(snapshot))
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_stats(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.stats()
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_find_symbol(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.find_symbol(args.text, limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_freshness(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.freshness(limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_affected_symbols(
    args: argparse.Namespace, environ: dict[str, str] | None
) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.affected_symbols(args.qname, limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_affected_tests(
    args: argparse.Namespace, environ: dict[str, str] | None
) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.affected_tests(args.qname, limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_literals(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.literals(args.value, limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_impact(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    """Blast radius for one or more symbols, answered by ONE process.

    Accepts several qnames because the cost of this command is process start
    plus consulting the KB, not the query. Measured 2026-08-13: an invocation
    costs 7.4s and ``--no-reindex`` changes that by 0.02s. A caller asking about
    a module's whole public surface therefore paid 7.4s PER SYMBOL, which is why
    ``module_context.py`` capped itself at 6 and left the rest unanswered — 456
    of 2485 public symbols across orchestrata and knowledge-wiki, 18%, on 11% of
    modules. Batching removes the reason for the cap rather than raising it.

    A single qname keeps the old single-object output, so every existing caller
    and every parsed script is unaffected. Several qnames produce a keyed
    mapping, because merging blast radii into one list would lose which symbol
    each dependent came from — the thing the caller asked.
    """
    kb = _prepare_query_kb(args, environ)
    qnames = args.qname if isinstance(args.qname, list) else [args.qname]

    if len(qnames) == 1:
        result, exit_code = _run_qname_query(
            kb, qnames[0], lambda q: kb.impact(q, limit=args.limit)
        )
        _emit(result, args.json)
        return exit_code

    results: dict[str, Any] = {}
    worst = 0
    for qname in qnames:
        result, code = _run_qname_query(
            kb, qname, lambda q: kb.impact(q, limit=args.limit)
        )
        results[qname] = result
        worst = max(worst, code)
    _emit({"batch": True, "results": results}, args.json)
    return worst


def _cmd_callers(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    query = (
        (lambda q: kb.callers_merged(q, limit=args.limit))
        if args.runtime
        else (lambda q: kb.callers(q, limit=args.limit))
    )
    result, exit_code = _run_qname_query(kb, args.qname, query)
    _emit(result, args.json)
    return exit_code


def _cmd_callees(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    query = (
        (lambda q: kb.callees_merged(q, limit=args.limit))
        if args.runtime
        else (lambda q: kb.callees(q, limit=args.limit))
    )
    result, exit_code = _run_qname_query(kb, args.qname, query)
    _emit(result, args.json)
    return exit_code


def _cmd_trace_path(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    """Shortest end-to-end call path between two qnames.

    Walks the MERGED static+runtime graph by default (the point of this
    subcommand: crossing DI/dispatch/CXP edges pure static analysis
    cannot see) -- pass ``--static-only`` to walk ``calls_resolved`` alone.
    On an unknown endpoint, falls back to ``find_symbol`` suggestions for
    each unresolved qname (same ergonomics as ``_run_qname_query``).
    """
    kb = _prepare_query_kb(args, environ)
    from_q = _strip_prefix(args.from_qname)
    to_q = _strip_prefix(args.to_qname)
    result = kb.trace_path(from_q, to_q, static_only=args.static_only)
    if isinstance(result, dict) and result.get("error") in _NOT_FOUND_ERRORS:
        result = {
            **result,
            "from_suggestions": kb.find_symbol(args.from_qname, limit=5),
            "to_suggestions": kb.find_symbol(args.to_qname, limit=5),
        }
        _emit(result, args.json)
        return 1
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_dep_path(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.dep_path(args.from_module, args.to_module)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_cycles(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.cycles(limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_unused(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.unused_symbols(limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


def _cmd_module_api(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result, exit_code = _run_qname_query(kb, args.module, kb.module_api)
    _emit(result, args.json)
    return exit_code


def _cmd_hotspots(args: argparse.Namespace, environ: dict[str, str] | None) -> int:
    kb = _prepare_query_kb(args, environ)
    result = kb.hotspots(limit=args.limit)
    _emit(result, args.json)
    return _result_exit_code(result)


# ---------------------------------------------------------------------------
# argparse wiring
# ---------------------------------------------------------------------------


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="code_intel",
        description=(
            "CLI + freshness wrapper over the Prolog code-intelligence KB "
            "(orchestrata_ultimate.tools.codebase_tools.CodebaseTools)."
        ),
    )
    parser.add_argument(
        "--json", action="store_true", help="Emit JSON instead of human-readable text."
    )
    parser.add_argument(
        "--no-reindex",
        action="store_true",
        help="Skip the automatic staleness reindex on query subcommands.",
    )
    parser.add_argument(
        "--runtime",
        action="store_true",
        help="For callers/callees: include runtime-captured edges (the "
        "merged static+runtime graph) in addition to calls_resolved. "
        "Default stays static-only -- no behavior change without this flag.",
    )

    sub = parser.add_subparsers(dest="command", required=True)

    rx = sub.add_parser(
        "reindex", help="Rebuild the KB from a CodebaseAST JSON snapshot."
    )
    rx.add_argument(
        "ast_json_path",
        nargs="?",
        default=None,
        help=f"Path to the AST JSON snapshot (default: {_DEFAULT_AST_SNAPSHOT}).",
    )

    sub.add_parser("stats", help="Overview of the indexed codebase.")

    fs = sub.add_parser(
        "find-symbol",
        help="Find a class, function, or method by exact or partial name.",
    )
    fs.add_argument("text", help="Search text.")
    fs.add_argument("--limit", type=int, default=20, help="Result limit (default 20).")

    lt = sub.add_parser(
        "literals",
        help="Where a declared constant comes from (parameter defaults, "
        "module vars, class vars) — with the enclosing symbol.",
        description="AST-precise constant search: matches only a DECLARED "
        "literal, never the same text inside a comment, docstring or "
        "identifier. Body-internal literals (sleep(60)) are not in the AST "
        "snapshot, so an empty result means 'not declared anywhere', NOT "
        "'absent from the source'.",
    )
    lt.add_argument("value", help="Literal as written, e.g. 60, 5.0, True.")
    lt.add_argument("--limit", type=int, default=50, help="Result limit (default 50).")

    fr = sub.add_parser(
        "freshness",
        help="Has the code moved under the KB? Lists the changed modules.",
        description="Freshness as inference over file_meta/5 vs disk. "
        "THREE-VALUED: fresh (checked, current) / stale (checked, N differ) "
        "/ unknown (snapshot carries no file_meta, so nothing was checked). "
        "unknown is NOT fresh.",
    )
    fr.add_argument("--limit", type=int, default=50, help="Result limit (default 50).")

    af = sub.add_parser(
        "affected-symbols",
        help="Blast radius as CLASSES and METHODS, not modules/test paths.",
        description="Every symbol transitively reachable from this one via "
        "RESOLVED calls and inheritance — what a change to it can break. "
        "Per-kind counts are computed before truncation.",
    )
    af.add_argument("qname", help="Fully-qualified symbol name.")
    af.add_argument(
        "--limit", type=int, default=100, help="Result limit (default 100)."
    )

    at = sub.add_parser(
        "affected-tests",
        help="Test modules to run for a change, selected by SYMBOL.",
        description="A test module qualifies because one of its own symbols "
        "is in the affected closure — finer than module-import selection. "
        "UNDER-SELECTS where call resolution is incomplete (function-local "
        "imports produce no resolved edge): a LOWER BOUND on what to run, "
        "never proof the omitted tests are unaffected.",
    )
    at.add_argument("qname", help="Fully-qualified symbol name.")
    at.add_argument(
        "--limit", type=int, default=100, help="Result limit (default 100)."
    )

    im = sub.add_parser(
        "impact", help="Everything that transitively depends on a symbol."
    )
    im.add_argument(
        "qname",
        nargs="+",
        help="Fully-qualified symbol name(s), with or without the leading "
        "'orchestrata_ultimate.' prefix. Several may be given: this command's "
        "cost is process start plus consulting the KB (7.4s measured), not the "
        "query (0.02s), so asking about a whole module's public surface in one "
        "invocation costs the same as asking about one symbol. One qname keeps "
        "the single-object output; several return {'batch': true, 'results': "
        "{qname: result}}.",
    )
    im.add_argument("--limit", type=int, default=25, help="Result limit (default 25).")

    ca = sub.add_parser("callers", help="Direct callers of a symbol.")
    ca.add_argument("qname", help="Fully-qualified symbol name.")
    ca.add_argument("--limit", type=int, default=50, help="Result limit (default 50).")

    ce = sub.add_parser("callees", help="Symbols directly called by this symbol.")
    ce.add_argument("qname", help="Fully-qualified symbol name.")
    ce.add_argument("--limit", type=int, default=50, help="Result limit (default 50).")

    dp = sub.add_parser(
        "dep-path", help="One concrete dependency chain between modules."
    )
    dp.add_argument("from_module", help="Source internal module path.")
    dp.add_argument("to_module", help="Target internal module path.")

    tp = sub.add_parser(
        "trace-path",
        help="Shortest end-to-end call path between two qnames (merged "
        "static+runtime graph by default -- see --static-only).",
    )
    tp.add_argument("from_qname", help="Source function/class/method qname.")
    tp.add_argument("to_qname", help="Target function/class/method qname.")
    tp.add_argument(
        "--static-only",
        action="store_true",
        help="Walk only the static calls_resolved graph (no runtime edges).",
    )

    cy = sub.add_parser("cycles", help="All import cycles among internal modules.")
    cy.add_argument("--limit", type=int, default=20, help="Result limit (default 20).")

    un = sub.add_parser("unused", help="Dead-code candidates.")
    un.add_argument("--limit", type=int, default=50, help="Result limit (default 50).")

    ma = sub.add_parser("module-api", help="The public API of a module.")
    ma.add_argument(
        "module",
        help="Internal module path (with or without the leading "
        "'orchestrata_ultimate.' prefix).",
    )

    ho = sub.add_parser("hotspots", help="Functions ranked by cyclomatic complexity.")
    ho.add_argument("--limit", type=int, default=20, help="Result limit (default 20).")

    return parser


def main(argv: list[str] | None = None, environ: dict[str, str] | None = None) -> None:
    parser = _build_parser()
    args = parser.parse_args(argv)

    dispatch = {
        "reindex": _cmd_reindex,
        "stats": _cmd_stats,
        "find-symbol": _cmd_find_symbol,
        "literals": _cmd_literals,
        "freshness": _cmd_freshness,
        "affected-symbols": _cmd_affected_symbols,
        "affected-tests": _cmd_affected_tests,
        "impact": _cmd_impact,
        "callers": _cmd_callers,
        "callees": _cmd_callees,
        "dep-path": _cmd_dep_path,
        "trace-path": _cmd_trace_path,
        "cycles": _cmd_cycles,
        "unused": _cmd_unused,
        "module-api": _cmd_module_api,
        "hotspots": _cmd_hotspots,
    }
    handler = dispatch.get(args.command)
    if handler is None:
        parser.print_help()
        sys.exit(1)

    try:
        exit_code = handler(args, environ)
    except Exception as exc:  # noqa: BLE001
        print(f"ERROR: {exc}", file=sys.stderr)
        sys.exit(1)

    sys.exit(exit_code)


if __name__ == "__main__":
    main()