scripts.check_incus_not_docker

Reference › scripts › check_incus_not_docker

The Incus-not-Docker gate, the mechanism for ruling RUL-004: containers run on Incus (LXD), not Docker. tests/scripts/test_check_incus_not_docker.py imports it.

Invocation: main takes --repo-root (default the repository root), --json and --ruling, which prints the ratified ruling text and exits 0. Otherwise run_check returns an exit code and a report, printed as text or JSON.

The scan covers the whole repository file by file (iter_scanned_files), restricted to the content scope. Allowed roots (AllowedRoot) and quarantined entries (QuarantineEntry) are the declared exceptions, and a few detection-predicate files are exempt because they must name Docker to detect it. Shell files are scanned line by line after comments and noise are stripped (scan_shell_file, find_shell_invocations); only invocations in an execution position count, so an echoed or quoted mention is not flagged. Python files are parsed: _ImportResolver resolves import aliases, and scan_python_file flags subprocess and shell calls that run Docker, including through shell strings. Config files (scan_config_file) and newly introduced files (scan_file_introduction) get their own rules. Every hit is recorded in a ScanState with rule, path, line and message, and becomes a Violation.

At a glance

Class diagram

Dependency diagram

No internal imports and no importers.

API

Constants

Classes

AllowedRoot (dataclass)

Decorators: dataclass(frozen=True) · Line 153

One allowlist entry — and the repo document that ratifies it.

QuarantineEntry (dataclass)

Decorators: dataclass(frozen=True) · Line 229

A live in-tree instance of the forbidden behaviour that is NOT excused.

Violation (dataclass)

Decorators: dataclass · Line 433

ScanState (dataclass)

Decorators: dataclass · Line 455

_ImportResolver

Bases: ast.NodeVisitor · Line 696

Map local names to canonical dotted names (sp -> subprocess).

Functions

Source

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

#!/usr/bin/env python3
"""Incus-not-Docker gate — the mechanism for ruling **RUL-004** ("Incus (LXD),
not Docker, for containers"), which was `none (judgment)` / prose-only until
this script existed.

See ``~/workspace/knowledge-wiki/rulings/RUL-004-incus-not-docker.md`` for the
ruling text, and ``RULING_TEXT`` below for the verbatim ratified design this
gate implements (also emitted by ``--ruling`` and by ``--json``).

**Shape: a SCOPE-based introduction ratchet, not a repo-wide grep.** The repo
deliberately carries six Docker-bearing path trees (topology reference, dev
stand-in backend, GitLab appliance nesting, CI substrate, frozen handover
artifacts, documentation). A naive scan is red at birth and gets reverted
within a day. This gate instead defines the LIVE SURFACE that determines how
orchestrata actually runs, and fails only when a Docker RUNTIME DEPENDENCY is
introduced there.

Four rules, all restricted to the enforced scope:

    A  shell / systemd / provisioning invocation (``*.sh``, ``*.bash``,
       ``*.service``, ``*.timer``, ``*.socket``) — comments and here-doc
       bodies stripped FIRST, then docker in an execution position,
       ``ExecStart=``-family with docker as argv[0], or a docker dependency
       probe (``command -v docker``, ``apt-get install docker.io``, ...).
       ``incus exec c -- docker compose up`` is NOT an escape: an in-scope
       file shelling docker through incus still FAILS.
    B  Python, AST-POSITION based (never a regex over source text) — a
       ``docker`` string in a subprocess-EXECUTING position, ``import
       docker`` (the PyPI Docker SDK), or a docker detection predicate used
       to branch orchestrata's own provisioning.
    C  config / env — ``DOCKER_HOST=``, ``/var/run/docker.sock``, a
       ``docker.io/`` image reference, or a ``docker`` distribution
       requirement, in an in-scope env file, ``deploy/**``,
       ``pyproject.toml``, ``.env.example``, or Terraform under ``infra/``.
    D  file introduction — a new ``Dockerfile*`` / ``docker-compose*.y*ml`` /
       ``compose.y*ml`` anywhere outside the allowlisted ``docker/`` and
       ``infra/docker/``.

Rule B is AST-position based precisely because raw text is not good enough:
measured over the enforced scope of this checkout, a naive regex returns 3
hits, 3 of them false positives (a comment, a docstring, and a prose string
literal in ``scripts/gen_level0.py``) and 0 true positives. The designed
rules return 0. That 3-to-0 delta is the whole value of the AST design, and
those three sites are regression-locked in the gate's own test module.

The allowlist protects FILES, never INVOCATIONS FROM LIVE CODE: if
``deploy/install.sh`` or ``scripts/start_test_services.py`` calls ``docker
compose -f docker/docker-compose.yml up``, the CALLER is in scope and the
gate fires. That seam is what stops the allowlist from becoming an evasion
route.

QUARANTINE is reported, never excused: ``rtm/scripts/*_test_infrastructure.sh``
are a live in-tree instance of the exact behaviour RUL-004 forbids (``docker
run -d`` for redis-test / neo4j-test / rabbitmq-test). They are dead code, so
they are not a blocking failure — but silently allowlisting them is the
false-confidence failure this gate exists to prevent, so every run prints
them as a non-blocking QUARANTINE line.

Known, deliberate gaps (stated rather than hidden):
  * a Docker deployment path added ENTIRELY inside an allowlisted tree with
    no in-scope caller is not detected — bounded by the caller seam above;
  * dot-directories (``.github/``, ``.harness/``) are pruned from the walk;
    they are not part of orchestrata's deployment path;
  * ``src/orchestrata_tools/`` is EXEMPT from rule B's detection-predicate
    clause (it analyses foreign projects for a living, which may legitimately
    be Docker projects) — it stays fully covered for subprocess-argv and for
    ``import docker``.

Exit 0 clean. Exit 1 on any violation, one ``RUL-004: <path>:<line> — ...``
line per finding. Quarantine lines never affect the exit code.

Usage:
    python scripts/check_incus_not_docker.py [--repo-root PATH] [--json]
                                             [--ruling]
"""

