settled-reach/tooling/test_conformance.py

#!/usr/bin/env python3
"""The D-263 invariants, as a test rather than a style guide (T-1270).

D-263 lists the rules that make the layering real and then says plainly that a
contract nothing checks is a style guide. This is the check.

It matters most *later*. With one domain, every rule here is obvious and nobody
would break one. Once T-1250 lands ~120 commands, nobody re-reads a decision
record before adding a verb — this is what tells them, at the moment it is cheap
to fix rather than after the pattern has been copied forty times.

Invariants:

  1. Transport stays out of the logic layers. No typer/click import in any
     service.py, schemas.py or helpers.py — ever. Transport lives in main.py,
     router.py, and core/cli.py (the single designated transport module).
  2. Nothing prints but console. No bare print()/sys.stdout.write outside
     core/console.py — two output paths drift, and the second is always the one
     that ends up unformatted on stdout inside a hook.
  3. Every registered command carries @command, so no command can have logging
     without error handling or vice versa.
  4. Every command has help at its own level.
  5. Every ReachError names a remedy — a `fix=` on every raise site. The hardest
     to enforce and the most valuable: an error that only says "no" is the thing
     D-263 exists to replace.

The import-graph invariant (nothing heavy reachable from main.py) lives in
test_lazy_domains.py rather than being duplicated here.

Run: python3 tooling/test_conformance.py
"""

import ast
import shutil
import subprocess
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
PACKAGE = REPO_ROOT / "tooling"

# Files permitted to import the CLI framework. See core/cli.py for why it is on
# the list: a shared group class is transport by definition, and the alternative
# is a copy of it in every router that drifts.
TRANSPORT_FILES = {"main.py", "router.py", "cli.py"}
LOGIC_FILES = {"service.py", "schemas.py", "helpers.py", "dependencies.py"}
FORBIDDEN_IMPORTS = {"typer", "click"}


# The invariants govern the PACKAGE, not the legacy tree. The ~123 loose scripts
# under tooling/ predate all of this and use bare print() throughout; holding
# them to a contract they were never written against would mean 500 failures on
# day one and a suite nobody runs.
#
# This is not a permanent carve-out. As T-1250 moves each script into
# domains/<name>/, it lands inside this scope and the invariants start applying
# automatically — so the test's reach grows with the migration rather than
# needing to be widened by hand.
PACKAGE_ROOTS = ("main.py", "__init__.py", "core", "domains")


def _package_files() -> list[Path]:
    """Every .py that is part of the reach package — not the legacy scripts."""
    files: list[Path] = []
    for entry in PACKAGE_ROOTS:
        target = PACKAGE / entry
        if target.is_dir():
            files.extend(target.rglob("*.py"))
        elif target.is_file():
            files.append(target)
    return sorted(path for path in files if not path.name.startswith("test_"))


def _imports(tree: ast.AST) -> set[str]:
    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) and node.module and node.level == 0:
            found.add(node.module.split(".")[0])
    return found


def check_transport_isolation(failures: list[str]) -> None:
    """(1) No typer/click outside the designated transport files."""
    for path in _package_files():
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        offending = _imports(tree) & FORBIDDEN_IMPORTS
        if not offending:
            continue
        if path.name in TRANSPORT_FILES:
            continue
        rel = path.relative_to(REPO_ROOT)
        failures.append(
            f"[transport] {rel} imports {', '.join(sorted(offending))} — "
            f"only {', '.join(sorted(TRANSPORT_FILES))} may. "
            "A service must not know it was called from a CLI."
        )
        if path.name in LOGIC_FILES:
            failures.append(
                f"[transport] {rel} is a LOGIC file — this is the invariant that "
                "makes services callable from tests and from each other"
            )


def check_jobs_stay_ambient(failures: list[str]) -> None:
    """No domain imports core.jobs — job identity is the decorator's business.

    The whole point of carrying the job ambiently is that a command never has to
    open one, tag one, or remember to close one. The moment a domain imports
    `core.jobs`, that has become call-site discipline again, and it will fail the
    same way: one command forgets and its output loses correlation silently.

    Exempt: core/ (command.py and console.py implement the ambience) and
    main.py. main.py is not a command — it is the entry point, and it already
    owns the invocation-level concerns --detach, --verbose and --no-input.
    Recording a detached child's completion at the PROCESS's exit belongs there
    for the same reason, and is far less coupled than the --detach flag it
    already carries.
    """
    for path in _package_files():
        if path.parent.name == "core" or path.name == "main.py":
            continue
        source = path.read_text(encoding="utf-8")
        tree = ast.parse(source, filename=str(path))
        for node in ast.walk(tree):
            # Three forms reach the same module and all three must be caught:
            #   from tooling.core import jobs        -> module, names
            #   from tooling.core.jobs import ...    -> module
            #   import tooling.core.jobs             -> names
            hit = False
            if isinstance(node, ast.ImportFrom) and node.module:
                hit = node.module == "tooling.core.jobs" or (
                    node.module == "tooling.core"
                    and any(alias.name == "jobs" for alias in node.names)
                )
            elif isinstance(node, ast.Import):
                hit = any(alias.name == "tooling.core.jobs" for alias in node.names)
            if hit:
                failures.append(
                    f"[jobs] {path.relative_to(REPO_ROOT)}:{node.lineno} imports core.jobs — "
                    "job identity is ambient and belongs to @command; a command that "
                    "touches it has reintroduced the call-site discipline this replaced"
                )


