work-graph-self-planning-and-loop-back-design.md 47.5 KB

Planner 动态自规划 Work Graph(含节点判断与失败回路)——详细设计(SDD)

日期: 2026-08-08 性质: 详细设计,对应需求文档 docs/todo/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 扩展

{
  "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/todo/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. 实施步骤建议

  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。