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