EVEngine/scripts/check_architecture_contracts.py

#!/usr/bin/env python3
"""Check the executable architecture-contract policy.

This gate has two deliberately different jobs:

* validate the reviewed contract catalogue (ownership, links, ECS systems,
  time, persistence, capabilities, backends and debt); and
* lint only *added* C/C++ lines for the high-signal API mistakes which are
  cheap to detect without a C++ parser (ambiguous operation ``bool``,
  ``lastError`` channels and undocumented raw-pointer APIs).

The changed-line mode is important.  It makes the rule a no-net-growth gate
for a dirty worktree or a pull request without pretending that a regular
expression can prove every property of old C++ code.  Existing debt belongs
in the existing reviewed baseline/allowlist; it is never silently moved into
this gate's baseline.

Usage::

    python3 scripts/check_architecture_contracts.py
    python3 scripts/check_architecture_contracts.py --base origin/dev
    python3 scripts/check_architecture_contracts.py --all
    python3 scripts/check_architecture_contracts.py --json

``--all`` validates the catalogue and reports the same API smells over every
source file.  CI uses the default changed-only mode so an old violation cannot
hide a newly introduced one.
"""

from __future__ import annotations

import argparse
import fnmatch
import json
import os
import re
import subprocess
import sys
from dataclasses import dataclass
from datetime import date, datetime
from pathlib import Path
from typing import Any, Iterable, Mapping
import utf8_stdio

ROOT = Path(__file__).resolve().parent.parent
DEFAULT_METADATA = ROOT / "scripts" / "architecture_contracts.json"

RULES = (
    "api-shape",
    "link",
    "state-owner",
    "ecs-system",
    "time-rng",
    "api-lifetime",
    "persistence",
    "optional-capability",
    "backend-contract",
    "debt-metadata",
)

COMMON_REQUIRED = {
    "owner",
    "issue",
    "reason",
    "expiry",
    "evidence",
    "tests",
}
RULE_REQUIRED = {
    "api-shape": {"result_policy", "nodiscard_policy", "pointer_policy"},
    "link": {"symbols", "create", "ownership", "destroy_order", "restore", "stale"},
    "state-owner": {"state", "authoritative_owner", "projections"},
    "ecs-system": {
        "entity_scope",
        "view",
        "read_set",
        "write_set",
        "structural_changes",
        "events",
        "services",
        "phase",
        "systems",
    },
    "time-rng": {"time_source", "rng_stream", "determinism", "tolerance"},
    "api-lifetime": {
        "thread_affinity",
        "reentrancy",
        "ownership_contract",
        "lifetime",
        "lock_callback_policy",
    },
    "persistence": {
        "schema",
        "version",
        "migration",
        "unknown_fields",
        "restore_atomicity",
    },
    "optional-capability": {
        "present_test",
        "absent_test",
        "fallback_observable",
        "fallback_policy",
    },
    "backend-contract": {
        "contract",
        "providers",
        "shared_contract_tests",
        "failure_injection",
    },
    "debt-metadata": {
        "removal_condition",
        "max_net_growth",
    },
}


@dataclass(frozen=True)
class SourceLine:
    path: str
    line: int
    text: str


@dataclass(frozen=True)
class Finding:
    rule: str
    code: str
    path: str
    line: int
    message: str

    def as_dict(self) -> dict[str, Any]:
        return {
            "rule": self.rule,
            "code": self.code,
            "path": self.path,
            "line": self.line,
            "message": self.message,
        }


def load_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(f"cannot read {path}: {error}") from error


def relative(path: Path) -> str:
    return path.relative_to(ROOT).as_posix()


def path_matches(path: str, scope: str) -> bool:
    # fnmatch's ** behaviour differs slightly between Python versions.  Try
    # both the spelling supplied by the metadata and a slash-normalized one.
    candidate = path.replace("\\", "/")
    pattern = scope.replace("\\", "/")
    return fnmatch.fnmatch(candidate, pattern) or fnmatch.fnmatch(
        candidate, pattern.replace("**/", "*/")
    )


