U-031: wire c_work_graph_plan/execute/run_work_node together + U-024.5 WS reconnect refetch

U-031: closes a structural integration gap found during the REQ-01~14 regression audit -- plan_work_graph/execute_work_graph/run_work_node were each individually tested but never actually called together in production code (e2e_pipeline.py's only validated closed loop bypassed all three via flat org_node_trigger calls). Adds work_graph_orchestrate.py (topological sort over graph_manifest + one one-node dynamic-workflow agent() call per Work Node, needed for REQ-06's per-node breakpoint recovery), rewires e2e_pipeline.py's plan_documents/ implement_code onto the real chain, and fixes two bugs a real API run surfaced: graph_manifest's id/node_id/depends_on normalization (work_graph_plan._normalize_graph_manifest) and a missing repo_path in run_work_node's cwd resolution. work_node_run.py also now marks a node 'blocked' instead of leaving it stuck 'running' on any unretryable run_fn exception. U-024.5: the regression audit also found REQ-12's reconnect boundary case (refetch a snapshot after a dropped WS reconnects) was never implemented -- only the dashboard reacted to onStatusChange, and even it never refetched. Fixes mountDashboard/mountTaskBoard/mountTaskDetail (which embeds v_graph_visualizer) to refetch their current snapshot on reconnect. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-031: wire c_work_graph_plan/execute/run_work_node together + U-024.5 WS reconnect refetch
U-031: closes a structural integration gap found during the REQ-01~14 regression audit -- plan_work_graph/execute_work_graph/run_work_node were each individually tested but never actually called together in production code (e2e_pipeline.py's only validated closed loop bypassed all three via flat org_node_trigger calls). Adds work_graph_orchestrate.py (topological sort over graph_manifest + one one-node dynamic-workflow agent() call per Work Node, needed for REQ-06's per-node breakpoint recovery), rewires e2e_pipeline.py's plan_documents/ implement_code onto the real chain, and fixes two bugs a real API run surfaced: graph_manifest's id/node_id/depends_on normalization (work_graph_plan._normalize_graph_manifest) and a missing repo_path in run_work_node's cwd resolution. work_node_run.py also now marks a node 'blocked' instead of leaving it stuck 'running' on any unretryable run_fn exception. U-024.5: the regression audit also found REQ-12's reconnect boundary case (refetch a snapshot after a dropped WS reconnects) was never implemented -- only the dashboard reacted to onStatusChange, and even it never refetched. Fixes mountDashboard/mountTaskBoard/mountTaskDetail (which embeds v_graph_visualizer) to refetch their current snapshot on reconnect. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent e696cd48
Showing 13 changed files with 770 additions and 92 deletions
docs/development-plan.md
docs/sdd/05-实施约束与依赖.md
docs/sdd/unit_matrix.json
frontend/app.js
src/e2e_pipeline.py
src/test_e2e_pipeline.py
src/test_e2e_pipeline_e2e.py
src/test_work_graph_orchestrate.py
src/test_work_graph_plan.py
src/test_work_node_run.py
src/work_graph_orchestrate.py
src/work_graph_plan.py
src/work_node_run.py
@@ -237,6 +237,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-024.5（2026-07-30 新增）[Value Path]** REQ-12 完整验收（事件多样性联调）
   - 依赖：U-024, U-015, U-016, U-018, U-019, U-020, U-021（需要 PH2 大部分 Unit 已上线，事件类型覆盖足够丰富才能真正验证 boundary/failure 场景——例如 usage-limit 等待、护栏拦截、Skill 回滚各自产生的 `execution_event` 都要真实出现过）
   - Done：REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准，U-024 本身只解决"链路通不通"
+  - **实现补充（2026-08-01）——U-031 回归审计中发现的真实缺口**：核对本 Unit 是否真的 Done 时，重新对照 06 节 REQ-12 的 boundary 用例原文——"重连后能补齐断线期间遗漏的关键状态（通过重新拉取 task/work_graph_run 当前快照兜底）"——grep 了 `frontend/app.js` 里所有 `onStatusChange` 用法，发现只有 `v_dashboard` 做了一点反应（重连时把断线提示 banner 隐藏），而且连它自己也没有真的重新拉取数据；`v_task_board`/`v_task_detail`（含内嵌的 `v_graph_visualizer`）的 `onStatusChange` 全是空函数 `() => {}`。也就是说这条 boundary 用例此前从未被真正实现过，U-024/U-024.5 之前的"REQ-12 完整通过"是不准确的。修复：`mountDashboard`/`mountTaskBoard`/`mountTaskDetail` 三处的 `connectEventSocket` 都加了 `wasDisconnected` 标志，在状态从 `disconnected` 转回 `connected` 时调用各自页面已有的刷新函数（`refreshTasksAndOrgGraph`+`refreshBudget` / `refresh` / `refresh`）重新拉取当前快照。`v_artifact_browser` 的 `onStatusChange` 保持空——REQ-12 命名的验收范围只到 `v_dashboard`/`v_graph_visualizer`/`v_task_detail`，artifact 浏览器不在其中，不属于这次修复范围。浏览器扩展本次会话不可用，验证方式：`node --check` 过语法 + 全量 `unittest` 回归（361 tests）；WS 重连本身的行为逻辑此前已被 `src/test_backend_server.py`/Node 原生 WebSocket 端到端验证过（U-024/U-027），这次只是前端消费侧新增的刷新调用，未改动服务端重连协议本身。
   - 覆盖：REQ-12（完整）
  
 - **U-025 [Value Path]** `v_dashboard` + `v_task_board` + `v_task_detail`
@@ -288,6 +289,15 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
  
 - **U-031 [Risk Gate]** REQ-01~14 全量自动化回归套件
   - 依赖：PH1/PH2/PH3 全部 Unit
