Org Graph unification Phase 2c: Validator self-plans its verification graph

Validator's terminal verdict is now a real self-planned node (node_type= 'final_verdict', judgment.kind='mechanical', on_fail.action= 'dynamic_loop_back') appended into the SAME shared work_graph_run Planner and Coder already contribute to (plan_work_graph(..., append_to_run_id= run_id)), instead of a flat post-graph org_node_trigger call against a separate hardcoded Validator manifest. This is what lets a failing verdict genuinely trigger Reflection-review + Fix-Skill + a precise loop back to whichever real ancestor node is at fault, reusing the existing dynamic_loop_back machinery instead of a parallel retry loop. The 2026-08-06 hard gate (a real non-zero exit from Validator's own test script force-overrides the verdict to FAIL, Python-enforced not just prompted) is preserved via a build_prompt wrapper + mechanical check closure sharing mutable state. INCONCLUSIVE is now treated the same as FAIL (triggers a bounded retry via dynamic_loop_back) rather than accepted as a terminal result. _VALIDATOR_MANIFEST/_VALIDATOR_DOC_PROMPTS/_VALIDATOR_VERDICT_PROMPT/ _validator_prompt_builder are kept unused (same precedent as _CODER_PROMPT). org_node column removal (§2.1), index narrowing (§2.2), and the new work-graph-node-review skill (§7.1) remain for a later sub-phase. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Org Graph unification Phase 2c: Validator self-plans its verification graph
Validator's terminal verdict is now a real self-planned node (node_type= 'final_verdict', judgment.kind='mechanical', on_fail.action= 'dynamic_loop_back') appended into the SAME shared work_graph_run Planner and Coder already contribute to (plan_work_graph(..., append_to_run_id= run_id)), instead of a flat post-graph org_node_trigger call against a separate hardcoded Validator manifest. This is what lets a failing verdict genuinely trigger Reflection-review + Fix-Skill + a precise loop back to whichever real ancestor node is at fault, reusing the existing dynamic_loop_back machinery instead of a parallel retry loop. The 2026-08-06 hard gate (a real non-zero exit from Validator's own test script force-overrides the verdict to FAIL, Python-enforced not just prompted) is preserved via a build_prompt wrapper + mechanical check closure sharing mutable state. INCONCLUSIVE is now treated the same as FAIL (triggers a bounded retry via dynamic_loop_back) rather than accepted as a terminal result. _VALIDATOR_MANIFEST/_VALIDATOR_DOC_PROMPTS/_VALIDATOR_VERDICT_PROMPT/ _validator_prompt_builder are kept unused (same precedent as _CODER_PROMPT). org_node column removal (§2.1), index narrowing (§2.2), and the new work-graph-node-review skill (§7.1) remain for a later sub-phase. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 89482e7c
Showing 6 changed files with 663 additions and 359 deletions
docs/todo/org-graph-unification-phase2-universal-self-planning-design.md
src/e2e_pipeline.py
src/task_execution.py
src/test_e2e_pipeline.py
src/test_task_execution.py
src/work_graph_plan.py
@@ -3,7 +3,7 @@
 **日期**: 2026-08-08
 **性质**: 详细设计，对应 `docs/todo/org-graph-unification-and-universal-self-planning-requirements.md` 的 Phase 2（FR-2/FR-3/FR-4），按 §4/§4.1 已拍板的合图（方案 A）路径设计。
 **依赖**: 硬依赖 Phase 1（`docs/done/org-graph-unification-phase1-review-gate-design.md`）——已实现（commit `ea19c7d`，2026-08-08）。§9 接口核对清单已于 2026-08-12 逐条核对完成：三项全部吻合（`plan_and_review_documents()` 经 `_review_planner_output_with_retries()` 透传 `plan_documents()` 的 `{"run_id", "artifact_ids"}` 返回形状，含 `run_id`；`task.status` 新值确认是 `reviewing_planner_output`；`roles/planner_reviewer/role.json` 目前 `skills: ["planner-output-review", "unattended-agent-operations"]`，未预留 §7.1 新 skill 位置，Phase 2 实施时新增一行即可，不是不兼容）。
-**状态**: Phase 2a（commit `86fc0bb`）+ Phase 2b（commit `7c46a2f`）已实现，2026-08-12。Phase 2c（Validator 接入自规划 + `org_node` 列删除 + 精确 loop_back 收口）待实施，详见 §10。**2026-08-08 修订**：写作过程中，需求文档 §7 的四个开放问题已被用户直接拍板（详见该文档 §7 对应条目），其中两条推翻了本文档最初的设计判断——Coder/Validator 自规划改为"建议节点清单"而非完全自由（§3），`work_graph_run.org_node` 改为彻底移除而非保留（§2.1）——本版本已按拍板结果改写，不是原始草稿。**2026-08-08 追加**（同日，见需求文档新增 §3.1）：FR-4 的目标被明确为"失败即进化"完整方法论（判断失败 → Reflection-review 动态诊断 → 系统性缺陷则先 Fix-Skill → 跳回执行），当时 §5 设计的 `dynamic_loop_back` 只覆盖第 1/2/4 步，第 3 步（Fix-Skill）是已知缺口。**2026-08-12 补齐**：§5 已扩写为完整四步设计（§5.4 Reflection-review 现在同时判断目标节点和缺陷性质，§5.5 新增 Fix-Skill，真实调用既有但从未真正投入使用的 `skill_update.py::update_skill()`），过程中发现两个既有事实（`skills:` frontmatter 只按名字动态加载内容、不需要重新部署；`org_level_loop.py` 的 `_ROUTE_FOR_CATEGORY` 从来没有 `skill_defect` 这个 key，`update_skill()` 在这个代码库里投产以来从未被真正调用过）——详见 §5 开头。由哪个身份执笔改 Skill 已通过 `AskUserQuestion` 确认（2026-08-12）：统一用 Validator，见 §5.5。
+**状态**: Phase 2a（commit `86fc0bb`）+ Phase 2b（commit `7c46a2f`）已实现，2026-08-12。Phase 2c 的 Validator 自规划改造（§6 核心部分）已实现，2026-08-12：`validate_implementation_with_own_graph()` 改为 `plan_work_graph(conn, "validator", ..., append_to_run_id=run_id)` 追加进 Planner/Coder 共享的同一个 `work_graph_run`，终审 verdict 变成图内一个真实节点（`node_type='final_verdict'`，`judgment.kind='mechanical'`、`check_ref='verdict_says_pass'`，`on_fail.action='dynamic_loop_back'`——Validator 图里唯一允许用这个 action 的节点），硬门槛（脚本真实非零退出码强制 FAIL，不听 LLM 辩解）通过 `_verdict_node_build_prompt`/`_verdict_says_pass_check` 的共享可变状态闭包保留，Python 强制而非只是 prompt 里说说。`_VALIDATOR_MANIFEST`/`_VALIDATOR_DOC_PROMPTS`/`_VALIDATOR_VERDICT_PROMPT`/`_validator_prompt_builder` 保留但不再使用（同 `_CODER_PROMPT` 先例）。`python3 -m unittest discover -s src`（756 tests）+ `discover -s supervisor`（126 tests）全绿。**尚未实施**：§2.1（`org_node` 列删除）、§2.2（`ux_work_graph_run_active` 收紧到 `UNIQUE(task_id)`）、§7.1（新 skill `work-graph-node-review` + 三个 role.json 追加）、§7.2 的收口验证（见 §10 Phase 2c 小节末尾新增的分析：一旦 verdict 节点自己的 `dynamic_loop_back` 内部消化重试，`validate_implementation_with_own_graph()` 正常返回时 verdict 必为 PASS，`org_level_loop.py` 的旧粗粒度路由对这条新路径自然不可达，不需要显式新旧流程判断——决定保留旧分支代码不删，加注释说明，不做额外改动）。**2026-08-08 修订**：写作过程中，需求文档 §7 的四个开放问题已被用户直接拍板（详见该文档 §7 对应条目），其中两条推翻了本文档最初的设计判断——Coder/Validator 自规划改为"建议节点清单"而非完全自由（§3），`work_graph_run.org_node` 改为彻底移除而非保留（§2.1）——本版本已按拍板结果改写，不是原始草稿。**2026-08-08 追加**（同日，见需求文档新增 §3.1）：FR-4 的目标被明确为"失败即进化"完整方法论（判断失败 → Reflection-review 动态诊断 → 系统性缺陷则先 Fix-Skill → 跳回执行），当时 §5 设计的 `dynamic_loop_back` 只覆盖第 1/2/4 步，第 3 步（Fix-Skill）是已知缺口。**2026-08-12 补齐**：§5 已扩写为完整四步设计（§5.4 Reflection-review 现在同时判断目标节点和缺陷性质，§5.5 新增 Fix-Skill，真实调用既有但从未真正投入使用的 `skill_update.py::update_skill()`），过程中发现两个既有事实（`skills:` frontmatter 只按名字动态加载内容、不需要重新部署；`org_level_loop.py` 的 `_ROUTE_FOR_CATEGORY` 从来没有 `skill_defect` 这个 key，`update_skill()` 在这个代码库里投产以来从未被真正调用过）——详见 §5 开头。由哪个身份执笔改 Skill 已通过 `AskUserQuestion` 确认（2026-08-12）：统一用 Validator，见 §5.5。
 ---
@@ -200,6 +200,15 @@ _VALIDATOR_PROMPT = (
     "Goal: {goal_text}\n\nBenchmark:\n\n{benchmark_content}"
 )
+# 2026-08-12 (Phase 2c): retired now that Validator self-plans its own
+# verification graph (validate_implementation_with_own_graph() below) via
+# plan_work_graph(conn, "validator", ..., append_to_run_id=run_id) instead
+# of running this hardcoded fixed chain -- the "held-out eval set, not the
+# Coder's own dev set" framing that motivated this dict lives on in
+# _VALIDATOR_PLANNING_CONTEXT's guidance text instead. Kept here, unused,
+# for the same "a future revert would want it back verbatim" reason
+# _CODER_PROMPT above is.
+#
 # 2026-08-06: Validator gets its OWN benchmark/test_dataset/test_script,
 # generated independently from SDD/PRD -- NOT reusing the Planner's/Coder's
 # own copies. If the Coder could game the Planner's test_script (e.g.
@@ -240,10 +249,12 @@ _VALIDATOR_DOC_PROMPTS = {
     ),
 }
-# validator_test_script node temporarily disabled (2026-08-07, same
-# stopgap as work_graph_plan._PROMPT_TEMPLATE -- see its comment for the
-# full rationale/upgrade path): 6/6 real repro rate for narrative-instead-
-# of-code output, docs/todo/validator-test-script-narrative-output-issue.md.
+# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_DOC_PROMPTS above --
+# see that comment. validator_test_script node temporarily disabled
+# (2026-08-07, same stopgap as work_graph_plan._PROMPT_TEMPLATE -- see its
+# comment for the full rationale/upgrade path): 6/6 real repro rate for
+# narrative-instead-of-code output, docs/todo/validator-test-script-
+# narrative-output-issue.md.
 _VALIDATOR_MANIFEST = {
     "nodes": [
         {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []},
@@ -255,6 +266,8 @@ _VALIDATOR_MANIFEST = {
     "edges": [],
 }
+# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above -- see
+# that comment.
 _VALIDATOR_VERDICT_PROMPT = (
     "Give your final verdict for this goal, informed by your own independent Benchmark below "
     "and the REAL execution result of your own test script (if you wrote one -- 'N/A' means you "
@@ -303,7 +316,12 @@ def _planner_prompt_builder(goal_text, node_artifact_types):
 def _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types):
-    """Same shape as _planner_prompt_builder, but the FIRST node
+    """2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above --
+    see that comment; validate_implementation_with_own_graph() now uses
+    dispatch_build_prompt(_planner_prompt_builder(...), generic_node_prompt_builder)
+    like every other self-planning org_node.
+
+    Same shape as _planner_prompt_builder, but the FIRST node
     (validator_benchmark) needs sdd_content/prd_content from OUTSIDE
     validator's own tiny graph -- those aren't a dependency within this
     graph, they come from the Planner's already-completed one. Passed
@@ -644,12 +662,16 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
     now carries its own `judgment` field (work_graph_plan.py's
     `_PLANNER_NODE_CATALOG` suggests one per node type, the Planner can
     reuse or override it), which `run_graph()` resolves itself per node.
-    `_combined_planner_verify_fn`/`_prd_consistency_verify_fn`/
-    `_test_script_verify_fn` are NOT deleted -- `validate_implementation_
-    with_own_graph()` (Validator's still-hardcoded manifest, unaffected by
-    this round) continues to use them (FR-A8). `mechanical_check_registry`
-    is still passed unconditionally: a self-planned node declaring
-    `judgment.kind == "mechanical"` (e.g. reusing the catalog's suggested
+    `_combined_planner_verify_fn`/`_prd_consistency_verify_fn` are NOT
+    deleted -- this function still uses them below (FR-A8).
+    `_test_script_verify_fn` is ALSO still used below for Planner's own
+    `test_script` node (2026-08-12: Validator's own equivalent node no
+    longer needs it -- validate_implementation_with_own_graph() now
+    self-plans too, and its `final_verdict` node's hard gate is enforced
+    via a mechanical_check_registry closure instead, see that function's
+    own docstring). `mechanical_check_registry` is still passed
+    unconditionally: a self-planned node declaring `judgment.kind ==
+    "mechanical"` (e.g. reusing the catalog's suggested
     `check_script_compiles` for a test_script node) needs it registered,
     same as before.
     """
@@ -1089,105 +1111,214 @@ def _format_script_execution_result(result):
     ).format(result["exit_code"], result["stdout"][-4000:], result["stderr"][-4000:])
+def _verdict_says_pass_check(shared_state):
+    """Phase 2c (2026-08-12): mechanical judgment for the final_verdict
+    node -- `shared_state` is a plain dict the matching build_prompt
+    wrapper (_verdict_node_build_prompt) populates with the REAL script
+    execution result as a side effect, read back here since a mechanical
+    check callable only ever receives `content`, no side channel of its
+    own. Two ways to fail:
+    1. Hard gate (user-confirmed 2026-08-06, preserved verbatim): a real
+       non-zero exit from Validator's own test script is not up for LLM
+       debate -- force-reject regardless of what the verdict text says.
+    2. The verdict itself isn't PASS -- INCONCLUSIVE is treated the SAME
+       as FAIL here (ok=False), not specially accepted: this node's job
+       is to keep trying (bounded by the loop_back attempt budget, same
+       as everything else on_fail.action='dynamic_loop_back' governs) to
+       actually resolve the goal, not to settle for genuine uncertainty
+       on the first attempt. If the budget is exhausted without reaching
+       PASS, run_graph() raises ConsistencyCheckFailedError same as any
+       other exhausted loop -- this function then never returns a non-PASS
+       verdict normally, it either returns PASS or raises (see this
+       function's own docstring for why that's the right contract)."""
+    def check(content):
+        script_execution = shared_state.get("script_execution")
+        if script_execution and script_execution["ran"] and script_execution["exit_code"] not in (0, None):
+            return False, "real test script exited non-zero (exit_code={}) -- hard gate, not up for debate".format(
+                script_execution["exit_code"]
+            )
+        verdict = _parse_verdict_line(content)
+        if verdict == "PASS":
+            return True, None
+        return False, "verdict was {}, not PASS".format(verdict)
+    return check
+
+
+def _verdict_node_build_prompt(inner_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout):
+    """Wraps a build_prompt(node, outputs) callable: for the ONE node
+    declaring node_type='final_verdict', performs REAL test script
+    execution (if a node_type='test_script' dependency produced real
+    content) as a side effect before returning the prompt -- embedding
+    the real result into both the prompt text (so the model sees it) and
+    `shared_state` (so _verdict_says_pass_check's hard gate can enforce it
+    in Python, not just ask nicely). Every other node passes through to
+    `inner_build_prompt` completely unchanged."""
+    def build_prompt(node, outputs):
+        if node.get("node_type") != "final_verdict":
+            return inner_build_prompt(node, outputs)
+        script_node_id = next(
+            (nid for nid, node_type in node_types.items() if node_type == "test_script" and nid in outputs), None
+        )
+        if script_node_id:
+            script_execution = run_test_script_for_real(
+                outputs[script_node_id].get("content", ""), target_folder_path, task_id,
+                "validator_final_verdict_script", timeout=timeout,
+            )
+        else:
+            script_execution = {
+                "ran": False, "exit_code": None, "stdout": "", "stderr": "",
+                "note": "N/A: this self-planned graph has no test_script-typed node",
+            }
+        shared_state["script_execution"] = script_execution
+        return inner_build_prompt(node, outputs) + (
+            "\n\nThe REAL execution result of your own test script (if any):\n"
+            + _format_script_execution_result(script_execution)
+        )
+    return build_prompt
+
+
+_VALIDATOR_PLANNING_CONTEXT = (
+    "{goal_text}\n\n"
+    "Approved SDD:\n\n{sdd_content}\n\n"
+    "PRD:\n\n{prd_content}\n\n"
+    "Guidance: you are independently verifying this goal's implementation. Form your OWN "
+    "acceptance criteria informed by (but not copied from) the SDD/PRD above -- don't assume "
+    "what the Coder actually did, you have no visibility into that beyond what you inspect "
+    "yourself. A held-out test script (if you plan one) will be executed for real against the "
+    "actual current repository state -- write real commands, not a description of what a script "
+    "would do. Your final verdict must be its own node (see the final_verdict entry in the "
+    "suggested node types), depending on whatever earlier nodes it genuinely needs."
+)
+
+
 def validate_implementation_with_own_graph(
-    conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200,
+    conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200,
 ):
-    """2026-08-06: Validator's own Work Graph (validator_benchmark ->
-    validator_test_dataset -> validator_test_script), independently
-    generated from SDD/PRD -- NOT reusing the Planner's/Coder's own
-    benchmark/dataset/script (see _VALIDATOR_DOC_PROMPTS's module comment
-    for why: a held-out eval set, not a re-run of the dev set the Coder
-    already developed against). The resulting test_script, if real (not
-    N/A), is then actually EXECUTED against the current repository state --
-    a non-zero real exit code force-overrides the verdict to FAIL in code,
-    not just by asking the LLM nicely (user-confirmed 2026-08-06: script is
-    the hard gate, subjective judgment can only add INCONCLUSIVE/risk notes
-    on top of a passing script, never override a failing one to PASS).
-
-    Returns {"verdict", "raw_response", "artifact_ids", "script_execution"}.
+    """Phase 2c (2026-08-12): Validator self-plans its own verification
+    graph via plan_work_graph(conn, "validator", ..., append_to_run_id=
+    run_id) -- appended onto the SAME shared run Planner/Coder already
+    contributed to, replacing the old hardcoded _VALIDATOR_MANIFEST/
+    _VALIDATOR_DOC_PROMPTS/_validator_prompt_builder fixed-chain approach
+    (kept in the module, unused, for the same "a future revert would want
+    it back verbatim" reason implement_code()'s retired _CODER_PROMPT is).
+
+    The final verdict is now a REAL node inside this graph (node_type=
+    'final_verdict', judgment.kind='mechanical', check_ref=
+    'verdict_says_pass', on_fail.action='dynamic_loop_back') instead of a
+    flat call after the graph finishes -- this is what lets a FAIL
+    genuinely trigger Reflection-review + [Fix-Skill] + a precise loop
+    back to whichever real ancestor node (possibly Planner's SDD, possibly
+    one of Coder's own implementation nodes, anywhere in the shared run)
+    is actually at fault, reusing work_graph_orchestrate.py's existing
+    dynamic_loop_back machinery (Phase 2a) rather than reimplementing a
+    parallel Python-level retry loop. The real test-script hard gate
+    (2026-08-06, user-confirmed: script is the hard gate, subjective
+    judgment can only add INCONCLUSIVE/risk notes on top of a passing
+    script, never override a failing one to PASS) is preserved via
+    _verdict_node_build_prompt/_verdict_says_pass_check's shared mutable
+    state -- still genuinely Python-enforced, not just prompted.
+
+    **Contract change from the old flat-call version**: because the
+    final_verdict node's own mechanical judgment treats anything other
+    than PASS as ok=False (INCONCLUSIVE included, see _verdict_says_pass_
+    check's own docstring for why) and retries via dynamic_loop_back up to
+    its attempt budget, this function can now only ever RETURN normally
+    with verdict='PASS' -- a goal that can't be resolved within the loop
+    budget raises ConsistencyCheckFailedError instead (same as any other
+    exhausted work_graph_orchestrate.py loop) rather than returning a
+    'FAIL'/'INCONCLUSIVE' verdict. This is what naturally retires org_
+    level_loop.py's involvement for tasks using this path: scheduler.py::
+    finish_task_lifecycle's "verdict != PASS -> run_fix_loop()" branch
+    becomes unreachable here, not because of an explicit new-flow-vs-old-
+    flow check, but because the condition it's gated on can no longer
+    occur on this path (resolves design doc §8 point 4 without adding one).
+    A caller that genuinely needs the old three-way PASS/FAIL/INCONCLUSIVE
+    contract (e.g. this function failing for a reason unrelated to the
+    goal itself) still gets that distinction via scheduler.py's existing
+    exception-based blocked_reason routing, same mechanism prd_consistency_
+    check_failed/test_script_verification_failed already use.
+
+    Returns {"verdict": "PASS", "raw_response", "artifact_ids", "script_execution"}.
     """
-    run_id = str(uuid.uuid4())
-    conn.execute(
-        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-        "VALUES (?, ?, 'validator', ?, 'running')",
-        (run_id, task_id, json.dumps(_VALIDATOR_MANIFEST)),
+    planning_context = _VALIDATOR_PLANNING_CONTEXT.format(
+        goal_text=goal_text, sdd_content=sdd_content, prd_content=prd_content,
+    )
+    plan_result = work_graph_plan.plan_work_graph(
+        conn, "validator", task_id, planning_context, target_folder_path,
+        timeout=300, append_to_run_id=run_id,
+    )
+    graph_manifest = plan_result["graph_manifest"]
+    new_node_ids = set(plan_result["new_node_ids"])
+    node_types = {n["node_id"]: n.get("node_type") for n in graph_manifest["nodes"]}
+
+    verdict_node_ids = [nid for nid, nt in node_types.items() if nt == "final_verdict" and nid in new_node_ids]
+    if len(verdict_node_ids) != 1:
+        raise work_graph_plan.PlanningFailedError(
+            "validator's self-planned graph must have exactly one node_type='final_verdict' node, "
+            "found {}: {}".format(len(verdict_node_ids), verdict_node_ids)
+        )
+    verdict_node_id = verdict_node_ids[0]
+
+    node_artifact_types = {}
+    for n in graph_manifest["nodes"]:
+        if "artifact_type" in n:
+            node_artifact_types[n["node_id"]] = n["artifact_type"]
+        elif n.get("promotes_to_artifact"):
+            node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"]
+
+    planner_builder = _planner_prompt_builder(goal_text, node_artifact_types)
+    base_build_prompt = work_graph_plan.dispatch_build_prompt(planner_builder, work_graph_plan.generic_node_prompt_builder)
+    shared_state = {}
+    build_prompt = _verdict_node_build_prompt(
+        base_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout,
     )
-    conn.commit()
-    node_artifact_types = {n["node_id"]: n["artifact_type"] for n in _VALIDATOR_MANIFEST["nodes"]}
-    # 2026-08-07: same run_graph()-partial-failure salvage as
-    # plan_documents() -- see _write_completed_node_artifacts's own
-    # docstring and docs/todo/work-graph-partial-artifact-persistence-
-    # requirements.md. `outputs` is only used below on the SUCCESS
-    # continuation (script execution + verdict prompt); the failure path
-    # re-raises before ever touching it.
     run_graph_error = None
     outputs = None
     try:
         outputs = work_graph_orchestrate.run_graph(
-            conn, run_id, "validator", "validator-independent-verification", target_folder_path, _VALIDATOR_MANIFEST,
-            _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types),
-            _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
-            verify_fn=_test_script_verify_fn(node_artifact_types, script_artifact_type="validator_test_script"),
-            timeout=timeout,
+            conn, run_id, "validator", "validator-independent-verification", target_folder_path, graph_manifest,
+            build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout,
+            mechanical_check_registry={
+                "check_script_compiles": check_script_compiles,
+                "verdict_says_pass": _verdict_says_pass_check(shared_state),
+            },
         )
     except Exception as e:
         run_graph_error = e
     artifact_ids = _write_completed_node_artifacts(
-        conn, target_folder_path, task_id, run_id, _VALIDATOR_MANIFEST, node_artifact_types,
-        lambda artifact_type: "txt" if artifact_type != "validator_benchmark" else "md", run_graph_error,
+        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
+        _planner_ext_for, run_graph_error, only_node_ids=new_node_ids,
     )
     if run_graph_error is not None:
         raise run_graph_error
-    if "validator_test_script" in outputs:
-        script_content = outputs["validator_test_script"]["content"]
-        script_execution = run_test_script_for_real(
-            script_content, target_folder_path, task_id, "validator_test_script", timeout=timeout,
-        )
-    else:
-        script_execution = {
-            "ran": False, "exit_code": None, "stdout": "", "stderr": "",
-            "note": "N/A: validator_test_script node is currently disabled (see _VALIDATOR_MANIFEST)",
-        }
+    # Reaching here means the final_verdict node's own mechanical judgment
+    # passed -- see this function's own docstring for why that means
+    # verdict is unconditionally 'PASS' at this point, not something that
+    # needs re-parsing/re-checking.
+    response = outputs[verdict_node_id]["content"]
+    script_execution = shared_state.get("script_execution") or {
+        "ran": False, "exit_code": None, "stdout": "", "stderr": "",
+        "note": "N/A: no test_script-typed node in this self-planned graph",
+    }
-    response = org_node_trigger.trigger_org_node(
-        "validator",
-        _VALIDATOR_VERDICT_PROMPT.format(
-            goal_text=goal_text,
-            validator_benchmark_content=outputs["validator_benchmark"]["content"],
-            script_execution_result=_format_script_execution_result(script_execution),
-        ),
-        target_folder_path, timeout=timeout,
-    )
-    verdict = _parse_verdict_line(response)
-
-    # Hard gate (user-confirmed 2026-08-06): a real non-zero exit from
-    # validator's OWN test script is not up for LLM debate -- force FAIL
-    # regardless of what the verdict line said. A passing/not-run script
-    # leaves the LLM's own judgment (including INCONCLUSIVE) untouched.
-    if script_execution["ran"] and script_execution["exit_code"] not in (0, None):
-        verdict = "FAIL"
-
-    # 2026-08-07 (real incident): this function INSERTs its work_graph_run
-    # row as status='running' at the top and, on the happy path (verify_fn
-    # returns ok=True, so _verify_and_maybe_regenerate never touches
-    # work_graph_run.status at all -- it only writes 'reflecting'/
-    # 'verifying'/'blocked' on the RETRY path), never wrote anything past
-    # that -- regardless of PASS/FAIL/INCONCLUSIVE, a validation run that
-    # got this far genuinely finished executing. `ux_work_graph_run_active`
-    # (schema.py) is a unique index on (task_id, org_node) covering exactly
-    # this non-terminal status set, so a real re-verify later (org_level_
-    # loop.run_fix_loop after a FAIL) collided with this row on INSERT --
-    # "UNIQUE constraint failed: work_graph_run.task_id, work_graph_run.
-    # org_node" -- found for real via a task reconciliation resume.
+    # 2026-08-07 (real incident, preserved): this run_id is now the SHARED
+    # Planner+Coder+Validator run (Phase 2c), not a Validator-only one --
+    # marking it 'done' here is still correct: reaching this line means the
+    # goal genuinely resolved to PASS, the task really is done.
+    # ux_work_graph_run_active's real re-verify collision this comment
+    # originally described no longer applies the same way once §2.2 lands
+    # (Phase 2c also narrows that index) -- kept here as the historical
+    # record of why this UPDATE exists at all, not because the exact old
+    # failure mode is still reachable.
     conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,))
     conn.commit()
     return {
-        "verdict": verdict, "raw_response": response,
+        "verdict": "PASS", "raw_response": response,
         "artifact_ids": artifact_ids, "script_execution": script_execution,
     }
