Org Graph unification Phase 2c §2.1/§2.2: drop work_graph_run.org_node

A merged graph's one run genuinely spans several org_nodes now (Planner/ Coder/Validator all append their own nodes into the SAME run) -- a single run-level org_node column was never accurate once Phase 2b/2c landed, and real per-node ownership already lives in work_node_execution.subagent_type. Schema: work_graph_run.org_node column removed; ux_work_graph_run_active narrowed from UNIQUE(task_id, org_node) to UNIQUE(task_id) WHERE status NOT IN ('done', 'failed') -- 'blocked' now counts as active on purpose, since resuming a blocked run always appends into the same row rather than inserting a fresh one. New migrate_drop_work_graph_run_org_node() handles already-populated databases, wired into backend_server.py's startup path. Call sites updated: work_graph_plan.py/work_graph_fork.py/ root_cause_trace.py's INSERTs; work_graph_run_query.py/task_query.py now return "org_nodes_involved" (derived from work_node_execution.subagent_type) instead of the removed single-value "org_node"; frontend/app.js's org-strip and run-list rendering updated to the list-based shape. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Org Graph unification Phase 2c §2.1/§2.2: drop work_graph_run.org_node
A merged graph's one run genuinely spans several org_nodes now (Planner/ Coder/Validator all append their own nodes into the SAME run) -- a single run-level org_node column was never accurate once Phase 2b/2c landed, and real per-node ownership already lives in work_node_execution.subagent_type. Schema: work_graph_run.org_node column removed; ux_work_graph_run_active narrowed from UNIQUE(task_id, org_node) to UNIQUE(task_id) WHERE status NOT IN ('done', 'failed') -- 'blocked' now counts as active on purpose, since resuming a blocked run always appends into the same row rather than inserting a fresh one. New migrate_drop_work_graph_run_org_node() handles already-populated databases, wired into backend_server.py's startup path. Call sites updated: work_graph_plan.py/work_graph_fork.py/ root_cause_trace.py's INSERTs; work_graph_run_query.py/task_query.py now return "org_nodes_involved" (derived from work_node_execution.subagent_type) instead of the removed single-value "org_node"; frontend/app.js's org-strip and run-list rendering updated to the list-based shape. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent a68f0adb
Showing 26 changed files with 564 additions and 178 deletions
docs/todo/org-graph-unification-phase2-universal-self-planning-design.md
frontend/app.js
frontend/test_org_click_through_smoke.js
src/backend_server.py
src/root_cause_trace.py
src/scheduler.py
src/schema.py
src/task_query.py
src/test_artifact_query.py
src/test_artifact_write.py
src/test_backend_server.py
src/test_e2e_pipeline.py
src/test_execution_log.py
src/test_org_node_lifecycle.py
src/test_org_node_lifecycle_e2e.py
src/test_schema.py
src/test_task_execution.py
src/test_task_query.py
src/test_work_graph_fork.py
src/test_work_graph_orchestrate.py
@@ -3,7 +3,9 @@
 **日期**: 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 自规划改造（§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。