+  - Done：06 节全部 14 条 REQ 的全部用例通过一次完整回归（原文档缺 Done/覆盖行，2026-08-01 补上）
+  - **实现补充（2026-08-01）——发现并修复一个结构性集成缺口，这是本 Unit 目前为止最重要的产出**：逐条核对 REQ-01~14 验收用例对应的测试覆盖时，发现 REQ-02/REQ-03 验收锚点明确列出的控制器覆盖（`c_work_graph_plan`/`c_work_graph_execute`/`c_work_node_run`）——也就是"Org Graph 触发 → Work Graph 规划 → Work Graph 执行 → 每个 Work Node 独立运行"这套两层架构——虽然在 U-011/U-012/U-013/U-020/U-021 各自都建好、也各自做过完整单元测试（U-013/U-021 甚至做过真实 git worktree 验证），但 `plan_work_graph()`/`execute_work_graph()`/`run_work_node()` 这三个函数彼此之间、以及和唯一真正跑通验证过的闭环（U-014 的 `e2e_pipeline.py`，snake-game 那次真实生成）之间，从未被真正接起来过——`e2e_pipeline.py` 走的是完全独立、更简单的路径（对 Planner/Coder/Validator 各打一次 `org_node_trigger.trigger_org_node()` 直连调用），完全绕开了这套两层架构。**这个决策已跟用户确认（AskUserQuestion），选择立即补上集成而不是把现状订正进文档了事**，因为 REQ-06（断点恢复）的验收要求"重启后不重跑已完成节点"，只有 `work_node_run.py` 的按节点 Python 层持久化状态能做到，这和"整个 Work Graph 包在一次不透明的 `claude -p` 调用里"（`execute_work_graph()` 原本设想的用法）根本互斥，所以这个缺口不是"锦上添花"，是 REQ-06 能不能算真正满足的前提。
+    - **设计**：新增 `src/work_graph_orchestrate.py`，每个 Work Node 的真实执行走**它自己单独的一次动态工作流调用**（`write_workflow_script` + `execute_work_graph`），不是把整张图塞进一次调用——这是刻意的取舍，牺牲了 03-C §3.5 原本"多节点编排摊薄到一次调用里，编排本身不消耗 token"的效率立意，换来 REQ-06 要求的按节点级持久化/断点恢复。`topological_order()` 对 `graph_manifest` 做拓扑排序，同时读 `depends_on` 和 `edges` 并合并（原因见下一条发现）。`build_prompt(node, outputs)` 在拓扑序上逐节点调用，`outputs` 里已经有该节点全部上游依赖的真实产出内容，天然支持"prd 的 prompt 需要嵌入 brd 的真实内容"这种链式依赖，也天然支持断点恢复（`run_work_node` 对已 `done` 节点直接返回持久化结果，恢复执行时上游内容照样能取到）。
+    - **`work_node_run.py` 补了一个真实 bug**：原来只捕获 `_UsageLimitSignal`，任何其他异常（比如新集成会遇到的 `ClassifierBlockedError`）会直接从 `run_work_node()` 冒泡出去，`work_node_execution.status` 永远停在 `running`，没有任何"为什么卡住了"的记录。改成任何非 usage-limit 异常都先把节点标记 `blocked` 再重新抛出。
+    - **真实调用验证时连续发现两个集成 bug，都已修复**：① 第一次真实跑 `plan_work_graph()` 时，模型返回的 `graph_manifest` 每个节点用的是 `id` 而不是文档要求的 `node_id`，而且完全没填任何节点的 `depends_on`（依赖关系只体现在 `edges` 里）——`plan_work_graph()` 本身从来没做过 schema 校验（`--json-schema` 配 `--agent` 不生效，05 §5.4.1.1 早就记录过），这是真实模型输出会偏离文档契约的活例子。新增 `work_graph_plan._normalize_graph_manifest()`，在 parse 完、持久化前统一把 `id`→`node_id`、从 `edges` 反推缺失的 `depends_on`——这样不只是我这个新集成受益，`v_task_detail` 的 to-do 树、`v_graph_visualizer` 的 DAG 分层以后遇到真实（而不是测试里手写整齐的）manifest 时也不会因为字段名对不上而显示错误。② 修好第一个 bug 后，第二次真实跑又炸了：`work_graph_orchestrate.run_graph()` 调 `run_work_node()` 时忘了传 `repo_path`，`cwd` 解析成 `None`，写工作流脚本直接崩——两次真实调用暴露的都是"单元测试全绿但没人真正把这几个函数接起来跑过"的典型症状，修完重跑，真实 4 篇文档（BRD→PRD→SDD→Benchmark）全部生成成功、血缘链正确、内容长度合理。
+    - **`agentType` + `schema` 能否在动态工作流的 `agent()` 调用里正确共存，是这次集成最大的未知数，做了两次独立真实验证**：先用一个最小 spike（一个只有一次 `agent({agentType:'planner', schema:{...}})` 调用的一节点工作流）验证——不仅 `structured_output` 正确返回，追问角色身份时模型明确自称"Planning Agent (PL)"，说明迁移后的 SOUL.md 人设真的通过 `agentType` 正确加载了（不是走到某个通用 agent）；后续完整 `plan_documents()` 真实跑通又在真实的四节点场景下再次确认。记录在 05 §5.4.1.5。
+    - **Coder 阶段的范围决策**：REQ-03 的验收锚点只写了 `c_work_node_run`，没写 `c_work_graph_plan`/`c_work_graph_execute`——因为"规划一个真正的多模块代码图"是另一个完全不同的 LLM 能力，这次没有人要求建，也没有 SDD 规定怎么建，属于超出这次集成范围的新能力，不是"补上就完事"的小活。`implement_code()` 因此只用一个合成的单节点 manifest 走 `work_graph_orchestrate.run_graph()`，让 `c_work_node_run` 对 Coder 也真正生效，但没有尝试造一个"Coder 图规划器"。`needs_worktree` 参数留在 `run_graph()` 里但暂时两处调用都没启用——worktree 隔离要求 `.claude/agents/<org_node>.md` 已经**提交**到分支上（`git worktree add` 只看已提交内容，`deploy_agent_definitions()` 目前只写工作区文件不 commit），这个"部署后要不要顺手提交"的问题这次没有一并解决，留作后续发现，不是遗漏是有意不在这次范围内展开。真实验证也只覆盖了 Planner 全流程（Coder/Validator 走同一套机制，机制本身已经被 Planner 的真实跑通证明，没有再花真钱跑一次完整 Coder 会话，跟 `test_e2e_pipeline_e2e.py` 一贯的成本节制原则一致）。
+    - **Validator 阶段未改动**：REQ-04 的覆盖从来没提过 Work Graph 控制器（只有 `c_bug_record_create`/`p_validator_root_cause_trace`），`validate_implementation()` 原来的直连调用已经如实对应它自己的 REQ，不需要动。
+  - 覆盖：01-M `work_graph_run`/`work_node_execution`（真正被生产代码写入，不再只是测试 fixture）；03-C `c_work_graph_plan`/`c_work_graph_execute`/`c_work_node_run`（首次真正接通）；REQ-01~14 全部
  
 - **U-032 [Risk Gate]** 多 Goal 并发（`org_concurrency_config`）+ 长时间运行稳定性压测
   - 依赖：U-031
@@ -61,6 +61,13 @@
 - **应对方式**：参照 `create_server()` 已有的 `db_path` 参数模式，加了 `skill_repo_path` 参数：默认值是 `skill_query._REPO_ROOT`（graphlemon 真实仓库根，生产环境下这就是唯一正确的行为——Skill 变更本来就应该落到这里），但可以在构造 server 时覆盖；测试统一传一个 `tempfile` 建的临时仓库。**这个参数只在 Python 层（进程启动/测试构造）可设，绝不通过 HTTP 请求体/query string 暴露给客户端**——否则等于开了一个"客户端指定任意目录、服务端替它跑 git commit"的口子，属于路径注入/任意写入范畴。
 - 这不是"发现"的约束（不是实测跑出来的意外行为），是设计阶段就该考虑到、写代码时主动做的防护，记录下来是为了以后改这块代码的人不要图省事把 `skill_repo_path` 从请求里读出来。
  
+## 5.4.1.5 动态工作流内部 `agent()` 的 `agentType`+`schema` 组合已实测可用（发现，2026-08-01，U-031）
+
+- 背景：U-031 补 Work Graph 三模块集成（`work_graph_orchestrate.py`）时，每个 Work Node 的真实执行都要用一份一次性的动态工作流脚本，脚本内唯一的一次 `agent()` 调用需要同时做到两件事——① 加载目标 org_node（Planner/Coder/Validator）迁移后的人设（`agentType: '<org_node>'`，对应 `.claude/agents/<org_node>.md`），② 拿到结构化输出（`schema: {...}`）。这和 §5.4.1.1 记录的"顶层 `--agent` + `--json-schema` 不能同时生效"表面上是同一类组合，但机制完全不同：§5.4.1.1 是 CLI 顶层参数 `claude --agent ... --json-schema ...`；这里是动态工作流脚本**内部** JS DSL 的 `agent(prompt, {agentType, schema})` 调用，二者互不影响，不能假设前者的限制也适用于后者——必须单独验证，不能凭直觉套用。
+- **实测确认（两次独立验证）**：① 先用一个最小 spike（仅一个一节点工作流，唯一动作是 `agent({agentType:'planner', schema:{type:'object',...}})`）验证——`structured_output` 正确按 schema 返回，且额外追问角色身份时模型明确自称"Planning Agent (PL)"，证明迁移后的 SOUL.md 人设确实通过 `agentType` 正确加载（不是退化成某个通用 agent 后蒙对了格式）；② 后续 `e2e_pipeline.plan_documents()` 完整四节点（BRD→PRD→SDD→Benchmark）真实跑通，再次在真实多节点场景下确认这个组合稳定可用。
+- **实际用法**：`work_graph_orchestrate._render_node_script()` 渲染的脚本固定形如 `const result = await agent(prompt, { agentType: <org_node>, schema: <output_schema> })`，`execute_work_graph()` 直接把这个一次性脚本文件写到 `.claude/workflows/` 后调用 `claude -p 'Run /<name>...'` 执行，不复用同一个脚本名跑第二个节点（每个 `node_exec_id` 生成自己独立的脚本文件名，避免并发/重试之间互相覆盖）。
+- 同 §5.4.1.1/§5.4.1.2：这条结论是实测发现的，官方文档没有专门说明动态工作流 `agent()` 调用里 `agentType`+`schema` 的组合行为，不排除后续 Claude Code 版本变化后需要重新验证。
+
 ## 5.4.2 Claude Code plan 额度等待重试参数（架构文档 7.5 节决策，2026-07-29）
  
 - 退避轮询间隔（5→10→20→30 分钟，上限 30 分钟）、超时兜底阈值（session 8 小时、weekly 9 天）是 MVP 阶段的硬编码默认值，不在管理后台开放用户配置，和 `log_retention_config` 的"保留天数可配、清理频率不可配"是同一个 MVP 简化原则
