"""muse check-attr — query merge-strategy attributes for paths. Reads ``.museattributes``, resolves the applicable rules for each supplied path, and reports the strategy that would be applied per dimension. Useful for verifying that attribute rules are wired up correctly before a merge, and for scripting domain-aware merge drivers. Output (JSON, default):: { "domain": "midi", "rules_loaded": 3, "dimension": "*", "summary": { "total": 2, "matched": 1, "unmatched": 1, "by_strategy": {"ours": 1, "auto": 1} }, "results": [ { "path": "tracks/drums.mid", "dimension": "*", "strategy": "ours", "rule": { "path_pattern": "drums/*", "dimension": "*", "strategy": "ours", "comment": "Drums always prefer ours.", "priority": 10, "source_index": 0 } }, { "path": "tracks/melody.mid", "dimension": "*", "strategy": "auto", "rule": null } ], "duration_ms": 0.000123, "exit_code": 0 } Text output (``--format text``):: tracks/drums.mid dimension=* strategy=ours (rule 0: drums/*) tracks/melody.mid dimension=* strategy=auto (no matching rule) With ``--rules-only`` (emit the loaded rule list without testing paths):: { "domain": "midi", "rules_loaded": 3, "dimension": "*", "rules": [ {"path_pattern": "drums/*", "dimension": "*", "strategy": "ours", ...} ], "duration_ms": 0.000042, "exit_code": 0 } Output contract --------------- - Exit 0: attributes resolved and emitted (even when no rules match). - Exit 1: bad ``--format`` value; missing path arguments when not using ``--rules-only`` or ``--stdin``; ``--unmatched-only`` combined with ``--rules-only``. - Exit 3: I/O or TOML parse error reading ``.museattributes``. JSON fields present in every successful response ------------------------------------------------ ``duration_ms`` Wall-clock time in seconds from argument parsing to output. Useful for agents monitoring attribute resolution latency as the rule set grows. ``exit_code`` Always ``0`` on success. Lets agents parse a single JSON payload instead of inspecting the process exit code separately. ``summary`` *(default mode only)* ``total`` — number of results emitted (after ``--unmatched-only`` filter). ``matched`` — paths where a rule fired (strategy ≠ ``"auto"``). ``unmatched`` — paths that fell through to the default ``"auto"`` strategy. ``by_strategy`` — ``{strategy: count}`` map for the emitted results. Strategies ---------- ``ours`` Conflict resolution keeps the current-branch version. ``theirs`` Conflict resolution keeps the incoming-branch version. ``union`` Conflict resolution takes the union of both sides (additive, e.g. note sets). ``base`` Conflict resolution falls back to the common ancestor. ``auto`` No rule matched; the merge engine selects the best strategy automatically. ``manual`` Conflict must be resolved manually; merge engine halts and surfaces the conflict for human or agent inspection. Agent use --------- Inspect active rules before a merge:: muse check-attr --rules-only --json Pipe paths from staging area:: muse check-attr --stdin < staged_paths.txt Query a specific dimension across many files:: muse check-attr tracks/drums.mid tracks/melody.mid --dimension notes Discover all rules that would fire for a path:: muse check-attr tracks/drums.mid --all-rules --json Find paths with no rule coverage (attribute gap analysis):: muse check-attr --unmatched-only --stdin < staged_paths.txt --json Check coverage summary without inspecting individual results:: muse check-attr foo.py bar.py baz.py --json | python3 -c \\ "import sys,json; s=json.load(sys.stdin)['summary']; print(s)" """ import argparse import fnmatch import json import logging import sys from typing import TypedDict from muse.core.attributes import AttributeRule, load_attributes from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.validation import sanitize_display, validate_workspace_path from muse.core.timing import start_timer from muse.core.envelope import EnvelopeJson, make_envelope from muse.plugins.registry import read_domain type _PerPath = dict[str, list["_RuleDict"]] logger = logging.getLogger(__name__) class _RuleDict(TypedDict): path_pattern: str dimension: str strategy: str comment: str priority: int source_index: int class _PathResult(TypedDict): path: str dimension: str strategy: str rule: _RuleDict | None class _SummaryDict(TypedDict): total: int matched: int unmatched: int by_strategy: dict[str, int] class _RulesListJson(EnvelopeJson): """Wire shape for --rules-only --json.""" domain: str rules_loaded: int dimension: str rules: list[_RuleDict] class _AllRulesJson(EnvelopeJson): """Wire shape for --all-rules --json.""" domain: str rules_loaded: int dimension: str results: list[dict] class _CheckAttrJson(EnvelopeJson): """Wire shape for default first-match --json output.""" domain: str rules_loaded: int dimension: str summary: _SummaryDict results: list[dict] def _dim_match(rule: AttributeRule, dimension: str) -> bool: """Return True when *rule* applies to *dimension*.""" return rule.dimension == "*" or rule.dimension == dimension or dimension == "*" def _resolve_with_rule( rules: list[AttributeRule], path: str, dimension: str, ) -> tuple[str, AttributeRule | None]: """Single-pass resolution: return ``(strategy, first_matching_rule)``. Replaces the previous pattern of calling ``resolve_strategy`` then ``_find_matching_rule`` separately, which iterated the rule list twice. Returns ``("auto", None)`` when no rule matches. """ for rule in rules: if fnmatch.fnmatch(path, rule.path_pattern) and _dim_match(rule, dimension): return rule.strategy, rule return "auto", None def _all_matching_rules( rules: list[AttributeRule], path: str, dimension: str, ) -> list[AttributeRule]: """Return every rule that matches *path* and *dimension* (for ``--all-rules``).""" return [ rule for rule in rules if fnmatch.fnmatch(path, rule.path_pattern) and _dim_match(rule, dimension) ] def _rule_to_dict(rule: AttributeRule) -> _RuleDict: return { "path_pattern": rule.path_pattern, "dimension": rule.dimension, "strategy": rule.strategy, "comment": rule.comment, "priority": rule.priority, "source_index": rule.source_index, } def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the check-attr subcommand.""" parser = subparsers.add_parser( "check-attr", help="Query merge-strategy attributes for workspace paths.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "paths", nargs="*", help=( "Workspace-relative paths to check. " "Required unless --stdin or --rules-only is used." ), ) parser.add_argument( "--stdin", action="store_true", dest="from_stdin", help=( "Read additional paths from stdin, one per line. " "Blank lines and '#'-comments are skipped. " "Combines with positional path arguments." ), ) parser.add_argument( "--dimension", "-d", default="*", dest="dimension", metavar="DIMENSION", help=( "Domain dimension to query (e.g. 'notes', 'pitch_bend'). " "Use '*' to match any dimension. (default: *)" ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON.", ) parser.add_argument( "--all-rules", "-A", action="store_true", dest="all_rules", help="For each path, list all matching rules (not just the first).", ) parser.add_argument( "--rules-only", action="store_true", dest="rules_only", help=( "Emit the full loaded rule list without testing any paths. " "No path arguments needed. " "Useful for agents inspecting attribute configuration before staging." ), ) parser.add_argument( "--unmatched-only", action="store_true", dest="unmatched_only", help=( "Show only paths that have no matching rule (strategy=auto). " "Useful for identifying gaps in attribute coverage before a merge." ), ) parser.set_defaults(func=run) def run(args: argparse.Namespace) -> None: """Query merge-strategy attributes for one or more paths. Reads ``.museattributes`` from the repository root and reports the effective merge strategy for each path in the requested dimension. Default mode returns the first matching rule per path in O(N); ``--all-rules`` returns every matching rule; ``--rules-only`` lists all loaded rules without requiring path arguments. Agent quickstart ---------------- :: muse check-attr tracks/drums.mid --json muse check-attr --rules-only --json muse check-attr foo.py bar.py --json JSON fields ----------- domain Repository domain (e.g. ``"code"``). rules_loaded Total number of rules loaded from ``.museattributes``. dimension Attribute dimension queried (e.g. ``"merge"``). results List of per-path result objects, each with: ``path``, ``dimension``, ``strategy``, ``rule`` (matched rule or ``null``), ``matching_rules`` (with ``--all-rules``). summary Summary counts: ``total``, ``matched``, ``unmatched``, ``by_strategy`` (only in default mode, not ``--rules-only``). rules List of all loaded rules (only with ``--rules-only``). Exit codes ---------- 0 Query completed successfully. 1 Conflicting flags or path validation error. 2 Not inside a Muse repository. 3 Internal error reading ``.museattributes``. """ elapsed = start_timer() json_out: bool = args.json_out cli_paths: list[str] = args.paths or [] from_stdin: bool = args.from_stdin dimension: str = args.dimension all_rules_mode: bool = args.all_rules rules_only: bool = args.rules_only unmatched_only: bool = getattr(args, "unmatched_only", False) if unmatched_only and rules_only: print( json.dumps({"error": "--unmatched-only and --rules-only are mutually exclusive."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() domain = read_domain(root) try: rules = load_attributes(root, domain=domain) except ValueError as exc: print(json.dumps({"error": str(exc)}), file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) # --rules-only: emit loaded rule list without requiring path args. if rules_only: rule_dicts = [_rule_to_dict(r) for r in rules] if not json_out: if not rules: print("(no rules)") for rd in rule_dicts: print( f"{sanitize_display(rd['path_pattern'])} " f"dimension={sanitize_display(rd['dimension'])} " f"strategy={sanitize_display(rd['strategy'])}" ) else: print(json.dumps(_RulesListJson( **make_envelope(elapsed), domain=domain, rules_loaded=len(rules), dimension=dimension, rules=rule_dicts, ))) return # Collect paths: positional args + optional stdin. # Strip \r\n (not just \n) so CRLF-terminated input on Windows or from # carriage-return-injecting agents does not embed \r in path strings. all_paths: list[str] = list(cli_paths) if from_stdin: for line in sys.stdin: stripped = line.rstrip("\r\n") if stripped and not stripped.startswith("#"): all_paths.append(stripped) if not all_paths: print( json.dumps({"error": "At least one path argument is required."}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Validate paths: reject traversal sequences, null bytes, absolute paths, # control characters, and excessively long values. for p in all_paths: try: validate_workspace_path(p) except ValueError as exc: print( json.dumps({"error": f"Invalid path {p!r}: {exc}"}), file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # --all-rules: every rule that fires for each path. if all_rules_mode: per_path: _PerPath = { path: [_rule_to_dict(r) for r in _all_matching_rules(rules, path, dimension)] for path in all_paths } if not json_out: for path, matched_rules in per_path.items(): if not matched_rules: print(f"{sanitize_display(path)} (no matching rules)") else: for rd in matched_rules: print( f"{sanitize_display(path)} " f"dimension={sanitize_display(rd['dimension'])} " f"strategy={sanitize_display(rd['strategy'])} " f"(rule {rd['source_index']}: " f"{sanitize_display(rd['path_pattern'])})" ) return print(json.dumps(_AllRulesJson( **make_envelope(elapsed), domain=domain, rules_loaded=len(rules), dimension=dimension, results=[ {"path": path, "matching_rules": per_path[path]} for path in all_paths ], ))) return # Default: first-match winner per path — single O(N) pass per path. results: list[_PathResult] = [] for path in all_paths: strategy, matched_rule = _resolve_with_rule(rules, path, dimension) results.append({ "path": path, "dimension": dimension, "strategy": strategy, "rule": _rule_to_dict(matched_rule) if matched_rule else None, }) # --unmatched-only: retain only paths that fell through to auto. if unmatched_only: results = [r for r in results if r["strategy"] == "auto"] # Build summary counts over the (possibly filtered) result set. by_strategy: dict[str, int] = {} for r in results: by_strategy[r["strategy"]] = by_strategy.get(r["strategy"], 0) + 1 summary: _SummaryDict = { "total": len(results), "matched": sum(1 for r in results if r["strategy"] != "auto"), "unmatched": sum(1 for r in results if r["strategy"] == "auto"), "by_strategy": by_strategy, } if not json_out: for res in results: rule_entry = res["rule"] if rule_entry is not None: rule_info = ( f"(rule {rule_entry['source_index']}: " f"{sanitize_display(rule_entry['path_pattern'])})" ) else: rule_info = "(no matching rule)" print( f"{sanitize_display(res['path'])} " f"dimension={sanitize_display(res['dimension'])} " f"strategy={sanitize_display(res['strategy'])} " f"{rule_info}" ) return print(json.dumps(_CheckAttrJson( **make_envelope(elapsed), domain=domain, rules_loaded=len(rules), dimension=dimension, summary=summary, results=[dict(r) for r in results], )))