feat: filesystem grep, read, write, edit file and workspace support#7402
Open
feat: filesystem grep, read, write, edit file and workspace support#7402
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request introduces new filesystem tools—ReadFileTool, FileEditTool, and GrepTool—and updates the local and sandbox booters to support these operations. It also implements a security layer to restrict file access for non-admin users in local environments. The review feedback highlights potential memory issues when reading or editing large files in their entirety and suggests applying the documented default limit for file reads to prevent excessive memory consumption.
| limit: int | None = None, | ||
| ) -> dict[str, Any]: | ||
| _ = encoding | ||
| content = await self._sandbox.filesystem.read_file(path) |
Contributor
There was a problem hiding this comment.
This implementation reads the full content of the file from the sandbox into the bot's memory before slicing. This is inefficient for large files. If the Shipyard Neo SDK supports range-based reads, they should be used here to fetch only the requested slice.
Soulter
changed the title
… including validation for large files
5 tasks
whatevertogo
reviewed
Apr 8, 2026
Contributor
There was a problem hiding this comment.
代码审查反馈(claude code生成的不要喷我呜呜呜~)
我仔细审查了这个 PR 的代码实现,发现了几个需要关注的问题。已在相关代码行添加了详细评论。
主要发现
✅ 做得好的地方:
- read_file 在 local booter 中使用了逐行迭代读取,内存安全
- GrepTool 有完善的 result_limit 机制(默认100),防止结果过多
- 测试覆盖充分,包含了大文件、图片、二进制文件等场景
- 工作空间隔离和权限控制设计合理
- edit_file 内存问题 - 使用 f.read() 读取整个文件,大文件会 OOM
- 默认 limit 未生效 - _validate_read_window 没有应用文档中的默认值 4000
- Shipyard Neo 性能 - 先读取完整文件再切片,效率低
这些问题都有具体的修复建议,请查看代码行评论。
总体评价
这是一个功能完整、架构良好的 PR,修复上述问题后可以合并。建议优先处理前两个问题,它们对内存安全有直接影响。
| self, | ||
| path: str, | ||
| encoding: str = "utf-8", | ||
| offset: int | None = None, |
| def _run() -> dict[str, Any]: | ||
| abs_path = os.path.abspath(path) | ||
| with open(abs_path, encoding=encoding) as f: | ||
| content = f.read() |
Contributor
There was a problem hiding this comment.
内存安全问题: 使用 f.read() 一次性读取整个文件到内存,对于大文件可能导致 OOM。
建议改为逐行处理或使用临时文件:
# 方案1: 逐行处理(适合简单替换)
with open(abs_path, 'r', encoding=encoding) as f_in:
with open(temp_path, 'w', encoding=encoding) as f_out:
for line in f_in:
if replace_all:
line = line.replace(old_string, new_string)
else:
if old_string in line and replacements < 1:
line = line.replace(old_string, new_string, 1)
replacements += 1
f_out.write(line)
# 方案2: 限制可编辑文件大小
file_size = os.path.getsize(abs_path)
if file_size > 10 * 1024 * 1024: # 10MB
return {"success": False, "error": "File too large for editing"}| path: str, | ||
| offset: int | None = None, | ||
| limit: int | None = None, | ||
| ) -> ToolExecResult: |
Contributor
There was a problem hiding this comment.
默认值未生效: 文档中 limit 参数说明默认为 4000,但这里的验证逻辑没有应用这个默认值。当 LLM 不提供 limit 参数时,会传递 None 给 booter,可能导致读取整个文件。
建议应用默认值:
def _validate_read_window(
self,
offset: int | None,
limit: int | None,
) -> tuple[int | None, int | None]:
if offset is not None and offset < 0:
raise ValueError("`offset` must be greater than or equal to 0.")
if limit is not None and limit < 1:
raise ValueError("`limit` must be greater than or equal to 1.")
# 应用默认值 4000
return offset, limit if limit is not None else 4000这样可以确保即使 LLM 忘记提供 limit,也不会尝试读取超大文件。
| limit: int | None = None, | ||
| ) -> dict[str, Any]: | ||
| _ = encoding | ||
| content = await self._sandbox.filesystem.read_file(path) |
Contributor
There was a problem hiding this comment.
性能问题: 这里先通过 self._sandbox.filesystem.read_file(path) 读取完整文件内容到内存,然后再用 _slice_content_by_lines 切片。对于大文件这样做效率很低且可能导致 OOM。
如果 Shipyard Neo SDK 支持范围读取(range-based read),应该优先使用 SDK 的原生能力,只获取需要的部分。
如果 SDK 不支持范围读取,建议:
- 在工具层面添加文件大小检查,拒绝读取过大的文件
- 或者在文档中明确说明 Shipyard Neo 环境下的文件大小限制
Contributor
|
太大了会不会炸喵，还是说你会从下载方面限制呢
|
Member
Author
最大 128KB 或者 25000 token，如果超出了就会提示错误。 |
dosubot
bot
added
size:XXL
|
Related Documentation 1 document(s) may need updating based on files changed in this PR: AstrBotTeam's Space pr4697的改动View Suggested Changes@@ -121,6 +121,34 @@
该优化避免了在运行时类型已明确时进行不必要的沙箱配置检查，提高了边缘场景的性能。
+**Workspace 支持（PR #7402）**
+
+[PR #7402](https://github.com/AstrBotDevs/AstrBot/pull/7402) 为文件系统工具引入了 workspace 支持，确保用户文件操作在隔离的工作区中进行。
+
+**路径解析规则：**
+
+- **本地运行时**（`runtime="local"`）：相对路径解析到用户专属工作区目录 `data/workspaces/{normalized_umo}/`，其中 `normalized_umo` 是经过规范化的统一消息来源标识符
+- **沙盒运行时**（`runtime="sandbox"`）：相对路径保持不变，由沙盒运行时自行处理
+
+**路径限制（非管理员用户）：**
+
+当 `provider_settings.computer_use_require_admin=True` 且用户角色非管理员时，在本地运行时下，文件系统工具（读取、写入、编辑、搜索）的路径访问受以下限制：
+- `data/skills`：技能目录
+- `data/workspaces/{normalized_umo}`：用户专属工作区
+- `/tmp/.astrbot`：临时文件目录
+
+尝试访问这些目录之外的路径时，工具会返回权限错误，并明确提示允许访问的目录列表。
+
+**工具能力：**
+
+- **`astrbot_read_file_tool`**：支持可选的 `offset` 和 `limit` 参数，用于读取文件的特定行范围（基于 0 的行索引）
+- **`astrbot_file_write_tool`**：写入 UTF-8 文本内容到文件
+- **`astrbot_file_edit_tool`**：使用字符串替换编辑文件，支持 `replace_all` 参数控制是否替换所有匹配项
+- **`astrbot_grep_tool`**：使用 ripgrep 搜索文件内容，支持以下可选参数：
+ - `glob`：glob 过滤器（如 `*.py`、`*.{ts,tsx}`）
+ - `-A`、`-B`、`-C`：上下文行数参数（after、before、context）
+ - `result_limit`：返回的结果组数上限（默认 100）
+
**运行时工具列表：**
- **`runtime="sandbox"`** 提供以下工具：
@@ -129,6 +157,10 @@
- 工具描述中包含当前操作系统信息（通过 `platform.system()` 获取），帮助 LLM 生成与平台兼容的代码
- `FILE_UPLOAD_TOOL`：上传文件到沙盒环境
- `FILE_DOWNLOAD_TOOL`：从沙盒环境下载文件
+ - `astrbot_read_file_tool`：读取文件内容，支持可选的行偏移（offset）和行数限制（limit）参数，用于读取指定行范围
+ - `astrbot_file_write_tool`：写入 UTF-8 文本内容到文件
+ - `astrbot_file_edit_tool`：使用字符串替换编辑文件，支持可选的 `replace_all` 参数
+ - `astrbot_grep_tool`：使用 ripgrep 搜索文件内容，支持可选的 glob 过滤器和上下文行参数（`-A`、`-B`、`-C`）
- **`runtime="local"`** 或 **`runtime="local_sandboxed"`** 提供以下工具：
- `LOCAL_EXECUTE_SHELL_TOOL`：在本地环境中执行 Shell 命令
@@ -137,6 +169,10 @@
- `LOCAL_PYTHON_TOOL`：在本地环境中执行 Python 代码
- 在 `local_sandboxed` 模式下，文件写入受限于工作区目录（`~/.astrbot/workspace/<session>`）
- 工具描述中包含当前操作系统信息（通过 `platform.system()` 获取），帮助 LLM 生成与平台兼容的代码
+ - `astrbot_read_file_tool`：读取文件内容，支持可选的行偏移（offset）和行数限制（limit）参数，用于读取指定行范围
+ - `astrbot_file_write_tool`：写入 UTF-8 文本内容到文件
+ - `astrbot_file_edit_tool`：使用字符串替换编辑文件，支持可选的 `replace_all` 参数
+ - `astrbot_grep_tool`：使用 ripgrep 搜索文件内容，支持可选的 glob 过滤器和上下文行参数（`-A`、`-B`、`-C`）
这些工具在 SubAgent handoff 场景下可正常使用，与主 Agent 运行时动态挂载的工具保持一致。
@@ -244,15 +280,19 @@
[PR #5402](https://github.com/AstrBotDevs/AstrBot/pull/5402) 强化了所有 Computer Use 工具的管理员权限强制执行，确保统一的权限控制策略：
-当 `provider_settings.computer_use_require_admin` 设置为 `true`（默认值）时，以下所有 Computer Use 工具均强制要求管理员权限：
-
-- **Shell 执行工具**：`EXECUTE_SHELL_TOOL`（沙盒）、`LOCAL_EXECUTE_SHELL_TOOL`（本地）
-- **Python 执行工具**：`PYTHON_TOOL`（沙盒）、`LOCAL_PYTHON_TOOL`（本地）
-- **文件传输工具**：`FILE_UPLOAD_TOOL`（`astrbot_upload_file`）、`FILE_DOWNLOAD_TOOL`（`astrbot_download_file`）
+当 `provider_settings.computer_use_require_admin` 设置为 `true`（默认值）时：
+
+- **文件传输工具**（`FILE_UPLOAD_TOOL`、`FILE_DOWNLOAD_TOOL`）：要求管理员权限
+- **读写编辑搜索工具**（`astrbot_read_file_tool`、`astrbot_file_write_tool`、`astrbot_file_edit_tool`、`astrbot_grep_tool`）：
+ - 在本地运行时（`runtime="local"`）：非管理员用户受路径限制，仅可访问以下目录：
+ - `data/skills`
+ - `data/workspaces/{normalized_umo}`
+ - `/tmp/.astrbot`
+ - 在沙盒运行时（`runtime="sandbox"`）：此模块不进行路径限制，由沙盒运行时边界强制执行
权限控制行为：
-- 非管理员用户尝试使用这些工具时，会收到权限拒绝错误，并提示联系管理员或使用 `/sid` 命令查看自己的用户 ID
-- 该修复确保所有 Computer Use 操作（Shell、Python、文件传输）受到统一的权限管理，防止非管理员用户在 Computer Use 工具可用时执行任何 sandbox 或本地环境操作
+- 非管理员用户尝试上传/下载文件时，会收到权限拒绝错误，并提示联系管理员或使用 `/sid` 命令查看自己的用户 ID
+- 非管理员用户在本地运行时尝试访问受限目录外的文件时，会收到权限错误提示，并显示允许访问的目录列表
**Docker Compose 配置优化（PR #6191）**
Note: You must be authenticated to accept/decline updates. How did I do? Any feedback? |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 6 issues, and left some high level feedback:
- Relaxing
_ensure_safe_pathinLocalBooterso that allcreate_file/read_file/write_file/delete_file/list_dircalls now accept arbitrary absolute paths shifts safety entirely to the tool layer; if any other code path usesLocalFileSystemComponentdirectly (bypassing the new tools), it can now read/write outside the Astrbot roots, so it’s worth either reinstating a minimal guard there or clearly constraining its callers. - In
GrepTool, using JSON argument names like"-A","-B", and"-C"works only because they are funneled through**kwargs; consider renaming these to identifier-like keys (e.g.after_context,before_context,context) for more robust schema compatibility and easier consumption by clients.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Relaxing `_ensure_safe_path` in `LocalBooter` so that all `create_file/read_file/write_file/delete_file/list_dir` calls now accept arbitrary absolute paths shifts safety entirely to the tool layer; if any other code path uses `LocalFileSystemComponent` directly (bypassing the new tools), it can now read/write outside the Astrbot roots, so it’s worth either reinstating a minimal guard there or clearly constraining its callers.
- In `GrepTool`, using JSON argument names like `"-A"`, `"-B"`, and `"-C"` works only because they are funneled through `**kwargs`; consider renaming these to identifier-like keys (e.g. `after_context`, `before_context`, `context`) for more robust schema compatibility and easier consumption by clients.
## Individual Comments
### Comment 1
<location path="astrbot/core/computer/booters/shipyard_neo.py" line_range="49-50" />
<code_context>
+ return "".join(selected)
+
+
class NeoPythonComponent(PythonComponent):
- def __init__(self, sandbox: Any) -> None:
+ def __init__(self, sandbox: Sandbox) -> None:
self._sandbox = sandbox
</code_context>
<issue_to_address>
**issue (bug_risk):** Using `Sandbox` as a type annotation without postponed evaluation can break module import when the SDK is missing.
Because this module doesn’t use `from __future__ import annotations`, the `sandbox: Sandbox` annotations on `NeoPythonComponent`, `NeoShellComponent`, `NeoFileSystemComponent`, and `NeoBrowserComponent` are evaluated at import time. When `shipyard_neo` isn’t installed, the `ImportError` branch runs and `Sandbox` is never defined, so these annotations raise a `NameError` during class definition and prevent the module from importing. Please either add `from __future__ import annotations` at the top of the file or use a string annotation (e.g. `sandbox: "Sandbox"`) to avoid this.
</issue_to_address>
### Comment 2
<location path="tests/test_computer_fs_tools.py" line_range="92-93" />
<code_context>
+ return "".join(f"line-{index:05d}-{'x' * 48}\n" for index in range(6000))
+
+
+@pytest.mark.asyncio
+async def test_file_read_tool_rejects_large_full_text_read_before_local_stream_read(
+ monkeypatch: pytest.MonkeyPatch,
+ tmp_path,
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for FileWriteTool and FileEditTool, including replace_all behavior and not-found cases
The new tests exercise read and grep behavior but don’t cover `FileWriteTool` or `FileEditTool`. Please add tests that: (1) confirm `FileWriteTool` writes UTF‑8 content to a normalized workspace path; (2) verify `FileEditTool` replaces only the first occurrence when `replace_all=False` and all occurrences when `replace_all=True`; and (3) check that when `old` is not found, the tool returns the expected error and `replacements=0`. You can use `_setup_local_fs_tools` and `LocalBooter` here to cover the full path‑normalization and `fs.edit_file` flow.
Suggested implementation:
```python
def _make_large_text() -> str:
return "".join(f"line-{index:05d}-{'x' * 48}\n" for index in range(6000))
@pytest.mark.asyncio
async def test_file_write_tool_writes_utf8_content_to_normalized_workspace_path(
monkeypatch: pytest.MonkeyPatch,
tmp_path: "Path",
) -> None:
# Arrange: set up local fs tools and workspace
tools, fs_tools, workspaces_root, _temp_root = _setup_local_fs_tools(
monkeypatch=monkeypatch,
tmp_path=tmp_path,
umo="qq:friend:user-1",
)
# Find the FileWriteTool in the registered tools
file_write_tool = next(
tool for tool in tools if getattr(tool, "name", "") == "file_write"
)
relative_path = "subdir/utf8-file.txt"
content = "héllo ☃ – UTF‑8 content"
# Act: run FileWriteTool to write the file
result = await file_write_tool.run(
path=relative_path,
content=content,
)
# Assert: tool reports success
assert result.success, result
normalized_umo = fs_tools._normalize_umo_for_workspace("qq:friend:user-1")
workspace = workspaces_root / normalized_umo
written_path = workspace / relative_path
assert written_path.is_file()
# Ensure the file was written as UTF‑8 and the content round‑trips correctly
data = written_path.read_text(encoding="utf-8")
assert data == content
@pytest.mark.asyncio
async def test_file_edit_tool_replace_first_only_when_replace_all_false(
monkeypatch: pytest.MonkeyPatch,
tmp_path: "Path",
) -> None:
# Arrange
tools, fs_tools, workspaces_root, _temp_root = _setup_local_fs_tools(
monkeypatch=monkeypatch,
tmp_path=tmp_path,
umo="qq:friend:user-1",
)
file_edit_tool = next(
tool for tool in tools if getattr(tool, "name", "") == "file_edit"
)
normalized_umo = fs_tools._normalize_umo_for_workspace("qq:friend:user-1")
workspace = workspaces_root / normalized_umo
workspace.mkdir(parents=True, exist_ok=True)
target_path = workspace / "replace-first-only.txt"
original_text = "spam eggs spam spam\n"
target_path.write_text(original_text, encoding="utf-8")
# Act: edit with replace_all=False
result = await file_edit_tool.run(
path="replace-first-only.txt",
old="spam",
new="ham",
replace_all=False,
)
# Assert: only the first occurrence is replaced
assert result.success, result
assert result.replacements == 1
edited_text = target_path.read_text(encoding="utf-8")
assert edited_text == "ham eggs spam spam\n"
@pytest.mark.asyncio
async def test_file_edit_tool_replace_all_when_replace_all_true(
monkeypatch: pytest.MonkeyPatch,
tmp_path: "Path",
) -> None:
# Arrange
tools, fs_tools, workspaces_root, _temp_root = _setup_local_fs_tools(
monkeypatch=monkeypatch,
tmp_path=tmp_path,
umo="qq:friend:user-1",
)
file_edit_tool = next(
tool for tool in tools if getattr(tool, "name", "") == "file_edit"
)
normalized_umo = fs_tools._normalize_umo_for_workspace("qq:friend:user-1")
workspace = workspaces_root / normalized_umo
workspace.mkdir(parents=True, exist_ok=True)
target_path = workspace / "replace-all.txt"
original_text = "spam eggs spam spam\n"
target_path.write_text(original_text, encoding="utf-8")
# Act: edit with replace_all=True
result = await file_edit_tool.run(
path="replace-all.txt",
old="spam",
new="ham",
replace_all=True,
)
# Assert: all occurrences are replaced
assert result.success, result
# We know there are 3 "spam" occurrences
assert result.replacements == 3
edited_text = target_path.read_text(encoding="utf-8")
assert edited_text == "ham eggs ham ham\n"
@pytest.mark.asyncio
async def test_file_edit_tool_not_found_returns_error_and_zero_replacements(
monkeypatch: pytest.MonkeyPatch,
tmp_path: "Path",
) -> None:
# Arrange
tools, fs_tools, workspaces_root, _temp_root = _setup_local_fs_tools(
monkeypatch=monkeypatch,
tmp_path=tmp_path,
umo="qq:friend:user-1",
)
file_edit_tool = next(
tool for tool in tools if getattr(tool, "name", "") == "file_edit"
)
normalized_umo = fs_tools._normalize_umo_for_workspace("qq:friend:user-1")
workspace = workspaces_root / normalized_umo
workspace.mkdir(parents=True, exist_ok=True)
target_path = workspace / "no-match.txt"
original_text = "there is no keyword here\n"
target_path.write_text(original_text, encoding="utf-8")
# Act: attempt to replace a string that does not exist
result = await file_edit_tool.run(
path="no-match.txt",
old="nonexistent",
new="replacement",
replace_all=True,
)
# Assert: tool signals a not-found error and does not modify the file
assert not result.success
assert getattr(result, "replacements", 0) == 0
assert "not found" in (result.error or "").lower()
edited_text = target_path.read_text(encoding="utf-8")
assert edited_text == original_text
```
The exact helper and tool APIs may differ slightly in your codebase. Please adjust the following as needed:
1. `_setup_local_fs_tools(...)`:
- The tests assume it returns `(tools, fs_tools, workspaces_root, temp_root)` and accepts `umo` as a keyword argument. If the actual signature or return values differ, update the unpacking and arguments accordingly.
2. Tool discovery:
- The tests look for tools via `tool.name == "file_write"` / `"file_edit"` and call `tool.run(...)`. If your tools expose a different identifier (e.g., `tool.tool_name`) or method (`tool._run` / `tool.__call__`), update the attribute names and invocation accordingly.
3. Result object:
- The tests assume the result has `success: bool`, `replacements: int` for `FileEditTool`, and `error: Optional[str]` for failures. If your result type uses different field names, adapt the assertions (`result.ok`, `result.status`, etc.).
4. Path semantics:
- The tests pass `path` as a workspace-relative path (e.g., `"replace-all.txt"`). If your tools expect absolute paths or a different base, adjust the `path` argument to match how other tests in this file invoke `FileWriteTool` / `FileEditTool`.
</issue_to_address>
### Comment 3
<location path="tests/unit/test_astr_main_agent.py" line_range="1494" />
<code_context>
"""Tests for _apply_sandbox_tools function."""
- def test_apply_sandbox_tools_creates_toolset_if_none(self):
+ def test_apply_sandbox_tools_creates_toolset_if_none(self, mock_context):
"""Test that ToolSet is created when func_tool is None."""
module = ama
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for _apply_local_env_tools and runtime tool selection covering filesystem and grep tools
The new logic wires in local filesystem tools via `_apply_local_env_tools` and changes `_get_runtime_computer_tools` to build tool sets from the tool manager, but tests still only cover `_apply_sandbox_tools`. Please add tests for `_apply_local_env_tools` (ensuring it creates a `ToolSet` when `func_tool` is `None` and that tool names include shell/python plus the new filesystem/grep tools), and add a small test for `_get_runtime_computer_tools` for both `runtime='local'` and `'sandbox'` using a fake `tool_mgr` to assert expected tool names and correct distinction between local and sandbox python tools. This will help catch regressions in the new registration flow.
Suggested implementation:
```python
module = ama
config = module.MainAgentBuildConfig(
assert "astrbot_upload_file" in tool_names
class TestApplyLocalEnvTools:
"""Tests for _apply_local_env_tools function."""
def test_apply_local_env_tools_creates_toolset_if_none(self, mock_context):
"""ToolSet is created and populated when func_tool is None."""
module = ama
class DummyReq:
pass
req = DummyReq()
req.func_tool = None
# Exercise
module._apply_local_env_tools(req=req, context=mock_context)
# Verify ToolSet created
assert req.func_tool is not None
assert isinstance(req.func_tool, ToolSet)
# Verify expected categories of tools are present
tool_names = {t.name for t in req.func_tool.tools}
# Shell / python tools
assert any("shell" in name.lower() for name in tool_names)
assert any("python" in name.lower() for name in tool_names)
# Filesystem and grep tools
assert any("file" in name.lower() or "fs" in name.lower() or "filesystem" in name.lower() for name in tool_names)
assert any("grep" in name.lower() for name in tool_names)
def test_apply_local_env_tools_extends_existing_toolset(self, mock_context):
"""Existing ToolSet is extended with local env tools."""
module = ama
class DummyReq:
pass
# Start with a ToolSet containing a single dummy tool
dummy_tool = MagicMock()
dummy_tool.name = "pre_existing_tool"
existing_toolset = ToolSet([dummy_tool])
req = DummyReq()
req.func_tool = existing_toolset
# Exercise
module._apply_local_env_tools(req=req, context=mock_context)
# Verify toolset is the same instance but extended
assert req.func_tool is existing_toolset
tool_names = {t.name for t in req.func_tool.tools}
assert "pre_existing_tool" in tool_names
assert any("shell" in name.lower() for name in tool_names)
assert any("python" in name.lower() for name in tool_names)
class TestGetRuntimeComputerTools:
"""Tests for _get_runtime_computer_tools runtime selection."""
class _FakeTool:
def __init__(self, name: str) -> None:
self.name = name
class _FakeToolManager:
"""Minimal fake tool manager for testing runtime selection."""
def __init__(self) -> None:
# Distinct tool instances so we can check identity/name
self.local_shell = TestGetRuntimeComputerTools._FakeTool("local_shell")
self.local_python = TestGetRuntimeComputerTools._FakeTool("local_python")
self.sandbox_shell = TestGetRuntimeComputerTools._FakeTool("sandbox_shell")
self.sandbox_python = TestGetRuntimeComputerTools._FakeTool("sandbox_python")
def get_tool(self, name: str):
return {
"local_shell": self.local_shell,
"local_python": self.local_python,
"sandbox_shell": self.sandbox_shell,
"sandbox_python": self.sandbox_python,
}[name]
def test_get_runtime_computer_tools_local_uses_local_tools(self):
module = ama
tool_mgr = self._FakeToolManager()
toolset = module._get_runtime_computer_tools(tool_mgr=tool_mgr, runtime="local")
assert isinstance(toolset, ToolSet)
tool_names = {t.name for t in toolset.tools}
assert "local_shell" in tool_names
assert "local_python" in tool_names
# Ensure sandbox tools are not wired for local runtime
assert "sandbox_shell" not in tool_names
assert "sandbox_python" not in tool_names
def test_get_runtime_computer_tools_sandbox_uses_sandbox_tools(self):
module = ama
tool_mgr = self._FakeToolManager()
toolset = module._get_runtime_computer_tools(tool_mgr=tool_mgr, runtime="sandbox")
assert isinstance(toolset, ToolSet)
tool_names = {t.name for t in toolset.tools}
assert "sandbox_shell" in tool_names
assert "sandbox_python" in tool_names
# Ensure local tools are not wired for sandbox runtime
assert "local_shell" not in tool_names
assert "local_python" not in tool_names
# Distinct python tools for local vs sandbox runtimes
local_toolset = module._get_runtime_computer_tools(tool_mgr=tool_mgr, runtime="local")
local_python = {t for t in local_toolset.tools if "python" in t.name}.pop()
sandbox_python = {t for t in toolset.tools if "python" in t.name}.pop()
assert local_python is not sandbox_python
```
These edits assume:
1. `_apply_local_env_tools` has a signature compatible with `module._apply_local_env_tools(req=req, context=mock_context)` and that `req.func_tool` is the attribute used for the function tool set. If the actual signature differs (for example, requiring `config` or different parameter names), adjust the test calls accordingly to match the real function signature and request object shape.
2. `ToolSet` and `MagicMock` are already imported in `tests/unit/test_astr_main_agent.py`, as they are used in the existing `_apply_sandbox_tools` tests. If they are not, you will need to add appropriate imports at the top of the file.
3. `_get_runtime_computer_tools` accepts `tool_mgr` and `runtime` keyword arguments and returns a `ToolSet`. If the real function signature differs (e.g., positional-only parameters or additional config/context args), update the test calls and the `_FakeToolManager` to match the way the production code actually requests tools from the tool manager.
4. In production, the names used when calling `tool_mgr.get_tool(...)` for local vs sandbox shell/python may differ from `"local_shell"`, `"local_python"`, `"sandbox_shell"`, `"sandbox_python"`. Adjust the keys in `_FakeToolManager.get_tool` and the assertions on `tool_names` to reflect the actual tool names used by `_get_runtime_computer_tools`.
</issue_to_address>
### Comment 4
<location path="tests/test_computer_tool_permissions.py" line_range="7-8" />
<code_context>
from astrbot.core.agent.run_context import ContextWrapper
-from astrbot.core.computer.tools.browser import BrowserExecTool
-from astrbot.core.computer.tools.neo_skills import GetExecutionHistoryTool
+from astrbot.core.tools.computer_tools.shipyard_neo.browser import BrowserExecTool
+from astrbot.core.tools.computer_tools.shipyard_neo.neo_skills import (
+ GetExecutionHistoryTool,
+)
</code_context>
<issue_to_address>
**suggestion (testing):** Consider permission tests for the new filesystem tools (upload/download and local read/write/edit/search restrictions)
This module currently only checks that browser and Neo skill tools respect `check_admin_permission`, but the new filesystem tools (`FileUploadTool`, `FileDownloadTool`, and the local read/write/edit/grep tools with restricted path rules`) add new permission surfaces. Please add tests that (1) ensure `FileUploadTool` and `FileDownloadTool` reject non‑admin callers, and (2) in a restricted local environment (`computer_use_require_admin=True`, non‑admin role) verify the read/write/edit/grep tools block paths outside allowed roots and return the expected errors. This will keep file‑access permissions covered alongside the existing browser/Neo tests.
Suggested implementation:
```python
import pytest
from types import SimpleNamespace
from astrbot.core.agent.run_context import ContextWrapper
from astrbot.core.tools.computer_tools.shipyard_neo.browser import BrowserExecTool
from astrbot.core.tools.computer_tools.shipyard_neo.neo_skills import (
GetExecutionHistoryTool,
FileUploadTool,
FileDownloadTool,
)
from astrbot.core.tools.computer_tools.shipyard_neo.local_fs import (
LocalReadTool,
LocalWriteTool,
LocalEditTool,
LocalSearchTool,
)
```
```python
class _FakeBrowser:
return SimpleNamespace(browser=_FakeBrowser())
monkeypatch.setattr(
"astrbot.core.tools.computer_tools.shipyard_neo.browser.get_booter",
_fake_get_booter,
)
def test_file_upload_download_require_admin(monkeypatch):
"""
Ensure FileUploadTool and FileDownloadTool reject non‑admin callers
when computer_use_require_admin=True.
"""
# Non‑admin context with computer use locked down
ctx = ContextWrapper(
role="user",
computer_use_require_admin=True,
)
upload_tool = FileUploadTool()
download_tool = FileDownloadTool()
# Upload should be rejected for non‑admin
with pytest.raises(PermissionError):
upload_tool.execute(
ctx,
file_path="/tmp/non_admin_upload.txt",
content=b"test",
)
# Download should be rejected for non‑admin
with pytest.raises(PermissionError):
download_tool.execute(
ctx,
file_path="/tmp/non_admin_download.txt",
)
def test_local_fs_tools_reject_outside_allowed_roots_for_non_admin(monkeypatch, tmp_path):
"""
In a restricted environment (computer_use_require_admin=True, non‑admin role),
local read/write/edit/search tools must block access outside allowed roots.
We simulate an allowed workspace root and verify that:
* access under that root is allowed; and
* access outside it raises/returns a permission error.
"""
# Create a fake "workspace" root and an "outside" path
workspace_root = tmp_path / "workspace"
workspace_root.mkdir()
inside_file = workspace_root / "inside.txt"
outside_file = tmp_path / "outside.txt"
ctx = ContextWrapper(
role="user",
computer_use_require_admin=True,
# Assuming ContextWrapper exposes a way to configure allowed local roots
allowed_local_roots=[str(workspace_root)],
)
read_tool = LocalReadTool()
write_tool = LocalWriteTool()
edit_tool = LocalEditTool()
search_tool = LocalSearchTool()
# Inside the allowed root should be permitted
write_tool.execute(
ctx,
path=str(inside_file),
content="hello",
)
result = read_tool.execute(
ctx,
path=str(inside_file),
)
assert "hello" in result
# Outside the allowed root should be blocked for all tools
for tool, kwargs in [
(read_tool, {"path": str(outside_file)}),
(write_tool, {"path": str(outside_file), "content": "blocked"}),
(edit_tool, {"path": str(outside_file), "patch": "blocked"}),
(search_tool, {"path": str(outside_file), "pattern": "blocked"}),
]:
with pytest.raises(PermissionError):
tool.execute(ctx, **kwargs)
```
The above edits assume several details that should be aligned to your existing code:
1. **Tool import paths and names**
* Confirm the concrete module and class names for the filesystem tools:
* `FileUploadTool`, `FileDownloadTool`
* `LocalReadTool`, `LocalWriteTool`, `LocalEditTool`, `LocalSearchTool`
* If they live in different modules or have different names, adjust the import block accordingly.
2. **Tool invocation API**
* The tests use `.execute(ctx, ...)` as a generic call surface.
* If your tools instead use a different method (`.run`, `.invoke`, `__call__`, etc.) or a different signature, update the calls in both tests to match your existing browser/Neo skill tests.
3. **ContextWrapper configuration**
* The tests assume `ContextWrapper`:
* accepts `role` and `computer_use_require_admin` arguments, and
* accepts an `allowed_local_roots` list to restrict local filesystem access.
* If the actual configuration is different (e.g. `user_role` vs `role`, or allowed roots live in `ctx.config`), adjust the `ContextWrapper(...)` construction accordingly.
4. **Error type / behavior**
* The tests use `pytest.raises(PermissionError)` as the expected behavior when a permission check fails.
* If your implementation uses a project-specific exception class or returns an error object / result instead of raising, update the assertions to match:
* Either change `PermissionError` to your concrete exception type, or
* Replace the `pytest.raises(...)` blocks with result assertions (e.g. checking an `.error` field or error code/message).
5. **Result structure for read/search tools**
* The `LocalReadTool` assertion assumes `read_tool.execute(...)` returns a string or an object whose string representation includes the file content.
* If your tool returns a structured object (e.g. `{"content": "..."}`), update the assertion to inspect the correct field.
6. **Existing browser/Neo fixtures**
* If the browser/Neo tests in this file already use shared fixtures or helper functions for constructing `ContextWrapper` instances with admin/non‑admin roles, refactor the new tests to use the same fixtures to keep style and behavior consistent.
Once these adjustments are made to match your actual tool and context APIs, the tests will cover the new file‑system permission surfaces in parity with the existing browser and Neo skill permission tests.
</issue_to_address>
### Comment 5
<location path="astrbot/core/tools/computer_tools/fs.py" line_range="166" />
<code_context>
+ )
+
+
+@builtin_tool
+@dataclass
+class FileReadTool(FunctionTool):
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the filesystem tools to centralize path/runtime policy, shared validation, and escaped-text handling so their behavior is easier to understand and maintain.
You can significantly cut repetition and make the behavior easier to reason about by centralizing the path/runtime policy and the common fs+path boilerplate. The functionality can stay identical; only wiring changes.
### 1. Centralize runtime/restriction/path logic
Right now each tool does:
- compute `local_env = _is_local_runtime(context)`
- compute `restricted = _is_restricted_env(context)`
- per-tool path normalization (`_normalize_rw_path` vs `_normalize_search_paths`)
- duplicate `if local_env else path.strip()` branches
- call `get_booter(...)` with the same args.
You can encapsulate this in a small, testable helper class:
```python
from dataclasses import dataclass
@dataclass
class PathPolicy:
context: ContextWrapper[AstrAgentContext]
@property
def umo(self) -> str:
return str(self.context.context.event.unified_msg_origin)
@property
def local_env(self) -> bool:
return _is_local_runtime(self.context)
@property
def restricted(self) -> bool:
return _is_restricted_env(self.context)
def normalize_rw(self, path: str) -> str:
raw = path.strip()
if not raw:
raise ValueError("`path` must be a non-empty string.")
if not self.local_env:
return raw
return _normalize_rw_path(
raw,
restricted=self.restricted,
local_env=self.local_env,
umo=self.umo,
)
def normalize_grep_paths(self, path: str | None) -> list[str]:
if not self.local_env:
return [path.strip()] if path and path.strip() else ["."]
return GrepTool._normalize_search_paths( # or extract logic to standalone fn
GrepTool, # type: ignore[arg-type] - using method as function
path,
restricted=self.restricted,
local_env=self.local_env,
umo=self.umo,
)
```
Then a **single** helper to get booter + normalized path:
```python
async def get_fs_and_path(
context: ContextWrapper[AstrAgentContext],
path: str,
*,
for_grep: bool = False,
) -> tuple[PathPolicy, "SandboxBooter", str | list[str]]:
policy = PathPolicy(context)
sb = await get_booter(
context.context.context,
context.context.event.unified_msg_origin,
)
if for_grep:
normalized = policy.normalize_grep_paths(path)
else:
normalized = policy.normalize_rw(path)
return policy, sb, normalized
```
Usage in `FileReadTool.call` becomes much simpler and consistent:
```python
async def call(self, context, path, offset=None, limit=None) -> ToolExecResult:
try:
policy, sb, normalized_path = await get_fs_and_path(context, path)
offset, limit = self._validate_read_window(offset, limit)
return await read_file_tool_result(
sb,
local_mode=policy.local_env,
path=normalized_path,
offset=offset,
limit=limit,
)
except PermissionError as exc:
return f"Error: {exc}"
except Exception as exc:
logger.error(f"Error reading file: {exc}")
return f"Error reading file: {exc}"
```
`FileWriteTool` and `FileEditTool` can reuse the same helper, eliminating repeated `local_env/restricted` branching and manual `path.strip()`.
For `GrepTool`:
```python
async def call(..., path: str | None = None, ...) -> ToolExecResult:
normalized_pattern = pattern.strip()
if not normalized_pattern:
return "Error: `pattern` must be a non-empty string."
try:
policy = PathPolicy(context)
sb = await get_booter(
context.context.context,
context.context.event.unified_msg_origin,
)
search_paths = policy.normalize_grep_paths(path)
...
```
This keeps all existing behavior (including restrictions) but centralizes the policy and booter wiring.
### 2. Reuse generic window/context validators
`FileReadTool._validate_read_window` and `GrepTool._resolve_context_options` are generic and can live in a shared utility to reduce duplication and make future read-like tools reuse them:
```python
def validate_window(offset: int | None, limit: int | None) -> tuple[int | None, int | None]:
if offset is not None and offset < 0:
raise ValueError("`offset` must be greater than or equal to 0.")
if limit is not None and limit < 1:
raise ValueError("`limit` must be greater than or equal to 1.")
return offset, limit
def resolve_context_options(a: int | None, b: int | None, c: int | None) -> tuple[int | None, int | None]:
if c is not None and c < 0:
raise ValueError("`-C` must be greater than or equal to 0.")
if a is not None and a < 0:
raise ValueError("`-A` must be greater than or equal to 0.")
if b is not None and b < 0:
raise ValueError("`-B` must be greater than or equal to 0.")
resolved_after = c if a is None else a
resolved_before = c if b is None else b
return resolved_after, resolved_before
```
Then:
```python
# FileReadTool
offset, limit = validate_window(offset, limit)
# GrepTool
after_context, before_context = resolve_context_options(
kwargs.get("-A"),
kwargs.get("-B"),
kwargs.get("-C"),
)
```
### 3. Make escaped text decoding less ad hoc
`_decode_escaped_text` is currently manual and partial. You can keep the behavior but leverage Python’s existing escape decoding for better coverage:
```python
import codecs
def _decode_escaped_text(value: str) -> str:
# Preserve current behavior for already-raw text; only decode backslash escapes.
try:
# 'unicode_escape' handles \n, \r, \t, etc.
return codecs.decode(value, "unicode_escape")
except Exception:
# Fallback to the current simple replacement to avoid breaking behavior
return (
value.replace("\\r\\n", "\n")
.replace("\\n", "\n")
.replace("\\r", "\r")
.replace("\\t", "\t")
)
```
This keeps all existing callers intact but reduces the chance of surprising edge cases and documents a single decoding strategy in one place.
</issue_to_address>
### Comment 6
<location path="astrbot/core/computer/file_read_utils.py" line_range="388" />
<code_context>
+ return None
+
+
+async def read_file_tool_result(
+ booter: ComputerBooter,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider introducing small helper functions that encapsulate the local vs sandbox filesystem logic so `read_file_tool_result` only orchestrates probe/read/validate steps at a high level.
You can reduce a fair bit of complexity by consolidating the duplicated *local vs sandbox* flows into a small internal FS abstraction, without changing any external behavior.
Right now `read_file_tool_result` knows about:
- Local vs sandbox branching
- Script generation (`_build_*_script`)
- Script execution (`_exec_python_json`)
- Local helpers (`_probe_local_file`, `read_local_text_range`, `_read_local_image_base64`)
You can hide that behind a minimal internal API and make `read_file_tool_result` only deal with `probe`, `read_text`, and `read_image` results.
### 1. Unify probing paths
Instead of branching on `local_mode` in `read_file_tool_result` and calling either `_probe_local_file` or `_exec_python_json(..._build_probe_script...)`, introduce a single helper:
```python
async def _probe_path(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
) -> dict[str, str | int]:
if local_mode:
return await _probe_local_file(path)
return await _exec_python_json(
booter,
_build_probe_script(path),
action="file probe",
)
```
Then `read_file_tool_result` can simplify to:
```python
probe_payload = await _probe_path(
booter,
local_mode=local_mode,
path=path,
)
sample_b64 = str(probe_payload.get("sample_b64", "") or "")
sample = base64.b64decode(sample_b64) if sample_b64 else b""
size_bytes = int(probe_payload.get("size_bytes", 0) or 0)
probe = _probe_file(sample, size_bytes=size_bytes)
```
This keeps functionality identical but removes one explicit local/sandbox branch and script wiring from `read_file_tool_result`.
### 2. Unify text read paths
Do the same for text reads: move the local vs sandbox branching into a helper, so `read_file_tool_result` no longer needs to know about script generation:
```python
async def _read_text_range(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
encoding: str,
offset: int | None,
limit: int | None,
) -> str:
if local_mode:
return await read_local_text_range(
path,
encoding=encoding,
offset=offset,
limit=limit,
)
payload = await _exec_python_json(
booter,
_build_text_read_script(
path,
encoding=encoding,
offset=offset,
limit=limit,
),
action="text read",
)
return str(payload.get("content", "") or "")
```
Then in `read_file_tool_result`:
```python
content = await _read_text_range(
booter,
local_mode=local_mode,
path=path,
encoding=probe.encoding or "utf-8",
offset=offset,
limit=limit,
)
```
### 3. Unify image read paths
Same idea for image reads:
```python
async def _read_image_base64(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
) -> dict[str, str | int]:
if local_mode:
return await _read_local_image_base64(path)
return await _exec_python_json(
booter,
_build_image_read_script(path),
action="image read",
)
```
And in `read_file_tool_result`:
```python
image_payload = await _read_image_base64(
booter,
local_mode=local_mode,
path=path,
)
raw_base64_data = str(image_payload.get("base64", "") or "")
...
```
### 4. Benefit
These small adapters:
- Keep all existing behavior, including the Python-script-based sandbox integration.
- Remove `local_mode` branching and script details from `read_file_tool_result`, so it reads like:
1. `probe_path(...)`
2. `probe_file(...)`
3. `read_text_range(...)` or `read_image_base64(...)`
4. validate + format result
This makes the high‑level flow much easier to follow and localizes the local vs sandbox differences into three small helpers, without reverting or changing any feature.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return None | ||
|
|
||
|
|
||
| async def read_file_tool_result( |
Contributor
There was a problem hiding this comment.
issue (complexity): Consider introducing small helper functions that encapsulate the local vs sandbox filesystem logic so read_file_tool_result only orchestrates probe/read/validate steps at a high level.
You can reduce a fair bit of complexity by consolidating the duplicated local vs sandbox flows into a small internal FS abstraction, without changing any external behavior.
Right now read_file_tool_result knows about:
- Local vs sandbox branching
- Script generation (
_build_*_script) - Script execution (
_exec_python_json) - Local helpers (
_probe_local_file,read_local_text_range,_read_local_image_base64)
You can hide that behind a minimal internal API and make read_file_tool_result only deal with probe, read_text, and read_image results.
1. Unify probing paths
Instead of branching on local_mode in read_file_tool_result and calling either _probe_local_file or _exec_python_json(..._build_probe_script...), introduce a single helper:
async def _probe_path(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
) -> dict[str, str | int]:
if local_mode:
return await _probe_local_file(path)
return await _exec_python_json(
booter,
_build_probe_script(path),
action="file probe",
)Then read_file_tool_result can simplify to:
probe_payload = await _probe_path(
booter,
local_mode=local_mode,
path=path,
)
sample_b64 = str(probe_payload.get("sample_b64", "") or "")
sample = base64.b64decode(sample_b64) if sample_b64 else b""
size_bytes = int(probe_payload.get("size_bytes", 0) or 0)
probe = _probe_file(sample, size_bytes=size_bytes)This keeps functionality identical but removes one explicit local/sandbox branch and script wiring from read_file_tool_result.
2. Unify text read paths
Do the same for text reads: move the local vs sandbox branching into a helper, so read_file_tool_result no longer needs to know about script generation:
async def _read_text_range(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
encoding: str,
offset: int | None,
limit: int | None,
) -> str:
if local_mode:
return await read_local_text_range(
path,
encoding=encoding,
offset=offset,
limit=limit,
)
payload = await _exec_python_json(
booter,
_build_text_read_script(
path,
encoding=encoding,
offset=offset,
limit=limit,
),
action="text read",
)
return str(payload.get("content", "") or "")Then in read_file_tool_result:
content = await _read_text_range(
booter,
local_mode=local_mode,
path=path,
encoding=probe.encoding or "utf-8",
offset=offset,
limit=limit,
)3. Unify image read paths
Same idea for image reads:
async def _read_image_base64(
booter: ComputerBooter,
*,
local_mode: bool,
path: str,
) -> dict[str, str | int]:
if local_mode:
return await _read_local_image_base64(path)
return await _exec_python_json(
booter,
_build_image_read_script(path),
action="image read",
)And in read_file_tool_result:
image_payload = await _read_image_base64(
booter,
local_mode=local_mode,
path=path,
)
raw_base64_data = str(image_payload.get("base64", "") or "")
...4. Benefit
These small adapters:
- Keep all existing behavior, including the Python-script-based sandbox integration.
- Remove
local_modebranching and script details fromread_file_tool_result, so it reads like:
probe_path(...)probe_file(...)read_text_range(...)orread_image_base64(...)- validate + format result
This makes the high‑level flow much easier to follow and localizes the local vs sandbox differences into three small helpers, without reverting or changing any feature.
Member
改动
值得注意的点
|
Member
Author
这个分叉是预期的，但是都是 rg / grep 来进行搜索。 |
…ng, including workspace storage for long documents
…iting and read tool integration
…on and related help information
dosubot
bot
added
size:XL
Soulter
force-pushed
the
feat/fs-grep-read-edit
branch
from
3996649 to
31846cb
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Todo
Modifications / 改动点
Screenshots or Test Results / 运行截图或测试结果
Checklist / 检查清单
😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
/ 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。
👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
/ 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。
🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in
requirements.txtandpyproject.toml./ 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到
requirements.txt和pyproject.toml文件相应位置。😮 My changes do not introduce malicious code.
/ 我的更改没有引入恶意代码。
Summary by Sourcery
Add filesystem read/write/edit/grep tools across local and sandbox runtimes, introduce workspace-based path handling and safety limits, and refactor computer tools into a centralized builtin-tool framework.
New Features:
Bug Fixes:
Enhancements:
Build: