scripts.tools_import_closure

Reference › scripts › tools_import_closure

scripts/tools_import_closure.py computes the static import closure of the tools-deploy entry modules over src/: the set of modules the deployed tools tree needs, and the third-party packages those modules pull in. The deploy tooling (see vendor_members_hook for the packaging side) and a human use it to decide what to vendor. It has no internal imports, because it reads the source with ast rather than importing it.

build_closure returns a walker; compute(entries) follows imports from the entry modules, treating imports under a TYPE_CHECKING guard as non-runtime (_is_type_checking) and resolving relative names (_names, _top_package). With --eager only module-level imports count, so lazy function-level imports are excluded. The entries come from --entries, defaulting to DEPLOYED_ENTRIES. Output modes, one per run: --json, --manifest (vendor manifest lines), --third-party (third-party names), --pip (the distributions the virtual environment needs to import the deployed tree, which uses the eager closure plus the entry scripts) and --check, which prints "closure OK" or lists every unresolved import on standard error and exits 1, naming a vendoring gap. --src overrides the source root.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

ImportCollector (protocol)

Bases: Protocol · Line 85

Strategy: which import statements of a parsed module count.

_Candidate

Bases: str · Line 93

from m import n yields m.n: a module only if one exists, else an attribute of m.

AllScopesImportCollector

Line 116

Every import anywhere in the module -- function bodies and TYPE_CHECKING too.

EagerImportCollector

Line 128

Only what import executes at runtime: module/class level, outside if TYPE_CHECKING: and outside function bodies.

ClosureReport (dataclass)

Decorators: dataclass(frozen=True) · Line 153

The closure: modules, per-package counts, third-party names, unresolved names.

ImportClosure

Line 191

Walks entries over src_root with an injected :class:ImportCollector.

Functions

Source

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

#!/usr/bin/env python3
"""Static import closure of the tools-deploy entry modules over ``src/``.

``deploy/install-tools.sh`` used to vendor a hand-written list of four subtrees.
Measured 2026-10-08: the codemod daemon imports in ONE eager module, but its first
default review (``InMemoryReviewEngine(registry=None)`` -> ``review_checks``) pulls
103 eager modules across six packages (core 40, refactor 34, verification 14,
openspec 7, knowledge_graph 2, orchestrata_ast 5) -- ``verification`` and
``openspec`` were never vendored, so the first ship of 3bb5319df failed with
``ModuleNotFoundError: orchestrata_ultimate.verification``.  Deferring that import
hid the gap instead of closing it.  This script COMPUTES what must be vendored and
which third-party distributions the deployed venv needs, so neither list is guessed.

STDLIB ONLY: it runs inside the installer, where the repo venv is not guaranteed.

Task 580 / plans/recorded-run-2026-10-08-eight-failures-design.md section 3.2.

Usage::

    tools_import_closure.py [--entries MOD ...] [--eager] [--json] [--src DIR]
    tools_import_closure.py --manifest     # vendor manifest lines (one per line)
    tools_import_closure.py --third-party  # third-party import names (one per line)
"""

from __future__ import annotations

import argparse
import ast
import json
import sys
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol

OURS: tuple[str, ...] = (
    "orchestrata_ultimate",
    "orchestrata_ast",
    "orchestrata_tools",
    "scripts",
)
ROOT_PACKAGE = "orchestrata_ultimate"

# Every module the deployed tree executes: the codemod daemon, the review checks its
# first default review loads lazily (review_engine.py), and the pytest observer pair
# plus the KG syncer that ast_live_sync imports.  Measured 2026-10-08 on this tree.
DEPLOYED_ENTRIES: tuple[str, ...] = (
    "orchestrata_ultimate.refactor.codemod._daemon_entry",
    "orchestrata_ultimate.refactor.codemod.review_checks",
    "orchestrata_ultimate.core.threading.pytest_live_reader",
    "orchestrata_ultimate.core.threading.pytest_live_observer",
    "orchestrata_ultimate.knowledge_graph.sync",
)

# The deployed entry scripts (install-tools.sh archive list).  They live outside
# src/, so they feed only the third-party computation (``--pip``), never the vendor
# manifest.
SCRIPT_ENTRIES: tuple[str, ...] = (
    "scripts.swarm_supervisor",
    "scripts.observe_pytest",
    "scripts.ast_live_sync",
    "scripts.ast_daemon",
    "scripts.ast_api",
    "scripts.serialize_codebase_ast",
    "scripts.build_runtime_callgraph",
)

# Third-party imports built from the staged sources by their own installer step
# (not installed from an index), and import-name -> distribution-name spellings.
LOCAL_NATIVES: frozenset[str] = frozenset({"codemod_fast", "ctok"})
# pytest is the TARGET project's own: `pytest-live` execs it, and the deployed venv
# ships no pytest BY DESIGN (install-tools.sh proves the reader runs without it).
PROJECT_PROVIDED: frozenset[str] = frozenset({"pytest", "_pytest"})
DIST_NAMES: Mapping[str, str] = {
    "yaml": "pyyaml",
    "pydantic_settings": "pydantic-settings",
    "prometheus_client": "prometheus-client",
    "janus_swi": "janus-swi",
}

