scripts.codemod

Reference › scripts › codemod

The command-line front end of the SPEC-110 codemod engine, which lives in codemod. It runs transformation passes over the codebase and answers dependency-graph questions. A human runs it. tests/scripts/test_codemod_cli.py and tests/scripts/test_codemod_cli_daemon.py import it, and it uses @traces from traces.

Invocation: main has four subcommands. load-bench measures how long loading the workspace takes (_cmd_load_bench). run applies one or more named transformation passes with their arguments (_cmd_run, _pass_spec, _resolve_pass) and reports a commit report (_finish_run). report shows a previous report (_cmd_report). analyze runs analysis passes or dependency queries over a module, such as dependencies, cycles, topological order and centrality (_cmd_analyze_passes, _LibraryQueries, the _print_* helpers), and can compare against a base revision (_changes_since). Common options include the repository path, --include roots, --no-daemon and --auto-start. A parse error returns 2.

Two execution modes exist. When a resident daemon serves the repository (REQ-110-140), _connect_or_warn connects to its socket with a CodemodClient and the work runs there (_run_via_daemon). Otherwise the script parses the code in its own process (_run_in_process, library mode), says so with _library_mode_warning, and --no-daemon selects that mode quietly. --auto-start first starts the daemon through _auto_start_daemon, and a failure to start it returns 1.

At a glance

Class diagram

Dependency diagram

API

Classes

_UnknownModule

Bases: Exception · Line 954

--module names no node of the code intelligence graph.

_LibraryQueries

Line 958

The graph queries of analyze, answered from a workspace parsed here.

_DaemonQueries

Line 999

The same queries, answered by the daemon serving the repo.

Functions

Source

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

#!/usr/bin/env python3
"""SPEC-110 codemod engine CLI — load-bench, run, report, analyze subcommands."""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
import uuid
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    # Static-only: the real imports stay lazy inside the functions below
    # (avoids loading the codemod engine for CLI paths that never touch
    # it), but mypy needs the names in scope to check the signatures.
    from orchestrata_ultimate.refactor.codemod._cig import CodeIntelligenceGraph
    from orchestrata_ultimate.refactor.codemod.client import CodemodClient
    from orchestrata_ultimate.refactor.codemod.passes import TransformPass
    from orchestrata_ultimate.refactor.codemod.report import ChangeSet, CommitReport
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="codemod",
        description="In-memory code intelligence platform",
    )
    parser.add_argument(
        "--repo",
        type=Path,
        default=Path(os.environ.get("CODEMOD_ROOT", ".")),
        help="Repository root path (default: $CODEMOD_ROOT or .)",
    )
    parser.add_argument(
        "--include",
        action="append",
        default=None,
        help="Directories to parse (repeatable; default: src, scripts, tests)",
    )
    parser.add_argument(
        "--parallel",
        action="store_true",
        default=False,
        help="Use parallel parsing",
    )
    parser.add_argument(
        "--auto-start",
        action="store_true",
        help="Auto-connect to daemon or start it if not running",
    )
    parser.add_argument(
        "--no-daemon",
        action="store_true",
        help="Force library mode (parse on every invocation)",
    )

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

    subparsers.add_parser(
        "load-bench",
        help="Measure workspace load time, RSS delta, file count",
    )

    run_parser = subparsers.add_parser(
        "run",
        help="Execute named transform passes",
    )
    run_parser.add_argument(
        "--pass",
        dest="passes",
        action="append",
        required=True,
        help="Transform pass name (repeatable)",
    )
    run_parser.add_argument(
        "--arg",
        action="append",
        default=[],
        help="Pass argument as key=value (repeatable)",
    )
    run_parser.add_argument(
        "--apply",
        action="store_true",
        default=False,
        help="Write changes to disk (default: dry-run)",
    )
    run_parser.add_argument(
        "--dry-run",
        action="store_true",
        default=True,
        help="Produce diff and report without writing (default)",
    )
    run_parser.add_argument(
        "--observe",
        action="store_true",
        default=False,
        help=(
            "Attach the five REQ-110-026 observers (tracing, metrics, debug "
            "panel, error detector, graph writer) for this run and commit; "
            "the debug panel writes NDJSON traces under "
            "pipeline_output/traces/ relative to the working directory"
        ),
    )
    run_parser.add_argument(
        "--baseline",
        choices=("delta", "absolute"),
        default="delta",
        help=(
            "Review-engine baseline (ADR-0073): 'delta' (default) aborts the "
            "commit only on error findings the change introduced; 'absolute' "
            "aborts on any error finding in a mutated module"
        ),
    )
    run_parser.add_argument(
        "--output",
        type=Path,
        default=None,
        help="Report output path (default: reports/codemod-<id>.json)",
    )

    report_parser = subparsers.add_parser(
        "report",
        help="Pretty-print a CommitReport JSON file",
    )
    report_parser.add_argument(
        "path",
        type=Path,
        help="Path to CommitReport JSON file",
    )

    analyze_parser = subparsers.add_parser(
        "analyze",
        help="Execute CIG queries",
    )
    analyze_parser.add_argument(
        "--query",
        required=True,
        choices=[
            "dependencies",
            "dependents",
            "blast_radius",
            "cycles",
            "topological_order",
            "coupling_score",
            "abstraction_level",
            "centrality",
            "impact_score",
            "breaking_changes",
            "complexity",
            "test_coverage",
            "dead_code",
            "all-passes",
        ],
        help="CIG query type",
    )
    analyze_parser.add_argument(
        "--module",
        type=str,
        default=None,
        help="Module path or dotted name for query",
    )
    analyze_parser.add_argument(
        "--hops",
        type=int,
        default=3,
        help="BFS depth for blast_radius (default: 3)",
    )
    analyze_parser.add_argument(
        "--top-k",
        type=int,
        default=20,
        help="Number of results for centrality (default: 20)",
    )
    analyze_parser.add_argument(
        "--exact",
        action="store_true",
        help="Exact Brandes centrality over all sources (default: "
        "pivot-sampled approximation)",
    )
    analyze_parser.add_argument(
        "--base",
        type=str,
        default=None,
        help="Git ref the working tree is diffed against for breaking_changes "
        "(used with --module)",
    )
    analyze_parser.add_argument(
        "--json",
        action="store_true",
        help="Emit the result of an analysis pass (blast_radius, coupling_score, "
        "abstraction_level, breaking_changes, complexity, test_coverage, "
        "dead_code, all-passes) as JSON",
    )

    return parser


