orchestrata_ast.analyzer

Reference › orchestrata_ast › analyzer

This is the engine of the code-AST component: it walks a source tree with the standard ast module and builds the CodebaseAST described in models, using nothing outside the standard library. It was extracted from the serializer script. Its users are scripts.serialize_codebase_ast, the entry point of the deployed ast command in /opt/orchestrata-tools/bin, scripts.ast_live_sync, the gate fact_extractor, the implementation_scanner and ast_pipeline.

Main parts:

• CodebaseAnalyzer takes a name, a root, optional extra roots and an optional anchor directory. analyze finds the Python files, calls _analyze_file on each in sorted order, then _compute_derived.
• File discovery skips build-output and cache directories listed in EXCLUDED_SOURCE_DIRS and any nested directory that carries a packaging marker from DISTRIBUTION_MARKERS, so vendored projects stay out of the snapshot.
• _analyze_file parses one file. A file with a syntax or decoding error is skipped with a printed note. It records classes, functions, module variables, __all__, imports, environment accesses, spawn-site program literals and a residue hash.
• Imports are collected by a scope-tracking walk, so lazy imports inside functions and imports under TYPE_CHECKING are labelled and not lost. The dependency graph counts only unguarded module-scope imports, so lazy imports that break cycles do not create false cycles.
• _compute_derived resolves base names into the inheritance graph, propagates the pydantic flag through ancestors, builds the dependency graph and sets totals.
• compute_file_meta records sha1, size and mtime per module file, skipping unreadable files.
• load_or_analyze is the warm start. It reuses cached modules whose file hash is unchanged, reparses new or changed files and drops removed ones. Any mismatch in name, root, captures or the file_meta map falls back to a full analysis. Its stats result lists what was reused, reparsed and removed.
• derive_subtree projects a single-tree snapshot from a wider one without reparsing, renaming dotted ids and rebuilding the derived graphs.
• module_info_from_dict rebuilds a ModuleInfo from cached JSON.

Relative paths are read under the anchor directory when one is given and never under the process working directory, while stored paths keep repository-relative form.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

CodebaseAnalyzer

Line 188

Full AST analysis of a Python codebase.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ast/analyzer.py

"""Source-tree -> CodebaseAST analyzer — stdlib-only.

Extracted verbatim from scripts/serialize_codebase_ast.py (2026-07-10);
see orchestrata_ast.models for the interchange contract.
"""

from __future__ import annotations

import ast
import dataclasses
import hashlib
import json
import os
import sys
from collections.abc import Iterator
from pathlib import Path
from typing import Any, cast

from orchestrata_ast.models import (
    ArgvLiteralInfo,
    ClassInfo,
    ClassVarInfo,
    CodebaseAST,
    EnvAccessInfo,
    FunctionInfo,
    ImportInfo,
    ModuleInfo,
    ParamInfo,
)
from orchestrata_ast.serialize import serialize

#: Directory names that hold BUILD OUTPUT rather than source, at any depth.
#: A copy of a module under ``build/`` parses fine and produces a second module
#: with the same dotted path, so every consumer keyed on that path silently
#: takes whichever won the walk order.
EXCLUDED_SOURCE_DIRS = frozenset(
    {
        ".git",
        ".mypy_cache",
        ".pytest_cache",
        ".ruff_cache",
        ".venv",
        "__pycache__",
        "build",
        "dist",
        "node_modules",
        "site-packages",
        "venv",
    }
)

#: Files whose presence declares a directory a SEPARATE distribution.
#: ``scripts/_cxpt``, ``scripts/_ctok``, ``scripts/_codemod_fast`` and
#: ``scripts/_janus_patched`` are upstream projects vendored here to be
#: compiled. Their contents are somebody else's source — and somebody else's
#: test suite — not this codebase's. Detecting them by the build metadata they
#: carry, rather than by name, keeps the next vendored tree from silently
#: rejoining the snapshot.
DISTRIBUTION_MARKERS = frozenset({"pyproject.toml", "setup.py", "setup.cfg"})


def _is_nested_distribution(directory: Path) -> bool:
    """Does *directory* declare itself a distribution independent of ours?"""
    return any((directory / marker).is_file() for marker in DISTRIBUTION_MARKERS)


# ---------------------------------------------------------------------------
# Byte-exact capture (2026-08-02 user mandate: "the AST must capture
# byte-identical code -- all of the code, leave nothing behind"). Two
# helpers: a source-segment hash for class/function bodies, and a residue
# hash for everything else in a module (imports, module vars, decorators of
# top-level defs, module-level statements). Together every byte of a file
# lands in at least one hash. See orchestrata_ast.models docstring.
# ---------------------------------------------------------------------------


def _source_segment_hash(text: str, node: ast.AST) -> str:
    """``sha256(ast.get_source_segment(text, node))[:16]`` — byte-exact hash
    of *node*'s exact source span within *text*.

    Empty string when *text* is empty (nothing to segment — e.g. a
    synthetic/detached node analyzed with no source available) or when
    ``ast.get_source_segment`` itself returns ``None`` (node missing
    lineno/col_offset attributes).
    """
    if not text:
        return ""
    try:
        segment = ast.get_source_segment(text, node)
    except IndexError, AttributeError, ValueError:
        return ""
    if segment is None:
        return ""
    return hashlib.sha256(segment.encode("utf-8", "surrogateescape")).hexdigest()[:16]


def _residue_hash(text: str, top_level_spans: list[tuple[int, int]]) -> str:
    """``sha256(text minus the given 1-based line spans)[:16]`` —
    ``ModuleInfo.residue_hash``: everything in a file OUTSIDE its top-level
    class/function defs (imports, module vars, decorators of top-level defs
    [decorator lines precede ``node.lineno`` so they stay in residue],
    module-level statements).

    *top_level_spans* is ``(node.lineno, node.end_lineno)`` per top-level
    class/function; for each, lines ``[lineno - 1, end_lineno)`` (0-based,
    exclusive end) are dropped from ``text.splitlines(keepends=True)``
    before hashing the remainder.
    """
    lines = text.splitlines(keepends=True)
    drop: set[int] = {
        i
        for lineno, end_lineno in top_level_spans
        for i in range(lineno - 1, end_lineno)
    }
    residue = "".join(line for i, line in enumerate(lines) if i not in drop)
    return hashlib.sha256(residue.encode("utf-8", "surrogateescape")).hexdigest()[:16]


# ---------------------------------------------------------------------------
# Callee names at process-spawn sites (2026-10-11, the bare-toolchain gate —
# Mohamed: "fix bare python make it prolog and atoms"). These name ONLY the
# stdlib spawn APIs whose argument POSITIONS are fixed by the OS contract —
# where the program token sits. Which PROGRAMS count as bare toolchain names
# is NOT decided here or anywhere in Python; that is the Prolog rule's job
# (Q2b), over the facts ArgvLiteralInfo records. See models.py.
# ---------------------------------------------------------------------------

#: Keywords a command list conventionally travels in when a call takes no
#: positional arguments: ``subprocess.Popen(args=("gcc", "-c", "a.c"))``.
_ARGV_KEYWORDS: tuple[str, ...] = ("args", "cmd", "command", "argv")

#: ``os.execv``/``os.execve``/``os.execvp``/``os.execvpe`` and the ``execl``
#: family take the program as the FIRST positional argument.
_EXEC_PROGRAM_FIRST: frozenset[str] = frozenset(
    {
        "os.execv",
        "os.execve",
        "os.execvp",
        "os.execvpe",
        "os.execl",
        "os.execle",
        "os.execlp",
        "os.execlpe",
    }
)

#: ``os.spawnv``/``os.spawnl`` families take the mode FIRST; the program is
#: the SECOND positional argument.
_EXEC_PROGRAM_SECOND: frozenset[str] = frozenset(
    {
        "os.spawnv",
        "os.spawnve",
        "os.spawnvp",
        "os.spawnvpe",
        "os.spawnl",
        "os.spawnle",
        "os.spawnlp",
        "os.spawnlpe",
    }
)

#: Last segments that mark the exec form on ANY base — the asyncio API and
#: any wrapper that kept its name.
_EXEC_LAST_SEGMENTS: frozenset[str] = frozenset({"create_subprocess_exec"})

