invariants.py
python
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2
fix: remove commit_exists filter from have anchors — server…
Sonnet 4.6
patch
21 days ago
| 1 | """Domain-agnostic invariants engine for Muse. |
| 2 | |
| 3 | An *invariant* is a semantic rule that a domain's state must satisfy. Rules |
| 4 | are declared in TOML, evaluated against commit snapshots, and reported with |
| 5 | structured violations. Any domain plugin can implement invariant checking |
| 6 | by satisfying the :class:`InvariantChecker` protocol and wiring a CLI command. |
| 7 | |
| 8 | This module defines the **shared vocabulary** — TypedDicts and protocols that |
| 9 | are domain-agnostic. Domain-specific implementations (MIDI, code, genomics…) |
| 10 | |
| 11 | import these types and add their own rule types and evaluators. |
| 12 | |
| 13 | Architecture |
| 14 | ------------ |
| 15 | :: |
| 16 | |
| 17 | muse/core/invariants.py ← this file: shared protocol |
| 18 | muse/plugins/midi/_invariants.py ← MIDI-specific rules + evaluator |
| 19 | muse/plugins/code/_invariants.py ← code-specific rules + evaluator |
| 20 | muse/cli/commands/midi_check.py ← CLI wiring for MIDI |
| 21 | muse/cli/commands/code_check.py ← CLI wiring for code |
| 22 | |
| 23 | TOML rule file format (shared across all domains):: |
| 24 | |
| 25 | [[rule]] |
| 26 | name = "my_rule" # unique human-readable identifier |
| 27 | severity = "error" # "info" | "warning" | "error" |
| 28 | scope = "file" # domain-specific scope tag |
| 29 | rule_type = "max_complexity" # domain-specific rule type string |
| 30 | |
| 31 | [rule.params] |
| 32 | threshold = 10 # rule-specific numeric / string params |
| 33 | |
| 34 | Severity levels |
| 35 | --------------- |
| 36 | - ``"error"`` — must be resolved before committing (when ``--strict`` is set). |
| 37 | - ``"warning"`` — reported but does not block commits. |
| 38 | - ``"info"`` — informational; surfaced in ``muse check`` output only. |
| 39 | |
| 40 | Public API |
| 41 | ---------- |
| 42 | - :data:`InvariantSeverity` — severity literal type alias. |
| 43 | - :class:`BaseViolation` — domain-agnostic violation record. |
| 44 | - :class:`BaseReport` — full check report for one commit. |
| 45 | - :class:`InvariantChecker` — Protocol every domain checker must satisfy. |
| 46 | - :func:`make_report` — build a ``BaseReport`` from a violation list. |
| 47 | - :func:`load_rules_toml` — parse any ``[[rule]]`` TOML file. |
| 48 | - :func:`format_report` — human-readable report text. |
| 49 | """ |
| 50 | |
| 51 | import logging |
| 52 | import pathlib |
| 53 | from typing import Literal, Protocol, TypedDict, runtime_checkable |
| 54 | |
| 55 | logger = logging.getLogger(__name__) |
| 56 | |
| 57 | type _RuleScalar = str | int | float |
| 58 | type _RuleParams = dict[str, _RuleScalar] |
| 59 | type _RuleDict = dict[str, _RuleScalar | _RuleParams] |
| 60 | type _RuleList = list[_RuleDict] |
| 61 | |
| 62 | # --------------------------------------------------------------------------- |
| 63 | # Shared severity literal |
| 64 | # --------------------------------------------------------------------------- |
| 65 | |
| 66 | InvariantSeverity = Literal["info", "warning", "error"] |
| 67 | |
| 68 | # --------------------------------------------------------------------------- |
| 69 | # Domain-agnostic violation + report TypedDicts |
| 70 | # --------------------------------------------------------------------------- |
| 71 | |
| 72 | class BaseViolation(TypedDict): |
| 73 | """A single invariant violation, domain-agnostic. |
| 74 | |
| 75 | Domain implementations extend this with additional fields (e.g. ``track`` |
| 76 | for MIDI, ``file`` and ``symbol`` for code). |
| 77 | |
| 78 | ``rule_name`` The name of the rule that fired. |
| 79 | ``severity`` Violation severity inherited from the rule declaration. |
| 80 | ``address`` Dotted path to the violating element |
| 81 | (e.g. ``"src/utils.py::my_fn"`` or ``"piano.mid/bar:4"``). |
| 82 | ``description`` Human-readable explanation of the violation. |
| 83 | """ |
| 84 | |
| 85 | rule_name: str |
| 86 | severity: InvariantSeverity |
| 87 | address: str |
| 88 | description: str |
| 89 | |
| 90 | class BaseReport(TypedDict): |
| 91 | """Full invariant check report for one commit, domain-agnostic. |
| 92 | |
| 93 | ``commit_id`` The commit that was checked. |
| 94 | ``domain`` Domain tag (e.g. ``"midi"``, ``"code"``). |
| 95 | ``violations`` All violations found, sorted by address. |
| 96 | ``rules_checked`` Number of rules evaluated. |
| 97 | ``has_errors`` ``True`` when any violation has severity ``"error"``. |
| 98 | ``has_warnings`` ``True`` when any violation has severity ``"warning"``. |
| 99 | """ |
| 100 | |
| 101 | commit_id: str |
| 102 | domain: str |
| 103 | violations: list[BaseViolation] |
| 104 | rules_checked: int |
| 105 | has_errors: bool |
| 106 | has_warnings: bool |
| 107 | |
| 108 | # --------------------------------------------------------------------------- |
| 109 | # InvariantChecker protocol |
| 110 | # --------------------------------------------------------------------------- |
| 111 | |
| 112 | @runtime_checkable |
| 113 | class InvariantChecker(Protocol): |
| 114 | """Protocol every domain invariant checker must satisfy. |
| 115 | |
| 116 | Domain plugins implement this by providing :meth:`check` — a function that |
| 117 | loads and evaluates the domain's invariant rules against a commit, returning |
| 118 | a :class:`BaseReport`. The CLI ``muse check`` command dispatches to the |
| 119 | domain's registered checker via this protocol. |
| 120 | |
| 121 | Example implementation:: |
| 122 | |
| 123 | class MyDomainChecker: |
| 124 | def check( |
| 125 | self, |
| 126 | repo_root: pathlib.Path, |
| 127 | commit_id: str, |
| 128 | *, |
| 129 | rules_file: pathlib.Path | None = None, |
| 130 | ) -> BaseReport: |
| 131 | rules = load_rules_toml(rules_file or default_path) |
| 132 | violations = _evaluate(repo_root, commit_id, rules) |
| 133 | return make_report(commit_id, "mydomain", violations, len(rules)) |
| 134 | """ |
| 135 | |
| 136 | def check( |
| 137 | self, |
| 138 | repo_root: pathlib.Path, |
| 139 | commit_id: str, |
| 140 | *, |
| 141 | rules_file: pathlib.Path | None = None, |
| 142 | ) -> BaseReport: |
| 143 | """Evaluate invariant rules and return a structured report. |
| 144 | |
| 145 | Args: |
| 146 | repo_root: Repository root (contains ``.muse/``). |
| 147 | commit_id: Commit to check. |
| 148 | rules_file: Path to a TOML rule file. ``None`` → use the |
| 149 | domain's default location. |
| 150 | |
| 151 | Returns: |
| 152 | A :class:`BaseReport` with all violations and summary flags. |
| 153 | """ |
| 154 | ... |
| 155 | |
| 156 | # --------------------------------------------------------------------------- |
| 157 | # Helpers |
| 158 | # --------------------------------------------------------------------------- |
| 159 | |
| 160 | def make_report( |
| 161 | commit_id: str, |
| 162 | domain: str, |
| 163 | violations: list[BaseViolation], |
| 164 | rules_checked: int, |
| 165 | ) -> BaseReport: |
| 166 | """Build a :class:`BaseReport` from a flat violation list. |
| 167 | |
| 168 | Sorts violations by address then rule name for deterministic output. |
| 169 | |
| 170 | Args: |
| 171 | commit_id: Commit that was checked. |
| 172 | domain: Domain tag. |
| 173 | violations: All violations found. |
| 174 | rules_checked: Number of rules that were evaluated. |
| 175 | |
| 176 | Returns: |
| 177 | A fully populated :class:`BaseReport`. |
| 178 | """ |
| 179 | sorted_violations = sorted(violations, key=lambda v: (v["address"], v["rule_name"])) |
| 180 | return BaseReport( |
| 181 | commit_id=commit_id, |
| 182 | domain=domain, |
| 183 | violations=sorted_violations, |
| 184 | rules_checked=rules_checked, |
| 185 | has_errors=any(v["severity"] == "error" for v in violations), |
| 186 | has_warnings=any(v["severity"] == "warning" for v in violations), |
| 187 | ) |
| 188 | |
| 189 | def load_rules_toml(path: pathlib.Path) -> _RuleList: |
| 190 | """Parse a ``[[rule]]`` TOML file and return the raw rule dicts. |
| 191 | |
| 192 | Returns an empty list when the file does not exist (domain then uses |
| 193 | built-in defaults). |
| 194 | |
| 195 | Args: |
| 196 | path: Path to the TOML file. |
| 197 | |
| 198 | Returns: |
| 199 | List of raw rule dicts (``{"name": ..., "severity": ..., ...}``). |
| 200 | """ |
| 201 | if not path.exists(): |
| 202 | logger.debug("Invariants rules file not found at %s — using defaults", path) |
| 203 | return [] |
| 204 | import tomllib # stdlib on Python ≥ 3.11; Muse requires 3.14 |
| 205 | try: |
| 206 | data = tomllib.loads(path.read_text()) |
| 207 | rules: _RuleList = data.get("rule", []) |
| 208 | return rules |
| 209 | except Exception as exc: |
| 210 | logger.warning("Failed to parse invariants file %s: %s", path, exc) |
| 211 | return [] |
| 212 | |
| 213 | def diff_reports(current: BaseReport, base: BaseReport) -> BaseReport: |
| 214 | """Return a report containing only violations that are new in *current*. |
| 215 | |
| 216 | A violation is considered *new* when no violation with the same |
| 217 | (rule_name, address) tuple exists in *base*. This lets CI gates report |
| 218 | only regressions rather than pre-existing noise. |
| 219 | |
| 220 | Args: |
| 221 | current: Report for the commit being checked. |
| 222 | base: Report for the reference commit (e.g. HEAD~1, main). |
| 223 | |
| 224 | Returns: |
| 225 | A :class:`BaseReport` with the same metadata as *current* but only |
| 226 | the net-new violations. |
| 227 | """ |
| 228 | base_keys: set[tuple[str, str]] = { |
| 229 | (v["rule_name"], v["address"]) for v in base["violations"] |
| 230 | } |
| 231 | new_violations = [ |
| 232 | v for v in current["violations"] |
| 233 | if (v["rule_name"], v["address"]) not in base_keys |
| 234 | ] |
| 235 | return make_report( |
| 236 | current["commit_id"], |
| 237 | current["domain"], |
| 238 | new_violations, |
| 239 | current["rules_checked"], |
| 240 | ) |
| 241 | |
| 242 | def format_report(report: BaseReport, *, color: bool = True) -> str: |
| 243 | """Return a human-readable multi-line report string. |
| 244 | |
| 245 | Args: |
| 246 | report: The report to format. |
| 247 | color: If ``True``, prefix error/warning/info lines with emoji. |
| 248 | |
| 249 | Returns: |
| 250 | Formatted string ready for printing to stdout. |
| 251 | """ |
| 252 | lines: list[str] = [] |
| 253 | prefix = { |
| 254 | "error": "❌" if color else "[error]", |
| 255 | "warning": "⚠️ " if color else "[warn] ", |
| 256 | "info": "ℹ️ " if color else "[info] ", |
| 257 | } |
| 258 | for v in report["violations"]: |
| 259 | p = prefix.get(v["severity"], " ") |
| 260 | lines.append(f" {p} [{v['rule_name']}] {v['address']}: {v['description']}") |
| 261 | |
| 262 | checked = report["rules_checked"] |
| 263 | total = len(report["violations"]) |
| 264 | errors = sum(1 for v in report["violations"] if v["severity"] == "error") |
| 265 | warnings = sum(1 for v in report["violations"] if v["severity"] == "warning") |
| 266 | |
| 267 | summary = f"\n{checked} rules checked — {total} violation(s)" |
| 268 | if errors: |
| 269 | summary += f", {errors} error(s)" |
| 270 | if warnings: |
| 271 | summary += f", {warnings} warning(s)" |
| 272 | if not total: |
| 273 | summary = f"\n✅ {checked} rules checked — no violations" |
| 274 | |
| 275 | return "\n".join(lines) + summary |
File History
4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2
fix: remove commit_exists filter from have anchors — server…
Sonnet 4.6
patch
21 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e
fix: rename objects→blobs in push client and all stale test…
Sonnet 4.6
patch
23 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a
fix: repair four test failures from post-migration audit
Sonnet 4.6
patch
29 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf
fix: unified object store migration — idempotent writes, JS…
Sonnet 4.6
minor
⚠
29 days ago