# 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 分支**： 1. **`judgment`**：节点自己说"怎么判断我的输出对不对"（机械检查 or 触发一次额外 LLM 判断）。 2. **`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 扩展 ```jsonc { "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`： ```jsonc // 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`（会成环）。机械检查改成调用方注入的白名单字典： ```python 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` 的隔离机制，泛化提示词 ```python 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 解析 ```python 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`： ```python 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` 两种既有类型继续按现状合并（它们表达的就是真实数据依赖）： ```python 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 相悖，成本可控且实现不复杂： ```python 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()` 主循环改造 ```python 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`）： ```python 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` 风格） ```sql -- 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 ); ``` ```python # 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 查找顺序 + 硬编码兜底 ```python # 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 手法： ```python # 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 写入 + 查询 ```python 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） ```python # 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` ```python _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()` 执行到一半才报错更早暴露问题。 ```python 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. 实施步骤建议 1. `schema.py` 两处迁移（`work_graph_loop_limit_config` 建表 + `migrate_add_loop_back_event_type`）+ `settings_config.py` 两个新函数 + `_ensure_schema()` 接入——纯数据层改动，先落地，不碰任何执行逻辑。 2. `work_graph_orchestrate.py`：`_resolve_node_verify_fn`/机械检查注册表注入/`_llm_judgment_verify_fn`——先只做"judgment 解析"，不做 loop_back（这一步验证：没有 `on_fail` 时行为完全不变，第 9 节核对表跑一遍）。 3. `work_graph_orchestrate.py`：`_transitive_dependents`/`_LoopBackRequested`/`run_graph()` 主循环回退逻辑/attempt 后缀机制——loop_back 的核心，单独用 mock 场景验证第 10 节测试计划 1-5、9、10。 4. `work_graph_plan.py`：删除 `_CANONICAL_DOC_CHAIN` 强制顺序逻辑、`_PLANNER_NODE_CATALOG`、新 `_PROMPT_TEMPLATE`、`_validate_loop_back_targets`——Part B，第 10 节测试计划 6-8、新增两项。 5. `e2e_pipeline.py::plan_documents()` 切到 `verify_fn=None` + `mechanical_check_registry`，端到端跑一次 mock 场景确认整条链路。 6. 真实场景验证（`RUN_REAL_AGENT_E2E=1`，小心控制次数/成本）：观察一次真实 Planner 自规划调用产出的 `graph_manifest` 长什么样，是否合理、是否触发过 `PlanningFailedError`。