"""muse code invariants — enforce architectural rules from .muse/code_invariants.toml. Loads declarative invariant rules and evaluates them against the committed snapshot at HEAD (or any historical commit). Rules are architectural constraints enforced by static analysis — no code is executed. Rules file ---------- Create ``.muse/code_invariants.toml`` using ``[[rule]]`` blocks: .. code-block:: toml [[rule]] name = "complexity gate" severity = "warning" scope = "function" rule_type = "max_complexity" [rule.params] threshold = 10 [[rule]] name = "no import cycles" severity = "error" scope = "file" rule_type = "no_circular_imports" [[rule]] name = "no dead exports" severity = "warning" scope = "file" rule_type = "no_dead_exports" [[rule]] name = "test coverage floor" severity = "warning" scope = "repo" rule_type = "test_coverage_floor" [rule.params] min_ratio = 0.30 [[rule]] name = "core must not import cli" severity = "error" scope = "file" rule_type = "forbidden_dependency" [rule.params] source_pattern = "muse/core/" forbidden_pattern = "muse/cli/" [[rule]] name = "plugins must not import from cli" severity = "error" scope = "file" rule_type = "layer_boundary" [rule.params] lower = "muse/plugins/" upper = "muse/cli/" Supported rule types -------------------- ``max_complexity`` Functions whose branch-count exceeds *threshold*. Default threshold: 10. ``no_circular_imports`` Import cycles among Python files in the snapshot. ``no_dead_exports`` Top-level functions/classes never imported by any other file. Exempt: test files, ``__init__.py``, files with ``__all__``. ``test_coverage_floor`` Requires ≥ *min_ratio* of public functions to have a ``test_`` counterpart. ``forbidden_dependency`` Files matching *source_pattern* must not import from *forbidden_pattern*. Enforces hard layer boundaries. ``layer_boundary`` *lower*-layer files must not import from *upper*-layer files. When no rules file is found, three built-in defaults are applied: ``complexity_gate`` (warning, threshold=10), ``no_cycles`` (error), ``dead_exports`` (warning). Usage:: muse code invariants muse code invariants --commit HEAD~5 muse code invariants --commit feat/my-branch muse code invariants --rule no_circular_imports muse code invariants --strict muse code invariants --json Flags: ``--commit, -c REF`` Check a historical snapshot (branch name, commit SHA, or ``HEAD~N``). ``--rule NAME_OR_TYPE`` Run only rules whose name or rule_type matches this value. ``--strict`` Exit non-zero if any warning-level violations are found. ``--json`` Emit a machine-readable JSON object. """ import argparse import json import logging import pathlib import sys from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.paths import code_invariants_path from muse.core.repo import require_repo from muse.core.timing import start_timer from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import resolve_commit_ref from muse.core.snapshots import get_commit_snapshot_manifest from muse.plugins.code._invariants import CodeViolation, load_invariant_rules, run_invariants from muse.core.validation import sanitize_display type _StrMap = dict[str, str] type _ByRule = dict[str, list[str]] logger = logging.getLogger(__name__) class _InvariantsOutputJson(EnvelopeJson): """JSON envelope emitted by ``muse code invariants --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ commit Abbreviated commit ID of the snapshot analysed. branch Current branch name. ref Human-readable ref label (branch + short SHA). using_defaults True when no .muse/code_invariants.toml was found. rule_filter The --rule filter applied, or None. strict True when --strict was passed. rules_checked Number of rules evaluated. violations_total Total number of violations across all rules. errors Count of error-severity violations. warnings_count Count of warning-severity violations. violations List of violation dicts ({rule_name, address, description, severity}). """ commit: str branch: str ref: str using_defaults: bool rule_filter: str | None strict: bool rules_checked: int violations_total: int errors: int warnings_count: int violations: list[CodeViolation] class _InvariantsErrorJson(EnvelopeJson): """JSON error output for ``muse code invariants --json`` on early-exit paths.""" error: str message: str # --------------------------------------------------------------------------- # CLI registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``invariants`` subcommand.""" parser = subparsers.add_parser( "invariants", help="Check architectural invariants from .muse/code_invariants.toml.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--commit", "-c", dest="ref", default=None, metavar="REF", help="Check a historical snapshot (branch, SHA, or HEAD~N).", ) parser.add_argument( "--rule", dest="rule_filter", default=None, metavar="NAME_OR_TYPE", help="Run only rules whose name or rule_type matches this value.", ) parser.add_argument( "--strict", dest="strict", action="store_true", help="Exit non-zero if any warning-level violations are found.", ) parser.add_argument( "--json", "-j", dest="json_out", action="store_true", help="Emit results as JSON.", ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Check architectural invariants and exit non-zero on violations. Evaluates rules from ``.muse/code_invariants.toml`` (or built-in defaults) against the committed snapshot. Rule types: ``max_complexity``, ``no_circular_imports``, ``no_dead_exports``, ``test_coverage_floor``, ``forbidden_dependency``, ``layer_boundary``. Agent quickstart ---------------- :: muse code check --json muse code check --strict --json muse code check --rule no_circular_imports --json muse code check --ref HEAD~5 --json JSON fields ----------- commit Short commit ID checked. branch Branch name. using_defaults ``true`` if no ``.muse/code_invariants.toml`` exists. rules_checked Number of rules evaluated. violations_total Total number of violations found. errors Error-severity violation count. warnings_count Warning-severity violation count. violations List of violation objects: ``rule``, ``message``, ``severity``. Exit codes ---------- 0 All rules pass (or only warnings without ``--strict``). 1 Error violations found; or warnings found under ``--strict``. 2 Not inside a Muse repository. """ elapsed = start_timer() ref: str | None = args.ref json_out: bool = args.json_out rule_filter: str | None = getattr(args, "rule_filter", None) strict: bool = getattr(args, "strict", False) root = require_repo() branch = read_current_branch(root) # Resolve ref — support branch names in addition to SHAs and HEAD~N. # Only attempt branch-name lookup when the ref looks like a plain name # (no ~, ^, @, : or whitespace) — those characters signal relative ref # syntax that validate_branch_name rejects but resolve_commit_ref handles. resolved_ref: str | None = ref if ref is not None and not any(c in ref for c in "~^@: \t"): branch_head = get_head_commit_id(root, ref) if branch_head is not None: resolved_ref = branch_head commit = resolve_commit_ref(root, branch, resolved_ref) if commit is None: print(f"❌ No commit found for ref '{ref or 'HEAD'}'.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Guard: refuse to proceed if the manifest is unreadable. manifest = get_commit_snapshot_manifest(root, commit.commit_id) if manifest is None: print( f"❌ Cannot read snapshot for commit {commit.commit_id} — " "repository may be corrupt.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # Load rules — built-in defaults apply when the file is absent. using_defaults = not code_invariants_path(root).exists() rules = load_invariant_rules(root, None) # Apply optional rule filter. if rule_filter: rules = [ r for r in rules if rule_filter in (r.get("name", ""), r.get("rule_type", "")) ] if not rules: msg = f"No rules match filter '{rule_filter}'." if json_out: print(json.dumps(_InvariantsErrorJson( **make_envelope(elapsed), error="no_matching_rules", message=msg, ))) else: print(f"⚠️ {msg}", file=sys.stderr) raise SystemExit(0) if not rules: msg = ( "No rules defined. " "Create .muse/code_invariants.toml with [[rule]] blocks to define architectural constraints." ) if json_out: print(json.dumps(_InvariantsOutputJson( **make_envelope(elapsed), commit=commit.commit_id, branch=branch, ref=commit.commit_id, using_defaults=False, rule_filter=rule_filter, strict=strict, rules_checked=0, violations_total=0, errors=0, warnings_count=0, violations=[], ))) else: print(f"⚠️ {msg}") raise SystemExit(0) # Run all rules via the plugin engine. report = run_invariants(root, commit.commit_id, rules) violations = report["violations"] errors = sum(1 for v in violations if v["severity"] == "error") warnings = sum(1 for v in violations if v["severity"] == "warning") exit_code = 1 if (errors > 0 or (strict and warnings > 0)) else 0 ref_label = commit.commit_id if ref and ref != commit.commit_id: ref_label = f"{ref} ({commit.commit_id})" if json_out: print(json.dumps(_InvariantsOutputJson( **make_envelope(elapsed, exit_code=exit_code), commit=commit.commit_id, branch=branch, ref=ref_label, using_defaults=using_defaults, rule_filter=rule_filter, strict=strict, rules_checked=report["rules_checked"], violations_total=len(violations), errors=errors, warnings_count=warnings, violations=list(violations), ))) raise SystemExit(exit_code) print(f"\nInvariant check — {ref_label}") if using_defaults: print(" (using built-in default rules — create .muse/code_invariants.toml to customise)") if rule_filter: print(f" (filter: {rule_filter})") print("─" * 62) if not violations: print(f"\n ✅ All {report['rules_checked']} rule(s) passed.") raise SystemExit(0) # Group violations by rule name for clean output. from collections import defaultdict by_rule: _ByRule = defaultdict(list) severity_by_rule: _StrMap = {} for v in violations: by_rule[v["rule_name"]].append( f" {sanitize_display(v['address'])}: {sanitize_display(v['description'])}" if v["address"] != "repo" else f" {sanitize_display(v['description'])}" ) severity_by_rule[v["rule_name"]] = v["severity"] # Show rules that passed (from original rules list) and rules that violated. violated_names = set(by_rule) passed_count = report["rules_checked"] - len(violated_names) for rule in rules: rule_name = rule.get("name", "unnamed") if rule_name not in violated_names: print(f"\n ✅ {sanitize_display(rule_name)}") else: sev = severity_by_rule[rule_name] icon = "🔴" if sev == "error" else "⚠️ " print(f"\n{icon} {sanitize_display(rule_name)} ({sev})") for line in by_rule[rule_name][:5]: print(f" {sanitize_display(line)}") if len(by_rule[rule_name]) > 5: print(f" … and {len(by_rule[rule_name]) - 5} more") print(f"\n {passed_count} rule(s) passed · {len(violated_names)} rule(s) violated") print(f" {errors} error(s) · {warnings} warning(s)") raise SystemExit(exit_code)