from __future__ import annotations

import argparse
import ast
import json
import os
import re
import sys
from dataclasses import asdict, dataclass, field
from pathlib import Path

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

RULE_ID = "RUL-004"

#: The ratified design this gate implements, verbatim. Emitted by ``--ruling``
#: and by ``--json`` so the mechanism always carries the reasoning it enforces.
RULING_TEXT = """RUL-004 — Incus (LXD), not Docker, for containers. Currently `none (judgment)`; prose-only. It IS mechanizable as a scoped introduction-ratchet gate, and the enforcement bullet should become:

  - gate scripts/check_incus_not_docker.py RUL-004
  - banned-deps docker

The gate is a SCOPE-based ratchet, not a repo-wide grep. The repo deliberately carries six Docker-bearing path trees (measured, sourced below); a naive scan is red at birth and would be reverted within a day. The design instead defines the LIVE SURFACE that determines how orchestrata actually runs, and fails only when a Docker RUNTIME DEPENDENCY is introduced there.

ENFORCED SCOPE (in scope — a docker runtime dep here is a FAIL):
  src/  scripts/  tests/  deploy/  env/  infra/ (minus infra/docker/)  conftest.py  pyproject.toml  .env.example

OUT OF SCOPE (allowlist, each with its source of authority — encoded as a literal ALLOWED_ROOTS table in the script, each entry carrying its justification string, printed in --json output so the excuse is auditable):
  - `docker/` — service-topology REFERENCE only. Source: plans/implementation-status-and-plan.md:31 "Containers = Incus, not Docker. docker-compose.yml is the topology reference only". Already gated separately by scripts/validate_ci.py Gate 8 (env vars, no hardcoded IPs).
  - `infra/docker/infra-compose.yml` + `claude-env/` — the ratified DEVELOPMENT STAND-IN backend for a machine with no Incus. Source: docs/environment-replication.md:47-55 ("--backend incus|docker", "Development stand-in. No Raft HA") and claude-env/README.md:10-12.
  - `incus/gitlab/*.sh`, `incus/launch.sh` — Docker nested INSIDE an Incus container to run the third-party GitLab CE appliance. Incus remains the container runtime; docker is appliance packaging. Same nesting shown canonically in docker/REFERENCE.md (`incus exec orchestrata-docker -- docker compose ...`).
  - `.gitlab-ci.yml` — builds the CI RUNNER image (docker:dind, `docker build -f docker/Dockerfile.ci`). GitLab shared runners are docker-executor; this is CI substrate, not orchestrata's deployment path.
  - `rtm/` — frozen handover artifact tree.
  - All `*.md`, `docs/`, `plans/`, `specs/` anywhere — documentation.

The allowlist protects FILES, never INVOCATIONS FROM LIVE CODE: if `deploy/install.sh` or `scripts/start_test_services.py` calls `docker compose -f docker/docker-compose.yml up`, the CALLER is in scope and the gate fires. That seam is what stops the allowlist from becoming an evasion route.

QUARANTINE (reported, not excused): `rtm/scripts/setup_test_infrastructure.sh` runs `docker run -d` for redis-test / neo4j-test / rabbitmq-test and `docker exec ... cypher-shell` (lines 50-51, 72-137, 551-617), and `rtm/scripts/check_test_infrastructure.sh:28` the same. That is EXACTLY the "test-isolation targeting Docker" RUL-004 names. It is dead — nothing invokes it; the only reference is prose in plans/tools_useful_to_plan.md:540-567. Silently allowlisting it is the false-confidence failure this task exists to prevent, so the script carries it as an explicit QUARANTINE entry (path + reason "frozen handover artifact, invoked by no runner" + the ruling clause it violates) printed on every run as a non-blocking QUARANTINE line. Visible, not invisible. Recommendation to surface separately: delete rtm/scripts/{setup,check}_test_infrastructure.sh — the live equivalent is scripts/start_test_services.py against the Incus test fleet."""


# --------------------------------------------------------------------------
# Scope
# --------------------------------------------------------------------------

#: Path roots whose content determines how orchestrata actually runs.
ENFORCED_ROOTS: tuple[str, ...] = (
    "src",
    "scripts",
    "tests",
    "deploy",
    "env",
    "infra",
)

#: Individual in-scope files that live at the repository root.
ENFORCED_FILES: tuple[str, ...] = (
    "conftest.py",
    "pyproject.toml",
    ".env.example",
)

#: Directory names never walked: VCS/build/venv/cache/vendor noise, plus every
#: dot-directory (``.git``, ``.venv``, ``.harness`` worktree copies, ...).
PRUNED_DIR_NAMES: frozenset[str] = frozenset(
    {
        "__pycache__",
        "node_modules",
        "vendor",
        "dist-janus",
        "htmlcov",
        "site-packages",
    }
)


@dataclass(frozen=True)
class AllowedRoot:
    """One allowlist entry — and the repo document that ratifies it.

    An allowlist entry with no citation is how a gate quietly becomes a
    no-op, so ``justification`` and ``source`` are required fields and are
    printed by ``--json``.
    """

    patterns: tuple[str, ...]
    justification: str
    source: str


ALLOWED_ROOTS: tuple[AllowedRoot, ...] = (
    AllowedRoot(
        patterns=("docker/",),
        justification=(
            "service-topology REFERENCE only; not a deployment path. Already "
            "gated separately by scripts/validate_ci.py Gate 8 (env vars, no "
            "hardcoded IPs)."
        ),
        source=(
            "plans/implementation-status-and-plan.md:31 — 'Containers = Incus, "
            "not Docker. docker-compose.yml is the topology reference only'"
        ),
    ),
    AllowedRoot(
        patterns=("infra/docker/", "claude-env/"),
        justification=(
            "ratified DEVELOPMENT STAND-IN backend for a machine with no "
            "Incus; not the deployment path (no Raft HA)."
        ),
        source=(
            "docs/environment-replication.md:47-55 ('--backend incus|docker', "
            "'Development stand-in. No Raft HA') and claude-env/README.md:10-12"
        ),
    ),
    AllowedRoot(
        patterns=("incus/",),
        justification=(
            "Docker nested INSIDE an Incus container to run the third-party "
            "GitLab CE appliance. Incus remains the container runtime; docker "
            "is appliance packaging."
        ),
        source=(
            "docker/REFERENCE.md — canonical nesting "
            "'incus exec orchestrata-docker -- docker compose ...'"
        ),
    ),
    AllowedRoot(
        patterns=(".gitlab-ci.yml",),
        justification=(
            "builds the CI RUNNER image (docker:dind, "
            "'docker build -f docker/Dockerfile.ci'). GitLab shared runners "
            "are docker-executor; CI substrate, not orchestrata's deployment "
            "path."
        ),
        source=".gitlab-ci.yml build stage",
    ),
    AllowedRoot(
        patterns=("rtm/",),
        justification=(
            "frozen handover artifact tree — see the QUARANTINE table, which "
            "reports (rather than excuses) its Docker test-isolation scripts."
        ),
        source="rtm/ handover snapshot",
    ),
    AllowedRoot(
        patterns=("docs/", "plans/", "specs/", "*.md", "*.markdown", "*.rst"),
        justification="documentation — prose about Docker is not a runtime dep.",
        source="RUL-004 design, out-of-scope table",
    ),
)


@dataclass(frozen=True)
class QuarantineEntry:
    """A live in-tree instance of the forbidden behaviour that is NOT excused.

    Reported on every run as a non-blocking line so it stays visible instead
    of disappearing into the allowlist.
    """

    path: str
    reason: str
    clause: str
    recommendation: str


QUARANTINE: tuple[QuarantineEntry, ...] = (
    QuarantineEntry(
        path="rtm/scripts/setup_test_infrastructure.sh",
        reason="frozen handover artifact, invoked by no runner",
        clause=(
            "'actual deployment and test-isolation should target Incus' — this "
            "script runs 'docker run -d' for redis-test / neo4j-test / "
            "rabbitmq-test and 'docker exec ... cypher-shell'"
        ),
        recommendation=(
            "delete it; the live equivalent is scripts/start_test_services.py "
            "against the Incus test fleet"
        ),
    ),
    QuarantineEntry(
        path="rtm/scripts/check_test_infrastructure.sh",
        reason="frozen handover artifact, invoked by no runner",
        clause=(
            "'actual deployment and test-isolation should target Incus' — line "
            "28 probes the test graph with 'docker exec neo4j-test cypher-shell'"
        ),
        recommendation=(
            "delete it; the live equivalent is scripts/start_test_services.py "
            "against the Incus test fleet"
        ),
    ),
)

#: Rule B detection-predicate exemption: this package's CONTRACT is describing
#: arbitrary foreign projects, which may legitimately be Docker projects. It
#: stays fully covered for subprocess-argv invocation and for ``import docker``.
DETECTION_PREDICATE_EXEMPT_PREFIXES: tuple[tuple[str, str], ...] = (
    (
        "src/orchestrata_tools/",
        "generic toolkit that DESCRIBES foreign projects (env_collector probes "
        "for a Dockerfile by design); exempt from the detection-predicate "
        "clause only",
    ),
)


# --------------------------------------------------------------------------
# Patterns
# --------------------------------------------------------------------------

_DOCKER_SUBCOMMANDS = (
    "compose",
    "run",
    "exec",
    "ps",
    "build",
    "push",
    "pull",
    "up",
    "down",
    "start",
    "stop",
    "rm",
    "kill",
    "login",
    "logs",
    "restart",
    "network",
    "volume",
    "image",
    "images",
    "info",
    "inspect",
    "cp",
    "tag",
    "save",
    "load",
)