+**状态**: 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` 先例）。§7.2 的收口验证已完成分析（见 §10 Phase 2c 小节末尾：一旦 verdict 节点自己的 `dynamic_loop_back` 内部消化重试，`validate_implementation_with_own_graph()` 正常返回时 verdict 必为 PASS，`org_level_loop.py` 的旧粗粒度路由对这条新路径自然不可达，不需要显式新旧流程判断——决定保留旧分支代码不删，已在 `scheduler.py::finish_task_lifecycle` 加注释说明）。
+
+**2026-08-13 §2.1/§2.2 已实施**：`work_graph_run.org_node` 列彻底删除，`ux_work_graph_run_active` 收紧为 `UNIQUE(task_id) WHERE status NOT IN ('done', 'failed')`（'blocked' 故意算作活跃——恢复一个 blocked 的 run 现在总是走 `append_to_run_id` 追加进同一行，不会再 INSERT 新行）。新迁移函数 `schema.migrate_drop_work_graph_run_org_node()`（内部先幂等调用 `migrate_relax_org_node_constraints()`，再做 create-copy-drop-rename，`PRAGMA foreign_keys=OFF` 全程包裹，理由同 `migrate_add_reviewing_planner_output_status` 的真实重启事故记录——`work_graph_run` 是被多张表 REFERENCES 的父表）已接入 `backend_server.py::_ensure_schema()`。下游改动：`work_graph_plan.py`/`work_graph_fork.py`（含 `create_fork()`/`_reconstruct_build_prompt()`/`execute_fork()`，`fork_info["org_node"]` 改名 `fork_info["subagent_type"]`，源节点自己的执行身份）/`root_cause_trace.py` 的三处 INSERT 全部去掉 `org_node` 列；`work_graph_run_query.py`（`list_work_graph_runs`/`get_work_graph_run`）与 `task_query.py::get_task` 的返回形状从单值 `"org_node"` 改成 `"org_nodes_involved"`（从 `work_node_execution.subagent_type` 去重派生的列表，`org_node=` 查询参数语义从等值匹配改成 EXISTS 子查询），前端 `frontend/app.js` 对应改造（`getLatestRunForOrgNode`/`orgNodesLabel`/`renderOrgGraphTopology` 等消费点全部改用列表语义）。发现一处设计文档本节 §2.1.3 没预料到的遗漏点：`task_query.py::get_task` 也有同样的单值 `org_node` 读取，一并修完。`root_cause_trace.py::judge_root_cause_via_work_graph` 的 INSERT 有一个已知、记录在代码注释里的边界情况：索引收紧后，若这个函数被调到时对应 task 已有一个仍处于活跃/blocked 状态的共享 run，会撞索引——目前不是真实风险（这个函数唯一的调用方 `org_level_loop.run_fix_loop` 只在一个已经不可达的分支里被调用，且它自己的测试夹具里对应 task 从未预先插入过 work_graph_run 行），但如果未来重新接上这条路径需要重新评估。`python3 -m unittest discover -s src`（768 tests）+ `discover -s supervisor`（126 tests）+ 4 个 frontend smoke test 全绿。**仍未实施**：§7.1（新 skill `work-graph-node-review` + 三个 role.json 追加）。**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。
 ---
@@ -692,10 +692,23 @@ const ORG_NODE_ORDER = [&quot;planner&quot;, &quot;coder&quot;, &quot;validator&quot;];
 // where on the already-rendered page to scroll/highlight.
 function getLatestRunForOrgNode(task, orgNode) {
-  const runs = task.work_graph_runs.filter((r) => r.org_node === orgNode);
+  // 2026-08-12 (Org Graph unification Phase 2c §2.1): a run's single
+  // org_node field is gone -- a merged run genuinely spans several
+  // org_nodes now (Planner/Coder/Validator can all append nodes into the
+  // SAME run), so "the run for this org_node" is membership in
+  // org_nodes_involved (a list, one entry per distinct subagent_type
+  // among the run's nodes), not equality against a single value.
+  const runs = task.work_graph_runs.filter((r) => (r.org_nodes_involved || []).includes(orgNode));
   return runs.length ? runs[runs.length - 1] : null;
 }
+// Display label for a run's org_nodes_involved list -- everywhere this
+// file used to show the single old org_node value directly.
+function orgNodesLabel(run) {
+  const nodes = run.org_nodes_involved || [];
+  return nodes.length ? nodes.join(" + ") : "(no nodes yet)";
+}
+
 // v_dashboard's org-strip: role + hint text, only ever links out to
 // v_task_detail (a genuine page change, so plain hash assignment is fine).
 function renderDashboardOrgStripHtml(activeOrgNode, runStatusByOrgNode) {
@@ -894,7 +907,7 @@ function mountDashboard() {
     const runStatusByOrgNode = {};
     let waitingNode = null;
     for (const run of detail.work_graph_runs) {
-      runStatusByOrgNode[run.org_node] = run.status;
+      for (const orgNode of run.org_nodes_involved || []) runStatusByOrgNode[orgNode] = run.status;
       if (!waitingNode) waitingNode = run.nodes.find((n) => n.status === "waiting_for_quota");
     }
     renderOrgStrip(active.assigned_org_node, runStatusByOrgNode);
@@ -1207,7 +1220,7 @@ function renderGraphVisualizerPanel(run, options) {
     // id="graph-panel-<run_id>": Org Graph 点击可达详情's scroll/highlight
     // target (focusOrgNodePanel above) -- additive, doesn't change any
     // existing class/data-* this function already emitted.
-    '<div class="panel" id="graph-panel-' + run.run_id + '"><h2>' + run.org_node + " Work Graph DAG 视图（v_graph_visualizer，run_id=" + run.run_id + "）</h2>"
+    '<div class="panel" id="graph-panel-' + run.run_id + '"><h2>' + orgNodesLabel(run) + " Work Graph DAG 视图（v_graph_visualizer，run_id=" + run.run_id + "）</h2>"
     + '<div class="work-graph" data-on-failure-edges="' + edgesDataJson + '">'
     + '<svg class="graph-edges-svg"></svg>'
     + rowsHtml + "</div>"
@@ -1456,7 +1469,7 @@ function mountTaskDetail(taskId, initialOrgNode) {
     return (
       // id="todo-tree-<run_id>": secondary scroll/highlight target when the
       // DAG panel above didn't render (graph_manifest missing).
-      '<div class="panel" id="todo-tree-' + run.run_id + '"><h2>' + run.org_node + " Work Graph 的 to-do 树（run_id=" + run.run_id + "）</h2>"
+      '<div class="panel" id="todo-tree-' + run.run_id + '"><h2>' + orgNodesLabel(run) + " Work Graph 的 to-do 树（run_id=" + run.run_id + "）</h2>"
       + '<ul class="todo-tree">' + items + "</ul>"
       + '<div class="see-more">DAG 视图（同一份 graph_manifest 的另一种呈现）见下方 v_graph_visualizer</div>'
       + "</div>"
@@ -1467,7 +1480,9 @@ function mountTaskDetail(taskId, initialOrgNode) {
     closeAllLogViewers();
     const activeSet = new Set([task.assigned_org_node]);
     for (const run of task.work_graph_runs) {
-      if (run.status !== "done" && run.status !== "failed") activeSet.add(run.org_node);
+      if (run.status !== "done" && run.status !== "failed") {
+        for (const orgNode of run.org_nodes_involved || []) activeSet.add(orgNode);
+      }
     }
     const orgStripHtml = renderTaskDetailOrgStripHtml(activeSet);
@@ -1848,7 +1863,7 @@ function mountRunDetail(runId) {
   function renderDetail(run) {
     closeAllLogViewers();
     content.innerHTML = (
-      "<h1>" + escapeHtml(run.org_node) + " Work Graph Run</h1>"
+      "<h1>" + escapeHtml(orgNodesLabel(run)) + " Work Graph Run</h1>"
       + renderForkBadgeHtml(run)
       + '<div class="meta-row">'
       + "<div><b>RUN_ID</b>" + escapeHtml(run.run_id) + "</div>"
@@ -1979,7 +1994,7 @@ async function mountArtifactsOverviewPanel(container, projectId) {
   const runsHtml = runs.length === 0 ? '<div class="hint">暂无 Graph 运行记录</div>' : (
     '<ul class="simple-list" id="gc_runs_list">' + runs.slice(0, 10).map((r) =>
-      '<li data-run-id="' + r.run_id + '" style="cursor:pointer;"><span>' + escapeHtml(r.org_node) + " · run "
+      '<li data-run-id="' + r.run_id + '" style="cursor:pointer;"><span>' + escapeHtml(orgNodesLabel(r)) + " · run "
       + r.run_id.slice(0, 8) + (r.parent_run_id ? " 🔀" : "") + "</span><span>"
       + escapeHtml(r.status) + " · " + r.node_count + " 节点</span></li>"
     ).join("") + "</ul>"
@@ -2108,7 +2123,9 @@ function mountGraphConsole(queryParams) {
     const task = await resp.json();
     const runStatusByOrgNode = {};
-    for (const run of task.work_graph_runs) runStatusByOrgNode[run.org_node] = run.status;
+    for (const run of task.work_graph_runs) {
+      for (const orgNode of run.org_nodes_involved || []) runStatusByOrgNode[orgNode] = run.status;
+    }
     orgTopologyEl.innerHTML = renderOrgGraphTopology(task.assigned_org_node, runStatusByOrgNode, task.route);
     const panelsHtml = task.work_graph_runs
@@ -96,9 +96,9 @@ function check(condition, message) {
 {
   const task = {
     work_graph_runs: [
-      { run_id: "run-old", org_node: "planner", status: "done", started_at: "2026-08-01 00:00:00" },
-      { run_id: "run-new", org_node: "planner", status: "done", started_at: "2026-08-02 00:00:00" },
-      { run_id: "run-coder", org_node: "coder", status: "running", started_at: "2026-08-02 01:00:00" },
+      { run_id: "run-old", org_nodes_involved: ["planner"], status: "done", started_at: "2026-08-01 00:00:00" },
+      { run_id: "run-new", org_nodes_involved: ["planner"], status: "done", started_at: "2026-08-02 00:00:00" },
+      { run_id: "run-coder", org_nodes_involved: ["coder"], status: "running", started_at: "2026-08-02 01:00:00" },
     ],
   };
   const latestPlanner = sandbox.getLatestRunForOrgNode(task, "planner");
@@ -119,7 +119,7 @@ function check(condition, message) {
   const taskWithRun = {
     work_graph_runs: [
-      { run_id: "run-1", org_node: "coder", status: "done", graph_manifest: { nodes: [{ node_id: "n1", depends_on: [] }] } },
+      { run_id: "run-1", org_nodes_involved: ["coder"], status: "done", graph_manifest: { nodes: [{ node_id: "n1", depends_on: [] }] } },
     ],
     artifacts: [
       { artifact_id: "art-1", produced_by_run_id: "run-1", file_path: "src/x.py", artifact_type: "code_diff", validation_status: "passed" },
@@ -132,7 +132,7 @@ function check(condition, message) {
   check(!populatedHtml.includes('data-artifact-id="art-2"'), "artifact from a different run_id must be filtered out");
   const taskNoArtifacts = {
-    work_graph_runs: [{ run_id: "run-2", org_node: "coder", status: "failed", graph_manifest: null }],
+    work_graph_runs: [{ run_id: "run-2", org_nodes_involved: ["coder"], status: "failed", graph_manifest: null }],
     artifacts: [],
   };
   const noArtifactsHtml = sandbox.renderOrgComboPanelContent(taskNoArtifacts, "coder");
@@ -1041,6 +1041,7 @@ def _ensure_schema(conn):
     schema.migrate_add_reviewing_planner_output_status(conn)
     schema.migrate_add_planner_review_verdict_event_type(conn)
     schema.migrate_relax_agent_profile_migrated_from_profile(conn)
+    schema.migrate_drop_work_graph_run_org_node(conn)
 class _Server(ThreadingHTTPServer):
@@ -219,10 +219,21 @@ def judge_root_cause_via_work_graph(
     manifest = _build_jury_graph_manifest(list(perspective_prompts))
     run_id = str(uuid.uuid4())
+    # org_node column removed (Org Graph unification Phase 2c §2.1,
+    # 2026-08-12) -- see work_graph_plan.py's own INSERT for the general
+    # rationale. Note: ux_work_graph_run_active is now UNIQUE(task_id) (was
+    # (task_id, org_node)), so this INSERT would now collide with a task's
+    # own still-active shared Planner/Coder/Validator run if one existed --
+    # not a real concern today since this function's only caller
+    # (org_level_loop.run_fix_loop) is only reachable from a task that
+    # never got a work_graph_run row in the first place (see scheduler.py::
+    # finish_task_lifecycle's own comment on why that branch is effectively
+    # dead now that validate_implementation_with_own_graph() only ever
+    # returns PASS or raises), but worth knowing if this function is ever
+    # wired up to a live task with its own in-flight run again.
     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)),
+        "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, ?, 'running')",
+        (run_id, task_id, json.dumps(manifest)),
     )
     conn.commit()
@@ -162,7 +162,22 @@ def _record_core_engine_audit_if_self_target(conn, task_id, target_folder_path, 
 def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
     """PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
     run_fix_loop` (root-cause judge + reroute + fix + reverify, U-017).