@@ -1246,7 +1377,7 @@ def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
     org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT)
     verdict = validate_implementation_with_own_graph(
-        conn, task_id, sdd_content, prd_content, goal_text, target_folder_path,
+        conn, task_id, plan_result["run_id"], sdd_content, prd_content, goal_text, target_folder_path,
     )
     return {
@@ -64,6 +64,21 @@ def _latest_artifact_ids(conn, task_id, artifact_types):
     return {r[0]: r[1] for r in rows}
+def _shared_run_id_for_task(conn, task_id):
+    """Phase 2c: recovers the shared Planner+Coder+Validator work_graph_run
+    id for a task resumed straight from 'coding'/'validating' (no plan_result
+    in scope at that point, unlike the 'reviewing_planner_output' branch
+    above) -- same "read it off a real artifact row's produced_by_run_id"
+    trick e2e_pipeline._review_planner_output_with_retries() already uses,
+    since every real artifact row has a non-null FK to the run that made it."""
+    row = conn.execute(
+        "SELECT produced_by_run_id FROM artifact WHERE task_id = ? AND artifact_type = 'sdd' "
+        "ORDER BY created_at DESC LIMIT 1",
+        (task_id,),
+    ).fetchone()
+    return row[0] if row else None
+
+
 def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default):
     """Soft-fail artifact content lookup: returns `default` instead of
     raising when no such artifact exists for this task yet -- e.g. the
@@ -164,8 +179,9 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
         _check_not_paused(conn, task_id)
         org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
+        run_id = plan_result["run_id"]
         code_artifact_ids = e2e_pipeline.implement_code(
-            conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path,
+            conn, task_id, run_id, sdd_content, goal_text, target_folder_path,
             benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
             test_script_content=test_script_content,
         )
@@ -180,6 +196,12 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
         sdd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "sdd", "N/A: not provided")
         prd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "prd", "N/A: not provided")
         code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES)
+        run_id = _shared_run_id_for_task(conn, task_id)
+        if run_id is None:
+            raise TaskExecutionError(
+                "task {} is status={} but has no sdd artifact to recover a work_graph_run id from"
+                .format(task_id, status)
+            )
     _check_not_paused(conn, task_id)
     if status != "validating":
@@ -188,7 +210,7 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
         # hit org_node_lifecycle's own PRE_STATE guard (needs "coding").
         org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
     verdict = e2e_pipeline.validate_implementation_with_own_graph(
-        conn, task_id, sdd_content, prd_content, goal_text, target_folder_path,
+        conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path,
     )
     return {
@@ -115,24 +115,45 @@ _FAKE_CODER_GRAPH_MANIFEST = {
     "edges": [],
 }
+# Phase 2c (2026-08-12): Validator now self-plans too, into the SAME shared
+# run as Planner/Coder -- depends_on "sdd" for the same "avoid landing in
+# Planner's own already-'done' in-degree-zero layer" reason
+# _FAKE_CODER_GRAPH_MANIFEST's own comment gives. A single node_type=
+# 'final_verdict' node is the minimal real shape validate_implementation_
+# with_own_graph() accepts (see work_graph_plan._VALIDATOR_NODE_CATALOG's
+# own catalog entry) -- no test_script node here, this fixture is about
+# pipeline wiring, not the hard-gate mechanism (see VerdictHardGateTest/
+# ValidateImplementationWithOwnGraphTest above for that).
+_FAKE_VALIDATOR_GRAPH_MANIFEST = {
+    "nodes": [{
+        "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict",
+        "node_type": "final_verdict",
+        "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+        "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
+    }],
+    "edges": [],
+}
+
 _real_subprocess_run = subprocess.run  # captured before any test patches subprocess.run globally
 def _fake_plan_subprocess_run_dispatch(*args, **kwargs):
     """Real plan_work_graph() dispatches by the --agent argv value -- this