def _auto_start_daemon(
    repo_root: Path,
    *,
    tools_prefix: Path | None = None,
    unit_name: str | None = None,
    wait_seconds: float | None = None,
) -> str | None:
    """Ensure the deployed codemod daemon serves *repo_root* (``--auto-start``).

    ADR-0067: the daemon only ever runs deployed code through its systemd
    unit. Unless given explicitly, the prefix and unit come from the
    identity-shaped seams ``CODEMOD_TOOLS_PREFIX`` / ``CODEMOD_DAEMON_UNIT``
    (default ``/opt/orchestrata-tools`` / ``orchestrata-codemod-daemon``).

    The daemon loads its workspace before it binds the socket, so a started
    unit is not yet a serving daemon: this waits for a running daemon's socket
    file, polling every 0.2 s up to *wait_seconds* (default
    ``CODEMOD_AUTO_START_WAIT``, 300: REQ-110-133's 120-s parse plus margin).
    A unit that is not running after its first second, or has failed, ends the
    wait at once.

    Returns:
        ``None`` when the daemon is serving afterwards, else the
        ``RuntimeError`` text verbatim (e.g. the not-deployed message), the
        failed-before-serving text, or the not-serving text after the wait.
    """
    from orchestrata_ultimate.refactor.codemod.daemon import (
        DEFAULT_TOOLS_PREFIX,
        DEFAULT_UNIT_NAME,
        CodemodDaemon,
    )

    daemon = CodemodDaemon(
        tools_prefix=tools_prefix
        or Path(os.environ.get("CODEMOD_TOOLS_PREFIX", str(DEFAULT_TOOLS_PREFIX))),
        unit_name=unit_name or os.environ.get("CODEMOD_DAEMON_UNIT", DEFAULT_UNIT_NAME),
    )
    daemon._repo_root = repo_root
    try:
        if not daemon.status()["running"]:
            daemon.start(repo_root)
    except RuntimeError as e:
        return str(e)

    wait = (
        wait_seconds
        if wait_seconds is not None
        else float(os.environ.get("CODEMOD_AUTO_START_WAIT", "300"))
    )
    started = time.monotonic()
    deadline = started + wait
    while True:
        status = daemon.status()
        running = bool(status.get("running"))
        if running and status.get("socket_exists"):
            return None
        # A unit that has not yet reported a main pid is given its first
        # second; after that a unit that is not running (exited, or waiting
        # to restart) or has failed will never serve, and neither will a
        # socket file left behind by a killed daemon.
        state = status.get("active_state")
        if state == "failed" or (not running and time.monotonic() - started >= 1.0):
            return f"daemon {status.get('unit')} failed before serving ({state})"
        if time.monotonic() >= deadline:
            return (
                f"daemon {status.get('unit')} started but not serving after {wait:g}s"
            )
        time.sleep(0.2)


def _daemon_socket_path(repo_root: Path) -> Path:
    """Where the daemon serving *repo_root* listens."""
    from orchestrata_ultimate.refactor.codemod.daemon import DAEMON_DIR, SOCKET_NAME

    return repo_root / DAEMON_DIR / SOCKET_NAME


def _connect_daemon(socket_path: Path) -> CodemodClient | None:
    """A client connected to the daemon at *socket_path*, or None.

    None covers every way there is nobody to talk to: no socket file, a
    socket file with no listener, a path that is not a socket.
    """
    from orchestrata_ultimate.refactor.codemod.client import CodemodClient

    try:
        return CodemodClient().connect(socket_path)
    except ConnectionError:
        return None


