""" pipeline-service: sdlc_ability — SDLC 开发产线能力包 以「可插拔能力包」形式定义 SDLC 产线专属能力: - 工具定义(ToolDefinition):任务/交付件/问答/仓库/进度/Bug - prompt 片段(追加到 core 通用心智之后) - 工具 handler(独立函数,签名 async def handler(sor, params, ctx) -> str) 通过 register_ability 注册到 pipeline-core 的 PipelineAbility 注册表, AgentExecutor 按 pipeline_id 动态挂载。 未来 B 产线 = 新增 b_ability.py + register_ability,零侵入 core / AgentExecutor。 """ import json import os from pipeline_core import ( ToolDefinition, PipelineAbility, RoleSpec, register_ability, SlashCommand, register_slash_command, ) # ── 工具定义 ── SDL_TOOLS = [ ToolDefinition( name="create_task", description="创建开发任务,分配给角色agent执行", parameters={ "title": "任务标题", "role": "目标角色(requirement/design/develop/test/deploy)", "description": "任务详细描述", }, category="task", ), ToolDefinition( name="list_tasks", description="查看当前项目的任务列表", parameters={"role": "按角色筛选(可选)", "state": "按状态筛选(可选)"}, category="task", ), ToolDefinition( name="task_detail", description="查看任务详情", parameters={"task_id": "任务ID"}, category="task", ), ToolDefinition( name="start_agents", description="启动角色agent执行已提交的任务", parameters={}, category="agent", ), ToolDefinition( name="diagnose_project", description="诊断项目:汇总卡点/失败/问题/运行中任务", parameters={}, category="agent", ), ToolDefinition( name="list_deliverables", description="查看交付件列表", parameters={"role": "按角色筛选(可选)"}, category="task", ), ToolDefinition( name="view_deliverable", description="查看交付件详细内容", parameters={"deliverable_id": "交付件ID"}, category="task", ), ToolDefinition( name="list_questions", description="查看待回答的问题", parameters={}, category="agent", ), ToolDefinition( name="answer_question", description="回答agent提出的问题", parameters={"question_id": "问题ID", "answer": "回答内容"}, category="agent", ), ToolDefinition( name="add_repo", description="添加项目关联的Git仓库", parameters={"repo_url": "仓库URL", "repo_name": "仓库名称", "default_branch": "默认分支(默认main)"}, category="repo", ), ToolDefinition( name="list_repos", description="查看项目关联的仓库列表", parameters={}, category="repo", ), ToolDefinition( name="clone_repo", description="克隆项目关联的仓库到工作空间repos/下(不传url则clone全部,传则clone指定)", parameters={"repo_url": "仓库URL(可选)"}, category="repo", ), ToolDefinition( name="check_progress", description="查看项目整体进展", parameters={}, category="agent", ), ToolDefinition( name="add_bug", description="提交Bug", parameters={"title": "Bug标题", "description": "描述", "severity": "严重程度"}, category="task", ), ] # ── prompt 片段(追加到 core 通用心智之后) ── SDL_PROMPT = """你是「开发产线」的驾驶舱 agent,负责软件项目的全流程管理(需求 → 设计 → 开发 → 测试 → 部署)。 ## 典型场景(帮助判断何时用哪个工具,非强制) - 用户提出新的开发需求/功能("实现XX""XX系统要做XX")→ 用 create_task 创建任务(可按 requirement→design→develop 拆分),再 start_agents 启动。 - 用户问进展/状态 → list_tasks / diagnose_project / check_progress。 - 用户问待回答问题 → list_questions / answer_question。 - 用户报告异常/故障 → 先用 diagnose_project 定位根因,再用 task_detail 查详情。 - 仓库管理 → add_repo / list_repos / clone_repo。 - 发现卡点(任务卡死、审核超时、失败、僵尸 claimed_by)时,主动定位根因并推动修复。""" # ── 产线角色集(可插拔,替代硬编码 ROLE_SPECIFICS/ROLE_ALIASES/ROLE_CHAIN) ── SDL_ROLES = [ RoleSpec( name="requirement", description="需求分析师", aliases=["requirements", "requirement_analysis"], system_prompt="""你是需求分析师。按 SDLC 仓库标准产出文档。 产出路径: docs/00-requirement/requirement-spec.md 同时创建(项目至少一个应用): - apps/<应用名>.md: 应用描述、部署环境、端口 - modules/<模块名>.md: 模块功能、仓库URL(先填写规划地址)、技术栈、依赖 - 每个应用至少关联一个模块 内容: 项目概述、用户角色及权限、功能列表(每个功能:输入/处理/输出/验收标准)、非功能需求、业务流程。""", next_role="design", ), RoleSpec( name="design", description="系统设计师", aliases=["designer", "ui", "ux"], system_prompt="""你是系统设计师。按 SDLC 仓库标准产出文档。 产出路径: docs/01-design/ 目录下: - architecture.md: 系统架构、技术选型理由 - database-design.md: ER图描述、表结构DDL - api-design.md: 接口列表(method/path/request/response) - ui-design.md: 页面结构、组件树(如适用) 产出后用 result 输出文档,files 列出所有文件路径。""", next_role="develop", ), RoleSpec( name="develop", description="开发工程师", aliases=["developer", "dev", "development", "coding"], system_prompt="""你是开发工程师。源码写入模块独立仓库,不在项目仓库。 准备工作: 1. read_file 读 docs/01-design/ 下的设计文档 2. read_file 读 modules/<模块名>.md 获取模块仓库 URL 3. git_clone 克隆模块仓库到工作空间 开发流程: 4. write_file 写代码到模块仓库目录下 5. run_shell 编译/运行验证 6. git_commit_push 提交 "develop: <简述>" 7. write_file 更新 modules/<模块名>.md 状态为已完成 8. write_file 写 docs/02-develop/dev-notes.md 记录开发内容 9. deliver 交付,files 列出所有产出文件路径 PM审核: git_status 检查模块仓库有提交记录。""", next_role="test", ), RoleSpec( name="test", description="测试工程师", aliases=["testing", "qa", "tester"], system_prompt="""你是测试工程师。按 SDLC 仓库标准产出文档。 产出路径: docs/03-test/ 目录下: - test-plan.md: 测试策略(单元/集成/端到端) - test-cases.md: 用例清单(编号/前置条件/步骤/预期结果) - test-report.md: 执行结果、Bug清单、覆盖率 测试脚本放到 files。""", next_role="deploy", ), RoleSpec( name="deploy", description="部署运维工程师", aliases=["deployment", "release"], system_prompt="""你是部署运维工程师。按 SDLC 仓库标准产出文档和配置。 产出路径: - docs/04-deploy/deploy-guide.md: 环境要求、部署步骤、回滚方案 - docs/04-deploy/release-notes.md: 发布说明 - config/ 下: Dockerfile、nginx配置、docker-compose.yml 等 产出后用 result 输出部署说明,files 列出所有配置文件路径。""", next_role="", ), ] # ── handler 独立函数(签名统一 async def handler(sor, params, ctx) -> str) ── async def _h_create_task(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" from appPublic.uniqueID import getID title = p.get("title", "").strip() role = p.get("role", "develop") desc = p.get("description", "") if not title: return "需要任务标题" tid = getID() await sor.C("pipeline_tasks", { "id": tid, "tenant_id": pid, "pipeline_id": "role_task", "owner_id": ctx.get("user_id") or "user", "title": title, "state": "submitted", "role": role, "params": json.dumps({"description": desc}, ensure_ascii=False), }) return f"OK: 已创建任务 {title}(角色: {role})" async def _h_list_tasks(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" role = p.get("role", "") state = p.get("state", "") sql = "SELECT id, title, role, state, created_at FROM pipeline_tasks WHERE tenant_id=${pid}$" params_dict = {"pid": pid} if role: sql += " AND role=${role}$" params_dict["role"] = role if state: sql += " AND state=${state}$" params_dict["state"] = state sql += " ORDER BY created_at DESC LIMIT 30" recs = await sor.sqlExe(sql, params_dict) if not recs: return "暂无任务" icons = { "submitted": "⏳", "running": "🔄", "review": "👀", "approved": "✅", "completed": "✔️", "failed": "❌", "waiting": "⏸️", } lines = ["| 状态 | 任务 | 角色 | ID |", "|------|------|------|----|"] for r in recs: st = getattr(r, 'state', '?') icon = icons.get(st, "❓") title = (getattr(r, 'title', '') or '')[:60] rrole = getattr(r, 'role', '') tid = getattr(r, 'id', '') lines.append(f"| {icon} {st} | {title} | {rrole} | {tid} |") return "\n".join(lines) async def _h_task_detail(sor, p, ctx): tid = p.get("task_id", "") if not tid: return "需要任务ID" recs = await sor.sqlExe( "SELECT * FROM pipeline_tasks WHERE id=${tid}$", {"tid": tid}) if not recs and len(tid) >= 6: recs = await sor.sqlExe( "SELECT * FROM pipeline_tasks WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": tid + "%"}) if not recs: return f"任务不存在: {tid}" r = recs[0] return json.dumps({ "id": getattr(r, "id", ""), "title": getattr(r, "title", ""), "state": getattr(r, "state", ""), "role": getattr(r, "role", ""), "params": getattr(r, "params", "{}"), }, ensure_ascii=False, indent=2) async def _h_start_agents(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" recs = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks " "WHERE tenant_id=${pid}$ AND state='submitted' ORDER BY created_at ASC LIMIT 10", {"pid": pid}) if not recs: return "没有待执行的任务" results = [] for r in recs: tid = getattr(r, "id", "") role = getattr(r, "role", "develop") try: from pipeline_service.agent_loop import role_agent_run result = await role_agent_run(pid, role) results.append(f"- {getattr(r, 'title', '')}: {result.get('status', '?')}") except Exception as e: results.append(f"- {getattr(r, 'title', '')}: error={str(e)[:100]}") return "Agent 执行结果:\n" + "\n".join(results) async def _h_diagnose_project(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" async def _cnt(state): recs = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state=${st}$", {"pid": pid, "st": state}) return getattr(recs[0], "c", 0) if recs else 0 sc = await _cnt("submitted") rc = await _cnt("running") fc = await _cnt("failed") vc = await _cnt("review") qc = 0 try: questions = await sor.sqlExe( "SELECT COUNT(*) as c FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND status='pending'", {"pid": pid}) qc = getattr(questions[0], "c", 0) if questions else 0 except Exception: pass lines = [ "项目诊断:", f"- 待执行: {sc}", f"- 运行中: {rc}", f"- 待审核: {vc}", f"- 失败: {fc}", f"- 待回答问题: {qc}", ] if fc: failed_rows = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='failed' ORDER BY created_at DESC LIMIT 10", {"pid": pid}) lines.append("失败任务:") for fr in (failed_rows or []): lines.append(f" - [{getattr(fr, 'role', '?')}] {getattr(fr, 'title', '')} (id={getattr(fr, 'id', '')})") if sc: submitted_rows = await sor.sqlExe( "SELECT id, title, role FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='submitted' ORDER BY created_at ASC LIMIT 10", {"pid": pid}) lines.append("待执行任务:") for sr in (submitted_rows or []): lines.append(f" - [{getattr(sr, 'role', '?')}] {getattr(sr, 'title', '')} (id={getattr(sr, 'id', '')})") if rc: running_rows = await sor.sqlExe( "SELECT id, title, role, claimed_by, " "TIMESTAMPDIFF(MINUTE, updated_at, NOW()) AS mins " "FROM pipeline_tasks WHERE tenant_id=${pid}$ AND state='running' " "ORDER BY updated_at ASC LIMIT 10", {"pid": pid}) lines.append("运行中任务:") for rr in (running_rows or []): rrole = getattr(rr, 'role', '?') rtitle = getattr(rr, 'title', '') rid = getattr(rr, 'id', '') rcb = getattr(rr, 'claimed_by', '') or '' try: mins = int(getattr(rr, 'mins', 0) or 0) except (TypeError, ValueError): mins = 0 flag = f"⚠️心跳超时{mins}分钟(疑似僵尸,将被回收重跑)" if mins >= 20 else f"已运行{mins}分钟" lines.append(f" - [{rrole}] {rtitle} (id={rid}) | {flag} | claimed_by={rcb[:8]}") return "\n".join(lines) async def _h_list_deliverables(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" sql = "SELECT id, title, deliverable_type, review_status FROM pipeline_deliverables WHERE project_id=${pid}$ ORDER BY created_at DESC LIMIT 20" recs = await sor.sqlExe(sql, {"pid": pid}) if not recs: return "暂无交付件" lines = [] for r in recs: lines.append( f"- [{getattr(r, 'review_status', '?')}] {getattr(r, 'title', '')} " f"({getattr(r, 'deliverable_type', '')}) id={getattr(r, 'id', '')}") return "\n".join(lines) async def _h_view_deliverable(sor, p, ctx): did = p.get("deliverable_id", "") if not did: return "需要交付件ID" recs = await sor.sqlExe( "SELECT * FROM pipeline_deliverables WHERE id=${did}$", {"did": did}) if not recs and len(did) >= 6: recs = await sor.sqlExe( "SELECT * FROM pipeline_deliverables WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": did + "%"}) if not recs: return f"交付件不存在: {did}" return getattr(recs[0], "content", "")[:4000] or "(空)" async def _h_list_questions(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" try: recs = await sor.sqlExe( "SELECT id, question, answer_source FROM pipeline_agent_questions " "WHERE tenant_id=${pid}$ AND status='pending' ORDER BY created_at DESC LIMIT 10", {"pid": pid}) except Exception: try: recs = await sor.sqlExe( "SELECT id, question, answer_source FROM pipeline_agent_questions " "WHERE tenant_id=${pid}$ ORDER BY created_at DESC LIMIT 10", {"pid": pid}) except Exception: return "暂无问题数据" if not recs: return "没有待回答问题" lines = [] for r in recs: lines.append(f"- [{getattr(r, 'id', '')}] {getattr(r, 'question', '')[:200]}") return "\n".join(lines) async def _h_answer_question(sor, p, ctx): qid = p.get("question_id", "") answer = p.get("answer", "") if not qid or not answer: return "需要问题ID和回答内容" recs = await sor.sqlExe( "SELECT id, task_id FROM pipeline_agent_questions WHERE id=${qid}$", {"qid": qid}) if not recs and len(qid) >= 6: recs = await sor.sqlExe( "SELECT id, task_id FROM pipeline_agent_questions WHERE id LIKE ${prefix}$ LIMIT 1", {"prefix": qid + "%"}) if not recs: return f"问题不存在: {qid}" full_qid = getattr(recs[0], "id", qid) task_id = getattr(recs[0], "task_id", "") or "" await sor.sqlExe( "UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', " "answered_by='main_agent', status='answered' WHERE id=${qid}$", {"a": answer, "qid": full_qid}) resumed = False if task_id: await sor.sqlExe( "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL " "WHERE id=${tid}$ AND state='waiting'", {"tid": task_id}) resumed = True suffix = ",任务已恢复执行" if resumed else "" return "OK: 已回答" + suffix async def _h_add_repo(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" from appPublic.uniqueID import getID url = p.get("repo_url", "").strip() name = p.get("repo_name", "").strip() branch = p.get("default_branch", "main") if not url: return "需要仓库URL" await sor.C("sd_project_repos", { "id": getID(), "project_id": pid, "repo_url": url, "repo_name": name or url.split("/")[-1].replace(".git", ""), "default_branch": branch, }) return f"OK: 已添加仓库 {name or url}" async def _h_list_repos(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" recs = await sor.sqlExe( "SELECT repo_name, repo_url, default_branch FROM sd_project_repos WHERE project_id=${pid}$", {"pid": pid}) if not recs: return "暂无关联仓库" lines = [] for r in recs: lines.append(f"- {getattr(r, 'repo_name', '')}: {getattr(r, 'repo_url', '')}") return "\n".join(lines) async def _h_clone_repo(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" from pipeline_service.agent_loop import ( _setup_repos, _get_workspace_dir, _git_clone) workspace_dir = await _get_workspace_dir(sor, pid) url = (p.get("repo_url", "") or "").strip() if url: name = url.rstrip("/").split("/")[-1].replace(".git", "") target = os.path.join(workspace_dir, "repos", name) r = await _git_clone(url, target, p.get("branch", "main")) return f"rc={r['rc']} {r['message']}" results = await _setup_repos(sor, workspace_dir, pid) if not results: return "无关联仓库可克隆" return "\n".join( f"- {x['repo']}: rc={x.get('rc', '?')} {x.get('message', '')}" for x in results) async def _h_check_progress(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" recs = await sor.sqlExe( "SELECT role, state, COUNT(*) as c FROM pipeline_tasks " "WHERE tenant_id=${pid}$ GROUP BY role, state ORDER BY role, state", {"pid": pid}) if not recs: return "暂无进度数据" lines = [] for r in recs: lines.append( f"- {getattr(r, 'role', '?')}: {getattr(r, 'state', '?')} x{getattr(r, 'c', 0)}") return "\n".join(lines) async def _h_add_bug(sor, p, ctx): pid = ctx.get("project_id", "") if not pid: return "请先切换到项目" from appPublic.uniqueID import getID title = p.get("title", "").strip() desc = p.get("description", "") severity = p.get("severity", "medium") if not title: return "需要Bug标题" iteration_id = p.get("iteration_id", "") if not iteration_id: its = await sor.sqlExe( "SELECT id FROM sd_iterations WHERE project_id=${pid}$ ORDER BY created_at ASC LIMIT 1", {"pid": pid}) if its: iteration_id = getattr(its[0], "id", "") if not iteration_id: return "ERROR: 项目无迭代,无法提交Bug" await sor.C("sd_bugs", { "id": getID(), "title": title, "description": desc, "severity": severity, "status": "open", "iteration_id": iteration_id, }) return f"OK: 已提交Bug: {title}" # ── 注册能力包 ── SDL_HANDLERS = { "create_task": _h_create_task, "list_tasks": _h_list_tasks, "task_detail": _h_task_detail, "start_agents": _h_start_agents, "diagnose_project": _h_diagnose_project, "list_deliverables": _h_list_deliverables, "view_deliverable": _h_view_deliverable, "list_questions": _h_list_questions, "answer_question": _h_answer_question, "add_repo": _h_add_repo, "list_repos": _h_list_repos, "clone_repo": _h_clone_repo, "check_progress": _h_check_progress, "add_bug": _h_add_bug, } def register_sdlc_ability(): """注册 SDLC 产线能力包(幂等)。""" ability = PipelineAbility( pipeline_id="sdlc_general", name="通用软件开发产线", tools=SDL_TOOLS, system_prompt=SDL_PROMPT, handlers=SDL_HANDLERS, roles=SDL_ROLES, menus=[ {"label": "📁 项目", "icon": "", "url": "/pipeline-sdlc/sd_projects/index.ui", "type": "popup", "width": "85%", "height": "80%"}, {"label": "🐛 Bug", "icon": "", "url": "/pipeline-sdlc/sd_bugs/index.ui", "type": "popup", "width": "85%", "height": "80%"}, {"label": "🧪 测试用例", "icon": "", "url": "/pipeline-sdlc/sd_test_cases/index.ui", "type": "popup", "width": "85%", "height": "80%"}, {"label": "🔄 迭代", "icon": "", "url": "/pipeline-sdlc/sd_iterations/index.ui", "type": "popup", "width": "85%", "height": "80%"}, ], ) register_ability(ability) return ability # ── SDLC 产线 slash 命令(产线专属命名) ── async def _slash_tasks(args, ctx): ex = ctx.get("executor") r = await ex._execute_ability_tool("list_tasks", {"role": args.strip() or ""}) if ex else None return r or "无任务" async def _slash_diagnose(args, ctx): ex = ctx.get("executor") r = await ex._execute_ability_tool("diagnose_project", {}) if ex else None return r or "无诊断数据" def register_sdlc_slash_commands(): for cmd in [ SlashCommand("tasks", "列出任务(可按角色筛选)", _slash_tasks, "pipeline", "sdlc_general"), SlashCommand("diagnose", "诊断项目状态", _slash_diagnose, "pipeline", "sdlc_general"), ]: register_slash_command(cmd) # import 即注册(模块加载副作用),pipeline-app 启动时 import pipeline_service 即生效 register_sdlc_ability() register_sdlc_slash_commands()