scripts.check_env_contract

Reference › scripts › check_env_contract

The env-contract gate, the mechanized half of ruling RUL-019. It statically analyses the source tree to check how program entrypoints load their environment configuration. tests/scripts/test_check_env_contract.py imports it; a gate or a human runs it.

Invocation: main takes --repo-root (default this checkout), --max-depth and --json. --max-depth is the import-closure depth searched for an infrastructure trigger; the default is 1, meaning the entrypoint and what it imports directly, and -1 means unlimited. The exit code is 0 when clean, 1 for violations and 2 when the tree cannot be analysed (AnalysisError).

The analysis has three layers. First, module_map lists the modules and load_infra_keys reads the set of infrastructure variable names. A _ModuleVisitor walks each module's syntax tree and records, as ModuleFacts, which names it reads from the environment, whether it loads a dotenv file and whether it writes to the environment; a module is then classed as a mutating loader or a conforming loader. Second, import_edges builds the module import graph and collect_entrypoints gathers entrypoints from the project's script table, the deployment unit files (execstart_commands) and modules with a main guard. Third, closure_with_witness computes what each entrypoint reaches within the depth limit, with a witness path, and check_rule_a and the later rule checks produce a Violation for each breach. Only variable names are handled; values are never read.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

AnalysisError

Bases: Exception · Line 166

The tree cannot be analysed at all — exit 2, never a silent pass.

Violation (dataclass)

Decorators: dataclass · Line 171

One finding. witness carries the import path that proves it, so a false positive is diagnosable in one read instead of being argued about.

ModuleFacts (dataclass)

Decorators: dataclass · Line 182

Everything the gate needs from one src/ module, extracted once.

Entrypoint (dataclass)

Decorators: dataclass · Line 235

One member of the E1/E2/E3 entrypoint universe, with its provenance.

_ModuleVisitor

Bases: ast.NodeVisitor · Line 338

Collects import edges, resolved call targets, and env facts.

Functions

Source

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

