aaronrene / gabriel-muse public
check.py python
409 lines 14.5 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """``muse check`` — generic domain invariant enforcement.
2
3 Dispatches to the domain plugin's registered
4 :class:`~muse.core.invariants.InvariantChecker` and reports all
5 violations. Works for any domain that has registered a checker.
6
7 Currently supported domains:
8
9 - ``code`` — complexity, circular imports, dead exports, test coverage.
10 - ``midi`` — polyphony, pitch range, key consistency, parallel fifths.
11
12 Commit reference syntax::
13
14 muse check # HEAD of current branch
15 muse check abc1234 # short SHA prefix
16 muse check HEAD~2 # two commits before HEAD
17 muse check main # tip of another branch
18
19 Usage::
20
21 muse check # HEAD, auto-detect domain
22 muse check abc1234 # specific commit (short or full SHA)
23 muse check HEAD~2 # ancestor relative to HEAD
24 muse check --branch dev # HEAD of another branch
25 muse check --strict # exit 1 on any error-severity violation
26 muse check --warn # exit 2 on any warning-severity violation
27 muse check --base HEAD~1 # report only violations NEW since HEAD~1
28 muse check --filter-severity error # show only errors
29 muse check --filter-rule no_cycles # run only one named rule
30 muse check --json # machine-readable JSON (all fields)
31 muse check --rules my.toml # custom rules file
32 muse check --summary # one-line pass/fail for scripts
33
34 Exit codes::
35
36 0 — all rules passed (or no checker registered)
37 1 — one or more error-severity violations found (with --strict)
38 2 — one or more warning-severity violations found (with --warn)
39 """
40
41 from __future__ import annotations
42
43 import argparse
44 import json
45 import logging
46 import pathlib
47 import sys
48 import time
49 from typing import TypedDict
50
51 from muse.core.errors import ExitCode
52 from muse.core.invariants import BaseReport, InvariantChecker, diff_reports, format_report
53 from muse.core.repo import read_repo_id, require_repo
54 from muse.core.store import read_current_branch, resolve_commit_ref
55 from muse.core.validation import sanitize_display
56 from muse.plugins.registry import read_domain
57
58 logger = logging.getLogger(__name__)
59
60
61 # ---------------------------------------------------------------------------
62 # Typed JSON schema
63 # ---------------------------------------------------------------------------
64
65
66 class _CheckJson(TypedDict):
67 """Machine-readable output of ``muse check --json``."""
68
69 commit_id: str
70 domain: str
71 rules_checked: int
72 has_errors: bool
73 has_warnings: bool
74 error_count: int
75 warning_count: int
76 info_count: int
77 total_violations: int
78 violations: list[dict]
79 base_commit_id: str | None
80 elapsed_seconds: float
81
82
83 # ---------------------------------------------------------------------------
84 # Helpers
85 # ---------------------------------------------------------------------------
86
87
88 def _get_checker(domain: str) -> InvariantChecker | None:
89 """Return the domain's InvariantChecker instance, or None.
90
91 Lazy-imports the domain checker to keep startup cost near-zero for repos
92 in domains that don't need checking.
93 """
94 if domain == "code":
95 from muse.plugins.code._invariants import CodeChecker
96 return CodeChecker()
97 if domain == "midi":
98 from muse.plugins.midi._invariants import MidiChecker
99 return MidiChecker()
100 return None
101
102
103 def _resolve_ref(
104 root: pathlib.Path,
105 ref: str | None,
106 branch: str,
107 ) -> str | None:
108 """Resolve *ref* (short SHA, HEAD~N, full SHA, or None → HEAD) to a commit ID.
109
110 Uses the full ``resolve_commit_ref`` machinery from the store layer so that
111 all reference syntax works consistently across all muse commands.
112 """
113 repo_id = read_repo_id(root)
114 commit = resolve_commit_ref(root, repo_id, branch, ref)
115 return commit.commit_id if commit is not None else None
116
117
118 def _filter_report(
119 report: BaseReport,
120 *,
121 filter_severity: str | None,
122 filter_rule: str | None,
123 filter_path: str | None,
124 ) -> BaseReport:
125 """Return a copy of *report* with violations filtered by severity/rule/path."""
126 import fnmatch
127 from muse.core.invariants import make_report
128
129 violations = report["violations"]
130 if filter_severity:
131 violations = [v for v in violations if v["severity"] == filter_severity]
132 if filter_rule:
133 violations = [v for v in violations if v["rule_name"] == filter_rule]
134 if filter_path:
135 violations = [v for v in violations if fnmatch.fnmatch(v["address"], filter_path)]
136 return make_report(
137 report["commit_id"],
138 report["domain"],
139 violations,
140 report["rules_checked"],
141 )
142
143
144 # ---------------------------------------------------------------------------
145 # Registration
146 # ---------------------------------------------------------------------------
147
148
149 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
150 """Register the ``check`` subcommand on *subparsers*."""
151 parser = subparsers.add_parser(
152 "check",
153 help="Run invariant checks for the current domain against a commit.",
154 description=__doc__,
155 formatter_class=argparse.RawDescriptionHelpFormatter,
156 )
157 parser.add_argument(
158 "commit_arg",
159 nargs="?",
160 default=None,
161 metavar="COMMIT",
162 help=(
163 "Commit to check: full/short SHA, HEAD~N, or branch name. "
164 "Defaults to HEAD of the current branch."
165 ),
166 )
167 parser.add_argument(
168 "--branch", "-b",
169 default=None,
170 dest="branch",
171 metavar="BRANCH",
172 help="Branch whose HEAD to check (defaults to the current branch).",
173 )
174 parser.add_argument(
175 "--base",
176 default=None,
177 dest="base_ref",
178 metavar="REF",
179 help=(
180 "Compare against a base commit and report only NEW violations. "
181 "Accepts the same ref syntax as COMMIT (HEAD~1, branch name, short SHA)."
182 ),
183 )
184 parser.add_argument(
185 "--strict",
186 action="store_true",
187 help="Exit 1 when any error-severity violation is found.",
188 )
189 parser.add_argument(
190 "--warn",
191 action="store_true",
192 help="Exit 2 when any warning-severity violation is found (implies --strict).",
193 )
194 parser.add_argument(
195 "--filter-severity",
196 default=None,
197 dest="filter_severity",
198 choices=("error", "warning", "info"),
199 metavar="SEVERITY",
200 help="Show only violations at this severity level (error|warning|info).",
201 )
202 parser.add_argument(
203 "--filter-rule",
204 default=None,
205 dest="filter_rule",
206 metavar="RULE",
207 help="Show only violations from this named rule.",
208 )
209 parser.add_argument(
210 "--filter-path",
211 default=None,
212 dest="filter_path",
213 metavar="GLOB",
214 help="Show only violations whose address matches this fnmatch glob.",
215 )
216 parser.add_argument(
217 "--rules",
218 default=None,
219 dest="rules_file",
220 metavar="FILE",
221 help="Path to a TOML invariants file (overrides the domain default).",
222 )
223 parser.add_argument(
224 "--summary",
225 action="store_true",
226 help="Print a single pass/fail summary line and exit (no violation details).",
227 )
228 parser.add_argument(
229 "--format", "-f",
230 default="text",
231 dest="fmt",
232 choices=("text", "json"),
233 help="Output format: text (default) or json.",
234 )
235 parser.add_argument(
236 "--json",
237 action="store_const",
238 const="json",
239 dest="fmt",
240 help="Shorthand for --format json.",
241 )
242 parser.set_defaults(func=run)
243
244
245 # ---------------------------------------------------------------------------
246 # Command implementation
247 # ---------------------------------------------------------------------------
248
249
250 def run(args: argparse.Namespace) -> None:
251 """Run invariant checks for the current domain against a commit.
252
253 Resolves the target commit using the full reference machinery (short SHA,
254 HEAD~N, branch name) then dispatches to the domain's registered
255 :class:`~muse.core.invariants.InvariantChecker`.
256
257 With ``--base REF``, only violations that are *new* relative to REF are
258 reported (uses :func:`~muse.core.invariants.diff_reports` internally).
259
260 With ``--filter-severity``, ``--filter-rule``, or ``--filter-path`` the
261 violation list is narrowed before display, useful for CI gates that care
262 about only a subset of rules.
263
264 JSON output (``--json``) includes ``elapsed_seconds``, ``error_count``,
265 ``warning_count``, and ``exit_code`` for agent-friendly consumption.
266 """
267 t0 = time.monotonic()
268
269 commit_arg: str | None = args.commit_arg
270 branch_arg: str | None = args.branch
271 base_ref: str | None = args.base_ref
272 strict: bool = args.strict
273 warn_flag: bool = args.warn
274 filter_severity: str | None = args.filter_severity
275 filter_rule: str | None = args.filter_rule
276 filter_path: str | None = args.filter_path
277 rules_file: str | None = args.rules_file
278 summary_only: bool = args.summary
279 fmt: str = args.fmt
280
281 root = require_repo()
282 domain = read_domain(root)
283
284 # Determine branch context.
285 branch = branch_arg or read_current_branch(root)
286
287 # Resolve target commit — full ref syntax (HEAD~N, short SHA, branch tip).
288 commit_id = _resolve_ref(root, commit_arg, branch)
289 if commit_id is None:
290 msg = f"Cannot resolve ref {commit_arg!r}" if commit_arg else "No commits on this branch yet"
291 if fmt == "json":
292 print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR}))
293 else:
294 print(f"❌ {sanitize_display(msg)}.", file=sys.stderr)
295 raise SystemExit(ExitCode.USER_ERROR)
296
297 # Validate rules_file path (no directory traversal).
298 rules_path: pathlib.Path | None = None
299 if rules_file is not None:
300 rules_path = pathlib.Path(rules_file)
301 if not rules_path.is_absolute():
302 rules_path = root / rules_path
303 # Contain within the repo root — reject anything that resolves outside.
304 try:
305 rules_path.resolve().relative_to(root.resolve())
306 except ValueError:
307 msg = f"--rules path {rules_file!r} is outside the repository"
308 if fmt == "json":
309 print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR}))
310 else:
311 print(f"❌ {sanitize_display(msg)}.", file=sys.stderr)
312 raise SystemExit(ExitCode.USER_ERROR)
313
314 checker = _get_checker(domain)
315 if checker is None:
316 msg = f"No invariant checker registered for domain {sanitize_display(domain)!r}. Supported: code, midi"
317 if fmt == "json":
318 print(json.dumps({"error": msg, "exit_code": 0}))
319 else:
320 print(f"⚠️ {msg}.", file=sys.stderr)
321 raise SystemExit(0)
322
323 # Run the checker.
324 report = checker.check(root, commit_id, rules_file=rules_path)
325
326 # Diff mode: strip violations already present in the base commit.
327 base_commit_id: str | None = None
328 if base_ref is not None:
329 base_commit_id = _resolve_ref(root, base_ref, branch)
330 if base_commit_id is None:
331 msg = f"Cannot resolve base ref {base_ref!r}"
332 if fmt == "json":
333 print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR}))
334 else:
335 print(f"❌ {sanitize_display(msg)}.", file=sys.stderr)
336 raise SystemExit(ExitCode.USER_ERROR)
337 base_report = checker.check(root, base_commit_id, rules_file=rules_path)
338 report = diff_reports(report, base_report)
339
340 # Apply post-run filters (severity/rule/path narrowing).
341 has_active_filter = filter_severity or filter_rule or filter_path
342 if has_active_filter:
343 report = _filter_report(
344 report,
345 filter_severity=filter_severity,
346 filter_rule=filter_rule,
347 filter_path=filter_path,
348 )
349
350 elapsed = time.monotonic() - t0
351 error_count = sum(1 for v in report["violations"] if v["severity"] == "error")
352 warning_count = sum(1 for v in report["violations"] if v["severity"] == "warning")
353 info_count = sum(1 for v in report["violations"] if v["severity"] == "info")
354
355 # Determine exit code before output so agents get it in JSON.
356 exit_code = 0
357 if strict and error_count:
358 exit_code = 1
359 if warn_flag and warning_count:
360 exit_code = max(exit_code, 2)
361
362 # ── JSON output ───────────────────────────────────────────────────────────
363 if fmt == "json":
364 payload = _CheckJson(
365 commit_id=commit_id,
366 domain=domain,
367 rules_checked=report["rules_checked"],
368 has_errors=report["has_errors"],
369 has_warnings=report["has_warnings"],
370 error_count=error_count,
371 warning_count=warning_count,
372 info_count=info_count,
373 total_violations=len(report["violations"]),
374 violations=list(report["violations"]),
375 base_commit_id=base_commit_id,
376 elapsed_seconds=round(elapsed, 4),
377 )
378 print(json.dumps(payload))
379 raise SystemExit(exit_code)
380
381 # ── Text output ───────────────────────────────────────────────────────────
382 safe_commit = sanitize_display(commit_id[:12])
383 header = f"\ncheck [{sanitize_display(domain)}] {safe_commit}"
384 if base_commit_id:
385 header += f" vs {sanitize_display(base_commit_id[:12])}"
386 header += f" — {report['rules_checked']} rules"
387 if has_active_filter:
388 parts = []
389 if filter_severity:
390 parts.append(f"severity={filter_severity}")
391 if filter_rule:
392 parts.append(f"rule={sanitize_display(filter_rule)}")
393 if filter_path:
394 parts.append(f"path={sanitize_display(filter_path)}")
395 header += f" (filtered: {', '.join(parts)})"
396 print(header)
397
398 if summary_only:
399 total = len(report["violations"])
400 if total == 0:
401 print(f"✅ No violations.")
402 else:
403 print(f"❌ {total} violation(s): {error_count} error(s), {warning_count} warning(s), {info_count} info(s)")
404 raise SystemExit(exit_code)
405
406 print(format_report(report))
407 print(f" ({elapsed:.3f}s)")
408
409 raise SystemExit(exit_code)
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago