Planner 动态自规划 Work Graph（含节点判断与失败回路）——详细设计（SDD）
日期: 2026-08-08
性质: 详细设计，对应需求文档 docs/done/work-graph-self-planning-and-loop-back-requirements.md（已确认）。本设计给出可直接实施的 schema/接口/算法/测试方案，逐条落实需求文档「8. 已确认的决策」。
依赖代码：src/work_graph_orchestrate.py（run_graph/topological_layers/_run_one_node/_verify_and_maybe_regenerate/_ensure_node_row/_LockedConn）、src/work_graph_plan.py（_PROMPT_TEMPLATE/_normalize_graph_manifest/GRAPH_MANIFEST_SCHEMA）、src/e2e_pipeline.py（plan_documents/check_script_compiles/_check_prd_consistency）、src/settings_config.py（get_/upsert_concurrency_config 先例）、src/schema.py（migrate_expand_artifact_types 先例）。
状态: ✅ 已实现（2026-08-08）。实施顺序按第 11 节执行，python3 -m unittest discover -s src（692 tests）+ discover -s supervisor -p "test_*.py"（126 tests）全绿。实现过程中用真实测试发现并修正了本设计原文的几处问题（设计文档本身未回填，实现是最终准确来源）：
- §2 最初的
on_fail.target_node_id字段设计，在实现前按用户反馈改成了从on_failure边读取（已反映在正文），但 §4.0（topological_layers()必须过滤on_failure边）和 §8.3（_validate_loop_back_targets改成校验边而不是字段）是开始写代码之后才发现遗漏并现场修的两处，不是设计阶段就想全的。 _next_exec_node_id（原设计称_count_prior_attempts，按行数计数）在真实测试中暴露两个问题：① 按 DB 行数计数会把"resume 一个从未失败过的节点"误判成需要新开 attempt 后缀；② 只统计"自己是不是直接 loop_back 目标"会漏掉"作为下游节点被牵连重跑"的情况。最终实现改成按execution_event里真实的 loop_back 触发次数、且用_transitive_dependents判断是否被牵连来计数，两个问题都已修复并有专门测试覆盖。_run_one_node最初设计里"先统一调用 verify_fn 判断、再决定走哪条分支"的写法，会导致legacy（无 judgment 声明）节点的 verify_fn 被调用两次（_verify_and_maybe_regenerate内部自己还会再调一次）——测试跑出来直接暴露，改成先判断has_own_judgment决定路径，只有新路径才提前调用 verify_fn。work_graph_plan.py::_PROMPT_TEMPLATE最初设计的"先.format()填 catalog_json、留{goal_text}给以后再填一次"写法有真实的转义 bug（catalog JSON 自带的花括号会被第二次.format()误当成占位符），实现时改成用字符串拼接嵌入 catalog、最终用.replace("{goal_text}", goal_text)而不是.format()填充目标文本，彻底绕开转义问题。- 顺带在真实测试中发现一个不在本设计范围内、
work_node_run.py自己的既有缺口：run_work_node()每次调用都无条件往execution_log_file插入一行新记录，没有检查该node_exec_id是否已有记录——如果一个节点在execute_work_graph()真正返回之前就进程崩溃（不是"干净的 done"，也不是全新节点），下次对同一个node_exec_id调用run_work_node()会撞execution_log_file.node_exec_id的 UNIQUE 约束报错。这个缺口在 loop_back 机制之前就存在（任何节点崩溃在这个时间窗口都会触发,不是 loop_back 特有的),只是这次因为测试真实构造了"进程崩溃后重启"的场景才被发现——不在本次范围内修复,记录在案供后续单独立项。
1. 设计概览
现状（work_graph_orchestrate.py，见其自身 docstring 与代码）：run_graph() 把 graph_manifest 用 topological_layers() 分层，层间屏障、层内（G，2026-08-06）用 ThreadPoolExecutor 并发；每个节点的判断逻辑完全靠调用方传入的单一 verify_fn(node, output) -> (ok, feedback) 回调，失败处理固定死在 _verify_and_maybe_regenerate() 里——原地重新生成同一节点恰好一次，还失败就 ConsistencyCheckFailedError，没有"跳回更早节点"这条路径。
改造后新增两条能力，都以 graph_manifest 节点自身携带的声明式字段驱动，不再要求调用方为每种 artifact_type 手写 Python 分支：
judgment：节点自己说"怎么判断我的输出对不对"（机械检查 or 触发一次额外 LLM 判断）。on_fail：节点自己说"判断失败了怎么办"（原地重试 / 跳回更早节点重新执行 / 直接判失败）。
不改变的部分（刻意保持稳定）：
- 每个节点仍然是"渲染成一个单节点 workflow 脚本 + 独立
claude -p子进程调用"（_make_run_fn/_render_node_script不变）。 - G（并行执行）的层内并发机制、
_LockedConn、WorkGraphLayerError完全不变——回路是"层与层之间的控制流"，跟"层内节点并发执行"是正交的两层机制，互不冲突（详见第 4 节）。 - 不声明
judgment/on_fail的现有graph_manifest（Coder 的合成单节点图）逐字节行为不变（FR-A8，第 8 节有完整核对表）。 run_work_node()/work_node_run.py的断点续跑状态机完全不变，回路复用它现成的"新 node_id → 新行 → 真实重跑"机制（第 5 节），不需要改这一层。
2. graph_manifest 节点 schema 扩展
{
"node_id": "benchmark",
"artifact_type": "benchmark", // 沿用现状；_GENERIC_PROMPT_TEMPLATE 路径用 task_prompt 代替，同样适用
"depends_on": ["sdd"],
// 新增，可选。缺省等价于 {"kind": "none"} —— 不做任何自动判断，
// 节点输出直接被接受（如果调用方仍然传了外部 verify_fn，外部的继续生效，见第 3 节）。
"judgment": {
"kind": "mechanical", // "mechanical" | "llm" | "none"
"check_ref": "check_script_compiles" // 仅 kind="mechanical" 时必填，见第 3.1 节的白名单机制
// kind="llm" 时改用：
// "judge_prompt": "Review the following content for ...: {content}"
// —— 由规划者（Planner 自己，或人工在固定链里手写）自己撰写，
// self-contained，不依赖调用方额外拼接上下文（同 _GENERIC_PROMPT_TEMPLATE
// 的 task_prompt 设计原则）。
},
// 新增，可选。缺省等价于 {"action": "hard_fail"} 当没有 judgment 时；
// 当声明了 judgment 但没声明 on_fail 时，缺省等价于 {"action": "retry_self", "max_attempts": 1}
// ——即完全复刻现状 _verify_and_maybe_regenerate() 的行为（原地重试恰好一次）。
"on_fail": {
"action": "loop_back", // "retry_self" | "loop_back" | "hard_fail"
"max_attempts": 3 // 可选，覆盖 FR-A5 配置表给出的默认值
// 注意：没有 target_node_id 字段——action="loop_back" 的目标节点不是
// node 自己声明的一个独立字段，而是从 graph_manifest 的 edges 里读出来
// 的（见下）。retry_self/hard_fail 不需要任何 edge，target 就是自己/无。
}
}
2026-08-08 修订（用户反馈）：失败路径的目标节点不应该是节点自己声明的、跟 edges 脱节的一个独立字段——"下一个节点是谁（成功路径）、失败了回到哪个节点（失败路径），都应该在规划这个节点的时候，用边（edge）定义好"，跟成功路径的 depends_on/edges 是同一套结构，不是节点身上另开的一个影子字段。所以 on_fail.loop_back 的目标改成从一条新的 edge_type: "on_failure" 边读出来，不再单独存 target_node_id：
// graph_manifest 顶层 edges 数组：
{"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"}, // 成功路径：sdd 产出喂给 benchmark（既有语义，depends_on 的另一种表达）
{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"} // 失败路径：benchmark 判断失败时回到 sdd 重新执行
即：一个节点的"下一步去哪"完全由它参与的 edges 决定——needs_judgment/zero_token 两种既有类型继续表达"成功之后数据往哪流"，新增的 on_failure 表达"判断失败之后控制流跳去哪"。节点自己的 on_fail.action 只声明"失败了要不要 loop_back"这个决策本身，不重复存目标是谁，避免两处数据不一致的风险（比如 on_fail.target_node_id 和实际 edges 没对上）。GRAPH_MANIFEST_SCHEMA 的 edge_type 枚举（work_graph_plan.py:60）加上 "on_failure"。
GRAPH_MANIFEST_SCHEMA（work_graph_plan.py:33）同步补上 judgment/on_fail 的形状 + on_failure edge_type（文档性质，跟现状一样不接入真正的 jsonschema 校验——--json-schema 和 --agent 不能同时用，见该文件已有说明）。
3. run_graph() 改造：judgment 解析
3.1 机械检查白名单——注入而不是硬编码
work_graph_orchestrate.py 现在对 e2e_pipeline.py 零依赖（依赖方向是反过来的），不能直接 import e2e_pipeline 去拿 check_script_compiles（会成环）。机械检查改成调用方注入的白名单字典：
def run_graph(
conn, run_id, org_node, skill_id, target_folder_path, graph_manifest,
build_prompt, input_schema, output_schema, timeout=600, needs_worktree=False,
verify_fn=None, sleep=time.sleep, now=time.time, max_workers=_DEFAULT_MAX_WORKERS,
mechanical_check_registry=None, # 新增：{check_ref: callable(content) -> (ok, detail)}
):
e2e_pipeline.py 的调用点传 mechanical_check_registry={"check_script_compiles": check_script_compiles}。judgment.kind="mechanical" 但 check_ref 不在注册表里，视为配置错误，直接在 run_graph() 一开始（还没跑任何节点前）校验并抛 ValueError——不要等跑到那个节点才发现拼错了名字，浪费一整条链的 token。LLM 只能从白名单里选，不能发明新的机械检查——这是有意的限制，不做"LLM 生成检查代码后拿去执行"这种沙箱执行的大工程。
3.2 LLM 判断——复用 _check_prd_consistency 的隔离机制，泛化提示词
def _llm_judgment_verify_fn(judgment, org_node, target_folder_path, timeout=180):
"""通用化的 _check_prd_consistency（e2e_pipeline.py）：同样的 self-target
worktree 隔离 + 独立 agent() 调用机制，但判断提示词来自 judgment['judge_prompt']
（节点自己声明的，不是硬编码的 PRD 专属提示词），返回 schema 也通用化成
{"ok": bool, "issues": [str, ...]}（PRD 检查原来的 "consistent"/"issues" 改名
成更通用的 "ok"，语义不变）。"""
is_self = paths.is_self_target(target_folder_path)
def verify_fn(node, output):
node_exec_id = "judgment-" + uuid.uuid4().hex[:12]
worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None
cwd = worktree_path or target_folder_path
try:
prompt = judgment["judge_prompt"].format(content=output["content"])
script_name = "graphlemon-judgment-" + uuid.uuid4().hex
script_content = (
"export const meta = {{ name: {name}, description: 'graphlemon node judgment' }}\n\n"
"const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema} }})\n\n"
"return result\n"
).format(
name=json.dumps(script_name), prompt=json.dumps(prompt),
org_node_json=json.dumps(org_node), schema=json.dumps(_JUDGMENT_SCHEMA),
)
work_graph_execute.write_workflow_script(cwd, script_name, script_content)
result = work_graph_execute.execute_work_graph(cwd, script_name, {}, _JUDGMENT_SCHEMA, timeout=timeout)
if worktree_path:
work_node_run.ensure_worktree_committed(worktree_path, node_exec_id)
work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift")
work_node_run.merge_worktree(target_folder_path, node_exec_id)
ok = _coerce_bool(result["ok"])
return ok, (result.get("issues") or [])
finally:
if worktree_path:
work_node_run.remove_worktree(target_folder_path, worktree_path)
return verify_fn
_JUDGMENT_SCHEMA = {
"type": "object", "required": ["ok", "issues"],
"properties": {"ok": {"type": "boolean"}, "issues": {"type": "array", "items": {"type": "string"}}},
}
_coerce_bool（原属 e2e_pipeline.py，见其 2026-08-02 的"LLM 把 boolean 回成字符串"教训）需要下沉成 work_graph_orchestrate.py 自己的小工具函数（或 paths.py 这类共享工具模块），两边都用同一份，不重复实现。e2e_pipeline.py::_check_prd_consistency/_prd_consistency_verify_fn 保留不动（FR-A8 向后兼容，现有固定链路径不强制迁移），新代码只是提供一个可选的、更通用的平行实现。
3.3 节点级 verify_fn 解析
def _resolve_node_verify_fn(node, external_verify_fn, org_node, target_folder_path, mechanical_check_registry):
judgment = node.get("judgment")
if not judgment or judgment.get("kind", "none") == "none":
return external_verify_fn # 完全现状：没声明 judgment，谁传了外部 verify_fn 就用谁的（可能是 None）
if judgment["kind"] == "mechanical":
check_fn = (mechanical_check_registry or {}).get(judgment["check_ref"])
if check_fn is None:
raise ValueError("unknown mechanical check_ref: {!r}".format(judgment["check_ref"]))
def verify_fn(node, output, _check_fn=check_fn):
ok, detail = _check_fn(output["content"])
return ok, ([detail] if detail else None)
return verify_fn
if judgment["kind"] == "llm":
return _llm_judgment_verify_fn(judgment, org_node, target_folder_path)
raise ValueError("unknown judgment.kind: {!r}".format(judgment["kind"]))
_run_one_node() 改成对每个节点各自调用 _resolve_node_verify_fn(...) 拿到"这个节点该用哪个 verify_fn"，而不是像现在这样整张图共用调用方传入的同一个 verify_fn。向后兼容关键点：现有调用方（plan_documents/validate_implementation_with_own_graph）继续传它们现在的 _combined_planner_verify_fn(...)/_test_script_verify_fn(...) 作为 external_verify_fn，只要节点没有 judgment 字段（现状 100% 如此），_resolve_node_verify_fn 原样返回这个外部函数，行为逐字节不变（FR-A8 验收标准 5）。
4. on_fail 执行：回路引擎
4.0 前提修复：topological_layers() 必须过滤 on_failure 边（FR-A7）
这是把 loop_back 目标从节点字段改成 edges 里的一条边（第 2 节 2026-08-08 修订）之后，必须同步修的一处，否则会引入真环：现有 topological_layers()（work_graph_orchestrate.py:196-200）把 graph_manifest["edges"] 里每一条边、不分 edge_type都当成数据依赖合并进 deps：
for edge in graph_manifest.get("edges", []):
to_id, from_id = edge.get("to"), edge.get("from")
if to_id in deps and from_id in nodes:
deps[to_id].add(from_id)
如果不改，一条 {"from": "benchmark", "to": "sdd", "edge_type": "on_failure"} 会被这段代码当成"sdd 依赖 benchmark"塞进 deps["sdd"]——而 benchmark 本来就依赖 sdd（正常数据依赖），两条方向相反的依赖凑在一起就是一个真环，topological_layers() 会在图规划阶段还没触发任何回路之前，就先被静态校验直接拒绝，FR-A7 的验收标准 4 会失败。
修复：合并 deps 时跳过 edge_type == "on_failure" 的边——只有 on_failure 需要特殊处理，zero_token/needs_judgment 两种既有类型继续按现状合并（它们表达的就是真实数据依赖）：
def topological_layers(graph_manifest):
nodes = {n["node_id"]: n for n in graph_manifest["nodes"]}
deps = {node_id: set(n.get("depends_on") or []) for node_id, n in nodes.items()}
for edge in graph_manifest.get("edges", []):
if edge.get("edge_type") == "on_failure":
continue # 新增：控制流边不参与静态拓扑排序
to_id, from_id = edge.get("to"), edge.get("from")
if to_id in deps and from_id in nodes:
deps[to_id].add(from_id)
... # 其余逻辑不变
topological_order()（薄封装）不用改。这条修复是本设计里唯一一处直接改动 G（并行执行）已经上线的现有函数的地方，改动是纯粹的"多一个 continue 条件"，不影响任何现有（不含 on_failure 边的）graph_manifest 的分层结果——src/test_work_graph_orchestrate.py 里 G 自己的既有测试用例不需要改一行断言即可继续通过。
4.1 关键简化：retry_self 是 loop_back 的特例
retry_self ≡ loop_back 且 target_node_id = 自己。原因：retry_self 触发时这个节点本身还没成功过，不可能已经有下游节点跑完并读取过它的（不存在的）输出，所以"重跑 target 到失败节点之间的下游链"这条规则应用到"target=自己"时，受影响集合就是 {自己}，跟现状的"原地重跑"完全等价。一套机制两种声明方式，不用维护两条独立代码路径。
4.2 受影响子图：只重跑真正的下游依赖，不是整层
需求文档非目标已声明"不追求完全通用的图执行框架"，但"回路触发时把同层不相关的兄弟节点也无谓重跑一遍"没必要，也和已完成节点白白重花 token 相悖，成本可控且实现不复杂：
def _transitive_dependents(graph_manifest, node_id):
"""node_id 自身 + 所有直接/间接依赖它的下游节点 id 集合（沿 depends_on/edges
正向遍历）。用于 loop_back 时判断哪些节点的输出已经"过期"、必须重跑。"""
dependents = {} # node_id -> [直接依赖它的节点 id, ...]
for node in graph_manifest["nodes"]:
dependents.setdefault(node["node_id"], [])
deps = {n["node_id"]: set(n.get("depends_on") or []) for n in graph_manifest["nodes"]}
for edge in graph_manifest.get("edges", []):
to_id, from_id = edge.get("to"), edge.get("from")
if to_id in deps and from_id in deps:
deps[to_id].add(from_id)
for node_id_, dep_set in deps.items():
for dep in dep_set:
dependents.setdefault(dep, []).append(node_id_)
affected, stack = set(), [node_id]
while stack:
current = stack.pop()
if current in affected:
continue
affected.add(current)
stack.extend(dependents.get(current, []))
return affected
def _on_failure_target(graph_manifest, node_id):
"""节点 node_id 声明了 on_fail.action='loop_back' 时，目标节点不是节点
自己存的字段，而是 graph_manifest['edges'] 里唯一一条
{"from": node_id, "to": ..., "edge_type": "on_failure"} 边的 "to"——
2026-08-08 修订：失败路径跟成功路径一样，都是"边"，不是节点身上另开
的一个跟 edges 脱节的影子字段。恰好 0 条或 2+ 条这样的边是规划错误，
在 _validate_loop_back_targets（8.3 节）里提前拦截，这里假设已经过
校验，直接找第一条匹配的。"""
for edge in graph_manifest.get("edges", []):
if edge.get("from") == node_id and edge.get("edge_type") == "on_failure":
return edge["to"]
raise ValueError("node {} declares on_fail.action='loop_back' but has no on_failure edge".format(node_id))
4.3 层级回退循环——run_graph() 主循环改造
class _LoopBackRequested(Exception):
"""节点内部信号：这个节点的 on_fail.action=loop_back 触发了，需要外层
run_graph() 主循环回退重跑，不是真正的失败——不通过正常返回值传递是
因为它可能发生在 ThreadPoolExecutor 的 worker 线程里（层内并发场景），
异常是唯一能干净地从 future.result() 里带出这个信号的方式。"""
def __init__(self, source_node_id, target_node_id, feedback):
self.source_node_id = source_node_id
self.target_node_id = target_node_id
self.feedback = feedback
super().__init__("node {} requests loop_back to {}".format(source_node_id, target_node_id))
def run_graph(..., mechanical_check_registry=None):
ensure_placeholder_skill(conn, skill_id, org_node)
effective_needs_worktree = needs_worktree or paths.is_self_target(target_folder_path)
layers = topological_layers(graph_manifest)
node_layer_index = {node["node_id"]: i for i, layer in enumerate(layers) for node in layer}
outputs = {}
locked_conn = _LockedConn(conn)
attempt_counts = {} # (target_node_id, source_node_id) -> 已尝试次数，本次 run_graph() 调用内的内存计数
# 断点续跑时从 execution_event 重建，见第 6.3 节
layer_index = 0
while layer_index < len(layers):
layer = layers[layer_index]
try:
_run_layer(
conn, locked_conn, run_id, org_node, skill_id, target_folder_path, layer, outputs,
build_prompt, input_schema, output_schema, timeout, effective_needs_worktree,
verify_fn, mechanical_check_registry, max_workers, sleep, now,
)
except _LoopBackRequested as sig:
key = (sig.target_node_id, sig.source_node_id)
attempt_counts[key] = attempt_counts.get(key, 0) + _loop_attempt_count_from_events(conn, run_id, key)
limit = _resolve_loop_limit(conn, org_node, graph_manifest, sig.target_node_id)
if attempt_counts[key] > limit:
conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
conn.commit()
raise ConsistencyCheckFailedError(
"loop_back from {} to {} exceeded max_attempts={}".format(
sig.source_node_id, sig.target_node_id, limit
),
node_id=sig.source_node_id,
)
_record_loop_back_event(conn, run_id, sig, attempt_counts[key])
affected = _transitive_dependents(graph_manifest, sig.target_node_id)
for affected_id in affected:
outputs.pop(affected_id, None) # 强制这些节点在重跑时被当成"未完成"
layer_index = node_layer_index[sig.target_node_id]
continue
layer_index += 1
return outputs
_run_layer(...) 是现有 run_graph() 主循环体（单节点同步 / 多节点 ThreadPoolExecutor）原样搬过来的辅助函数，唯一的改动是：只重跑本轮 affected 集合里的节点（第一次进入某层时 affected 概念不适用，正常跑全部；loop_back 回退重入某层时，只有 outputs 里被 .pop() 掉的节点需要重新执行，其余节点直接复用已有的 outputs[node_id]，不重新调用 _run_one_node）——outputs.pop() 天然标记了"这个需要重跑"，_run_layer 判断 node["node_id"] not in outputs 决定要不要真的执行这个节点，跳过的节点保留原值。
_run_one_node() 里判断 verify_fn 失败后该走哪条路径（原地重试 vs 抛 _LoopBackRequested）：
def _run_one_node(..., node, outputs, ...):
node_id = node["node_id"]
# 每次真正执行都用一个新的 attempt 后缀 node_id，绝不复用同一个已经
# 'done' 的 node_id——_ensure_node_row 按 (run_id, node_id) 去重，复用
# 同一个 node_id 会被 run_work_node 的断点续跑短路成"已完成，直接返回
# 旧输出"，这正是必须避免的（详见 5 节）。
attempt = _count_prior_attempts(conn, run_id, node_id) + 1
exec_node_id = node_id if attempt == 1 else "{}-attempt{}".format(node_id, attempt)
node_exec_id = _ensure_node_row(conn, run_id, exec_node_id, org_node, skill_id, ..., {"node_id": node_id})
prompt = build_prompt(node, outputs) + _loop_history_prompt_suffix(conn, run_id, node_id)
run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout)
output = work_node_run.run_work_node(conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn, ...)
resolved_verify_fn = _resolve_node_verify_fn(node, verify_fn, org_node, target_folder_path, mechanical_check_registry)
if resolved_verify_fn is None:
return output
ok, feedback = resolved_verify_fn(node, output)
if ok:
return output
on_fail = node.get("on_fail") or {"action": "retry_self" if node.get("judgment") else "hard_fail"}
action = on_fail.get("action", "retry_self")
if action == "hard_fail" or (action == "retry_self" and not node.get("judgment")):
# 完全现状路径：没声明 judgment 的老节点，外部 verify_fn 失败时走
# 原有的 _verify_and_maybe_regenerate()（恰好一次原地重试），逐字节不变。
return _verify_and_maybe_regenerate(conn, run_id, org_node, skill_id, target_folder_path,
node, node_id, prompt, output, input_schema, output_schema,
timeout, needs_worktree, resolved_verify_fn, sleep, now)
target = node_id if action == "retry_self" else _on_failure_target(graph_manifest, node_id)
raise _LoopBackRequested(source_node_id=node_id, target_node_id=target, feedback=feedback)
（_run_one_node 需要拿到 graph_manifest 才能查 on_failure 边——现有签名没有这个参数，改造时顺带传进去；retry_self 完全不查边，target_node_id 直接等于自己，第 4.1 节已经说明原因。）
关键取舍：hard_fail、以及"没声明 judgment 的老节点走 retry_self"这两种情况，继续复用现有的 _verify_and_maybe_regenerate() 一字不改——不是所有失败路径都要改造成新机制，只有节点主动声明了 judgment（即 Part B 自规划出的新节点，或 Part A 上线后手工升级的固定链节点）才会走新的 _LoopBackRequested 回路路径。这保证 FR-A8 的向后兼容不是"理论上兼容"而是"代码路径都没变过"。
5. 断点续跑兼容性：新 attempt 后缀就是全部答案
不需要专门为回路设计新的续跑逻辑。原理：_run_one_node() 每次真正执行前，都先查询"这个 node_id 已经有几次真正的执行尝试"（_count_prior_attempts，扫 work_node_execution 里 node_id 匹配 original_id 或 original_id-attemptN 模式的行数），用下一个序号构造 exec_node_id。进程崩溃重启后重新调用 run_graph()：
- 还没跑到的层：
outputs里没有对应 key，_run_layer判断需要执行，正常走一遍——_count_prior_attempts会看到之前留下的部分尝试记录（如果崩溃发生在某次 attempt 执行中途），用下一个序号继续，不会跟已存在的行冲突（_ensure_node_row按exec_node_id去重，exec_node_id本身按 attempt 序号递增，天然不冲突）。 - 已经
done的 attempt：run_work_node()现有的"已完成短路"逻辑对每个exec_node_id各自生效，不需要改。 attempt_counts（内存字典，第 4.3 节）：崩溃后是空的，但run_graph()一旦真的触发_LoopBackRequested时会先调用_loop_attempt_count_from_events(conn, run_id, key)（第 6.3 节，从execution_event里查询真实发生过的次数），不依赖内存状态——重启后第一次遇到某个 loop_back key 时，从 DB 查出来的次数就是崩溃前的真实累计值，不会被内存清零重置为 0。
6. 循环上限配置（FR-A5）
6.1 新表（settings_config.py 风格）
-- schema.py 新增
CREATE TABLE work_graph_loop_limit_config (
config_id TEXT PRIMARY KEY, -- "{org_node}:{node_type}" / "{org_node}:*" / "default"
max_loop_attempts INTEGER NOT NULL,
updated_by TEXT,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
# settings_config.py 新增，跟 get_/upsert_concurrency_config 完全同构
def get_loop_limit_config(conn, config_id):
row = conn.execute(
"SELECT config_id, max_loop_attempts, updated_at FROM work_graph_loop_limit_config WHERE config_id = ?",
(config_id,),
).fetchone()
return None if row is None else {"config_id": row[0], "max_loop_attempts": row[1], "updated_at": row[2]}
def upsert_loop_limit_config(conn, config_id, max_loop_attempts, updated_by):
if max_loop_attempts < 1:
raise InvalidParamError("max_loop_attempts must be >= 1, got {!r}".format(max_loop_attempts))
exists = conn.execute(
"SELECT 1 FROM work_graph_loop_limit_config WHERE config_id = ?", (config_id,)
).fetchone() is not None
if exists:
conn.execute(
"UPDATE work_graph_loop_limit_config SET max_loop_attempts = ?, updated_by = ?, "
"updated_at = CURRENT_TIMESTAMP WHERE config_id = ?",
(max_loop_attempts, updated_by, config_id),
)
else:
conn.execute(
"INSERT INTO work_graph_loop_limit_config (config_id, max_loop_attempts, updated_by) VALUES (?, ?, ?)",
(config_id, max_loop_attempts, updated_by),
)
conn.commit()
return get_loop_limit_config(conn, config_id)
6.2 查找顺序 + 硬编码兜底
# work_graph_orchestrate.py
_DEFAULT_LOOP_LIMIT = 3 # 只在配置表连 "default" 行都没有时才会用到（比如全新数据库还没跑过 seed）
def _resolve_loop_limit(conn, org_node, graph_manifest, target_node_id):
node = next(n for n in graph_manifest["nodes"] if n["node_id"] == target_node_id)
node_level = (node.get("on_fail") or {}).get("max_attempts")
if node_level is not None:
return node_level # 节点自己声明的优先级最高（FR-A5）
node_type = node.get("artifact_type", target_node_id)
for config_id in ("{}:{}".format(org_node, node_type), "{}:*".format(org_node), "default"):
config = settings_config.get_loop_limit_config(conn, config_id)
if config:
return config["max_loop_attempts"]
return _DEFAULT_LOOP_LIMIT
初始 seed：schema.create_schema()/迁移里插入一条 config_id="default", max_loop_attempts=3 的行（跟 org_loop_route._LOOP_LIMIT=3 取值一致，作为一个"暂时先抄现有跨 Org loop 的经验值，后续按真实调用数据调整"的起点——呼应需求文档「7. 风险」第二条，这不是拍脑袋，是有意选择一个已经在生产验证过的数量级作为起点，且明确设计成之后可以不改代码调整）。
7. loop 历史持久化（FR-A4）：execution_event 迁移
7.1 新增 event_type：loop_back_triggered
execution_event.event_type 是 CHECK 约束枚举（schema.py:262-266），SQLite 无法直接 ALTER，沿用 migrate_expand_artifact_types 的 rebuild-copy-swap 手法：
# schema.py 新增
def migrate_add_loop_back_event_type(conn):
"""2026-08-08: 给 execution_event.event_type 的 CHECK 枚举加
'loop_back_triggered'，用于记录 Work Graph 内部回路——见
docs/done/work-graph-self-planning-and-loop-back-design.md §7。
幂等，安全对已有数据库无条件调用（同 migrate_expand_artifact_types 的写法）。
"""
row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone()
if row is None or "'loop_back_triggered'" in row[0]:
return
conn.executescript(
"""
CREATE TABLE execution_event_new (
event_id TEXT PRIMARY KEY,
project_id TEXT NOT NULL REFERENCES project(project_id),
task_id TEXT REFERENCES task(task_id),
run_id TEXT REFERENCES work_graph_run(run_id),
node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
event_type TEXT NOT NULL CHECK (event_type IN (
'task_status_changed', 'node_status_changed', 'artifact_created',
'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
'log_appended', 'quota_wait_started', 'loop_back_triggered'
)),
payload TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO execution_event_new SELECT * FROM execution_event;
DROP TABLE execution_event;
ALTER TABLE execution_event_new RENAME TO execution_event;
CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);
"""
)
conn.commit()
_ensure_schema()（backend_server.py:1020）加一行无条件调用，跟 migrate_expand_artifact_types/migrate_relax_org_node_constraints 并列——这正是 2026-08-06 那次"忘记接进 _ensure_schema导致生产 500"的教训要求的（该函数自己的注释已经把这条约定写死）。
7.2 写入 + 查询
def _record_loop_back_event(conn, run_id, sig, attempt_number):
task_id, project_id = conn.execute(
"SELECT task_id, project_id FROM work_graph_run wgr JOIN task t ON wgr.task_id = t.task_id "
"WHERE wgr.run_id = ?", (run_id,)
).fetchone()
conn.execute(
"INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) "
"VALUES (?, ?, ?, ?, 'loop_back_triggered', ?)",
(str(uuid.uuid4()), project_id, task_id, run_id, json.dumps({
"source_node_id": sig.source_node_id, "target_node_id": sig.target_node_id,
"attempt_number": attempt_number, "feedback": sig.feedback,
})),
)
conn.commit()
def _loop_attempt_count_from_events(conn, run_id, key):
target_node_id, source_node_id = key
rows = conn.execute(
"SELECT payload FROM execution_event WHERE run_id = ? AND event_type = 'loop_back_triggered'", (run_id,)
).fetchall()
return sum(
1 for (payload,) in rows
if json.loads(payload)["target_node_id"] == target_node_id
and json.loads(payload)["source_node_id"] == source_node_id
)
def _loop_history_prompt_suffix(conn, run_id, node_id):
"""target_node_id 重新执行时，把它过去每一轮被打回的完整历史（不只
最近一轮）拼进提示词——FR-A4。"""
rows = conn.execute(
"SELECT payload FROM execution_event WHERE run_id = ? AND event_type = 'loop_back_triggered' "
"ORDER BY created_at", (run_id,)
).fetchall()
history = [json.loads(p) for (p,) in rows if json.loads(p)["target_node_id"] == node_id]
if not history:
return ""
lines = [
"Attempt {}: rejected because: {}".format(h["attempt_number"], "; ".join(h["feedback"] or []))
for h in history
]
return "\n\nThis node was sent back for revision before, with this history:\n" + "\n".join(lines)
不新增表，execution_event 一张表回答"这个节点被打回几次、每次谁触发的、为什么、按什么顺序"——满足决策 #4，且天然带 run_id/node_exec_id/created_at，现有图可视化机制（v_graph_visualizer，若按 run_id 拉取 execution_event 渲染）不需要新的数据源就能把 source_node_id → target_node_id 画成一条回路边。
8. Part B：Planner 自规划节点集合
8.1 节点类型清单——数据，不是硬编码提示词字符串（FR-B2）
# work_graph_plan.py 新增
_PLANNER_NODE_CATALOG = [
{"artifact_type": "brd", "description": "商业需求文档...", "judgment": {"kind": "none"}},
{"artifact_type": "prd", "description": "产品需求文档...",
"judgment": {"kind": "llm", "judge_prompt": _PRD_CONSISTENCY_JUDGE_PROMPT}},
{"artifact_type": "sdd", "description": "详细设计文档...", "judgment": {"kind": "none"}},
{"artifact_type": "benchmark", "description": "验收基准...", "judgment": {"kind": "none"}},
{"artifact_type": "test_dataset", "description": "测试数据集（可选，goal 不需要时回复 N/A）...",
"judgment": {"kind": "none"}},
{"artifact_type": "test_script", "description": "可执行验收脚本（可选，goal 不需要时回复 N/A）...",
"judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"}},
]
_PRD_CONSISTENCY_JUDGE_PROMPT 是现有 _PRD_CONSISTENCY_PROMPT（e2e_pipeline.py:72）改写成 judge_prompt 的通用形状（{content} 占位符替代原来的 {prd_content}），供 Planner 在自规划时"抄"进它给 prd 节点声明的 judgment 里——这份清单只是给 LLM 的参考建议，不是强制：Planner 完全可以给某个节点声明跟清单建议不同的 judgment/on_fail（FR-B4 已确认"全部自由规划"），清单只是降低它每次都要从零发明判断方式的成本。
8.2 新 _PROMPT_TEMPLATE
_PROMPT_TEMPLATE = (
"Plan (do not execute) the Work Graph for this goal: {goal_text}\n"
"Available node types (use any subset, in any order that makes sense -- "
"this goal may not need all of them):\n{catalog_json}\n"
"For each node you include, decide: node_id, artifact_type, depends_on, "
"a judgment config (how to tell if this node's output is acceptable -- you "
"may reuse a catalog entry's suggested judgment, or write your own 'llm' "
"judge_prompt), and an on_fail config (what happens if judgment fails: "
"retry_self, loop_back, or hard_fail). If a node's on_fail.action is "
"'loop_back', you MUST also add a corresponding edge in the top-level "
"'edges' array with edge_type 'on_failure', from that node to the exact "
"earlier node_id it should return to -- the target must be a real "
"upstream dependency of this node (something that genuinely produced "
"input this node used), not an arbitrary node. This mirrors how the "
"success path already works: every node's 'yes, this succeeded, what's "
"next' is expressed as a normal dependency edge, and 'no, this failed, "
"go back to X' is expressed the same way, just with edge_type "
"'on_failure' instead. Every other edge needs edge_type: 'zero_token' "
"for pure aggregation/reduce steps, 'needs_judgment' for anything "
"requiring the agent to actually reason about content.\n"
"End your reply with the graph_manifest as a single fenced ```json code "
"block shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no "
"extra wrapper key."
).format(catalog_json=json.dumps(_PLANNER_NODE_CATALOG), goal_text="{goal_text}")
（.format() 两次是因为 catalog_json 在模块加载时就能确定、goal_text 要留到调用时才填——用 {{/}} 转义加上先格式化一次 catalog_json，再对外暴露带 {goal_text} 占位符的最终模板字符串，写法上等价于现有 _GENERIC_PROMPT_TEMPLATE 的处理方式，不是新模式。）
8.3 _normalize_graph_manifest 的变化：删除，不是加固
_CANONICAL_DOC_CHAIN 常量、以及"当六个/五个规范类型全部出现时强制线性依赖顺序"那段逻辑（work_graph_plan.py，2026-08-07 已经因为禁用 test_script 缩成 5 元组）整体删除——FR-B4 已确认"全部自由规划，不保留任何代码级固定顺序兜底"。_normalize_graph_manifest 保留的部分：
id/node_id键名兼容（2026-08-01 真实教训）。- 从
edges反推depends_on（同上）。
新增的部分：
- 2026-08-08 修订（用户反馈）：失败路径的目标节点不是节点身上一个跟
edges脱节的独立字段，而是跟成功路径一样，由规划这个节点的时候一并定好的边决定——"yes 走哪个节点、no 走哪个节点，都是边定义好的"。所以校验对象变成：每个声明了on_fail.action="loop_back"的节点，必须恰好有一条从它出发、edge_type="on_failure"的边（见第 4.2 节_on_failure_target），且这条边指向的节点必须是该节点的（直接或间接）上游依赖——LLM 规划时可能"跳"到一个跟自己毫无依赖关系的节点（逻辑上没有意义：loop_back 的语义是"回到更早、真正影响了我的节点"），这种情况在_normalize_graph_manifest里直接拒绝并抛PlanningFailedError（复用现有异常类），比让它在run_graph()执行到一半才报错更早暴露问题。
def _validate_loop_back_targets(graph_manifest):
node_ids = {n["node_id"] for n in graph_manifest["nodes"]}
for node in graph_manifest["nodes"]:
on_fail = node.get("on_fail") or {}
if on_fail.get("action") != "loop_back":
continue
node_id = node["node_id"]
on_failure_edges = [
e for e in graph_manifest.get("edges", [])
if e.get("from") == node_id and e.get("edge_type") == "on_failure"
]
if len(on_failure_edges) != 1:
raise PlanningFailedError(
"node {} declares on_fail.action='loop_back' but has {} on_failure edge(s), expected exactly 1".format(
node_id, len(on_failure_edges)
)
)
target = on_failure_edges[0]["to"]
if target not in node_ids:
raise PlanningFailedError(
"node {}'s on_failure edge targets {!r}, which does not exist in this graph_manifest".format(
node_id, target
)
)
if target not in _transitive_ancestors(graph_manifest, node_id):
raise PlanningFailedError(
"node {}'s on_failure edge targets {!r}, which is not an upstream dependency -- "
"loop_back must target a node that genuinely influenced this one".format(node_id, target)
)
（_transitive_ancestors 是 _transitive_dependents——第 4.2 节，work_graph_orchestrate.py——的反向版本；两边各自独立实现一份即可，work_graph_plan.py 不需要因此新引入对 work_graph_orchestrate.py 的依赖，两个模块现在也没有互相 import，保持这个边界。_on_failure_target——第 4.2 节——运行时可以放心假设"恰好一条"已经在这里校验过，不用再处理 0 条/多条的情况。）
8.4 plan_documents() 调用点改动
e2e_pipeline.py::plan_documents() 不再需要 _combined_planner_verify_fn（PRD 一致性检查 + test_script 编译检查两个手写分支）——改成 verify_fn=None，因为自规划出的每个节点都自带 judgment，run_graph() 内部会自动解析。mechanical_check_registry={"check_script_compiles": check_script_compiles} 照常传入。旧的 _combined_planner_verify_fn/_prd_consistency_verify_fn/_test_script_verify_fn 函数不删除——validate_implementation_with_own_graph()（Validator 自己的硬编码 3/2 节点图，本轮不改，见需求文档非目标）继续用它们，且它们仍然是"外部 verify_fn 向后兼容路径"的真实调用方，删除会破坏 FR-A8。
9. 向后兼容性核对表（FR-A8）
| 现有调用点 | 是否声明 judgment/on_fail |
改造后走的路径 |
|---|---|---|
implement_code() 的合成单节点图 |
否 | _resolve_node_verify_fn 直接返回 None（该调用点本来就不传 verify_fn），零改动 |
validate_implementation_with_own_graph()（_VALIDATOR_MANIFEST） |
否 | _resolve_node_verify_fn 返回外部 _test_script_verify_fn(...)，失败时走 hard_fail/现状 _verify_and_maybe_regenerate，逐字节不变 |
work_graph_fork.py（_GENERIC_PROMPT_TEMPLATE 路径） |
否 | 同上，无 verify_fn 传入，无判断 |
plan_documents()（Part B 改造后） |
是（每个节点自带） | 走新的 judgment/on_fail 解析路径 |
10. 测试计划（对应需求文档验收标准 1-10）
| 验收标准 | 测试方式 |
|---|---|
| 1-2. loop_back 累积历史 + 触发上限 | mock execute_work_graph，构造 A→B、B 声明 on_fail=loop_back(target=A)，B 连续失败 N 次；断言 A 被重跑 N 次、execution_event 里有 N 条 loop_back_triggered、A 第 N 次收到的 prompt 包含前 N-1 次的 feedback 文本 |
| 3. 崩溃重启后正确恢复轮次 | 跑到第 2 次 loop_back 时中断（mock 抛异常模拟进程退出），重新调用 run_graph()，断言 _loop_attempt_count_from_events 从 DB 读到 2，不是从 0 开始 |
| 4. 静态图校验不因 loop_back 报环 | 一个声明了 on_fail.action="loop_back" 且带一条指向上游节点的 on_failure 边的 graph_manifest，topological_layers() 正常返回分层，不抛 ValueError（因为 on_failure 边类型根本不参与 topological_layers 的依赖计算，只有 depends_on/普通数据依赖 edge 参与） |
| 5. 未声明字段的图逐字节不变 | 现有 test_work_graph_orchestrate.py/test_e2e_pipeline.py 全部既有用例不改断言即通过 |
| 6-7. Part B 自由规划 + 健壮性兜底 | 同需求文档，mock LLM 规划响应缺 test_dataset / 缺 depends_on，断言分别正确处理 |
| 8. test_script 可选 | mock 规划响应包含 test_script 节点，断言正常规划执行 |
| 9. 按 org_node/node_type 配置上限生效 | settings_config.upsert_loop_limit_config 写两条不同配置，构造两个 loop_back 场景，断言各自在配置的上限触发 |
10. loop 历史只在 execution_event |
触发若干轮 loop_back 后，只查 execution_event（不查任何新表）复原完整历史 |
新增：_validate_loop_back_targets 拒绝非法目标 |
分别构造"on_failure 边指向的节点不是上游依赖"、"声明 loop_back 但没有 on_failure 边"、"同一个节点有 2 条 on_failure 边"三种场景，断言 plan_work_graph() 都抛 PlanningFailedError |
| 新增：mechanical_check_registry 未注册报错 | judgment.check_ref 指向一个不存在的名字，断言 run_graph() 在执行任何节点前就报错，不是跑到那个节点才发现 |
新增：topological_layers() 正确忽略 on_failure 边（4.0 节修复） |
构造 A→B（普通依赖）+ B→A（on_failure 边）的 graph_manifest，断言 topological_layers() 返回 [[A], [B]]（不因为 B→A 那条边被误判成环）；另跑一遍 G 既有的 test_work_graph_orchestrate.py 全部用例确认零回归 |
11. 实施步骤建议
schema.py两处迁移（work_graph_loop_limit_config建表 +migrate_add_loop_back_event_type）+settings_config.py两个新函数 +_ensure_schema()接入——纯数据层改动，先落地，不碰任何执行逻辑。work_graph_orchestrate.py：_resolve_node_verify_fn/机械检查注册表注入/_llm_judgment_verify_fn——先只做"judgment 解析"，不做 loop_back（这一步验证：没有on_fail时行为完全不变，第 9 节核对表跑一遍）。work_graph_orchestrate.py：_transitive_dependents/_LoopBackRequested/run_graph()主循环回退逻辑/attempt 后缀机制——loop_back 的核心，单独用 mock 场景验证第 10 节测试计划 1-5、9、10。work_graph_plan.py：删除_CANONICAL_DOC_CHAIN强制顺序逻辑、_PLANNER_NODE_CATALOG、新_PROMPT_TEMPLATE、_validate_loop_back_targets——Part B，第 10 节测试计划 6-8、新增两项。e2e_pipeline.py::plan_documents()切到verify_fn=None+mechanical_check_registry，端到端跑一次 mock 场景确认整条链路。- 真实场景验证（
RUN_REAL_AGENT_E2E=1，小心控制次数/成本）：观察一次真实 Planner 自规划调用产出的graph_manifest长什么样，是否合理、是否触发过PlanningFailedError。