def _library_mode_warning(repo_root: Path, socket_path: Path) -> str:
    """The one stderr line printed when no daemon serves *repo_root*."""
    from orchestrata_ultimate.refactor.codemod.daemon import (
        DEFAULT_UNIT_NAME,
        CodemodDaemon,
    )

    daemon = CodemodDaemon(
        unit_name=os.environ.get("CODEMOD_DAEMON_UNIT", DEFAULT_UNIT_NAME)
    )
    try:
        instance = daemon._unit_instance(repo_root)
    except RuntimeError:
        instance = f"{daemon.unit_name}@<escaped-repo-path>.service"
    return (
        f"codemod: no daemon serving {repo_root} (socket {socket_path}); "
        "library mode, parsing on every invocation. "
        f"Start it with --auto-start or systemctl --user start {instance}; "
        "--no-daemon selects library mode without this warning."
    )


def _roots_text(roots: tuple[str, ...]) -> str:
    return ", ".join(roots) or "the whole repo"


def _folded_roots(repo_root: Path, include: list[str]) -> tuple[str, ...]:
    """*include* as the sorted repo-relative roots a workspace load stores.

    Folds the way ``CodemodWorkspace.load`` does (an include naming the repo
    root lifts the restriction, which a workspace stores as no roots). An
    include the load would refuse is kept as spelled: it matches no daemon.
    """
    from orchestrata_ultimate.refactor.codemod.workspace import _include_root

    try:
        folded = {_include_root(repo_root, subdir) for subdir in include}
    except ValueError:
        return tuple(sorted(set(include)))
    return () if None in folded else tuple(sorted(r for r in folded if r))


@traces(spec="110", story="US-110-013", reqs=["REQ-110-140"])
def _connect_or_warn(
    repo_root: Path, include: list[str] | None = None
) -> CodemodClient | None:
    """Auto-connect to the daemon serving *repo_root*; say so when none does.

    Library mode answers (with one stderr line) when no daemon is listening,
    and when one is but loaded other roots than the ``--include`` asked for:
    a daemon never answers for roots it did not parse. Without ``--include``
    the daemon's roots are the roots (CLI and daemon default alike).
    """
    socket_path = _daemon_socket_path(repo_root)
    client = _connect_daemon(socket_path)
    # A server between its bind and its listen refuses connections on a socket
    # file that exists: retry before taking it for a stale one.
    for _ in range(3):
        if client is not None or not socket_path.exists():
            break
        time.sleep(0.05)
        client = _connect_daemon(socket_path)
    if client is None:
        print(_library_mode_warning(repo_root, socket_path), file=sys.stderr)
        return None
    if not include:
        return client
    # A daemon older than the ``include`` status field parsed the defaults.
    served = tuple(sorted(set(client.status().get("include", _resolve_include(None)))))
    asked = _folded_roots(repo_root, include)
    if served != asked:
        client.close()
        print(
            f"codemod: daemon serves {_roots_text(served)} but --include asks "
            f"for {_roots_text(asked)}; library mode for this run.",
            file=sys.stderr,
        )
        return None
    return client


def _resolve_include(include: list[str] | None) -> tuple[str, ...]:
    if include:
        return tuple(include)
    return ("src", "scripts", "tests")


def _load_workspace(args: argparse.Namespace) -> CodemodWorkspace:
    """Parse the repo into a workspace in this process (library mode)."""
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    return CodemodWorkspace.load(
        args.repo.resolve(),
        include=_resolve_include(args.include),
        parallel=args.parallel,
    )


_PASS_CLASSES = {
    "replace-string": "ReplaceStringLiteralPass",
    "add-import": "AddImportPass",
    "remove-unused-import": "RemoveUnusedImportPass",
    "docstring-sweep": "DocstringSweepPass",
    "rename-symbol": "RenameSymbolPass",
}


def _pass_spec(name: str, args: dict[str, str]) -> tuple[str, dict[str, Any]]:
    """The pass CLASS name (as ``passes`` exports it) and its constructor kwargs.

    The kwargs are JSON-serializable, so the same spec builds the pass in
    this process (``_resolve_pass``) or inside the daemon (``run_pass``).
    """
    cls_name = _PASS_CLASSES.get(name)
    if cls_name is None:
        print(
            f"Error: unknown pass '{name}'. "
            f"Available: {', '.join(sorted(_PASS_CLASSES))}",
            file=sys.stderr,
        )
        sys.exit(2)

    if name == "replace-string":
        return cls_name, {
            "old": args.get("old", ""),
            "new": args.get("new", ""),
            "regex": args.get("regex", "false").lower() == "true",
            "keep_patterns": [p for p in args.get("keep", "").split(",") if p],
        }
    if name == "add-import":
        return cls_name, {
            "module": args.get("module", ""),
            "name": args.get("name"),
            "alias": args.get("alias"),
        }
    if name == "docstring-sweep":
        replacements = {}
        for k, v in args.items():
            if k.startswith("old_"):
                suffix = k[4:]
                new_key = f"new_{suffix}"
                if new_key in args:
                    replacements[v] = args[new_key]
        if not replacements and "old" in args and "new" in args:
            replacements[args["old"]] = args["new"]
        return cls_name, {"replacements": replacements}
    if name == "rename-symbol":
        print(
            "Error: rename-symbol requires a WorkspaceEditPlan (use Python API)",
            file=sys.stderr,
        )
        sys.exit(2)
    return cls_name, {}


