diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 5371dab..143c6a8 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -1,32 +1,23 @@ -"""Agent loop — 角色agent循环认领执行任务（v3.3.0）。 +"""Agent loop — 角色agent循环认领执行任务（v3.4.0）。 -v3.3.0 新增： -- PM（项目经理）角色：审核交付件，决定通过/驳回 -- 任务链：requirement → design → develop → test → deploy - 前序角色完成后 → PM审核通过 → 自动创建下一角色任务 -- 产物文件输出：交付件写入项目工作目录 deliverables/{role}/ -- 工作目录自动创建：确保 workspace_dir 存在 +v3.4.0 新增： +- 所有agent具备文件读写 + git pull/push 能力 +- agent 产出实际代码文件到项目仓库，而非仅文档 +- PM agent 负责 clone 项目关联仓库、初始化工作区 +- develop agent 产出可直接运行的源码文件 +- 每个agent完成后自动 git commit + push -架构： -- 主agent（cockpit_chat）理解客户意图，开发类意图写入 pipeline_tasks（带 role）。 -- 各角色agent（需求/设计/开发/测试/运维）调 role_agent_run/role_agent_loop - 循环认领自己角色的任务并执行。 -- PM agent 调 pm_review_loop 循环审核处于 review 状态的任务。 -- 认领是原子的：UPDATE ... WHERE state='submitted' + claimed_by 令牌校验， - 多agent并发不会重复认领。 -- 角色agent缺信息时返回 need_info → 写入 pipeline_agent_questions， - 任务置 waiting，等主agent/客户回答后回到 submitted 重新认领继续。 -- 只认领没有步骤记录的任务（带步骤的 DAG 任务归 executor 管）。 +任务链：requirement → design → develop → test → deploy """ import asyncio import json import os +import subprocess import logging logger = logging.getLogger("pipeline.agent_loop") -# 角色别名归一化：前端/主agent传的角色名统一到认领用的规范名 ROLE_ALIASES = { 'developer': 'develop', 'dev': 'develop', 'development': 'develop', 'coding': 'develop', 'requirement': 'requirement', 'requirements': 'requirement', 'requirement_analysis': 'requirement', @@ -37,19 +28,24 @@ ROLE_ALIASES = { 'pm': 'pm', 'project_manager': 'pm', '项目经理': 'pm', 'manager': 'pm', } -# 任务链：前序角色完成后，PM审核通过 → 自动创建下一角色任务 -# None 表示链末端 ROLE_CHAIN = { 'requirement': 'design', 'design': 'develop', 'develop': 'test', 'test': 'deploy', - 'deploy': None, # 部署完成 = 链结束 + 'deploy': None, } -# PM 审核后任务状态 -TASK_REVIEW = 'review' # 角色agent完成，待PM审核 -TASK_APPROVED = 'approved' # PM审核通过 +TASK_REVIEW = 'review' +TASK_APPROVED = 'approved' + +# 允许的 shell 工作目录前缀（安全限制） +_ALLOWED_WORKDIRS = [ + os.path.expanduser('~/pipeline_ws'), + '/d/pipeline/workspaces', + '/tmp/pipeline_workspaces', + '/tmp/pipeline_ws', +] def _normalize_role(role): @@ -57,66 +53,6 @@ def _normalize_role(role): return ROLE_ALIASES.get(r, r) -ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。 - -当前认领到的任务： -标题：__TITLE__ -参数：__PARAMS__ -__QNA__ - -## 项目信息 -- 工作目录：__WORKSPACE__ -- 你的产出物将保存到：__DELIVERABLE_DIR__ - -请站在你的角色立场完成这项工作。 - -## 产出规范 -1. 产出内容必须完整、可交付、可直接使用 -2. 产出格式：根据角色不同，产出对应格式： - - requirement（需求分析）：需求规格文档（Markdown格式） - - design（设计）：设计文档 / 架构图描述（Markdown格式） - - develop（开发）：可运行的代码 + 代码说明 - - test（测试）：测试用例 + 测试报告（Markdown格式） - - deploy（部署）：部署文档 / 配置 / 脚本 - -输出要求——返回纯 JSON（不要 markdown 包裹）： -- 能完成时：{"status":"done","deliverable_type":"文档类型(requirement_doc/design_doc/code/test_report/deploy_doc)","result":"完整交付内容","summary":"一段话概述产出的关键结论"} -- 缺少关键信息无法继续时：{"status":"need_info","question":"要向主agent/客户提出的具体问题","partial":"已完成的部分"}""" - -PM_PROMPT = """你是软件开发产线项目中的「项目经理（PM）」角色agent。 - -你的职责是审核其他角色agent产出的交付件，确保质量和完整性，并推动项目前进。 - -## 任务链 -项目按照以下阶段推进： -1. requirement（需求分析）→ 2. design（设计）→ 3. develop（开发）→ 4. test（测试）→ 5. deploy（部署） - -当前待审核的任务： -标题：__TITLE__ -角色：__ROLE__ -交付件内容： -__DELIVERABLE__ - -## 你的决策 -请基于交付件内容做出判断，输出纯JSON（不要markdown包裹）： - -1. 审核通过，启动下一阶段： - {"status":"approved","comment":"审核意见","next_task_title":"下一阶段任务标题","next_task_description":"下一阶段任务详细描述"} - -2. 审核不通过，需要修改： - {"status":"rejected","comment":"驳回原因和具体修改要求","questions":"需要角色agent澄清的具体问题"} - -3. 项目已完成（deploy阶段通过后）： - {"status":"completed","comment":"项目完成总结"} - -## 审核标准 -- requirement：需求是否清晰、完整、可量化 -- design：方案是否合理、技术选型是否恰当、是否覆盖需求 -- develop：代码是否可运行、是否遵循规范、是否实现设计 -- test：测试覆盖是否充分、是否发现关键问题 -- deploy：部署文档是否完整、配置是否正确""" - - def _get_db(): from sqlor.dbpools import DBPools db = DBPools() @@ -129,7 +65,6 @@ def _get_db(): def _resolve_workspace(workspace_dir): - """解析工作目录，如果不可写则回退到 ~/pipeline_ws。确保目录存在。""" if not workspace_dir: workspace_dir = os.path.expanduser('~/pipeline_ws/default') try: @@ -143,22 +78,186 @@ def _resolve_workspace(workspace_dir): return workspace_dir -async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted', match_role: bool = True): - """原子认领一条符合角色的任务。返回 task record 或 None。 +# ── Agent 工具函数 ── - 认领策略： - 1. 选出最早一条符合状态和角色的任务， - 且排除有步骤记录的 DAG 任务（那些归 executor）。 - 2. UPDATE ... SET state='running', claimed_by=令牌 WHERE id=? AND state=原状态 - 3. 用令牌回查校验——别的agent抢先时校验失败，返回 None。 - - match_role=False 时不过滤角色（PM审核所有角色的review任务）。 - """ +def _is_safe_workdir(workdir): + """检查目录是否在允许范围内。""" + wd = os.path.abspath(workdir) + for allowed in _ALLOWED_WORKDIRS: + awd = os.path.abspath(os.path.expanduser(allowed)) + if wd.startswith(awd): + return True + return False + + +async def _run_shell(command, workdir, timeout=120): + """安全执行 shell 命令。返回 {"rc": int, "stdout": str, "stderr": str}""" + cwd = os.path.abspath(workdir) if workdir else os.path.expanduser('~/pipeline_ws') + if not _is_safe_workdir(cwd): + return {"rc": -1, "stdout": "", "stderr": f"安全限制：目录 {cwd} 不在允许范围"} + if not os.path.isdir(cwd): + return {"rc": -1, "stdout": "", "stderr": f"目录不存在: {cwd}"} + try: + proc = await asyncio.create_subprocess_shell( + command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + cwd=cwd, executable='/bin/bash') + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return {"rc": -1, "stdout": "", "stderr": f"命令超时（{timeout}s）"} + return {"rc": proc.returncode or 0, + "stdout": stdout.decode('utf-8', 'replace')[-8000:], + "stderr": stderr.decode('utf-8', 'replace')[-4000:]} + except Exception as e: + return {"rc": -1, "stdout": "", "stderr": str(e)[:500]} + + +async def _write_code_file(filepath, content): + """写代码文件，自动创建父目录。""" + try: + d = os.path.dirname(filepath) + if d: + os.makedirs(d, exist_ok=True) + with open(filepath, 'w', encoding='utf-8') as f: + f.write(content or '') + return True, filepath + except Exception as e: + return False, str(e) + + +async def _git_setup(workdir): + """确保 git 用户已配置。""" + r = await _run_shell('git config user.email', workdir, 5) + if 'opencomputing' not in r.get('stdout', '') and '@' not in r.get('stdout', ''): + await _run_shell('git config user.email "pipeline@opencomputing.cn"', workdir, 5) + await _run_shell('git config user.name "Pipeline Agent"', workdir, 5) + + +async def _git_commit_push(workdir, commit_message, branch='main'): + """git add → commit → push。""" + await _git_setup(workdir) + r1 = await _run_shell('git add -A', workdir, 10) + r2 = await _run_shell(f'git diff --cached --stat', workdir, 10) + if not r2.get('stdout', '').strip(): + return {"rc": 0, "message": "没有变更需要提交"} + r3 = await _run_shell(f'git commit -m "{commit_message}"', workdir, 15) + if r3['rc'] != 0: + return {"rc": r3['rc'], "message": f"commit 失败: {r3['stderr'][:200]}"} + r4 = await _run_shell(f'git push origin {branch}', workdir, 30) + return {"rc": r4['rc'], "message": f"push {'成功' if r4['rc']==0 else '失败'}: {r4['stderr'][:200]}"} + + +async def _git_clone(repo_url, target_dir, branch='main'): + """克隆仓库到目标目录。已存在则 pull。""" + if os.path.isdir(os.path.join(target_dir, '.git')): + r = await _run_shell(f'git checkout {branch} && git pull origin {branch}', target_dir, 30) + return {"rc": r['rc'], "message": f"已存在，pull: {r['stdout'][:200]}"} + parent = os.path.dirname(target_dir) + os.makedirs(parent, exist_ok=True) + r = await _run_shell(f'git clone -b {branch} {repo_url} {target_dir}', parent, 120) + return {"rc": r['rc'], "message": f"clone: {r['stdout'][:200] if r['rc']==0 else r['stderr'][:200]}"} + + +# ── Prompts ── + +ROLE_PROMPT = """你是软件开发产线项目中的「__ROLE__」角色agent。 + +当前任务：__TITLE__ +参数：__PARAMS__ +__QNA__ + +## 工作环境 +- 项目根目录：__WORKSPACE__ +- 代码仓库位于：__WORKSPACE__/repos/ 下各子目录 +- 你有能力读写文件、执行git操作 + +## 产出要求 +__ROLE_SPECIFIC__ + +## 输出格式（严格JSON，不要markdown包裹） +{ + "status": "done", + "deliverable_type": "类型(requirement_doc/design_doc/code_files/test_report/deploy_doc)", + "summary": "一段话概述产出", + "result": "当产出是文档时，这里是完整文档内容", + "files": [ + {"path": "repos/仓库名/src/相对路径/文件.java", "content": "文件完整内容"}, + {"path": "repos/仓库名/pom.xml", "content": "..."} + ], + "git_commit_message": "feat: 简短描述本次变更", + "need_more_info": false, + "question": "" +} + +- 如果是文档类产出(requirement/design)，用 result 字段输出文档正文 +- 如果是代码产出(develop)，必须用 files 数组输出每个源码文件，每个元素含 path 和 content +- 测试产出用 result 输出测试报告，如有测试脚本则用 files +- 部署产出用 files 输出部署配置/脚本，用 result 输出部署说明 +- 缺少关键信息时：{"status":"need_info","question":"问题"} + +## Git 操作说明 +- 产出代码文件将自动写入 files 中指定的路径并 git commit + push +- 你只需要在 files 中指定正确的仓库路径即可""" + +ROLE_SPECIFICS = { + 'requirement': """你是需求分析师。输出完整的需求规格文档（Markdown），用 result 字段。 +文档应包含：项目概述、功能需求、非功能需求、用户角色、业务流程、验收标准。""", + + 'design': """你是系统设计师。输出完整的设计文档（Markdown），用 result 字段。 +文档应包含：系统架构、技术选型、数据库设计(ER图描述)、API接口设计、前端页面规划、安全设计。 +同时可输出 proto/接口定义文件到 files。""", + + 'develop': """你是开发工程师。你必须输出实际可运行的代码文件！ +用 files 数组输出每个源码文件，准确指定在仓库中的路径。 +files 中的 path 格式：repos/{仓库名}/src/... 或 repos/{仓库名}/pom.xml 等。 +每个文件包含完整可运行的代码。 +同时用 result 字段输出代码说明。""", + + 'test': """你是测试工程师。输出测试报告（Markdown）到 result 字段。 +如有测试脚本/用例代码，输出到 files。 +报告应包含：测试策略、测试用例、执行结果、Bug清单、覆盖率。""", + + 'deploy': """你是部署运维工程师。输出部署文档（Markdown）到 result 字段。 +必须输出部署配置/脚本到 files：Dockerfile/docker-compose.yml/nginx配置等。 +files 中的 path 格式：repos/{仓库名}/deploy/...""", +} + +PM_PROMPT = """你是项目经理（PM）。审核交付件并推动项目前进。 + +## 项目 +工作目录：__WORKSPACE__ +关联仓库：__REPOS__ + +## 待审核任务 +标题：__TITLE__ +角色：__ROLE__ +交付件摘要：__SUMMARY__ +交付件内容(前6000字)：__DELIVERABLE__ + +## 仓库状态 +__REPO_STATE__ + +## 决策 +输出纯JSON： +1. 通过：{"status":"approved","comment":"审核意见","next_task_title":"下阶段标题","next_task_description":"详细描述"} +2. 驳回：{"status":"rejected","comment":"驳回原因","questions":"修改要求"} +3. 完成：{"status":"completed","comment":"项目总结"} + +审核标准： +- requirement：需求是否清晰完整可量化 +- design：方案合理、覆盖需求、技术可行 +- develop：代码文件是否实际产出、目录结构是否合理、是否可编译运行 +- test：测试覆盖充分、发现问题记录完整 +- deploy：部署配置完整、可一键部署""" + + +# ── Agent 核心逻辑 ── + +async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True): role = _normalize_role(role) - if match_role: - where_role = "AND (role=${role}$ OR role='')" - else: - where_role = "" + where_role = "AND (role=${role}$ OR role='')" if match_role else "" recs = await sor.sqlExe( "SELECT id, title, params, pipeline_id, tenant_id, role FROM pipeline_tasks " "WHERE tenant_id=${tid}$ AND state=${state}$ " + where_role + " " @@ -167,53 +266,43 @@ async def _claim_task(sor, tenant_id: str, role: str, state: str = 'submitted', {"tid": tenant_id, "state": state, "role": role}) if not recs: return None - task = recs[0] task_id = task.id - from appPublic.uniqueID import getID claim_token = getID() await sor.sqlExe( "UPDATE pipeline_tasks SET state='running', claimed_by=${cb}$ " "WHERE id=${tid}$ AND state=${state}$", {"cb": claim_token, "tid": task_id, "state": state}) - - # 令牌校验：确认是本agent抢到的 check = await sor.sqlExe( "SELECT id FROM pipeline_tasks WHERE id=${tid}$ AND state='running' AND claimed_by=${cb}$", {"tid": task_id, "cb": claim_token}) if not check: - logger.info(f"claim lost race: task={task_id} role={role}") + logger.info(f"claim lost race: task={task_id}") return None return task async def _get_workspace_dir(sor, project_id): - """获取项目的工作目录，确保可写并存在。""" - recs = await sor.sqlExe( - "SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", - {"pid": project_id}) + recs = await sor.sqlExe("SELECT workspace_dir FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) ws = getattr(recs[0], 'workspace_dir', '') if recs else '' return _resolve_workspace(ws) -async def _build_qna_section(sor, task_id: str) -> str: - """把该任务已回答的问答历史注入prompt，让角色agent带着答案继续。""" +async def _build_qna_section(sor, task_id): from .questions import get_task_qna qna = await get_task_qna(task_id) if not qna: return "" - lines = ["历史问答（这些问题已得到回答，请结合答案继续工作）："] + lines = ["历史问答："] for item in qna: - src = item.get('answer_source') or '' - src_label = "客户" if src == "customer" else "主agent" - lines.append(f"问：{item.get('question', '')}") - lines.append(f"答（{src_label}）：{item.get('answer', '')}") + src = "客户" if item.get('answer_source') == "customer" else "主agent" + lines.append(f"问：{item.get('question','')}") + lines.append(f"答({src})：{item.get('answer','')}") return "\n".join(lines) -def _parse_result(raw: str) -> dict: - """解析角色agent的 LLM 返回。解析失败时按 done + 原文处理。""" +def _parse_result(raw): raw = (raw or "").strip() if raw.startswith("```"): raw = raw.split("\n", 1)[1].rsplit("```", 1)[0] @@ -226,42 +315,58 @@ def _parse_result(raw: str) -> dict: return {"status": "done", "result": raw} -async def _write_deliverable_file(workspace_dir, role, task_id, deliverable_type, content): - """将交付件写入项目工作目录的文件中。""" - role_dir = os.path.join(workspace_dir, 'deliverables', role) - os.makedirs(role_dir, exist_ok=True) - # 文件名：任务ID_类型.md - safe_type = (deliverable_type or 'deliverable').replace('/', '_') - filename = f"{task_id}_{safe_type}.md" - file_path = os.path.join(role_dir, filename) - try: - with open(file_path, 'w', encoding='utf-8') as f: - f.write(content or '') - logger.info(f"deliverable written: {file_path}") - return file_path - except Exception as e: - logger.error(f"failed to write deliverable file: {file_path} err={e}") - return '' - - async def _get_deliverable_content(sor, task_id): - """获取任务的最新交付件内容。""" recs = await sor.sqlExe( "SELECT content, deliverable_type FROM pipeline_deliverables " - "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", - {"tid": task_id}) + "WHERE task_id=${tid}$ ORDER BY created_at DESC LIMIT 1", {"tid": task_id}) if recs: return getattr(recs[0], 'content', '') or '', getattr(recs[0], 'deliverable_type', '') or '' return '', '' +async def _get_project_repos(sor, project_id): + """获取项目关联的所有仓库。""" + recs = await sor.sqlExe( + "SELECT id, repo_name, repo_url, default_branch, local_path FROM sd_project_repos " + "WHERE project_id=${pid}$", {"pid": project_id}) + return [{'id': getattr(r, 'id', ''), 'name': getattr(r, 'repo_name', ''), + 'url': getattr(r, 'repo_url', ''), 'branch': getattr(r, 'default_branch', 'main'), + 'local_path': getattr(r, 'local_path', '') or ''} for r in (recs or [])] + + +async def _setup_repos(sor, workspace_dir, project_id): + """PM：clone 所有项目关联仓库到 workspace/repos/。""" + repos = await _get_project_repos(sor, project_id) + results = [] + repos_dir = os.path.join(workspace_dir, 'repos') + os.makedirs(repos_dir, exist_ok=True) + for repo in repos: + target = os.path.join(repos_dir, repo['name']) + r = await _git_clone(repo['url'], target, repo['branch']) + results.append({'repo': repo['name'], **r}) + return results + + +async def _get_repo_state(workspace_dir): + """获取仓库当前状态（供PM审核时查看）。""" + repos_dir = os.path.join(workspace_dir, 'repos') + if not os.path.isdir(repos_dir): + return "暂无仓库" + lines = [] + for name in sorted(os.listdir(repos_dir)): + rp = os.path.join(repos_dir, name) + if os.path.isdir(rp) and os.path.isdir(os.path.join(rp, '.git')): + r = await _run_shell('git log --oneline -3', rp, 5) + lines.append(f"\n[{name}]") + lines.append(r.get('stdout', '') or '(空仓库)') + return "\n".join(lines) if lines else "仓库无提交记录" + + async def _get_next_role(current_role): - """获取任务链中的下一个角色。""" return ROLE_CHAIN.get(_normalize_role(current_role)) async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): - """为下一个角色创建任务。""" from appPublic.uniqueID import getID title = getattr(task, 'title', '') or '' params_str = getattr(task, 'params', '{}') or '{}' @@ -269,47 +374,26 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): params = json.loads(params_str) if isinstance(params_str, str) else params_str except (json.JSONDecodeError, TypeError): params = {} - new_title = f"{title}（{next_role}阶段）" - new_params = { - **params, - 'previous_role': _normalize_role(getattr(task, 'role', '')), - 'previous_task_id': getattr(task, 'id', ''), - 'pm_comment': pm_comment, - } - + new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')), + 'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment} new_task_id = getID() await sor.C('pipeline_tasks', { - 'id': new_task_id, - 'tenant_id': project_id, - 'pipeline_id': 'role_task', - 'owner_id': 'pm', - 'title': new_title, + 'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task', + 'owner_id': 'pm', 'title': new_title, 'params': json.dumps(new_params, ensure_ascii=False), - 'role': next_role, - 'state': 'submitted', - 'claimed_by': None, + 'role': next_role, 'state': 'submitted', 'claimed_by': None, }) - logger.info(f"_create_next_task: {next_role} task={new_task_id} from {getattr(task, 'id', '')}") + logger.info(f"_create_next_task: {next_role} task={new_task_id}") return new_task_id, new_title -async def role_agent_run(project_id: str, role: str, agent_id: str = None, model_name: str = None) -> dict: - """角色agent单次迭代：认领一条任务 → 执行 → 交付。 +# ── 角色 Agent ── - 非PM角色完成 → 任务置 review（等PM审核） - PM角色 → 由 pm_review_run 处理 - - Returns: - {"status": "idle"} 没有待办任务 - {"status": "completed", "task_id", "deliverable_id"} 完成交付（已进入review） - {"status": "need_info", "task_id", "question_id", "question"} 缺信息已提问 - {"status": "failed", "task_id", "error"} 执行失败 - """ +async def role_agent_run(project_id, role, agent_id=None, model_name=None): role = _normalize_role(role) if role == 'pm': - # PM不执行角色任务，走pm_review_run - return {"status": "idle", "message": "PM agent请使用 pm_review_run"} + return {"status": "idle", "message": "PM请使用 pm_review_run"} db = _get_db() async with db.sqlorContext("pipeline") as sor: @@ -320,74 +404,198 @@ async def role_agent_run(project_id: str, role: str, agent_id: str = None, model task_id = task.id title = getattr(task, "title", "") or "" params_str = getattr(task, "params", "{}") or "{}" - - # 获取工作目录 workspace_dir = await _get_workspace_dir(sor, project_id) - deliverable_dir = os.path.join(workspace_dir, 'deliverables', role) - os.makedirs(deliverable_dir, exist_ok=True) + # 确保 repos 目录存在 + repos_dir = os.path.join(workspace_dir, 'repos') + if not os.path.isdir(repos_dir): + os.makedirs(repos_dir, exist_ok=True) + + role_specific = ROLE_SPECIFICS.get(role, ROLE_SPECIFICS.get('develop', '')) qna_section = await _build_qna_section(sor, task_id) + prompt = (ROLE_PROMPT .replace('__ROLE__', role) .replace('__TITLE__', title) .replace('__PARAMS__', params_str) .replace('__QNA__', qna_section) .replace('__WORKSPACE__', workspace_dir) - .replace('__DELIVERABLE_DIR__', deliverable_dir)) + .replace('__ROLE_SPECIFIC__', role_specific)) from .llm_bridge import llm_call try: raw = await llm_call(prompt, model=model_name, temperature=0.4) except Exception as e: - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", - {"tid": task_id}) + await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) logger.error(f"role_agent_run llm failed: task={task_id} err={e}") return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} parsed = _parse_result(raw) - # 缺信息 → 提问回路：任务置 waiting，等主agent/客户回答 if parsed.get("status") == "need_info" and parsed.get("question"): from .questions import agent_ask qid = await agent_ask(project_id, task_id, role, parsed["question"], context={"partial": parsed.get("partial", ""), "title": title}) - return {"status": "need_info", "task_id": task_id, - "question_id": qid, "question": parsed["question"]} + return {"status": "need_info", "task_id": task_id, "question_id": qid, "question": parsed["question"]} - # 完成 → 写交付件到文件 + DB + # ── 处理产出 ── from appPublic.uniqueID import getID did = getID() result_text = parsed.get("result") or raw deliverable_type = parsed.get("deliverable_type") or role summary = parsed.get("summary", "") - # 写文件 - file_path = await _write_deliverable_file( - workspace_dir, role, task_id, deliverable_type, result_text) + # 写入实际代码文件（如果有 files 数组） + files_written = [] + code_files = parsed.get("files") or [] + if isinstance(code_files, list): + for f in code_files: + if isinstance(f, dict) and f.get("path") and f.get("content"): + abs_path = os.path.join(workspace_dir, f["path"]) + ok, msg = await _write_code_file(abs_path, f["content"]) + if ok: + files_written.append(abs_path) + logger.info(f"code file written: {abs_path}") + else: + logger.error(f"code file failed: {abs_path} err={msg}") + + # 写交付件文档 + file_path = '' + if result_text: + deliverable_dir = os.path.join(workspace_dir, 'deliverables', role) + os.makedirs(deliverable_dir, exist_ok=True) + safe_type = (deliverable_type or 'deliverable').replace('/', '_') + file_path = os.path.join(deliverable_dir, f"{task_id}_{safe_type}.md") + try: + with open(file_path, 'w', encoding='utf-8') as f: + f.write(result_text or '') + logger.info(f"deliverable written: {file_path}") + except Exception as e: + logger.error(f"deliverable write failed: {file_path} err={e}") + file_path = '' # 写DB await sor.C("pipeline_deliverables", { "id": did, "project_id": project_id, "task_id": task_id, - "deliverable_type": deliverable_type, - "title": title, "content": result_text, - "file_path": file_path, + "deliverable_type": deliverable_type, "title": title, + "content": result_text, "file_path": file_path, "quality_score": 80, "review_status": "pending", "created_by": agent_id or role, }) - # 置为 review 状态，等待PM审核（清除 claimed_by 让 PM poller 能认领） + # Git commit + push（如果有产出文件） + commit_msg = parsed.get("git_commit_message") or f"{role}: {title[:80]}" + git_result = {"rc": 0, "message": "无git操作"} + if files_written: + # 在 repos 目录下的每个 git 仓库各自 commit + for repo_name in os.listdir(repos_dir): + repo_path = os.path.join(repos_dir, repo_name) + if os.path.isdir(os.path.join(repo_path, '.git')): + git_result = await _git_commit_push(repo_path, commit_msg) + + # 置为 review 状态 await sor.sqlExe( "UPDATE pipeline_tasks SET state=${st}$, claimed_by=NULL WHERE id=${tid}$", {"st": TASK_REVIEW, "tid": task_id}) - logger.info(f"role_agent_run completed(review): task={task_id} role={role} deliverable={did} file={file_path}") + logger.info(f"role_agent_run completed: task={task_id} role={role} " + f"deliverable={did} files={len(files_written)} git={git_result.get('rc',-1)}") return {"status": "completed", "task_id": task_id, "deliverable_id": did, - "next_state": TASK_REVIEW, "file_path": file_path} + "files_written": len(files_written), "git_result": git_result} -async def role_agent_loop(project_id: str, role: str, agent_id: str = None, - model_name: str = None, max_iterations: int = 10) -> list: - """角色agent循环：连续认领执行，直到没有任务、需要提问或达到上限。""" +# ── PM 审核 ── + +async def pm_review_run(project_id, agent_id=None, model_name=None): + db = _get_db() + async with db.sqlorContext("pipeline") as sor: + task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False) + if not task: + return {"status": "idle", "message": "没有待审核任务"} + + task_id = task.id + title = getattr(task, "title", "") or "" + task_role = _normalize_role(getattr(task, "role", "") or "") + workspace_dir = await _get_workspace_dir(sor, project_id) + + # 首次审核时，确保仓库已 clone + repos = await _get_project_repos(sor, project_id) + if repos: + await _setup_repos(sor, workspace_dir, project_id) + + deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) + if not deliverable_content: + await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) + return {"status": "rejected", "task_id": task_id, "reason": "没有交付件"} + + summary = deliverable_content[:200].replace('\n', ' ') + content_preview = deliverable_content[:6000] + if len(deliverable_content) > 6000: + content_preview += "\n...（截断）" + + repo_state = await _get_repo_state(workspace_dir) + repos_str = ", ".join([r['name'] for r in repos]) if repos else "无" + + prompt = (PM_PROMPT + .replace('__TITLE__', title) + .replace('__ROLE__', task_role) + .replace('__SUMMARY__', summary) + .replace('__DELIVERABLE__', content_preview) + .replace('__WORKSPACE__', workspace_dir) + .replace('__REPOS__', repos_str) + .replace('__REPO_STATE__', repo_state)) + + from .llm_bridge import llm_call + try: + raw = await llm_call(prompt, model=model_name, temperature=0.3) + except Exception as e: + await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id}) + return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} + + parsed = _parse_result(raw) + decision = parsed.get("status", "rejected") + comment = parsed.get("comment", "") + + if decision == "approved": + from appPublic.uniqueID import getID + pm_did = getID() + await sor.C("pipeline_deliverables", { + "id": pm_did, "project_id": project_id, "task_id": task_id, + "deliverable_type": "pm_review", + "title": f"PM审核：{title}", + "content": json.dumps(parsed, ensure_ascii=False), + "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', f"{task_id}_review.md"), + "quality_score": 100, "review_status": "approved", + "created_by": agent_id or "pm", + }) + await sor.sqlExe( + "UPDATE pipeline_deliverables SET review_status='approved', review_comment=${cm}$ " + "WHERE task_id=${tid}$", {"cm": comment, "tid": task_id}) + await sor.sqlExe( + "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", + {"st": TASK_APPROVED, "tid": task_id}) + + next_role = await _get_next_role(task_role) + if next_role: + next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) + return {"status": "approved", "task_id": task_id, + "next_task_id": next_tid, "next_role": next_role, "comment": comment} + else: + return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} + + elif decision == "rejected": + from .questions import agent_ask + rejection_q = parsed.get("questions") or comment or "交付件不满足要求" + await agent_ask(project_id, task_id, "pm", rejection_q, + context={"pm_comment": comment, "deliverable_type": deliverable_type}) + await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", {"tid": task_id}) + return {"status": "rejected", "task_id": task_id, "comment": comment, "question": rejection_q} + + else: + await sor.sqlExe("UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", {"tid": task_id}) + return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} + + +async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10): results = [] for _ in range(max_iterations): r = await role_agent_run(project_id, role, agent_id, model_name) @@ -398,133 +606,7 @@ async def role_agent_loop(project_id: str, role: str, agent_id: str = None, return results -# ── PM 审核 ── - -async def pm_review_run(project_id: str, agent_id: str = None, model_name: str = None) -> dict: - """PM审核单次迭代：认领一条 review 状态的任务 → 审核交付件 → 批准/驳回。 - - Returns: - {"status": "idle"} 没有待审核任务 - {"status": "approved", "task_id", "next_task_id", "next_role"} 审核通过，已创建下一阶段任务 - {"status": "rejected", "task_id"} 驳回，任务回到 submitted - {"status": "completed", "task_id"} 项目完成（链末端） - """ - db = _get_db() - async with db.sqlorContext("pipeline") as sor: - # 认领 review 状态的任务（不限角色，PM审核所有角色产出） - task = await _claim_task(sor, project_id, '', state=TASK_REVIEW, match_role=False) - if not task: - return {"status": "idle", "message": "没有待审核任务"} - - task_id = task.id - title = getattr(task, "title", "") or "" - task_role = _normalize_role(getattr(task, "role", "") or "") - - # 获取交付件内容 - deliverable_content, deliverable_type = await _get_deliverable_content(sor, task_id) - if not deliverable_content: - # 没有交付件 → 驳回 - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", - {"tid": task_id}) - return {"status": "rejected", "task_id": task_id, - "reason": "没有找到交付件内容，任务已退回"} - - # 截断过长内容（LLM有token上限） - content_preview = deliverable_content[:6000] - if len(deliverable_content) > 6000: - content_preview += "\n\n...（内容过长已截断）" - - prompt = (PM_PROMPT - .replace('__TITLE__', title) - .replace('__ROLE__', task_role) - .replace('__DELIVERABLE__', content_preview)) - - from .llm_bridge import llm_call - try: - raw = await llm_call(prompt, model=model_name, temperature=0.3) - except Exception as e: - logger.error(f"pm_review_run llm failed: task={task_id} err={e}") - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", - {"tid": task_id}) - return {"status": "failed", "task_id": task_id, "error": str(e)[:200]} - - parsed = _parse_result(raw) - decision = parsed.get("status", "rejected") - comment = parsed.get("comment", "") - - if decision == "approved": - # 获取工作目录 - workspace_dir = await _get_workspace_dir(sor, project_id) - - # 写PM审核记录到交付件 - from appPublic.uniqueID import getID - pm_did = getID() - await sor.C("pipeline_deliverables", { - "id": pm_did, "project_id": project_id, "task_id": task_id, - "deliverable_type": "pm_review", - "title": f"PM审核：{title}", - "content": json.dumps(parsed, ensure_ascii=False), - "file_path": os.path.join(workspace_dir, 'deliverables', 'pm', - f"{task_id}_review.md"), - "quality_score": 100, "review_status": "approved", - "created_by": agent_id or "pm", - }) - - # 更新原交付件审核状态 - await sor.sqlExe( - "UPDATE pipeline_deliverables SET review_status='approved', " - "review_comment=${cm}$ WHERE task_id=${tid}$", - {"cm": comment, "tid": task_id}) - - # 任务置为 approved - await sor.sqlExe( - "UPDATE pipeline_tasks SET state=${st}$ WHERE id=${tid}$", - {"st": TASK_APPROVED, "tid": task_id}) - - # 检查是否有下一阶段 - next_role = await _get_next_role(task_role) - if next_role: - next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) - logger.info(f"pm_review: approved task={task_id}, created next={next_tid} role={next_role}") - return {"status": "approved", "task_id": task_id, - "next_task_id": next_tid, "next_role": next_role, - "next_title": next_title, "comment": comment} - else: - # 链末端（deploy审核通过 = 项目完成） - logger.info(f"pm_review: project completed, task={task_id} role={task_role}") - return {"status": "completed", "task_id": task_id, - "comment": comment or "项目所有阶段已完成"} - - elif decision == "rejected": - # 驳回：任务回到 submitted，附带PM驳回原因作为question - from .questions import agent_ask - rejection_question = parsed.get("questions") or comment or "交付件不满足要求，请修改后重新提交" - await agent_ask(project_id, task_id, "pm", rejection_question, - context={"pm_comment": comment, "deliverable_type": deliverable_type}) - # 任务回到 submitted（通过回答问题的resume机制） - # 但agent_ask已经把任务置waiting，需要让主agent/客户确认后才能回到submitted - # 简化为：直接回 submitted - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='submitted' WHERE id=${tid}$", - {"tid": task_id}) - logger.info(f"pm_review: rejected task={task_id}") - return {"status": "rejected", "task_id": task_id, - "comment": comment, "question": rejection_question} - - else: - # completed（项目完成） - await sor.sqlExe( - "UPDATE pipeline_tasks SET state='completed' WHERE id=${tid}$", - {"tid": task_id}) - return {"status": "completed", "task_id": task_id, - "comment": comment or "项目完成"} - - -async def pm_review_loop(project_id: str, agent_id: str = None, - model_name: str = None, max_iterations: int = 20) -> list: - """PM审核循环：连续审核处于 review 状态的任务。""" +async def pm_review_loop(project_id, agent_id=None, model_name=None, max_iterations=20): results = [] for _ in range(max_iterations): r = await pm_review_run(project_id, agent_id, model_name) @@ -535,10 +617,8 @@ async def pm_review_loop(project_id: str, agent_id: str = None, return results -# ── 向后兼容别名（旧前端按 run_agent_loop/agent_loop 调用）── - +# 向后兼容 async def run_agent_loop(project_id, role_name, model_name=None): - """旧接口：单次迭代。role_name 为空时按 'dev' 处理。""" role = _normalize_role(role_name or "develop") if role == 'pm': return await pm_review_run(project_id, model_name=model_name) @@ -546,9 +626,7 @@ async def run_agent_loop(project_id, role_name, model_name=None): async def agent_loop(project_id, role_name, model_name=None, max_iterations=10): - """旧接口：循环执行。role_name 为空时按 'dev' 处理。""" role = _normalize_role(role_name or "develop") if role == 'pm': return await pm_review_loop(project_id, model_name=model_name, max_iterations=max_iterations) - return await role_agent_loop(project_id, role_name or "develop", - model_name=model_name, max_iterations=max_iterations) + return await role_agent_loop(project_id, role_name or "develop", model_name=model_name, max_iterations=max_iterations) diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 029e447..3dc4952 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -41,7 +41,7 @@ from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_l from .agent_loop import pm_review_run, pm_review_loop MODULE_NAME = "pipeline_service" -MODULE_VERSION = "3.3.0" +MODULE_VERSION = "3.4.0" async def pipeline_submit(tenant_id, pipeline_id, owner_id, title, params=None): @@ -378,6 +378,7 @@ def pipeline_unregister_step_type(step_type): _SHELL_BASE_DIR = '/d/pipeline/workspaces' _WORKDIR_FALLBACK = '/tmp/pipeline_workspaces' +_PIPELINE_WS = os.path.expanduser('~/pipeline_ws') def _resolve_workdir(): """选择可写的工作目录根。""" @@ -401,8 +402,9 @@ async def shell_exec(command: str, workdir: str = None, timeout: int = 60): cwd = os.path.abspath(cwd) base = os.path.abspath(_SHELL_BASE_DIR) fallback = os.path.abspath(_WORKDIR_FALLBACK) - if not (cwd.startswith(base) or cwd.startswith(fallback)): - return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在 {_SHELL_BASE_DIR} 下"} + pipeline_ws = os.path.abspath(_PIPELINE_WS) + if not (cwd.startswith(base) or cwd.startswith(fallback) or cwd.startswith(pipeline_ws)): + return {"rc": -1, "stdout": "", "stderr": f"安全限制：工作目录必须在允许范围内"} try: proc = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,