F: three-perspective jury as a real Work Graph fan-out

Implements docs/todo/verifier-fan-out-jury-design.md. root_cause_trace.py: _tally_perspectives() extracts the existing >=2/3 rule out of judge_root_cause so the closure-based version and the new judge_root_cause_via_work_graph() share one implementation, not two that could drift apart. The fan-in vote stays plain Python (zero_token aggregation per the design-dynamic-work-graph Skill's own edge_type distinction) rather than becoming a fourth Work Node -- no reason to pay for an Agent call just to count votes. judge_root_cause_via_work_graph(): each perspective becomes an independent Work Node with real schema-constrained structured output ({"category", "detail"} directly), not real_validator_perspective's manual "CATEGORY:"/"DETAIL:" text-line parsing. judge_root_cause / real_validator_perspective are kept, not removed -- documented as the lightweight fallback for scenarios that don't need worktree/Work Graph overhead. org_level_loop.run_fix_loop (the one real production caller) now calls the Work Graph version; the fixer/verify calls are unchanged flat trigger_org_node calls. Real bug found and fixed along the way: this is the first production code path that actually triggers a run_graph() multi-node CONCURRENT layer (every graph_manifest before F was single-node or a strict serial chain, so G's ThreadPoolExecutor path had never really run in production). scheduler.py::run_goal_to_completion and task_execution.py::start_task_execution_in_background both created their conn without check_same_thread=False -- fixed both, since their conn can now reach run_graph()'s worker threads via run_fix_loop. Test boundary migrated in test_org_level_loop.py/test_scheduler.py: the 3 perspective calls are now mocked at work_graph_execute.write_workflow_script/ execute_work_graph (matching test_work_graph_orchestrate.py's pattern) instead of org_node_trigger.trigger_org_node. New tests cover: three real independent Work Node executions, judgment-rule equivalence with the closure version, clean per-node input_payload, and concurrent wall-clock overlap across the three perspectives. Full src suite (586 tests) + supervisor suite (125 tests) pass with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

F: three-perspective jury as a real Work Graph fan-out
Implements docs/todo/verifier-fan-out-jury-design.md. root_cause_trace.py: _tally_perspectives() extracts the existing >=2/3 rule out of judge_root_cause so the closure-based version and the new judge_root_cause_via_work_graph() share one implementation, not two that could drift apart. The fan-in vote stays plain Python (zero_token aggregation per the design-dynamic-work-graph Skill's own edge_type distinction) rather than becoming a fourth Work Node -- no reason to pay for an Agent call just to count votes. judge_root_cause_via_work_graph(): each perspective becomes an independent Work Node with real schema-constrained structured output ({"category", "detail"} directly), not real_validator_perspective's manual "CATEGORY:"/"DETAIL:" text-line parsing. judge_root_cause / real_validator_perspective are kept, not removed -- documented as the lightweight fallback for scenarios that don't need worktree/Work Graph overhead. org_level_loop.run_fix_loop (the one real production caller) now calls the Work Graph version; the fixer/verify calls are unchanged flat trigger_org_node calls. Real bug found and fixed along the way: this is the first production code path that actually triggers a run_graph() multi-node CONCURRENT layer (every graph_manifest before F was single-node or a strict serial chain, so G's ThreadPoolExecutor path had never really run in production). scheduler.py::run_goal_to_completion and task_execution.py::start_task_execution_in_background both created their conn without check_same_thread=False -- fixed both, since their conn can now reach run_graph()'s worker threads via run_fix_loop. Test boundary migrated in test_org_level_loop.py/test_scheduler.py: the 3 perspective calls are now mocked at work_graph_execute.write_workflow_script/ execute_work_graph (matching test_work_graph_orchestrate.py's pattern) instead of org_node_trigger.trigger_org_node. New tests cover: three real independent Work Node executions, judgment-rule equivalence with the closure version, clean per-node input_payload, and concurrent wall-clock overlap across the three perspectives. Full src suite (586 tests) + supervisor suite (125 tests) pass with zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 8712aef4
Showing 9 changed files with 472 additions and 47 deletions
docs/todo/graph-engineering-open-items.md
docs/todo/verifier-fan-out-jury-design.md
src/org_level_loop.py
src/root_cause_trace.py
src/scheduler.py
src/task_execution.py
src/test_org_level_loop.py
src/test_root_cause_trace.py
src/test_scheduler.py
@@ -90,7 +90,7 @@
 ## F. 三视角评审团机制要不要推广到其他验证点
-**状态**：✅ 已确认，且明确了实现方式
+**状态**：✅ 已实现（2026-08-06）→ 需求文档 `docs/todo/verifier-fan-out-jury-requirements.md` + 设计文档 `docs/todo/verifier-fan-out-jury-design.md`
 **来源**：综述版对照 §7 开放问题 4 + 14 步版对照 §3 开放问题 4（同一主题）
@@ -98,7 +98,7 @@
 **确认结果（2026-08-05）**：要做，而且明确了实现方式——**不是**继续用现在 `root_cause_trace.judge_root_cause` 那种 Python 层直接顺序调用三个 `perspective_fn` 的写法，而是把三视角评审团**做成一张真正的 Work Graph fan-out/fan-in 菱形拓扑**：fan-out 出 N 个独立的验证节点（每个节点是一次独立的 `claude -p` 调用，带自己的视角提示词），fan-in 节点做投票/汇总。用户原话："三视角评审团可以做，做成 work graph 的 fan out 图来实现吧"。