def nonempty_string(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def nonempty_list(value: Any) -> bool:
    return isinstance(value, list) and bool(value) and all(nonempty_string(item) for item in value)


def catalogue_sort_key(entry: Mapping[str, Any]) -> tuple[int, str]:
    """Return the canonical review order for one catalogue entry."""

    rule = entry.get("rule")
    rule_index = RULES.index(rule) if rule in RULES else len(RULES)
    entry_id = entry.get("id")
    return rule_index, entry_id if isinstance(entry_id, str) else ""


def validate_catalogue(metadata: Any, today: date | None = None) -> list[str]:
    """Validate the contract catalogue without inspecting source code."""

    today = today or date.today()
    errors: list[str] = []
    if not isinstance(metadata, Mapping):
        return ["metadata root must be an object"]
    if metadata.get("schema_version") != 1:
        errors.append("metadata schema_version must be 1")
    entries = metadata.get("entries")
    if not isinstance(entries, list):
        return errors + ["metadata entries must be an array"]

    sortable_entries = [entry for entry in entries if isinstance(entry, Mapping)]
    if len(sortable_entries) == len(entries) and sortable_entries != sorted(
        sortable_entries, key=catalogue_sort_key
    ):
        errors.append(
            "metadata entries must use canonical rule/id order; run "
            "scripts/sort_architecture_contracts.py"
        )

    seen: set[str] = set()
    covered: set[str] = set()
    for index, entry in enumerate(entries):
        prefix = f"entries[{index}]"
        if not isinstance(entry, Mapping):
            errors.append(f"{prefix} must be an object")
            continue
        entry_id = entry.get("id")
        rule = entry.get("rule")
        if not nonempty_string(entry_id):
            errors.append(f"{prefix}.id must be a non-empty string")
        elif entry_id in seen:
            errors.append(f"duplicate contract id: {entry_id}")
        else:
            seen.add(entry_id)
        if rule not in RULES:
            errors.append(f"{prefix}.rule must be one of {', '.join(RULES)}")
        else:
            covered.add(rule)
        for field in ("scope", "owner", "issue", "reason"):
            if not nonempty_string(entry.get(field)):
                errors.append(f"{prefix}.{field} must be a non-empty string")
        scope = entry.get("scope")
        if nonempty_string(scope):
            scoped_files = [
                candidate
                for candidate in ROOT.rglob("*")
                if candidate.is_file() and path_matches(relative(candidate), scope)
            ]
            if not scoped_files:
                errors.append(f"{prefix}.scope matches no repository file: {scope}")
            if rule in {"link", "ecs-system"} and scope in {"src/**", "src/*", "src/**/*"}:
                errors.append(f"{prefix}.{rule} scope must be path-specific, not {scope}")
            names_field = "symbols" if rule == "link" else "systems" if rule == "ecs-system" else None
            if names_field is not None:
                names = entry.get(names_field)
                if not nonempty_list(names):
                    errors.append(f"{prefix}.{names_field} must be a non-empty string array")
                else:
                    source_text = "\n".join(
                        candidate.read_text(encoding="utf-8", errors="replace")
                        for candidate in scoped_files
                    )
                    for name in names:
                        if re.search(r"\b" + re.escape(name) + r"\b", source_text) is None:
                            errors.append(f"{prefix}.{names_field} symbol not found in scope: {name}")
        for field in ("evidence", "tests"):
            if not nonempty_list(entry.get(field)):
                errors.append(f"{prefix}.{field} must be a non-empty string array")
            else:
                for item in entry[field]:
                    evidence_path = ROOT / item
                    if not evidence_path.is_file():
                        errors.append(f"{prefix}.{field} references missing file {item}")
        expiry = entry.get("expiry")
        try:
            expiry_date = datetime.strptime(expiry, "%Y-%m-%d").date()
            if expiry_date < today:
                errors.append(f"{prefix}.expiry is past: {expiry}")
        except (TypeError, ValueError):
            errors.append(f"{prefix}.expiry must be YYYY-MM-DD")
        if rule in RULES:
            for field in COMMON_REQUIRED | RULE_REQUIRED[rule]:
                if field not in entry:
                    errors.append(f"{prefix} ({rule}) is missing {field}")
            if rule == "debt-metadata":
                growth = entry.get("max_net_growth")
                if not isinstance(growth, int) or growth < 0:
                    errors.append(f"{prefix}.max_net_growth must be a non-negative integer")

    missing = sorted(set(RULES) - covered)
    if missing:
        errors.append("catalogue has no entry for rule(s): " + ", ".join(missing))
    return errors


def _git(args: list[str]) -> str:
    try:
        result = subprocess.run(
            ["git", *args],
            cwd=ROOT,
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            # git writes UTF-8; the locale default (cp936 on a Chinese Windows)
            # would mangle or reject a non-ASCII path instead.
            encoding="utf-8",
            errors="replace",
        )
    except (OSError, subprocess.CalledProcessError):
        return ""
    return result.stdout


def _changed_lines(base: str | None) -> list[SourceLine]:
    """Return added C/C++ lines from a git diff and untracked source files."""

    diff_args = ["diff", "--no-color", "--unified=0"]
    if base:
        diff_args.append(base)
    diff_args.extend(["--", "src/**/*.h", "src/**/*.hpp", "src/**/*.cpp"])
    diff = _git(diff_args)
    lines: list[SourceLine] = []
    current: str | None = None
    new_line = 0
    for raw in diff.splitlines():
        if raw.startswith("+++ b/"):
            current = raw[6:]
            continue
        if raw.startswith("@@"):
            match = re.search(r"\+(\d+)(?:,(\d+))?", raw)
            if match:
                new_line = int(match.group(1))
            continue
        if current is None or not raw.startswith("+") or raw.startswith("+++"):
            if current is not None and raw.startswith("-"):
                continue
            continue
        lines.append(SourceLine(current, new_line, raw[1:]))
        new_line += 1

    untracked = _git(["ls-files", "--others", "--exclude-standard", "--", "src"])
    known = {(item.path, item.line) for item in lines}
    for name in untracked.splitlines():
        path = ROOT / name
        if path.suffix not in {".h", ".hpp", ".cpp"} or not path.is_file():
            continue
        for number, text in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), 1):
            if (name, number) not in known:
                lines.append(SourceLine(name, number, text))
    return lines


