E: cross-Org-Agent dynamic Work Graph design capability

Implements docs/todo/dynamic-work-graph-design-skill-design.md. G already answered FR-2 (how to mark a parallel-eligible node group): no new field needed -- topological_layers() infers independence purely from depends_on/edges, so the new Skill's job is just "express real dependencies honestly, don't add fake ones." New Skill skills/design-dynamic-work-graph/SKILL.md: the node-boundary / zero_token-vs-needs_judgment / per-node self-contained task_prompt methodology, wired into agent_profile.py's FRONTMATTER for coder and validator (Planner's own doc-chain path is untouched -- FR-4 only names Coder/Validator). work_graph_plan.py: new _GENERIC_PROMPT_TEMPLATE + generic_node_prompt_builder(), used whenever org_node != 'planner'. Planner keeps its existing fixed brd/prd/sdd/benchmark/... chain (_PROMPT_TEMPLATE) completely unchanged -- zero regression risk to the one real production planning path. Scoped per the requirements doc's non-goal #1 ("prove the mechanism works, don't cover every real scenario yet"): does NOT wire into e2e_pipeline.py::implement_code()'s existing single-node production path. New tests prove the mechanism itself: a Coder scenario that naturally splits into 3 independent modules produces a correctly-shaped graph_manifest, topological_layers() puts all three in one concurrent layer (ties E directly to G), and generic_node_prompt_builder drives it through run_graph() to completion. Full src suite (580 tests) passes with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