-    mock does the same, so both Planner's and Coder's planning calls (which
-    now share the same mocked subprocess.run patch target) each get their
-    own realistic fake manifest instead of one hardcoded response serving
-    both callers (which would hand Coder Planner's own brd/prd/... nodes,
-    colliding with the ones already in the run it's appending onto).
+    mock does the same, so Planner's/Coder's/Validator's planning calls
+    (which now all share the same mocked subprocess.run patch target) each
+    get their own realistic fake manifest instead of one hardcoded response
+    serving every caller (which would hand Coder/Validator Planner's own
+    brd/prd/... nodes, colliding with the ones already in the run they're
+    appending onto).
     patch("src.work_graph_plan.subprocess.run", ...) patches subprocess.run
     GLOBALLY, not just within work_graph_plan.py -- `subprocess` is the same
     singleton module object everywhere it's imported, so this side_effect
     also intercepts every OTHER real subprocess.run call this test's own
     call chain makes (artifact_write.write_artifact's `git add`/`git
-    commit`, e2e_pipeline's `git diff`/`git rev-parse`, etc). Those aren't
+    commit`, e2e_pipeline's `git diff`/`git rev-parse`, the final_verdict
+    node's own REAL test-script execution, etc). Those aren't
     `claude --agent ... -p ...` calls -- pass them through to the REAL
     subprocess.run instead of returning a canned fake "claude planning
     response" CompletedProcess for them too (which happened to not visibly
@@ -144,7 +165,12 @@ def _fake_plan_subprocess_run_dispatch(*args, **kwargs):
     if not argv or argv[0] != "claude":
         return _real_subprocess_run(*args, **kwargs)
     org_node = argv[argv.index("--agent") + 1]
-    manifest = _FAKE_CODER_GRAPH_MANIFEST if org_node == "coder" else _FAKE_GRAPH_MANIFEST
+    if org_node == "coder":
+        manifest = _FAKE_CODER_GRAPH_MANIFEST
+    elif org_node == "validator":
+        manifest = _FAKE_VALIDATOR_GRAPH_MANIFEST
+    else:
+        manifest = _FAKE_GRAPH_MANIFEST
     payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
     return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
@@ -384,11 +410,10 @@ class RunEndToEndTest(PipelineTestBase):
                 # content-keyed like every other self-planned node, not the
                 # old single synthetic node's response-keyed output.
                 {"content": "Implemented. Ran `pytest`: 3 passed."},
-                # Validator's OWN independent benchmark/dataset (2026-08-06) --
-                # deliberately different content from the Planner's, to also prove
-                # nothing here accidentally reuses the Planner's copies.
-                {"content": "# Validator's own Benchmark\nfake"},
-                {"content": "N/A: validator judged no dataset needed either"},
+                # Phase 2c (2026-08-12): Validator's own verdict is now a
+                # self-planned node too (_FAKE_VALIDATOR_GRAPH_MANIFEST),
+                # not a flat org_node_trigger.trigger_org_node call.
+                {"content": "VERDICT: PASS\nAll scenarios pass."},
             ]
             result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏")
@@ -406,35 +431,38 @@ class RunEndToEndTest(PipelineTestBase):
         all_artifacts = self.conn.execute(
             "SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],)
         ).fetchone()[0]
-        # brd, prd, sdd, benchmark, test_dataset, code_diff, test_report,
-        # validator_benchmark, validator_test_dataset
-        # (test_script/validator_test_script disabled, see _VALIDATOR_MANIFEST)
-        self.assertEqual(all_artifacts, 9)
-
-        # Phase 2 §6 (2026-08-12): Coder no longer gets its own work_graph_run
-        # -- implement_code() appends its self-planned nodes onto Planner's
-        # SAME run (plan_work_graph(..., append_to_run_id=...)), so only TWO
-        # rows exist now, not three: the shared Planner+Coder run (still
-        # labeled org_node='planner' -- that column reflects who ORIGINATED
-        # the run, not every org_node that has since contributed nodes to it;
-        # Phase 2c is expected to fold Validator into the same run too, at
-        # which point this column stops being a reliable "who's involved"
-        # signal at all, see design doc §2.1) and Validator's own separate run.
+        # brd, prd, sdd, benchmark, test_dataset, code_diff, test_report --
+        # Phase 2c: Validator's self-planned "verdict" node deliberately
+        # does NOT declare promotes_to_artifact (a verdict is a judgment,
+        # not a persisted deliverable -- see _generic_prompt_template's own
+        # instruction), so no validator_* artifacts exist anymore.
+        self.assertEqual(all_artifacts, 7)
+
+        # Phase 2c (2026-08-12): Validator now ALSO appends its self-planned
+        # nodes onto the SAME shared run (plan_work_graph(..., append_to_
+        # run_id=...)) instead of creating its own separate row -- only ONE
+        # work_graph_run row exists for this task now, still labeled
+        # org_node='planner' (that column reflects who ORIGINATED the run,
+        # not every org_node that has since contributed nodes to it -- §2.1
+        # is expected to retire this column's meaning once it's no longer
+        # a reliable "who's involved" signal at all).
         run_rows = self.conn.execute(
-            "SELECT run_id, org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
+            "SELECT run_id, org_node, status FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
         ).fetchall()
-        self.assertEqual(len(run_rows), 2)
-        self.assertEqual({row[1] for row in run_rows}, {"planner", "validator"})
+        self.assertEqual(len(run_rows), 1)
+        self.assertEqual(run_rows[0][1], "planner")
+        self.assertEqual(run_rows[0][2], "done")
-        # Coder's self-planned "impl" node genuinely landed in the SAME run
-        # as Planner's nodes, not a separate one.
-        planner_run_id = next(row[0] for row in run_rows if row[1] == "planner")
+        # Coder's "impl" AND Validator's "verdict" nodes both genuinely
+        # landed in the SAME run as Planner's own nodes.
+        shared_run_id = run_rows[0][0]
         node_ids = {
             row[0] for row in self.conn.execute(
-                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (planner_run_id,)
+                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,)
             ).fetchall()
         }
         self.assertIn("impl", node_ids)