-    Adds 'final_status' to `run_result` and returns it."""
+    Adds 'final_status' to `run_result` and returns it.
+
+    2026-08-12 (Org Graph unification Phase 2c): the FAIL/INCONCLUSIVE
+    branch below is effectively dead code for tasks going through the
+    current e2e_pipeline.py flow -- `validate_implementation_with_own_
+    graph()`'s verdict is now a self-planned node with `judgment.kind=
+    'mechanical'`/`on_fail.action='dynamic_loop_back'` (see that function's
+    own docstring), which retries internally via work_graph_orchestrate.py's
+    Reflection-review + [Fix-Skill] + loop-back machinery; a normal return
+    from that function can therefore only ever carry verdict='PASS' -- a
+    goal that can't resolve within the retry budget raises
+    ConsistencyCheckFailedError instead, which routes to
+    blocked_pending_human upstream of this function ever being called with
+    a non-PASS verdict. Kept, not deleted, per minimal-change philosophy
+    (a caller using the older, still-live `validate_implementation()`/
+    hardcoded-manifest path -- none currently do -- would still need it)."""
     task_id = run_result["task_id"]
     verdict = run_result["verdict"]["verdict"]
     if verdict == "PASS":
@@ -17,7 +17,9 @@ Interpretive conventions used below, where 01-M leaves the physical type implici
 - task.assigned_org_node is documented as "not null, fk" but there is no standalone
   org-node lookup table to reference (planner/coder/validator only exist as entries
   inside org_graph_definition.nodes, a JSON column) -- implemented as a CHECK enum
