Add unified diff patch tool

README.md

@@ -28,7 +28,7 @@ Qwen OAuth + OpenAI-compatible endpoint
 - хранение токенов в `~/.qwen/oauth_creds.json`
 - HTTP API сервера
 - агентный цикл с tool calling
-- инструменты: `list_files`, `glob_search`, `grep_text`, `stat_path`, `read_file`, `replace_in_file`, `write_file`, `make_directory`, `exec_command`
+- инструменты: `list_files`, `glob_search`, `grep_text`, `stat_path`, `read_file`, `apply_unified_diff`, `replace_in_file`, `write_file`, `make_directory`, `exec_command`
 - Telegram polling без внешних библиотек
 - JSON-хранилище сессий
 - API списка и просмотра сессий

serv/tools.py

@@ -5,6 +5,7 @@ import json
 import os
 import re
 import subprocess
+import tempfile
 from pathlib import Path
 from typing import Any, Callable
 
@@ -25,6 +26,7 @@ class ToolRegistry:
             "grep_text": self.grep_text,
             "stat_path": self.stat_path,
             "read_file": self.read_file,
+            "apply_unified_diff": self.apply_unified_diff,
             "replace_in_file": self.replace_in_file,
             "write_file": self.write_file,
             "make_directory": self.make_directory,
@@ -107,6 +109,21 @@ class ToolRegistry:
                     },
                 },
             },
+            {
+                "type": "function",
+                "function": {
+                    "name": "apply_unified_diff",
+                    "description": "Apply a unified diff patch inside the workspace using the system patch command.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "patch": {"type": "string"},
+                            "strip": {"type": "integer"},
+                        },
+                        "required": ["patch"],
+                    },
+                },
+            },
             {
                 "type": "function",
                 "function": {
@@ -201,7 +218,12 @@ class ToolRegistry:
 
     def requires_approval(self, tool_name: str) -> bool:
         policy = self.config.tool_policy
-        write_tools = {"replace_in_file", "write_file", "make_directory"}
+        write_tools = {
+            "apply_unified_diff",
+            "replace_in_file",
+            "write_file",
+            "make_directory",
+        }
         shell_tools = {"exec_command"}
         if policy == "ask-all":
             return True
@@ -351,6 +373,52 @@ class ToolRegistry:
             "replacements": count,
         }
 
+    def apply_unified_diff(self, arguments: dict[str, Any]) -> dict[str, Any]:
+        patch_text = arguments["patch"]
+        strip = int(arguments.get("strip", 0))
+        if not patch_text.strip():
+            raise ToolError("Patch is empty")
+        if "\x00" in patch_text:
+            raise ToolError("Patch contains NUL byte")
+
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            suffix=".patch",
+            delete=False,
+        ) as handle:
+            patch_file = Path(handle.name)
+            handle.write(patch_text)
+
+        try:
+            completed = subprocess.run(
+                ["patch", f"-p{strip}", "--forward", "--batch", "-i", str(patch_file)],
+                cwd=str(self.workspace_root),
+                capture_output=True,
+                text=True,
+                timeout=120,
+            )
+        except FileNotFoundError as exc:
+            raise ToolError("System command 'patch' is not available") from exc
+        finally:
+            try:
+                patch_file.unlink()
+            except OSError:
+                pass
+
+        if completed.returncode != 0:
+            raise ToolError(
+                "patch failed: "
+                + (completed.stderr.strip() or completed.stdout.strip() or "unknown error")
+            )
+
+        return {
+            "applied": True,
+            "strip": strip,
+            "stdout": completed.stdout[-self.config.max_command_output_bytes :],
+            "stderr": completed.stderr[-self.config.max_command_output_bytes :],
+        }
+
     def make_directory(self, arguments: dict[str, Any]) -> dict[str, Any]:
         target = self._resolve(arguments["path"])
         target.mkdir(parents=True, exist_ok=True)