scripts.check_wiring

Reference › scripts › check_wiring

The wiring gate, ruling RUL-023: when a feature is requested, that means wiring it in, not leaving an unconnected file. The gate looks at the files a change adds and fails when an added piece is not connected. It reads the declared gates through rulings_lint, and tests/scripts/test_check_wiring.py imports it.

Invocation: main takes --repo-root (default the current directory), --rulings-dir (override of the rulings directory) and --json. run_check returns an exit code and a report.

diff_entries reads the change from git (_git, with DiffError on failure) and added_paths selects the added files. Four checks follow. W1 (check_w1_claude_hooks) requires an added Claude hook to be registered in the settings, using _iter_hook_commands. W2 (check_w2_scripts_hooks) requires an added script that works as a hook to be wired. W3 (check_w3_src_modules) requires an added source module to be imported by something, and _orphan_modules finds the modules nothing imports. W4 (check_w4_gate_scripts) requires every gate script declared in a ruling to be named in a runner. Each failure is a Violation.

At a glance

Class diagram

Dependency diagram

API

Classes

DiffError

Bases: Exception · Line 80

git diff --name-status master...HEAD could not be computed — not a git repository, git is unavailable, or master doesn't exist.

Violation (dataclass)

Decorators: dataclass · Line 86

Functions

Source

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

#!/usr/bin/env python3
"""Wiring gate — RUL-023 ("when I ask for a feature that means wiring the
damn thing not just code vomit", verbatim, 2026-07-27).

See ``~/workspace/knowledge-wiki/rulings/RUL-023-feature-means-wired.md`` for
the ruling text and the incident (in a single session three capabilities were
built and reported complete while inert: a git hook whose ``core.hooksPath``
pointed elsewhere, a ``.claude/hooks/`` script registered only in a
worktree's tracked ``settings.json``, and a gate no runner invoked). RUL-012
already defines *done* as wired+deployed; this mechanically checks it for the
artifact classes that are easiest to leave inert.

Scope: everything this gate flags is scoped to **the current branch's diff
against ``master``** (``git diff --name-status master...HEAD``), so it grades
what THIS change added, never the tree's accumulated history. No diff against
master -> exit 0, nothing to grade.

Checks (see the module docstring of each ``check_w*`` function for detail):

    W1  a file added under ``.claude/hooks/`` must be registered in
        ``.claude/settings.json``.
    W2  a file added under ``scripts/hooks/`` (other than the installer
        itself) must be referenced by ``scripts/hooks/install-git-hooks.sh``.
    W3  a new ``.py`` module added under ``src/`` must be TRANSITIVELY
        reachable from a production entrypoint — the same reachability BFS
        the AST toolkit's ``orphans`` query runs for AGENTS.md Hard Rule 8
        (``find_orphan_modules``: roots are first-party imports of top-level
        ``scripts/*.py``, ``pyproject.toml`` entry-points, ``__main__``-
        guarded modules, and ``orchestrata_ultimate.__main__``). An import
        from a file that is itself unreachable does not count — this does
        NOT hand-roll a second reachability engine.
    W4  a ``scripts/*.py`` file added in this diff that some ruling under
        RULINGS_DIR declares as its ``gate`` mechanism must be invoked by
        ``scripts/health_check.py``, a file under ``.claude/hooks/``, a file
        under ``scripts/hooks/``, or ``scripts/run_tests.py``.

RULINGS_DIR is read with ``os.environ.get(...)`` at call time (RUL-003) via
the SAME resolution + missing-directory SKIP behaviour as
``scripts/rulings_lint.py`` (imported from there, not reimplemented) — a
machine without the knowledge-wiki clone must not go false-red: W4 is
skipped, W1-W3 still run.

Exit 0 clean (or skipped, with a reason). Exit 1 on any violation, one line
per finding: check id, artifact path, and the exact wiring site it is
missing. Not a git repository, git unavailable, or no divergence from
master: exit 0 with a printed reason, never crash.

Usage:
    python scripts/check_wiring.py [--repo-root PATH] [--rulings-dir PATH] [--json]
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from dataclasses import asdict, dataclass
from pathlib import Path

# Repo root on sys.path so `scripts` (imported below) resolves regardless of
# how this file is invoked (direct `python scripts/check_wiring.py`, `-m`,
# or as a test import) — same pattern as scripts/refactor.py.
_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from scripts.rulings_lint import (  # noqa: E402
    _default_rulings_dir,
    _extract_enforcement_lines,
)

_RUL_ID_RE = re.compile(r"^(RUL-\d+)")
_ORPHAN_LINE_RE = re.compile(r"^  ([A-Za-z_][\w.]*)$")
_RUNNER_RELATIVE_PATHS = ("scripts/health_check.py", "scripts/run_tests.py")
_HOOK_DIRS = (".claude/hooks", "scripts/hooks")


class DiffError(Exception):
    """``git diff --name-status master...HEAD`` could not be computed —
    not a git repository, git is unavailable, or ``master`` doesn't exist."""


