hermes-agent/scripts/check

#!/usr/bin/env python3
"""Run every blocking lint check CI runs, locally, with the same pinned tools: the lint
workflow's guards, ruff and the code-health ratchet, plus the case-collision, lazy-deps and
profile-archive guards that CI runs in workflows of their own.

    python scripts/check                # everything, against the merge-base with origin/main
    python scripts/check --only health  # just the code-health ratchet
    python scripts/check --staged       # judge the index (the pre-commit hook runs this)
    python scripts/check --commit REV   # judge one commit (the pre-push hook runs this)
    python scripts/check --install-hook [pre-commit|pre-push]   # run it automatically
    python scripts/check --uninstall-hook

CI runs the same command (``.github/workflows/lint.yml``), so a green local run means a green
lint lane. Hooks are skippable (``--no-verify``); CI is not. The code-health part honours the
``ENFORCEMENT`` switch in ``scripts/code_health/config.py`` (blocking / advisory / off).
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import tempfile
from collections.abc import Iterator
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO))

from scripts.code_health import gitio  # noqa: E402
from scripts.code_health.ruff_runner import resolve_ruff  # noqa: E402

_HOOK_MARK = "# hermes-agent scripts/check hook"
_HOOK_HEAD = f"""#!/usr/bin/env bash
{_HOOK_MARK} (`scripts/check --install-hook`; skip once with --no-verify)
repo="$(git rev-parse --show-toplevel)"
# Hooks are shared by every worktree; a branch that predates scripts/check is not checked.
[ -f "$repo/scripts/check" ] || exit 0
py=python3
command -v python3 >/dev/null 2>&1 || py=python
"""
# pre-push reads "<local ref> <local sha> <remote ref> <remote sha>" per pushed ref; an all-zero
# local sha is a branch deletion. Each pushed commit is judged as CI will judge the PR.
_HOOKS = {
    "pre-commit": _HOOK_HEAD + 'exec "$py" "$repo/scripts/check" --staged\n',
    "pre-push": _HOOK_HEAD + """status=0
while read -r _ local_sha _ _; do
  case "$local_sha" in
    *[!0]*) "$py" "$repo/scripts/check" --commit "$local_sha" </dev/null || status=1 ;;
  esac
