"""``muse check`` — generic domain invariant enforcement. Dispatches to the domain plugin's registered :class:`~muse.core.invariants.InvariantChecker` and reports all violations. Works for any domain that has registered a checker. Currently supported domains: - ``code`` — complexity, circular imports, dead exports, test coverage. - ``midi`` — polyphony, pitch range, key consistency, parallel fifths. Commit reference syntax:: muse check # HEAD of current branch muse check abc1234 # short SHA prefix muse check HEAD~2 # two commits before HEAD muse check main # tip of another branch Usage:: muse check # HEAD, auto-detect domain muse check abc1234 # specific commit (short or full SHA) muse check HEAD~2 # ancestor relative to HEAD muse check --branch dev # HEAD of another branch muse check --strict # exit 1 on any error-severity violation muse check --warn # exit 2 on any warning-severity violation muse check --base HEAD~1 # report only violations NEW since HEAD~1 muse check --filter-severity error # show only errors muse check --filter-rule no_cycles # run only one named rule muse check --json # machine-readable JSON (all fields) muse check --rules my.toml # custom rules file muse check --summary # one-line pass/fail for scripts Exit codes:: 0 — all rules passed (or no checker registered) 1 — one or more error-severity violations found (with --strict) 2 — one or more warning-severity violations found (with --warn) """ from __future__ import annotations import argparse import json import logging import pathlib import sys import time from typing import TypedDict from muse.core.errors import ExitCode from muse.core.invariants import BaseReport, InvariantChecker, diff_reports, format_report from muse.core.repo import read_repo_id, require_repo from muse.core.store import read_current_branch, resolve_commit_ref from muse.core.validation import sanitize_display from muse.plugins.registry import read_domain logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Typed JSON schema # --------------------------------------------------------------------------- class _CheckJson(TypedDict): """Machine-readable output of ``muse check --json``.""" commit_id: str domain: str rules_checked: int has_errors: bool has_warnings: bool error_count: int warning_count: int info_count: int total_violations: int violations: list[dict] base_commit_id: str | None elapsed_seconds: float # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_checker(domain: str) -> InvariantChecker | None: """Return the domain's InvariantChecker instance, or None. Lazy-imports the domain checker to keep startup cost near-zero for repos in domains that don't need checking. """ if domain == "code": from muse.plugins.code._invariants import CodeChecker return CodeChecker() if domain == "midi": from muse.plugins.midi._invariants import MidiChecker return MidiChecker() return None def _resolve_ref( root: pathlib.Path, ref: str | None, branch: str, ) -> str | None: """Resolve *ref* (short SHA, HEAD~N, full SHA, or None → HEAD) to a commit ID. Uses the full ``resolve_commit_ref`` machinery from the store layer so that all reference syntax works consistently across all muse commands. """ repo_id = read_repo_id(root) commit = resolve_commit_ref(root, repo_id, branch, ref) return commit.commit_id if commit is not None else None def _filter_report( report: BaseReport, *, filter_severity: str | None, filter_rule: str | None, filter_path: str | None, ) -> BaseReport: """Return a copy of *report* with violations filtered by severity/rule/path.""" import fnmatch from muse.core.invariants import make_report violations = report["violations"] if filter_severity: violations = [v for v in violations if v["severity"] == filter_severity] if filter_rule: violations = [v for v in violations if v["rule_name"] == filter_rule] if filter_path: violations = [v for v in violations if fnmatch.fnmatch(v["address"], filter_path)] return make_report( report["commit_id"], report["domain"], violations, report["rules_checked"], ) # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``check`` subcommand on *subparsers*.""" parser = subparsers.add_parser( "check", help="Run invariant checks for the current domain against a commit.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "commit_arg", nargs="?", default=None, metavar="COMMIT", help=( "Commit to check: full/short SHA, HEAD~N, or branch name. " "Defaults to HEAD of the current branch." ), ) parser.add_argument( "--branch", "-b", default=None, dest="branch", metavar="BRANCH", help="Branch whose HEAD to check (defaults to the current branch).", ) parser.add_argument( "--base", default=None, dest="base_ref", metavar="REF", help=( "Compare against a base commit and report only NEW violations. " "Accepts the same ref syntax as COMMIT (HEAD~1, branch name, short SHA)." ), ) parser.add_argument( "--strict", action="store_true", help="Exit 1 when any error-severity violation is found.", ) parser.add_argument( "--warn", action="store_true", help="Exit 2 when any warning-severity violation is found (implies --strict).", ) parser.add_argument( "--filter-severity", default=None, dest="filter_severity", choices=("error", "warning", "info"), metavar="SEVERITY", help="Show only violations at this severity level (error|warning|info).", ) parser.add_argument( "--filter-rule", default=None, dest="filter_rule", metavar="RULE", help="Show only violations from this named rule.", ) parser.add_argument( "--filter-path", default=None, dest="filter_path", metavar="GLOB", help="Show only violations whose address matches this fnmatch glob.", ) parser.add_argument( "--rules", default=None, dest="rules_file", metavar="FILE", help="Path to a TOML invariants file (overrides the domain default).", ) parser.add_argument( "--summary", action="store_true", help="Print a single pass/fail summary line and exit (no violation details).", ) parser.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) parser.add_argument( "--json", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command implementation # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Run invariant checks for the current domain against a commit. Resolves the target commit using the full reference machinery (short SHA, HEAD~N, branch name) then dispatches to the domain's registered :class:`~muse.core.invariants.InvariantChecker`. With ``--base REF``, only violations that are *new* relative to REF are reported (uses :func:`~muse.core.invariants.diff_reports` internally). With ``--filter-severity``, ``--filter-rule``, or ``--filter-path`` the violation list is narrowed before display, useful for CI gates that care about only a subset of rules. JSON output (``--json``) includes ``elapsed_seconds``, ``error_count``, ``warning_count``, and ``exit_code`` for agent-friendly consumption. """ t0 = time.monotonic() commit_arg: str | None = args.commit_arg branch_arg: str | None = args.branch base_ref: str | None = args.base_ref strict: bool = args.strict warn_flag: bool = args.warn filter_severity: str | None = args.filter_severity filter_rule: str | None = args.filter_rule filter_path: str | None = args.filter_path rules_file: str | None = args.rules_file summary_only: bool = args.summary fmt: str = args.fmt root = require_repo() domain = read_domain(root) # Determine branch context. branch = branch_arg or read_current_branch(root) # Resolve target commit — full ref syntax (HEAD~N, short SHA, branch tip). commit_id = _resolve_ref(root, commit_arg, branch) if commit_id is None: msg = f"Cannot resolve ref {commit_arg!r}" if commit_arg else "No commits on this branch yet" if fmt == "json": print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {sanitize_display(msg)}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Validate rules_file path (no directory traversal). rules_path: pathlib.Path | None = None if rules_file is not None: rules_path = pathlib.Path(rules_file) if not rules_path.is_absolute(): rules_path = root / rules_path # Contain within the repo root — reject anything that resolves outside. try: rules_path.resolve().relative_to(root.resolve()) except ValueError: msg = f"--rules path {rules_file!r} is outside the repository" if fmt == "json": print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {sanitize_display(msg)}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) checker = _get_checker(domain) if checker is None: msg = f"No invariant checker registered for domain {sanitize_display(domain)!r}. Supported: code, midi" if fmt == "json": print(json.dumps({"error": msg, "exit_code": 0})) else: print(f"⚠️ {msg}.", file=sys.stderr) raise SystemExit(0) # Run the checker. report = checker.check(root, commit_id, rules_file=rules_path) # Diff mode: strip violations already present in the base commit. base_commit_id: str | None = None if base_ref is not None: base_commit_id = _resolve_ref(root, base_ref, branch) if base_commit_id is None: msg = f"Cannot resolve base ref {base_ref!r}" if fmt == "json": print(json.dumps({"error": msg, "exit_code": ExitCode.USER_ERROR})) else: print(f"❌ {sanitize_display(msg)}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) base_report = checker.check(root, base_commit_id, rules_file=rules_path) report = diff_reports(report, base_report) # Apply post-run filters (severity/rule/path narrowing). has_active_filter = filter_severity or filter_rule or filter_path if has_active_filter: report = _filter_report( report, filter_severity=filter_severity, filter_rule=filter_rule, filter_path=filter_path, ) elapsed = time.monotonic() - t0 error_count = sum(1 for v in report["violations"] if v["severity"] == "error") warning_count = sum(1 for v in report["violations"] if v["severity"] == "warning") info_count = sum(1 for v in report["violations"] if v["severity"] == "info") # Determine exit code before output so agents get it in JSON. exit_code = 0 if strict and error_count: exit_code = 1 if warn_flag and warning_count: exit_code = max(exit_code, 2) # ── JSON output ─────────────────────────────────────────────────────────── if fmt == "json": payload = _CheckJson( commit_id=commit_id, domain=domain, rules_checked=report["rules_checked"], has_errors=report["has_errors"], has_warnings=report["has_warnings"], error_count=error_count, warning_count=warning_count, info_count=info_count, total_violations=len(report["violations"]), violations=list(report["violations"]), base_commit_id=base_commit_id, elapsed_seconds=round(elapsed, 4), ) print(json.dumps(payload)) raise SystemExit(exit_code) # ── Text output ─────────────────────────────────────────────────────────── safe_commit = sanitize_display(commit_id[:12]) header = f"\ncheck [{sanitize_display(domain)}] {safe_commit}" if base_commit_id: header += f" vs {sanitize_display(base_commit_id[:12])}" header += f" — {report['rules_checked']} rules" if has_active_filter: parts = [] if filter_severity: parts.append(f"severity={filter_severity}") if filter_rule: parts.append(f"rule={sanitize_display(filter_rule)}") if filter_path: parts.append(f"path={sanitize_display(filter_path)}") header += f" (filtered: {', '.join(parts)})" print(header) if summary_only: total = len(report["violations"]) if total == 0: print(f"✅ No violations.") else: print(f"❌ {total} violation(s): {error_count} error(s), {warning_count} warning(s), {info_count} info(s)") raise SystemExit(exit_code) print(format_report(report)) print(f" ({elapsed:.3f}s)") raise SystemExit(exit_code)