-**待办**：✅ 需求文档已产出 → `docs/verifier-fan-out-jury-requirements.md`。明确标注**强依赖 G**——没有并行执行能力，把 `judge_root_cause` 改成 Work Graph fan-out 反而更慢（多了 Work Node 调度开销，没有并行收益），建议排在 G 之后。
+**待办**：✅ 已实现（2026-08-06）。汇总节点没有做成第四个 Work Node——投票是纯 `zero_token` 聚合（跟 E 里 `edge_type` 的区分直接对应），不需要 Agent 推理，做成普通 Python 函数在 `run_graph()` 返回后直接对 `outputs` 计数，不产生额外 `claude -p` 调用/token 成本，设计文档第 1 节记录了这个取舍。实现内容：`root_cause_trace.py` 新增 `_tally_perspectives()`（>=2/3 判定规则从 `judge_root_cause` 抽出来，闭包版和 Work Graph 版共用同一份规则，不是两份容易失配的实现）+ `judge_root_cause_via_work_graph()`（N 个视角节点用真实 schema 约束的结构化输出，不再像 `real_validator_perspective` 那样手工解析"CATEGORY:/DETAIL:"文本行）。闭包版 `judge_root_cause`/`real_validator_perspective` **保留**，不废弃（轻量备选，见设计文档第 4 节）。`org_level_loop.run_fix_loop`（唯一真实生产调用方）已迁移到 Work Graph 版本，fixer/verify 两次调用保持不变的 flat `trigger_org_node` 调用。**顺带修了一个真实的潜在 bug**：`judge_root_cause_via_work_graph` 是第一个真正触发 `run_graph()` 多节点并发层的生产代码路径（之前所有真实 `graph_manifest` 要么单节点要么严格串行链，G 的并发调度从未在生产里真正跑过），暴露出 `scheduler.py::run_goal_to_completion`/`task_execution.py::start_task_execution_in_background` 的 `schema.connect(db_path)` 都没传 `check_same_thread=False`——已一并修复。测试改造范围（设计文档第 5 节标注的最高风险点）：`test_org_level_loop.py`/`test_scheduler.py` 的 3 个测试把 mock 边界从 `org_node_trigger.trigger_org_node`（拦截"CATEGORY:"提示词）改成 `work_graph_execute.write_workflow_script`/`execute_work_graph`。新增测试证明验收标准 1-3（三次真实独立 Work Node 执行、判定规则与闭包版等价、每个视角节点的 `input_payload` 不含其他视角内容）+ 一个直接验证并发重叠的测试（`ConcurrentPerspectiveExecutionTest`）。`python3 -m unittest discover -s src`（586 tests）+ `discover -s supervisor`（125 tests）全绿，零回归。FR-4（`verify_fn` 通用对接其他验证点）本轮不实现，标注"应该"优先级，留待后续。
 ---
@@ -134,7 +134,7 @@
 |---|---|---|---|---|---|
 | **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 已满足） |