#: ``docker`` / ``docker-compose`` (optionally as an absolute path) followed by
#: a real subcommand. The lookbehind stops ``orchestrata-docker`` and
#: ``docker/Dockerfile.ci`` from matching.
_DOCKER_CMD_RE = re.compile(
    r"(?<![\w.-])(?P<path>/[\w./-]*/)?(?P<bin>docker(?:-compose)?)\s+"
    r"(?:compose\s+)?(?P<sub>" + "|".join(_DOCKER_SUBCOMMANDS) + r")(?![\w-])"
)

#: Text that may legitimately precede a command word. ``--`` is included on
#: purpose: ``incus exec c -- docker compose up`` must FAIL, not pass.
#:
#: A QUOTE is deliberately NOT an execution position here, and neither is
#: ``=``. Treating them as one made every search pattern and string comparison
#: read as an invocation — measured 2026-08-05, ``grep -q "docker run"``,
#: ``grep -rn 'docker run' src/``, ``[ "$CMD" = "docker run" ]``,
#: ``jq '... == "docker run"'`` and the bare assignment
#: ``LEGACY_CMD="docker run redis"`` all reported "shell invocation of
#: docker", and those idioms already appear in 11 of the 25 in-scope shell
#: files. A quote only opens a COMMAND when a shell was asked to interpret the
#: string, which _SHELL_DASH_C_RE below recognises.
_EXEC_PREFIX_RE = re.compile(
    r"(?:^|[;&|(){}`]|\$\(|&&|\|\||--|"
    r"\b(?:sudo|exec|env|nohup|time|then|do|else|eval|xargs)\b)\s*$"
)

#: ``bash -c "docker run ..."`` / ``sudo sh -lc 'docker compose up'`` — the one
#: shape where an opening quote really does begin a command.
_SHELL_DASH_C_RE = re.compile(
    r"\b(?:sh|bash|zsh|dash|ksh|env)\b[^\"']*\s-[a-z]*c[a-z]*\s*[\"']\s*$"
)

_COMMAND_SEPARATOR_RE = re.compile(r"[;&|]|&&|\|\|")

_HEREDOC_RE = re.compile(r"<<-?\s*(?P<q>['\"]?)(?P<term>[A-Za-z_][\w-]*)(?P=q)")

_SYSTEMD_EXEC_RE = re.compile(
    r"^\s*(?P<key>Exec(?:Start|StartPre|StartPost|Stop|StopPost|Reload|Condition))"
    r"\s*=\s*(?P<value>.*)$"
)

_DOCKER_PROBE_RES: tuple[tuple[re.Pattern[str], str], ...] = (
    (
        re.compile(r"\bcommand\s+-v\s+docker(?:-compose)?\b"),
        "probes for the docker binary (command -v docker)",
    ),
    (
        re.compile(r"\bwhich\s+docker(?:-compose)?\b"),
        "probes for the docker binary (which docker)",
    ),
    (
        re.compile(
            r"\b(?:apt|apt-get|dnf|yum|apk|zypper)\s+(?:-\S+\s+)*"
            r"(?:install|add)\b[^;&|]*\b(?:docker|docker\.io|docker-ce|"
            r"docker-compose)\b"
        ),
        "installs the docker package as a dependency",
    ),
    (
        re.compile(r"\bpacman\s+-S\S*\b[^;&|]*\bdocker\b"),
        "installs the docker package as a dependency",
    ),
)

_DOCKER_BINARIES = frozenset({"docker", "docker-compose"})

_DOCKER_FILENAME_RE = re.compile(
    r"^(?:Dockerfile(?:[.\-].*)?|docker-compose(?:[.\-].*)?\.ya?ml|compose\.ya?ml)$",
    re.IGNORECASE,
)

_DOCKER_FILENAME_CONSTANTS = frozenset(
    {
        "Dockerfile",
        "dockerfile",
        "docker-compose.yml",
        "docker-compose.yaml",
        "compose.yml",
        "compose.yaml",
    }
)

_CONFIG_MARKER_RES: tuple[tuple[re.Pattern[str], str], ...] = (
    (
        re.compile(r"\bDOCKER_HOST\s*="),
        "declares DOCKER_HOST — a Docker daemon endpoint",
    ),
    (
        re.compile(r"(?:unix://)?/var/run/docker\.sock"),
        "references the Docker daemon socket",
    ),
    (
        re.compile(r"\bdocker\.io/[\w./-]+"),
        "pins a docker.io/ image reference",
    ),
)

#: A ``docker`` distribution requirement in pyproject.toml (also blocked at
#: write time by .claude/hooks/no-banned-deps.sh via banned-libraries.txt).
_PYPROJECT_DOCKER_DEP_RE = re.compile(
    r"^\s*[\"']?(?P<name>docker|docker-compose|docker-py)\s*"
    r"(?:[<>=!~\[][^\"']*)?[\"']?\s*,?\s*$",
    re.IGNORECASE,
)

_SHELL_SUFFIXES = frozenset({".sh", ".bash", ".service", ".timer", ".socket"})

_ECHO_COMMANDS = frozenset({"echo", "printf", ":", "#"})

_INFRA_TERRAFORM_SUFFIXES = frozenset({".tf", ".tfvars", ".tftpl"})