#: Callees whose first positional argument is a SHELL STRING, by full dotted
#: text...
_SHELL_DOTTED: frozenset[str] = frozenset({"os.system", "os.popen"})
#: ...or by last segment on a known base.
_SHELL_LAST_SEGMENTS: frozenset[str] = frozenset(
    {
        "getoutput",
        "getstatusoutput",
        "create_subprocess_shell",
        "run",
        "Popen",
        "call",
        "check_call",
        "check_output",
    }
)
#: The base those last segments are shell-string callees on. ``asyncio``
#: qualifies ONLY ``create_subprocess_shell``.
_SHELL_BASES: frozenset[str] = frozenset({"subprocess"})
_SHELL_ASYNCIO_SEGMENTS: frozenset[str] = frozenset({"create_subprocess_shell"})


class CodebaseAnalyzer:
    """Full AST analysis of a Python codebase."""

    def __init__(
        self,
        name: str,
        root: Path,
        monorepo: bool = False,
        extra_roots: tuple[Path, ...] = (),
        anchor: Path | None = None,
    ):
        self.name = name
        self.root = root
        self.monorepo = monorepo
        #: The checkout relative roots are read UNDER (ADR-0097) — never
        #: ``os.getcwd()``. Every path the analyzer STORES keeps the form it
        #: was given (repository-root-relative for a daemon publish, exactly
        #: like a CLI run from the repository root); only the reads are
        #: anchored, via :meth:`_io`. ``None`` keeps the CLI's cwd behaviour.
        self.anchor = anchor
        #: Additional trees to analyze alongside *root*. Mohamed, 2026-08-08:
        #: "the ast must see tests and scripts too." A snapshot of ``src``
        #: alone cannot answer a question about a test file, and the Prolog
        #: gate proved it: with only ``src`` serialized, ``file_class`` was
        #: 850/850 ``src`` and every rule was structurally blind to the 1,688
        #: test modules and 191 scripts — a PASS that meant "not looked at".
        self.extra_roots = tuple(extra_roots)
        #: file_path -> parsed AST, populated by ``_analyze_file`` as it
        #: parses each module. Reused by ``_augment_end_line_and_complexity``
        #: (scripts/serialize_codebase_ast.py) so the end_line/complexity
        #: enrichment pass doesn't re-read+re-parse a file this analyzer
        #: already parsed once.
        self.trees: dict[str, ast.Module] = {}
        #: How many Python files ``analyze()`` DISCOVERED, as distinct from
        #: how many it successfully parsed (``len(self.trees)``). A file
        #: that fails to parse is skipped with a stderr note and simply
        #: contributes nothing, so without this the two counts are
        #: indistinguishable to a caller and "found nothing in this file"
        #: reads identically to "could not read this file".
        self.discovered_python_files = 0
        self.result = CodebaseAST(
            name=name,
            root_path=str(root),
            extra_roots=[str(extra) for extra in self.extra_roots],
            captures=list(SNAPSHOT_CAPTURES),
        )

    def analyze(self) -> CodebaseAST:
        """Analyze the full codebase."""
        py_files = self._find_python_files()
        self.discovered_python_files = len(py_files)
        print(f"Found {len(py_files)} Python files in {self.name}")

        for py_file in sorted(py_files):
            module = self._analyze_file(py_file)
            if module:
                self.result.modules.append(module)

        self._compute_derived()
        return self.result

    def _io(self, path: Path) -> Path:
        """The form of *path* to READ: anchored, never cwd-relative."""
        if self.anchor is not None and not path.is_absolute():
            return self.anchor / path
        return path

    def _display(self, path: Path) -> Path:
        """The form of *path* to STORE: relative to the anchor when under it."""
        if self.anchor is None:
            return path
        try:
            return path.relative_to(self.anchor)
        except ValueError:
            return path

    def _find_python_files(self) -> list[Path]:
        """Find all Python source files, across every declared root.

        Walks the anchored (readable) roots but returns the DISPLAY form of
        every file, so each stored ``file_path`` keeps the shape a CLI run
        from the repository root would produce.
        """
        if self.monorepo:
            # Scan packages/orchestrata-*/src/orchestrata/**/*.py
            files: list[Path] = []
            for pkg_dir in sorted(self._io(self.root).iterdir()):
                if pkg_dir.is_dir() and pkg_dir.name.startswith("orchestrata-"):
                    src = pkg_dir / "src"
                    if src.exists():
                        files.extend(self._display(p) for p in self._walk(src))
            return files
        files = [self._display(p) for p in self._walk(self._io(self.root))]
        for extra in self.extra_roots:
            files.extend(self._display(p) for p in self._walk(self._io(extra)))
        return files

    @staticmethod
    def _walk(root: Path) -> list[Path]:
        """Every ``.py`` under *root* that is OUR SOURCE rather than output.

        ``src/orchestrata_ultimate`` needed no exclusions — it holds nothing
        but source. ``scripts`` does, in two distinct ways.

        The vendored native extensions each carry a ``build/`` tree holding
        COPIES of their own modules, so a naive walk serializes the same module
        twice under two paths and hands every consumer keyed on a dotted path a
        silent collision.

        And the vendored trees are not this codebase at all: they are upstream
        distributions parked here to be compiled, carrying upstream's package
        AND upstream's test suite. Serializing them told the coverage report
        that ``janus.query.next`` was an untested method of ours and put
        ``scripts._janus_patched.tests.dog`` in our test inventory.
        """
        found: list[Path] = []
        for dirpath, dirnames, filenames in os.walk(root):
            here = Path(dirpath)
            # In-place, so os.walk never descends into what we drop.
            dirnames[:] = [
                name
                for name in dirnames
                if name not in EXCLUDED_SOURCE_DIRS
                and not _is_nested_distribution(here / name)
            ]
            found.extend(here / name for name in filenames if name.endswith(".py"))
        # os.walk order is filesystem order; the snapshot must not depend on it.
        return sorted(found)

    def _owning_extra_root(self, file_path: Path) -> Path | None:
        """Which extra root holds this file, if any."""
        for extra in self.extra_roots:
            try:
                file_path.relative_to(extra)
            except ValueError:
                continue
            return extra
        return None

    def _file_to_module_path(self, file_path: Path) -> str:
        """Convert file path to Python module path."""
        if self.monorepo:
            # packages/orchestrata-agents/src/orchestrata/agents/base.py
            # → orchestrata.agents.base
            parts = file_path.parts
            try:
                src_idx = list(parts).index("src")
                module_parts = parts[src_idx + 1 :]
            except ValueError:
                module_parts = file_path.relative_to(self.root).parts
        elif (extra := self._owning_extra_root(file_path)) is not None:
            # tests/gates/test_gate_rules.py (extra root=tests)
            # → tests.gates.test_gate_rules
            #
            # The tree's OWN directory name leads, deliberately. The primary
            # root's paths drop their package prefix (``agents.base_agent``),
            # and reusing that scheme for a second tree would collide the
            # moment two trees share a subdirectory name — ``core``, ``agents``
            # and ``gateway`` all exist under both ``src/orchestrata_ultimate``
            # and ``tests``. Prefixing keeps the primary tree's 850 ids
            # byte-identical while making the new ones unambiguous.
            module_parts = (extra.name, *file_path.relative_to(extra).parts)
        else:
            # src/orchestrata_ultimate/agents/base_agent.py (root=src/orchestrata_ultimate)
            # → agents.base_agent  (root-relative, no package prefix)
            try:
                rel = file_path.relative_to(self.root)
                module_parts = rel.parts
            except ValueError:
                parts = file_path.parts
                try:
                    src_idx = list(parts).index("src")
                    module_parts = parts[src_idx + 1 :]
                except ValueError:
                    module_parts = file_path.parts

        # Convert to dotted path, remove .py
        path_str = ".".join(module_parts)
        if path_str.endswith(".py"):
            path_str = path_str[:-3]
        if path_str.endswith(".__init__"):
            path_str = path_str[:-9]
        return path_str

    def _analyze_file(self, file_path: Path) -> ModuleInfo | None:
        """Analyze a single Python file."""
        try:
            content = self._io(file_path).read_text(encoding="utf-8")
            tree = ast.parse(content, filename=str(file_path))
        except (SyntaxError, UnicodeDecodeError) as e:
            print(f"  Skip {file_path}: {e}")
            return None
        self.trees[str(file_path)] = tree

        module_path = self._file_to_module_path(file_path)
        module = ModuleInfo(
            path=module_path,
            file_path=str(file_path),
            line_count=len(content.splitlines()),
            docstring=ast.get_docstring(tree),
        )

        top_level_spans: list[tuple[int, int]] = []
        for node in ast.iter_child_nodes(tree):
            if isinstance(node, ast.ClassDef):
                cls = self._analyze_class(node, module_path, str(file_path), content)
                module.classes.append(cls)
                top_level_spans.append((node.lineno, node.end_lineno or node.lineno))
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                func = self._analyze_function(
                    node, module_path, str(file_path), text=content
                )
                module.functions.append(func)
                top_level_spans.append((node.lineno, node.end_lineno or node.lineno))
            elif isinstance(node, ast.AnnAssign):
                var = self._analyze_annotation(node)
                if var:
                    module.module_vars.append(var)
            elif isinstance(node, ast.Assign):
                # Check for __all__
                for target in node.targets:
                    if isinstance(target, ast.Name) and target.id == "__all__":
                        if isinstance(node.value, (ast.List, ast.Tuple)):
                            for elt in node.value.elts:
                                if isinstance(elt, ast.Constant) and isinstance(
                                    elt.value, str
                                ):
                                    module.all_exports.append(elt.value)

        # Whole-tree, not the top-level loop above: an env read inside a
        # function body is exactly as much a RUL-003 subject as one at module
        # scope, and the loop above only walks direct children of the Module.
        module.env_accesses = self._analyze_env_accesses(tree, str(file_path))

        # Same reasoning, different subject: a spawn inside a function body is
        # exactly as much a bare-toolchain subject as one at module scope, and
        # `calls` carries the callee with NO arguments — the program literal
        # is only recoverable from the tree. See ArgvLiteralInfo.
        module.argv_literals = self._analyze_argv_literals(tree, str(file_path))

        # Same reasoning, and the same bug the line above already fixed once:
        # an import inside a function body is still an import. Collected by a
        # scope-tracking walk rather than the top-level loop, which sees only
        # direct children of the Module. See _collect_scoped_imports.
        module.imports = self._collect_scoped_imports(tree)

        module.residue_hash = _residue_hash(content, top_level_spans)
        return module

    def _analyze_class(
        self, node: ast.ClassDef, module_path: str, file_path: str, text: str = ""
    ) -> ClassInfo:
        """Analyze a class definition.

        *text* is the full source of the file *node* was parsed from, used
        to compute a byte-exact ``body_hash`` over the class's source
        segment (``ast.get_source_segment``) — covers class vars + methods,
        every byte of the class. Defaults to ``""`` (body_hash left empty)
        for callers analyzing a synthetic/detached node with no source text.
        """
        qualified = f"{module_path}.{node.name}"
        bases = []
        for base in node.bases:
            try:
                bases.append(ast.unparse(base))
            except Exception:
                if isinstance(base, ast.Name):
                    bases.append(base.id)

        decorators = self._get_decorators(node)

        cls = ClassInfo(
            name=node.name,
            qualified_name=qualified,
            module_path=module_path,
            file_path=file_path,
            line=node.lineno,
            bases=bases,
            decorators=decorators,
            docstring=ast.get_docstring(node),
            is_abstract=any("ABC" in b or "Abstract" in b for b in bases),
            is_dataclass=any("dataclass" in d for d in decorators),
            is_pydantic=any("BaseModel" in b or "BaseSettings" in b for b in bases),
            is_protocol=any("Protocol" in b for b in bases),
            is_enum=any("Enum" in b or "IntEnum" in b for b in bases),
            end_line=node.end_lineno if node.end_lineno is not None else 0,
            body_hash=_source_segment_hash(text, node),
        )

        for item in node.body:
            if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                method = self._analyze_function(
                    item, module_path, file_path, class_name=node.name, text=text
                )
                cls.methods.append(method)
            elif isinstance(item, ast.AnnAssign):
                var = self._analyze_annotation(item)
                if var:
                    cls.class_vars.append(var)
            elif isinstance(item, ast.ClassDef):
                cls.inner_classes.append(item.name)

        return cls

    def _analyze_function(
        self,
        node: ast.FunctionDef | ast.AsyncFunctionDef,
        module_path: str,
        file_path: str,
        class_name: str | None = None,
        text: str = "",
    ) -> FunctionInfo:
        """Analyze a function or method definition (top-level function OR
        method — methods are FunctionInfo too, distinguished by class_name).

        *text* is the full source of the file *node* was parsed from, used
        to compute a byte-exact ``body_hash`` over the def..end span
        (``ast.get_source_segment``). Defaults to ``""`` (body_hash left
        empty) for callers analyzing a synthetic/detached node.
        """
        if class_name:
            qualified = f"{module_path}.{class_name}.{node.name}"
        else:
            qualified = f"{module_path}.{node.name}"

        decorators = self._get_decorators(node)

        func = FunctionInfo(
            name=node.name,
            qualified_name=qualified,
            module_path=module_path,
            file_path=file_path,
            line=node.lineno,
            is_async=isinstance(node, ast.AsyncFunctionDef),
            is_method=class_name is not None,
            is_classmethod=any("classmethod" in d for d in decorators),
            is_staticmethod=any("staticmethod" in d for d in decorators),
            is_property=any("property" in d for d in decorators),
            is_abstract=any("abstractmethod" in d for d in decorators),
            decorators=decorators,
            parameters=self._extract_params(node.args),
            return_type=ast.unparse(node.returns) if node.returns else None,
            docstring=ast.get_docstring(node),
            class_name=class_name,
            calls=self._extract_calls(node),
            end_line=node.end_lineno if node.end_lineno is not None else 0,
            body_hash=_source_segment_hash(text, node),
        )

        return func

    def _extract_params(self, args: ast.arguments) -> list[ParamInfo]:
        """Extract parameter information."""
        params = []
        num_defaults = len(args.defaults)
        num_args = len(args.args)

        for i, arg in enumerate(args.args):
            if arg.arg in ("self", "cls"):
                continue
            default_idx = i - (num_args - num_defaults)
            default = None
            if default_idx >= 0:
                try:
                    default = ast.unparse(args.defaults[default_idx])
                except Exception:
                    default = "..."

            params.append(
                ParamInfo(
                    name=arg.arg,
                    type_annotation=(
                        ast.unparse(arg.annotation) if arg.annotation else None
                    ),
                    default=default,
                    kind="positional",
                )
            )

        if args.vararg:
            params.append(
                ParamInfo(
                    name=f"*{args.vararg.arg}",
                    type_annotation=(
                        ast.unparse(args.vararg.annotation)
                        if args.vararg.annotation
                        else None
                    ),
                    kind="varargs",
                )
            )

        for i, arg in enumerate(args.kwonlyargs):
            default = None
            kw_default = args.kw_defaults[i] if i < len(args.kw_defaults) else None
            if kw_default is not None:
                try:
                    default = ast.unparse(kw_default)
                except Exception:
                    default = "..."
            params.append(
                ParamInfo(
                    name=arg.arg,
                    type_annotation=(
                        ast.unparse(arg.annotation) if arg.annotation else None
                    ),
                    default=default,
                    kind="keyword_only",
                )
            )

        if args.kwarg:
            params.append(
                ParamInfo(
                    name=f"**{args.kwarg.arg}",
                    type_annotation=(
                        ast.unparse(args.kwarg.annotation)
                        if args.kwarg.annotation
                        else None
                    ),
                    kind="kwargs",
                )
            )

        return params

    def _get_decorators(
        self, node: ast.ClassDef | ast.FunctionDef | ast.AsyncFunctionDef
    ) -> list[str]:
        """Extract decorator names."""
        decorators = []
        for dec in node.decorator_list:
            try:
                decorators.append(ast.unparse(dec))
            except Exception:
                if isinstance(dec, ast.Name):
                    decorators.append(dec.id)
                elif isinstance(dec, ast.Attribute):
                    decorators.append(dec.attr)
        return decorators

    def _extract_calls(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> list[str]:
        """Extract function/method calls within a function body."""
        calls = []
        for child in ast.walk(node):
            if isinstance(child, ast.Call):
                try:
                    calls.append(ast.unparse(child.func))
                except Exception:
                    pass
        # Deduplicate, keep order
        seen = set()
        unique = []
        for c in calls:
            if c not in seen:
                seen.add(c)
                unique.append(c)
        return unique

    @staticmethod
    def _is_type_checking_test(test: ast.expr) -> bool:
        """``TYPE_CHECKING`` or ``typing.TYPE_CHECKING`` as an ``if`` test.

        Both spellings, unlike ``architecture_checker._is_inside_type_checking``
        which only matches the bare ``ast.Name`` form.
        """
        if isinstance(test, ast.Name):
            return test.id == "TYPE_CHECKING"
        if isinstance(test, ast.Attribute):
            return test.attr == "TYPE_CHECKING"
        return False

    def _collect_scoped_imports(self, tree: ast.Module) -> list[ImportInfo]:
        """Every import in the file, labelled with the scope it executes in.

        ``_analyze_file``'s top-level loop iterates ``ast.iter_child_nodes`` on
        the Module, so it only ever saw imports that were *direct* children of
        it. Everything nested was silently dropped: function-local (lazy)
        imports, class-body imports, ``if TYPE_CHECKING:`` imports, and the
        ``try: ... except ImportError:`` optional-dependency shape.

        That gap had a measured consequence. ``from_name/4`` — the Prolog fact
        that maps a local name back to the module it came from, and therefore
        the thing that resolves a call to its target — is built from this list.
        With lazy imports missing, ``code_intel.py callers
        core.redis_factory.create_sync_redis`` answered **0 callers** while real
        callers existed, because every one of them imports it inside a function.

        Scope is recorded, not flattened. ``_compute_derived`` still builds
        ``dependency_graph`` from module-scope imports alone: a lazy import is
        real runtime coupling but is NOT an import-time dependency, and feeding
        it to the dependency graph would resurrect precisely the cycles that
        lazy imports are written to break.
        """
        found: list[ImportInfo] = []

        def qualify(owner: str | None, name: str) -> str:
            return f"{owner}.{name}" if owner else name

        def emit(
            node: ast.Import | ast.ImportFrom,
            owner: str | None,
            scope: str,
            conditional: bool,
        ) -> None:
            imp = self._analyze_import(node)
            imp.scope = scope
            imp.owner = owner
            imp.conditional = conditional
            found.append(imp)

        def visit_body(
            stmts: list[ast.stmt], owner: str | None, scope: str, conditional: bool
        ) -> None:
            for stmt in stmts:
                visit(stmt, owner, scope, conditional)

        def visit(
            stmt: ast.stmt, owner: str | None, scope: str, conditional: bool
        ) -> None:
            if isinstance(stmt, (ast.Import, ast.ImportFrom)):
                emit(stmt, owner, scope, conditional)
            elif isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
                # Anything under a def runs only when called, however the def
                # itself was reached.
                visit_body(
                    stmt.body, qualify(owner, stmt.name), "function", conditional
                )
            elif isinstance(stmt, ast.ClassDef):
                # A class body runs at definition time — which is import time
                # at module scope, but stays lazy inside a function.
                inner = "function" if scope == "function" else "class"
                visit_body(stmt.body, qualify(owner, stmt.name), inner, conditional)
            elif isinstance(stmt, ast.If):
                # `if TYPE_CHECKING:` never executes; its `else:` branch does.
                # A def-scoped TYPE_CHECKING block is still never executed.
                taken = (
                    "type_checking" if self._is_type_checking_test(stmt.test) else scope
                )
                visit_body(stmt.body, owner, taken, True)
                visit_body(stmt.orelse, owner, scope, True)
            elif isinstance(stmt, (ast.Try, ast.TryStar)):
                visit_body(stmt.body, owner, scope, True)
                for handler in stmt.handlers:
                    visit_body(handler.body, owner, scope, True)
                visit_body(stmt.orelse, owner, scope, True)
                visit_body(stmt.finalbody, owner, scope, True)
            elif isinstance(stmt, (ast.For, ast.AsyncFor, ast.While)):
                visit_body(stmt.body, owner, scope, True)
                visit_body(stmt.orelse, owner, scope, True)
            elif isinstance(stmt, (ast.With, ast.AsyncWith)):
                visit_body(stmt.body, owner, scope, True)
            elif isinstance(stmt, ast.Match):
                for case in stmt.cases:
                    visit_body(case.body, owner, scope, True)

        visit_body(tree.body, None, "module", False)
        return found

    def _analyze_import(self, node: ast.Import | ast.ImportFrom) -> ImportInfo:
        """Analyze an import statement."""
        if isinstance(node, ast.Import):
            names = [alias.name for alias in node.names]
            # ImportInfo has a single alias slot, so asname is only representable
            # for single-name imports (`import a as b, c as d` stays None).
            asname = node.names[0].asname if len(node.names) == 1 else None
            return ImportInfo(
                module=names[0] if names else "",
                names=names,
                is_from=False,
                alias=asname,
                line=node.lineno,
            )
        else:
            names = [alias.name for alias in node.names]
            return ImportInfo(
                module=node.module or "",
                names=names,
                is_from=True,
                is_relative=node.level > 0,
                level=node.level,
                line=node.lineno,
            )

    @staticmethod
    def _analyze_env_accesses(tree: ast.AST, file_path: str) -> list[EnvAccessInfo]:
        """Every access to the process environment in one parsed module.

        Walks the WHOLE tree, not just module level: RUL-003 applies to a
        function-local `os.environ["X"]` exactly as much as a top-level one.

        Only the literal key is recorded. `os.environ.get(name)` where `name`
        is a variable yields key="" — reported as unknown rather than guessed,
        because a rule that invents a key name is worse than one that abstains.
        """
        accesses: list[EnvAccessInfo] = []

        # Which BARE names in this module are bound to os.environ. Only
        # `from os import environ` (or an `as` alias of it) creates one.
        #
        # Matching the bare name `environ` unconditionally was a false-positive
        # generator, and it fired on the exact pattern RUL-003 mandates: a test
        # that builds `environ: dict[str, str] = {}` and passes it through the
        # code's `env=` injection seam got eight findings telling it to add
        # defaults to a local dict. Measured 2026-08-08 across tests/, the hour
        # the snapshot first covered them. A binding is resolvable from this
        # module's own imports, so there is no reason to guess from the name.
        environ_names: set[str] = set()
        for imported in ast.walk(tree):
            if isinstance(imported, ast.ImportFrom) and imported.module == "os":
                environ_names.update(
                    alias.asname or alias.name
                    for alias in imported.names
                    if alias.name == "environ"
                )

        def _literal(node: ast.AST | None) -> str:
            return (
                node.value
                if isinstance(node, ast.Constant) and isinstance(node.value, str)
                else ""
            )

        def _is_environ(node: ast.AST) -> bool:
            # `os.environ` by attribute chain, or a bare name this module
            # actually imported FROM os — never a bare name that merely reads
            # like one.
            if isinstance(node, ast.Attribute):
                return node.attr == "environ"
            return isinstance(node, ast.Name) and node.id in environ_names

        for node in ast.walk(tree):
            # os.environ["X"] — a Subscript, which never appears in `calls`.
            if isinstance(node, ast.Subscript) and _is_environ(node.value):
                # Load vs Store is recorded because they are DIFFERENT policies
                # with different remedies, and one Form value conflated them.
                # `x = os.environ["K"]` risks KeyError -> use .get(); but
                # `os.environ["K"] = v` MUTATES the process environment, and
                # the gate told the reader to ".get()" an assignment — advice
                # that, if followed, silently drops the write. Measured
                # 2026-08-08 on a two-line corpus.
                accesses.append(
                    EnvAccessInfo(
                        file_path=file_path,
                        line=node.lineno,
                        key=_literal(node.slice),
                        form=(
                            "subscript"
                            if isinstance(node.ctx, ast.Load)
                            else "subscript_store"
                        ),
                    )
                )
                continue
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if isinstance(func, ast.Attribute) and _is_environ(func.value):
                form = "get" if func.attr == "get" else "other"
                accesses.append(
                    EnvAccessInfo(
                        file_path=file_path,
                        line=node.lineno,
                        key=_literal(node.args[0]) if node.args else "",
                        form=form,
                        has_default=len(node.args) > 1 or bool(node.keywords),
                    )
                )
            elif isinstance(func, ast.Attribute) and func.attr == "getenv":
                accesses.append(
                    EnvAccessInfo(
                        file_path=file_path,
                        line=node.lineno,
                        key=_literal(node.args[0]) if node.args else "",
                        form="getenv",
                        has_default=len(node.args) > 1 or bool(node.keywords),
                    )
                )
        return accesses

    @staticmethod
    def _analyze_argv_literals(tree: ast.AST, file_path: str) -> list[ArgvLiteralInfo]:
        """Every literal program name at a spawn-shaped site in one module.

        Walks the WHOLE tree, like ``_analyze_env_accesses``: a spawn inside a
        function body is exactly as much a bare-toolchain subject as one at
        module scope, and ``calls`` carries the callee with NO arguments, so
        the program token is recoverable only from the tree.

        LITERALS ONLY. A head that is not a ``str`` ``Constant`` is never
        recorded: ``sys.executable``, a variable, ``[*PY, ...]``, an ``argv``
        parameter are skipped, not guessed — a fact that invents a program
        name is worse than one that abstains.

        THIS FUNCTION CLASSIFIES NOTHING. Every literal program token is
        recorded verbatim with its callee — absolute paths, non-toolchain
        heads, and non-spawning callees included (``sorted(["python", "b"])``
        IS a record): a project wrapper such as ``_run([...])`` is a spawn
        that no fixed callee list can name, so which callees spawn processes
        and which programs are bare toolchain names is decided by the Prolog
        rule (Q2b) over these facts. Over-capture is the design; there is
        deliberately no callee allowlist or denylist for the ``argv`` form.

        The forms (see ``ArgvLiteralInfo`` for the full table):

        ``argv``   a literal-headed List/Tuple (element 0 a str Constant) or a
                   BinOp Add whose LEFT is one, among a call's positional
                   arguments — ANY position, not only the first:
                   ``os.execvp("gcc", ["gcc", "a.c"])`` carries its argv list
                   at position 1 — or, when the call takes no positional
                   arguments, its ``args``/``cmd``/``command``/``argv``
                   keyword; or a same-scope assignment of such a list to a
                   Name that a call in the same scope passes as its command
                   argument.
        ``exec``   last segment ``create_subprocess_exec``, or a dotted
                   ``os.execv*``/``os.execl*`` (program = first positional
                   str Constant), or ``os.spawnv*``/``os.spawnl*`` (program =
                   SECOND positional — the first is the mode).
        ``shell``  dotted ``os.system``/``os.popen``, or a
                   ``_SHELL_LAST_SEGMENTS`` callee on a ``subprocess`` base
                   (``asyncio`` qualifies only ``create_subprocess_shell``),
                   whose first positional is a str Constant or a JoinedStr
                   whose leading literal chunk contains whitespace; the
                   program is the string's first whitespace-separated token.

        KNOWN LIMITATION. A shell string prefixed by ``env`` (``"env -u X
        python y"``) records ``env``: the first token IS the program the
        shell execs. Looking past ``env`` is a later refinement (Q2b can do
        it in Prolog), not this layer's job.
        """

        def _callee_text(node: ast.expr) -> str:
            """Dotted callee text: ``Name`` gives its id, ``Attribute`` gives
            ``<render(value)>.<attr>``, any other node in the chain renders
            as ``?`` — ``factory().run`` gives ``?.run``."""
            if isinstance(node, ast.Name):
                return node.id
            if isinstance(node, ast.Attribute):
                return f"{_callee_text(node.value)}.{node.attr}"
            return "?"

        def _literal_head(node: ast.expr) -> tuple[ast.expr, str] | None:
            """``(seq, program)`` when *node* is a literal-headed command
            list: a List/Tuple whose element 0 is a str Constant, or a BinOp
            Add whose LEFT is such a List/Tuple (``["python"] + extra`` —
            the right operand is runtime data, the head is a fact). A Starred
            or computed element 0 yields None."""
            seq = (
                node.left
                if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add)
                else node
            )
            if isinstance(seq, (ast.List, ast.Tuple)) and seq.elts:
                first = seq.elts[0]
                if isinstance(first, ast.Constant) and isinstance(first.value, str):
                    return seq, first.value
            return None

        def _command_arg(call: ast.Call) -> ast.expr | None:
            """The slot a command travels in: the first positional argument,
            or — when the call has no positional arguments — the first of the
            conventional command keywords."""
            if call.args:
                return call.args[0]
            for keyword in call.keywords:
                if keyword.arg in _ARGV_KEYWORDS:
                    return keyword.value
            return None

        records: list[ArgvLiteralInfo] = []

        def _add(line: int, site: str, form: str, program: str) -> None:
            records.append(
                ArgvLiteralInfo(
                    file_path=file_path,
                    line=line,
                    site=site,
                    form=form,
                    program=program,
                )
            )

        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            site = _callee_text(node.func)
            base, _, last = site.rpartition(".")

            # argv (direct): every literal-headed command list among the
            # positional arguments — see the docstring for why not only the
            # first. With no positional arguments, the conventional command
            # keywords are the slot.
            slots: list[ast.expr] = list(node.args) or [
                keyword.value
                for keyword in node.keywords
                if keyword.arg in _ARGV_KEYWORDS
            ]
            for slot in slots:
                head = _literal_head(slot)
                if head is not None:
                    seq, program = head
                    _add(seq.lineno, site, "argv", program)

            # exec: the OS API fixes the position of the program token.
            if last in _EXEC_LAST_SEGMENTS or site in _EXEC_PROGRAM_FIRST:
                if node.args:
                    first = node.args[0]
                    if isinstance(first, ast.Constant) and isinstance(first.value, str):
                        _add(first.lineno, site, "exec", first.value)
            elif site in _EXEC_PROGRAM_SECOND and len(node.args) >= 2:
                second = node.args[1]
                if isinstance(second, ast.Constant) and isinstance(second.value, str):
                    _add(second.lineno, site, "exec", second.value)

            # shell: the program is the first whitespace-separated token of
            # the command string.
            if node.args and (
                site in _SHELL_DOTTED
                or (
                    last in _SHELL_LAST_SEGMENTS
                    and (
                        base in _SHELL_BASES
                        or (base == "asyncio" and last in _SHELL_ASYNCIO_SEGMENTS)
                    )
                )
            ):
                arg = node.args[0]
                text = ""
                if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                    text = arg.value
                elif isinstance(arg, ast.JoinedStr) and arg.values:
                    chunk = arg.values[0]
                    if (
                        isinstance(chunk, ast.Constant)
                        and isinstance(chunk.value, str)
                        and any(character.isspace() for character in chunk.value)
                    ):
                        # Only the LEADING literal chunk of an f-string is
                        # knowable from the snapshot; an f-string that starts
                        # with an interpolation has no literal token and is
                        # skipped, not guessed.
                        text = chunk.value
                tokens = text.strip().split()
                if tokens:
                    _add(arg.lineno, site, "shell", tokens[0])

        # argv (named): a literal command list assigned to a Name and passed
        # to a call IN THE SAME SCOPE — the module body or one function body.
        # A nested function or class body is a scope of its own and is NOT
        # searched from the outer one: an assignment in `f` consumed by a call
        # in `f`'s nested `g` is not a same-scope fact, and resolving names
        # across the boundary is exactly the guessing this capture abstains
        # from.
        def _walk_scope(scope: ast.AST) -> Iterator[ast.AST]:
            """Every node in *scope*, stopping at nested scope boundaries.

            ``Lambda`` prunes like the statement scopes: its body is an
            expression evaluated in its own context, visited as part of its
            own walk, never as part of the enclosing scope's."""
            stack = list(ast.iter_child_nodes(scope))
            while stack:
                current = stack.pop()
                if isinstance(
                    current,
                    (
                        ast.FunctionDef,
                        ast.AsyncFunctionDef,
                        ast.ClassDef,
                        ast.Lambda,
                    ),
                ):
                    continue
                yield current
                stack.extend(ast.iter_child_nodes(current))

        scopes: list[ast.AST] = [tree]
        scopes.extend(
            found
            for found in ast.walk(tree)
            if isinstance(found, (ast.FunctionDef, ast.AsyncFunctionDef))
        )
        for scope in scopes:
            assigned: dict[str, list[tuple[ast.expr, str]]] = {}
            consumers: dict[str, list[ast.Call]] = {}
            for node in _walk_scope(scope):
                if (
                    isinstance(node, ast.Assign)
                    and len(node.targets) == 1
                    and isinstance(node.targets[0], ast.Name)
                ):
                    head = _literal_head(node.value)
                    if head is not None:
                        assigned.setdefault(node.targets[0].id, []).append(head)
                elif (
                    isinstance(node, ast.AnnAssign)
                    and isinstance(node.target, ast.Name)
                    and node.value is not None
                ):
                    head = _literal_head(node.value)
                    if head is not None:
                        assigned.setdefault(node.target.id, []).append(head)
                elif isinstance(node, ast.Call):
                    command = _command_arg(node)
                    if isinstance(command, ast.Name):
                        consumers.setdefault(command.id, []).append(node)
            for assigned_name, heads in assigned.items():
                calls = consumers.get(assigned_name)
                if not calls:
                    # A literal list no call consumes is not a spawn: a REPL
                    # language table (LANGS = ("python", "bash")) is data.
                    continue
                first_call = min(calls, key=lambda call: (call.lineno, call.col_offset))
                consumer_site = _callee_text(first_call.func)
                for seq, program in heads:
                    _add(seq.lineno, consumer_site, "argv", program)

        # Determinism: the snapshot is committed; unstable ordering creates
        # churn. Sorted and de-duplicated on the 4-tuple that distinguishes
        # two records (file_path is constant within one module).
        seen: set[tuple[int, str, str, str]] = set()
        ordered: list[ArgvLiteralInfo] = []
        for record in sorted(
            records,
            key=lambda record: (record.line, record.site, record.form, record.program),
        ):
            key = (record.line, record.site, record.form, record.program)
            if key in seen:
                continue
            seen.add(key)
            ordered.append(record)
        return ordered

    def _analyze_annotation(self, node: ast.AnnAssign) -> ClassVarInfo | None:
        """Analyze an annotated assignment."""
        if not isinstance(node.target, ast.Name):
            return None
        name = node.target.id
        type_ann = ast.unparse(node.annotation) if node.annotation else None
        default = None
        if node.value:
            try:
                default = ast.unparse(node.value)
            except Exception:
                default = "..."

        is_cv = False
        if type_ann and "ClassVar" in type_ann:
            is_cv = True

        return ClassVarInfo(
            name=name,
            type_annotation=type_ann,
            default=default,
            is_class_var=is_cv,
        )

    def _compute_derived(self) -> None:
        """Compute inheritance graph, dependency graph, and totals."""
        # Inheritance graph: class_qualified_name → [base_qualified_names]
        # We resolve base names to qualified names where possible
        class_by_name: dict[str, str] = {}  # short name → qualified name
        for mod in self.result.modules:
            for cls in mod.classes:
                class_by_name[cls.name] = cls.qualified_name

        for mod in self.result.modules:
            for cls in mod.classes:
                resolved_bases = []
                for base in cls.bases:
                    # Try to resolve short name to qualified
                    short = base.split(".")[-1]
                    if short in class_by_name:
                        resolved_bases.append(class_by_name[short])
                    else:
                        resolved_bases.append(base)
                self.result.inheritance_graph[cls.qualified_name] = resolved_bases

        # Propagate is_pydantic through inheritance chains
        # First, build a map of qualified_name -> ClassInfo
        class_by_qualified: dict[str, ClassInfo] = {}
        for mod in self.result.modules:
            for cls in mod.classes:
                class_by_qualified[cls.qualified_name] = cls

        def _has_pydantic_ancestor(cls_info: ClassInfo, visited: set[str]) -> bool:
            if cls_info.is_pydantic:
                return True
            for base in cls_info.bases:
                short = base.split(".")[-1]
                # Try to find the base class by short name in the same module
                base_qualified = None
                for qn, c in class_by_qualified.items():
                    if c.name == short:
                        base_qualified = qn
                        break
                if base_qualified and base_qualified not in visited:
                    visited.add(base_qualified)
                    if _has_pydantic_ancestor(
                        class_by_qualified[base_qualified], visited
                    ):
                        return True
            return False

        for mod in self.result.modules:
            for cls in mod.classes:
                if not cls.is_pydantic:
                    if _has_pydantic_ancestor(cls, set()):
                        cls.is_pydantic = True

        # Dependency graph: module_path → [imported_module_paths]
        # Import targets are absolute import strings (e.g.
        # "orchestrata_ultimate.agents.constants") while module paths are
        # root-relative (e.g. "agents.constants"). Strip the root-package
        # prefix from project-internal targets so both ends of an edge live
        # in the same namespace; external imports pass through unchanged.
        #
        # SCOPE FILTER (see _collect_scoped_imports): this graph means "what
        # does importing this module drag in", and `find-cycles` reads it. A
        # lazy function-local import is the standard way to BREAK an import
        # cycle, so counting one as a dependency would report cycles that do
        # not exist at import time. Unguarded module-scope imports only — the
        # exact set this loop saw before nested imports were captured.
        module_paths = {m.path for m in self.result.modules}
        root_pkg_prefix = self.root.name + "."

        def _normalize_dep(target: str, importer: str = "") -> str:
            if target not in module_paths and target.startswith(root_pkg_prefix):
                stripped = target[len(root_pkg_prefix) :]
                if stripped in module_paths:
                    return stripped
            # The mirror case: a module run as a SCRIPT (claude/hooks/*.py,
            # scripts/*.py) puts its own directory on sys.path and imports its
            # siblings BARE, so the recorded target is SHORT by the importer's
            # package prefix -- "error_gate.checks_lint" where the module path
            # is "claude.hooks.error_gate.checks_lint". Walk the importer's
            # package ancestors most-specific first and take the first that
            # names a real module. Without this, every intra-subtree edge in
            # such a tree is silently dropped and the graph reports the whole
            # subtree as orphaned (measured 2026-08-23: 240 claude.hooks
            # targets, 0 resolving to a known module, 0 IMPORTS edges).
            #
            # The walk must NOT fire for a name that already resolves as an
            # absolute import, or a project module SHADOWING a stdlib name
            # captures it: `import logging` inside core.base.service became an
            # edge to core.logging, and `import logging` inside core.logging
            # became a self-edge. Measured 2026-08-23 before this guard: 32
            # fabricated edges in plans/ast-ultimate.json, every one of them
            # stdlib `logging` or `threading`.
            if (
                target not in module_paths
                and importer
                and target.split(".")[0] not in sys.stdlib_module_names
            ):
                parts = importer.split(".")
                for i in range(len(parts) - 1, 0, -1):
                    candidate = f"{'.'.join(parts[:i])}.{target}"
                    if candidate in module_paths and candidate != importer:
                        return candidate
            return target

        for mod in self.result.modules:
            deps = set()
            for imp in mod.imports:
                if imp.scope != "module" or imp.conditional:
                    continue
                if imp.module:
                    deps.add(_normalize_dep(imp.module, mod.path))
            self.result.dependency_graph[mod.path] = sorted(deps)

        # Totals
        for mod in self.result.modules:
            self.result.total_classes += len(mod.classes)
            self.result.total_functions += len(mod.functions)
            for cls in mod.classes:
                self.result.total_methods += len(cls.methods)
            self.result.total_loc += mod.line_count


