scripts.validate_ci

Reference › scripts › validate_ci

scripts/validate_ci.py is the deterministic CI configuration validator: it checks .gitlab-ci.yml, the Docker compose file and the repository's environment handling against a numbered list of gates, and stops at the first failure through fail. A human or a CI job runs it; nothing imports it. It uses orchestrata_ultimate.core.infra_env_rule for the infrastructure-environment rule.

main takes only keyword arguments (a repository root and an environment mapping, for tests) and runs these gates in order: 1, the YAML parses; 2, the stages lint, test and build exist; 3, no stale --cov=src coverage path; 4, pytest-xdist is banned, because it deadlocks this suite, so job script lines are scanned for an xdist invocation; 5, a timeout is present; 6, service containers or environment variables for Redis and the graph; 7, an architecture checker or mypy is present; 8, the compose file uses environment variables rather than hard-coded addresses; 9, test-execution jobs call the canonical runner run_tests; 10, the AST outputs exist and are not stale; 11, no hard-coded production connection literals (_scan_hardcoded_prod_ip); 12, no literal fallback for required infrastructure environment keys, per RUL-003 (_scan_infra_env_defaults); 13, every environment-profile overlay loader honours one ownership set (_scan_env_overlay_loaders, _check_loader_consults_marker, _check_fallback_marker_name); and 14, every module that claims to own an environment key actually declares it (_scan_env_ownership_writers, _declared_keys_in_module).

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

#!/usr/bin/env python3
"""Deterministic CI configuration validator.

Verifies .gitlab-ci.yml and docker-compose.yml against hard requirements.
Exit 0 = all gates pass. Exit 1 = first failure printed to stderr.
"""

from __future__ import annotations

import os
import pathlib
import re
import sys
from collections.abc import Mapping
from typing import NoReturn

import yaml

# A job's script clearly runs the test suite if it invokes pytest (directly,
# via ``uv run pytest`` / ``python -m pytest``, or via the threadpool runner)
# or passes the project's coverage target -- any of those without also
# calling scripts/run_tests.py bypasses the canonical runner's infra check,
# blocking gates, and solo-retry round (Phase R6, plans/reproducibility-plan.md).
# Deliberately NOT matched on a bare "tests/" path reference: lint/static
# jobs (ruff/black/mypy over "src/ tests/") legitimately mention that path
# without executing the suite, and must not trip this gate.
_TEST_EXEC_PATTERN = re.compile(
    r"\bpytest\b|run_tests_threadpool\.py|--cov=orchestrata_ultimate"
)
_RESERVED_CI_KEYS = {"stages", "variables", "include", "workflow", "default"}

# Gate 11: a *_HOST/*_URI/*_URIS dict/env key, or a host=/uri= keyword
# argument, immediately followed by a quoted 10.53.1.x literal — the exact
# shape of the bug a 2026-08 audit found in scripts/e2e_run.py and
# scripts/refactor.py (``os.environ.setdefault("REDIS_HOST", "10.53.1.10")``,
# the PRODUCTION Redis). Deliberately narrower than Gate 8's whole-text
# docker-compose.yml scan: a bare string occurrence isn't enough to flag here
# (comments, docstrings/CLI usage examples, and classification sets like
# gen_level0.py's `_ORCHGRAPH_COORDINATORS` frozenset legitimately mention the
# same IPs without ever using them as a connection default).
_PROD_IP_CONNECTION_CONTEXT = re.compile(
    r'["\'][A-Za-z_]*(?:HOST|URI|URIS)["\']\s*[:,]\s*f?["\'](?:[a-z][a-z0-9+.\-]*://)?10\.53\.1\.\d{1,3}'
    r'|\b(?:host|uri|redis_host|memgraph_uri)\s*=\s*f?["\'](?:[a-z][a-z0-9+.\-]*://)?10\.53\.1\.\d{1,3}',
    re.IGNORECASE,
)


