diff --git a/docs/LOCAL_PUBLIC_TOOLS.md b/docs/LOCAL_PUBLIC_TOOLS.md index d6e8cb5d..cc3ea03e 100644 --- a/docs/LOCAL_PUBLIC_TOOLS.md +++ b/docs/LOCAL_PUBLIC_TOOLS.md @@ -9,6 +9,20 @@ This project exposes local atomic capabilities as shared `public` tools for all - `local_write_file` - `local_edit_file` +## Included P1/P2/P3 tools + +- `list_directory` +- `move_file` +- `delete_file` +- `mkdir` +- `search_files` +- `get_cwd` +- `cd` +- `get_env` +- `set_env` +- `list_processes` +- `kill_process` + ## Loading path - Files live under `runtime/tools/public/`. diff --git a/runtime/tools/local_sdk/adapter.py b/runtime/tools/local_sdk/adapter.py index fa32da16..76e62d99 100644 --- a/runtime/tools/local_sdk/adapter.py +++ b/runtime/tools/local_sdk/adapter.py @@ -1,8 +1,12 @@ from __future__ import annotations import os +import re +import shutil import subprocess from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path from typing import Any from oclaw.runtime.tools.experts.workspace.workspace_base import resolve_workspace_path, truncate_text @@ -22,6 +26,45 @@ class LocalAdapter: def __init__(self) -> None: self._init_error: LocalAdapterError | None = None + self._cwd: str = "." + + def get_cwd(self) -> dict[str, Any]: + try: + p = resolve_workspace_path(self._cwd or ".") + return {"ok": True, "cwd": str(p)} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def cd(self, *, cwd: str) -> dict[str, Any]: + raw = str(cwd or "").strip() + if not raw: + return {"ok": False, "error_code": "cwd_required", "error": "cwd_required"} + try: + p = resolve_workspace_path(raw) + if not p.exists() or not p.is_dir(): + return {"ok": False, "error_code": "not_a_directory", "error": "not_a_directory", "path": str(p)} + # Store as workspace-relative-ish string; resolve_workspace_path handles both. + self._cwd = str(p) + return {"ok": True, "cwd": str(p)} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def get_env(self, *, key: str, default: str | None = None) -> dict[str, Any]: + k = str(key or "").strip() + if not k: + return {"ok": False, "error_code": "key_required", "error": "key_required"} + val = os.environ.get(k, default) + return {"ok": True, "key": k, "value": val, "present": k in os.environ} + + def set_env(self, *, key: str, value: str | None = None) -> dict[str, Any]: + k = str(key or "").strip() + if not k: + return {"ok": False, "error_code": "key_required", "error": "key_required"} + if value is None: + os.environ.pop(k, None) + return {"ok": True, "key": k, "deleted": True} + os.environ[k] = str(value) + return {"ok": True, "key": k, "value": os.environ.get(k), "deleted": False} def run_command(self, *, command: str, cwd: str | None = None, timeout: int = 30) -> dict[str, Any]: cmd = str(command or "").strip() @@ -29,7 +72,7 @@ class LocalAdapter: return {"ok": False, "error_code": "command_required", "error": "command_required"} try: timeout_s = max(1, min(int(timeout or 30), 600)) - workdir = str(resolve_workspace_path(cwd or ".")) + workdir = str(resolve_workspace_path(cwd or (self._cwd or "."))) run_kwargs: dict[str, Any] = { "cwd": workdir, "shell": True, @@ -79,7 +122,7 @@ class LocalAdapter: if not target: return {"ok": False, "error_code": "path_required", "error": "path_required"} try: - p = resolve_workspace_path(target) + p = resolve_workspace_path(target if Path(target).is_absolute() else str(Path(self._cwd) / target)) if not p.exists() or not p.is_file(): return {"ok": False, "error_code": "file_not_found", "error": "file_not_found", "path": str(p)} lines = p.read_text(encoding="utf-8", errors="replace").splitlines() @@ -106,7 +149,7 @@ class LocalAdapter: if write_mode not in {"overwrite", "append"}: return {"ok": False, "error_code": "invalid_mode", "error": "invalid_mode"} try: - p = resolve_workspace_path(target) + p = resolve_workspace_path(target if Path(target).is_absolute() else str(Path(self._cwd) / target)) p.parent.mkdir(parents=True, exist_ok=True) if write_mode == "append" and p.exists(): merged = p.read_text(encoding="utf-8", errors="replace") + str(content or "") @@ -131,7 +174,7 @@ class LocalAdapter: if not target: return {"ok": False, "error_code": "path_required", "error": "path_required"} try: - p = resolve_workspace_path(target) + p = resolve_workspace_path(target if Path(target).is_absolute() else str(Path(self._cwd) / target)) if not p.exists() or not p.is_file(): return {"ok": False, "error_code": "file_not_found", "error": "file_not_found", "path": str(p)} text = p.read_text(encoding="utf-8", errors="replace") @@ -159,6 +202,193 @@ class LocalAdapter: except Exception as exc: return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + def mkdir(self, *, path: str, parents: bool = True, exist_ok: bool = True) -> dict[str, Any]: + raw = str(path or "").strip() + if not raw: + return {"ok": False, "error_code": "path_required", "error": "path_required"} + try: + p = resolve_workspace_path(raw if Path(raw).is_absolute() else str(Path(self._cwd) / raw)) + p.mkdir(parents=bool(parents), exist_ok=bool(exist_ok)) + return {"ok": True, "path": str(p)} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def delete_path(self, *, path: str, missing_ok: bool = True) -> dict[str, Any]: + raw = str(path or "").strip() + if not raw: + return {"ok": False, "error_code": "path_required", "error": "path_required"} + try: + p = resolve_workspace_path(raw if Path(raw).is_absolute() else str(Path(self._cwd) / raw)) + if not p.exists(): + return {"ok": bool(missing_ok), "path": str(p), "missing": True} + if p.is_dir(): + shutil.rmtree(p) + else: + p.unlink() + return {"ok": True, "path": str(p), "deleted": True} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def move_path(self, *, src: str, dst: str, overwrite: bool = False) -> dict[str, Any]: + s = str(src or "").strip() + d = str(dst or "").strip() + if not s or not d: + return {"ok": False, "error_code": "src_dst_required", "error": "src_dst_required"} + try: + p_src = resolve_workspace_path(s if Path(s).is_absolute() else str(Path(self._cwd) / s)) + p_dst = resolve_workspace_path(d if Path(d).is_absolute() else str(Path(self._cwd) / d)) + if not p_src.exists(): + return {"ok": False, "error_code": "src_not_found", "error": "src_not_found", "src": str(p_src)} + if p_dst.exists() and not overwrite: + return {"ok": False, "error_code": "dst_exists", "error": "dst_exists", "dst": str(p_dst)} + p_dst.parent.mkdir(parents=True, exist_ok=True) + if p_dst.exists() and overwrite: + if p_dst.is_dir(): + shutil.rmtree(p_dst) + else: + p_dst.unlink() + out = shutil.move(str(p_src), str(p_dst)) + return {"ok": True, "src": str(p_src), "dst": str(Path(out))} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def list_directory(self, *, path: str = ".", max_entries: int = 500) -> dict[str, Any]: + try: + raw = str(path or ".").strip() or "." + p = resolve_workspace_path(raw if Path(raw).is_absolute() else str(Path(self._cwd) / raw)) + if not p.exists() or not p.is_dir(): + return {"ok": False, "error_code": "not_a_directory", "error": "not_a_directory", "path": str(p)} + entries: list[dict[str, Any]] = [] + cap = max(1, min(int(max_entries or 500), 5000)) + for child in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())): + try: + st = child.stat() + except Exception: + continue + mtime = datetime.fromtimestamp(st.st_mtime, tz=timezone.utc).isoformat().replace("+00:00", "Z") + entries.append( + { + "name": child.name, + "path": str(child), + "is_dir": child.is_dir(), + "size": int(st.st_size), + "mtime_utc": mtime, + } + ) + if len(entries) >= cap: + break + return {"ok": True, "path": str(p), "count": len(entries), "entries": entries} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def search_files( + self, + *, + root: str = ".", + pattern: str, + file_glob: str = "**/*", + regex: bool = True, + max_matches: int = 200, + ) -> dict[str, Any]: + pat = str(pattern or "").strip() + if not pat: + return {"ok": False, "error_code": "pattern_required", "error": "pattern_required"} + try: + base_raw = str(root or ".").strip() or "." + base = resolve_workspace_path(base_raw if Path(base_raw).is_absolute() else str(Path(self._cwd) / base_raw)) + if not base.exists() or not base.is_dir(): + return {"ok": False, "error_code": "not_a_directory", "error": "not_a_directory", "path": str(base)} + rx = re.compile(pat) if regex else None + matches: list[dict[str, Any]] = [] + cap = max(1, min(int(max_matches or 200), 5000)) + fg = str(file_glob or "**/*").strip() or "**/*" + for p in base.glob(fg): + if p.is_dir(): + continue + try: + lines = p.read_text(encoding="utf-8", errors="replace").splitlines() + except Exception: + continue + for i, line in enumerate(lines, start=1): + ok = bool(rx.search(line)) if rx else (pat in line) + if ok: + matches.append({"file": str(p.relative_to(base)), "line": i, "text": line[:400]}) + if len(matches) >= cap: + return {"ok": True, "root": str(base), "count": len(matches), "matches": matches} + return {"ok": True, "root": str(base), "count": len(matches), "matches": matches} + except re.error as exc: + return {"ok": False, "error_code": "invalid_regex", "error": "invalid_regex", "detail": str(exc)} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def list_processes(self, *, max_results: int = 200) -> dict[str, Any]: + cap = max(1, min(int(max_results or 200), 2000)) + try: + rows: list[dict[str, Any]] = [] + if os.name == "nt": + cp = subprocess.run( + "tasklist /FO CSV /NH", + shell=True, + capture_output=True, + text=True, + timeout=10.0, + ) + out = str(cp.stdout or "") + for line in out.splitlines(): + line = line.strip() + if not line: + continue + # "Image Name","PID","Session Name","Session#","Mem Usage" + parts = [p.strip().strip('"') for p in line.split('","')] + if len(parts) < 2: + continue + name = parts[0].strip('"') + pid_raw = parts[1].strip('"') + if pid_raw.isdigit(): + rows.append({"pid": int(pid_raw), "name": name}) + if len(rows) >= cap: + break + else: + cp = subprocess.run( + "ps -eo pid=,comm=", + shell=True, + capture_output=True, + text=True, + timeout=10.0, + ) + for line in str(cp.stdout or "").splitlines(): + s = line.strip() + if not s: + continue + pid_str, _, name = s.partition(" ") + if pid_str.isdigit(): + rows.append({"pid": int(pid_str), "name": name.strip()}) + if len(rows) >= cap: + break + return {"ok": True, "count": len(rows), "processes": rows} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}"} + + def kill_process(self, *, pid: int, force: bool = True) -> dict[str, Any]: + try: + p = int(pid) + if p <= 0: + return {"ok": False, "error_code": "pid_invalid", "error": "pid_invalid"} + except Exception: + return {"ok": False, "error_code": "pid_invalid", "error": "pid_invalid"} + try: + if os.name == "nt": + cmd = f"taskkill /PID {p} " + ("/F" if force else "") + cp = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10.0) + ok = cp.returncode == 0 + return {"ok": ok, "pid": p, "stdout": str(cp.stdout or ""), "stderr": str(cp.stderr or ""), "exit_code": int(cp.returncode)} + import signal + + os.kill(p, signal.SIGKILL if force else signal.SIGTERM) + return {"ok": True, "pid": p} + except Exception as exc: + return {"ok": False, "error_code": "local_execution_error", "error": f"{type(exc).__name__}: {exc}", "pid": p} + _ADAPTER_SINGLETON: LocalAdapter | None = None diff --git a/runtime/tools/public/cd_tool.py b/runtime/tools/public/cd_tool.py new file mode 100644 index 00000000..a5c88660 --- /dev/null +++ b/runtime/tools/public/cd_tool.py @@ -0,0 +1,30 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def cd_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + cwd = str(args.get("cwd") or "").strip() + return get_local_adapter().cd(cwd=cwd) + + return ToolSpec( + name="cd", + description="Change current working directory for local tools (best-effort, per-process).", + parameters={ + "type": "object", + "properties": {"cwd": {"type": "string", "description": "Directory to change into."}}, + "required": ["cwd"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "write"}), + risk_level="high", + timeout_s=5.0, + ) + + +__all__ = ["cd_tool"] diff --git a/runtime/tools/public/delete_file_tool.py b/runtime/tools/public/delete_file_tool.py new file mode 100644 index 00000000..a6354264 --- /dev/null +++ b/runtime/tools/public/delete_file_tool.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def delete_file_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or "").strip() + missing_ok = bool(args.get("missing_ok", True)) + return get_local_adapter().delete_path(path=path, missing_ok=missing_ok) + + return ToolSpec( + name="delete_file", + description="Delete a file or directory inside the workspace.", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Path to delete."}, + "missing_ok": {"type": "boolean", "default": True, "description": "If true, missing path is ok."}, + }, + "required": ["path"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "workspace", "write"}), + risk_level="high", + timeout_s=20.0, + ) + + +__all__ = ["delete_file_tool"] diff --git a/runtime/tools/public/get_cwd_tool.py b/runtime/tools/public/get_cwd_tool.py new file mode 100644 index 00000000..33dce1e6 --- /dev/null +++ b/runtime/tools/public/get_cwd_tool.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def get_cwd_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + del args + return get_local_adapter().get_cwd() + + return ToolSpec( + name="get_cwd", + description="Get current working directory for local tools.", + parameters={"type": "object", "properties": {}, "additionalProperties": False}, + handler=_handler, + tags=frozenset({"public", "local", "read"}), + risk_level="low", + read_only=True, + timeout_s=2.0, + ) + + +__all__ = ["get_cwd_tool"] diff --git a/runtime/tools/public/get_env_tool.py b/runtime/tools/public/get_env_tool.py new file mode 100644 index 00000000..044e57b9 --- /dev/null +++ b/runtime/tools/public/get_env_tool.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def get_env_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + key = str(args.get("key") or "").strip() + default = args.get("default") + return get_local_adapter().get_env(key=key, default=str(default) if default is not None else None) + + return ToolSpec( + name="get_env", + description="Get an environment variable value.", + parameters={ + "type": "object", + "properties": { + "key": {"type": "string", "description": "Environment variable key."}, + "default": {"type": "string", "description": "Default value if missing."}, + }, + "required": ["key"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "read"}), + risk_level="low", + read_only=True, + timeout_s=2.0, + ) + + +__all__ = ["get_env_tool"] diff --git a/runtime/tools/public/kill_process_tool.py b/runtime/tools/public/kill_process_tool.py new file mode 100644 index 00000000..4f94e0e3 --- /dev/null +++ b/runtime/tools/public/kill_process_tool.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def kill_process_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + pid = int(args.get("pid") or 0) + force = bool(args.get("force", True)) + return get_local_adapter().kill_process(pid=pid, force=force) + + return ToolSpec( + name="kill_process", + description="Kill a process by PID (best-effort).", + parameters={ + "type": "object", + "properties": { + "pid": {"type": "integer", "description": "Process id."}, + "force": {"type": "boolean", "default": True, "description": "Force kill if true."}, + }, + "required": ["pid"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "write"}), + risk_level="high", + timeout_s=10.0, + ) + + +__all__ = ["kill_process_tool"] diff --git a/runtime/tools/public/list_directory_tool.py b/runtime/tools/public/list_directory_tool.py new file mode 100644 index 00000000..5a9ec6bf --- /dev/null +++ b/runtime/tools/public/list_directory_tool.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def list_directory_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or ".").strip() or "." + max_entries = int(args.get("max_entries") or 500) + return get_local_adapter().list_directory(path=path, max_entries=max_entries) + + return ToolSpec( + name="list_directory", + description="List a directory with file metadata (name/size/mtime).", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "default": ".", "description": "Directory path."}, + "max_entries": {"type": "integer", "default": 500, "description": "Max entries to return."}, + }, + "required": [], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "workspace", "read"}), + risk_level="low", + read_only=True, + timeout_s=10.0, + ) + + +__all__ = ["list_directory_tool"] diff --git a/runtime/tools/public/list_processes_tool.py b/runtime/tools/public/list_processes_tool.py new file mode 100644 index 00000000..ca5db1b8 --- /dev/null +++ b/runtime/tools/public/list_processes_tool.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def list_processes_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + max_results = int(args.get("max_results") or 200) + return get_local_adapter().list_processes(max_results=max_results) + + return ToolSpec( + name="list_processes", + description="List running processes (best-effort).", + parameters={ + "type": "object", + "properties": {"max_results": {"type": "integer", "default": 200}}, + "required": [], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "read"}), + risk_level="low", + read_only=True, + timeout_s=10.0, + ) + + +__all__ = ["list_processes_tool"] diff --git a/runtime/tools/public/mkdir_tool.py b/runtime/tools/public/mkdir_tool.py new file mode 100644 index 00000000..2fb51b03 --- /dev/null +++ b/runtime/tools/public/mkdir_tool.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def mkdir_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or "").strip() + parents = bool(args.get("parents", True)) + exist_ok = bool(args.get("exist_ok", True)) + return get_local_adapter().mkdir(path=path, parents=parents, exist_ok=exist_ok) + + return ToolSpec( + name="mkdir", + description="Create a directory inside the workspace.", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "Directory path to create."}, + "parents": {"type": "boolean", "default": True}, + "exist_ok": {"type": "boolean", "default": True}, + }, + "required": ["path"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "workspace", "write"}), + risk_level="high", + timeout_s=10.0, + ) + + +__all__ = ["mkdir_tool"] diff --git a/runtime/tools/public/move_file_tool.py b/runtime/tools/public/move_file_tool.py new file mode 100644 index 00000000..e1b0a0c3 --- /dev/null +++ b/runtime/tools/public/move_file_tool.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def move_file_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + src = str(args.get("src") or "").strip() + dst = str(args.get("dst") or "").strip() + overwrite = bool(args.get("overwrite", False)) + return get_local_adapter().move_path(src=src, dst=dst, overwrite=overwrite) + + return ToolSpec( + name="move_file", + description="Move/rename a file or directory inside the workspace.", + parameters={ + "type": "object", + "properties": { + "src": {"type": "string", "description": "Source path."}, + "dst": {"type": "string", "description": "Destination path."}, + "overwrite": {"type": "boolean", "default": False, "description": "Overwrite destination if exists."}, + }, + "required": ["src", "dst"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "workspace", "write"}), + risk_level="high", + timeout_s=20.0, + ) + + +__all__ = ["move_file_tool"] diff --git a/runtime/tools/public/search_files_tool.py b/runtime/tools/public/search_files_tool.py new file mode 100644 index 00000000..9f05bc8f --- /dev/null +++ b/runtime/tools/public/search_files_tool.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def search_files_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + pattern = str(args.get("pattern") or "").strip() + root = str(args.get("root") or ".").strip() or "." + file_glob = str(args.get("file_glob") or "**/*").strip() or "**/*" + regex = bool(args.get("regex", True)) + max_matches = int(args.get("max_matches") or 200) + return get_local_adapter().search_files( + root=root, + pattern=pattern, + file_glob=file_glob, + regex=regex, + max_matches=max_matches, + ) + + return ToolSpec( + name="search_files", + description="Search file contents under a directory (regex or substring).", + parameters={ + "type": "object", + "properties": { + "pattern": {"type": "string", "description": "Regex or substring pattern."}, + "root": {"type": "string", "default": ".", "description": "Root directory to search."}, + "file_glob": {"type": "string", "default": "**/*", "description": "File glob under root."}, + "regex": {"type": "boolean", "default": True, "description": "Treat pattern as regex if true."}, + "max_matches": {"type": "integer", "default": 200, "description": "Max matches to return."}, + }, + "required": ["pattern"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "workspace", "read"}), + risk_level="low", + read_only=True, + timeout_s=20.0, + ) + + +__all__ = ["search_files_tool"] diff --git a/runtime/tools/public/set_env_tool.py b/runtime/tools/public/set_env_tool.py new file mode 100644 index 00000000..79188096 --- /dev/null +++ b/runtime/tools/public/set_env_tool.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.local_sdk import get_local_adapter + + +def set_env_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + key = str(args.get("key") or "").strip() + value = args.get("value") + return get_local_adapter().set_env(key=key, value=str(value) if value is not None else None) + + return ToolSpec( + name="set_env", + description="Set or delete an environment variable in the current process.", + parameters={ + "type": "object", + "properties": { + "key": {"type": "string", "description": "Environment variable key."}, + "value": {"type": "string", "description": "Value to set. If omitted, deletes the key."}, + }, + "required": ["key"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "local", "write"}), + risk_level="high", + timeout_s=2.0, + ) + + +__all__ = ["set_env_tool"] diff --git a/tests/test_local_public_tools.py b/tests/test_local_public_tools.py index eac7356a..35dfdb37 100644 --- a/tests/test_local_public_tools.py +++ b/tests/test_local_public_tools.py @@ -10,15 +10,32 @@ from oclaw.runtime.tools.public.local_read_file_tool import local_read_file_tool from oclaw.runtime.tools.public.local_run_command_tool import local_run_command_tool from oclaw.runtime.tools.public.local_write_file_tool import local_write_file_tool from oclaw.runtime.tools.public_registry import clear_public_tool_cache +from oclaw.runtime.tools.public.list_directory_tool import list_directory_tool +from oclaw.runtime.tools.public.search_files_tool import search_files_tool +from oclaw.runtime.tools.public.get_cwd_tool import get_cwd_tool +from oclaw.runtime.tools.public.cd_tool import cd_tool +from oclaw.runtime.tools.public.get_env_tool import get_env_tool +from oclaw.runtime.tools.public.set_env_tool import set_env_tool def test_local_public_read_tool_visible_by_default() -> None: clear_public_tool_cache() names = [t.name for t in default_registry(expert="network_ops+memory", specialist="ops").list()] assert "local_read_file" in names + assert "list_directory" in names + assert "search_files" in names + assert "get_cwd" in names + assert "get_env" in names + assert "list_processes" in names assert "local_run_command" not in names assert "local_write_file" not in names assert "local_edit_file" not in names + assert "mkdir" not in names + assert "delete_file" not in names + assert "move_file" not in names + assert "cd" not in names + assert "set_env" not in names + assert "kill_process" not in names def test_local_public_high_risk_tools_visible_when_enabled(monkeypatch) -> None: @@ -26,9 +43,20 @@ def test_local_public_high_risk_tools_visible_when_enabled(monkeypatch) -> None: monkeypatch.setenv("AIA_PUBLIC_TOOLS_ALLOW_HIGH", "1") names = [t.name for t in default_registry(expert="network_ops+memory", specialist="ops").list()] assert "local_read_file" in names + assert "list_directory" in names + assert "search_files" in names + assert "get_cwd" in names + assert "get_env" in names + assert "list_processes" in names assert "local_run_command" in names assert "local_write_file" in names assert "local_edit_file" in names + assert "mkdir" in names + assert "delete_file" in names + assert "move_file" in names + assert "cd" in names + assert "set_env" in names + assert "kill_process" in names def test_local_adapter_backend_roundtrip(tmp_path: Path, monkeypatch) -> None: @@ -145,3 +173,33 @@ def test_local_tool_integration_roundtrip(monkeypatch) -> None: stdout = str(out_run.get("stdout") or "") assert "12345" in stdout + +def test_p1_p2_read_tools_smoke(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("OPS_WORKSPACE_ROOT", str(tmp_path)) + # list_directory + (tmp_path / "d").mkdir() + (tmp_path / "d" / "a.txt").write_text("hello", encoding="utf-8") + out_ls = list_directory_tool().handler({"path": "d", "max_entries": 50}) + assert out_ls.get("ok") is True + assert any(e.get("name") == "a.txt" for e in (out_ls.get("entries") or [])) + # search_files + out_s = search_files_tool().handler({"pattern": "hell", "root": "d", "regex": False}) + assert out_s.get("ok") is True + assert (out_s.get("count") or 0) >= 1 + # get_cwd / cd + out_cwd0 = get_cwd_tool().handler({}) + assert out_cwd0.get("ok") is True + out_cd = cd_tool().handler({"cwd": "d"}) + assert out_cd.get("ok") is True + out_cwd1 = get_cwd_tool().handler({}) + assert out_cwd1.get("ok") is True + assert str(out_cwd1.get("cwd") or "").replace("\\", "/").endswith("/d") + # get_env / set_env + out_get0 = get_env_tool().handler({"key": "LOCAL_PUBLIC_TOOLS_TEST_KEY", "default": "x"}) + assert out_get0.get("ok") is True + out_set = set_env_tool().handler({"key": "LOCAL_PUBLIC_TOOLS_TEST_KEY", "value": "y"}) + assert out_set.get("ok") is True + out_get1 = get_env_tool().handler({"key": "LOCAL_PUBLIC_TOOLS_TEST_KEY"}) + assert out_get1.get("ok") is True + assert out_get1.get("value") == "y" +