# --------------------------------------------------------------------------
# Findings
# --------------------------------------------------------------------------


@dataclass
class Violation:
    rule: str
    path: str
    line: int
    message: str

    def render(self) -> str:
        return f"{RULE_ID}: {self.path}:{self.line} — {self.message}"


#: Substrings without which no rule A/B/C pattern can match. ``compose.y`` is
#: listed separately because ``compose.yml`` / ``compose.yaml`` (the
#: unprefixed Compose v2 filenames) do not contain the word "docker".
_NEEDLES: tuple[str, ...] = ("docker", "compose.y")


def _has_docker_needle(text: str) -> bool:
    lowered = text.lower()
    return any(needle in lowered for needle in _NEEDLES)


@dataclass
class ScanState:
    violations: list[Violation] = field(default_factory=list)
    unparsed_python: list[str] = field(default_factory=list)
    scanned: int = 0
    graded: int = 0

    def add(self, rule: str, path: str, line: int, message: str) -> None:
        for existing in self.violations:
            if existing.path == path and existing.line == line:
                return
        self.violations.append(Violation(rule, path, line, message))


# --------------------------------------------------------------------------
# Scope helpers
# --------------------------------------------------------------------------


def _matches_pattern(rel: str, pattern: str) -> bool:
    if pattern.endswith("/"):
        return rel == pattern[:-1] or rel.startswith(pattern)
    if pattern.startswith("*."):
        return rel.endswith(pattern[1:])
    return rel == pattern


def allowlist_entry_for(rel: str) -> AllowedRoot | None:
    """Return the allowlist entry covering ``rel``, or ``None``."""
    for entry in ALLOWED_ROOTS:
        for pattern in entry.patterns:
            if _matches_pattern(rel, pattern):
                return entry
    return None


def _in_content_scope(rel: str) -> bool:
    """True when the file's CONTENT is graded (rules A, B, C)."""
    if allowlist_entry_for(rel) is not None:
        return False
    if rel in ENFORCED_FILES:
        return True
    return rel.split("/", 1)[0] in ENFORCED_ROOTS


def _detection_predicate_exemption(rel: str) -> str | None:
    for prefix, reason in DETECTION_PREDICATE_EXEMPT_PREFIXES:
        if rel.startswith(prefix):
            return reason
    return None


def iter_scanned_files(repo_root: Path) -> list[tuple[Path, str]]:
    """Enforced roots (recursive, pruned) plus repo-root top-level files.

    Top-level files are included so that a newly introduced ``./Dockerfile``
    or ``./docker-compose.yml`` is caught by rule D even though only
    ``conftest.py`` / ``pyproject.toml`` / ``.env.example`` are graded for
    content there.
    """
    found: list[tuple[Path, str]] = []

    for entry in sorted(repo_root.iterdir()) if repo_root.is_dir() else []:
        if entry.is_file():
            found.append((entry, entry.name))

    for root_name in ENFORCED_ROOTS:
        root_dir = repo_root / root_name
        if not root_dir.is_dir():
            continue
        for dirpath, dirnames, filenames in os.walk(root_dir):
            dirnames[:] = sorted(
                d
                for d in dirnames
                if not d.startswith(".") and d not in PRUNED_DIR_NAMES
            )
            for filename in sorted(filenames):
                abs_path = Path(dirpath) / filename
                rel = abs_path.relative_to(repo_root).as_posix()
                found.append((abs_path, rel))

    return found


def _read_text(path: Path) -> str | None:
    try:
        return path.read_text(encoding="utf-8")
    except OSError, UnicodeDecodeError:
        return None


# --------------------------------------------------------------------------
# Rule A — shell / systemd / provisioning invocation
# --------------------------------------------------------------------------


def strip_shell_comment(line: str) -> str:
    """Drop an unquoted ``#`` comment, preserving the code before it."""
    quote: str | None = None
    for index, char in enumerate(line):
        if quote is not None:
            if char == quote:
                quote = None
            continue
        if char in ("'", '"'):
            quote = char
            continue
        if char == "#" and (index == 0 or line[index - 1].isspace()):
            return line[:index]
    return line


def strip_shell_noise(text: str) -> list[str]:
    """Return the lines of ``text`` with comments and here-doc BODIES blanked.

    Line numbering is preserved (blanked lines become empty strings) so a
    finding still cites the real line number.
    """
    out: list[str] = []
    terminator: str | None = None
    for line in text.splitlines():
        if terminator is not None:
            if line.strip() == terminator:
                terminator = None
            out.append("")
            continue
        stripped = strip_shell_comment(line)
        match = _HEREDOC_RE.search(stripped)
        if match:
            terminator = match.group("term")
        out.append(stripped)
    return out


def _is_execution_position(prefix: str) -> bool:
    if _EXEC_PREFIX_RE.search(prefix):
        return True
    # An opening quote counts only when a shell was handed the string:
    # ``bash -c "docker run"``. Everything else quoting the words is a search
    # pattern, a comparison, or an assignment — not an invocation.
    return bool(_SHELL_DASH_C_RE.search(prefix))


def _is_echoed(prefix: str) -> bool:
    """True when the docker text is an argument of a leading echo/printf.

    Only applies when no command separator intervenes — ``echo hi && docker
    run x`` is still an invocation.
    """
    if _COMMAND_SEPARATOR_RE.search(prefix):
        return False
    tokens = prefix.split()
    return bool(tokens) and tokens[0] in _ECHO_COMMANDS


def find_shell_invocations(line: str) -> list[tuple[str, str]]:
    """Return ``(binary, subcommand)`` pairs invoked on this (stripped) line."""
    hits: list[tuple[str, str]] = []
    for match in _DOCKER_CMD_RE.finditer(line):
        prefix = line[: match.start()]
        if _is_echoed(prefix):
            continue
        if not _is_execution_position(prefix):
            continue
        binary = (match.group("path") or "") + match.group("bin")
        hits.append((binary, match.group("sub")))
    return hits


def _systemd_argv0(value: str) -> str | None:
    cleaned = value.lstrip("-+@!:").strip()
    if not cleaned:
        return None
    return cleaned.split()[0]


def scan_shell_file(rel: str, text: str, state: ScanState) -> None:
    for lineno, line in enumerate(strip_shell_noise(text), start=1):
        if not line.strip():
            continue

        systemd = _SYSTEMD_EXEC_RE.match(line)
        if systemd:
            argv0 = _systemd_argv0(systemd.group("value"))
            if argv0 and Path(argv0).name in _DOCKER_BINARIES:
                state.add(
                    "A",
                    rel,
                    lineno,
                    f"systemd {systemd.group('key')} invokes docker "
                    f"({argv0}); Incus is the container runtime "
                    f"(see INFRASTRUCTURE.md, deploy/orchestrata-test/)",
                )
                continue

        for binary, subcommand in find_shell_invocations(line):
            state.add(
                "A",
                rel,
                lineno,
                f"shell invocation of {binary} ({subcommand}); orchestrata "
                f"runs on Incus (LXD) — see INFRASTRUCTURE.md and "
                f"scripts/start_test_services.py",
            )

        for pattern, description in _DOCKER_PROBE_RES:
            if pattern.search(line):
                state.add("A", rel, lineno, f"{description}; Incus is the runtime")
                break


# --------------------------------------------------------------------------
# Rule B — Python, AST position based
# --------------------------------------------------------------------------

_EXEC_FUNCS = frozenset(
    {
        "subprocess.run",
        "subprocess.Popen",
        "subprocess.call",
        "subprocess.check_call",
        "subprocess.check_output",
        "os.system",
        "os.popen",
        "os.execv",
        "os.execvp",
        "os.execvpe",
        "asyncio.create_subprocess_exec",
        "asyncio.create_subprocess_shell",
    }
)

_VARIADIC_EXEC_FUNCS = frozenset(
    {
        "asyncio.create_subprocess_exec",
        "os.execvp",
        "os.execv",
        "os.execvpe",
    }
)


class _ImportResolver(ast.NodeVisitor):
    """Map local names to canonical dotted names (``sp`` -> ``subprocess``)."""

    def __init__(self) -> None:
        self.aliases: dict[str, str] = {}

    def visit_Import(self, node: ast.Import) -> None:
        for alias in node.names:
            local = alias.asname or alias.name.split(".")[0]
            canonical = alias.name if alias.asname else alias.name.split(".")[0]
            self.aliases[local] = canonical
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
        module = node.module or ""
        for alias in node.names:
            local = alias.asname or alias.name
            self.aliases[local] = f"{module}.{alias.name}" if module else alias.name
        self.generic_visit(node)


def _dotted_name(node: ast.expr) -> str | None:
    parts: list[str] = []
    current: ast.expr = node
    while isinstance(current, ast.Attribute):
        parts.append(current.attr)
        current = current.value
    if not isinstance(current, ast.Name):
        return None
    parts.append(current.id)
    return ".".join(reversed(parts))


def _canonical_call_name(node: ast.Call, aliases: dict[str, str]) -> str | None:
    dotted = _dotted_name(node.func)
    if dotted is None:
        return None
    head, _, tail = dotted.partition(".")
    resolved = aliases.get(head)
    if resolved is None:
        return dotted
    return f"{resolved}.{tail}" if tail else resolved


def _const_str(node: ast.expr) -> str | None:
    if isinstance(node, ast.Constant) and isinstance(node.value, str):
        return node.value
    return None


def _argv_strings(node: ast.Call, canonical: str) -> tuple[list[str | None], bool]:
    """Return (argv-ish strings, resolved) for a subprocess-executing call."""
    if canonical in _VARIADIC_EXEC_FUNCS:
        return [_const_str(arg) for arg in node.args], bool(node.args)

    target: ast.expr | None = node.args[0] if node.args else None
    if target is None:
        for keyword in node.keywords:
            if keyword.arg in ("args", "cmd", "command"):
                target = keyword.value
                break
    if target is None:
        return [], False
    if isinstance(target, (ast.List, ast.Tuple)):
        return [_const_str(el) for el in target.elts], True
    literal = _const_str(target)
    if literal is not None:
        return [literal], True
    return [], False


def _shell_string_hit(text: str) -> tuple[str, str] | None:
    for line in strip_shell_noise(text):
        hits = find_shell_invocations(line)
        if hits:
            return hits[0]
    return None


#: Binaries that INTERPRET their argument as a shell command line.
_SHELL_BINARIES = frozenset({"sh", "bash", "zsh", "dash", "ksh", "fish"})

