_invariants.py
python
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839
chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm…
Sonnet 5
patch
13 days ago
| 1 | """Code-domain invariants engine for Muse. |
| 2 | |
| 3 | Evaluates semantic rules against code snapshots. Rules are declared in |
| 4 | ``.muse/code_invariants.toml`` and evaluated at commit time, merge time, or |
| 5 | on-demand via ``muse code-check``. |
| 6 | |
| 7 | Rule types |
| 8 | ---------- |
| 9 | |
| 10 | ``max_complexity`` |
| 11 | Detects functions / methods whose estimated cyclomatic complexity exceeds |
| 12 | *threshold*. Complexity is approximated by counting control-flow branch |
| 13 | points (``if``, ``elif``, ``for``, ``while``, ``except``, ``with``, |
| 14 | ``and``, ``or``) inside each symbol's body. This correlates well with |
| 15 | real cyclomatic complexity for Python and is language-agnostic for other |
| 16 | tree-sitter-parsed languages. |
| 17 | |
| 18 | ``no_circular_imports`` |
| 19 | Detects import cycles among Python files in the snapshot. Builds a |
| 20 | directed graph (file → files it imports) and runs DFS cycle detection. |
| 21 | Reports each cycle as one violation at the root file of the cycle. |
| 22 | |
| 23 | ``no_dead_exports`` |
| 24 | Detects top-level functions and classes that are never imported by any |
| 25 | other file in the snapshot (dead exports / unreachable public API). |
| 26 | Only applies to semantic files; test files and ``__init__.py`` are exempt. |
| 27 | |
| 28 | ``test_coverage_floor`` |
| 29 | Requires that at least *min_ratio* of non-test functions have a |
| 30 | corresponding test function (detected by ``test_`` prefix convention). |
| 31 | Reports the actual vs required coverage ratio when the floor is not met. |
| 32 | |
| 33 | TOML example |
| 34 | ------------ |
| 35 | :: |
| 36 | |
| 37 | [[rule]] |
| 38 | name = "complexity_gate" |
| 39 | severity = "error" |
| 40 | scope = "function" |
| 41 | rule_type = "max_complexity" |
| 42 | [rule.params] |
| 43 | threshold = 15 |
| 44 | |
| 45 | [[rule]] |
| 46 | name = "no_cycles" |
| 47 | severity = "error" |
| 48 | scope = "file" |
| 49 | rule_type = "no_circular_imports" |
| 50 | |
| 51 | [[rule]] |
| 52 | name = "dead_exports" |
| 53 | severity = "warning" |
| 54 | scope = "file" |
| 55 | rule_type = "no_dead_exports" |
| 56 | |
| 57 | [[rule]] |
| 58 | name = "test_coverage" |
| 59 | severity = "warning" |
| 60 | scope = "repo" |
| 61 | rule_type = "test_coverage_floor" |
| 62 | [rule.params] |
| 63 | min_ratio = 0.30 |
| 64 | |
| 65 | Public API |
| 66 | ---------- |
| 67 | - :class:`CodeInvariantRule` — code-specific rule declaration. |
| 68 | - :class:`CodeViolation` — violation with file + symbol address. |
| 69 | - :class:`CodeInvariantReport` — full report for one commit. |
| 70 | - :class:`CodeChecker` — satisfies :class:`~muse.core.invariants.InvariantChecker`. |
| 71 | - :func:`load_invariant_rules` — load from TOML with built-in defaults. |
| 72 | - :func:`run_invariants` — top-level runner. |
| 73 | """ |
| 74 | |
| 75 | import ast |
| 76 | import logging |
| 77 | import pathlib |
| 78 | from typing import Literal, TypedDict |
| 79 | |
| 80 | from muse.core.cache_base import MsgpackCache, _RawCacheMap |
| 81 | from muse.core.paths import code_invariants_path, muse_dir as _muse_dir |
| 82 | from muse.core.invariants import ( |
| 83 | BaseReport, |
| 84 | BaseViolation, |
| 85 | InvariantSeverity, |
| 86 | load_rules_toml, |
| 87 | make_report, |
| 88 | ) |
| 89 | from muse.core.object_store import read_object |
| 90 | from muse.core.types import Manifest |
| 91 | from muse.core.snapshots import get_commit_snapshot_manifest |
| 92 | from muse.core.validation import MAX_AST_BYTES |
| 93 | |
| 94 | logger = logging.getLogger(__name__) |
| 95 | |
| 96 | type ComplexityMap = dict[str, int] # symbol_address → branch-count score |
| 97 | type FileDataMap = dict[str, _FileData] # content_hash → parsed file data |
| 98 | type ImportGraph = dict[str, set[str]] # file_path → set of imported file_paths |
| 99 | type ParamsDict = dict[str, str | int | float] # rule params (dynamic keys) |
| 100 | |
| 101 | _FILE_CACHE_VERSION = 2 |
| 102 | |
| 103 | # --------------------------------------------------------------------------- |
| 104 | # Per-file AST cache — two-level: intra-run (in-memory) + cross-run (JSON) |
| 105 | # --------------------------------------------------------------------------- |
| 106 | |
| 107 | class _FileData(TypedDict): |
| 108 | """All AST-derived data for one Python file, derived from a single parse. |
| 109 | |
| 110 | Keyed by file content hash in :class:`_InvariantFileCache` so a file that |
| 111 | has not changed between ``muse code invariants`` runs costs zero parse time. |
| 112 | |
| 113 | ``raw_module_imports`` |
| 114 | Module names from ``import X`` statements. e.g. ``"import os"`` → ``["os"]``. |
| 115 | ``from_module`` / ``from_name`` |
| 116 | Parallel lists for ``from M import N`` statements. Same index = one import pair. |
| 117 | ``top_level_fns`` |
| 118 | Public (non-underscore, non-``main``) top-level function names. |
| 119 | ``top_level_classes`` |
| 120 | Public top-level class names. |
| 121 | ``has_all`` |
| 122 | ``True`` when the file declares ``__all__``. |
| 123 | ``complexity`` |
| 124 | ``{symbol_address: score}`` — branch-count cyclomatic complexity proxy. |
| 125 | """ |
| 126 | |
| 127 | raw_module_imports: list[str] |
| 128 | from_module: list[str] |
| 129 | from_name: list[str] |
| 130 | top_level_fns: list[str] |
| 131 | top_level_classes: list[str] |
| 132 | has_all: bool |
| 133 | complexity: ComplexityMap |
| 134 | |
| 135 | def _empty_file_data() -> _FileData: |
| 136 | return _FileData( |
| 137 | raw_module_imports=[], |
| 138 | from_module=[], |
| 139 | from_name=[], |
| 140 | top_level_fns=[], |
| 141 | top_level_classes=[], |
| 142 | has_all=False, |
| 143 | complexity={}, |
| 144 | ) |
| 145 | |
| 146 | def _parse_file_info(source: bytes, file_path: str) -> _FileData: |
| 147 | """Parse one Python file and extract everything all rules need. |
| 148 | |
| 149 | A single ``ast.parse`` call populates all fields. Non-Python files and |
| 150 | files that fail to parse return an empty :class:`_FileData`. |
| 151 | """ |
| 152 | if not file_path.endswith((".py", ".pyi")): |
| 153 | return _empty_file_data() |
| 154 | if len(source) > MAX_AST_BYTES: |
| 155 | return _empty_file_data() |
| 156 | try: |
| 157 | tree = ast.parse(source, filename=file_path) |
| 158 | except SyntaxError: |
| 159 | return _empty_file_data() |
| 160 | |
| 161 | branch_nodes = ( |
| 162 | ast.If, ast.For, ast.While, ast.ExceptHandler, |
| 163 | ast.With, ast.AsyncWith, ast.AsyncFor, |
| 164 | ast.BoolOp, ast.Assert, ast.comprehension, |
| 165 | ) |
| 166 | |
| 167 | raw_module_imports: list[str] = [] |
| 168 | from_module: list[str] = [] |
| 169 | from_name: list[str] = [] |
| 170 | top_level_fns: list[str] = [] |
| 171 | top_level_classes: list[str] = [] |
| 172 | has_all = False |
| 173 | complexity: ComplexityMap = {} |
| 174 | |
| 175 | def _score(node: ast.FunctionDef | ast.AsyncFunctionDef, prefix: str) -> None: |
| 176 | qualified = f"{prefix}{node.name}" if prefix else node.name |
| 177 | addr = f"{file_path}::{qualified}" |
| 178 | score = 1 |
| 179 | for child in ast.walk(node): |
| 180 | if isinstance(child, branch_nodes): |
| 181 | score += 1 |
| 182 | complexity[addr] = score |
| 183 | |
| 184 | # Single walk: collect imports, complexity, top-level names. |
| 185 | for node in ast.walk(tree): |
| 186 | if isinstance(node, ast.Import): |
| 187 | for alias in node.names: |
| 188 | raw_module_imports.append(alias.name) |
| 189 | elif isinstance(node, ast.ImportFrom) and node.module: |
| 190 | for alias in node.names: |
| 191 | from_module.append(node.module) |
| 192 | from_name.append(alias.name) |
| 193 | elif isinstance(node, ast.ClassDef): |
| 194 | for item in ast.walk(node): |
| 195 | if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 196 | _score(item, f"{node.name}.") |
| 197 | elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 198 | _score(node, "") |
| 199 | |
| 200 | # Top-level names only (not recursive — iter_child_nodes). |
| 201 | for node in ast.iter_child_nodes(tree): |
| 202 | if isinstance(node, ast.Assign): |
| 203 | if any( |
| 204 | isinstance(t, ast.Name) and t.id == "__all__" |
| 205 | for t in node.targets |
| 206 | ): |
| 207 | has_all = True |
| 208 | elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 209 | if not node.name.startswith("_") and node.name != "main": |
| 210 | top_level_fns.append(node.name) |
| 211 | elif isinstance(node, ast.ClassDef): |
| 212 | if not node.name.startswith("_"): |
| 213 | top_level_classes.append(node.name) |
| 214 | |
| 215 | return _FileData( |
| 216 | raw_module_imports=raw_module_imports, |
| 217 | from_module=from_module, |
| 218 | from_name=from_name, |
| 219 | top_level_fns=top_level_fns, |
| 220 | top_level_classes=top_level_classes, |
| 221 | has_all=has_all, |
| 222 | complexity=complexity, |
| 223 | ) |
| 224 | |
| 225 | class _InvariantFileCache(MsgpackCache): |
| 226 | """Persistent per-file AST analysis cache keyed by content hash. |
| 227 | |
| 228 | Stored at ``.muse/cache/invariants.json``. Inherits load/save/get/ |
| 229 | put/prune/size/empty from :class:`~muse.core.cache_base.MsgpackCache` |
| 230 | (Pattern A — mkstemp + replace). |
| 231 | |
| 232 | On a warm cache (unchanged snapshot), ``muse code invariants`` skips |
| 233 | all ``ast.parse`` calls — reducing parse time from O(N×R) to O(1). |
| 234 | |
| 235 | Attributes |
| 236 | ---------- |
| 237 | _cache_dir : pathlib.Path | None |
| 238 | Absolute path to ``.muse/cache/``. ``None`` for in-memory-only |
| 239 | instances created via :meth:`empty`. :meth:`save` is a no-op |
| 240 | when ``None``. |
| 241 | _dirty : bool |
| 242 | Set to ``True`` by :meth:`put` and :meth:`prune` whenever entries |
| 243 | change. Reset to ``False`` by a successful :meth:`save`. |
| 244 | """ |
| 245 | |
| 246 | _CACHE_FILENAME = "invariants.json" |
| 247 | _CACHE_VERSION = _FILE_CACHE_VERSION |
| 248 | _TEMP_PREFIX = ".invariants_" |
| 249 | |
| 250 | @classmethod |
| 251 | def _deserialize_entries(cls, raw: _RawCacheMap) -> _RawCacheMap: |
| 252 | """Validate and convert raw JSON entries to typed _FileData entries. |
| 253 | |
| 254 | Each value must be a dict; invalid entries are skipped so a partially |
| 255 | corrupt file does not poison the whole cache. |
| 256 | """ |
| 257 | data: FileDataMap = {} |
| 258 | for content_hash, v in raw.items(): |
| 259 | if not isinstance(content_hash, str) or not isinstance(v, dict): |
| 260 | continue |
| 261 | raw_mi = v.get("raw_module_imports") |
| 262 | raw_fm = v.get("from_module") |
| 263 | raw_fn = v.get("from_name") |
| 264 | raw_tf = v.get("top_level_fns") |
| 265 | raw_tc = v.get("top_level_classes") |
| 266 | raw_cx = v.get("complexity") |
| 267 | data[content_hash] = _FileData( |
| 268 | raw_module_imports=[str(x) for x in raw_mi] if isinstance(raw_mi, list) else [], |
| 269 | from_module=[str(x) for x in raw_fm] if isinstance(raw_fm, list) else [], |
| 270 | from_name=[str(x) for x in raw_fn] if isinstance(raw_fn, list) else [], |
| 271 | top_level_fns=[str(x) for x in raw_tf] if isinstance(raw_tf, list) else [], |
| 272 | top_level_classes=[str(x) for x in raw_tc] if isinstance(raw_tc, list) else [], |
| 273 | has_all=bool(v.get("has_all", False)), |
| 274 | complexity={ |
| 275 | str(k): val |
| 276 | for k, val in raw_cx.items() |
| 277 | if isinstance(val, int) |
| 278 | } if isinstance(raw_cx, dict) else {}, |
| 279 | ) |
| 280 | return data |
| 281 | |
| 282 | def _serialize_entries(self) -> _RawCacheMap: |
| 283 | """Return ``_entries`` as-is — _FileData TypedDicts are JSON-serialisable.""" |
| 284 | return dict(self._entries) |
| 285 | |
| 286 | @classmethod |
| 287 | def from_root(cls, root: pathlib.Path) -> "_InvariantFileCache": # type: ignore[override] |
| 288 | """Load for a repository root, returning ``empty()`` when ``.muse/`` is absent. |
| 289 | |
| 290 | Delegates to :meth:`load`, which already accepts a ``repo_root`` and |
| 291 | handles the ``is_dir()`` check internally. |
| 292 | |
| 293 | Parameters |
| 294 | ---------- |
| 295 | root: |
| 296 | Repository root (the directory that contains ``.muse/``). |
| 297 | """ |
| 298 | return cls.load(root) # type: ignore[return-value] |
| 299 | |
| 300 | @classmethod |
| 301 | def load(cls, repo_root: pathlib.Path) -> "_InvariantFileCache": # type: ignore[override] |
| 302 | """Load from ``repo_root/.muse/cache/invariants.json``. |
| 303 | |
| 304 | Takes ``repo_root`` (not ``muse_dir``) to maintain API consistency |
| 305 | with the rest of the invariants module. Returns an empty cache on any |
| 306 | failure — missing file, corrupt bytes, version mismatch — and never |
| 307 | raises. Invalid individual entries are skipped so a partially corrupt |
| 308 | file does not poison the whole cache. |
| 309 | |
| 310 | Parameters |
| 311 | ---------- |
| 312 | repo_root: |
| 313 | Root directory of the repository (the directory that contains |
| 314 | ``.muse/``). |
| 315 | """ |
| 316 | muse_dir = _muse_dir(repo_root) |
| 317 | if not muse_dir.is_dir(): |
| 318 | return cls(None, {}) |
| 319 | return super().load(muse_dir) # type: ignore[return-value] |
| 320 | |
| 321 | # Type-narrowing overrides |
| 322 | |
| 323 | def get(self, content_hash: str) -> _FileData | None: |
| 324 | """Return cached ``_FileData`` for *content_hash*, or ``None`` on miss.""" |
| 325 | return self._entries.get(content_hash) # type: ignore[return-value] |
| 326 | |
| 327 | def put(self, content_hash: str, data: _FileData) -> None: |
| 328 | """Store *data* under *content_hash* and mark the cache dirty.""" |
| 329 | super().put(content_hash, data) |
| 330 | |
| 331 | def prune(self, live_hashes: set[str]) -> None: |
| 332 | """Remove entries whose content hashes are not in *live_hashes*. |
| 333 | |
| 334 | Sets ``_dirty`` only when at least one entry is removed. Call this |
| 335 | after computing the current manifest to prevent unbounded growth. |
| 336 | """ |
| 337 | super().prune(live_hashes) |
| 338 | |
| 339 | def _build_file_data( |
| 340 | manifest: Manifest, |
| 341 | repo_root: pathlib.Path, |
| 342 | cache: _InvariantFileCache, |
| 343 | ) -> FileDataMap: |
| 344 | """Return :class:`_FileData` for every Python file in *manifest*. |
| 345 | |
| 346 | Serves from *cache* on a hit; reads + parses the file on a miss, then |
| 347 | stores the result back into the cache (which is later persisted). |
| 348 | |
| 349 | This is the entry point for both intra-run sharing (all rules receive |
| 350 | the same dict) and cross-run caching (unchanged files are never re-parsed). |
| 351 | """ |
| 352 | result: FileDataMap = {} |
| 353 | for file_path, content_hash in manifest.items(): |
| 354 | if not file_path.endswith((".py", ".pyi")): |
| 355 | continue |
| 356 | cached = cache.get(content_hash) |
| 357 | if cached is not None: |
| 358 | result[file_path] = cached |
| 359 | continue |
| 360 | source = read_object(repo_root, content_hash) |
| 361 | if source is None: |
| 362 | continue |
| 363 | info = _parse_file_info(source, file_path) |
| 364 | cache.put(content_hash, info) |
| 365 | result[file_path] = info |
| 366 | return result |
| 367 | |
| 368 | # --------------------------------------------------------------------------- |
| 369 | # Types |
| 370 | # --------------------------------------------------------------------------- |
| 371 | |
| 372 | class _RuleRequired(TypedDict): |
| 373 | name: str |
| 374 | severity: InvariantSeverity |
| 375 | scope: Literal["function", "file", "repo", "global"] |
| 376 | rule_type: str |
| 377 | |
| 378 | class CodeInvariantRule(_RuleRequired, total=False): |
| 379 | """A single code invariant rule declaration. |
| 380 | |
| 381 | ``name`` Unique human-readable identifier. |
| 382 | ``severity`` ``"info"``, ``"warning"``, or ``"error"``. |
| 383 | ``scope`` Granularity: ``"function"``, ``"file"``, ``"repo"``. |
| 384 | ``rule_type`` Built-in type: ``"max_complexity"``, |
| 385 | ``"no_circular_imports"``, ``"no_dead_exports"``, |
| 386 | ``"test_coverage_floor"``. |
| 387 | ``params`` Rule-specific numeric / string parameters. |
| 388 | """ |
| 389 | |
| 390 | params: ParamsDict |
| 391 | |
| 392 | class CodeViolation(TypedDict): |
| 393 | """A code invariant violation with precise source location. |
| 394 | |
| 395 | ``rule_name`` Rule that fired. |
| 396 | ``severity`` Inherited from the rule. |
| 397 | ``address`` ``"file.py::symbol_name"`` or ``"file.py"`` for file-level. |
| 398 | ``description`` Human-readable explanation. |
| 399 | ``file`` Workspace-relative file path. |
| 400 | ``symbol`` Symbol name (empty string for file-level violations). |
| 401 | ``detail`` Additional context (e.g. complexity score, cycle path). |
| 402 | """ |
| 403 | |
| 404 | rule_name: str |
| 405 | severity: InvariantSeverity |
| 406 | address: str |
| 407 | description: str |
| 408 | file: str |
| 409 | symbol: str |
| 410 | detail: str |
| 411 | |
| 412 | # --------------------------------------------------------------------------- |
| 413 | # Built-in default rules |
| 414 | # --------------------------------------------------------------------------- |
| 415 | |
| 416 | _BUILTIN_DEFAULTS: list[CodeInvariantRule] = [ |
| 417 | CodeInvariantRule( |
| 418 | name="complexity_gate", |
| 419 | severity="warning", |
| 420 | scope="function", |
| 421 | rule_type="max_complexity", |
| 422 | params={"threshold": 10}, |
| 423 | ), |
| 424 | CodeInvariantRule( |
| 425 | name="no_cycles", |
| 426 | severity="error", |
| 427 | scope="file", |
| 428 | rule_type="no_circular_imports", |
| 429 | params={}, |
| 430 | ), |
| 431 | CodeInvariantRule( |
| 432 | name="dead_exports", |
| 433 | severity="warning", |
| 434 | scope="file", |
| 435 | rule_type="no_dead_exports", |
| 436 | params={}, |
| 437 | ), |
| 438 | ] |
| 439 | |
| 440 | # --------------------------------------------------------------------------- |
| 441 | # Rule implementations |
| 442 | # --------------------------------------------------------------------------- |
| 443 | |
| 444 | def _estimate_complexity(source: bytes, file_path: str) -> ComplexityMap: |
| 445 | """Return {symbol_address: complexity_score} for a Python source file. |
| 446 | |
| 447 | Uses a simple branch-count heuristic — same logic as :func:`_parse_file_info`. |
| 448 | Prefer passing ``file_data`` to :func:`check_max_complexity` instead of |
| 449 | calling this directly, so the parse cost is shared across all rules. |
| 450 | """ |
| 451 | return _parse_file_info(source, file_path)["complexity"] |
| 452 | |
| 453 | def check_max_complexity( |
| 454 | manifest: Manifest, |
| 455 | repo_root: pathlib.Path, |
| 456 | rule_name: str, |
| 457 | severity: InvariantSeverity, |
| 458 | *, |
| 459 | threshold: int = 10, |
| 460 | file_data: FileDataMap | None = None, |
| 461 | ) -> list[CodeViolation]: |
| 462 | """Detect functions whose estimated cyclomatic complexity exceeds *threshold*. |
| 463 | |
| 464 | When *file_data* is supplied (built by :func:`_build_file_data`), no I/O |
| 465 | or parsing is performed — complexity scores come directly from the cache. |
| 466 | """ |
| 467 | violations: list[CodeViolation] = [] |
| 468 | |
| 469 | if file_data is not None: |
| 470 | # Fast path: scores already computed, no I/O needed. |
| 471 | for file_path, fd in sorted(file_data.items()): |
| 472 | for addr, score in sorted(fd["complexity"].items()): |
| 473 | if score > threshold: |
| 474 | symbol = addr.split("::", 1)[-1] if "::" in addr else "" |
| 475 | violations.append(CodeViolation( |
| 476 | rule_name=rule_name, |
| 477 | severity=severity, |
| 478 | address=addr, |
| 479 | description=( |
| 480 | f"Complexity {score} exceeds threshold {threshold}. " |
| 481 | "Consider extracting helper functions." |
| 482 | ), |
| 483 | file=file_path, |
| 484 | symbol=symbol, |
| 485 | detail=f"score={score} threshold={threshold}", |
| 486 | )) |
| 487 | return violations |
| 488 | |
| 489 | # Slow path: read and parse each file individually. |
| 490 | for file_path, content_hash in manifest.items(): |
| 491 | if not file_path.endswith((".py", ".pyi")): |
| 492 | continue |
| 493 | source = read_object(repo_root, content_hash) |
| 494 | if source is None: |
| 495 | continue |
| 496 | scores = _estimate_complexity(source, file_path) |
| 497 | for addr, score in sorted(scores.items()): |
| 498 | if score > threshold: |
| 499 | symbol = addr.split("::", 1)[-1] if "::" in addr else "" |
| 500 | violations.append(CodeViolation( |
| 501 | rule_name=rule_name, |
| 502 | severity=severity, |
| 503 | address=addr, |
| 504 | description=( |
| 505 | f"Complexity {score} exceeds threshold {threshold}. " |
| 506 | "Consider extracting helper functions." |
| 507 | ), |
| 508 | file=file_path, |
| 509 | symbol=symbol, |
| 510 | detail=f"score={score} threshold={threshold}", |
| 511 | )) |
| 512 | return violations |
| 513 | |
| 514 | def _build_module_index(manifest: Manifest) -> Manifest: |
| 515 | """Return a module-name → file-path index for intra-repo import resolution.""" |
| 516 | module_to_file: Manifest = {} |
| 517 | for fp in manifest: |
| 518 | if fp.endswith((".py", ".pyi")): |
| 519 | mod = fp.removesuffix(".pyi").removesuffix(".py").replace("/", ".").replace("\\", ".") |
| 520 | module_to_file[mod] = fp |
| 521 | module_to_file.setdefault(mod.rsplit(".", 1)[-1], fp) |
| 522 | return module_to_file |
| 523 | |
| 524 | def _build_import_graph( |
| 525 | manifest: Manifest, |
| 526 | repo_root: pathlib.Path, |
| 527 | *, |
| 528 | file_data: FileDataMap | None = None, |
| 529 | ) -> ImportGraph: |
| 530 | """Build a directed import graph: {file → set of imported files}. |
| 531 | |
| 532 | Only tracks intra-repo imports (files that exist in the manifest). |
| 533 | |
| 534 | When *file_data* is supplied (built by :func:`_build_file_data`), no I/O |
| 535 | or ``ast.parse`` calls are made — imports come directly from the cache. |
| 536 | """ |
| 537 | module_to_file = _build_module_index(manifest) |
| 538 | graph: ImportGraph = {fp: set() for fp in manifest if fp.endswith(".py")} |
| 539 | |
| 540 | if file_data is not None: |
| 541 | # Fast path: resolve imports from pre-parsed _FileData — zero I/O. |
| 542 | for file_path, fd in file_data.items(): |
| 543 | if not file_path.endswith(".py"): |
| 544 | continue |
| 545 | for mod_name in fd["raw_module_imports"]: |
| 546 | t = module_to_file.get(mod_name) |
| 547 | if t and t != file_path: |
| 548 | graph[file_path].add(t) |
| 549 | for module, name in zip(fd["from_module"], fd["from_name"]): |
| 550 | t = module_to_file.get(module) |
| 551 | if t and t != file_path: |
| 552 | graph[file_path].add(t) |
| 553 | t2 = module_to_file.get(f"{module}.{name}") |
| 554 | if t2 and t2 != file_path: |
| 555 | graph[file_path].add(t2) |
| 556 | return graph |
| 557 | |
| 558 | # Slow path: read and parse each file individually. |
| 559 | for file_path, content_hash in manifest.items(): |
| 560 | if not file_path.endswith(".py"): |
| 561 | continue |
| 562 | source = read_object(repo_root, content_hash) |
| 563 | if source is None: |
| 564 | continue |
| 565 | if len(source) > MAX_AST_BYTES: |
| 566 | continue |
| 567 | if len(source) > MAX_AST_BYTES: |
| 568 | continue |
| 569 | try: |
| 570 | tree = ast.parse(source, filename=file_path) |
| 571 | except SyntaxError: |
| 572 | continue |
| 573 | |
| 574 | for node in ast.walk(tree): |
| 575 | if isinstance(node, ast.Import): |
| 576 | for alias in node.names: |
| 577 | target = module_to_file.get(alias.name) |
| 578 | if target and target != file_path: |
| 579 | graph[file_path].add(target) |
| 580 | elif isinstance(node, ast.ImportFrom) and node.module: |
| 581 | # Direct module match: `from foo.bar import baz` where foo.bar is a file. |
| 582 | target = module_to_file.get(node.module) |
| 583 | if target and target != file_path: |
| 584 | graph[file_path].add(target) |
| 585 | # Submodule match: `from pkg import submod` where pkg.submod is a file. |
| 586 | # This covers both `from src import b` (→ src.b → src/b.py) and |
| 587 | # `from src.cli import app` (→ src.cli.app → src/cli/app.py). |
| 588 | for alias in node.names: |
| 589 | sub = f"{node.module}.{alias.name}" |
| 590 | sub_target = module_to_file.get(sub) |
| 591 | if sub_target and sub_target != file_path: |
| 592 | graph[file_path].add(sub_target) |
| 593 | |
| 594 | return graph |
| 595 | |
| 596 | def _find_cycles(graph: ImportGraph) -> list[list[str]]: |
| 597 | """DFS cycle detection; returns list of cycles as file-path lists.""" |
| 598 | WHITE, GRAY, BLACK = 0, 1, 2 |
| 599 | color = {n: WHITE for n in graph} |
| 600 | stack: list[str] = [] |
| 601 | cycles: list[list[str]] = [] |
| 602 | |
| 603 | def dfs(node: str) -> None: |
| 604 | color[node] = GRAY |
| 605 | stack.append(node) |
| 606 | for neighbor in sorted(graph.get(node, set())): |
| 607 | if color.get(neighbor, BLACK) == WHITE: |
| 608 | dfs(neighbor) |
| 609 | elif color.get(neighbor, BLACK) == GRAY: |
| 610 | # Found a cycle — extract from stack. |
| 611 | idx = stack.index(neighbor) |
| 612 | cycle = stack[idx:] |
| 613 | # Deduplicate: only add if not already seen. |
| 614 | cycle_key = frozenset(cycle) |
| 615 | if not any(frozenset(c) == cycle_key for c in cycles): |
| 616 | cycles.append(list(cycle)) |
| 617 | stack.pop() |
| 618 | color[node] = BLACK |
| 619 | |
| 620 | for node in sorted(graph): |
| 621 | if color[node] == WHITE: |
| 622 | dfs(node) |
| 623 | |
| 624 | return cycles |
| 625 | |
| 626 | def check_no_circular_imports( |
| 627 | manifest: Manifest, |
| 628 | repo_root: pathlib.Path, |
| 629 | rule_name: str, |
| 630 | severity: InvariantSeverity, |
| 631 | *, |
| 632 | import_graph: ImportGraph | None = None, |
| 633 | ) -> list[CodeViolation]: |
| 634 | """Detect import cycles among Python files in the snapshot.""" |
| 635 | graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root) |
| 636 | cycles = _find_cycles(graph) |
| 637 | violations: list[CodeViolation] = [] |
| 638 | for cycle in cycles: |
| 639 | root_file = cycle[0] |
| 640 | cycle_str = " → ".join([*cycle, cycle[0]]) |
| 641 | violations.append(CodeViolation( |
| 642 | rule_name=rule_name, |
| 643 | severity=severity, |
| 644 | address=root_file, |
| 645 | description=f"Circular import cycle detected: {cycle_str}", |
| 646 | file=root_file, |
| 647 | symbol="", |
| 648 | detail=cycle_str, |
| 649 | )) |
| 650 | return violations |
| 651 | |
| 652 | def check_no_dead_exports( |
| 653 | manifest: Manifest, |
| 654 | repo_root: pathlib.Path, |
| 655 | rule_name: str, |
| 656 | severity: InvariantSeverity, |
| 657 | *, |
| 658 | file_data: FileDataMap | None = None, |
| 659 | ) -> list[CodeViolation]: |
| 660 | """Detect top-level functions/classes never imported by any other file. |
| 661 | |
| 662 | Exempt: test files, ``__init__.py``, files with ``__all__`` declarations |
| 663 | (which signal deliberate public API), and ``main`` functions. |
| 664 | |
| 665 | When *file_data* is supplied (built by :func:`_build_file_data`), no I/O |
| 666 | or ``ast.parse`` calls are made. |
| 667 | """ |
| 668 | violations: list[CodeViolation] = [] |
| 669 | |
| 670 | if file_data is not None: |
| 671 | # Fast path: derive imported names and top-level symbols from _FileData. |
| 672 | imported_names: set[str] = set() |
| 673 | for fd in file_data.values(): |
| 674 | imported_names.update(fd["from_name"]) |
| 675 | for mod in fd["raw_module_imports"]: |
| 676 | imported_names.add(mod.split(".")[-1]) |
| 677 | |
| 678 | for file_path, fd in file_data.items(): |
| 679 | if not file_path.endswith(".py"): |
| 680 | continue |
| 681 | base = pathlib.PurePosixPath(file_path).name |
| 682 | if base.startswith("test_") or base == "__init__.py": |
| 683 | continue |
| 684 | if fd["has_all"]: |
| 685 | continue |
| 686 | for fn_name in fd["top_level_fns"]: |
| 687 | if fn_name not in imported_names: |
| 688 | addr = f"{file_path}::{fn_name}" |
| 689 | violations.append(CodeViolation( |
| 690 | rule_name=rule_name, severity=severity, address=addr, |
| 691 | description=( |
| 692 | f"'{fn_name}' is never imported by any other file. " |
| 693 | "Consider removing, making private (prefix _), or adding to __all__." |
| 694 | ), |
| 695 | file=file_path, symbol=fn_name, detail="no importers found", |
| 696 | )) |
| 697 | for cls_name in fd["top_level_classes"]: |
| 698 | if cls_name not in imported_names: |
| 699 | addr = f"{file_path}::{cls_name}" |
| 700 | violations.append(CodeViolation( |
| 701 | rule_name=rule_name, severity=severity, address=addr, |
| 702 | description=( |
| 703 | f"Class '{cls_name}' is never imported by any other file. " |
| 704 | "Consider removing or making private." |
| 705 | ), |
| 706 | file=file_path, symbol=cls_name, detail="no importers found", |
| 707 | )) |
| 708 | return violations |
| 709 | |
| 710 | # Slow path: read and parse each file individually. |
| 711 | slow_imported: set[str] = set() |
| 712 | for file_path, content_hash in manifest.items(): |
| 713 | if not file_path.endswith(".py"): |
| 714 | continue |
| 715 | source = read_object(repo_root, content_hash) |
| 716 | if source is None: |
| 717 | continue |
| 718 | if len(source) > MAX_AST_BYTES: |
| 719 | continue |
| 720 | try: |
| 721 | tree = ast.parse(source, filename=file_path) |
| 722 | except SyntaxError: |
| 723 | continue |
| 724 | for node in ast.walk(tree): |
| 725 | if isinstance(node, ast.Import): |
| 726 | for alias in node.names: |
| 727 | slow_imported.add(alias.asname or alias.name.split(".")[-1]) |
| 728 | elif isinstance(node, ast.ImportFrom): |
| 729 | for alias in node.names: |
| 730 | slow_imported.add(alias.asname or alias.name) |
| 731 | |
| 732 | for file_path, content_hash in manifest.items(): |
| 733 | if not file_path.endswith(".py"): |
| 734 | continue |
| 735 | base = pathlib.PurePosixPath(file_path).name |
| 736 | if base.startswith("test_") or base == "__init__.py": |
| 737 | continue |
| 738 | source = read_object(repo_root, content_hash) |
| 739 | if source is None: |
| 740 | continue |
| 741 | if len(source) > MAX_AST_BYTES: |
| 742 | continue |
| 743 | try: |
| 744 | tree = ast.parse(source, filename=file_path) |
| 745 | except SyntaxError: |
| 746 | continue |
| 747 | has_all = any( |
| 748 | isinstance(n, ast.Assign) |
| 749 | and any(isinstance(t, ast.Name) and t.id == "__all__" for t in n.targets) |
| 750 | for n in ast.walk(tree) |
| 751 | ) |
| 752 | if has_all: |
| 753 | continue |
| 754 | for node in ast.iter_child_nodes(tree): |
| 755 | if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 756 | if node.name.startswith("_") or node.name == "main": |
| 757 | continue |
| 758 | if node.name not in slow_imported: |
| 759 | addr = f"{file_path}::{node.name}" |
| 760 | violations.append(CodeViolation( |
| 761 | rule_name=rule_name, severity=severity, address=addr, |
| 762 | description=( |
| 763 | f"'{node.name}' is never imported by any other file. " |
| 764 | "Consider removing, making private (prefix _), or adding to __all__." |
| 765 | ), |
| 766 | file=file_path, symbol=node.name, detail="no importers found", |
| 767 | )) |
| 768 | elif isinstance(node, ast.ClassDef): |
| 769 | if node.name.startswith("_"): |
| 770 | continue |
| 771 | if node.name not in slow_imported: |
| 772 | addr = f"{file_path}::{node.name}" |
| 773 | violations.append(CodeViolation( |
| 774 | rule_name=rule_name, severity=severity, address=addr, |
| 775 | description=( |
| 776 | f"Class '{node.name}' is never imported by any other file. " |
| 777 | "Consider removing or making private." |
| 778 | ), |
| 779 | file=file_path, symbol=node.name, detail="no importers found", |
| 780 | )) |
| 781 | |
| 782 | return violations |
| 783 | |
| 784 | def check_test_coverage_floor( |
| 785 | manifest: Manifest, |
| 786 | repo_root: pathlib.Path, |
| 787 | rule_name: str, |
| 788 | severity: InvariantSeverity, |
| 789 | *, |
| 790 | min_ratio: float = 0.30, |
| 791 | ) -> list[CodeViolation]: |
| 792 | """Require that at least *min_ratio* of functions have a test counterpart. |
| 793 | |
| 794 | A function ``foo`` is considered "tested" if any test file contains a |
| 795 | function named ``test_foo`` or a class method containing ``foo`` in its |
| 796 | name. This is a naming-convention heuristic, not true coverage. |
| 797 | """ |
| 798 | test_fn_names: set[str] = set() |
| 799 | all_fn_names: set[str] = set() |
| 800 | |
| 801 | for file_path, content_hash in manifest.items(): |
| 802 | if not file_path.endswith(".py"): |
| 803 | continue |
| 804 | source = read_object(repo_root, content_hash) |
| 805 | if source is None: |
| 806 | continue |
| 807 | if len(source) > MAX_AST_BYTES: |
| 808 | continue |
| 809 | try: |
| 810 | tree = ast.parse(source, filename=file_path) |
| 811 | except SyntaxError: |
| 812 | continue |
| 813 | |
| 814 | base = pathlib.PurePosixPath(file_path).name |
| 815 | is_test = base.startswith("test_") or base.endswith("_test.py") |
| 816 | |
| 817 | for node in ast.walk(tree): |
| 818 | if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 819 | if is_test and node.name.startswith("test_"): |
| 820 | test_fn_names.add(node.name.removeprefix("test_")) |
| 821 | elif not is_test and not node.name.startswith("_"): |
| 822 | all_fn_names.add(node.name) |
| 823 | |
| 824 | if not all_fn_names: |
| 825 | return [] |
| 826 | |
| 827 | covered = all_fn_names & test_fn_names |
| 828 | ratio = len(covered) / len(all_fn_names) |
| 829 | |
| 830 | if ratio < min_ratio: |
| 831 | pct_actual = round(ratio * 100, 1) |
| 832 | pct_required = round(min_ratio * 100, 1) |
| 833 | return [CodeViolation( |
| 834 | rule_name=rule_name, |
| 835 | severity=severity, |
| 836 | address="repo", |
| 837 | description=( |
| 838 | f"Test coverage floor not met: {pct_actual}% of functions have test counterparts " |
| 839 | f"(required {pct_required}%). Untested: " |
| 840 | f"{', '.join(sorted(all_fn_names - covered)[:10])}" |
| 841 | f"{'…' if len(all_fn_names - covered) > 10 else ''}" |
| 842 | ), |
| 843 | file="", |
| 844 | symbol="", |
| 845 | detail=f"ratio={ratio:.3f} required={min_ratio:.3f}", |
| 846 | )] |
| 847 | return [] |
| 848 | |
| 849 | # --------------------------------------------------------------------------- |
| 850 | # Rule dispatch |
| 851 | # --------------------------------------------------------------------------- |
| 852 | |
| 853 | def _dispatch_rule( |
| 854 | rule: CodeInvariantRule, |
| 855 | manifest: Manifest, |
| 856 | repo_root: pathlib.Path, |
| 857 | *, |
| 858 | file_data: FileDataMap | None = None, |
| 859 | import_graph: ImportGraph | None = None, |
| 860 | ) -> list[CodeViolation]: |
| 861 | """Dispatch a single rule to its implementation function. |
| 862 | |
| 863 | When *file_data* and *import_graph* are supplied, all rule implementations |
| 864 | use the fast (cached) code path — no I/O or ``ast.parse`` calls are made. |
| 865 | """ |
| 866 | params = rule.get("params", {}) |
| 867 | rule_name = rule["name"] |
| 868 | severity = rule["severity"] |
| 869 | rt = rule["rule_type"] |
| 870 | |
| 871 | if rt == "max_complexity": |
| 872 | threshold = int(params.get("threshold", 10)) |
| 873 | return check_max_complexity( |
| 874 | manifest, repo_root, rule_name, severity, |
| 875 | threshold=threshold, file_data=file_data, |
| 876 | ) |
| 877 | |
| 878 | if rt == "no_circular_imports": |
| 879 | return check_no_circular_imports( |
| 880 | manifest, repo_root, rule_name, severity, |
| 881 | import_graph=import_graph, |
| 882 | ) |
| 883 | |
| 884 | if rt == "no_dead_exports": |
| 885 | return check_no_dead_exports( |
| 886 | manifest, repo_root, rule_name, severity, |
| 887 | file_data=file_data, |
| 888 | ) |
| 889 | |
| 890 | if rt == "test_coverage_floor": |
| 891 | min_ratio = float(params.get("min_ratio", 0.30)) |
| 892 | return check_test_coverage_floor(manifest, repo_root, rule_name, severity, min_ratio=min_ratio) |
| 893 | |
| 894 | if rt == "forbidden_dependency": |
| 895 | return check_forbidden_dependency( |
| 896 | manifest, repo_root, rule_name, severity, |
| 897 | source_pattern=str(params.get("source_pattern", "")), |
| 898 | forbidden_pattern=str(params.get("forbidden_pattern", "")), |
| 899 | import_graph=import_graph, |
| 900 | ) |
| 901 | |
| 902 | if rt == "layer_boundary": |
| 903 | return check_layer_boundary( |
| 904 | manifest, repo_root, rule_name, severity, |
| 905 | lower=str(params.get("lower", "")), |
| 906 | upper=str(params.get("upper", "")), |
| 907 | import_graph=import_graph, |
| 908 | ) |
| 909 | |
| 910 | logger.warning("Unknown code invariant rule_type: %r — skipping", rt) |
| 911 | return [] |
| 912 | |
| 913 | # --------------------------------------------------------------------------- |
| 914 | # Public entry points |
| 915 | # --------------------------------------------------------------------------- |
| 916 | |
| 917 | def load_invariant_rules( |
| 918 | repo_root: pathlib.Path, |
| 919 | rules_file: pathlib.Path | None = None, |
| 920 | ) -> list[CodeInvariantRule]: |
| 921 | """Load code invariant rules from TOML, falling back to built-in defaults. |
| 922 | |
| 923 | When *rules_file* is ``None`` the default path (``.muse/code_invariants.toml`` |
| 924 | inside *repo_root*) is used; if that file is absent or empty, the built-in |
| 925 | defaults are returned. |
| 926 | |
| 927 | When *rules_file* is explicitly provided, the file is used as-is. An empty |
| 928 | file means "no rules" — the built-in defaults are NOT applied. This |
| 929 | allows callers to pass ``--rules empty.toml`` and receive an empty rule set. |
| 930 | |
| 931 | Args: |
| 932 | repo_root: Root of the Muse repository. |
| 933 | rules_file: Path to the TOML file. ``None`` → use default path. |
| 934 | |
| 935 | Returns: |
| 936 | List of :class:`CodeInvariantRule` dicts. |
| 937 | """ |
| 938 | path = rules_file or code_invariants_path(repo_root) |
| 939 | raw = load_rules_toml(path) |
| 940 | if not raw: |
| 941 | # Only fall back to defaults when using the implicit default path. |
| 942 | # An explicit file that is empty means "no rules". |
| 943 | if rules_file is None: |
| 944 | return list(_BUILTIN_DEFAULTS) |
| 945 | return [] |
| 946 | |
| 947 | rules: list[CodeInvariantRule] = [] |
| 948 | for r in raw: |
| 949 | name = str(r.get("name", "unnamed")) |
| 950 | raw_sev = str(r.get("severity", "warning")) |
| 951 | _SevMap = dict[str, InvariantSeverity] |
| 952 | _sev_map: _SevMap = {"info": "info", "warning": "warning", "error": "error"} |
| 953 | severity: InvariantSeverity = _sev_map.get(raw_sev, "warning") |
| 954 | scope_raw = str(r.get("scope", "function")) |
| 955 | _ScopeMap = dict[str, Literal["function", "file", "repo", "global"]] |
| 956 | _scope_map: _ScopeMap = { |
| 957 | "function": "function", "file": "file", "repo": "repo", "global": "global", |
| 958 | } |
| 959 | scope: Literal["function", "file", "repo", "global"] = _scope_map.get(scope_raw, "function") |
| 960 | rule_type = str(r.get("rule_type", "")) |
| 961 | raw_params = r.get("params", {}) |
| 962 | params: ParamsDict = ( |
| 963 | {k: v for k, v in raw_params.items()} |
| 964 | if isinstance(raw_params, dict) |
| 965 | else {} |
| 966 | ) |
| 967 | rule = CodeInvariantRule( |
| 968 | name=name, severity=severity, scope=scope, rule_type=rule_type, params=params |
| 969 | ) |
| 970 | rules.append(rule) |
| 971 | return rules |
| 972 | |
| 973 | def run_invariants( |
| 974 | repo_root: pathlib.Path, |
| 975 | commit_id: str, |
| 976 | rules: list[CodeInvariantRule], |
| 977 | ) -> BaseReport: |
| 978 | """Evaluate all rules against the snapshot of *commit_id*. |
| 979 | |
| 980 | Args: |
| 981 | repo_root: Repository root. |
| 982 | commit_id: Commit to check. |
| 983 | rules: Rules to evaluate (from :func:`load_invariant_rules`). |
| 984 | |
| 985 | Returns: |
| 986 | A :class:`~muse.core.invariants.BaseReport` with all violations. |
| 987 | """ |
| 988 | manifest = get_commit_snapshot_manifest(repo_root, commit_id) |
| 989 | if manifest is None: |
| 990 | logger.warning("Could not load snapshot for commit %s", commit_id) |
| 991 | return make_report(commit_id, "code", [], 0) |
| 992 | |
| 993 | mutable_manifest = dict(manifest) |
| 994 | |
| 995 | # Pre-flight: parse every Python file exactly once and cache the results. |
| 996 | # All rules that need AST data share this pre-parsed context — no file is |
| 997 | # read or parsed more than once per run_invariants call. |
| 998 | inv_cache = _InvariantFileCache.load(repo_root) |
| 999 | file_data = _build_file_data(mutable_manifest, repo_root, inv_cache) |
| 1000 | # Build the import graph once; shared by no_circular_imports, |
| 1001 | # forbidden_dependency, and layer_boundary rules. |
| 1002 | import_graph = _build_import_graph(mutable_manifest, repo_root, file_data=file_data) |
| 1003 | # Persist any newly parsed entries for future runs. |
| 1004 | inv_cache.save() |
| 1005 | |
| 1006 | all_violations: list[BaseViolation] = [] |
| 1007 | for rule in rules: |
| 1008 | try: |
| 1009 | code_violations = _dispatch_rule( |
| 1010 | rule, mutable_manifest, repo_root, |
| 1011 | file_data=file_data, |
| 1012 | import_graph=import_graph, |
| 1013 | ) |
| 1014 | for cv in code_violations: |
| 1015 | all_violations.append(BaseViolation( |
| 1016 | rule_name=cv["rule_name"], |
| 1017 | severity=cv["severity"], |
| 1018 | address=cv["address"], |
| 1019 | description=cv["description"], |
| 1020 | )) |
| 1021 | except Exception: |
| 1022 | logger.exception("Error evaluating rule %r on commit %s", rule["name"], commit_id) |
| 1023 | |
| 1024 | return make_report(commit_id, "code", all_violations, len(rules)) |
| 1025 | |
| 1026 | class CodeChecker: |
| 1027 | """Satisfies :class:`~muse.core.invariants.InvariantChecker` for the code domain.""" |
| 1028 | |
| 1029 | def check( |
| 1030 | self, |
| 1031 | repo_root: pathlib.Path, |
| 1032 | commit_id: str, |
| 1033 | *, |
| 1034 | rules_file: pathlib.Path | None = None, |
| 1035 | ) -> BaseReport: |
| 1036 | """Run code invariant checks against *commit_id*.""" |
| 1037 | rules = load_invariant_rules(repo_root, rules_file) |
| 1038 | return run_invariants(repo_root, commit_id, rules) |
| 1039 | |
| 1040 | def check_forbidden_dependency( |
| 1041 | manifest: Manifest, |
| 1042 | repo_root: pathlib.Path, |
| 1043 | rule_name: str, |
| 1044 | severity: InvariantSeverity, |
| 1045 | *, |
| 1046 | source_pattern: str = "", |
| 1047 | forbidden_pattern: str = "", |
| 1048 | import_graph: ImportGraph | None = None, |
| 1049 | ) -> list[CodeViolation]: |
| 1050 | """Detect files matching *source_pattern* that import from *forbidden_pattern*. |
| 1051 | |
| 1052 | Both patterns are simple substring matches against workspace-relative file |
| 1053 | paths. An empty pattern matches nothing, so a misconfigured rule with |
| 1054 | missing patterns produces no violations rather than matching everything. |
| 1055 | """ |
| 1056 | if not source_pattern or not forbidden_pattern: |
| 1057 | logger.warning( |
| 1058 | "Rule %r: forbidden_dependency requires source_pattern and " |
| 1059 | "forbidden_pattern params — skipping", |
| 1060 | rule_name, |
| 1061 | ) |
| 1062 | return [] |
| 1063 | |
| 1064 | graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root) |
| 1065 | violations: list[CodeViolation] = [] |
| 1066 | for file_path, deps in sorted(graph.items()): |
| 1067 | if source_pattern not in file_path: |
| 1068 | continue |
| 1069 | for dep in sorted(deps): |
| 1070 | if forbidden_pattern in dep: |
| 1071 | violations.append(CodeViolation( |
| 1072 | rule_name=rule_name, |
| 1073 | severity=severity, |
| 1074 | address=file_path, |
| 1075 | description=( |
| 1076 | f"'{file_path}' imports '{dep}' which matches " |
| 1077 | f"forbidden pattern '{forbidden_pattern}'" |
| 1078 | ), |
| 1079 | file=file_path, |
| 1080 | symbol="", |
| 1081 | detail=f"dep={dep}", |
| 1082 | )) |
| 1083 | return violations |
| 1084 | |
| 1085 | def check_layer_boundary( |
| 1086 | manifest: Manifest, |
| 1087 | repo_root: pathlib.Path, |
| 1088 | rule_name: str, |
| 1089 | severity: InvariantSeverity, |
| 1090 | *, |
| 1091 | lower: str = "", |
| 1092 | upper: str = "", |
| 1093 | import_graph: ImportGraph | None = None, |
| 1094 | ) -> list[CodeViolation]: |
| 1095 | """Detect lower-layer files that import from upper-layer files. |
| 1096 | |
| 1097 | The convention: *lower* (e.g. ``muse/core/``) must not import from |
| 1098 | *upper* (e.g. ``muse/cli/``). Both are simple substring matches against |
| 1099 | workspace-relative file paths. |
| 1100 | """ |
| 1101 | if not lower or not upper: |
| 1102 | logger.warning( |
| 1103 | "Rule %r: layer_boundary requires lower and upper params — skipping", |
| 1104 | rule_name, |
| 1105 | ) |
| 1106 | return [] |
| 1107 | |
| 1108 | graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root) |
| 1109 | violations: list[CodeViolation] = [] |
| 1110 | for file_path, deps in sorted(graph.items()): |
| 1111 | if lower not in file_path: |
| 1112 | continue |
| 1113 | for dep in sorted(deps): |
| 1114 | if upper in dep: |
| 1115 | violations.append(CodeViolation( |
| 1116 | rule_name=rule_name, |
| 1117 | severity=severity, |
| 1118 | address=file_path, |
| 1119 | description=( |
| 1120 | f"Lower-layer '{file_path}' imports upper-layer '{dep}' " |
| 1121 | f"('{lower}' must not import from '{upper}')" |
| 1122 | ), |
| 1123 | file=file_path, |
| 1124 | symbol="", |
| 1125 | detail=f"lower={lower} upper={upper} dep={dep}", |
| 1126 | )) |
| 1127 | return violations |
File History
1 commit
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839
chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm…
Sonnet 5
patch
13 days ago