@@ -39,7 +39,7 @@
     {"id": "U-022", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_project_setup + c_pick_folder 全平台（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-007", "U-008"], "coverage": {"model": [], "view": ["v_project_setup"], "controller": ["c_pick_folder", "c_project_create"], "process": [], "requirement": ["REQ-11"]}, "done": "REQ-11 里 v_project_setup/c_pick_folder/c_project_create 部分通过；c_pick_folder 补齐 Windows/Linux；c_project_create 首次接上 HTTP 路由；REQ-11 完整 happy（含建 task）随 U-023 累计满足"},
     {"id": "U-023", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_goal_composer + v_bug_feedback（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-022"], "coverage": {"model": [], "view": ["v_goal_composer", "v_bug_feedback"], "controller": ["c_task_create", "c_bug_report_create"], "process": [], "requirement": []}, "done": "能通过 UI 创建 task 和人工 bug_record；补上 c_bug_report_create 的 project_id 必填缺口（03-C input_contract 遗漏）；related_task_id 简化为文本输入，等 U-025 的 c_task_list 落地后换真下拉"},
     {"id": "U-024", "phase": "PH3", "sub_phase": "PH3-S2", "name": "c_ws_subscribe + execution_event 实时推送（最小连通版，见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-012", "U-013"], "coverage": {"model": ["execution_event"], "view": [], "controller": ["c_ws_subscribe"], "process": [], "requirement": ["REQ-12"]}, "done": "推送链路本身连通，≤2 秒延迟达成；REQ-12 happy 用例通过（boundary/failure 留给 U-024.5）；手写 RFC 6455 握手+帧编解码（无第三方 websockets 依赖），轮询式推送（无单一写路径可插桩），Node 原生 WebSocket 客户端验证过真实互操作性"},
-    {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（2026-07-30 新增）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准"},
+    {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（见 development-plan.md 2026-08-01 实现补充）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准。boundary 用例（重连补齐断线期间遗漏状态）此前从未真正实现，仅 v_dashboard 有 UI banner 切换、无数据重拉——2026-08-01 于 U-031 回归审计中发现并补上：v_dashboard/v_task_board/v_task_detail（含内嵌的 v_graph_visualizer）的 connectEventSocket onStatusChange 均已在 disconnected→connected 转换时重新拉取当前快照；v_artifact_browser 不在 REQ-12 命名范围内，未改动"},
     {"id": "U-025", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_dashboard + v_task_board + v_task_detail（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_dashboard", "v_task_board", "v_task_detail"], "controller": ["c_task_list", "c_task_get", "c_org_loop_route"], "process": [], "requirement": []}, "done": "三个页面可正常展示实时数据；补上此前无 Unit 认领的 c_task_list/c_task_get HTTP 路由，c_org_loop_route 首次接上 HTTP；v_graph_visualizer/v_log_viewer 子模块占位等 U-026/U-027"},
     {"id": "U-026", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_graph_visualizer（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_graph_visualizer"], "controller": [], "process": [], "requirement": []}, "done": "Work Graph DAG 分层视图含 waiting_for_quota 信息态；Org Graph 层复用 v_task_detail 已有的 org-strip 面板，不重复渲染"},
     {"id": "U-027", "phase": "PH3", "sub_phase": "PH3-S3", "name": "execution_log_file + 日志查看器（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": ["execution_log_file", "log_retention_config"], "view": ["v_log_viewer"], "controller": ["c_execution_log_get", "c_ws_log_tail"], "process": ["p_log_retention_cleanup"], "requirement": ["REQ-13"]}, "done": "REQ-13 happy/boundary/failure 通过；补上此前从未存在的 append_log 写路径；顺带修了 U-024 就存在的 server_close() 并发 segfault 隐患（conn_lock 未覆盖）"},
@@ -47,7 +47,7 @@
     {"id": "U-029", "phase": "PH3", "sub_phase": "PH3-S3", "name": "v_skill_library + v_skill_audit_log（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-018"], "coverage": {"model": [], "view": ["v_skill_library", "v_skill_audit_log"], "controller": ["c_skill_diff", "c_skill_list", "c_skill_get", "c_skill_rollback"], "process": [], "requirement": []}, "done": "Skill 库与变更审计日志可查看；c_skill_rollback 首次接上 HTTP（skill_repo_path 可注入，测试绝不碰真实仓库）"},
     {"id": "U-030", "phase": "PH3", "sub_phase": "PH3-S4", "name": "并发/预算/日志保留配置页（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-019", "U-021"], "coverage": {"model": ["org_concurrency_config"], "view": ["v_concurrency_settings", "v_budget_settings", "v_log_retention_settings"], "controller": ["c_concurrency_config_update", "c_budget_config_update", "c_log_retention_config_update"], "process": [], "requirement": []}, "done": "三个配置页可用且改动生效；PH3 至此全部完成"},
  
-    {"id": "U-031", "phase": "PH4", "sub_phase": "PH4-S1", "name": "REQ-01~14 全量自动化回归", "classification": "Risk Gate", "depends_on": ["U-014", "U-017", "U-020", "U-021", "U-022", "U-023", "U-024.5", "U-025", "U-026", "U-027", "U-028", "U-029", "U-030"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-07", "REQ-08", "REQ-09", "REQ-10", "REQ-11", "REQ-12", "REQ-13", "REQ-14"]}, "done": "06 节全部 14 条 REQ 的全部用例通过一次完整回归"},
+    {"id": "U-031", "phase": "PH4", "sub_phase": "PH4-S1", "name": "REQ-01~14 全量自动化回归（见 development-plan.md 2026-08-01 实现补充）", "classification": "Risk Gate", "depends_on": ["U-014", "U-017", "U-020", "U-021", "U-022", "U-023", "U-024.5", "U-025", "U-026", "U-027", "U-028", "U-029", "U-030"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-07", "REQ-08", "REQ-09", "REQ-10", "REQ-11", "REQ-12", "REQ-13", "REQ-14"]}, "done": "06 节全部 14 条 REQ 的全部用例通过一次完整回归；回归审计中发现 c_work_graph_plan/c_work_graph_execute/c_work_node_run 三个模块彼此隔离测试通过但从未被生产代码真正串联（唯一验证过的闭环 e2e_pipeline.py 绕开了三者，直接用 org_node_trigger），已按用户决策现场补上集成：新增 src/work_graph_orchestrate.py（拓扑排序 + 每节点一份独立动态工作流脚本调用 run_work_node），e2e_pipeline.py 的 plan_documents/implement_code 改为真调用 plan_work_graph→run_graph；真实 API 验证过程中发现并修复两个真实 bug（graph_manifest 的 id/node_id 字段名及 depends_on 缺失见 work_graph_plan._normalize_graph_manifest；run_work_node 缺 repo_path 导致 cwd=None）"},
     {"id": "U-032", "phase": "PH4", "sub_phase": "PH4-S2", "name": "多 Goal 并发 + 长时间运行稳定性压测", "classification": "Risk Gate", "depends_on": ["U-031"], "coverage": {"model": ["org_concurrency_config"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "并发数调高后多个独立 Goal 可同时稳定运行"},
     {"id": "U-033", "phase": "PH4", "sub_phase": "PH4-S3", "name": "A-2 假设收口", "classification": "Risk Gate", "depends_on": ["U-004"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": [], "assumption": ["A-2"]}, "done": "A-2 状态确认为 confirmed 或记录实际偏差"},
  
@@ -548,6 +548,7 @@ function mountDashboard() {
   refreshBudget();
   eventTimeline.innerHTML = '<div class="hint">暂无事件</div>';
  
+  let wasDisconnected = false;
   const socket = connectEventSocket(projectId, null, {
     onEvent: (event) => {
       prependEvent(event);
@@ -556,6 +557,16 @@ function mountDashboard() {
     },
     onStatusChange: (status) => {
       wsBanner.style.display = status === "disconnected" ? "block" : "none";
+      if (status === "disconnected") {
+        wasDisconnected = true;
+      } else if (status === "connected" && wasDisconnected) {
+        // REQ-12 boundary: reconnect doesn't replay every missed event, but
+        // must at least catch up on current state via a fresh snapshot
+        // fetch (03-C §3.14's own documented reconnect story).
+        wasDisconnected = false;
+        refreshTasksAndOrgGraph();
+        refreshBudget();
+      }
     },
   });
   registerRouteCleanup(() => socket.close());
@@ -637,11 +648,20 @@ function mountTaskBoard() {
   orgNodeEl.addEventListener("change", refresh);
   refresh();
  
+  let wasDisconnected = false;
   const socket = connectEventSocket(projectId, null, {
     onEvent: (event) => {
       if (event.event_type === "task_status_changed") refresh();
     },
-    onStatusChange: () => {},
+    onStatusChange: (status) => {
+      if (status === "disconnected") {
+        wasDisconnected = true;
+      } else if (status === "connected" && wasDisconnected) {
+        // REQ-12 boundary: catch up via a fresh snapshot on reconnect.
+        wasDisconnected = false;
+        refresh();
+      }
+    },
   });
   registerRouteCleanup(() => socket.close());
 }
@@ -969,9 +989,18 @@ function mountTaskDetail(taskId) {
  
   refresh();
  
+  let wasDisconnected = false;
   const socket = connectEventSocket(projectId, taskId, {
     onEvent: () => refresh(),
-    onStatusChange: () => {},
+    onStatusChange: (status) => {
+      if (status === "disconnected") {
+        wasDisconnected = true;
+      } else if (status === "connected" && wasDisconnected) {
+        // REQ-12 boundary: catch up via a fresh snapshot on reconnect.
+        wasDisconnected = false;
+        refresh();
+      }
+    },
   });
   registerRouteCleanup(() => {
     socket.close();
@@ -5,51 +5,82 @@ benchmark/tests and produces a root-cause-attributed verdict). Criteria 4-7
 (Org-level Loop, breakpoint recovery, ...) are explicitly out of this Unit's
 scope -- PH2 (U-017/U-021).
  
+**Rewired 2026-08-01 (U-031 regression audit)**: `plan_documents` used to
+call `org_node_trigger.trigger_org_node()` directly, bypassing
+`work_graph_plan.plan_work_graph()`/`work_graph_execute.execute_work_graph()`
+/`work_node_run.run_work_node()` entirely -- those three modules were each
+fully unit-tested (and even real-API-validated) in isolation since PH1/PH2,
+but nothing in production ever called any of them; this was the one real,
+validated closed loop, and it used a completely separate, simpler mechanism.
+REQ-02's own coverage line names `c_work_graph_plan`/`c_work_graph_execute`
+explicitly, so that gap is now closed: `plan_documents` calls
+`plan_work_graph()` for a real graph_manifest, then
+`work_graph_orchestrate.run_graph()` executes each node for real (see that
+module's docstring for why each node gets its own one-node dynamic-workflow
+script rather than one big multi-node script -- REQ-06 breakpoint recovery
+needs it). `implement_code` (Coder) goes through the same per-node
+mechanism with a synthetic single-node manifest, since REQ-03's own
+coverage only names `c_work_node_run`, not graph planning -- there is no
+real "plan a Coder module graph" LLM capability built (a much bigger,
+separate undertaking; planning docs and planning code modules are
+different problems), so a one-node graph is the honest minimum that makes
+`c_work_node_run` genuinely load-bearing for Coder too.
+`validate_implementation` (Validator) is unchanged -- REQ-04's own
+coverage never named the Work Graph controllers at all.
+
 Status-transition bookkeeping (task.status, precondition guards) stays
 U-010's job (org_node_lifecycle.trigger_org_node_full) -- called once per
-phase with a lightweight "confirm ready" prompt. Content generation is
-separate, plain org_node_trigger.trigger_org_node calls (no status
-side-effects) so each document/step's real text is available to write as an
-artifact -- trigger_org_node_full's own contract (a status dict, no text)
-isn't changed to accommodate this.
+phase with a lightweight "confirm ready" prompt.
 """
  
+import json
 import os
 import subprocess
+import uuid
  
-from src import artifact_write, org_node_lifecycle, org_node_trigger, task_create
+from src import (
+    artifact_write, org_node_lifecycle, org_node_trigger, task_create,
+    work_graph_orchestrate, work_graph_plan,
+)
  
 _READY_PROMPT = "Confirm you're ready to begin. Take no other action yet."
  
+_DOC_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
+_DOC_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]}
+
 _DOC_PROMPTS = {
     "brd": (
         "Write a BRD (Business Requirement Document) for this goal, following your own "
         "SOUL.md methodology: {goal_text}\n"
-        "Output ONLY the BRD content in Markdown -- no preamble, no meta-commentary about "
-        "what you're about to do."
+        "Reply with the document content (Markdown) via the required schema -- no preamble, "
+        "no meta-commentary about what you're about to do."
     ),
     "prd": (
         "Here is the approved BRD for goal '{goal_text}':\n\n{brd_content}\n\n"
         "Write the PRD (Product Requirement Document) based on this BRD. "
-        "Output ONLY the PRD content in Markdown."
+        "Reply with the document content (Markdown) via the required schema."
     ),
     "sdd": (
         "Here is the approved PRD for goal '{goal_text}':\n\n{prd_content}\n\n"
         "Write the SDD (System Design Document) based on this PRD, including a Development "
-        "Unit breakdown. Output ONLY the SDD content in Markdown."
+        "Unit breakdown. Reply with the document content (Markdown) via the required schema."
     ),
     "benchmark": (
         "Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n"
         "Write the Benchmark/acceptance-test document based on this SDD: concrete test "
         "scenarios (happy/boundary/failure) with observable pass criteria. "
-        "Output ONLY the Benchmark content in Markdown."
+        "Reply with the document content (Markdown) via the required schema."
     ),
 }
  
+_CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
+_CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]}
+
 _CODER_PROMPT = (
     "Implement this goal per the approved SDD below. Write real code and real tests, run the "
     "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts "
-    "and can explain why not). Goal: {goal_text}\n\nSDD:\n\n{sdd_content}"
+    "and can explain why not). Reply via the required schema with a `response` field describing "
+    "what you did and the real test commands/output. Goal: {goal_text}\n\nSDD:\n\n{sdd_content}"
 )
  
 _VALIDATOR_PROMPT = (
@@ -61,42 +92,86 @@ _VALIDATOR_PROMPT = (
 )
  
  
-def plan_documents(conn, task_id, run_id, goal_text, target_folder_path, timeout=300):
-    """Architecture §11 criterion 1: real Planner calls producing BRD->PRD->SDD
-    ->Benchmark, each written via c_artifact_write with correct lineage.
-    Returns {"brd": artifact_id, "prd": ..., "sdd": ..., "benchmark": ...}.
+def _planner_prompt_builder(goal_text, node_artifact_types):
+    """node_artifact_types: dict[node_id] -> artifact_type, read off the
+    real graph_manifest before running it (build_prompt only ever sees one
+    node/outputs-so-far pair at a time, not the whole manifest)."""
+    def build_prompt(node, outputs):
+        node_id = node["node_id"]
+        artifact_type = node_artifact_types.get(node_id, node_id)
+        content_kwargs = {}
+        for dep_id in (node.get("depends_on") or []):
+            dep_output = outputs.get(dep_id)
+            if dep_output is not None:
+                dep_type = node_artifact_types.get(dep_id, dep_id)
+                content_kwargs[dep_type + "_content"] = dep_output["content"]
+        return _DOC_PROMPTS[artifact_type].format(goal_text=goal_text, **content_kwargs)
+    return build_prompt
+
+
+def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
+    """Architecture §11 criterion 1: a real Planner Work Graph
+    (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
+    work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark,
+    each written via c_artifact_write with correct lineage.
+    Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}.
     """
+    plan_result = work_graph_plan.plan_work_graph(
+        conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout
+    )
+    run_id = plan_result["run_id"]
+    graph_manifest = plan_result["graph_manifest"]
+    node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}
+
+    outputs = work_graph_orchestrate.run_graph(
+        conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
+        _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
+    )
+
     artifact_ids = {}
-    contents = {}
-    prev_type = None
-    for artifact_type in ("brd", "prd", "sdd", "benchmark"):
-        prompt = _DOC_PROMPTS[artifact_type].format(
-            goal_text=goal_text, **{k + "_content": v for k, v in contents.items()}
-        )
-        content = org_node_trigger.trigger_org_node("planner", prompt, target_folder_path, timeout=timeout)
-        deps = [artifact_ids[prev_type]] if prev_type else []
+    for node in work_graph_orchestrate.topological_order(graph_manifest):
+        node_id = node["node_id"]
+        artifact_type = node_artifact_types[node_id]
+        deps = [
+            artifact_ids[node_artifact_types[dep_id]]
+            for dep_id in (node.get("depends_on") or [])
+            if node_artifact_types.get(dep_id) in artifact_ids
+        ]
         result = artifact_write.write_artifact(
             conn, target_folder_path, artifact_type, task_id, run_id,
-            ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), content,
+            ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), outputs[node_id]["content"],
             depends_on_artifact_ids=deps,
         )
         artifact_ids[artifact_type] = result["artifact_id"]
-        contents[artifact_type] = content
-        prev_type = artifact_type
-    return artifact_ids
+    return {"run_id": run_id, "artifact_ids": artifact_ids}
  
  
-def implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200):
-    """Architecture §11 criterion 2: real Coder call, then registers whatever
-    the Coder actually committed as code_diff (the diff itself) and
-    test_report (the Coder's own response text, which per its migrated SOUL.md
-    is required to include real test commands + real output) artifacts.
+def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200):
+    """Architecture §11 criterion 2: a real Coder call via c_work_node_run
+    (work_graph_orchestrate.run_graph with a synthetic single-node
+    manifest -- see module docstring for why there's no real Coder graph
+    planner), then registers whatever the Coder actually committed as
+    code_diff (the diff itself) and test_report (the Coder's own response
+    text, which per its migrated SOUL.md is required to include real test
+    commands + real output) artifacts.
     """
     before_sha = _current_commit(target_folder_path)
-    response = org_node_trigger.trigger_org_node(
-        "coder", _CODER_PROMPT.format(goal_text=goal_text, sdd_content=sdd_content),
-        target_folder_path, timeout=timeout,
+    run_id = str(uuid.uuid4())
+    manifest = {"nodes": [{"node_id": "implement", "depends_on": []}], "edges": []}
+    conn.execute(
+        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+        "VALUES (?, ?, 'coder', ?, 'running')",
+        (run_id, task_id, json.dumps(manifest)),
+    )
+    conn.commit()
+
+    outputs = work_graph_orchestrate.run_graph(
+        conn, run_id, "coder", "coder-implementation", target_folder_path, manifest,
+        lambda node, outputs: _CODER_PROMPT.format(goal_text=goal_text, sdd_content=sdd_content),
+        _CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout,
     )
+    response = outputs["implement"]["response"]
+
     diff = subprocess.run(
         ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True
     ).stdout
@@ -115,7 +190,11 @@ def implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_
 def validate_implementation(task_id, benchmark_content, goal_text, target_folder_path, timeout=1200):
     """Architecture §11 criterion 3: real Validator call, parses the
     'VERDICT: PASS|FAIL|INCONCLUSIVE' line this Unit's prompt asks for.
-    Returns {"verdict", "raw_response"}.
+    Returns {"verdict", "raw_response"}. Unchanged by the 2026-08-01
+    rewiring -- REQ-04's own coverage never named the Work Graph
+    controllers (c_bug_record_create/p_validator_root_cause_trace only),
+    so the simpler flat trigger_org_node call is already faithful to what
+    this REQ actually requires.
     """
     response = org_node_trigger.trigger_org_node(
         "validator",
@@ -145,30 +224,19 @@ def run_end_to_end(conn, target_folder_path, goal_text):
     validate_implementation. Returns a dict with every intermediate result,
     for inspection/debugging rather than hiding it behind a single verdict.
     """
-    project_id = conn.execute("SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)).fetchone()[0]
+    project_id = conn.execute(
+        "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
+    ).fetchone()[0]
     task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
  
     org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT)
-    # trigger_org_node_full itself doesn't create a work_graph_run row (that's
-    # U-011's plan_work_graph, a separate call this pipeline doesn't invoke,
-    # since REQ-02's graph_manifest and this pipeline's actual document
-    # content are orthogonal concerns) -- artifact.produced_by_run_id is a
-    # real FK, so a row has to exist; this one just records that the e2e
-    # pipeline (not a real planned Work Graph) produced these artifacts
-    run_id = "e2e-" + task_id
-    conn.execute(
-        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-        "VALUES (?, ?, 'planner', '{}', 'running')",
-        (run_id, task_id),
-    )
-    conn.commit()
-
-    artifact_ids = plan_documents(conn, task_id, run_id, goal_text, target_folder_path)
+    plan_result = plan_documents(conn, task_id, goal_text, target_folder_path)
+    artifact_ids = plan_result["artifact_ids"]
     sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
     benchmark_content = _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"])
  
     org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, _READY_PROMPT)
-    code_artifact_ids = implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path)
+    code_artifact_ids = implement_code(conn, task_id, sdd_content, goal_text, target_folder_path)
  
     org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT)
     verdict = validate_implementation(task_id, benchmark_content, goal_text, target_folder_path)
@@ -6,6 +6,19 @@ correctly, not what a real Planner/Coder/Validator would actually produce.
 Real coverage (as much of it as is safe/affordable to run) is in
 src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1).
  
+**Rewired 2026-08-01 (U-031 regression audit)**: plan_documents/
+implement_code now go through work_graph_plan.plan_work_graph +
+work_graph_orchestrate.run_graph (which calls work_graph_execute.
+execute_work_graph per node) instead of a flat org_node_trigger.
+trigger_org_node call -- see e2e_pipeline.py's module docstring for why.
+plan_work_graph is mocked at its own subprocess boundary (same pattern as
+test_work_graph_plan.py) so its real graph_manifest-parsing logic and its
+real work_graph_run INSERT both still run for real; execute_work_graph is
+mocked directly (one level higher than its own subprocess boundary) since
+this file's job is testing e2e_pipeline's wiring, not re-proving
+execute_work_graph's own mechanics (covered in test_work_graph_execute.py)
+or work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py).
+
 Run: python3 -m unittest src.test_e2e_pipeline -v
 """
  
@@ -15,27 +28,33 @@ import tempfile
 import unittest
 from unittest.mock import patch
  
-from src import e2e_pipeline, project_scaffold, schema
+from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute
  
 _FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"}
  
+_FAKE_GRAPH_MANIFEST = {
+    "nodes": [
+        {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
+        {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
+        {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]},
+        {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]},
+    ],
+    "edges": [
+        {"from": "brd", "to": "prd", "edge_type": "needs_judgment"},
+        {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"},
+        {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"},
+    ],
+}
+
+
+def _fake_plan_subprocess_run(*args, **kwargs):
+    payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_GRAPH_MANIFEST) + "\n```"}
+    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
  
-def _fake_trigger(org_node, prompt, target_folder_path, timeout=300):
-    if prompt == e2e_pipeline._READY_PROMPT:
-        return "Ready."
-    if "BRD (Business" in prompt:
-        return _FAKE_DOCS["brd"]
-    if "PRD (Product" in prompt:
-        return _FAKE_DOCS["prd"]
-    if "SDD (System" in prompt:
-        return _FAKE_DOCS["sdd"]
-    if "Benchmark/acceptance-test" in prompt:
-        return _FAKE_DOCS["benchmark"]
-    if prompt.startswith("Implement this goal"):
-        return "Implemented. Ran `pytest`: 3 passed."
-    if prompt.startswith("Independently verify"):
-        return "VERDICT: PASS\nAll benchmark scenarios reproduced and passing."
-    raise AssertionError("unexpected prompt: " + prompt[:80])
+
+def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
+    assert prompt == e2e_pipeline._READY_PROMPT
+    return "Ready."
  
  
 class PipelineTestBase(unittest.TestCase):
@@ -58,17 +77,18 @@ class PlanDocumentsTest(PipelineTestBase):
             "VALUES ('task-1', (SELECT project_id FROM project), '实现贪吃蛇游戏', 'goal', "
             "'planning_brd_prd_sdd', 'planner')"
         )
-        self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
-        )
         self.conn.commit()
  
-        with patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
-            artifact_ids = e2e_pipeline.plan_documents(
-                self.conn, "task-1", "run-1", "实现贪吃蛇游戏", self.tmp.name
-            )
+        with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [
+                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
+                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
+            ]
+            plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name)
  
+        artifact_ids = plan_result["artifact_ids"]
         self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark"})
         prd_deps = json.loads(
             self.conn.execute(
@@ -83,11 +103,25 @@ class PlanDocumentsTest(PipelineTestBase):
         )
         self.assertEqual(sdd_deps, [artifact_ids["prd"]])
  
+        # plan_work_graph's own real work_graph_run INSERT actually ran (not bypassed)
+        run_row = self.conn.execute(
+            "SELECT org_node, status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
+        ).fetchone()
+        self.assertEqual(run_row, ("planner", "planning"))
+
  
 class RunEndToEndTest(PipelineTestBase):
     def test_full_pipeline_wiring(self):
-        with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_trigger), \
-             patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger):
+        with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
+             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."):
+            mock_execute.side_effect = [
+                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
+                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
+                {"response": "Implemented. Ran `pytest`: 3 passed."},
+            ]
             result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏")
  
         self.assertEqual(result["verdict"]["verdict"], "PASS")
@@ -104,6 +138,14 @@ class RunEndToEndTest(PipelineTestBase):
         ).fetchone()[0]
         self.assertEqual(all_artifacts, 6)  # brd, prd, sdd, benchmark, code_diff, test_report
  
+        # planner and coder each got their own work_graph_run (not sharing one)
+        run_org_nodes = {
+            row[0] for row in self.conn.execute(
+                "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
+            ).fetchall()
+        }
+        self.assertEqual(run_org_nodes, {"planner", "coder"})
+
  
 class ValidateImplementationTest(unittest.TestCase):
     def test_parses_pass_verdict(self):
@@ -29,14 +29,12 @@ class RealPlanDocumentsTest(unittest.TestCase):
             agent_profile.deploy_agent_definitions(tmp)
             project_id = conn.execute("SELECT project_id FROM project").fetchone()[0]
             task_id = task_create.create_task(conn, project_id, "实现贪吃蛇游戏")["task_id"]
-            conn.execute(
-                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-                "VALUES ('run-1', ?, 'planner', '{}', 'running')",
-                (task_id,),
-            )
-            conn.commit()
  
-            artifact_ids = e2e_pipeline.plan_documents(conn, task_id, "run-1", "实现贪吃蛇游戏", tmp)
+            # plan_documents (2026-08-01 rewiring) creates its own real
+            # work_graph_run via plan_work_graph() -- no manual seed row
+            # needed anymore.
+            plan_result = e2e_pipeline.plan_documents(conn, task_id, "实现贪吃蛇游戏", tmp)
+            artifact_ids = plan_result["artifact_ids"]
  
             for artifact_type, artifact_id in artifact_ids.items():
                 row = conn.execute(
+"""U-031 tests: work_graph_orchestrate.py -- the graph_manifest -> per-node
+execution bridge closed during the REQ-01~14 regression audit. Real sqlite,
+run_fn mocked at the claude subprocess boundary (execute_work_graph is
+tested for real elsewhere, in test_work_graph_execute.py / the 2026-08-01
+spike -- this module's own job is orchestration, not re-proving that).
+
+Run: python3 -m unittest src.test_work_graph_orchestrate -v
+"""
+
+import unittest
+from unittest.mock import patch
+
+from src import schema, work_graph_execute, work_graph_orchestrate as wgo
+
+
+class TopologicalOrderTest(unittest.TestCase):
+    def test_linear_chain_via_depends_on(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "brd", "depends_on": []},
+                {"node_id": "prd", "depends_on": ["brd"]},
+                {"node_id": "sdd", "depends_on": ["prd"]},
+            ],
+            "edges": [],
+        }
+        self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["brd", "prd", "sdd"])
+
+    def test_linear_chain_via_edges_only(self):
+        # real LLM output might populate edges but not each node's own
+        # depends_on -- must still order correctly
+        manifest = {
+            "nodes": [{"node_id": "a"}, {"node_id": "b"}, {"node_id": "c"}],
+            "edges": [{"from": "a", "to": "b"}, {"from": "b", "to": "c"}],
+        }
+        self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b", "c"])
+
+    def test_diamond_fan_out_fan_in(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "n1", "depends_on": []},
+                {"node_id": "n2", "depends_on": ["n1"]},
+                {"node_id": "n3", "depends_on": ["n1"]},
+                {"node_id": "n4", "depends_on": ["n2", "n3"]},
+            ],
+            "edges": [],
+        }
+        order = [n["node_id"] for n in wgo.topological_order(manifest)]
+        self.assertEqual(order[0], "n1")
+        self.assertEqual(order[-1], "n4")
+        self.assertEqual(set(order[1:3]), {"n2", "n3"})
+
+    def test_cycle_raises(self):
+        manifest = {
+            "nodes": [{"node_id": "x", "depends_on": ["y"]}, {"node_id": "y", "depends_on": ["x"]}],
+            "edges": [],
+        }
+        with self.assertRaises(ValueError):
+            wgo.topological_order(manifest)
+
+    def test_independent_nodes_ordered_deterministically(self):
+        manifest = {"nodes": [{"node_id": "b"}, {"node_id": "a"}], "edges": []}
+        self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b"])
+
+
+class RunGraphTestBase(unittest.TestCase):
+    def setUp(self):
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+        )
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+
+_OUT_SCHEMA = {"type": "object", "required": ["content"]}
+_IN_SCHEMA = {"type": "object", "required": ["node_id"]}
+
+
+class RunGraphHappyPathTest(RunGraphTestBase):
+    def test_runs_nodes_in_order_and_returns_outputs(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
+                {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
+            ],
+            "edges": [],
+        }
+        calls = []
+
+        def build_prompt(node, outputs):
+            calls.append((node["node_id"], dict(outputs)))
+            return "prompt for " + node["node_id"]
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [{"content": "brd text"}, {"content": "prd text"}]
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                build_prompt, _IN_SCHEMA, _OUT_SCHEMA,
+            )
+
+        self.assertEqual(outputs, {"brd": {"content": "brd text"}, "prd": {"content": "prd text"}})
+        self.assertEqual(calls[0], ("brd", {}))
+        self.assertEqual(calls[1], ("prd", {"brd": {"content": "brd text"}}))
+
+    def test_creates_work_node_execution_rows_and_placeholder_skill(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}):
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+            )
+        row = self.conn.execute(
+            "SELECT status, subagent_type, skill_ref FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'"
+        ).fetchone()
+        self.assertEqual(row, ("done", "planner", "planner-doc-writing"))
+        skill_row = self.conn.execute("SELECT owner_org_node FROM skill WHERE skill_id = 'planner-doc-writing'").fetchone()
+        self.assertEqual(skill_row[0], "planner")
+
+    def test_resumes_without_rerunning_done_nodes(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "brd", "depends_on": []},
+                {"node_id": "prd", "depends_on": ["brd"]},
+            ],
+            "edges": [],
+        }
+        calls = []
+
+        def build_prompt(node, outputs):
+            return "p-" + node["node_id"]
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = lambda cwd, name, args, schema, timeout=600: calls.append(name) or {"content": "c-" + name}
+            wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                build_prompt, _IN_SCHEMA, _OUT_SCHEMA,
+            )
+        self.assertEqual(len(calls), 2)
+
+        # simulate a restart: run_graph called again for the same run_id
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute2:
+            mock_execute2.side_effect = AssertionError("should not be called -- both nodes are already done")
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                build_prompt, _IN_SCHEMA, _OUT_SCHEMA,
+            )
+        self.assertEqual(set(outputs.keys()), {"brd", "prd"})
+
+
+class RunGraphUsageLimitTest(RunGraphTestBase):
+    def test_usage_limit_reached_translates_to_quota_wait(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        attempts = {"n": 0}
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            attempts["n"] += 1
+            if attempts["n"] == 1:
+                raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session")
+            return {"content": "recovered"}
+
+        sleeps = []
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                sleep=lambda s: sleeps.append(s), now=lambda: 0.0,
+            )
+        self.assertEqual(outputs["brd"], {"content": "recovered"})
+        self.assertEqual(sleeps, [5 * 60])
+
+
+class RunGraphClassifierBlockedTest(RunGraphTestBase):
+    def test_classifier_blocked_marks_node_blocked_and_propagates(self):
+        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 side_effect=work_graph_execute.ClassifierBlockedError("blocked by classifier"),
+             ):
+            with self.assertRaises(work_graph_execute.ClassifierBlockedError):
+                wgo.run_graph(
+                    self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                    lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                )
+        node_exec_id = self.conn.execute(
+            "SELECT node_exec_id FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'"
+        ).fetchone()[0]
+        status = self.conn.execute(
+            "SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,)
+        ).fetchone()[0]
+        self.assertEqual(status, "blocked")
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -72,6 +72,63 @@ class ExtractGraphManifestTest(unittest.TestCase):
         self.assertIsNone(work_graph_plan._extract_graph_manifest("sorry, I can't do that"))
  
  
+class NormalizeGraphManifestTest(unittest.TestCase):
+    """U-031 regression: a real Planner call (2026-08-01) came back using
+    `id` instead of the documented `node_id`, and never populated any
+    node's own `depends_on` -- dependencies existed only as `edges`. This
+    is a real, observed deviation, not a hypothetical one; the fixture
+    below is the actual shape captured from that call (fields trimmed to
+    what normalization touches)."""
+
+    def test_already_well_formed_manifest_untouched(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
+                {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
+            ],
+            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
+        }
+        result = work_graph_plan._normalize_graph_manifest(manifest)
+        self.assertEqual(result["nodes"][1]["depends_on"], ["brd"])
+
+    def test_id_key_normalized_to_node_id(self):
+        manifest = {
+            "nodes": [{"id": "brd", "artifact_type": "brd"}, {"id": "prd", "artifact_type": "prd"}],
+            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
+        }
+        result = work_graph_plan._normalize_graph_manifest(manifest)
+        self.assertEqual([n["node_id"] for n in result["nodes"]], ["brd", "prd"])
+
+    def test_depends_on_derived_from_edges_when_absent(self):
+        manifest = {
+            "nodes": [{"id": "brd"}, {"id": "prd"}, {"id": "sdd"}, {"id": "benchmark"}],
+            "edges": [
+                {"from": "brd", "to": "prd", "edge_type": "needs_judgment"},
+                {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"},
+                {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"},
+            ],
+        }
+        result = work_graph_plan._normalize_graph_manifest(manifest)
+        by_id = {n["node_id"]: n for n in result["nodes"]}
+        self.assertEqual(by_id["brd"]["depends_on"], [])
+        self.assertEqual(by_id["prd"]["depends_on"], ["brd"])
+        self.assertEqual(by_id["sdd"]["depends_on"], ["prd"])
+        self.assertEqual(by_id["benchmark"]["depends_on"], ["sdd"])
+
+    def test_does_not_duplicate_dependency_already_present(self):
+        manifest = {
+            "nodes": [{"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}],
+            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
+        }
+        result = work_graph_plan._normalize_graph_manifest(manifest)
+        self.assertEqual(result["nodes"][1]["depends_on"], ["brd"])
+
+    def test_edge_referencing_unknown_node_id_ignored(self):
+        manifest = {"nodes": [{"id": "brd"}], "edges": [{"from": "brd", "to": "no-such-node"}]}
+        result = work_graph_plan._normalize_graph_manifest(manifest)
+        self.assertEqual(result["nodes"][0]["depends_on"], [])
+
+
 class HappyPathTest(PlanWorkGraphTestBase):
     def test_returns_run_id_and_graph_manifest(self):
         with patch("src.work_graph_plan.subprocess.run") as run:
@@ -109,6 +109,24 @@ class HappyPathTest(WorkNodeRunTestBase):
         final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0]
         self.assertEqual(final_status, "closed")
  
+    def test_unretryable_run_fn_failure_marks_node_blocked_not_stuck_running(self):
+        # any exception from run_fn other than a recognized usage-limit
+        # signal (e.g. CLASSIFIER_BLOCKED surfacing from
+        # work_graph_execute.execute_work_graph, 03-C §3.5 point 6) must
+        # not leave the row stuck at status='running' with no record of why
+        class SomeUnretryableError(Exception):
+            pass
+
+        def boom(cwd, model=None):
+            raise SomeUnretryableError("classifier blocked or similar")
+
+        with self.assertRaises(SomeUnretryableError):
+            work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=boom)
+        status = self.conn.execute(
+            "SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'"
+        ).fetchone()[0]
+        self.assertEqual(status, "blocked")
+
  
 class RetryOnSchemaFailureTest(WorkNodeRunTestBase):
     def test_retries_then_succeeds(self):
+"""Bridges `plan_work_graph()`'s `graph_manifest` to real per-node execution
+via `execute_work_graph()` + `run_work_node()` -- the missing integration
+between three previously isolated, individually-tested modules (U-011/
+U-012/U-013), closed during U-031's regression audit (2026-08-01): each of
+the three had its own unit tests (and even real-API validation) but none of
+them had ever actually been called by production code, and the one
+genuinely-validated closed loop (`e2e_pipeline.py`, U-014) used a much
+simpler, separate mechanism (flat `org_node_trigger.trigger_org_node()`
+calls) that bypassed all three.
+
+Design: each node's actual work happens via its OWN one-node dynamic
+workflow script (`write_workflow_script` + `execute_work_graph`), not one
+big multi-node script for the whole graph. This is a deliberate trade-off,
+not the "amortize orchestration cost across many nodes in one `claude -p`
+call" pattern 03-C §3.5 originally motivated `execute_work_graph` with --
+REQ-06's breakpoint-recovery requirement (resume after a process restart
+without re-running already-`done` nodes) is only achievable with per-node,
+Python-level persisted state (`work_node_run.py`'s whole design), which is
+fundamentally incompatible with wrapping the WHOLE graph in one opaque
+`claude` subprocess call Python has no visibility into mid-flight.
+
+Validated for real (2026-08-01): a one-node workflow script with a custom
+`agentType` + `schema` genuinely loads the target org_node's migrated
+persona (confirmed via a real call asking the agent to self-identify) and
+returns real `structured_output` -- see 05 §5.4.1.5.
+"""
+
+import json
+import time
+import uuid
+
+from src import work_graph_execute, work_node_run
+
+
+def topological_order(graph_manifest):
+    """Kahn's algorithm. Dependencies are read from BOTH `node['depends_on']`
+    and `graph_manifest['edges']` and merged -- `plan_work_graph()`'s own
+    output isn't schema-validated (it can't be, `--json-schema` doesn't
+    work with `--agent`, see work_graph_plan.py), so a real LLM response
+    might populate one and not the other; trusting only one would silently
+    mis-order nodes rather than raise. Raises ValueError on a cycle or a
+    dependency referencing an unknown node_id.
+    """
+    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)
+
+    ready = sorted(node_id for node_id, degree in in_degree.items() if degree == 0)
+    ordered = []
+    while ready:
+        node_id = ready.pop(0)
+        ordered.append(nodes[node_id])
+        for dependent in sorted(dependents[node_id]):
+            in_degree[dependent] -= 1
+            if in_degree[dependent] == 0:
+                ready.append(dependent)
+        ready.sort()
+    if len(ordered) != len(nodes):
+        raise ValueError(
+            "graph_manifest has a dependency cycle or a depends_on/edge references an unknown node_id"
+        )
+    return ordered
+
+
+def ensure_placeholder_skill(conn, skill_id, owner_org_node):
+    """No Skill content has actually been migrated from the Hermes profiles
+    yet (skill_update.py's own documented gap) -- work_node_execution.
+    skill_ref is a required FK, so a real row has to exist regardless.
+    Idempotent."""
+    exists = conn.execute("SELECT 1 FROM skill WHERE skill_id = ?", (skill_id,)).fetchone()
+    if exists is None:
+        conn.execute(
+            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+            "VALUES (?, ?, ?, 'unknown', 'new', 'active')",
+            (skill_id, owner_org_node, "skills/" + skill_id + "/"),
+        )
+        conn.commit()
+
+
+def _ensure_node_row(conn, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, input_payload):
+    existing = conn.execute(
+        "SELECT node_exec_id FROM work_node_execution WHERE run_id = ? AND node_id = ?", (run_id, node_id)
+    ).fetchone()
+    if existing:
+        return existing[0]
+    node_exec_id = str(uuid.uuid4())
+    conn.execute(
+        "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+        "input_schema_ref, output_schema_ref, input_payload, status) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')",
+        (node_exec_id, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, json.dumps(input_payload)),
+    )
+    conn.commit()
+    return node_exec_id
+
+
+def _render_node_script(script_name, org_node, prompt, output_schema, model):
+    model_opt = ", model: " + json.dumps(model) if model else ""
+    return (
+        "export const meta = {{ name: {name}, description: 'graphlemon Work Node ({org_node})' }}\n\n"
+        "const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema}{model_opt} }})\n\n"
+        "return result\n"
+    ).format(
+        name=json.dumps(script_name), org_node=org_node, org_node_json=json.dumps(org_node),
+        prompt=json.dumps(prompt), schema=json.dumps(output_schema), model_opt=model_opt,
+    )
+
+
+def _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout):
+    script_name = "graphlemon-node-" + node_exec_id.replace("-", "")
+
+    def run_fn(cwd, model=None):
+        script_content = _render_node_script(script_name, org_node, prompt, output_schema, model)
+        work_graph_execute.write_workflow_script(cwd, script_name, script_content)
+        try:
+            return work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout)
+        except work_graph_execute.UsageLimitReachedError as e:
+            work_node_run.raise_usage_limit_signal(e.quota_limit_type)
+        # ClassifierBlockedError / generic WorkGraphExecuteError propagate
+        # as-is -- run_work_node's own `except Exception` clause marks the
+        # node 'blocked' and re-raises, no special handling needed here.
+
+    return run_fn
+
+
+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,
+    sleep=time.sleep, now=time.time,
+):
+    """Runs every node in `graph_manifest` (topologically ordered) through
+    `run_work_node()`, each backed by a real one-node dynamic-workflow
+    `agent()` call using `org_node`'s migrated persona.
+
+    `build_prompt(node, outputs)` is called right before running that
+    node -- `outputs` contains every already-completed node's
+    output_payload so far (topological order guarantees a node's
+    dependencies are always already present), so prompts needing upstream
+    content (e.g. "here is the approved BRD...") can be built just-in-time
+    instead of requiring the whole graph's prompts precomputed before
+    anything runs. This also makes resume-after-restart work correctly:
+    `run_work_node` short-circuits already-`done` nodes and returns their
+    persisted output, which lands in `outputs` the same as a freshly-run
+    one would.
+
+    `needs_worktree=True` (03-C §3.6: "并行写代码的节点必须先申请
+    git_worktree_ref") isolates each node's actual file changes in its own
+    git worktree, merged back only once the node succeeds -- callers
+    writing code (Coder) should pass this; callers only producing document
+    content (Planner) don't need it.
+
+    Returns {node_id: output_payload}.
+    """
+    ensure_placeholder_skill(conn, skill_id, org_node)
+    ordered = topological_order(graph_manifest)
+    outputs = {}
+    for node in ordered:
+        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(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,
+        )
+        outputs[node_id] = output
+    return outputs
@@ -102,6 +102,35 @@ def _extract_graph_manifest(text):
     return None
  
  
+def _normalize_graph_manifest(manifest):
+    """Real LLM output has been observed (2026-08-01, U-031 integration
+    validation) to use `id` instead of the documented `node_id` key, and to
+    express dependencies purely via `edges` (from/to) without ever
+    populating each node's own `depends_on` array -- even though this
+    schema requires both. Normalizing here, right after parsing and before
+    persisting, means every downstream consumer (work_graph_orchestrate.py's
+    topological sort, v_task_detail's to-do tree, v_graph_visualizer's DAG
+    layering) gets a manifest that actually matches the documented shape,
+    instead of each one needing its own defensive "id or node_id,
+    depends_on or edges" fallback logic repeated independently. Mutates and
+    returns `manifest`.
+    """
+    nodes = manifest.get("nodes", [])
+    for node in nodes:
+        if "node_id" not in node and "id" in node:
+            node["node_id"] = node["id"]
+        node.setdefault("depends_on", [])
+
+    node_ids = {n.get("node_id") for n in nodes}
+    for edge in manifest.get("edges", []):
+        from_id, to_id = edge.get("from"), edge.get("to")
+        if from_id in node_ids and to_id in node_ids:
+            for node in nodes:
+                if node.get("node_id") == to_id and from_id not in node["depends_on"]:
+                    node["depends_on"].append(from_id)
+    return manifest
+
+
 def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300):
     """Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row
     (status='planning') -- moving it to 'running'/executing it is out of scope."""
@@ -134,6 +163,7 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time
     graph_manifest = _extract_graph_manifest(payload.get("result", ""))
     if not graph_manifest:
         raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload))
+    graph_manifest = _normalize_graph_manifest(graph_manifest)
  
     run_id = str(uuid.uuid4())
     conn.execute(
@@ -291,6 +291,22 @@ def run_work_node(
                 backoff_index = min(int(elapsed // (30 * 60)), len(_BACKOFF_SCHEDULE_MINUTES) - 1)
                 sleep(_BACKOFF_SCHEDULE_MINUTES[backoff_index] * 60)
                 continue
+            except Exception:
+                # Anything else run_fn raises (e.g. CLASSIFIER_BLOCKED from
+                # work_graph_execute.execute_work_graph, 03-C §3.5 point 6)
+                # is an unretryable node-level failure -- mark 'blocked'
+                # before re-raising so the row doesn't sit stuck at
+                # 'running' forever with no record of why. This is
+                # deliberately broad (not just ClassifierBlockedError):
+                # REQ-06's "系统应明确报错并转 blocked_pending_human，而不是
+                # 静默丢失进度" applies to any node-level catastrophic
+                # failure work_node_run.py doesn't have specific retry
+                # logic for, not just the one named error type.
+                conn.execute(
+                    "UPDATE work_node_execution SET status = 'blocked' WHERE node_exec_id = ?", (node_exec_id,)
+                )
+                conn.commit()
+                raise
  
             if _validate_schema_lite(output_payload, output_schema):
                 conn.execute(
@@ -347,6 +363,15 @@ def raise_if_usage_limited(text):
         raise _UsageLimitSignal(quota_type)
  
  
+def raise_usage_limit_signal(quota_limit_type):
+    """Same signal as raise_if_usage_limited, for callers that already know
+    the quota type structurally (e.g. work_graph_orchestrate.py wrapping
+    work_graph_execute.UsageLimitReachedError, which already parsed this
+    from the same detection regex at its own layer) rather than needing to
+    regex-match free text a second time."""
+    raise _UsageLimitSignal(quota_limit_type)
+
+
 def resume_work_graph_run(conn, run_id):
     """REQ-06 (04-P §4.2 p_work_graph_execution, breakpoint recovery):
     after a process restart, determine which nodes in `run_id` still need