From da2413afad850bc5480b32a1b132a167ad018db7 Mon Sep 17 00:00:00 2001 From: oliver Date: Fri, 1 May 2026 02:45:00 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=85=AC=E5=85=B1=E5=B7=A5?= =?UTF-8?q?=E5=85=B7=EF=BC=9A=E6=89=B9=E9=87=8F=E5=86=99=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E3=80=81=E5=B7=A5=E4=BD=9C=E5=8C=BA=E6=A0=91=E3=80=81=E4=B8=8A?= =?UTF-8?q?=E4=B8=8B=E6=96=87=E6=90=9C=E7=B4=A2=E3=80=81=E5=A4=9A=E6=AE=B5?= =?UTF-8?q?=E6=9B=BF=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- runtime/tools/public/bulk_apply_patch_tool.py | 98 +++++++++++++++++ .../tools/public/list_workspace_tree_tool.py | 83 ++++++++++++++ runtime/tools/public/multi_edit_file_tool.py | 93 ++++++++++++++++ .../tools/public/search_files_context_tool.py | 103 ++++++++++++++++++ 4 files changed, 377 insertions(+) create mode 100644 runtime/tools/public/bulk_apply_patch_tool.py create mode 100644 runtime/tools/public/list_workspace_tree_tool.py create mode 100644 runtime/tools/public/multi_edit_file_tool.py create mode 100644 runtime/tools/public/search_files_context_tool.py diff --git a/runtime/tools/public/bulk_apply_patch_tool.py b/runtime/tools/public/bulk_apply_patch_tool.py new file mode 100644 index 00000000..fe79dcce --- /dev/null +++ b/runtime/tools/public/bulk_apply_patch_tool.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.path_guard import resolve_workspace_path + + +def bulk_apply_patch_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + patches = args.get("patches") + if not isinstance(patches, list) or not patches: + return {"ok": False, "error": "patches_required"} + + results: list[dict[str, Any]] = [] + + for i, item in enumerate(patches): + if not isinstance(item, dict): + results.append({"index": i, "ok": False, "error": "invalid_item"}) + continue + + path_raw = str(item.get("path") or "").strip() + content = item.get("content") + expected_sha256 = item.get("expected_sha256") + + if not path_raw: + results.append({"index": i, "ok": False, "error": "path_required"}) + continue + if not isinstance(content, str): + results.append({"index": i, "ok": False, "error": "content_required"}) + continue + + try: + p = resolve_workspace_path(path_raw) + except Exception as exc: + results.append({"index": i, "ok": False, "error": "invalid_path", "detail": str(exc)}) + continue + + # Optional optimistic concurrency (match existing apply_patch semantics) + if expected_sha256: + import hashlib + + if p.exists() and p.is_file(): + cur = p.read_bytes() + cur_sha = hashlib.sha256(cur).hexdigest() + if cur_sha != expected_sha256: + results.append( + { + "index": i, + "ok": False, + "error": "sha256_mismatch", + "path": str(p), + "expected_sha256": expected_sha256, + "actual_sha256": cur_sha, + } + ) + continue + + try: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(content, encoding="utf-8") + results.append({"index": i, "ok": True, "path": str(p), "bytes": p.stat().st_size}) + except Exception as exc: + results.append({"index": i, "ok": False, "error": "write_failed", "path": str(p), "detail": str(exc)}) + + return {"ok": True, "results": results} + + return ToolSpec( + name="bulk_apply_patch", + description="Apply multiple file overwrites in one call (with optional sha256 preconditions).", + parameters={ + "type": "object", + "properties": { + "patches": { + "type": "array", + "items": { + "type": "object", + "properties": { + "path": {"type": "string"}, + "content": {"type": "string"}, + "expected_sha256": {"type": "string"}, + }, + "required": ["path", "content"], + "additionalProperties": False, + }, + } + }, + "required": ["patches"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace", "write"}), + read_only=False, + risk_level="high", + ) + + +__all__ = ["bulk_apply_patch_tool"] diff --git a/runtime/tools/public/list_workspace_tree_tool.py b/runtime/tools/public/list_workspace_tree_tool.py new file mode 100644 index 00000000..3341b26e --- /dev/null +++ b/runtime/tools/public/list_workspace_tree_tool.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.path_guard import resolve_workspace_path + + +def list_workspace_tree_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + root_raw = str(args.get("root") or ".").strip() + max_depth = int(args.get("max_depth") or 4) + max_entries = int(args.get("max_entries") or 300) + show_files = bool(args.get("show_files", True)) + if max_depth < 0: + max_depth = 0 + if max_entries <= 0: + max_entries = 1 + + try: + root = resolve_workspace_path(root_raw) + except Exception as exc: + return {"ok": False, "error": "invalid_root", "detail": str(exc)} + + entries: list[dict[str, Any]] = [] + + def walk(p: Path, depth: int) -> None: + if len(entries) >= max_entries or depth > max_depth: + return + try: + children = sorted(list(p.iterdir()), key=lambda x: (not x.is_dir(), x.name.lower())) + except Exception: + return + for child in children: + if len(entries) >= max_entries: + return + is_dir = child.is_dir() + if is_dir or show_files: + entries.append( + { + "path": str(child), + "name": child.name, + "type": "dir" if is_dir else "file", + "depth": depth, + "size": child.stat().st_size if child.exists() and child.is_file() else None, + } + ) + if is_dir: + walk(child, depth + 1) + + if root.exists(): + walk(root, 0) + + return { + "ok": True, + "root": str(root), + "max_depth": max_depth, + "max_entries": max_entries, + "entries": entries, + } + + return ToolSpec( + name="list_workspace_tree", + description="List a workspace tree with depth-limited entries.", + parameters={ + "type": "object", + "properties": { + "root": {"type": "string", "description": "Directory to start from.", "default": "."}, + "max_depth": {"type": "integer", "description": "Maximum recursion depth.", "default": 4}, + "max_entries": {"type": "integer", "description": "Maximum entries to return.", "default": 300}, + "show_files": {"type": "boolean", "description": "Whether to include files as well as directories.", "default": True}, + }, + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace"}), + read_only=True, + risk_level="low", + ) + + +__all__ = ["list_workspace_tree_tool"] diff --git a/runtime/tools/public/multi_edit_file_tool.py b/runtime/tools/public/multi_edit_file_tool.py new file mode 100644 index 00000000..f0c22837 --- /dev/null +++ b/runtime/tools/public/multi_edit_file_tool.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.path_guard import resolve_workspace_path + + +def multi_edit_file_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or "").strip() + replacements = args.get("replacements") or [] + if not path: + return {"ok": False, "error": "path_required"} + if not isinstance(replacements, list) or not replacements: + return {"ok": False, "error": "replacements_required"} + + try: + p = resolve_workspace_path(path) + except Exception as exc: + return {"ok": False, "error": "invalid_path", "detail": str(exc)} + + if not p.exists() or not p.is_file(): + return {"ok": False, "error": "file_not_found", "path": str(p)} + + original = p.read_text(encoding="utf-8", errors="replace") + text = original + applied: list[dict[str, Any]] = [] + + for item in replacements: + if not isinstance(item, dict): + return {"ok": False, "error": "invalid_replacement_item"} + search = str(item.get("search") or "") + replace = str(item.get("replace") or "") + count = int(item.get("count") or 0) + if not search: + return {"ok": False, "error": "search_required"} + occurrences = text.count(search) + if occurrences == 0: + applied.append({"search": search, "replaced": 0, "found": 0}) + continue + if count and count > 0: + text = text.replace(search, replace, count) + applied.append({"search": search, "replaced": min(count, occurrences), "found": occurrences}) + else: + text = text.replace(search, replace) + applied.append({"search": search, "replaced": occurrences, "found": occurrences}) + + if text != original: + p.write_text(text, encoding="utf-8") + + return { + "ok": True, + "path": str(p), + "changed": text != original, + "bytes": p.stat().st_size, + "applied": applied, + } + + return ToolSpec( + name="multi_edit_file", + description="Apply multiple search/replace edits to a single workspace file.", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Target file path."}, + "replacements": { + "type": "array", + "description": "List of replacement operations in order.", + "items": { + "type": "object", + "properties": { + "search": {"type": "string", "description": "Text to find."}, + "replace": {"type": "string", "description": "Replacement text."}, + "count": {"type": "integer", "description": "Optional max occurrences to replace.", "default": 0}, + }, + "required": ["search", "replace"], + "additionalProperties": False, + }, + }, + }, + "required": ["path", "replacements"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace", "write"}), + read_only=False, + risk_level="high", + ) + + +__all__ = ["multi_edit_file_tool"] diff --git a/runtime/tools/public/search_files_context_tool.py b/runtime/tools/public/search_files_context_tool.py new file mode 100644 index 00000000..a0af4903 --- /dev/null +++ b/runtime/tools/public/search_files_context_tool.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.path_guard import resolve_workspace_path, truncate_text + + +def search_files_context_tool() -> ToolSpec: + def _iter_files(root: Path, file_glob: str) -> list[Path]: + if root.is_file(): + return [root] + try: + return [p for p in root.rglob(file_glob) if p.is_file()] + except Exception: + return [] + + def _handler(args: dict[str, Any]) -> dict[str, Any]: + root_raw = str(args.get("root") or ".").strip() + file_glob = str(args.get("file_glob") or "**/*").strip() or "**/*" + pattern = str(args.get("pattern") or "").strip() + regex = bool(args.get("regex", True)) + max_matches = int(args.get("max_matches") or 200) + context_lines = int(args.get("context_lines") or 2) + if max_matches <= 0: + max_matches = 1 + if context_lines < 0: + context_lines = 0 + + try: + root = resolve_workspace_path(root_raw) + except Exception as exc: + return {"ok": False, "error": "invalid_root", "detail": str(exc)} + + if not pattern: + return {"ok": False, "error": "pattern_required"} + + matcher = re.compile(pattern) if regex else None + out: list[dict[str, Any]] = [] + scanned = 0 + + for fp in _iter_files(root, file_glob): + scanned += 1 + if len(out) >= max_matches: + break + try: + lines = fp.read_text(encoding="utf-8", errors="replace").splitlines() + except Exception: + continue + + for idx, line in enumerate(lines): + hit = matcher.search(line) if matcher else (pattern in line) + if not hit: + continue + start = max(0, idx - context_lines) + end = min(len(lines), idx + context_lines + 1) + out.append( + { + "path": str(fp), + "line": idx + 1, + "match": line, + "context": [f"{i + 1}|{lines[i]}" for i in range(start, end)], + } + ) + if len(out) >= max_matches: + break + + return { + "ok": True, + "root": str(root), + "file_glob": file_glob, + "pattern": pattern, + "regex": regex, + "scanned_files": scanned, + "matches": out, + } + + return ToolSpec( + name="search_files_context", + description="Search files under a root and return matched lines with surrounding context.", + parameters={ + "type": "object", + "properties": { + "root": {"type": "string", "description": "Directory to search under."}, + "file_glob": {"type": "string", "description": "File glob under root.", "default": "**/*"}, + "pattern": {"type": "string", "description": "Regex or substring pattern."}, + "regex": {"type": "boolean", "description": "Treat pattern as regex if true.", "default": True}, + "max_matches": {"type": "integer", "description": "Max matches to return.", "default": 200}, + "context_lines": {"type": "integer", "description": "Number of lines of context on each side.", "default": 2}, + }, + "required": ["pattern"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "search", "workspace"}), + read_only=True, + risk_level="low", + ) + + +__all__ = ["search_files_context_tool"]