#!/usr/bin/env python3
"""Env-contract gate — the mechanized half of RUL-019.

Ruling text, verbatim (this is the ``gate scripts/check_env_contract.py
RUL-019`` mechanism declared by
``~/workspace/knowledge-wiki/rulings/RUL-019-pydantic-dotenv-config-pattern.md``):

RUL-019 — Config modules use Pydantic BaseSettings, and env files must be auto-loaded by every entrypoint via _load_dotenv() (secrets-only .env, then the active ORCHESTRATA_ENV_PROFILE overlay env/{dev,test,prod}.env with override=True) — never left to direnv or manual sourcing. Currently declared `none (judgment)`, i.e. prose only.

MEASURED STATE (why this is mechanizable and worth mechanizing):
- There is NO canonical two-phase loader in src/. The only correct implementations of the ruling's contract live OUTSIDE the package: scripts/start_test_services.py::_load_dotenv (lines 95-127, .env setdefault then env/<profile>.env with override) and tests/conftest.py (lines 263/277).
- src/ carries three DIVERGENT copies, none of which does the profile overlay: gateway/cli.py::_load_dotenv (.env only, setdefault), kernel/cli_client.py::load_dotenv (explicitly "mirrors gateway.cli._load_dotenv"), llm/client.py::_load_dotenv (returns a mapping, never mutates os.environ).
- Two production entrypoints reach a REQUIRED raising infra seam and load nothing: agents/runtime_main.py (CXP_GRPC_PORT via cxp.endpoint_config.cxp_grpc_port) and knowledge_graph/server.py (KG_GRPC_PORT via cxp.endpoint_config.require_port). Both are launched by systemd ExecStart under deploy/ and survive only on EnvironmentFile= injection.
- Verified empirically, not theoretically: `env -i .venv/bin/python -m orchestrata_ultimate.agents.runtime_main` → rc=1, CXPConfigError "…define one in .env." The process names .env in its own failure message and never reads it.

RULING: mechanizable as a gate. The ruling is NOT "config modules should look like infra_config.py" (unfalsifiable, correctly judgment-only). It IS "every process that needs the infra contract must load the infra contract" — that is a reachability question over an enumerable entrypoint set, and it has a real answer today: two failures.

WHAT THIS GATE CHECKS — two rules. Rule B is the ruling; Rule A is what stops
Rule B being satisfiable by a stub.

ENTRYPOINT UNIVERSE (enumerable, deterministic; never "the file is called
cli.py"):

    E1  ``[project.scripts]`` targets in ``pyproject.toml`` that resolve to a
        module under ``src/``.
    E2  ``ExecStart=`` lines under ``deploy/`` invoking ``python -m <mod>`` or
        a path under ``src/``. Backslash line-continuations are joined before
        matching (``deploy/systemd/*.service`` wraps every ExecStart) — a
        line-wise regex would silently miss the kernel unit entirely, and a
        missed entrypoint is a FALSE NEGATIVE, the worse direction here.
    E3  ``src/**/*.py`` carrying a top-level ``if __name__ == "__main__":``.

IN-SCOPE FILTER: an entrypoint is in scope iff its transitive intra-package
import closure reaches a *read* of one of ``core.infra_env_rule
.INFRA_ENDPOINT_KEYS`` (the RUL-003 set — parsed out of that module, never
re-listed here, so the two rules can never drift) or a *call* to one of the
raising resolver seams in :data:`RESOLVER_SEAMS`. Everything else — the
infra-free CLIs, ``session/examples/*``, ``session/hooks/*`` — is silent.

RULE B — COVERAGE (the ruling): every in-scope entrypoint's closure must reach
a call to a REGISTERED env loader (:data:`REGISTERED_LOADERS` — an explicit
short list, never a name heuristic). Reachability, not unconditional
execution: ``gateway/cli.py`` guards its call with ``if env is None:`` for the
hermetic-test seam, and that legitimately counts.

RULE A — LOADER CONFORMANCE / NO DUPLICATES: at most ONE module under ``src/``
may define a MUTATING dotenv loader — defined precisely as a module that both
names a file whose basename is ``.env`` AND writes into ``os.environ``
(subscript assign / ``setdefault`` / ``update`` / ``pop`` / ``clear``). That
one module must statically reference ``ORCHESTRATA_ENV_PROFILE``, an ``env/``
profile path, and apply the profile with OVERRIDE (a bare ``os.environ[k] = v``
write, not only ``setdefault``). A second such module is a violation and must
import the canonical one.

DELIBERATELY NOT FLAGGED: non-mutating ``.env`` READERS (``llm/client.py``
``_load_dotenv`` returns a merged mapping; ``harness/history.py``
``_read_dotenv_graph_vars`` likewise) — the mutation predicate is exactly what
keeps them out. Non-entrypoint modules that read infra keys (``gateway/
server.py`` reads ``CXP_GRPC_PORT`` but is not a process; it is reached via
``gateway/cli.py``, which loads) — the analysis is transitive-from-entrypoint,
never per-file. ``scripts/*.py`` — operator tooling, out of scope entirely.

NOT MECHANIZED: the "config modules use Pydantic BaseSettings" clause. It has
no falsifiable subject — there is no enumerable set of "config modules"
distinct from "modules that read configuration", so a gate for it would either
flag every module touching ``os.environ`` (noise) or merely re-assert that
``core/infra_config.py`` exists (vacuous). That clause stays a second
``none (judgment)`` bullet on the ruling page, so the page states exactly which
half is covered.

CLOSURE DEPTH — why the trigger search is bounded (``--max-depth``, default 1),
measured, not assumed. Import reachability is a PROXY for "this process reads
the infra contract at startup", and at unlimited depth the proxy decouples from
the truth it stands for. Measured on this tree: unbounded, the gate reports 16
violations; 14 of them trace to ONE trigger site,
``core/graph/adapters/bolt_async.py:88``, where ``resolve_kg_database()`` is
called inside ``_session_kwargs`` — a method that runs only when a graph
session is actually opened. Importing the adapter reads nothing. Those 14
include ``session/examples/basic_usage.py``, which is not a process anybody
runs against infra. A gate that fires there gets switched off, and a
switched-off gate enforces nothing — the exact RUL-023 failure this exercise
exists to prevent.

At depth 1 the trigger must appear in the entrypoint module itself or in a
module it names in its OWN import statements — the process's declared startup
surface. That keeps the two real findings (``agents/runtime_main.py`` and
``knowledge_graph/server.py`` both call the seam in their own module) and
``gateway/cli.py`` (whose directly-imported ``gateway/server.py`` reads
``CXP_GRPC_PORT``).

The bound errs toward FALSE NEGATIVES: an entrypoint that delegates its infra
read two or more hops down is not flagged. That is deliberate and it is
VISIBLE, never silently dropped — every run reports ``deep_scope_only``, the
entrypoints that would be in scope at unlimited depth, with the witness path
to each trigger. Raise the bound with ``--max-depth`` (``-1`` = unlimited) to
audit that region.

Trigger search is depth-bounded; LOADER search is not. Scoping narrowly and
crediting coverage broadly both bias away from false alarms, which is the only
direction a gate survives in.

SCOPE LIMIT, STATED: this gate proves an entrypoint REACHES a conforming
loader; it does not prove the loader RAN before the first infra read
(ordering). An ordering proof needs runtime evidence. Do not claim ordering
coverage.

Exit 0 clean, 1 on any violation, 2 when the tree cannot be analysed at all
(no ``src/``, unparseable ``infra_env_rule.py``).

Usage:
    python scripts/check_env_contract.py [--repo-root PATH] [--max-depth N] [--json]
"""

from __future__ import annotations

import argparse
import ast
import json
import re
import sys
import tomllib
from collections import deque
from dataclasses import asdict, dataclass, field
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

RULE_ID = "RUL-019"

#: The raising resolver seams. A call to any of these means the caller depends
#: on a REQUIRED infra key being present in the environment — the seam raises
#: rather than defaulting (RUL-003), so an unloaded process dies at startup.
RESOLVER_SEAMS: frozenset[str] = frozenset(
    {
        "orchestrata_ultimate.cxp.endpoint_config.require_port",
        "orchestrata_ultimate.cxp.endpoint_config.cxp_grpc_port",
        "orchestrata_ultimate.core.graph_factory.resolve_uri",
        "orchestrata_ultimate.core.graph_factory.resolve_kg_database",
        "orchestrata_ultimate.core.redis_factory.resolve_redis_host",
        "orchestrata_ultimate.core.redis_factory.resolve_redis_port",
    }
)

