fix(orchestration): 元景三问题根治——task_title归位+问题通道统一冒泡+依赖启动策略+确认去重

问题1(部署早于开发启动):
- _check_orchestration_gaps 重构为结构化缺口(kind/fingerprint/task_id/fix_ids)
- 缺口不再只打日志:派发后当场回给PM + 持久化pipeline_pm_notices注入PM下回合上下文
- 新增 update_task_deps 原语:PM核实缺口后补依赖(修正是LLM决策,代码只查漏)

问题2(env补齐后waiting不唤醒):
- _bubble_deploy_env_missing 从自建人工任务死胡同迁移到统一问题通道(raise_problem)
- owner在待办回答→resolve_problem自动唤醒任务→QA注入agent prompt,信息真正交给LLM

问题3(task_title错位+重复确认):
- SDL_ROLES task_title 各归其位(需求分析/架构设计/脚手架开发/部署测试/功能测试)
- 确认去重:同迭代已确认且非回退重做→跳过重复确认直接派发
- _create_next_task 单例阶段幂等:requirement/design复用同迭代任意活跃任务

机制增强:
- 依赖启动策略 params.dep_policy: all(默认)/any/at_least{n},纯函数_eval_deps_policy
- init.py poller 手写依赖求值收敛到同一纯函数(消除两处语义漂移)
This commit is contained in:
ymq 2026-08-27 22:52:27 +08:00
parent 98b90d4f21
commit d9827c391b
4 changed files with 434 additions and 90 deletions

View File

@ -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"]}
]
}

View File

@ -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 个前置结束即启动
# 不变量(任何策略下都成立,代码守门):
# · 依赖任务不存在 → 阻塞(可能是编造 IDD1/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_policydict 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 任务自动 waitingsubmitted
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 修正
不需要再反查任务 IDG2 元景故障的教训告警只进日志没人看事实必须送到 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流转决策权归 LLMPM 核实后用本原语修正
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}

View File

@ -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:

View File

@ -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_proddeploy_prod 由用户明确指令触发
next_role="",
task_title="功能测试",
),
RoleSpec(
name="agent.deploy_prod",