diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 71eeb53..8dbe92d 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/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]}" + # ═══════════════════════════════════════════════════════ # 上下文压缩 # ═══════════════════════════════════════════════════════