def _all_lines() -> list[SourceLine]:
    result: list[SourceLine] = []
    for path in sorted((ROOT / "src").rglob("*")):
        if path.suffix not in {".h", ".hpp", ".cpp"} or not path.is_file():
            continue
        for number, text in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), 1):
            result.append(SourceLine(relative(path), number, text))
    return result


BOOL_RETURN = re.compile(
    r"^\s*(?:(?:\[\[(?:nodiscard|deprecated)(?:\([^]]*\))?\]\]\s*)*"
    r"(?:(?:static|virtual|inline|constexpr|explicit)\s+)*)bool\s+"
    r"([A-Za-z_]\w*)\s*\("
)
POINTER_API = re.compile(
    r"^\s*(?:(?:\[\[nodiscard(?:\([^]]*\))?\]\]\s*)?"
    r"(?:static|virtual|inline|constexpr|explicit|const)\s+)*"
    r"(?:[A-Za-z_]\w*(?:::[A-Za-z_]\w*)*(?:\s*<[^;{}()]+>)?\s*)?\*\s*"
    r"[A-Za-z_]\w*\s*(?:\(|;|=)"
)
# A bool query is a legitimate API shape when its name follows the project's
# query vocabulary.  Keep this list deliberately lexical: this checker does
# not have a C++ type system, so it must not infer intent from a return value's
# callers.  The first-token forms cover names such as ``activeExecuted`` and
# ``usingGpu``; the last-token forms cover names such as
# ``transactionStateEquals``.  Exact names cover state predicates whose
# natural spelling has no ``is`` prefix (``ok``, ``active``, ...).
PREDICATE_FIRST_TOKENS = frozenset(
    {
        "is",
        "get",
        "has",
        "can",
        "should",
        "supports",
        "contains",
        "empty",
        "valid",
        "matches",
        "equals",
        "ok",
        "active",
        "paused",
        "passed",
        "owns",
        "using",
        "used",
        "critical",
        "disposed",
        "changed",
        "ready",
        "available",
        "enabled",
        "visible",
        "playing",
        "finished",
        "stale",
        "bound",
        "dirty",
        "as",
        "exact",
        "initialized",
    }
)
PREDICATE_LAST_TOKENS = frozenset(
    {
        "matches",
        "equals",
        "changed",
        "active",
        "paused",
        "passed",
        "disposed",
        "critical",
        "owned",
        "ready",
        "available",
        "valid",
        "visible",
        "enabled",
        "stale",
        "dirty",
    }
)


def _identifier_tokens(name: str) -> list[str]:
    """Split a C++ identifier into lower-case semantic name tokens."""

    words = re.sub(r"([a-z0-9])([A-Z])", r"\1 \2", name).replace("_", " ")
    return [word.lower() for word in words.split() if word]


def _is_predicate_name(name: str) -> bool:
    tokens = _identifier_tokens(name)
    if not tokens:
        return False
    return tokens[0] in PREDICATE_FIRST_TOKENS or tokens[-1] in PREDICATE_LAST_TOKENS