def _scan_hardcoded_prod_ip(repo_root: pathlib.Path) -> list[str]:
    """Find live 10.53.1.x connection literals in scripts/*.py and src/**/*.py.

    Line-scanned (not AST-walked): only lines matching
    ``_PROD_IP_CONNECTION_CONTEXT`` are flagged, and comment lines are
    skipped outright, so explanatory comments/docstrings referencing a
    retired or historical address (there are many, deliberately) do not trip
    this gate — only a literal actually wired up as a connection default.
    """
    hits: list[str] = []
    targets = sorted((repo_root / "scripts").glob("*.py")) + sorted(
        (repo_root / "src").rglob("*.py")
    )
    for path in targets:
        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except OSError, UnicodeDecodeError:
            continue
        for lineno, line in enumerate(lines, start=1):
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            if _PROD_IP_CONNECTION_CONTEXT.search(line):
                hits.append(f"{path.relative_to(repo_root)}:{lineno}: {stripped}")
    return hits


def _scan_infra_env_defaults(repo_root: pathlib.Path) -> list[str]:
    """Find literal fallbacks for REQUIRED infra-connection env keys (RUL-003)
    across scripts/*.py, src/**/*.py and tests/**/*.py.

    Generalizes Gate 11 (prod-IP literals) to ANY literal default on a required
    endpoint key — including ``"localhost"`` / ``""`` — the strict rule the
    2026-08-03 audit adopted. Shares its key set + AST scanner with the
    ``check_infra_env_defaults`` review check via ``core.infra_env_rule`` so the
    gate and the review can never drift. Imported lazily (with ``src`` added to
    the path) so this otherwise stdlib-only validator does not hard-depend on
    the package being importable in every context.

    ``tests`` joined the scan on 2026-08-07. It was outside it until then, and
    that hole is the whole reason 118 sites accumulated there — every one of
    them a ``os.environ.get("REDIS_HOST", "localhost")`` re-derivation of an
    endpoint ``tests/infrastructure`` already resolves. A test that defaults to
    localhost does not fail loudly on a config gap; it connects to nothing and
    reports the resulting timeout as a product bug. The rule is identical in
    all three trees, so the scan scope is now identical too.
    """
    import ast

    src_dir = str(repo_root / "src")
    if src_dir not in sys.path:
        sys.path.insert(0, src_dir)
    from orchestrata_ultimate.core.infra_env_rule import find_infra_env_defaults

    hits: list[str] = []
    targets = (
        sorted((repo_root / "scripts").glob("*.py"))
        + sorted((repo_root / "src").rglob("*.py"))
        + sorted((repo_root / "tests").rglob("*.py"))
    )
    for path in targets:
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except OSError, UnicodeDecodeError, SyntaxError:
            continue
        hits.extend(
            f"{path.relative_to(repo_root)}:{v.lineno}: "
            f'.{v.method}("{v.key}", <default>)'
            for v in find_infra_env_defaults(tree)
        )
    return hits


# (file, function that applies the overlay, name it must CALL)
_ENV_OVERLAY_LOADERS = (
    (
        "src/orchestrata_ultimate/core/env_bootstrap.py",
        "load_env_contract",
        "owned_keys",
    ),
    ("scripts/start_test_services.py", "_load_dotenv", "_owned_keys"),
)