def _resolve_pass(name: str, args: dict[str, str]) -> TransformPass:
    from orchestrata_ultimate.refactor.codemod import passes

    cls_name, kwargs = _pass_spec(name, args)
    pass_cls: type[TransformPass] = getattr(passes, cls_name)
    return pass_cls(**kwargs)


def _parse_args(raw_args: list[str] | None = None) -> dict[str, str]:
    result: dict[str, str] = {}
    if not raw_args:
        return result
    for arg in raw_args:
        if "=" in arg:
            k, v = arg.split("=", 1)
            result[k] = v
    return result


def _cmd_load_bench(args: argparse.Namespace) -> int:
    from orchestrata_ultimate.refactor.codemod.workspace import CodemodWorkspace

    root = args.repo.resolve()
    include = _resolve_include(args.include)

    t0 = time.monotonic()
    CodemodWorkspace.load(root, include=include, parallel=False)
    serial_time = time.monotonic() - t0

    t0 = time.monotonic()
    ws_parallel = CodemodWorkspace.load(root, include=include, parallel=True)
    parallel_time = time.monotonic() - t0

    stats = ws_parallel.stats()
    cores = os.cpu_count() or 1
    print(f"Load time (serial):   {serial_time:.1f}s")
    print(f"Load time (parallel): {parallel_time:.1f}s ({cores} cores)")
    print(f"Files loaded:         {stats.file_count}")
    print(f"Total bytes:          {stats.total_bytes:,}")
    print(f"RSS delta:            {stats.rss_delta_bytes / (1024**3):.2f} GiB")
    print(f"Unparsable:           {stats.unparsable_count}")
    return 0


@traces(spec="110", story="US-110-013", reqs=["REQ-110-140"])
def _cmd_run(args: argparse.Namespace, client: CodemodClient | None = None) -> int:
    """``run``: in this process, or inside the daemon ``client`` is connected to."""
    from orchestrata_ultimate.refactor.codemod.client import CodemodRpcError

    pass_args = _parse_args(args.arg)
    if client is None:
        return _run_in_process(args, pass_args)
    try:
        return _run_via_daemon(args, client, pass_args)
    except CodemodRpcError as e:
        return _fail_run(e.rpc_message)


def _run_in_process(args: argparse.Namespace, pass_args: dict[str, str]) -> int:
    """Library mode: parse the repo, run, commit, all in this process."""
    from orchestrata_ultimate.refactor.codemod.review_engine import (
        InMemoryReviewEngine,
    )
    from orchestrata_ultimate.refactor.codemod.workspace import default_observers

    ws = _load_workspace(args)
    pass_instances = [_resolve_pass(pass_name, pass_args) for pass_name in args.passes]

    # --observe: the five REQ-110-026 observers watch this run and commit
    # (the workspace default stays "nobody", REQ-110-027).
    observers = default_observers() if args.observe else None
    try:
        changes = ws.run(pass_instances, observers=observers)
        try:
            report = ws.commit(
                changes,
                dry_run=not args.apply,
                observers=observers,
                review_engine=InMemoryReviewEngine(baseline=args.baseline),
            )
        except Exception as exc:
            return _fail_run(f"{type(exc).__name__}: {exc}")
    finally:
        # The debug panel holds open NDJSON handles; release them.
        for observer in observers or ():
            close = getattr(observer, "close", None)
            if callable(close):
                close()

    return _finish_run(
        args,
        [p.name for p in pass_instances],
        changes.file_count,
        report,
    )


def _run_via_daemon(
    args: argparse.Namespace, client: CodemodClient, pass_args: dict[str, str]
) -> int:
    """Daemon mode: the passes stack into one run the daemon holds by reference.

    The daemon restores its forest after a dry run, so this leaves the
    served workspace as it found it.
    """
    specs = [_pass_spec(pass_name, pass_args) for pass_name in args.passes]
    run_id: str | None = None
    names: list[str] = []
    file_count = 0
    for cls_name, kwargs in specs:
        reply = client.run_pass(cls_name, kwargs, run_id=run_id, observe=args.observe)
        run_id = reply["run_id"]
        file_count = reply["file_count"]
        names.append(reply["pass_name"])
    assert run_id is not None  # argparse requires at least one --pass
    report = client.commit(
        run_id,
        dry_run=not args.apply,
        baseline=args.baseline,
        observe=args.observe,
    )
    return _finish_run(args, names, file_count, report)


def _fail_run(message: str) -> int:
    """The one ``Error:`` line a failed commit prints, in either mode."""
    print(f"Error: {message}", file=sys.stderr)
    return 1