REPO_ROOT = Path(__file__).resolve().parent.parent
REPO_SRC = REPO_ROOT / "src"


class ImportCollector(Protocol):
    """Strategy: which import statements of a parsed module count."""

    def imports_of(self, tree: ast.Module) -> Iterator[str]:
        """Yield absolute dotted names; ``from m import n`` yields ``m`` and ``m.n``."""
        ...


class _Candidate(str):
    """``from m import n`` yields ``m.n``: a module only if one exists, else an
    attribute of ``m``.  Marked so a missing one is not reported as unresolved."""


def _names(node: ast.Import | ast.ImportFrom) -> Iterator[str]:
    """Absolute names an import statement can bind (relative imports are banned)."""
    if isinstance(node, ast.Import):
        for alias in node.names:
            yield alias.name
    elif node.module and node.level == 0:
        yield node.module
        for alias in node.names:
            yield _Candidate(f"{node.module}.{alias.name}")


def _is_type_checking(test: ast.expr) -> bool:
    """True for ``TYPE_CHECKING`` / ``typing.TYPE_CHECKING``."""
    if isinstance(test, ast.Name):
        return test.id == "TYPE_CHECKING"
    return isinstance(test, ast.Attribute) and test.attr == "TYPE_CHECKING"


class AllScopesImportCollector:
    """Every import anywhere in the module -- function bodies and TYPE_CHECKING too.

    What a long-lived daemon can execute eventually, so it decides what is vendored.
    """

    def imports_of(self, tree: ast.Module) -> Iterator[str]:
        for node in ast.walk(tree):
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                yield from _names(node)


class EagerImportCollector:
    """Only what ``import`` executes at runtime: module/class level, outside
    ``if TYPE_CHECKING:`` and outside function bodies."""

    def imports_of(self, tree: ast.Module) -> Iterator[str]:
        yield from self._block(tree.body)

    def _block(self, body: Sequence[ast.stmt]) -> Iterator[str]:
        for node in body:
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                yield from _names(node)
            elif isinstance(node, ast.If):
                if not _is_type_checking(node.test):
                    yield from self._block(node.body)
                yield from self._block(node.orelse)
            elif isinstance(node, ast.Try):
                for part in (node.body, node.orelse, node.finalbody):
                    yield from self._block(part)
                for handler in node.handlers:
                    yield from self._block(handler.body)
            elif isinstance(node, (ast.With, ast.ClassDef)):
                yield from self._block(node.body)


@dataclass(frozen=True)
class ClosureReport:
    """The closure: modules, per-package counts, third-party names, unresolved names."""

    modules: frozenset[str]
    top_packages: Mapping[str, int]
    third_party: frozenset[str]
    unresolved: frozenset[str]
    vendor_roots: frozenset[str] = frozenset()

    def vendor_manifest(self) -> list[str]:
        """Sorted ``src/...`` paths (subtree dirs or single files) to archive."""
        return sorted(self.vendor_roots)

    def pip_requirements(self) -> list[str]:
        """Distributions to ``uv pip install``: third-party names minus the local
        natives (built separately) and the project-provided pytest."""
        names = self.third_party - LOCAL_NATIVES - PROJECT_PROVIDED
        return sorted(DIST_NAMES.get(n, n) for n in names)

    def as_json(self) -> dict[str, object]:
        return {
            "module_count": len(self.modules),
            "modules": sorted(self.modules),
            "top_packages": dict(sorted(self.top_packages.items())),
            "third_party": sorted(self.third_party),
            "pip_requirements": self.pip_requirements(),
            "unresolved": sorted(self.unresolved),
            "vendor_manifest": self.vendor_manifest(),
        }


def _top_package(module: str) -> str:
    parts = module.split(".")
    return (
        ".".join(parts[:2]) if parts[0] == ROOT_PACKAGE and len(parts) > 1 else parts[0]
    )