def _check_loader_consults_marker(
    repo_root: pathlib.Path, rel: str, func_name: str, symbol: str
) -> list[str]:
    """One loader: it must CALL *symbol* inside *func_name*, and own no list."""
    import ast

    hits: list[str] = []
    path = repo_root / rel
    if not path.exists():
        return [f"{rel}: env-overlay loader is missing"]
    try:
        source = path.read_text(encoding="utf-8")
        tree = ast.parse(source)
    except (OSError, UnicodeDecodeError, SyntaxError) as exc:
        return [f"{rel}: unparseable ({exc})"]

    func = next(
        (
            n
            for n in ast.walk(tree)
            if isinstance(n, ast.FunctionDef) and n.name == func_name
        ),
        None,
    )
    if func is None:
        hits.append(f"{rel}: overlay function `{func_name}` not found")
    else:
        # A CALL inside the function body — not a substring anywhere in the
        # file. The first draft of this gate matched the bare name and stayed
        # green after the call was removed, because the name still appeared in
        # a docstring and in declare_owned().
        called = any(
            isinstance(n, ast.Call)
            and isinstance(n.func, ast.Name)
            and n.func.id == symbol
            for n in ast.walk(func)
        )
        if not called:
            hits.append(
                f"{rel}:{func.lineno}: `{func_name}` overlays the env profile "
                f"but never CALLS `{symbol}()` — a key this process claimed "
                "via declare_owned() would be silently overwritten"
            )
    if "_PER_PROCESS_AUTHORITATIVE_KEYS" in source:
        hits.append(
            f"{rel}: reintroduced a private hardcoded ownership list "
            "(_PER_PROCESS_AUTHORITATIVE_KEYS). The claim travels in the "
            "environment so BOTH loaders see it; a per-loader list cannot "
            "protect the other loader."
        )
    return hits


def _check_fallback_marker_name(repo_root: pathlib.Path) -> list[str]:
    """The fallback copy of the marker NAME must match the canonical constant.

    ``start_test_services`` carries one for the path where the package is not
    importable, so a rename on one side silently splits the ownership set in
    two. Both reads are guarded: a missing loader is reported by
    :func:`_check_loader_consults_marker`, and this check must add to that
    report rather than raise over it.
    """
    boot_path = repo_root / "src/orchestrata_ultimate/core/env_bootstrap.py"
    start_path = repo_root / "scripts/start_test_services.py"
    if not (boot_path.exists() and start_path.exists()):
        return []
    canonical = ""
    for line in boot_path.read_text(encoding="utf-8").splitlines():
        if line.startswith("ENV_OWNED_KEYS_VAR"):
            canonical = line.split("=", 1)[1].strip().strip('"').strip("'")
            break
    starter = start_path.read_text(encoding="utf-8")
    if not (canonical and '_ENV_OWNED_KEYS_VAR = "' in starter):
        return []
    mirrored = starter.split('_ENV_OWNED_KEYS_VAR = "', 1)[1].split('"', 1)[0]
    if mirrored == canonical:
        return []
    return [
        "scripts/start_test_services.py: fallback marker name "
        f"{mirrored!r} != env_bootstrap.ENV_OWNED_KEYS_VAR {canonical!r}"
    ]


def _scan_env_overlay_loaders(repo_root: pathlib.Path) -> list[str]:
    """Every loader that overlays the env profile must honour ONE ownership set.

    Two modules apply ``env/<profile>.env`` over the running environment:
    ``core.env_bootstrap.load_env_contract`` and
    ``scripts.start_test_services._load_dotenv``. Until 2026-08-07 they
    disagreed — start_test_services skipped three keys by name, env_bootstrap
    skipped none — and the gap silently reset the suite's per-worker isolation
    (``KG_DATABASE`` ``orchkg_w<N>`` -> ``orchkg``) whenever any entrypoint ran
    mid-process. Writes went to one database and read-backs to another; four
    authoritative gate runs were misdiagnosed as Memgraph contention.

    So this is not a style check. Both loaders must consult ``owned_keys`` (the
    marker), and neither may reintroduce a private hardcoded key list — a
    second list is exactly how the policies diverged the first time.
    """
    hits: list[str] = []
    for rel, func_name, symbol in _ENV_OVERLAY_LOADERS:
        hits.extend(_check_loader_consults_marker(repo_root, rel, func_name, symbol))
    hits.extend(_check_fallback_marker_name(repo_root))
    return hits


