# task_tree.dspy - 任务树（树状结构：项目 → 迭代 → 阶段 → 任务） # 懒加载协议： # 无 id → 返回项目根节点 # __root__ → 返回迭代列表（sd_iterations + 未分配迭代） # iter:{iter_id} → 返回阶段列表（需求/设计/开发/测试/部署，只显示有任务的） # iter:{iter_id}:phase:{role} → 返回该迭代该阶段的任务列表（叶子） import json uid = await get_user() if not uid: uid = 'user-01' node_id = (params_kw or {}).get('id', '').strip() dbname = get_module_dbname('pipeline-sdlc') _state_icons = { 'completed': '✅', 'approved': '✅', 'running': '🟡', 'failed': '❌', 'waiting': '⚠️', 'submitted': '⬜', 'review': '👀', 'rejected': '🚫', 'paused': '⏸️', 'cancelled': '🛑', 'qc_review': '🔍', } _state_zh = { 'completed': '已完成', 'approved': '已批准', 'running': '运行中', 'failed': '失败', 'waiting': '等待中', 'submitted': '已提交', 'review': '审核中', 'rejected': '已驳回', 'paused': '已暂停', 'cancelled': '已取消', } # 阶段顺序：从产线角色集（SDL_ROLES）动态生成，用 next_role 链推导线性角色（自动排除 qc 门禁）。 def _get_phase_order(): try: from pipeline_core import list_roles roles = list_roles("sdlc_general") or [] role_map = {r.name: r for r in roles} next_set = {getattr(r, 'next_role', '') for r in roles if getattr(r, 'next_role', '')} heads = [r for r in roles if r.name not in next_set and getattr(r, 'next_role', '')] order = [] seen = set() cur = heads[0] if heads else None while cur and cur.name not in seen: seen.add(cur.name) desc = getattr(cur, 'description', '') or cur.name order.append((cur.name, desc)) cur = role_map.get(getattr(cur, 'next_role', '')) return order except Exception: return [ ('agent.requirement', '需求阶段'), ('agent.design', '设计阶段'), ('agent.develop', '开发阶段'), ('agent.deploy_test', '测试环境部署'), ('agent.test', '测试阶段'), ('agent.deploy_prod', '生产环境部署'), ] def _iter_of(task): """从任务 params 里提取 iteration_id（存的是迭代名称，非 id）。""" params_str = getattr(task, 'params', '') or '' try: p = json.loads(params_str) if params_str else {} return (p.get('iteration_id') or '').strip() except Exception: return '' async with DBPools().sqlorContext(dbname) as sor: # 当前项目（2026-09-11 升级：会话级解析——session 优先→全局兜底， # 含悬空自愈+产线隔离；旧实现只读全局 pipeline_agent_settings，多 tab 串项目） _sid = (params_kw or {}).get('session_id', '') or '' pid = '' try: from pipeline_service.workspace import get_session_project_id pid = await get_session_project_id(sor, uid, _sid, 'sdlc_general') or '' except Exception: pid = '' if not pid: recs = await sor.sqlExe( "SELECT current_project_id FROM pipeline_agent_settings WHERE user_id=${u}$", {"u": uid}) pid = getattr(recs[0], 'current_project_id', '') if recs else '' if not pid: if not node_id: return json.dumps([{"id": "__root__", "label": "请先选择或新建项目", "is_leaf": True}], ensure_ascii=False) return json.dumps([], ensure_ascii=False) # 项目名 pname = '' precs = await sor.sqlExe("SELECT name FROM sd_projects WHERE id=${p}$", {"p": pid}) if precs: pname = getattr(precs[0], 'name', '') or '' # 一次查所有任务 + 所有迭代（任务数不大，Python 分组避免 SQL JSON 函数坑） all_tasks = await sor.sqlExe( "SELECT id, title, role, state, params, parent_id FROM pipeline_tasks WHERE tenant_id=${p}$ ORDER BY created_at ASC", {"p": pid}) or [] iters = await sor.sqlExe( "SELECT id, iteration_name, status FROM sd_iterations WHERE project_id=${p}$ ORDER BY created_at ASC", {"p": pid}) or [] if not node_id: # 根节点 return json.dumps([{"id": "__root__", "label": pname or "项目", "is_leaf": False}], ensure_ascii=False) if node_id == '__root__': # 迭代列表 nodes = [] for it in iters: iid = getattr(it, 'id', '') iname = getattr(it, 'iteration_name', '') or '' istatus = getattr(it, 'status', '') or '' if not iid: continue cnt = sum(1 for t in all_tasks if _iter_of(t) == iname) nodes.append({ "id": "iter:" + iid, "label": "🗂 {} [{}] ({})".format(iname, istatus, cnt), "is_leaf": False, }) # 未分配迭代（无 iteration_id 的任务）—— 有未分配任务才显示 none_cnt = sum(1 for t in all_tasks if not _iter_of(t)) if none_cnt > 0: nodes.append({ "id": "iter:__none__", "label": "🗂 未分配迭代 ({})".format(none_cnt), "is_leaf": False, }) return json.dumps(nodes, ensure_ascii=False) if node_id.startswith('iter:'): rest = node_id[5:] parts = rest.split(':phase:') iter_id = parts[0] phase_role = parts[1] if len(parts) > 1 else '' # 该迭代的迭代名 iter_name = '' if iter_id != '__none__': for it in iters: if getattr(it, 'id', '') == iter_id: iter_name = getattr(it, 'iteration_name', '') or '' break # 该迭代的任务 def _in_iter(t): if iter_id == '__none__': return not _iter_of(t) return _iter_of(t) == iter_name iter_tasks = [t for t in all_tasks if _in_iter(t)] if not phase_role: # 阶段列表 role_cnt = {} for t in iter_tasks: # 子任务（有 parent_id）挂在父任务下展示，不计入阶段计数， # 否则阶段节点计数与点开后的平铺任务数不一致（计数含子任务、平铺只列顶层任务）。 if (getattr(t, 'parent_id', '') or '').strip(): continue r = (getattr(t, 'role', '') or '').strip() role_cnt[r] = role_cnt.get(r, 0) + 1 nodes = [] for rn, rzh in _get_phase_order(): if rn in role_cnt: nodes.append({ "id": "iter:{}:phase:{}".format(iter_id, rn), "label": "📌 {} ({})".format(rzh, role_cnt[rn]), "is_leaf": False, }) if '' in role_cnt: nodes.append({ "id": "iter:{}:phase:__none__".format(iter_id), "label": "📌 其他 ({})".format(role_cnt['']), "is_leaf": False, }) return json.dumps(nodes, ensure_ascii=False) # 任务列表（只列顶层任务：parent_id 为空的；子任务挂在父任务下展开，避免平铺重复） nodes = [] for t in iter_tasks: r = (getattr(t, 'role', '') or '').strip() if phase_role == '__none__': if r: continue elif r != phase_role: continue tid = getattr(t, 'id', '') title = getattr(t, 'title', '') or '' state = getattr(t, 'state', '') or '' parent = (getattr(t, 'parent_id', '') or '').strip() if parent: continue # 子任务不在阶段平铺，挂在父任务下展示 if not tid or not title: continue icon = _state_icons.get(state, '⬜') zh = _state_zh.get(state, state) has_child = any((getattr(c, 'parent_id', '') or '').strip() == tid for c in all_tasks) nodes.append({ "id": tid, "label": "{} [{}] {}".format(icon, zh, title), "is_leaf": not has_child, }) return json.dumps(nodes, ensure_ascii=False) # 子任务层：裸任务ID → 返回该任务的子任务（parent_id == 该ID），递归支持多级分解 nodes = [] for t in all_tasks: if (getattr(t, 'parent_id', '') or '').strip() != node_id: continue tid = getattr(t, 'id', '') title = getattr(t, 'title', '') or '' state = getattr(t, 'state', '') or '' if not tid or not title: continue icon = _state_icons.get(state, '⬜') zh = _state_zh.get(state, state) has_child = any((getattr(c, 'parent_id', '') or '').strip() == tid for c in all_tasks) nodes.append({ "id": tid, "label": "{} [{}] {}".format(icon, zh, title), "is_leaf": not has_child, }) return json.dumps(nodes, ensure_ascii=False)