@dataclass
class Violation:
    check: str
    artifact: str
    message: str


def _git(repo_root: Path, *argv: str) -> str:
    """Run one git command, or raise :class:`DiffError`."""
    try:
        result = subprocess.run(
            ["git", *argv],
            cwd=repo_root,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise DiffError(str(exc)) from exc
    if result.returncode != 0:
        raise DiffError(result.stderr.strip() or f"git {argv[0]} failed")
    return result.stdout


def diff_entries(repo_root: Path) -> list[tuple[str, str]]:
    """``(status, path)`` for everything THIS BRANCH introduces over master.

    Graded against the MERGE BASE, so master's own history is never graded,
    and against the WORKING TREE, so work that has not been committed yet is
    graded too.

    That second half is the whole point. The previous implementation ran
    ``git diff --name-status master...HEAD`` — a commit-range diff. Agents in
    this repo are forbidden to commit, so a session's entire body of work sits
    uncommitted in the working tree with ``HEAD == master``; the range was
    empty and this gate reported ``SKIP — no diff against master`` and exited
    0. Measured 2026-08-12 on feat/session-262-gate-and-provenance: 147 files
    changed vs master, and the gate graded ZERO of them. A gate that
    self-disables exactly when there is work to grade is a false green.

    Three sources are unioned, later ones winning on conflict:

    1. ``<merge-base>..HEAD`` — commits the branch already carries.
    2. ``git diff --name-status <merge-base>`` — the working tree (staged and
       unstaged alike) against that same base.
    3. ``git ls-files --others --exclude-standard`` — brand-new untracked
       files, graded ``A``. Without these a newly written script is invisible.

    Raises ``DiffError`` when git cannot run at all (missing git, not a repo,
    no ``master`` ref) — the caller turns that into a SKIP, never a crash.
    """
    base = _git(repo_root, "merge-base", "master", "HEAD").strip()
    if not base:
        raise DiffError("could not resolve merge-base of master and HEAD")

    # Ordered, so a later source overrides an earlier one for the same path
    # while the first-seen order is preserved for stable reporting.
    merged: dict[str, str] = {}

    def _absorb(raw: str) -> None:
        for line in raw.splitlines():
            if not line.strip():
                continue
            parts = line.split("\t")
            merged[parts[-1]] = parts[0]  # rename lines: STATUS OLD NEW

    _absorb(_git(repo_root, "diff", "--name-status", f"{base}..HEAD"))
    _absorb(_git(repo_root, "diff", "--name-status", base))

    for line in _git(
        repo_root, "ls-files", "--others", "--exclude-standard"
    ).splitlines():
        path = line.strip()
        if path:
            merged[path] = "A"

    return [(status, path) for path, status in merged.items()]


def added_paths(entries: list[tuple[str, str]]) -> list[str]:
    return [path for status, path in entries if status == "A"]


def _iter_hook_commands(settings: object) -> list[str]:
    """Every ``command`` string nested under the ``hooks`` table of a parsed
    ``.claude/settings.json`` — any event, any matcher, per RUL-023's W1."""
    commands: list[str] = []
    if not isinstance(settings, dict):
        return commands
    hooks_section = settings.get("hooks")
    if not isinstance(hooks_section, dict):
        return commands
    for entries in hooks_section.values():
        if not isinstance(entries, list):
            continue
        for entry in entries:
            if not isinstance(entry, dict):
                continue
            for h in entry.get("hooks", []):
                if isinstance(h, dict) and isinstance(h.get("command"), str):
                    commands.append(h["command"])
    return commands


def check_w1_claude_hooks(added: list[str], repo_root: Path) -> list[Violation]:
    """A file newly added under ``.claude/hooks/`` must be registered in
    ``.claude/settings.json`` (any event/matcher) — otherwise the live
    dispatcher never loads it, exactly the ``require-design-before-build.sh``
    incident RUL-023 names."""
    hook_files = [p for p in added if p.startswith(".claude/hooks/")]
    if not hook_files:
        return []
    settings_path = repo_root / ".claude" / "settings.json"
    commands: list[str] = []
    if settings_path.exists():
        try:
            commands = _iter_hook_commands(
                json.loads(settings_path.read_text(encoding="utf-8"))
            )
        except json.JSONDecodeError:
            commands = []
    violations = []
    for rel in hook_files:
        basename = Path(rel).name
        if not any(basename in cmd for cmd in commands):
            violations.append(
                Violation(
                    "W1",
                    rel,
                    f"{basename} must be registered under a PreToolUse/"
                    f"PostToolUse/Stop matcher in .claude/settings.json",
                )
            )
    return violations


def check_w2_scripts_hooks(added: list[str], repo_root: Path) -> list[Violation]:
    """A file newly added under ``scripts/hooks/`` (other than the installer
    itself) must be referenced by ``scripts/hooks/install-git-hooks.sh`` —
    otherwise running the installer never activates it."""
    hook_files = [
        p
        for p in added
        if p.startswith("scripts/hooks/") and Path(p).name != "install-git-hooks.sh"
    ]
    if not hook_files:
        return []
    installer = repo_root / "scripts" / "hooks" / "install-git-hooks.sh"
    text = installer.read_text(encoding="utf-8") if installer.exists() else ""
    violations = []
    for rel in hook_files:
        basename = Path(rel).name
        if basename not in text:
            violations.append(
                Violation(
                    "W2",
                    rel,
                    f"{basename} is not referenced by "
                    f"scripts/hooks/install-git-hooks.sh",
                )
            )
    return violations


def _module_dotted_name(rel_path: str) -> str | None:
    """``src/pkg/sub/mod.py`` -> ``pkg.sub.mod``. ``None`` for anything not a
    Python file under ``src/`` — INCLUDING ``__init__.py``: a package marker
    is never independently "imported" by name (``from pkg.sub.mod import X``
    does not import bare ``pkg.sub`` per ``who_imports``'s matching, even
    though Python's own import machinery loads it transitively). Its
    reachability is entirely a function of its sibling modules', which W3
    already checks — grading the marker file itself would false-positive on
    every brand-new package's ``__init__.py``."""
    if not (rel_path.startswith("src/") and rel_path.endswith(".py")):
        return None
    parts = rel_path[len("src/") :].split("/")
    if parts[-1] == "__init__.py":
        return None
    parts[-1] = parts[-1][: -len(".py")]
    if not parts or any(not p for p in parts):
        return None
    return ".".join(parts)


def _orphan_modules(repo_root: Path) -> set[str]:
    """Shell out to the sanctioned AST toolkit's ``orphans`` query — the SAME
    production-reachability BFS that gates AGENTS.md Hard Rule 8 elsewhere
    (``find_orphan_modules``: roots are first-party imports of top-level
    ``scripts/*.py``, ``pyproject.toml`` entry-points, ``__main__``-guarded
    modules, and ``orchestrata_ultimate.__main__``; reachability follows the
    first-party import graph with package-``__init__`` re-export semantics).
    W3 must not hand-roll a second reachability engine — reusing this one
    keeps "is X wired" answered identically everywhere in the project.

    One subprocess call for the WHOLE diff, not one per added module:
    ``orphans`` always recomputes the full graph regardless of scope, so
    batching amortises that cost across every module W3 needs to grade
    instead of paying it once per module. The AST tool is resolved next to
    THIS script (this checkout's own tooling), never under *repo_root* —
    *repo_root* is the tree being examined (real project root in
    production, a throwaway fixture in tests), which need not carry its own
    copy of the toolkit. ``--source`` is *repo_root*'s bare ``src/`` (not
    ``src/orchestrata_ultimate``): the real tree also carries sibling
    packages (``orchestrata_ast``, ``orchestrata_tools``) that
    :func:`_module_dotted_name` grades too, and a narrower root would drop
    them from the reachability graph entirely."""
    ast_tool = Path(__file__).resolve().parent / "serialize_codebase_ast.py"
    result = subprocess.run(
        [
            sys.executable,
            str(ast_tool),
            "orphans",
            "--source",
            str(repo_root / "src"),
            "--scripts",
            str(repo_root / "scripts"),
            "--pyproject",
            str(repo_root / "pyproject.toml"),
        ],
        capture_output=True,
        text=True,
        timeout=120,
    )
    orphans: set[str] = set()
    for line in result.stdout.splitlines():
        m = _ORPHAN_LINE_RE.match(line)
        if m:
            orphans.add(m.group(1))
    return orphans


def check_w3_src_modules(added: list[str], repo_root: Path) -> list[Violation]:
    """A newly added ``src/`` module must be reachable from at least one
    production entrypoint via the transitive first-party import graph — an
    import from a file that is itself unreachable does not count (that was
    the false-positive-prone proxy this replaced: "something outside my own
    package imports me" said nothing about whether that importer was ever
    itself wired in)."""
    modules = {rel: m for rel in added if (m := _module_dotted_name(rel))}
    if not modules:
        return []
    orphans = _orphan_modules(repo_root)
    violations = []
    for rel, module in modules.items():
        if module in orphans:
            violations.append(
                Violation(
                    "W3",
                    rel,
                    f"{module!r} is unreachable from every production "
                    f"entrypoint (scripts/*.py imports, pyproject.toml "
                    f"entry-points, __main__ guards) — no live path wires "
                    f"it in",
                )
            )
    return violations


def _gate_declarations(rulings_dir: Path) -> list[tuple[str, str, str]]:
    """``(ruling_id, script_rel_path, ruling_file)`` for every ``gate
    <script> <rule-id>`` enforcement directive across every ruling."""
    declarations = []
    for path in sorted(rulings_dir.glob("RUL-*.md")):
        m = _RUL_ID_RE.match(path.stem)
        if not m:
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        for directive in _extract_enforcement_lines(text):
            parts = directive.split(None, 2)
            if len(parts) >= 2 and parts[0] == "gate":
                declarations.append((m.group(1), parts[1], str(path)))
    return declarations


def _runner_texts(repo_root: Path) -> list[str]:
    texts = []
    for rel in _RUNNER_RELATIVE_PATHS:
        p = repo_root / rel
        if p.exists():
            texts.append(p.read_text(encoding="utf-8", errors="replace"))
    for hook_dir in _HOOK_DIRS:
        d = repo_root / hook_dir
        if d.is_dir():
            for f in sorted(d.iterdir()):
                if f.is_file():
                    texts.append(f.read_text(encoding="utf-8", errors="replace"))
    return texts


def check_w4_gate_scripts(
    added: list[str], repo_root: Path, rulings_dir: Path
) -> list[Violation]:
    """A ``scripts/*.py`` gate script that THIS diff adds, and that some
    ruling declares as its ``gate`` mechanism, must be invoked by at least
    one runner (health_check.py / .claude/hooks / scripts/hooks /
    run_tests.py) — otherwise nothing ever activates it."""
    added_set = set(added)
    declarations = [d for d in _gate_declarations(rulings_dir) if d[1] in added_set]
    if not declarations:
        return []
    texts = _runner_texts(repo_root)
    violations = []
    for ruling_id, script_rel, ruling_path in declarations:
        if not any(script_rel in text for text in texts):
            violations.append(
                Violation(
                    "W4",
                    script_rel,
                    f"{ruling_id} ({ruling_path}) declares this as its gate "
                    f"mechanism, but no runner invokes it",
                )
            )
    return violations


def run_check(repo_root: Path, rulings_dir: Path | None = None) -> tuple[int, dict]:
    """Injection seam: tests pass a throwaway tmp_path repo root (and,
    optionally, an explicit rulings directory) — never touches the live
    checkout."""
    try:
        entries = diff_entries(repo_root)
    except DiffError as exc:
        return 0, {
            "status": "skipped",
            "reason": f"could not compute diff against master at {repo_root}: {exc}",
        }
    if not entries:
        return 0, {"status": "skipped", "reason": "no diff against master"}
    added = added_paths(entries)

    rdir = rulings_dir if rulings_dir is not None else _default_rulings_dir()
    w4_skipped = not rdir.exists()

    violations = [
        *check_w1_claude_hooks(added, repo_root),
        *check_w2_scripts_hooks(added, repo_root),
        *check_w3_src_modules(added, repo_root),
    ]
    if not w4_skipped:
        violations.extend(check_w4_gate_scripts(added, repo_root, rdir))

    report = {
        "status": "pass" if not violations else "fail",
        "repo_root": str(repo_root),
        "added_files": added,
        "w4_skipped": w4_skipped,
        "violations": [asdict(v) for v in violations],
    }
    return (0 if not violations else 1), report


def _print_text_report(report: dict) -> None:
    if report["status"] == "skipped":
        print(f"[wiring] SKIP — {report['reason']}", flush=True)
        return
    if report["status"] == "pass":
        print("[wiring] OK — RUL-023 satisfied", flush=True)
        return
    print(f"[wiring] FAIL — {len(report['violations'])} violation(s):", flush=True)
    for v in report["violations"]:
        print(f"  [{v['check']}] {v['artifact']}: {v['message']}", flush=True)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--repo-root", default=None, help="Directory to gate (default: cwd)"
    )
    parser.add_argument("--rulings-dir", default=None, help="Override RULINGS_DIR")
    parser.add_argument("--json", action="store_true", help="Machine-readable output")
    args = parser.parse_args(argv)

    repo_root = Path(args.repo_root).resolve() if args.repo_root else Path.cwd()
    rulings_dir = Path(args.rulings_dir).expanduser() if args.rulings_dir else None
    exit_code, report = run_check(repo_root, rulings_dir=rulings_dir)

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