#: The env-contract loaders an in-scope entrypoint may reach. EXPLICIT — never
#: a name heuristic ("anything called ``*load_dotenv*``"), because the whole
#: point of Rule A is that hand-rolled look-alikes are the defect. Keep this
#: list in the same commit as any new loader; Rule A's one-canonical-loader
#: constraint is what keeps it from growing.
REGISTERED_LOADERS: frozenset[str] = frozenset(
    {
        # THE single canonical two-phase loader. The two migration entries that
        # used to sit beside it (``gateway.cli._load_dotenv`` and
        # ``kernel.cli_client.load_dotenv``) were hand-rolled ``.env``-only
        # walk-ups that never applied the ``env/<profile>.env`` overlay; both
        # now delegate here and are reached transitively, so the allowlist is
        # back to the one name Rule A intends.
        "orchestrata_ultimate.core.env_bootstrap.load_env_contract",
    }
)

_ENVIRON_MUTATORS = frozenset({"setdefault", "update", "pop", "clear", "popitem"})
_EXECSTART_RE = re.compile(r"^ExecStart\s*=\s*(.*)$", re.MULTILINE)
_CONTINUATION_RE = re.compile(r"\\\s*\n\s*")


class AnalysisError(Exception):
    """The tree cannot be analysed at all — exit 2, never a silent pass."""


@dataclass
class Violation:
    """One finding. *witness* carries the import path that proves it, so a
    false positive is diagnosable in one read instead of being argued about."""

    rule: str  # "A" or "B"
    artifact: str  # repo-relative path
    message: str
    witness: str = ""


@dataclass
class ModuleFacts:
    """Everything the gate needs from one ``src/`` module, extracted once."""

    dotted: str
    path: Path
    imports: set[str] = field(default_factory=set)
    calls: set[str] = field(default_factory=set)
    infra_reads: dict[str, int] = field(default_factory=dict)  # key -> lineno
    seam_calls: dict[str, int] = field(default_factory=dict)  # seam -> lineno
    loader_calls: dict[str, int] = field(default_factory=dict)  # loader -> lineno
    names_dotenv_file: bool = False
    mutates_environ: bool = False
    #: A single SCOPE both names a ``.env`` file and writes ``os.environ``.
    #: Module-wide co-occurrence is NOT enough — see ``_names_dotenv_and_writes``.
    dotenv_write_scope: bool = False
    overrides_environ: bool = False
    names_env_profile: bool = False
    names_profile_overlay_path: bool = False
    is_main_guarded: bool = False

    @property
    def is_mutating_loader(self) -> bool:
        """A hand-rolled ``.env`` loader: ONE SCOPE both names a ``.env`` file
        and writes into ``os.environ``. Non-mutating readers
        (``llm/client.py``) are deliberately excluded — they serve a different
        job (returning a mapping to a caller) and are not a second source of
        truth for the process environment.

        The scope correlation is load-bearing. Module-wide co-occurrence of
        the two facts is not evidence of anything: measured 2026-08-05, adding
        the project-mandated free-threading guard
        ``os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")`` to
        ``core/infra_config.py`` — whose only ``.env`` mention is pydantic's
        ``env_file=".env"`` in a class body — reported it as a rogue loader
        and told the author to "fold this onto" the canonical env loader.
        Eight live modules name a ``.env`` path; each was one unrelated
        ``os.environ`` write away from the same false accusation."""
        return self.dotenv_write_scope

    @property
    def is_conforming_loader(self) -> bool:
        """The full RUL-019 two-phase contract: a mutating loader that is
        profile-aware AND applies the overlay with override (a bare
        ``os.environ[k] = v`` write, not only ``setdefault``)."""
        return (
            self.is_mutating_loader
            and self.names_env_profile
            and self.names_profile_overlay_path
            and self.overrides_environ
        )


@dataclass
class Entrypoint:
    """One member of the E1/E2/E3 entrypoint universe, with its provenance."""

    module: str
    origin: str  # human-readable provenance, e.g. "deploy/x.service ExecStart"


# --------------------------------------------------------------------------
# module map
# --------------------------------------------------------------------------


def module_map(src_root: Path) -> dict[str, Path]:
    """``dotted name -> file`` for every ``.py`` under *src_root*.

    ``pkg/__init__.py`` maps to the bare package name so ``from pkg import x``
    resolves an edge onto the package marker as well as the submodule.
    """
    out: dict[str, Path] = {}
    for path in sorted(src_root.rglob("*.py")):
        rel = path.relative_to(src_root)
        parts = list(rel.parts)
        if parts[-1] == "__init__.py":
            parts = parts[:-1]
            if not parts:
                continue
        else:
            parts[-1] = parts[-1][: -len(".py")]
        out[".".join(parts)] = path
    return out