# ---------------------------------------------------------------------------
# Warm-cache read path (kills the cold-reparse design flaw — see
# docs/ast-toolkit.md "Warm-cache read path"). Two pieces:
#
#   compute_file_meta() — the additive freshness map embedded into a
#   serialized CodebaseAST at `serialize` time (top-level `file_meta` key:
#   module_path -> {file_path, sha1, size, mtime_ns}). Content-hash (sha1)
#   is the authority, never mtime — the JSON is a committed artifact and
#   mtimes don't survive a fresh clone; mtime_ns is carried only as a local
#   fast-path hint for future tooling, never trusted on its own here.
#
#   load_or_analyze() — used by `serialize --incremental` to warm-start a
#   CodebaseAST from a previous run's cache: unchanged files reuse their
#   cached ModuleInfo verbatim (zero re-parse), changed/new files are
#   re-parsed via the analyzer's own per-file methods, deleted files drop
#   out, and derived data (inheritance/dependency graphs, totals) is
#   recomputed over the patched module list. Falls back to a full analyze
#   on any mismatch (different name/root, corrupt/old-format cache).
# ---------------------------------------------------------------------------


def _sha1_of_source(path: Path) -> str:
    """Content hash of a source file — the freshness authority (never mtime).

    RAW BYTES, deliberately. Hashing ``read_text()`` instead hashes the
    UNIVERSAL-NEWLINE translation of the file: a CRLF file decodes to LF and
    produces a hash no byte-level reader can reproduce. The gate's freshness
    layer hashes bytes, so the two disagreed on every CRLF file in the tree —
    invisible while the snapshot covered only ``src`` (all LF), and immediately
    visible when it widened to ``scripts``: the five vendored
    ``scripts/_janus_patched/xsb_tests/*.py`` files reported ``content_changed``
    against a snapshot built seconds earlier, and the gate went INVALID for a
    tree that had not moved.
    """
    return hashlib.sha1(path.read_bytes()).hexdigest()