# Modules whose declare_owned() calls exist to test the FUNCTION ITSELF,
# against synthetic dicts, unconditionally on every run -- never a real
# ownership decision. Counting this module as evidence that "someone
# declares ownership" would make _scan_env_ownership_writers pass even with
# every REAL caller deleted -- the identical trap
# _check_loader_consults_marker's own docstring already recounts (a bare
# name match staying green because the name still appeared in a docstring
# and in declare_owned() itself).
_ENV_OWNERSHIP_SELF_TEST = "tests/core/test_env_bootstrap.py"

# A module's own citation style for a key it has decided to specialize --
# the word OWNS (never the bare possessive "own", which this repo uses
# constantly for "its own ``X``" -- ``knowledge_graph/sync.py``'s "syncer's
# own ``SYNC_DDL``" is not an ownership claim) immediately followed by a
# double-backtick ALL-CAPS token, e.g. a docstring saying a script owns the
# KG_DATABASE key (written here with single backticks so this comment itself
# is not a match). Deliberately narrow and text-scanned, like every other
# prose-matching gate in this file (Gate 11/12) -- not a claim to parse
# arbitrary English.
_OWNERSHIP_CLAIM_RE = re.compile(r"\bowns\s+``([A-Z][A-Z0-9_]*)``")


def _declared_keys_in_module(path: pathlib.Path) -> tuple[str, ...]:
    """String-literal keys passed to real ``declare_owned(...)`` calls in *path*.

    Alias-aware: resolves ``from ...env_bootstrap import declare_owned`` AND
    the aliased spelling ``... import declare_owned as _declare_owned`` --
    ``tests/conftest.py``'s own form. This function's approach was
    generalized (2026-08-19, after this exact alias undercounted
    ``declare_owned``'s real callers in a dead-code audit) into
    ``serialize_codebase_ast.py``'s ``refs`` / ``alias_aware_references`` --
    an arbitrary-symbol, four-binding-form resolver (this one is hardcoded to
    one symbol, from-import form only). ``refs declare_owned`` is alias-aware
    now too; this narrow, single-purpose version stays here rather than
    importing the 13k-line toolkit module into a CI gate script, and because
    it also extracts the ``declare_owned(...)`` call's STRING-LITERAL
    ARGUMENTS (the owned keys), which is Gate 13's own job and out of scope
    for a generic reference finder.

    Returns an empty tuple when *path* never imports ``declare_owned`` at
    all -- unparseable files included, so a syntax error elsewhere in the
    repo cannot crash this gate (matching
    ``_check_loader_consults_marker``'s own error handling).
    """
    import ast

    try:
        tree = ast.parse(path.read_text(encoding="utf-8"))
    except OSError, UnicodeDecodeError, SyntaxError:
        return ()

    local_name: str | None = None
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and (node.module or "").endswith(
            "core.env_bootstrap"
        ):
            for alias in node.names:
                if alias.name == "declare_owned":
                    local_name = alias.asname or alias.name
    if local_name is None:
        return ()

    keys: list[str] = []
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Name)
            and node.func.id == local_name
        ):
            keys.extend(
                arg.value
                for arg in node.args
                if isinstance(arg, ast.Constant) and isinstance(arg.value, str)
            )
    return tuple(keys)


