From 849a761b763056bfa482256ad9e55cddd7c2b7dc Mon Sep 17 00:00:00 2001 From: ymq Date: Sat, 15 Aug 2026 19:11:49 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=88=A0=E9=99=A4=E6=97=A7?= =?UTF-8?q?=E7=89=88cockpit=E7=95=8C=E9=9D=A2(sd=5Fcockpit/index.ui+cockpi?= =?UTF-8?q?t=5Fchat.dspy),=E7=BB=9F=E4=B8=80=E5=88=B0v2=E6=9E=B6=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- wwwroot/api/cockpit_chat.dspy | 656 ---------------------------------- wwwroot/sd_cockpit/index.ui | 190 ---------- 2 files changed, 846 deletions(-) delete mode 100644 wwwroot/api/cockpit_chat.dspy delete mode 100644 wwwroot/sd_cockpit/index.ui diff --git a/wwwroot/api/cockpit_chat.dspy b/wwwroot/api/cockpit_chat.dspy deleted file mode 100644 index b101ee6..0000000 --- a/wwwroot/api/cockpit_chat.dspy +++ /dev/null @@ -1,656 +0,0 @@ -# cockpit_chat.dspy - SDLC Agent Loop v3.1 (streaming NDJSON) -import aiohttp -import os -import re -import shutil -import zipfile -from ahserver.filestorage import FileStorage - - -def _extract_text(path, name): - """提取文件文本内容（docx/txt/md等），返回文本或空字符串。二进制/无法解析返回空。""" - ext = os.path.splitext(name)[1].lower() - try: - if ext in ('.txt', '.md', '.json', '.csv', '.py', '.log', '.yaml', '.yml', '.xml', '.html', '.ini'): - with open(path, 'r', encoding='utf-8', errors='ignore') as f: - return f.read()[:15000] - if ext == '.docx': - with zipfile.ZipFile(path) as z: - xml = z.read('word/document.xml').decode('utf-8', errors='ignore') - texts = re.findall(r']*>(.*?)', xml) - return '\n'.join(texts)[:15000] - except Exception: - pass - return '' - -action = (params_kw or {}).get('action', 'send_message') -# AgentIO sends JSON body without action field — detect from message_text presence -if action != 'send_message': - p = params_kw or {} - msg = (p.get('message_text') or '').strip() - if not msg: - inner = p.get('params', {}) - if isinstance(inner, dict): - msg = (inner.get('prompt') or inner.get('message_text') or '').strip() - if msg: - action = 'send_message' -dbname = get_module_dbname('pipeline-sdlc') - -AGENT_PROMPT = """你是开发产线驾驶舱 Agent。用工具获取数据，用 reply 回复。 - -## 回复格式 -- 用换行分段，不要一大段文字 -- 列表项前加 - 或数字 -- 重要信息用 **粗体** -- 代码/命令用 `` 包裹 -- 简洁直接，先说结论再展开 - -## 核心原则 -- 环境信息中已显示当前项目。所有操作在当前项目内完成 -- 用户输入为主要依据，历史记录仅作参考 -- **reply 只用于汇报已完成的结果，不要用 reply 描述"我将要做XX"——想做就直接调 tool_call 去做** -- 诊断 → 调 diagnose_project。回答 → 调 answer_question。启动 → 调 start_agents。执行，不描述 - -## 工作流 -- 切换项目：用户说"切换到XXX项目" → switch_project -- 删除项目：delete_project → 看到确认提示后，用户回复「确认删除 xxx」→ 再调 _sys_delete_project -- 查看进度：list_tasks / task_detail / list_deliverables -- 提交任务：先确保在正确项目，再 create_task -- 诊断项目：diagnose_project 找出卡点/失败任务/待回答问题 -- 启动Agent：任务提交后调 start_agents 让角色Agent执行 -- 回答问题：list_questions 看问题 → answer_question 回答。待回答问题在环境信息中已列出，主动提示用户 -- clone_repo 用于克隆 git 仓库到工作空间 -- run_command 用于在工作空间中执行 shell 命令 - -输出格式二选一： -{"action":"tool_call","tool":"工具名","params":{}} -{"action":"reply","message":"中文回复"} - -示例——错误做法（用reply描述计划）： -{"action":"reply","message":"我来逐一回答这三个问题"} ← 错！应该直接调工具 - -示例——正确做法（用tool_call执行，最后用reply汇报）： -{"action":"tool_call","tool":"answer_question","params":{"question_id":"sx728Vhq","answer":"..."}} -{"action":"tool_call","tool":"answer_question","params":{"question_id":"C0ujsOUj","answer":"..."}} -{"action":"tool_call","tool":"start_agents","params":{}} -{"action":"reply","message":"三个问题已回答，Agent已启动"} - -工具: __TOOLS__ -环境: __ENV__""" - -TOOLS = [ - {"name":"switch_project","description":"切换到指定项目","params":{"project":"项目名称"}}, - {"name":"create_project","description":"创建新项目","params":{"name":"项目名称","description":"项目描述(可选)"}}, - {"name":"delete_project","description":"删除项目（返回确认提示，用户确认后再调用_sys_delete_project执行）","params":{"project":"项目名称"}}, - {"name":"list_tasks","description":"列出当前项目任务","params":{"state":"状态(可选)"}}, - {"name":"task_detail","description":"查看任务详情（含交付件和问答）","params":{"task_id":"任务ID"}}, - {"name":"list_deliverables","description":"列出当前项目交付件","params":{"task_id":"任务ID(可选)"}}, - {"name":"view_deliverable","description":"查看交付件内容","params":{"deliverable_id":"交付件ID"}}, - {"name":"list_questions","description":"列出待回答的问题","params":{}}, - {"name":"answer_question","description":"回答agent提出的问题","params":{"question_id":"问题ID","answer":"回答内容"}}, - {"name":"create_task","description":"提交开发任务到指定角色","params":{"title":"标题","description":"描述","role":"requirement/design/develop/test/deploy"}}, - {"name":"start_agents","description":"启动项目角色agents执行待办任务","params":{}}, - {"name":"diagnose_project","description":"诊断项目状态：找出卡点、失败任务、待回答问题，给出推进建议","params":{}}, - {"name":"check_progress","description":"查看项目整体进度","params":{"task_id":"任务ID(可选)"}}, - {"name":"add_repo","description":"关联仓库到当前项目","params":{"url":"git地址","name":"名称(可选)"}}, - {"name":"list_repos","description":"列出当前项目仓库","params":{}}, - {"name":"clone_repo","description":"克隆仓库到工作空间","params":{"url":"git地址","branch":"分支名(可选,默认main)"}}, - {"name":"add_bug","description":"报告Bug","params":{"title":"标题","description":"描述"}}, - {"name":"run_command","description":"在工作空间中执行shell命令","params":{"cmd":"命令"}}, -] -TOOLS_TEXT = json.dumps(TOOLS, ensure_ascii=False) - -def _security_scan(text): - t = (text or '').lower() - for kw in ('drop table','truncate','rm -rf','密钥','api_key'): - if kw in t: return True, '危险操作' - return False, '' - -def _guess_role(title): - t = (title or '').lower() - for kw,r in [('需求','requirement'),('设计','design'),('开发','develop'),('实现','develop'),('测试','test'),('部署','deploy')]: - if kw in t: return r - return 'develop' - -async def _find_project(sor, name): - """Exact match only — LLM has full project list in env and should pick exact name.""" - if not name: return None - recs = await sor.sqlExe("SELECT id,name FROM sd_projects WHERE name=${n}$",{"n":name}) - return recs[0] if recs else None - -async def _load_ctx(sor, uid): - if not uid: - return {'pid':'','pname':'','ws':'','skills_dir':'','repos':[]} - recs = await sor.sqlExe("SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$",{"u":uid}) - ctx = {'pid':'','pname':'','ws':'','skills_dir':'','repos':[]} - if not recs: return ctx - ctx['pid'] = getattr(recs[0],'current_project_id','') or '' - if not ctx['pid']: - # Fallback: recover from most recent conversation's iteration_id (which stores project_id) - last = await sor.sqlExe("SELECT iteration_id FROM pipeline_conversations WHERE created_by=${u}$ AND iteration_id != '' ORDER BY created_at DESC LIMIT 1",{"u":uid}) - if last: - ctx['pid'] = getattr(last[0],'iteration_id','') or '' - if ctx['pid']: - projs = await sor.sqlExe("SELECT name,workspace_dir,org_id FROM sd_projects WHERE id=${p}$",{"p":ctx['pid']}) - if projs: - ctx['pname'] = getattr(projs[0],'name','') - ctx['ws'] = getattr(projs[0],'workspace_dir','') or '' - oid = getattr(projs[0],'org_id','') or '0' - orgs = await sor.sqlExe("SELECT skills_dir FROM sd_org_settings WHERE org_id=${o}$",{"o":oid}) - if orgs: ctx['skills_dir'] = getattr(orgs[0],'skills_dir','') or '' - repos = await sor.sqlExe("SELECT repo_name,repo_url FROM sd_project_repos WHERE project_id=${p}$",{"p":ctx['pid']}) - ctx['repos'] = [{'n':r.repo_name,'u':r.repo_url} for r in (repos or [])] - return ctx - -async def _save_ctx(sor, uid, pid): - if not uid: - return - try: - await sor.sqlExe("UPDATE pipeline_agent_settings SET current_project_id=${p}$ WHERE user_id=${u}$",{"p":pid or '','u':uid}) - ex = await sor.sqlExe("SELECT id FROM pipeline_agent_settings WHERE user_id=${u}$",{"u":uid}) - if not ex: - await sor.C('pipeline_agent_settings',{'id':getID(),'user_id':uid,'default_llm_id':'','current_project_id':pid or '','current_iteration_id':''}) - except: - # Duplicate key race — update instead - try: - await sor.sqlExe("UPDATE pipeline_agent_settings SET current_project_id=${p}$ WHERE user_id=${u}$",{"p":pid or '','u':uid}) - except: - pass - -async def _sel_model(sor, pref): - if pref: - recs = await sor.sqlExe("SELECT id,name,provider,model_id,api_base,api_key FROM llm WHERE (id=${l}$ OR name=${l}$) AND status='active'",{"l":pref}) - if recs: return recs[0] - recs = await sor.sqlExe("SELECT id,name,provider,model_id,api_base,api_key FROM llm WHERE status='active' LIMIT 1",{}) - return recs[0] if recs else None - -async def _call_llm_raw(prompt, temp=0.0): - """Lightweight LLM call for classification tasks. Uses first available model.""" - to = aiohttp.ClientTimeout(total=30) - async with DBPools().sqlorContext(dbname) as sor: - model = await _sel_model(sor, None) - if not model: - return "不存在" - hdrs = {"Authorization": f"Bearer {model.api_key or ''}", "Content-Type": "application/json"} - payload = {"model": model.model_id, "messages": [{"role":"user","content":prompt}], "temperature": temp} - async with aiohttp.ClientSession(timeout=to) as s: - async with s.post(f"{model.api_base.rstrip('/')}/chat/completions", headers=hdrs, json=payload) as r: - if r.status != 200: raise ValueError(f"LLM {r.status}") - return (await r.json())["choices"][0]["message"]["content"] - -async def _call_llm(model, msgs, temp): - hdrs = {"Authorization": f"Bearer {model.api_key or ''}", "Content-Type": "application/json"} - payload = {"model": model.model_id, "messages": msgs, "temperature": temp} - to = aiohttp.ClientTimeout(total=180) - async with aiohttp.ClientSession(timeout=to) as s: - async with s.post(f"{model.api_base.rstrip('/')}/chat/completions", headers=hdrs, json=payload) as r: - if r.status != 200: raise ValueError(f"LLM {r.status}") - return (await r.json())["choices"][0]["message"]["content"] - -def _parse(raw): - raw = (raw or "").strip() - if raw.startswith("```"): raw = raw.split("\n",1)[1].rsplit("```",1)[0].strip() - try: - d = json.loads(raw) - if isinstance(d,dict) and 'action' in d: return d - except: pass - # Strip any {"action":"tool_call"...} JSON fragments from reply text - import re as _re2 - clean = _re2.sub(r'\s*\{[^}]*"action"\s*:\s*"tool_call"[^}]*\}\s*', ' ', raw).strip() - # Remove stray trailing braces from malformed JSON attempts - clean = clean.rstrip().rstrip('}').strip() - return {"action":"reply","message":clean or raw} - -def _w_text(t): return {"widgettype":"Text","options":{"text":t,"css":"agent-text","halign":"left"}} -def _w_md(t): return {"widgettype":"MdWidget","options":{"mdtext":t,"css":"resp-content","width":"100%"}} -def _w_card(title, body, kind="success"): - colors = {"success":"#10b981","error":"#ef4444","reply":"#6366f1","progress":"#f59e0b"} - c = colors.get(kind,"#6b7280") - return {"widgettype":"VBox","options":{"style":{"borderLeft":f"3px solid {c}","padding":"8px 12px","margin":"4px 0"}},"subwidgets":[ - {"widgettype":"Text","options":{"text":title,"style":{"fontWeight":"bold","fontSize":"13px","color":c}}}, - body - ]} -def _w_progress(text): return {"widgettype":"Text","options":{"text":text,"style":{"color":"#f59e0b","fontSize":"12px","padding":"2px 8px"},"halign":"left"}} - -_tool_labels = {'switch_project':'切换项目','create_project':'创建项目','delete_project':'删除项目','list_tasks':'查询任务','task_detail':'任务详情','list_deliverables':'查看交付件','view_deliverable':'交付件内容','list_questions':'查看问题','answer_question':'回答问题','create_task':'创建任务','start_agents':'启动角色Agent','diagnose_project':'项目诊断','check_progress':'检查进度','add_repo':'关联仓库','list_repos':'查看仓库','clone_repo':'克隆仓库','add_bug':'报告Bug','run_command':'执行命令'} - -async def _exec_tool(sor, tool, params, ctx, uid, org_id): - p = params or {} - try: - if tool == 'create_project': - name = (p.get('name','') or '').strip() - if not name: return 'FAIL: 需要项目名称' - ex = await sor.sqlExe("SELECT id FROM sd_projects WHERE name=${n}$",{"n":name}) - if ex: return f'FAIL: 项目「{name}」已存在' - pid = getID() - wid = getID() - _wsb = await get_workspace_base(sor) - ws = _wsb + '/' + str(org_id) + '/' + name - await sor.C('sd_projects',{'id':pid,'name':name,'description':p.get('description',''),'workspace_dir':ws,'org_id':org_id,'created_by':uid}) - await sor.C('sd_iterations',{'id':wid,'project_id':pid,'iteration_name':'默认迭代','iteration_type':'default','status':'active','priority':1,'org_id':org_id}) - await _save_ctx(sor, uid, pid) - ctx['pid'] = pid; ctx['pname'] = name; ctx['ws'] = ws - return f'OK: 已创建并切换到「{name}」' - elif tool == 'delete_project': - name = (p.get('project','') or '').strip() - if not name: return 'FAIL: 需要项目名称' - proj = await _find_project(sor, name) - if not proj: return f'FAIL: 项目「{name}」不存在' - return f'CONFIRM: 确定要删除项目「{name}」吗？此操作不可撤销，将清除所有关联数据。请回复「确认删除 {name}」继续，回复其他内容取消。' - elif tool == '_sys_delete_project': - name = (p.get('project','') or '').strip() - if not name: return 'FAIL: 缺少项目名称' - proj = await _find_project(sor, name) - if not proj: return f'FAIL: 项目「{name}」不存在' - pid = proj.id - # 级联删除关联数据（表名及对应的project关联字段） - tables = [ - ('pipeline_deliverables', 'project_id'), - ('pipeline_tasks', 'tenant_id'), - ('pipeline_agent_questions', 'tenant_id'), - ('sd_project_repos', 'project_id'), - ('sd_iterations', 'project_id'), - ('sd_bugs', 'iteration_id IN (SELECT id FROM sd_iterations WHERE project_id=${p}$)'), - ] - count = 0 - for tbl, cond in tables: - if cond == '1=0': continue - sql = f"DELETE FROM {tbl} WHERE {cond}=${{p}}$" - await sor.sqlExe(sql, {"p": pid}) - count += 1 - await sor.sqlExe("DELETE FROM pipeline_agent_settings WHERE current_project_id=${p}$", {"p": pid}) - await sor.D('sd_projects', {'id': pid}) - if ctx['pid'] == pid: - ctx['pid'] = ''; ctx['pname'] = ''; ctx['ws'] = ''; ctx['repos'] = [] - return f'OK: 已删除「{name}」及其所有关联数据' - elif tool == 'switch_project': - name = (p.get('project','') or '').strip() - if not name: - # No project specified — return available projects - allp = await sor.sqlExe("SELECT id,name FROM sd_projects ORDER BY created_at DESC LIMIT 20",{}) - if not allp: return '无可用项目' - lines = [] - for r in allp: - marker = '★' if getattr(r,'id','') == ctx['pid'] else ' ' - lines.append(f"{marker} {getattr(r,'name','')}") - return '可用项目:\n'+'\n'.join(lines) - proj = await _find_project(sor, name) - if not proj: - # LLM classification: ask LLM to match user input against project list - allp = await sor.sqlExe("SELECT id,name FROM sd_projects ORDER BY created_at DESC LIMIT 20",{}) - pnames = [getattr(r,'name','') for r in (allp or [])] - classify_prompt = f"用户输入: {name}\n项目列表: {', '.join(pnames)}\n\n判断用户想要哪个项目。只回复项目名或\"不存在\"。" - try: - raw = await _call_llm_raw(classify_prompt, 0.0) - matched = raw.strip().strip('"').strip("'") - for r in (allp or []): - if getattr(r,'name','') == matched: - proj = r; break - except: - pass - if not proj: - allp2 = await sor.sqlExe("SELECT name FROM sd_projects ORDER BY created_at DESC LIMIT 15",{}) - return 'FAIL: 未找到。可用: '+', '.join([getattr(r,'name','') for r in (allp2 or [])]) - if proj.id == ctx['pid']: - return f'已在「{proj.name}」项目中，无需切换' - await _save_ctx(sor, uid, proj.id) - ctx['pid'] = proj.id; ctx['pname'] = proj.name - return 'OK: 已切换到「'+proj.name+'」' - elif tool == 'create_task': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - title = p.get('title','')[:100] - if not title: return 'FAIL: 缺少标题' - role = p.get('role','') or _guess_role(title) - tp = {'description':p.get('description',''),'project_id':ctx['pid']} - r = await pipeline_role_submit(ctx['pid'],'role_task',uid,title,tp,role) - rd = json.loads(r) - if rd.get('success'): return f"OK: 任务「{title}」已创建(角色:{role})" - return f"FAIL: {rd.get('message','')}" - elif tool == 'check_progress': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - tid = p.get('task_id','') - if tid: - ts = await sor.sqlExe("SELECT title,state,role FROM pipeline_tasks WHERE id=${t}$",{"t":tid}) - if not ts: return 'FAIL: 任务不存在' - t = ts[0]; return f"[{t.state}][{t.role}] {t.title}" - ts = await sor.sqlExe("SELECT state,count(*)c FROM pipeline_tasks WHERE tenant_id=${p}$ GROUP BY state",{"p":ctx['pid']}) - summary = ', '.join([f"{r.state}:{r.c}" for r in (ts or [])]) or '无' - ags = await sor.sqlExe("SELECT role,count(*)c FROM pipeline_tasks WHERE tenant_id=${p}$ AND state IN ('submitted','running') GROUP BY role",{"p":ctx['pid']}) - al = ', '.join([f"{getattr(r,'role','')}:{getattr(r,'c','')}" for r in (ags or [])]) or '无活跃任务' - ds = await sor.sqlExe("SELECT deliverable_type,title,review_status FROM pipeline_deliverables WHERE project_id=${p}$ ORDER BY created_at DESC LIMIT 3",{"p":ctx['pid']}) - dl = '\n'.join([str(d.deliverable_type)+' '+str(d.title)[:40] for d in (ds or [])]) or '无' - return f"任务: {summary}\nAgent: {al}\n交付:\n{dl}" - elif tool == 'list_tasks': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - sf = p.get('state','') - sql = "SELECT title,state,role FROM pipeline_tasks WHERE tenant_id=${p}$" - if sf: sql += " AND state=${s}$" - sql += " ORDER BY created_at DESC LIMIT 20" - ts = await sor.sqlExe(sql, {"p":ctx['pid'],"s":sf} if sf else {"p":ctx['pid']}) - em = {'submitted':'⏳','running':'🔄','review':'👀','approved':'✅','completed':'🏁','failed':'❌'} - return '\n'.join([f"{em.get(t.state,'?')}[{t.state}][{t.role}]{t.title}" for t in ts]) if ts else '无任务' - elif tool == 'add_repo': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - url = p.get('url','') - if not url: return 'FAIL: 需要仓库地址' - name = p.get('name','') or url.rstrip('/').split('/')[-1].replace('.git','') - ex = await sor.sqlExe("SELECT id FROM sd_project_repos WHERE project_id=${p}$ AND repo_url=${u}$",{"p":ctx['pid'],"u":url}) - if ex: return '仓库已关联' - await sor.C('sd_project_repos',{'id':getID(),'project_id':ctx['pid'],'repo_name':name,'repo_url':url,'default_branch':'main','local_path':'','org_id':org_id}) - ctx['repos'].append({'n':name,'u':url}) - return f"OK: 已关联 {name}" - elif tool == 'list_repos': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - rs = await sor.sqlExe("SELECT repo_name,repo_url FROM sd_project_repos WHERE project_id=${p}$",{"p":ctx['pid']}) - return '\n'.join([f"{r.repo_name}: {r.repo_url}" for r in rs]) if rs else '无仓库' - elif tool == 'clone_repo': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - url = p.get('url','') - if not url: return 'FAIL: 需要仓库地址' - branch = p.get('branch','main') - ws = ctx.get('ws','') - if not ws: return 'FAIL: 项目无工作空间路径' - import subprocess, os - repo_name = url.rstrip('/').split('/')[-1].replace('.git','') - target = os.path.join(ws, repo_name) - if os.path.exists(target): return f'FAIL: 目录已存在 {target}' - os.makedirs(ws, exist_ok=True) - cmd = f'git clone -b {branch} {url} {target}' - r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=120, cwd=ws) - if r.returncode != 0: return f'FAIL: git clone失败: {r.stderr[:200]}' - # Auto-add repo to project - ex = await sor.sqlExe("SELECT id FROM sd_project_repos WHERE project_id=${p}$ AND repo_url=${u}$",{"p":ctx['pid'],"u":url}) - if not ex: - await sor.C('sd_project_repos',{'id':getID(),'project_id':ctx['pid'],'repo_name':repo_name,'repo_url':url,'default_branch':branch,'local_path':target,'org_id':org_id}) - ctx['repos'].append({'n':repo_name,'u':url}) - return f'OK: 已克隆到 {target}' - elif tool == 'run_command': - import subprocess - ws = ctx.get('ws','') or '/tmp' - cmd = p.get('cmd','') - if not cmd: return 'FAIL: 需要命令' - # Safety: block destructive commands - blocked = ['rm -rf /','mkfs.','dd if=',':(){','chmod 777 /'] - for b in blocked: - if b in cmd: return 'FAIL: 危险命令被拦截' - r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60, cwd=ws) - out = (r.stdout + r.stderr)[:2000] - return f'OK (rc={r.returncode}):\n{out}' if out else f'OK (rc={r.returncode})' - elif tool == 'task_detail': - tid = p.get('task_id','') - if not tid: - # No task_id provided — show first few pending/running tasks for context - if not ctx['pid']: return 'FAIL: 请先切换到项目并指定任务ID' - ts = await sor.sqlExe("SELECT id,title,state,role FROM pipeline_tasks WHERE tenant_id=${p}$ AND state IN ('submitted','running','waiting') ORDER BY created_at ASC LIMIT 5",{"p":ctx['pid']}) - if not ts: return '无活跃任务，请用 list_tasks 查看全部' - return '当前活跃任务（请指定 task_id 查看详情）:\n'+'\n'.join([f" [{getattr(t,'state','')}] {getattr(t,'id','')[:12]} [{getattr(t,'role','')}] {getattr(t,'title','')[:60]}" for t in ts]) - ts = await sor.sqlExe("SELECT * FROM pipeline_tasks WHERE id=${t}$",{"t":tid}) - if not ts: return 'FAIL: 任务不存在' - t = ts[0] - ds = await sor.sqlExe("SELECT deliverable_type,title,review_status,quality_score FROM pipeline_deliverables WHERE task_id=${t}$",{"t":tid}) - dl = '\n'.join([f" [{getattr(d,'review_status','')}] {getattr(d,'deliverable_type','')}: {getattr(d,'title','')[:60]}" for d in (ds or [])]) or ' 无' - qs = await sor.sqlExe("SELECT question,answer,status FROM pipeline_agent_questions WHERE task_id=${t}$ ORDER BY created_at DESC LIMIT 5",{"t":tid}) - ql = '\n'.join([f" [{getattr(q,'status','')}] Q:{getattr(q,'question','')[:60]}\n A:{getattr(q,'answer','') or '待回答'}" for q in (qs or [])]) or ' 无' - return f"任务: {getattr(t,'title','')}\n状态: {getattr(t,'state','')}\n角色: {getattr(t,'role','')}\n交付件:\n{dl}\n问答:\n{ql}" - elif tool == 'list_deliverables': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - tid = p.get('task_id','') - if tid: - ds = await sor.sqlExe("SELECT id,deliverable_type,title,review_status,quality_score FROM pipeline_deliverables WHERE task_id=${t}$ ORDER BY created_at DESC",{"t":tid}) - else: - ds = await sor.sqlExe("SELECT id,deliverable_type,title,review_status,quality_score FROM pipeline_deliverables WHERE project_id=${p}$ ORDER BY created_at DESC LIMIT 20",{"p":ctx['pid']}) - if not ds: return '无交付件' - return '\n'.join([f"[{getattr(d,'review_status','?')}] {getattr(d,'id','')[:8]} {getattr(d,'deliverable_type','')}: {getattr(d,'title','')[:60]} ({getattr(d,'quality_score','-')}分)" for d in ds]) - elif tool == 'view_deliverable': - did = p.get('deliverable_id','') - if not did: return 'FAIL: 需要交付件ID' - ds = await sor.sqlExe("SELECT deliverable_type,title,content,review_status,quality_score FROM pipeline_deliverables WHERE id=${d}$",{"d":did}) - if not ds: return 'FAIL: 交付件不存在' - d = ds[0] - return f"类型: {getattr(d,'deliverable_type','')}\n标题: {getattr(d,'title','')}\n状态: {getattr(d,'review_status','')} | 评分: {getattr(d,'quality_score','')}\n\n{getattr(d,'content','')[:3000]}" - elif tool == 'list_questions': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - qs = await sor.sqlExe("SELECT id,task_id,question,answer,status,created_at FROM pipeline_agent_questions WHERE tenant_id=${p}$ AND status='pending' ORDER BY created_at DESC LIMIT 10",{"p":ctx['pid']}) - if not qs: return '无待回答问题' - return '\n'.join([f"[{getattr(q,'id','')[:8]}] 任务:{getattr(q,'task_id','')[:8]} Q:{getattr(q,'question','')[:80]}" for q in qs]) - elif tool == 'answer_question': - qid = p.get('question_id','') - ans = p.get('answer','') - if not qid: return 'FAIL: 需要问题ID' - if not ans: return 'FAIL: 需要回答内容' - qs = await sor.sqlExe("SELECT id,task_id FROM pipeline_agent_questions WHERE id=${q}$",{"q":qid}) - if not qs: return 'FAIL: 问题不存在' - await sor.sqlExe("UPDATE pipeline_agent_questions SET answer=${a}$, answer_source='main_agent', answered_by='main_agent', status='answered' WHERE id=${q}$",{"a":ans,"q":qid}) - tid = getattr(qs[0],'task_id','') or '' - if tid: - await sor.sqlExe("UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL WHERE id=${t}$ AND state='waiting'",{"t":tid}) - return 'OK: 已回答' + ('，任务已恢复执行' if tid else '') - elif tool == 'start_agents': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - from pipeline_service.agent_loop import role_agent_run, pm_review_run - import asyncio as _asyncio - results = [] - for role in ['requirement','design','develop','test','deploy']: - r = await role_agent_run(ctx['pid'], role) - if r['status'] != 'idle': - results.append(f"{role}: {r['status']} task={r.get('task_id','')[:8]}") - pm_r = await pm_review_run(ctx['pid']) - if pm_r['status'] != 'idle': - results.append(f"pm_review: {pm_r['status']} task={pm_r.get('task_id','')[:8]}") - return '\n'.join(results) if results else '无待执行任务' - elif tool == 'diagnose_project': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - pid = ctx['pid'] - # 1) Stuck submitted tasks - stuck = await sor.sqlExe("SELECT id,title,role,state,created_at FROM pipeline_tasks WHERE tenant_id=${p}$ AND state IN ('submitted','waiting') ORDER BY created_at ASC LIMIT 10",{"p":pid}) - # 2) Failed tasks - failed = await sor.sqlExe("SELECT id,title,role,created_at FROM pipeline_tasks WHERE tenant_id=${p}$ AND state='failed' ORDER BY created_at DESC LIMIT 5",{"p":pid}) - # 3) Pending questions - qs = await sor.sqlExe("SELECT id,question,status,task_id FROM pipeline_agent_questions WHERE tenant_id=${p}$ AND status='pending' LIMIT 5",{"p":pid}) - # 4) Running tasks - running = await sor.sqlExe("SELECT id,title,role,claimed_by,TIMESTAMPDIFF(MINUTE,updated_at,NOW()) AS mins FROM pipeline_tasks WHERE tenant_id=${p}$ AND state='running' LIMIT 5",{"p":pid}) - # 5) Review tasks - review = await sor.sqlExe("SELECT id,title,role FROM pipeline_tasks WHERE tenant_id=${p}$ AND state='review' LIMIT 5",{"p":pid}) - lines = [f"项目诊断: {ctx['pname']}"] - if stuck: - lines.append(f"\n📌 卡住的任务({len(stuck)}个):") - for t in stuck: - lines.append(f" [{t.state}] [{t.role}] {t.title[:60]}") - if failed: - lines.append(f"\n❌ 失败任务({len(failed)}个):") - for t in failed: - lines.append(f" [{t.role}] {t.title[:60]}") - if running: - lines.append(f"\n🔄 运行中({len(running)}个):") - for t in running: - m = 0 - try: - m = int(getattr(t,'mins',0) or 0) - except Exception: - pass - cb = (getattr(t,'claimed_by','') or '')[:8] - flag = f"⚠️心跳超时{m}分钟(疑似僵尸)" if m >= 20 else f"已运行{m}分钟" - lines.append(f" [{t.role}] {t.title[:50]} | {flag} | claimed={cb}") - if review: - lines.append(f"\n👀 待审核({len(review)}个):") - for t in review: - lines.append(f" [{t.role}] {t.title[:60]}") - if qs: - lines.append(f"\n❓ 待回答问题({len(qs)}个):") - for q in qs: - lines.append(f" [{q.id[:8]}] {q.question[:80]}") - if not stuck and not failed and not qs: - lines.append("\n✅ 项目运行正常，无卡点") - else: - lines.append(f"\n💡 建议:") - if stuck: - lines.append(f" - {len(stuck)}个任务待执行 → 调用 start_agents 启动角色Agent") - if failed: - lines.append(f" - {len(failed)}个任务失败 → 检查失败原因，考虑重新提交") - if qs: - lines.append(f" - {len(qs)}个问题待回答 → 调用 list_questions + answer_question") - if review: - lines.append(f" - {len(review)}个交付件待PM审核 → PM agent会自动处理") - return '\n'.join(lines) - elif tool == 'add_bug': - if not ctx['pid']: return 'FAIL: 请先切换到项目' - title = p.get('title','')[:100] - if not title: return 'FAIL: 缺少标题' - await sor.C('sd_bugs',{'id':getID(),'iteration_id':'','title':title,'description':p.get('description',''),'severity':'major','priority':'P1','status':'open','reporter_type':'human','reporter_id':uid}) - return f"OK: Bug已记录 {title}" - return f'未实现: {tool}' - except Exception as e: - return f'ERROR: {str(e)[:300]}' - - -if action == 'send_message': - uid = await get_user() - if not uid: - return json.dumps({"error":"请先登录"},ensure_ascii=False) - org_id = await get_userorgid() or '0' - p = params_kw or {} - message_text = (p.get('message_text') or '').strip() - user_model_id = p.get('model_id', '') - # Try prompt at top level (bricks AgentIO) - if not message_text: - message_text = (p.get('prompt') or '').strip() - # Try JSON body from AgentIO - if not message_text: - inner = p.get('params', {}) - if isinstance(inner, dict): - message_text = (inner.get('prompt') or inner.get('message_text') or '').strip() - user_model_id = inner.get('model_id') or inner.get('llmid') or user_model_id - if not message_text: return json.dumps({"error":"message_text is required"},ensure_ascii=False) - - blocked, reason = _security_scan(message_text) - if blocked: return json.dumps({"success":True,"agent_reply":f"⚠️ {reason}"},ensure_ascii=False) - - debug(f"AUTH: uid={uid}, org_id={org_id}") - - async def agent_stream(): - async with DBPools().sqlorContext(dbname) as sor: - ctx = await _load_ctx(sor, uid) - model = await _sel_model(sor, user_model_id) - if not model: - d = json.dumps({"error": "没有可用的LLM模型配置"}, ensure_ascii=False)+'\n' - debug(f"YIELD error: {d[:80]}") - yield d - return - - repos_str = ', '.join([r['n'] for r in ctx['repos']]) or '无' - # Check for pending questions from role agents - pending_qs = [] - if ctx['pid']: - qs = await sor.sqlExe("SELECT id,question,task_id FROM pipeline_agent_questions WHERE tenant_id=${p}$ AND status='pending' ORDER BY created_at ASC LIMIT 5",{"p":ctx['pid']}) - pending_qs = [f" [{getattr(q,'id','')[:8]}] {getattr(q,'question','')[:100]}" for q in (qs or [])] - qinfo = '\n'.join(pending_qs) if pending_qs else '无' - env_text = f"【当前项目: {ctx['pname'] or '未选择'}】— 所有操作在当前项目内完成\n工作空间: {ctx['ws'] or '未设置'} | 仓库: {repos_str}\n待回答问题: {qinfo}" - system = AGENT_PROMPT.replace('__TOOLS__',TOOLS_TEXT).replace('__ENV__',env_text) - debug(f"SYSTEM PROMPT env: {env_text}") - msgs = [{"role":"system","content":system}] - # Aggregate history as reference, not as independent messages - hist_parts = [] - if uid: - sql = "SELECT role,content FROM pipeline_conversations WHERE created_by=${u}$ ORDER BY created_at ASC LIMIT 10" - params = {"u":uid} - if ctx['pid']: - sql = "SELECT role,content FROM pipeline_conversations WHERE created_by=${u}$ AND iteration_id=${p}$ ORDER BY created_at ASC LIMIT 10" - params = {"u":uid,"p":ctx['pid']} - hist_rows = await sor.sqlExe(sql, params) - debug(f"HISTORY: loaded {len(hist_rows or [])} rows, project={ctx['pid'][:8] if ctx['pid'] else 'none'}") - for h in (hist_rows or []): - c = getattr(h,'content','') or '' - r = getattr(h,'role','') or '' - if c.startswith('{"action":"tool_call"') or c.startswith('{"action": "tool_call"'): - continue - prefix = "用户" if r == 'user' else "助手" - hist_parts.append(f"{prefix}: {c[:300]}") - if hist_parts: - history_text = "参考历史记录:\n" + '\n'.join(hist_parts[-6:]) # last 6 exchanges only - msgs.append({"role":"system","content":history_text}) - # 处理用户上传的文件（multipart file 字段 → web_path），复制到工作空间 uploads/ 并作为中性上下文注入 - uploaded_files = [] - _fval = p.get('file') - _fpaths = _fval if isinstance(_fval, list) else ([_fval] if _fval else []) - for _fp in _fpaths: - try: - _abs = FileStorage().realPath(_fp) - _name = os.path.basename(_abs) - _ws = ctx.get('ws') or '' - if _ws and os.path.isdir(_ws): - _udir = os.path.join(_ws, 'uploads') - os.makedirs(_udir, exist_ok=True) - _dest = os.path.join(_udir, _name) - shutil.copy(_abs, _dest) - uploaded_files.append((_name, _dest)) - except Exception as _e: - debug(f"UPLOAD FAIL {_fp}: {_e}") - if uploaded_files: - _parts = [] - for _n, _d in uploaded_files: - _txt = _extract_text(_d, _n) - if _txt: - _parts.append(f"【文件 {_n} 内容】\n{_txt}") - else: - _parts.append(f"【文件 {_n}】无法直接读取文本（可能是二进制），相对路径 uploads/{_n}，可用 run_command 处理") - msgs.append({"role": "system", "content": "用户本次上传了文件，内容如下：\n\n" + "\n\n".join(_parts)}) - msgs.append({"role":"user","content":message_text}) - if uid: - await sor.C('pipeline_conversations',{'id':getID(),'role':'user','content':message_text,'msg_type':'text','org_id':org_id,'created_by':uid,'iteration_id':ctx['pid'] or ''}) - - agent_reply = '' - # Push active notifications before LLM processing - if pending_qs: - d = json.dumps(_w_md(f"📬 **待回答问题({len(pending_qs)}个)**:\n" + '\n'.join(pending_qs)), ensure_ascii=False)+'\n' - yield d - for turn in range(10): - raw = await _call_llm(model, msgs, 0.4) - act = _parse(raw) - debug(f"LLM turn {turn}: raw={raw[:200]}, action={act.get('action','?')}, tool={act.get('tool','?')}") - if act.get('action') == 'reply': - agent_reply = act.get('message', raw); break - if act.get('action') == 'tool_call': - tool = act.get('tool','') - label = _tool_labels.get(tool,tool) - debug(f"TOOL CALL: {tool} params={act.get('params',{})}") - d = json.dumps(_w_progress(f"🔄 {label}..."), ensure_ascii=False)+'\n' - debug(f"YIELD progress: {d[:60]}") - yield d - result = await _exec_tool(sor, tool, act.get('params',{}), ctx, uid, org_id) - # If result is a raw widget descriptor, yield it directly and break loop - if result.startswith('{"widgettype":'): - d = result + '\n' - debug(f"YIELD widget: {d[:60]}") - yield d - agent_reply = '__widget_handled__' - break - else: - ok = result.startswith("OK:") - d = json.dumps(_w_card(label, _w_text(result), "success" if ok else "error"), ensure_ascii=False)+'\n' - debug(f"YIELD result: {d[:60]}") - yield d - msgs.append({"role":"assistant","content":f"已调用 {tool}"}) - msgs.append({"role":"user","content":f"工具 {tool} 结果:\n{result}"}) - continue - agent_reply = raw; break - - if not agent_reply: agent_reply = "处理超时" - d = json.dumps(_w_md(agent_reply), ensure_ascii=False)+'\n' - debug(f"YIELD final: {d[:60]}") - yield d - if uid: - await sor.C('pipeline_conversations',{'id':getID(),'role':'agent','content':agent_reply,'msg_type':'text','org_id':org_id,'created_by':'system','iteration_id':ctx['pid'] or ''}) - - return await stream_response(request, agent_stream, 'text/plain; charset=utf-8') - -elif action == 'list_messages': - uid = await get_user() - if not uid: - return json.dumps({"error":"请先登录"},ensure_ascii=False) - async with DBPools().sqlorContext(dbname) as sor: - ms = await sor.sqlExe("SELECT role,content,created_at FROM pipeline_conversations ORDER BY created_at ASC LIMIT 100",{}) - result = [{"role":getattr(m,'role',''),"content":getattr(m,'content','')} for m in (ms or [])] - return json.dumps({"success":True,"messages":result},ensure_ascii=False,default=str) - -else: - return json.dumps({"error":f"Unknown: {action}"},ensure_ascii=False) diff --git a/wwwroot/sd_cockpit/index.ui b/wwwroot/sd_cockpit/index.ui deleted file mode 100644 index 5d8a0a4..0000000 --- a/wwwroot/sd_cockpit/index.ui +++ /dev/null @@ -1,190 +0,0 @@ -{ - "widgettype": "VBox", - "options": { - "width": "100%", - "height": "100%", - "padding": "0" - }, - "subwidgets": [ - { - "widgettype": "HBox", - "options": { - "width": "100%", - "alignItems": "center", - "padding": "16px 24px 8px 24px", - "cheight": 6, - "gap": "12px" - }, - "subwidgets": [ - { - "widgettype": "Title2", - "options": { - "text": "\u5f00\u53d1\u4ea7\u7ebf\u9a7e\u9a76\u8231" - } - }, - { - "widgettype": "Text", - "options": { - "text": "AI\u9a71\u52a8\u7684\u5bf9\u8bdd\u5f0f\u5f00\u53d1", - "cfontsize": 0.9, - "color": "#94a3b8" - } - }, - { - "widgettype": "Filler" - }, - { - "widgettype": "Form", - "id": "model_selector", - "options": { - "name": "model_selector", - "cols": 1, - "fields": [ - { - "name": "model_id", - "uitype": "code", - "label": "", - "placeholder": "\u9009\u62e9\u6a21\u578b", - "cwidth": 12, - "dataurl": "/pipeline-sdlc/api/cockpit_model_options.dspy", - "valueField": "model_id", - "textField": "model_id_text", - "params": { - "valueField": "model_id", - "textField": "model_id_text" - } - } - ] - } - }, - { - "widgettype": "Button", - "options": { - "name": "model_config", - "label": "\u6a21\u578b\u914d\u7f6e", - "css": "small" - }, - "binds": [ - { - "wid": "self", - "event": "click", - "actiontype": "script", - "target": "self", - "script": "var pw=new bricks.PopupWindow({title:'\u6a21\u578b\u914d\u7f6e',cwidth:36,cheight:26,auto_open:true});bricks.widgetBuild({widgettype:'urlwidget',options:{url:'/pipeline_core/llm/index.ui',method:'GET'}},pw.content_w).then(function(w){if(w)pw.content_w.add_widget(w);});" - } - ] - }, - { - "widgettype": "Button", - "options": { - "name": "wechat_config", - "label": "\u5fae\u4fe1\u901a\u9053", - "css": "small" - }, - "binds": [ - { - "wid": "self", - "event": "click", - "actiontype": "script", - "target": "self", - "script": "var pw=new bricks.PopupWindow({title:'\u5fae\u4fe1\u901a\u9053\u914d\u7f6e',cwidth:44,cheight:34,auto_open:true});bricks.widgetBuild({widgettype:'urlwidget',options:{url:'/pipeline-sdlc/wechat_config/index.ui',method:'GET'}},pw.content_w).then(function(w){if(w)pw.content_w.add_widget(w);});" - } - ] - } - ] - }, - { - "widgettype": "HBox", - "options": { - "width": "100%", - "padding": "8px 24px 8px 24px", - "cheight": 4, - "gap": "16px" - }, - "subwidgets": [ - { - "widgettype": "urlwidget", - "id": "stats_row", - "options": { - "url": "{{entire_url('/pipeline-sdlc/api/cockpit_stats.dspy')}}", - "method": "GET" - } - } - ] - }, - { - "widgettype": "VBox", - "options": { - "css": "filler", - "width": "100%", - "height": "100%", - "padding": "0 24px 0 24px", - "gap": "0" - }, - "subwidgets": [ - { - "widgettype": "VScrollPanel", - "id": "chat_scroll", - "options": { - "css": "filler", - "width": "100%", - "height": "100%", - "bgcolor": "#f8fafc", - "border": "1px solid #e2e8f0", - "borderBottom": "none", - "borderRadius": "8px 8px 0 0", - "padding": "16px" - }, - "subwidgets": [ - { - "widgettype": "VBox", - "options": { - "alignItems": "center", - "padding": "40px 0", - "gap": "12px" - }, - "subwidgets": [ - { - "widgettype": "Text", - "options": { - "text": "\u9009\u62e9\u4e00\u4e2a\u8fed\u4ee3\u540e\uff0c\u5728\u4e0b\u65b9\u8f93\u5165\u4f60\u7684\u9700\u6c42\u3002", - "cfontsize": 1, - "color": "#64748b" - } - }, - { - "widgettype": "Text", - "options": { - "text": "Agent \u5c06\u81ea\u52a8\u5206\u6790\u9700\u6c42\u3001\u751f\u6210\u8bbe\u8ba1\u3001\u9a71\u52a8\u4ea7\u7ebf\u6267\u884c\u3002", - "cfontsize": 0.85, - "color": "#94a3b8" - } - } - ] - } - ] - }, - { - "widgettype": "TextFiles", - "id": "chat_input", - "options": { - "bgcolor": "#fff", - "border": "1px solid #e2e8f0", - "borderTop": "none", - "borderRadius": "0 0 8px 8px", - "padding": "10px 14px 10px 14px" - }, - "binds": [ - { - "wid": "self", - "event": "inputed", - "actiontype": "script", - "target": "self", - "script": "var p=params.prompt;var files=params.add_files||[];var iid='';try{var iter=bricks.getWidgetById('current_iteration_id',bricks.app);if(iter)iid=iter.options.value||'';}catch(e){}var mid='';try{var ms=bricks.getWidgetById('model_selector',bricks.app);if(ms){var f=ms.form_element;if(f){var el=f.querySelector('[name=model_id]')||f.querySelector('select');if(el)mid=el.value||'';}}}catch(e){}var chat=bricks.getWidgetById('chat_scroll',bricks.app);var at=null;if(chat){var ub=new bricks.HBox({width:'100%'});var um=new bricks.VBox({width:'85%',alignSelf:'flex-end',bgcolor:'#dbeafe',borderRadius:'12px',padding:'12px 16px',marginBottom:'10px',gap:'4px'});um.add_widget(new bricks.Text({text:'\\u4f60',cfontsize:0.75,color:'#2563eb',fontWeight:'bold'}));um.add_widget(new bricks.Text({text:p,cfontsize:0.95,color:'#1e293b',whiteSpace:'pre-wrap'}));ub.add_widget(new bricks.VBox({css:'filler'}));ub.add_widget(um);ub.add_widget(new bricks.Svg({rate:2,url:bricks_resource('imgs/chat-user.svg')}));chat.add_widget(ub);var ab=new bricks.HBox({width:'100%'});var am=new bricks.VBox({width:'85%',alignSelf:'flex-start',bgcolor:'#e8f0fe',borderRadius:'12px',padding:'12px 16px',marginBottom:'10px',gap:'4px'});am.add_widget(new bricks.Text({text:'Agent',cfontsize:0.75,color:'#3b82f6',fontWeight:'bold'}));at=new bricks.Text({text:'\\u6b63\\u5728\\u5206\\u6790\\u9700\\u6c42...',cfontsize:0.85,color:'#64748b'});am.add_widget(at);ab.add_widget(new bricks.Svg({rate:2,url:bricks_resource('imgs/llm.svg')}));ab.add_widget(am);ab.add_widget(new bricks.VBox({css:'filler'}));chat.add_widget(ab);}var fd=new FormData();fd.append('message_text',p);fd.append('iteration_id',iid);fd.append('action','send_message');files.forEach(function(f){fd.append('file',f);});fetch('/pipeline-sdlc/api/cockpit_chat.dspy',{method:'POST',body:fd}).then(function(r){return r.json()}).then(function(r){if(r.success){if(at)at.set_text(r.agent_reply||'\\u5df2\\u5904\\u7406');var s=bricks.getWidgetById('stats_row',bricks.app);if(s)s.render({});}else{if(at)at.set_text('\\u9519\\u8bef: '+(r.error||'\\u672a\\u77e5\\u9519\\u8bef'));}});" - } - ] - } - ] - } - ] -} \ No newline at end of file