def compute_file_meta(
    modules: list[ModuleInfo], root: Path | None = None
) -> dict[str, dict[str, Any]]:
    """``module_path -> {file_path, sha1, size, mtime_ns}`` for *modules*.

    Additive companion to a serialized :class:`CodebaseAST`, embedded as the
    top-level ``file_meta`` key at serialize time. Files that can no longer
    be read (moved/deleted between analysis and this call) are silently
    skipped — best-effort, matching ``_augment_end_line_and_complexity``'s
    own skip-on-unreadable behaviour in serialize_codebase_ast.py.

    *root* anchors the stat/hash reads the same way :func:`load_or_analyze`'s
    does (ADR-0097): a relative ``file_path`` is read under *root* — never
    ``os.getcwd()`` — while the string STORED in the map stays the module's
    own ``file_path`` verbatim, so the artifact keeps its repository-root
    conventions no matter where the publishing process runs.
    """
    meta: dict[str, dict[str, Any]] = {}
    for mod in modules:
        path = Path(mod.file_path)
        if root is not None and not path.is_absolute():
            path = root / path
        try:
            sha1 = _sha1_of_source(path)
            stat = path.stat()
        except OSError:
            continue
        meta[mod.path] = {
            "file_path": mod.file_path,
            "sha1": sha1,
            "size": stat.st_size,
            "mtime_ns": stat.st_mtime_ns,
        }
    return meta


