aaronrene / gabriel-muse public
invariants.py python
339 lines 10.9 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """muse code invariants — enforce architectural rules from .muse/code_invariants.toml.
2
3 Loads declarative invariant rules and evaluates them against the committed
4 snapshot at HEAD (or any historical commit). Rules are architectural
5 constraints enforced by static analysis — no code is executed.
6
7 Rules file
8 ----------
9 Create ``.muse/code_invariants.toml`` using ``[[rule]]`` blocks:
10
11 .. code-block:: toml
12
13 [[rule]]
14 name = "complexity gate"
15 severity = "warning"
16 scope = "function"
17 rule_type = "max_complexity"
18 [rule.params]
19 threshold = 10
20
21 [[rule]]
22 name = "no import cycles"
23 severity = "error"
24 scope = "file"
25 rule_type = "no_circular_imports"
26
27 [[rule]]
28 name = "no dead exports"
29 severity = "warning"
30 scope = "file"
31 rule_type = "no_dead_exports"
32
33 [[rule]]
34 name = "test coverage floor"
35 severity = "warning"
36 scope = "repo"
37 rule_type = "test_coverage_floor"
38 [rule.params]
39 min_ratio = 0.30
40
41 [[rule]]
42 name = "core must not import cli"
43 severity = "error"
44 scope = "file"
45 rule_type = "forbidden_dependency"
46 [rule.params]
47 source_pattern = "muse/core/"
48 forbidden_pattern = "muse/cli/"
49
50 [[rule]]
51 name = "plugins must not import from cli"
52 severity = "error"
53 scope = "file"
54 rule_type = "layer_boundary"
55 [rule.params]
56 lower = "muse/plugins/"
57 upper = "muse/cli/"
58
59 Supported rule types
60 --------------------
61 ``max_complexity``
62 Functions whose branch-count exceeds *threshold*. Default threshold: 10.
63
64 ``no_circular_imports``
65 Import cycles among Python files in the snapshot.
66
67 ``no_dead_exports``
68 Top-level functions/classes never imported by any other file.
69 Exempt: test files, ``__init__.py``, files with ``__all__``.
70
71 ``test_coverage_floor``
72 Requires ≥ *min_ratio* of public functions to have a ``test_`` counterpart.
73
74 ``forbidden_dependency``
75 Files matching *source_pattern* must not import from *forbidden_pattern*.
76 Enforces hard layer boundaries.
77
78 ``layer_boundary``
79 *lower*-layer files must not import from *upper*-layer files.
80
81 When no rules file is found, three built-in defaults are applied:
82 ``complexity_gate`` (warning, threshold=10), ``no_cycles`` (error),
83 ``dead_exports`` (warning).
84
85 Usage::
86
87 muse code invariants
88 muse code invariants --commit HEAD~5
89 muse code invariants --commit feat/my-branch
90 muse code invariants --rule no_circular_imports
91 muse code invariants --strict
92 muse code invariants --json
93
94 Flags:
95
96 ``--commit, -c REF``
97 Check a historical snapshot (branch name, commit SHA, or ``HEAD~N``).
98
99 ``--rule NAME_OR_TYPE``
100 Run only rules whose name or rule_type matches this value.
101
102 ``--strict``
103 Exit non-zero if any warning-level violations are found.
104
105 ``--json``
106 Emit a machine-readable JSON object.
107 """
108
109 from __future__ import annotations
110
111 import argparse
112 import json
113 import logging
114 import pathlib
115 import sys
116
117 from muse._version import __version__
118 from muse.core.errors import ExitCode
119 from muse.core.repo import read_repo_id, require_repo
120 from muse.core.store import (
121 get_commit_snapshot_manifest,
122 get_head_commit_id,
123 read_current_branch,
124 resolve_commit_ref,
125 )
126 from muse.plugins.code._invariants import load_invariant_rules, run_invariants
127 from muse.core.validation import sanitize_display
128
129
130
131 type _StrMap = dict[str, str]
132 type _ByRule = dict[str, list[str]]
133 logger = logging.getLogger(__name__)
134
135 _RULES_FILE = pathlib.Path(".muse") / "code_invariants.toml"
136
137
138 # ---------------------------------------------------------------------------
139 # CLI plumbing
140 # ---------------------------------------------------------------------------
141
142
143 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
144 """Register the ``invariants`` subcommand."""
145 parser = subparsers.add_parser(
146 "invariants",
147 help="Check architectural invariants from .muse/code_invariants.toml.",
148 description=__doc__,
149 formatter_class=argparse.RawDescriptionHelpFormatter,
150 )
151 parser.add_argument(
152 "--commit", "-c",
153 dest="ref",
154 default=None,
155 metavar="REF",
156 help="Check a historical snapshot (branch, SHA, or HEAD~N).",
157 )
158 parser.add_argument(
159 "--rule",
160 dest="rule_filter",
161 default=None,
162 metavar="NAME_OR_TYPE",
163 help="Run only rules whose name or rule_type matches this value.",
164 )
165 parser.add_argument(
166 "--strict",
167 dest="strict",
168 action="store_true",
169 help="Exit non-zero if any warning-level violations are found.",
170 )
171 parser.add_argument(
172 "--json",
173 dest="as_json",
174 action="store_true",
175 help="Emit results as JSON.",
176 )
177 parser.set_defaults(func=run)
178
179
180 def run(args: argparse.Namespace) -> None:
181 """Check architectural invariants and exit non-zero on violations.
182
183 Delegates rule evaluation to the code-domain plugin engine
184 (``muse.plugins.code._invariants``). All rule types are supported:
185 ``max_complexity``, ``no_circular_imports``, ``no_dead_exports``,
186 ``test_coverage_floor``, ``forbidden_dependency``, ``layer_boundary``.
187
188 Exits with code 0 when all checked rules pass (or only warnings when
189 ``--strict`` is not set). Exits with code 1 when any error-severity
190 rule is violated, or when ``--strict`` is set and warnings are present.
191 """
192 ref: str | None = args.ref
193 as_json: bool = args.as_json
194 rule_filter: str | None = getattr(args, "rule_filter", None)
195 strict: bool = getattr(args, "strict", False)
196
197 root = require_repo()
198 repo_id = read_repo_id(root)
199 branch = read_current_branch(root)
200
201 # Resolve ref — support branch names in addition to SHAs and HEAD~N.
202 resolved_ref: str | None = ref
203 if ref is not None:
204 branch_head = get_head_commit_id(root, ref)
205 if branch_head is not None:
206 resolved_ref = branch_head
207
208 commit = resolve_commit_ref(root, repo_id, branch, resolved_ref)
209 if commit is None:
210 print(f"❌ No commit found for ref '{ref or 'HEAD'}'.", file=sys.stderr)
211 raise SystemExit(ExitCode.USER_ERROR)
212
213 # Guard: refuse to proceed if the manifest is unreadable.
214 manifest = get_commit_snapshot_manifest(root, commit.commit_id)
215 if manifest is None:
216 print(
217 f"❌ Cannot read snapshot for commit {commit.commit_id[:8]} — "
218 "repository may be corrupt.",
219 file=sys.stderr,
220 )
221 raise SystemExit(ExitCode.INTERNAL_ERROR)
222
223 # Load rules — built-in defaults apply when the file is absent.
224 rules_file = root / _RULES_FILE
225 rules = load_invariant_rules(rules_file if rules_file.exists() else None)
226 using_defaults = not rules_file.exists()
227
228 # Apply optional rule filter.
229 if rule_filter:
230 rules = [
231 r for r in rules
232 if rule_filter in (r.get("name", ""), r.get("rule_type", ""))
233 ]
234 if not rules:
235 msg = f"No rules match filter '{rule_filter}'."
236 if as_json:
237 print(json.dumps({
238 "schema_version": __version__,
239 "error": "no_matching_rules",
240 "message": msg,
241 }, indent=2))
242 else:
243 print(f"⚠️ {msg}", file=sys.stderr)
244 raise SystemExit(0)
245
246 if not rules:
247 msg = (
248 "No rules defined. "
249 f"Create {_RULES_FILE} with [[rule]] blocks to define architectural constraints."
250 )
251 if as_json:
252 print(json.dumps({
253 "schema_version": __version__,
254 "commit": commit.commit_id[:8],
255 "branch": branch,
256 "using_defaults": False,
257 "rules_checked": 0,
258 "errors": 0,
259 "warnings": 0,
260 "violations": [],
261 }, indent=2))
262 else:
263 print(f"⚠️ {msg}")
264 raise SystemExit(0)
265
266 # Run all rules via the plugin engine.
267 report = run_invariants(root, commit.commit_id, rules)
268
269 violations = report["violations"]
270 errors = sum(1 for v in violations if v["severity"] == "error")
271 warnings = sum(1 for v in violations if v["severity"] == "warning")
272 exit_code = 1 if (errors > 0 or (strict and warnings > 0)) else 0
273
274 ref_label = commit.commit_id[:8]
275 if ref and ref != commit.commit_id:
276 ref_label = f"{ref} ({commit.commit_id[:8]})"
277
278 if as_json:
279 print(json.dumps(
280 {
281 "schema_version": __version__,
282 "commit": commit.commit_id[:8],
283 "branch": branch,
284 "ref": ref_label,
285 "using_defaults": using_defaults,
286 "rule_filter": rule_filter,
287 "strict": strict,
288 "rules_checked": report["rules_checked"],
289 "violations_total": len(violations),
290 "errors": errors,
291 "warnings": warnings,
292 "violations": list(violations),
293 },
294 indent=2,
295 ))
296 raise SystemExit(exit_code)
297
298 print(f"\nInvariant check — {ref_label}")
299 if using_defaults:
300 print(" (using built-in default rules — create .muse/code_invariants.toml to customise)")
301 if rule_filter:
302 print(f" (filter: {rule_filter})")
303 print("─" * 62)
304
305 if not violations:
306 print(f"\n ✅ All {report['rules_checked']} rule(s) passed.")
307 raise SystemExit(0)
308
309 # Group violations by rule name for clean output.
310 from collections import defaultdict
311 by_rule: _ByRule = defaultdict(list)
312 severity_by_rule: _StrMap = {}
313 for v in violations:
314 by_rule[v["rule_name"]].append(
315 f" {sanitize_display(v['address'])}: {sanitize_display(v['description'])}" if v["address"] != "repo"
316 else f" {sanitize_display(v['description'])}"
317 )
318 severity_by_rule[v["rule_name"]] = v["severity"]
319
320 # Show rules that passed (from original rules list) and rules that violated.
321 violated_names = set(by_rule)
322 passed_count = report["rules_checked"] - len(violated_names)
323
324 for rule in rules:
325 rule_name = rule.get("name", "unnamed")
326 if rule_name not in violated_names:
327 print(f"\n ✅ {sanitize_display(rule_name)}")
328 else:
329 sev = severity_by_rule[rule_name]
330 icon = "🔴" if sev == "error" else "⚠️ "
331 print(f"\n{icon} {sanitize_display(rule_name)} ({sev})")
332 for line in by_rule[rule_name][:5]:
333 print(f" {sanitize_display(line)}")
334 if len(by_rule[rule_name]) > 5:
335 print(f" … and {len(by_rule[rule_name]) - 5} more")
336
337 print(f"\n {passed_count} rule(s) passed · {len(violated_names)} rule(s) violated")
338 print(f" {errors} error(s) · {warnings} warning(s)")
339 raise SystemExit(exit_code)
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago