From 835830200251284eb4fc659c8ddfee41322768df Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Tue, 1 Sep 2026 10:16:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(v3):=20T22-T23=20harness=20=E7=BA=A7?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E6=89=A9=E5=B1=95=20+=20=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E5=8C=BA=E9=80=89=E6=8B=A9=E5=90=8E=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- gateway/agent.py | 24 +++-- gateway/api.py | 99 +++++++++++++++++--- gateway/settings.py | 3 + router_system/tools.py | 199 +++++++++++++++++++++++++++++++++++++++- tests/test_agent_api.py | 76 +++++++++++++++ tests/test_tools.py | 63 +++++++++++++ 6 files changed, 440 insertions(+), 24 deletions(-) diff --git a/gateway/agent.py b/gateway/agent.py index 6eef4c4..21831a1 100644 --- a/gateway/agent.py +++ b/gateway/agent.py @@ -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: diff --git a/gateway/api.py b/gateway/api.py index e49b380..86a55b1 100644 --- a/gateway/api.py +++ b/gateway/api.py @@ -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(): diff --git a/gateway/settings.py b/gateway/settings.py index c836a38..414622f 100644 --- a/gateway/settings.py +++ b/gateway/settings.py @@ -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 命令超时 }, } diff --git a/router_system/tools.py b/router_system/tools.py index 3f72b41..e75dc48 100644 --- a/router_system/tools.py +++ b/router_system/tools.py @@ -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)} diff --git a/tests/test_agent_api.py b/tests/test_agent_api.py index 8a8c9a6..99f7139 100644 --- a/tests/test_agent_api.py +++ b/tests/test_agent_api.py @@ -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 diff --git a/tests/test_tools.py b/tests/test_tools.py index fec1e27..e3802d8 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -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