# field_name -> nested dataclass type, for the handful of dataclass fields
# that hold ``list[<dataclass>]`` rather than primitives — reconstructing a
# dict back into one of these needs each item rebuilt too.
_DATACLASS_LIST_FIELDS: dict[type, dict[str, type]] = {
    FunctionInfo: {"parameters": ParamInfo},
    ClassInfo: {"methods": FunctionInfo, "class_vars": ClassVarInfo},
    ModuleInfo: {
        "classes": ClassInfo,
        "functions": FunctionInfo,
        "imports": ImportInfo,
        "module_vars": ClassVarInfo,
        "env_accesses": EnvAccessInfo,
        "argv_literals": ArgvLiteralInfo,
    },
}

#: The per-module CAPTURE fields this analyzer emits — the fields whose facts
#: a whole-tree walk recovers because ``calls``/``imports`` cannot express
#: them (see ``EnvAccessInfo``, ``ArgvLiteralInfo``). Adding a capture field
#: means adding its name HERE: the list is part of the warm-start cache
#: identity (``CodebaseAST.captures``, checked by ``load_or_analyze``), so a
#: snapshot written before the field existed forces ONE full reparse instead
#: of reusing module dicts that lack the key — a reused dict deserializes the
#: missing field as ``[]``, indistinguishable from "captured, found none",
#: and the snapshot reads half-captured while looking complete (2026-10-11).
SNAPSHOT_CAPTURES: tuple[str, ...] = ("env_accesses", "argv_literals")