def _is_compatibility_declaration(text: str, context: str) -> bool:
    """Recognize an explicitly documented one-way compatibility facade.

    Looking for these words anywhere in a class made the old checker suppress
    unrelated declarations after a single legacy method.  Restrict the
    exemption to the declaration's own attribute or nearest Doxygen block.
    """

    if re.search(r"\[\[deprecated(?:\([^]]*\))?\]\]", text, re.IGNORECASE):
        return True
    blocks = re.findall(r"/\*\*.*?\*/", context, flags=re.DOTALL)
    if not blocks:
        return False
    return bool(re.search(r"\b(?:deprecated|compatibility|legacy)\b", blocks[-1], re.IGNORECASE))


def _context(lines: list[SourceLine], index: int) -> str:
    """Read the real preceding source, not only the diff hunk.

    A Doxygen contract is often unchanged while its declaration is edited. A
    changed-line lint must therefore inspect the complete file context or it
    would report a false violation merely because the documentation was not
    touched in the same patch.
    """

    item = lines[index]
    source = ROOT / item.path
    if source.is_file():
        content = source.read_text(encoding="utf-8", errors="replace").splitlines()
        start = max(0, item.line - 25)
        return "\n".join(content[start : item.line])
    start = max(0, index - 24)
    return "\n".join(candidate.text for candidate in lines[start : index + 1] if candidate.path == item.path)


def _is_non_public_declaration(item: SourceLine) -> bool:
    """Return whether a header declaration is under ``private``/``protected``.

    API-shape lint is intentionally about public contracts.  A private helper
    in a header may use a scalar control result internally without creating a
    public compatibility surface.  This small access-label check is the
    narrow parser-free approximation; it only treats an explicit access label
    immediately preceding the declaration as authoritative.
    """

    source = ROOT / item.path
    if not source.is_file() or item.line <= 0:
        return False
    prefix = source.read_text(encoding="utf-8", errors="replace").splitlines()[: item.line - 1]
    # Remove comments while retaining line breaks.  The small brace scanner
    # below tracks class scopes instead of taking the last access label in the
    # whole file (which would misclassify a later public class after an earlier
    # private section).
    text = "\n".join(prefix)
    text = re.sub(r"/\*.*?\*/", lambda match: "\n" * match.group(0).count("\n"), text, flags=re.DOTALL)
    text = re.sub(r"//[^\n]*", "", text)

    scopes: list[tuple[bool, str]] = []
    pending_class: str | None = None
    for line in text.splitlines():
        class_match = re.search(r"\b(class|struct)\s+[A-Za-z_]\w*[^;{]*", line)
        if class_match:
            pending_class = "public" if class_match.group(1) == "struct" else "private"
        access_match = re.match(r"\s*(public|protected|private)\s*:", line)
        if access_match and scopes and scopes[-1][0]:
            scopes[-1] = (True, access_match.group(1))
        for character in line:
            if character == "{":
                if pending_class is not None:
                    scopes.append((True, pending_class))
                    pending_class = None
                else:
                    scopes.append((False, ""))
            elif character == "}" and scopes:
                scopes.pop()
    return bool(scopes and scopes[-1][0] and scopes[-1][1] in {"private", "protected"})


def lint_api_shapes(lines: list[SourceLine]) -> list[Finding]:
    findings: list[Finding] = []
    for index, item in enumerate(lines):
        text = item.text
        context = _context(lines, index).lower()
        # Reads/assignments in an implementation are often the compatibility
        # facade's internal plumbing.  The dangerous API shape is declaring a
        # second error channel in a public header; keep this check high-signal.
        if re.search(
            r"\b(?:bool|Status|int)\s+[A-Za-z_]\w*\s*\([^;{}]*"
            r"(?:lastError|last_error|\b(?:err|error)\b)\s*\*",
            text,
            re.IGNORECASE,
        ):
            findings.append(
                Finding(
                    "api-shape",
                    "bool-error-pointer-channel",
                    item.path,
                    item.line,
                    "operation combines a scalar result with an error pointer; return structured Result/Diagnostic",
                )
            )
        declares_last_error = item.path.endswith((".h", ".hpp")) and re.search(
            r"\b(?:lastError|last_error)(?:_|\b)\s*(?:[;=,)])", text
        )
        if declares_last_error:
            findings.append(
                Finding(
                    "api-shape",
                    "last-error-channel",
                    item.path,
                    item.line,
                    "new code mentions lastError; return structured Result/Diagnostic instead",
                )
            )
        # Definitions inside an anonymous namespace are implementation
        # predicates/helpers, not public API.  Public declarations live in
        # headers and are the safe, parser-free surface to lint here.
        match = BOOL_RETURN.match(text) if item.path.endswith((".h", ".hpp")) else None
        if match:
            name = match.group(1)
            if _is_non_public_declaration(item):
                continue
            lower_name = name.lower()
            predicate = _is_predicate_name(name) or lower_name in {
                "operator bool",
                "operator==",
                "operator!=",
            }
            compatibility = _is_compatibility_declaration(text, context)
            if not predicate and not compatibility:
                findings.append(
                    Finding(
                        "api-shape",
                        "ambiguous-operation-bool",
                        item.path,
                        item.line,
                        f"operation {name} returns bool; use a Result or named status enum",
                    )
                )
        if item.path.endswith((".h", ".hpp")) and "*" in text and "(" in text:
            if POINTER_API.search(text) and not re.search(
                r"@(?:ownership|lifetime|borrowed|owned|outlives|thread)",
                context,
                re.IGNORECASE,
            ):
                findings.append(
                    Finding(
                        "api-lifetime",
                        "undocumented-raw-pointer-api",
                        item.path,
                        item.line,
                        "public raw-pointer API needs Doxygen ownership and lifetime contract",
                    )
                )
    return findings


