Allow safe patch hunk relocation

tests/test_worktree_mutation_tools.py

@@ -43,6 +43,17 @@ def test_apply_patch_multiple_hunks(tmp_path: Path) -> None:
     assert (tmp_path / "repo" / "file.txt").read_text(encoding="utf-8") == "A\nb\nc\nD\ne\n"
 
 
+def test_apply_patch_relocates_matching_hunk_when_line_number_is_stale(tmp_path: Path) -> None:
+    tools = init_repo(tmp_path / "repo")
+    (tmp_path / "repo" / "file.txt").write_text("header\nalpha\nbeta\ngamma\n", encoding="utf-8")
+    commit_baseline(tmp_path / "repo")
+
+    result = tools.apply_patch("file.txt", "@@ -1,3 +1,3 @@\n alpha\n-beta\n+BETA\n gamma\n")
+
+    assert result.success
+    assert (tmp_path / "repo" / "file.txt").read_text(encoding="utf-8") == "header\nalpha\nBETA\ngamma\n"
+
+
 def test_apply_patch_context_mismatch_is_not_partial(tmp_path: Path) -> None:
     tools = init_repo(tmp_path / "repo")
     path = tmp_path / "repo" / "file.txt"

tools/runtime.py

@@ -170,16 +170,22 @@ class WorktreeTools:
             if not match:
                 raise ValueError(f"Invalid hunk header: {line.strip()}")
             old_start = int(match.group(1)) - 1
-            if old_start < source_index:
-                raise ValueError("Overlapping or out-of-order patch hunk")
-            output.extend(original[source_index:old_start])
-            source_index = old_start
             index += 1
-            hunks += 1
+            hunk_lines: list[str] = []
             while index < len(patch_lines) and not patch_lines[index].startswith("@@"):
-                hunk_line = patch_lines[index]
+                hunk_lines.append(patch_lines[index])
+                index += 1
+            old_lines = self._patch_old_lines(hunk_lines)
+            actual_start = self._locate_hunk(original, old_lines, old_start, source_index)
+            if actual_start is None:
+                raise ValueError("Patch context does not match target file")
+            if actual_start < source_index:
+                raise ValueError("Overlapping or out-of-order patch hunk")
+            output.extend(original[source_index:actual_start])
+            source_index = actual_start
+            hunks += 1
+            for hunk_line in hunk_lines:
                 if hunk_line.startswith(("---", "+++")):
-                    index += 1
                     continue
                 if hunk_line.startswith(" "):
                     expected = hunk_line[1:]
@@ -200,12 +206,44 @@ class WorktreeTools:
                     raise ValueError("Blank patch lines must be prefixed with context/add/remove marker")
                 else:
                     raise ValueError(f"Invalid hunk line: {hunk_line[:40]}")
-                index += 1
         if hunks == 0:
             raise ValueError("Patch contains no hunks")
         output.extend(original[source_index:])
         return output, hunks
 
+    def _patch_old_lines(self, hunk_lines: list[str]) -> list[str]:
+        old_lines: list[str] = []
+        for hunk_line in hunk_lines:
+            if hunk_line.startswith(("---", "+++", "+", "\\ No newline at end of file")):
+                continue
+            if hunk_line.startswith((" ", "-")):
+                old_lines.append(hunk_line[1:])
+            elif hunk_line.strip() == "":
+                raise ValueError("Blank patch lines must be prefixed with context/add/remove marker")
+            else:
+                raise ValueError(f"Invalid hunk line: {hunk_line[:40]}")
+        return old_lines
+
+    def _locate_hunk(self, original: list[str], old_lines: list[str], old_start: int, source_index: int) -> int | None:
+        if not old_lines:
+            return max(old_start, source_index) if old_start >= source_index else None
+        if self._matches_at(original, old_lines, old_start):
+            return old_start
+        window_start = max(source_index, old_start - 200)
+        window_end = min(len(original), old_start + 200)
+        for candidate in range(window_start, window_end + 1):
+            if self._matches_at(original, old_lines, candidate):
+                return candidate
+        for candidate in range(source_index, len(original) + 1):
+            if self._matches_at(original, old_lines, candidate):
+                return candidate
+        return None
+
+    def _matches_at(self, original: list[str], old_lines: list[str], start: int) -> bool:
+        if start < 0 or start + len(old_lines) > len(original):
+            return False
+        return original[start : start + len(old_lines)] == old_lines
+
     def run(self, command: list[str], timeout: int = 60) -> ToolResult:
         self._require(Capability.RUN_TESTS)
         env = os.environ.copy()