feat: code-health ratchet with per-unit caps for Python and TypeScript

scripts/code_health measures the files a change touches on both sides (base revision and head) and fails when: - a function or file over target grows past its own value on the base (CC 20, 300 own lines, nesting 6, file 2,000 lines); - a new function or file is over target; - a pattern rule gains a fingerprint (rule + enclosing function + line text), so fixing one violation never buys another. Code that moves keeps its cap: functions match by name-independent body hash, pattern hits by (rule, line text) once their function has left the old place. Nested functions are their own units, so a closure edit never counts against the hook or function around it. Python CC is ruff's C901 minus the direct nested defs; TypeScript is parsed with the lockfile's typescript (no type check). Rules: swallowed exceptions (S110, BLE001, E722), correctness (F821/F823/ F811/F841, RUF006), hangs (ASYNC230/240, S113, subprocess/urlopen without timeout, sync config I/O in async def, get_event_loop, gather results checked against Exception), profile scope (hardcoded ~/.hermes, a new HERMES_* env var, UnscopedSecretError -> os.getenv, import-time capture, raw threading.Thread, P05/P06 from profile_scope_patterns.json), argv process identity, bool() of an env string, and if/elif ladders. Growth of a file already over 2,000 lines is advisory: replaying 300 merged PRs, a blocking file-size ratchet would have stopped 46 of them before the 119 over-target files are split. Everything else blocks; the same replay finds 18 PRs that would have needed a change, every finding a real hit for its rule. No baseline file: base is measured on the fly, so an improvement that lands on main is the new cap at once and parallel PRs never conflict on ratchet state. `# health: allow <RULE> -- <why>` is the reviewed escape.
2026-10-04 00:37:45 -07:00
parent a6b1fb4e66
commit 4156715960
17 changed files with 1815 additions and 0 deletions
--- a/scripts/check
+++ 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())
--- a/scripts/code_health/init.py
+++ b/scripts/code_health/init.py
@@ -0,0 +1 @@
+"""Code-health ratchet: per-unit caps that only go down. Entry point: ``scripts/check``."""
--- a/scripts/code_health/main.py
+++ b/scripts/code_health/main.py
@@ -0,0 +1,3 @@
+from scripts.code_health.cli import main
+
+raise SystemExit(main())
--- a/scripts/code_health/cli.py
+++ 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 <RULE> -- <why>`. 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
--- a/scripts/code_health/compare.py
+++ 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
--- a/scripts/code_health/config.py
+++ 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 (`<stem>_<topic>.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 `<stem>_<topic>` (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 <RULE> -- <why>"
+
+
+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
--- a/scripts/code_health/gitio.py
+++ 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 <tree>` prefixes "<tree>:" even with -h on some versions.
+        names.add(line.rsplit(":", 1)[-1])
+    return names
--- a/scripts/code_health/measure.py
+++ 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"],
+            )
--- a/scripts/code_health/model.py
+++ 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 = "<module>"
+
+
+@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
--- a/scripts/code_health/py_rules.py
+++ 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,
+}
--- a/scripts/code_health/py_structure.py
+++ 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 ``<locals>``; 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)
--- a/scripts/code_health/replay.py
+++ 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 <dir>
+
+Writes ``<dir>/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())
--- a/scripts/code_health/report.py
+++ 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)
--- a/scripts/code_health/ruff_runner.py
+++ 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==<version> 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>``."""
+    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))
--- a/scripts/code_health/ts_measure.py
+++ 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)
--- a/scripts/code_health/ts_units.mjs
+++ 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 <typescript-module-path> <root>  < 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) ?? '<class>']
+    } else if (FUNCTION_KINDS.has(node.kind)) {
+      const qual = unique([...stack, ownName(node) ?? '<anon>'].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))
--- a/tests/scripts/test_code_health.py
+++ 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
				`@@ -0,0 +1 @@`
				"""Code-health ratchet: per-unit caps that only go down. Entry point: ``scripts/check``."""