def _scan_env_ownership_writers(repo_root: pathlib.Path) -> list[str]:
    """Every module that DECIDES it owns a key must actually declare it.

    Gate 13 (above) only proves the two overlay loaders CONSULT
    ``owned_keys()``; nothing ever proved anyone POPULATES it. Two failure
    classes, both regressions of RUL-019's write side:

    1. No REAL ``declare_owned(...)`` call exists anywhere in the repo. Then
       ``owned_keys()`` is unconditionally empty and the whole per-process
       protection is a no-op -- as if ``declare_owned`` had never been
       written, just with the function still sitting there looking wired.
    2. A module's docstring claims ownership of a named key (see
       ``_OWNERSHIP_CLAIM_RE``) but that SAME module never actually calls
       ``declare_owned(KEY, ...)`` -- a claim with nothing behind it.

    ``_ENV_OWNERSHIP_SELF_TEST`` is excluded from class 1's evidence -- see
    its own comment.

    Class 1 scans ``tests/`` too (``tests/conftest.py`` is the real,
    production-grade owner of the per-worker isolation keys), but class 2
    scans only ``src/`` and ``scripts/`` -- production entrypoints, the only
    modules that can genuinely CLAIM ownership in the RUL-019 sense. A test
    module is never an entrypoint, and scanning it would flag its OWN
    ownership-claim fixtures (this file's gate-14 tests among them) as if
    they were real, unbacked claims about ``tests/scripts/test_validate_ci.py``
    itself.
    """
    hits: list[str] = []
    caller_targets = (
        sorted((repo_root / "scripts").glob("*.py"))
        + sorted((repo_root / "src").rglob("*.py"))
        + sorted((repo_root / "tests").rglob("*.py"))
    )
    claim_targets = sorted((repo_root / "scripts").glob("*.py")) + sorted(
        (repo_root / "src").rglob("*.py")
    )

    declared_by: dict[str, tuple[str, ...]] = {}
    for path in caller_targets:
        rel = path.relative_to(repo_root).as_posix()
        declared_by[rel] = _declared_keys_in_module(path)

    real_caller = any(
        keys for rel, keys in declared_by.items() if rel != _ENV_OWNERSHIP_SELF_TEST
    )
    if not real_caller:
        hits.append(
            "no module anywhere calls core.env_bootstrap.declare_owned() with "
            "a real key -- owned_keys() is unconditionally empty and the "
            "RUL-019 per-process protection is a no-op"
        )

    for path in claim_targets:
        rel = path.relative_to(repo_root).as_posix()
        try:
            text = path.read_text(encoding="utf-8")
        except OSError, UnicodeDecodeError:
            continue
        claimed = set(_OWNERSHIP_CLAIM_RE.findall(text))
        if not claimed:
            continue
        declared = set(declared_by.get(rel, ()))
        for key in sorted(claimed - declared):
            hits.append(
                f"{rel}: docstring claims ownership of `{key}` but this "
                f"module never calls declare_owned({key!r}, ...)"
            )
    return hits


def fail(msg: str) -> NoReturn:
    print(f"FAIL: {msg}", file=sys.stderr)
    sys.exit(1)