def _finish_run(
    args: argparse.Namespace,
    pass_names: list[str],
    files_mutated: int,
    report: CommitReport,
) -> int:
    """Print the run summary, write the report file, return the exit code.

    Both modes end here, so the two cannot drift.
    """
    root = args.repo.resolve()
    dry_run = not args.apply
    correlation_id = (
        f"codemod-{time.strftime('%Y-%m-%dT%H-%M-%S')}-{uuid.uuid4().hex[:8]}"
    )

    if report.diffstat:
        print(report.diffstat)

    error_count = len([f for f in report.findings if f.severity == "error"])
    warn_count = len([f for f in report.findings if f.severity == "warning"])

    print(f"Passes run: {', '.join(pass_names)}")
    if dry_run:
        print(f"Files mutated: {files_mutated}")
    else:
        print(f"Files written: {len(report.files_written)}")
    print(f"Findings: {error_count} errors, {warn_count} warnings")
    print(f"Duration: {report.duration_seconds:.1f}s")

    if args.output:
        report_path = args.output
    else:
        report_dir = root / "reports"
        report_dir.mkdir(exist_ok=True)
        report_path = report_dir / f"{correlation_id}.json"

    report_path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.loads(report.to_json())
    payload["baseline"] = args.baseline
    report_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    print(f"Commit report: {report_path}")

    # The exit code follows ``success`` in a dry run too: a review that
    # aborts the commit is a failure whether or not --apply was given.
    if not report.success or (error_count > 0 and not dry_run):
        return 1
    return 0


def _cmd_report(args: argparse.Namespace) -> int:
    if not args.path.exists():
        print(f"Error: file not found: {args.path}", file=sys.stderr)
        return 2

    data = json.loads(args.path.read_text(encoding="utf-8"))

    passes = data.get("passes_run", [])
    files = data.get("files_written", [])
    findings = data.get("findings", [])
    duration = data.get("duration_seconds", 0.0)
    dry_run = data.get("dry_run", True)
    correlation_id = data.get("correlation_id", "")
    success = data.get("success", True)
    diffstat = data.get("diffstat", "")

    print(f"Passes run:     {', '.join(passes) if passes else '(none)'}")
    print(f"Files written:  {len(files)}")
    print(f"Duration:       {duration:.1f}s")
    print(f"Dry run:        {dry_run}")
    print(f"Success:        {success}")
    print(f"Correlation ID: {correlation_id}")

    if diffstat:
        print(f"\nDiff:\n{diffstat}")

    if findings:
        error_count = len([f for f in findings if f.get("severity") == "error"])
        warn_count = len([f for f in findings if f.get("severity") == "warning"])
        print(
            f"\nFindings ({len(findings)}): {error_count} errors, {warn_count} warnings"
        )
        for f in findings:
            line = f.get("line", "?")
            print(
                f"  [{f.get('severity', '?')}] {f.get('rule', '?')} "
                f"at {f.get('path', '?')}:{line} — {f.get('message', '?')}"
            )
    else:
        print("\nNo findings.")

    return 0


def _resolve_cig_node(cig: CodeIntelligenceGraph, module: str) -> int | None:
    from orchestrata_ultimate.refactor.codemod.analysis.protocol import (
        resolve_node_id,
    )

    return resolve_node_id(cig, module)


_ROUTED_QUERIES = ("blast_radius", "coupling_score", "abstraction_level")
_PASS_QUERIES = ("breaking_changes", "complexity", "test_coverage", "dead_code")


def _changes_since(ws: CodemodWorkspace, module: str, base: str) -> ChangeSet:
    """A one-entry ChangeSet: ``module`` at git ref ``base`` vs the loaded tree.

    Raises:
        ValueError: the module is not in the forest, or ``git show`` /
            parsing of the base version fails.
    """
    from orchestrata_ultimate.refactor.codemod.report import ChangeEntry, ChangeSet
    from orchestrata_ultimate.refactor.ctok import LosslessModule

    node_id = _resolve_cig_node(ws.cig, module)
    path = ws.cig.nodes[node_id].path if node_id is not None else module
    mutated = ws.forest.get(path)
    if mutated is None:
        raise ValueError(f"module '{module}' not found in the workspace")
    shown = subprocess.run(
        # ``./`` makes git resolve the path from the -C directory, not the
        # repository top level (``--repo`` may be a subdirectory).
        ["git", "-C", str(ws.root), "show", f"{base}:./{path}"],
        capture_output=True,
        check=False,
    )
    if shown.returncode != 0:
        detail = shown.stderr.decode("utf-8", "replace").strip()
        raise ValueError(f"cannot read {path} at {base}: {detail}")
    try:
        original = LosslessModule.parse(shown.stdout)
    except SyntaxError as exc:
        raise ValueError(f"{path} at {base} does not parse: {exc}") from exc
    entry = ChangeEntry(
        path=path,
        original_tree=original,
        mutated_tree=mutated,
        pass_name="working-tree-diff",
    )
    return ChangeSet(entries=(entry,))


