fix(orchestration): 多依赖门控 4 处放行漏洞 + description 继承 + 幂等 + 完备性查漏

【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 处放行漏洞。
This commit is contained in:
ymq 2026-08-26 11:49:24 +08:00
parent c03469bf7a
commit 31eab89234
2 changed files with 341 additions and 26 deletions

View File

@ -595,30 +595,103 @@ __ROLE_SKILLS__
# ── Agent 核心逻辑 ── # ── Agent 核心逻辑 ──
async def _task_deps_satisfied(sor, depends_on_raw): async def _task_deps_satisfied(sor, depends_on_raw, task_id=''):
"""依赖门控:depends_on 是 JSON 数组(依赖任务ID列表),全部处于终态(completed/approved) """依赖门控:depends_on 全部处于终态(completed/approved)才可认领。
才可认领(返回 True)。空/无依赖 → 并行,立即满足。任一依赖未完成 → 串行等待(False)。
PM 分解任务时用 depends_on 表达「不能并行的串行依赖」;agent 认领前必须先过这道门, 返回 (satisfied: bool, reason: str)。reason 非空说明未满足的原因,供逃逸阀冒泡用。
否则串行依赖会被 poller 提前认领、破坏编排顺序。
⚠️ 多依赖正确性(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 = [] deps = []
if depends_on_raw: if depends_on_raw:
try: try:
d = json.loads(depends_on_raw) if isinstance(depends_on_raw, str) else depends_on_raw d = json.loads(depends_on_raw) if isinstance(depends_on_raw, str) else depends_on_raw
if isinstance(d, list): 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): except (json.JSONDecodeError, TypeError, ValueError):
deps = [] # 解析失败不能当「无依赖」放行——那是把编排约束静默丢掉
return False, f"depends_on 解析失败(原始值:{str(depends_on_raw)[:60]}),保守阻塞"
deps = list(dict.fromkeys(deps)) # 去重,保持顺序
if not deps: if not deps:
return True return True, ''
# 任一依赖不在终态 → 未满足。终态 = completed/approved(approved 视为已验收、可驱动下游)。 if task_id and task_id in deps:
return False, f"依赖自身({task_id[:8]}),永久阻塞,需人工修正 depends_on"
ids = ",".join(["'" + x.replace("'", "''") + "'" for x in deps]) ids = ",".join(["'" + x.replace("'", "''") + "'" for x in deps])
recs = await sor.sqlExe( recs = await sor.sqlExe(
"SELECT id FROM pipeline_tasks WHERE id IN (" + ids + ") " "SELECT id, state, title FROM pipeline_tasks WHERE id IN (" + ids + ")", {})
"AND state NOT IN ('completed','approved')", {})
await sor.sqlExe("COMMIT", {}) 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'): 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 等依赖完成后下一轮认领。) # (并行任务不被前面的串行任务阻塞;被依赖阻塞的任务保持 submitted 等依赖完成后下一轮认领。)
task = None task = None
blocked = [] # [(task_id, title, reason)] 依赖未满足的任务,供逃逸阀冒泡
for rec in recs: 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 task = rec
break 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: if task is None:
return None return None
task_id = task.id task_id = task.id
@ -939,7 +1025,124 @@ async def _get_next_role(current_role, project_id=""):
return "" 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 from appPublic.uniqueID import getID
title = getattr(task, 'title', '') or '' title = getattr(task, 'title', '') or ''
params_str = getattr(task, 'params', '{}') 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 = {} params = {}
# 标题:不再继承上一任务 title + 追加「({next_role}阶段)」——那是 title 继承 bug 的根源, # 标题:不再继承上一任务 title + 追加「({next_role}阶段)」——那是 title 继承 bug 的根源,
# 导致 design 任务名变成「需求规格说明书(agent.design阶段)」跟需求名混同,回退重做后更叠加成 # 导致 design 任务名变成「需求规格说明书(agent.design阶段)」跟需求名混同,回退重做后更叠加成
# 「...(回退重做)(agent.develop阶段)(agent.deploy_test阶段)」。改为按 next_role 的声明式 # 「...(回退重做)(agent.develop阶段)(agent.deploy_test阶段)」。优先用 PM 在 review_approve
# task_title 生成语义标题(项目名 + 阶段语义),阶段标题模板由 RoleSpec.task_title 声明。 # 里给出的 next_task_title(LLM 懂语义,流转决策权归它),缺失时按 RoleSpec.task_title 兜底。
stage = next_role stage = next_role
try: try:
pid = await _resolve_pipeline_id(project_id) 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 '' pname = getattr(_p[0], 'name', '') if _p else ''
except Exception: except Exception:
pname = '' 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', '')), new_params = {**params, 'previous_role': _normalize_role(getattr(task, 'role', '')),
'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment} 'previous_task_id': getattr(task, 'id', ''), 'pm_comment': pm_comment}
# 任务来源标记:系统自动派生(design→develop→deploy_test→test)都是「新开发」链, # ⚠️ description 不能继承(2026-08-25 修复):原实现 new_params = {**params} 把上游任务的
# 不含 bug 修复语义。develop 角色据此判断是否要走 fix_bug 状态机。 # 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' new_params['task_kind'] = 'new_dev'
# 清除 pm_assigned:它是「PM 按模块清单派发」的标记,只作用于当前任务; # 清除 pm_assigned:它是「PM 按模块清单派发」的标记,只作用于当前任务;
# 下一角色任务是系统自动创建(非 PM 派发),若继承会污染——例如 PM 派发的 design 任务 # 下一角色任务是系统自动创建(非 PM 派发),若继承会污染——例如 PM 派发的 design 任务
# 带 pm_assigned=True,design approved 后自动创建的 develop 任务继承了它,被 1785 行的 # 带 pm_assigned=True,design approved 后自动创建的 develop 任务继承了它,被 1785 行的
# deploy_test 触发判断误判为「模块级 develop」,导致应用脚手架 develop approved 后跳过 deploy_test。 # deploy_test 触发判断误判为「模块级 develop」,导致应用脚手架 develop approved 后跳过 deploy_test。
new_params.pop('pm_assigned', None) 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() new_task_id = getID()
await sor.C('pipeline_tasks', { await sor.C('pipeline_tasks', {
'id': new_task_id, 'tenant_id': project_id, 'pipeline_id': 'role_task', '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(key 或 任务ID)→ 真实任务ID,回填 depends_on 列。
# 串行依赖 = depends_on 非空(依赖完成才认领);并行 = 空(不填)。 # 串行依赖 = depends_on 非空(依赖完成才认领);并行 = 空(不填)。
# 父任务本身在 PM approve 后即为 approved 终态,子任务无需再显式依赖父。 # 父任务本身在 PM approve 后即为 approved 终态,子任务无需再显式依赖父。
#
# ⚠️ 依赖完整性(2026-08-25 修复 D2/D3):原实现对未知引用**只警告、静默丢弃**,
# 对 len>=20 的引用**不校验存在性**。两者都会让任务以「更少的依赖」落库 →
# 依赖门控失效 → 依赖未完成就开跑(正是「多依赖时有依赖没完成就已经开始」的成因)。
# 现在:未解析成功的引用一律阻塞该任务(state='waiting' + 冒泡),绝不静默降级为并行。
dep_errors = [] # [(tid, title, [坏引用])]
for t, tid, title, role in created: for t, tid, title, role in created:
deps = t.get('depends_on') or [] deps = t.get('depends_on') or []
if isinstance(deps, str): if isinstance(deps, str):
try: try:
deps = json.loads(deps) deps = json.loads(deps)
except (json.JSONDecodeError, ValueError): except (json.JSONDecodeError, ValueError):
deps = [] deps = [deps] if deps.strip() else []
resolved = [] resolved = []
bad = []
for d in (deps or []): for d in (deps or []):
d = (str(d) or '').strip() d = (str(d) or '').strip()
if not d: if not d:
continue continue
if d in key_map: if d in key_map:
resolved.append(key_map[d]) # 同批 key 引用 resolved.append(key_map[d]) # 同批 key 引用
elif d == tid:
bad.append(f"{d}(依赖自身)")
elif len(d) >= 20: 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: else:
warnings.append(f"「{title}」的 depends_on 未知引用「{d}」已忽略") bad.append(f"{d}(未知 key)")
if resolved: 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( await sor.sqlExe(
"UPDATE pipeline_tasks SET depends_on=${deps}$ WHERE id=${tid}$", "UPDATE pipeline_tasks SET depends_on=${deps}$ WHERE id=${tid}$",
{"deps": json.dumps(resolved, ensure_ascii=False), "tid": 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)}") f"给 owner={owner_id}, human_task={'OK' if ok else 'FAIL:' + str(hid)}")
return {"status": "approved", "task_id": task_id, "next_role": next_role, return {"status": "approved", "task_id": task_id, "next_role": next_role,
"human_confirm": hid if ok else "", "comment": comment} "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} return {"status": "approved", "task_id": task_id, "next_task_id": next_tid, "next_role": next_role, "comment": comment}
else: else:
return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"} return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}

View File

@ -702,11 +702,51 @@ def load_pipeline_service():
await sor.sqlExe("COMMIT", {}) await sor.sqlExe("COMMIT", {})
ready = [] ready = []
blocked_dead = [] # 依赖永不可达 → 冒泡人工(逃逸阀)
DONE_ = ('completed', 'approved')
DEAD_ = ('cancelled', 'failed')
for rec in (recs or []): for rec in (recs or []):
_rid = getattr(rec, 'id', '')
deps = _deps_of(getattr(rec, 'depends_on', '') or '') deps = _deps_of(getattr(rec, 'depends_on', '') or '')
# 依赖 id 查不到(任务已删)→ 视为未满足,保守不放行 if not deps:
if all(dep_state.get(d, '') in ('completed', 'approved') for d in deps):
ready.append(rec) 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 + 每项目并发上限):每个项目轮流取一个就绪任务, # 项目轮询(round-robin + 每项目并发上限):每个项目轮流取一个就绪任务,
# 保证项目间公平 + 单项目不占满全局并发名额、饿死其他项目。(dict 保持插入序) # 保证项目间公平 + 单项目不占满全局并发名额、饿死其他项目。(dict 保持插入序)