E: cross-Org-Agent dynamic Work Graph design capability
Implements docs/todo/dynamic-work-graph-design-skill-design.md. G already answered FR-2 (how to mark a parallel-eligible node group): no new field needed -- topological_layers() infers independence purely from depends_on/edges, so the new Skill's job is just "express real dependencies honestly, don't add fake ones." New Skill skills/design-dynamic-work-graph/SKILL.md: the node-boundary / zero_token-vs-needs_judgment / per-node self-contained task_prompt methodology, wired into agent_profile.py's FRONTMATTER for coder and validator (Planner's own doc-chain path is untouched -- FR-4 only names Coder/Validator). work_graph_plan.py: new _GENERIC_PROMPT_TEMPLATE + generic_node_prompt_builder(), used whenever org_node != 'planner'. Planner keeps its existing fixed brd/prd/sdd/benchmark/... chain (_PROMPT_TEMPLATE) completely unchanged -- zero regression risk to the one real production planning path. Scoped per the requirements doc's non-goal #1 ("prove the mechanism works, don't cover every real scenario yet"): does NOT wire into e2e_pipeline.py::implement_code()'s existing single-node production path. New tests prove the mechanism itself: a Coder scenario that naturally splits into 3 independent modules produces a correctly-shaped graph_manifest, topological_layers() puts all three in one concurrent layer (ties E directly to G), and generic_node_prompt_builder drives it through run_graph() to completion. Full src suite (580 tests) passes with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent bc0eaa5c
Showing 6 changed files with 305 additions and 22 deletions
docs/todo/dynamic-work-graph-design-skill-design.md
docs/todo/graph-engineering-open-items.md
skills/design-dynamic-work-graph/SKILL.md
src/agent_profile.py
src/test_work_graph_plan.py
src/work_graph_plan.py
+# 跨 Org Agent 的动态 Work Graph 设计能力——详细设计（SDD）
+
+**日期**: 2026-08-06
+**性质**: 详细设计，对应需求文档 `docs/todo/dynamic-work-graph-design-skill-requirements.md`（条目 E，已确认）。回答该需求文档第 8 节的开放问题。
+**依赖**：条目 G（`docs/work-graph-parallel-execution-design.md`）已实现——`topological_layers()`/`run_graph()` 已经在生产代码里。
+**状态**: 设计草案，待评审后实施。
+
+## 1. G 已经回答了 FR-2（并行分组字段格式）
+
+需求文档 FR-2 留空的"哪些节点属于同一个可并行分组"字段格式，G 的实际实现给出了确定答案：**不需要新字段**。`topological_layers()` 完全靠 `depends_on`/`edges` 结构本身推导独立性——没有相互依赖的节点自动落入同一层，`run_graph()` 自动并发执行同层节点。Planning Skill 的方法论职责因此简化为："如实表达真正的依赖关系,不要为了图好看而人为加上并不存在的 depends_on"（14 步版对照笔记"边是数据契约"这条原则的直接应用），不需要额外发明一个"parallel_group"标记。
+
+## 2. 新 Skill：`design-dynamic-work-graph`
+
+`skills/design-dynamic-work-graph/SKILL.md`，与 `planning-requirement-expansion`、`code-task-development` 等现有 Skill 平级存放。内容方法论（不含代码）：
+
+1. 节点边界怎么划：一个节点 = 一次独立的、有明确输入输出契约的 Work Node 执行（一次 `claude -p` 调用）。任务小到一次调用就能完成时不需要拆图（呼应需求文档第 6 节约束）。
+2. `edge_type` 怎么选：`zero_token`（纯聚合/规约，不需要 Agent 推理，比如"把三份判断投票汇总"）vs `needs_judgment`（下游需要真正读懂上游内容才能继续，比如"PRD 依赖 BRD 的实际内容"）。
+3. 如何识别可并行的候选点：如实列出每对节点之间是否存在真实数据流动——没有数据流动就不要加 `depends_on`，G 会自动把它们并发调度。
+4. 每个节点必须自带 `task_prompt`（该节点执行时使用的完整提示词，自包含，不依赖调用方另外拼接）——这是 Coder/Validator 场景与 Planner 场景的关键区别：Planner 的四个节点类型固定（brd/prd/sdd/benchmark...），下游代码用一个查表 `_DOC_PROMPTS[artifact_type]` 就能拿到提示词；Coder/Validator 的子任务没有固定的小词表，必须由 Claude 自己在规划阶段把每个节点的提示词写出来。
+
+## 3. `work_graph_plan.py` 改造（FR-3）
+
+**不改动 Planner 现有路径**：`_PROMPT_TEMPLATE`（brd/prd/sdd/benchmark 固定链）与 `_normalize_graph_manifest` 的 canonical-chain 强制逻辑完全保留，`org_node == "planner"` 时走原有分支，零回归风险。
+
+新增：
+
+- `_GENERIC_PROMPT_TEMPLATE`：当 `org_node != "planner"` 时使用，要求 Claude 依据 `design-dynamic-work-graph` Skill 的方法论，把 `goal_text` 拆成一个 `graph_manifest`，每个节点除 `node_id`/`depends_on` 外必须包含 `task_prompt`（自包含提示词）；输出格式（fenced ```json 块）跟现有约定一致，复用同一套 `_extract_graph_manifest` 解析逻辑。
+- `plan_work_graph()` 内部按 `org_node == "planner"` 二选一模板，函数签名/返回值不变（需求文档非目标 #2）。
+- `generic_node_prompt_builder(graph_manifest)`：新增的公开辅助函数，返回一个 `build_prompt(node, outputs)` 闭包，直接读取 `node["task_prompt"]`（`run_graph()` 的 `build_prompt` 约定不变，仍然是"给定 node + 已完成 outputs，返回这次调用的提示词"）——之所以放在 `work_graph_plan.py` 而不是 `e2e_pipeline.py`，是因为它跟生成侧的字段约定（`task_prompt` 这个 key 名）强耦合，两者应该在同一个模块里维护，避免字段改名时两处不同步。
+
+`_normalize_graph_manifest` 不需要改动——canonical-chain 强制分支本来就有 `all(t in by_artifact_type for t in _CANONICAL_DOC_CHAIN)` 前置判断，Coder/Validator 的动态图没有 brd/prd/sdd/benchmark 节点，天然不触发，其余的 `id`→`node_id`/`depends_on` 兜底逻辑对新路径同样适用，不需要区分。
+
+## 4. Skill 接入 Coder/Validator
+
+`src/agent_profile.py::FRONTMATTER["coder"]["skills"]`/`["validator"]["skills"]` 各自追加 `"design-dynamic-work-graph"`。`.claude/agents/{coder,validator}.md` 是从这里渲染出来的（`render_agent_definition`），不需要手改部署产物。
+
+## 5. 验证范围（对应需求文档验收标准，按非目标 #1 收窄）
+
+需求文档 FR-5/验收标准 1 优先级是"应该"，且非目标 #1 明确"不要求第一版就覆盖 Coder/Validator 的全部真实工作场景，只需要证明机制本身走得通"。本轮实现范围：
+
+- 证明 `plan_work_graph(conn, "coder", ...)` 能生成一个正确带 `task_prompt`、互不依赖节点的 `graph_manifest`（mock `claude` 子进程返回一个"给三个独立模块各自补单元测试"的真实场景响应）。
+- 证明该 `graph_manifest` 经 `topological_layers()` 分层后，三个节点确实落在同一层（跟 G 的实现直接挂钩，不是空谈）。
+- 证明 `generic_node_prompt_builder` 生成的 `build_prompt` 能驱动 `run_graph()` 端到端跑完（mock `execute_work_graph`）。
+- **不改动** `e2e_pipeline.py::implement_code()` 的现有单节点生产路径——那是更大的、需要独立评估的改动（如何决定"这个 Coder 任务复杂到该规划成图"本身就是一个开放问题，需求文档风险第 3 条已经点出），留给后续任务。
+
+## 6. 回答需求文档第 8 节开放问题
+
+- 新 Skill 命名：`design-dynamic-work-graph`，目录 `skills/design-dynamic-work-graph/`。
+- `_PROMPT_TEMPLATE` 改造方式：按 `org_node` 二选一模板（Planner 固定链 vs 通用动态规划），不是让 Claude 完全自由决定 Planner 要生成哪些节点——Planner 那条链是"架构常量"，如 `_normalize_graph_manifest` 自己的注释所说,不应该变成每次都不同。
+- FR-2 并行分组字段格式：不需要,G 靠 `depends_on`/`edges` 结构自动推导（见第 1 节）。
+- Coder/Validator 第一个接入场景：mock 场景"给三个独立模块各自补一个单元测试"，用于验证机制可用性，不接入真实生产路径。
@@ -76,7 +76,7 @@
  
 ## E. Planning Skill 提示词强化——识别可并行节点 + 通用 Work Graph 设计能力
  
-**状态**：✅ 已确认，且范围比原开放问题更大
+**状态**：✅ 已实现（2026-08-06）→ 需求文档 `docs/todo/dynamic-work-graph-design-skill-requirements.md` + 设计文档 `docs/todo/dynamic-work-graph-design-skill-design.md`
  
 **来源**：综述版对照 §7 开放问题 3 + 14 步版对照 §3 开放问题 2（同一主题：让 Planning 阶段更早识别并行机会）
  
@@ -84,9 +84,7 @@
  
 **确认结果（2026-08-05）**：提示词强化肯定要做——把"对每个箭头问是否真的有数据流动"的提问方式、以及"并行可分解 / 单一上下文装不下 / 收益覆盖成本"这三条尺子写进 Planning 相关 prompt。**但范围比原开放问题更大**：不能只强化 Planner 自己写 BRD/PRD/SDD/Benchmark 那条链的 prompt——架构文档 §5 本来就说"每个 Org Agent 拥有一个 Planning Skill"（Coder 和 Validator 各自也要规划自己的 Work Graph），"如何识别 fan-out/fan-in、设计一张动态 Work Graph"这套方法论应该是**跨 Org Agent 通用**的能力，而不是分别在 Planner/Coder/Validator 三份 Planning Skill 里各写一遍。用户原话："planner agent 里面要有多个 skill，不仅仅是 brd prd sdd benchmark 还有给其他 org graph 的 node 节点的 agent 做各种动态 work graph 的设计"。
  
-**待办**：✅ 需求文档已产出 → `docs/dynamic-work-graph-design-skill-requirements.md`。深挖代码后发现真实缺口比原开放问题更大：`work_graph_plan.py::plan_work_graph()` 的提示词目前**硬编码**成永远只生成 brd/prd/sdd/benchmark 四个节点，且只有 Planner（`e2e_pipeline.py::plan_documents`）真正调用它——**Coder 和 Validator 现在根本没有走"先规划图再执行"这条路径**，走的是更简化的单次 `org_node_trigger.trigger_org_node()` 调用。所以 E 不只是"强化提示词"，而是要先把动态 Work Graph 规划能力从 Planner 专属扩展成三个 Org Agent 通用。
-
-**依赖关系**：这条和「G. Work Graph 节点并行执行」是互补的——G 解决"执行层能不能并发跑"，E 解决"规划层会不会生成值得并发的图"。两条需要一起排期,建议先看 G 的需求文档,E 的需求文档要在 G 确认的数据结构（`graph_manifest` 里怎么表达"这组节点可并行"）基础上写,避免两边字段对不上。
+**待办**：✅ 已实现（2026-08-06）。G 已经回答了 FR-2（并行分组字段格式）：不需要新字段，`topological_layers()` 完全靠 `depends_on`/`edges` 结构本身推导独立性——设计文档第 1 节记录了这个结论。实现内容：新 Skill `skills/design-dynamic-work-graph/SKILL.md`（跨 Org Agent 通用方法论，接入 `agent_profile.py::FRONTMATTER["coder"]["skills"]`/`["validator"]["skills"]`）；`work_graph_plan.py` 新增 `_GENERIC_PROMPT_TEMPLATE`（`org_node != "planner"` 时使用，每节点自带 `task_prompt`，Planner 原有固定链路 `_PROMPT_TEMPLATE` 完全不动，零回归）+ `generic_node_prompt_builder()`。按需求文档非目标 #1（"不要求第一版就覆盖 Coder/Validator 全部真实工作场景，只需要证明机制走得通"）收窄范围：**没有**改动 `e2e_pipeline.py::implement_code()` 的现有单节点生产路径——新增测试证明"给 Coder 一个可拆成 3 个独立模块补测试的场景 → 生成正确的 `graph_manifest` → `topological_layers()` 确认三个节点落在同一层 → `run_graph()` 端到端跑完"这条链路本身成立（`src/test_work_graph_plan.py`：`GenericPlanningPromptTest`/`CoderThreeIndependentModulesScenarioTest`/`GenericNodePromptBuilderTest`/`GenericPathEndToEndViaRunGraphTest`），不是接入真实生产路径。`python3 -m unittest discover -s src`（580 tests）全绿，零回归。
  
 ---
  
@@ -118,20 +116,40 @@
  
 ---
  
-## 确认结果汇总（2026-08-05）
+## H. 统一 Role Package 抽象层
+
+**状态**：🟡 需求文档已产出，待评审/排期
+
+**来源**：不属于三份 Graph Engineering 对照笔记，是用户对照 Agentic Enterprise 产品目标（"role + context + skills + applications + workflows + permissions + acceptance criteria；all essentially are context for intelligent computing"）单独提出的差距检查，2026-08-06。
+
+**背景**：逐维度核对代码发现，七个维度里六个都有对应实现（applications 是真空白），但没有被收敛成一个统一的"role package"格式——理解"Validator 这个角色完整定义是什么"现在得同时读 `agent_profile.py` + `skills/qa-task-validation/SKILL.md` + `.claude/agent-memory/validator/` + `docs/sdd/06-验收与测试锚点.md` 四个地方拼出来。更根本的问题：`role` 本身不是数据驱动的——`ORG_NODES = ("planner", "coder", "validator")` 硬编码元组，`schema.py` 四张表的 `CHECK (org_node IN ('planner','coder','validator'))` 约束——新增角色需要改代码和 schema，不是"包一个新角色进来"。
+
+**待办**：需求文档已产出 → `docs/role-package-abstraction-requirements.md`。核心可验证标准：能不能只靠"写一份新的 Role Package"、不改 `schema.py` 约束、不改 `agent_profile.py` 硬编码字典，就让系统认识一个新角色。已知这条改动面比 B/D/E/F/G 任何一条都大（4 张表的 CHECK 约束、`agent_profile.py` 被多处代码依赖），且 workflows 维度依赖 E 的产出。
+
+---
+
+## 工作量与优先级评估（2026-08-06）
+
+| 项 | 一句话 | 阶段 | 工作量 | 优先级 | 依赖 |
+|---|---|---|---|---|---|
+| **G** 并行执行 | Work Graph 层内并发、层间屏障 | ✅ **已实现**（`bc0eaa5`，575 测试全绿） | M | 已完成 | 无 |
+| **E** 动态 Work Graph 设计能力 | 让 Coder/Validator 也能规划自己的图，Planning 提示词识别可并行节点 | ✅ **已实现**（2026-08-06，580 测试全绿） | L | 已完成 | G（已满足） |
+| **F** 三视角评审团 fan-out | `judge_root_cause` 从 Python 函数改成 Work Graph 菱形拓扑 | 设计已产出 → `docs/todo/verifier-fan-out-jury-design.md`，实现中 | M | P0（G/E 都已落地，正在推进） | 无（G 已满足） |
+| **B** loop-until-dry Skill | Validator 未知规模扫描收敛能力 | 仅需求 | S/M | P2 | 无 |
+| **H** Role Package 抽象层 | 统一 role/context/skills/applications/workflows/permissions/acceptance 打包格式 | 仅需求 | XL | P2（战略级，不阻塞其他项） | workflows 字段依赖 E 的产出 |
+| **D** Time-travel 调试 | 回到历史节点分叉重跑 | 仅需求 | L | P3 | 无强依赖，但价值随图变复杂（E 落地后）才会显现 |
+| C 按节点选模型 | — | ⛔ 已确认不做 | — | — | — |
+| A 术语对照 | — | ✅ 已完成 | — | — | — |
  
-| 项 | 结果 | 需求文档 |
-|---|---|---|
-| A | ✅ 已完成 | 无需求文档，直接改了 `docs/graphlemon_architecture.md` §2 术语表 |
-| B | ✅ 已确认，需求文档已产出 | `docs/validator-loop-until-dry-skill-requirements.md` |
-| C | ⛔ 暂不做 | 无 |
-| D | ✅ 已确认，需求文档已产出 | `docs/work-graph-time-travel-debugging-requirements.md` |
-| E | ✅ 已确认，需求文档已产出（范围比原开放问题更大） | `docs/dynamic-work-graph-design-skill-requirements.md` |
-| F | ✅ 已确认，需求文档已产出 | `docs/verifier-fan-out-jury-requirements.md` |
-| G | ✅ 已实现（2026-08-06） | `docs/work-graph-parallel-execution-requirements.md` + `docs/work-graph-parallel-execution-design.md` |
+**2026-08-06 更新的关键判断**：G 已经被 graphlemon 自己的自举循环实现并验证（不是本轮对话做的——`git log` 显示提交 `bc0eaa5`，`src/work_graph_orchestrate.py` 里 `topological_layers()`/`_LockedConn`/`WorkGraphLayerError` 都是真代码，`python3 -m unittest discover -s src` 跑出 575 测试全绿，跟 G 条目自己记录的实现细节吻合）。这把之前"G 和 E 必须一起排"的判断更新为：**E 现在是唯一的剩余瓶颈**——并行执行引擎已经是真的了，但 `plan_work_graph()` 的提示词依然硬编码只产出一条 BRD→PRD→SDD→Benchmark 直链，没有 E，G 造好的并行能力仍然没有真实的图可以运行。F 也从"强依赖 G，排在后面"变成"G 的执行引擎已经存在，可以直接排"，工作量估算不变但不再需要等待。
  
-**当前状态**：A 已落地；B/C/D/E/F/G 六项全部完成需求文档（BRD/PRD 层面），均未进入详细设计（SDD 层面）。
+**工作量估算依据**：G（已完成，实际证实是 M 量级——6 步实施计划，改动集中在一个模块）；E（L——Coder/Validator 从未真正走过 `plan_work_graph`/`run_graph` 路径，是全新集成，还要防止破坏 Planner 现有生产路径）；F（M——判定逻辑已验证过，只是把 Python 函数调用改造成图形状）；B（S/M——不碰执行引擎，纯提示词工程）；H（XL——自己的需求文档写明"改动面比 B/D/E/F/G 任何一条都大"，4 张表 CHECK 约束+多处代码依赖+从零定义 applications）；D（L——动一条有连锁风险的唯一约束，且控制台可视化工作量可能比后端还大）。
  
-**推进顺序建议**：G 是 E、F 的前置依赖（E 的并行分组字段格式、F 的 fan-out 时间收益都需要 G 先定型），建议优先把 G 推进到详细设计。B、D 相对独立，可以随时单独推进详细设计。E、F 的详细设计需要等 G 的数据结构定下来。
+**当前推进顺序建议**：
+1. **E**（P0）——现在是把 G 的投入变成真实收益的唯一剩余步骤，优先推进详细设计。
+2. **F**（P1）——G 的执行引擎已就绪，可以直接进入详细设计，不必再等。
+3. **B**（P2）——随时可插空，跟其他项无依赖冲突。
+4. **H**（P2，建议独立设计周期）——价值是长期的（系统现在被硬编码死在 3 个角色上），不阻塞任何短期工作。
+5. **D**（P3）——现在的图都还很简单，time-travel 要解决的"大图不好调"痛点要等 E 落地、图真的变复杂后才会显现。
  
-下一步要我先把哪一份需求文档推进到详细设计（SDD 层面）？还是先把六份需求文档过一遍评审，确认没有需要调整的地方，再决定设计阶段的排期？
+下一步要我先把 E 推进到详细设计吗？
+---
+name: design-dynamic-work-graph
+description: |
+  跨 Org Agent（Planner/Coder/Validator 共用）的通用方法论：如何把一个目标拆成一张可执行的
+  graph_manifest（节点、依赖、每节点自包含提示词），供 work_graph_plan.py::plan_work_graph()
+  的通用规划路径驱动、work_graph_orchestrate.py::run_graph() 执行。
+version: 1.0.0
+tags:
+  - planning
+  - work-graph
+  - orchestration
+metadata:
+  origin: graphlemon-native
+  related: [G（并行执行）docs/work-graph-parallel-execution-design.md, E docs/todo/dynamic-work-graph-design-skill-design.md]
+platforms: [linux, macos, windows]
+---
+
+# SKILL：design-dynamic-work-graph
+
+## 适用场景
+- 任务复杂到值得先规划一张图、再按图执行，而不是一次单次调用就能完成（不是所有任务都需要上图——任务很小或孤立时，直接一次调用更合适）。
+- 调用方是 Planner 自身以外的 Org Agent（Coder/Validator），需要为自己的一批子任务生成 `graph_manifest`。
+
+## 输出：一个 `graph_manifest`
+```json
+{
+  "nodes": [
+    {"node_id": "...", "depends_on": ["..."], "task_prompt": "..."}
+  ],
+  "edges": [
+    {"from": "...", "to": "...", "edge_type": "zero_token" | "needs_judgment"}
+  ]
+}
+```
+
+每个节点必须自带 `task_prompt`：该节点执行时使用的完整、自包含提示词，不依赖调用方另外拼接上下文。这是跟 Planner 自己的 BRD/PRD/SDD/Benchmark 链的关键区别——Planner 的节点类型固定在一个小词表里，下游代码按 `artifact_type` 查表就能拿到提示词；Coder/Validator 的子任务没有这种固定词表，提示词必须在规划阶段由你自己写清楚。
+
+## 规则（强制）
+
+1. **节点边界**：一个节点 = 一次独立的、有明确输入输出契约的 Work Node 执行（一次 `claude -p` 调用）。不要把一个连贯的思考过程拆成多个人为的小节点。
+2. **如实表达依赖，不要多加**：`depends_on` 只写真正存在数据流动的依赖——下游节点的 `task_prompt` 真的需要引用上游节点的产出内容，才算依赖。**不要为了图看起来更"有序"而人为串行化本来互不依赖的节点**：执行引擎（`topological_layers()`）会把没有依赖关系的节点自动分到同一层并发执行，多余的依赖会白白拖慢总执行时间，还会让"这几个节点该并行"这条信息从图里消失。
+3. **`edge_type` 二选一**：
+   - `zero_token`：纯聚合/规约边，下游不需要 Agent 推理就能处理上游输出（比如"把几份判断汇总计数"）。这类聚合优先用普通代码完成，不要为它专门生成一个节点。
+   - `needs_judgment`：下游真的需要读懂上游内容、做出判断才能继续（比如"根据 SDD 内容写实现"）。
+4. **任务小就不要上图**：如果目标本身一次 Work Node 调用就能完成，直接返回一个单节点 `graph_manifest`（或者告知调用方不需要走图规划路径），不要为了"用上这套机制"强行拆分。
+5. **`node_id` 唯一、可读**：用能表达该节点做什么的短标识符（如 `test-module-a`），不要用无意义的序号。
+
+## 分解方法
+
+1. 先列出这批子任务里，每一对任务之间是否存在真实的数据流动（"B 的 `task_prompt` 是否真的需要 A 的产出内容"）。没有则不连边。
+2. 对确实存在的依赖，判断是 `zero_token` 还是 `needs_judgment`。
+3. 为每个节点写出自包含的 `task_prompt`——假设执行它的 Agent 看不到这次规划的任何上下文，只能看到这一个节点的 `task_prompt` 和它依赖节点的产出。
+4. 检查是否有环（`topological_layers()` 遇到环会报错）——如果两个节点互相需要对方的产出，说明边界划错了，需要合并成一个节点或重新拆分。
+
+## 自审清单
+- [ ] 每个节点都有 `task_prompt`，且不依赖这次规划对话里才有的隐含上下文。
+- [ ] 没有为了让图"看起来有序"而添加的虚假 `depends_on`。
+- [ ] 没有本来可以用普通代码完成的聚合被写成了一个需要 Agent 推理的节点。
+- [ ] 图里没有环。
+- [ ] 如果这批任务本质上一次调用就能做完，没有为了走图机制而强行拆分。
@@ -65,13 +65,14 @@ FRONTMATTER = {
             "code-agent-engineering",
             "debugging-toolkit",
             "software-planning-execution",
+            "design-dynamic-work-graph",
         ],
         "tools": ["Read", "Write", "Edit", "Bash", "Grep", "Glob"],
     },
     "validator": {
         "memory": "project",
         "model": "claude-sonnet-5",
-        "skills": ["qa-task-validation", "unattended-agent-operations"],
+        "skills": ["qa-task-validation", "unattended-agent-operations", "design-dynamic-work-graph"],
         "tools": ["Read", "Bash", "Grep", "Glob"],
         # 2026-07-31 fix, found for real running U-014's full closed loop:
         # scoped Bash patterns here ("Bash(git commit:*)") don't just block
@@ -12,7 +12,7 @@ import subprocess
 import unittest
 from unittest.mock import patch
  
-from src import schema, work_graph_plan
+from src import schema, work_graph_execute, work_graph_orchestrate, work_graph_plan
  
 _MANIFEST = {
     "nodes": [
@@ -41,7 +41,12 @@ def _fenced_result(manifest, wrapper_key=None):
  
 class PlanWorkGraphTestBase(unittest.TestCase):
     def setUp(self):
-        self.conn = schema.connect(":memory:")
+        # check_same_thread=False: GenericPathEndToEndViaRunGraphTest below
+        # drives a real multi-node concurrent layer through run_graph()
+        # (G), which calls into this conn from ThreadPoolExecutor worker
+        # threads via _LockedConn -- see test_work_graph_orchestrate.py's
+        # RunGraphTestBase for the same reasoning.
+        self.conn = schema.connect(":memory:", check_same_thread=False)
         schema.create_schema(self.conn)
         self.conn.execute(
             "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
@@ -268,5 +273,105 @@ class FailureTest(PlanWorkGraphTestBase):
         self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM work_graph_run").fetchone()[0], 0)
  
  
+# E (2026-08-06): cross-Org-Agent dynamic planning path -- org_node != 'planner'
+# gets _GENERIC_PROMPT_TEMPLATE instead of the fixed doc-chain prompt, and each
+# node carries its own task_prompt instead of a fixed artifact_type.
+
+_CODER_THREE_MODULE_MANIFEST = {
+    "nodes": [
+        {"node_id": "test-module-a", "depends_on": [], "task_prompt": "Write unit tests for module_a.py."},
+        {"node_id": "test-module-b", "depends_on": [], "task_prompt": "Write unit tests for module_b.py."},
+        {"node_id": "test-module-c", "depends_on": [], "task_prompt": "Write unit tests for module_c.py."},
+    ],
+    "edges": [],
+}
+
+
+class GenericPlanningPromptTest(PlanWorkGraphTestBase):
+    def test_planner_still_gets_fixed_doc_chain_template(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertIn("BRD, PRD, SDD, Benchmark", prompt)
+        self.assertNotIn("design-dynamic-work-graph", prompt)
+
+    def test_coder_gets_generic_template_referencing_the_skill(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertIn("design-dynamic-work-graph", prompt)
+        self.assertNotIn("BRD, PRD, SDD, Benchmark", prompt)
+        self.assertIn("task_prompt", prompt)
+
+
+class CoderThreeIndependentModulesScenarioTest(PlanWorkGraphTestBase):
+    """E acceptance criterion 1: a Coder scenario that naturally splits into
+    2-3 independent sub-tasks produces a graph_manifest via the new generic
+    path (not the old single trigger_org_node call), correctly marking those
+    nodes as mutually independent."""
+
+    def test_three_independent_modules_produce_correctly_shaped_manifest(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            result = work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
+
+        manifest = result["graph_manifest"]
+        self.assertEqual(len(manifest["nodes"]), 3)
+        for node in manifest["nodes"]:
+            self.assertEqual(node["depends_on"], [])
+            self.assertIn("task_prompt", node)
+
+        # E's whole point ties directly into G: independence is expressed
+        # structurally (empty depends_on), so topological_layers() must put
+        # all three in ONE concurrent layer, not three serial ones.
+        layers = work_graph_orchestrate.topological_layers(manifest)
+        self.assertEqual(len(layers), 1)
+        self.assertEqual({n["node_id"] for n in layers[0]}, {"test-module-a", "test-module-b", "test-module-c"})
+
+
+class GenericNodePromptBuilderTest(unittest.TestCase):
+    def test_returns_the_node_own_task_prompt(self):
+        node = {"node_id": "test-module-a", "task_prompt": "Write unit tests for module_a.py."}
+        self.assertEqual(
+            work_graph_plan.generic_node_prompt_builder(node, {}), "Write unit tests for module_a.py."
+        )
+
+
+class GenericPathEndToEndViaRunGraphTest(PlanWorkGraphTestBase):
+    """E acceptance criterion (FR-5): the generated graph_manifest, driven by
+    generic_node_prompt_builder, actually runs to completion through
+    run_graph() -- not just a standalone parsing/shape check."""
+
+    def test_three_independent_nodes_execute_and_return_outputs(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            plan_result = work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
+
+        # plan_work_graph() already inserted the work_graph_run row
+        # (status='planning') -- run_graph() doesn't require any particular
+        # status on it, so nothing more to set up here.
+        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: {"content": "done: " + name}
+            outputs = work_graph_orchestrate.run_graph(
+                self.conn, plan_result["run_id"], "coder", "coder-implementation", "/tmp/proj-1",
+                plan_result["graph_manifest"], work_graph_plan.generic_node_prompt_builder,
+                {"type": "object", "required": ["node_id"]}, {"type": "object", "required": ["content"]},
+            )
+
+        self.assertEqual(set(outputs.keys()), {"test-module-a", "test-module-b", "test-module-c"})
+        self.assertEqual(mock_execute.call_count, 3)
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -77,6 +77,34 @@ _PROMPT_TEMPLATE = (
     "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key."
 )
  
+# E (2026-08-06): the generic, cross-Org-Agent path -- used for every
+# org_node OTHER than 'planner'. Planner's own doc-writing chain stays on
+# _PROMPT_TEMPLATE above unchanged (zero regression risk, see
+# docs/todo/dynamic-work-graph-design-skill-design.md §3): that chain is a
+# fixed architectural constant (brd->prd->sdd->benchmark->...), not
+# something that should vary call to call, whereas Coder/Validator's
+# sub-tasks have no such fixed small vocabulary of node types. Each node
+# here carries its own self-contained `task_prompt` (see the
+# design-dynamic-work-graph Skill this prompt points at) instead of an
+# `artifact_type` that a fixed lookup table like _DOC_PROMPTS could resolve.
+_GENERIC_PROMPT_TEMPLATE = (
+    "Plan (do not execute) a Work Graph for this goal: {goal_text}\n"
+    "Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge "
+    "when a node's task_prompt genuinely needs another node's output content -- nodes with "
+    "no real data dependency between them will be scheduled to run CONCURRENTLY, so adding "
+    "an edge just to make the graph look ordered actively slows execution down. If this goal "
+    "is small enough to finish in one call, output a single-node graph_manifest rather than "
+    "forcing a split.\n"
+    "Output a graph_manifest whose nodes each have: node_id (a short readable identifier), "
+    "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt "
+    "for that node's own execution -- assume the agent running it sees nothing except this "
+    "task_prompt and its dependencies' outputs, not this planning conversation). Every edge "
+    "needs edge_type: 'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for "
+    "anything requiring the agent to actually reason about content.\n"
+    "End your reply with the graph_manifest as a single fenced ```json code block "
+    "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key."
+)
+
 _JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)
  
  
