Work Graph 节点并行执行——详细设计（SDD）
日期: 2026-08-05
性质: 详细设计，对应需求文档 docs/work-graph-parallel-execution-requirements.md（已确认，核心优先级）。本设计逐条回答该需求文档第 8 节的开放问题，并给出可直接实施的接口/算法/测试方案。
依赖代码：src/work_graph_orchestrate.py（run_graph/topological_order/_ensure_node_row/_make_run_fn/_verify_and_maybe_regenerate）、src/work_node_run.py（run_work_node/create_worktree）、src/backend_server.py（conn_lock 先例）。
状态: 设计草案，待评审后实施。
1. 设计概览
现状：run_graph()（work_graph_orchestrate.py:196）用 topological_order() 把 graph_manifest 拍平成一条扁平列表，纯 Python for 循环逐节点执行。
改造后：新增 topological_layers()，把 Kahn 算法的中间产物（每一轮 in-degree 归零的节点集合）保留为"层"而不是拍平；run_graph() 改成"层间顺序、层内并发"——同一层内的节点用 ThreadPoolExecutor 并发触发各自的 run_work_node()，层与层之间保持屏障（下一层要等上一层全部完成）。
不改变的部分（刻意保持稳定，降低改动面）：
- 每个节点仍然是"渲染成一个单节点 workflow 脚本 + 独立
claude -p子进程调用"（_make_run_fn/_render_node_script不变）。 run_work_node()的状态机、断点续跑逻辑、重试/配额等待逻辑完全不变。run_graph()的对外签名和返回值格式（{node_id: output_payload}）不变（需求文档 FR-9）。verify_fn的语义不变（需求文档 FR-8）。
2. topological_layers()：新增分层拓扑排序
def topological_layers(graph_manifest):
"""Kahn 算法的分层版本：返回 List[List[node]]，每个内层 list 是同一轮
in-degree 归零的节点集合（互相之间保证没有依赖，可以并发执行）。
层内按 node_id 排序（保持确定性，便于测试断言）。
依赖来源、报错行为跟现有 topological_order() 完全一致（合并
depends_on 和 edges 两个来源；有环或引用未知 node_id 时 raise ValueError）。
"""
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", []):
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)
in_degree = {node_id: len(d) for node_id, d in deps.items()}
dependents = {node_id: [] for node_id in nodes}
for node_id, d in deps.items():
for dep in d:
if dep in dependents:
dependents[dep].append(node_id)
layers = []
current = sorted(node_id for node_id, degree in in_degree.items() if degree == 0)
seen_count = 0
while current:
layers.append([nodes[node_id] for node_id in current])
seen_count += len(current)
next_layer = set()
for node_id in current:
for dependent in dependents[node_id]:
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
next_layer.add(dependent)
current = sorted(next_layer)
if seen_count != len(nodes):
raise ValueError(
"graph_manifest has a dependency cycle or a depends_on/edge references an unknown node_id"
)
return layers
topological_order() 保留（其他调用方，如 e2e_pipeline.py:233 直接用它拿扁平顺序写 artifact），但改成对 topological_layers() 的薄封装，保证行为/顺序完全不变：
def topological_order(graph_manifest):
return [node for layer in topological_layers(graph_manifest) for node in layer]
（需要跑一次现有 test_work_graph_orchestrate.py 里 topological_order 的既有单测，确认拍平后的顺序逐字节相同——这两版算法的排序规则一致，理论上不会有差异，但改动了实现路径，必须实测验证不是"理论上一致"就够。）
3. SQLite 并发写入：_LockedConn 代理
问题：run_work_node()/_ensure_node_row()/execution_log.append_log() 等函数内部有大量分散的 conn.execute(...) + conn.commit() 调用点。层内并发后，这些调用可能从多个线程同时打到同一个 sqlite3.Connection 对象上。
关键判断：SQLite 本身同一时刻只允许一个写者（这是 SQLite 的架构限制，不是 Python 线程模型能绕开的），所以不需要设计"细粒度多把锁"，一把锁保护整个连接的读写足够——真正的并发收益来自于"锁只在单条 SQL 语句执行期间持有，claude -p 子进程调用（决定墙钟时间的部分）完全不持锁"，而不是来自"多个线程能同时写 SQLite"（这本来就不可能）。
方案：不去逐个包裹现有几十处 conn.execute() 调用点（改动面大、容易漏改），而是引入一个跟 sqlite3.Connection 接口兼容的轻量代理对象，在并发路径上把 conn 参数替换成这个代理——调用方代码（work_node_run.py/work_graph_orchestrate.py/execution_log.py）完全不用改一行：
class _LockedConn:
"""sqlite3.Connection 的线程安全代理：execute/commit/executemany 各自
持锁执行，可以直接替代 conn 参数传给现有代码，调用方无需感知。复用
backend_server.py::_Server 已经验证过的"共享连接 + threading.Lock"模式，
只是把锁的粒度做到了每条语句级别（backend_server 现有的 conn_lock 只
保护 server_close() 一处，不是通用查询锁——本类是把同一个思路做成可
在任意并发场景复用的通用工具，两者不冲突，backend_server 未来可以直
接换成这个类）。
"""
def __init__(self, conn, lock=None):
self._conn = conn
self._lock = lock or threading.Lock()
def execute(self, *args, **kwargs):
with self._lock:
return self._conn.execute(*args, **kwargs)
def executemany(self, *args, **kwargs):
with self._lock:
return self._conn.executemany(*args, **kwargs)
def commit(self):
with self._lock:
return self._conn.commit()
def __getattr__(self, name):
# cursor()/rollback()/close() 等未显式包装的方法透传给底层连接，
# 调用方目前不会从多线程并发调用这些（没有并发路径用到它们）。
return getattr(self._conn, name)
run_graph() 在进入并发执行前，用一把 threading.Lock() 包出一个 _LockedConn 实例，传给每个并发任务内部实际调用的 _ensure_node_row/run_work_node/_make_run_fn（后者内部会调用 execution_log.append_log）——注意：只有并发的那一层用 _LockedConn，层外（比如 run_graph() 自己开头调用 ensure_placeholder_skill）继续用原始 conn，因为那部分本来就不在并发路径上，没必要多一层锁的开销。
4. Worktree 创建竞态
work_node_run.py::create_worktree()（git worktree add -b <branch> <path>）在并发场景下可能对同一个 repo 的 .git/worktrees/ 元数据产生竞态——这是推测，需求文档已标注"未经验证"。设计上采用保守做法：给 create_worktree() 包一把进程级模块锁（跟 SQLite 锁分开，因为这个锁保护的是文件系统/git 内部状态，不是数据库）：
_worktree_creation_lock = threading.Lock()
def create_worktree(repo_path, node_exec_id):
with _worktree_creation_lock:
... # 原逻辑不变
这个锁只包住 git worktree add 这一条命令本身（毫秒级），不包住节点其余的执行时间，所以即使锁確实是必要的，对整体并行收益的影响可以忽略——claude -p 调用（秒级到分钟级）才是墙钟时间的主体。验证方式：写一个真实的并发 create_worktree 调用测试（多线程同时对同一个临时 git 仓库调用），先不加锁跑一遍看是否真的出错，出错则保留锁、不出错也保留锁（保守优先，反正代价极小）。
5. 并发调度器：run_graph() 改造
import concurrent.futures
_DEFAULT_MAX_WORKERS = 4 # 保守默认值，见第 9 节
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,
):
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)
outputs = {}
db_lock = threading.Lock()
locked_conn = _LockedConn(conn, db_lock)
for layer in layers:
if len(layer) == 1:
# 单节点层：不走线程池，直接同步执行——省掉线程池调度开销，
# 也让现有单节点场景（目前观察到的真实 graph_manifest 全都是
# 单节点链）行为路径尽量贴近改造前，降低回归风险。
node = layer[0]
outputs[node["node_id"]] = _run_one_node(
conn, run_id, org_node, skill_id, target_folder_path, node, outputs,
build_prompt, input_schema, output_schema, timeout,
effective_needs_worktree, verify_fn, sleep, now,
)
continue
worker_count = min(max_workers, len(layer))
errors = []
with concurrent.futures.ThreadPoolExecutor(max_workers=worker_count) as pool:
futures = {
pool.submit(
_run_one_node, locked_conn, run_id, org_node, skill_id, target_folder_path,
node, outputs, build_prompt, input_schema, output_schema, timeout,
effective_needs_worktree, verify_fn, sleep, now,
): node
for node in layer
}
# 等整层全部完成（成功或失败）再继续——层内节点互不依赖，
# 一个节点的失败不该取消已经在跑的兄弟节点（需求文档 FR-5）。
done, _ = concurrent.futures.wait(futures, return_when=concurrent.futures.ALL_COMPLETED)
for future in done:
node = futures[future]
try:
outputs[node["node_id"]] = future.result()
except Exception as e:
errors.append((node["node_id"], e))
if errors:
# 聚合报错：全部记录，抛出的异常携带完整清单，不只是"第一个"，
# 方便排查同一层里到底几个节点失败、分别是什么原因。
raise WorkGraphLayerError(errors)
return outputs
def _run_one_node(
conn, run_id, org_node, skill_id, target_folder_path, node, outputs,
build_prompt, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
):
"""原 run_graph() 循环体单节点逻辑原样保留，只是抽成一个独立函数,
好让同步路径（单节点层）和并发路径（多节点层）共用同一份实现,
不重复维护两份"跑一个节点+可选 verify_fn"的逻辑。"""
node_id = node["node_id"]
node_exec_id = _ensure_node_row(
conn, run_id, node_id, org_node, skill_id,
"work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
{"node_id": node_id},
)
prompt = build_prompt(node, outputs)
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,
needs_worktree=needs_worktree, repo_path=target_folder_path, sleep=sleep, now=now,
)
if verify_fn is not None:
output = _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, verify_fn, sleep, now,
)
return output
class WorkGraphLayerError(Exception):
"""一层内一个或多个节点执行失败——携带 [(node_id,原始异常), ...] 完整
清单，调用方可以按需检查是不是全是同一种可恢复错误。"""
def __init__(self, errors):
self.errors = errors
summary = "; ".join("{}: {}".format(node_id, e) for node_id, e in errors)
super().__init__("{} node(s) failed in one layer: {}".format(len(errors), summary))
关键设计取舍——build_prompt(node, outputs) 里的 outputs 读取时机：层内并发时，outputs 字典在整层的所有 _run_one_node 调用里都是同一个共享字典，但层内节点互相没有依赖，build_prompt 不会读到"同层兄弟节点"的输出（拓扑保证），只会读到"更早的层"已经写完的输出——这些更早层的写入在当前层开始前已经全部完成（层屏障保证），所以层内并发读 outputs 是安全的，不需要给 outputs 字典本身加锁（Python 字典的读操作本身也是线程安全的，只要没有并发写同一个 key，而层内不同节点写的是不同的 node_id key，不冲突）。每个节点各自的 outputs[node_id] = ... 写入统一放到层执行完、concurrent.futures.wait 返回之后、在主线程里做（上面伪代码里 for future in done: outputs[node["node_id"]] = future.result()），不在子线程里直接写共享字典，进一步避免任何并发写字典的疑虑，即使实际上没有真正的 key 冲突。
6. 错误/配额处理的真实限制（诚实记录，不回避）
work_node_run.run_work_node() 内部对配额限制（UsageLimitReachedError→_UsageLimitSignal）的处理是在函数内部原地 sleep-retry，退避时间可以长达 30 分钟一次、session 级超时 8 小时、weekly 级超时 9 天（work_node_run.py:42-43）。这意味着：如果同一层里节点 A 触发了配额等待、节点 B 几秒内就正常完成，concurrent.futures.wait(..., return_when=ALL_COMPLETED) 会让整层卡在等 A——B 的结果虽然已经算出来了，但因为层屏障语义（下一层要等这一层全部完成），实际上整个 run_graph() 调用也会被拖住。
这是本设计接受的限制，不是本轮要解决的问题：Python 线程一旦启动就没有干净的"取消并回收"机制（Future.cancel() 只能取消还没开始跑的任务），要让 B 提前继续、A 在后台继续等配额而不阻塞整层，需要给 run_work_node() 引入可中断/可分离的配额等待机制——这是比本次改造大得多的另一个课题，不在本次范围内（需求文档"非目标"第 1 条已经声明"不做完整的数据流式调度器"，这个限制是同一类取舍的自然延伸）。如果未来这个限制成为真实瓶颈，需要单独立项评估。
7. 断点续跑兼容性——不需要新逻辑
run_work_node() 已有的"已 done 节点短路返回持久化输出"逻辑对并发路径天然兼容，不需要改动：进程重启后重新调用 run_graph()，topological_layers() 重新计算分层，对每一层内的节点，_run_one_node() 里的 run_work_node() 调用会对"已完成"的节点立即返回持久化结果（几乎不占用线程池工作时间），对"未完成"的节点正常重新执行——同一层里有的节点秒回、有的节点真正重跑，线程池天然处理这种混合情况，不需要额外判断"这层要不要重跑"。
8. 可观测性
_log_event（work_graph_orchestrate.py:74）现有的 node_start/node_completed/node_failed/quota_limit 四类事件保持不变，每条事件本身已经带 node_exec_id，事件的时间戳（execution_log.append_log 底层写入时机）已经足够回答"这次运行到底并发了多少节点"——只要看同一个 run_id 下多个 node_start 事件的时间戳有没有重叠，不需要新增专门的"分组开始/分组结束"事件类型（呼应需求文档开放问题第 4 条：复用现有事件、靠时间戳重叠反推，不新增事件类型，保持改动面最小）。
9. 并发度默认值
_DEFAULT_MAX_WORKERS = 4。理由：现在观察到的真实 graph_manifest 规模是个位数节点，4 是一个不会让"一次任务同时拉起过多 claude 子进程"的保守值,同时足够覆盖当前能想到的 fan-out 场景（比如 Planner 未来 PRD 阶段 3-5 个场景并行）。作为 run_graph() 的一个具名参数（而不是硬编码常量或环境变量),调用方可以按场景覆盖,但不新增管理后台配置项（需求文档非目标已声明不做这类配置暴露，YAGNI——等真的有需要按项目/按 Org Agent 类型区分并发度时再加）。
10. 测试计划（对应需求文档验收标准 1-7）
| 验收标准 | 测试方式 |
|---|---|
| 1. 同层节点确实并发触发 | mock run_fn，每次调用记录 (node_id, 开始时间, 结束时间) 到共享列表（用锁保护这个记录列表本身，测试代码自己的锁，不是被测代码的），断言同层两个节点的时间区间有重叠 |
| 2. 单测断言并发+层间顺序 | 同上，另外断言"下一层任意节点的开始时间 >= 上一层全部节点的结束时间"（层屏障） |
| 3. 断点续跑：已完成节点不重跑 | 第一次调用 mock 让某层其中一个节点抛异常模拟中断（WorkGraphLayerError 传播出来），检查 DB 里其余节点确实落库为 done；第二次调用同一个 run_graph()，断言 run_fn 没有被再次调用去跑那些已 done 的节点（用调用计数 mock 断言） |
| 4. SQLite 并发写入无异常 | 用真实 :memory:（或临时文件）SQLite 库跑一个多节点并发层，不 mock conn，跑多轮（并发问题往往概率复现）确认不抛 database is locked |
| 5. worktree 并发创建互不冲突 | 真实临时 git 仓库，多线程调用 create_worktree，断言 git worktree list 输出的路径数量和唯一性符合预期 |
| 6. 配额限制 + 正常完成共存 | mock 一个节点的 run_fn 触发 _UsageLimitSignal 且配置极短的超时（依赖注入的 sleep/now，不真的等待），另一个节点正常返回，断言正常节点的输出被保留在最终 outputs 里，且整体调用最终正确抛出/传播配额相关信号 |
| 7. 既有测试兼容性 | 跑 src/test_work_graph_orchestrate.py、src/test_work_graph_execute.py、src/test_e2e_pipeline.py 全部既有用例，逐个确认通过；对断言"严格串行"这一实现细节的用例（如果存在）需要单独识别并更新断言方式（改成断言"层间顺序"而不是"节点间顺序"） |
11. 实施步骤建议（拆成可独立验证的小步）
topological_layers()新增 +topological_order()改写成基于它的薄封装，跑既有测试确认零回归（不改run_graph()，先证明分层算法本身正确）。- 引入
_LockedConn，先只在新代码路径里用，不碰现有串行路径。 create_worktree()加锁，写第 10 节测试用例 5 验证是否真的需要（哪怕结论是"其实不需要"，锁本身开销极小，可以保留作为防御性设计）。run_graph()改造成层级调度（第 5 节的完整实现），单节点层走同步路径、多节点层走线程池路径。- 补全第 10 节全部测试用例。
- 找一个真实存在多节点并行结构的
graph_manifest场景做一次端到端验证（如果当前还没有这样的真实场景，构造一个人工测试场景即可，不强求等 E 落地后才能验证 G 本身）。
12. 开放问题的最终决定
回答需求文档 docs/work-graph-parallel-execution-requirements.md §8 列的开放问题：
topological_layers()实现方式：新增函数，topological_order()改写成基于它的薄封装（第 2 节已给出实现）。max_workers默认值：4，作为run_graph()的具名参数，不做管理后台配置项（第 9 节）。- FR-5 异常聚合规则：层内全部完成后统一收集，抛出携带完整
[(node_id, exception), ...]清单的WorkGraphLayerError（第 5 节），调用方可以按需检查列表内容做进一步判断（比如是否全是可重试的配额问题）。 execution_event是否需要新事件类型：不需要，复用现有事件+时间戳重叠反推（第 8 节）。- worktree 加锁粒度：只锁
create_worktree()内部git worktree add这一条命令,不锁整个节点执行（第 4 节）。