gabriel / muse public
apply.py python
429 lines 13.8 KB
Raw
sha256:a317886dc0496c4af7b285b3e41c86c4c34ea2e79afc63b8829aadb1ada7903f chore: bump version to 0.2.0rc15 to match musehub#113 fix release Sonnet 4.6 patch 70 days ago
1 """``muse apply <patch-file>`` — apply .patch files to the working tree.
2
3 Reads unified diff patch files (as produced by ``muse format-patch``) and
4 applies them to the working tree. Supports mail-format patches with commit
5 metadata headers.
6
7 Flags
8 -----
9 ``--check``
10 Validate the patch can apply cleanly without modifying any files.
11 Exits 0 if the patch applies cleanly, non-zero if there are conflicts.
12
13 ``--staged``
14 After applying, stage the modified files (add them to the index).
15
16 ``--json``
17 Emit a JSON summary::
18
19 {"applied": ["a.py", "b.py"], "failed": ["c.py"]}
20
21 Exit codes::
22
23 0 — success (all hunks applied)
24 1 — user error: patch file not found, conflicts, path traversal
25 2 — not a Muse repository
26
27 Examples::
28
29 muse apply 0001-feat-add-login.patch
30 muse apply 0001-feat.patch --check
31 muse apply 0001-feat.patch --json
32 muse apply 0001-feat.patch --staged
33 """
34
35 import argparse
36 import json as _json
37 import logging
38 import pathlib
39 import re
40 import sys
41 from typing import Iterator, TypedDict
42
43 from muse.core.envelope import EnvelopeJson, make_envelope
44 from muse.core.errors import ExitCode
45 from muse.core.repo import require_repo
46 from muse.core.timing import start_timer
47 from muse.core.validation import contain_path, sanitize_display
48
49 logger = logging.getLogger(__name__)
50
51 class _PatchHunk(TypedDict, total=False):
52 old_start: int
53 context_before: list[str]
54 removes: list[str]
55 adds: list[str]
56 context_after: list[str]
57 _phase: str
58
59 class _FileDiff(TypedDict):
60 path: str | None
61 hunks: list[_PatchHunk]
62 is_new: bool
63 is_delete: bool
64
65 class _ApplyJson(EnvelopeJson):
66 """JSON output of ``muse apply --json``."""
67
68 applied: list[str]
69 failed: list[str]
70
71 # Matches unified diff file headers: --- a/path or +++ b/path
72 _FROM_RE = re.compile(r"^--- (?:a/)?(.+?)(?:\t.*)?$")
73 _TO_RE = re.compile(r"^\+\+\+ (?:b/)?(.+?)(?:\t.*)?$")
74 # Matches hunk headers: @@ -old_start[,old_count] +new_start[,new_count] @@
75 _HUNK_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@")
76 # Mail-format headers to skip
77 _MAIL_HEADERS = re.compile(r"^(From |Date: |Subject: |X-Muse-Commit-ID: |From: )")
78
79 # ---------------------------------------------------------------------------
80 # Internal helpers
81 # ---------------------------------------------------------------------------
82
83 def _strip_mail_headers(lines: list[str]) -> list[str]:
84 """Remove mail-format header block from the beginning of a patch."""
85 # Find the start of the actual diff content (--- line or diff --git line)
86 for i, line in enumerate(lines):
87 if line.startswith("--- ") or line.startswith("diff --"):
88 return lines[i:]
89 return lines
90
91 def _parse_patch(patch_text: str) -> list[_FileDiff]:
92 """Parse a unified diff patch string into a list of per-file diffs.
93
94 Args:
95 patch_text: Full patch file content (may include mail-format headers).
96
97 Returns:
98 List of dicts, each with keys:
99 - ``path``: target file path (str)
100 - ``hunks``: list of hunk dicts
101 - ``is_new``: True if this is a new file (no base)
102 - ``is_delete``: True if this deletes a file
103 """
104 lines = patch_text.splitlines(keepends=True)
105 lines = _strip_mail_headers(lines)
106
107 file_diffs: list[_FileDiff] = []
108 current: _FileDiff | None = None
109 current_hunk: _PatchHunk | None = None
110 i = 0
111
112 while i < len(lines):
113 line = lines[i]
114
115 # File header: --- a/path
116 if line.startswith("--- "):
117 m = _FROM_RE.match(line.rstrip("\n"))
118 from_path = m.group(1) if m else None
119 i += 1
120 # Next should be +++ b/path
121 if i < len(lines) and lines[i].startswith("+++ "):
122 m2 = _TO_RE.match(lines[i].rstrip("\n"))
123 to_path = m2.group(1) if m2 else None
124 i += 1
125 else:
126 to_path = from_path
127
128 # Flush previous file
129 if current is not None:
130 if current_hunk is not None:
131 current["hunks"].append(current_hunk)
132 current_hunk = None
133 file_diffs.append(current)
134
135 path = to_path if to_path and to_path != "/dev/null" else from_path
136 is_new = from_path == "/dev/null"
137 is_delete = to_path == "/dev/null"
138
139 current = {
140 "path": path,
141 "hunks": [],
142 "is_new": is_new,
143 "is_delete": is_delete,
144 }
145 continue
146
147 # Hunk header: @@ -old_start,old_count +new_start,new_count @@
148 if line.startswith("@@") and current is not None:
149 m = _HUNK_RE.match(line)
150 if m:
151 if current_hunk is not None:
152 current["hunks"].append(current_hunk)
153 old_start = int(m.group(1))
154 current_hunk = {
155 "old_start": old_start,
156 "context_before": [],
157 "removes": [],
158 "adds": [],
159 "context_after": [],
160 "_phase": "removes", # internal state
161 }
162 i += 1
163 continue
164
165 # Diff content lines
166 if current_hunk is not None:
167 stripped = line.rstrip("\n")
168 # Mail trailer "-- " or "--" signals end of patch content
169 if stripped in ("--", "-- "):
170 current["hunks"].append(current_hunk)
171 current_hunk = None
172 elif stripped.startswith("+"):
173 current_hunk["adds"].append(f"{stripped[1:]}\n")
174 current_hunk["_phase"] = "adds_seen"
175 elif stripped.startswith("-"):
176 current_hunk["removes"].append(f"{stripped[1:]}\n")
177 elif stripped.startswith(" "):
178 ctx_line = f"{stripped[1:]}\n"
179 if current_hunk["_phase"] == "adds_seen":
180 current_hunk["context_after"].append(ctx_line)
181 else:
182 current_hunk["context_before"].append(ctx_line)
183 elif stripped == "":
184 # Bare blank line: encoding artifact from lineterm="" — skip.
185 pass
186 elif stripped.startswith("\\"):
187 pass # "\ No newline at end of file"
188 else:
189 # Non-diff line — end of hunk
190 if current_hunk is not None:
191 current["hunks"].append(current_hunk)
192 current_hunk = None
193
194 i += 1
195
196 # Flush last hunk and file
197 if current is not None:
198 if current_hunk is not None:
199 current["hunks"].append(current_hunk)
200 file_diffs.append(current)
201
202 # Clean up internal state keys
203 for fd in file_diffs:
204 for h in fd["hunks"]:
205 h.pop("_phase", None)
206
207 return file_diffs
208
209 def _apply_hunk(lines: list[str], hunk: _PatchHunk) -> tuple[list[str], bool]:
210 """Apply a single unified diff hunk to a list of file lines.
211
212 Searches for the hunk's context + remove lines starting near ``old_start``,
213 then replaces them with the add lines.
214
215 Args:
216 lines: Current file lines (with newlines).
217 hunk: Hunk dict as returned by ``_parse_patch``.
218
219 Returns:
220 ``(result_lines, success)`` — modified lines and a success flag.
221 """
222 old_start = hunk["old_start"]
223 removes = hunk["removes"]
224 adds = hunk["adds"]
225 ctx_before = hunk["context_before"]
226 ctx_after = hunk["context_after"]
227
228 # Build the sequence of lines we expect to find (context + removes).
229 expected = list(ctx_before) + list(removes)
230
231 if not expected and not removes:
232 # Pure insertion — find insertion point at old_start
233 insert_at = min(old_start - 1, len(lines))
234 return lines[:insert_at] + list(adds) + lines[insert_at:], True
235
236 # Search for the expected lines near old_start (1-indexed → 0-indexed).
237 search_start = max(0, old_start - 1)
238
239 for offset in range(len(lines) + 1):
240 for direction in (0, 1, -1):
241 pos = search_start + direction * offset
242 if pos < 0 or pos + len(expected) > len(lines):
243 continue
244 if lines[pos : pos + len(expected)] == expected:
245 # Found it — replace removes with adds, keep context
246 result = (
247 lines[:pos]
248 + list(ctx_before)
249 + list(adds)
250 + lines[pos + len(expected):]
251 )
252 return result, True
253 if offset > max(10, len(lines)):
254 break
255
256 return lines, False
257
258 def _validate_path(root: pathlib.Path, path: str) -> pathlib.Path | None:
259 """Validate and resolve a patch path against the repo root.
260
261 Returns the absolute path if safe, or None if the path is unsafe.
262 """
263 # Reject paths that are absolute or contain traversal components
264 if ".." in pathlib.PurePosixPath(path).parts:
265 return None
266 if pathlib.Path(path).is_absolute():
267 return None
268
269 # Reject writes to .muse/ internals
270 parts = pathlib.PurePosixPath(path).parts
271 if parts and parts[0] == ".muse":
272 return None
273
274 try:
275 abs_path = contain_path(root, path)
276 return abs_path
277 except (ValueError, Exception):
278 return None
279
280 # ---------------------------------------------------------------------------
281 # Registration
282 # ---------------------------------------------------------------------------
283
284 def register(
285 subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]",
286 ) -> None:
287 """Register the ``muse apply`` subcommand."""
288 parser = subparsers.add_parser(
289 "apply",
290 help="Apply .patch files to the working tree.",
291 description=__doc__,
292 formatter_class=argparse.RawDescriptionHelpFormatter,
293 )
294 parser.add_argument(
295 "patch_file",
296 metavar="PATCH",
297 help="Path to the .patch file to apply.",
298 )
299 parser.add_argument(
300 "--check",
301 action="store_true",
302 help="Validate the patch without modifying files.",
303 )
304 parser.add_argument(
305 "--staged",
306 action="store_true",
307 help="Stage modified files after applying.",
308 )
309 parser.add_argument(
310 "--json", "-j",
311 action="store_true",
312 dest="json_out",
313 help="Emit a JSON summary of applied/failed files.",
314 )
315 parser.set_defaults(func=run)
316
317 # ---------------------------------------------------------------------------
318 # Run
319 # ---------------------------------------------------------------------------
320
321 def run(args: argparse.Namespace) -> None:
322 """Apply a unified diff patch file to the working tree.
323
324 Use ``--check`` to validate the patch without writing any files.
325 Generates patches with ``muse format-patch``.
326
327 Agent quickstart
328 ----------------
329 ::
330
331 muse apply my.patch --json
332 muse apply my.patch --check --json # dry-run: validate only
333
334 JSON fields
335 -----------
336 applied List of file paths successfully patched.
337 skipped List of file paths skipped (already applied or conflicts).
338 conflicts List of file paths with unresolvable conflicts.
339 check_only ``true`` when ``--check`` was passed (no writes occurred).
340
341 Exit codes
342 ----------
343 0 Patch applied successfully.
344 1 Patch conflicts or bad paths.
345 2 Not inside a Muse repository.
346 """
347 elapsed = start_timer()
348 patch_path = pathlib.Path(args.patch_file)
349 check_only: bool = args.check
350 json_out: bool = args.json_out
351
352 if not patch_path.exists():
353 print(f"❌ Patch file not found: {sanitize_display(str(patch_path))}", file=sys.stderr)
354 raise SystemExit(ExitCode.USER_ERROR)
355
356 try:
357 patch_text = patch_path.read_text(encoding="utf-8")
358 except OSError as exc:
359 print(f"❌ Cannot read patch file: {exc}", file=sys.stderr)
360 raise SystemExit(ExitCode.IO_ERROR)
361
362 root = require_repo()
363
364 file_diffs = _parse_patch(patch_text)
365
366 if not file_diffs:
367 print("❌ No file diffs found in patch.", file=sys.stderr)
368 raise SystemExit(ExitCode.USER_ERROR)
369
370 applied: list[str] = []
371 failed: list[str] = []
372
373 for fd in file_diffs:
374 raw_path = fd["path"]
375 abs_path = _validate_path(root, raw_path)
376
377 if abs_path is None:
378 print(
379 f"❌ Unsafe path in patch: {sanitize_display(raw_path)}",
380 file=sys.stderr,
381 )
382 failed.append(raw_path)
383 if not json_out:
384 raise SystemExit(ExitCode.USER_ERROR)
385 continue
386
387 # Load current file content
388 if abs_path.exists():
389 current_lines = abs_path.read_text(encoding="utf-8").splitlines(keepends=True)
390 else:
391 current_lines = []
392
393 # Apply all hunks in order
394 result_lines = list(current_lines)
395 all_ok = True
396 for hunk in fd["hunks"]:
397 result_lines, ok = _apply_hunk(result_lines, hunk)
398 if not ok:
399 all_ok = False
400 break
401
402 if not all_ok:
403 failed.append(raw_path)
404 if check_only and not json_out:
405 raise SystemExit(ExitCode.USER_ERROR)
406 continue
407
408 applied.append(raw_path)
409
410 if not check_only:
411 try:
412 abs_path.parent.mkdir(parents=True, exist_ok=True)
413 abs_path.write_text("".join(result_lines), encoding="utf-8")
414 except OSError as exc:
415 print(f"❌ Cannot write {sanitize_display(raw_path)}: {exc}", file=sys.stderr)
416 failed.append(raw_path)
417 applied.remove(raw_path)
418
419 # Output
420 if json_out:
421 print(_json.dumps(_ApplyJson(**make_envelope(elapsed), applied=applied, failed=failed)))
422 else:
423 for p in applied:
424 print(f"✓ {p}")
425 for p in failed:
426 print(f"✗ {p}")
427
428 if failed:
429 raise SystemExit(ExitCode.USER_ERROR)
File History 2 commits
sha256:a317886dc0496c4af7b285b3e41c86c4c34ea2e79afc63b8829aadb1ada7903f chore: bump version to 0.2.0rc15 to match musehub#113 fix release Sonnet 4.6 patch 70 days ago
sha256:f3b726b50f0aee3622bba751e0a67aa7ae4cf75a798477dbce581940b6a9cf70 feat: migrate invariants cache to .muse/cache/invariants.ms… Sonnet 4.6 patch 137 days ago