tzt/projectAIpopular
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v3): T22-T23 harness 级工具扩展 + 工作区选择后端

Browse Source 
- T22 工具内核：edit_file（old_string 唯一命中才替换，防误改）、search_files
  （跨文件内容搜索，跳过 .git/node_modules 与二进制大文件）、run_command
  （allow_shell 默认关；超时+输出截断+Windows CREATE_NO_WINDOW）
- T23 工作区选择（参考 deepseek-harness 打开文件夹体验）：/agent/fs 磁盘目录浏览
  （空 path 列 Windows 盘符）、/agent/workspaces 最近列表持久化、
  POST /agent 接受 workspace（须存在目录），运行状态记录所用工作区
- 新增测试 12 项，全量 274 passed
This commit is contained in:
tzt
2026-09-01 10:16:58 +08:00
parent 10bd4cc71d
commit 8358302002
6 changed files with 440 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/agent.py
+16 -8
View File
@@ -32,10 +32,13 @@ STATE_DONE = "done"
STATE_FAILED = "failed"
AGENT_SYSTEM_PROMPT = (
"你是端云协同 LLM 系统中的智能体（Agent）。你拥有工作区文件工具："
"list_dir（列目录）、read_file（读文件）、write_file（写文件）。"
"像程序员助手一样工作：先列目录/读文件了解现状，需要时再写文件；"
"任务完成或给出结论后，直接输出给用户的最终答复（中文，不要再调用工具）。"
"你是端云协同 LLM 系统中的智能体（Agent），正在操作用户选择的**真实项目工作目录**。"
"你拥有的工具：list_dir（列目录）、read_file（读文件）、write_file（写文件/新建）、"
"edit_file（精确替换编辑：old_string 须唯一匹配）、search_files（跨文件搜索内容）、"
"run_command（执行 shell 命令，仅当系统开启 allow_shell 时可用，否则不要尝试）。"
"像编程助手一样工作：先列目录/搜索了解项目结构，读文件核对原文后再用 edit_file 小步修改"
"（或 write_file 新建），需要时运行命令验证。任务完成或给出结论后，"
"直接输出给用户的最终答复（中文，不要再调用工具）。"
)
@@ -125,6 +128,7 @@ class AgentRunInfo:
prompt_tokens: int = 0
completion_tokens: int = 0
pool_id: str = ""
workspace: str = "" # 本次运行使用的工作区根目录（绝对路径）
asyncio_task: Optional[asyncio.Task] = field(default=None, repr=False)
def to_dict(self) -> Dict[str, Any]:
@@ -141,6 +145,7 @@ class AgentRunInfo:
"prompt_tokens": self.prompt_tokens,
"completion_tokens": self.completion_tokens,
"pool_id": self.pool_id,
"workspace": self.workspace,
}
@@ -163,12 +168,13 @@ class AgentService:
return self._dir(request_id) / "status.json"
# ---------- 注册与查询 ----------
def register(self, request_id: str, task: str, model: str, pool_id: str) -> Optional[AgentRunInfo]:
def register(self, request_id: str, task: str, model: str, pool_id: str,
workspace: str = "") -> Optional[AgentRunInfo]:
running = [r for r in self._runs.values() if r.state == STATE_RUNNING]
if len(running) >= self.max_running:
return None
info = AgentRunInfo(request_id=request_id, task=task, model=model,
pool_id=pool_id, started_at=time.time())
pool_id=pool_id, workspace=workspace, started_at=time.time())
self._runs[request_id] = info
self._dir(request_id).mkdir(parents=True, exist_ok=True)
self._write_status(info)
@@ -179,9 +185,11 @@ class AgentService:
# ---------- 执行 ----------
async def run(self, info: AgentRunInfo, chat: Any, workspace_dir: str | Path,
max_rounds: int = 8, token_cap: int = 0) -> None:
max_rounds: int = 8, token_cap: int = 0,
allow_shell: bool = False, shell_timeout_s: int = 20) -> None:
"""执行智能体任务（由调用方包成后台协程）。"""
tools = WorkspaceTools(workspace_dir)
tools = WorkspaceTools(workspace_dir, allow_shell=allow_shell,
shell_timeout_s=shell_timeout_s)
loop = ToolLoop(tools, chat, max_rounds=max_rounds, token_cap=token_cap,
on_event=self._make_event_writer(info))
try:
gateway/api.py
+84 -15
View File
@@ -501,11 +501,13 @@ try:
@app.post("/agent", tags=["agent"])
async def agent_run(req: dict):
"""提交智能体任务：{"task": "...", "pool_id": "可选模型条目"}。
"""提交智能体任务：{"task", "pool_id"?, "workspace"?}。
workspace 为用户选择的工作目录（绝对路径）；缺省用设置里的 agent.workspace_dir。
立即返回 request_id；过程事件经 GET /agent/{id}/stream (SSE) 推送。
"""
from gateway.agent import get_agent_service, new_request_id
from router_system.tools import WorkspaceTools
task = str((req or {}).get("task") or "").strip()
if not task:
@@ -514,23 +516,36 @@ try:
raise HTTPException(status_code=400, detail="task 过长（>8000）")
pool_id = str((req or {}).get("pool_id") or "")
s = settings_store().to_dict()
agent_cfg = s.get("agent", {})
ws_raw = str((req or {}).get("workspace") or "").strip()
if ws_raw:
ws_path = Path(ws_raw)
if not ws_path.exists():
raise HTTPException(status_code=400, detail=f"工作目录不存在: {ws_raw}")
if not ws_path.is_dir():
raise HTTPException(status_code=400, detail=f"不是目录: {ws_raw}")
workspace_dir = str(ws_path.resolve())
else:
workspace_dir = agent_cfg.get("workspace_dir", "agent_workspace")
chat, model, used_pool_id = _resolve_agent_chat(pool_id)
service = get_agent_service()
request_id = new_request_id()
info = service.register(request_id, task, model, used_pool_id)
info = service.register(request_id, task, model, used_pool_id,
workspace=workspace_dir)
if info is None:
raise HTTPException(status_code=503, detail="智能体同时运行任务已达上限")
s = settings_store().to_dict()
agent_cfg = s.get("agent", {})
async def _run():
try:
await service.run(
info, chat,
workspace_dir=agent_cfg.get("workspace_dir", "agent_workspace"),
workspace_dir=workspace_dir,
max_rounds=int(agent_cfg.get("max_rounds", 8)),
token_cap=int(agent_cfg.get("token_cap", 20000)),
allow_shell=bool(agent_cfg.get("allow_shell", False)),
shell_timeout_s=int(agent_cfg.get("shell_timeout_s", 20)),
)
except Exception as exc:
import traceback
@@ -541,7 +556,50 @@ try:
service._write_status(info)
info.asyncio_task = asyncio.create_task(_run())
return {"request_id": request_id, "status": "running", "model": model}
return {"request_id": request_id, "status": "running", "model": model,
"workspace": workspace_dir}
@app.get("/agent/fs", tags=["agent"])
async def agent_fs_browse(path: str = ""):
"""目录选择器：浏览本地文件系统（只列子目录，不读文件内容）。"""
from router_system.tools import browse_directories
return browse_directories(path)
@app.get("/agent/workspaces", tags=["agent"])
async def agent_workspaces():
"""当前工作区 + 最近打开列表。"""
s = settings_store().to_dict()
agent_cfg = s.get("agent", {})
current = agent_cfg.get("workspace_dir", "agent_workspace")
return {"current": current, "recent": list(agent_cfg.get("recent_workspaces", []))}
@app.post("/agent/workspaces", tags=["agent"])
async def agent_open_workspace(req: dict):
"""打开（或创建）一个工作目录：设为当前并记入最近列表。"""
path = str((req or {}).get("path") or "").strip()
create = bool((req or {}).get("create", False))
if not path:
raise HTTPException(status_code=400, detail="path 不能为空")
p = Path(path)
if not p.exists():
if not create:
raise HTTPException(status_code=400,
detail=f"目录不存在: {path}（可勾选“新建目录”）")
try:
p.mkdir(parents=True, exist_ok=True)
except OSError as e:
raise HTTPException(status_code=400, detail=f"创建失败: {e}")
elif not p.is_dir():
raise HTTPException(status_code=400, detail=f"不是目录: {path}")
resolved = str(p.resolve())
store = settings_store()
store.update({"agent": {"workspace_dir": resolved}})
merged = store.to_dict().get("agent", {})
recent = [w for w in merged.get("recent_workspaces", []) if w != resolved]
recent.insert(0, resolved)
store.update({"agent": {"recent_workspaces": recent[:8]}})
return {"ok": True, "current": resolved,
"recent": store.to_dict().get("agent", {}).get("recent_workspaces", [])}
@app.get("/agent/{request_id}/status", tags=["agent"])
async def agent_status(request_id: str):
@@ -585,22 +643,22 @@ try:
)
@app.get("/agent/workspace", tags=["agent"])
async def agent_workspace(path: str = ""):
"""列出智能体工作区（默认根目录；path 指定子目录）。越界返回 400。"""
async def agent_workspace(path: str = "", root: str = ""):
"""列出智能体工作区内容。root 可指定其他已选工作目录（默认用设置值）。越界/非法返回 400。"""
from router_system.tools import ToolError, WorkspaceTools
s = settings_store().to_dict()
tools = WorkspaceTools(s.get("agent", {}).get("workspace_dir", "agent_workspace"))
base = _agent_workspace_root(root)
tools = WorkspaceTools(base)
try:
return tools.list_dir(path)
except ToolError as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/agent/file", tags=["agent"])
async def agent_file(path: str):
"""读取智能体工作区内文件（前端预览，越界即 400）。"""
async def agent_file(path: str, root: str = ""):
"""读取智能体工作区内文件（前端预览，越界即 400）。root 同 /agent/workspace。"""
from router_system.tools import ToolError, WorkspaceTools
s = settings_store().to_dict()
tools = WorkspaceTools(s.get("agent", {}).get("workspace_dir", "agent_workspace"))
base = _agent_workspace_root(root)
tools = WorkspaceTools(base)
try:
result = tools.read_file(path)
except ToolError as e:
@@ -609,6 +667,17 @@ try:
raise HTTPException(status_code=404, detail=result.get("error", "读取失败"))
return result
def _agent_workspace_root(root: str = "") -> str:
"""解析工作区根：显式 root（须为已存在目录）> 设置值。"""
root = (root or "").strip()
if root:
p = Path(root)
if not p.is_dir():
raise HTTPException(status_code=400, detail=f"工作目录不存在: {root}")
return str(p.resolve())
s = settings_store().to_dict()
return s.get("agent", {}).get("workspace_dir", "agent_workspace")
# ---------------- 模型设置（用户可调整） ----------------
@app.get("/config", tags=["settings"])
async def get_config():
gateway/settings.py
+3
View File
@@ -38,8 +38,11 @@ DEFAULTS: Dict[str, Any] = {
},
"agent": {
"workspace_dir": "agent_workspace", # 智能体工作区根目录（越界即拒）
"recent_workspaces": [], # 最近打开的工作区（供快速切换）
"max_rounds": 8, # 工具循环轮数上限
"token_cap": 20000, # 单次智能体任务 token 熔断
"allow_shell": False, # 允许 run_command 执行 shell（默认关）
"shell_timeout_s": 20, # shell 命令超时
},
}
router_system/tools.py
+198 -1
View File
@@ -24,6 +24,15 @@ MAX_LIST_ENTRIES = 200
MAX_RESULT_CHARS = 8000
MAX_WRITE_CHARS = 200_000
# search_files 上限
SEARCH_MAX_MATCHES = 30
SEARCH_MAX_FILES = 400
SEARCH_MAX_FILE_BYTES = 512 * 1024
SEARCH_SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}
# run_command 上限
SHELL_OUTPUT_CHARS = 4000
# 默认循环上限
DEFAULT_MAX_ROUNDS = 8
@@ -71,11 +80,89 @@ TOOLS_SPEC: List[Dict[str, Any]] = [
},
},
},
{
"type": "function",
"function": {
"name": "edit_file",
"description": "对工作区内已有文件做精确替换编辑：old_string 必须在文件中恰好出现一次，"
"被替换为 new_string。适合小改动；大改用 write_file 整体重写。",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "工作区内的相对路径"},
"old_string": {"type": "string", "description": "要替换的原文（须唯一匹配）"},
"new_string": {"type": "string", "description": "替换后的新文"},
},
"required": ["path", "old_string", "new_string"],
},
},
},
{
"type": "function",
"function": {
"name": "search_files",
"description": "在工作区（或其子目录）内按关键词跨文件搜索文本内容，"
"返回匹配的文件/行号/行内容（自动跳过 .git、node_modules 等目录与二进制大文件）。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索的关键词（大小写敏感）"},
"path": {"type": "string", "description": "限定的子目录，默认整个工作区"},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "run_command",
"description": "在工作区根目录执行一条 shell 命令并返回退出码与输出（如运行测试、查看版本）。"
"仅当系统开启 allow_shell 时可用。",
"parameters": {
"type": "object",
"properties": {
"command": {"type": "string", "description": "要执行的命令行"},
},
"required": ["command"],
},
},
},
]
TOOL_NAMES = {t["function"]["name"] for t in TOOLS_SPEC}
def browse_directories(path: str = "") -> Dict[str, Any]:
"""目录选择器的本地文件系统浏览（只列目录，不读文件内容）。
path 为空时列出 Windows 盘符（POSIX 列根目录）。返回
{"ok", "path", "parent", "dirs": [名称]}；用于智能体"选择工作区"。
"""
import os
if not path or not path.strip():
if os.name == "nt":
drives = [f"{c}:\\" for c in "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
if os.path.exists(f"{c}:\\")]
return {"ok": True, "path": "", "parent": "", "dirs": drives}
return {"ok": True, "path": "/", "parent": "",
"dirs": sorted(os.listdir("/"))}
p = Path(path).resolve()
if not p.exists():
return {"ok": False, "error": f"路径不存在: {path}"}
if not p.is_dir():
return {"ok": False, "error": f"不是目录: {path}"}
dirs = []
for child in sorted(p.iterdir(), key=lambda c: c.name.lower()):
try:
if child.is_dir():
dirs.append(child.name)
except OSError:
continue # 无权限/符号链接坏点，跳过
parent = str(p.parent) if p.parent != p else ""
return {"ok": True, "path": str(p), "parent": parent, "dirs": dirs}
class ToolError(Exception):
"""工具执行失败（路径越界/不存在/参数非法）。"""
@@ -87,9 +174,12 @@ class WorkspaceTools:
（根目录自身允许），否则抛 ToolError——防 ../ 越界与绝对路径逃逸。
"""
def __init__(self, root: str | Path):
def __init__(self, root: str | Path,
allow_shell: bool = False, shell_timeout_s: int = 20):
self.root = Path(root).resolve()
self.root.mkdir(parents=True, exist_ok=True)
self.allow_shell = bool(allow_shell)
self.shell_timeout_s = max(1, int(shell_timeout_s))
# ---------- 路径关押 ----------
def resolve(self, rel_path: str) -> Path:
@@ -147,6 +237,103 @@ class WorkspaceTools:
p.write_text(content, encoding="utf-8")
return {"ok": True, "path": rel_path, "bytes_written": len(content.encode("utf-8"))}
def edit_file(self, rel_path: str, old_string: str, new_string: str) -> Dict[str, Any]:
"""精确替换编辑：old_string 必须在文件中恰好出现一次（harness 式安全编辑）。"""
if not old_string:
return {"ok": False, "error": "old_string 不能为空"}
if len(old_string) > MAX_READ_CHARS:
return {"ok": False, "error": "old_string 过长（先 read_file 分段定位）"}
p = self.resolve(rel_path)
if not p.exists() or not p.is_file():
return {"ok": False, "error": f"文件不存在: {rel_path}"}
try:
text = p.read_text(encoding="utf-8")
except UnicodeDecodeError:
return {"ok": False, "error": f"非文本文件: {rel_path}"}
count = text.count(old_string)
if count == 0:
return {"ok": False, "error": "old_string 未在文件中找到（先 read_file 核对原文）"}
if count > 1:
return {"ok": False,
"error": f"old_string 出现 {count} 次（要求唯一）；请扩大上下文使其唯一"}
new_text = text.replace(old_string, new_string, 1)
p.write_text(new_text, encoding="utf-8")
return {
"ok": True, "path": rel_path,
"replaced": 1,
"changed_chars": len(new_text) - len(text),
}
def search_files(self, query: str, rel_path: str = "") -> Dict[str, Any]:
"""跨文件文本搜索（跳过依赖/构建目录与二进制大文件，限量返回）。"""
if not query:
return {"ok": False, "error": "query 不能为空"}
base = self.resolve(rel_path or "")
if not base.exists() or not base.is_dir():
return {"ok": False, "error": f"目录不存在: {rel_path}"}
matches: List[Dict[str, Any]] = []
scanned = 0
truncated = False
for p in sorted(base.rglob("*")):
if len(matches) >= SEARCH_MAX_MATCHES:
truncated = True
break
if not p.is_file():
continue
rel_parts = p.relative_to(base).parts
if any(part in SEARCH_SKIP_DIRS for part in rel_parts):
continue
try:
if p.stat().st_size > SEARCH_MAX_FILE_BYTES:
continue
text = p.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
continue # 二进制/不可读，跳过
scanned += 1
if scanned > SEARCH_MAX_FILES:
truncated = True
break
for lineno, line in enumerate(text.splitlines(), 1):
if query in line:
rel = Path(*p.relative_to(self.root).parts).as_posix()
matches.append({
"file": rel, "line": lineno,
"text": line.strip()[:300],
})
if len(matches) >= SEARCH_MAX_MATCHES:
truncated = True
break
return {"ok": True, "query": query, "matches": matches,
"scanned_files": scanned, "truncated": truncated}
def run_command(self, command: str) -> Dict[str, Any]:
"""在工作区根目录执行 shell 命令（默认关闭，allow_shell 开启后可用）。"""
if not self.allow_shell:
return {"ok": False,
"error": "run_command 未启用（系统设置 allow_shell 为关）。"
"请让用户在智能体页打开「允许执行命令」后重试。"}
command = (command or "").strip()
if not command:
return {"ok": False, "error": "command 不能为空"}
import subprocess
creationflags = 0x08000000 if __import__("os").name == "nt" else 0 # CREATE_NO_WINDOW
try:
proc = subprocess.run(
command, shell=True, cwd=str(self.root), capture_output=True,
timeout=self.shell_timeout_s, creationflags=creationflags,
)
out = (proc.stdout or b"").decode("utf-8", errors="replace")
err = (proc.stderr or b"").decode("utf-8", errors="replace")
combined = (out + ("\n[stderr]\n" + err if err.strip() else "")).strip()
if len(combined) > SHELL_OUTPUT_CHARS:
combined = combined[:SHELL_OUTPUT_CHARS] + "…(输出截断)"
return {"ok": True, "exit_code": proc.returncode,
"output": combined or "(无输出)", "command": command}
except subprocess.TimeoutExpired:
return {"ok": False, "error": f"命令超时（>{self.shell_timeout_s}s），已终止: {command}"}
except OSError as e:
return {"ok": False, "error": f"命令执行失败: {type(e).__name__}: {e}"}
# ---------- 统一执行入口 ----------
def execute(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""按名字执行工具；任何异常折叠为 {"ok": False, "error": ...}。"""
@@ -158,6 +345,16 @@ class WorkspaceTools:
if name == "write_file":
return self.write_file(
str(arguments.get("path", "")), str(arguments.get("content", "")))
if name == "edit_file":
return self.edit_file(
str(arguments.get("path", "")),
str(arguments.get("old_string", "")),
str(arguments.get("new_string", "")))
if name == "search_files":
return self.search_files(
str(arguments.get("query", "")), str(arguments.get("path", "")))
if name == "run_command":
return self.run_command(str(arguments.get("command", "")))
return {"ok": False, "error": f"未知工具: {name}"}
except ToolError as e:
return {"ok": False, "error": str(e)}
tests/test_agent_api.py
+76
View File
@@ -168,3 +168,79 @@ def test_agent_task_validation(agent_env, client):
def test_agent_404(agent_env, client):
assert client.get("/agent/ghost/status").status_code == 404
assert client.get("/agent/ghost/events").json() == []
# ---------------- 工作区选择（T23） ----------------
def test_agent_run_with_selected_workspace(agent_env, client, tmp_path):
"""显式 workspace 应成为本次运行的工作目录（文件写进去，状态记录目录）。"""
target = tmp_path / "my_project"
target.mkdir()
agent_env["set_script"]([
{"content": None,
"tool_calls": [{"id": "c1", "name": "write_file",
"arguments": {"path": "build.py", "content": "print('ok')"}}],
"usage": {"prompt_tokens": 10, "completion_tokens": 2}},
{"content": "已写入 build.py。", "tool_calls": [], "usage": {}},
])
r = client.post("/agent", json={"task": "写 build.py", "workspace": str(target)})
assert r.status_code == 200
rid = r.json()["request_id"]
info = _wait_done(agent_env["service"], rid)
assert info.state == "done"
assert (target / "build.py").read_text(encoding="utf-8") == "print('ok')"
st = client.get(f"/agent/{rid}/status").json()
assert st["workspace"] == str(target.resolve())
def test_agent_workspace_not_exists(agent_env, client, tmp_path):
r = client.post("/agent", json={"task": "t", "workspace": str(tmp_path / "ghost")})
assert r.status_code == 400
assert "不存在" in r.json()["detail"]
def test_workspace_open_and_recent(agent_env, client, tmp_path):
"""打开目录：设为当前 + 记入最近列表；支持 create 新建。"""
d1 = tmp_path / "proj_a"
d1.mkdir()
r1 = client.post("/agent/workspaces", json={"path": str(d1)})
assert r1.status_code == 200
assert r1.json()["current"] == str(d1.resolve())
assert str(d1.resolve()) in r1.json()["recent"]
# create 新建
new_dir = tmp_path / "proj_b" / "nested"
r2 = client.post("/agent/workspaces", json={"path": str(new_dir), "create": True})
assert r2.status_code == 200
assert new_dir.is_dir()
assert r2.json()["current"] == str(new_dir.resolve())
# 不存在且不建 -> 400
r3 = client.post("/agent/workspaces", json={"path": str(tmp_path / "nope")})
assert r3.status_code == 400
# 列表端点
lst = client.get("/agent/workspaces").json()
assert lst["current"] == str(new_dir.resolve())
assert len(lst["recent"]) >= 2
def test_fs_browse_endpoint(agent_env, client, tmp_path):
r = client.get("/agent/fs", params={"path": str(tmp_path)})
assert r.status_code == 200
assert r.json()["ok"] is True
assert "dirs" in r.json()
r2 = client.get("/agent/fs", params={"path": str(tmp_path / "nope")})
assert r2.json()["ok"] is False
def test_agent_workspace_and_file_accept_root(agent_env, client, tmp_path):
"""浏览/读取端点可指定 root（选中工作区）。"""
other = tmp_path / "other_ws"
other.mkdir()
(other / "x.txt").write_text("外部工作区", encoding="utf-8")
ls = client.get("/agent/workspace", params={"root": str(other)}).json()
assert ls["ok"] is True
assert any(e["name"] == "x.txt" for e in ls["entries"])
f = client.get("/agent/file", params={"path": "x.txt", "root": str(other)}).json()
assert f["content"] == "外部工作区"
# 非法 root -> 400
r = client.get("/agent/workspace", params={"root": str(tmp_path / "nope")})
assert r.status_code == 400
tests/test_tools.py
+63
View File
@@ -174,3 +174,66 @@ def test_toolloop_chat_error(ws):
result = asyncio_run(loop.run("任务"))
assert result["reason"] == "error"
assert "RuntimeError" in result["error"]
# ---------------- 扩展工具（T22）：edit_file / search_files / run_command ----------------
def test_edit_file_unique_replace(ws):
ws.write_file("app.py", "def main():\n print('v1')\n return 0\n")
r = ws.execute("edit_file", {"path": "app.py", "old_string": "print('v1')",
"new_string": "print('v2 — 已修复')"})
assert r["ok"] is True and r["replaced"] == 1
assert "print('v2 — 已修复')" in ws.read_file("app.py")["content"]
def test_edit_file_rejects_ambiguous_and_missing(ws):
ws.write_file("dup.txt", "abc-abc")
r1 = ws.execute("edit_file", {"path": "dup.txt", "old_string": "abc", "new_string": "x"})
assert r1["ok"] is False and "2 次" in r1["error"]
r2 = ws.execute("edit_file", {"path": "dup.txt", "old_string": "zzz", "new_string": "x"})
assert r2["ok"] is False and "未在文件中找到" in r2["error"]
assert ws.read_file("dup.txt")["content"] == "abc-abc" # 原文未被破坏
def test_search_files(ws):
ws.write_file("a.py", "DEFAULT_PORT = 8000\n")
ws.write_file("docs/note.md", "端口 8000 是默认值\n")
ws.write_file("node_modules/pkg/index.js", "port 8000\n") # 应被跳过
r = ws.execute("search_files", {"query": "8000"})
assert r["ok"] is True
files = {m["file"] for m in r["matches"]}
assert files == {"a.py", "docs/note.md"}
assert all("node_modules" not in f for f in files)
# 空查询
assert ws.execute("search_files", {"query": ""})["ok"] is False
def test_run_command_disabled_by_default(ws):
r = ws.execute("run_command", {"command": "echo hi"})
assert r["ok"] is False and "allow_shell" in r["error"]
def test_run_command_enabled(tmp_path):
import sys
ws2 = WorkspaceTools(tmp_path / "ws2", allow_shell=True, shell_timeout_s=15)
r = ws2.execute("run_command", {"command": f'"{sys.executable}" -c "print(40+2)"'})
assert r["ok"] is True and r["exit_code"] == 0
assert "42" in r["output"]
def test_run_command_timeout(tmp_path):
import sys
ws2 = WorkspaceTools(tmp_path / "ws3", allow_shell=True, shell_timeout_s=2)
r = ws2.execute("run_command",
{"command": f'"{sys.executable}" -c "import time; time.sleep(30)"'})
assert r["ok"] is False and "超时" in r["error"]
def test_browse_directories(tmp_path):
from router_system.tools import browse_directories
(tmp_path / "sub").mkdir()
(tmp_path / "file.txt").write_text("x", encoding="utf-8")
r = browse_directories(str(tmp_path))
assert r["ok"] is True and r["dirs"] == ["sub"] # 只列目录不列文件
assert r["parent"] # 可以上级
assert browse_directories(str(tmp_path / "ghost"))["ok"] is False