scripts.architecture_checker

Reference › scripts › architecture_checker

Validates import rules, module boundaries and layering constraints across the main source package. tests/scripts/test_architecture_checker.py imports it; its sibling architecture_static_check takes the same path argument and adds a baseline ratchet. For the layers themselves, see the architecture overview.

Invocation: main takes an optional path (default src/orchestrata_ultimate), -v/--verbose for per-violation detail, -o/--output FILE to write the report, and --strict. Without --strict it exits 0 even when violations exist, so the gate passes; with --strict any violation gives a non-zero exit. A missing path is an error.

check_project parses every module and applies several checks. get_layer maps a module to a layer number and check_import flags an import that breaks the layering; protocol imports and re-exports from an __init__ are exempt (is_protocol_import, is_internal_init_reexport). Imports inside TYPE_CHECKING blocks, and imports scoped to a function, are handled separately: function-scoped cross-layer imports are counted and reported as an accepted pattern. Three further checks look at construction: check_di_instantiation finds code that instantiates the dependency-injection container outside its package, check_direct_driver_construction finds drivers built directly instead of through the factory, and check_graph_seam finds breaches of the graph seam. Test paths are excluded. The report lists files checked, the violation count and each violation.

At a glance

Source scripts/architecture_checker.py
Lines of code 691
Classes 0
Functions 22
Imports (internal) none
Imported by none
Tests that import it tests/scripts/test_architecture_checker.py

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Constants