+| **F** 三视角评审团 fan-out | `judge_root_cause` 从 Python 函数改成 Work Graph 菱形拓扑 | ✅ **已实现**（2026-08-06，586 测试全绿） | M | 已完成 | 无（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 落地后）才会显现 |
+# 三视角评审团——Work Graph Fan-out 实现——详细设计（SDD）
+
+**日期**: 2026-08-06
+**性质**: 详细设计，对应需求文档 `docs/todo/verifier-fan-out-jury-requirements.md`（条目 F，已确认）。回答该需求文档第 8 节的开放问题。
+**依赖代码**：`src/root_cause_trace.py`（`judge_root_cause`/`_PERSPECTIVE_FRAMINGS`）、`src/org_level_loop.py`（`run_fix_loop`，唯一真实生产调用方）、`src/work_graph_orchestrate.py`（G 已实现的 `run_graph()`/`topological_layers()`）。
+**状态**: 设计草案，待评审后实施。
+
+## 1. 汇总节点不是 Work Node——这是"zero_token 边"原则的直接应用
+
+需求文档背景里"1 个汇总节点"的表述容易被误读成"汇总也要是一次 `claude -p` 调用"。实际不需要：`>=2/3` 投票本身是纯粹的计数聚合，不需要 Agent 推理——这正是 E 设计文档里强调的 `edge_type: zero_token` 场景本身。汇总逻辑保持为一个普通 Python 函数，在 `run_graph()` 返回之后，在调用方所在的进程里直接对 `outputs` 做计数，不产生额外的 `claude -p` 调用、不产生额外 token 成本（呼应需求文档第 7 节风险"token 成本上升"的顾虑，此处直接不引入这块新增成本）。
+
+验收标准 1 要求"能在 execution_event/日志里看到三次独立的节点执行记录"——这只要求**三个视角节点**是真 Work Node（`run_graph()`/`run_work_node()` 本来就会为每个节点写 `node_start`/`node_completed` 日志），不要求汇总也是。
+
+## 2. 判定规则复用，不重写（FR-3）
+
+`root_cause_trace.judge_root_cause` 现有的 `>=2/3` 计数逻辑被抽成一个共享私有函数 `_tally_perspectives(judgments)`，`judge_root_cause`（保留，闭包版本）和新的 `judge_root_cause_via_work_graph`（Work Graph 版本）都调用它——两套判定规则物理上是同一份代码，不是两份逻辑各自实现、容易失配。`judgments` 的 dict 形状（`{"category", "detail"}`）两个版本完全一致，`_tally_perspectives` 不需要知道调用方是闭包还是 Work Node 输出。
+
+## 3. `judge_root_cause_via_work_graph`：新的 Work Graph 版本
+
+```python
+def judge_root_cause_via_work_graph(
+    conn, task_id, target_folder_path, bug_description, context,
+    org_node="validator", skill_id="qa-task-validation", timeout=300,
+):
+```
+
+内部结构：
+
+1. `_PERSPECTIVE_FRAMINGS`（已存在，三个视角的 framing 文本）驱动构造每个视角的完整提示词（复用现有 framing 文本，不改视角内容本身）。
+2. 构造一个菱形（严格说是"N 节点单层，无边"）`graph_manifest`：`nodes = [{"node_id": "perspective-" + name, "depends_on": []} for name in perspectives]`，`edges = []`——三个节点互不依赖，`topological_layers()` 天然把它们分进同一层，G 自动并发调度，不需要额外的调度代码。
+3. 像 `e2e_pipeline.py::implement_code`/`validate_implementation_with_own_graph` 一样，手动 `INSERT INTO work_graph_run (..., status='running')`（不经过 `plan_work_graph()`——这个图的形状是固定的、代码已知的，不需要一次额外的 LLM 规划调用，同样是"机制用代码"原则的应用）。
+4. `run_graph()` 执行：每个视角节点用**真实 schema 约束的结构化输出**（`{"type": "object", "required": ["category", "detail"]}`），不再像现有 `real_validator_perspective` 那样手工解析"CATEGORY:/DETAIL:"文本行——`work_graph_orchestrate._make_run_fn` 走的是单节点 dynamic-workflow 脚本机制（`agent(prompt, {{ agentType, schema }})`），这条路径已经被 2026-08-01 的真实调用验证过 `structured_output` 真的能拿到（`work_graph_orchestrate.py` 模块 docstring 有记录），跟 `work_graph_plan.py` 那条"`--json-schema` 对 `--agent` 无效"的已知限制是两回事，不受它影响。
+5. `_tally_perspectives([outputs["perspective-" + name] for name in perspectives])`，返回值形状跟现有 `judge_root_cause` 完全一致（`{"root_cause_category", "root_cause_detail", "judgments"}`），调用方不需要区分。
+
+每个视角节点获得独立、干净的上下文（验收标准 3）：`_ensure_node_row`/`run_work_node` 本身就是每个 node_exec_id 一次全新的 workflow 脚本渲染 + 独立子进程调用，`input_payload` 只含 `{"node_id": ...}`，视角提示词通过 `build_prompt` 的闭包参数传入，节点之间不共享对话历史——这是 `run_graph()` 现有机制的天然属性，不需要额外设计。
+
+## 4. `org_level_loop.run_fix_loop` 迁移（FR-5，唯一真实生产场景）
+
+现状：
+
+```python
+perspective_fns = [root_cause_trace.real_validator_perspective(...) for name in (...)]
+judgment = root_cause_trace.judge_root_cause(perspective_fns)
+```
+
+改为：
+
+```python
+judgment = root_cause_trace.judge_root_cause_via_work_graph(
+    conn, task_id, target_folder_path, bug_description, context,
+)
+```
+
+`real_validator_perspective`/闭包版 `judge_root_cause` **不删除**——需求文档第 8 节开放问题已问过"要不要保留作为轻量备选"，答案是保留：`src/test_root_cause_trace.py` 现有单测（`JudgeRootCauseTest`/`RealValidatorPerspectiveParsingTest`）覆盖的是判定规则本身的正确性，这些测试不需要真实 `claude` 子进程调用就能跑（用纯 Python 闭包 mock），继续作为规则正确性的快速回归保障（`_tally_perspectives` 被两边共用，这些测试事实上也在验证 Work Graph 版本的判定逻辑）；闭包版本本身作为"不需要 worktree/Work Graph 开销的场景"的备选实现继续存在，不强制废弃。
+
+## 5. 测试改造范围（真实影响面，如实记录）
+
+`src/test_org_level_loop.py` 现有 3 个测试全部通过 mock `org_node_trigger.trigger_org_node`（拦截"CATEGORY:"提示词）来伪造三视角判断——这个 mock 边界在迁移后失效（新实现走 `work_graph_execute.execute_work_graph`，不再调用 `trigger_org_node` 做视角判断）。必须同步把这三个测试的 mock 边界改成 `work_graph_execute.write_workflow_script`/`execute_work_graph`（跟 `test_work_graph_orchestrate.py` 现有模式一致），`org_node_trigger.trigger_org_node` 的 mock 保留但只用于 fixer/verify 两次调用（迁移范围明确不包括这两处）。这是本次改动风险最高的一步——`run_fix_loop` 是真实生产闭环的一部分，测试改造必须逐条比对新旧断言等价，不能只是"能跑通"就算数。
+
+## 6. 不实现的部分（诚实记录，对应需求文档非目标 + FR-4 的"应该"优先级）
+
+- **FR-4（`verify_fn` 通用对接）不在本轮实现**：把评审团机制包装成一个可以传给 `run_graph(verify_fn=...)` 的工厂函数，用于升级 `_check_prd_consistency` 这类现有单一验证器场景——需求文档标注"应该"而非"必须"，且"哪些验证点该升级成评审团"本身是待定的判定标准（需求文档第 8 节开放问题未回答）。本轮只交付 FR-5（根因溯源场景本身）这一个"必须"级别的真实落地点，`judge_root_cause_via_work_graph` 的函数签名设计上保持通用（视角 framing/数量可参数化），为未来接 FR-4 留出空间，但不在本轮消费它。
+- **视角节点并发时间收益的实测**（验收标准 4）：G 已实现，`topological_layers()` 会自动把三个视角节点分进同一层走 `ThreadPoolExecutor`，机制上必然并发；但"在真实 `claude -p` 调用下测出时间重叠"需要 `RUN_REAL_AGENT_E2E=1` 真实调用，不在 mock 测试范围内验证，留给后续真实运行观察确认（跟 G 自己的 `RunGraphConcurrencyTest` 用 mock+`time.sleep` 验证并发的方式已经覆盖了"机制确实并发"这件事，不需要为 F 重新证明一遍 G 的能力）。
+
+## 7. 回答需求文档第 8 节开放问题
+
+- `judge_root_cause`（闭包版）保留作为轻量备选，不废弃（第 4 节）。
+- FR-4 接口设计：本轮不实现，见第 6 节。
+- 除根因溯源外还有哪些验证点该升级：本轮不判定，留待 FR-4 一起决定。
+- 视角数量 N 是否可配置：`judge_root_cause_via_work_graph`/`_tally_perspectives` 设计上已经是对任意数量视角通用的（不硬编码 3），根因溯源场景固定传 3 个，未来场景可以传不同数量而不需要改动这两个函数本身。
@@ -15,15 +15,18 @@ def run_fix_loop(conn, task_id, target_folder_path, bug_description, context, fi
     "verify_response"}. Raises root_cause_trace.InconclusiveEscalateError if
     the 3 perspectives don't reach 2/3 agreement (same as U-015 -- this
     function doesn't add its own fallback for that, matching 04-P's own
