diff --git a/models/pipeline_pm_notices.json b/models/pipeline_pm_notices.json new file mode 100644 index 0000000..194940d --- /dev/null +++ b/models/pipeline_pm_notices.json @@ -0,0 +1,26 @@ +{ + "summary": [ + { + "name": "pipeline_pm_notices", + "title": "PM 编排查漏通知表", + "primary": ["id"], + "catelog": "entity" + } + ], + "fields": [ + {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "tenant_id", "title": "项目ID", "type": "str", "length": 32, "nullable": "no"}, + {"name": "task_id", "title": "关联任务ID", "type": "str", "length": 32, "nullable": "yes"}, + {"name": "fingerprint", "title": "缺口指纹(去重)", "type": "str", "length": 64, "nullable": "no"}, + {"name": "gap_kind", "title": "缺口类型(G1/G2/G3)", "type": "str", "length": 8, "nullable": "yes"}, + {"name": "gap_text", "title": "缺口事实描述", "type": "text"}, + {"name": "status", "title": "通知状态", "type": "str", "length": 16, "nullable": "no", "default": "pending"}, + {"name": "created_at", "title": "创建时间", "type": "timestamp", "nullable": "no"}, + {"name": "updated_at", "title": "更新时间", "type": "timestamp", "nullable": "no"} + ], + "indexes": [ + {"name": "idx_ppn_tenant", "idxtype": "index", "idxfields": ["tenant_id"]}, + {"name": "idx_ppn_status", "idxtype": "index", "idxfields": ["status"]}, + {"name": "idx_ppn_fp", "idxtype": "unique", "idxfields": ["tenant_id", "fingerprint"]} + ] +} diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index 17b5593..c637ad4 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -524,9 +524,10 @@ __ROLE_SKILLS__ - list_files(path) — 列目录 - git_status() — 查看git状态 - run_shell(command) — 执行命令（编译/测试验证） -- create_tasks(tasks) — 批量创建并派发后续任务（tasks 是 JSON 数组，每项 {title, role, description, key?, depends_on?, parent_id?}；key 供同批任务间 depends_on 引用，depends_on 是前序 key 或任务ID 数组，空=并行/非空=串行） +- create_tasks(tasks) — 批量创建并派发后续任务（tasks 是 JSON 数组，每项 {title, role, description, key?, depends_on?, dep_policy?, parent_id?}；key 供同批任务间 depends_on 引用，depends_on 是前序 key 或任务ID 数组，空=并行/非空=串行；dep_policy 是可选的启动策略：{"mode":"all"}（默认，前置全部结束才启动）/{"mode":"any"}（任一前置结束即启动）/{"mode":"at_least","n":k}（至少 k 个前置结束即启动）） - list_tasks(role, state) — 列出项目现有任务（派发前先查，避免重复） - cancel_task(task_id) — 取消任务（重做/作废前必须先取消旧任务，避免两个相同任务并存） +- update_task_deps(task_id, add_depends_on, dep_policy?) — 给已存在任务补前置依赖（系统发现编排缺口时通知你，核实后用此工具修正；add_depends_on 是要追加的任务ID数组，dep_policy 可选、同上） ## 审核流程 1. 先用工具检查代码/交付件是否实际产出（git_status / list_files / read_file） @@ -597,8 +598,8 @@ __ROLE_SKILLS__ # ── Agent 核心逻辑 ── -async def _task_deps_satisfied(sor, depends_on_raw, task_id=''): - """依赖门控：depends_on 全部处于终态(completed/approved)才可认领。 +async def _task_deps_satisfied(sor, depends_on_raw, task_id='', dep_policy=None): + """依赖门控：按启动策略求值（默认 all=全部终态；见 _eval_deps_policy）。 返回 (satisfied: bool, reason: str)。reason 非空说明未满足的原因，供逃逸阀冒泡用。 @@ -630,23 +631,92 @@ async def _task_deps_satisfied(sor, depends_on_raw, task_id=''): recs = await sor.sqlExe( "SELECT id, state, title FROM pipeline_tasks WHERE id IN (" + ids + ")", {}) await sor.sqlExe("COMMIT", {}) - found = {getattr(r, 'id', ''): (getattr(r, 'state', ''), getattr(r, 'title', '') or '') - for r in (recs or [])} + state_of = {} + title_of = {} + for r in (recs or []): + _id = getattr(r, 'id', '') + state_of[_id] = getattr(r, 'state', '') + title_of[_id] = getattr(r, 'title', '') or '' - DONE = ('completed', 'approved') - DEAD = ('cancelled', 'failed') # 永不可达终态 → 死锁，必须冒泡 - missing, dead, pending = [], [], [] + satisfied, reason, _kind = _eval_deps_policy(deps, state_of, dep_policy, title_of) + return satisfied, reason + + +# ── 依赖启动策略（2026-08-27 新增）────────────────────────────────────────── +# 每个任务可带 params.dep_policy 声明前置任务的启动策略（通用数据模型， +# 代码不编码任何「某角色必须依赖某角色」的业务规则，依赖编排决策权归 PM/LLM）： +# {"mode": "all"} 前置全部结束才启动（默认，等同历史行为） +# {"mode": "any"} 任一前置结束即启动 +# {"mode": "at_least", "n": k} 至少 k 个前置结束即启动 +# 不变量（任何策略下都成立，代码守门）： +# · 依赖任务不存在 → 阻塞（可能是编造 ID，D1/D3） +# · 依赖处于 cancelled/failed → 永不可达，阻塞并冒泡（D4） +# · 自依赖 / depends_on 解析失败 → 阻塞 +DEP_DONE = ('completed', 'approved') +DEP_DEAD = ('cancelled', 'failed') + + +def _parse_dep_policy(policy): + """解析 params.dep_policy → (mode, n)。缺省/非法 → ('all', 0)（保守兜底）。""" + if isinstance(policy, str): + try: + policy = json.loads(policy) + except (json.JSONDecodeError, TypeError, ValueError): + policy = None + if not isinstance(policy, dict): + return 'all', 0 + mode = str(policy.get('mode') or 'all').strip().lower() + if mode not in ('all', 'any', 'at_least'): + return 'all', 0 + n = 0 + if mode == 'at_least': + try: + n = int(policy.get('n') or 0) + except (TypeError, ValueError): + n = 0 + if n < 1: + return 'all', 0 # 非法 n → 保守退回 all + return mode, n + + +def _eval_deps_policy(deps, state_of, policy=None, title_of=None): + """纯函数：对依赖列表按启动策略求值（不访问 DB，两处调用方共用，语义单一来源）。 + + Args: + deps: 依赖任务 id 列表（已去重） + state_of: {dep_id: state} 状态映射；id 不在映射中 = 任务不存在 + policy: params.dep_policy（dict 或 JSON 字符串，可为空=默认 all） + title_of: 可选 {dep_id: title}，用于生成可读 reason + + Returns: + (satisfied: bool, reason: str, kind: str) + kind ∈ ('', 'dead', 'wait')： + '' 已满足（或无依赖） + 'dead' 永不可达（不存在/已作废/失败）→ 调用方应冒泡人工 + 'wait' 仅等待中 → 保持等待，不冒泡 + """ + title_of = title_of or {} + if not deps: + return True, '', '' + + mode, n = _parse_dep_policy(policy) + + # at_least 越界保护：n > 依赖总数时该策略永不可达（死等）， + # 保守钳到依赖总数——宁可提前放行也不能制造永久阻塞。 + if mode == 'at_least' and n > len(deps): + n = len(deps) + + missing, dead, pending, done = [], [], [], [] for d in deps: - if d not in found: + st = state_of.get(d) + if st is None: missing.append(d) # D1/D3：依赖不存在 → 阻塞（不再放行） - continue - st, ti = found[d] - if st in DONE: - continue - if st in DEAD: - dead.append(f"{ti or d[:8]}({st})") + elif st in DEP_DONE: + done.append(d) + elif st in DEP_DEAD: + dead.append(f"{title_of.get(d) or d[:8]}({st})") else: - pending.append(f"{ti or d[:8]}({st})") + pending.append(f"{title_of.get(d) or d[:8]}({st})") if missing or dead: parts = [] @@ -654,10 +724,21 @@ async def _task_deps_satisfied(sor, depends_on_raw, task_id=''): parts.append("依赖任务不存在：" + "、".join(x[:8] for x in missing)) if dead: parts.append("依赖已作废/失败，永不可达：" + "、".join(dead)) - return False, "；".join(parts) + "（需人工修正依赖或重建被依赖任务）" + return False, "；".join(parts) + "（需人工修正依赖或重建被依赖任务）", 'dead' + + if mode == 'any': + if done: + return True, '', '' + return False, "等待任一前置完成（启动策略=any）：" + "、".join(pending), 'wait' + if mode == 'at_least': + if len(done) >= n: + return True, '', '' + return False, (f"等待至少 {n} 个前置完成（启动策略=at_least {n}，" + f"已完成 {len(done)}）：" + "、".join(pending)), 'wait' + # mode == 'all'（默认） if pending: - return False, "等待依赖完成：" + "、".join(pending) - return True, '' + return False, "等待依赖完成：" + "、".join(pending), 'wait' + return True, '', '' PLACEHOLDER_MARKS = ('待明确', '待确认', '待补充', 'TODO', 'todo', 'xxx', 'XXX', '占位', '<', '未知') @@ -740,15 +821,24 @@ async def _check_deploy_env_ready(sor, project_id, role): async def _bubble_deploy_env_missing(sor, project_id, task_id, task_title, missing, env_file): - """部署环境信息缺失 → 冒泡给 owner 补齐（去重）。 + """部署环境信息缺失 → 走统一问题通道冒泡给项目 owner（去重）。 - 这是「逃逸阀」而非「门禁」：任务不会静默死等，owner 待办里会出现一条明确的 - 「补齐部署环境信息」任务，写清缺哪些字段、文件在哪。 + 统一冒泡机制（2026-08-27 定调：缺外部信息的问题一律走问题通道，不再按类型各写补丁）： + ① raise_problem 建问题记录、任务置 waiting（清 claimed_by）； + first_handler_agentid=owner → owner 的「我的待办」里出现一条待答问题 + （list_my_human_todos 已 union pipeline_agent_questions，按 agentid 命中）。 + ② owner 在待办里回答（待办详情已支持 question 渲染 + 回答框）→ + question_answer → resolve_problem → 任务自动 waiting→submitted。 + ③ poller 下轮重新派发 → 本函数调用方重跑 env 检查：补齐 → 认领， + _build_qna_section 把 owner 的回答注入 agent prompt（信息真正交给 LLM）； + 仍未补齐 → 旧问题已 answered、去重不拦 → 冒新一条。诚实循环，不静默死等。 """ - from .human_task_capability import create_human_task + from .communication import raise_problem + # 去重只看系统冒泡的（from_role='system.orchestrator'）——need_info 是通用类型， + # 角色 agent 缺信息也会用 need_info 提问，不加此条件会误拦。 exists = await sor.sqlExe( - "SELECT id FROM pipeline_human_tasks WHERE project_id=${pid}$ AND task_id=${tid}$ " - "AND task_type='deploy_env_required' AND status='pending' LIMIT 1", + "SELECT id FROM pipeline_agent_questions WHERE tenant_id=${pid}$ AND task_id=${tid}$ " + "AND problem_type='need_info' AND from_role='system.orchestrator' AND status='pending' LIMIT 1", {"pid": project_id, "tid": task_id}) await sor.sqlExe("COMMIT", {}) if exists: @@ -756,20 +846,22 @@ async def _bubble_deploy_env_missing(sor, project_id, task_id, task_title, missi _own = await sor.sqlExe("SELECT created_by FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) await sor.sqlExe("COMMIT", {}) owner_id = (getattr(_own[0], 'created_by', '') if _own else '') or 'user-01' - await create_human_task( - project_id, - f"补齐部署环境信息：{task_title or task_id[:8]}", - "部署任务无法开始：真实部署环境信息尚未提供（这属于外部输入，agent 无法自行产出，" - "因此不回退需求/设计/开发阶段，只等这里补齐）。\n\n" + question = ( + f"部署任务「{task_title or task_id[:8]}」无法开始：真实部署环境信息尚未提供。" + "这属于外部输入（agent 无法自行产出），因此不回退需求/设计/开发阶段，只等这里补齐。\n\n" f"配置文件：{env_file}\n\n" "缺失字段：\n" + "\n".join(f" · {m}" for m in missing) + - "\n\n补齐后本任务会自动继续（无需重跑上游阶段）。", - task_type='deploy_env_required', - assignee_id=owner_id, - created_by='system.orchestrator', - task_id=task_id, + "\n\n请补齐以上字段后提交回答（回答内容会随任务上下文交给执行 agent，" + "任务将自动恢复执行、重新校验环境信息）。" ) - logger.warning(f"部署环境信息缺失冒泡: task={task_id} missing={missing}") + await raise_problem( + "need_info", question, "system.orchestrator", + tenant_id=project_id, task_id=task_id, + first_handler_role="owner.superuser", first_handler_agentid=owner_id, + context={"env_file": env_file, "missing": missing}, + suspend_task=True, + ) + logger.warning(f"部署环境信息缺失冒泡(问题通道): task={task_id} missing={missing} owner={owner_id}") async def _bubble_blocked_dependency(sor, project_id, task_id, task_title, reason): @@ -825,8 +917,14 @@ async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True, blocked = [] # [(task_id, title, reason)] 依赖未满足的任务，供逃逸阀冒泡 for rec in recs: _tid = getattr(rec, 'id', '') + # 启动策略：任务可在 params.dep_policy 声明 all/any/at_least（缺省 all） + try: + _tp = json.loads(getattr(rec, 'params', '{}') or '{}') + except (json.JSONDecodeError, TypeError): + _tp = {} ok_dep, why = await _task_deps_satisfied( - sor, getattr(rec, 'depends_on', '') or '', task_id=_tid) + sor, getattr(rec, 'depends_on', '') or '', task_id=_tid, + dep_policy=_tp.get('dep_policy')) if not ok_dep: blocked.append((_tid, getattr(rec, 'title', '') or '', why)) continue @@ -1173,6 +1271,10 @@ async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_ta → 脚手架会先跑完，打破「应用级 develop approved = 编码完成」这个 deploy_test 门控前提 G3 design 产出的模块清单（spec.json generated_modules）有模块没有对应 develop 任务 → PM 漏派，模块不会被开发却照样部署 + + 返回结构化缺口列表 [{kind, fingerprint, text, task_id, fix_ids}]： + fingerprint 用于通知去重；task_id/fix_ids 让 PM 能直接调 update_task_deps 修正， + 不需要再反查任务 ID（G2 元景故障的教训：告警只进日志没人看，事实必须送到 PM 手上）。 """ gaps = [] _iter = _task_iteration_name(task) @@ -1206,9 +1308,11 @@ async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_ta title_of = {t['id']: t['title'] for t in tasks} for tid, ds in dep_map.items(): if tid in ds: - gaps.append(f"G1 任务「{title_of.get(tid, tid[:8])}」依赖自身，必然死锁") + gaps.append({'kind': 'G1', 'fingerprint': f"G1:{tid}", 'task_id': tid, 'fix_ids': [], + 'text': f"G1 任务「{title_of.get(tid, tid[:8])}」({tid}) 依赖自身，必然死锁"}) WHITE, GRAY, BLACK = 0, 1, 2 color = {k: WHITE for k in dep_map} + _cyc_seen = set() def _dfs(n, path): color[n] = GRAY @@ -1217,10 +1321,16 @@ async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_ta continue if color[m] == GRAY: cyc = path[path.index(m):] if m in path else [m] - gaps.append("G1 依赖成环，必然死锁：" + - " → ".join(title_of.get(x, x[:8]) for x in cyc + [m])) + fp = "G1:cycle:" + ",".join(sorted(cyc)) + if fp in _cyc_seen: + continue + _cyc_seen.add(fp) + gaps.append({'kind': 'G1', 'fingerprint': fp, 'task_id': (cyc[0] if cyc else ''), + 'fix_ids': [], + 'text': "G1 依赖成环，必然死锁：" + + " → ".join(title_of.get(x, x[:8]) for x in cyc + [m])}) elif color[m] == WHITE: - _dfs(m, path + [m]) + _dfs(m, path + [n]) color[n] = BLACK for n in list(dep_map.keys()): @@ -1237,13 +1347,17 @@ async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_ta for a in app_devs: if a['state'] not in ACTIVE and a['state'] != 'approved': continue - missing = [m['title'] for m in unfinished_mods if m['id'] not in a['deps']] - if missing: - gaps.append( - f"G2 应用级 develop「{a['title']}」未依赖以下未完成的模块任务：" - + "、".join(missing) - + " → 脚手架可能先于模块完成，使 deploy_test 在编码未完成时启动" - + f"（修法：给该任务 depends_on 补上这些模块任务 id）") + miss_tasks = [m for m in unfinished_mods if m['id'] not in a['deps']] + if miss_tasks: + miss_ids = [m['id'] for m in miss_tasks] + gaps.append({ + 'kind': 'G2', 'fingerprint': "G2:" + a['id'] + ":" + ",".join(sorted(miss_ids)), + 'task_id': a['id'], 'fix_ids': miss_ids, + 'text': (f"G2 应用级 develop「{a['title']}」({a['id']}) 未依赖以下未完成的模块任务：" + + "、".join(f"{m['title']}({m['id']})" for m in miss_tasks) + + " → 脚手架可能先于模块完成，使 deploy_test 在编码未完成时启动" + + "（修法：update_task_deps(task_id=\"" + a['id'] + + "\", add_depends_on=" + json.dumps(miss_ids) + ")）")}) # G3 设计模块清单 vs 实际派发的 develop 任务（漏派检测） try: @@ -1263,16 +1377,85 @@ async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_ta hit = any(str(m).lower() in (t['title'] or '').lower() for t in tasks if t['role'] == 'agent.develop') if not hit: - gaps.append( - f"G3 设计清单里的模块「{m}」没有对应的 develop 任务" - f"（{_os.path.basename(spec_path)} generated_modules）→ PM 漏派，" - f"该模块不会被开发却会进入部署") + gaps.append({ + 'kind': 'G3', 'fingerprint': f"G3:{m}", 'task_id': '', 'fix_ids': [], + 'text': (f"G3 设计清单里的模块「{m}」没有对应的 develop 任务" + f"（{_os.path.basename(spec_path)} generated_modules）→ PM 漏派，" + f"该模块不会被开发却会进入部署（修法：create_tasks 补派该模块）")}) except Exception as e: logger.debug(f"_check_orchestration_gaps G3 跳过: {e}") return gaps +def _gap_texts(gaps): + """结构化缺口 → 纯文本列表（日志用）。""" + return [g['text'] for g in (gaps or []) if isinstance(g, dict) and g.get('text')] + + +async def _save_gap_notices(sor, project_id, gaps, source_task_id=''): + """把编排查漏结果持久化为 PM 通知（pending），供 PM 下一回合注入上下文。 + + 去重：(tenant_id, fingerprint) 逻辑唯一——同一缺口重复检出只留一条。 + 实现用 SELECT 先查再 UPDATE/INSERT，不依赖 unique 索引（create_tables.py + 的幂等建表会跳过 CREATE INDEX 语句，索引可能不存在）。 + 已存在（含已 delivered）→ 重置回 pending：缺口未修复再次检出时， + PM 必须再看到（若用 INSERT IGNORE 会被挡住，缺口就永远静默了）。 + """ + from appPublic.uniqueID import getID + for g in (gaps or []): + if not isinstance(g, dict) or not g.get('fingerprint'): + continue + try: + _ex = await sor.sqlExe( + "SELECT id FROM pipeline_pm_notices WHERE tenant_id=${pid}$ AND fingerprint=${fp}$", + {"pid": project_id, "fp": g['fingerprint']}) + _tid = g.get('task_id') or source_task_id or '' + if _ex: + await sor.sqlExe( + "UPDATE pipeline_pm_notices SET status='pending', gap_text=${t}$, " + "task_id=${tid}$, gap_kind=${k}$, updated_at=NOW() WHERE id=${id}$", + {"t": g.get('text') or '', "tid": _tid, "k": g.get('kind') or '', + "id": getattr(_ex[0], 'id', '')}) + else: + await sor.sqlExe( + "INSERT INTO pipeline_pm_notices " + "(id, tenant_id, task_id, fingerprint, gap_kind, gap_text, status, created_at, updated_at) " + "VALUES (${id}$, ${pid}$, ${tid}$, ${fp}$, ${k}$, ${t}$, 'pending', NOW(), NOW())", + {"id": getID(), "pid": project_id, "tid": _tid, + "fp": g['fingerprint'], "k": g.get('kind') or '', "t": g.get('text') or ''}) + await sor.sqlExe("COMMIT", {}) + except Exception as e: + logger.debug(f"_save_gap_notices 跳过 {g.get('fingerprint')}: {e}") + + +async def _load_pm_notices(sor, project_id): + """取本项目全部 pending 编排通知并标记 delivered（每个缺口只进一次 PM 上下文，避免重复噪声）。 + + 返回拼接好的文本块；无通知返回 ''。 + """ + recs = await sor.sqlExe( + "SELECT id, gap_text FROM pipeline_pm_notices " + "WHERE tenant_id=${pid}$ AND status='pending' ORDER BY created_at ASC", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return '' + ids = [getattr(r, 'id', '') for r in recs] + ph = ",".join("${i%d}$" % n for n in range(len(ids))) + kw = {("i%d" % n): v for n, v in enumerate(ids)} + await sor.sqlExe( + "UPDATE pipeline_pm_notices SET status='delivered', updated_at=NOW() " + "WHERE id IN (" + ph + ")", kw) + await sor.sqlExe("COMMIT", {}) + lines = ["⚠️ 编排完备性检查发现以下缺口（代码查漏，修正决策权在你）："] + for r in recs: + lines.append("· " + (getattr(r, 'gap_text', '') or '')) + lines.append("请核实后用 update_task_deps / create_tasks / cancel_task 修正；" + "若缺口是误报（如任务已另行安排），忽略即可。") + return "\n".join(lines) + + async def _create_next_task(sor, project_id, task, next_role, pm_comment='', next_title='', next_desc=''): from appPublic.uniqueID import getID @@ -1337,13 +1520,20 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment='', "ORDER BY created_at DESC LIMIT 20", {"pid": project_id, "r": next_role}) await sor.sqlExe("COMMIT", {}) + # 单例阶段：需求/设计每迭代只有一个（不像 develop 按模块并行多任务）。 + # 系统派生时若同迭代已有活跃任务——无论来源（含 PM create_tasks 派发、无 previous_role）—— + # 都复用，不重复创建。元景故障：PM 派的真设计任务（无 previous_role）不在旧幂等匹配范围， + # 导致自动派生的错位设计任务与它并存、各自 approved 各触发一次设计确认。 + _single_stage = next_role in ('agent.requirement', 'agent.design') for _d in (_dup or []): try: _dp = json.loads(getattr(_d, 'params', '{}') or '{}') except (json.JSONDecodeError, TypeError): _dp = {} + if (_dp.get('iteration_id') or '') != _iter: + continue # 同迭代 + 同角色 + 同为系统派生（有 previous_role）→ 认定为同一阶段任务，幂等复用 - if (_dp.get('iteration_id') or '') == _iter and _dp.get('previous_role'): + if _dp.get('previous_role') or _single_stage: _did = getattr(_d, 'id', '') logger.info(f"_create_next_task 幂等命中：{next_role} 同迭代已有活跃任务 {_did}，跳过创建") return _did, getattr(_d, 'title', '') or new_title @@ -2101,6 +2291,12 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): tparams['task_kind'] = 'new_dev' if iteration_name: tparams["iteration_id"] = iteration_name + # 启动策略（可选）：PM 可为任务声明前置任务的启动方式 + # {"mode": "all"} / {"mode": "any"} / {"mode": "at_least", "n": k} + # 非法值会被 _parse_dep_policy 保守兜底为 all，不会破坏编排。 + _dp = t.get('dep_policy') + if isinstance(_dp, dict) and _dp.get('mode'): + tparams['dep_policy'] = _dp tid = getID() await sor.C('pipeline_tasks', { 'id': tid, 'tenant_id': project_id, 'pipeline_id': 'role_task', @@ -2164,10 +2360,28 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): return 'FAIL: 没有可创建的任务（缺少 title）' # C 之后立即 COMMIT，释放行锁，供 agent_poller 下一轮可见认领 await sor.sqlExe("COMMIT", {}) + # 编排查漏：派发后立即跑 G1/G2/G3，缺口当场返回给 PM（同回合即可修正）+ + # 持久化通知兜底（本回合未修，下一回合仍能看到）。元景故障教训：只打日志没人看。 + gap_note = '' + try: + _anchor = None + for _, _tid0, _, _ in created: + _ar = await sor.sqlExe("SELECT id, params FROM pipeline_tasks WHERE id=${i}$", {"i": _tid0}) + if _ar: + _anchor = _ar[0] + break + _gaps = await _check_orchestration_gaps(sor, project_id, _anchor) if _anchor else [] + if _gaps: + await _save_gap_notices(sor, project_id, _gaps, source_task_id=parent_task_id or '') + logger.warning(f"编排完备性告警(派发后) project={project_id}: {_gap_texts(_gaps)}") + gap_note = ("\n⚠️ 编排完备性检查发现缺口：\n" + "\n".join("· " + g['text'] for g in _gaps) + + "\n请核实后立即用 update_task_deps/create_tasks 修正。") + except Exception as _ge: + logger.warning(f"派发后编排查漏失败: {_ge}") note = f"（已自动取消 {len(cancelled)} 个同标题活跃任务）" if cancelled else "" warn_note = f"（{len(warnings)} 个未知依赖引用已忽略）" if warnings else "" return f"OK: 已派发 {len(created)} 个任务" + note + warn_note + "：" + \ - "；".join([f"{title}({role})" for _, _, title, role in created]) + "；".join([f"{title}({role})" for _, _, title, role in created]) + gap_note async def _pm_list_tasks(sor, project_id, params): @@ -2223,6 +2437,81 @@ async def _pm_cancel_task(sor, project_id, params): return f'OK: 已取消任务「{title}」({task_id})，正在执行的 agent 将在下一轮心跳中止' +async def _pm_update_task_deps(sor, project_id, params): + """PM 给已存在任务补前置依赖（G2/G3 查漏的修正出口）。 + + 代码只发现缺口并把事实交给 PM（流转决策权归 LLM），PM 核实后用本原语修正。 + params: {task_id, add_depends_on: [任务ID...], dep_policy?: {mode, n?}} + """ + task_id = (params.get('task_id') or params.get('id') or '').strip() + add = params.get('add_depends_on') or [] + if isinstance(add, str): + try: + add = json.loads(add) + except (json.JSONDecodeError, ValueError): + add = [] + add = [str(x).strip() for x in (add or []) if str(x).strip()] + if not task_id: + return 'FAIL: 需要 task_id' + if not add: + return 'FAIL: 需要 add_depends_on（要追加的任务ID数组）' + + recs = await sor.sqlExe( + "SELECT id, title, state, depends_on, params FROM pipeline_tasks " + "WHERE id=${tid}$ AND tenant_id=${pid}$", + {"tid": task_id, "pid": project_id}) + await sor.sqlExe("COMMIT", {}) + if not recs: + return 'FAIL: 任务不存在或不属于当前项目' + rec = recs[0] + title = getattr(rec, 'title', '') + state = getattr(rec, 'state', '') + if state in ('completed', 'cancelled', 'failed', 'approved'): + return f'FAIL: 任务「{title}」已是终态 {state}，改依赖无意义' + + # 校验追加的依赖真实存在（防编造 ID，与 _pm_create_tasks 的 D3 校验同标准） + resolved, bad = [], [] + for d in add: + _ex = await sor.sqlExe("SELECT id FROM pipeline_tasks WHERE id=${i}$", {"i": d}) + if _ex: + resolved.append(d) + else: + bad.append(d[:12]) + if bad: + return 'FAIL: 以下任务ID不存在：' + '、'.join(bad) + + # 合并现有 depends_on（去重、防自依赖） + try: + cur = json.loads(getattr(rec, 'depends_on', '') or '[]') + cur = [str(x) for x in cur if x] if isinstance(cur, list) else [] + except (json.JSONDecodeError, TypeError): + cur = [] + if task_id in resolved: + return 'FAIL: 不能依赖自身' + merged = cur + [d for d in resolved if d not in cur] + + # 可选更新启动策略 + try: + tparams = json.loads(getattr(rec, 'params', '{}') or '{}') + except (json.JSONDecodeError, TypeError): + tparams = {} + _dp = params.get('dep_policy') + if isinstance(_dp, dict) and _dp.get('mode'): + tparams['dep_policy'] = _dp + + await sor.sqlExe( + "UPDATE pipeline_tasks SET depends_on=${deps}$, params=${p}$, updated_at=NOW() WHERE id=${tid}$", + {"deps": json.dumps(merged, ensure_ascii=False), + "p": json.dumps(tparams, ensure_ascii=False), "tid": task_id}) + await sor.sqlExe("COMMIT", {}) + from .audit import record_audit + await record_audit(project_id, 'pipeline_tasks', task_id, 'update_deps', + who='agent.pm', + detail=f"depends_on += {resolved}", sor=sor) + return (f'OK: 任务「{title}」depends_on 已更新为 {merged}' + + (f'，启动策略={_dp}' if isinstance(_dp, dict) and _dp.get('mode') else '')) + + async def pm_review_run(project_id, agent_id=None, model_name=None): db = _get_db() async with db.sqlorContext("pipeline") as sor: @@ -2288,6 +2577,14 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): msgs = [{"role": "system", "content": pm_system}] msgs.append({"role": "user", "content": f"请审核以下交付件（类型:{deliverable_type}）：\n\n{content_preview}"}) + # 编排缺口通知注入：代码查漏发现的 G1/G2/G3 缺口（此前只进日志没人看）， + # 这里随任务上下文交给 PM——PM 可在本回合核实后用 update_task_deps 等修正。 + try: + _notices = await _load_pm_notices(sor, project_id) + if _notices: + msgs.append({"role": "user", "content": _notices}) + except Exception as _ne: + logger.warning(f"_load_pm_notices failed: {_ne}") from .llm_bridge import llm_call_msgs @@ -2344,6 +2641,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): result = await _pm_list_tasks(sor, project_id, params) elif tool in ('cancel_task', 'cancel'): result = await _pm_cancel_task(sor, project_id, params) + elif tool in ('update_task_deps', 'update_deps'): + result = await _pm_update_task_deps(sor, project_id, params) elif tool == 'load_skill': result = await _load_skill_by_name(sor, project_id, 'pm', org_id, params.get('name', ''), params.get('file_path') or None) else: @@ -2452,20 +2751,42 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): # 而是发「需求/设计确认」人工任务给项目 owner。owner 确认通过 → confirm_stage_gate # 继续派发下一角色；不确定 → 退回原角色重做（带修改意见）。 if task_role in ('agent.requirement', 'agent.design'): - _own = await sor.sqlExe( - "SELECT created_by FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) - await sor.sqlExe("COMMIT", {}) - owner_id = getattr(_own[0], 'created_by', '') if _own else '' - if not owner_id: - owner_id = 'user-01' # 历史项目 NULL 回填 admin confirm_type = 'requirement_confirmation' if task_role == 'agent.requirement' else 'design_confirmation' confirm_label = '需求' if task_role == 'agent.requirement' else '设计' + # 确认去重（2026-08-27 修复）：同迭代已确认过（status=done）且本任务非回退重做 + # → 不再重复发确认任务，直接走确认后派发。元景故障：错位的重复 design 任务各被 + # approved 一次，owner 连续收到两条「设计确认」待办。回退重做（有 rollback_from） + # 产出的是新交付件，必须重新确认，不走去重。 iter_id = '' if task_iter: _it2 = await sor.sqlExe( "SELECT id FROM sd_iterations WHERE project_id=${pid}$ AND iteration_name=${n}$", {"pid": project_id, "n": task_iter}) iter_id = getattr(_it2[0], 'id', '') if _it2 else '' + is_redo = bool(_tp.get('rollback_from')) + if iter_id and not is_redo: + _done_conf = await sor.sqlExe( + "SELECT id FROM pipeline_human_tasks WHERE iteration_id=${iid}$ " + "AND task_type=${ct}$ AND status='done' LIMIT 1", + {"iid": iter_id, "ct": confirm_type}) + await sor.sqlExe("COMMIT", {}) + if _done_conf: + logger.info( + f"确认去重：{confirm_type} 本迭代已确认过（非回退重做），" + f"跳过重复确认，直接派发 {next_role}: task={task_id}") + next_tid, next_title = await _create_next_task( + sor, project_id, task, next_role, comment, + next_title=decision.get('next_title', '') or '', + next_desc=decision.get('next_desc', '') or '') + return {"status": "approved", "task_id": task_id, + "next_task_id": next_tid, "next_role": next_role, + "comment": comment, "confirm_skipped": "already_confirmed"} + _own = await sor.sqlExe( + "SELECT created_by FROM sd_projects WHERE id=${pid}$", {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + owner_id = getattr(_own[0], 'created_by', '') if _own else '' + if not owner_id: + owner_id = 'user-01' # 历史项目 NULL 回填 admin from .human_task_capability import create_human_task ok, hid = await create_human_task( project_id, @@ -2492,12 +2813,13 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): sor, project_id, task, next_role, comment, next_title=decision.get('next_title', '') or '', next_desc=decision.get('next_desc', '') or '') - # 完备性校验（代码只查漏、不替 PM 决策）：把结构化事实回给 PM/日志， - # 漏派模块、脚手架缺依赖、依赖成环都在这里暴露。 + # 完备性校验（代码只查漏、不替 PM 决策）：缺口事实持久化为通知， + # 注入 PM 下一回合上下文（元景故障教训：只打日志没人看，事实必须送到 PM 手上）。 try: _gaps = await _check_orchestration_gaps(sor, project_id, task, next_role, next_tid) if _gaps: - logger.warning(f"编排完备性告警 task={next_tid}: {_gaps}") + logger.warning(f"编排完备性告警 task={next_tid}: {_gap_texts(_gaps)}") + await _save_gap_notices(sor, project_id, _gaps, source_task_id=next_tid) except Exception as _e: logger.warning(f"_check_orchestration_gaps failed: {_e}") return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment} diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 1f00bd4..4b51621 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -672,7 +672,7 @@ def load_pipeline_service(): # 排除 paused 项目：paused = 完全暂停推进（不派发/不审查/不审核），三个 poller 语义统一， # 否则会出现「任务照常派发执行、却卡在 PM 审核」的半吊子状态（2026-08 实测）。 recs = await sor.sqlExe( - "SELECT t.id, t.tenant_id, t.role, t.depends_on FROM pipeline_tasks t " + "SELECT t.id, t.tenant_id, t.role, t.depends_on, t.params FROM pipeline_tasks t " "WHERE t.state='submitted' AND t.pipeline_id='role_task' " "AND t.claimed_by IS NULL " "AND NOT EXISTS (SELECT 1 FROM sd_projects p WHERE p.id=t.tenant_id AND p.status='paused') " @@ -689,7 +689,11 @@ def load_pipeline_service(): # 依赖门控提前到候选筛选：批量查 depends_on 里所有依赖任务的状态， # 排除依赖未满足的「僵尸 submitted」任务——否则它们占住队列头、饿死其他就绪任务 - # （一项目卡死 → 全平台瘫痪）。终态 = completed/approved，与 _task_deps_satisfied 一致。 + # （一项目卡死 → 全平台瘫痪）。 + # 求值统一走 agent_loop._eval_deps_policy（纯函数，支持 all/any/at_least 启动策略）—— + # 此前这里是手写 all 语义，与 _task_deps_satisfied 并存构成两处实现， + # 语义漂移过（2026-08-25 修复注释可查）。现在单一来源，两处必一致。 + from .agent_loop import _eval_deps_policy all_dep_ids = set() for rec in (recs or []): all_dep_ids.update(_deps_of(getattr(rec, 'depends_on', '') or '')) @@ -703,35 +707,26 @@ def load_pipeline_service(): ready = [] blocked_dead = [] # 依赖永不可达 → 冒泡人工（逃逸阀） - DONE_ = ('completed', 'approved') - DEAD_ = ('cancelled', 'failed') for rec in (recs or []): _rid = getattr(rec, 'id', '') deps = _deps_of(getattr(rec, 'depends_on', '') or '') if not deps: ready.append(rec) continue - # 多依赖门控（2026-08-25 修复）：必须**逐个**确认依赖处于终态。 - # 原实现用 dep_state.get(d,'') 判定，虽能拦住「不存在」，但与 agent_loop 的 - # _task_deps_satisfied 语义不一致（后者查「不在终态」的行，依赖不存在时 - # 查出 0 行 → 误放行）。两处实现必须同语义，否则一处拦住另一处放行。 - _miss = [d for d in deps if d not in dep_state] - _dead = [d for d in deps if dep_state.get(d, '') in DEAD_] - _self = _rid in deps - if _miss or _dead or _self: - why = [] - if _miss: - why.append("依赖任务不存在：" + "、".join(x[:8] for x in _miss)) - if _dead: - why.append("依赖已作废/失败，永不可达：" + "、".join( - f"{x[:8]}({dep_state.get(x,'')})" for x in _dead)) - if _self: - why.append("依赖自身") + if _rid in deps: blocked_dead.append((_rid, getattr(rec, 'tenant_id', ''), - "；".join(why))) - continue # 永不放行，交人工 - if all(dep_state.get(d, '') in DONE_ for d in deps): + "依赖自身（" + _rid[:8] + "），永久阻塞，需人工修正 depends_on")) + continue + try: + _pp = json.loads(getattr(rec, 'params', '{}') or '{}') + except (json.JSONDecodeError, TypeError): + _pp = {} + _ok, _why, _kind = _eval_deps_policy(deps, dep_state, _pp.get('dep_policy')) + if _ok: ready.append(rec) + elif _kind == 'dead': + blocked_dead.append((_rid, getattr(rec, 'tenant_id', ''), _why)) + continue # 永不放行，交人工 # 逃逸阀：依赖永不可达的任务冒泡人工（去重由 _bubble_blocked_dependency 内部做） if blocked_dead: diff --git a/pipeline_service/sdlc_ability.py b/pipeline_service/sdlc_ability.py index 7b7103b..9a8047e 100644 --- a/pipeline_service/sdlc_ability.py +++ b/pipeline_service/sdlc_ability.py @@ -272,7 +272,7 @@ SDL_ROLES = [ aliases=["requirements", "requirement_analysis"], system_prompt="""你是需求分析师。先 load_skill 加载 role 技能，按其中的职责与应遵守规范执行任务。""", next_role="agent.design", - task_title="应用架构与模块设计", + task_title="需求分析", ), RoleSpec( name="agent.design", @@ -280,7 +280,7 @@ SDL_ROLES = [ aliases=["designer", "ui", "ux"], system_prompt="""你是系统设计师。先 load_skill 加载 role 技能，按其中的职责与应遵守规范执行任务。""", next_role="agent.develop", - task_title="应用脚手架开发", + task_title="应用架构与模块设计", ), RoleSpec( name="agent.develop", @@ -288,7 +288,7 @@ SDL_ROLES = [ aliases=["developer", "dev", "development", "coding"], system_prompt="""你是开发工程师。先 load_skill 加载 role 技能，按其中的职责与应遵守规范执行任务。""", next_role="agent.deploy_test", - task_title="部署测试", + task_title="应用脚手架开发", ), RoleSpec( name="agent.deploy_test", @@ -296,7 +296,7 @@ SDL_ROLES = [ aliases=["deploy_staging", "staging"], system_prompt="""你是测试环境部署工程师。先 load_skill 加载 role 技能，按其中的职责与应遵守规范执行任务。""", next_role="agent.test", - task_title="功能测试", + task_title="部署测试", ), RoleSpec( name="agent.test", @@ -305,6 +305,7 @@ SDL_ROLES = [ system_prompt="""你是测试工程师。先 load_skill 加载 role 技能，按其中的职责与应遵守规范执行任务。""", # 生产部署需人工指令：test 通过后不自动派发 deploy_prod，deploy_prod 由用户明确指令触发 next_role="", + task_title="功能测试", ), RoleSpec( name="agent.deploy_prod",