class ImportClosure:
    """Walks ``entries`` over ``src_root`` with an injected :class:`ImportCollector`."""

    def __init__(
        self,
        src_root: Path,
        collector: ImportCollector,
        extra_roots: Sequence[Path] = (),
    ) -> None:
        self._src_root = Path(src_root)
        self._collector = collector
        # ``scripts.*`` entries resolve under the repo root, not under src/.
        self._roots = (self._src_root, *map(Path, extra_roots))

    def _locate(self, name: str) -> Path | None:
        """Module file for ``name``; a directory with no ``__init__.py`` is NOT a module."""
        rel = Path(*name.split("."))
        for root in self._roots:
            for cand in (root / rel.with_suffix(".py"), root / rel / "__init__.py"):
                if cand.exists():
                    return cand
        return None

    def _is_dir(self, name: str) -> bool:
        rel = Path(*name.split("."))
        return any((root / rel).is_dir() for root in self._roots)

    def compute(self, entries: Sequence[str]) -> ClosureReport:
        seen: dict[str, Path | None] = {}
        third: set[str] = set()
        hard: set[str] = set(entries)  # names that MUST be modules, not attributes
        stack = list(entries)
        while stack:
            name = stack.pop()
            if name in seen:
                continue
            path = self._locate(name)
            seen[name] = path
            if path is None:
                continue
            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
            for dep in self._collector.imports_of(tree):
                top = dep.split(".")[0]
                if top in OURS:
                    parts = dep.split(".")
                    # importing a.b.c executes a and a.b first
                    chain = [".".join(parts[:i]) for i in range(1, len(parts) + 1)]
                    if isinstance(dep, _Candidate):
                        chain = [str(dep)]  # its prefixes came with ``from m import``
                    else:
                        hard.update(chain)
                    stack.extend(chain)
                elif top not in sys.stdlib_module_names:
                    third.add(top)
        modules = frozenset(m for m, p in seen.items() if p is not None)
        counts: dict[str, int] = {}
        for module in modules:
            top = _top_package(module)
            counts[top] = counts.get(top, 0) + 1
        return ClosureReport(
            modules=modules,
            top_packages=counts,
            third_party=frozenset(third),
            unresolved=frozenset(self._unresolved(seen, hard)),
            vendor_roots=frozenset(self._vendor_roots(seen)),
        )

    def _unresolved(
        self, seen: Mapping[str, Path | None], hard: set[str]
    ) -> Iterator[str]:
        """Names with no module file that had to be one.

        ``from pkg import Name`` yields the candidate ``pkg.Name``, legitimately not a
        module when ``Name`` is an attribute, so a candidate is skipped -- EXCEPT a
        directory with no ``__init__.py`` (an accidental namespace package), which is
        always reported.
        """
        for name, path in seen.items():
            if path is None and (name in hard or self._is_dir(name)):
                yield name

    def _vendor_roots(self, seen: Mapping[str, Path | None]) -> Iterator[str]:
        """``src/`` subtrees to archive; modules outside ``src/`` (scripts) have none."""
        for module, path in seen.items():
            if path is None or not path.is_relative_to(self._src_root):
                continue
            parts = module.split(".")
            if parts[0] == ROOT_PACKAGE and len(parts) == 1:
                yield f"src/{ROOT_PACKAGE}/__init__.py"
            elif parts[0] == ROOT_PACKAGE:
                two = self._src_root / ROOT_PACKAGE / parts[1]
                suffix = "" if two.is_dir() else ".py"
                yield f"src/{ROOT_PACKAGE}/{parts[1]}{suffix}"
            else:
                yield f"src/{parts[0]}"


def build_closure(src_root: Path, *, eager: bool) -> ImportClosure:
    """The closure walker with the collector strategy ``--eager`` selects.

    ``scripts.*`` entries resolve against the directory above ``src_root``.
    """
    collector: ImportCollector = (
        EagerImportCollector() if eager else AllScopesImportCollector()
    )
    return ImportClosure(src_root, collector, extra_roots=(Path(src_root).parent,))


def main(argv: Sequence[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=(__doc__ or "").split("\n\n")[0])
    ap.add_argument("--entries", nargs="+", default=None)
    ap.add_argument("--eager", action="store_true", help="module-level imports only")
    ap.add_argument("--json", action="store_true")
    ap.add_argument(
        "--manifest", action="store_true", help="print vendor manifest lines"
    )
    ap.add_argument(
        "--third-party", action="store_true", help="print third-party names"
    )
    ap.add_argument(
        "--pip", action="store_true", help="print pip distributions to install"
    )
    ap.add_argument(
        "--check",
        action="store_true",
        help="exit 1 naming every unresolved import (a vendoring gap)",
    )
    ap.add_argument("--src", type=Path, default=REPO_SRC)
    args = ap.parse_args(argv)
    # --pip answers "what must the venv hold to IMPORT the deployed tree": the eager
    # closure of the entries plus the entry scripts.  Everything else is lazy.
    entries = args.entries or list(DEPLOYED_ENTRIES)
    if args.pip and not args.entries:
        entries = [*DEPLOYED_ENTRIES, *SCRIPT_ENTRIES]
    report = build_closure(args.src, eager=args.eager or args.pip).compute(entries)
    if args.check:
        if report.unresolved:
            print("unresolved:", *sorted(report.unresolved), file=sys.stderr)
            return 1
        print(f"closure OK: {len(report.modules)} modules, nothing unresolved")
    elif args.manifest:
        print("\n".join(report.vendor_manifest()))
    elif args.pip:
        print("\n".join(report.pip_requirements()))
    elif args.third_party:
        print("\n".join(sorted(report.third_party)))
    elif args.json:
        print(json.dumps(report.as_json(), indent=2))
    else:
        print(f"modules in closure: {len(report.modules)}")
        for top, n in sorted(report.top_packages.items()):
            print(f"  {top:55s} {n}")
        print(f"third-party: {sorted(report.third_party)}")
        print(f"unresolved: {sorted(report.unresolved)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())