feat: AgentExecutor实现通用工具集——文件读写/搜索/会话搜索/todo/子代理委派+workspace_dir解析

pipeline_service/agent_loop_v2.py

@@ -20,6 +20,7 @@ pipeline-service v2: Agent Executor — 智能执行引擎
 import asyncio
 import json
 import logging
+import os
 import time
 from typing import AsyncGenerator, Dict, List, Optional
 
@@ -73,6 +74,7 @@ class AgentExecutor:
         self._auto_push_count: int = 0   # 限制 auto-inject 次数
         self._session_id: str = ""
         self._started_at: float = 0.0
+        self._todos: List[dict] = []     # 会话内任务清单 [{done, content}]
 
         # 懒加载
         self._tool_registry = None
@@ -290,17 +292,20 @@ class AgentExecutor:
         except ImportError:
             self._memory_store = None
 
-        # 加载 org_id（从项目上下文）
+        # 加载 org_id + workspace_dir（从项目上下文）
         if self.project_id:
             try:
                 from sqlor.dbpools import DBPools
                 db = DBPools()
                 async with db.sqlorContext("pipeline") as sor:
                     recs = await sor.sqlExe(
-                        "SELECT org_id FROM sd_projects WHERE id=${pid}$",
+                        "SELECT org_id, workspace_dir FROM sd_projects WHERE id=${pid}$",
                         {"pid": self.project_id})
                     if recs:
                         self.org_id = getattr(recs[0], 'org_id', '') or ''
+                        ws = getattr(recs[0], 'workspace_dir', '') or ''
+                        if ws:
+                            self.workspace_dir = ws
             except Exception:
                 pass
 
@@ -555,6 +560,14 @@ class AgentExecutor:
             "run_command": self._t_run_command,
             "check_progress": self._t_check_progress,
             "add_bug": self._t_add_bug,
+            # ── 通用工具集（Hermes CLI 能力子集）──
+            "read_file": self._t_read_file,
+            "write_file": self._t_write_file,
+            "list_files": self._t_list_files,
+            "search_files": self._t_search_files,
+            "session_search": self._t_session_search,
+            "todo": self._t_todo,
+            "delegate_subtask": self._t_delegate_subtask,
         }
 
         handler = handlers.get(tool_name)
@@ -1077,6 +1090,167 @@ class AgentExecutor:
         })
         return f"OK: 已提交Bug: {title}"
 
