scripts.health_check

Reference › scripts › health_check

scripts/health_check.py is the project health check that must pass before any bulk change or commit. A human runs it, and with --ci a CI job runs it; nothing imports it. It aggregates many independent checks into one verdict and one exit code.

Options: --fast runs only the fast checks, --ci adds pytest, and --fix applies ruff auto-fixes. build_checks assembles the list from those flags. The checks are small functions that each run a subprocess through _run (with a timeout, default 120 seconds) and return a pass flag and error lines: check_ruff, check_syntax (every .py file compiles), check_architecture (runs architecture_checker), check_orphan_modules (the orphan-module ratchet, AGENTS.md Hard Rule 8), check_rulings_lint (see rulings_lint), check_worktree_envs (RUL-021, see check_worktree_envs), check_wiring (RUL-023, see check_wiring), check_ai_attribution (RUL-007, over the whole reachable history), check_native_build_freshness, check_incus_not_docker (RUL-004) and check_env_contract (RUL-019), plus others further down the module.

main runs the checks concurrently in a ThreadPoolExecutor of at most eight workers, since they are independent subprocesses; wall-clock time is the longest check rather than the sum. Progress lines print as each starts, results are collected as they finish, and an exception in a check is reported as an error and counts as a failure. It exits non-zero when any check fails.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

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

#!/usr/bin/env python3
"""Project health check — mandatory before any bulk change or commit.

This script is the single entry point for all validation. It runs
fast checks first, then slower ones, and exits non-zero on any failure.

Usage:
    python scripts/health_check.py          # Full check
    python scripts/health_check.py --fast   # Only fast checks (ruff, syntax)
    python scripts/health_check.py --ci     # CI mode (includes pytest)

Enforcement:
    - The pre-commit hook runs this automatically and BLOCKS commits on failure.
    - The session pre-close hook requires this to pass before ending a session
      with >10 files changed.
    - AGENTS.md mandates running this before any bulk change (>10 files).
"""

from __future__ import annotations

import argparse
import ast
import os
import subprocess
import sys
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path


def _run(
    cmd: list[str], cwd: Path | None = None, timeout: int = 120
) -> tuple[int, str, str]:
    """Run a subprocess and return (returncode, stdout, stderr)."""
    try:
        result = subprocess.run(
            cmd,
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
        return result.returncode, result.stdout, result.stderr
    except FileNotFoundError as e:
        return 127, "", str(e)
    except subprocess.TimeoutExpired:
        return 124, "", f"Command timed out after {timeout}s"


def check_ruff(fix: bool = False) -> tuple[bool, list[str]]:
    """Run ruff check."""
    cmd = ["uv", "run", "ruff", "check", "src/", "tests/"]
    if fix:
        cmd.append("--fix")
    rc, out, err = _run(cmd)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["RUFF: " + line for line in lines if line.strip()]


def check_syntax() -> tuple[bool, list[str]]:
    """Verify all .py files compile."""
    errors: list[str] = []
    for path in Path("src").rglob("*.py"):
        try:
            ast.parse(path.read_text())
        except SyntaxError as e:
            errors.append(f"SYNTAX: {path}:{e.lineno}: {e.msg}")
    for path in Path("tests").rglob("*.py"):
        try:
            ast.parse(path.read_text())
        except SyntaxError as e:
            errors.append(f"SYNTAX: {path}:{e.lineno}: {e.msg}")
    return not errors, errors


def check_architecture() -> tuple[bool, list[str]]:
    """Run architecture checker."""
    rc, out, err = _run([sys.executable, "scripts/architecture_checker.py"])
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["ARCH: " + line for line in lines if line.strip()]


def check_orphan_modules() -> tuple[bool, list[str]]:
    """Run the orphan-module ratchet (AGENTS.md Hard Rule 8): fail the check
    when a NEW src module has become production-unreachable — nothing outside
    tests imports it, transitively, from any production entrypoint. The
    committed ``plans/orphan-modules-baseline.txt`` is the known unwired-debt
    ledger: growth fails, shrink is advisory (see ``orphans --check`` in
    scripts/serialize_codebase_ast.py).
    """
    rc, out, err = _run(
        [
            sys.executable,
            "scripts/serialize_codebase_ast.py",
            "orphans",
            "--source",
            "src/orchestrata_ultimate",
            "--baseline",
            "plans/orphan-modules-baseline.txt",
            "--check",
        ]
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["ORPHAN: " + line for line in lines if line.strip()]


def check_rulings_lint() -> tuple[bool, list[str]]:
    """Run the rulings-enforcement lint (plans/workflow-improvements-plan.md
    Phase 2, RUL-014): every ruling declared in the git-backed wiki rulings
    canon must resolve to a real mechanism (hook/gate/banned-deps/
    constitution) or an explicit ``none (judgment)`` mark.

    SKIPS (does not fail health_check) when RULINGS_DIR /
    ~/workspace/knowledge-wiki/rulings is absent, so a machine without the
    knowledge-wiki checkout never goes false-red on this check.
    """
    rc, out, err = _run([sys.executable, "scripts/rulings_lint.py"], timeout=30)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["RULINGS-LINT: " + line for line in lines if line.strip()]


def check_worktree_envs() -> tuple[bool, list[str]]:
    """Run the RUL-021 worktree-environment gate: every non-primary worktree
    must own a real (non-symlinked) ``.venv``, resolve its own relative
    ``vendor/`` ``[tool.uv.sources]`` pins, and not still exist once its
    branch has merged into master (see scripts/check_worktree_envs.py)."""
    rc, out, err = _run([sys.executable, "scripts/check_worktree_envs.py"], timeout=30)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["WORKTREE-ENV: " + line for line in lines if line.strip()]


def check_wiring() -> tuple[bool, list[str]]:
    """Run the RUL-023 wiring gate: a branch that adds something must also
    activate it — a ``.claude/hooks/`` file registered in
    ``.claude/settings.json``, a ``scripts/hooks/`` file referenced by the
    installer, a new ``src/`` module imported from outside its own package,
    and any newly added gate script a ruling declares as its mechanism
    actually invoked by a runner (see scripts/check_wiring.py)."""
    rc, out, err = _run([sys.executable, "scripts/check_wiring.py"], timeout=60)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["WIRING: " + line for line in lines if line.strip()]


def check_ai_attribution() -> tuple[bool, list[str]]:
    """Run the RUL-007 AI-attribution gate over the whole reachable history:
    no ``Co-Authored-By: Claude ...``/``Claude-Session:``/"Generated with
    Claude Code" marker may appear in any commit message or annotated tag.
    Every commit and tag carries Mohamed's identity only. Findings pinned in
    ``scripts/rul007-baseline.txt`` (published history that cannot be
    rewritten) are printed but do not fail (see
    scripts/check_ai_attribution.py)."""
    rc, out, err = _run(
        [
            sys.executable,
            "scripts/check_ai_attribution.py",
            "--range",
            "--all",
            "--tags",
        ],
        timeout=60,
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["AI-ATTRIBUTION: " + line for line in lines if line.strip()]


def check_native_build_freshness() -> tuple[bool, list[str]]:
    """Run the native-build freshness gate: every vendored C extension's
    installed ``.so`` must be at least as new as its sources. uv keys its build
    cache on package VERSION, and all four vendored extensions are pinned at a
    version that never changes, so a reinstall can serve a previously-built
    wheel — shipping a stale binary with a FRESH mtime while the source shows
    the fix. Measured 2026-08-05 on the cxpt ``Server_dealloc`` fix (see
    scripts/check_native_build_freshness.py)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_native_build_freshness.py"], timeout=60
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["NATIVE-BUILD: " + line for line in lines if line.strip()]


def check_incus_not_docker() -> tuple[bool, list[str]]:
    """Run the RUL-004 container-runtime gate: deployment and test isolation
    target Incus (LXD), never Docker. Blocks a shell/systemd docker
    invocation, a Python subprocess/import of docker, a docker endpoint in
    config/env, and a new docker-compose file outside the allowlisted
    reference trees (see scripts/check_incus_not_docker.py)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_incus_not_docker.py"], timeout=60
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["INCUS-NOT-DOCKER: " + line for line in lines if line.strip()]


def check_env_contract() -> tuple[bool, list[str]]:
    """Run the RUL-019 env-contract gate: every entrypoint that reaches a
    required-infra-key seam must also reach the canonical env loader
    (secrets-only ``.env`` first, then the ``ORCHESTRATA_ENV_PROFILE``
    overlay ``env/{dev,test,prod}.env`` with ``override=True``), and no
    second hand-rolled mutating ``.env`` loader may exist (see
    scripts/check_env_contract.py)."""
    rc, out, err = _run([sys.executable, "scripts/check_env_contract.py"], timeout=60)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["ENV-CONTRACT: " + line for line in lines if line.strip()]


def check_generated_artifacts() -> tuple[bool, list[str]]:
    """Run the RUL-030 gate: every generated artifact is tracked by git.

    A generated artifact is a snapshot of project state at a commit, so an
    artifact that is gitignored or present-but-untracked makes the commit
    stop describing the project (see scripts/check_generated_artifacts.py
    for the derived/read/function-of-the-commit boundary). Exit 2 means the
    gate could not do its job -- an empty or unresolvable registry -- and is
    surfaced as a failure rather than folded into a pass.
    """
    rc, out, err = _run(
        [sys.executable, "scripts/check_generated_artifacts.py"], timeout=60
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    prefix = "GENERATED-ARTIFACTS-INVALID: " if rc == 2 else "GENERATED-ARTIFACTS: "
    return False, [prefix + line for line in lines if line.strip()]


def check_requirement_corpus() -> tuple[bool, list[str]]:
    """Run the requirement-corpus gate: no src/ @traces cites an ungradeable id.

    Two readers of one corpus drifted apart -- the AST toolkit scanned
    checklist + spec.md + addendum while the RTM generator scanned checklists
    only -- so ``trace-lint`` called 21 citations clean while the RTM graded
    them GHOST, and 68 requirement ids never entered the matrix's denominator.
    This asserts the two agree, measured against @traces decorators harvested
    from the AST, which can genuinely disagree with either side. Exit 2 means
    the gate examined nothing and is surfaced as a failure, never a pass.
    """
    rc, out, err = _run(
        [sys.executable, "scripts/check_requirement_corpus.py"], timeout=180
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    prefix = "REQUIREMENT-CORPUS-INVALID: " if rc == 2 else "REQUIREMENT-CORPUS: "
    return False, [prefix + line for line in lines if line.strip()]


def check_graph_write_swallow() -> tuple[bool, list[str]]:
    """Run the graph-write-swallow gate: a graph write inside a ``try`` body
    (``execute_write``/``unit_of_work``/``execute_autocommit``/``session.run``,
    or a MERGE/CREATE/DETACH DELETE/SET Cypher literal) whose ``except``
    handler cannot inform the caller of failure — no raise, no return, no
    outliving assignment/mutation — makes contention on the shared Memgraph
    indistinguishable from a bug. Ratcheted against
    ``reports/graph-write-swallow-baseline.json``; only NEW findings fail
    (see scripts/check_graph_write_swallow.py)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_graph_write_swallow.py", "--check"], timeout=60
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["GRAPH-WRITE-SWALLOW: " + line for line in lines if line.strip()]


def check_kimi_agent_level() -> tuple[bool, list[str]]:
    """Run the RUL-018 agent-level-identity gate: kimi's IRC presence is the
    single systemd USER unit ``kimi-irc-bot.service`` — this repo must never
    start a project-level kimi listener, and must never bind ``IRC_NICK`` to
    another agent's identity (see scripts/check_kimi_agent_level.py)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_kimi_agent_level.py"], timeout=30
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["KIMI-AGENT-LEVEL: " + line for line in lines if line.strip()]


def check_spec_authority() -> tuple[bool, list[str]]:
    """Run the RUL-013 spec-authority gate: a requirement id cited by a test
    must be defined by a spec under ``specs/`` — a test may never invent an
    id to justify its own behaviour.

    Pinned at ``--max 0``: the gate landed with a backlog of 8 ungrounded
    ids (the ``REQ-003-FR24-100..106`` family, cited by end-to-end pipeline
    tests before anyone wrote them down), and that backlog was resolved the
    RUL-013 way — by writing the requirements into
    ``specs/003-intent/checklists/requirements.md`` rather than re-pointing
    the tests at unrelated ids. There is no remaining backlog to tolerate,
    so the ceiling stays 0 and the ratchet cannot silently re-widen
    (see scripts/check_spec_authority.py)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_spec_authority.py", "--max", "0"], timeout=120
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["SPEC-AUTHORITY: " + line for line in lines if line.strip()]


def check_validate_ci() -> tuple[bool, list[str]]:
    """Run CI validation."""
    rc, out, err = _run([sys.executable, "scripts/validate_ci.py"])
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["CI: " + line for line in lines if line.strip()]


def check_test_purity() -> tuple[bool, list[str]]:
    """Run the test-purity audit (RUL-002: no mocks / doubles / monkeypatch).

    Graded on the AUDIT'S EXIT CODE, not on grepping its output for the words
    "ERROR"/"FAIL". Measured 2026-08-05: a planted ``MagicMock()`` makes
    ``audit_test_purity.py`` exit 1 and print "SUMMARY: 1 files with 1 total
    violations" — a report that contains NEITHER word, so the old
    substring-grep produced an empty error list and this check returned PASS
    while the audit was failing. That silently disarmed the repo's most
    fundamental test ruling inside its own health check.
    """
    rc, out, err = _run([sys.executable, "scripts/audit_test_purity.py"])
    if rc == 0:
        return True, []
    lines = [line for line in (out + err).splitlines() if line.strip()]
    # Keep the violation detail: per-file "Line N:" citations and the summary.
    detail = [
        line
        for line in lines
        if "violation" in line.lower() or "Line " in line or ".py" in line
    ]
    return False, ["PURITY: " + line for line in (detail or lines)[-25:]]


def check_prolog_boundary() -> tuple[bool, list[str]]:
    """Run the Prolog boundary gate — every Python Prolog query site must
    target a predicate/arity that exists in the consulted rule base
    (reports/proposal-wire-prolog-governance.md §7 Phase 1)."""
    rc, out, err = _run(
        [sys.executable, "scripts/check_prolog_boundary.py"], timeout=90
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["PROLOG-BOUNDARY: " + line for line in lines if line.strip()]


def check_prolog_meta_lint() -> tuple[bool, list[str]]:
    """Run the Prolog governance-rulebase meta-lint — SPEC-033 §9 (task B17):
    ADR-0025's construct ban (no cut / no dynamic call / no rule-body side
    effects) and ADR-0010's stratification check over the governance
    rulebase (constitutional.pl, agents.pl, cxp.pl, multiagent.pl, ethics.pl,
    governance.pl, knowledge.pl, completion.pl, intent_policy.pl).

    NOTE (2026-07-24): as of this wiring, the governance rulebase carries 19
    known, tracked violations pending a separate remediation lane (Lane 3) —
    15 cuts the Lane-0 inventory marked needs-rewrite/safe-remove, plus 4
    real assert/retract-in-rule-body sites this lint independently found
    (constitutional.pl:81-82's human_override/3; cxp.pl:410,414's
    register_session/3 / unregister_session/1) that the given inventory did
    not enumerate — the latter sits in constitutional.pl, which no lane may
    modify in this task, so it needs an orchestrator/human decision (ADR
    exception vs. restructure vs. accepted debt). Until that remediation
    lands, THIS GATE FAILS on every full (non-``--fast``) health_check run,
    matching the brief's explicit "failure fails health_check" instruction
    rather than being softened to advisory (contrast ``check_kg_schema_
    advisory`` below, which IS deliberately non-blocking) — flagged loudly
    here rather than silently decided.
    """
    rc, out, err = _run([sys.executable, "scripts/prolog_meta_lint.py"], timeout=60)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["PROLOG-META-LINT: " + line for line in lines if line.strip()]


def check_kg_schema_advisory() -> tuple[bool, list[str]]:
    """Canonical KG schema conformance — ADVISORY, never fails the run.

    CI already runs ``kg_schema.py check`` REQUIRED (ci.yml); this surfaces
    live-graph drift during local health checks without blocking (ratified
    O-2, ADR-023 — pre-commit is permanently out, CI is the hard gate).
    """
    rc, out, err = _run([sys.executable, "scripts/kg_schema.py", "check"], timeout=120)
    if rc != 0:
        lines = [ln for ln in (out + err).splitlines() if ln.strip()][:10]
        print("\n  ⚠ ADVISORY (non-blocking) — kg_schema check reports drift:")
        for ln in lines:
            print(f"    {ln}")
        print("  (CI enforces this as a required gate; fix before pushing.)")
    return True, []


def check_kg_connectivity() -> tuple[bool, list[str]]:
    """Every node in the operational KG must be reachable from a root
    ``:Project`` node (session-254 ruling). REQUIRED, not advisory (contrast
    ``check_kg_schema_advisory`` above) -- its failure marks health_check
    FAILED. As of the session-254 WCC census this gate is EXPECTED to fail
    on the live graph (2,654 disconnected islands); the writer-side fixes
    landing alongside this wiring stop the bleeding, the full data repair is
    a separate, later effort.
    """
    rc, out, err = _run(
        [sys.executable, "scripts/check_kg_connectivity.py"],
        timeout=120,
    )
    if rc == 0:
        return True, []
    lines = [ln for ln in (out + err).splitlines() if ln.strip()]
    return False, ["KG-CONNECTIVITY: " + line for line in lines]


def check_adr_graph_sync() -> tuple[bool, list[str]]:
    """Read-only three-way drift check (repo docs/adr/ / wiki decisions.md /
    live ``ArchitectureDecision`` nodes) — ``scripts/check_adr_graph_sync.py``.
    REQUIRED, not advisory, same reasoning as ``check_kg_connectivity`` above
    (a live-infra check that can legitimately stay red pending a human
    decision does not belong in the fast pre-commit / ``scripts/gates.py``
    battery, which would then block every relevant commit on a known,
    pending issue -- it belongs here, in the mandatory-but-slow report).

    2026-08-19 audit (session finding, not yet reconciled): ``docs/adr/`` and
    the knowledge-wiki's ``decisions.md`` are two independent ADR registers
    that number their entries the same way and disagree on the decision
    itself for every one of the 25 overlapping ids -- the live graph's
    ``ArchitectureDecision`` nodes carry no property distinguishing which
    register produced them. **This gate is EXPECTED to fail** until that id
    collision is resolved (a human decision -- this gate only detects it,
    per its own module docstring; it never writes or deletes anything).
    """
    rc, out, err = _run(
        [sys.executable, "scripts/check_adr_graph_sync.py"],
        timeout=60,
    )
    if rc == 0:
        return True, []
    lines = [ln for ln in (out + err).splitlines() if ln.strip()]
    return False, ["ADR-GRAPH-SYNC: " + line for line in lines]


def check_env_loaded() -> tuple[bool, list[str]]:
    """Check that required env vars are present."""
    import os

    required = [
        "REDIS_HOST",
        "REDIS_PORT",
        # KG_ROUTING_URI/KG_DATABASE, not NEO4J_* — retired 2026-07-27.
        "KG_ROUTING_URI",
        "KG_DATABASE",
    ]
    missing = [v for v in required if not os.environ.get(v)]
    if missing:
        return False, [
            f"ENV: Missing {', '.join(missing)} — run 'source .env' or "
            "ensure direnv is active"
        ]
    return True, []


def check_hardcoded_ips() -> tuple[bool, list[str]]:
    """Check for hardcoded PRODUCTION connection IPs in active Python code.

    Delegates to ``validate_ci._scan_hardcoded_prod_ip`` — the single
    line-level, connection-context scanner (a ``*_HOST``/``*_URI`` key or
    ``host=``/``uri=`` kwarg adjacent to a ``10.53.1.x`` literal), covering BOTH
    ``src/**`` and ``scripts/*``. This replaces the previous whole-file heuristic
    which (a) never scanned ``scripts/`` and (b) exempted any file containing
    ``os.environ`` ANYWHERE — i.e. every env-config file, making the check a
    no-op for exactly the prod-default bug it was meant to catch.
    """
    # A local import keeps the dependency scoped to this one check. scripts/ is
    # on sys.path when health_check runs AS a script, but NOT when a test imports
    # it as a module — so ensure the sibling directory is importable first
    # (fixes ModuleNotFoundError: No module named 'validate_ci' under pytest).
    # Scan the CWD tree (repo root in normal use; overridable by chdir in tests).
    import sys as _sys

    _scripts_dir = str(Path(__file__).resolve().parent)
    if _scripts_dir not in _sys.path:
        _sys.path.insert(0, _scripts_dir)
    from validate_ci import _scan_hardcoded_prod_ip

    hits = _scan_hardcoded_prod_ip(Path.cwd())
    errors = [f"IP: {hit}" for hit in hits]
    return not errors, errors


def check_infra_drift() -> tuple[bool, list[str]]:
    """Run the ADR-0077 infra drift gate (scripts/check_infra_drift.py).

    OpenTofu owns the Incus topology of all three environments and the live
    fleets are imported into the host-owned state at
    ``/var/lib/orchestrata/terraform/fleet.tfstate``. The gate runs ``tofu plan
    -detailed-exitcode`` over ``infra/terraform`` and proves the bridge subnets
    live with ``incus network get``. A MISSING state or binary is a FAIL,
    never a skip (ADR-0077 item 4); only a checkout with no ``infra/terraform``
    at all passes with a printed NOTE. The same script runs hourly from
    deploy/systemd/orchestrata-infra-drift.timer.
    """
    rc, out, err = _run([sys.executable, "scripts/check_infra_drift.py"], timeout=360)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["INFRA: " + line for line in lines if line.strip()]


def check_env_fleet_agreement() -> tuple[bool, list[str]]:
    """Run the env/fleet agreement gate (scripts/check_env_fleet_agreement.py).

    ``env/test.env``, ``env/dev.env`` and the ``TEST_*`` keys of ``.env.example``
    are DERIVED from ``tofu output -json`` (infra/gen_env.py). The gate fails on
    any disagreement, any commented-out owned key, and a missing tofu binary or
    state (ADR-0077 item 4). Measured 2026-10-08: ``.env.example`` carried a
    retired prod-bridge ``TEST_MEMGRAPH_URI`` for two months with nothing checking it.
    """
    rc, out, err = _run(
        [sys.executable, "scripts/check_env_fleet_agreement.py"], timeout=360
    )
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["ENV-FLEET: " + line for line in lines if line.strip()]


def check_systemd_units(installed_dir: Path | None = None) -> tuple[bool, list[str]]:
    """Fail if an installed user unit no longer matches its ``deploy/systemd``
    copy, or if a unit the repo declares is missing from the host.

    Those units are the live schedule this project runs on, not documentation
    — see ``deploy/systemd/README.md``. This gate closes both drift directions:
    an edited installed unit, and a declared unit that was never installed.
    Skips (passes with a note) on a host with no user units, so it is a no-op
    in CI/containers rather than a false failure.

    Args:
        installed_dir: Where installed user units live. Defaults to
            ``$XDG_CONFIG_HOME/systemd/user``; injectable so the gate is
            testable without touching the real host.
    """
    unit_dir = Path("deploy/systemd")
    if installed_dir is None:
        config_home = os.environ.get("XDG_CONFIG_HOME") or str(Path.home() / ".config")
        installed_dir = Path(config_home) / "systemd/user"
    if not unit_dir.is_dir():
        return True, ["SYSTEMD: skipped (deploy/systemd absent)"]
    if not installed_dir.is_dir():
        return True, ["SYSTEMD: skipped (no user unit directory on this host)"]

    errors: list[str] = []
    for declared in sorted(unit_dir.glob("*.service")) + sorted(
        unit_dir.glob("*.timer")
    ):
        installed = installed_dir / declared.name
        if not installed.exists():
            errors.append(
                f"SYSTEMD: {declared.name} is version-controlled but NOT "
                f"installed (run deploy/systemd/install.sh)"
            )
            continue
        if installed.read_text() != declared.read_text():
            errors.append(
                f"SYSTEMD DRIFT: {declared.name} differs from the installed "
                f"unit (diff {installed} {declared})"
            )
    return not errors, errors


def check_pytest(skip: bool = False) -> tuple[bool, list[str]]:
    """Run tests (non-Neo4j subset) via the sanctioned threadpool runner.

    pytest-xdist (``-n auto``) is BANNED in this project — it deadlocks the
    suite ~78%. Parallelism comes from scripts/run_tests_threadpool.py
    (ThreadPoolExecutor, one pytest process per module). Invoked via
    ``sys.executable`` so it always uses this interpreter's venv (with all
    plugins) regardless of cwd or whether VIRTUAL_ENV is exported.
    """
    if skip:
        return True, ["PYTEST: skipped (use --ci to run)"]
    runner = Path(__file__).resolve().parent / "run_tests_threadpool.py"
    cmd = [
        sys.executable,
        str(runner),
        "tests/",
        "--ignore=tests/knowledge_graph",
        "--ignore=tests/session/kg_sync",
        "--ignore=tests/gateway",
        "--ignore=tests/orchestrator/test_coverage_boost.py",
        "--ignore=tests/integration",
        "--ignore=tests/agents/volks/architect",
        "--pytest-arg=--timeout=120",
        "--pytest-arg=-q",
    ]
    rc, out, err = _run(cmd, timeout=300)
    if rc == 0:
        return True, []
    lines = (out + err).splitlines()
    return False, ["PYTEST: " + line for line in lines if line.strip()]


def build_checks(
    args: argparse.Namespace,
) -> list[tuple[str, Callable[[], tuple[bool, list[str]]]]]:
    """THE registry of health checks, in run order, for the selected mode.

    The base block always runs and includes every ruling-enforcement gate
    (RUL-004/007/013/018/019/021/023): a gate that exists but is never
    invoked is not a delivery (RUL-023), and this list is what invokes them.
    ``--fast`` omits the slow/infra block; ``--ci`` adds the pytest run.
    """
    checks: list[tuple[str, Callable[[], tuple[bool, list[str]]]]] = [
        ("Environment variables", check_env_loaded),
        ("Python syntax", check_syntax),
        ("Ruff lint", lambda: check_ruff(fix=args.fix)),
        ("Architecture rules", check_architecture),
        ("CI validation", check_validate_ci),
        ("Hardcoded IPs", check_hardcoded_ips),
        ("Orphan modules (Rule 8 wiring ratchet)", check_orphan_modules),
        ("Rulings lint", check_rulings_lint),
        ("Worktree envs (RUL-021)", check_worktree_envs),
        ("Wiring (RUL-023)", check_wiring),
        ("Native build freshness", check_native_build_freshness),
        ("Incus not Docker (RUL-004)", check_incus_not_docker),
        ("AI attribution (RUL-007)", check_ai_attribution),
        ("Spec authority (RUL-013)", check_spec_authority),
        ("Kimi agent-level (RUL-018)", check_kimi_agent_level),
        ("Env contract (RUL-019)", check_env_contract),
        ("Generated artifacts (RUL-030)", check_generated_artifacts),
        ("Requirement corpus (RTM gradeability)", check_requirement_corpus),
        ("Graph write swallow", check_graph_write_swallow),
    ]

    if not args.fast:
        checks.append(("Test purity", check_test_purity))
        checks.append(("Infra drift (Terraform)", check_infra_drift))
        checks.append(("Env/fleet agreement (Terraform)", check_env_fleet_agreement))
        checks.append(("Systemd units (deploy/systemd)", check_systemd_units))
        checks.append(("Prolog boundary", check_prolog_boundary))
        checks.append(("Prolog meta-lint (SPEC-033 B17)", check_prolog_meta_lint))
        checks.append(("KG schema (advisory)", check_kg_schema_advisory))
        checks.append(("KG connectivity", check_kg_connectivity))
        checks.append(("ADR graph sync (repo/wiki/graph)", check_adr_graph_sync))

    if args.ci:
        checks.append(("Pytest (non-Neo4j)", lambda: check_pytest(skip=False)))

    return checks


def main() -> int:
    """CLI entry point — runs every registered check and aggregates the
    pass/fail result (see the module docstring for the mode flags)."""
    parser = argparse.ArgumentParser(description="Project health check")
    parser.add_argument("--fast", action="store_true", help="Only fast checks")
    parser.add_argument("--ci", action="store_true", help="CI mode (includes pytest)")
    parser.add_argument("--fix", action="store_true", help="Auto-fix ruff issues")
    args = parser.parse_args()

    checks = build_checks(args)

    all_passed = True
    all_errors: list[str] = []

    print("=" * 60)
    print("HEALTH CHECK")
    print("=" * 60)

    # Run all checks concurrently — they are independent subprocesses.
    # Wall-clock becomes the longest single check, not the sum.
    # Results are collected in submission order for deterministic output.
    futures = {}
    t_start = time.monotonic()
    with ThreadPoolExecutor(max_workers=min(len(checks), 8)) as pool:
        for name, check_fn in checks:
            fut = pool.submit(check_fn)
            futures[fut] = name
            print(f"⏳ {name} ... started", flush=True)

        for fut in as_completed(futures):
            name = futures[fut]
            elapsed = time.monotonic() - t_start
            try:
                passed, errors = fut.result()
            except Exception as e:
                print(f"💥 ERROR {name} ({elapsed:.1f}s): {e}", flush=True)
                all_passed = False
                all_errors.append(f"{name}: {e}")
                continue
            if passed:
                print(f"✅ PASS {name} ({elapsed:.1f}s)", flush=True)
            else:
                print(f"❌ FAIL {name} ({elapsed:.1f}s)", flush=True)
                all_passed = False
                all_errors.extend(errors)

    total_elapsed = time.monotonic() - t_start

    print(f"\n{'=' * 60}")
    print(f"Total elapsed: {total_elapsed:.1f}s")
    if all_passed:
        print("RESULT: ALL CHECKS PASSED")
        print("=" * 60)
        return 0
    else:
        print(f"RESULT: {len(all_errors)} ERROR(S)")
        print("=" * 60)
        for e in all_errors[:50]:
            print(f"  {e}")
        if len(all_errors) > 50:
            print(f"  ... and {len(all_errors) - 50} more")
        return 1


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