-    "升级人工" answer to disagreement)."""
-    perspective_fns = [
-        root_cause_trace.real_validator_perspective(
-            name, bug_description, context, target_folder_path,
-            trigger_fn=org_node_trigger.trigger_org_node, timeout=timeout,
-        )
-        for name in ("correctness", "requirement_completeness", "skill_defect")
-    ]
-    judgment = root_cause_trace.judge_root_cause(perspective_fns)
+    "升级人工" answer to disagreement).
+
+    **F (2026-08-06, docs/todo/verifier-fan-out-jury-design.md)**: root
+    cause judgment now runs as a real Work Graph fan-out (each perspective
+    is its own Work Node with its own clean context) instead of three
+    sequential Python closures calling org_node_trigger.trigger_org_node --
+    see root_cause_trace.judge_root_cause_via_work_graph. The fixer/verify
+    calls below are unchanged flat trigger_org_node calls; only root-cause
+    judgment moved."""
+    judgment = root_cause_trace.judge_root_cause_via_work_graph(
+        conn, task_id, target_folder_path, bug_description, context, timeout=timeout,
+    )
     category = judgment["root_cause_category"]
     route = _ROUTE_FOR_CATEGORY.get(category, "full_replan")
@@ -14,11 +14,27 @@ perspectives 04-P names (æ£ç¡®æ€§è§†è§’/éœ€æ±‚å®Œæ•´æ€§è§†è§’/Skill ç³»ç»Ÿæ€§ç¼
 concrete question each perspective asks is a prompt-engineering concern for
 the caller (see real_validator_perspective below for the real-call version),
 not this module's.
+
+**F (2026-08-06, docs/todo/verifier-fan-out-jury-design.md)**: judge_root_cause
+above is the closure-based implementation -- each perspective is a Python
+function call, sequentially, in the SAME process/context. Kept as-is (not
+replaced): it's the cheap path for scenarios that don't need worktree/Work
+Graph overhead, and its already-proven >=2/3 rule is now shared (via
+_tally_perspectives) with judge_root_cause_via_work_graph below, the real
+Work Graph fan-out version -- each perspective becomes an independent Work
+Node (own claude -p call, own clean context, own execution_event log
+entries), scheduled by G's topological_layers()/run_graph() same as any
+other graph_manifest. The fan-in vote itself stays plain Python (a
+zero_token aggregation per the design-dynamic-work-graph Skill's own
+edge_type distinction -- counting doesn't need an Agent call), not a
+fourth Work Node.
 """
+import json
+import uuid
 from collections import Counter
-from src import bug_record
+from src import bug_record, work_graph_orchestrate
 class InconclusiveEscalateError(Exception):
@@ -31,6 +47,19 @@ class InconclusiveEscalateError(Exception):
         super().__init__("perspectives did not reach 2/3 agreement: " + repr(judgments))
+def _tally_perspectives(judgments):
+    """>=2/3 agreement rule (04-P: "è‡³å°‘ 2/3 è§†è§’ä¸€è‡´æ‰å®šæ¡ˆ") -- shared by both
+    judge_root_cause (closures) and judge_root_cause_via_work_graph (Work
+    Graph fan-out), so the rule is one piece of code, not two that could
+    drift apart. `judgments`: list of {"category", "detail"}."""
+    counts = Counter(j["category"] for j in judgments)
+    winner, count = counts.most_common(1)[0]
+    if count * 2 < len(judgments):  # strictly "at least 2/3", not just plurality
+        raise InconclusiveEscalateError(judgments)
+    detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner)
+    return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments}
+
+
 def judge_root_cause(perspective_fns):
     """Runs all perspective_fns, requires >=2/3 to agree on the same
     category (04-P guard: "è‡³å°‘ 2/3 è§†è§’ä¸€è‡´æ‰å®šæ¡ˆ"). Returns
@@ -38,12 +67,7 @@ def judge_root_cause(perspective_fns):
     raises InconclusiveEscalateError otherwise.
     """
     judgments = [fn() for fn in perspective_fns]
-    counts = Counter(j["category"] for j in judgments)
-    winner, count = counts.most_common(1)[0]
-    if count * 2 < len(judgments):  # strictly "at least 2/3", not just plurality
-        raise InconclusiveEscalateError(judgments)
-    detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner)
-    return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments}
+    return _tally_perspectives(judgments)
 def trace_root_cause_test_failed(conn, task_id, description, perspective_fns, suggested_route, related_artifact_ids=None):
@@ -126,3 +150,86 @@ def real_validator_perspective(perspective_name, bug_description, context, targe
         return {"category": category, "detail": detail or response}
     return perspective_fn
+
+
+_PERSPECTIVE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
+_PERSPECTIVE_OUTPUT_SCHEMA = {"type": "object", "required": ["category", "detail"]}
+
+
+def _perspective_prompt(perspective_name, bug_description, context):
+    framing = _PERSPECTIVE_FRAMINGS[perspective_name]
+    return (
+        "{framing}\n\nBug: {bug_description}\n\nContext:\n{context}\n\n"
+        "Reply with your judgment: category must be exactly one of 'planning', 'coder', or "
+        "'skill_defect'; detail is one sentence citing the specific SDD section or code "
+        "location that supports it."
+    ).format(framing=framing, bug_description=bug_description, context=context)
+
+
+def _node_id_for_perspective(name):
+    return "perspective-" + name
+
+
+def _build_jury_graph_manifest(perspective_names):
+    """FR-1: a reusable fan-out graph_manifest shape -- N mutually
+    independent perspective nodes, generic over any set of perspective
+    names (not hardcoded to root-cause's three). No fan-in node: see this
+    module's own docstring / design doc Â§1 for why the vote stays plain
+    Python (zero_token aggregation, not worth an Agent call)."""
+    return {
+        "nodes": [{"node_id": _node_id_for_perspective(name), "depends_on": []} for name in perspective_names],
+        "edges": [],
+    }
+
+
+def _jury_prompt_builder(perspective_prompts):
+    """FR-2: perspective content is parameterized -- this scheduling code
+    has no idea what the perspectives actually ask, it just runs whatever
+    prompt each node was given."""
+    def build_prompt(node, outputs):
+        name = node["node_id"][len("perspective-"):]
+        return perspective_prompts[name]
+    return build_prompt
+
+
+def judge_root_cause_via_work_graph(
+    conn, task_id, target_folder_path, bug_description, context,
+    org_node="validator", skill_id="qa-task-validation", timeout=300,
+):
+    """Work Graph fan-out version of judge_root_cause (FR-5's real landing
+    scenario) -- each perspective is an independent Work Node (own
+    `claude -p` call with real schema-constrained structured output, own
+    clean context -- work_graph_orchestrate._make_run_fn's dynamic-
+    workflow-script mechanism, validated for real 2026-08-01, see that
+    module's docstring) instead of a Python closure called in a loop, so
+    unlike real_validator_perspective there's no 'CATEGORY:'/'DETAIL:'
+    text parsing needed -- the node's output_payload already IS
+    {"category", "detail"}. Same return shape as judge_root_cause, and
+    raises InconclusiveEscalateError on <2/3 agreement via the same shared
+    _tally_perspectives call -- callers don't need to know which
+    implementation they're getting.
+
+    No plan_work_graph() call: this graph's shape (N independent nodes) is
+    fixed and known ahead of time, same reasoning e2e_pipeline.py's
+    implement_code()/validate_implementation_with_own_graph() already use
+    for skipping an LLM planning step of their own."""
+    perspective_prompts = {
+        name: _perspective_prompt(name, bug_description, context) for name in _PERSPECTIVE_FRAMINGS
+    }
+    manifest = _build_jury_graph_manifest(list(perspective_prompts))
+
+    run_id = str(uuid.uuid4())
+    conn.execute(
+        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+        "VALUES (?, ?, ?, ?, 'running')",
+        (run_id, task_id, org_node, json.dumps(manifest)),
+    )
+    conn.commit()
+
+    outputs = work_graph_orchestrate.run_graph(
+        conn, run_id, org_node, skill_id, target_folder_path, manifest,
+        _jury_prompt_builder(perspective_prompts), _PERSPECTIVE_INPUT_SCHEMA, _PERSPECTIVE_OUTPUT_SCHEMA,
+        timeout=timeout,
+    )
+    judgments = [outputs[_node_id_for_perspective(name)] for name in perspective_prompts]
+    return _tally_perspectives(judgments)
@@ -193,7 +193,17 @@ def run_goal_to_completion(db_path, target_folder_path, goal_text):
     task always lands on a real terminal-ish status with a real error
     detail, never a caller-crashes-and-the-task-is-just-stuck outcome.
     """
-    conn = schema.connect(db_path)
+    # check_same_thread=False (2026-08-06, F): this conn can reach
+    # work_graph_orchestrate.run_graph() via finish_task_lifecycle ->
+    # org_level_loop.run_fix_loop -> root_cause_trace.
+    # judge_root_cause_via_work_graph, whose 3-perspective layer runs on
+    # its OWN ThreadPoolExecutor worker threads (G), not this function's
+    # thread -- a plain schema.connect() here would raise "SQLite objects
+    # created in a thread can only be used in that same thread" the first
+    # time a real multi-node concurrent layer actually runs (previously
+    # never exercised in production: every graph_manifest before F was
+    # either single-node or a strictly serial chain).
+    conn = schema.connect(db_path, check_same_thread=False)
     conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
     try:
         project_id = conn.execute(
@@ -157,7 +157,13 @@ def start_task_execution_in_background(db_path, task_id, target_folder_path, goa
     started one reaches the same real terminal state. Returns the Thread
     (already started) so callers/tests can `.join()` it if needed."""
     def run():
-        conn = schema.connect(db_path)
+        # check_same_thread=False (2026-08-06, F): same reasoning as
+        # scheduler.run_goal_to_completion's own conn -- this thread's conn
+        # can reach work_graph_orchestrate.run_graph()'s multi-node
+        # concurrent layers (G) via finish_task_lifecycle -> run_fix_loop ->
+        # judge_root_cause_via_work_graph, whose worker threads aren't this
+        # one.
+        conn = schema.connect(db_path, check_same_thread=False)
         try:
             result = run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text)
             scheduler.finish_task_lifecycle(conn, result, target_folder_path, goal_text)
@@ -3,23 +3,35 @@ fix -&gt; re-verify -&gt; done), mocked at the subprocess boundary (CLAUDE.md
 Testing constraint). Real coverage is in src/test_org_level_loop_e2e.py
 (RUN_REAL_AGENT_E2E=1).
+**F (2026-08-06)**: root-cause judgment now runs through
+root_cause_trace.judge_root_cause_via_work_graph (a real Work Graph fan-out,
+see docs/todo/verifier-fan-out-jury-design.md), so the 3 perspective calls
+are mocked at work_graph_execute.write_workflow_script/execute_work_graph
+(same boundary as test_work_graph_orchestrate.py), not at
+org_node_trigger.trigger_org_node -- that mock is now only used for the
+fixer/verify calls, which are unchanged.
+
 Run: python3 -m unittest src.test_org_level_loop -v
 """
 import unittest
 from unittest.mock import patch
-from src import org_level_loop, schema
+from src import org_level_loop, root_cause_trace, schema, work_graph_execute
+
+def _make_perspective_execute(perspective_category):
+    """Fakes work_graph_execute.execute_work_graph for the 3 perspective
+    Work Nodes -- all agree on perspective_category, real schema-shaped
+    output (no more 'CATEGORY:'/'DETAIL:' text parsing)."""
+    def fake_execute(cwd, name, args, output_schema, timeout=300):
+        return {"category": perspective_category, "detail": "because reasons"}
+    return fake_execute
-def _make_trigger(perspective_category, fix_ok=True):
-    """Fakes org_node_trigger.trigger_org_node for the whole loop: 3
-    perspective calls (all agreeing on perspective_category), one fixer
-    call, one verify call."""
+def _make_trigger(fix_ok=True):
+    """Fakes org_node_trigger.trigger_org_node for the fixer/verify calls only."""
     def fake(org_node, prompt, target_folder_path, timeout=300):
-        if "CATEGORY:" in prompt:
-            return "CATEGORY: {}\nDETAIL: because reasons".format(perspective_category)
         if org_node in ("coder", "planner"):
             return "Fixed it."
         if org_node == "validator":
@@ -31,7 +43,9 @@ def _make_trigger(perspective_category, fix_ok=True):
 class RunFixLoopTestBase(unittest.TestCase):
     def setUp(self):
-        self.conn = schema.connect(":memory:")
+        # check_same_thread=False: the 3 root-cause perspective nodes form
+        # one concurrent layer under G, run via ThreadPoolExecutor.
+        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) "
@@ -46,13 +60,21 @@ class RunFixLoopTestBase(unittest.TestCase):
     def tearDown(self):
         self.conn.close()
+    def _run_fix_loop(self, perspective_category, fix_ok=True):
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 side_effect=_make_perspective_execute(perspective_category),
+             ), \
+             patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=fix_ok)):
+            return org_level_loop.run_fix_loop(
+                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
+            )
+
 class CoderCategoryFastLaneTest(RunFixLoopTestBase):
     def test_coder_category_routes_fast_lane_and_reaches_done(self):
-        with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=True)):
-            result = org_level_loop.run_fix_loop(
-                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
-            )
+        result = self._run_fix_loop("coder", fix_ok=True)
         self.assertEqual(result["root_cause_category"], "coder")
         self.assertEqual(result["route"], "fast_lane")
         self.assertTrue(result["verified"])
@@ -69,10 +91,7 @@ class CoderCategoryFastLaneTest(RunFixLoopTestBase):
 class PlanningCategoryFullReplanTest(RunFixLoopTestBase):
     def test_planning_category_routes_full_replan(self):
-        with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("planning", fix_ok=True)):
-            result = org_level_loop.run_fix_loop(
-                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
-            )
+        result = self._run_fix_loop("planning", fix_ok=True)
         self.assertEqual(result["route"], "full_replan")
         self.assertEqual(result["final_status"], "done")
         assigned = self.conn.execute("SELECT assigned_org_node FROM task WHERE task_id = 'task-1'").fetchone()[0]
@@ -83,15 +102,36 @@ class PlanningCategoryFullReplanTest(RunFixLoopTestBase):
 class VerificationFailsTest(RunFixLoopTestBase):
     def test_fix_not_verified_leaves_task_out_of_done(self):
-        with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=False)):
-            result = org_level_loop.run_fix_loop(
-                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
-            )
+        result = self._run_fix_loop("coder", fix_ok=False)
         self.assertFalse(result["verified"])
         self.assertNotEqual(result["final_status"], "done")
         status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0]
         self.assertNotEqual(status, "done")
+class InconclusiveDisagreementTest(RunFixLoopTestBase):
+    """Real Work Graph fan-out still surfaces InconclusiveEscalateError the
+    same way the closure-based version did -- run_fix_loop adds no fallback
+    of its own (04-P: disagreement escalates to a human, it isn't silently
+    routed anywhere)."""
+
+    def test_three_way_split_raises_and_creates_no_bug_record(self):
+        categories = iter(["planning", "coder", "skill_defect"])
+
+        def fake_execute(cwd, name, args, output_schema, timeout=300):
+            return {"category": next(categories), "detail": "d"}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute), \
+             patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=True)):
+            with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
+                org_level_loop.run_fix_loop(
+                    self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it"
+                )
+
+        count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0]
+        self.assertEqual(count, 0)
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -6,9 +6,15 @@ coverage is in src/test_root_cause_trace_e2e.py (RUN_REAL_AGENT_E2E=1).
 Run: python3 -m unittest src.test_root_cause_trace -v
 """
+import json
+import os
+import tempfile
+import threading
+import time
 import unittest
+from unittest.mock import patch
-from src import bug_record, root_cause_trace, schema
+from src import bug_record, root_cause_trace, schema, work_graph_execute
 def _fixed(category, detail="detail"):
@@ -128,5 +134,180 @@ class RealValidatorPerspectiveParsingTest(unittest.TestCase):
             perspective_fn()
+# F (2026-08-06): Work Graph fan-out version -- each perspective becomes a
+# real Work Node (mocked at the claude subprocess boundary, same as
+# test_work_graph_orchestrate.py) instead of a Python closure.
+
+class JudgeRootCauseViaWorkGraphTestBase(unittest.TestCase):
+    def setUp(self):
+        self.home_tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.home_tmp.cleanup)
+        self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME")
+        os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name
+        self.addCleanup(self._restore_graphlemon_home)
+
+        # check_same_thread=False: the three perspective nodes form one
+        # concurrent layer under G, run via ThreadPoolExecutor.
+        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) "
+            "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', 'validating', 'validator')"
+        )
+        self.conn.commit()
+
+    def _restore_graphlemon_home(self):
+        if self.old_graphlemon_home is None:
+            os.environ.pop("GRAPHLEMON_HOME", None)
+        else:
+            os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home
+
+    def tearDown(self):
+        self.conn.close()
+
+    def _perspective_for_script(self, prompt_by_script, script_name):
+        content = prompt_by_script[script_name]
+        for name, framing in root_cause_trace._PERSPECTIVE_FRAMINGS.items():
+            if framing in content:
+                return name
+        raise AssertionError("could not identify perspective for script " + script_name)
+
+
+class RealWorkNodesPerPerspectiveTest(JudgeRootCauseViaWorkGraphTestBase):
+    """Acceptance criterion 1: three independent Work Node executions (real
+    work_node_execution rows, real execution_event logs), not three closure
+    calls inside one function."""
+
+    def test_three_perspectives_are_three_real_work_node_rows(self):
+        prompt_by_script = {}
+
+        def fake_write_script(cwd, script_name, script_content):
+            prompt_by_script[script_name] = script_content
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            perspective = self._perspective_for_script(prompt_by_script, name)
+            return {"category": "coder", "detail": "agreed via " + perspective}
+
+        with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            result = root_cause_trace.judge_root_cause_via_work_graph(
+                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
+            )
+
+        self.assertEqual(result["root_cause_category"], "coder")
+        self.assertEqual(len(result["judgments"]), 3)
+
+        rows = self.conn.execute(
+            "SELECT node_id, status, output_payload FROM work_node_execution WHERE run_id = ("
+            "  SELECT run_id FROM work_graph_run WHERE task_id = 'task-1'"
+            ")"
+        ).fetchall()
+        self.assertEqual(
+            {r[0] for r in rows},
+            {"perspective-correctness", "perspective-requirement_completeness", "perspective-skill_defect"},
+        )
+        for node_id, status, output_payload in rows:
+            self.assertEqual(status, "done")
+            self.assertEqual(json.loads(output_payload)["category"], "coder")
+
+
+class JudgmentRuleEquivalenceTest(JudgeRootCauseViaWorkGraphTestBase):
+    """Acceptance criterion 2: same >=2/3 rule, same results, as the
+    closure-based judge_root_cause on equivalent inputs -- both call the
+    same shared _tally_perspectives."""
+
+    def _run_with_categories(self, categories_by_perspective):
+        prompt_by_script = {}
+
+        def fake_write_script(cwd, script_name, script_content):
+            prompt_by_script[script_name] = script_content
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            perspective = self._perspective_for_script(prompt_by_script, name)
+            return {"category": categories_by_perspective[perspective], "detail": "d-" + perspective}
+
+        with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            return root_cause_trace.judge_root_cause_via_work_graph(
+                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
+            )
+
+    def test_two_of_three_agreement_wins(self):
+        result = self._run_with_categories(
+            {"correctness": "planning", "requirement_completeness": "planning", "skill_defect": "coder"}
+        )
+        self.assertEqual(result["root_cause_category"], "planning")
+
+        closure_result = root_cause_trace.judge_root_cause(
+            [_fixed("planning"), _fixed("planning"), _fixed("coder")]
+        )
+        self.assertEqual(result["root_cause_category"], closure_result["root_cause_category"])
+
+    def test_three_way_split_is_inconclusive(self):
+        with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
+            self._run_with_categories(
+                {"correctness": "planning", "requirement_completeness": "coder", "skill_defect": "skill_defect"}
+            )
+        with self.assertRaises(root_cause_trace.InconclusiveEscalateError):
+            root_cause_trace.judge_root_cause([_fixed("planning"), _fixed("coder"), _fixed("skill_defect")])
+
+
+class IndependentCleanContextTest(JudgeRootCauseViaWorkGraphTestBase):
+    """Acceptance criterion 3: each perspective node's input_payload only
+    carries what it needs to judge -- not other perspectives' reasoning."""
+
+    def test_input_payload_has_no_cross_perspective_content(self):
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 return_value={"category": "coder", "detail": "d"},
+             ):
+            root_cause_trace.judge_root_cause_via_work_graph(
+                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
+            )
+
+        rows = self.conn.execute("SELECT node_id, input_payload FROM work_node_execution").fetchall()
+        self.assertEqual(len(rows), 3)
+        for node_id, input_payload in rows:
+            payload = json.loads(input_payload)
+            self.assertEqual(payload, {"node_id": node_id})
+
+
+class ConcurrentPerspectiveExecutionTest(JudgeRootCauseViaWorkGraphTestBase):
+    """G is already in place -- three mutually-independent perspective
+    nodes land in one topological layer, so they run concurrently, not
+    queued one after another."""
+
+    def test_three_perspective_calls_overlap_in_wall_clock_time(self):
+        events = []
+        events_lock = threading.Lock()
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            start = time.monotonic()
+            time.sleep(0.05)
+            end = time.monotonic()
+            with events_lock:
+                events.append((start, end))
+            return {"category": "coder", "detail": "d"}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            root_cause_trace.judge_root_cause_via_work_graph(
+                self.conn, "task-1", "/tmp/proj-1", "bug desc", "context",
+            )
+
+        self.assertEqual(len(events), 3)
+        # at least two of the three calls must have overlapped
+        overlaps = sum(
+            1 for i in range(len(events)) for j in range(i + 1, len(events))
+            if events[i][0] < events[j][1] and events[j][0] < events[i][1]
+        )
+        self.assertGreater(overlaps, 0)
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -377,10 +377,10 @@ class UnroutedExceptionStillBlocksTaskTest(RoutableExceptionTestBase):
 def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300):
     """org_node_trigger.trigger_org_node fake for org_level_loop.run_fix_loop's
-    3 perspective calls (all agree 'coder') + fixer + verify -- verify always
-    fails, so route_org_loop keeps incrementing loop_round on retries."""
-    if "CATEGORY:" in prompt:
-        return "CATEGORY: coder\nDETAIL: because reasons"
+    fixer + verify calls -- verify always fails, so route_org_loop keeps
+    incrementing loop_round on retries. Root-cause judgment itself (F,
+    2026-08-06) no longer goes through trigger_org_node -- see
+    _fix_loop_perspective_execute below for that boundary."""
     if org_node == "coder":
         return "Fixed it."
     if org_node == "validator":
@@ -388,6 +388,12 @@ def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300):
     raise AssertionError("unexpected call: " + org_node)
+def _fix_loop_perspective_execute(cwd, name, args, output_schema, timeout=300):
+    """work_graph_execute.execute_work_graph fake for run_fix_loop's 3
+    root-cause perspective Work Nodes (F) -- all agree 'coder'."""
+    return {"category": "coder", "detail": "because reasons"}
+
+
 class SchedulerContinuesPastBlockedTaskTest(unittest.TestCase):
     """REQ-05 failure / REQ-09 boundary, against the REAL scheduler (U-032) --
     test_org_loop_route.py's own test for this literally says "no scheduler
@@ -430,7 +436,9 @@ class SchedulerContinuesPastBlockedTaskTest(unittest.TestCase):
         goals = [("/tmp/proj-1", "goal-ok-1"), ("/tmp/proj-1", "goal-doomed"), ("/tmp/proj-1", "goal-ok-2")]
         with patch.object(scheduler.e2e_pipeline, "run_task_lifecycle", side_effect=fake_run_task_lifecycle), \
              patch.object(scheduler.e2e_pipeline, "_read_artifact_content", return_value="benchmark content"), \
-             patch("src.org_node_trigger.trigger_org_node", side_effect=_fix_loop_trigger):
+             patch("src.org_node_trigger.trigger_org_node", side_effect=_fix_loop_trigger), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=_fix_loop_perspective_execute):
             results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3)
         by_goal = {r["goal_text"]: r for r in results}