-  instead, matching work_graph_run.org_node / skill.owner_org_node / agent_profile.org_node.
+  instead, matching skill.owner_org_node / agent_profile.org_node / work_node_execution.subagent_type
+  (work_graph_run.org_node existed the same way until it was removed, Org Graph unification
+  Phase 2c §2.1, 2026-08-12 -- see that column's own removal comment above work_graph_run).
 - bug_record.root_cause_category gained 'core_engine' in U-034 (2026-08-02, PH5) --
   05 §5.8: self-target scenarios (`.claude/agents/*.md` identity file changes, Graph
   Engine scheduler/executor core logic changes, this schema's own changes), routed by
@@ -126,8 +128,13 @@ CREATE INDEX idx_task_project_id ON task(project_id);
 CREATE TABLE work_graph_run (
     run_id TEXT PRIMARY KEY,
     task_id TEXT NOT NULL REFERENCES task(task_id),
-    -- H (2026-08-06): see skill.owner_org_node's comment above -- same relaxation.
-    org_node TEXT NOT NULL CHECK (org_node != ''),
+    -- org_node column removed (Org Graph unification Phase 2c §2.1,
+    -- 2026-08-12, user-confirmed 2026-08-08): a merged graph's one run
+    -- genuinely spans several org_nodes (planner/coder/validator all
+    -- append their own nodes into the SAME run), so a single run-level
+    -- value was never accurate once Phase 2b/2c landed -- the real
+    -- per-node ownership already lives in work_node_execution.subagent_type
+    -- (see §1.2 of the design doc). See migrate_drop_work_graph_run_org_node.
     graph_manifest TEXT NOT NULL,
     orchestration_script_ref TEXT,
     status TEXT NOT NULL CHECK (status IN (
@@ -141,16 +148,23 @@ CREATE TABLE work_graph_run (
     -- Set together when this run was forked from a historical node instead
     -- of planned fresh -- see src/work_graph_fork.py. v1 only forks from a
     -- parent run that has already reached a terminal status, so a forked
-    -- run's (task_id, org_node) pair never collides with
-    -- ux_work_graph_run_active below (that index only covers non-terminal
-    -- statuses) -- no change needed to the index itself.
+    -- run's task_id never collides with ux_work_graph_run_active below
+    -- (that index only covers non-terminal statuses) -- no change needed
+    -- to the index itself.
     parent_run_id TEXT REFERENCES work_graph_run(run_id),
     fork_of_node_exec_id TEXT REFERENCES work_node_execution(node_exec_id)
 );
--- only one active (non-terminal) run per (task_id, org_node) at a time
-CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id, org_node)
-    WHERE status IN ('planning', 'running', 'merging', 'verifying');
+-- Org Graph unification Phase 2c §2.2 (2026-08-12): narrowed from
+-- (task_id, org_node) to plain task_id -- a merged graph means a task has
+-- ONE run spanning its whole lifecycle now, not one per org_node. 'blocked'
+-- stays counted as active on purpose (not just 'planning'/'running'/
+-- 'merging'/'verifying'/'reflecting'): resuming a blocked run appends into
+-- the SAME row (plan_work_graph(..., append_to_run_id=...)), it never
+-- INSERTs a fresh one, so a blocked run must still hold the (task_id) slot.
+-- Only 'done'/'failed' are genuinely terminal.
+CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id)
+    WHERE status NOT IN ('done', 'failed');
 -- 1.4 work_node_execution
 CREATE TABLE work_node_execution (
@@ -802,3 +816,84 @@ def migrate_add_planner_review_verdict_event_type(conn):
         """
     )
     conn.commit()
+
+
+def migrate_drop_work_graph_run_org_node(conn):
+    """Org Graph unification Phase 2c §2.1/§2.2 (2026-08-12, user-confirmed
+    2026-08-08): drops work_graph_run.org_node on an ALREADY-POPULATED
+    database and narrows ux_work_graph_run_active from (task_id, org_node)
+    to plain task_id in the SAME rebuild -- see both columns'/the index's
+    own comments on the CREATE TABLE above for why (a merged graph's one
+    run genuinely spans several org_nodes now, so neither the column nor
+    the old per-org_node uniqueness grain is accurate anymore). Both
+    changes have to happen in one rebuild since the old index is defined
+    on the column being dropped.
+
+    work_graph_run is a PARENT table (work_node_execution/artifact/
+    execution_event/bug_record/work_graph_run's own self-reference all
+    REFERENCES it) -- same PRAGMA foreign_keys=OFF/rebuild/
+    foreign_key_check/ON dance as migrate_add_reviewing_planner_output_status
+    for the exact same reason (see that function's own docstring for the
+    real-restart incident this precedent comes from); a plain executescript()
+    without toggling foreign_keys would trip FK enforcement on DROP TABLE
+    the instant any child row exists.
+
+    Depends on migrate_relax_org_node_constraints() having already run
+    against this database (called here unconditionally, it's idempotent) --
+    that migration is what first relaxed org_node's CHECK from the old
+    3-value enum; this one goes further and removes the column outright,
+    but only after that intermediate state, matching the accumulated-history
+    convention every other CHECK-enum migration in this file already follows
+    (each embeds the full column/constraint set valid as of ITS place in the
+    sequence, not retroactively edited by later migrations).
+
+    Idempotent: skipped if work_graph_run no longer has an org_node column,
+    so a fresh create_schema()'d database (already on this final shape) is a
+    no-op, same convention as every other migrate_* function here."""
+    migrate_relax_org_node_constraints(conn)
+    conn.commit()
+
+    cols = {row[1] for row in conn.execute("PRAGMA table_info(work_graph_run)").fetchall()}
+    if "org_node" not in cols:
+        return
+    conn.execute("PRAGMA foreign_keys = OFF")
+    try:
+        conn.executescript(
+            """
+            CREATE TABLE work_graph_run_new (
+                run_id TEXT PRIMARY KEY,
+                task_id TEXT NOT NULL REFERENCES task(task_id),
+                graph_manifest TEXT NOT NULL,
+                orchestration_script_ref TEXT,
+                status TEXT NOT NULL CHECK (status IN (
+                    'planning', 'running', 'merging', 'verifying',
+                    'reflecting', 'blocked', 'done', 'failed'
+                )),
+                started_at TEXT,
+                completed_at TEXT,
+                result_artifact_refs TEXT,
+                parent_run_id TEXT REFERENCES work_graph_run(run_id),
+                fork_of_node_exec_id TEXT REFERENCES work_node_execution(node_exec_id)
+            );
+            INSERT INTO work_graph_run_new (
+                run_id, task_id, graph_manifest, orchestration_script_ref, status,
+                started_at, completed_at, result_artifact_refs, parent_run_id, fork_of_node_exec_id
+            )
+            SELECT
+                run_id, task_id, graph_manifest, orchestration_script_ref, status,
+                started_at, completed_at, result_artifact_refs, parent_run_id, fork_of_node_exec_id
+            FROM work_graph_run;
+            DROP TABLE work_graph_run;
+            ALTER TABLE work_graph_run_new RENAME TO work_graph_run;
+            CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id)
+                WHERE status NOT IN ('done', 'failed');
+            """
+        )
+        conn.commit()
+        violations = conn.execute("PRAGMA foreign_key_check").fetchall()
+        if violations:
+            raise RuntimeError(
+                "migrate_drop_work_graph_run_org_node left dangling foreign keys: {}".format(violations)
+            )
+    finally:
+        conn.execute("PRAGMA foreign_keys = ON")
@@ -155,20 +155,30 @@ def get_task(conn, task_id):
     ]
     runs = conn.execute(
-        "SELECT run_id, org_node, graph_manifest, status, started_at, completed_at, "
+        "SELECT run_id, graph_manifest, status, started_at, completed_at, "
         "parent_run_id, fork_of_node_exec_id "
         "FROM work_graph_run WHERE task_id = ? ORDER BY started_at",
         (task_id,),
     ).fetchall()
-    task["work_graph_runs"] = [
-        {
-            "run_id": w[0], "org_node": w[1],
-            "graph_manifest": json.loads(w[2]) if w[2] else None,
-            "status": w[3], "started_at": w[4], "completed_at": w[5],
+    task["work_graph_runs"] = []
+    for w in runs:
+        nodes = _work_nodes_for_run(conn, w[0])
+        task["work_graph_runs"].append({
+            "run_id": w[0],
+            "graph_manifest": json.loads(w[1]) if w[1] else None,
+            "status": w[2], "started_at": w[3], "completed_at": w[4],
             # D (2026-08-06): both None for a normal run; both set when this
             # run was forked from a historical node (src/work_graph_fork.py).
-            "parent_run_id": w[6], "fork_of_node_exec_id": w[7],
-            "nodes": _work_nodes_for_run(conn, w[0]),
+            "parent_run_id": w[5], "fork_of_node_exec_id": w[6],
+            # 2026-08-12 (Org Graph unification Phase 2c §2.1.1): replaces
+            # the removed single-value work_graph_run.org_node -- a merged
+            # run genuinely spans several org_nodes now, computed from this
+            # run's own nodes (already fetched below for "nodes") rather
+            # than a second query. Same shape as work_graph_run_query.py's
+            # get_work_graph_run/list_work_graph_runs, kept in lockstep per
+            # this module's own docstring.
+            "org_nodes_involved": sorted({n["subagent_type"] for n in nodes if n["subagent_type"]}),
+            "nodes": nodes,
             # 2026-08-08 (Work Graph self-planning + loop-back): the console's
             # v_graph_visualizer needs this to actually draw a loop_back's
             # on_failure edge / a retry_self node's "sent back Nx" badge --
@@ -176,9 +186,7 @@ def get_task(conn, task_id):
             # alone (loop_back_triggered is execution_event-only by design,
             # see docs/done/work-graph-self-planning-and-loop-back-design.md §7).
             "loop_back_events": _loop_back_events_for_run(conn, w[0]),
-        }
-        for w in runs
-    ]
+        })
     return task
@@ -28,8 +28,8 @@ class ArtifactQueryTestBase(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         self.conn.commit()
@@ -35,8 +35,8 @@ class ArtifactWriteTestBase(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'planning_brd_prd_sdd', 'planner')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.commit()
@@ -1135,8 +1135,8 @@ class ExecutionLogRouteTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
         )
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -1291,8 +1291,8 @@ class ArtifactRouteTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
         )
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         with open(os.path.join(self.tmp.name, "sdd.md"), "w") as f:
             f.write("# SDD content")
@@ -1366,8 +1366,8 @@ class WorkGraphRunListRouteTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
         )
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, started_at) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'done', '2026-08-01T00:00:00')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, started_at) "
+            "VALUES ('run-1', 'task-1', '{}', 'done', '2026-08-01T00:00:00')"
         )
         conn.commit()
@@ -1403,6 +1403,11 @@ class WorkGraphRunListRouteTest(unittest.TestCase):
         self.assertEqual(body["error"], "INVALID_PARAM")
     def test_list_org_node_filter(self):
+        # 2026-08-12 (Org Graph unification Phase 2c §2.1.1): org_node now
+        # filters by whether any node in the run has that subagent_type --
+        # this fixture's run-1 has zero nodes, so ANY org_node filter
+        # (including one that would have matched the old run-level column)
+        # now excludes it.
         status, body = self._get("/api/work-graph-runs?project_id=proj-1&org_node=planner")
         self.assertEqual(status, 200)
         self.assertEqual(body, [])
@@ -1412,7 +1417,10 @@ class WorkGraphRunListRouteTest(unittest.TestCase):
         self.assertEqual(status, 200)
         self.assertEqual(body["run_id"], "run-1")
         self.assertEqual(body["task_id"], "task-1")
-        self.assertEqual(body["org_node"], "coder")
+        # run-1 has zero nodes seeded above -- org_nodes_involved is derived
+        # from work_node_execution.subagent_type (Phase 2c §2.1.1), so it's
+        # empty here, not the old run-level org_node column's fixed value.
+        self.assertEqual(body["org_nodes_involved"], [])
         self.assertEqual(body["nodes"], [])
     def test_get_nonexistent_run_returns_404(self):
@@ -1444,8 +1452,8 @@ class WorkNodeForkRouteTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'validating', 'planner')"
         )
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -218,8 +218,8 @@ class WriteCompletedNodeArtifactsTest(PipelineTestBase):
             "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', 'coding', 'coder')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -330,9 +330,9 @@ class PlanDocumentsTest(PipelineTestBase):
         # plan_work_graph's own real work_graph_run INSERT actually ran (not bypassed)
         run_row = self.conn.execute(
-            "SELECT org_node, status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
+            "SELECT status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
         ).fetchone()
-        self.assertEqual(run_row, ("planner", "planning"))
+        self.assertEqual(run_row, ("planning",))
     def test_earlier_nodes_survive_a_later_node_failing_verification(self):
         # 2026-08-07 real incident (docs/todo/work-graph-partial-artifact-
@@ -441,17 +441,14 @@ class RunEndToEndTest(PipelineTestBase):
         # 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).
+        # work_graph_run row exists for this task now. org_node column
+        # removed entirely (§2.1) -- real per-node ownership lives in
+        # work_node_execution.subagent_type, checked below via node_ids.
         run_rows = self.conn.execute(
-            "SELECT run_id, org_node, status FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
+            "SELECT run_id, status FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
         ).fetchall()
         self.assertEqual(len(run_rows), 1)
-        self.assertEqual(run_rows[0][1], "planner")
-        self.assertEqual(run_rows[0][2], "done")
+        self.assertEqual(run_rows[0][1], "done")
         # Coder's "impl" AND Validator's "verdict" nodes both genuinely
         # landed in the SAME run as Planner's own nodes.
@@ -558,8 +555,8 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase):
     def test_pass_on_first_attempt_reuses_existing_artifacts_no_replan(self):
         for artifact_type, content in (("sdd", "# SDD\nreal"), ("brd", "# BRD\nreal")):
             self.conn.execute(
-                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-                "VALUES ('run-1', ?, 'planner', '{}', 'done') ON CONFLICT(run_id) DO NOTHING",
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('run-1', ?, '{}', 'done') ON CONFLICT(run_id) DO NOTHING",
                 (self.task_id,),
             )
             file_path = ".graphlemon/runs/{}/{}.md".format(self.task_id, artifact_type)
@@ -1093,8 +1090,8 @@ class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
         # 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')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-shared', 'task-1', '{\"nodes\": [], \"edges\": []}', 'running')"
         )
         self.conn.commit()
@@ -29,8 +29,8 @@ class ExecutionLogTestBase(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'coding', 'coder')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -28,8 +28,8 @@ class LifecycleTestBase(unittest.TestCase):
     def _pass_artifact(self, artifact_type, run_id="run-1"):
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, 'planner', '{}', 'done') ON CONFLICT(run_id) DO NOTHING",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES (?, ?, '{}', 'done') ON CONFLICT(run_id) DO NOTHING",
             (run_id, self.task_id),
         )
         self.conn.execute(
@@ -49,8 +49,7 @@ class RealChainedTriggerTest(unittest.TestCase):
     def _seed_passing_artifacts(conn, task_id, artifact_types):
         run_id = str(uuid.uuid4())
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, 'planner', '{}', 'done')",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, '{}', 'done')",
             (run_id, task_id),
         )
         for artifact_type in artifact_types:
@@ -91,11 +91,10 @@ class ConstraintTestBase(unittest.TestCase):
         self.conn.commit()
         return skill_id
-    def insert_work_graph_run(self, run_id="run-1", task_id="task-1", org_node="coder", status="planning"):
+    def insert_work_graph_run(self, run_id="run-1", task_id="task-1", status="planning"):
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, ?, '{}', ?)",
-            (run_id, task_id, org_node, status),
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, '{}', ?)",
+            (run_id, task_id, status),
         )
         self.conn.commit()
         return run_id
@@ -167,24 +166,41 @@ class ForeignKeyEnforcementTest(ConstraintTestBase):
 class WorkGraphRunActiveUniqueIndexTest(ConstraintTestBase):
-    """01-M 1.3: only one active (planning/running/merging/verifying) run per
-    (task_id, org_node) at a time -- terminal-status runs don't count."""
+    """Org Graph unification Phase 2c §2.2 (2026-08-12): only one active run
+    per task_id at a time now (narrowed from (task_id, org_node) -- a merged
+    graph means one task has ONE run spanning its whole lifecycle, not one
+    per org_node). Active = any status except 'done'/'failed' -- 'blocked'
+    now counts as active on purpose (a blocked run is resumed by appending
+    into the SAME row, never by inserting a fresh one)."""
     def setUp(self):
         super().setUp()
         self.insert_project()
         self.insert_task()
-    def test_second_active_run_for_same_task_and_org_node_rejected(self):
-        self.insert_work_graph_run("run-1", "task-1", "coder", status="running")
+    def test_second_active_run_for_same_task_rejected(self):
+        self.insert_work_graph_run("run-1", "task-1", status="running")
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.insert_work_graph_run("run-2", "task-1", status="planning")
+
+    def test_blocked_run_still_counts_as_active(self):
+        self.insert_work_graph_run("run-1", "task-1", status="blocked")
         with self.assertRaises(sqlite3.IntegrityError):
-            self.insert_work_graph_run("run-2", "task-1", "coder", status="planning")
+            self.insert_work_graph_run("run-2", "task-1", status="planning")
+
+    def test_second_run_allowed_once_first_is_done(self):
+        self.insert_work_graph_run("run-1", "task-1", status="done")
+        self.insert_work_graph_run("run-2", "task-1", status="planning")
+        count = self.conn.execute(
+            "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'"
+        ).fetchone()[0]
+        self.assertEqual(count, 2)
-    def test_second_run_allowed_once_first_is_terminal(self):
-        self.insert_work_graph_run("run-1", "task-1", "coder", status="done")
-        self.insert_work_graph_run("run-2", "task-1", "coder", status="planning")
+    def test_second_run_allowed_once_first_is_failed(self):
+        self.insert_work_graph_run("run-1", "task-1", status="failed")
+        self.insert_work_graph_run("run-2", "task-1", status="planning")
         count = self.conn.execute(
-            "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1' AND org_node = 'coder'"
+            "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'"
         ).fetchone()[0]
         self.assertEqual(count, 2)
@@ -247,8 +263,8 @@ class MigrateExpandArtifactTypesTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'planning')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'planning')"
         )
         self.conn.execute(
             "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, "
@@ -309,8 +325,8 @@ class MigrateExpandArtifactTypesTest(unittest.TestCase):
                 "VALUES ('t', 'p', 'g', 'goal', 'queued', 'planner')"
             )
             fresh.execute(
-                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-                "VALUES ('r', 't', 'planner', '{}', 'planning')"
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('r', 't', '{}', 'planning')"
             )
             fresh.execute(
                 "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, "
@@ -519,8 +535,8 @@ class MigrateAddWorkGraphForkColumnsTest(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         self.conn.commit()
@@ -552,6 +568,28 @@ class MigrateAddWorkGraphForkColumnsTest(unittest.TestCase):
             fresh.close()
+_OLD_WORK_GRAPH_RUN_BLOCK = """-- 1.3 work_graph_run
+CREATE TABLE work_graph_run (
+    run_id TEXT PRIMARY KEY,
+    task_id TEXT NOT NULL REFERENCES task(task_id),
+    org_node TEXT NOT NULL CHECK (org_node IN ('planner', 'coder', 'validator')),
+    graph_manifest TEXT NOT NULL,
+    orchestration_script_ref TEXT,
+    status TEXT NOT NULL CHECK (status IN (
+        'planning', 'running', 'merging', 'verifying',
+        'reflecting', 'blocked', 'done', 'failed'
+    )),
+    started_at TEXT,
+    completed_at TEXT,
+    result_artifact_refs TEXT
+);
+
+CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id, org_node)
+    WHERE status IN ('planning', 'running', 'merging', 'verifying');
+
+"""
+
+
 def _old_org_node_enum_schema_sql():
     """H (2026-08-06): simulates a pre-migration database whose 4
     org_node-ish columns still have the old CHECK(x IN ('planner','coder',
@@ -559,9 +597,22 @@ def _old_org_node_enum_schema_sql():
     too (migrate_relax_org_node_constraints depends on
     migrate_add_work_graph_fork_columns having run, so the test fixture
     should start from a database that's had NEITHER migration, same as a
-    real old production db would be)."""
-    sql = schema.SCHEMA_SQL
-    for column in ("owner_org_node", "assigned_org_node", "org_node"):
+    real old production db would be).
+
+    2026-08-12 (Org Graph unification Phase 2c §2.1): work_graph_run.org_node
+    no longer exists in schema.SCHEMA_SQL at all (removed outright, not just
+    relaxed -- see migrate_drop_work_graph_run_org_node), so this function
+    can no longer derive the 'old' work_graph_run shape via regex
+    substitution the way it still does for owner_org_node/assigned_org_node/
+    agent_profile.org_node below (those 3 are unaffected by Phase 2c, still
+    present in SCHEMA_SQL). The 1.3 section is spliced out and replaced with
+    a hardcoded legacy block instead -- the historical shape this test
+    exercises no longer has a live counterpart to derive from."""
+    before, _, rest = schema.SCHEMA_SQL.partition("-- 1.3 work_graph_run")
+    _, _, after = rest.partition("-- 1.4 work_node_execution")
+    sql = before + _OLD_WORK_GRAPH_RUN_BLOCK + "-- 1.4 work_node_execution" + after
+
+    for column in ("owner_org_node", "assigned_org_node"):
         sql = re.sub(
             r"(?:    --[^\n]*\n)*    " + column + r" TEXT( NOT NULL)? CHECK \(" + column + r" != ''\)",
             column + r" TEXT\1 CHECK (" + column + " IN ('planner', 'coder', 'validator'))",
@@ -572,12 +623,6 @@ def _old_org_node_enum_schema_sql():
         "org_node TEXT PRIMARY KEY CHECK (org_node IN ('planner', 'coder', 'validator'))",
         sql,
     )
-    sql = re.sub(
-        r",\n(    --[^\n]*\n)*    parent_run_id TEXT REFERENCES work_graph_run\(run_id\),\n"
-        r"    fork_of_node_exec_id TEXT REFERENCES work_node_execution\(node_exec_id\)\n\);",
-        "\n);",
-        sql,
-    )
     return sql
@@ -739,6 +784,138 @@ class MigrateRelaxOrgNodeConstraintsTest(unittest.TestCase):
             fresh.close()
+class MigrateDropWorkGraphRunOrgNodeTest(unittest.TestCase):
+    """Org Graph unification Phase 2c §2.1/§2.2 (2026-08-12): real upgrade
+    path for an already-populated database whose work_graph_run still has
+    the org_node column (with the OLDEST possible starting shape -- the
+    strict 3-value enum, same fixture as MigrateRelaxOrgNodeConstraintsTest
+    -- to also exercise this migration's own internal dependency call to
+    migrate_relax_org_node_constraints())."""
+
+    def setUp(self):
+        old_sql = _old_org_node_enum_schema_sql()
+        self.conn = schema.connect(":memory:")
+        self.conn.execute("PRAGMA foreign_keys = ON")
+        self.conn.executescript(old_sql)
+        self.conn.commit()
+        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', 'g', 'goal', 'queued', 'planner')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+        )
+        self.conn.execute(
+            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+            "VALUES ('skill-1', 'planner', 'skills/x/', 'sha', 'new', 'active')"
+        )
+        self.conn.commit()
+
+    def tearDown(self):
+        self.conn.close()
+
+    def test_org_node_column_actually_dropped(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        cols = {row[1] for row in self.conn.execute("PRAGMA table_info(work_graph_run)").fetchall()}
+        self.assertNotIn("org_node", cols)
+
+    def test_migration_preserves_existing_rows(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.assertEqual(
+            self.conn.execute(
+                "SELECT run_id, task_id, status FROM work_graph_run WHERE run_id = 'run-1'"
+            ).fetchone(),
+            ("run-1", "task-1", "done"),
+        )
+
+    def test_ux_work_graph_run_active_narrowed_to_task_id(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-2', 'task-1', '{}', 'running')"
+        )
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.conn.execute(
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('run-3', 'task-1', '{}', 'running')"
+            )
+
+    def test_blocked_run_still_counts_as_active_after_migration(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-2', 'task-1', '{}', 'blocked')"
+        )
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.conn.execute(
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('run-3', 'task-1', '{}', 'planning')"
+            )
+
+    def test_second_run_allowed_once_first_is_done(self):
+        # run-1 (seeded in setUp) is already status='done' -- a second run
+        # for the SAME task_id must be allowed.
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-2', 'task-1', '{}', 'planning')"
+        )
+        self.conn.commit()
+        count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(count, 2)
+
+    def test_foreign_key_relationships_survive_rebuild(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.conn.execute(
+            "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+            "input_schema_ref, output_schema_ref, input_payload, status) "
+            "VALUES ('node-1', 'run-1', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', 'done')"
+        )
+        self.conn.commit()
+        with self.assertRaises(sqlite3.IntegrityError):
+            self.conn.execute(
+                "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+                "input_schema_ref, output_schema_ref, input_payload, status) "
+                "VALUES ('node-2', 'no-such-run', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', 'done')"
+            )
+
+    def test_idempotent_on_a_database_already_migrated(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        schema.migrate_drop_work_graph_run_org_node(self.conn)  # must not raise
+        count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run").fetchone()[0]
+        self.assertEqual(count, 1)
+
+    def test_foreign_keys_pragma_restored_after_migration(self):
+        schema.migrate_drop_work_graph_run_org_node(self.conn)
+        self.assertEqual(self.conn.execute("PRAGMA foreign_keys").fetchone()[0], 1)
+
+    def test_noop_on_a_freshly_created_database(self):
+        fresh = schema.connect(":memory:")
+        try:
+            schema.create_schema(fresh)
+            schema.migrate_drop_work_graph_run_org_node(fresh)  # must not raise
+            fresh.execute(
+                "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+                "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+            )
+            fresh.execute(
+                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+                "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')"
+            )
+            fresh.execute(
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('run-1', 'task-1', '{}', 'planning')"
+            )
+            fresh.commit()
+        finally:
+            fresh.close()
+
+
 class MigrateAddWorkGraphLoopLimitConfigTest(unittest.TestCase):
     """2026-08-08: real upgrade path for an already-populated database that
     predates work_graph_loop_limit_config (plain CREATE TABLE IF NOT
@@ -880,8 +1057,8 @@ class MigrateAddReviewingPlannerOutputStatusTest(unittest.TestCase):
         # every test below previously ran against a task row with ZERO
         # referencing children, so FK enforcement was never truly exercised.
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         self.conn.execute(
             "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, "
@@ -128,8 +128,8 @@ class TaskExecutionTestBase(unittest.TestCase):
             "edges": [],
         })
         self.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, manifest),
+            "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES (?, ?, ?, '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
@@ -527,8 +527,8 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase):
             "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, manifest),
