diff --git a/docs/LOCAL_PUBLIC_TOOLS.md b/docs/LOCAL_PUBLIC_TOOLS.md index cc3ea03e..a2d92840 100644 --- a/docs/LOCAL_PUBLIC_TOOLS.md +++ b/docs/LOCAL_PUBLIC_TOOLS.md @@ -4,10 +4,10 @@ This project exposes local atomic capabilities as shared `public` tools for all ## Included P0 tools -- `local_run_command` -- `local_read_file` -- `local_write_file` -- `local_edit_file` +- `run_command` +- `read_file` +- `write_file` +- `edit_file` ## Included P1/P2/P3 tools @@ -31,8 +31,8 @@ This project exposes local atomic capabilities as shared `public` tools for all ## Risk gating -- `local_read_file` is `risk_level=low` and visible by default. -- `local_run_command`, `local_write_file`, `local_edit_file` are `risk_level=high`. +- `read_file` is `risk_level=low` and visible by default. +- `run_command`, `write_file`, `edit_file` are `risk_level=high`. - High-risk public tools are hidden unless `AIA_PUBLIC_TOOLS_ALLOW_HIGH=1`. ## Local backend adapter diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index 1dd442b4..bbc65e91 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -22,7 +22,7 @@ from oclaw.runtime.tools.base import ToolRegistry from oclaw.platform.llm.chat_models import LLMToolCall from oclaw.runtime.tools.tool_validation import validate_tool_arguments from oclaw.runtime.chat.media_redact import ingest_embedded_image_blobs_as_refs -from oclaw.runtime.tools.experts.workspace.workspace_base import ( +from oclaw.runtime.tools.path_guard import ( workspace_path_access_scope, workspace_write_namespace_scope, ) diff --git a/runtime/operations/hooks_cmd.py b/runtime/operations/hooks_cmd.py index 05057927..c793d0b4 100644 --- a/runtime/operations/hooks_cmd.py +++ b/runtime/operations/hooks_cmd.py @@ -33,7 +33,7 @@ from oclaw.runtime.hooks.user_config_hooks import ( ) from oclaw.runtime.hooks.workspace import load_workspace_hook_entries from oclaw.runtime.hooks_runtime import resolve_runtime_config -from oclaw.runtime.tools.experts.workspace.workspace_base import workspace_root +from oclaw.runtime.tools.path_guard import workspace_root def _resolve_cli_workspace(ns: argparse.Namespace) -> str: diff --git a/runtime/tools/experts/workspace/__init__.py b/runtime/tools/experts/workspace/__init__.py deleted file mode 100644 index a04cd9d0..00000000 --- a/runtime/tools/experts/workspace/__init__.py +++ /dev/null @@ -1,4 +0,0 @@ -from __future__ import annotations - -__all__ = [] - diff --git a/runtime/tools/experts/workspace/fs_tools.py b/runtime/tools/experts/workspace/fs_tools.py deleted file mode 100644 index 08f2edfb..00000000 --- a/runtime/tools/experts/workspace/fs_tools.py +++ /dev/null @@ -1,178 +0,0 @@ -from __future__ import annotations - -import hashlib -from pathlib import Path -from typing import Any - -from oclaw.runtime.tools.base import ToolSpec -from oclaw.runtime.tools.experts.workspace.workspace_base import ( - resolve_workspace_path, -) - - -def read_file_tool() -> ToolSpec: - def handler(args: dict[str, Any]) -> dict[str, Any]: - path = str(args.get("path") or "").strip() - offset = int(args.get("offset") or 1) - limit = int(args.get("limit") or 400) - if offset == 0: - offset = 1 - if limit <= 0: - limit = 1 - p = resolve_workspace_path(path) - if not p.exists() or not p.is_file(): - return {"ok": False, "error": "file_not_found", "path": str(p)} - text = p.read_text(encoding="utf-8", errors="replace").splitlines() - # 1-indexed offsets; negative counts from end - if offset < 0: - start = max(0, len(text) + offset) - else: - start = max(0, offset - 1) - end = min(len(text), start + min(limit, 2000)) - out_lines = [f"{i+1}|{text[i]}" for i in range(start, end)] - blob = p.read_bytes() - sha = hashlib.sha256(blob).hexdigest() - return { - "ok": True, - "path": str(p), - "start_line": start + 1, - "end_line": end, - "total_lines": len(text), - "sha256": sha, - "content": "\n".join(out_lines), - } - - return ToolSpec( - name="read_file", - description="Read a text file from the workspace with line numbers.", - parameters={ - "type": "object", - "properties": { - "path": {"type": "string", "description": "File path, relative to workspace root."}, - "offset": {"type": "integer", "description": "1-indexed start line; negative counts from end.", "default": 1}, - "limit": {"type": "integer", "description": "Max lines to return (capped).", "default": 400}, - }, - "required": ["path"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace"}), - read_only=True, - ) - - -def write_file_tool() -> ToolSpec: - def _sandbox_base_dir() -> Path: - return Path("data") / "workspace" - - def _normalize_write_path(path: str) -> str: - raw = str(path or "").strip().strip('"').strip("'") - if not raw: - raise ValueError("path_required") - p = Path(raw) - base = _sandbox_base_dir() - # Enforce sandbox for absolute paths as well. - if p.is_absolute(): - # Collapse absolute user path into sandbox-relative target to prevent - # writes to repo root or arbitrary host locations. - name = str(p.name or "").strip() - if not name: - raise ValueError("path_required") - return str(base / name) - # Keep generated files out of repo root: default relative writes go under data/workspace/... - rel = raw.lstrip("./\\") - if not rel: - raise ValueError("path_required") - return str(base / rel) - - def handler(args: dict[str, Any]) -> dict[str, Any]: - path = str(args.get("path") or "").strip() - content = str(args.get("content") or "") - mode = str(args.get("mode") or "overwrite").strip().lower() - try: - normalized = _normalize_write_path(path) - except ValueError as exc: - return {"ok": False, "error": str(exc)} - p = resolve_workspace_path(normalized) - p.parent.mkdir(parents=True, exist_ok=True) - if mode not in ("overwrite", "append"): - return {"ok": False, "error": "invalid_mode", "allowed": ["overwrite", "append"]} - if mode == "append": - p.write_text(p.read_text(encoding="utf-8", errors="replace") + content, encoding="utf-8") - else: - p.write_text(content, encoding="utf-8") - return {"ok": True, "path": str(p), "bytes": p.stat().st_size} - - return ToolSpec( - name="write_file", - description="Write text content to a workspace file (overwrite or append).", - parameters={ - "type": "object", - "properties": { - "path": {"type": "string", "description": "File path, relative to workspace root."}, - "content": {"type": "string", "description": "Full text content to write."}, - "mode": {"type": "string", "enum": ["overwrite", "append"], "default": "overwrite"}, - }, - "required": ["path", "content"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace", "write"}), - ) - - -def list_files_tool() -> ToolSpec: - def handler(args: dict[str, Any]) -> dict[str, Any]: - pattern = str(args.get("pattern") or "**/*").strip() or "**/*" - max_results = int(args.get("max_results") or 200) - root_arg = str(args.get("root") or "").strip() - if not root_arg: - base = resolve_workspace_path(".") - else: - base = resolve_workspace_path(root_arg) - if not base.is_dir(): - return {"ok": False, "error": "not_a_directory", "path": str(base)} - out: list[str] = [] - for p in base.glob(pattern): - if p.is_dir(): - continue - rel = str(p.relative_to(base)) - out.append(rel) - if len(out) >= max(1, min(max_results, 2000)): - break - return { - "ok": True, - "root": str(base), - "pattern": pattern, - "count": len(out), - "files": out, - } - - return ToolSpec( - name="glob", - description=( - "List files under a directory matching a glob pattern. " - "Default root is the workspace root; set `root` to an absolute path (e.g. D:\\\\download) when the user names a folder outside the repo — " - "this respects gateway workspace path policy. Prefer this over MCP filesystem list_directory when the user path may be outside MCP's configured roots." - ), - parameters={ - "type": "object", - "properties": { - "pattern": {"type": "string", "description": "Glob pattern relative to root, e.g. '**/*' or '*.pdf'.", "default": "**/*"}, - "root": { - "type": "string", - "description": "Optional directory to search under (absolute or workspace-relative). If omitted, uses workspace root.", - }, - "max_results": {"type": "integer", "default": 200, "description": "Max number of files to return."}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace"}), - read_only=True, - ) - - -__all__ = ["read_file_tool", "write_file_tool", "list_files_tool"] - diff --git a/runtime/tools/experts/workspace/search_tools.py b/runtime/tools/experts/workspace/search_tools.py deleted file mode 100644 index 8074868e..00000000 --- a/runtime/tools/experts/workspace/search_tools.py +++ /dev/null @@ -1,92 +0,0 @@ -from __future__ import annotations - -import re -from typing import Any - -from oclaw.runtime.tools.base import ToolSpec -from oclaw.runtime.tools.experts.workspace.workspace_base import resolve_workspace_path - - -def grep_tool() -> ToolSpec: - def handler(args: dict[str, Any]) -> dict[str, Any]: - pattern = str(args.get("pattern") or "").strip() - file_glob = str(args.get("file_glob") or "**/*").strip() or "**/*" - max_matches = int(args.get("max_matches") or 200) - if not pattern: - return {"ok": False, "error": "pattern_required"} - root = resolve_workspace_path(".") - try: - rx = re.compile(pattern) - except re.error as e: - return {"ok": False, "error": "invalid_regex", "detail": str(e)} - matches: list[dict[str, Any]] = [] - for p in root.glob(file_glob): - if p.is_dir(): - continue - try: - text = p.read_text(encoding="utf-8", errors="replace").splitlines() - except Exception: - continue - for i, line in enumerate(text, start=1): - if rx.search(line): - matches.append({"file": str(p.relative_to(root)), "line": i, "text": line[:400]}) - if len(matches) >= max(1, min(max_matches, 5000)): - return {"ok": True, "pattern": pattern, "count": len(matches), "matches": matches} - return {"ok": True, "pattern": pattern, "count": len(matches), "matches": matches} - - return ToolSpec( - name="grep", - description="Search files in the workspace for a regex pattern.", - parameters={ - "type": "object", - "properties": { - "pattern": {"type": "string", "description": "Regex pattern."}, - "file_glob": {"type": "string", "default": "**/*", "description": "Glob of files to search."}, - "max_matches": {"type": "integer", "default": 200, "description": "Max number of matches."}, - }, - "required": ["pattern"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace"}), - read_only=True, - ) - - -def index_workspace_tool() -> ToolSpec: - def handler(args: dict[str, Any]) -> dict[str, Any]: - max_files = int(args.get("max_files") or 120) - try: - # Lazy import to avoid heavy deps during tool discovery. - from oclaw.platform.config.paths import db_path - except Exception: - pass - # Indexer uses store passed via closure? ToolSpec doesn't carry store. - # We index using the global SqliteStore path (same as app runtime). - try: - from oclaw.platform.persistence.sqlite_store import SqliteStore - from oclaw.platform.config.paths import db_path - from oclaw.runtime.tools.workspace_indexer import index_workspace - - store = SqliteStore(db_path()) - st = index_workspace(store, max_files=max(1, min(max_files, 800))) - return {"ok": True, "files_seen": st.files_seen, "chunks_upserted": st.chunks_upserted, "embeddings_upserted": st.embeddings_upserted} - except Exception as e: - return {"ok": False, "error": f"{type(e).__name__}: {e}"} - - return ToolSpec( - name="index_workspace", - description="Index workspace files into the vector knowledge base for RAG (may be slow).", - parameters={ - "type": "object", - "properties": {"max_files": {"type": "integer", "default": 120, "description": "Max files to index."}}, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace", "rag"}), - ) - - -__all__ = ["grep_tool", "index_workspace_tool"] - diff --git a/runtime/tools/experts/workspace/shell_tools.py b/runtime/tools/experts/workspace/shell_tools.py deleted file mode 100644 index f5b32b68..00000000 --- a/runtime/tools/experts/workspace/shell_tools.py +++ /dev/null @@ -1,315 +0,0 @@ -from __future__ import annotations - -import subprocess -import re -from pathlib import Path -from typing import Any - -from oclaw.platform.config.paths import db_path -from oclaw.platform.persistence.sqlite_store import SqliteStore -from oclaw.runtime.tools.base import ToolSpec -from oclaw.runtime.tools.experts.workspace.workspace_base import ( - resolve_workspace_path, - truncate_text, - workspace_root, -) - - -def run_command_tool() -> ToolSpec: - _LEADING_CD_CHAIN_RE = re.compile( - r"^\s*(?:(?:[A-Za-z]:)\s*&&\s*)?(?:@echo\s+off\s*&&\s*)?cd\s+(?:/d\s+)?(?:\"[^\"]+\"|[^&]+?)\s*&&\s*(.+)$", - re.IGNORECASE | re.DOTALL, - ) - - def _run_command_enabled() -> bool: - import os - - try: - raw_setting = str(SqliteStore(db_path()).get_setting("AIA_ENABLE_RUN_COMMAND") or "").strip().lower() - if raw_setting in ("0", "false", "no", "off"): - return False - if raw_setting in ("1", "true", "yes", "on"): - return True - except Exception: - pass - - raw_env = str(os.getenv("AIA_ENABLE_RUN_COMMAND") or "").strip().lower() - if raw_env in ("0", "false", "no", "off"): - return False - if raw_env in ("1", "true", "yes", "on"): - return True - # Default disabled when unset (explicit opt-in only). - return False - - def handler(args: dict[str, Any]) -> dict[str, Any]: - import os - - def _default_exec_dir() -> str: - # Keep command execution in the same sandbox namespace as write_file. - try: - return str(resolve_workspace_path("data/workspace")) - except Exception: - return str(workspace_root()) - - def _strip_leading_cd_chain(cmd: str) -> tuple[str, bool]: - raw = str(cmd or "") - changed = False - out = raw - # Strip repeated leading "cd ... &&" so default sandbox cwd cannot be bypassed by habit. - for _ in range(3): - m = _LEADING_CD_CHAIN_RE.match(out) - if not m: - break - tail = str(m.group(1) or "").strip() - if not tail: - break - out = tail - changed = True - return out, changed - - def _rewrite_workspace_absolute_refs(cmd: str, *, workdir: str) -> tuple[str, bool]: - raw = str(cmd or "") - root = str(workspace_root()) - if not raw or not root: - return raw, False - root_norm = root.rstrip("\\/") - changed = False - out = raw - marker = root_norm + "\\" - if marker.lower() not in out.lower(): - return out, False - idx = 0 - rebuilt = [] - low = out.lower() - marker_low = marker.lower() - while True: - pos = low.find(marker_low, idx) - if pos < 0: - rebuilt.append(out[idx:]) - break - rebuilt.append(out[idx:pos]) - tail_start = pos + len(marker) - tail_end = tail_start - while tail_end < len(out) and out[tail_end] not in ('"', "'", " ", "\t", "\r", "\n"): - tail_end += 1 - rel_tail = out[tail_start:tail_end] - candidate = str(Path(workdir) / rel_tail) - if Path(candidate).exists(): - rebuilt.append(candidate) - changed = True - else: - rebuilt.append(out[pos:tail_end]) - idx = tail_end - return "".join(rebuilt), changed - - def _rewrite_python_script_arg(cmd: str, *, workdir: str) -> tuple[str, bool]: - raw = str(cmd or "").strip() - if not raw: - return raw, False - m = re.match(r'^\s*(python|py)\s+("([^"]+\.py)"|([^\s]+\.py))(\s+.*)?$', raw, flags=re.IGNORECASE) - if not m: - return raw, False - script = str(m.group(3) or m.group(4) or "").strip() - if not script: - return raw, False - # Absolute path is handled by workspace-absolute rewrite already. - sp = Path(script) - if sp.is_absolute(): - return raw, False - base = str(Path(script).name or "").strip() - if not base: - return raw, False - # Deterministic policy: always bind python script arg to sandbox root. - rel = base - quote = '"' if " " in rel else "" - prefix = str(m.group(1) or "python") - rest = str(m.group(5) or "") - return f"{prefix} {quote}{rel}{quote}{rest}", True - - if not _run_command_enabled(): - return { - "ok": False, - "error": "disabled", - "hint": "Enable run_command in Admin -> Plugins -> Tool Policy.", - } - command = str(args.get("command") or "").strip() - cwd = str(args.get("cwd") or "").strip() - timeout_s = float(args.get("timeout_s") or 30.0) - max_output_chars = int(args.get("max_output_chars") or 20000) - if not command: - return {"ok": False, "error": "command_required"} - normalized_cd_removed = False - command_rewritten = False - cwd_redirected_to_sandbox = False - script_path_rewritten = False - original_command = command - # Execute in caller-provided cwd (guarded by resolve_workspace_path), otherwise workspace root. - try: - workdir = str(resolve_workspace_path(cwd)) if cwd else _default_exec_dir() - except Exception: - # If caller path is rejected by workspace guard, fall back to workspace root. - workdir = _default_exec_dir() - if cwd: - cwd_redirected_to_sandbox = True - if cwd: - try: - requested_path = Path(cwd).expanduser().resolve() - if requested_path == workspace_root().resolve(): - # Explicit repo-root cwd still gets sandboxed to avoid writes/exec at repo root. - workdir = _default_exec_dir() - cwd_redirected_to_sandbox = True - except Exception: - pass - if cwd: - try: - requested = str(Path(cwd).expanduser().resolve()) - resolved = str(Path(workdir).expanduser().resolve()) - if requested != resolved: - cwd_redirected_to_sandbox = True - except Exception: - # Conservative signal: explicit cwd provided but could not preserve same path. - cwd_redirected_to_sandbox = True - command, normalized_cd_removed = _strip_leading_cd_chain(command) - command, command_rewritten = _rewrite_workspace_absolute_refs(command, workdir=workdir) - command, script_path_rewritten = _rewrite_python_script_arg(command, workdir=workdir) - - def _external_skill_install_cli_blocked(raw_cmd: str) -> bool: - s = str(raw_cmd or "").strip() - if not s: - return False - low = s.lower() - if re.match(r"^\s*cocoloop(?:\.cmd|\.exe)?\s+install(?:\s|$)", s, flags=re.IGNORECASE): - return True - if re.match(r"^\s*clawhub(?:\.cmd|\.exe)?\s+install(?:\s|$)", s, flags=re.IGNORECASE): - return True - if re.search(r"\bnpx\b", low) and "clawhub" in low: - return True - if re.match(r"^\s*npm(?:\.cmd|\.exe)?\s+install\b", low) and "clawhub" in low: - return True - return False - - if _external_skill_install_cli_blocked(str(command or "")): - return { - "ok": False, - "error_code": "skill_install_cli_blocked", - "error": "skill_install_cli_blocked", - "hint": "Oclaw has no shell skill installer. Use Admin POST /admin/api/skills/market/install or install-registry, or skill_auto_install.", - "command": command, - "cwd": str(workdir), - "normalized_cd_removed": bool(normalized_cd_removed), - "cwd_redirected_to_sandbox": bool(cwd_redirected_to_sandbox), - "command_rewritten": bool(command_rewritten), - "script_path_rewritten": bool(script_path_rewritten), - "original_command": original_command, - } - try: - os.makedirs(workdir, exist_ok=True) - except Exception: - pass - try: - run_kwargs: dict[str, Any] = { - "cwd": str(workdir), - "shell": True, - "capture_output": True, - "text": True, - "timeout": max(1.0, min(timeout_s, 600.0)), - } - if os.name == "nt": - startupinfo = subprocess.STARTUPINFO() - startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW - startupinfo.wShowWindow = 0 # SW_HIDE - run_kwargs["startupinfo"] = startupinfo - run_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW - cp = subprocess.run( - command, - **run_kwargs, - ) - stdout_text = str(cp.stdout or "") - stderr_text = str(cp.stderr or "") - out_raw = stdout_text + (("\n" + stderr_text) if stderr_text else "") - out_limit = max(1000, min(max_output_chars, 200000)) - out = truncate_text(out_raw, limit=out_limit) - out_truncated = len(out_raw) > out_limit - out_empty = (len(str(out_raw or "").strip()) == 0) - exit_code = int(cp.returncode) - ok_flag = exit_code == 0 - return { - "ok": bool(ok_flag), - "command": command, - "cwd": str(workdir), - "exit_code": exit_code, - "stdout": stdout_text, - "stderr": stderr_text, - "output": out, - "output_chars": int(len(out_raw)), - "output_empty": bool(out_empty), - "output_truncated": bool(out_truncated), - "output_not_truncated": bool(not out_truncated), - "normalized_cd_removed": bool(normalized_cd_removed), - "cwd_redirected_to_sandbox": bool(cwd_redirected_to_sandbox), - "command_rewritten": bool(command_rewritten), - "script_path_rewritten": bool(script_path_rewritten), - "original_command": original_command, - "error_code": ("" if ok_flag else "command_exit_nonzero"), - "output_hint": ( - "Command produced empty stdout/stderr; this is not system truncation. " - "Do not claim truncation unless output_truncated=true." - if out_empty - else "" - ), - } - except subprocess.TimeoutExpired as e: - partial = "" - try: - partial = ((e.stdout or "") + ("\n" + (e.stderr or "") if e.stderr else "")).strip() - except Exception: - partial = "" - return { - "ok": False, - "error": "timeout", - "command": command, - "cwd": str(workdir), - "timeout_s": timeout_s, - "output": truncate_text(partial, limit=max_output_chars), - "output_empty": len(str(partial or "").strip()) == 0, - "output_truncated": len(str(partial or "")) > int(max_output_chars or 0), - "normalized_cd_removed": bool(normalized_cd_removed), - "cwd_redirected_to_sandbox": bool(cwd_redirected_to_sandbox), - "command_rewritten": bool(command_rewritten), - "script_path_rewritten": bool(script_path_rewritten), - "original_command": original_command, - } - except Exception as e: - return { - "ok": False, - "error": f"{type(e).__name__}: {e}", - "command": command, - "cwd": str(workdir), - "normalized_cd_removed": bool(normalized_cd_removed), - "cwd_redirected_to_sandbox": bool(cwd_redirected_to_sandbox), - "command_rewritten": bool(command_rewritten), - "script_path_rewritten": bool(script_path_rewritten), - "original_command": original_command, - } - - return ToolSpec( - name="run_command", - description="Run a shell command inside the workspace (captured output, timeout).", - parameters={ - "type": "object", - "properties": { - "command": {"type": "string", "description": "Shell command to run."}, - "cwd": {"type": "string", "description": "Working directory relative to workspace.", "default": "."}, - "timeout_s": {"type": "number", "default": 30.0, "description": "Command timeout in seconds."}, - "max_output_chars": {"type": "integer", "default": 20000, "description": "Max characters to return."}, - }, - "required": ["command"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"workspace", "exec"}), - ) - - -__all__ = ["run_command_tool"] - diff --git a/runtime/tools/local_sdk/adapter.py b/runtime/tools/local_sdk/adapter.py index 76e62d99..5561bc53 100644 --- a/runtime/tools/local_sdk/adapter.py +++ b/runtime/tools/local_sdk/adapter.py @@ -9,7 +9,7 @@ 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 +from oclaw.runtime.tools.path_guard import resolve_workspace_path, truncate_text @dataclass(frozen=True) @@ -72,7 +72,9 @@ 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 (self._cwd or "."))) + # run_command never follows adapter cd state. + # It only uses explicit cwd; otherwise defaults to workspace root ("."). + workdir = str(resolve_workspace_path(cwd or ".")) run_kwargs: dict[str, Any] = { "cwd": workdir, "shell": True, diff --git a/runtime/tools/experts/workspace/workspace_base.py b/runtime/tools/path_guard.py similarity index 84% rename from runtime/tools/experts/workspace/workspace_base.py rename to runtime/tools/path_guard.py index 4df3f996..268af312 100644 --- a/runtime/tools/experts/workspace/workspace_base.py +++ b/runtime/tools/path_guard.py @@ -18,8 +18,6 @@ def _env_truthy(name: str) -> bool: def workspace_root() -> Path: - # Allow explicit override (recommended when running as a packaged app) - # Prefer legacy OPS_* overrides when explicitly provided (tests + backwards compatibility). override = (os.getenv("OPS_WORKSPACE_ROOT") or os.getenv("AIA_WORKSPACE_ROOT") or "").strip() if override: p = Path(override).expanduser() @@ -44,7 +42,7 @@ def _parse_pipe_separated_roots(raw: str) -> list[Path]: @dataclass(frozen=True) class WorkspacePathAccess: - """Effective path guard for the current tool invocation (env + optional per-user DB).""" + """Effective path guard for current tool invocation.""" extra_roots: tuple[Path, ...] allow_any_path: bool @@ -79,16 +77,6 @@ def build_workspace_path_access( allowlist_tenant_id: str | None = None, allowlist_user_id: str | None = None, ) -> WorkspacePathAccess: - """Resolve per-user ``extra_roots`` / ``allow_any_path`` from ``user_workspace_path_allowlist``. - - ``session_id`` is usually the chat row messages are written to (may be a specialist temp session - without ``ui_session_owner``). In that case pass ``owner_fallback_session_id`` = the user's - UI-owned session id so DB allowlist still applies. - - If ``get_ui_session_owner`` yields nothing, ``allowlist_tenant_id`` + ``allowlist_user_id`` - (from the authenticated user / request metadata) can be used to load the same allowlist, so - a missing ``ui_session_owner`` row does not drop per-user extra roots. - """ base = access_from_env() if store is None: return base @@ -125,8 +113,6 @@ def build_workspace_path_access( ) return _merge_access(base, db_access) - # Fallback: use explicit tenant / user (e.g. wecom or admin ``metadata``) when session is not - # linked in ``ui_session_owner`` (legacy session or data repair in progress). t2 = str(allowlist_tenant_id or "").strip() u2 = str(allowlist_user_id or "").strip() if not t2 or not u2: @@ -209,11 +195,6 @@ def clear_workspace_path_access_for_tests() -> None: def _is_subpath(path: Path, root: Path) -> bool: - """``path`` is under ``root`` (treated as a directory), including the root itself. - - On Windows, comparison is case- and path-separator-insensitive; ``resolve`` may - not normalize casing consistently across all drives, so we use normcase. - """ try: pr = path.resolve() rr = root.resolve() @@ -258,7 +239,6 @@ def truncate_text(s: str, *, limit: int = 20000) -> str: return s[: max(0, limit - 12)] + "\n..." -# NOTE: put '-' at end or escape it to avoid "bad character range" on Windows Python regex. _SAFE_GIT_REF_RE = re.compile(r"^[A-Za-z0-9._/\\-]{1,80}$") diff --git a/runtime/tools/experts/workspace/patch_tools.py b/runtime/tools/public/apply_patch_tool.py similarity index 85% rename from runtime/tools/experts/workspace/patch_tools.py rename to runtime/tools/public/apply_patch_tool.py index cccfbbdc..654c97bf 100644 --- a/runtime/tools/experts/workspace/patch_tools.py +++ b/runtime/tools/public/apply_patch_tool.py @@ -4,7 +4,7 @@ import hashlib from typing import Any from oclaw.runtime.tools.base import ToolSpec -from oclaw.runtime.tools.experts.workspace.workspace_base import resolve_workspace_path +from oclaw.runtime.tools.path_guard import resolve_workspace_path def apply_patch_tool() -> ToolSpec: @@ -30,7 +30,7 @@ def apply_patch_tool() -> ToolSpec: return ToolSpec( name="apply_patch", - description="Apply a full-file patch by overwriting a file with new content (optional sha256 precondition).", + description="Overwrite a file with new content (optional sha256 precondition).", parameters={ "type": "object", "properties": { @@ -38,16 +38,17 @@ def apply_patch_tool() -> ToolSpec: "new_content": {"type": "string", "description": "New full file content."}, "expected_sha256": { "type": "string", - "description": "If provided, the current file sha256 must match (precondition).", + "description": "If provided, the current file sha256 must match.", }, }, "required": ["path", "new_content"], "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "write"}), + tags=frozenset({"public", "write"}), + risk_level="high", + read_only=False, ) __all__ = ["apply_patch_tool"] - diff --git a/runtime/tools/public/local_edit_file_tool.py b/runtime/tools/public/edit_file_tool.py similarity index 93% rename from runtime/tools/public/local_edit_file_tool.py rename to runtime/tools/public/edit_file_tool.py index 48f641ba..3f8d3a7a 100644 --- a/runtime/tools/public/local_edit_file_tool.py +++ b/runtime/tools/public/edit_file_tool.py @@ -6,7 +6,7 @@ from oclaw.runtime.tools.base import ToolSpec from oclaw.runtime.tools.local_sdk import get_local_adapter -def local_edit_file_tool() -> ToolSpec: +def edit_file_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: path = str(args.get("path") or "").strip() if not path: @@ -43,7 +43,7 @@ def local_edit_file_tool() -> ToolSpec: ) return ToolSpec( - name="local_edit_file", + name="edit_file", description="Edit partial file content via local backend.", parameters={ "type": "object", @@ -59,11 +59,11 @@ def local_edit_file_tool() -> ToolSpec: "additionalProperties": False, }, handler=_handler, - tags=frozenset({"public", "local", "workspace", "write", "edit"}), + tags=frozenset({"public", "write", "edit"}), risk_level="high", timeout_s=30.0, read_only=False, ) -__all__ = ["local_edit_file_tool"] +__all__ = ["edit_file_tool"] diff --git a/runtime/tools/experts/workspace/git_tools.py b/runtime/tools/public/git_tools.py similarity index 82% rename from runtime/tools/experts/workspace/git_tools.py rename to runtime/tools/public/git_tools.py index 8b905c6e..83b2e7c7 100644 --- a/runtime/tools/experts/workspace/git_tools.py +++ b/runtime/tools/public/git_tools.py @@ -4,7 +4,7 @@ import subprocess from typing import Any from oclaw.runtime.tools.base import ToolSpec -from oclaw.runtime.tools.experts.workspace.workspace_base import resolve_workspace_path, truncate_text, sanitize_git_ref +from oclaw.runtime.tools.path_guard import resolve_workspace_path, sanitize_git_ref, truncate_text def _git(command: str, *, cwd: str) -> dict[str, Any]: @@ -25,8 +25,7 @@ def git_status_tool() -> ToolSpec: def handler(args: dict[str, Any]) -> dict[str, Any]: cwd = str(args.get("cwd") or ".").strip() res = _git("status --porcelain=v1 -b", cwd=cwd) - ok = res["exit_code"] == 0 - return {"ok": ok, **res} + return {"ok": res["exit_code"] == 0, **res} return ToolSpec( name="git_status", @@ -38,7 +37,9 @@ def git_status_tool() -> ToolSpec: "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "git"}), + tags=frozenset({"public", "git"}), + read_only=True, + risk_level="low", ) @@ -48,8 +49,7 @@ def git_diff_tool() -> ToolSpec: ref = sanitize_git_ref(str(args.get("ref") or "").strip()) if args.get("ref") else "" cmd = "diff" if not ref else f"diff {ref}...HEAD" res = _git(cmd, cwd=cwd) - ok = res["exit_code"] == 0 - return {"ok": ok, **res} + return {"ok": res["exit_code"] == 0, **res} return ToolSpec( name="git_diff", @@ -64,7 +64,9 @@ def git_diff_tool() -> ToolSpec: "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "git"}), + tags=frozenset({"public", "git"}), + read_only=True, + risk_level="low", ) @@ -74,8 +76,7 @@ def git_log_tool() -> ToolSpec: n = int(args.get("n") or 10) n = max(1, min(n, 50)) res = _git(f"log -{n} --oneline --decorate", cwd=cwd) - ok = res["exit_code"] == 0 - return {"ok": ok, **res} + return {"ok": res["exit_code"] == 0, **res} return ToolSpec( name="git_log", @@ -87,7 +88,9 @@ def git_log_tool() -> ToolSpec: "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "git"}), + tags=frozenset({"public", "git"}), + read_only=True, + risk_level="low", ) @@ -97,18 +100,16 @@ def git_commit_tool() -> ToolSpec: message = str(args.get("message") or "").strip() if not message: return {"ok": False, "error": "message_required"} - # stage all changes (simple default) s1 = _git("add -A", cwd=cwd) if s1["exit_code"] != 0: return {"ok": False, "error": "git_add_failed", **s1} msg_esc = message.replace('"', '\\"') s2 = _git(f'commit -m "{msg_esc}"', cwd=cwd) - ok = s2["exit_code"] == 0 - return {"ok": ok, **s2} + return {"ok": s2["exit_code"] == 0, **s2} return ToolSpec( name="git_commit", - description="Stage all and create a git commit (requires confirmation by policy).", + description="Stage all and create a git commit.", parameters={ "type": "object", "properties": { @@ -119,7 +120,9 @@ def git_commit_tool() -> ToolSpec: "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "git", "write"}), + tags=frozenset({"public", "git", "write"}), + risk_level="high", + read_only=False, ) @@ -129,12 +132,11 @@ def git_push_tool() -> ToolSpec: remote = str(args.get("remote") or "origin").strip() or "origin" refspec = str(args.get("refspec") or "HEAD").strip() or "HEAD" res = _git(f"push {remote} {refspec}", cwd=cwd) - ok = res["exit_code"] == 0 - return {"ok": ok, **res} + return {"ok": res["exit_code"] == 0, **res} return ToolSpec( name="git_push", - description="Push current branch (requires confirmation by policy).", + description="Push current branch to remote.", parameters={ "type": "object", "properties": { @@ -146,9 +148,10 @@ def git_push_tool() -> ToolSpec: "additionalProperties": False, }, handler=handler, - tags=frozenset({"workspace", "git", "write"}), + tags=frozenset({"public", "git", "write"}), + risk_level="high", + read_only=False, ) __all__ = ["git_status_tool", "git_diff_tool", "git_log_tool", "git_commit_tool", "git_push_tool"] - diff --git a/runtime/tools/public/glob_tool.py b/runtime/tools/public/glob_tool.py new file mode 100644 index 00000000..10556954 --- /dev/null +++ b/runtime/tools/public/glob_tool.py @@ -0,0 +1,61 @@ +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 glob_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + pattern = str(args.get("pattern") or "**/*").strip() or "**/*" + max_results = int(args.get("max_results") or 200) + root_arg = str(args.get("root") or "").strip() + if not root_arg: + base = resolve_workspace_path(".") + else: + base = resolve_workspace_path(root_arg) + if not base.is_dir(): + return {"ok": False, "error": "not_a_directory", "path": str(base)} + out: list[str] = [] + for p in base.glob(pattern): + if p.is_dir(): + continue + out.append(str(p.relative_to(base))) + if len(out) >= max(1, min(max_results, 2000)): + break + return { + "ok": True, + "root": str(base), + "pattern": pattern, + "count": len(out), + "files": out, + } + + return ToolSpec( + name="glob", + description=( + "List files under a directory matching a glob pattern. " + "Default root is workspace root; set `root` to an absolute path (e.g. D:\\download) when needed." + ), + parameters={ + "type": "object", + "properties": { + "pattern": {"type": "string", "description": "Glob pattern relative to root, e.g. '**/*' or '*.pdf'.", "default": "**/*"}, + "root": { + "type": "string", + "description": "Optional directory to search under (absolute or workspace-relative). If omitted, uses workspace root.", + }, + "max_results": {"type": "integer", "default": 200, "description": "Max number of files to return."}, + }, + "required": [], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace"}), + read_only=True, + risk_level="low", + ) + + +__all__ = ["glob_tool"] diff --git a/runtime/tools/public/index_workspace_tool.py b/runtime/tools/public/index_workspace_tool.py new file mode 100644 index 00000000..3cc826ad --- /dev/null +++ b/runtime/tools/public/index_workspace_tool.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec + + +def index_workspace_tool() -> ToolSpec: + def handler(args: dict[str, Any]) -> dict[str, Any]: + max_files = int(args.get("max_files") or 120) + try: + from oclaw.platform.persistence.sqlite_store import SqliteStore + from oclaw.platform.config.paths import db_path + from oclaw.runtime.tools.workspace_indexer import index_workspace + + store = SqliteStore(db_path()) + st = index_workspace(store, max_files=max(1, min(max_files, 800))) + return {"ok": True, "files_seen": st.files_seen, "chunks_upserted": st.chunks_upserted, "embeddings_upserted": st.embeddings_upserted} + except Exception as e: + return {"ok": False, "error": f"{type(e).__name__}: {e}"} + + return ToolSpec( + name="index_workspace", + description="Index workspace files into vector knowledge base for RAG.", + parameters={ + "type": "object", + "properties": {"max_files": {"type": "integer", "default": 120, "description": "Max files to index."}}, + "required": [], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"public", "rag"}), + risk_level="high", + read_only=False, + ) + + +__all__ = ["index_workspace_tool"] diff --git a/runtime/tools/public/local_read_file_tool.py b/runtime/tools/public/local_read_file_tool.py deleted file mode 100644 index a460124c..00000000 --- a/runtime/tools/public/local_read_file_tool.py +++ /dev/null @@ -1,43 +0,0 @@ -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 local_read_file_tool() -> ToolSpec: - def _handler(args: dict[str, Any]) -> dict[str, Any]: - path = str(args.get("path") or "").strip() - if not path: - return {"ok": False, "error_code": "path_required", "error": "path_required"} - start_line = args.get("start_line") - end_line = args.get("end_line") - return get_local_adapter().read_file( - path=path, - start_line=int(start_line) if start_line is not None else None, - end_line=int(end_line) if end_line is not None else None, - ) - - return ToolSpec( - name="local_read_file", - description="Read file content via local backend.", - parameters={ - "type": "object", - "properties": { - "path": {"type": "string", "description": "Target file path."}, - "start_line": {"type": "integer", "description": "Optional start line (1-indexed)."}, - "end_line": {"type": "integer", "description": "Optional end line (1-indexed)."}, - }, - "required": ["path"], - "additionalProperties": False, - }, - handler=_handler, - tags=frozenset({"public", "local", "workspace", "read"}), - risk_level="low", - timeout_s=20.0, - read_only=True, - ) - - -__all__ = ["local_read_file_tool"] diff --git a/runtime/tools/public/local_write_file_tool.py b/runtime/tools/public/local_write_file_tool.py deleted file mode 100644 index e7d5aef3..00000000 --- a/runtime/tools/public/local_write_file_tool.py +++ /dev/null @@ -1,39 +0,0 @@ -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 local_write_file_tool() -> ToolSpec: - def _handler(args: dict[str, Any]) -> dict[str, Any]: - path = str(args.get("path") or "").strip() - if not path: - return {"ok": False, "error_code": "path_required", "error": "path_required"} - content = str(args.get("content") or "") - mode = str(args.get("mode") or "overwrite").strip().lower() - return get_local_adapter().write_file(path=path, content=content, mode=mode) - - return ToolSpec( - name="local_write_file", - description="Write or append file content via local backend.", - parameters={ - "type": "object", - "properties": { - "path": {"type": "string", "description": "Target file path."}, - "content": {"type": "string", "description": "Content to write."}, - "mode": {"type": "string", "enum": ["overwrite", "append"], "default": "overwrite"}, - }, - "required": ["path", "content"], - "additionalProperties": False, - }, - handler=_handler, - tags=frozenset({"public", "local", "workspace", "write"}), - risk_level="high", - timeout_s=30.0, - read_only=False, - ) - - -__all__ = ["local_write_file_tool"] diff --git a/runtime/tools/public/read_file_tool.py b/runtime/tools/public/read_file_tool.py new file mode 100644 index 00000000..a49bbfaa --- /dev/null +++ b/runtime/tools/public/read_file_tool.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import hashlib +from typing import Any + +from oclaw.runtime.tools.base import ToolSpec +from oclaw.runtime.tools.path_guard import resolve_workspace_path + + +def read_file_tool() -> ToolSpec: + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or "").strip() + offset = int(args.get("offset") or 1) + limit = int(args.get("limit") or 400) + if offset == 0: + offset = 1 + if limit <= 0: + limit = 1 + p = resolve_workspace_path(path) + if not p.exists() or not p.is_file(): + return {"ok": False, "error": "file_not_found", "path": str(p)} + text = p.read_text(encoding="utf-8", errors="replace").splitlines() + if offset < 0: + start = max(0, len(text) + offset) + else: + start = max(0, offset - 1) + end = min(len(text), start + min(limit, 2000)) + out_lines = [f"{i + 1}|{text[i]}" for i in range(start, end)] + sha = hashlib.sha256(p.read_bytes()).hexdigest() + return { + "ok": True, + "path": str(p), + "start_line": start + 1, + "end_line": end, + "total_lines": len(text), + "sha256": sha, + "content": "\n".join(out_lines), + } + + return ToolSpec( + name="read_file", + description="Read a text file from the workspace with line numbers.", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "File path, relative to workspace root."}, + "offset": {"type": "integer", "description": "1-indexed start line; negative counts from end.", "default": 1}, + "limit": {"type": "integer", "description": "Max lines to return (capped).", "default": 400}, + }, + "required": ["path"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace"}), + read_only=True, + risk_level="low", + ) + + +__all__ = ["read_file_tool"] diff --git a/runtime/tools/public/local_run_command_tool.py b/runtime/tools/public/run_command_tool.py similarity index 87% rename from runtime/tools/public/local_run_command_tool.py rename to runtime/tools/public/run_command_tool.py index 5c05219a..6f231cf9 100644 --- a/runtime/tools/public/local_run_command_tool.py +++ b/runtime/tools/public/run_command_tool.py @@ -6,7 +6,7 @@ from oclaw.runtime.tools.base import ToolSpec from oclaw.runtime.tools.local_sdk import get_local_adapter -def local_run_command_tool() -> ToolSpec: +def run_command_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: command = str(args.get("command") or "").strip() if not command: @@ -16,7 +16,7 @@ def local_run_command_tool() -> ToolSpec: return get_local_adapter().run_command(command=command, cwd=cwd, timeout=timeout) return ToolSpec( - name="local_run_command", + name="run_command", description="Run a shell command via local backend.", parameters={ "type": "object", @@ -29,11 +29,11 @@ def local_run_command_tool() -> ToolSpec: "additionalProperties": False, }, handler=_handler, - tags=frozenset({"public", "local", "exec", "workspace"}), + tags=frozenset({"public", "exec"}), risk_level="high", timeout_s=90.0, read_only=False, ) -__all__ = ["local_run_command_tool"] +__all__ = ["run_command_tool"] diff --git a/runtime/tools/public/write_file_tool.py b/runtime/tools/public/write_file_tool.py new file mode 100644 index 00000000..f35633f4 --- /dev/null +++ b/runtime/tools/public/write_file_tool.py @@ -0,0 +1,68 @@ +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 write_file_tool() -> ToolSpec: + def _sandbox_base_dir() -> Path: + return Path("data") / "workspace" + + def _normalize_write_path(path: str) -> str: + raw = str(path or "").strip().strip('"').strip("'") + if not raw: + raise ValueError("path_required") + p = Path(raw) + base = _sandbox_base_dir() + if p.is_absolute(): + name = str(p.name or "").strip() + if not name: + raise ValueError("path_required") + return str(base / name) + rel = raw.lstrip("./\\") + if not rel: + raise ValueError("path_required") + return str(base / rel) + + def _handler(args: dict[str, Any]) -> dict[str, Any]: + path = str(args.get("path") or "").strip() + content = str(args.get("content") or "") + mode = str(args.get("mode") or "overwrite").strip().lower() + try: + normalized = _normalize_write_path(path) + except ValueError as exc: + return {"ok": False, "error": str(exc)} + p = resolve_workspace_path(normalized) + p.parent.mkdir(parents=True, exist_ok=True) + if mode not in ("overwrite", "append"): + return {"ok": False, "error": "invalid_mode", "allowed": ["overwrite", "append"]} + if mode == "append": + p.write_text(p.read_text(encoding="utf-8", errors="replace") + content, encoding="utf-8") + else: + p.write_text(content, encoding="utf-8") + return {"ok": True, "path": str(p), "bytes": p.stat().st_size} + + return ToolSpec( + name="write_file", + description="Write text content to a workspace file (overwrite or append).", + parameters={ + "type": "object", + "properties": { + "path": {"type": "string", "description": "File path, relative to workspace root."}, + "content": {"type": "string", "description": "Full text content to write."}, + "mode": {"type": "string", "enum": ["overwrite", "append"], "default": "overwrite"}, + }, + "required": ["path", "content"], + "additionalProperties": False, + }, + handler=_handler, + tags=frozenset({"public", "workspace", "write"}), + risk_level="high", + read_only=False, + ) + + +__all__ = ["write_file_tool"] diff --git a/tests/test_local_public_tools.py b/tests/test_local_public_tools.py index 35dfdb37..0965afef 100644 --- a/tests/test_local_public_tools.py +++ b/tests/test_local_public_tools.py @@ -5,10 +5,10 @@ from pathlib import Path from oclaw.runtime.tools.catalog import default_registry from oclaw.runtime.tools.local_sdk.adapter import LocalAdapter -from oclaw.runtime.tools.public.local_edit_file_tool import local_edit_file_tool -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.edit_file_tool import edit_file_tool +from oclaw.runtime.tools.public.run_command_tool import run_command_tool +from oclaw.runtime.tools.public.read_file_tool import read_file_tool +from oclaw.runtime.tools.public.write_file_tool import 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 @@ -21,15 +21,15 @@ 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 "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 "run_command" not in names + assert "write_file" not in names + assert "edit_file" not in names assert "mkdir" not in names assert "delete_file" not in names assert "move_file" not in names @@ -42,15 +42,15 @@ def test_local_public_high_risk_tools_visible_when_enabled(monkeypatch) -> None: clear_public_tool_cache() 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 "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 "run_command" in names + assert "write_file" in names + assert "edit_file" in names assert "mkdir" in names assert "delete_file" in names assert "move_file" in names @@ -71,13 +71,13 @@ def test_local_adapter_backend_roundtrip(tmp_path: Path, monkeypatch) -> None: assert "hi" in str(out2.get("stdout") or "").lower() -def test_local_run_command_tool_handler(monkeypatch) -> None: +def test_run_command_tool_handler(monkeypatch) -> None: class _Adapter: def run_command(self, *, command: str, cwd: str | None = None, timeout: int = 30): return {"ok": True, "command": command, "cwd": cwd, "timeout": timeout} - monkeypatch.setattr("oclaw.runtime.tools.public.local_run_command_tool.get_local_adapter", lambda: _Adapter()) - spec = local_run_command_tool() + monkeypatch.setattr("oclaw.runtime.tools.public.run_command_tool.get_local_adapter", lambda: _Adapter()) + spec = run_command_tool() out = spec.handler({"command": "echo hi", "cwd": "repo", "timeout": 9}) assert out.get("ok") is True assert out.get("command") == "echo hi" @@ -85,35 +85,7 @@ def test_local_run_command_tool_handler(monkeypatch) -> None: assert out.get("timeout") == 9 -def test_local_read_file_tool_handler(monkeypatch) -> None: - class _Adapter: - def read_file(self, *, path: str, start_line: int | None = None, end_line: int | None = None): - return {"ok": True, "path": path, "start_line": start_line, "end_line": end_line} - - monkeypatch.setattr("oclaw.runtime.tools.public.local_read_file_tool.get_local_adapter", lambda: _Adapter()) - spec = local_read_file_tool() - out = spec.handler({"path": "a.py", "start_line": 2, "end_line": 5}) - assert out.get("ok") is True - assert out.get("path") == "a.py" - assert out.get("start_line") == 2 - assert out.get("end_line") == 5 - - -def test_local_write_file_tool_handler(monkeypatch) -> None: - class _Adapter: - def write_file(self, *, path: str, content: str, mode: str = "overwrite"): - return {"ok": True, "path": path, "content": content, "mode": mode} - - monkeypatch.setattr("oclaw.runtime.tools.public.local_write_file_tool.get_local_adapter", lambda: _Adapter()) - spec = local_write_file_tool() - out = spec.handler({"path": "a.py", "content": "x=1", "mode": "append"}) - assert out.get("ok") is True - assert out.get("path") == "a.py" - assert out.get("content") == "x=1" - assert out.get("mode") == "append" - - -def test_local_edit_file_tool_handler(monkeypatch) -> None: +def test_edit_file_tool_handler(monkeypatch) -> None: class _Adapter: def edit_file( self, @@ -135,8 +107,8 @@ def test_local_edit_file_tool_handler(monkeypatch) -> None: "replacement": replacement, } - monkeypatch.setattr("oclaw.runtime.tools.public.local_edit_file_tool.get_local_adapter", lambda: _Adapter()) - spec = local_edit_file_tool() + monkeypatch.setattr("oclaw.runtime.tools.public.edit_file_tool.get_local_adapter", lambda: _Adapter()) + spec = edit_file_tool() out = spec.handler({"path": "a.py", "search": "foo", "replace": "bar"}) assert out.get("ok") is True assert out.get("path") == "a.py" @@ -145,26 +117,26 @@ def test_local_edit_file_tool_handler(monkeypatch) -> None: def test_local_tool_integration_roundtrip(monkeypatch) -> None: - run_spec = local_run_command_tool() - read_spec = local_read_file_tool() - write_spec = local_write_file_tool() - edit_spec = local_edit_file_tool() + run_spec = run_command_tool() + read_spec = read_file_tool() + write_spec = write_file_tool() + edit_spec = edit_file_tool() tmpdir = Path(tempfile.mkdtemp(prefix="local_it_")) monkeypatch.setenv("OPS_WORKSPACE_ROOT", str(tmpdir)) - target = tmpdir / "it_sample.txt" + target_rel = "data/workspace/it_sample.txt" - out_write = write_spec.handler({"path": str(target), "content": "line1\nline2\n", "mode": "overwrite"}) + out_write = write_spec.handler({"path": "it_sample.txt", "content": "line1\nline2\n", "mode": "overwrite"}) assert out_write.get("ok") is True, out_write - out_read_before = read_spec.handler({"path": str(target), "start_line": 1, "end_line": 10}) + out_read_before = read_spec.handler({"path": target_rel, "offset": 1, "limit": 10}) assert out_read_before.get("ok") is True, out_read_before assert "line2" in str(out_read_before.get("content") or "") - out_edit = edit_spec.handler({"path": str(target), "search": "line2", "replace": "line2_edited"}) + out_edit = edit_spec.handler({"path": target_rel, "search": "line2", "replace": "line2_edited"}) assert out_edit.get("ok") is True, out_edit - out_read_after = read_spec.handler({"path": str(target), "start_line": 1, "end_line": 10}) + out_read_after = read_spec.handler({"path": target_rel, "offset": 1, "limit": 10}) assert out_read_after.get("ok") is True, out_read_after assert "line2_edited" in str(out_read_after.get("content") or "") @@ -203,3 +175,21 @@ def test_p1_p2_read_tools_smoke(tmp_path: Path, monkeypatch) -> None: assert out_get1.get("ok") is True assert out_get1.get("value") == "y" + +def test_run_command_does_not_follow_cd_state(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("OPS_WORKSPACE_ROOT", str(tmp_path)) + (tmp_path / "subdir").mkdir(parents=True, exist_ok=True) + (tmp_path / "subdir" / "echo_dir.py").write_text( + "import os\nprint(os.path.basename(os.getcwd()))\n", + encoding="utf-8", + ) + + adapter = LocalAdapter() + out_cd = adapter.cd(cwd="subdir") + assert out_cd.get("ok") is True + + out_run = adapter.run_command(command='python -c "import os; print(os.path.basename(os.getcwd()))"', timeout=20) + assert out_run.get("ok") is True, out_run + # If run_command follows cd state this would be "subdir"; we expect workspace root name instead. + assert str(out_run.get("cwd") or "").replace("\\", "/").rstrip("/").endswith(str(tmp_path.name)) + diff --git a/tests/test_workspace_path_guard.py b/tests/test_workspace_path_guard.py index 939e0c24..498b3980 100644 --- a/tests/test_workspace_path_guard.py +++ b/tests/test_workspace_path_guard.py @@ -12,9 +12,9 @@ from fastapi.testclient import TestClient from oclaw.interfaces.http.fastapi_app import create_app from oclaw.platform.config.paths import db_path from oclaw.platform.persistence.sqlite_store import SqliteStore -from oclaw.runtime.tools.experts.workspace.fs_tools import list_files_tool, write_file_tool -from oclaw.runtime.tools.experts.workspace.shell_tools import run_command_tool -from oclaw.runtime.tools.experts.workspace.workspace_base import ( +from oclaw.runtime.tools.public.glob_tool import glob_tool +from oclaw.runtime.tools.public.write_file_tool import write_file_tool +from oclaw.runtime.tools.path_guard import ( access_from_env, build_workspace_path_access, clear_workspace_path_access_for_tests, @@ -79,7 +79,7 @@ class WorkspacePathGuardTests(unittest.TestCase): clear=False, ): clear_workspace_path_access_for_tests() - spec = list_files_tool() + spec = glob_tool() with workspace_path_access_scope(None, None): r = spec.handler({"root": str(sub), "pattern": "**/*", "max_results": 50}) self.assertTrue(r.get("ok"), r) @@ -150,224 +150,6 @@ class WorkspacePathGuardTests(unittest.TestCase): self.assertEqual(str(expected), str(r.get("path"))) self.assertTrue(expected.exists()) - def test_run_command_default_cwd_uses_workspace_namespace_sandbox(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - with workspace_path_access_scope(None, None), workspace_write_namespace_scope("ops"): - r = spec.handler({"command": "python -c \"print('ok')\""}) - self.assertTrue(r.get("ok"), r) - expected_cwd = (self.root / "data" / "workspace").resolve() - self.assertEqual(str(expected_cwd), str(r.get("cwd"))) - - def test_run_command_strips_leading_cd_chain_in_default_sandbox(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - cmd = f'cd /d "{self.root}" && python -c "print(123)"' - with workspace_path_access_scope(None, None), workspace_write_namespace_scope("ops"): - r = spec.handler({"command": cmd}) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("normalized_cd_removed"))) - expected_cwd = (self.root / "data" / "workspace").resolve() - self.assertEqual(str(expected_cwd), str(r.get("cwd"))) - self.assertIn("123", str(r.get("output") or "")) - - def test_run_command_strips_windows_drive_prefix_cd_chain(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - ws_root = self.root / "data" / "workspace" - ws_root.mkdir(parents=True, exist_ok=True) - (ws_root / "count_directory.py").write_text("print('drive-cd-ok')\n", encoding="utf-8") - spec = run_command_tool() - cmd = f'D: && cd /d "{self.root}" && python count_directory.py' - with workspace_path_access_scope(None, None): - r = spec.handler({"command": cmd}) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("normalized_cd_removed")), r) - self.assertTrue(bool(r.get("script_path_rewritten")), r) - self.assertIn("drive-cd-ok", str(r.get("output") or "")) - - def test_run_command_output_flags_distinguish_empty_from_truncation(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - with workspace_path_access_scope(None, None), workspace_write_namespace_scope("ops"): - r = spec.handler({"command": 'python -c "pass"'}) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("output_empty"))) - self.assertFalse(bool(r.get("output_truncated"))) - self.assertTrue(bool(r.get("output_not_truncated"))) - self.assertEqual(str(r.get("error_code") or ""), "") - - def test_run_command_nonzero_exit_marks_failure_not_truncation(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - with workspace_path_access_scope(None, None), workspace_write_namespace_scope("ops"): - r = spec.handler({"command": 'python -c "import sys; sys.exit(3)"'}) - self.assertFalse(bool(r.get("ok"))) - self.assertEqual(int(r.get("exit_code") or 0), 3) - self.assertEqual(str(r.get("error_code") or ""), "command_exit_nonzero") - self.assertFalse(bool(r.get("output_truncated"))) - - def test_run_command_blocks_cocoloop_install_cli(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - with workspace_path_access_scope(None, None): - r = spec.handler({"command": "cocoloop install 7288"}) - self.assertFalse(bool(r.get("ok")), r) - self.assertEqual("skill_install_cli_blocked", str(r.get("error_code") or "")) - self.assertIn("market/install", str(r.get("hint") or "")) - - def test_run_command_blocks_npx_clawhub_install_pattern(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - spec = run_command_tool() - with workspace_path_access_scope(None, None): - r = spec.handler({"command": "npx -y clawhub@latest install foo"}) - self.assertFalse(bool(r.get("ok")), r) - self.assertEqual("skill_install_cli_blocked", str(r.get("error_code") or "")) - - def test_run_command_rewrites_workspace_absolute_script_path_to_sandbox(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - # Prepare script inside sandbox, but command will reference repo-root absolute path. - ws_script = self.root / "data" / "workspace" / "count_files.py" - ws_script.parent.mkdir(parents=True, exist_ok=True) - ws_script.write_text("print('sandbox-ok')\n", encoding="utf-8") - spec = run_command_tool() - absolute_repo_script = str((self.root / "count_files.py").resolve()) - with workspace_path_access_scope(None, None), workspace_write_namespace_scope("ops"): - r = spec.handler({"command": f'python "{absolute_repo_script}"'}) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("command_rewritten")), r) - self.assertIn("sandbox-ok", str(r.get("output") or "")) - - def test_run_command_explicit_repo_root_cwd_is_redirected_to_sandbox(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - ws_script = self.root / "data" / "workspace" / "count_files.py" - ws_script.parent.mkdir(parents=True, exist_ok=True) - ws_script.write_text("print('redirect-ok')\n", encoding="utf-8") - spec = run_command_tool() - abs_repo_script = str((self.root / "count_files.py").resolve()) - with workspace_path_access_scope(None, None): - r = spec.handler( - { - "command": f'python "{abs_repo_script}"', - "cwd": str(self.root), - } - ) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("cwd_redirected_to_sandbox")), r) - self.assertTrue(bool(r.get("command_rewritten")), r) - self.assertIn("redirect-ok", str(r.get("output") or "")) - - def test_run_command_rewrites_relative_python_script_to_sandbox_root(self) -> None: - with mock.patch.dict( - os.environ, - { - "OPS_WORKSPACE_ROOT": str(self.root), - "OPS_WORKSPACE_EXTRA_ROOTS": "", - "OPS_WORKSPACE_ALLOW_ANY_PATH": "", - "AIA_ENABLE_RUN_COMMAND": "1", - }, - clear=False, - ): - clear_workspace_path_access_for_tests() - ws_root = self.root / "data" / "workspace" - ws_root.mkdir(parents=True, exist_ok=True) - (ws_root / "count_directory.py").write_text("print('found-in-sandbox-root')\n", encoding="utf-8") - spec = run_command_tool() - with workspace_path_access_scope(None, None): - r = spec.handler({"command": "python count_directory.py"}) - self.assertTrue(r.get("ok"), r) - self.assertTrue(bool(r.get("script_path_rewritten")), r) - self.assertIn("found-in-sandbox-root", str(r.get("output") or "")) - def test_per_user_extra_roots_from_db(self) -> None: f = self.extra / "u.txt" f.write_text("u", encoding="utf-8")