def check_one_guarded_exec(failures: list[str]) -> None:
    """Only core/process.py may touch subprocess or os.system.

    A guarded exec is the right answer for genuine OS work (D-263) — but the
    guards only hold if there is one door. Every per-domain `subprocess.run` is
    a place where the argv-list rule, the missing-binary message, or turning a
    non-zero exit into a remedy quietly goes missing, and none of those
    omissions announces itself.

    core/process.py is exempt because it IS the door. Test scripts are outside
    the package scope already.
    """
    for path in _package_files():
        if path.name == "process.py" and path.parent.name == "core":
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        if "subprocess" in _imports(tree):
            failures.append(
                f"[exec] {path.relative_to(REPO_ROOT)} imports subprocess — go through "
                "core/process.run, which is where the argv-list rule, the "
                "missing-binary message and the failure remedy live"
            )
        for node in ast.walk(tree):
            # The receiver has to be checked, not just the attribute name:
            # `platform.system()` is a legitimate call and shares a name with
            # `os.system()`. A check that fires on the wrong thing gets muted.
            if (
                isinstance(node, ast.Call)
                and isinstance(node.func, ast.Attribute)
                and isinstance(node.func.value, ast.Name)
                and node.func.value.id == "os"
                and node.func.attr in {"system", "popen", "execv", "execvp"}
            ):
                failures.append(
                    f"[exec] {path.relative_to(REPO_ROOT)}:{node.lineno} calls "
                    f"os.{node.func.attr} — unguarded, and shell-interpreting in the "
                    "case of system()"
                )


def check_single_output_path(failures: list[str]) -> None:
    """(2) Nothing prints but console."""
    for path in _package_files():
        if path.name == "console.py":
            continue
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if isinstance(func, ast.Name) and func.id == "print":
                failures.append(
                    f"[output] {path.relative_to(REPO_ROOT)}:{node.lineno} calls print() — "
                    "core/console.py is the single output path (D-263)"
                )
            elif (
                isinstance(func, ast.Attribute)
                and func.attr == "write"
                and isinstance(func.value, ast.Attribute)
                and func.value.attr in {"stdout", "stderr"}
            ):
                failures.append(
                    f"[output] {path.relative_to(REPO_ROOT)}:{node.lineno} writes to "
                    "sys.stdout/stderr directly — go through core/console.py"
                )


def check_commands_decorated(failures: list[str]) -> None:
    """(3) and (4): every registered command carries @command and has help."""
    probe = """
import json, sys
from tooling.core.command import MARKER
from tooling.main import DOMAINS, _load_domain

report = []
for name in sorted(DOMAINS):
    group = _load_domain(name)
    ctx = None
    for verb in group.list_commands(ctx):
        cmd = group.get_command(ctx, verb)
        callback = getattr(cmd, "callback", None)
        report.append({
            "domain": name,
            "verb": verb,
            "decorated": bool(getattr(callback, MARKER, False)),
            "help": (cmd.help or cmd.short_help or "").strip(),
        })
print(json.dumps(report))
"""
    result = subprocess.run(
        [sys.executable, "-c", probe], capture_output=True, text=True, cwd=REPO_ROOT
    )
    if result.returncode != 0:
        failures.append(f"[commands] could not introspect the CLI:\n{result.stderr}")
        return

    import json

    report = json.loads(result.stdout)
    if not report:
        failures.append(
            "[commands] no commands found — every assertion here would pass vacuously"
        )
    for entry in report:
        where = f"{entry['domain']} {entry['verb']}"
        if not entry["decorated"]:
            failures.append(
                f"[commands] `reach {where}` is missing @command — it would run "
                "without the error contract or the invocation record"
            )
        if not entry["help"]:
            failures.append(f"[commands] `reach {where}` has no help text")


def check_errors_name_a_remedy(failures: list[str]) -> None:
    """(5) Every ReachError raise site passes fix=."""
    for path in _package_files():
        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Raise) or not isinstance(node.exc, ast.Call):
                continue
            func = node.exc.func
            name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "")
            if name != "ReachError":
                continue
            if not any(kw.arg == "fix" for kw in node.exc.keywords):
                failures.append(
                    f"[remedy] {path.relative_to(REPO_ROOT)}:{node.lineno} raises "
                    "ReachError without fix= — an error that only says 'no' is "
                    "what D-263 exists to replace"
                )


def check_exit_codes(failures: list[str]) -> None:
    """(6) Discovery succeeds; being wrong fails.

    The distinction is easy to break in either direction, and both directions
    are bad in ways nothing else would catch. Make discovery a usage error and
    an agent reads its own onboarding as a failure. Make a wrong name succeed
    and a typo in a hook passes silently — the exit-0 trap D-263 opens with.
    """
    cases = [
        ([], 0, "bare `reach` is discovery, not a usage error"),
        (["check"], 0, "bare `reach <domain>` is discovery, not a usage error"),
        (["--help"], 0, "--help succeeds"),
        (["definitely-not-a-domain"], 2, "an unknown domain is a usage error"),
        (["check", "definitely-not-a-verb"], 2, "an unknown verb is a usage error"),
    ]
    for args, expected, why in cases:
        result = subprocess.run(
            ["reach", *args], capture_output=True, text=True, cwd=REPO_ROOT
        )
        if result.returncode != expected:
            failures.append(
                f"[exit] `reach {' '.join(args)}` exited {result.returncode}, "
                f"expected {expected} — {why}"
            )


def main() -> int:
    if shutil.which("reach") is None:
        print("test_conformance: `reach` is not on PATH.\n  Fix: make install-reach", file=sys.stderr)
        return 1

    failures: list[str] = []
    check_transport_isolation(failures)
    check_jobs_stay_ambient(failures)
    check_one_guarded_exec(failures)
    check_single_output_path(failures)
    check_commands_decorated(failures)
    check_errors_name_a_remedy(failures)
    check_exit_codes(failures)

    if failures:
        print("test_conformance: FAIL", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1

    print("test_conformance: OK — transport isolated, one output path, "
          "every command decorated and helped, every error names a remedy")
    return 0


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