diff --git a/scripts/check b/scripts/check new file mode 100755 index 0000000000..a81d4c23de --- /dev/null +++ b/scripts/check @@ -0,0 +1,104 @@ +#!/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 --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 subprocess +import sys +from concurrent.futures import ThreadPoolExecutor +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" +""" + +# 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"], +} + + +def _run(argv: list[str]) -> tuple[int, str]: + proc = subprocess.run( + argv, cwd=REPO, 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 _jobs(only: str | None, base: str | None) -> dict[str, list[str]]: + jobs = {name: [sys.executable, *argv] for name, argv in _GUARDS.items()} + jobs["ruff"] = [*resolve_ruff(REPO), "check", "."] + health = [sys.executable, "-m", "scripts.code_health"] + jobs["health"] = health + (["--base", base] if base else []) + if only: + return {name: argv for name, argv in jobs.items() if name in only.split(",")} + return jobs + + +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("--install-hook", action="store_true") + args = parser.parse_args(argv) + if args.install_hook: + return install_hook() + try: + jobs = _jobs(args.only, args.base) + except RuntimeError as exc: + print(f"scripts/check: {exc}", file=sys.stderr) + return 2 + with ThreadPoolExecutor(max_workers=len(jobs) or 1) as pool: + results = dict(zip(jobs, pool.map(_run, jobs.values()), strict=True)) + failed = [name for name, (code, _) in results.items() if code != 0] + for name in failed: + print(f"\n=== {name} (failed) ===\n{results[name][1]}") + 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()) diff --git a/scripts/code_health/__init__.py b/scripts/code_health/__init__.py new file mode 100644 index 0000000000..8241edf494 --- /dev/null +++ b/scripts/code_health/__init__.py @@ -0,0 +1 @@ +"""Code-health ratchet: per-unit caps that only go down. Entry point: ``scripts/check``.""" diff --git a/scripts/code_health/__main__.py b/scripts/code_health/__main__.py new file mode 100644 index 0000000000..8d1b3f7895 --- /dev/null +++ b/scripts/code_health/__main__.py @@ -0,0 +1,3 @@ +from scripts.code_health.cli import main + +raise SystemExit(main()) diff --git a/scripts/code_health/cli.py b/scripts/code_health/cli.py new file mode 100644 index 0000000000..d025586890 --- /dev/null +++ b/scripts/code_health/cli.py @@ -0,0 +1,93 @@ +"""``python -m scripts.code_health`` / ``scripts/check``: the code-health ratchet.""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +from dataclasses import asdict +from pathlib import Path + +from scripts.code_health import gitio +from scripts.code_health.compare import compare +from scripts.code_health.config import in_scope +from scripts.code_health.measure import Measurer +from scripts.code_health.report import ( + apply_allows, + format_findings, + summarize_tree, + verdict, +) +from scripts.code_health.ruff_runner import pinned_version, resolve_ruff + +_FOOTER = """ +Each function and file has its own cap: its value on main if it is already over target, +otherwise the target. New code meets the target; existing code may only go down. Fix the code +(the `fix:` line is the repo's remedy); a genuine exception takes a reviewed comment on the line, +`# health: allow -- `. Rules and targets: scripts/code_health/config.py.""" + + +def _parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="code_health", description=__doc__) + p.add_argument("--base", help="base revision (default: merge-base of HEAD with origin/main)") + p.add_argument("--head", help="head revision (default: the working tree, untracked included)") + p.add_argument("--report", action="store_true", help="burn-down summary of the whole head tree") + p.add_argument("--json", action="store_true", help="machine-readable findings") + p.add_argument("--print-pins", action="store_true", + help="print the pinned ruff requirement (CI installs exactly this)") + return p + + +def _report(repo: Path, head: str | None) -> int: + paths = [p for p in gitio.tracked_files(repo, head) if in_scope(p)] + measurer = Measurer(repo, resolve_ruff(repo), known_env=gitio.known_env_names(repo, head)) + print(summarize_tree(measurer.measure(head, paths))) + return 0 + + +def run(repo: Path, base: str, head: str | None, as_json: bool = False) -> int: + started = time.monotonic() + changes = gitio.changed_files(repo, base, head) + head_paths = sorted({c.new for c in changes if c.new and in_scope(c.new)}) + base_paths = sorted({c.old for c in changes if c.old and in_scope(c.old)}) + if not head_paths: + print("code health: no measured files changed") + return 0 + needs_ruff = any(p.endswith(".py") for p in head_paths) + ruff = resolve_ruff(repo) if needs_ruff else [] + measurer = Measurer(repo, ruff, known_env=gitio.known_env_names(repo, base)) + base_m = measurer.measure(base, base_paths) + head_m = measurer.measure(head, head_paths) + findings = compare(base_m, head_m, changes) + apply_allows(findings, head_m) + blocking, advisory = verdict(findings) + if as_json: + print(json.dumps([asdict(f) for f in findings], indent=2)) + return 1 if blocking else 0 + body = format_findings(findings) + if body: + print(body) + elapsed = time.monotonic() - started + print(f"\ncode health: {len(head_paths)} files vs {base[:12]}: {blocking} blocking, " + f"{advisory} advisory ({elapsed:.1f}s)") + if blocking: + print(_FOOTER) + return 1 if blocking else 0 + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + repo = gitio.repo_root(Path.cwd()) + if args.print_pins: + print(f"ruff=={pinned_version(repo)}") + return 0 + head = gitio.resolve_rev(repo, args.head) if args.head else None + try: + if args.report: + return _report(repo, head) + base = gitio.resolve_rev(repo, args.base) if args.base else gitio.default_base(repo) + return run(repo, base, head, as_json=args.json) + except RuntimeError as exc: + print(f"code health: {exc}", file=sys.stderr) + return 2 diff --git a/scripts/code_health/compare.py b/scripts/code_health/compare.py new file mode 100644 index 0000000000..a03100e41e --- /dev/null +++ b/scripts/code_health/compare.py @@ -0,0 +1,121 @@ +"""Base vs head: the ratchet verdict. + +Every unit has its own cap: a function or file already over target may not grow past the value +it has on the base revision; anything new must meet the target. Pattern rules compare multisets +of fingerprints per file, so fixing one violation and adding another still fails. + +Code that moves keeps its cap and its existing violations, matched by content: a function by +its name-independent body hash, a pattern hit by (rule, line text) once its enclosing function +has vanished from the old place. Nothing else crosses units. +""" + +from __future__ import annotations + +from collections import Counter, defaultdict +from dataclasses import dataclass, field + +from scripts.code_health.config import ADVISORY_GROWTH, RULES_BY_ID, TARGETS +from scripts.code_health.gitio import Change +from scripts.code_health.model import MODULE_SCOPE, FileMeasure, Finding, Hit, Unit + + +@dataclass +class _Pool: + """Units and hits that disappeared from their base location in this diff.""" + + units: dict[str, list[Unit]] = field(default_factory=lambda: defaultdict(list)) + hits: Counter[tuple[str, str]] = field(default_factory=Counter) + + def take_unit(self, body_hash: str) -> Unit | None: + bucket = self.units.get(body_hash) + return bucket.pop() if bucket else None + + def take_hit(self, rule: str, text: str) -> bool: + key = (rule, text) + if self.hits[key] > 0: + self.hits[key] -= 1 + return True + return False + + +def _build_pool(base: dict[str, FileMeasure], head: dict[str, FileMeasure], + head_of: dict[str, str | None]) -> _Pool: + pool = _Pool() + for bpath, bf in base.items(): + hpath = head_of.get(bpath, bpath) + hf = head.get(hpath) if hpath else None + live = set(hf.units) if hf else set() + for qual, unit in bf.units.items(): + if qual not in live: + pool.units[unit.body_hash].append(unit) + for hit, count in bf.hits.items(): + gone = hf is None or (hit.scope != MODULE_SCOPE and hit.scope not in live) + if gone: + pool.hits[(hit.rule, hit.text)] += count + return pool + + +def _metric_findings(path: str, hf: FileMeasure, bf: FileMeasure | None, pool: _Pool) -> list[Finding]: + findings: list[Finding] = [] + lines = hf.metrics.get("FILE_LINES", 0) + if lines > TARGETS["FILE_LINES"]: + base_lines = bf.metrics.get("FILE_LINES", 0) if bf else 0 + cap = max(TARGETS["FILE_LINES"], base_lines) + if lines > cap: + grew_over = base_lines > TARGETS["FILE_LINES"] and "FILE_LINES" in ADVISORY_GROWTH + findings.append(Finding(path, "FILE_LINES", MODULE_SCOPE, 1, _detail( + lines, TARGETS["FILE_LINES"], base_lines if bf else None, "lines"), + blocking=not grew_over)) + for qual, unit in hf.units.items(): + prior = bf.units.get(qual) if bf else None + if prior is None: + prior = pool.take_unit(unit.body_hash) + for metric, value in unit.metrics.items(): + target = TARGETS[metric] + if value <= target: + continue + was = prior.metrics.get(metric) if prior else None + cap = max(target, was or 0) + if value > cap: + findings.append(Finding(path, metric, qual, unit.line, + _detail(value, target, was, metric))) + return findings + + +def _detail(value: int, target: int, was: int | None, what: str) -> str: + if was is None: + return f"{what} {value} > target {target} (new code must meet the target)" + if was <= target: + return f"{what} {value} > target {target} (was {was})" + return f"{what} {value} > {was}, its value on main (over target {target}: it may only go down)" + + +def _hit_findings(path: str, hf: FileMeasure, bf: FileMeasure | None, pool: _Pool) -> list[Finding]: + findings: list[Finding] = [] + base_hits: Counter[Hit] = bf.hits if bf else Counter() + for hit, count in sorted(hf.hits.items(), key=lambda kv: hf.hit_lines[kv[0]][0]): + extra = count - base_hits.get(hit, 0) + while extra > 0 and pool.take_hit(hit.rule, hit.text): + extra -= 1 + if extra <= 0: + continue + rule = RULES_BY_ID[hit.rule] + for line in hf.hit_lines[hit][-extra:]: + findings.append(Finding(path, hit.rule, hit.scope, line, rule.title, + blocking=rule.blocking)) + return findings + + +def compare(base: dict[str, FileMeasure], head: dict[str, FileMeasure], + changes: list[Change]) -> list[Finding]: + base_of = {c.new: c.old for c in changes if c.new} + head_of = {c.old: c.new for c in changes if c.old} + pool = _build_pool(base, head, head_of) + findings: list[Finding] = [] + for path in sorted(head): + hf = head[path] + bpath = base_of.get(path, path) + bf = base.get(bpath) if bpath else None + findings += _metric_findings(path, hf, bf, pool) + findings += _hit_findings(path, hf, bf, pool) + return findings diff --git a/scripts/code_health/config.py b/scripts/code_health/config.py new file mode 100644 index 0000000000..115549046a --- /dev/null +++ b/scripts/code_health/config.py @@ -0,0 +1,138 @@ +"""Targets, file scope and the rule table: the one place a rule's id, scope and fix text live. + +A rule's ``fix`` is the text the agent reads at the moment it trips the rule, so it names the +repo-specific remedy rather than restating the rule. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from fnmatch import fnmatch + +# Per-unit targets. A unit already over its target keeps its own current value as its cap +# (measured on the base revision) and may only go down; new units must meet the target. +TARGETS = { + "CC": 20, # cyclomatic complexity per function (ruff C901 numbers for Python) + "FUNC_LINES": 300, + "NESTING": 6, # nested control-flow blocks inside one function + "FILE_LINES": 2000, +} + +# Growth of a unit ALREADY over target is advisory for these metrics; crossing the target still +# blocks. Replay of 300 merged PRs: a blocking FILE_LINES ratchet would have stopped 46 of them +# for +2..+70 lines on one of the 119 files already over 2,000, before those files are split. +ADVISORY_GROWTH = {"FILE_LINES"} + +METRIC_FIX = { + "CC": "split it: extract phases into a topical sibling (`_.py`), or replace a" + " branch ladder with a dict/table -> handler", + "FUNC_LINES": "extract phases into named helpers or a topical sibling module", + "NESTING": "return early, or extract the inner block into a helper", + "FILE_LINES": "split along `_` (facade + siblings, see root AGENTS.md); new" + " behaviour goes in a new or topical sibling, never appended to a facade", +} + +PY_EXCLUDE = ("website/*", "*/node_modules/*", "node_modules/*") +TS_EXCLUDE = PY_EXCLUDE + ("*.d.ts", "*/dist/*", "*/build/*", "*/generated/*", "*.generated.ts") +TESTS = ("tests/*", "*/tests/*", "tests-js/*", "*/conftest.py") +# One-shot processes that serve a single profile (CI/release scripts, skill scripts run as a +# child with a scoped env, installers): the profile-scope rules don't apply to them. +SINGLE_PROFILE = TESTS + ("scripts/*", "skills/*", "optional-skills/*", "evals/*", "setup.py") + + +@dataclass(frozen=True) +class Rule: + id: str + title: str + fix: str + source: str # "ruff" | "ast" | "regex" + exclude: tuple[str, ...] = TESTS + blocking: bool = True + pattern_id: str = "" # regex rules: id in scripts/ci/profile_scope_patterns.json + + +RULES: tuple[Rule, ...] = ( + # Swallowed exceptions: the top statically-catchable bug cause in the fix sample. + Rule("S110", "try/except/pass", "log it or narrow the except; a silent pass hides the" + " failure the next bug report is about", "ruff"), + Rule("BLE001", "blind `except Exception`", "catch the exceptions this code can actually" + " raise; if a boundary truly needs a catch-all, log it with context", "ruff"), + Rule("E722", "bare `except:`", "catch a named exception (bare except also swallows" + " KeyboardInterrupt/SystemExit)", "ruff"), + # Correctness basics. + # tui_gateway method modules get the server's names injected (`bind_module(globals(), ...)`). + Rule("F821", "undefined name", "import or define it", "ruff", exclude=("tui_gateway/*",)), + Rule("F823", "local used before assignment", "assign before use or declare global", + "ruff", exclude=()), + Rule("F811", "redefinition of unused name", "remove the shadowed definition", "ruff", + exclude=()), + Rule("F841", "unused local variable", "delete it or use it", "ruff", exclude=()), + Rule("RUF006", "dangling asyncio task", "keep a reference to the task (it can be garbage" + " collected mid-run) and handle its exception", "ruff"), + # Hangs. + Rule("ASYNC230", "blocking open() in async def", "`await asyncio.to_thread(...)`", "ruff"), + Rule("ASYNC240", "blocking path call in async def", "`await asyncio.to_thread(...)`", + "ruff"), + Rule("S113", "HTTP request without timeout", "pass `timeout=`", "ruff"), + Rule("HX006", "subprocess/urlopen without timeout", "pass `timeout=` (a child or socket" + " that never answers hangs this thread forever)", "ast"), + Rule("HX007", "sync config I/O inside async def", "`await asyncio.to_thread(load_config)`" + " or read it before entering the event loop", "ast"), + Rule("HX008", "asyncio.get_event_loop()", "`asyncio.get_running_loop()` inside a" + " coroutine; `asyncio.run()` at an entry point", "ast"), + Rule("HX009", "isinstance(r, Exception) on gather results", "check `BaseException`:" + " `gather(return_exceptions=True)` also returns CancelledError", "ast"), + # Profile scope (one process serves many profiles). + Rule("HX001", "hardcoded Hermes home", "`get_hermes_home()` for paths," + " `display_hermes_home()` for user-facing text (`hermes_constants`)", "ast", + exclude=SINGLE_PROFILE + ("hermes_constants.py",)), + Rule("HX002", "new HERMES_* environment variable", "behavioural settings go in" + " config.yaml, secrets through the secret scope; `.env` is for credentials only", + "ast", exclude=SINGLE_PROFILE), + Rule("HX004", "UnscopedSecretError falls back to os.getenv", "bind the owning profile's" + " secret scope at the spawn site; never read the launch profile's environment", + "ast", exclude=SINGLE_PROFILE), + Rule("HX005", "home/config/env captured at import time", "resolve it at call time" + " (`get_hermes_home()`, `load_config()`) or key the slot by `hermes_home_key()`;" + " module constants hold the launch profile's value", "ast", + exclude=SINGLE_PROFILE), + Rule("HX012", "raw threading.Thread", "`spawn_context_thread(...)` so the thread keeps" + " the caller's profile scope (ContextVars do not cross a bare Thread)", "ast", + exclude=SINGLE_PROFILE), + Rule("PS-P05", "child env built from os.environ", "`served_profile_child_env()`; the" + " child otherwise inherits the launch profile's home and secrets", "regex", + exclude=SINGLE_PROFILE, pattern_id="P05"), + Rule("PS-P06", "raw platform credential getenv", "read it through the profile's secret" + " scope (`get_secret`)", "regex", exclude=SINGLE_PROFILE, pattern_id="P06"), + # Process identity. + Rule("HX003", "process identity from argv substrings", "use" + " `gateway.status.looks_like_gateway_command_line` /" + " `hermes_cli.update_cmd._hermes_holder_subcommand` and match full cmdlines", "ast"), + # Config truthiness. + Rule("HX010", "bool() of a config/env string", "`bool(\"false\")` is True: parse it" + " (`is_truthy_value`) or compare explicitly", "ast"), + # Structure. + Rule("HX011", "if/elif ladder on one name", "use a dict/table -> handler (`_SLASH_DISPATCH`" + " is the shape)", "ast", exclude=()), +) + +RULES_BY_ID = {rule.id: rule for rule in RULES} +RUFF_CODES = tuple(rule.id for rule in RULES if rule.source == "ruff") +ALLOW_SYNTAX = "# health: allow -- " + + +def _matches(path: str, globs: tuple[str, ...]) -> bool: + return any(fnmatch(path, glob) for glob in globs) + + +def rule_applies(rule: Rule, path: str) -> bool: + return not _matches(path, rule.exclude) + + +def in_scope(path: str) -> str | None: + """Language of a tracked path the engine measures, or None.""" + if path.endswith(".py"): + return None if _matches(path, PY_EXCLUDE) else "py" + if path.endswith((".ts", ".tsx", ".mts", ".cts")): + return None if _matches(path, TS_EXCLUDE) else "ts" + return None diff --git a/scripts/code_health/gitio.py b/scripts/code_health/gitio.py new file mode 100644 index 0000000000..bd99b586f3 --- /dev/null +++ b/scripts/code_health/gitio.py @@ -0,0 +1,124 @@ +"""Git plumbing: which files changed between two trees, and their contents on either side. + +A tree is a revision (``"abc123"``) or ``None`` for the working tree. Both sides are read +through the same functions so base and head are measured identically. +""" + +from __future__ import annotations + +import subprocess +from dataclasses import dataclass +from pathlib import Path + + +def git(repo: Path, *args: str, check: bool = True) -> str: + proc = subprocess.run( + ["git", *args], + cwd=repo, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + stdin=subprocess.DEVNULL, + timeout=120, + check=False, + ) + if check and proc.returncode != 0: + raise RuntimeError(f"git {' '.join(args)} failed: {proc.stderr.strip()}") + return proc.stdout + + +def repo_root(start: Path) -> Path: + return Path(git(start, "rev-parse", "--show-toplevel").strip()) + + +def resolve_rev(repo: Path, rev: str) -> str: + return git(repo, "rev-parse", "--verify", f"{rev}^{{commit}}").strip() + + +def default_base(repo: Path) -> str: + """Merge-base of HEAD with origin/main (local runs); CI passes ``--base`` explicitly.""" + for ref in ("origin/main", "main"): + out = git(repo, "merge-base", "HEAD", ref, check=False).strip() + if out: + return out + raise RuntimeError("no merge-base with origin/main or main; pass --base") + + +@dataclass(frozen=True) +class Change: + status: str # A, M, D, R + old: str | None + new: str | None + + +def _parse_name_status(out: str) -> list[Change]: + fields = out.split("\0") + changes: list[Change] = [] + i = 0 + while i < len(fields) and fields[i]: + code = fields[i][0] + if code in "RC": + old, new = fields[i + 1], fields[i + 2] + changes.append(Change("R" if code == "R" else "A", old if code == "R" else None, new)) + i += 3 + continue + path = fields[i + 1] + if code == "D": + changes.append(Change("D", path, None)) + elif code == "A": + changes.append(Change("A", None, path)) + else: + changes.append(Change("M", path, path)) + i += 2 + return changes + + +def changed_files(repo: Path, base: str, head: str | None) -> list[Change]: + args = ["diff", "--name-status", "-M", "-z", "--no-color", base] + if head is not None: + args.append(head) + changes = _parse_name_status(git(repo, *args)) + if head is None: + untracked = git(repo, "ls-files", "--others", "--exclude-standard", "-z") + changes.extend(Change("A", None, p) for p in untracked.split("\0") if p) + return changes + + +def read_file(repo: Path, tree: str | None, path: str) -> str | None: + if tree is None: + try: + return (repo / path).read_text(encoding="utf-8-sig", errors="replace") + except (FileNotFoundError, IsADirectoryError): + return None + proc = subprocess.run( + ["git", "show", f"{tree}:{path}"], + cwd=repo, + capture_output=True, + stdin=subprocess.DEVNULL, + timeout=60, + check=False, + ) + if proc.returncode != 0: + return None + return proc.stdout.decode("utf-8", errors="replace") + + +def tracked_files(repo: Path, tree: str | None) -> list[str]: + if tree is None: + return [p for p in git(repo, "ls-files", "-z").split("\0") if p] + return [p for p in git(repo, "ls-tree", "-r", "--name-only", "-z", tree).split("\0") if p] + + +def known_env_names(repo: Path, tree: str | None) -> set[str]: + """Every HERMES_* literal in non-test Python of a tree (the HX002 allow set).""" + args = ["grep", "-h", "-o", "-I", "-E", r"HERMES_[A-Z0-9_]+"] + if tree is not None: + args.append(tree) + args += ["--", "*.py", ":!tests/**", ":!**/tests/**"] + out = git(repo, *args, check=False) + names = set() + for line in out.splitlines(): + # `git grep ` prefixes ":" even with -h on some versions. + names.add(line.rsplit(":", 1)[-1]) + return names diff --git a/scripts/code_health/measure.py b/scripts/code_health/measure.py new file mode 100644 index 0000000000..2ab82e5008 --- /dev/null +++ b/scripts/code_health/measure.py @@ -0,0 +1,121 @@ +"""Measure a set of files in one tree (a revision or the working tree).""" + +from __future__ import annotations + +import ast +import json +import re +import tempfile +import warnings +from pathlib import Path + +from scripts.code_health import py_rules, py_structure +from scripts.code_health.config import RULES, RULES_BY_ID, in_scope, rule_applies +from scripts.code_health.gitio import read_file +from scripts.code_health.model import FileMeasure, Unit +from scripts.code_health.ruff_runner import run_ruff +from scripts.code_health.ts_measure import measure_ts + +_PATTERNS_FILE = Path("scripts/ci/profile_scope_patterns.json") + + +def _regex_rules(repo: Path) -> list[tuple[str, re.Pattern[str], re.Pattern[str] | None]]: + data = json.loads((repo / _PATTERNS_FILE).read_text(encoding="utf-8-sig")) + by_id = {p["id"]: p for p in data["patterns"]} + out = [] + for rule in RULES: + if rule.source != "regex": + continue + pat = by_id[rule.pattern_id] + path_re = re.compile(pat["path_regex"]) if pat.get("path_regex") else None + out.append((rule.id, re.compile(pat["pattern_regex"]), path_re)) + return out + + +def _own_complexity(fm: FileMeasure, cc_by_line: dict[int, int]) -> None: + """Ruff's C901 for a function includes every nested def; subtract the direct children so + each function carries only its own branches (nested defs are separate units).""" + full = {q: cc_by_line[u.line] for q, u in fm.units.items() if u.line in cc_by_line} + own = dict(full) + for qual, unit in fm.units.items(): + if unit.parent in own and qual in full: + own[unit.parent] -= full[qual] + for qual, value in own.items(): + fm.units[qual].metrics["CC"] = value + + +class Measurer: + def __init__(self, repo: Path, ruff: list[str], known_env: set[str]) -> None: + self.repo = repo + self.ruff = ruff + self.ctx = py_rules.Ctx(known_env=known_env) + self.regex_rules = _regex_rules(repo) + + def measure(self, tree: str | None, paths: list[str]) -> dict[str, FileMeasure]: + contents = {} + for path in paths: + if in_scope(path): + text = read_file(self.repo, tree, path) + if text is not None: + contents[path] = text + if tree is None: + return self._measure_in(self.repo, contents) + with tempfile.TemporaryDirectory(prefix="code-health-") as tmp: + root = Path(tmp) + for path, text in contents.items(): + dest = root / path + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_text(text, encoding="utf-8") + return self._measure_in(root, contents) + + def _measure_in(self, root: Path, contents: dict[str, str]) -> dict[str, FileMeasure]: + result = {path: FileMeasure(path, lines=text.splitlines()) for path, text in contents.items()} + py = sorted(p for p in contents if in_scope(p) == "py") + ts = sorted(p for p in contents if in_scope(p) == "ts") + ruff_out = run_ruff(self.ruff, root, py) if py else {} + for path in py: + self._python(result[path], contents[path], ruff_out.get(path)) + for path, data in measure_ts(self.repo, root, ts).items(): + self._typescript(result[path], data) + return result + + def _python(self, fm: FileMeasure, text: str, ruff_file) -> None: + fm.metrics["FILE_LINES"] = len(fm.lines) + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", SyntaxWarning) + tree = ast.parse(text) + except SyntaxError: + return + scopes = py_structure.measure_structure(fm, tree) + if ruff_file is not None: + _own_complexity(fm, ruff_file.cc_by_line) + for code, row in ruff_file.hits: + if code in RULES_BY_ID and rule_applies(RULES_BY_ID[code], fm.path): + fm.add_hit(code, scopes.scope(row), row) + for rule_id, checker in py_rules.CHECKERS.items(): + if rule_applies(RULES_BY_ID[rule_id], fm.path): + for row in sorted(set(checker(tree, self.ctx))): + fm.add_hit(rule_id, scopes.scope(row), row) + self._regex(fm, scopes) + + def _regex(self, fm: FileMeasure, scopes) -> None: + for rule_id, pattern, path_re in self.regex_rules: + if not rule_applies(RULES_BY_ID[rule_id], fm.path): + continue + if path_re and not path_re.search(fm.path): + continue + for index, line in enumerate(fm.lines, start=1): + if pattern.search(line): + fm.add_hit(rule_id, scopes.scope(index), index) + + @staticmethod + def _typescript(fm: FileMeasure, data: dict) -> None: + fm.metrics["FILE_LINES"] = len(fm.lines) + for unit in data.get("units", []): + fm.units[unit["q"]] = Unit( + qualname=unit["q"], + line=unit["line"], + metrics={"CC": unit["cc"], "FUNC_LINES": unit["lines"], "NESTING": unit["nesting"]}, + body_hash=unit["hash"], + ) diff --git a/scripts/code_health/model.py b/scripts/code_health/model.py new file mode 100644 index 0000000000..6cb9ad9788 --- /dev/null +++ b/scripts/code_health/model.py @@ -0,0 +1,62 @@ +"""Data shapes shared by the measurers and the comparison.""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass, field + +MODULE_SCOPE = "" + + +@dataclass +class Unit: + """One function (Python def, TS function/method/arrow) and its metrics.""" + + qualname: str + line: int + metrics: dict[str, int] + body_hash: str + parent: str | None = None # nearest enclosing function (nested defs are their own units) + + +@dataclass(frozen=True) +class Hit: + """A pattern-rule fingerprint: never a line number, so edits above don't move it.""" + + rule: str + scope: str + text: str + + +@dataclass +class FileMeasure: + path: str + metrics: dict[str, int] = field(default_factory=dict) + units: dict[str, Unit] = field(default_factory=dict) + hits: Counter[Hit] = field(default_factory=Counter) + # Line numbers of each hit occurrence, for reporting only. + hit_lines: dict[Hit, list[int]] = field(default_factory=dict) + # 1-based line -> source text, for allow comments and messages. + lines: list[str] = field(default_factory=list) + + def add_hit(self, rule: str, scope: str, line_no: int) -> None: + text = " ".join(self.source_line(line_no).split()) + hit = Hit(rule, scope, text) + self.hits[hit] += 1 + self.hit_lines.setdefault(hit, []).append(line_no) + + def source_line(self, line_no: int) -> str: + if 1 <= line_no <= len(self.lines): + return self.lines[line_no - 1] + return "" + + +@dataclass +class Finding: + path: str + rule: str + scope: str + line: int + detail: str + blocking: bool = True + allowed_reason: str | None = None diff --git a/scripts/code_health/py_rules.py b/scripts/code_health/py_rules.py new file mode 100644 index 0000000000..95bbc15d4f --- /dev/null +++ b/scripts/code_health/py_rules.py @@ -0,0 +1,305 @@ +"""Hermes-specific AST rules (the ``HX`` ids in ``config.RULES``). + +Each checker is a small function ``(tree, ctx) -> iterable of line numbers``; the table at the +bottom maps rule ids to checkers. Checkers favour precision: a rule that cries wolf gets an +allow comment on every hit and stops meaning anything. +""" + +from __future__ import annotations + +import ast +from collections.abc import Callable, Iterable, Iterator +from dataclasses import dataclass, field + +_FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef) +_CAPTURE_CALLS = { + "get_hermes_home", + "display_hermes_home", + "load_config", + "load_config_readonly", + "read_raw_config", + "get_secret", + "getcwd", + "expanduser", + "_float_env", + "_int_env", + "_bool_env", + "_env_float", + "_env_int", + "_env_bool", +} +_SYNC_CONFIG_CALLS = {"load_config", "save_config", "read_raw_config", "load_config_readonly"} +_SUBPROCESS_WAITS = {"run", "call", "check_call", "check_output"} +# health: allow HX003 -- the detector's own pattern list +_SHELL_IDENTITY = ("pgrep -f", "ps aux", "ps -ef", "ps -eo") + + +@dataclass +class Ctx: + known_env: set[str] = field(default_factory=set) + + +def _dotted(node: ast.AST) -> str: + """``os.environ.get`` for an Attribute chain, ``""`` for anything else.""" + parts: list[str] = [] + while isinstance(node, ast.Attribute): + parts.append(node.attr) + node = node.value + if isinstance(node, ast.Name): + parts.append(node.id) + return ".".join(reversed(parts)) + if isinstance(node, ast.Call): + inner = _dotted(node.func) + return ".".join([f"{inner}()", *reversed(parts)]) if inner else "" + return "" + + +def _call_name(call: ast.Call) -> str: + return _dotted(call.func) + + +def _str_arg(call: ast.Call, index: int = 0) -> str | None: + if len(call.args) > index: + arg = call.args[index] + if isinstance(arg, ast.Constant) and isinstance(arg.value, str): + return arg.value + return None + + +def _has_kw(call: ast.Call, name: str) -> bool: + return any(kw.arg in (name, None) for kw in call.keywords) + + +def _env_read_name(node: ast.AST) -> str | None: + """Env var name for ``os.getenv("X")`` / ``os.environ.get("X")`` / ``os.environ["X"]``.""" + if isinstance(node, ast.Call): + if _call_name(node) in ("os.getenv", "os.environ.get", "environ.get", "getenv"): + return _str_arg(node) + return None + if isinstance(node, ast.Subscript) and isinstance(node.ctx, ast.Load): + if _dotted(node.value) in ("os.environ", "environ"): + key = node.slice + if isinstance(key, ast.Constant) and isinstance(key.value, str): + return key.value + if isinstance(node, ast.Compare) and len(node.ops) == 1: + if isinstance(node.ops[0], (ast.In, ast.NotIn)): + if _dotted(node.comparators[0]) in ("os.environ", "environ"): + left = node.left + if isinstance(left, ast.Constant) and isinstance(left.value, str): + return left.value + return None + + +def _walk_skipping(node: ast.AST, skip: tuple[type, ...]) -> Iterator[ast.AST]: + """``ast.walk`` that does not descend into ``skip`` node types (except the root).""" + stack = [node] + while stack: + current = stack.pop() + yield current + for child in ast.iter_child_nodes(current): + if not isinstance(child, skip): + stack.append(child) + + +def hardcoded_home(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div): + right = node.right + if ( + isinstance(node.left, ast.Call) + and _call_name(node.left).endswith("Path.home") + and isinstance(right, ast.Constant) + and isinstance(right.value, str) + and right.value.strip("/").startswith(".hermes") + ): + yield node.lineno + elif isinstance(node, ast.Call): + name = _call_name(node) + arg = _str_arg(node) + if arg and arg.startswith("~/.hermes") and name.endswith(("expanduser", "Path")): + yield node.lineno + + +def new_env_var(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + name = _env_read_name(node) + if name and name.startswith("HERMES_") and name not in ctx.known_env: + yield getattr(node, "lineno", 0) + + +def argv_identity(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if isinstance(node, ast.Compare) and len(node.ops) == 1: + if isinstance(node.ops[0], (ast.In, ast.NotIn)): + left = node.left + target = ast.unparse(node.comparators[0]) + if isinstance(left, ast.Constant) and isinstance(left.value, str): + if "cmdline" in target: + yield node.lineno + elif isinstance(node, ast.Constant) and isinstance(node.value, str): + if any(cmd in node.value for cmd in _SHELL_IDENTITY): + yield node.lineno + + +def unscoped_secret_fallback(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if not isinstance(node, ast.ExceptHandler) or node.type is None: + continue + if "UnscopedSecretError" not in ast.unparse(node.type): + continue + for stmt in node.body: + if any(_env_read_name(inner) for inner in ast.walk(stmt)): + yield node.lineno + break + + +def _capture_lines(expr: ast.AST) -> Iterator[int]: + for node in _walk_skipping(expr, (ast.Lambda, *_FUNCS)): + if isinstance(node, ast.Call): + name = _call_name(node) + leaf = name.rsplit(".", 1)[-1] + if leaf in _CAPTURE_CALLS or name.endswith("Path.home"): + yield node.lineno + return + if _env_read_name(node): + yield getattr(node, "lineno", 0) + return + + +def _is_main_guard(stmt: ast.stmt) -> bool: + return isinstance(stmt, ast.If) and "__main__" in ast.unparse(stmt.test) + + +def _import_time_statements(body: list[ast.stmt]) -> Iterator[ast.stmt]: + """Statements that run at import: module/class bodies and their if/try/with blocks.""" + for stmt in body: + if _is_main_guard(stmt): + continue + yield stmt + if isinstance(stmt, ast.ClassDef): + yield from _import_time_statements(stmt.body) + elif isinstance(stmt, (ast.If, ast.Try, ast.With)): + for name in ("body", "orelse", "finalbody"): + yield from _import_time_statements(getattr(stmt, name, None) or []) + for handler in getattr(stmt, "handlers", None) or []: + yield from _import_time_statements(handler.body) + + +def import_time_capture(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for stmt in _import_time_statements(tree.body): + if isinstance(stmt, (ast.Assign, ast.AnnAssign, ast.AugAssign)) and stmt.value: + yield from _capture_lines(stmt.value) + elif isinstance(stmt, _FUNCS): + for default in [*stmt.args.defaults, *stmt.args.kw_defaults]: + if default is not None: + yield from _capture_lines(default) + + +def missing_timeout(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or _has_kw(node, "timeout"): + continue + name = _call_name(node) + head, _, leaf = name.rpartition(".") + if head == "subprocess" and leaf in _SUBPROCESS_WAITS: + yield node.lineno + elif leaf == "urlopen" and len(node.args) < 3: + yield node.lineno + elif leaf == "communicate" and head and not node.args: + yield node.lineno + + +def sync_config_in_async(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for func in ast.walk(tree): + if not isinstance(func, ast.AsyncFunctionDef): + continue + for stmt in func.body: + for node in _walk_skipping(stmt, (ast.Lambda, *_FUNCS)): + if isinstance(node, ast.Call): + if _call_name(node).rsplit(".", 1)[-1] in _SYNC_CONFIG_CALLS: + yield node.lineno + + +def get_event_loop(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _call_name(node) == "asyncio.get_event_loop": + yield node.lineno + + +def _gathers_exceptions(func: ast.AST) -> bool: + for node in ast.walk(func): + if isinstance(node, ast.Call) and _call_name(node).endswith("gather"): + for kw in node.keywords: + if kw.arg == "return_exceptions" and getattr(kw.value, "value", False) is True: + return True + return False + + +def gather_exception_check(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for func in ast.walk(tree): + if not isinstance(func, _FUNCS) or not _gathers_exceptions(func): + continue + for node in ast.walk(func): + if ( + isinstance(node, ast.Call) + and _call_name(node) == "isinstance" + and len(node.args) == 2 + and _dotted(node.args[1]) == "Exception" + ): + yield node.lineno + + +def bool_of_env(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _call_name(node) == "bool" and len(node.args) == 1: + if isinstance(node.args[0], ast.Call) and _env_read_name(node.args[0]): + yield node.lineno + + +def _ladder_key(test: ast.expr) -> str | None: + if isinstance(test, ast.Compare) and len(test.ops) == 1: + if isinstance(test.ops[0], (ast.Eq, ast.In, ast.Is)): + right = test.comparators[0] + if isinstance(right, (ast.Constant, ast.Tuple, ast.Set, ast.List)): + return ast.dump(test.left) + return None + + +def elif_ladder(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + elifs: set[int] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.If) or id(node) in elifs: + continue + keys, current = [], node + while isinstance(current, ast.If): + keys.append(_ladder_key(current.test)) + nxt = current.orelse + if len(nxt) == 1 and isinstance(nxt[0], ast.If): + elifs.add(id(nxt[0])) + current = nxt[0] + else: + break + if len(keys) >= 4 and keys[0] is not None and len(set(keys)) == 1: + yield node.lineno + + +def raw_thread(tree: ast.Module, ctx: Ctx) -> Iterable[int]: + for node in ast.walk(tree): + if isinstance(node, ast.Call) and _call_name(node) in ("threading.Thread", "Thread"): + yield node.lineno + + +CHECKERS: dict[str, Callable[[ast.Module, Ctx], Iterable[int]]] = { + "HX001": hardcoded_home, + "HX002": new_env_var, + "HX003": argv_identity, + "HX004": unscoped_secret_fallback, + "HX005": import_time_capture, + "HX006": missing_timeout, + "HX007": sync_config_in_async, + "HX008": get_event_loop, + "HX009": gather_exception_check, + "HX010": bool_of_env, + "HX011": elif_ladder, + "HX012": raw_thread, +} diff --git a/scripts/code_health/py_structure.py b/scripts/code_health/py_structure.py new file mode 100644 index 0000000000..b228a934ac --- /dev/null +++ b/scripts/code_health/py_structure.py @@ -0,0 +1,144 @@ +"""Python structure: functions (qualname, length, nesting, body hash) and scope lookup. + +Cyclomatic complexity comes from ruff (``ruff_runner``) so the number matches what +``ruff check --select C901`` prints; everything else is measured here from the AST. +""" + +from __future__ import annotations + +import ast +import hashlib +from bisect import bisect_right +from dataclasses import dataclass + +from scripts.code_health.model import MODULE_SCOPE, FileMeasure, Unit + +_FUNCS = (ast.FunctionDef, ast.AsyncFunctionDef) +_BLOCKS = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.With, ast.AsyncWith, ast.Match) +_TRY_STAR = getattr(ast, "TryStar", None) +if _TRY_STAR is not None: + _BLOCKS = (*_BLOCKS, _TRY_STAR) + + +@dataclass +class Span: + start: int + end: int + qualname: str + + +def _child_blocks(node: ast.AST): + """Statement lists directly owned by a compound statement.""" + for name in ("body", "orelse", "finalbody"): + yield getattr(node, name, None) or [] + for handler in getattr(node, "handlers", None) or []: + yield handler.body + for case in getattr(node, "cases", None) or []: + yield case.body + + +def nesting_depth(stmts: list[ast.stmt], depth: int = 0) -> int: + deepest = depth + for stmt in stmts: + if isinstance(stmt, (*_FUNCS, ast.ClassDef)) or not isinstance(stmt, _BLOCKS): + continue + for index, block in enumerate(_child_blocks(stmt)): + # `elif` is an If alone in orelse: same visual depth as its `if`. + is_elif = ( + isinstance(stmt, ast.If) + and index == 1 + and len(block) == 1 + and isinstance(block[0], ast.If) + ) + deepest = max(deepest, nesting_depth(block, depth if is_elif else depth + 1)) + return deepest + + +def body_hash(node: ast.AST) -> str: + """Name-independent hash, so a function moved or renamed unchanged keeps its cap.""" + dumped = ast.dump(node, annotate_fields=False) + name = getattr(node, "name", "") + dumped = dumped.replace(repr(name), "''", 1) + return hashlib.sha1(dumped.encode("utf-8")).hexdigest()[:16] + + +class _UnitCollector(ast.NodeVisitor): + def __init__(self) -> None: + self.stack: list[str] = [] + self.units: dict[str, Unit] = {} + self.spans: list[Span] = [] + self.seen: set[str] = set() + self.funcs: list[str] = [] + self.total_lines: dict[str, int] = {} + + def _qualname(self, name: str) -> str: + """Dotted path without ````; a repeated name (property setter, conditional + def) gets ``#2``, ``#3`` in source order.""" + base = ".".join([*self.stack, name]) + qual, n = base, 1 + while qual in self.seen: + n += 1 + qual = f"{base}#{n}" + self.seen.add(qual) + return qual + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + qual = self._qualname(node.name) + self.spans.append(Span(node.lineno, node.end_lineno or node.lineno, qual)) + self.stack.append(qual.rsplit(".", 1)[-1]) + self.generic_visit(node) + self.stack.pop() + + def _visit_func(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None: + qual = self._qualname(node.name) + end = node.end_lineno or node.lineno + first = min([node.lineno, *(d.lineno for d in node.decorator_list)]) + self.total_lines[qual] = end - first + 1 + self.units[qual] = Unit( + qualname=qual, + line=node.lineno, + metrics={"FUNC_LINES": end - first + 1, "NESTING": nesting_depth(node.body)}, + body_hash=body_hash(node), + parent=self.funcs[-1] if self.funcs else None, + ) + self.spans.append(Span(node.lineno, end, qual)) + self.stack.append(qual.rsplit(".", 1)[-1]) + self.funcs.append(qual) + self.generic_visit(node) + self.funcs.pop() + self.stack.pop() + + visit_FunctionDef = _visit_func + visit_AsyncFunctionDef = _visit_func + + +class ScopeIndex: + """Innermost enclosing def/class qualname for a line.""" + + def __init__(self, spans: list[Span]) -> None: + self.spans = sorted(spans, key=lambda s: (s.start, -s.end)) + self.starts = [s.start for s in self.spans] + + def scope(self, line: int) -> str: + best = MODULE_SCOPE + best_size = None + for span in self.spans[: bisect_right(self.starts, line)]: + if span.start <= line <= span.end: + size = span.end - span.start + if best_size is None or size <= best_size: + best, best_size = span.qualname, size + return best + + +def measure_structure(fm: FileMeasure, tree: ast.Module) -> ScopeIndex: + collector = _UnitCollector() + collector.visit(tree) + # FUNC_LINES is a function's OWN lines: a nested def is its own unit, so editing a closure + # never counts against the function that encloses it. + for unit in collector.units.values(): + if unit.parent is not None: + parent = collector.units[unit.parent] + parent.metrics["FUNC_LINES"] -= collector.total_lines[unit.qualname] + fm.units = collector.units + fm.metrics["FILE_LINES"] = len(fm.lines) + return ScopeIndex(collector.spans) diff --git a/scripts/code_health/replay.py b/scripts/code_health/replay.py new file mode 100644 index 0000000000..dbcc597a8e --- /dev/null +++ b/scripts/code_health/replay.py @@ -0,0 +1,116 @@ +"""Replay the ratchet over recently merged PRs: what would it have blocked? + + python -m scripts.code_health.replay --limit 300 --out + +Writes ``/replay.jsonl`` (one row per PR) and prints per-rule totals. Use it before +promoting a rule to blocking or changing a target: a rule whose hits on merged PRs are mostly +legitimate code ships advisory until its checker is fixed. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +from collections import Counter +from dataclasses import asdict +from pathlib import Path + +from scripts.code_health import gitio +from scripts.code_health.compare import compare +from scripts.code_health.config import in_scope +from scripts.code_health.measure import Measurer +from scripts.code_health.report import apply_allows +from scripts.code_health.ruff_runner import resolve_ruff + +_QUERY = """ +query($q: String!, $after: String) { + search(query: $q, type: ISSUE, first: 50, after: $after) { + pageInfo { hasNextPage endCursor } + nodes { ... on PullRequest { + number title mergedAt + mergeCommit { oid } + commits(last: 100) { totalCount nodes { commit { messageHeadline } } } + } } + } +}""" + + +def merged_prs(repo: Path, limit: int) -> list[dict]: + prs: list[dict] = [] + after = None + while len(prs) < limit: + args = ["gh", "api", "graphql", "-f", f"query={_QUERY}", + "-f", "q=repo:NousResearch/hermes-agent is:pr is:merged base:main sort:updated-desc"] + if after: + args += ["-f", f"after={after}"] + out = subprocess.run(args, cwd=repo, capture_output=True, text=True, encoding="utf-8", check=True, + timeout=120, stdin=subprocess.DEVNULL).stdout + data = json.loads(out)["data"]["search"] + prs += [n for n in data["nodes"] if n.get("mergeCommit")] + if not data["pageInfo"]["hasNextPage"]: + break + after = data["pageInfo"]["endCursor"] + return prs[:limit] + + +def pr_range(repo: Path, pr: dict) -> tuple[str, str]: + """Base/head on main for a rebase-merged PR (its commits' subjects), else the merge parent.""" + head = pr["mergeCommit"]["oid"] + subjects = {n["commit"]["messageHeadline"] for n in pr["commits"]["nodes"]} + log = gitio.git(repo, "log", "--first-parent", "--format=%H%x00%s", "-n", + str(pr["commits"]["totalCount"] + 1), head).splitlines() + k = 0 + for line in log: + _, subject = line.split("\0", 1) + if subject not in subjects: + break + k += 1 + return gitio.resolve_rev(repo, f"{head}~{max(k, 1)}"), head + + +def replay_one(repo: Path, measurer: Measurer, base: str, head: str) -> list[dict]: + changes = gitio.changed_files(repo, base, head) + head_paths = sorted({c.new for c in changes if c.new and in_scope(c.new)}) + base_paths = sorted({c.old for c in changes if c.old and in_scope(c.old)}) + if not head_paths: + return [] + measurer.ctx.known_env = gitio.known_env_names(repo, base) + base_m = measurer.measure(base, base_paths) + head_m = measurer.measure(head, head_paths) + findings = compare(base_m, head_m, changes) + apply_allows(findings, head_m) + return [asdict(f) for f in findings if f.allowed_reason is None] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="code_health.replay") + parser.add_argument("--limit", type=int, default=300) + parser.add_argument("--out", required=True) + args = parser.parse_args(argv) + repo = gitio.repo_root(Path.cwd()) + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + measurer = Measurer(repo, resolve_ruff(repo), known_env=set()) + per_rule: Counter[str] = Counter() + prs_blocked: Counter[str] = Counter() + with (out_dir / "replay.jsonl").open("w", encoding="utf-8") as fh: + for pr in merged_prs(repo, args.limit): + try: + base, head = pr_range(repo, pr) + findings = replay_one(repo, measurer, base, head) + except RuntimeError as exc: + findings, base, head = [{"error": str(exc)}], "", "" + row = {"number": pr["number"], "title": pr["title"], "base": base, "head": head, + "findings": findings} + fh.write(json.dumps(row) + "\n") + rules = {f["rule"] for f in findings if "rule" in f} + prs_blocked.update(rules) + per_rule.update(f["rule"] for f in findings if "rule" in f) + for rule, count in per_rule.most_common(): + print(f"{rule:<11} findings {count:>4} PRs {prs_blocked[rule]:>4}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/code_health/report.py b/scripts/code_health/report.py new file mode 100644 index 0000000000..02722b84e7 --- /dev/null +++ b/scripts/code_health/report.py @@ -0,0 +1,98 @@ +"""Allow comments and the agent-facing report.""" + +from __future__ import annotations + +import re +from collections import Counter + +from scripts.code_health.config import ALLOW_SYNTAX, METRIC_FIX, RULES_BY_ID, TARGETS +from scripts.code_health.model import FileMeasure, Finding + +_ALLOW = re.compile(r"health:\s*allow\s+([A-Za-z0-9_,\- ]+?)\s*(?:--|—|:)\s*(\S.*)$") +_ALLOW_BARE = re.compile(r"health:\s*allow\b") + + +def _allow_lines(finding: Finding) -> list[int]: + if finding.rule == "FILE_LINES": + return list(range(1, 6)) + return [finding.line, finding.line - 1] + + +def apply_allows(findings: list[Finding], head: dict[str, FileMeasure]) -> None: + """Mark findings whose line carries ``# health: allow RULE -- why`` as allowed. + + A bare allow without a reason does not count; the finding stays and says why. + """ + for finding in findings: + fm = head.get(finding.path) + if fm is None: + continue + for line_no in _allow_lines(finding): + text = fm.source_line(line_no) + match = _ALLOW.search(text) + if match and finding.rule in {r.strip() for r in match.group(1).split(",")}: + finding.allowed_reason = match.group(2).strip() + break + if _ALLOW_BARE.search(text) and finding.rule in text: + finding.detail += f" [allow comment ignored: it needs a reason, {ALLOW_SYNTAX}]" + + +def _fix_text(rule_id: str) -> str: + if rule_id in METRIC_FIX: + return METRIC_FIX[rule_id] + return RULES_BY_ID[rule_id].fix + + +def format_findings(findings: list[Finding]) -> str: + out: list[str] = [] + for f in sorted(findings, key=lambda f: (f.path, f.line, f.rule)): + if f.allowed_reason is not None: + continue + tag = "" if f.blocking else " (advisory)" + out.append(f"{f.path}:{f.line} {f.rule}{tag} {f.scope}") + out.append(f" {f.detail}") + out.append(f" fix: {_fix_text(f.rule)}") + allowed = [f for f in findings if f.allowed_reason is not None] + if allowed: + out.append("") + out.append("Allowed by comment (reviewers: check the reasons):") + for f in allowed: + out.append(f" {f.path}:{f.line} {f.rule} {f.scope} -- {f.allowed_reason}") + return "\n".join(out) + + +def verdict(findings: list[Finding]) -> tuple[int, int]: + blocking = sum(1 for f in findings if f.blocking and f.allowed_reason is None) + advisory = sum(1 for f in findings if not f.blocking and f.allowed_reason is None) + return blocking, advisory + + +def summarize_tree(measures: dict[str, FileMeasure]) -> str: + """Burn-down view of a whole tree: units over target and hits per rule.""" + over: Counter[str] = Counter() + worst: dict[str, tuple[int, str]] = {} + hits: Counter[str] = Counter() + for path, fm in measures.items(): + if fm.metrics.get("FILE_LINES", 0) > TARGETS["FILE_LINES"]: + over["FILE_LINES"] += 1 + _track(worst, "FILE_LINES", fm.metrics["FILE_LINES"], path) + for unit in fm.units.values(): + for metric, value in unit.metrics.items(): + if value > TARGETS[metric]: + over[metric] += 1 + _track(worst, metric, value, f"{path}::{unit.qualname}") + for hit, count in fm.hits.items(): + hits[hit.rule] += count + lines = ["Over target (units that are baselined at their current value):"] + for metric, target in TARGETS.items(): + value, where = worst.get(metric, (0, "-")) + lines.append(f" {metric:<11} target {target:<5} over: {over[metric]:<5} worst: {value} {where}") + lines.append("Pattern hits (existing ones are baselined; new ones fail):") + for rule_id, count in sorted(hits.items(), key=lambda kv: -kv[1]): + lines.append(f" {rule_id:<9} {count:>6} {RULES_BY_ID[rule_id].title}") + return "\n".join(lines) + + +def _track(worst: dict[str, tuple[int, str]], metric: str, value: int, where: str) -> None: + if value > worst.get(metric, (0, ""))[0]: + worst[metric] = (value, where) diff --git a/scripts/code_health/ruff_runner.py b/scripts/code_health/ruff_runner.py new file mode 100644 index 0000000000..b53048dda6 --- /dev/null +++ b/scripts/code_health/ruff_runner.py @@ -0,0 +1,101 @@ +"""Run the pinned ruff and split its output into per-function CC and pattern hits. + +The pin lives in pyproject's ``dev`` dependency group (the version CI installs); a different +local ruff is never used silently, because rule behaviour and CC counts drift between releases. +""" + +from __future__ import annotations + +import json +import re +import shutil +import subprocess +import sys +import tomllib +from dataclasses import dataclass, field +from pathlib import Path + +from scripts.code_health.config import RUFF_CODES + +_CC_MESSAGE = re.compile(r"is too complex \((\d+) >") +_CHUNK = 400 + + +@dataclass +class RuffFile: + cc_by_line: dict[int, int] = field(default_factory=dict) + hits: list[tuple[str, int]] = field(default_factory=list) + + +def pinned_version(repo: Path) -> str: + data = tomllib.loads((repo / "pyproject.toml").read_text(encoding="utf-8-sig")) + for spec in data.get("dependency-groups", {}).get("dev", []): + if isinstance(spec, str) and spec.startswith("ruff=="): + return spec.split("==", 1)[1].strip() + raise RuntimeError("pyproject.toml [dependency-groups].dev has no ruff== pin") + + +def _version_of(cmd: list[str]) -> str | None: + try: + out = subprocess.run( + [*cmd, "--version"], capture_output=True, text=True, timeout=60, + stdin=subprocess.DEVNULL, check=False, + ).stdout + except (OSError, subprocess.TimeoutExpired): + return None + parts = out.split() + return parts[1] if len(parts) >= 2 and parts[0] == "ruff" else None + + +def resolve_ruff(repo: Path) -> list[str]: + """Command prefix for the pinned ruff: a matching install, else ``uvx ruff@``.""" + pin = pinned_version(repo) + candidates = [Path(sys.prefix) / "bin" / "ruff", Path(sys.prefix) / "Scripts" / "ruff.exe"] + on_path = shutil.which("ruff") + if on_path: + candidates.insert(0, Path(on_path)) + for cand in candidates: + if cand.exists() and _version_of([str(cand)]) == pin: + return [str(cand)] + uvx = shutil.which("uvx") + if uvx and _version_of([uvx, f"ruff@{pin}"]) == pin: + return [uvx, f"ruff@{pin}"] + found = _version_of(["ruff"]) if on_path else None + raise RuntimeError( + f"code health needs ruff {pin} (pyproject dev group); found {found or 'none'}. " + f"Install it (`uv tool install ruff=={pin}`) or put uv on PATH so `uvx ruff@{pin}` works." + ) + + +def run_ruff(ruff: list[str], root: Path, paths: list[str]) -> dict[str, RuffFile]: + """Ruff over ``paths`` (relative to ``root``), keyed by the same relative path.""" + results: dict[str, RuffFile] = {p: RuffFile() for p in paths} + select = ",".join(("C901", *RUFF_CODES)) + for start in range(0, len(paths), _CHUNK): + chunk = paths[start : start + _CHUNK] + proc = subprocess.run( + [*ruff, "check", "--isolated", "--ignore-noqa", "--no-cache", "--preview", + "--target-version", "py311", "--select", select, + "--config", "lint.mccabe.max-complexity=0", + "--output-format", "json", "--exit-zero", *chunk], + cwd=root, capture_output=True, text=True, encoding="utf-8", timeout=600, + stdin=subprocess.DEVNULL, check=False, + ) + if proc.returncode != 0: + raise RuntimeError(f"ruff failed: {proc.stderr.strip()[:2000]}") + for diag in json.loads(proc.stdout or "[]"): + _record(results, root, diag) + return results + + +def _record(results: dict[str, RuffFile], root: Path, diag: dict) -> None: + rel = Path(diag["filename"]).resolve().relative_to(root.resolve()).as_posix() + entry = results.setdefault(rel, RuffFile()) + row = diag["location"]["row"] + code = diag.get("code") + if code == "C901": + match = _CC_MESSAGE.search(diag.get("message", "")) + if match: + entry.cc_by_line[row] = int(match.group(1)) + elif code: + entry.hits.append((code, row)) diff --git a/scripts/code_health/ts_measure.py b/scripts/code_health/ts_measure.py new file mode 100644 index 0000000000..6ed7fd436b --- /dev/null +++ b/scripts/code_health/ts_measure.py @@ -0,0 +1,69 @@ +"""TypeScript measurement: runs ``ts_units.mjs`` under node with the lockfile's typescript.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path + +from scripts.code_health.gitio import git + +_SCRIPT = Path(__file__).with_name("ts_units.mjs") + + +def pinned_typescript(repo: Path) -> str: + lock = json.loads((repo / "package-lock.json").read_text(encoding="utf-8-sig")) + return lock["packages"]["node_modules/typescript"]["version"] + + +def _version_at(pkg_dir: Path) -> str | None: + try: + return json.loads((pkg_dir / "package.json").read_text(encoding="utf-8-sig"))["version"] + except (OSError, ValueError, KeyError): + return None + + +def _cache_root() -> Path: + base = os.environ.get("XDG_CACHE_HOME") or str(Path.home() / ".cache") + return Path(base) / "hermes-code-health" + + +def resolve_typescript(repo: Path) -> Path: + """Directory of the pinned ``typescript`` package, installing it into a cache if absent.""" + pin = pinned_typescript(repo) + common = Path(git(repo, "rev-parse", "--path-format=absolute", "--git-common-dir").strip()) + for root in (repo, common.parent): + cand = root / "node_modules" / "typescript" + if _version_at(cand) == pin: + return cand + prefix = _cache_root() / f"typescript-{pin}" + cand = prefix / "node_modules" / "typescript" + if _version_at(cand) != pin: + npm = shutil.which("npm") + if not npm: + raise RuntimeError("code health needs node + npm to measure TypeScript files") + subprocess.run( + [npm, "install", "--no-save", "--no-package-lock", "--no-audit", "--no-fund", + "--prefix", str(prefix), f"typescript@{pin}"], + check=True, capture_output=True, timeout=300, stdin=subprocess.DEVNULL, + ) + return cand + + +def measure_ts(repo: Path, root: Path, paths: list[str]) -> dict[str, dict]: + if not paths: + return {} + node = shutil.which("node") + if not node: + raise RuntimeError("code health needs node to measure TypeScript files") + ts_dir = resolve_typescript(repo) + proc = subprocess.run( + [node, str(_SCRIPT), str(ts_dir), str(root)], + input=json.dumps(paths), capture_output=True, text=True, encoding="utf-8", + timeout=600, check=False, + ) + if proc.returncode != 0: + raise RuntimeError(f"ts_units.mjs failed: {proc.stderr.strip()[:2000]}") + return json.loads(proc.stdout) diff --git a/scripts/code_health/ts_units.mjs b/scripts/code_health/ts_units.mjs new file mode 100644 index 0000000000..caed121c42 --- /dev/null +++ b/scripts/code_health/ts_units.mjs @@ -0,0 +1,141 @@ +// TypeScript structure for the code-health ratchet: per-function CC, length, nesting. +// Parses only (no type check), so it is fast and needs nothing but the `typescript` package. +// +// node ts_units.mjs < paths.json > units.json +// +// CC is classic McCabe, like ruff's: 1 + if, ?:, case, loops, catch, &&, ||, ??. +import { createHash } from 'node:crypto' +import { readFileSync } from 'node:fs' +import { createRequire } from 'node:module' +import { join } from 'node:path' + +const [tsPath, root] = process.argv.slice(2) +const ts = createRequire(import.meta.url)(tsPath) +const K = ts.SyntaxKind + +const FUNCTION_KINDS = new Set([ + K.FunctionDeclaration, K.MethodDeclaration, K.ArrowFunction, K.FunctionExpression, + K.Constructor, K.GetAccessor, K.SetAccessor +]) +const BRANCH_KINDS = new Set([ + K.IfStatement, K.ConditionalExpression, K.ForStatement, K.ForInStatement, K.ForOfStatement, + K.WhileStatement, K.DoStatement, K.CatchClause +]) +const BLOCK_KINDS = new Set([ + K.IfStatement, K.ForStatement, K.ForInStatement, K.ForOfStatement, K.WhileStatement, + K.DoStatement, K.SwitchStatement, K.TryStatement +]) +const LOGICAL_OPS = new Set([ + K.AmpersandAmpersandToken, K.BarBarToken, K.QuestionQuestionToken +]) + +function complexity(fn) { + let cc = 1 + const visit = node => { + if (node !== fn && FUNCTION_KINDS.has(node.kind)) return + if (BRANCH_KINDS.has(node.kind)) cc++ + else if (node.kind === K.CaseClause) cc++ + else if (node.kind === K.BinaryExpression && LOGICAL_OPS.has(node.operatorToken.kind)) cc++ + ts.forEachChild(node, visit) + } + visit(fn) + return cc +} + +function nesting(fn) { + let deepest = 0 + const visit = (node, depth) => { + if (node !== fn && FUNCTION_KINDS.has(node.kind)) return + let next = depth + // `else if` stays at the depth of its `if`. + const isElseIf = node.kind === K.IfStatement && node.parent?.kind === K.IfStatement && + node.parent.elseStatement === node + if (BLOCK_KINDS.has(node.kind) && !isElseIf) { + next = depth + 1 + deepest = Math.max(deepest, next) + } + ts.forEachChild(node, child => visit(child, next)) + } + visit(fn, 0) + return deepest +} + +function propertyName(name) { + if (!name) return null + if (ts.isIdentifier(name) || ts.isPrivateIdentifier(name) || ts.isStringLiteral(name) || + ts.isNumericLiteral(name)) return name.text + return null +} + +// Name an anonymous function from where it is bound: `const x = () => {}`, +// `x: () => {}`, `const x = useCallback(() => {})`. +function bindingName(node) { + let current = node.parent + while (current && (ts.isCallExpression(current) || ts.isParenthesizedExpression(current) || + ts.isAsExpression(current) || ts.isSatisfiesExpression?.(current))) { + if (ts.isCallExpression(current) && current.parent && ts.isExpressionStatement(current.parent)) { + return null + } + current = current.parent + } + if (!current) return null + if (ts.isVariableDeclaration(current) || ts.isPropertyAssignment(current) || + ts.isPropertyDeclaration(current)) return propertyName(current.name) + return null +} + +function ownName(node) { + if (node.kind === K.Constructor) return 'constructor' + const named = propertyName(node.name) + if (named) return node.kind === K.GetAccessor ? `get ${named}` : node.kind === K.SetAccessor ? `set ${named}` : named + return bindingName(node) +} + +function measureFile(path) { + const text = readFileSync(join(root, path), 'utf8') + const kind = path.endsWith('x') ? ts.ScriptKind.TSX : ts.ScriptKind.TS + const sf = ts.createSourceFile(path, text, ts.ScriptTarget.Latest, true, kind) + const units = [] + const seen = new Map() + const unique = q => { + const n = (seen.get(q) ?? 0) + 1 + seen.set(q, n) + return n === 1 ? q : `${q}#${n}` + } + // `lines` is a function's OWN lines: nested functions are their own units, so editing a + // closure inside a hook never counts against the hook itself. + const walk = (node, stack, parent) => { + let next = stack + let nextParent = parent + if (ts.isClassDeclaration(node) || ts.isClassExpression(node)) { + next = [...stack, propertyName(node.name) ?? ''] + } else if (FUNCTION_KINDS.has(node.kind)) { + const qual = unique([...stack, ownName(node) ?? ''].join('.')) + const start = sf.getLineAndCharacterOfPosition(node.getStart(sf)).line + 1 + const end = sf.getLineAndCharacterOfPosition(node.getEnd()).line + 1 + const body = node.getText(sf).replace(/\s+/g, ' ') + const unit = { + q: qual, line: start, lines: end - start + 1, cc: complexity(node), nesting: nesting(node), + hash: createHash('sha1').update(body).digest('hex').slice(0, 16) + } + if (parent) parent.lines -= end - start + 1 + units.push(unit) + next = [...stack, qual.split('.').pop()] + nextParent = unit + } + ts.forEachChild(node, child => walk(child, next, nextParent)) + } + walk(sf, [], null) + return { units } +} + +const paths = JSON.parse(readFileSync(0, 'utf8')) +const out = {} +for (const path of paths) { + try { + out[path] = measureFile(path) + } catch (error) { + out[path] = { error: String(error) } + } +} +process.stdout.write(JSON.stringify(out)) diff --git a/tests/scripts/test_code_health.py b/tests/scripts/test_code_health.py new file mode 100644 index 0000000000..29fc800252 --- /dev/null +++ b/tests/scripts/test_code_health.py @@ -0,0 +1,74 @@ +"""Invariants of the code-health ratchet (scripts/code_health), on a real git repo + pinned ruff.""" + +from __future__ import annotations + +import shutil +import subprocess +from pathlib import Path + +from scripts.code_health.cli import run + +REPO = Path(__file__).resolve().parents[2] +_LEGACY = "def legacy(x):\n" + "".join(f" if x == {i}:\n return {i}\n" for i in range(21)) +_SWALLOW = "def other():\n try:\n pass\n except Exception:\n pass\n" + + +def _git(repo: Path, *args: str) -> str: + return subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True, text=True, + encoding="utf-8", timeout=60).stdout.strip() + + +def _commit(repo: Path, files: dict[str, str | None]) -> str: + for rel, text in files.items(): + path = repo / rel + if text is None: + path.unlink() + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + _git(repo, "add", "-A") + _git(repo, "commit", "-q", "-m", "step") + return _git(repo, "rev-parse", "HEAD") + + +def _repo(tmp_path: Path) -> tuple[Path, str]: + repo = tmp_path / "repo" + (repo / "scripts" / "ci").mkdir(parents=True) + shutil.copy(REPO / "pyproject.toml", repo / "pyproject.toml") + shutil.copy(REPO / "scripts/ci/profile_scope_patterns.json", repo / "scripts/ci/") + _git(repo, "init", "-q", "-b", "main") + _git(repo, "config", "user.email", "t@example.com") + _git(repo, "config", "user.name", "t") + return repo, _commit(repo, {"pkg/a.py": _LEGACY + "\n\n" + _SWALLOW}) + + +def _verdict(repo: Path, base: str, files: dict[str, str | None], capsys) -> tuple[int, str]: + head = _commit(repo, files) + code = run(repo, base, head) + out = capsys.readouterr().out + _git(repo, "reset", "-q", "--hard", base) + return code, out + + +def test_ratchet_blocks_growth_new_and_swapped_violations(tmp_path, capsys): + repo, base = _repo(tmp_path) + legacy_plus = _LEGACY + " if x == 99:\n return 99\n" + ok = _LEGACY + "\n\n" + _SWALLOW + "\n\ndef added():\n return 1\n" + grown = legacy_plus + "\n\n" + _SWALLOW + swapped = _LEGACY + "\n\ndef other():\n return 1\n\n\n" + _SWALLOW.replace("other", "fresh") + + assert _verdict(repo, base, {"pkg/a.py": ok}, capsys)[0] == 0 # legacy debt never blocks + code, out = _verdict(repo, base, {"pkg/a.py": grown}, capsys) + assert code == 1 and "legacy" in out and "CC 23 > 22" in out + code, out = _verdict(repo, base, {"pkg/a.py": swapped}, capsys) + assert code == 1 and "BLE001" in out and "fresh" in out # fixing one doesn't buy another + + +def test_moved_code_keeps_its_cap(tmp_path, capsys): + repo, base = _repo(tmp_path) + split: dict[str, str | None] = {"pkg/a.py": _SWALLOW, "pkg/a_legacy.py": _LEGACY} + code, out = _verdict(repo, base, split, capsys) + assert code == 0, out + renamed: dict[str, str | None] = {"pkg/a.py": None, "pkg/b.py": _LEGACY + "\n\n" + _SWALLOW} + code, out = _verdict(repo, base, renamed, capsys) + assert code == 0, out