+        self.assertIn("verdict", node_ids)
         self.assertIn("sdd", node_ids)
@@ -603,7 +631,7 @@ class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase):
             captured["benchmark_content"] = kwargs.get("benchmark_content")
             return {"code_diff": "diff-1", "test_report": "report-1"}
-        def fake_validate(conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, **kwargs):
+        def fake_validate(conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, **kwargs):
             captured["prd_content"] = prd_content
             return {"verdict": "PASS", "raw_response": "ok"}
@@ -919,205 +947,234 @@ class RunTestScriptForRealTest(unittest.TestCase):
         self.assertTrue(os.path.exists(expected))
-# validator_test_script disabled by default 2026-08-07 (see
-# e2e_pipeline._VALIDATOR_MANIFEST's own comment) -- tests below that need
-# to exercise the hard-gate/salvage mechanisms involving that node patch
-# _VALIDATOR_MANIFEST back to this 3-node shape for their own duration, so
-# that real code (still in the codebase, just not planned by default) stays
-# covered instead of silently rotting.
-_VALIDATOR_MANIFEST_WITH_TEST_SCRIPT = {
+def _fake_validator_plan_subprocess_run(manifest):
+    # Same real-passthrough-for-non-'claude'-argv requirement as this
+    # file's other subprocess.run fakes (see _fake_plan_subprocess_run_
+    # dispatch's own docstring) -- the final_verdict node's build_prompt
+    # wrapper actually EXECUTES a real test script via a real subprocess.run
+    # call of its own (run_test_script_for_real/_detect_script_language),
+    # which would otherwise also get intercepted by this same globally-
+    # patched target and handed the fake planning JSON instead of a real
+    # exit code.
+    def run(*args, **kwargs):
+        argv = args[0] if args else kwargs.get("args")
+        if not argv or argv[0] != "claude":
+            return _real_subprocess_run(*args, **kwargs)
+        payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
+        return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
+    return run
+
+
+# on_fail.action='hard_fail' below (not the real catalog's suggested
+# 'dynamic_loop_back') is a deliberate test-fixture choice, not something a
+# real self-planned graph would use: it isolates the hard-gate mechanical
+# check itself (this file's job) from the separate dynamic_loop_back/
+# Reflection-review retry machinery, which already has its own dedicated
+# coverage in test_work_graph_orchestrate.py (DynamicLoopBackContentErrorTest/
+# DynamicLoopBackSkillDefectTest/TallyReflectionTest) -- mocking a 3-perspective
+# real reflection jury here would be expensive and wouldn't test anything new.
+_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY = {
+    "nodes": [{
+        "node_id": "verdict", "depends_on": [], "task_prompt": "give your final verdict",
+        "node_type": "final_verdict",
+        "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+        "on_fail": {"action": "hard_fail"},
+    }],
+    "edges": [],
+}
+
+_FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT = {
     "nodes": [
-        {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []},
         {
-            "node_id": "validator_test_dataset", "artifact_type": "validator_test_dataset",
-            "depends_on": ["validator_benchmark"],
+            "node_id": "test_script", "depends_on": [], "task_prompt": "write a real test script",
+            "node_type": "test_script",
+            "judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
         },
         {
-            "node_id": "validator_test_script", "artifact_type": "validator_test_script",
-            "depends_on": ["validator_test_dataset", "validator_benchmark"],
+            "node_id": "verdict", "depends_on": ["test_script"], "task_prompt": "give your final verdict",
+            "node_type": "final_verdict",
+            "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+            "on_fail": {"action": "hard_fail"},
         },
     ],
     "edges": [],
 }
+class VerdictHardGateTest(unittest.TestCase):
+    """Phase 2c (2026-08-12): unit coverage of the two functions carrying
+    the safety-critical hard gate (2026-08-06, user-confirmed: a real
+    non-zero exit from Validator's own test script forces FAIL regardless
+    of what the LLM's verdict line says) forward into the self-planned
+    final_verdict node design -- direct/isolated rather than through the
+    full self-planning + run_graph() + work_node_run machinery, which the
+    integration tests below cover for the wiring, not this specific logic."""
+
+    def test_real_script_failure_overrides_pass_verdict_text(self):
+        shared_state = {"script_execution": {"ran": True, "exit_code": 1, "stdout": "", "stderr": "boom"}}
+        ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good")
+        self.assertFalse(ok)
+        self.assertIn("exit_code=1", detail)
+
+    def test_passing_script_and_pass_verdict_is_ok(self):
+        shared_state = {"script_execution": {"ran": True, "exit_code": 0, "stdout": "", "stderr": ""}}
+        ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good")
+        self.assertTrue(ok)
+        self.assertIsNone(detail)
+
+    def test_no_script_and_pass_verdict_is_ok(self):
+        ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: PASS\nlooks good")
+        self.assertTrue(ok)
+
+    def test_fail_verdict_is_not_ok_even_with_no_script(self):
+        ok, detail = e2e_pipeline._verdict_says_pass_check({})("VERDICT: FAIL\nreal bug found")
+        self.assertFalse(ok)
+        self.assertIn("FAIL", detail)
+
+    def test_inconclusive_verdict_is_not_ok(self):
+        # 2026-08-12 (Phase 2c, design decision): INCONCLUSIVE is treated
+        # the same as FAIL here, not specially accepted -- see
+        # validate_implementation_with_own_graph's own docstring for why.
+        ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: INCONCLUSIVE\nneeds human review")
+        self.assertFalse(ok)
+
+    def test_build_prompt_passes_through_unchanged_for_non_verdict_nodes(self):
+        shared_state = {}
+        inner = lambda node, outputs: "inner prompt for " + node["node_id"]
+        wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60)
+        result = wrapped({"node_id": "impl"}, {})
+        self.assertEqual(result, "inner prompt for impl")
+        self.assertEqual(shared_state, {})
+
+    def test_build_prompt_runs_real_script_for_verdict_node_with_script_dependency(self):
+        shared_state = {}
+        inner = lambda node, outputs: "base prompt"
+        node_types = {"test_script": "test_script"}
+        outputs = {"test_script": {"content": "import sys\nsys.exit(0)\n"}}
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
+            wrapped = e2e_pipeline._verdict_node_build_prompt(inner, node_types, "task-1", tmp, shared_state, 60)
+            result = wrapped({"node_id": "verdict", "node_type": "final_verdict"}, outputs)
+        self.assertTrue(shared_state["script_execution"]["ran"])
+        self.assertEqual(shared_state["script_execution"]["exit_code"], 0)
+        self.assertIn("base prompt", result)
+        self.assertIn("REAL execution result", result)
+
+    def test_build_prompt_na_stub_for_verdict_node_with_no_script_dependency(self):
+        shared_state = {}
+        inner = lambda node, outputs: "base prompt"
+        wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60)
+        wrapped({"node_id": "verdict", "node_type": "final_verdict"}, {})
+        self.assertFalse(shared_state["script_execution"]["ran"])
+
+
 class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
-    """2026-08-06: the safety-critical piece -- a real non-zero exit from
-    Validator's own script must force FAIL no matter what the LLM's verdict
-    line says (user-confirmed hard gate)."""
+    """Phase 2c (2026-08-12): Validator now self-plans its own verification
+    graph, appended onto the SAME shared run Planner/Coder already
+    contributed to (plan_work_graph(..., append_to_run_id=run_id)) -- see
+    e2e_pipeline.validate_implementation_with_own_graph's own docstring.
+    VerdictHardGateTest above covers the hard-gate logic itself in
+    isolation; these tests cover the wiring: planning dispatch, mechanical
+    check registration, run_id reuse, and result unwrapping."""
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)
         subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
-        self.conn = schema.connect(":memory:")
+        self.conn = schema.connect(":memory:", check_same_thread=False)
         schema.create_schema(self.conn)
         project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
         self.conn.execute(
             "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
             "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', 'validating', 'validator')"
         )
