Skills › Compound Engineering › ce-code-review › review-scope.py
review-scope.py
419 lines · skills/ce-code-review/scripts/review-scope.py at 3449230
1#!/usr/bin/env python3
2"""Compute fail-closed, deterministic scope facts for ce-code-review.
4The helper never awards lite. It reports counts, path classes, and floors
5the skill's Review depth gate reads. `hard_block_full` forces the full
6spine: a path class the script can name, a file it could not count, or a
7change whose executable non-test lines reach the full floor. Below that
8floor, size is a fact the gate reads, never a decision; the agent judges
9consequence.
10"""
12from __future__ import annotations
14import argparse
15import functools
16import json
17import os
18import re
19import subprocess
20import sys
21from pathlib import Path
24CODE_EXTENSIONS = {
25 ".rb", ".py", ".js", ".mjs", ".cjs", ".jsx", ".ts", ".tsx", ".go", ".rs",
26 ".java", ".swift", ".kt", ".c", ".cc", ".cpp", ".cs", ".php",
27 ".ex", ".exs", ".scala",
28 ".sh", ".bash", ".zsh", ".fish", ".ps1", ".pl", ".pm", ".lua", ".dart",
29 ".vue", ".svelte",
30}
32SIGNAL_PATTERNS = {
33 "migrations": re.compile(
34 r"db/migrate/|schema\.(rb|sql)|/migrations?/|alembic|flyway|liquibase",
35 re.I,
36 ),
37 "frontend": re.compile(
38 r"\.(tsx|jsx|vue|svelte|css|scss|html|erb|haml)$|/components?/|stimulus|turbo",
39 re.I,
40 ),
41 "api": re.compile(
42 r"/(routes?|controllers?|api|serializers?|graphql)/|\.proto$|openapi|swagger",
43 re.I,
44 ),
45 "swift-ios": re.compile(r"\.(swift|kt|pbxproj|xcconfig|entitlements)$", re.I),
46}
48# Classes the script can name from paths alone. These force full; they do not
49# award lite.
50HARD_BLOCK_PATTERNS = {
51 "migrations": SIGNAL_PATTERNS["migrations"],
52}
54# Silent-pass guards the script can name from paths alone. These forbid lite
55# (the change needs the adversarial read the focused path carries) but do not
56# force full; consequence still decides focused versus full. Silent-pass guards
57# outside these paths are the agent's question.
58SILENT_PASS_PATTERNS = {
59 "ci": re.compile(
60 r"(^|/)\.github/workflows/|(^|/)\.gitlab-ci\.yml$|(^|/)\.gitlab-ci/"
61 r"|(^|/)Jenkinsfile$|(^|/)\.circleci/|(^|/)\.buildkite/",
62 re.I,
63 ),
64}
66# Executable non-test changed lines at or above this run the full spine; it
67# matches the maintainability reviewer's trigger. Below it, consequence decides.
68# Whole-file deletions never count toward it: they band separately, and that
69# band can forbid lite but never forces full.
70FULL_EXEC_LINE_MIN = 200
72# Conventions recognized: tests?/spec/__tests__ directories; a .test./.spec.
73# suffix; a test_*.py / conftest.py Python prefix; and a case-sensitive
74# Test/Tests/Spec class-file suffix (Java/C#/Scala/Swift/Kotlin), so that
75# Contest.java or Manifest.cs stays production code.
76TEST_PATTERN = re.compile(
77 r"(^|/)(tests?|spec|__tests__)/"
78 r"|(^|/)[^/]+[._-](test|spec)\.[^/]+$"
79 r"|(^|/)(test_[^/]+|conftest)\.[^/]+$"
80 r"|(?-i:(^|/)[^/]+(Test|Tests|Spec)\.(java|kt|scala|swift|cs)$)",
81 re.I,
82)
83AGENT_SURFACE_PATTERN = re.compile(
84 r"(^|/)(skills?|agents?|prompts?|tools?|mcp|commands?)(/|$)|SKILL\.md$|"
85 r"(^|/)(AGENTS|CLAUDE|GEMINI)\.md$|\.cursor/|\.codex-plugin/|\.claude-plugin/",
86 re.I,
87)
90def git(*args: str) -> subprocess.CompletedProcess[str]:
91 return subprocess.run(
92 ["git", *args], capture_output=True, text=True, check=False
93 )
96def valid_commit(ref: str | None) -> bool:
97 if not ref:
98 return False
99 return git("rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}").returncode == 0
102def unique_merge_base(base: str, head: str) -> str | None:
103 result = git("merge-base", "--all", base, head)
104 candidates = [line for line in result.stdout.splitlines() if line]
105 if result.returncode != 0 or len(candidates) != 1:
106 return None
107 return candidates[0]
110DEFAULT_DOCS_ROOT = "docs"
113def normalize_docs_root(docs_root: str | None) -> str:
114 """Fall back to the default root for an unset, empty, or unsubstituted value.
116 The calling skill substitutes a resolved path for the ``<root>`` placeholder
117 before invoking this script. If that substitution is missing — the value is
118 empty, or still contains angle brackets (a literal ``<root>``) — treat it as
119 unset and use the default ``docs``, which is exactly the block's unset
120 behavior. This keeps the common default-config case correct even when the
121 caller forgets to substitute.
122 """
123 if not docs_root or "<" in docs_root or ">" in docs_root:
124 return DEFAULT_DOCS_ROOT
125 return docs_root
128@functools.lru_cache(maxsize=None)
129def repo_root() -> Path:
130 """The repository root, matching how docs_root is resolved everywhere else.
132 docs_root is repo-relative (``<repo-root>/<docs_root>``), so the corpus
133 check must resolve against the git toplevel, not the current working
134 directory. ce-code-review can run from a subdirectory (``git diff`` still
135 works there), where ``Path.cwd()`` would join docs_root under the subdir and
136 wrongly report the corpus absent. Fall back to cwd when git can't answer.
137 """
138 result = git("rev-parse", "--show-toplevel")
139 if result.returncode == 0 and result.stdout.strip():
140 return Path(result.stdout.strip()).resolve()
141 return Path.cwd().resolve()
144def has_learnings_corpus(docs_root: str | None) -> bool:
145 """Whether a `<docs_root>/solutions` learnings corpus exists.
147 docs_root is the artifact root resolved by the calling skill (default
148 ``docs``). Guard it the way the skill-prose rule does: normalize an
149 unset/placeholder value to the default, and treat a value that is absolute
150 or escapes the repository as absent rather than probing an out-of-repo path.
151 """
152 docs_root = normalize_docs_root(docs_root)
153 if os.path.isabs(docs_root):
154 return False
155 repo = repo_root()
156 candidate = (repo / docs_root / "solutions").resolve()
157 if repo not in candidate.parents and candidate != repo:
158 return False
159 return candidate.is_dir()
162PACKS_RESOLVER = Path(__file__).resolve().parent / "packs-resolve.py"
163# Parse-only mode does no git or cache work, so this bound only guards against a
164# wedged interpreter; the helper is meant to be cheap and must never hang scope.
165PACKS_RESOLVER_TIMEOUT = 30.0
168def declared_packs() -> tuple[bool | None, int]:
169 """Whether the local CE config declares Compound Packs, from the config alone.
171 Runs the sibling resolver in `--declared-only` mode, which parses the
172 `packs:` list from both CE config layers and shape-checks each entry with no
173 git or cache work. Its `declared` is true when any entry parsed or the block
174 is malformed -- a broken declaration is still one the learnings pass must
175 surface in Coverage. ``None`` means the helper could not tell (resolver
176 missing, crashed, timed out, or answered without `declared`); the caller
177 then falls closed to reading the config's `packs:` key itself. The second
178 value keeps the `pack_roots` output slot and is always 0: nothing resolves
179 here.
180 """
181 if not PACKS_RESOLVER.is_file():
182 return None, 0
183 try:
184 proc = subprocess.run(
185 [sys.executable, str(PACKS_RESOLVER), "--declared-only"],
186 capture_output=True,
187 text=True,
188 check=False,
189 cwd=repo_root(),
190 timeout=PACKS_RESOLVER_TIMEOUT,
191 )
192 except (OSError, subprocess.TimeoutExpired):
193 return None, 0
194 if proc.returncode != 0:
195 return None, 0
196 try:
197 data = json.loads(proc.stdout)
198 except ValueError:
199 return None, 0
200 declared = data.get("declared") if isinstance(data, dict) else None
201 if not isinstance(declared, bool):
202 return None, 0
203 return declared, 0
206def repo_signals(docs_root: str | None, local_scope: bool) -> dict[str, object]:
207 """Facts about the repo, not the diff: present in every result shape.
209 `declared_packs` describes the local checkout's config, which is not the
210 reviewed tree's config in remote scope, so it is evaluated only when
211 `local_scope` is true and reported as ``None`` otherwise.
212 """
213 learnings_corpus = has_learnings_corpus(docs_root)
214 declared, pack_roots = declared_packs() if local_scope else (None, 0)
215 return {
216 "has_learnings_corpus": learnings_corpus,
217 "declared_packs": declared,
218 "pack_roots": pack_roots,
219 }
222def fail_closed(reason: str, signals: dict[str, object]) -> dict[str, object]:
223 return {
224 "status": "unknown",
225 "reason": reason,
226 "exec_lines": None,
227 "exec_nontest_lines": None,
228 "unclassified_lines": {},
229 "changed_lines": None,
230 "uncounted_files": 1,
231 "changed_files": [],
232 "signals": [],
233 "hard_block_classes": ["unknown-scope"],
234 "hard_block_full": True,
235 "silent_pass_classes": [],
236 "size_band": "unknown",
237 "deleted_files": [],
238 "deleted_exec_nontest_lines": None,
239 "deleted_band": "unknown",
240 "test_files_changed": False,
241 "agent_surface": False,
242 **signals,
243 }
246def size_band_for(exec_nontest_lines: int | None) -> str:
247 """Band the executable non-test lines: `large` is a full-spine floor.
249 Sources the extension list cannot name are not banded; they are reported
250 in `unclassified_lines` for the gate's consequence judgment.
251 """
252 if exec_nontest_lines is None:
253 return "unknown"
254 if exec_nontest_lines >= FULL_EXEC_LINE_MIN:
255 return "large"
256 return "small"
259def deleted_band_for(deleted_files: list[str], deleted_exec_nontest_lines: int) -> str:
260 if not deleted_files:
261 return "none"
262 if deleted_exec_nontest_lines >= FULL_EXEC_LINE_MIN:
263 return "large"
264 return "small"
267def matching_classes(
268 files: list[str], patterns: dict[str, re.Pattern[str]]
269) -> list[str]:
270 return [
271 name
272 for name, pattern in patterns.items()
273 if any(pattern.search(file) for file in files)
274 ]
277def numstat_path(name: str) -> str:
278 """Return the destination path from a `git diff --numstat` rename display name."""
279 if " => " not in name:
280 return name
281 if "{" in name and "}" in name:
282 prefix, rest = name.split("{", 1)
283 old_new, suffix = rest.split("}", 1)
284 _, new = old_new.split(" => ", 1)
285 # A collapsed segment (`a/{b => }/c`) leaves an empty side, so the
286 # rebuilt path would carry `//` and miss every path-class pattern.
287 return re.sub(r"/{2,}", "/", f"{prefix}{new}{suffix}")
288 _, new = name.split(" => ", 1)
289 return new
292def main() -> int:
293 parser = argparse.ArgumentParser()
294 parser.add_argument("--base", required=True)
295 parser.add_argument("--head")
296 parser.add_argument("--docs-root", default="docs")
297 args = parser.parse_args()
299 # Remote scope (pr-remote / branch-remote) always passes --head, even when a
300 # best-effort fetch left it empty; the local config is not that tree's config.
301 repo = repo_signals(args.docs_root, local_scope=args.head is None)
303 if not valid_commit(args.base):
304 print(json.dumps(fail_closed("invalid base endpoint", repo), sort_keys=True))
305 return 0
306 if args.head is not None and not valid_commit(args.head):
307 print(json.dumps(fail_closed("invalid head endpoint", repo), sort_keys=True))
308 return 0
310 diff_args = [args.base]
311 if args.head:
312 merge_base = unique_merge_base(args.base, args.head)
313 if merge_base is None:
314 print(json.dumps(fail_closed("merge base unavailable or ambiguous", repo), sort_keys=True))
315 return 0
316 diff_args = [merge_base, args.head]
318 # Unquoted non-ASCII paths keep extension checks and manifest paths usable.
319 numstat = git("-c", "core.quotePath=false", "diff", "--numstat", *diff_args)
320 raw = git("-c", "core.quotePath=false", "diff", "--raw", *diff_args)
321 if numstat.returncode != 0 or raw.returncode != 0:
322 print(json.dumps(fail_closed("git diff failed", repo), sort_keys=True))
323 return 0
325 executable_mode_paths: set[str] = set()
326 deleted_paths: set[str] = set()
327 for line in raw.stdout.splitlines():
328 if "\t" not in line:
329 continue
330 meta, path_field = line.split("\t", 1)
331 fields = meta.lstrip(":").split(" ")
332 if len(fields) < 2:
333 continue
334 if fields[-1] == "D":
335 deleted_paths.add(path_field)
336 old_mode, new_mode = fields[0], fields[1]
337 mode = old_mode if new_mode == "000000" else new_mode
338 if not mode.endswith("755"):
339 continue
340 for path in path_field.split("\t"):
341 executable_mode_paths.add(path)
343 files: list[str] = []
344 executable_lines = 0
345 executable_nontest_lines = 0
346 unclassified_lines: dict[str, int] = {}
347 changed_lines = 0
348 uncounted = 0
349 deleted_files: list[str] = []
350 deleted_exec_nontest_lines = 0
351 for line in numstat.stdout.splitlines():
352 parts = line.split("\t")
353 if len(parts) < 3:
354 continue
355 added, deleted, name = parts[0], parts[1], parts[2]
356 resolved_name = numstat_path(name)
357 files.append(resolved_name)
358 ext = Path(resolved_name).suffix.lower()
359 is_exec = ext in CODE_EXTENSIONS or resolved_name in executable_mode_paths
360 is_test = bool(TEST_PATTERN.search(resolved_name))
361 if resolved_name in deleted_paths:
362 deleted_files.append(resolved_name)
363 if deleted.isdigit():
364 changed_lines += int(deleted)
365 if is_exec and not is_test:
366 deleted_exec_nontest_lines += int(deleted)
367 continue
368 if added == "-" or deleted == "-":
369 uncounted += 1
370 continue
371 try:
372 total = int(added) + int(deleted)
373 except ValueError:
374 uncounted += 1
375 continue
376 changed_lines += total
377 if is_exec:
378 executable_lines += total
379 if not is_test:
380 executable_nontest_lines += total
381 elif not is_test:
382 unclassified_lines[ext] = unclassified_lines.get(ext, 0) + total
384 files.sort()
385 deleted_files.sort()
386 signals = matching_classes(files, SIGNAL_PATTERNS)
387 hard_block_classes = matching_classes(files, HARD_BLOCK_PATTERNS)
388 silent_pass_classes = matching_classes(files, SILENT_PASS_PATTERNS)
389 if uncounted:
390 hard_block_classes.append("uncounted")
391 band = size_band_for(executable_nontest_lines)
393 result = {
394 "status": "complete",
395 "reason": None,
396 "exec_lines": executable_lines,
397 "exec_nontest_lines": executable_nontest_lines,
398 "unclassified_lines": unclassified_lines,
399 "changed_lines": changed_lines,
400 "uncounted_files": uncounted,
401 "changed_files": files,
402 "signals": signals,
403 "hard_block_classes": hard_block_classes,
404 "hard_block_full": bool(hard_block_classes) or band != "small",
405 "silent_pass_classes": silent_pass_classes,
406 "size_band": band,
407 "deleted_files": deleted_files,
408 "deleted_exec_nontest_lines": deleted_exec_nontest_lines,
409 "deleted_band": deleted_band_for(deleted_files, deleted_exec_nontest_lines),
410 "test_files_changed": any(TEST_PATTERN.search(file) for file in files),
411 "agent_surface": any(AGENT_SURFACE_PATTERN.search(file) for file in files),
412 **repo,
413 }
414 print(json.dumps(result, sort_keys=True))
415 return 0
418if __name__ == "__main__":
419 raise SystemExit(main())