| 31 | |
| 32 | |
| 33 | class WorkspaceEditor: |
| 34 | def __init__(self, root: Path, approvals: ApprovalTracker, auto_approve: bool) -> None: |
| 35 | self._root = root.resolve() |
| 36 | self._approvals = approvals |
| 37 | self._auto_approve = auto_approve or os.environ.get("APPLY_PATCH_AUTO_APPROVE") == "1" |
| 38 | |
| 39 | def create_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 40 | relative = self._relative_path(operation.path) |
| 41 | self._require_approval(operation, relative) |
| 42 | target = self._resolve(operation.path, ensure_parent=True) |
| 43 | diff = operation.diff or "" |
| 44 | content = apply_diff("", diff, mode="create") |
| 45 | target.write_text(content, encoding="utf-8") |
| 46 | return ApplyPatchResult(output=f"Created {relative}") |
| 47 | |
| 48 | def update_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 49 | relative = self._relative_path(operation.path) |
| 50 | self._require_approval(operation, relative) |
| 51 | target = self._resolve(operation.path) |
| 52 | original = target.read_text(encoding="utf-8") |
| 53 | diff = operation.diff or "" |
| 54 | patched = apply_diff(original, diff) |
| 55 | target.write_text(patched, encoding="utf-8") |
| 56 | return ApplyPatchResult(output=f"Updated {relative}") |
| 57 | |
| 58 | def delete_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 59 | relative = self._relative_path(operation.path) |
| 60 | self._require_approval(operation, relative) |
| 61 | target = self._resolve(operation.path) |
| 62 | target.unlink(missing_ok=True) |
| 63 | return ApplyPatchResult(output=f"Deleted {relative}") |
| 64 | |
| 65 | def _relative_path(self, value: str) -> str: |
| 66 | resolved = self._resolve(value) |
| 67 | return resolved.relative_to(self._root).as_posix() |
| 68 | |
| 69 | def _resolve(self, relative: str, ensure_parent: bool = False) -> Path: |
| 70 | candidate = Path(relative) |
| 71 | target = candidate if candidate.is_absolute() else (self._root / candidate) |
| 72 | target = target.resolve() |
| 73 | try: |
| 74 | target.relative_to(self._root) |
| 75 | except ValueError: |
| 76 | raise RuntimeError(f"Operation outside workspace: {relative}") from None |
| 77 | if ensure_parent: |
| 78 | target.parent.mkdir(parents=True, exist_ok=True) |
| 79 | return target |
| 80 | |
| 81 | def _require_approval(self, operation: ApplyPatchOperation, display_path: str) -> None: |
| 82 | fingerprint = self._approvals.fingerprint(operation, display_path) |
| 83 | if self._auto_approve or self._approvals.is_approved(fingerprint): |
| 84 | self._approvals.remember(fingerprint) |
| 85 | return |
| 86 | |
| 87 | print("\n[apply_patch] approval required") |
| 88 | print(f"- type: {operation.type}") |
| 89 | print(f"- path: {display_path}") |
| 90 | if operation.diff: |