def _dataclass_from_dict(cls: type, d: dict[str, Any]) -> Any:
    """Reconstruct a dataclass instance from its ``dataclasses.asdict()``
    dict (the inverse of ``orchestrata_ast.serialize.serialize``),
    recursively rebuilding nested dataclass ``list[...]`` fields.

    Unknown keys are dropped rather than raising ``TypeError``: a cached
    module dict may carry additive JSON-only enrichment (``end_line`` /
    ``complexity``, injected post-serialize by
    ``_augment_end_line_and_complexity`` in serialize_codebase_ast.py) that
    is not part of the dataclass model itself.
    """
    known = {f.name for f in dataclasses.fields(cls)}
    kwargs: dict[str, Any] = {k: v for k, v in d.items() if k in known}
    for field_name, item_cls in _DATACLASS_LIST_FIELDS.get(cls, {}).items():
        items = kwargs.get(field_name)
        if items is not None:
            kwargs[field_name] = [
                _dataclass_from_dict(item_cls, item) for item in items
            ]
    return cls(**kwargs)


def module_info_from_dict(d: dict[str, Any]) -> ModuleInfo:
    """Reconstruct a :class:`ModuleInfo` (+ nested classes/functions/imports)
    from its serialized dict — the inverse of ``serialize(module)``."""
    return cast(ModuleInfo, _dataclass_from_dict(ModuleInfo, d))