def main(
    *,
    repo_root: pathlib.Path | None = None,
    env: Mapping[str, str] | None = None,
) -> int:
    if repo_root is None:
        repo_root = pathlib.Path(__file__).resolve().parent.parent
    if env is None:
        env = os.environ
    ci_file = repo_root / ".gitlab-ci.yml"
    compose_file = repo_root / "docker" / "docker-compose.yml"

    # --- Gate 1: YAML parseable ---
    if not ci_file.exists():
        fail(".gitlab-ci.yml not found")
    with ci_file.open() as f:
        try:
            ci = yaml.safe_load(f) or {}
        except yaml.YAMLError as exc:
            fail(f"Invalid YAML: {exc}")

    # --- Gate 2: Required stages present ---
    stages = ci.get("stages", [])
    required_stages = {"lint", "test", "build"}
    missing = required_stages - set(stages)
    if missing:
        fail(f"Missing stages: {missing}")

    # --- Gate 3: No stale coverage path ---
    ci_text = ci_file.read_text()
    if "--cov=src" in ci_text:
        fail("Found '--cov=src' (should be '--cov=orchestrata_ultimate')")

    # --- Gate 4: pytest-xdist is BANNED (deadlocks this suite ~78%) ---
    # Parallelism comes from scripts/run_tests.py's batched executor, NOT from
    # pytest-xdist. Scan actual job SCRIPT lines (never comments) for an xdist
    # invocation. The prior gate REQUIRED "-n auto" — obsolete and backwards:
    # it passed only because a warning comment happens to mention the flag.
    #
    # Two match classes, checked PER LINE:
    #  - UNAMBIGUOUS: --numprocesses, --dist (but not --dist-dir), pytest-xdist.
    #  - The short `-n auto`/`-n4`/`-nauto` form is ambiguous with head/grep/tail/
    #    sort `-n`, so it only counts on a test-EXECUTION line (one matching
    #    _TEST_EXEC_PATTERN: pytest / run_tests_threadpool.py / the cov target).
    _XDIST_UNAMBIGUOUS = re.compile(r"--numprocesses\b|--dist(?![\w-])|pytest-xdist")
    _XDIST_SHORT_N = re.compile(r"(?<![\w-])-n\s*=?\s*(?:auto|\d)")
    for job_name, job_def in ci.items():
        if job_name in _RESERVED_CI_KEYS or job_name.startswith("."):
            continue
        if not isinstance(job_def, Mapping):
            continue
        script = job_def.get("script", [])
        script_lines = script if isinstance(script, list) else [script]
        for raw_line in script_lines:
            line = str(raw_line)
            if _XDIST_UNAMBIGUOUS.search(line) or (
                _TEST_EXEC_PATTERN.search(line) and _XDIST_SHORT_N.search(line)
            ):
                fail(
                    f"CI job '{job_name}' invokes pytest-xdist (-n/--numprocesses/"
                    "--dist) — BANNED (deadlocks this suite ~78%). Use "
                    "scripts/run_tests.py."
                )

    # --- Gate 5: Timeout present ---
    if "--timeout=" not in ci_text:
        fail("Missing '--timeout=' in pytest invocation")

    # --- Gate 6: Service containers or env vars for Redis/Neo4j ---
    if "redis" not in ci_text.lower() and "REDIS_HOST" not in ci_text:
        fail("No Redis reference in CI config")
    if "neo4j" not in ci_text.lower() and "NEO4J_URI" not in ci_text:
        fail("No graph reference in CI config")

    # --- Gate 7: Architecture checker or mypy present ---
    if "architecture_checker" not in ci_text and "mypy" not in ci_text:
        fail("No architecture_checker or mypy stage")

    # --- Gate 8: Docker Compose uses env vars, not hardcoded IPs ---
    if compose_file.exists():
        compose_text = compose_file.read_text()
        if "10.222.9" in compose_text:
            fail("docker-compose.yml still contains stale 10.222.9.x IPs")
        if "10.53.1" in compose_text:
            fail(
                "docker-compose.yml contains hardcoded 10.53.1.x IPs (use ${REDIS_HOST} / ${NEO4J_URI})"
            )
        # Accept bare ${VAR} or bash default-substitution ${VAR:-default}
        if "${REDIS_HOST" not in compose_text:
            fail("docker-compose.yml missing ${REDIS_HOST} env var reference")
        if "${NEO4J_URI" not in compose_text:
            fail("docker-compose.yml missing ${NEO4J_URI} env var reference")
    else:
        fail("docker-compose.yml not found")

    # --- Gate 9: test-execution jobs must invoke the canonical runner ---
    # scripts/run_tests.py is THE single entry point (infra check -> blocking
    # gates -> batched executor -> solo-retry round). A job whose script
    # clearly runs the suite (pytest / tests/ / the coverage target) but
    # never calls scripts/run_tests.py has invented its own invocation --
    # a raw pytest call, or even scripts/run_tests_threadpool.py directly --
    # and silently loses the safety gates and solo-retry flake handling.
    for job_name, job_def in ci.items():
        if job_name in _RESERVED_CI_KEYS or job_name.startswith("."):
            continue
        if not isinstance(job_def, Mapping):
            continue
        script = job_def.get("script", [])
        script_lines = script if isinstance(script, list) else [script]
        script_text = "\n".join(str(line) for line in script_lines)
        if not _TEST_EXEC_PATTERN.search(script_text):
            continue
        if "scripts/run_tests.py" not in script_text:
            fail(
                f"CI job '{job_name}' runs the test suite without calling "
                "scripts/run_tests.py (the canonical runner) — found a raw "
                "pytest/run_tests_threadpool.py invocation instead"
            )

    # --- Gate 10: AST outputs exist and are not stale ---
    plans_dir = repo_root / "plans"
    source_ultimate = repo_root / "src" / "orchestrata_ultimate"
    source_orchestrata = pathlib.Path(
        env.get(
            "ORCHESTRATA_MONOREPO_ROOT", "/home/mohamed/workspace/orchestrata/packages"
        )
    )
    ast_files = {
        "ast-ultimate.json": source_ultimate,
        "ast-orchestrata.json": source_orchestrata,
        "ast-deep-comparison.json": None,
    }
    for name, src_dir in ast_files.items():
        path = plans_dir / name
        if not path.exists():
            fail(f"Missing AST output: plans/{name} (run uv run ast-sync)")
        if src_dir is not None and src_dir.exists():
            # Stale check: AST mtime < newest .py in source
            newest_py = max(
                (p.stat().st_mtime for p in src_dir.rglob("*.py")),
                default=0,
            )
            if newest_py > path.stat().st_mtime:
                fail(
                    f"plans/{name} is stale (source changed since last run). Run: uv run ast-sync"
                )

    # --- Gate 11: no hardcoded PRODUCTION 10.53.1.x connection literals ---
    # Gate 8 already catches this in docker-compose.yml; this extends the
    # same doctrine (RUL-003: infra endpoints come from env, no default) to
    # scripts/*.py and src/**/*.py, so a future setdefault()/dict-default
    # like the one Gate 1 of this audit removed from scripts/e2e_run.py and
    # scripts/refactor.py is caught automatically instead of by hand.
    ip_hits = _scan_hardcoded_prod_ip(repo_root)
    if ip_hits:
        fail(
            "Hardcoded 10.53.1.x connection literal(s) found (read from env "
            "with no default — RUL-003):\n  " + "\n  ".join(ip_hits)
        )

    # --- Gate 12: no literal fallback for REQUIRED infra-connection env keys ---
    # The strict generalization of Gate 11 (user ruling 2026-08-03): a required
    # endpoint key (REDIS_HOST/PORT, KG_ROUTING_URI/KG_URI/KG_DATABASE,
    # MEMGRAPH_URI, CXP_GRPC_PORT, KG_GRPC_PORT) must RAISE when unset — no
    # `.get(KEY, default)` fallback, not even "localhost"/"". Same key set +
    # scanner as the `check_infra_env_defaults` review check (core.infra_env_rule).
    default_hits = _scan_infra_env_defaults(repo_root)
    if default_hits:
        fail(
            "Literal fallback for required infra-endpoint key(s) found "
            "(read from env with no default and raise — RUL-003, standards "
            "rule 30d; route through core.graph_factory.resolve_uri / "
            "core.redis_factory.resolve_redis_host):\n  " + "\n  ".join(default_hits)
        )

    # --- Gate 13: every env-profile overlay loader honours ONE ownership set ---
    # Divergence between the two loaders reset per-worker isolation mid-process
    # and cost four gate runs misdiagnosed as Memgraph contention (2026-08-07).
    overlay_hits = _scan_env_overlay_loaders(repo_root)
    if overlay_hits:
        fail(
            "Env-profile overlay loaders disagree on ownership (a key claimed "
            "with core.env_bootstrap.declare_owned() must survive EVERY "
            "loader):\n  " + "\n  ".join(overlay_hits)
        )

    # --- Gate 14: env-key ownership claims must have a real declare_owned() ---
    # Gate 13 proves both loaders CONSULT owned_keys(); nothing proved anyone
    # POPULATES it. declare_owned() had zero real (non-self-test) call sites
    # in src/ or scripts/ before this gate existed to catch that regression.
    ownership_hits = _scan_env_ownership_writers(repo_root)
    if ownership_hits:
        fail(
            "RUL-019 env-key ownership has nothing behind it:\n  "
            + "\n  ".join(ownership_hits)
        )

    print("OK: All 14 CI validation gates passed")
    return 0


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