+            "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES (?, ?, ?, 'done')", (run_id, task_id, manifest),
         )
         work_graph_orchestrate.ensure_placeholder_skill(conn, "planner-doc-writing", "planner")
         conn.execute(
@@ -94,8 +94,7 @@ class GetTaskTest(TaskQueryTestBase):
     def test_includes_artifacts(self):
         self._insert_task("task-1")
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         self.conn.execute(
             "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, file_path, "
@@ -121,8 +120,8 @@ class GetTaskTest(TaskQueryTestBase):
         self._insert_task("task-1")
         manifest = {"nodes": [{"node_id": "n1", "artifact_type": "brd", "depends_on": []}], "edges": []}
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', ?, 'running')",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', ?, 'running')",
             (json.dumps(manifest),),
         )
         self.conn.execute(
@@ -152,8 +151,8 @@ class GetTaskTest(TaskQueryTestBase):
         # from graph_manifest or work_node_execution alone.
         self._insert_task("task-1")
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.execute(
             "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) "
@@ -177,8 +176,7 @@ class GetTaskTest(TaskQueryTestBase):
     def test_no_loop_back_events_is_an_empty_list_not_missing_key(self):
         self._insert_task("task-1")
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES ('run-1', 'task-1', '{}', 'done')"
         )
         self.conn.commit()
         result = task_query.get_task(self.conn, "task-1")
@@ -189,8 +187,8 @@ class GetTaskTest(TaskQueryTestBase):
         # apart from the original run it descends from.
         self._insert_task("task-1")
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-orig', 'task-1', 'planner', '{}', 'done')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-orig', 'task-1', '{}', 'done')"
         )
         self.conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -202,9 +200,9 @@ class GetTaskTest(TaskQueryTestBase):
             "VALUES ('node-1', 'run-orig', 'n1', 'planner', 'skill-1', 'in.json', 'out.json', '{}', '{}', 'done')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, "
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, "
             "parent_run_id, fork_of_node_exec_id) "
