新增公共工具：批量写文件、工作区树、上下文搜索、多段替换

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"]

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"]

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"]

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"]