#: Exec functions whose command argument is always shell-interpreted.
_SHELL_EXEC_FUNCS = frozenset(
    {"os.system", "os.popen", "asyncio.create_subprocess_shell"}
)


def _is_shell_exec(canonical: str, node: ast.Call, argv: list[str | None]) -> bool:
    """True when this call hands its string(s) to a shell for interpretation.

    Three ways that happens: the function is inherently shell-based
    (``os.system``), ``shell=True`` is passed, or argv[0] is a shell binary
    (``["bash", "-lc", "docker run ..."]``). Anything else receives its
    arguments verbatim, so a docker-shaped string in there is data — a search
    pattern, a message, a config value — not a command.
    """
    if canonical in _SHELL_EXEC_FUNCS:
        return True
    for keyword in node.keywords:
        if keyword.arg == "shell":
            value = keyword.value
            if isinstance(value, ast.Constant) and value.value is True:
                return True
    argv0 = argv[0] if argv else None
    if argv0 and Path(argv0).name in _SHELL_BINARIES:
        return True
    return False


def scan_python_file(rel: str, text: str, state: ScanState) -> None:
    try:
        tree = ast.parse(text, filename=rel)
    except SyntaxError:
        state.unparsed_python.append(rel)
        return

    resolver = _ImportResolver()
    resolver.visit(tree)
    aliases = resolver.aliases
    predicate_exemption = _detection_predicate_exemption(rel)

    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                root = alias.name.split(".")[0]
                if root == "docker":
                    state.add(
                        "B",
                        rel,
                        node.lineno,
                        f"imports the PyPI Docker SDK ({alias.name}); "
                        f"orchestrata drives Incus, not Docker",
                    )
            continue

        if isinstance(node, ast.ImportFrom):
            module = node.module or ""
            if module == "docker" or module.startswith("docker."):
                state.add(
                    "B",
                    rel,
                    node.lineno,
                    f"imports the PyPI Docker SDK (from {module}); "
                    f"orchestrata drives Incus, not Docker",
                )
            continue

        if not isinstance(node, ast.Call):
            continue

        # Detection predicate FIRST: ``(root / "Dockerfile").exists()`` has a
        # BinOp receiver, so it has no resolvable dotted name at all — testing
        # it after the ``canonical is None`` guard below would make it dead.
        if (
            predicate_exemption is None
            and isinstance(node.func, ast.Attribute)
            and node.func.attr == "exists"
            and isinstance(node.func.value, ast.BinOp)
            and isinstance(node.func.value.op, ast.Div)
        ):
            probe = _const_str(node.func.value.right)
            if probe in _DOCKER_FILENAME_CONSTANTS:
                state.add(
                    "B",
                    rel,
                    node.lineno,
                    f"branches on the presence of {probe!r}; orchestrata's own "
                    f"provisioning is Incus-based",
                )
                continue

        canonical = _canonical_call_name(node, aliases)
        if canonical is None:
            continue

        if canonical in _EXEC_FUNCS:
            argv, _resolved = _argv_strings(node, canonical)
            argv0 = argv[0] if argv else None
            if argv0 and Path(argv0).name in _DOCKER_BINARIES:
                state.add(
                    "B",
                    rel,
                    node.lineno,
                    f"{canonical} executes docker (argv[0]={argv0!r}); "
                    f"test isolation and deployment target Incus "
                    f"(scripts/start_test_services.py, harness IncusIsolation)",
                )
                continue
            # Only scan argv STRINGS as shell commands when a shell is
            # actually going to interpret them. Applying the shell grammar to
            # every argv element makes a search PATTERN byte-indistinguishable
            # from a command: measured 2026-08-05,
            # ``subprocess.run(["grep", "-rn", "docker run", root])`` — a
            # companion audit that ENFORCES this very ruling — reported
            # "subprocess.run runs a shell command invoking docker (run)".
            # 327 such calls exist in 93 in-scope Python files, including four
            # sibling ruling gates.
            if _is_shell_exec(canonical, node, argv):
                for element in argv:
                    if element is None:
                        continue
                    hit = _shell_string_hit(element)
                    if hit is not None:
                        state.add(
                            "B",
                            rel,
                            node.lineno,
                            f"{canonical} runs a shell command invoking "
                            f"{hit[0]} ({hit[1]}); orchestrata runs on Incus",
                        )
                        break
            continue

        if predicate_exemption is not None:
            continue

        if canonical == "shutil.which":
            probe = _const_str(node.args[0]) if node.args else None
            if probe in _DOCKER_BINARIES:
                state.add(
                    "B",
                    rel,
                    node.lineno,
                    f"probes for the docker binary (shutil.which({probe!r})) "
                    f"to branch provisioning; the runtime is Incus",
                )
            continue


# --------------------------------------------------------------------------
# Rule C — config / env
# --------------------------------------------------------------------------


def _is_rule_c_target(rel: str) -> bool:
    if rel in ("pyproject.toml", ".env.example"):
        return True
    if rel.startswith("deploy/"):
        return True
    if rel.startswith("env/") and rel.endswith(".env"):
        return True
    if rel.startswith("infra/"):
        return Path(rel).suffix in _INFRA_TERRAFORM_SUFFIXES
    return False