def load_infra_keys(src_root: Path) -> frozenset[str]:
    """Parse ``INFRA_ENDPOINT_KEYS`` out of ``core/infra_env_rule.py``.

    Parsed statically rather than imported: the gate must work against a
    throwaway fixture tree that is not installed, and against a checkout whose
    package may not be importable. Reusing that already-canonical set (the
    RUL-003 keys) instead of re-listing it here is what stops RUL-003 and
    RUL-019 drifting apart.
    """
    path = src_root / "orchestrata_ultimate" / "core" / "infra_env_rule.py"
    if not path.exists():
        raise AnalysisError(f"missing canonical key set: {path}")
    try:
        tree = ast.parse(path.read_text(encoding="utf-8"))
    except SyntaxError as exc:
        raise AnalysisError(f"cannot parse {path}: {exc}") from exc
    for node in tree.body:
        target = None
        value = None
        if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
            target, value = node.target.id, node.value
        elif isinstance(node, ast.Assign) and len(node.targets) == 1:
            if isinstance(node.targets[0], ast.Name):
                target, value = node.targets[0].id, node.value
        if target != "INFRA_ENDPOINT_KEYS" or value is None:
            continue
        payload: ast.expr = value
        if (
            isinstance(value, ast.Call)
            and isinstance(value.func, ast.Name)
            and value.func.id == "frozenset"
            and value.args
        ):
            payload = value.args[0]
        try:
            keys = ast.literal_eval(payload)
        except ValueError as exc:
            raise AnalysisError(f"INFRA_ENDPOINT_KEYS is not a literal: {exc}") from exc
        return frozenset(str(k) for k in keys)
    raise AnalysisError(f"INFRA_ENDPOINT_KEYS not found in {path}")


# --------------------------------------------------------------------------
# per-module static analysis
# --------------------------------------------------------------------------


def _dotted_of(node: ast.expr) -> str | None:
    """``a.b.c`` for a Name/Attribute chain, else ``None``."""
    parts: list[str] = []
    cur: ast.expr = node
    while isinstance(cur, ast.Attribute):
        parts.append(cur.attr)
        cur = cur.value
    if not isinstance(cur, ast.Name):
        return None
    parts.append(cur.id)
    return ".".join(reversed(parts))


def _resolve_relative(module: str | None, level: int, package_parts: list[str]) -> str:
    base = (
        package_parts[: len(package_parts) - (level - 1)]
        if level > 1
        else list(package_parts)
    )
    if module:
        base = base + module.split(".")
    return ".".join(base)


class _ModuleVisitor(ast.NodeVisitor):
    """Collects import edges, resolved call targets, and env facts.

    Call targets are resolved through the module's own IMPORT BINDINGS (and
    its top-level ``def``s), never by bare name matching — a local helper that
    happens to be called ``require_port`` must not read as the CXP seam.
    """

    def __init__(self, dotted: str, infra_keys: frozenset[str]) -> None:
        self.dotted = dotted
        self.infra_keys = infra_keys
        self.package_parts = dotted.split(".")[:-1]
        self.bindings: dict[str, str] = {}
        self.environ_aliases: set[str] = set()
        self.facts = ModuleFacts(dotted=dotted, path=Path())
        self.local_defs: set[str] = set()

    # -- imports ---------------------------------------------------------
    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            local = alias.asname or alias.name.split(".")[0]
            target = alias.name if alias.asname else alias.name.split(".")[0]
            self.bindings[local] = target
            self.facts.imports.add(alias.name)
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        module = (
            _resolve_relative(node.module, node.level, self.package_parts)
            if node.level
            else (node.module or "")
        )
        if module:
            self.facts.imports.add(module)
        for alias in node.names:
            local = alias.asname or alias.name
            if module == "os" and alias.name == "environ":
                self.environ_aliases.add(local)
            qualified = f"{module}.{alias.name}" if module else alias.name
            self.bindings[local] = qualified
            self.facts.imports.add(qualified)
        self.generic_visit(node)

    # -- defs ------------------------------------------------------------
    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
        self.local_defs.add(node.name)
        self.generic_visit(node)

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
        self.local_defs.add(node.name)
        self.generic_visit(node)

    # -- env facts -------------------------------------------------------
    def is_environ(self, node: ast.expr) -> bool:
        """Public alias — the scope-correlation pass reuses this resolution."""
        return self._is_environ(node)

    def _is_environ(self, node: ast.expr) -> bool:
        dotted = _dotted_of(node)
        if dotted in ("os.environ", "environ"):
            return dotted == "os.environ" or "environ" in self.environ_aliases
        return False

    def visit_Constant(self, node: ast.Constant) -> None:
        if isinstance(node.value, str):
            value = node.value
            if value == ".env" or value.endswith("/.env"):
                self.facts.names_dotenv_file = True
            if value == "ORCHESTRATA_ENV_PROFILE":
                self.facts.names_env_profile = True
            if value == "env" or "env/" in value:
                self.facts.names_profile_overlay_path = True
        self.generic_visit(node)

    def visit_JoinedStr(self, node: ast.JoinedStr) -> None:
        # f"{profile}.env" — the profile-overlay filename never appears as a
        # plain constant, so the overlay check must see through f-strings.
        tail = node.values[-1] if node.values else None
        if isinstance(tail, ast.Constant) and isinstance(tail.value, str):
            if tail.value.endswith(".env"):
                self.facts.names_profile_overlay_path = True
        self.generic_visit(node)

    def visit_Assign(self, node: ast.Assign) -> None:
        for target in node.targets:
            if isinstance(target, ast.Subscript) and self._is_environ(target.value):
                self.facts.mutates_environ = True
                # A bare subscript write is the OVERRIDE form; setdefault is
                # the only-if-unset form. The distinction is the whole
                # difference between the base .env tier and the profile tier.
                self.facts.overrides_environ = True
        self.generic_visit(node)

    def visit_Subscript(self, node: ast.Subscript) -> None:
        if self._is_environ(node.value) and isinstance(node.slice, ast.Constant):
            key = node.slice.value
            if isinstance(key, str) and key in self.infra_keys:
                self.facts.infra_reads.setdefault(key, node.lineno)
        self.generic_visit(node)

    # -- calls -----------------------------------------------------------
    def _resolve_call(self, func: ast.expr) -> str | None:
        if isinstance(func, ast.Name):
            if func.id in self.bindings:
                return self.bindings[func.id]
            if func.id in self.local_defs:
                return f"{self.dotted}.{func.id}"
            return None
        dotted = _dotted_of(func)
        if dotted is None:
            return None
        head, _, rest = dotted.partition(".")
        if head in self.bindings and rest:
            return f"{self.bindings[head]}.{rest}"
        return None

    def visit_Call(self, node: ast.Call) -> None:
        resolved = self._resolve_call(node.func)
        if resolved:
            self.facts.calls.add(resolved)
            if resolved in RESOLVER_SEAMS:
                self.facts.seam_calls.setdefault(resolved, node.lineno)
            if resolved in REGISTERED_LOADERS:
                self.facts.loader_calls.setdefault(resolved, node.lineno)
        # os.environ.setdefault(...) / .update(...) — mutation, not a read.
        if isinstance(node.func, ast.Attribute):
            if node.func.attr in _ENVIRON_MUTATORS and self._is_environ(
                node.func.value
            ):
                self.facts.mutates_environ = True
            if node.func.attr in ("get", "setdefault") and self._is_environ(
                node.func.value
            ):
                self._record_key_arg(node)
        # A key STRING is an infra READ only when something RESOLVES it: an
        # os.environ lookup (above) or a raising resolver seam such as
        # require_port("KG_GRPC_PORT"). Recording it for ANY call whose first
        # argument merely equals a key name turns pure help text into an infra
        # dependency — measured 2026-08-05, a one-line
        # ``_key_help("REDIS_HOST")`` returning an f-string pulled
        # ``postman.cli`` into scope and failed it for "reads required infra
        # key REDIS_HOST ... depends on systemd EnvironmentFile= injection".
        # The two genuine findings are seam calls, so they are unaffected.
        if resolved in RESOLVER_SEAMS:
            self._record_key_arg(node)
        self.generic_visit(node)

    def _record_key_arg(self, node: ast.Call) -> None:
        if not node.args:
            return
        first = node.args[0]
        if isinstance(first, ast.Constant) and isinstance(first.value, str):
            if first.value in self.infra_keys:
                self.facts.infra_reads.setdefault(first.value, node.lineno)


