# 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()`：新增分层拓扑排序

```python
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()` 的薄封装，保证行为/顺序完全不变：

```python
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`）完全不用改一行：

```python
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 内部状态，不是数据库）：

```python
_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()` 改造

```python
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. 实施步骤建议（拆成可独立验证的小步）

1. `topological_layers()` 新增 + `topological_order()` 改写成基于它的薄封装，跑既有测试确认零回归（不改 `run_graph()`，先证明分层算法本身正确）。
2. 引入 `_LockedConn`，先只在新代码路径里用，不碰现有串行路径。
3. `create_worktree()` 加锁，写第 10 节测试用例 5 验证是否真的需要（哪怕结论是"其实不需要"，锁本身开销极小，可以保留作为防御性设计）。
4. `run_graph()` 改造成层级调度（第 5 节的完整实现），单节点层走同步路径、多节点层走线程池路径。
5. 补全第 10 节全部测试用例。
6. 找一个真实存在多节点并行结构的 `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 节）。