"""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. """ from __future__ import annotations import argparse import json import logging import pathlib import sys from muse._version import __version__ from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_commit_snapshot_manifest, get_head_commit_id, read_current_branch, resolve_commit_ref, ) from muse.plugins.code._invariants import 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__) _RULES_FILE = pathlib.Path(".muse") / "code_invariants.toml" # --------------------------------------------------------------------------- # CLI plumbing # --------------------------------------------------------------------------- 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", dest="as_json", 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. Delegates rule evaluation to the code-domain plugin engine (``muse.plugins.code._invariants``). All rule types are supported: ``max_complexity``, ``no_circular_imports``, ``no_dead_exports``, ``test_coverage_floor``, ``forbidden_dependency``, ``layer_boundary``. Exits with code 0 when all checked rules pass (or only warnings when ``--strict`` is not set). Exits with code 1 when any error-severity rule is violated, or when ``--strict`` is set and warnings are present. """ ref: str | None = args.ref as_json: bool = args.as_json rule_filter: str | None = getattr(args, "rule_filter", None) strict: bool = getattr(args, "strict", False) root = require_repo() repo_id = read_repo_id(root) branch = read_current_branch(root) # Resolve ref — support branch names in addition to SHAs and HEAD~N. resolved_ref: str | None = ref if ref is not None: branch_head = get_head_commit_id(root, ref) if branch_head is not None: resolved_ref = branch_head commit = resolve_commit_ref(root, repo_id, 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[:8]} — " "repository may be corrupt.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # Load rules — built-in defaults apply when the file is absent. rules_file = root / _RULES_FILE rules = load_invariant_rules(rules_file if rules_file.exists() else None) using_defaults = not rules_file.exists() # 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 as_json: print(json.dumps({ "schema_version": __version__, "error": "no_matching_rules", "message": msg, }, indent=2)) else: print(f"⚠️ {msg}", file=sys.stderr) raise SystemExit(0) if not rules: msg = ( "No rules defined. " f"Create {_RULES_FILE} with [[rule]] blocks to define architectural constraints." ) if as_json: print(json.dumps({ "schema_version": __version__, "commit": commit.commit_id[:8], "branch": branch, "using_defaults": False, "rules_checked": 0, "errors": 0, "warnings": 0, "violations": [], }, indent=2)) 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[:8] if ref and ref != commit.commit_id: ref_label = f"{ref} ({commit.commit_id[:8]})" if as_json: print(json.dumps( { "schema_version": __version__, "commit": commit.commit_id[:8], "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": warnings, "violations": list(violations), }, indent=2, )) 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)