hermes-agent/scripts/check

#!/usr/bin/env python3
"""Run every blocking lint check CI runs, locally, with the same pinned tools.

    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 --install-hook # run it as the git pre-commit hook

CI runs the same command (``.github/workflows/lint.yml``), so a green local run means a green
lint lane. The pre-commit hook is skippable (``git commit --no-verify``); CI is not.
"""

from __future__ import annotations

import argparse
import os
import shutil
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 pre-commit hook"
_HOOK = f"""#!/usr/bin/env bash
{_HOOK_MARK} (installed by `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 commits unchecked.
[ -f "$repo/scripts/check" ] || exit 0
py=python3
command -v python3 >/dev/null 2>&1 || py=python
exec "$py" "$repo/scripts/check" --staged
"""

# name -> argv (relative to the repo root); mirrors the blocking steps in lint.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"],
}


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


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

    Unstaged edits stay where they are (no stash, no checkout); the index is read from a copy,
    so nothing a checker runs can rewrite the user's real index. 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 = Path(gitio.git(REPO, "rev-parse", "--absolute-git-dir").strip())
    index = Path(gitio.git(REPO, "rev-parse", "--path-format=absolute", "--git-path", "index").strip())
    with tempfile.TemporaryDirectory(prefix="hermes-check-index-") as tmp:
        root = Path(tmp) / "tree"
        root.mkdir()  # Git for Windows will not create the --prefix directory itself
        gitio.git(REPO, "checkout-index", "-a", f"--prefix={root}{os.sep}")
        index_copy = Path(tmp) / "index"
        shutil.copyfile(index, index_copy)
        env = {**os.environ, "GIT_DIR": str(git_dir), "GIT_WORK_TREE": str(root),
               "GIT_INDEX_FILE": str(index_copy)}
        yield root, env


def _jobs(only: str | None, base: str | None, head: str | None,
          root: Path = REPO, env: dict[str, str] | None = None) -> dict[str, _Job]:
    names = [*_GUARDS, "ruff", "health"]
    if only:
        names = [n for n in names if n in only.split(",")]
    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(REPO), "check", "."], root, env)
        else:
            # The ratchet reads both sides from git: real repo, judged against the head tree.
            health = [sys.executable, "-m", "scripts.code_health"]
            health += ["--base", base] if base else []
            health += ["--head", head] if head else []
            jobs[name] = _Job(health)
    return jobs


def _run_all(args: argparse.Namespace) -> dict[str, tuple[int, str]]:
    if not args.staged:
        jobs = _jobs(args.only, args.base, None)
        with ThreadPoolExecutor(max_workers=len(jobs) or 1) as pool:
            return dict(zip(jobs, pool.map(_run, jobs.values()), strict=True))
    # Pre-commit: judge the bytes being committed, not the working tree.
    tree = gitio.git(REPO, "write-tree").strip()
    with _index_snapshot() as (root, env):
        jobs = _jobs(args.only, args.base, tree, root, env)
        with ThreadPoolExecutor(max_workers=len(jobs) or 1) as pool:
            return dict(zip(jobs, pool.map(_run, jobs.values()), strict=True))


def install_hook() -> int:
    hooks = Path(gitio.git(REPO, "rev-parse", "--path-format=absolute", "--git-path", "hooks").strip())
    target = hooks / "pre-commit"
    if target.exists() and _HOOK_MARK not in target.read_text(encoding="utf-8-sig", errors="replace"):
        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(_HOOK, 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: pre-commit hook installed at {target}")
    return 0


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([*_GUARDS, 'ruff', 'health'])}")
    parser.add_argument("--base", help="base revision for the code-health ratchet")
    parser.add_argument("--staged", action="store_true",
                        help="check the index (what `git commit` records), not the working tree")
    parser.add_argument("--install-hook", action="store_true")
    args = parser.parse_args(argv)
    if args.install_hook:
        return install_hook()
    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())