def contract_matches(path: str, rule: str, entries: Iterable[Mapping[str, Any]]) -> list[Mapping[str, Any]]:
    return [entry for entry in entries if entry.get("rule") == rule and path_matches(path, entry.get("scope", ""))]


def lint_contract_coverage(lines: list[SourceLine], metadata: Mapping[str, Any]) -> list[Finding]:
    """Require a catalogue entry for newly introduced contract surfaces.

    This is intentionally a high-signal coverage check.  The broad policy
    entries in the catalogue cover established conventions; concrete Link and
    ECS declarations must additionally be represented by a path-scoped entry.
    """

    entries = [entry for entry in metadata.get("entries", []) if isinstance(entry, Mapping)]
    findings: list[Finding] = []
    seen: set[tuple[str, str]] = set()
    for item in lines:
        text = item.text
        triggers: list[tuple[str, str]] = []
        if re.search(r"\b(?:struct|class)\s+[A-Za-z_]\w*Link\b|\busing\s+\w*Link\b", text):
            triggers.append(("link", "new Link declaration"))
        if re.search(r"\b(?:class|struct)\s+[A-Za-z_]\w*System\b", text):
            triggers.append(("ecs-system", "new System declaration"))
        if re.search(r"\b(?:SimulationStep|Rng|RNG|seedFor)\b", text):
            triggers.append(("time-rng", "injected time/RNG surface"))
        if re.search(r"\b(?:restore|Snapshot|schema_version|migrate)\b", text, re.IGNORECASE):
            triggers.append(("persistence", "persistent or restore surface"))
        if re.search(r"\b(?:Capability|Provider|Backend)\b", text):
            triggers.append(("backend-contract", "provider/backend surface"))
        for rule, description in triggers:
            key = (item.path, rule)
            if key in seen:
                continue
            seen.add(key)
            matches = contract_matches(item.path, rule, entries)
            if not matches:
                findings.append(
                    Finding(rule, "missing-contract-entry", item.path, item.line, f"{description} has no catalogue entry")
                )
    return findings


def render(findings: list[Finding], catalogue_errors: list[str], json_output: bool) -> int:
    if json_output:
        print(
            json.dumps(
                {"catalogue_errors": catalogue_errors, "findings": [finding.as_dict() for finding in findings]},
                ensure_ascii=False,
                indent=2,
            )
        )
    else:
        for error in catalogue_errors:
            print(f"FAIL catalogue: {error}")
        for finding in findings:
            print(f"FAIL {finding.rule} {finding.path}:{finding.line}: {finding.message}")
        if not catalogue_errors and not findings:
            print("architecture contracts OK: catalogue valid; no new high-signal violations")
    return 1 if catalogue_errors or findings else 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--metadata", type=Path, default=DEFAULT_METADATA)
    parser.add_argument("--base", help="git base for changed-only mode (default: CI base or HEAD)")
    parser.add_argument("--all", action="store_true", help="lint all src C/C++ lines instead of only additions")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)
    try:
        metadata = load_json(args.metadata)
    except ValueError as error:
        return render([], [str(error)], args.json)
    catalogue_errors = validate_catalogue(metadata)
    if not isinstance(metadata, Mapping):
        return render([], catalogue_errors, args.json)
    if args.all:
        lines = _all_lines()
    else:
        base = args.base
        if base is None:
            base = os.environ.get("EVENGINE_ARCHITECTURE_BASE")
        lines = _changed_lines(base or "HEAD")
    findings = lint_api_shapes(lines) + lint_contract_coverage(lines, metadata)
    return render(findings, catalogue_errors, args.json)


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