+        # The shared Planner+Coder run Validator appends onto -- empty on
+        # purpose (these tests aren't about cross-stage node dependencies,
+        # see test_task_execution.py for that), just needs to genuinely
+        # exist for append_to_run_id's own existence check.
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-shared', 'task-1', 'planner', '{\"nodes\": [], \"edges\": []}', 'running')"
+        )
         self.conn.commit()
     def tearDown(self):
         self.conn.close()
-    def _run(self, script_content, verdict_line, with_test_script=False):
-        mock_responses = [
-            {"content": "# Validator's own Benchmark"},
-            {"content": "N/A: no dataset needed"},
-        ]
-        if with_test_script:
+    def _run(self, manifest, verdict_content, script_content=None):
+        mock_responses = []
+        if script_content is not None:
             mock_responses.append({"content": script_content})
-        with contextlib.ExitStack() as stack:
-            stack.enter_context(patch.object(work_graph_execute, "write_workflow_script"))
-            mock_execute = stack.enter_context(patch.object(work_graph_execute, "execute_work_graph"))
-            stack.enter_context(patch.object(org_node_trigger, "trigger_org_node", return_value=verdict_line))
-            if with_test_script:
-                stack.enter_context(
-                    patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT)
-                )
+        mock_responses.append({"content": verdict_content})
+        with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_validator_plan_subprocess_run(manifest)), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
             mock_execute.side_effect = mock_responses
             return e2e_pipeline.validate_implementation_with_own_graph(
-                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
+                self.conn, "task-1", "run-shared", "sdd content", "prd content", "goal text", self.tmp.name,
             )
     def test_llm_says_pass_and_script_really_passes_stays_pass(self):
-        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True)
+        result = self._run(
+            _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good",
+            script_content="import sys\nsys.exit(0)\n",
+        )
         self.assertEqual(result["verdict"], "PASS")
         self.assertTrue(result["script_execution"]["ran"])
         self.assertEqual(result["script_execution"]["exit_code"], 0)
-    def test_llm_says_pass_but_script_really_fails_is_force_overridden_to_fail(self):
+    def test_llm_says_pass_but_script_really_fails_raises(self):
         # the exact case the user's hard-gate rule exists for: a script
         # that objectively fails must not be argued away by the LLM's own
-        # verdict line.
-        result = self._run(
-            "import sys\nsys.exit(1)\n", "VERDICT: PASS\nlooks good to me", with_test_script=True
-        )
-        self.assertEqual(result["verdict"], "FAIL")
-
-    def test_llm_says_fail_and_script_fails_stays_fail(self):
-        result = self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")
-        self.assertEqual(result["verdict"], "FAIL")
+        # verdict line -- hard_fail (see the fixture's own comment) makes
+        # this deterministic without invoking the loop-back/reflection jury.
+        with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
+            self._run(
+                _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good to me",
+                script_content="import sys\nsys.exit(1)\n",
+            )
-    def test_na_script_leaves_llm_judgment_untouched(self):
-        result = self._run("N/A: no mechanical check possible", "VERDICT: INCONCLUSIVE\nneeds human review")
-        self.assertEqual(result["verdict"], "INCONCLUSIVE")
+    def test_no_script_node_leaves_llm_judgment_as_the_only_signal(self):
+        result = self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
+        self.assertEqual(result["verdict"], "PASS")
         self.assertFalse(result["script_execution"]["ran"])
-    def test_writes_two_validator_artifacts_with_lineage(self):
-        # test_script disabled by default -- see _VALIDATOR_MANIFEST_WITH_
-        # TEST_SCRIPT variant below for the 3-artifact/script-lineage case.
-        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
-        self.assertEqual(set(result["artifact_ids"]), {"validator_benchmark", "validator_test_dataset"})
-        dataset_deps = json.loads(
-            self.conn.execute(
-                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
-                (result["artifact_ids"]["validator_test_dataset"],),
-            ).fetchone()[0]
-        )
-        self.assertIn(result["artifact_ids"]["validator_benchmark"], dataset_deps)
-
-    def test_writes_three_validator_artifacts_with_lineage_when_test_script_re_enabled(self):
-        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True)
-        self.assertEqual(
-            set(result["artifact_ids"]),
-            {"validator_benchmark", "validator_test_dataset", "validator_test_script"},
-        )
-        script_deps = json.loads(
-            self.conn.execute(
-                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
-                (result["artifact_ids"]["validator_test_script"],),
-            ).fetchone()[0]
-        )
-        self.assertIn(result["artifact_ids"]["validator_benchmark"], script_deps)
-        self.assertIn(result["artifact_ids"]["validator_test_dataset"], script_deps)
-
-    def test_creates_its_own_work_graph_run_row(self):
-        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
-        run_row = self.conn.execute(
-            "SELECT org_node, status FROM work_graph_run WHERE task_id = 'task-1'"
-        ).fetchone()
-        self.assertEqual(run_row[0], "validator")
-        # 2026-08-07 (real incident): this used to stay 'running' forever on
-        # the happy path -- verify_fn returning ok=True means
-        # _verify_and_maybe_regenerate never touches work_graph_run.status
-        # at all, so nothing else in this function's old code ever marked
-        # it terminal. schema.py's `ux_work_graph_run_active` unique index
-        # covers exactly this non-terminal status set, so a stuck 'running'
-        # row silently blocked any later re-verify of the SAME task from
-        # ever creating its own new validator run.
-        self.assertEqual(run_row[1], "done")
-
-    def test_second_validation_run_does_not_collide_with_the_first(self):
-        # The actual regression: org_level_loop.run_fix_loop re-verifying
-        # after a FAIL (or a reconciliation-loop resume retrying) calls this
-        # function again for the SAME task_id -- must not raise
-        # "UNIQUE constraint failed: work_graph_run.task_id,
-        # work_graph_run.org_node" against a first run stuck non-terminal.
-        self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")
-
-        # Benchmark/dataset content must differ from the first call's --
-        # write_artifact does a real `git commit`, which fails with "nothing
-        # to commit" if a second write produces byte-identical file content
-        # (the file path isn't run_id-scoped). Real re-verify runs would
-        # naturally regenerate slightly different content each time; this
-        # just makes that explicit rather than accidentally relying on it.
-        with 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\nfixed now"):
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark (re-verify)"},
-                {"content": "N/A: no dataset needed, still"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
-            second = e2e_pipeline.validate_implementation_with_own_graph(
-                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
-            )
+    def test_appends_into_the_shared_run_without_creating_a_new_row(self):
+        self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
+        rows = self.conn.execute("SELECT run_id, status FROM work_graph_run WHERE task_id = 'task-1'").fetchall()
+        self.assertEqual(rows, [("run-shared", "done")])
+        node_ids = {
+            r[0] for r in self.conn.execute(
+                "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'"
+            ).fetchall()
+        }
+        self.assertIn("verdict", node_ids)
+
+    def test_second_call_appends_fresh_nodes_without_colliding(self):
+        # A real re-verify plans its OWN new node_ids each time (a fresh
+        # LLM planning call, not a replay) -- node_id collision is what
+        # plan_work_graph's append mode actively guards against (raises
+        # PlanningFailedError), so this uses a different node_id for the
+        # second call, same as a real second planning call naturally would.
+        second_manifest = {
+            "nodes": [{
+                "node_id": "verdict_v2", "depends_on": [], "task_prompt": "give your final verdict",
+                "node_type": "final_verdict",
+                "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+                "on_fail": {"action": "hard_fail"},
+            }],
+            "edges": [],
+        }
+        self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
+        second = self._run(second_manifest, "VERDICT: PASS\nfixed now")
         self.assertEqual(second["verdict"], "PASS")
-        rows = self.conn.execute(
-            "SELECT status FROM work_graph_run WHERE task_id = 'task-1' ORDER BY rowid"
-        ).fetchall()
-        self.assertEqual([r[0] for r in rows], ["done", "done"])
-
-    def test_earlier_nodes_survive_a_later_node_failing_verification(self):
-        # 2026-08-07 (same real incident/fix as PlanDocumentsTest's own
-        # test of this name): validator_test_script fails verify_fn twice
-        # -- ConsistencyCheckFailedError must still propagate, but
-        # validator_benchmark/validator_test_dataset (genuinely produced
-        # and accepted) must be persisted, not discarded along with it.
-        # validator_test_script itself is disabled by default (see
-        # _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT above) -- re-enabled here to
-        # keep the salvage mechanism (generic, not test_script-specific) covered.
-        with patch.object(work_graph_execute, "write_workflow_script"), \
-             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
-             patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT):
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "this is prose, not a real script"},  # fails compile check
-                {"content": "still not real code either"},  # the one regeneration attempt: also fails
-            ]
-            with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
-                e2e_pipeline.validate_implementation_with_own_graph(
-                    self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
-                )
-
-        rows = self.conn.execute("SELECT artifact_type FROM artifact WHERE task_id = 'task-1'").fetchall()
-        persisted = {r[0] for r in rows}
-        self.assertEqual(persisted, {"validator_benchmark", "validator_test_dataset"})
-        self.assertNotIn("validator_test_script", persisted)
-        # the work_graph_run itself must NOT be left stuck 'running' either
-        # (ux_work_graph_run_active would block any later re-verify) --
-        # _verify_and_maybe_regenerate's own 'blocked' write on this path
-        # already covers it; confirming it here catches a regression if
-        # that ever stops being true.
-        run_status = self.conn.execute(
-            "SELECT status FROM work_graph_run WHERE task_id = 'task-1'"
-        ).fetchone()[0]
-        self.assertEqual(run_status, "blocked")
+        node_ids = {
+            r[0] for r in self.conn.execute(
+                "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'"
+            ).fetchall()
+        }
+        self.assertEqual(node_ids, {"verdict", "verdict_v2"})
 if __name__ == "__main__":