-            "VALUES ('run-fork', 'task-1', 'planner', '{}', 'done', 'run-orig', 'node-1')"
+            "VALUES ('run-fork', 'task-1', '{}', 'done', 'run-orig', 'node-1')"
         )
         self.conn.commit()
@@ -219,8 +217,8 @@ class GetTaskTest(TaskQueryTestBase):
         # quota_limit_type + how long it's been waiting (02-V §2.7)
         self._insert_task("task-1")
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -64,9 +64,8 @@ class ForkTestBase(unittest.TestCase):
         row plus one 'done' work_node_execution row per node, each holding
         contents[node_id] as its output_payload."""
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, 'task-1', ?, ?, ?)",
-            (run_id, org_node, json.dumps(graph_manifest), status),
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, 'task-1', ?, ?)",
+            (run_id, json.dumps(graph_manifest), status),
         )
         node_exec_ids = {}
         for node in graph_manifest["nodes"]:
@@ -254,8 +253,8 @@ class MultipleForksFromSameNodeTest(ForkTestBase):
 class ForkValidationErrorsTest(ForkTestBase):
     def test_forking_a_non_done_node_raises(self):
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', ?, 'running')",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', ?, 'running')",
             (json.dumps(_BRD_PRD_SDD_BENCHMARK_MANIFEST),),
         )
         self.conn.execute(
@@ -321,8 +320,8 @@ class StartForkInBackgroundTest(unittest.TestCase):
             "VALUES ('planner-doc-writing', 'planner', 'skills/planner-doc-writing/', 'unknown', 'new', 'active')"
         )
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-orig', 'task-1', 'planner', ?, 'done')",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-orig', 'task-1', ?, 'done')",
             (json.dumps(_BRD_PRD_SDD_BENCHMARK_MANIFEST),),
         )
         contents = {
@@ -234,8 +234,8 @@ class RunGraphTestBase(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.commit()
@@ -829,8 +829,8 @@ class RunGraphRealSqliteConcurrencyTest(unittest.TestCase):
                 "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')"
             )
             conn.execute(
-                "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-                "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')"
+                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+                "VALUES ('run-1', 'task-1', '{}', 'running')"
             )
             conn.commit()
@@ -308,13 +308,12 @@ class HappyPathTest(PlanWorkGraphTestBase):
                 self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
             )
         row = self.conn.execute(
-            "SELECT task_id, org_node, status, graph_manifest FROM work_graph_run WHERE run_id = ?",
+            "SELECT task_id, status, graph_manifest FROM work_graph_run WHERE run_id = ?",
             (result["run_id"],),
         ).fetchone()
         self.assertEqual(row[0], "task-1")
-        self.assertEqual(row[1], "planner")
-        self.assertEqual(row[2], "planning")
-        self.assertEqual(json.loads(row[3]), _MANIFEST)
+        self.assertEqual(row[1], "planning")
+        self.assertEqual(json.loads(row[2]), _MANIFEST)
     def test_invokes_claude_with_expected_args(self):
         with patch("src.work_graph_plan.subprocess.run") as run:
@@ -448,8 +447,7 @@ class AppendToRunIdTest(PlanWorkGraphTestBase):
     def _seed_run(self, run_id, manifest):
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, 'task-1', 'planner', ?, 'running')",
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, 'task-1', ?, 'running')",
             (run_id, json.dumps(manifest)),
         )
         self.conn.commit()
@@ -29,24 +29,31 @@ class WorkGraphRunQueryTestBase(unittest.TestCase):
     def _insert_run(self, run_id, org_node, status="done", started_at="2026-08-01T00:00:00",
                      completed_at=None, task_id="task-1", node_count=0,
                      parent_run_id=None, fork_of_node_exec_id=None):
+        """2026-08-12 (Org Graph unification Phase 2c §2.1.1):
+        work_graph_run.org_node no longer exists -- `org_node` here now
+        means "the subagent_type to stamp this run's seeded nodes with"
+        (org_nodes_involved is derived from work_node_execution.
+        subagent_type, not a run-level column). A run with node_count=0
+        genuinely has no org_nodes_involved at all now; tests that care
+        about org_node-based filtering/listing must pass node_count >= 1."""
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, "
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, "
             "started_at, completed_at, parent_run_id, fork_of_node_exec_id) "
-            "VALUES (?, ?, ?, '{}', ?, ?, ?, ?, ?)",
-            (run_id, task_id, org_node, status, started_at, completed_at, parent_run_id, fork_of_node_exec_id),
+            "VALUES (?, ?, '{}', ?, ?, ?, ?, ?)",
+            (run_id, task_id, status, started_at, completed_at, parent_run_id, fork_of_node_exec_id),
         )
         for i in range(node_count):
             node_exec_id = "{}-node-{}".format(run_id, i)
             self.conn.execute(
                 "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
-                "VALUES (?, 'coder', 'skills/x.md', 'sha1', 'new', 'active')",
-                ("skill-{}".format(node_exec_id),),
+                "VALUES (?, ?, 'skills/x.md', 'sha1', 'new', 'active')",
+                ("skill-{}".format(node_exec_id), org_node),
             )
             self.conn.execute(
                 "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
                 "input_schema_ref, output_schema_ref, input_payload, status) "
-                "VALUES (?, ?, ?, 'coder', ?, 'schema', 'schema', '{}', 'done')",
-                (node_exec_id, run_id, "node-{}".format(i), "skill-{}".format(node_exec_id)),
+                "VALUES (?, ?, ?, ?, ?, 'schema', 'schema', '{}', 'done')",
+                (node_exec_id, run_id, "node-{}".format(i), org_node, "skill-{}".format(node_exec_id)),
             )
         self.conn.commit()
@@ -87,12 +94,30 @@ class ListWorkGraphRunsTest(WorkGraphRunQueryTestBase):
         self.assertEqual([r["run_id"] for r in result], ["run-1"])
     def test_org_node_filter(self):
-        self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
-        self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00")
-        self._insert_run("run-3", "coder", started_at="2026-08-03T00:00:00")
+        self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00", node_count=1)
+        self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00", node_count=1)
+        self._insert_run("run-3", "coder", started_at="2026-08-03T00:00:00", node_count=1)
         result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1", org_node="coder")
         self.assertEqual({r["run_id"] for r in result}, {"run-2", "run-3"})
+    def test_org_nodes_involved_reflects_merged_run_nodes(self):
+        # Org Graph unification Phase 2c §2.1.1: a merged run's nodes can
+        # span several org_nodes -- this is the whole reason the old
+        # single-value org_node column was removed.
+        self._insert_run("run-1", "planner", node_count=1)
+        self.conn.execute(
+            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+            "VALUES ('skill-extra', 'coder', 'skills/x.md', 'sha1', 'new', 'active')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
+            "input_schema_ref, output_schema_ref, input_payload, status) "
+            "VALUES ('run-1-extra', 'run-1', 'extra', 'coder', 'skill-extra', 'schema', 'schema', '{}', 'done')"
+        )
+        self.conn.commit()
+        result = work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")
+        self.assertEqual(sorted(result[0]["org_nodes_involved"]), ["coder", "planner"])
+
     def test_limit_one_returns_latest(self):
         self._insert_run("run-1", "planner", started_at="2026-08-01T00:00:00")
         self._insert_run("run-2", "coder", started_at="2026-08-02T00:00:00")
@@ -128,7 +153,7 @@ class GetWorkGraphRunTest(WorkGraphRunQueryTestBase):
         result = work_graph_run_query.get_work_graph_run(self.conn, "run-1")
         self.assertEqual(result["run_id"], "run-1")
         self.assertEqual(result["task_id"], "task-1")
-        self.assertEqual(result["org_node"], "coder")
+        self.assertEqual(result["org_nodes_involved"], ["coder"])
         self.assertEqual(result["status"], "running")
         self.assertEqual(result["graph_manifest"], {})
         self.assertEqual(len(result["nodes"]), 2)
@@ -49,8 +49,8 @@ class WorkNodeRunTestBase(unittest.TestCase):
             "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'coder')"
         )
         self.conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', '{}', 'running')"
         )
         self.conn.execute(
             "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
@@ -91,7 +91,7 @@ def _downstream_node_ids(graph_manifest, start_node_id):
     return seen
-def _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest):
+def _reconstruct_build_prompt(conn, task_id, graph_manifest):
     """Rebuilds a build_prompt(node, outputs) usable for the re-executed
     downstream subgraph, purely from persisted state -- no access to the
     original call site's Python closures. Two supported node shapes