+    # ═══════════════════════════════════════════════════════
+    # 通用工具集（Hermes CLI 能力子集）
+    # ═══════════════════════════════════════════════════════
+
+    def _resolve_ws_path(self, path: str) -> str:
+        """解析相对路径为工作空间内绝对路径（越界返回 ''）。"""
+        import os
+        ws = self.workspace_dir or "/tmp/pipeline_ws"
+        full = os.path.abspath(os.path.join(ws, path or "."))
+        # 限制在 workspace 内
+        if full != ws and not full.startswith(ws.rstrip("/") + "/"):
+            return ""
+        return full
+
+    async def _t_read_file(self, sor, p, pid):
+        path = p.get("path", "")
+        if not path:
+            return "FAIL: 需要文件路径"
+        full = self._resolve_ws_path(path)
+        if not full:
+            return f"FAIL: 路径越界 {path}"
+        try:
+            if not os.path.isfile(full):
+                return f"FAIL: 文件不存在 {path}"
+            with open(full, encoding="utf-8") as f:
+                return f.read()[:30000]
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
+    async def _t_write_file(self, sor, p, pid):
+        path = p.get("path", "")
+        content = p.get("content", "")
+        if not path:
+            return "FAIL: 需要文件路径"
+        full = self._resolve_ws_path(path)
+        if not full:
+            return f"FAIL: 路径越界 {path}"
+        try:
+            os.makedirs(os.path.dirname(full), exist_ok=True)
+            with open(full, "w", encoding="utf-8") as f:
+                f.write(content or "")
+            return f"OK: 已写入 {path} ({len(content)} 字符)"
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
+    async def _t_list_files(self, sor, p, pid):
+        path = p.get("path", "") or "."
+        full = self._resolve_ws_path(path)
+        if not full:
+            return f"FAIL: 路径越界 {path}"
+        try:
+            if not os.path.isdir(full):
+                return f"FAIL: 目录不存在 {path}"
+            items = sorted(os.listdir(full))[:50]
+            lines = []
+            for name in items:
+                fp = os.path.join(full, name)
+                if name.startswith("."):
+                    continue
+                t = "DIR" if os.path.isdir(fp) else "FILE"
+                size = os.path.getsize(fp) if os.path.isfile(fp) else 0
+                lines.append(f"[{t}] {name} ({size}B)")
+            return "\n".join(lines) if lines else "(空目录)"
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
+    async def _t_search_files(self, sor, p, pid):
+        pattern = p.get("pattern", "")
+        if not pattern:
+            return "FAIL: 需要搜索关键词"
+        path = p.get("path", "") or "."
+        full = self._resolve_ws_path(path)
+        if not full:
+            return f"FAIL: 路径越界 {path}"
+        try:
+            from pipeline_service.agent_loop import _run_shell
+            # grep -rn，排除 .git 和 __pycache__，限制输出
+            r = await _run_shell(
+                f"grep -rn --include='*.py' --include='*.md' --include='*.json' --include='*.txt' "
+                f"--exclude-dir=.git --exclude-dir=__pycache__ '{pattern}' . 2>/dev/null | head -50",
+                full, timeout=30)
+            out = r.get("stdout", "").strip()
+            return out[:4000] if out else f"未找到匹配 '{pattern}' 的内容"
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
+    async def _t_session_search(self, sor, p, pid):
+        query = p.get("query", "").strip()
+        if not query:
+            return "FAIL: 需要搜索关键词"
+        try:
+            recs = await sor.sqlExe(
+                "SELECT role, content, created_at FROM pipeline_conversations "
+                "WHERE created_by=${u}$ AND iteration_id=${pid}$ AND content LIKE ${q}$ "
+                "ORDER BY created_at DESC LIMIT 10",
+                {"u": self.user_id, "pid": pid, "q": f"%{query}%"})
+            if not recs:
+                return f"未找到包含 '{query}' 的会话记录"
+            lines = []
+            for r in recs:
+                content = (getattr(r, "content", "") or "")[:300]
+                lines.append(f"[{getattr(r, 'role', '?')}] {content}")
+            return "\n".join(lines)
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
+    async def _t_todo(self, sor, p, pid):
+        action = p.get("action", "list")
+        content = (p.get("content", "") or "").strip()
+        if action == "add":
+            if not content:
+                return "FAIL: add 需要任务内容"
+            self._todos.append({"done": False, "content": content})
+            return f"OK: 已添加任务（共 {len(self._todos)} 项）"
+        if action == "done":
+            if not content:
+                return "FAIL: done 需要任务序号或内容"
+            for t in self._todos:
+                if content in t["content"] or content == str(self._todos.index(t) + 1):
+                    t["done"] = True
+                    return f"OK: 已完成任务 '{t['content']}'"
+            return f"未找到任务 '{content}'"
+        # list（默认）
+        if not self._todos:
+            return "任务清单为空"
+        lines = []
+        for i, t in enumerate(self._todos):
+            mark = "✅" if t["done"] else "⏳"
+            lines.append(f"{i + 1}. {mark} {t['content']}")
+        return "\n".join(lines)
+
+    async def _t_delegate_subtask(self, sor, p, pid):
+        goal = (p.get("goal", "") or "").strip()
+        context = (p.get("context", "") or "").strip()
+        if not goal:
+            return "FAIL: 需要子任务目标"
+        try:
+            sub = AgentExecutor(
+                config=self.config,
+                project_id=self.project_id,
+                user_id=self.user_id,
+                workspace_dir=self.workspace_dir,
+                model_name=self.model_name,
+            )
+            prompt = goal if not context else f"{goal}\n\n背景：{context}"
+            result_parts = []
+            async for chunk in sub.run(prompt):
+                try:
+                    data = json.loads(chunk)
+                except Exception:
+                    continue
+                t = data.get("type", "")
+                if t == "reply":
+                    result_parts.append(data.get("message", ""))
+                elif t == "tool_result":
+                    result_parts.append(data.get("result", ""))
+            out = "\n".join(x for x in result_parts if x).strip()
+            return out[:3000] or "(子任务无输出)"
+        except Exception as e:
+            return f"ERROR: {str(e)[:300]}"
+
     # ═══════════════════════════════════════════════════════
     # 上下文压缩
     # ═══════════════════════════════════════════════════════