@@ -180,8 +208,16 @@ def _normalize_graph_manifest(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."""
-    prompt = _PROMPT_TEMPLATE.format(goal_text=goal_text)
+    (status='planning') -- moving it to 'running'/executing it is out of scope.
+
+    E (2026-08-06): `org_node == 'planner'` still gets the fixed
+    brd/prd/sdd/benchmark/... chain prompt (_PROMPT_TEMPLATE, unchanged).
+    Every other org_node gets _GENERIC_PROMPT_TEMPLATE -- the cross-Org-
+    Agent dynamic planning path described in the design-dynamic-work-graph
+    Skill, whose nodes carry their own task_prompt instead of a fixed
+    artifact_type. Function signature/return shape are unchanged either way."""
+    template = _PROMPT_TEMPLATE if org_node == "planner" else _GENERIC_PROMPT_TEMPLATE
+    prompt = template.format(goal_text=goal_text)
  
     result = subprocess.run(
         [
@@ -220,3 +256,15 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time
     )
     conn.commit()
     return {"run_id": run_id, "graph_manifest": graph_manifest}
+
+
+def generic_node_prompt_builder(node, outputs):
+    """E (2026-08-06): the `build_prompt(node, outputs)` run_graph() needs
+    for a graph_manifest produced via _GENERIC_PROMPT_TEMPLATE -- lives
+    here, not in e2e_pipeline.py, because it's tightly coupled to the
+    `task_prompt` field convention that prompt asks for (keeping both in
+    the same module means a field rename can't silently desync producer
+    and consumer). No factory/closure needed: every node is self-contained
+    per the design-dynamic-work-graph Skill's contract, so `outputs`
+    (already-completed dependency content) isn't needed here."""
+    return node["task_prompt"]