From 31eab892342a0fee9cf43a381c41245d5abae233 Mon Sep 17 00:00:00 2001 From: ymq Date: Wed, 26 Aug 2026 11:49:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(orchestration):=20=E5=A4=9A=E4=BE=9D?= =?UTF-8?q?=E8=B5=96=E9=97=A8=E6=8E=A7=204=20=E5=A4=84=E6=94=BE=E8=A1=8C?= =?UTF-8?q?=E6=BC=8F=E6=B4=9E=20+=20description=20=E7=BB=A7=E6=89=BF=20+?= =?UTF-8?q?=20=E5=B9=82=E7=AD=89=20+=20=E5=AE=8C=E5=A4=87=E6=80=A7?= =?UTF-8?q?=E6=9F=A5=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 【1. description 不再继承 + 用 PM 的 next_task_description】 _create_next_task 原实现 new_params={**params} 把上游 description 整个继承，导致 design 派生的「应用脚手架 develop」拿到需求任务的描述(产出需求规格说明书)，于是它交付 文档、4 分钟 approved —— 一行代码没写却走完 develop，打破「应用级 develop approved = 编码完成」这个 deploy_test 门控前提，造成 deploy_test 在编码未完成时提前启动。 同时 PM 在 review_approve 里给出的 next_task_title/next_task_description 一直被丢弃 (1935 行采集、2239 行调用时没传) —— 现已接上，流转决策权归 LLM，代码只做兜底。 【2. 多依赖门控 4 处漏洞(用户重点要求)】 D1 _task_deps_satisfied 查「不在终态的依赖」，依赖ID不存在时查出0行 → 误判满足而放行 D2 PM 派发时未知依赖引用只警告、静默丢弃 → 任务以更少依赖落库 → 并行开跑 D3 len>=20 就当有效任务ID收下、不校验存在性，配合 D1 让门控完全失效 D4 依赖处于 cancelled/failed 时依赖方永久 submitted，死锁且无人知晓 修法：逐个比对依赖存在性与状态；missing/dead/self 一律阻塞并冒泡；PM 派发时依赖 无法解析的任务置 waiting 而非静默降级为并行。init.py 的 poller 同步同语义(两处实现 必须一致，否则一处拦住另一处放行)。 【3. 幂等】并发多任务同时 approved 会各自创建下一阶段任务 → 同迭代重复任务(历史上出现 过「同一迭代两个 design 任务」)。按 (迭代, 角色, 系统派生) 先查后建。 【4. 逃逸阀通则】依赖永不可达时冒泡 dependency_blocked 人工任务(按 task_id 去重)， 消除静默死锁。对照：failed_poller 自动 pause 项目却无人通知，hrs7 因此卡近 1 小时。 【5. 完备性查漏 _check_orchestration_gaps】代码只查漏、不替 PM 决策： G1 依赖自引用/成环(DFS 回边) → 必然死锁 G2 应用级 develop 未依赖未完成的模块级 develop → 正是本次故障成因 G3 设计清单 generated_modules 有模块无对应 develop 任务 → PM 漏派 逻辑验证：/tmp 两组离线用例共 30 例全过，并以可执行方式证实原实现 4 处放行漏洞。 --- pipeline_service/agent_loop.py | 323 ++++++++++++++++++++++++++++++--- pipeline_service/init.py | 44 ++++- 2 files changed, 341 insertions(+), 26 deletions(-) diff --git a/pipeline_service/agent_loop.py b/pipeline_service/agent_loop.py index af74b58..dfe7a5d 100644 --- a/pipeline_service/agent_loop.py +++ b/pipeline_service/agent_loop.py @@ -595,30 +595,103 @@ __ROLE_SKILLS__ # ── Agent 核心逻辑 ── -async def _task_deps_satisfied(sor, depends_on_raw): - """依赖门控：depends_on 是 JSON 数组（依赖任务ID列表），全部处于终态(completed/approved) - 才可认领（返回 True）。空/无依赖 → 并行，立即满足。任一依赖未完成 → 串行等待（False）。 +async def _task_deps_satisfied(sor, depends_on_raw, task_id=''): + """依赖门控：depends_on 全部处于终态(completed/approved)才可认领。 - PM 分解任务时用 depends_on 表达「不能并行的串行依赖」；agent 认领前必须先过这道门， - 否则串行依赖会被 poller 提前认领、破坏编排顺序。 + 返回 (satisfied: bool, reason: str)。reason 非空说明未满足的原因，供逃逸阀冒泡用。 + + ⚠️ 多依赖正确性（2026-08-25 修复，原实现有 4 个漏洞会导致「依赖未完成就开始」）： + D1 原实现查「不在终态的依赖」，依赖 ID 不存在时查出 0 行 → 误判为已满足而放行。 + 现改为：先查回实际存在的依赖 id 集合，**逐个比对**——缺失的依赖视为未满足。 + D3 PM 可能编造 20 位 ID（_pm_create_tasks 对 len>=20 的引用不校验存在性）， + 配合 D1 会让门控完全失效。现在缺失即阻塞，编造 ID 不再能绕过。 + D4 依赖处于 cancelled/failed 等永不可达终态时，原实现让依赖方永久 submitted（死锁 + 且无人知晓）。现在识别为 dead 依赖并在 reason 里标注，由调用方冒泡人工处理。 + 自依赖：depends_on 含自身 id 会永久阻塞，识别为 dead 依赖。 """ deps = [] if depends_on_raw: try: d = json.loads(depends_on_raw) if isinstance(depends_on_raw, str) else depends_on_raw if isinstance(d, list): - deps = [str(x) for x in d if x] + deps = [str(x).strip() for x in d if x and str(x).strip()] except (json.JSONDecodeError, TypeError, ValueError): - deps = [] + # 解析失败不能当「无依赖」放行——那是把编排约束静默丢掉 + return False, f"depends_on 解析失败（原始值：{str(depends_on_raw)[:60]}），保守阻塞" + deps = list(dict.fromkeys(deps)) # 去重，保持顺序 if not deps: - return True - # 任一依赖不在终态 → 未满足。终态 = completed/approved（approved 视为已验收、可驱动下游）。 + return True, '' + if task_id and task_id in deps: + return False, f"依赖自身（{task_id[:8]}），永久阻塞，需人工修正 depends_on" + ids = ",".join(["'" + x.replace("'", "''") + "'" for x in deps]) recs = await sor.sqlExe( - "SELECT id FROM pipeline_tasks WHERE id IN (" + ids + ") " - "AND state NOT IN ('completed','approved')", {}) + "SELECT id, state, title FROM pipeline_tasks WHERE id IN (" + ids + ")", {}) await sor.sqlExe("COMMIT", {}) - return not recs + found = {getattr(r, 'id', ''): (getattr(r, 'state', ''), getattr(r, 'title', '') or '') + for r in (recs or [])} + + DONE = ('completed', 'approved') + DEAD = ('cancelled', 'failed') # 永不可达终态 → 死锁，必须冒泡 + missing, dead, pending = [], [], [] + for d in deps: + if d not in found: + 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})") + else: + pending.append(f"{ti or d[:8]}({st})") + + if missing or dead: + parts = [] + if missing: + parts.append("依赖任务不存在：" + "、".join(x[:8] for x in missing)) + if dead: + parts.append("依赖已作废/失败，永不可达：" + "、".join(dead)) + return False, "；".join(parts) + "（需人工修正依赖或重建被依赖任务）" + if pending: + return False, "等待依赖完成：" + "、".join(pending) + return True, '' + + +async def _bubble_blocked_dependency(sor, project_id, task_id, task_title, reason): + """逃逸阀：任务依赖「永不可达」时冒泡人工任务，避免静默死锁。 + + 设计原则（本次编排整改的通用不变量）： + **任何自动门禁都必须配一个「卡住 → 冒泡给谁」的出口。** + 没有出口的门禁 = 死锁制造机（对照：failed_poller 自动 pause 项目却无人通知， + 导致 hrs7 的 payroll/recruitment 卡了近 1 小时才被偶然发现）。 + + 去重：poller 每 10 秒扫一轮，同一任务不能每轮冒一个待办 —— 按 task_id 查已存在的 + pending 依赖阻塞任务，有则跳过。 + """ + from .human_task_capability import create_human_task + exists = await sor.sqlExe( + "SELECT id FROM pipeline_human_tasks WHERE project_id=${pid}$ AND task_id=${tid}$ " + "AND task_type='dependency_blocked' AND status='pending' LIMIT 1", + {"pid": project_id, "tid": task_id}) + await sor.sqlExe("COMMIT", {}) + if exists: + return + _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]}", + f"任务「{task_title}」的依赖永远无法满足，已停止等待，需人工处理。\n\n" + f"原因：{reason}\n\n" + f"处理方式：修正该任务的 depends_on，或重建被依赖的任务，或取消本任务。", + task_type='dependency_blocked', + assignee_id=owner_id, + created_by='system.orchestrator', + task_id=task_id, + ) + logger.warning(f"依赖阻塞冒泡: task={task_id} title={task_title} reason={reason}") async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True, set_state='running'): @@ -635,10 +708,23 @@ async def _claim_task(sor, tenant_id, role, state='submitted', match_role=True, # 依赖门控：跳过依赖未完成的任务，认领第一个依赖已满足的任务。 # （并行任务不被前面的串行任务阻塞；被依赖阻塞的任务保持 submitted 等依赖完成后下一轮认领。） task = None + blocked = [] # [(task_id, title, reason)] 依赖未满足的任务，供逃逸阀冒泡 for rec in recs: - if await _task_deps_satisfied(sor, getattr(rec, 'depends_on', '') or ''): + _tid = getattr(rec, 'id', '') + ok_dep, why = await _task_deps_satisfied( + sor, getattr(rec, 'depends_on', '') or '', task_id=_tid) + if ok_dep: task = rec break + blocked.append((_tid, getattr(rec, 'title', '') or '', why)) + # 逃逸阀：依赖「永不可达」（不存在/已作废/自依赖）的任务会永久 submitted 且无人知晓， + # 与「自动 pause 无出口」是同一类活性缺陷。这里冒泡人工任务，让 owner 能看到并修正。 + for _tid, _ti, _why in blocked: + if '永不可达' in _why or '不存在' in _why or '依赖自身' in _why or '解析失败' in _why: + try: + await _bubble_blocked_dependency(sor, tenant_id, _tid, _ti, _why) + except Exception as e: + logger.warning(f"_bubble_blocked_dependency failed task={_tid}: {e}") if task is None: return None task_id = task.id @@ -939,7 +1025,124 @@ async def _get_next_role(current_role, project_id=""): return "" -async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): +async def _check_orchestration_gaps(sor, project_id, task, next_role='', next_task_id=''): + """编排完备性校验：代码只**查漏**，不替 PM 决策。 + + 设计立场（2026-08-25 定调）：流转决策权归 LLM（PM 懂业务语义，知道模块怎么拆、依赖怎么连）， + 代码负责补 LLM 的固有短板——**不穷尽**。今天的实证：PM 给 3 个模块任务里的 2 个正确设了 + depends_on，却漏了应用脚手架那一个，导致脚手架先跑完、触发 deploy_test 提前启动。 + LLM 能做依赖判断（已证明），但会漏（也已证明）→ 代码查漏，把结构化事实回给 PM。 + + 返回 [告警字符串]，空 = 无问题。不阻断流程（阻断权仍在人和 PM），只暴露事实。 + + 检查项： + G1 依赖自引用 / 成环 → 必然死锁（我在方案走查阶段就发现自己的设计有这个洞） + G2 应用级 develop（脚手架）无 depends_on，但同迭代存在模块级 develop 未完成 + → 脚手架会先跑完，打破「应用级 develop approved = 编码完成」这个 deploy_test 门控前提 + G3 design 产出的模块清单（spec.json generated_modules）有模块没有对应 develop 任务 + → PM 漏派，模块不会被开发却照样部署 + """ + gaps = [] + _iter = _task_iteration_name(task) + + # 取同迭代全部活跃/终态任务（含 depends_on），一次查完供三项检查复用 + recs = await sor.sqlExe( + "SELECT id, title, role, state, depends_on, params FROM pipeline_tasks " + "WHERE tenant_id=${pid}$ AND state NOT IN ('cancelled') ORDER BY created_at", + {"pid": project_id}) + await sor.sqlExe("COMMIT", {}) + tasks = [] + for r in (recs or []): + try: + p = json.loads(getattr(r, 'params', '{}') or '{}') + except (json.JSONDecodeError, TypeError): + p = {} + if _iter and (p.get('iteration_id') or '') != _iter: + continue + deps = [] + try: + _d = json.loads(getattr(r, 'depends_on', '') or '[]') + deps = [str(x) for x in _d if x] if isinstance(_d, list) else [] + except (json.JSONDecodeError, TypeError): + deps = [] + tasks.append({'id': getattr(r, 'id', ''), 'title': getattr(r, 'title', '') or '', + 'role': getattr(r, 'role', '') or '', 'state': getattr(r, 'state', '') or '', + 'deps': deps, 'params': p}) + + # G1 自引用 / 成环（DFS 找回边） + dep_map = {t['id']: t['deps'] for t in tasks} + 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])}」依赖自身，必然死锁") + WHITE, GRAY, BLACK = 0, 1, 2 + color = {k: WHITE for k in dep_map} + + def _dfs(n, path): + color[n] = GRAY + for m in dep_map.get(n, []): + if m not in color: + 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])) + elif color[m] == WHITE: + _dfs(m, path + [m]) + color[n] = BLACK + + for n in list(dep_map.keys()): + if color.get(n) == WHITE: + _dfs(n, [n]) + + # G2 应用级 develop 缺依赖（本次故障的直接成因） + ACTIVE = ('submitted', 'running', 'review', 'qc_review', 'waiting') + module_devs = [t for t in tasks + if t['role'] == 'agent.develop' and not t['params'].get('previous_role')] + app_devs = [t for t in tasks + if t['role'] == 'agent.develop' and t['params'].get('previous_role')] + unfinished_mods = [t for t in module_devs if t['state'] in ACTIVE] + 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）") + + # G3 设计模块清单 vs 实际派发的 develop 任务（漏派检测） + try: + wdir = await _get_project_dir(sor, project_id) + if wdir: + import os as _os + import glob as _glob + for spec_path in _glob.glob(_os.path.join(wdir, '*_spec.json')): + with open(spec_path, encoding='utf-8') as f: + spec = json.load(f) + mods = spec.get('generated_modules') or [] + if not isinstance(mods, list): + continue + for m in mods: + if not m: + continue + 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"该模块不会被开发却会进入部署") + except Exception as e: + logger.debug(f"_check_orchestration_gaps G3 跳过: {e}") + + return gaps + + +async def _create_next_task(sor, project_id, task, next_role, pm_comment='', + next_title='', next_desc=''): from appPublic.uniqueID import getID title = getattr(task, 'title', '') or '' params_str = getattr(task, 'params', '{}') or '{}' @@ -949,8 +1152,8 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): params = {} # 标题：不再继承上一任务 title + 追加「（{next_role}阶段）」——那是 title 继承 bug 的根源， # 导致 design 任务名变成「需求规格说明书（agent.design阶段）」跟需求名混同，回退重做后更叠加成 - # 「...（回退重做）（agent.develop阶段）（agent.deploy_test阶段）」。改为按 next_role 的声明式 - # task_title 生成语义标题（项目名 + 阶段语义），阶段标题模板由 RoleSpec.task_title 声明。 + # 「...（回退重做）（agent.develop阶段）（agent.deploy_test阶段）」。优先用 PM 在 review_approve + # 里给出的 next_task_title（LLM 懂语义，流转决策权归它），缺失时按 RoleSpec.task_title 兜底。 stage = next_role try: pid = await _resolve_pipeline_id(project_id) @@ -967,17 +1170,52 @@ async def _create_next_task(sor, project_id, task, next_role, pm_comment=''): pname = getattr(_p[0], 'name', '') if _p else '' except Exception: pname = '' - new_title = f"{pname} {stage}" if pname else stage + new_title = (next_title or '').strip() or (f"{pname} {stage}" if pname else stage) + new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')), 'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment} - # 任务来源标记：系统自动派生（design→develop→deploy_test→test）都是「新开发」链， - # 不含 bug 修复语义。develop 角色据此判断是否要走 fix_bug 状态机。 + # ⚠️ description 不能继承（2026-08-25 修复）：原实现 new_params = {**params} 把上游任务的 + # description 整个带下来，导致 design 派生的「应用脚手架 develop」拿到的是**需求任务的描述** + # （"产出需求规格说明书与需求评审记录"），于是它交付了文档、4 分钟就 approved —— 一行代码没写 + # 却走完 develop 阶段，进而打破「应用级 develop approved = 编码已完成」这个 deploy_test 门控 + # 前提，造成 deploy_test 在编码未完成时提前启动。 + # 正确来源优先级：PM 的 next_task_description（LLM 按阶段语义写）> 阶段兜底描述。 + _desc = (next_desc or '').strip() + if not _desc: + _desc = (f"承接上一阶段「{title}」的产出，执行 {next_role} 阶段工作：{stage}。" + f"先 load_skill 加载 role 技能，按其中职责与规范执行，产出物按规范落盘。") + if pm_comment: + _desc += f"\n\n上一阶段 PM 审核意见：{pm_comment}" + new_params['description'] = _desc new_params['task_kind'] = 'new_dev' # 清除 pm_assigned：它是「PM 按模块清单派发」的标记，只作用于当前任务； # 下一角色任务是系统自动创建（非 PM 派发），若继承会污染——例如 PM 派发的 design 任务 # 带 pm_assigned=True，design approved 后自动创建的 develop 任务继承了它，被 1785 行的 # deploy_test 触发判断误判为「模块级 develop」，导致应用脚手架 develop approved 后跳过 deploy_test。 new_params.pop('pm_assigned', None) + + # 幂等（2026-08-25 新增）：并发场景下多个任务几乎同时 approved，每个都会走到这里尝试 + # 创建下一阶段任务 → 同一迭代同一角色出现多个重复任务（历史上已出现「同一迭代两个 design + # 任务」）。这里先查同迭代同角色的活跃任务，存在则复用、不重复创建。 + # 注：这是机制层不变量，不能靠 LLM 自觉（LLM 看不到并发）。 + _iter = new_params.get('iteration_id') or '' + _dup = await sor.sqlExe( + "SELECT id, title, params FROM pipeline_tasks WHERE tenant_id=${pid}$ AND role=${r}$ " + "AND state IN ('submitted','running','review','qc_review','waiting') " + "ORDER BY created_at DESC LIMIT 20", + {"pid": project_id, "r": next_role}) + await sor.sqlExe("COMMIT", {}) + for _d in (_dup or []): + try: + _dp = json.loads(getattr(_d, 'params', '{}') or '{}') + except (json.JSONDecodeError, TypeError): + _dp = {} + # 同迭代 + 同角色 + 同为系统派生（有 previous_role）→ 认定为同一阶段任务，幂等复用 + if (_dp.get('iteration_id') or '') == _iter and _dp.get('previous_role'): + _did = getattr(_d, 'id', '') + logger.info(f"_create_next_task 幂等命中：{next_role} 同迭代已有活跃任务 {_did}，跳过创建") + return _did, getattr(_d, 'title', '') or new_title + new_task_id = getID() await sor.C('pipeline_tasks', { 'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task', @@ -1745,25 +1983,47 @@ async def _pm_create_tasks(sor, project_id, params, parent_task_id=None): # 第二遍：解析 depends_on（key 或 任务ID）→ 真实任务ID，回填 depends_on 列。 # 串行依赖 = depends_on 非空（依赖完成才认领）；并行 = 空（不填）。 # 父任务本身在 PM approve 后即为 approved 终态，子任务无需再显式依赖父。 + # + # ⚠️ 依赖完整性（2026-08-25 修复 D2/D3）：原实现对未知引用**只警告、静默丢弃**， + # 对 len>=20 的引用**不校验存在性**。两者都会让任务以「更少的依赖」落库 → + # 依赖门控失效 → 依赖未完成就开跑（正是「多依赖时有依赖没完成就已经开始」的成因）。 + # 现在：未解析成功的引用一律阻塞该任务（state='waiting' + 冒泡），绝不静默降级为并行。 + dep_errors = [] # [(tid, title, [坏引用])] for t, tid, title, role in created: deps = t.get('depends_on') or [] if isinstance(deps, str): try: deps = json.loads(deps) except (json.JSONDecodeError, ValueError): - deps = [] + deps = [deps] if deps.strip() else [] resolved = [] + bad = [] for d in (deps or []): d = (str(d) or '').strip() if not d: continue if d in key_map: resolved.append(key_map[d]) # 同批 key 引用 + elif d == tid: + bad.append(f"{d}(依赖自身)") elif len(d) >= 20: - resolved.append(d) # 已有任务ID（跨批/父任务引用） + # D3：不能只看长度就当有效 ID —— 必须确认该任务真实存在 + _ex = await sor.sqlExe( + "SELECT id FROM pipeline_tasks WHERE id=${i}$", {"i": d}) + if _ex: + resolved.append(d) + else: + bad.append(f"{d[:12]}(任务不存在)") else: - warnings.append(f"「{title}」的 depends_on 未知引用「{d}」已忽略") - if resolved: + bad.append(f"{d}(未知 key)") + if bad: + # 依赖解析不完整 → 任务不进入可认领状态，冒泡人工修正，避免提前开跑 + dep_errors.append((tid, title, bad)) + await sor.sqlExe( + "UPDATE pipeline_tasks SET state='waiting', depends_on=${deps}$ WHERE id=${tid}$", + {"deps": json.dumps(resolved, ensure_ascii=False) if resolved else None, "tid": tid}) + warnings.append(f"「{title}」依赖无法解析（{'、'.join(bad)}）→ 已置 waiting 待人工修正") + elif resolved: await sor.sqlExe( "UPDATE pipeline_tasks SET depends_on=${deps}$ WHERE id=${tid}$", {"deps": json.dumps(resolved, ensure_ascii=False), "tid": tid}) @@ -2092,7 +2352,22 @@ async def pm_review_run(project_id, agent_id=None, model_name=None): f"给 owner={owner_id}, human_task={'OK' if ok else 'FAIL:' + str(hid)}") return {"status": "approved", "task_id": task_id, "next_role": next_role, "human_confirm": hid if ok else "", "comment": comment} - next_tid, next_title = await _create_next_task(sor, project_id, task, next_role, comment) + # 把 PM 给出的 next_task_title / next_task_description 传下去（原实现只传 comment， + # PM 的这两个字段被静默丢弃 → 派生任务只能继承上游 title/description，正是 + # 「design 任务名跟需求任务名一样」和「应用级 develop 拿到需求描述」的根源）。 + # 流转决策权归 LLM：PM 写了就用 PM 的；没写才用 RoleSpec 阶段模板兜底。 + 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 '') + # 完备性校验（代码只查漏、不替 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}") + 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} else: return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 135a973..1f00bd4 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -702,11 +702,51 @@ def load_pipeline_service(): await sor.sqlExe("COMMIT", {}) 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 '') - # 依赖 id 查不到（任务已删）→ 视为未满足，保守不放行 - if all(dep_state.get(d, '') in ('completed', 'approved') for d in deps): + 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("依赖自身") + blocked_dead.append((_rid, getattr(rec, 'tenant_id', ''), + "；".join(why))) + continue # 永不放行，交人工 + if all(dep_state.get(d, '') in DONE_ for d in deps): + ready.append(rec) + + # 逃逸阀：依赖永不可达的任务冒泡人工（去重由 _bubble_blocked_dependency 内部做） + if blocked_dead: + from .agent_loop import _bubble_blocked_dependency + for _bid, _bpid, _bwhy in blocked_dead: + if not (_bid and _bpid): + continue + try: + _bt = await sor.sqlExe( + "SELECT title FROM pipeline_tasks WHERE id=${i}$", {"i": _bid}) + await sor.sqlExe("COMMIT", {}) + _btitle = getattr(_bt[0], 'title', '') if _bt else '' + await _bubble_blocked_dependency(sor, _bpid, _bid, _btitle, _bwhy) + except Exception as _e: + debug(f"agent_poller 依赖阻塞冒泡失败 task={_bid}: {_e}") # 项目轮询（round-robin + 每项目并发上限）：每个项目轮流取一个就绪任务， # 保证项目间公平 + 单项目不占满全局并发名额、饿死其他项目。（dict 保持插入序）