user-engine/scripts/check_layer_conformance.py

#!/usr/bin/env python3
"""Check user-engine against the NetKingdom security layer model (§5, §11).

Read-only. user-engine is Engine/PIP and holds no catalogued Tooling client.
PostgreSQL is the modeled-concept store, declared under own_store.
INTENT.md frontmatter governs the layer; layer.yaml is its derived form.

The failure this exists to catch is a convenience: an OpenBao, Vault, LDAP,
or cluster client arriving as one import. That is an undeclared violation.

Exit 0 clean, 1 undeclared contact found, 2 declaration malformed.
"""
from __future__ import annotations

import argparse
import ast
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "user_engine"
DECL = ROOT / "layer.yaml"
INTENT = ROOT / "INTENT.md"
sys.path.insert(0, str(ROOT / "src"))

from user_engine.layer_yaml import load_mapping, load_mapping_text  # noqa: E402

TOOLING_IMPORTS = {
    "hvac": "OpenBao / Vault client",
    "bao": "OpenBao client",
    "kubernetes": "cluster client",
    "kubernetes_asyncio": "cluster client",
    "ldap3": "direct LDAP client (key-cape tooling)",
    "python_ldap": "direct LDAP client (key-cape tooling)",
    "docker": "container runtime client",
    "redis": "direct datastore connection",
}

OWN_STORE_IMPORTS = {
    "psycopg": "PostgreSQL modeled-concept store",
    "psycopg2": "PostgreSQL modeled-concept store",
    "asyncpg": "PostgreSQL modeled-concept store",
    "sqlalchemy": "database client",
    "pymysql": "database client",
}


# §3's vocabulary is closed at four tokens and compared after an ASCII
# case-fold (GH-DEC-2026-017 §2, §3, amendment A9). Nothing is re-spelled.
LAYER_VOCABULARY = {"taxonomy", "tooling", "engine", "staff"}


def fold(value: object) -> str:
    """ASCII case-fold only: non-ASCII letters are left untouched."""
    return "".join(
        chr(ord(c) + 32) if "A" <= c <= "Z" else c for c in str(value).strip()
    )


