From 41a0492e632f49f04fea103612071e4d3d963018 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Thu, 13 Aug 2026 00:56:34 +0800 Subject: [PATCH] Org Graph unification Phase 2c §2.1/§2.2: drop work_graph_run.org_node --- docs/todo/org-graph-unification-phase2-universal-self-planning-design.md | 4 +++- frontend/app.js | 33 +++++++++++++++++++++++++-------- frontend/test_org_click_through_smoke.js | 10 +++++----- src/backend_server.py | 1 + src/root_cause_trace.py | 17 ++++++++++++++--- src/scheduler.py | 17 ++++++++++++++++- src/schema.py | 113 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- src/task_query.py | 30 +++++++++++++++++++----------- src/test_artifact_query.py | 4 ++-- src/test_artifact_write.py | 4 ++-- src/test_backend_server.py | 26 +++++++++++++++++--------- src/test_e2e_pipeline.py | 29 +++++++++++++---------------- src/test_execution_log.py | 4 ++-- src/test_org_node_lifecycle.py | 4 ++-- src/test_org_node_lifecycle_e2e.py | 3 +-- src/test_schema.py | 237 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------ src/test_task_execution.py | 8 ++++---- src/test_task_query.py | 26 ++++++++++++-------------- src/test_work_graph_fork.py | 13 ++++++------- src/test_work_graph_orchestrate.py | 8 ++++---- src/test_work_graph_plan.py | 10 ++++------ src/test_work_graph_run_query.py | 47 ++++++++++++++++++++++++++++++++++++----------- src/test_work_node_run.py | 4 ++-- src/work_graph_fork.py | 37 +++++++++++++++++++++++-------------- src/work_graph_plan.py | 11 ++++++++--- src/work_graph_run_query.py | 42 ++++++++++++++++++++++++++++++++---------- 26 files changed, 564 insertions(+), 178 deletions(-) diff --git a/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md b/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md index d743639..4734690 100644 --- a/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md +++ b/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md @@ -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。 --- diff --git a/frontend/app.js b/frontend/app.js index eaf6426..945c374 100644 --- a/frontend/app.js +++ b/frontend/app.js @@ -692,10 +692,23 @@ const ORG_NODE_ORDER = ["planner", "coder", "validator"]; // 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-": Org Graph 点击可达详情's scroll/highlight // target (focusOrgNodePanel above) -- additive, doesn't change any // existing class/data-* this function already emitted. - '

' + run.org_node + " Work Graph DAG 视图（v_graph_visualizer，run_id=" + run.run_id + "）

" + '

' + orgNodesLabel(run) + " Work Graph DAG 视图（v_graph_visualizer，run_id=" + run.run_id + "）

" + '
' + '' + rowsHtml + "
" @@ -1456,7 +1469,7 @@ function mountTaskDetail(taskId, initialOrgNode) { return ( // id="todo-tree-": secondary scroll/highlight target when the // DAG panel above didn't render (graph_manifest missing). - '

' + run.org_node + " Work Graph 的 to-do 树（run_id=" + run.run_id + "）

" + '

' + orgNodesLabel(run) + " Work Graph 的 to-do 树（run_id=" + run.run_id + "）

" + '
    ' + items + "
" + '
DAG 视图（同一份 graph_manifest 的另一种呈现）见下方 v_graph_visualizer
' + "
" @@ -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 = ( - "

" + escapeHtml(run.org_node) + " Work Graph Run

" + "

" + escapeHtml(orgNodesLabel(run)) + " Work Graph Run

" + renderForkBadgeHtml(run) + '
' + "
RUN_ID" + escapeHtml(run.run_id) + "
" @@ -1979,7 +1994,7 @@ async function mountArtifactsOverviewPanel(container, projectId) { const runsHtml = runs.length === 0 ? '
暂无 Graph 运行记录
' : ( '
    ' + runs.slice(0, 10).map((r) => - '
  • ' + escapeHtml(r.org_node) + " · run " + '
  • ' + escapeHtml(orgNodesLabel(r)) + " · run " + r.run_id.slice(0, 8) + (r.parent_run_id ? " 🔀" : "") + "" + escapeHtml(r.status) + " · " + r.node_count + " 节点
    • " ).join("") + "