@@ -16,7 +16,7 @@ from unittest.mock import patch
 from src import (
     e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler,
-    task_execution, work_graph_execute, work_graph_plan,
+    task_execution, work_graph_execute, work_graph_orchestrate, work_graph_plan,
 )
@@ -41,9 +41,45 @@ _FAKE_CODER_MANIFEST = {
     "edges": [],
 }
+# Phase 2c (2026-08-12): Validator now self-plans too (validate_
+# implementation_with_own_graph() calls plan_work_graph(conn, "validator",
+# ..., append_to_run_id=...), through the SAME mocked subprocess boundary
+# as Planner's/Coder's own planning calls) -- a single node_type=
+# 'final_verdict' node is the minimal real shape this mechanism accepts
+# (see work_graph_plan._VALIDATOR_NODE_CATALOG's own catalog entry).
+# depends_on ["sdd"] for the same "avoid landing in the same in-degree-zero
+# topological layer as an already-'done' node" reason _FAKE_CODER_MANIFEST
+# above documents.
+_FAKE_VALIDATOR_MANIFEST = {
+    "nodes": [{
+        "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict",
+        "node_type": "final_verdict",
+        "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+        "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
+    }],
+    "edges": [],
+}
-def _fake_coder_plan_subprocess_run(*args, **kwargs):
-    payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_CODER_MANIFEST) + "\n```"}
+_real_subprocess_run = subprocess.run  # captured before any test patches subprocess.run globally
+
+
+def _fake_plan_subprocess_run_dispatch(*args, **kwargs):
+    """Real plan_work_graph() dispatches by the --agent argv value -- this
+    mock does the same, so Planner's/Coder's/Validator's planning calls
+    (which all now share the same mocked subprocess.run patch target) each
+    get their own realistic fake manifest instead of one hardcoded response
+    serving all three callers. Same dispatch-by-argv + real-passthrough-for-
+    non-'claude'-calls pattern as test_e2e_pipeline.py's helper of the same
+    name -- see that module's docstring for the full rationale (this file's
+    own git add/commit/diff calls inside the `with patch(...)` blocks below
+    would otherwise also get intercepted, since `subprocess` is one shared
+    module object)."""
+    argv = args[0]
+    if not argv or argv[0] != "claude":
+        return _real_subprocess_run(*args, **kwargs)
+    org_node = argv[argv.index("--agent") + 1]
+    manifest = _FAKE_VALIDATOR_MANIFEST if org_node == "validator" else _FAKE_CODER_MANIFEST
+    payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
     return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
@@ -95,6 +131,24 @@ class TaskExecutionTestBase(unittest.TestCase):
             "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
             "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest),
         )
+        # Phase 2c (2026-08-12, found while adding Validator self-planning
+        # tests): a real Planner run_graph() call ALSO leaves a real
+        # work_node_execution row behind for this node -- without one here
+        # too, a later self-planning append onto this SAME run_id (Coder's/
+        # Validator's own plan_work_graph(..., append_to_run_id=run_id))
+        # finds no existing (run_id, node_id) row for `artifact_type` and
+        # spuriously RE-EXECUTES it for real via the mocked execute_work_
+        # graph, burning an extra mock_execute.side_effect slot the calling
+        # test never accounted for (masked before this fix only because
+        # those tests happened to provide enough spare list items and never
+        # asserted on the exact content that ended up threaded through).
+        work_graph_orchestrate.ensure_placeholder_skill(self.conn, "planner-doc-writing", "planner")
+        self.conn.execute(
+            "INSERT OR IGNORE INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, "
+            "skill_ref, input_schema_ref, output_schema_ref, input_payload, output_payload, status) "
+            "VALUES (?, ?, ?, 'planner', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')",
+            ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})),
+        )
         rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
         abs_path = os.path.join(self.tmp.name, rel_path)
         os.makedirs(os.path.dirname(abs_path), exist_ok=True)
@@ -130,22 +184,17 @@ class ResumeFromCodingTest(TaskExecutionTestBase):
         self._write_artifact("task-1", "test_report", "report content")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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") as mock_trigger:
-            # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06)
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            # validator's self-planned final_verdict node (Phase 2c)
+            mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}]
             result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
         self.assertEqual(result["verdict"]["verdict"], "PASS")
         self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff")
-        # only the validator's final-verdict real call happened -- no coder call
-        mock_trigger.assert_called_once()
-        self.assertEqual(mock_trigger.call_args[0][0], "validator")
+        # only the validator's self-planned node ran -- no coder call
+        mock_execute.assert_called_once()
         row = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()
         self.assertEqual(row[0], "validating")
@@ -168,17 +217,14 @@ class ResumeFromPlanningTest(TaskExecutionTestBase):
         self._write_artifact("task-1", "benchmark", "# Benchmark content")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
-             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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") as mock_trigger, \
              patch.object(e2e_pipeline, "plan_documents") as mock_plan:
             mock_execute.side_effect = [
                 {"content": "done, tests pass"},  # coder's self-planned "impl" node
-                # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06)
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
+                {"content": "VERDICT: PASS\nlooks good"},  # validator's self-planned final_verdict node (Phase 2c)
             ]
             result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
@@ -187,11 +233,11 @@ class ResumeFromPlanningTest(TaskExecutionTestBase):
         # existing artifacts were reused -- plan_documents() must NOT have been
         # called again (that's the whole point of the existence-probing retry loop).
         mock_plan.assert_not_called()
-        # two real org_node_trigger.trigger_org_node calls: planner_reviewer's
-        # judgment, then validator's final verdict -- not just validator anymore.
-        self.assertEqual(mock_trigger.call_count, 2)
+        # one real org_node_trigger.trigger_org_node call: planner_reviewer's
+        # judgment -- Validator's own verdict is now a self-planned node (Phase
+        # 2c), read via mock_execute above, not a flat org_node_trigger call.
+        mock_trigger.assert_called_once()
         self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer")
-        self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator")
 class ResumeFromReviewingTest(TaskExecutionTestBase):
@@ -216,16 +262,14 @@ class ResumeFromReviewingTest(TaskExecutionTestBase):
             return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout)
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=tracking_ready_trigger), \
-             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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") as mock_trigger, \
              patch.object(e2e_pipeline, "plan_documents") as mock_plan:
             mock_execute.side_effect = [
                 {"content": "done, tests pass"},  # coder's self-planned "impl" node
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
+                {"content": "VERDICT: PASS\nlooks good"},  # validator's self-planned final_verdict node (Phase 2c)
             ]
             result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
@@ -236,9 +280,11 @@ class ResumeFromReviewingTest(TaskExecutionTestBase):
         # this task into 'reviewing_planner_output' in the first place).
         self.assertEqual(ready_calls, ["coder", "validator"])
         # but the real review judgment call DOES still run (that's the actual
-        # gap this status resumes into -- see the module docstring).
+        # gap this status resumes into -- see the module docstring). Validator's
+        # own verdict is now a self-planned node (Phase 2c), read via
+        # mock_execute above, not a second org_node_trigger call.
+        mock_trigger.assert_called_once()
         self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer")
-        self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator")
 class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase):
@@ -266,16 +312,14 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase):
             return {"run_id": "run-sdd", "artifact_ids": {"sdd": "art-sdd"}}
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
-             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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"), \
              patch.object(e2e_pipeline, "plan_documents", side_effect=fake_plan_documents) as mock_plan:
             mock_execute.side_effect = [
                 {"content": "done, tests pass"},  # coder's self-planned "impl" node
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
+                {"content": "VERDICT: PASS\nlooks good"},  # validator's self-planned final_verdict node (Phase 2c)
             ]
             # must not raise (this is the actual regression: it used to KeyError/
             # TaskExecutionError before ever reaching here)
@@ -304,14 +348,10 @@ class ResumeFromValidatingTest(TaskExecutionTestBase):
         self._write_artifact("task-1", "test_report", "report content")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger) as mock_ready, \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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") as mock_trigger:
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}]  # validator's self-planned final_verdict node
             result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal")
         self.assertEqual(result["verdict"]["verdict"], "PASS")
@@ -321,14 +361,19 @@ class ResumeFromValidatingTest(TaskExecutionTestBase):
         # get the task into 'validating' in the first place, and calling it
         # again would hit org_node_lifecycle's own "needs status=coding" guard.
         mock_ready.assert_not_called()
-        # only the real work_graph_run's own final call happened
-        mock_trigger.assert_called_once()
-        self.assertEqual(mock_trigger.call_args[0][0], "validator")
-        # a brand-new validator work_graph_run was created (none existed before)
-        rows = self.conn.execute(
-            "SELECT run_id FROM work_graph_run WHERE task_id = 'task-1' AND org_node = 'validator'"
-        ).fetchall()
-        self.assertEqual(len(rows), 1)
+        # only the validator's self-planned node ran
+        mock_execute.assert_called_once()
+        # Phase 2c: Validator appends its final_verdict node into the SAME
+        # shared run the sdd artifact points at (via _shared_run_id_for_task),
+        # not a brand-new work_graph_run row of its own -- confirm the merge
+        # landed on "run-sdd" and no extra row was created for this task.
+        row = self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-sdd'").fetchone()
+        node_ids = {n["node_id"] for n in json.loads(row[0])["nodes"]}
+        self.assertIn("verdict", node_ids)
+        total_rows = self.conn.execute(
+            "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'"
+        ).fetchone()[0]
+        self.assertEqual(total_rows, 5)
 class ResumeUnresumableStatusTest(TaskExecutionTestBase):
