|
| 1 | +import argparse |
| 2 | +import asyncio |
| 3 | +import hashlib |
| 4 | +import os |
| 5 | +import tempfile |
| 6 | +from pathlib import Path |
| 7 | + |
| 8 | +from agents import Agent, ApplyPatchTool, ModelSettings, Runner, apply_diff, trace |
| 9 | +from agents.editor import ApplyPatchOperation, ApplyPatchResult |
| 10 | + |
| 11 | + |
| 12 | +class ApprovalTracker: |
| 13 | + def __init__(self) -> None: |
| 14 | + self._approved: set[str] = set() |
| 15 | + |
| 16 | + def fingerprint(self, operation: ApplyPatchOperation, relative_path: str) -> str: |
| 17 | + hasher = hashlib.sha256() |
| 18 | + hasher.update(operation.type.encode("utf-8")) |
| 19 | + hasher.update(b"\0") |
| 20 | + hasher.update(relative_path.encode("utf-8")) |
| 21 | + hasher.update(b"\0") |
| 22 | + hasher.update((operation.diff or "").encode("utf-8")) |
| 23 | + return hasher.hexdigest() |
| 24 | + |
| 25 | + def remember(self, fingerprint: str) -> None: |
| 26 | + self._approved.add(fingerprint) |
| 27 | + |
| 28 | + def is_approved(self, fingerprint: str) -> bool: |
| 29 | + return fingerprint in self._approved |
| 30 | + |
| 31 | + |
| 32 | +class WorkspaceEditor: |
| 33 | + def __init__(self, root: Path, approvals: ApprovalTracker, auto_approve: bool) -> None: |
| 34 | + self._root = root.resolve() |
| 35 | + self._approvals = approvals |
| 36 | + self._auto_approve = auto_approve or os.environ.get("APPLY_PATCH_AUTO_APPROVE") == "1" |
| 37 | + |
| 38 | + def create_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 39 | + relative = self._relative_path(operation.path) |
| 40 | + self._require_approval(operation, relative) |
| 41 | + target = self._resolve(operation.path, ensure_parent=True) |
| 42 | + diff = operation.diff or "" |
| 43 | + content = apply_diff("", diff, mode="create") |
| 44 | + target.write_text(content, encoding="utf-8") |
| 45 | + return ApplyPatchResult(output=f"Created {relative}") |
| 46 | + |
| 47 | + def update_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 48 | + relative = self._relative_path(operation.path) |
| 49 | + self._require_approval(operation, relative) |
| 50 | + target = self._resolve(operation.path) |
| 51 | + original = target.read_text(encoding="utf-8") |
| 52 | + diff = operation.diff or "" |
| 53 | + patched = apply_diff(original, diff) |
| 54 | + target.write_text(patched, encoding="utf-8") |
| 55 | + return ApplyPatchResult(output=f"Updated {relative}") |
| 56 | + |
| 57 | + def delete_file(self, operation: ApplyPatchOperation) -> ApplyPatchResult: |
| 58 | + relative = self._relative_path(operation.path) |
| 59 | + self._require_approval(operation, relative) |
| 60 | + target = self._resolve(operation.path) |
| 61 | + target.unlink(missing_ok=True) |
| 62 | + return ApplyPatchResult(output=f"Deleted {relative}") |
| 63 | + |
| 64 | + def _relative_path(self, value: str) -> str: |
| 65 | + resolved = self._resolve(value) |
| 66 | + return resolved.relative_to(self._root).as_posix() |
| 67 | + |
| 68 | + def _resolve(self, relative: str, ensure_parent: bool = False) -> Path: |
| 69 | + candidate = Path(relative) |
| 70 | + target = candidate if candidate.is_absolute() else (self._root / candidate) |
| 71 | + target = target.resolve() |
| 72 | + try: |
| 73 | + target.relative_to(self._root) |
| 74 | + except ValueError: |
| 75 | + raise RuntimeError(f"Operation outside workspace: {relative}") from None |
| 76 | + if ensure_parent: |
| 77 | + target.parent.mkdir(parents=True, exist_ok=True) |
| 78 | + return target |
| 79 | + |
| 80 | + def _require_approval(self, operation: ApplyPatchOperation, display_path: str) -> None: |
| 81 | + fingerprint = self._approvals.fingerprint(operation, display_path) |
| 82 | + if self._auto_approve or self._approvals.is_approved(fingerprint): |
| 83 | + self._approvals.remember(fingerprint) |
| 84 | + return |
| 85 | + |
| 86 | + print("\n[apply_patch] approval required") |
| 87 | + print(f"- type: {operation.type}") |
| 88 | + print(f"- path: {display_path}") |
| 89 | + if operation.diff: |
| 90 | + preview = operation.diff if len(operation.diff) < 400 else f"{operation.diff[:400]}…" |
| 91 | + print("- diff preview:\n", preview) |
| 92 | + answer = input("Proceed? [y/N] ").strip().lower() |
| 93 | + if answer not in {"y", "yes"}: |
| 94 | + raise RuntimeError("Apply patch operation rejected by user.") |
| 95 | + self._approvals.remember(fingerprint) |
| 96 | + |
| 97 | + |
| 98 | +async def main(auto_approve: bool, model: str) -> None: |
| 99 | + with trace("apply_patch_example"): |
| 100 | + with tempfile.TemporaryDirectory(prefix="apply-patch-example-") as workspace: |
| 101 | + workspace_path = Path(workspace).resolve() |
| 102 | + approvals = ApprovalTracker() |
| 103 | + editor = WorkspaceEditor(workspace_path, approvals, auto_approve) |
| 104 | + tool = ApplyPatchTool(editor=editor) |
| 105 | + previous_response_id: str | None = None |
| 106 | + |
| 107 | + agent = Agent( |
| 108 | + name="Patch Assistant", |
| 109 | + model=model, |
| 110 | + instructions=( |
| 111 | + f"You can edit files inside {workspace_path} using the apply_patch tool. " |
| 112 | + "When modifying an existing file, include the file contents between " |
| 113 | + "<BEGIN_FILES> and <END_FILES> in your prompt." |
| 114 | + ), |
| 115 | + tools=[tool], |
| 116 | + model_settings=ModelSettings(tool_choice="required"), |
| 117 | + ) |
| 118 | + |
| 119 | + print(f"[info] Workspace root: {workspace_path}") |
| 120 | + print(f"[info] Using model: {model}") |
| 121 | + print("[run] Creating tasks.md") |
| 122 | + result = await Runner.run( |
| 123 | + agent, |
| 124 | + "Create tasks.md with a shopping checklist of 5 entries.", |
| 125 | + previous_response_id=previous_response_id, |
| 126 | + ) |
| 127 | + previous_response_id = result.last_response_id |
| 128 | + print(f"[run] Final response #1:\n{result.final_output}\n") |
| 129 | + notes_path = workspace_path / "tasks.md" |
| 130 | + if not notes_path.exists(): |
| 131 | + raise RuntimeError(f"{notes_path} was not created by the apply_patch tool.") |
| 132 | + updated_notes = notes_path.read_text(encoding="utf-8") |
| 133 | + print("[file] tasks.md after creation:\n") |
| 134 | + print(updated_notes) |
| 135 | + |
| 136 | + prompt = ( |
| 137 | + "<BEGIN_FILES>\n" |
| 138 | + f"===== tasks.md\n{updated_notes}\n" |
| 139 | + "<END_FILES>\n" |
| 140 | + "Check off the last two items from the file." |
| 141 | + ) |
| 142 | + print("\n[run] Updating tasks.md") |
| 143 | + result2 = await Runner.run( |
| 144 | + agent, |
| 145 | + prompt, |
| 146 | + previous_response_id=previous_response_id, |
| 147 | + ) |
| 148 | + print(f"[run] Final response #2:\n{result2.final_output}\n") |
| 149 | + if not notes_path.exists(): |
| 150 | + raise RuntimeError("tasks.md vanished unexpectedly before the second read.") |
| 151 | + print("[file] Final tasks.md:\n") |
| 152 | + print(notes_path.read_text(encoding="utf-8")) |
| 153 | + |
| 154 | + |
| 155 | +if __name__ == "__main__": |
| 156 | + parser = argparse.ArgumentParser() |
| 157 | + parser.add_argument( |
| 158 | + "--auto-approve", |
| 159 | + action="store_true", |
| 160 | + default=False, |
| 161 | + help="Skip manual confirmations for apply_patch operations.", |
| 162 | + ) |
| 163 | + parser.add_argument( |
| 164 | + "--model", |
| 165 | + default="gpt-5.1", |
| 166 | + help="Model ID to use for the agent.", |
| 167 | + ) |
| 168 | + args = parser.parse_args() |
| 169 | + asyncio.run(main(args.auto_approve, args.model)) |
0 commit comments