def _has_main_guard(tree: ast.Module) -> bool:
    for node in tree.body:
        if not isinstance(node, ast.If):
            continue
        test = node.test
        if (
            isinstance(test, ast.Compare)
            and isinstance(test.left, ast.Name)
            and test.left.id == "__name__"
            and len(test.comparators) == 1
            and isinstance(test.comparators[0], ast.Constant)
            and test.comparators[0].value == "__main__"
        ):
            return True
    return False


def _is_dotenv_const(node: ast.AST) -> bool:
    return (
        isinstance(node, ast.Constant)
        and isinstance(node.value, str)
        and (node.value == ".env" or node.value.endswith("/.env"))
    )


def _writes_environ(node: ast.AST, visitor: _ModuleVisitor) -> bool:
    """``os.environ[k] = v`` / ``os.environ.setdefault(...)`` / ``.update(...)``."""
    if isinstance(node, ast.Assign):
        return any(
            isinstance(t, ast.Subscript) and visitor.is_environ(t.value)
            for t in node.targets
        )
    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
        return node.func.attr in _ENVIRON_MUTATORS and visitor.is_environ(
            node.func.value
        )
    return False


def _scope_bodies(tree: ast.AST) -> list[list[ast.stmt]]:
    """Every scope that could hold a loader: each function body, plus the
    module's own top-level statements (walked WITHOUT descending into function
    or class bodies, which are separate scopes)."""
    scopes: list[list[ast.stmt]] = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            scopes.append(node.body)
    if isinstance(tree, ast.Module):
        top: list[ast.stmt] = []
        stack: list[ast.stmt] = list(tree.body)
        while stack:
            stmt = stack.pop()
            if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
                continue
            top.append(stmt)
            for child in ast.iter_child_nodes(stmt):
                if isinstance(child, ast.stmt):
                    stack.append(child)
        scopes.append(top)
    return scopes


def _names_dotenv_and_writes(tree: ast.AST, visitor: _ModuleVisitor) -> bool:
    """True when ONE scope both names a ``.env`` file and writes ``os.environ``.

    Requiring co-occurrence within a single scope is what separates a real
    hand-rolled loader (``def _load_dotenv(): ... os.environ.setdefault(k, v)``)
    from a module that merely happens to do both somewhere — e.g. a pydantic
    ``env_file=".env"`` declaration in a class body alongside an unrelated
    ``os.environ.setdefault("OPENBLAS_NUM_THREADS", "1")`` at module level.
    """
    for body in _scope_bodies(tree):
        names_dotenv = False
        writes = False
        for stmt in body:
            for node in ast.walk(stmt):
                if _is_dotenv_const(node):
                    names_dotenv = True
                elif _writes_environ(node, visitor):
                    writes = True
        if names_dotenv and writes:
            return True
    return False


