598 lines
37 KiB
Plaintext
598 lines
37 KiB
Plaintext
# cockpit_chat.dspy - SDLC Agent Loop v3.1 (streaming NDJSON)
|
||
import aiohttp
|
||
|
||
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()
|
||
ws = f'/d/pipeline/workspaces/{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,status FROM pipeline_agent_questions WHERE id=${q}$",{"q":qid})
|
||
if not qs: return 'FAIL: 问题不存在'
|
||
await sor.sqlExe("UPDATE pipeline_agent_questions SET answer=${a}$, status='answered', answered_at=NOW() WHERE id=${q}$",{"a":ans,"q":qid})
|
||
return f'OK: 已回答'
|
||
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 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:
|
||
lines.append(f" [{t.role}] {t.title[:60]}")
|
||
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})
|
||
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)
|