6AST import-graph extractor: closures, reverse edges, blind spots.
10Answer "what does this set of modules actually drag in?" with a number
11somebody else can reproduce. Built for plan 256, which had to re-derive
12the `oc_orchestrator` extraction closure and found that the V0.1
13analysis (2026-07-08) was computed by a script left in a session
14scratchpad, so none of its numbers could be re-run. This tool is the
15permanent fix: it is committed, it takes its seed set and its roots as
16arguments rather than hardcoding one question, and it is deterministic,
17so a reviewer can re-run it and diff the receipt.
19It parses with :mod:`ast` and NEVER imports the code it analyses. A
20module that raises at import time, or that does not even parse, is
21analysed (or listed as a blind spot) rather than crashing the run.
25- :class:`Module` / :class:`Edge` / :class:`BlindSpot` / :class:`Graph` —
26 the data model: one node per file, one edge per resolved import, one
27 blind spot per import this tool provably cannot resolve.
28- :func:`build_graph` — discover modules under the roots and extract
29 every edge, resolving relative imports and classifying each target as
30 standard library, third party, same package or sibling package.
31- :func:`closure` — the transitive closure of same-package edges from a
32 seed set, honouring `--cut` edge suppressions.
33- :func:`parse_layer` — several named seed sets measured against one
34 parsed tree, so a multi-layer question costs one receipt rather than
35 one whole-graph copy per layer.
36- :func:`reverse_edges` — who imports each in-tree module.
37- :func:`receipt` / :func:`render_table` — the JSON receipt and the
39- :func:`main` — the CLI.
45 L2 - lifecycle events (graph built, closure computed)
46 L3 - semantic details (per-module edge counts)
47 L4 - deep tracing (per-edge)
51- The code under measurement is never imported, only parsed.
52- Determinism is a requirement, not a nicety: two runs over an unchanged
53 tree produce byte-identical stdout and byte-identical JSON. The
54 receipt therefore carries no clock and no absolute path -- roots
55 appear as their directory NAME and module paths are root-relative
56 POSIX, so a receipt is machine-independent. The tree the numbers
57 belong to is pinned by the caller through ``--label``.
58- A lazy (function-body) import is a dependency and is counted as one;
59 it is flagged ``deferred`` so a reader can see which couplings are
60 cheap to cut. A ``TYPE_CHECKING`` import is likewise counted and
61 flagged through ``scope``.
62- Known limitation, stated rather than hidden: only the
63 ``importlib.import_module`` call shape is recognised as a string
64 import. ``__import__``, ``importlib.util.spec_from_file_location`` and
65 hand-rolled loaders are invisible to this tool. Grep for them
66 separately before trusting a closure to be complete.
67- Console output is ASCII only.
71- oc_diagnostics (structured debug logging via _dbg)
74from __future__
import annotations
81from dataclasses
import dataclass
82from pathlib
import Path
83from typing
import Iterable, Sequence
85from oc_diagnostics
import _dbg
87_dbg(
"OC_CHAT.MAIN",
"module",
"import graph extractor loaded", 4)
94THIRD_PARTY =
"third_party"
95SAME_PACKAGE =
"same_package"
96SIBLING =
"sibling_package"
101KIND_MODULE =
"module"
103KIND_STRING =
"string"
106SCOPE_RUNTIME =
"runtime"
107SCOPE_TYPE_CHECKING =
"type_checking"
110REASON_NON_LITERAL =
"import_module called with a non-literal argument"
111REASON_RELATIVE =
"relative import escapes the scanned roots"
112REASON_UNPARSABLE =
"file could not be parsed"
113REASON_SHADOWED =
"module name already claimed by an earlier root"
116_SKIP_DIRS = frozenset({
117 "__pycache__",
".git",
".venv",
".mypy_cache",
".pytest_cache",
118 ".ruff_cache",
"node_modules",
".formatter_archive",
123 """A named refusal: an unknown seed, an unreadable root, a bad cut."""
126@dataclass(frozen=
True)
128 """One parsed source file."""
136@dataclass(frozen=True)
138 """One resolved import, from an in-tree module to some target."""
149@dataclass(frozen=True)
151 """An import this tool provably cannot resolve, named not dropped."""
159@dataclass(frozen=True)
161 """The whole measured tree."""
163 modules: dict[str, Module]
164 edges: tuple[Edge, ...]
165 blind_spots: tuple[BlindSpot, ...]
168@dataclass(frozen=True)
170 """A suppressed edge class, as ``source_glob->target_glob``."""
176 """True when *edge* is suppressed by this cut."""
178 fnmatch.fnmatchcase(edge.source, self.
source)
179 and fnmatch.fnmatchcase(edge.target, self.
target)
183 """The round-trippable text form."""
184 return f
"{self.source}->{self.target}"
187@dataclass(frozen=True)
189 """The transitive same-package reach of one seed set."""
192 seeds: tuple[str, ...]
193 members: tuple[str, ...]
194 package_inits: tuple[str, ...]
195 cut_edges: tuple[Edge, ...]
196 external: dict[str, tuple[str, ...]]
201 """Modules in the closure, seeds included."""
206 """Modules in the closure that are not package markers."""
215 """Dotted name for *path* under *root*, and whether it is a package."""
216 parts = list(path.relative_to(root).parts)
217 if parts[-1] ==
"__init__.py":
218 return ".".join(parts[:-1]),
True
219 parts[-1] = parts[-1][: -len(
".py")]
220 return ".".join(parts),
False
223def _excluded(relative: str, excludes: Sequence[str]) -> bool:
224 """True when the root-relative POSIX path matches an exclude glob."""
225 return any(fnmatch.fnmatchcase(relative, pattern)
for pattern
in excludes)
229 roots: Sequence[Path],
231 excludes: Sequence[str] = (),
232) -> tuple[dict[str, Module], dict[str, Path], list[BlindSpot]]:
233 """Find every ``*.py`` under *roots*, first root wins on a clash."""
235 modules: dict[str, Module] = {}
236 files: dict[str, Path] = {}
237 shadowed: list[BlindSpot] = []
239 if not root.is_dir():
241 for path
in sorted(root.rglob(
"*.py")):
242 if any(part
in _SKIP_DIRS
for part
in path.relative_to(root).parts):
244 relative = path.relative_to(root).as_posix()
252 source=name, line=0, reason=REASON_SHADOWED,
259 is_package=is_package,
262 _dbg(
"OC_CHAT.MAIN", func, f
"{len(modules)} module(s) discovered", 2)
263 return modules, files, shadowed
267 """Read *path* as UTF-8, tolerating a stray decoding error."""
268 return path.read_text(encoding=
"utf-8", errors=
"replace")
272 """Physical line count, counting a final unterminated line."""
275 return text.count(
"\n") + (0
if text.endswith(
"\n")
else 1)
282def classify(target: str, *, package: str, siblings: Sequence[str]) -> str:
283 """Classify an import target by its top-level name."""
284 top = target.split(
".", 1)[0]
289 if top
in sys.stdlib_module_names:
299 module: str, is_package: bool, level: int, tail: str |
None,
301 """Absolute name for a relative import, or None when it escapes."""
302 parts = module.split(
".")
if module
else []
303 base = parts
if is_package
else parts[:-1]
305 if climb > len(base):
308 base = base[: len(base) - climb]
311 return ".".join([*base, tail]
if tail
else base)
315 """Names bound to ``importlib`` and to ``import_module``."""
316 modules: set[str] = set()
317 functions: set[str] = set()
318 for node
in ast.walk(tree):
319 if isinstance(node, ast.Import):
320 for alias
in node.names:
321 if alias.name ==
"importlib":
322 modules.add(alias.asname
or "importlib")
323 elif isinstance(node, ast.ImportFrom)
and node.module ==
"importlib":
324 for alias
in node.names:
325 if alias.name ==
"import_module":
326 functions.add(alias.asname
or "import_module")
327 return modules, functions
331 """True for ``if TYPE_CHECKING:`` in either spelling."""
332 if isinstance(test, ast.Name):
333 return test.id ==
"TYPE_CHECKING"
334 if isinstance(test, ast.Attribute):
335 return test.attr ==
"TYPE_CHECKING"
340 """Walk one module, recording edges and blind spots.
342 Tracks two pieces of context the plain visitor does not carry: are we
343 inside a function body (the import is deferred), and are we inside an
344 ``if TYPE_CHECKING:`` block (the import is typing-only).
351 index: Iterable[str],
353 siblings: Sequence[str],
354 importlib_names: tuple[set[str], set[str]],
355 source_lines: Sequence[str],
371 """A function body: everything inside it is deferred."""
374 self.generic_visit(node)
377 visit_FunctionDef = _visit_function
378 visit_AsyncFunctionDef = _visit_function
379 visit_Lambda = _visit_function
382 """``if TYPE_CHECKING:`` marks its body, never its else branch."""
386 for statement
in node.body:
387 self.visit(statement)
389 for statement
in node.orelse:
390 self.visit(statement)
392 self.generic_visit(node)
397 """The scope tag for an edge found at the current position."""
398 return SCOPE_TYPE_CHECKING
if self.
_type_checking else SCOPE_RUNTIME
400 def _emit(self, target: str, line: int, kind: str, seen: set[str]) ->
None:
401 """Record *target* plus every in-tree ancestor package of it."""
402 names: list[str] = []
403 parts = target.split(
".")
404 for depth
in range(1, len(parts) + 1):
405 candidate =
".".join(parts[:depth])
406 if candidate
in self.
index:
407 names.append(candidate)
408 if target
not in self.
index:
411 if name
in seen
or name == self.
module.name:
415 source=self.
module.name, target=name, line=line, kind=kind,
423 """Import-statement kind at the current position."""
424 return KIND_LAZY
if self.
_deferred else KIND_MODULE
427 """The source line, trimmed, for a blind-spot record."""
435 """``import a.b.c`` imports every package on the way down."""
436 seen: set[str] = set()
437 for alias
in node.names:
438 self.
_emit(alias.name, node.lineno, self.
_kind(), seen)
441 """``from X import y`` reaches X and, when it exists, ``X.y``."""
442 seen: set[str] = set()
450 source=self.
module.name, line=node.lineno,
451 reason=REASON_RELATIVE,
459 self.
_emit(base, node.lineno, self.
_kind(), seen)
460 for alias
in node.names:
461 if alias.name ==
"*":
463 candidate = f
"{base}.{alias.name}"
464 if candidate
in self.
index:
465 self.
_emit(candidate, node.lineno, self.
_kind(), seen)
470 """True for the ``importlib.import_module`` call shapes we know."""
472 if isinstance(target, ast.Name):
474 if isinstance(target, ast.Attribute)
and target.attr ==
"import_module":
476 return isinstance(base, ast.Name)
and (
482 """A string import is an edge; a computed one is a blind spot."""
484 first = node.args[0]
if node.args
else None
485 if isinstance(first, ast.Constant)
and isinstance(first.value, str):
486 self.
_emit(first.value, node.lineno, KIND_STRING, set())
489 source=self.
module.name, line=node.lineno,
490 reason=REASON_NON_LITERAL,
493 self.generic_visit(node)
497 roots: Sequence[Path | str],
500 siblings: Sequence[str] = (),
501 excludes: Sequence[str] = (),
503 """Discover every module under *roots* and extract its import edges."""
505 resolved = [Path(root)
for root
in roots]
506 modules, files, blind_spots =
discover(resolved, excludes=excludes)
507 edges: list[Edge] = []
508 for name
in sorted(modules):
509 module = modules[name]
510 text =
_read(files[name])
512 tree = ast.parse(text, filename=module.path)
513 except SyntaxError
as exc:
515 source=name, line=exc.lineno
or 0, reason=REASON_UNPARSABLE,
516 snippet=str(exc.msg)[:200],
520 module=module, index=modules, package=package,
522 source_lines=text.splitlines(),
524 extractor.visit(tree)
525 edges.extend(extractor.edges)
526 blind_spots.extend(extractor.blind_spots)
527 _dbg(
"OC_CHAT.MAIN", func, f
"{name}: {len(extractor.edges)} edge(s)", 4)
528 ordered = tuple(sorted(
530 key=
lambda edge: (edge.source, edge.line, edge.target, edge.kind),
532 spots = tuple(sorted(
533 blind_spots, key=
lambda spot: (spot.source, spot.line, spot.reason),
536 "OC_CHAT.MAIN", func,
537 f
"{len(ordered)} edge(s), {len(spots)} blind spot(s)", 2,
539 return Graph(modules=modules, edges=ordered, blind_spots=spots)
547 """Parse ``name=mod1,mod2`` into a named seed set."""
550 name, joined = spec.split(
"=", 1)
551 seeds = tuple(part.strip()
for part
in joined.split(
",")
if part.strip())
552 if not name.strip()
or not seeds:
554 return name.strip(), seeds
558 """Parse ``source_glob->target_glob`` into a :class:`Cut`."""
561 source, target = spec.split(
"->", 1)
562 if not source.strip()
or not target.strip():
564 return Cut(source=source.strip(), target=target.strip())
569 seeds: Sequence[str],
571 name: str =
"closure",
572 cuts: Sequence[Cut] = (),
574 """Transitive same-package reach of *seeds*, honouring *cuts*."""
576 unknown = [seed
for seed
in seeds
if seed
not in graph.modules]
579 "seed not found in the scanned tree: " +
", ".join(sorted(unknown)),
581 outgoing: dict[str, list[Edge]] = {}
582 for edge
in graph.edges:
583 outgoing.setdefault(edge.source, []).append(edge)
585 members: set[str] = set(seeds)
586 cut_edges: list[Edge] = []
587 pending = list(seeds)
589 current = pending.pop()
590 for edge
in outgoing.get(current, ()):
591 if any(cut.matches(edge)
for cut
in cuts):
592 cut_edges.append(edge)
594 if edge.classification != SAME_PACKAGE:
596 if edge.target
not in graph.modules
or edge.target
in members:
598 members.add(edge.target)
599 pending.append(edge.target)
601 ordered = tuple(sorted(members))
602 external: dict[str, set[str]] = {
603 STDLIB: set(), THIRD_PARTY: set(), SIBLING: set(),
605 for edge
in graph.edges:
606 if edge.source
not in members:
608 if any(cut.matches(edge)
for cut
in cuts):
610 if edge.classification
in external:
611 external[edge.classification].add(edge.target)
612 _dbg(
"OC_CHAT.MAIN", func, f
"{name}: closure of {len(ordered)}", 2)
615 seeds=tuple(sorted(seeds)),
618 member
for member
in ordered
if graph.modules[member].is_package
620 cut_edges=tuple(sorted(
622 key=
lambda edge: (edge.source, edge.line, edge.target),
624 external={key: tuple(sorted(value))
for key, value
in external.items()},
625 lines=sum(graph.modules[member].lines
for member
in ordered),
630 """For every in-tree module, the sorted set of modules importing it."""
631 importers: dict[str, set[str]] = {}
632 for edge
in graph.edges:
633 if edge.target
in graph.modules:
634 importers.setdefault(edge.target, set()).add(edge.source)
636 target: tuple(sorted(sources))
637 for target, sources
in sorted(importers.items())
645def _tally(values: Iterable[str]) -> dict[str, int]:
646 """Sorted count-by-value, for the receipt's summary blocks."""
647 counts: dict[str, int] = {}
649 counts[value] = counts.get(value, 0) + 1
650 return {key: counts[key]
for key
in sorted(counts)}
654 """One closure, as receipt JSON."""
657 "seeds": list(result.seeds),
659 "size_excluding_inits": result.size_excluding_inits,
660 "lines": result.lines,
661 "members": list(result.members),
662 "package_inits": list(result.package_inits),
663 "external": {key: list(value)
for key, value
in result.external.items()},
664 "cut_edges": [
_edge_payload(edge)
for edge
in result.cut_edges],
669 """One edge, as receipt JSON."""
671 "source": edge.source,
"target": edge.target,
"line": edge.line,
672 "kind": edge.kind,
"scope": edge.scope,
"deferred": edge.deferred,
673 "classification": edge.classification,
679 closures: Sequence[Closure],
682 params: dict[str, object],
683) -> dict[str, object]:
684 """Build the JSON receipt: no clock, no absolute path, fully sorted."""
686 _dbg(
"OC_CHAT.MAIN", func, f
"receipt for {len(closures)} closure(s)", 3)
688 "schema_version": SCHEMA_VERSION,
692 "modules": len(graph.modules),
693 "lines": sum(m.lines
for m
in graph.modules.values()),
694 "edges": len(graph.edges),
695 "edges_by_kind":
_tally(e.kind
for e
in graph.edges),
696 "edges_by_scope":
_tally(e.scope
for e
in graph.edges),
697 "edges_by_class":
_tally(e.classification
for e
in graph.edges),
698 "deferred_edges": sum(1
for e
in graph.edges
if e.deferred),
699 "blind_spots": len(graph.blind_spots),
703 "source": spot.source,
"line": spot.line,
704 "reason": spot.reason,
"snippet": spot.snippet,
706 for spot
in graph.blind_spots
711 "path": module.path,
"lines": module.lines,
712 "is_package": module.is_package,
714 for name, module
in sorted(graph.modules.items())
718 target: list(sources)
725 """The human table for one closure."""
728 f
"== closure: {result.name} ==",
729 f
"seeds ({len(result.seeds)}): {', '.join(result.seeds)}",
730 f
"size: {result.size} module(s), "
731 f
"{result.size_excluding_inits} excluding package markers, "
732 f
"{result.lines} line(s)",
736 for member
in result.members:
737 marker =
" (pkg)" if graph.modules[member].is_package
else ""
738 lines.append(f
" {graph.modules[member].lines:5d} {member}{marker}")
740 (SIBLING,
"sibling"), (THIRD_PARTY,
"third party"), (STDLIB,
"stdlib"),
742 values = result.external.get(key, ())
743 lines.append(f
" {title}: {', '.join(values) if values else 'none'}")
745 lines.append(f
" cut edges ({len(result.cut_edges)}):")
746 for edge
in result.cut_edges:
748 f
" {edge.source}:{edge.line} -> {edge.target} "
756 closures: Sequence[Closure],
759 roots: Sequence[str],
761 """The human report. ASCII only, and stable across runs."""
763 "== import graph ==",
765 f
"roots: {', '.join(roots)}",
766 f
"modules: {len(graph.modules)} "
767 f
"lines: {sum(m.lines for m in graph.modules.values())} "
768 f
"edges: {len(graph.edges)}",
769 f
"by kind: {_tally(e.kind for e in graph.edges)}",
770 f
"by class: {_tally(e.classification for e in graph.edges)}",
771 f
"deferred: {sum(1 for e in graph.edges if e.deferred)} "
773 f
"{sum(1 for e in graph.edges if e.scope == SCOPE_TYPE_CHECKING)}",
774 f
"blind spots: {len(graph.blind_spots)}",
776 for spot
in graph.blind_spots:
777 lines.append(f
" {spot.source}:{spot.line} {spot.reason}")
779 lines.append(f
" {spot.snippet}")
780 for result
in closures:
782 return "\n".join(lines) +
"\n"
790 """The CLI surface; every question is an argument, none is baked in."""
791 parser = argparse.ArgumentParser(
793 description=
"AST import-graph closures with reproducible receipts.",
796 "--root", action=
"append", required=
True, metavar=
"DIR",
797 help=
"directory on the import path; repeatable, first wins on clash",
800 "--package", required=
True,
801 help=
"top-level package whose edges the closure follows",
804 "--sibling", action=
"append", default=[], metavar=
"NAME",
805 help=
"top-level name to classify as a sibling package; repeatable",
808 "--seed", action=
"append", default=[], metavar=
"MODULE",
809 help=
"closure seed; repeatable",
812 "--closure-name", default=
"closure",
813 help=
"name recorded for this closure in the receipt",
816 "--layer", action=
"append", default=[], metavar=
"NAME=MODULES",
817 help=
"a named, comma-separated seed set; repeatable, so one run "
818 "and one receipt can cover several layers of the same tree",
821 "--cut", action=
"append", default=[], metavar=
"SRC->DST",
822 help=
"suppress matching edges; both sides accept glob wildcards. "
823 "Cuts are global to the run: a layer with no matching edge is "
827 "--exclude", action=
"append", default=[], metavar=
"GLOB",
828 help=
"skip root-relative POSIX paths matching this glob; repeatable",
831 "--json", metavar=
"PATH", help=
"write the JSON receipt here",
834 "--label", default=
"",
835 help=
"what tree these numbers belong to; recorded verbatim",
840def main(argv: Sequence[str] |
None =
None) -> int:
841 """Run one measurement. 0 on success, 1 on a named refusal."""
843 args =
build_parser().parse_args(list(argv)
if argv
is not None else None)
845 cuts = [
parse_cut(spec)
for spec
in args.cut]
846 layers: list[tuple[str, Sequence[str]]] = []
848 layers.append((args.closure_name, args.seed))
849 layers.extend(
parse_layer(spec)
for spec
in args.layer)
851 [Path(root)
for root
in args.root], package=args.package,
852 siblings=args.sibling, excludes=args.exclude,
854 closures: list[Closure] = []
855 for name, seeds
in layers:
856 closures.append(
closure(graph, seeds, name=name))
859 graph, seeds, name=f
"{name}+cuts", cuts=cuts,
861 except ImportGraphError
as exc:
862 print(f
"import_graph: {exc}", file=sys.stderr)
865 root_names = [Path(root).resolve().name
for root
in args.root]
866 print(
render_table(graph, closures, label=args.label, roots=root_names),
869 params: dict[str, object] = {
871 "package": args.package,
872 "siblings": sorted(args.sibling),
873 "excludes": sorted(args.exclude),
874 "seeds": sorted(args.seed),
875 "cuts": [cut.spec()
for cut
in cuts],
876 "closure_name": args.closure_name,
877 "layers": sorted(args.layer),
879 document =
receipt(graph, closures, label=args.label, params=params)
880 text = json.dumps(document, indent=2, sort_keys=
True) +
"\n"
881 Path(args.json).write_text(text, encoding=
"utf-8", newline=
"\n")
882 _dbg(
"OC_CHAT.MAIN", func, f
"receipt written to {args.json}", 2)
886if __name__ ==
"__main__":
887 raise SystemExit(
main())
int size_excluding_inits(self)
bool matches(self, Edge edge)
bool _is_type_checking(ast.expr test)
Closure closure(Graph graph, Sequence[str] seeds, *, str name="closure", Sequence[Cut] cuts=())
int _count_lines(str text)
tuple[set[str], set[str]] _importlib_aliases(ast.AST tree)
str|None _resolve_relative(str module, bool is_package, int level, str|None tail)
list[str] _closure_block(Graph graph, Closure result)
bool _excluded(str relative, Sequence[str] excludes)
int main(Sequence[str]|None argv=None)
dict[str, tuple[str,...]] reverse_edges(Graph graph)
argparse.ArgumentParser build_parser()
Graph build_graph(Sequence[Path|str] roots, *, str package, Sequence[str] siblings=(), Sequence[str] excludes=())
str classify(str target, *, str package, Sequence[str] siblings)
tuple[str, tuple[str,...]] parse_layer(str spec)
str render_table(Graph graph, Sequence[Closure] closures, *, str label, Sequence[str] roots)
dict[str, int] _tally(Iterable[str] values)
dict[str, object] _edge_payload(Edge edge)
dict[str, object] receipt(Graph graph, Sequence[Closure] closures, *, str label, dict[str, object] params)
tuple[str, bool] _module_name(Path root, Path path)
tuple[dict[str, Module], dict[str, Path], list[BlindSpot]] discover(Sequence[Path] roots, *, Sequence[str] excludes=())
dict[str, object] _closure_payload(Closure result)