def analyse_modules(
    modules: dict[str, Path], infra_keys: frozenset[str]
) -> dict[str, ModuleFacts]:
    """Extract :class:`ModuleFacts` for every module. A module that fails to
    parse yields EMPTY facts rather than aborting the run — one broken file
    must not blind the gate to the rest of the tree."""
    out: dict[str, ModuleFacts] = {}
    for dotted, path in modules.items():
        try:
            tree = ast.parse(path.read_text(encoding="utf-8", errors="replace"))
        except SyntaxError:
            out[dotted] = ModuleFacts(dotted=dotted, path=path)
            continue
        visitor = _ModuleVisitor(dotted, infra_keys)
        visitor.visit(tree)
        facts = visitor.facts
        facts.path = path
        facts.is_main_guarded = _has_main_guard(tree)
        facts.dotenv_write_scope = _names_dotenv_and_writes(tree, visitor)
        out[dotted] = facts
    return out


def import_edges(
    facts: dict[str, ModuleFacts], modules: dict[str, Path]
) -> dict[str, set[str]]:
    """Intra-package STATIC import edges only.

    No speculative call-graph resolution: an edge exists because an ``import``
    statement names a module that exists under ``src/``. A dotted name that
    resolves to a symbol inside a module (``pkg.mod.func``) folds onto its
    longest existing module prefix.
    """
    edges: dict[str, set[str]] = {}
    for dotted, mf in facts.items():
        targets: set[str] = set()
        for raw in mf.imports:
            parts = raw.split(".")
            for i in range(len(parts), 0, -1):
                candidate = ".".join(parts[:i])
                if candidate in modules and candidate != dotted:
                    targets.add(candidate)
                    break
        edges[dotted] = targets
    return edges


# --------------------------------------------------------------------------
# entrypoint universe
# --------------------------------------------------------------------------


def entrypoints_from_pyproject(
    repo_root: Path, modules: dict[str, Path]
) -> list[Entrypoint]:
    """E1: ``[project.scripts]`` / ``[project.gui-scripts]`` targets that
    resolve to a module under ``src/``."""
    path = repo_root / "pyproject.toml"
    if not path.exists():
        return []
    try:
        data = tomllib.loads(path.read_text(encoding="utf-8"))
    except tomllib.TOMLDecodeError:
        return []
    out: list[Entrypoint] = []
    for table in ("scripts", "gui-scripts"):
        entries = data.get("project", {}).get(table) or {}
        for name, target in sorted(entries.items()):
            module = str(target).split(":")[0]
            if module in modules:
                out.append(
                    Entrypoint(module, f"pyproject.toml [project.{table}] {name}")
                )
    return out


def execstart_commands(repo_root: Path) -> list[tuple[str, str]]:
    """``(unit_rel_path, command)`` for every ``ExecStart=`` under ``deploy/``.

    Backslash line-continuations are joined FIRST. ``deploy/systemd/*.service``
    wraps every ExecStart across two lines; a line-wise regex would resolve
    zero targets there and report a clean tree — a false negative.
    """
    deploy = repo_root / "deploy"
    if not deploy.is_dir():
        return []
    return [
        (str(path.relative_to(repo_root)), match.group(1).strip())
        for path in sorted(deploy.rglob("*"))
        if path.is_file() and path.suffix in (".service", ".timer", ".socket")
        for match in _EXECSTART_RE.finditer(
            _CONTINUATION_RE.sub(
                " ", path.read_text(encoding="utf-8", errors="replace")
            )
        )
    ]


def entrypoints_from_deploy(
    repo_root: Path, modules: dict[str, Path], src_root: Path
) -> tuple[list[Entrypoint], int]:
    """E2: ``ExecStart=`` targets under ``deploy/``, plus the count resolved
    (reported so a silent parse regression is visible, never inferred)."""
    out: list[Entrypoint] = []
    resolved = 0
    for unit, command in execstart_commands(repo_root):
        tokens = command.replace("=", " ").split()
        module: str | None = None
        for i, token in enumerate(tokens):
            if token == "-m" and i + 1 < len(tokens):
                candidate = tokens[i + 1]
                if candidate in modules:
                    module = candidate
                break
            if token.endswith(".py"):
                candidate_path = Path(token)
                try:
                    rel = candidate_path.resolve().relative_to(src_root.resolve())
                except ValueError, OSError:
                    continue
                parts = list(rel.parts)
                parts[-1] = parts[-1][: -len(".py")]
                candidate = ".".join(parts)
                if candidate in modules:
                    module = candidate
                break
        if module:
            resolved += 1
            out.append(Entrypoint(module, f"{unit} ExecStart"))
    return out, resolved


def entrypoints_from_main_guards(facts: dict[str, ModuleFacts]) -> list[Entrypoint]:
    """E3: ``src/`` modules carrying a top-level ``__main__`` guard."""
    return [
        Entrypoint(dotted, "__main__ guard")
        for dotted, mf in sorted(facts.items())
        if mf.is_main_guarded
    ]


def collect_entrypoints(
    repo_root: Path,
    src_root: Path,
    modules: dict[str, Path],
    facts: dict[str, ModuleFacts],
) -> tuple[dict[str, list[str]], int]:
    """``module -> [origins]`` for the union of E1/E2/E3, plus the count of
    ExecStart targets resolved (printed so a silent parse regression is
    visible)."""
    deploy_eps, resolved = entrypoints_from_deploy(repo_root, modules, src_root)
    universe: dict[str, list[str]] = {}
    for ep in (
        entrypoints_from_pyproject(repo_root, modules)
        + deploy_eps
        + entrypoints_from_main_guards(facts)
    ):
        universe.setdefault(ep.module, []).append(ep.origin)
    return universe, resolved


