diff --git a/README.md b/README.md index 4405d30..f09850c 100644 --- a/README.md +++ b/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 списка и просмотра сессий diff --git a/serv/tools.py b/serv/tools.py index bd1c582..1eb2f48 100644 --- a/serv/tools.py +++ b/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)