done
exit $status
""",
}

# name -> argv (relative to the repo root); mirrors the blocking steps in lint.yml, plus the
# unconditional guards CI runs in their own workflows (case-collision-check.yml,
# lazy-deps-guard.yml, profile-artifact-check.yml).
_GUARDS = {
    "footguns": ["scripts/check-windows-footguns.py", "--all"],
    "shebangs": ["scripts/check_bash_shebangs.py"],
    "no-tmp": ["scripts/check_no_tmp_literals.py"],
    "config-writers": ["scripts/check_config_yaml_writers.py"],
    "os-marker-fakes": ["scripts/ci/check_os_marker_fakes.py"],
    "case-collisions": ["scripts/check-case-collisions.py"],
    "lazy-deps": ["scripts/ci/check_lazy_deps_imports.py"],
    "profile-archives": ["scripts/ci/check_profile_archive_boundary.py"],
}
_CHECKS = (*_GUARDS, "ruff", "health")


@dataclass
class _Job:
    argv: list[str]
    cwd: Path = REPO
    env: dict[str, str] | None = None


def _run(job: _Job) -> tuple[int, str]:
    proc = subprocess.run(
        job.argv, cwd=job.cwd, env=job.env, capture_output=True, text=True, encoding="utf-8",
        errors="replace", stdin=subprocess.DEVNULL, timeout=900, check=False,
    )
    return proc.returncode, (proc.stdout + proc.stderr).strip()


def _git_in(env: dict[str, str], *args: str) -> None:
    proc = subprocess.run(["git", *args], cwd=REPO, env=env, capture_output=True, text=True,
                          stdin=subprocess.DEVNULL, timeout=300, check=False)
    if proc.returncode != 0:
        raise RuntimeError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")


@contextmanager
def _tree_snapshot(tree: str) -> Iterator[tuple[Path, dict[str, str]]]:
    """``tree`` checked out on its own, plus git env that sees exactly that tree.

    The user's worktree and index are never touched (no stash, no checkout): the tree is read
    into a private index and written to a temp dir. Always a snapshot, even when the worktree
    looks clean: ignored files (scratch, build output) are invisible to git status but not to
    the guards that walk the disk.
    """
    git_dir = gitio.git(REPO, "rev-parse", "--absolute-git-dir").strip()
    with tempfile.TemporaryDirectory(prefix="hermes-check-tree-") as tmp:
        root = Path(tmp) / "tree"
        root.mkdir()
        env = {**os.environ, "GIT_DIR": git_dir, "GIT_WORK_TREE": str(root),
               "GIT_INDEX_FILE": str(Path(tmp) / "index")}
        _git_in(env, "read-tree", tree)
        _git_in(env, "checkout-index", "-a")
        yield root, env


def _jobs(names: list[str], base: str | None, head: str | None,
          root: Path = REPO, env: dict[str, str] | None = None) -> dict[str, _Job]:
    jobs: dict[str, _Job] = {}
    for name in names:  # built only when selected: `--only shebangs` needs no ruff
        if name in _GUARDS:
            jobs[name] = _Job([sys.executable, *_GUARDS[name]], root, env)
        elif name == "ruff":
            jobs[name] = _Job([*resolve_ruff(root), "check", "."], root, env)
        else:
            # The ratchet reads both sides from git. It runs from `root` like every other job, so
            # in a snapshot its engine, rules and policy files (profile_scope_patterns.json, the
            # ruff pin) are the judged artifact's own, never unstaged working-tree bytes.
            health = [sys.executable, "-m", "scripts.code_health"]
            health += ["--base", base] if base else []
            health += ["--head", head] if head else []
            jobs[name] = _Job(health, root, env)
    return jobs


def _run_jobs(jobs: dict[str, _Job]) -> dict[str, tuple[int, str]]:
    with ThreadPoolExecutor(max_workers=len(jobs) or 1) as pool:
        return dict(zip(jobs, pool.map(_run, jobs.values()), strict=True))


def _run_all(args: argparse.Namespace) -> dict[str, tuple[int, str]]:
    if not (args.staged or args.commit):
        return _run_jobs(_jobs(args.only, args.base, None))
    # Hooks judge the bytes being committed or pushed, never the working tree. Without --base the
    # ratchet derives the base from the head (a commit's own merge-base) and the switch from main.
    if args.staged:
        tree = head = gitio.git(REPO, "write-tree").strip()
    else:
        head = gitio.resolve_rev(REPO, args.commit)
        tree = gitio.resolve_tree(REPO, head)
    with _tree_snapshot(tree) as (root, env):
        return _run_jobs(_jobs(args.only, args.base, head, root, env))


def _hooks_dir() -> Path:
    return Path(gitio.git(REPO, "rev-parse", "--path-format=absolute", "--git-path", "hooks").strip())


def _ours(target: Path) -> bool:
    return _HOOK_MARK in target.read_text(encoding="utf-8-sig", errors="replace")


def install_hook(kind: str) -> int:
    hooks = _hooks_dir()
    target = hooks / kind
    if target.exists() and not _ours(target):
        print(f"scripts/check: {target} exists and is not ours; add `python scripts/check` to it")
        return 1
    hooks.mkdir(parents=True, exist_ok=True)
    target.write_text(_HOOKS[kind], encoding="utf-8", newline="\n")
    os.chmod(target, 0o755)  # windows-footgun: ok — chmod +x is a no-op on Windows, harmless
    print(f"scripts/check: {kind} hook installed at {target} (every worktree of this clone)")
    return 0


def uninstall_hooks() -> int:
    removed = [t for t in (_hooks_dir() / k for k in _HOOKS) if t.exists() and _ours(t)]
    for target in removed:
        target.unlink()
        print(f"scripts/check: removed {target}")
    if not removed:
        print("scripts/check: no scripts/check hooks installed")
    return 0


def _selected(parser: argparse.ArgumentParser, only: str | None) -> list[str]:
    """Every check, or exactly the named ones. A typo must not turn into an empty green run."""
    if only is None:
        return list(_CHECKS)
    wanted = {name.strip() for name in only.split(",")} - {""}
    unknown = sorted(wanted - set(_CHECKS))
    if not wanted or unknown:
        parser.error(f"--only {only!r}: unknown or no check name ({', '.join(unknown) or 'empty'});"
                     f" valid: {', '.join(_CHECKS)}")
    return [name for name in _CHECKS if name in wanted]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="scripts/check", description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--only", help=f"comma list of: {', '.join(_CHECKS)}")
    parser.add_argument("--base", help="base revision for the code-health ratchet")
    target = parser.add_mutually_exclusive_group()
    target.add_argument("--staged", action="store_true",
                        help="check the index (what `git commit` records), not the working tree")
    target.add_argument("--commit", metavar="REV", help="check one commit's tree (pre-push)")
    parser.add_argument("--install-hook", nargs="?", const="pre-commit", choices=sorted(_HOOKS),
                        help="run on every commit (default) or every push")
    parser.add_argument("--uninstall-hook", action="store_true", help="remove our git hooks")
    args = parser.parse_args(argv)
    args.only = _selected(parser, args.only)
    if args.install_hook:
        return install_hook(args.install_hook)
    if args.uninstall_hook:
        return uninstall_hooks()
    try:
        results = _run_all(args)
    except RuntimeError as exc:
        print(f"scripts/check: {exc}", file=sys.stderr)
        return 2
    failed = [name for name, (code, _) in results.items() if code != 0]
    # A passing health run still carries advisories and comment-allowed findings that the
    # reviewer must see, so its report is always shown.
    for name, (code, output) in results.items():
        if code != 0 or name == "health":
            label = "failed" if code != 0 else "ok"
            print(f"\n=== {name} ({label}) ===\n{output.rstrip()}")
    status = "FAILED: " + ", ".join(failed) if failed else "ok"
    print(f"\nscripts/check: {len(results)} checks, {status}")
    return 1 if failed else 0


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