def load_intent_layer() -> str:
    """Return the governing `layer:` value from INTENT.md frontmatter (§11)."""
    try:
        text = INTENT.read_text()
    except OSError as exc:
        print(f"FAIL: cannot read {INTENT.name}: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc
    parts = text.split("---", 2)
    if not text.startswith("---") or len(parts) < 3:
        print(f"FAIL: {INTENT.name} has no frontmatter (§11)", file=sys.stderr)
        raise SystemExit(2)
    try:
        front = load_mapping_text(parts[1])
    except ValueError as exc:
        print(f"FAIL: {INTENT.name} frontmatter is not parseable: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc
    if "layer" not in front:
        print(f"FAIL: {INTENT.name} frontmatter has no 'layer:' key (§11)", file=sys.stderr)
        raise SystemExit(2)
    return str(front["layer"])


def load_declaration() -> dict:
    if not DECL.exists():
        print(f"FAIL: no declaration at {DECL.relative_to(ROOT)} (§11)", file=sys.stderr)
        raise SystemExit(2)
    try:
        data = load_mapping(DECL)
    except (ValueError, OSError) as exc:
        print(f"FAIL: {DECL.name} is not parseable: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc
    # No standard_version: a layer declaration MUST NOT carry one
    # (GH-DEC-2026-017 §4, amendment A12). Its presence is malformed.
    for key in ("layer", "role", "repository", "tooling_contacts", "derived", "derived_from"):
        if key not in data:
            print(f"FAIL: {DECL.name} missing required key '{key}' (§11)", file=sys.stderr)
            raise SystemExit(2)
    if "standard_version" in data:
        print(
            f"FAIL: {DECL.name} carries standard_version; a layer declaration "
            "MUST NOT carry a standard version (GH-DEC-2026-017, A12)",
            file=sys.stderr,
        )
        raise SystemExit(2)
    if data["derived"] is not True or data["derived_from"] != "INTENT.md":
        print(
            f"FAIL: {DECL.name} must be marked derived: true, derived_from: INTENT.md "
            "(§11 derived-artifact rule, GH-DEC-2026-017, A11)",
            file=sys.stderr,
        )
        raise SystemExit(2)
    if fold(data["layer"]) not in LAYER_VOCABULARY:
        print(
            f"FAIL: declared layer {data['layer']!r} is not a §3 token "
            f"({', '.join(sorted(LAYER_VOCABULARY))}, case-insensitive)",
            file=sys.stderr,
        )
        raise SystemExit(2)
    intent_layer = load_intent_layer()
    if fold(intent_layer) != fold(data["layer"]):
        # A finding in its own right; INTENT.md governs but the disagreement
        # is reported, not resolved away by precedence (A11).
        print(
            f"FAIL: INTENT.md declares layer {intent_layer!r} but {DECL.name} "
            f"declares {data['layer']!r}; they disagree after case-folding",
            file=sys.stderr,
        )
        raise SystemExit(2)
    if fold(intent_layer) != "engine":
        print(
            f"FAIL: declared layer is {intent_layer!r}, expected Engine",
            file=sys.stderr,
        )
        raise SystemExit(2)
    if str(data["role"]).lower() != "pip":
        print(f"FAIL: declared role is {data['role']!r}, expected 'pip'", file=sys.stderr)
        raise SystemExit(2)
    if data["tooling_contacts"] not in ([], None):
        print(
            "FAIL: tooling_contacts must be empty; catalogued Tooling clients "
            "are undeclared violations for this Engine",
            file=sys.stderr,
        )
        raise SystemExit(2)
    return data


def imported_modules(path: Path) -> set[str]:
    try:
        tree = ast.parse(path.read_text())
    except SyntaxError:
        return set()
    found: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            found.update(alias.name.split(".")[0] for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.level == 0 and node.module:
                found.add(node.module.split(".")[0])
    return found


def scan() -> list[tuple[Path, str, str]]:
    hits: list[tuple[Path, str, str]] = []
    for path in sorted(SRC.rglob("*.py")):
        for module in sorted(imported_modules(path)):
            if module in TOOLING_IMPORTS:
                hits.append((path, module, TOOLING_IMPORTS[module]))
    return hits


def scan_own_store() -> list[tuple[Path, str, str]]:
    hits: list[tuple[Path, str, str]] = []
    for path in sorted(SRC.rglob("*.py")):
        for module in sorted(imported_modules(path)):
            if module in OWN_STORE_IMPORTS:
                hits.append((path, module, OWN_STORE_IMPORTS[module]))
    return hits


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--report", action="store_true", help="print the declaration summary")
    args = parser.parse_args()

    decl = load_declaration()
    hits = scan()
    own_store_hits = scan_own_store()
    own_store = decl.get("own_store") or []

    if args.report:
        print(
            f"user-engine — layer {decl['layer']}/{decl['role']} "
            f"(derived from {decl['derived_from']})"
        )
        print("  tooling contacts declared: 0")
        print(f"  own-store declarations: {len(own_store)}")
        print(f"  own-store imports: {len(own_store_hits)}")
        print(f"  pep stance: {decl.get('pep_stance')}")

    if hits:
        print("", file=sys.stderr)
        print(
            "FAIL: undeclared Tooling-layer client (§11 undeclared violation)",
            file=sys.stderr,
        )
        for path, module, what in hits:
            print(f"  {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr)
        return 1

    if own_store_hits and not own_store:
        print(
            "FAIL: modeled-concept store import with no own_store declaration",
            file=sys.stderr,
        )
        for path, module, what in own_store_hits:
            print(f"  {path.relative_to(ROOT)}: imports {module!r} — {what}", file=sys.stderr)
        return 1

    if not args.report:
        print(
            f"OK: no catalogued Tooling client in {SRC.relative_to(ROOT)} "
            "(Engine/PIP, §11)"
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())