@@ -110,11 +110,12 @@ def _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest):
     (work_graph_plan.dispatch_build_prompt) instead of once for the whole
     graph -- every pre-Phase-2 graph_manifest still has just one shape
     throughout, so this is a strict generalization, not a behavior change
-    for anything that isn't a merged graph. `org_node` is unused by this
-    per-node dispatch itself (kept as a parameter -- callers still pass it
-    for other purposes, e.g. run_graph()'s own org_node argument;
-    removing it from work_graph_run/this signature is Phase 2c's job
-    alongside the actual org_node column removal, not this fix's)."""
+    for anything that isn't a merged graph.
+
+    2026-08-12 (Phase 2c §2.1.2): `org_node` parameter dropped -- it was
+    already unused by this per-node dispatch (kept only because Phase 2b
+    hadn't yet done the org_node column removal this fix was explicitly
+    deferring to)."""
     nodes = graph_manifest["nodes"]
     if not nodes:
         raise ForkGraphShapeUnsupportedError("graph_manifest has no nodes to reconstruct a build_prompt for")
@@ -147,7 +148,7 @@ def create_fork(conn, source_node_exec_id, new_output_payload):
     uses for the manual Start button."""
     row = conn.execute(
         "SELECT wne.run_id, wne.node_id, wne.status, wne.subagent_type, wne.skill_ref, "
-        "wgr.task_id, wgr.org_node, wgr.graph_manifest, wgr.status "
+        "wgr.task_id, wgr.graph_manifest, wgr.status "
         "FROM work_node_execution wne JOIN work_graph_run wgr ON wgr.run_id = wne.run_id "
         "WHERE wne.node_exec_id = ?",
         (source_node_exec_id,),
@@ -156,7 +157,7 @@ def create_fork(conn, source_node_exec_id, new_output_payload):
         raise ValueError("no such work_node_execution: " + source_node_exec_id)
     (
         source_run_id, source_node_id, source_status, subagent_type, skill_ref,
-        task_id, org_node, graph_manifest_json, parent_run_status,
+        task_id, graph_manifest_json, parent_run_status,
     ) = row
     if source_status != "done":
         raise ForkSourceNotDoneError(
@@ -173,7 +174,7 @@ def create_fork(conn, source_node_exec_id, new_output_payload):
         )
     graph_manifest = json.loads(graph_manifest_json)
-    build_prompt = _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest)
+    build_prompt = _reconstruct_build_prompt(conn, task_id, graph_manifest)
     downstream = _downstream_node_ids(graph_manifest, source_node_id) - {source_node_id}
     original_outputs = {
@@ -184,10 +185,13 @@ def create_fork(conn, source_node_exec_id, new_output_payload):
     }
     new_run_id = str(uuid.uuid4())
+    # org_node column removed (Org Graph unification Phase 2c §2.1,
+    # 2026-08-12) -- see work_graph_plan.py's own INSERT for the general
+    # rationale.
     conn.execute(
-        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, started_at, "
-        "parent_run_id, fork_of_node_exec_id) VALUES (?, ?, ?, ?, 'running', CURRENT_TIMESTAMP, ?, ?)",
-        (new_run_id, task_id, org_node, graph_manifest_json, source_run_id, source_node_exec_id),
+        "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status, started_at, "
+        "parent_run_id, fork_of_node_exec_id) VALUES (?, ?, ?, 'running', CURRENT_TIMESTAMP, ?, ?)",
+        (new_run_id, task_id, graph_manifest_json, source_run_id, source_node_exec_id),
     )
     conn.commit()
@@ -209,7 +213,12 @@ def create_fork(conn, source_node_exec_id, new_output_payload):
     conn.commit()
     return {
-        "run_id": new_run_id, "org_node": org_node, "skill_ref": skill_ref,
+        # "org_node" key renamed to "subagent_type" (Phase 2c §2.1.2): the
+        # source node's own execution identity, not a run-level org_node --
+        # execute_fork() passes this straight through as run_graph()'s
+        # org_node argument (its default for nodes with no per-node
+        # subagent_type override, see work_graph_orchestrate.py's own §4.1).
+        "run_id": new_run_id, "subagent_type": subagent_type, "skill_ref": skill_ref,
         "graph_manifest": graph_manifest, "build_prompt": build_prompt,
     }
@@ -228,7 +237,7 @@ def execute_fork(conn, fork_info, target_folder_path, timeout=600, sleep=time.sl
     run_id = fork_info["run_id"]
     try:
         outputs = work_graph_orchestrate.run_graph(
-            conn, run_id, fork_info["org_node"], fork_info["skill_ref"], target_folder_path,
+            conn, run_id, fork_info["subagent_type"], fork_info["skill_ref"], target_folder_path,
             fork_info["graph_manifest"], fork_info["build_prompt"], _FORK_INPUT_SCHEMA, _FORK_OUTPUT_SCHEMA,
             timeout=timeout, sleep=sleep, now=now,
         )
@@ -566,10 +566,15 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time
     if existing_manifest is None:
         run_id = str(uuid.uuid4())
+        # org_node column removed (Org Graph unification Phase 2c §2.1,
+        # 2026-08-12) -- this run's real per-node ownership lives in each
+        # node's own work_node_execution.subagent_type (set per node in
+        # _run_one_node()), not a single run-level value; a merged run
+        # genuinely spans several org_nodes by design (Coder/Validator
+        # append their own nodes into this SAME run via append_to_run_id).
         conn.execute(
-            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, ?, ?, 'planning')",
-            (run_id, task_id, org_node, json.dumps(new_manifest)),
+            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, ?, 'planning')",
+            (run_id, task_id, json.dumps(new_manifest)),
         )
         conn.commit()
         return {"run_id": run_id, "graph_manifest": new_manifest, "new_node_ids": []}