def _run_analysis(
    name: str, ws: CodemodWorkspace, module: str | None, args: argparse.Namespace
) -> object:
    """Run one analysis pass over the loaded workspace (never writes it)."""
    from orchestrata_ultimate.refactor.codemod import analysis
    from orchestrata_ultimate.refactor.codemod.report import ChangeSet

    cig = ws.cig
    if name == "blast_radius":
        return analysis.BlastRadiusAnalyzer(cig).analyze(module, hops=args.hops)
    if name == "coupling_score":
        return analysis.CouplingAnalyzer(cig).analyze(module)
    if name == "abstraction_level":
        return analysis.AbstractionAnalyzer(cig).analyze(module)
    if name == "complexity":
        return analysis.ComplexityAnalyzer(cig, ws.forest).analyze(module)
    if name == "test_coverage":
        return analysis.TestCoverageMapper(cig).analyze(module)
    if name == "dead_code":
        return analysis.DeadCodeDetector(cig).analyze(module)
    # breaking_changes: needs a ChangeSet; without --base there is no diff.
    changes = (
        _changes_since(ws, module, args.base) if module and args.base else ChangeSet()
    )
    return analysis.BreakingChangeDetector(cig).analyze(module, changes=changes)


def _module_paths(ws: CodemodWorkspace) -> list[str]:
    return sorted(
        node.path for node in ws.cig.nodes.values() if node.node_type == "Module"
    )


def _run_all_passes(
    ws: CodemodWorkspace, module: str | None, args: argparse.Namespace
) -> dict[str, object]:
    """The seven analyzers, keyed by analyzer name.

    The module-scoped ones (blast_radius, coupling_score, abstraction_level)
    run for ``module`` when given, else for every module (a mapping keyed by
    path).
    """
    results: dict[str, object] = {}
    for name in (*_ROUTED_QUERIES, *_PASS_QUERIES):
        if name in _ROUTED_QUERIES and module is None:
            results[name] = {
                path: _run_analysis(name, ws, path, args) for path in _module_paths(ws)
            }
        else:
            results[name] = _run_analysis(name, ws, module, args)
    return results


def _summarize(item: object) -> object:
    """Counts in place of the id/path collections of a BlastRadius."""
    from orchestrata_ultimate.refactor.codemod._types import BlastRadius

    if isinstance(item, BlastRadius):
        return {
            "affected_node_count": len(item.affected_nodes),
            "affected_file_count": len(item.affected_files),
            "impact_score": item.impact_score,
        }
    return item


def _encode_all(results: dict[str, object]) -> dict[str, object]:
    from orchestrata_ultimate.refactor.codemod.analysis import AnalysisResultEncoder

    encoded: dict[str, object] = {}
    for name, result in results.items():
        if isinstance(result, dict):
            # The per-module mapping carries counts, never node-id lists: a
            # whole-repo blast_radius id list was 147 MB on this repo.
            encoded[name] = {
                "type": "mapping",
                "data": {
                    key: AnalysisResultEncoder.encode(_summarize(item))["data"]
                    for key, item in result.items()
                },
            }
        else:
            encoded[name] = AnalysisResultEncoder.encode(result)
    return encoded


def _print_analysis_text(name: str, result: object, cig: CodeIntelligenceGraph) -> None:
    """Human-readable summary of one analysis result."""
    from orchestrata_ultimate.refactor.codemod._types import (
        ComplexityReport,
        DeadCodeReport,
        TestCoverageMap,
    )

    print(f"== {name} ==")
    if isinstance(result, dict):
        print(f"  {len(result)} modules analyzed (use --json for the full mapping)")
    elif isinstance(result, ComplexityReport):
        print(f"  Files:           {result.file_count}")
        print(f"  Functions:       {len(result.functions)}")
        print(f"  Max cyclomatic:  {result.max_cyclomatic}")
        print(f"  Mean cyclomatic: {result.mean_cyclomatic:.2f}")
    elif isinstance(result, TestCoverageMap):
        print(f"  Covered:         {len(result.covered)}")
        print(f"  Uncovered:       {len(result.uncovered)}")
        print(f"  Heuristic only:  {len(result.heuristic_only)}")
        print(f"  Coverage ratio:  {result.coverage_ratio:.2f}")
    elif isinstance(result, DeadCodeReport):
        print(f"  Granularity:     {result.granularity}")
        print(f"  Entry roots:     {len(result.roots)}")
        for nid in sorted(result.unreachable_modules):
            node = cig.nodes.get(nid)
            print(f"  unreachable: {node.path if node else nid}")
    elif isinstance(result, list):
        print(f"  Breaking changes: {len(result)}")
        for item in result:
            print(
                f"  {item.consumer_path}:{item.consumer_line} "
                f"<- {item.change_path} ({item.incompatibility})"
            )
    else:
        print(f"  {result}")


def _cmd_analyze_passes(
    args: argparse.Namespace, ws: CodemodWorkspace, module: str | None
) -> int:
    """The analysis-pass queries (REQ-110-031): print text or ``--json``."""
    if args.query == "breaking_changes" and not (module and args.base):
        print("Error: breaking_changes needs both --module and --base", file=sys.stderr)
        return 2
    try:
        if args.query == "all-passes":
            results = _run_all_passes(ws, module, args)
        else:
            results = {args.query: _run_analysis(args.query, ws, module, args)}
    except ValueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1
    if args.json:
        encoded = _encode_all(results)
        print(
            json.dumps(encoded if args.query == "all-passes" else encoded[args.query])
        )
        return 0
    for name, result in results.items():
        _print_analysis_text(name, result, ws.cig)
    return 0