" @@ -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 diff --git a/frontend/test_org_click_through_smoke.js b/frontend/test_org_click_through_smoke.js index 043319b..f9c23bf 100644 --- a/frontend/test_org_click_through_smoke.js +++ b/frontend/test_org_click_through_smoke.js @@ -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"); diff --git a/src/backend_server.py b/src/backend_server.py index 3817d50..c770636 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -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): diff --git a/src/root_cause_trace.py b/src/root_cause_trace.py index 7e5be79..341842d 100644 --- a/src/root_cause_trace.py +++ b/src/root_cause_trace.py @@ -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() diff --git a/src/scheduler.py b/src/scheduler.py index f0134cd..a0602a4 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -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": diff --git a/src/schema.py b/src/schema.py index b16ad39..99a226b 100644 --- a/src/schema.py +++ b/src/schema.py @@ -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") diff --git a/src/task_query.py b/src/task_query.py index a4374c7..5db9810 100644 --- a/src/task_query.py +++ b/src/task_query.py @@ -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 diff --git a/src/test_artifact_query.py b/src/test_artifact_query.py index 9ae8c8b..c93eb78 100644 --- a/src/test_artifact_query.py +++ b/src/test_artifact_query.py @@ -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() diff --git a/src/test_artifact_write.py b/src/test_artifact_write.py index 44569aa..5840599 100644 --- a/src/test_artifact_write.py +++ b/src/test_artifact_write.py @@ -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() diff --git a/src/test_backend_server.py b/src/test_backend_server.py index eccea7b..156b2a0 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -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) " diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 06d3bde..7760068 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -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() diff --git a/src/test_execution_log.py b/src/test_execution_log.py index 8e0c31f..2f9e916 100644 --- a/src/test_execution_log.py +++ b/src/test_execution_log.py @@ -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) " diff --git a/src/test_org_node_lifecycle.py b/src/test_org_node_lifecycle.py index deb6cb7..77d2078 100644 --- a/src/test_org_node_lifecycle.py +++ b/src/test_org_node_lifecycle.py @@ -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( diff --git a/src/test_org_node_lifecycle_e2e.py b/src/test_org_node_lifecycle_e2e.py index d323a4c..e15d0f2 100644 --- a/src/test_org_node_lifecycle_e2e.py +++ b/src/test_org_node_lifecycle_e2e.py @@ -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: diff --git a/src/test_schema.py b/src/test_schema.py index 271faad..e94cafb 100644 --- a/src/test_schema.py +++ b/src/test_schema.py @@ -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, " diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 2fd6f54..92e0cb0 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -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( diff --git a/src/test_task_query.py b/src/test_task_query.py index e50ba89..40f4400 100644 --- a/src/test_task_query.py +++ b/src/test_task_query.py @@ -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) " diff --git a/src/test_work_graph_fork.py b/src/test_work_graph_fork.py index 25dea98..605dde5 100644 --- a/src/test_work_graph_fork.py +++ b/src/test_work_graph_fork.py @@ -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 = { diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py index ee1a095..db049eb 100644 --- a/src/test_work_graph_orchestrate.py +++ b/src/test_work_graph_orchestrate.py @@ -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() diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index f82210e..9ea6ce0 100644 --- a/src/test_work_graph_plan.py +++ b/src/test_work_graph_plan.py @@ -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() diff --git a/src/test_work_graph_run_query.py b/src/test_work_graph_run_query.py index 2622435..3bd0c07 100644 --- a/src/test_work_graph_run_query.py +++ b/src/test_work_graph_run_query.py @@ -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) diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py index 8988670..640d868 100644 --- a/src/test_work_node_run.py +++ b/src/test_work_node_run.py @@ -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) " diff --git a/src/work_graph_fork.py b/src/work_graph_fork.py index edfbedf..90dd2a7 100644 --- a/src/work_graph_fork.py +++ b/src/work_graph_fork.py @@ -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, ) diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index fd9d0a8..073534f 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -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": []} diff --git a/src/work_graph_run_query.py b/src/work_graph_run_query.py index 824464b..6b3af70 100644 --- a/src/work_graph_run_query.py +++ b/src/work_graph_run_query.py @@ -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, } -- libgit2 0.21.4