def scan_config_file(rel: str, text: str, state: ScanState) -> None:
    in_dependency_block = rel == "pyproject.toml"
    for lineno, raw in enumerate(text.splitlines(), start=1):
        line = raw.split("#", 1)[0] if raw.lstrip().startswith("#") else raw
        if not line.strip():
            continue
        for pattern, description in _CONFIG_MARKER_RES:
            if pattern.search(line):
                state.add(
                    "C",
                    rel,
                    lineno,
                    f"{description}; infra endpoints are Incus containers "
                    f"(env/{{dev,test,prod}}.env)",
                )
                break
        if in_dependency_block and _PYPROJECT_DOCKER_DEP_RE.match(line):
            state.add(
                "C",
                rel,
                lineno,
                "declares a docker distribution dependency; 'docker' is a "
                "banned library (.claude/banned-libraries.txt)",
            )


# --------------------------------------------------------------------------
# Rule D — file introduction
# --------------------------------------------------------------------------


def scan_file_introduction(rel: str, state: ScanState) -> None:
    name = Path(rel).name
    if _DOCKER_FILENAME_RE.match(name):
        state.add(
            "D",
            rel,
            1,
            f"introduces {name} outside the allowlisted docker/ and "
            f"infra/docker/ trees; docker-compose is the topology REFERENCE, "
            f"not the deployment path",
        )


# --------------------------------------------------------------------------
# Driver
# --------------------------------------------------------------------------


def run_check(repo_root: Path) -> tuple[int, dict]:
    """Scan ``repo_root``; return ``(exit_code, report)``."""
    repo_root = repo_root.resolve()
    state = ScanState()

    for abs_path, rel in iter_scanned_files(repo_root):
        if allowlist_entry_for(rel) is not None:
            continue

        scan_file_introduction(rel, state)

        if not _in_content_scope(rel):
            continue

        suffix = Path(rel).suffix
        if (
            suffix not in _SHELL_SUFFIXES
            and suffix != ".py"
            and not _is_rule_c_target(rel)
        ):
            continue

        text = _read_text(abs_path)
        if text is None:
            continue
        state.scanned += 1

        # Necessary-condition prefilter, NOT a detection rule: every pattern in
        # rules A/B/C contains one of these needles somewhere in the file
        # (``DOCKER_HOST``, ``docker.sock``, ``docker.io/``, ``import docker``,
        # ``shutil.which("docker")``, ``Dockerfile``, ``compose.yml``...), so a
        # file without one cannot produce a finding. Skipping the AST parse for
        # the other ~2760 in-scope files takes the whole-tree scan from ~13.7s
        # to well under a second, which is what makes this affordable as a
        # BLOCKING gate. Rule D is filename-based and already ran above.
        if not _has_docker_needle(text):
            continue
        state.graded += 1

        if suffix in _SHELL_SUFFIXES:
            scan_shell_file(rel, text, state)
        elif suffix == ".py":
            scan_python_file(rel, text, state)

        if _is_rule_c_target(rel):
            scan_config_file(rel, text, state)

    quarantined = [
        asdict(entry) for entry in QUARANTINE if (repo_root / entry.path).exists()
    ]

    report = {
        "rule_id": RULE_ID,
        "repo_root": str(repo_root),
        "status": "fail" if state.violations else "pass",
        "files_scanned": state.scanned,
        "files_graded": state.graded,
        "enforced_scope": {
            "roots": list(ENFORCED_ROOTS),
            "files": list(ENFORCED_FILES),
        },
        "allowlist": [asdict(entry) for entry in ALLOWED_ROOTS],
        "exemptions": [
            {"prefix": prefix, "clause": "rule B detection predicate", "reason": reason}
            for prefix, reason in DETECTION_PREDICATE_EXEMPT_PREFIXES
        ],
        "quarantine": quarantined,
        "unparsed_python": state.unparsed_python,
        "violations": [asdict(v) for v in state.violations],
        "ruling": RULING_TEXT,
    }
    return (1 if state.violations else 0), report


def _print_text_report(report: dict) -> None:
    for entry in report["quarantine"]:
        print(
            f"QUARANTINE {entry['path']} — {entry['reason']}; violates "
            f"{entry['clause']}. Recommendation: {entry['recommendation']}",
            flush=True,
        )
    if report["status"] == "pass":
        print(
            f"[incus-not-docker] OK — {RULE_ID} satisfied "
            f"({report['files_scanned']} in-scope files scanned, "
            f"{report['files_graded']} carrying a docker token graded)",
            flush=True,
        )
        return
    print(
        f"[incus-not-docker] FAIL — {len(report['violations'])} violation(s):",
        flush=True,
    )
    for raw in report["violations"]:
        print(f"  {Violation(**raw).render()}", flush=True)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="RUL-004 gate — Incus (LXD), not Docker, for containers."
    )
    parser.add_argument(
        "--repo-root", default=None, help="Directory to gate (default: repo root)"
    )
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    parser.add_argument(
        "--ruling", action="store_true", help="Print the ratified ruling and exit 0"
    )
    args = parser.parse_args(argv)

    if args.ruling:
        print(RULING_TEXT)
        return 0

    repo_root = Path(args.repo_root).resolve() if args.repo_root else REPO_ROOT
    exit_code, report = run_check(repo_root)

    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())