@@ -38,6 +38,15 @@ class InvalidParamError(Exception):
 def list_work_graph_runs(conn, project_id, org_node=None, limit=20):
+    """`org_node` (2026-08-12, Org Graph unification Phase 2c §2.1.1):
+    work_graph_run.org_node no longer exists (a merged run genuinely spans
+    several org_nodes) -- this filter now means "runs with at least one
+    node whose subagent_type matches", via an EXISTS subquery against
+    work_node_execution instead of a plain column match. Each returned row
+    carries "org_nodes_involved" (a list, via GROUP_CONCAT DISTINCT over
+    the same table) instead of the old single-value "org_node" -- Phase 3's
+    Org-level meta-graph (FR-5a) needs exactly this "which stages does this
+    run span" signal anyway."""
     if not project_id:
         raise InvalidParamError("project_id is required")
     if org_node is not None and org_node not in _VALID_ORG_NODES:
@@ -52,14 +61,19 @@ def list_work_graph_runs(conn, project_id, org_node=None, limit=20):
         raise InvalidParamError("limit must be in [1, 100], got {!r}".format(limit))
     query = (
-        "SELECT wgr.run_id, wgr.task_id, wgr.org_node, wgr.status, wgr.started_at, wgr.completed_at, "
+        "SELECT wgr.run_id, wgr.task_id, wgr.status, wgr.started_at, wgr.completed_at, "
         "(SELECT COUNT(*) FROM work_node_execution wne WHERE wne.run_id = wgr.run_id) AS node_count, "
+        "(SELECT GROUP_CONCAT(DISTINCT wne.subagent_type) FROM work_node_execution wne "
+        " WHERE wne.run_id = wgr.run_id) AS org_nodes_involved, "
         "wgr.parent_run_id, wgr.fork_of_node_exec_id "
         "FROM work_graph_run wgr JOIN task t ON t.task_id = wgr.task_id WHERE t.project_id = ?"
     )
     params = [project_id]
     if org_node:
-        query += " AND wgr.org_node = ?"
+        query += (
+            " AND EXISTS (SELECT 1 FROM work_node_execution wne "
+            "WHERE wne.run_id = wgr.run_id AND wne.subagent_type = ?)"
+        )
         params.append(org_node)
     query += " ORDER BY wgr.started_at DESC LIMIT ?"
     params.append(limit)
@@ -67,8 +81,9 @@ def list_work_graph_runs(conn, project_id, org_node=None, limit=20):
     rows = conn.execute(query, params).fetchall()
     return [
         {
-            "run_id": r[0], "task_id": r[1], "org_node": r[2], "status": r[3],
-            "started_at": r[4], "completed_at": r[5], "node_count": r[6],
+            "run_id": r[0], "task_id": r[1], "status": r[2],
+            "started_at": r[3], "completed_at": r[4], "node_count": r[5],
+            "org_nodes_involved": r[6].split(",") if r[6] else [],
             # D (2026-08-06): both None for a normal run; both set when this
             # run was forked from a historical node (src/work_graph_fork.py).
             "parent_run_id": r[7], "fork_of_node_exec_id": r[8],
@@ -82,20 +97,27 @@ class WorkGraphRunNotFoundError(Exception):
 def get_work_graph_run(conn, run_id):
+    """`org_nodes_involved` (2026-08-12, Phase 2c §2.1.1): same replacement
+    for the removed single-value `org_node` as list_work_graph_runs above
+    -- computed from this one run's own nodes (already fetched via
+    _nodes_for_run below) rather than a second query."""
     row = conn.execute(
-        "SELECT run_id, task_id, org_node, graph_manifest, status, started_at, completed_at, "
+        "SELECT run_id, task_id, graph_manifest, status, started_at, completed_at, "
         "parent_run_id, fork_of_node_exec_id "
         "FROM work_graph_run WHERE run_id = ?",
         (run_id,),
     ).fetchone()
     if row is None:
         raise WorkGraphRunNotFoundError(run_id)
+    nodes = _nodes_for_run(conn, row[0])
+    org_nodes_involved = sorted({n["subagent_type"] for n in nodes if n["subagent_type"]})
     return {
-        "run_id": row[0], "task_id": row[1], "org_node": row[2],
-        "graph_manifest": json.loads(row[3]) if row[3] else None,
-        "status": row[4], "started_at": row[5], "completed_at": row[6],
-        "parent_run_id": row[7], "fork_of_node_exec_id": row[8],
-        "nodes": _nodes_for_run(conn, row[0]),
+        "run_id": row[0], "task_id": row[1],
+        "graph_manifest": json.loads(row[2]) if row[2] else None,
+        "status": row[3], "started_at": row[4], "completed_at": row[5],
+        "parent_run_id": row[6], "fork_of_node_exec_id": row[7],
+        "org_nodes_involved": org_nodes_involved,
+        "nodes": nodes,
     }