# --------------------------------------------------------------------------
# closure + rules
# --------------------------------------------------------------------------


def closure_with_witness(
    root: str, edges: dict[str, set[str]]
) -> tuple[dict[str, int], dict[str, str | None]]:
    """BFS over static import edges.

    Returns ``{module: depth}`` (breadth-first, so ``depth`` is the SHORTEST
    import distance from *root*) and a parent map, so every finding can print
    the exact witness path instead of asserting reachability the reader has to
    take on faith.
    """
    depth: dict[str, int] = {root: 0}
    parent: dict[str, str | None] = {root: None}
    queue = deque([root])
    while queue:
        current = queue.popleft()
        for nxt in sorted(edges.get(current, ())):
            if nxt not in depth:
                depth[nxt] = depth[current] + 1
                parent[nxt] = current
                queue.append(nxt)
    return depth, parent


def _witness_path(target: str, parent: dict[str, str | None]) -> str:
    chain = [target]
    cur = parent.get(target)
    while cur is not None:
        chain.append(cur)
        cur = parent.get(cur)
    return " -> ".join(reversed(chain))


def check_rule_a(facts: dict[str, ModuleFacts], repo_root: Path) -> list[Violation]:
    """At most ONE mutating dotenv loader in ``src/``, and it must be
    conforming (profile-aware, override-applying)."""
    loaders = [mf for mf in facts.values() if mf.is_mutating_loader]
    if not loaders:
        return []
    conforming = [mf for mf in loaders if mf.is_conforming_loader]
    violations: list[Violation] = []

    def rel(mf: ModuleFacts) -> str:
        try:
            return str(mf.path.relative_to(repo_root))
        except ValueError:
            return str(mf.path)

    if len(conforming) == 1:
        canonical = conforming[0]
        for mf in sorted(loaders, key=lambda m: m.dotted):
            if mf is canonical:
                continue
            violations.append(
                Violation(
                    "A",
                    rel(mf),
                    f"{mf.dotted} is a second hand-rolled mutating .env loader "
                    f"(names a .env file and writes os.environ); the canonical "
                    f"loader is {canonical.dotted} — import it instead of "
                    f"re-implementing the env contract",
                )
            )
        return violations

    if len(conforming) > 1:
        return [
            Violation(
                "A",
                rel(mf),
                f"{mf.dotted} is one of {len(conforming)} conforming env "
                f"loaders — exactly ONE module under src/ may own the env "
                f"contract",
            )
            for mf in sorted(conforming, key=lambda m: m.dotted)
        ]

    # Zero conforming: every mutating loader is a non-conforming duplicate of
    # a contract that does not exist yet.
    for mf in sorted(loaders, key=lambda m: m.dotted):
        missing = []
        if not mf.names_env_profile:
            missing.append("ORCHESTRATA_ENV_PROFILE")
        if not mf.names_profile_overlay_path:
            missing.append("env/<profile>.env overlay path")
        if not mf.overrides_environ:
            missing.append("override write (os.environ[k] = v)")
        violations.append(
            Violation(
                "A",
                rel(mf),
                f"{mf.dotted} writes os.environ from a .env file but does not "
                f"implement the RUL-019 two-phase contract — missing: "
                f"{', '.join(missing)}. Exactly ONE canonical loader may exist; "
                f"fold this onto it",
            )
        )
    return violations


def _first_trigger(
    depth: dict[str, int],
    facts: dict[str, ModuleFacts],
    max_depth: int,
) -> tuple[str, str, str] | None:
    """``(kind, detail, via_module)`` for the shallowest trigger within
    *max_depth* (``-1`` = unlimited), or ``None``.

    Shallowest first, so the witness path printed for a finding is the
    shortest one — the version a reader can actually check.
    """
    candidates = sorted(depth, key=lambda name: (depth[name], name))
    for name in candidates:
        if max_depth >= 0 and depth[name] > max_depth:
            break
        other = facts.get(name)
        if other is None:
            continue
        for seam, lineno in sorted(other.seam_calls.items()):
            return ("seam", f"{seam} (line {lineno})", name)
        for key, lineno in sorted(other.infra_reads.items()):
            return ("key", f"{key} (line {lineno})", name)
    return None