def derive_subtree(
    snapshot: dict[str, Any], *, tree: str, name: str, repo_root: Path
) -> dict[str, Any]:
    """A single-tree snapshot DERIVED from a wider one — never re-parsed.

    ``plans/ast-tests.json`` used to be a second ``serialize --source tests``
    run. Two snapshots of the same files, produced by different invocations at
    different times, is one truth too many: they drift, and the drift is
    invisible because each looks complete on its own. Measured 2026-08-09, the
    checked-in tests snapshot predated four serializer fields the wide one
    already carried — ``body_hash``, ``residue_hash``, ``env_accesses`` and
    per-import line numbers — so the RTM's evidence tooling and the
    affected-test selector were reading a format the rest of the repo had
    moved past.

    Everything here is a RENAME plus a re-derivation of graphs the analyzer
    owns. Nothing is re-parsed, so the two artifacts cannot disagree about a
    file: they are the same parse, projected twice.

    The dotted ids drop the tree's own prefix (``tests.gates.x`` ->
    ``gates.x``), which is what a ``--source tests`` run produced and what this
    artifact's consumers are keyed on. The derived graphs are then rebuilt by
    :meth:`CodebaseAnalyzer._compute_derived` over the NARROWED module set,
    rather than by filtering the wide graphs — that is the difference between
    reproducing the semantics and re-implementing them. It matters concretely:
    a dependency on ``orchestrata_ultimate.agents.swarm.coordinator`` stays
    fully qualified here (that module is not in this snapshot) while the wide
    snapshot normalises it to ``agents.swarm.coordinator`` (where it is).
    """
    prefix = f"{tree}."
    root = Path(repo_root).resolve()

    def _rename(dotted: str) -> str:
        return dotted[len(prefix) :] if dotted.startswith(prefix) else dotted

    def _relative(file_path: str) -> str:
        path = Path(file_path)
        if not path.is_absolute():
            return path.as_posix()
        try:
            return path.relative_to(root).as_posix()
        except ValueError:
            return path.as_posix()

    def _retarget(node: Any) -> None:
        """Rewrite every dotted/absolute field this projection owns, in place."""
        if isinstance(node, dict):
            for key, value in node.items():
                if key in ("module_path", "qualified_name") and isinstance(value, str):
                    node[key] = _rename(value)
                elif key == "file_path" and isinstance(value, str):
                    node[key] = _relative(value)
                else:
                    _retarget(value)
        elif isinstance(node, list):
            for item in node:
                _retarget(item)

    modules: list[dict[str, Any]] = []
    for module in snapshot.get("modules", []):
        path = str(module.get("path", ""))
        if path != tree and not path.startswith(prefix):
            continue
        projected = json.loads(json.dumps(module))
        _retarget(projected)
        # `tests/__init__.py` is `tests` in the wide snapshot (the trailing
        # `.__init__` is stripped there) and `__init__` in a --source tests run.
        projected["path"] = _rename(path) if path != tree else "__init__"
        modules.append(projected)

    analyzer = CodebaseAnalyzer(name, root / tree)
    analyzer.result = CodebaseAST(
        name=name,
        root_path=tree,
        modules=[module_info_from_dict(module) for module in modules],
    )
    analyzer._compute_derived()

    derived = cast("dict[str, Any]", serialize(analyzer.result))
    # The projection reports what the SOURCE snapshot captured, never more:
    # this CodebaseAST was built by hand (its captures default to []), while
    # the module dicts below are the source's own. Claiming a capture the
    # source lacks would make its absent per-module keys read as "captured,
    # found none" — the exact lie load_or_analyze's identity check exists to
    # prevent.
    derived["captures"] = list(snapshot.get("captures", []))
    # _compute_derived works on the MODEL, which drops fields the model does
    # not declare; the projected dicts are the authority on module content.
    derived["modules"] = modules
    # compute_file_meta OPENS each file_path, and the projected paths are
    # repo-relative — so hand it absolute ones and put the relative form back.
    # Resolving them against the process CWD instead is the same class of bug
    # as the original artifact's: correct only when the producer happens to run
    # from the repo root. Caught by the unit test, whose repo_root is a
    # tmp_path and whose file_meta came back empty.
    absolute = [
        dataclasses.replace(module, file_path=str(root / module.file_path))
        for module in analyzer.result.modules
    ]
    meta = compute_file_meta(absolute)
    for entry in meta.values():
        entry["file_path"] = Path(entry["file_path"]).relative_to(root).as_posix()
    derived["file_meta"] = meta
    return derived


