aaronrene / gabriel-muse public
check_attr.py python
399 lines 12.2 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
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 }))
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 33 days ago