"""``muse code code-check`` — code invariant enforcement. Evaluates semantic rules declared in ``.muse/code_invariants.toml`` against the code snapshot of the specified commit and reports violations. Built-in rule types (declared in TOML):: [[rule]] name = "complexity_gate" severity = "error" rule_type = "max_complexity" [rule.params] threshold = 10 [[rule]] name = "no_cycles" severity = "error" rule_type = "no_circular_imports" [[rule]] name = "dead_exports" severity = "warning" rule_type = "no_dead_exports" [[rule]] name = "coverage_floor" severity = "warning" rule_type = "test_coverage_floor" [rule.params] min_ratio = 0.30 Usage:: muse code code-check # check HEAD muse code code-check abc1234 # check specific commit muse code code-check --strict # exit 1 on any error-severity violation muse code code-check --json # machine-readable JSON output muse code code-check --rules my_rules.toml # custom rules file muse code code-check --filter error # show only error-severity violations muse code code-check --diff HEAD~1 # show only NEW violations vs reference muse code code-check --diff main --strict # CI ratchet: fail only on regressions """ from __future__ import annotations import argparse import json import logging import pathlib import sys from muse.core.invariants import diff_reports, format_report, make_report from muse.core.repo import require_repo from muse.core.store import get_commit_snapshot_manifest, get_head_commit_id, read_current_branch from muse.core.validation import contain_path from muse.plugins.code._invariants import load_invariant_rules, run_invariants logger = logging.getLogger(__name__) _SEVERITIES = ("error", "warning", "info") def _resolve_commit(root: pathlib.Path, ref: str | None) -> str | None: """Resolve *ref* to a full commit ID. Resolution order: 1. ``None`` → HEAD of the current branch. 2. Branch name that exists in the store → HEAD commit of that branch. 3. Anything else → returned as-is (assumed to be a full or partial commit ID; the caller validates via the snapshot manifest). """ if ref is None: branch = read_current_branch(root) return get_head_commit_id(root, branch) # Try treating ref as a branch name first. branch_commit = get_head_commit_id(root, ref) if branch_commit is not None: return branch_commit # Fall back: treat as a literal commit ID. return ref def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the code-check subcommand.""" parser = subparsers.add_parser( "code-check", help="Enforce code invariant rules against a commit snapshot.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "commit_arg", nargs="?", default=None, metavar="commit", help="Commit ID to check (default: HEAD).", ) parser.add_argument( "--strict", action="store_true", help="Exit 1 when any error-severity violation is found.", ) parser.add_argument( "--json", action="store_true", dest="as_json", help="Emit machine-readable JSON.", ) parser.add_argument( "--rules", default=None, dest="rules_file", metavar="RULES_FILE", help=( "Path to a TOML invariants file inside the repo " "(default: .muse/code_invariants.toml)." ), ) parser.add_argument( "--filter", dest="severity_filter", choices=_SEVERITIES, default=None, metavar="SEVERITY", help="Show only violations of this severity (error | warning | info).", ) parser.add_argument( "--diff", dest="diff_ref", default=None, metavar="REF", help=( "Show only violations that are NEW compared to REF. " "Ideal for CI: muse code code-check --diff HEAD~1 --strict " "fails only when you introduce a new violation." ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Enforce code invariant rules against a commit snapshot. Reports cyclomatic complexity violations, import cycles, dead exports, and test coverage shortfalls based on the rules in ``.muse/code_invariants.toml`` (or built-in defaults when the file is absent). The ``--rules`` path is validated against the repo root — paths that escape the repository (e.g. ``../../shared/rules.toml``) are rejected. """ commit_arg: str | None = args.commit_arg strict: bool = args.strict as_json: bool = args.as_json rules_file: str | None = args.rules_file severity_filter: str | None = args.severity_filter diff_ref: str | None = args.diff_ref root = require_repo() commit_id = _resolve_commit(root, commit_arg) if commit_id is None: print("❌ No commit found.", file=sys.stderr) raise SystemExit(1) rules_path: pathlib.Path | None = None if rules_file: try: rules_path = contain_path(root, rules_file) except ValueError as exc: print(f"❌ {exc}", file=sys.stderr) raise SystemExit(1) if not rules_path.exists(): print(f"❌ Rules file not found: {rules_path}", file=sys.stderr) raise SystemExit(1) rules = load_invariant_rules(rules_path) report = run_invariants(root, commit_id, rules) # --diff: subtract violations already present in the reference commit so # only regressions are reported. if diff_ref is not None: ref_id = _resolve_commit(root, diff_ref) if ref_id is None: print(f"❌ Could not resolve --diff ref: {diff_ref!r}", file=sys.stderr) raise SystemExit(1) if get_commit_snapshot_manifest(root, ref_id) is None: print( f"❌ Commit {diff_ref!r} not found — cannot use as --diff base.", file=sys.stderr, ) raise SystemExit(1) ref_report = run_invariants(root, ref_id, rules) report = diff_reports(report, ref_report) # --filter: drop violations that don't match the requested severity. if severity_filter is not None: filtered = [v for v in report["violations"] if v["severity"] == severity_filter] report = make_report( report["commit_id"], report["domain"], filtered, report["rules_checked"], ) if as_json: payload = dict(report) if diff_ref is not None: payload["diff_ref"] = diff_ref if severity_filter is not None: payload["severity_filter"] = severity_filter print(json.dumps(payload)) if strict and report["has_errors"]: raise SystemExit(1) return diff_label = f" (new since {diff_ref})" if diff_ref else "" filter_label = f" [{severity_filter} only]" if severity_filter else "" print( f"\ncode-check {commit_id[:8]}{diff_label}{filter_label}" f" — {report['rules_checked']} rules" ) print(format_report(report)) if strict and report["has_errors"]: raise SystemExit(1)