def _print_routed(args: argparse.Namespace, result: object) -> None:
    """Render a blast_radius / coupling_score / abstraction_level result.

    Shared by library and daemon mode: both hand it the same dataclasses.
    """
    from orchestrata_ultimate.refactor.codemod._types import (
        AbstractionLevel,
        BlastRadius,
        CouplingScore,
    )

    module = args.module
    if args.json:
        from orchestrata_ultimate.refactor.codemod.analysis import (
            AnalysisResultEncoder,
        )

        print(json.dumps(AnalysisResultEncoder.encode(result)))
        return
    if isinstance(result, BlastRadius):
        print(f"Blast radius of {module} ({args.hops} hops):")
        print(f"  Affected nodes: {len(result.affected_nodes)}")
        print(f"  Affected files: {len(result.affected_files)}")
        print(f"  Impact score:   {result.impact_score:.2f}")
    elif isinstance(result, CouplingScore):
        print(f"Coupling score for {module}:")
        print(f"  Efferent:     {result.efferent}")
        print(f"  Afferent:     {result.afferent}")
        print(f"  Instability:  {result.instability:.2f}")
    elif isinstance(result, AbstractionLevel):
        print(f"Abstraction level for {module}:")
        print(f"  Abstract:  {result.abstract_count}")
        print(f"  Concrete:  {result.concrete_count}")
        print(f"  Ratio:     {result.ratio:.2f}")


NameOf = Callable[[int], "str | None"]


def _print_node_list(title: str, ids: list[int], name_of: NameOf) -> None:
    print(title)
    for node_id in ids:
        name = name_of(node_id)
        if name is not None:
            print(f"  - {name}")


def _print_cycles(cycles: list[list[int]], name_of: NameOf) -> None:
    if not cycles:
        print("No import cycles found.")
        return
    print(f"Import cycles found: {len(cycles)}")
    for cycle in cycles:
        names = [n for n in (name_of(nid) for nid in cycle) if n is not None]
        print(f"  {' -> '.join(names)}")


def _print_topological_order(order: list[int], name_of: NameOf) -> None:
    print(f"Topological order ({len(order)} modules):")
    for nid in order[:20]:
        name = name_of(nid)
        if name is not None:
            print(f"  - {name}")
    if len(order) > 20:
        print(f"  ... and {len(order) - 20} more")


def _print_centrality(
    results: list[tuple[int, float]], name_of: NameOf, top_k: int, exact: bool
) -> None:
    mode = "exact" if exact else "approximate"
    print(f"Betweenness centrality ({mode}, top {top_k}):")
    for nid, centrality_score in results:
        name = name_of(nid)
        print(
            f"  {name if name is not None else f'node:{nid}'}: {centrality_score:.4f}"
        )


class _UnknownModule(Exception):
    """``--module`` names no node of the code intelligence graph."""


class _LibraryQueries:
    """The graph queries of ``analyze``, answered from a workspace parsed here."""

    def __init__(self, ws: CodemodWorkspace, args: argparse.Namespace) -> None:
        self._ws = ws
        self._cig = ws.cig
        self._args = args

    def _node(self, module: str) -> int:
        node_id = _resolve_cig_node(self._cig, module)
        if node_id is None:
            raise _UnknownModule(module)
        return node_id

    def name_of(self, ids: list[int]) -> NameOf:
        nodes = self._cig.nodes
        return lambda nid: nodes[nid].qualified_name if nid in nodes else None

    def dependencies(self, module: str) -> list[int]:
        return self._cig.dependencies(self._node(module))

    def dependents(self, module: str) -> list[int]:
        return self._cig.dependents(self._node(module))

    def cycles(self) -> list[list[int]]:
        return self._cig.cycles()

    def topological_order(self) -> list[int]:
        return self._cig.topological_order()

    def centrality(self) -> list[tuple[int, float]]:
        return self._cig.centrality(top_k=self._args.top_k, exact=self._args.exact)

    def impact_score(self, module: str) -> float:
        return self._cig.pagerank_score(self._node(module))

    def routed(self, module: str) -> object:
        self._node(module)
        return _run_analysis(self._args.query, self._ws, module, self._args)


class _DaemonQueries:
    """The same queries, answered by the daemon serving the repo."""

    def __init__(self, client: CodemodClient, args: argparse.Namespace) -> None:
        self._client = client
        self._args = args

    @staticmethod
    def _query(module: str, call: Callable[[str], Any]) -> Any:
        from orchestrata_ultimate.refactor.codemod.client import CodemodRpcError

        try:
            return call(module)
        except CodemodRpcError as exc:
            # -32602 on a query is the server's "Unknown node": the module
            # key resolved to nothing, the refusal library mode raises itself.
            if exc.code == -32602:
                raise _UnknownModule(module) from exc
            raise

    def name_of(self, ids: list[int]) -> NameOf:
        names = {
            record["id"]: record["qualified_name"]
            for record in self._client.resolve_nodes(ids)
        }
        return names.get

    def dependencies(self, module: str) -> list[int]:
        return self._query(module, self._client.query_dependencies)

    def dependents(self, module: str) -> list[int]:
        return self._query(module, self._client.query_dependents)

    def cycles(self) -> list[list[int]]:
        return self._client.query_cycles()

    def topological_order(self) -> list[int]:
        return self._client.query_topological_order()

    def centrality(self) -> list[tuple[int, float]]:
        return self._client.query_centrality(self._args.top_k, self._args.exact)

    def impact_score(self, module: str) -> float:
        return self._query(module, self._client.query_impact_score)

    def routed(self, module: str) -> object:
        from orchestrata_ultimate.refactor.codemod._types import (
            AbstractionLevel,
            BlastRadius,
            CouplingScore,
        )

        query = self._args.query
        if query == "blast_radius":
            record = self._query(
                module,
                lambda m: self._client.query_blast_radius(m, self._args.hops),
            )
            return BlastRadius(
                affected_nodes=frozenset(record["affected_nodes"]),
                affected_files=frozenset(record["affected_files"]),
                impact_score=record["impact_score"],
            )
        if query == "coupling_score":
            return CouplingScore(
                **self._query(module, self._client.query_coupling_score)
            )
        return AbstractionLevel(
            **self._query(module, self._client.query_abstraction_level)
        )


def _render_graph_query(
    args: argparse.Namespace, queries: _LibraryQueries | _DaemonQueries
) -> None:
    """Print one non-pass ``analyze`` query; both modes share every line."""
    query = args.query
    module = args.module
    if query in _ROUTED_QUERIES:
        _print_routed(args, queries.routed(module))
    elif query in ("dependencies", "dependents"):
        ids = getattr(queries, query)(module)
        label = "Dependencies" if query == "dependencies" else "Dependents"
        _print_node_list(f"{label} of {module}:", ids, queries.name_of(ids))
    elif query == "cycles":
        cycles = queries.cycles()
        _print_cycles(cycles, queries.name_of([nid for c in cycles for nid in c]))
    elif query == "topological_order":
        order = queries.topological_order()
        _print_topological_order(order, queries.name_of(order[:20]))
    elif query == "centrality":
        results = queries.centrality()
        _print_centrality(
            results,
            queries.name_of([nid for nid, _ in results]),
            args.top_k,
            args.exact,
        )
    elif query == "impact_score":
        impact = queries.impact_score(module)
        print(f"Impact score for {module}: {impact:.4f}")


_MODULE_QUERIES = (
    "dependencies",
    "dependents",
    "blast_radius",
    "coupling_score",
    "abstraction_level",
    "impact_score",
)


@traces(spec="110", story="US-110-013", reqs=["REQ-110-140"])
def _cmd_analyze(args: argparse.Namespace, client: CodemodClient | None = None) -> int:
    """``analyze``: graph queries from the daemon ``client`` is connected to, or
    from a workspace parsed in this process (library mode)."""
    query = args.query
    module = args.module

    if query in (*_PASS_QUERIES, "all-passes"):
        # The analysis passes walk the parsed forest; the daemon exposes no
        # method for them (S4 backlog), so they run here either way.
        if client is not None:
            print(
                f"codemod: '{query}' has no daemon method; parsed locally",
                file=sys.stderr,
            )
        return _cmd_analyze_passes(args, _load_workspace(args), module)

    if query in _MODULE_QUERIES and not module:
        print(f"Error: --module required for {query}", file=sys.stderr)
        return 2

    queries: _LibraryQueries | _DaemonQueries
    if client is not None:
        queries = _DaemonQueries(client, args)
    else:
        queries = _LibraryQueries(_load_workspace(args), args)
    try:
        _render_graph_query(args, queries)
    except _UnknownModule:
        print(f"Error: module '{module}' not found in CIG", file=sys.stderr)
        return 1
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = _build_parser()

    try:
        args = parser.parse_args(argv)
    except SystemExit as e:
        return 2 if e.code != 0 else 0

    if args.auto_start and not args.no_daemon:
        failure = _auto_start_daemon(args.repo.resolve())
        if failure is not None:
            print(failure, file=sys.stderr)
            return 1

    if args.command == "load-bench":
        return _cmd_load_bench(args)
    if args.command == "report":
        return _cmd_report(args)
    if args.command not in ("run", "analyze"):
        return 2

    # run / analyze: use the daemon serving the repo when one is running
    # (REQ-110-140); otherwise parse in this process and say so. --no-daemon
    # selects library mode quietly.
    client = (
        None if args.no_daemon else _connect_or_warn(args.repo.resolve(), args.include)
    )
    try:
        if args.command == "run":
            return _cmd_run(args, client)
        return _cmd_analyze(args, client)
    finally:
        if client is not None:
            client.close()


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