def load_or_analyze(
    name: str,
    source: Path,
    cache_path: Path,
    monorepo: bool = False,
    extra_roots: tuple[Path, ...] = (),
    root: Path | None = None,
) -> tuple[CodebaseAST, dict[str, Any]]:
    """Warm-start a :class:`CodebaseAST` from a previous ``serialize`` run.

    Reads *cache_path* — a serialized :class:`CodebaseAST` JSON carrying the
    additive ``file_meta`` freshness map (see :func:`compute_file_meta`). If
    its recorded ``name``/``root_path`` match this call AND it has a
    ``file_meta`` map, every live file under *source* is stat+hashed (content
    sha1 — cheap, tens of ms for hundreds of files) against that map:
    unchanged files reuse their cached :class:`ModuleInfo` verbatim (zero
    re-parse), new/changed files are re-parsed via the analyzer's own
    per-file methods, and modules for files that no longer exist are
    dropped. Derived data (``inheritance_graph``/``dependency_graph``/
    totals) is recomputed over the patched module list — cheap dict work
    over the in-memory model, no reparsing involved.

    On ANY mismatch — different name/root, missing/corrupt JSON, no
    ``file_meta`` key (an old-format cache) — this silently falls back to a
    full :meth:`CodebaseAnalyzer.analyze`: correctness over speed, never a
    stale or wrong answer from an inapplicable cache.

    Returns ``(result, stats)``. ``stats`` carries ``fallback`` (bool),
    ``reused``/``reparsed`` (module-path lists), ``removed`` (module paths
    dropped because their file is gone), and ``cached_modules``
    (``module_path -> raw cached dict`` for the reused modules — callers
    like ``serialize --incremental`` that want to reuse a module's on-disk
    enrichment (``end_line``/``complexity``) verbatim instead of just the
    dataclass model can look it up there instead of re-deriving it).

    *root* anchors the analysis to a checkout other than the process cwd
    (ADR-0097): every relative input is resolved against it — never
    ``os.getcwd()`` — so a resident daemon can publish a repository's
    snapshot without chdir-ing the whole process, and every path the result
    stores (``root_path``, ``extra_roots``, each module's ``file_path``) is
    relativised back against it, keeping the artifact byte-identical to a
    CLI run from the repository root. ``None`` keeps the cwd behaviour.
    """

    def _anchored(path: Path) -> Path:
        if root is not None and not path.is_absolute():
            return root / path
        return path

    def _display(path: Path) -> Path:
        if root is None:
            return path
        try:
            return path.relative_to(root)
        except ValueError:
            return path

    src_io = _anchored(Path(source))
    stored_src = _display(src_io)
    stored_extras = tuple(_display(_anchored(Path(extra))) for extra in extra_roots)
    analyzer = CodebaseAnalyzer(
        name, stored_src, monorepo=monorepo, extra_roots=stored_extras, anchor=root
    )
    stats: dict[str, Any] = {
        "fallback": False,
        "reused": [],
        "reparsed": [],
        "removed": [],
        "cached_modules": {},
    }

    def _full_fallback() -> tuple[CodebaseAST, dict[str, Any]]:
        result = analyzer.analyze()
        stats["fallback"] = True
        stats["reparsed"] = [m.path for m in result.modules]
        stats["trees"] = analyzer.trees
        return result, stats

    try:
        cached = json.loads(Path(cache_path).read_text(encoding="utf-8"))
    except OSError, ValueError:
        return _full_fallback()
    if not isinstance(cached, dict):
        return _full_fallback()

    try:
        cached_root = _anchored(Path(str(cached.get("root_path", ""))))
        same_root = cached_root.resolve() == src_io.resolve()
    except OSError:
        same_root = False
    if cached.get("name") != name or not same_root:
        return _full_fallback()
    # The set of trees is part of the cache's identity, not a detail of how it
    # was built. Without this check, widening the snapshot to ``tests`` and
    # ``scripts`` would warm-start from the src-only cache, keep its
    # ``file_meta``, and report every new module as unchanged — a narrower
    # snapshot wearing the wider one's name.
    if list(cached.get("extra_roots", [])) != [str(extra) for extra in stored_extras]:
        return _full_fallback()
    # The capture set is part of the cache's identity for the same reason: a
    # snapshot written BEFORE a capture field existed holds module dicts
    # WITHOUT the key, and a reused dict deserializes the field as [] —
    # "captured, found none", a lie wearing a complete snapshot's name. One
    # full reparse instead (2026-10-11, SNAPSHOT_CAPTURES).
    if list(cached.get("captures", [])) != list(SNAPSHOT_CAPTURES):
        return _full_fallback()

    file_meta = cached.get("file_meta")
    if not isinstance(file_meta, dict):
        return _full_fallback()
    cached_modules_raw = cached.get("modules")
    if not isinstance(cached_modules_raw, list):
        return _full_fallback()
    cached_by_path: dict[str, dict[str, Any]] = {
        m["path"]: m for m in cached_modules_raw if isinstance(m, dict) and "path" in m
    }

    py_files = sorted(analyzer._find_python_files())
    live_modules: list[ModuleInfo] = []
    live_paths: set[str] = set()
    for py_file in py_files:
        module_path = analyzer._file_to_module_path(py_file)
        live_paths.add(module_path)
        try:
            sha1 = _sha1_of_source(_anchored(py_file))
        except OSError:
            continue
        prior = file_meta.get(module_path)
        prior_dict = cached_by_path.get(module_path)
        if (
            isinstance(prior, dict)
            and prior.get("sha1") == sha1
            and prior_dict is not None
        ):
            live_modules.append(module_info_from_dict(prior_dict))
            stats["reused"].append(module_path)
            stats["cached_modules"][module_path] = prior_dict
        else:
            module = analyzer._analyze_file(py_file)
            if module:
                live_modules.append(module)
            stats["reparsed"].append(module_path)

    stats["removed"] = sorted(set(file_meta) - live_paths)

    analyzer.result = CodebaseAST(
        name=name,
        root_path=str(stored_src),
        modules=live_modules,
        extra_roots=[str(extra) for extra in stored_extras],
        captures=list(SNAPSHOT_CAPTURES),
    )
    analyzer._compute_derived()
    stats["trees"] = analyzer.trees
    return analyzer.result, stats