check_attr.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse plumbing check-attr — query merge-strategy attributes for paths. |
| 2 | |
| 3 | Reads ``.museattributes``, resolves the applicable rules for each supplied |
| 4 | path, and reports the strategy that would be applied per dimension. Useful |
| 5 | for verifying that attribute rules are wired up correctly before a merge, and |
| 6 | for scripting domain-aware merge drivers. |
| 7 | |
| 8 | Output (JSON, default):: |
| 9 | |
| 10 | { |
| 11 | "domain": "midi", |
| 12 | "rules_loaded": 3, |
| 13 | "results": [ |
| 14 | { |
| 15 | "path": "tracks/drums.mid", |
| 16 | "dimension": "*", |
| 17 | "strategy": "ours", |
| 18 | "rule": { |
| 19 | "path_pattern": "drums/*", |
| 20 | "dimension": "*", |
| 21 | "strategy": "ours", |
| 22 | "comment": "Drums always prefer ours.", |
| 23 | "priority": 10, |
| 24 | "source_index": 0 |
| 25 | } |
| 26 | }, |
| 27 | { |
| 28 | "path": "tracks/melody.mid", |
| 29 | "dimension": "*", |
| 30 | "strategy": "auto", |
| 31 | "rule": null |
| 32 | } |
| 33 | ] |
| 34 | } |
| 35 | |
| 36 | Text output (``--format text``):: |
| 37 | |
| 38 | tracks/drums.mid dimension=* strategy=ours (rule 0: drums/*) |
| 39 | tracks/melody.mid dimension=* strategy=auto (no matching rule) |
| 40 | |
| 41 | With ``--rules-only`` (emit the loaded rule list without testing paths):: |
| 42 | |
| 43 | { |
| 44 | "domain": "midi", |
| 45 | "rules_loaded": 3, |
| 46 | "dimension": "*", |
| 47 | "rules": [ |
| 48 | {"path_pattern": "drums/*", "dimension": "*", "strategy": "ours", ...} |
| 49 | ] |
| 50 | } |
| 51 | |
| 52 | Plumbing contract |
| 53 | ----------------- |
| 54 | |
| 55 | - Exit 0: attributes resolved and emitted (even when no rules match). |
| 56 | - Exit 1: bad ``--format`` value; missing path arguments when not using |
| 57 | ``--rules-only`` or ``--stdin``. |
| 58 | - Exit 3: I/O or TOML parse error reading ``.museattributes``. |
| 59 | |
| 60 | Strategies |
| 61 | ---------- |
| 62 | |
| 63 | ``ours`` |
| 64 | Conflict resolution keeps the current-branch version. |
| 65 | ``theirs`` |
| 66 | Conflict resolution keeps the incoming-branch version. |
| 67 | ``union`` |
| 68 | Conflict resolution takes the union of both sides (additive, e.g. note sets). |
| 69 | ``base`` |
| 70 | Conflict resolution falls back to the common ancestor. |
| 71 | ``auto`` |
| 72 | No rule matched; the merge engine selects the best strategy automatically. |
| 73 | ``manual`` |
| 74 | Conflict must be resolved manually; merge engine halts and surfaces the |
| 75 | conflict for human or agent inspection. |
| 76 | |
| 77 | Agent use |
| 78 | --------- |
| 79 | |
| 80 | Inspect active rules before a merge:: |
| 81 | |
| 82 | muse plumbing check-attr --rules-only --json |
| 83 | |
| 84 | Pipe paths from staging area:: |
| 85 | |
| 86 | muse plumbing check-attr --stdin < staged_paths.txt |
| 87 | |
| 88 | Query a specific dimension across many files:: |
| 89 | |
| 90 | muse plumbing check-attr tracks/drums.mid tracks/melody.mid --dimension notes |
| 91 | |
| 92 | Discover all rules that would fire for a path:: |
| 93 | |
| 94 | muse plumbing check-attr tracks/drums.mid --all-rules --json |
| 95 | """ |
| 96 | |
| 97 | from __future__ import annotations |
| 98 | |
| 99 | import argparse |
| 100 | import fnmatch |
| 101 | import json |
| 102 | import logging |
| 103 | import sys |
| 104 | from typing import TypedDict |
| 105 | |
| 106 | from muse.core.attributes import AttributeRule, load_attributes |
| 107 | from muse.core.errors import ExitCode |
| 108 | from muse.core.repo import require_repo |
| 109 | from muse.core.validation import sanitize_display, validate_workspace_path |
| 110 | from muse.plugins.registry import read_domain |
| 111 | |
| 112 | |
| 113 | type _PerPath = dict[str, list["_RuleDict"]] |
| 114 | logger = logging.getLogger(__name__) |
| 115 | |
| 116 | _FORMAT_CHOICES = ("json", "text") |
| 117 | |
| 118 | |
| 119 | class _RuleDict(TypedDict): |
| 120 | path_pattern: str |
| 121 | dimension: str |
| 122 | strategy: str |
| 123 | comment: str |
| 124 | priority: int |
| 125 | source_index: int |
| 126 | |
| 127 | |
| 128 | class _PathResult(TypedDict): |
| 129 | path: str |
| 130 | dimension: str |
| 131 | strategy: str |
| 132 | rule: _RuleDict | None |
| 133 | |
| 134 | |
| 135 | def _dim_match(rule: AttributeRule, dimension: str) -> bool: |
| 136 | """Return True when *rule* applies to *dimension*.""" |
| 137 | return rule.dimension == "*" or rule.dimension == dimension or dimension == "*" |
| 138 | |
| 139 | |
| 140 | def _resolve_with_rule( |
| 141 | rules: list[AttributeRule], |
| 142 | path: str, |
| 143 | dimension: str, |
| 144 | ) -> tuple[str, AttributeRule | None]: |
| 145 | """Single-pass resolution: return ``(strategy, first_matching_rule)``. |
| 146 | |
| 147 | Replaces the previous pattern of calling ``resolve_strategy`` then |
| 148 | ``_find_matching_rule`` separately, which iterated the rule list twice. |
| 149 | |
| 150 | Returns ``("auto", None)`` when no rule matches. |
| 151 | """ |
| 152 | for rule in rules: |
| 153 | if fnmatch.fnmatch(path, rule.path_pattern) and _dim_match(rule, dimension): |
| 154 | return rule.strategy, rule |
| 155 | return "auto", None |
| 156 | |
| 157 | |
| 158 | def _all_matching_rules( |
| 159 | rules: list[AttributeRule], |
| 160 | path: str, |
| 161 | dimension: str, |
| 162 | ) -> list[AttributeRule]: |
| 163 | """Return every rule that matches *path* and *dimension* (for ``--all-rules``).""" |
| 164 | return [ |
| 165 | rule for rule in rules |
| 166 | if fnmatch.fnmatch(path, rule.path_pattern) and _dim_match(rule, dimension) |
| 167 | ] |
| 168 | |
| 169 | |
| 170 | def _rule_to_dict(rule: AttributeRule) -> _RuleDict: |
| 171 | return { |
| 172 | "path_pattern": rule.path_pattern, |
| 173 | "dimension": rule.dimension, |
| 174 | "strategy": rule.strategy, |
| 175 | "comment": rule.comment, |
| 176 | "priority": rule.priority, |
| 177 | "source_index": rule.source_index, |
| 178 | } |
| 179 | |
| 180 | |
| 181 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 182 | """Register the check-attr subcommand.""" |
| 183 | parser = subparsers.add_parser( |
| 184 | "check-attr", |
| 185 | help="Query merge-strategy attributes for workspace paths.", |
| 186 | description=__doc__, |
| 187 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 188 | ) |
| 189 | parser.add_argument( |
| 190 | "paths", |
| 191 | nargs="*", |
| 192 | help=( |
| 193 | "Workspace-relative paths to check. " |
| 194 | "Required unless --stdin or --rules-only is used." |
| 195 | ), |
| 196 | ) |
| 197 | parser.add_argument( |
| 198 | "--stdin", |
| 199 | action="store_true", |
| 200 | dest="from_stdin", |
| 201 | help=( |
| 202 | "Read additional paths from stdin, one per line. " |
| 203 | "Blank lines and '#'-comments are skipped. " |
| 204 | "Combines with positional path arguments." |
| 205 | ), |
| 206 | ) |
| 207 | parser.add_argument( |
| 208 | "--dimension", "-d", |
| 209 | default="*", |
| 210 | dest="dimension", |
| 211 | metavar="DIMENSION", |
| 212 | help=( |
| 213 | "Domain dimension to query (e.g. 'notes', 'pitch_bend'). " |
| 214 | "Use '*' to match any dimension. (default: *)" |
| 215 | ), |
| 216 | ) |
| 217 | parser.add_argument( |
| 218 | "--format", "-f", |
| 219 | dest="fmt", |
| 220 | default="json", |
| 221 | metavar="FORMAT", |
| 222 | help="Output format: json or text. (default: json)", |
| 223 | ) |
| 224 | parser.add_argument( |
| 225 | "--json", action="store_const", const="json", dest="fmt", |
| 226 | help="Shorthand for --format json.", |
| 227 | ) |
| 228 | parser.add_argument( |
| 229 | "--all-rules", "-A", |
| 230 | action="store_true", |
| 231 | dest="all_rules", |
| 232 | help="For each path, list all matching rules (not just the first).", |
| 233 | ) |
| 234 | parser.add_argument( |
| 235 | "--rules-only", |
| 236 | action="store_true", |
| 237 | dest="rules_only", |
| 238 | help=( |
| 239 | "Emit the full loaded rule list without testing any paths. " |
| 240 | "No path arguments needed. " |
| 241 | "Useful for agents inspecting attribute configuration before staging." |
| 242 | ), |
| 243 | ) |
| 244 | parser.set_defaults(func=run) |
| 245 | |
| 246 | |
| 247 | def run(args: argparse.Namespace) -> None: |
| 248 | """Query merge-strategy attributes for one or more paths. |
| 249 | |
| 250 | Reads ``.museattributes`` from the repository root and reports the |
| 251 | strategy that would be applied to each path for the given dimension. |
| 252 | |
| 253 | Default mode performs a single rule-list pass per path (not two passes), |
| 254 | returning both the effective strategy and the first matching rule in O(N) |
| 255 | where N is the number of rules. |
| 256 | """ |
| 257 | fmt: str = args.fmt |
| 258 | cli_paths: list[str] = args.paths or [] |
| 259 | from_stdin: bool = args.from_stdin |
| 260 | dimension: str = args.dimension |
| 261 | all_rules_mode: bool = args.all_rules |
| 262 | rules_only: bool = args.rules_only |
| 263 | |
| 264 | if fmt not in _FORMAT_CHOICES: |
| 265 | print( |
| 266 | json.dumps( |
| 267 | {"error": f"Unknown format {fmt!r}. Valid: {', '.join(_FORMAT_CHOICES)}"} |
| 268 | ), |
| 269 | file=sys.stderr, |
| 270 | ) |
| 271 | raise SystemExit(ExitCode.USER_ERROR) |
| 272 | |
| 273 | root = require_repo() |
| 274 | domain = read_domain(root) |
| 275 | |
| 276 | try: |
| 277 | rules = load_attributes(root, domain=domain) |
| 278 | except ValueError as exc: |
| 279 | print(json.dumps({"error": str(exc)}), file=sys.stderr) |
| 280 | raise SystemExit(ExitCode.INTERNAL_ERROR) |
| 281 | |
| 282 | # --rules-only: emit loaded rule list without requiring path args. |
| 283 | if rules_only: |
| 284 | rule_dicts = [_rule_to_dict(r) for r in rules] |
| 285 | if fmt == "text": |
| 286 | if not rules: |
| 287 | print("(no rules)") |
| 288 | for rd in rule_dicts: |
| 289 | print( |
| 290 | f"{sanitize_display(rd['path_pattern'])} " |
| 291 | f"dimension={sanitize_display(rd['dimension'])} " |
| 292 | f"strategy={sanitize_display(rd['strategy'])}" |
| 293 | ) |
| 294 | else: |
| 295 | print(json.dumps({ |
| 296 | "domain": domain, |
| 297 | "rules_loaded": len(rules), |
| 298 | "dimension": dimension, |
| 299 | "rules": rule_dicts, |
| 300 | })) |
| 301 | return |
| 302 | |
| 303 | # Collect paths: positional args + optional stdin. |
| 304 | # Strip \r\n (not just \n) so CRLF-terminated input on Windows or from |
| 305 | # carriage-return-injecting agents does not embed \r in path strings. |
| 306 | all_paths: list[str] = list(cli_paths) |
| 307 | if from_stdin: |
| 308 | for line in sys.stdin: |
| 309 | stripped = line.rstrip("\r\n") |
| 310 | if stripped and not stripped.startswith("#"): |
| 311 | all_paths.append(stripped) |
| 312 | |
| 313 | if not all_paths: |
| 314 | print( |
| 315 | json.dumps({"error": "At least one path argument is required."}), |
| 316 | file=sys.stderr, |
| 317 | ) |
| 318 | raise SystemExit(ExitCode.USER_ERROR) |
| 319 | |
| 320 | # Validate paths: reject traversal sequences, null bytes, absolute paths, |
| 321 | # control characters, and excessively long values. |
| 322 | for p in all_paths: |
| 323 | try: |
| 324 | validate_workspace_path(p) |
| 325 | except ValueError as exc: |
| 326 | print( |
| 327 | json.dumps({"error": f"Invalid path {p!r}: {exc}"}), |
| 328 | file=sys.stderr, |
| 329 | ) |
| 330 | raise SystemExit(ExitCode.USER_ERROR) |
| 331 | |
| 332 | # --all-rules: every rule that fires for each path. |
| 333 | if all_rules_mode: |
| 334 | per_path: _PerPath = { |
| 335 | path: [_rule_to_dict(r) for r in _all_matching_rules(rules, path, dimension)] |
| 336 | for path in all_paths |
| 337 | } |
| 338 | |
| 339 | if fmt == "text": |
| 340 | for path, matched_rules in per_path.items(): |
| 341 | if not matched_rules: |
| 342 | print(f"{sanitize_display(path)} (no matching rules)") |
| 343 | else: |
| 344 | for rd in matched_rules: |
| 345 | print( |
| 346 | f"{sanitize_display(path)} " |
| 347 | f"dimension={sanitize_display(rd['dimension'])} " |
| 348 | f"strategy={sanitize_display(rd['strategy'])} " |
| 349 | f"(rule {rd['source_index']}: " |
| 350 | f"{sanitize_display(rd['path_pattern'])})" |
| 351 | ) |
| 352 | return |
| 353 | |
| 354 | print(json.dumps({ |
| 355 | "domain": domain, |
| 356 | "rules_loaded": len(rules), |
| 357 | "dimension": dimension, |
| 358 | "results": [ |
| 359 | {"path": path, "matching_rules": per_path[path]} |
| 360 | for path in all_paths |
| 361 | ], |
| 362 | })) |
| 363 | return |
| 364 | |
| 365 | # Default: first-match winner per path — single O(N) pass per path. |
| 366 | results: list[_PathResult] = [] |
| 367 | for path in all_paths: |
| 368 | strategy, matched_rule = _resolve_with_rule(rules, path, dimension) |
| 369 | results.append({ |
| 370 | "path": path, |
| 371 | "dimension": dimension, |
| 372 | "strategy": strategy, |
| 373 | "rule": _rule_to_dict(matched_rule) if matched_rule else None, |
| 374 | }) |
| 375 | |
| 376 | if fmt == "text": |
| 377 | for res in results: |
| 378 | rule_entry = res["rule"] |
| 379 | if rule_entry is not None: |
| 380 | rule_info = ( |
| 381 | f"(rule {rule_entry['source_index']}: " |
| 382 | f"{sanitize_display(rule_entry['path_pattern'])})" |
| 383 | ) |
| 384 | else: |
| 385 | rule_info = "(no matching rule)" |
| 386 | print( |
| 387 | f"{sanitize_display(res['path'])} " |
| 388 | f"dimension={sanitize_display(res['dimension'])} " |
| 389 | f"strategy={sanitize_display(res['strategy'])} " |
| 390 | f"{rule_info}" |
| 391 | ) |
| 392 | return |
| 393 | |
| 394 | print(json.dumps({ |
| 395 | "domain": domain, |
| 396 | "rules_loaded": len(rules), |
| 397 | "dimension": dimension, |
| 398 | "results": [dict(r) for r in results], |
| 399 | })) |