@@ -473,9 +518,24 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
     def _write_artifact(self, task_id, artifact_type, content):
         conn = schema.connect(self.db_path)
         run_id = "run-" + artifact_type
+        # Real "nodes"/"edges" shape + a matching 'done' work_node_execution
+        # row -- see TaskExecutionTestBase._write_artifact's own comment
+        # (same file) for why both are needed once Validator self-plans and
+        # appends onto this run via plan_work_graph(..., append_to_run_id=...).
+        manifest = json.dumps({
+            "nodes": [{"node_id": artifact_type, "artifact_type": artifact_type, "depends_on": []}],
+            "edges": [],
+        })
         conn.execute(
             "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id),
+            "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest),
+        )
+        work_graph_orchestrate.ensure_placeholder_skill(conn, "planner-doc-writing", "planner")
+        conn.execute(
+            "INSERT OR IGNORE INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, "
+            "skill_ref, input_schema_ref, output_schema_ref, input_payload, output_payload, status) "
+            "VALUES (?, ?, ?, 'planner', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')",
+            ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})),
         )
         rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
         abs_path = os.path.join(self.repo.name, rel_path)
@@ -506,14 +566,10 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
         self._write_artifact("task-1", "test_report", "report")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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"):
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}]  # validator's self-planned final_verdict node
             thread = task_execution.start_task_execution_in_background(
                 self.db_path, "task-1", self.repo.name, "a goal",
             )
@@ -561,14 +617,10 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
             return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout)
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=blocking_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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"):
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}]  # validator's self-planned final_verdict node
             thread = task_execution.start_task_execution_in_background(
                 self.db_path, "task-1", self.repo.name, "a goal",
             )
@@ -591,17 +643,14 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
         self._write_artifact("task-1", "test_report", "report")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \
              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", side_effect=RuntimeError("boom")):
-            # validator's own benchmark/dataset/script graph completes fine --
-            # the failure this test cares about is the FINAL verdict call
-            # (org_node_trigger.trigger_org_node), which happens after.
-            mock_execute.side_effect = [
-                {"content": "# Validator's own Benchmark"},
-                {"content": "N/A: no dataset needed"},
-                {"content": "import sys\nsys.exit(0)\n"},
-            ]
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=RuntimeError("boom")):
+            # Phase 2c: Validator's own verdict is now a self-planned node,
+            # not a flat org_node_trigger.trigger_org_node call -- the
+            # failure this test cares about (a real exception surfacing as
+            # blocked_pending_human/execution_error) is injected at its own
+            # node's real execution instead.
             thread = task_execution.start_task_execution_in_background(
                 self.db_path, "task-1", self.repo.name, "a goal",
             )
@@ -81,6 +81,12 @@ GRAPH_MANIFEST_SCHEMA = {
                     },
                     "subagent_type": {"type": "string"},
                     "skill_ref": {"type": "string"},
+                    # 2026-08-12 (Phase 2c): echoed-back catalog label, e.g.
+                    # "test_script"/"final_verdict" -- see _generic_prompt_
+                    # template's own comment for why this exists (finding
+                    # role-specific nodes among arbitrarily-named
+                    # self-planned ones).
+                    "node_type": {"type": "string"},
                 },
             },
         },
@@ -232,9 +238,22 @@ _VALIDATOR_NODE_CATALOG = [
     {"node_type": "test_dataset", "description": "Concrete test data, if this goal has a natural one."},
     {
         "node_type": "test_script",
-        "description": "A real, runnable script mechanically checking the independent benchmark's scenarios.",
+        "description": "A real, runnable script mechanically checking the independent benchmark's scenarios. "
+                        "It will actually be EXECUTED against the real repository after you plan it -- write "
+                        "real commands, not a description of what a script would do.",
         "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
     },
+    {
+        "node_type": "final_verdict",
+        "description": "Exactly ONE node of this type, depending on every other node whose content your "
+                        "verdict needs. Its content must be your final judgment: a line 'VERDICT: PASS', "
+                        "'VERDICT: FAIL', or 'VERDICT: INCONCLUSIVE', followed by your reasoning (and, if not "
+                        "PASS, a root_cause_category of planning|coder|skill_defect with detail). This is the "
+                        "ONLY node in a Validator graph allowed to use on_fail.action='dynamic_loop_back' -- "
+                        "see the on_fail instructions below for why.",
+        "suggested_judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
+        "suggested_on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
+    },
 ]
 _GENERIC_NODE_CATALOGS = {"coder": _CODER_NODE_CATALOG, "validator": _VALIDATOR_NODE_CATALOG}
@@ -245,15 +264,37 @@ def _generic_prompt_template(org_node):
     org_node so each gets its own suggested catalog (or none, for any
     future org_node this hasn't been extended for yet -- catalog_section
     is simply empty then, same as before this change existed).
-    node_type is a label in the SUGGESTION data only, not a required
-    graph_manifest node field -- node_id/task_prompt stay exactly as
-    freeform as before."""
+
+    2026-08-12 (Phase 2c): `node_type` is now asked to be ECHOED BACK onto
+    each planned node (not just a label in the suggestion data) -- e2e_
+    pipeline.py's Validator rework needs to find "the test_script node"/
+    "the final_verdict node" among self-planned, arbitrarily-named nodes,
+    and node_type is the one signal that survives self-planning for this
+    (node_id is freeform, task_prompt is prose). `dynamic_loop_back` is
+    now allowed, but ONLY for org_node='validator' AND only on a node
+    that also declares node_type='final_verdict' -- this reverses the
+    Phase 2a/2b instruction ("reserved for the terminal review stage
+    outside this graph, not for nodes you plan yourself"), because
+    Validator's terminal review IS now planned inside this graph, not a
+    flat call after it (see _VALIDATOR_NODE_CATALOG's own 'final_verdict'
+    entry for why: only a real graph node can use the loop_back mechanism
+    _run_one_node already implements). Every OTHER org_node still can't
+    use it -- Coder's catalog has no comparable terminal-judgment role."""
     catalog = _GENERIC_NODE_CATALOGS.get(org_node, [])
     catalog_section = (
         "Suggested node types for this goal (use any subset, in any order, invent your own if "
         "none fit -- these are starting points, not requirements):\n" + json.dumps(catalog) + "\n"
         if catalog else ""
     )
+    dynamic_loop_back_rule = (
+        "{\"action\": \"dynamic_loop_back\"} ONLY for the node also declaring "
+        "node_type='final_verdict' (see the final_verdict catalog entry above) -- no on_failure "
+        "edge needed for this one, the target is chosen dynamically at failure time, not declared "
+        "now.\n"
+        if org_node == "validator" else
+        "do NOT use \"dynamic_loop_back\" -- that's reserved for Validator's terminal review node, "
+        "not for nodes you plan yourself.\n"
+    )
     return (
         "Plan (do not execute) a Work Graph for this goal: {goal_text}\n"
         + catalog_section +
@@ -267,6 +308,9 @@ def _generic_prompt_template(org_node):
         "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), plus:\n"
+        "  - node_type (optional but STRONGLY suggested when a node matches one of the suggested "
+        "types above): echo back the matching node_type string (e.g. \"test_script\") so it's "
+        "identifiable later -- omit only for a node that doesn't match any suggested type.\n"
         "  - judgment (optional): how to tell if this node's output is acceptable -- "
         "{\"kind\": \"none\"} if nothing needs checking (the default if omitted), {\"kind\": "
         "\"llm\", \"judge_prompt\": \"...{content}...\"} for your own judgment prompt with a "
@@ -274,13 +318,14 @@ def _generic_prompt_template(org_node):
         "  - on_fail (only meaningful if judgment.kind != \"none\"): {\"action\": \"retry_self\"}, "
         "{\"action\": \"hard_fail\"}, or {\"action\": \"loop_back\"} with a matching on_failure "
         "edge to a genuine upstream dependency (same rule _PLANNER_NODE_CATALOG's own nodes "
-        "follow) -- do NOT use \"dynamic_loop_back\", that's reserved for the terminal review "
-        "stage outside this graph, not for nodes you plan yourself.\n"
+        "follow); " + dynamic_loop_back_rule +
         "  - promotes_to_artifact (optional): {\"artifact_type\": \"test_report\"} ONLY for a "
         "node producing this task's actual final deliverable text -- most nodes should omit "
         "this; it marks a node's content for persistence beyond this graph's own execution. "
         "(code_diff, if relevant, is derived separately from real file changes, not from any "
-        "node's own content -- never declare artifact_type: \"code_diff\" here.)\n"
+        "node's own content -- never declare artifact_type: \"code_diff\" here. The final_verdict "
+        "node, if any, should also NOT declare promotes_to_artifact -- a verdict is a judgment, "
+        "not a persisted deliverable.)\n"
         "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 "