def check_rule_b(
    universe: dict[str, list[str]],
    facts: dict[str, ModuleFacts],
    edges: dict[str, set[str]],
    repo_root: Path,
    max_depth: int,
) -> tuple[list[Violation], list[dict], list[dict]]:
    """Every in-scope entrypoint's closure must reach a registered loader.

    Returns ``(violations, in_scope, deep_scope_only)``. The third list is the
    region the depth bound discards — entrypoints that WOULD be in scope at
    unlimited depth. It is reported on every run so the bound's false-negative
    surface stays visible instead of becoming an invisible exemption.
    """
    violations: list[Violation] = []
    scoped: list[dict] = []
    deep_only: list[dict] = []
    for module in sorted(universe):
        mf = facts.get(module)
        if mf is None:
            continue
        depth, parent = closure_with_witness(module, edges)

        trigger = _first_trigger(depth, facts, max_depth)
        if trigger is None:
            deep_trigger = _first_trigger(depth, facts, -1)
            if deep_trigger is not None:
                kind, detail, via = deep_trigger
                deep_only.append(
                    {
                        "module": module,
                        "origins": universe[module],
                        "trigger": {"kind": kind, "detail": detail, "via": via},
                        "depth": depth[via],
                        "witness": _witness_path(via, parent),
                    }
                )
            continue

        # Loader search is deliberately UNBOUNDED — see the module docstring.
        loader: tuple[str, str] | None = None
        for name in sorted(depth, key=lambda n: (depth[n], n)):
            other = facts.get(name)
            if other is None or not other.loader_calls:
                continue
            loader = (sorted(other.loader_calls)[0], name)
            break

        kind, detail, via = trigger
        scoped.append(
            {
                "module": module,
                "origins": universe[module],
                "trigger": {"kind": kind, "detail": detail, "via": via},
                "loader": ({"call": loader[0], "via": loader[1]} if loader else None),
            }
        )
        if loader is not None:
            continue

        try:
            artifact = str(facts[module].path.relative_to(repo_root))
        except ValueError:
            artifact = str(facts[module].path)
        subject = (
            f"calls required infra seam {detail}"
            if kind == "seam"
            else f"reads required infra key {detail}"
        )
        origins = "; ".join(universe[module])
        violations.append(
            Violation(
                "B",
                artifact,
                f"entrypoint {module} ({origins}) {subject}, but its import "
                f"closure never reaches the env-contract loader — it depends on "
                f"systemd EnvironmentFile= injection and cannot start from a "
                f"bare shell",
                witness=_witness_path(via, parent),
            )
        )
    return violations, scoped, deep_only


# --------------------------------------------------------------------------
# driver
# --------------------------------------------------------------------------


def run_check(repo_root: Path, max_depth: int = 1) -> tuple[int, dict]:
    """Injection seam: tests pass a throwaway tmp_path repo root — this never
    touches the live checkout implicitly."""
    src_root = repo_root / "src"
    if not src_root.is_dir():
        raise AnalysisError(f"no src/ directory under {repo_root}")
    modules = module_map(src_root)
    if not modules:
        raise AnalysisError(f"no python modules under {src_root}")
    infra_keys = load_infra_keys(src_root)
    facts = analyse_modules(modules, infra_keys)
    edges = import_edges(facts, modules)
    universe, execstart_resolved = collect_entrypoints(
        repo_root, src_root, modules, facts
    )

    violations = check_rule_a(facts, repo_root)
    rule_b, scoped, deep_only = check_rule_b(
        universe, facts, edges, repo_root, max_depth
    )
    violations.extend(rule_b)

    report = {
        "status": "pass" if not violations else "fail",
        "rule_id": RULE_ID,
        "repo_root": str(repo_root),
        "max_depth": max_depth,
        "infra_keys": sorted(infra_keys),
        "modules": len(modules),
        "entrypoints": len(universe),
        "execstart_targets_resolved": execstart_resolved,
        "in_scope": scoped,
        "deep_scope_only": deep_only,
        "mutating_loaders": sorted(
            mf.dotted for mf in facts.values() if mf.is_mutating_loader
        ),
        "violations": [asdict(v) for v in violations],
    }
    return (0 if not violations else 1), report


def _print_text_report(report: dict) -> None:
    print(
        f"[env-contract] {report['modules']} module(s), "
        f"{report['entrypoints']} entrypoint(s), "
        f"{len(report['in_scope'])} in scope at depth "
        f"{report['max_depth']}, "
        f"{report['execstart_targets_resolved']} ExecStart target(s) resolved",
        flush=True,
    )
    if report["deep_scope_only"]:
        print(
            f"[env-contract] note: {len(report['deep_scope_only'])} further "
            f"entrypoint(s) would be in scope at unlimited depth — not graded "
            f"here (see --max-depth): "
            + ", ".join(d["module"] for d in report["deep_scope_only"]),
            flush=True,
        )
    if report["status"] == "pass":
        print(f"[env-contract] OK — {RULE_ID} satisfied", flush=True)
        return
    print(
        f"[env-contract] FAIL — {len(report['violations'])} violation(s):",
        flush=True,
    )
    for v in report["violations"]:
        print(
            f"  {RULE_ID} [rule {v['rule']}] {v['artifact']}: {v['message']}",
            flush=True,
        )
        if v["witness"]:
            print(f"      witness: {v['witness']}", flush=True)


def main(argv: list[str] | None = None) -> int:
    """CLI entry point. 0 clean, 1 violations, 2 unanalysable tree."""
    parser = argparse.ArgumentParser(description="RUL-019 env-contract gate")
    parser.add_argument(
        "--repo-root",
        default=None,
        help="Tree to analyse (default: this checkout's root)",
    )
    parser.add_argument(
        "--max-depth",
        type=int,
        default=1,
        help=(
            "Import-closure depth searched for an infra trigger (default 1: "
            "the entrypoint and what it imports directly; -1 = unlimited). "
            "See the module docstring for the measurement behind the default."
        ),
    )
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    args = parser.parse_args(argv)

    root = Path(args.repo_root).expanduser().resolve() if args.repo_root else REPO_ROOT
    try:
        exit_code, report = run_check(root, max_depth=args.max_depth)
    except AnalysisError as exc:
        print(f"[env-contract] ERROR — {exc}", file=sys.stderr, flush=True)
        return 2

    if args.json:
        print(json.dumps(report, indent=2, default=str))
    else:
        _print_text_report(report)
    return exit_code


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