Name Type Value
LAYER_ASSIGNMENTS dict[str, int] {'core.domain': 0, 'core.protocols': 0, 'core.packet': 0, 'core.enums': 0, 'cor…
UPWARD_IMPORT_LIMITS dict[int, int] {0: 0, 1: 1, 2: 2, 3: 3, 4: 4, 5: 5}
KNOWN_VIOLATIONS set[tuple[str, str]] {('orchestrata_ultimate.cxp.prolog_bridge', 'orchestrata_ultimate.orchestrator.…
DI_CONTAINER_HOME_PACKAGES frozenset[str] frozenset({'plugin', 'gateway', 'di'})
_GRAPH_SEAM_ALLOWED tuple[str, ...] ('core/graph_factory.py', 'core/graph/repository.py', 'core/graph/adapters/bolt…

Functions

Function Signature Summary
_is_test_path _is_test_path(path: Path) -> bool True if path lives under a tests directory or is a test module.
_di_container_top_level_package _di_container_top_level_package(path: Path) -> str | None Return the top-level orchestrata_ultimate.<pkg> package name for a source file path, or None if it isn't under an orchestrata_ultimate tree at all (e.g. a synthetic fixture rooted directly…
_instantiates_di_container _instantiates_di_container(tree: ast.AST) -> bool True if the AST contains a call shaped like DIContainer(...) or some.path.DIContainer(...).
check_di_instantiation check_di_instantiation(py_file: Path) -> str | None Return a violation description if py_file instantiates DIContainer(...) outside the plugin/gateway/di packages, else None.
_constructs_driver_directly _constructs_driver_directly(tree: ast.AST) -> str | None Return a short description of the first direct driver construction in the AST — GraphDatabase.driver(...) / AsyncGraphDatabase.driver(...) or redis.Redis(...) /…
check_direct_driver_construction check_direct_driver_construction(py_file: Path) -> str | None Return a violation if py_file constructs a graph/Redis client directly instead of via the core factories (D12), else None.
_import_probe_lines _import_probe_lines(tree: ast.AST) -> set[int] Line numbers inside try: bodies that have an ImportError (or ModuleNotFoundError, or bare except) handler.
_graph_seam_breach _graph_seam_breach(tree: ast.AST) -> str | None Return a description of the first raw-graph access in the AST — a neo4j import or a .session(...) call — else None.
check_graph_seam check_graph_seam(py_file: Path) -> str | None Return a violation if py_file reaches the graph outside the seam (D13), else None.
get_layer get_layer(module: str) -> int | None Return the numeric layer for a module path, or None if unassigned.
is_protocol_import is_protocol_import(imported_module: str) -> bool Heuristic: protocol imports live in Layer 0 modules.
is_internal_init_reexport is_internal_init_reexport(module: str, imported_module: str) -> bool Allow init.py files to re-export from their own package subtree.
check_import check_import(importing_module: str, imported_module: str) -> str | None Return violation description or None if allowed.
_is_inside_type_checking _is_inside_type_checking(node: ast.AST, tree: ast.AST) -> bool Check whether an import node is inside an if TYPE_CHECKING: block.
_is_inside_function_or_class _is_inside_function_or_class(node: ast.AST, tree: ast.AST) -> bool Check whether an import node is inside a function or class definition.
_is_inside_def _is_inside_def(node: ast.AST, tree: ast.AST) -> bool Check whether an import node is def-nested — inside a (sync or async) function body, as opposed to a bare class-body-level import.
_parse_module _parse_module(path: Path) -> tuple[str, ast.AST] | None Parse a source file under src/ and return (module_name, ast_tree).
_iter_ultimate_imports _iter_ultimate_imports(tree: ast.AST) -> list[tuple[ast.AST, str]] Return (node, imported_module) for every orchestrata_ultimate import node (import x or from x import y) in the tree, unfiltered.
extract_imports extract_imports(path: Path) -> list[tuple[str, str]] Return list of (importing_module, imported_module) from a Python file.
extract_function_scoped_imports extract_function_scoped_imports(path: Path) -> list[tuple[str, str]] Return (importing_module, imported_module) pairs for def-nested (function-scoped) orchestrata_ultimate imports.
check_project check_project(root: Path, verbose: bool = False) -> tuple[int, list[str], int, int] Check all Python files under root.
main main(argv: list[str] | None = None) -> int

Source

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

#!/usr/bin/env python3
"""Architecture checker — validates import rules, module boundaries, and layering constraints."""

from __future__ import annotations

import argparse
import ast
import sys
from pathlib import Path

# ── Layer model ──────────────────────────────────────────────────────────────
# Based on architecture-standards.md (the authoritative layer doc — this
# LAYER_ASSIGNMENTS table is the enforcement source; architecture-standards.md
# mirrors it for humans and MUST be kept in sync when this table changes).
# Layer 0 = Vocabulary (no orchestrata imports)
# Layer 1 = Infrastructure
# Layer 2 = Communication
# Layer 3 = Domain Services (no concrete peer imports)
# Layer 4 = Orchestration
# Layer 5 = Interface

LAYER_ASSIGNMENTS: dict[str, int] = {
    # Layer 0: Vocabulary
    "core.domain": 0,
    "core.protocols": 0,
    "core.packet": 0,
    "core.enums": 0,
    "core.interfaces": 0,
    "core.base": 0,
    "types": 0,
    "models": 0,
    "interfaces": 0,
    "exceptions": 0,
    # Layer 1: Infrastructure
    "core.config": 1,
    "core.logging": 1,
    "core.tracing": 1,
    "core.monitoring": 1,
    "core.patterns": 1,
    "core.state": 1,
    "core.threading": 1,
    "core.adapters": 1,
    "core.traceability": 1,
    "core.traceability_utils": 1,
    "core.redis_factory": 1,
    "core.graph_factory": 1,
    "core.infra_config": 1,
    "core.graph_directives": 1,
    "di": 1,
    "infra": 1,
    # Layer 2: Communication
    "cxp": 2,
    "postman": 2,
    "memory": 2,
    # Layer 3: Domain Services
    "agents": 3,
    "governance": 3,
    "knowledge_graph": 3,
    "providers": 3,
    "verification": 3,
    "llm": 3,
    "drivers": 3,
    "prolog": 3,
    # Layer 4: Orchestration
    "kernel": 4,
    "orchestrator": 4,
    "session": 4,
    "context": 4,
    # Layer 5: Interface
    "gateway": 5,
    "plugin": 5,
    "__main__": 5,
}

# Allowed upward imports per layer.
# Layer N may import from layers <= N.
# Layer 3 has special peer-import rules (see check_peer_import).
UPWARD_IMPORT_LIMITS: dict[int, int] = {
    0: 0,  # Layer 0 may only import from Layer 0 (but in practice only stdlib)
    1: 1,
    2: 2,
    3: 3,
    4: 4,
    5: 5,
}

# Known violations that are accepted as legacy / transitional.
# Format: (importing_module_pattern, imported_module_pattern)
KNOWN_VIOLATIONS: set[tuple[str, str]] = {
    # Conditional/lazy import for optional Prolog backend (layer-2 module
    # safely attempts layer-4 import inside try/except; runtime graceful
    # degradation when Prolog worker is unavailable).
    (
        "orchestrata_ultimate.cxp.prolog_bridge",
        "orchestrata_ultimate.orchestrator.prolog_worker",
    ),
}

# ── D4 composition rule ──────────────────────────────────────────────────────
# Ruled 2026-07-06 (plans/decisions-2026-07-06.md D4, evidence:
# plans/evidence/d4-di-container.md): an adoption audit found 8 of 9 real
# composition roots never touch DIContainer at all — they wire dependencies
# by hand (explicit constructor calls / setters). The container's scope was
# shrunk to the plugin SDK's resolve() touchpoint and the gateway's single
# assembly root (see specs/024-dependency-injection/spec.md § "Descoped
# (2026-07-06)"). This rule keeps that scope from silently regressing: any
# NEW ``DIContainer(...)`` construction outside plugin/gateway/di is flagged.
# Test files are exempt — constructing components directly in tests is the
# established, accepted pattern in this suite, not a violation.
DI_CONTAINER_HOME_PACKAGES: frozenset[str] = frozenset({"plugin", "gateway", "di"})


def _is_test_path(path: Path) -> bool:
    """True if ``path`` lives under a ``tests`` directory or is a test module."""
    return "tests" in path.parts or path.name.startswith("test_")


def _di_container_top_level_package(path: Path) -> str | None:
    """Return the top-level ``orchestrata_ultimate.<pkg>`` package name for a
    source file path, or None if it isn't under an ``orchestrata_ultimate``
    tree at all (e.g. a synthetic fixture rooted directly at a package dir)."""
    parts = path.parts
    if "orchestrata_ultimate" in parts:
        idx = parts.index("orchestrata_ultimate")
        return parts[idx + 1] if idx + 1 < len(parts) else None
    # Fall back to the first path component (covers fixtures/tests that
    # don't nest under a literal "orchestrata_ultimate" directory).
    return parts[0] if parts else None


def _instantiates_di_container(tree: ast.AST) -> bool:
    """True if the AST contains a call shaped like ``DIContainer(...)`` or
    ``some.path.DIContainer(...)``."""
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func = node.func
        if isinstance(func, ast.Name) and func.id == "DIContainer":
            return True
        if isinstance(func, ast.Attribute) and func.attr == "DIContainer":
            return True
    return False


def check_di_instantiation(py_file: Path) -> str | None:
    """Return a violation description if ``py_file`` instantiates
    ``DIContainer(...)`` outside the plugin/gateway/di packages, else None.

    Independent of the layer-import machinery below: this looks at *calls*,
    not imports, and doesn't require a ``src/`` prefix on ``py_file`` (so it
    degrades safely if ever pointed at a non-src root).
    """
    if _is_test_path(py_file):
        return None
    if _di_container_top_level_package(py_file) in DI_CONTAINER_HOME_PACKAGES:
        return None
    try:
        source = py_file.read_text(encoding="utf-8")
        tree = ast.parse(source)
    except OSError, SyntaxError, UnicodeDecodeError:
        return None
    if not _instantiates_di_container(tree):
        return None
    return (
        f"DI composition (D4): {py_file} instantiates DIContainer() outside "
        "plugin/gateway/di — composition roots must use explicit "
        "constructor/setter wiring (architecture-standards.md §composition)"
    )


# Files allowed to construct graph/Redis clients directly: the factories
# themselves (D12 — every other module must go through them).
_DRIVER_FACTORY_FILES = {"graph_factory.py", "redis_factory.py"}


def _constructs_driver_directly(tree: ast.AST) -> str | None:
    """Return a short description of the first direct driver construction in
    the AST — ``GraphDatabase.driver(...)`` / ``AsyncGraphDatabase.driver(...)``
    or ``redis.Redis(...)`` / ``redis.asyncio.Redis(...)`` — else None."""
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        func = node.func
        if not isinstance(func, ast.Attribute):
            continue
        if func.attr == "driver":
            base = func.value
            name = base.id if isinstance(base, ast.Name) else getattr(base, "attr", "")
            if name in ("GraphDatabase", "AsyncGraphDatabase"):
                return f"{name}.driver(...) at line {node.lineno}"
        if func.attr == "Redis":
            base = func.value
            name = base.id if isinstance(base, ast.Name) else getattr(base, "attr", "")
            if name in ("redis", "asyncio"):
                return f"{name}.Redis(...) at line {node.lineno}"
    return None


def check_direct_driver_construction(py_file: Path) -> str | None:
    """Return a violation if ``py_file`` constructs a graph/Redis client
    directly instead of via the core factories (D12), else None.

    Call-shape (AST) based like ``check_di_instantiation`` — string literals
    and docstrings never match. Tests and the factory modules are exempt.
    """
    if _is_test_path(py_file):
        return None
    if py_file.name in _DRIVER_FACTORY_FILES:
        return None
    try:
        source = py_file.read_text(encoding="utf-8")
        tree = ast.parse(source)
    except OSError, SyntaxError, UnicodeDecodeError:
        return None
    hit = _constructs_driver_directly(tree)
    if hit is None:
        return None
    return (
        f"Persistence factories (D12): {py_file} constructs {hit} directly — "
        "use core.graph_factory.create_graph/create_async_graph or "
        "core.redis_factory.create_sync_redis/create_async_redis"
    )


# ── D13: operational-graph repository seam (backlog #163 phase 6) ────────────
#
# After the phase 0-5b migration, raw graph access lives ONLY in the seam and
# the documented escape hatches. Two shapes are gated, because phase 5b proved
# imports alone miss attribute-coupled usage (an injected driver never imports
# neo4j — only a `.session(` sweep found QueryMixin, context_injector, etc.):
#   (a) `import neo4j` / `from neo4j import ...` module-level or nested;
#   (b) `.session(...)` attribute CALLS (AST-shape: strings/docstrings that
#       merely mention driver.session() never match).
#
# Allowlist rationale (plans/neo4j-seam-design.md):
#   core/graph_factory.py                 — THE driver factory
#   core/graph/repository.py             — seam factory/protocol layer
#   core/graph/adapters/bolt_sync.py     — the seam itself
#   core/graph/adapters/bolt_async.py    — the seam itself
#   knowledge_graph/graph_client.py      — plan ruling: "stays as-is — itself
#                                          nearly a repository"
#   knowledge_graph/kg_context.py        — documented raw-Driver escape hatch
#                                          (external inspection) + typing
#   knowledge_graph/semantic_bridge.py   — documented escape hatch: sync
#                                          Memgraph client (Bolt) for the
#                                          assembler's synchronous call path;
#                                          the proper semantic_graph adapter
#                                          is async-only, and the Neo4j side
#                                          already goes through
#                                          create_graph_repository()
#   session/kg_sync/_base.py             — documented escape hatch: bespoke
#                                          deadlock-retry + chunk-splitting
#                                          needs begin_transaction
#   semantic_graph/adapters/memgraph.py  — the MEMGRAPH adapter: the neo4j
#                                          Bolt package is its client library;
#                                          this file IS that graph's seam
#   semantic_graph/ha_controller.py      — Memgraph HA controller, same
#   factory/promotion/store.py           — documented escape hatch: owns the
#                                          ADR-0019 watermark write/read pair
#                                          and exposes its own database-bound
#                                          `session()` to callers (SPEC-033
#                                          B4a/B9). The original rationale —
#                                          "needs raw `Bookmarks` +
#                                          `driver.session(bookmarks=...)`" —
#                                          expired on 2026-07-26: the
#                                          orchgraph engine emits no bookmarks,
#                                          so the watermark is now an explicit
#                                          visibility token and `Bookmarks` is
#                                          gone from this file entirely.
#   factory/scheduler/scheduler.py       — parks on ADR-0019 watermarks between
#                                          DAG levels. Still allowlisted, but
#                                          for a NARROWER reason than before:
#                                          it no longer builds a bookmarked
#                                          driver session, it borrows the
#                                          promotion store's own
#                                          database-bound `store.session()`.
#                                          D13 matches any `.session(` call
#                                          shape, seam-obtained or not.
_GRAPH_SEAM_ALLOWED: tuple[str, ...] = (
    "core/graph_factory.py",
    "core/graph/repository.py",
    "core/graph/adapters/bolt_sync.py",
    "core/graph/adapters/bolt_async.py",
    "knowledge_graph/graph_client.py",
    "knowledge_graph/kg_context.py",
    "knowledge_graph/semantic_bridge.py",
    "session/kg_sync/_base.py",
    "semantic_graph/adapters/memgraph.py",
    "semantic_graph/ha_controller.py",
    "factory/promotion/store.py",
    "factory/scheduler/scheduler.py",
)


def _import_probe_lines(tree: ast.AST) -> set[int]:
    """Line numbers inside ``try:`` bodies that have an ``ImportError`` (or
    ``ModuleNotFoundError``, or bare ``except``) handler. A bare
    ``import neo4j`` there is an availability PROBE — it feature-gates on the
    optional package's presence and reaches nothing — so D13 exempts it.
    ``from neo4j import X`` stays gated even here: it binds driver symbols."""
    probe_lines: set[int] = set()
    for node in ast.walk(tree):
        if not isinstance(node, ast.Try):
            continue
        catches_import_error = False
        for handler in node.handlers:
            if handler.type is None:
                catches_import_error = True
                break
            names = []
            for t in (
                handler.type.elts
                if isinstance(handler.type, ast.Tuple)
                else [handler.type]
            ):
                if isinstance(t, ast.Name):
                    names.append(t.id)
            if "ImportError" in names or "ModuleNotFoundError" in names:
                catches_import_error = True
                break
        if not catches_import_error:
            continue
        for stmt in node.body:
            end = getattr(stmt, "end_lineno", stmt.lineno)
            probe_lines.update(range(stmt.lineno, end + 1))
    return probe_lines


def _graph_seam_breach(tree: ast.AST) -> str | None:
    """Return a description of the first raw-graph access in the AST —
    a ``neo4j`` import or a ``.session(...)`` call — else None."""
    probe_lines = _import_probe_lines(tree)
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name == "neo4j" or alias.name.startswith("neo4j."):
                    if node.lineno in probe_lines:
                        continue  # availability probe, not graph access
                    return f"import {alias.name} at line {node.lineno}"
        elif isinstance(node, ast.ImportFrom):
            mod = node.module or ""
            if node.level == 0 and (mod == "neo4j" or mod.startswith("neo4j.")):
                return f"from {mod} import ... at line {node.lineno}"
        elif isinstance(node, ast.Call):
            func = node.func
            if isinstance(func, ast.Attribute) and func.attr == "session":
                return f".session(...) call at line {node.lineno}"
    return None


def check_graph_seam(py_file: Path) -> str | None:
    """Return a violation if ``py_file`` reaches the graph outside the seam
    (D13), else None. Tests and the allowlisted seam/escape-hatch files are
    exempt."""
    if _is_test_path(py_file):
        return None
    posix = py_file.as_posix()
    if any(posix.endswith(allowed) for allowed in _GRAPH_SEAM_ALLOWED):
        return None
    try:
        source = py_file.read_text(encoding="utf-8")
        tree = ast.parse(source)
    except OSError, SyntaxError, UnicodeDecodeError:
        return None
    hit = _graph_seam_breach(tree)
    if hit is None:
        return None
    return (
        f"Graph seam (D13): {py_file} reaches the graph driver raw ({hit}) — go through "
        "core.graph (GraphRepository / create_graph_repository) or add the "
        "file to the documented escape-hatch allowlist with a reason "
        "(plans/neo4j-seam-design.md phase 6)"
    )


def get_layer(module: str) -> int | None:
    """Return the numeric layer for a module path, or None if unassigned."""
    # Strip package prefix so we look up relative to the package root.
    prefix = "orchestrata_ultimate."
    if module.startswith(prefix):
        module = module[len(prefix) :]
    parts = module.split(".")
    # Try progressively longer prefixes
    for i in range(len(parts), 0, -1):
        candidate = ".".join(parts[:i])
        if candidate in LAYER_ASSIGNMENTS:
            return LAYER_ASSIGNMENTS[candidate]
    return None


def is_protocol_import(imported_module: str) -> bool:
    """Heuristic: protocol imports live in Layer 0 modules."""
    layer = get_layer(imported_module)
    return layer == 0


def is_internal_init_reexport(module: str, imported_module: str) -> bool:
    """Allow __init__.py files to re-export from their own package subtree."""
    if not module.endswith(".__init__"):
        return False
    pkg = module.removesuffix(".__init__") + "."
    return imported_module.startswith(pkg)


def check_import(importing_module: str, imported_module: str) -> str | None:
    """Return violation description or None if allowed."""
    if not imported_module.startswith("orchestrata_ultimate"):
        return None

    importer_layer = get_layer(importing_module)
    imported_layer = get_layer(imported_module)

    if importer_layer is None or imported_layer is None:
        return None  # Unassigned modules are not checked

    # Allow __init__ re-exports
    if is_internal_init_reexport(importing_module, imported_module):
        return None

    # Upward import check
    if imported_layer > importer_layer:
        return (
            f"upward import: {importing_module} (layer {importer_layer}) "
            f"imports {imported_module} (layer {imported_layer})"
        )

    # (A former "layer-0 violation" branch here was unreachable: any
    # imported_layer > importer_layer already returned as an upward import
    # above, so a layer-0 importer could only reach it with imported_layer == 0,
    # which is allowed. Removed 2026-07-06.)

    # Layer 3 peer-import check: concrete imports from *different* peer Layer-3
    # top-level packages are forbidden.  Imports within the same package tree
    # (e.g. agents.foo -> agents.bar) are allowed.
    # Protocol imports (Layer 0) are always fine.
    if importer_layer == 3 and imported_layer == 3:
        if not is_protocol_import(imported_module):
            # Determine top-level package for each (e.g. "agents", "knowledge_graph")
            importer_pkg = importing_module.removeprefix("orchestrata_ultimate.").split(
                "."
            )[0]
            imported_pkg = imported_module.removeprefix("orchestrata_ultimate.").split(
                "."
            )[0]
            if importer_pkg != imported_pkg:
                return (
                    f"peer import: {importing_module} (layer 3) "
                    f"imports concrete {imported_module} (layer 3); "
                    f"use protocols instead"
                )

    return None


def _is_inside_type_checking(node: ast.AST, tree: ast.AST) -> bool:
    """Check whether an import node is inside an ``if TYPE_CHECKING:`` block."""
    for parent in ast.walk(tree):
        if isinstance(parent, ast.If):
            if isinstance(parent.test, ast.Name) and parent.test.id == "TYPE_CHECKING":
                for child in ast.walk(parent):
                    if child is node:
                        return True
    return False


def _is_inside_function_or_class(node: ast.AST, tree: ast.AST) -> bool:
    """Check whether an import node is inside a function or class definition."""
    for parent in ast.walk(tree):
        if isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
            for child in ast.walk(parent):
                if child is node:
                    return True
    return False


def _is_inside_def(node: ast.AST, tree: ast.AST) -> bool:
    """Check whether an import node is def-nested — inside a (sync or async)
    function body, as opposed to a bare class-body-level import. This is the
    "function-scoped" / lazy-import shape (RULED D7): imports guarded behind
    a call, used for optional/heavy deps or to sidestep a layer exception."""
    for parent in ast.walk(tree):
        if isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef)):
            for child in ast.walk(parent):
                if child is node:
                    return True
    return False


def _parse_module(path: Path) -> tuple[str, ast.AST] | None:
    """Parse a source file under ``src/`` and return (module_name, ast_tree).

    Returns None if the module name can't be derived, the file can't be
    read, or it fails to parse.
    """
    rel = path.relative_to("src")
    module = str(rel.with_suffix("")).replace("/", ".").replace("\\", ".")

    try:
        source = path.read_text(encoding="utf-8")
    except Exception:
        return None

    try:
        tree = ast.parse(source)
    except SyntaxError:
        return None

    return module, tree


def _iter_ultimate_imports(tree: ast.AST) -> list[tuple[ast.AST, str]]:
    """Return (node, imported_module) for every orchestrata_ultimate import
    node (``import x`` or ``from x import y``) in the tree, unfiltered."""
    found: list[tuple[ast.AST, str]] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom):
            if node.module and node.module.startswith("orchestrata_ultimate"):
                found.append((node, node.module))
        elif isinstance(node, ast.Import):
            for alias in node.names:
                if alias.name.startswith("orchestrata_ultimate"):
                    found.append((node, alias.name))
    return found


def extract_imports(path: Path) -> list[tuple[str, str]]:
    """Return list of (importing_module, imported_module) from a Python file.

    Excludes ``TYPE_CHECKING``-guarded imports and any import nested inside a
    function or class body — those are the accepted lazy-import shape, and
    are surfaced separately by :func:`extract_function_scoped_imports`.
    """
    imports: list[tuple[str, str]] = []
    parsed = _parse_module(path)
    if parsed is None:
        return imports
    module, tree = parsed

    for node, imported_module in _iter_ultimate_imports(tree):
        if _is_inside_type_checking(node, tree):
            continue
        if _is_inside_function_or_class(node, tree):
            continue
        imports.append((module, imported_module))

    return imports


def extract_function_scoped_imports(path: Path) -> list[tuple[str, str]]:
    """Return (importing_module, imported_module) pairs for def-nested
    (function-scoped) orchestrata_ultimate imports.

    This is the accepted lazy-import pattern (RULED D7 — see
    architecture-standards.md): function-scoped cross-layer imports for
    optional/heavy deps or layer-exception avoidance. They are excluded from
    the main violation check but reported at INFO level for visibility, since
    they still create real runtime coupling.
    """
    imports: list[tuple[str, str]] = []
    parsed = _parse_module(path)
    if parsed is None:
        return imports
    module, tree = parsed

    for node, imported_module in _iter_ultimate_imports(tree):
        if _is_inside_type_checking(node, tree):
            continue
        if _is_inside_def(node, tree):
            imports.append((module, imported_module))

    return imports


def check_project(root: Path, verbose: bool = False) -> tuple[int, list[str], int, int]:
    """Check all Python files under root.

    Returns (violation_count, messages, files_checked,
    function_scoped_cross_layer_count). The last figure is an INFO-level
    count of def-nested imports that reach a higher layer (RULED D7) — it
    never affects violation_count or the process exit code.
    """
    violations: list[str] = []
    files_checked = 0
    function_scoped_cross_layer_count = 0

    for py_file in sorted(root.rglob("*.py")):
        if "__pycache__" in py_file.parts:
            continue
        files_checked += 1
        for importing_module, imported_module in extract_imports(py_file):
            violation = check_import(importing_module, imported_module)
            if violation:
                # Check against known waivers
                waived = any(
                    importing_module.startswith(imp_pat)
                    and imported_module.startswith(exp_pat)
                    for imp_pat, exp_pat in KNOWN_VIOLATIONS
                )
                if not waived:
                    msg = f"{importing_module} -> {imported_module}: {violation}"
                    violations.append(msg)
                    if verbose:
                        print(f"  VIOLATION: {msg}")

        for importing_module, imported_module in extract_function_scoped_imports(
            py_file
        ):
            # Count every def-nested import that WOULD be a violation (upward
            # OR layer-3 peer) if it were eager — the full set of lazy imports
            # dodging the layer rules, per D7's visibility intent.
            if check_import(importing_module, imported_module) is not None:
                function_scoped_cross_layer_count += 1

        di_violation = check_di_instantiation(py_file)
        if di_violation is not None:
            violations.append(di_violation)
            if verbose:
                print(f"  VIOLATION: {di_violation}")

        driver_violation = check_direct_driver_construction(py_file)
        if driver_violation is not None:
            violations.append(driver_violation)
            if verbose:
                print(f"  VIOLATION: {driver_violation}")

        seam_violation = check_graph_seam(py_file)
        if seam_violation is not None:
            violations.append(seam_violation)
            if verbose:
                print(f"  VIOLATION: {seam_violation}")

    return (
        len(violations),
        violations,
        files_checked,
        function_scoped_cross_layer_count,
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Validate architecture import rules")
    parser.add_argument(
        "path", nargs="?", default="src/orchestrata_ultimate", help="Path to check"
    )
    parser.add_argument(
        "-v", "--verbose", action="store_true", help="Show per-violation details"
    )
    parser.add_argument("-o", "--output", help="Write report to file")
    parser.add_argument(
        "--strict", action="store_true", help="Exit non-zero on any violation"
    )
    args = parser.parse_args(argv)

    root = Path(args.path)
    if not root.exists():
        print(f"ERROR: path does not exist: {root}", file=sys.stderr)
        return 1

    print(f"Checking architecture rules under {root.resolve()} ...")
    violation_count, violations, files_checked, function_scoped_cross_layer_count = (
        check_project(root, verbose=args.verbose)
    )

    lines: list[str] = []
    lines.append(f"Files checked: {files_checked}")
    lines.append(f"Violations found: {violation_count}")
    lines.append(
        f"INFO: {function_scoped_cross_layer_count} function-scoped cross-layer "
        "imports (accepted pattern, see architecture-standards.md)"
    )
    if violations:
        lines.append("")
        lines.append("Violations:")
        lines.extend(f"  - {v}" for v in violations)

    report = "\n".join(lines)
    print(report)

    if args.output:
        Path(args.output).write_text(report + "\n", encoding="utf-8")
        print(f"\nReport written to {args.output}")

    # In non-strict mode we exit 0 so the gate passes while violations are
    # documented.  In strict mode any violation is a failure.
    if args.strict and violation_count > 0:
        return 1
    return 0


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