diff --git a/docs/todo/graph-engineering-open-items.md b/docs/todo/graph-engineering-open-items.md index 97aa8ce..a7ef909 100644 --- a/docs/todo/graph-engineering-open-items.md +++ b/docs/todo/graph-engineering-open-items.md @@ -62,7 +62,7 @@ ## D. Time-travel 调试要不要立项 -**状态**：✅ 已确认立项 +**状态**：✅ 已实现（2026-08-06）→ 需求文档 `docs/todo/work-graph-time-travel-debugging-requirements.md` + 设计文档 `docs/todo/work-graph-time-travel-debugging-design.md` **来源**：综述版对照 §6 @@ -70,7 +70,9 @@ **确认结果（2026-08-05）**：立项。 -**待办**：✅ 需求文档已产出 → `docs/work-graph-time-travel-debugging-requirements.md`。已明确跟 Deploy Supervisor 自动回滚（`docs/done/supervisor-auto-rollback-design.md`）的命名消歧。最大的数据模型影响点：`work_graph_run` 表现有的 `ux_work_graph_run_active` 唯一约束（同一 `task_id`+`org_node` 同时只能有一个非终态 run）会挡住分叉分支，需要在设计阶段决定怎么扩展。 +**待办**：✅ 已实现（2026-08-06，后端+控制台一起做，用户确认）。核心发现：`ux_work_graph_run_active` 唯一索引是**带 WHERE 条件的部分索引**（只约束非终态 run），比背景描述的风险小——v1 裁剪成只支持从**已终态**的父 run 分叉，完全绕开这条约束，不需要碰它。FR-1 的"覆盖 input_payload 重新触发执行"的歧义，最终决定成**替换输出、不重新提示**（分叉点本身不重新跑 `claude -p`，只把它的 `output_payload` 换成人工提供的新内容，只有严格下游节点真的重新执行）——设计文档第 2 节记录了为什么这样解决了跨调用方约定不统一的问题。实现依赖**零改动执行引擎**：预先把非下游节点插入成 `status='done'` 的 `work_node_execution` 行，`run_work_node()` 已有的 REQ-06 断点续跑短路逻辑自动完成"复用而不重跑"，`run_graph()`/`topological_layers()` 一行都没动。 + +新增：`src/work_graph_fork.py`（`create_fork`/`execute_fork`/`fork_work_graph_run`/`start_fork_in_background`，`_reconstruct_build_prompt` 支持 Planner 固定文档链和 E 的通用 `task_prompt` 两种图形状，其余形状明确报错而不是产出错误提示词）；`work_graph_run`新增 `parent_run_id`/`fork_of_node_exec_id`（`migrate_add_work_graph_fork_columns` 迁移已有数据库）；查询层（`work_graph_run_query.py`/`task_query.py`）透出这两个字段；新 HTTP 端点 `POST /api/work-nodes/:node_exec_id/fork`（`_handle_task_execute` 同款"同步创建+后台真实执行"模式）；控制台 `v_run_detail` 新增"从这里分叉"表单（仅 `done` 节点）+ 分叉徽标（链接回父 run），`v_graph_console` 的运行列表用 🔀 标出分叉分支。真实跑通了一次端到端验证（启动真实 backend、真实 HTTP fork 请求、确认新 run 正确落库并可追溯）。测试覆盖需求文档验收标准 1-3（PRD 换内容后下游真实重跑/BRD 不动、原始 run 完全不受影响、同一节点可分叉多次互不干扰）。`python3 -m unittest discover -s src`（608 tests）+ 前端 3 个 smoke test 全绿；`supervisor` 套件里一个真实候选进程冒烟测试用 `git stash create` 快照当前工作树来测试未提交状态，只快照 tracked 文件——本轮新增的 3 个文件当时还是 untracked，被这个测试技巧本身漏掉导致候选进程 import 失败，不是代码问题，提交后自然解决（真实生产的 `stage_deployment()` 用的是真实 commit hash，不会有这个问题）。 --- @@ -142,7 +144,7 @@ | **F** 三视角评审团 fan-out | `judge_root_cause` 从 Python 函数改成 Work Graph 菱形拓扑 | ✅ **已实现**（2026-08-06，586 测试全绿） | M | 已完成 | 无（G 已满足） | | **B** loop-until-dry Skill | Validator 未知规模扫描收敛能力 | ✅ **已实现**（2026-08-06，587 测试全绿） | S/M | 已完成 | 无 | | **H** Role Package 抽象层 | 统一 role/context/skills/applications/workflows/permissions/acceptance 打包格式 | ✅ 需求评审通过（2026-08-06），待设计 | XL | P2（战略级，不阻塞其他项） | workflows 字段依赖 E 的产出（已满足） | -| **D** Time-travel 调试 | 回到历史节点分叉重跑 | 仅需求 | L | P3 | 无强依赖，但价值随图变复杂（E 落地后）才会显现 | +| **D** Time-travel 调试 | 回到历史节点分叉重跑 | ✅ **已实现**（2026-08-06，608 测试全绿） | L | 已完成 | 无 | | C 按节点选模型 | — | ⛔ 已确认不做 | — | — | — | | A 术语对照 | — | ✅ 已完成 | — | — | — | diff --git a/docs/todo/work-graph-time-travel-debugging-design.md b/docs/todo/work-graph-time-travel-debugging-design.md new file mode 100644 index 0000000..178d5e8 --- /dev/null +++ b/docs/todo/work-graph-time-travel-debugging-design.md @@ -0,0 +1,61 @@ +# Work Graph 执行历史 Time-Travel 调试——详细设计（SDD） + +**日期**: 2026-08-06 +**性质**: 详细设计，对应需求文档 `docs/todo/work-graph-time-travel-debugging-requirements.md`（条目 D，已确认）。回答该需求文档第 8 节的开放问题。 +**状态**: 设计草案，实施中，后端+控制台一起做（用户 2026-08-06 确认）。 + +## 1. `ux_work_graph_run_active` 比需求文档描述的风险更小 + +`src/schema.py:134` 那条唯一索引是**带 WHERE 条件的部分索引**：`WHERE status IN ('planning', 'running', 'merging', 'verifying')`——'done'/'failed'/'blocked' 状态的 run 完全不受它约束。这意味着"从一个已经跑完很久的 task 分叉调试"（最常见的真实场景）根本不会撞上这条约束。 + +**v1 范围裁剪（跟用户讨论后确认）**：只支持从**父 run 已经到达终态**（`done`/`failed`/`blocked`）的节点分叉。这样新 fork 出的 run 完全落在这条部分索引的约束范围之外，不需要碰这条索引本身，也不需要梳理"还有哪些代码路径隐式假设了一个 task_id+org_node 只有一个活跃 run"这个更大的排查面（需求文档风险第 1 条点名的工作）。从还在跑的父 run 分叉，留给以后有真实需求再做。 + +## 2. 关键设计决定：替换输出，不重新提示（解决 FR-1 的歧义） + +需求文档 FR-1 原文："人工可以指定新的输入（覆盖原 `input_payload` 的部分或全部字段），触发从这个节点开始的重新执行。" 但 `input_payload` 在现有代码里几乎总是一个薄壳（`{"node_id": ...}`）——真正的提示词内容来自调用方各自不同的机制（Planner 靠 `artifact_type` 查 `_DOC_PROMPTS`，E 的通用路径靠节点自带的 `task_prompt` 字段，F 的评审团靠运行时闭包）。要让分叉机制对这些不同调用方都通用，"覆盖 input_payload 重新触发 LLM 调用"这条路走不通，会绑死在某一种调用方约定上。 + +**决定**：分叉点这个节点本身**不重新执行**——它的 `output_payload` 被人工提供的新内容直接替换（验收标准 1 的例子"换一个不同的 PRD 内容"本来就是这个意思：人工已经有了想要的新内容，不是要求 Agent 重新生成一个不一样的版本）。只有**严格下游**的节点才会真的重新跑一次 `claude -p`。这个决定让分叉机制完全不需要知道原图是哪个调用方产出的、用的是哪种 `build_prompt` 约定——只需要持久化的 `graph_manifest` 和各节点的 `output_payload`，这两样任何一条 Work Graph run 都有。 + +## 3. 实现机制：100% 复用 REQ-06 断点续跑，零改动执行引擎 + +`work_node_run.run_work_node()` 已有的逻辑：一个 `work_node_execution` 行如果 `status='done'`，直接返回持久化的 `output_payload`，不调用 `run_fn`。`work_graph_orchestrate.run_graph()`/`topological_layers()` 对此完全无感——它只是正常跑一遍拓扑分层调度，每层每个节点各自决定"是短路返回还是真的执行"是 `run_work_node()` 内部的事。 + +这意味着 `fork_work_graph_run()` 不需要裁剪 `graph_manifest`、不需要给 `run_graph()` 传一个"预填充的 outputs"参数（`run_graph()` 也没有这个参数，不需要新加）——**只需要在调用 `run_graph()` 之前，往新 `run_id` 下预先插入一批 `work_node_execution` 行**： + +- 分叉点本身：`status='done'`，`output_payload` = 人工提供的新内容。 +- 分叉点的**非下游**节点（严格下游子图之外的所有节点，包括分叉点的上游和跟分叉点无关的兄弟分支）：`status='done'`，`output_payload` = 原始 run 里这个节点已经产出的内容，原样复制。 +- 分叉点的**严格下游**节点：什么都不插入，留给 `run_graph()` 正常调度、真实执行。 + +`run_graph()` 拿到的还是**完整的原始 `graph_manifest`**（不裁剪），对每个节点走到 `_ensure_node_row()` 时，发现 `(new_run_id, node_id)` 已经有行了（我们预先插的），直接复用那个 `node_exec_id`；`run_work_node()` 一看 `status='done'`，立刻短路返回——这条路径是 `RunGraphHappyPathTest.test_resumes_without_rerunning_done_nodes` 已经在测的行为，不是新写的逻辑。严格下游的节点走的是全新的 `node_exec_id`，正常执行，`build_prompt(node, outputs)` 拿到的 `outputs` 里已经有了（无论是复用来的还是刚替换的）上游内容——这跟"进程重启后 resume"时 `outputs` 字典的填充方式完全一样，天然支持 G 的并发调度（如果严格下游子图里有互不依赖的节点，一样会被分进同一层并发执行，不需要额外代码）。 + +`_downstream_node_ids(graph_manifest, source_node_id)`：从 `depends_on`/`edges` 正向遍历依赖图（复用 `topological_layers()` 已经在用的同一套 dependents 构建逻辑，抽出来共享，不重复实现一遍）。 + +## 4. Schema 改动 + +`work_graph_run` 新增两个可空字段：`parent_run_id`（指向被分叉出去的原始 run）、`fork_of_node_exec_id`（指向具体是从哪个节点分叉的）。纯新增字段，**不改** `ux_work_graph_run_active`（见第 1 节，v1 范围内不需要）。新数据库直接建在 `SCHEMA_SQL` 里；已有数据库用 `migrate_add_work_graph_fork_columns()`（跟 `migrate_expand_artifact_types` 同一个先例，但更简单——纯 `ALTER TABLE ADD COLUMN`，不需要建新表复制数据那套，因为不是改 CHECK 约束）。 + +## 5. 查询层 + 控制台（FR-4） + +`work_graph_run_query.list_work_graph_runs`/`get_work_graph_run`、`task_query.get_task` 的 `work_graph_runs` 列表，都加上 `parent_run_id`/`fork_of_node_exec_id` 两个字段——前端靠这两个字段判断"这条是原始记录还是分叉分支"，以及"分叉自哪一条"。 + +控制台改动（`frontend/app.js`）： +- 运行历史列表/任务详情页的每条 `work_graph_run` 卡片，`parent_run_id` 非空时显示一个"🔀 从 `` 分叉自 ``"的标记，点击可跳转到父 run。 +- 节点详情面板（已有的 `renderGraphVisualizerPanel`/节点 hover 面板）里，`status='done'` 的节点新增一个"从这里分叉"按钮，点开一个表单填新的 output 内容（JSON），提交调用新端点。 +- 不做 FR-5 提到的"设为主线"之类的合并操作——需求文档非目标 #3 已经说了第一版不做 diff/合并 UI，只读展示 + 触发分叉。 + +## 6. HTTP 端点 + +`POST /api/work-node-executions/:node_exec_id/fork`，body `{"new_output_payload": {...}}`，返回 `{"run_id", "outputs"}`。放在 `backend_server.py` 现有的路由风格里（正则匹配路径 + 方法分派，参考已有的 `/api/tasks/:id/...` 系列端点）。 + +## 7. 不做的部分（如实记录） + +- **从还在跑的父 run 分叉**（第 1 节裁剪掉的部分）。 +- **FR-5 的"设为主线"交互**——只读展示分叉分支，人工另外决定怎么处理分叉产物（需求文档非目标 #3 已经说了 diff/比较视图留给后续）。 +- **分叉分支的生命周期/清理策略**（需求文档 §8 开放问题最后一条）——先不清理，跟原始执行日志的保留策略是两个独立的问题，等真的堆积成问题再处理。 + +## 8. 回答需求文档第 8 节开放问题 + +- `parent_run_id`/`fork_of_node_exec_id` 具体 schema：见第 4 节，纯新增可空字段，不动唯一索引（因为 v1 范围裁剪掉了会冲突的场景，见第 1 节）。 +- 触发入口：控制台 + 一个后端 API（`POST /api/work-node-executions/:node_exec_id/fork`），两者复用同一个 `fork_work_graph_run()`。 +- FR-5 交互：不做"设为主线"按钮，纯只读展示（见第 5、7 节）。 +- 生命周期/清理策略：本轮不做（见第 7 节）。 diff --git a/frontend/app.js b/frontend/app.js index da50ca9..1a5b3ec 100644 --- a/frontend/app.js +++ b/frontend/app.js @@ -1487,10 +1487,26 @@ function mountTaskDetail(taskId, initialOrgNode) { // existing auto-scroll checkbox keeps working without touching // mountLogViewer itself. +// D (2026-08-06, time-travel debugging, FR-4): only a 'done' node is a +// valid fork point (work_graph_fork.create_fork's own FR-1 requirement -- +// its output_payload has to exist to be substituted). The form itself is +// plain textarea + button, no native prompt()/confirm() (this codebase +// doesn't use those anywhere, matches the existing btn-rollback pattern's +// inline-form style instead). function renderRunNodeListHtml(run) { if (!run.nodes.length) return '
该运行尚无节点
'; const items = run.nodes.map((wn) => { const status = wn.status; + const forkHtml = status === "done" ? ( + '
' + + '' + + '" + + "
" + ) : ""; return ( '
  • ' + '
    ' @@ -1502,6 +1518,7 @@ function renderRunNodeListHtml(run) { + '
    ' + (wn.reflection_notes ? "reflection_notes: " + escapeHtml(wn.reflection_notes) + "
    " : "") + '
    ' + + forkHtml + "
    " + "
    • " ); @@ -1509,6 +1526,58 @@ function renderRunNodeListHtml(run) { return '"; } +// D (2026-08-06, FR-4): a forked run shows where it came from, linking +// back to the parent run -- read-only, no "set as primary" action (the +// requirements doc's own non-goal #3, see design doc §5/§7). +function renderForkBadgeHtml(run) { + if (!run.parent_run_id) return ""; + return ( + '
    🔀 从节点 ' + escapeHtml(run.fork_of_node_exec_id) + ' 分叉自 ' + + '' + escapeHtml(run.parent_run_id) + "
    " + ); +} + +function mountForkPanels(container) { + for (const panel of container.querySelectorAll(".fork-panel")) { + const nodeExecId = panel.closest(".todo-detail").id.replace("todo-detail-", ""); + const toggleBtn = panel.querySelector(".btn-fork"); + const form = panel.querySelector(".fork-form"); + const textarea = panel.querySelector(".fork-content"); + const submitBtn = panel.querySelector(".btn-fork-submit"); + const errorEl = panel.querySelector(".fork-error"); + + toggleBtn.addEventListener("click", (e) => { + e.stopPropagation(); + form.style.display = form.style.display === "none" ? "block" : "none"; + }); + // clicking inside the form must not toggle the parent .todo-detail row + form.addEventListener("click", (e) => e.stopPropagation()); + + submitBtn.addEventListener("click", async () => { + const content = textarea.value; + if (!content.trim()) { + errorEl.textContent = "请输入新的 content"; + errorEl.style.display = "block"; + return; + } + errorEl.style.display = "none"; + submitBtn.disabled = true; + const resp = await fetch("/api/work-nodes/" + encodeURIComponent(nodeExecId) + "/fork", { + method: "POST", + body: JSON.stringify({ new_output_payload: { content: content } }), + }); + submitBtn.disabled = false; + const body = await resp.json(); + if (resp.status !== 200) { + errorEl.textContent = body.error + (body.detail ? "：" + body.detail : ""); + errorEl.style.display = "block"; + return; + } + window.location.hash = "#/runs/" + encodeURIComponent(body.run_id); + }); + } +} + function renderRunDetail(runId) { if (!getCurrentProjectId()) return renderNoProjectSelected(); return ` @@ -1540,6 +1609,7 @@ function mountRunDetail(runId) { closeAllLogViewers(); content.innerHTML = ( "

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

    " + + renderForkBadgeHtml(run) + '
    ' + "
    RUN_ID" + escapeHtml(run.run_id) + "
    " + "
    STATUS" + escapeHtml(run.status) + "
    " @@ -1566,6 +1636,7 @@ function mountRunDetail(runId) { } }); } + mountForkPanels(content); mountGraphVisualizerTooltips(content); if (!socket) { @@ -1668,7 +1739,8 @@ async function mountArtifactsOverviewPanel(container, projectId) { const runsHtml = runs.length === 0 ? '
    暂无 Graph 运行记录
    ' : ( '
      ' + runs.slice(0, 10).map((r) => '
    • ' + escapeHtml(r.org_node) + " · run " - + r.run_id.slice(0, 8) + "" + escapeHtml(r.status) + " · " + r.node_count + " 节点
      • " + + r.run_id.slice(0, 8) + (r.parent_run_id ? " 🔀" : "") + "" + + escapeHtml(r.status) + " · " + r.node_count + " 节点" ).join("") + "
    " ); diff --git a/frontend/style.css b/frontend/style.css index c3b4179..bf48791 100644 --- a/frontend/style.css +++ b/frontend/style.css @@ -191,3 +191,12 @@ pre.code { background: #0b1120; color: #d1d5db; padding: 14px; border-radius: 8p .iteration .summary { font-size: 13px; margin-bottom: 8px; } .iteration .actions { display: flex; gap: 8px; } .iteration .diff-output { margin-top: 10px; } + +/* D (2026-08-06): v_run_detail time-travel fork controls, and the fork + marker shown on run list rows (v_graph_console) / the run-detail header + itself when the run being viewed is a fork. */ +.fork-badge { background: #eef2ff; color: #4338ca; border-radius: 8px; padding: 8px 12px; font-size: 13px; margin-bottom: 12px; } +.fork-panel { margin-top: 10px; } +.fork-form { margin-top: 8px; display: flex; flex-direction: column; gap: 8px; max-width: 480px; } +.fork-content { font-family: inherit; font-size: 13px; padding: 8px; border: 1px solid #e5e7eb; border-radius: 6px; resize: vertical; } +.btn-fork-submit { align-self: flex-start; } diff --git a/frontend/test_run_detail_smoke.js b/frontend/test_run_detail_smoke.js index 58befa9..55a34bc 100644 --- a/frontend/test_run_detail_smoke.js +++ b/frontend/test_run_detail_smoke.js @@ -53,6 +53,9 @@ const src = fs.readFileSync(path.join(__dirname, "app.js"), "utf8"); // reach the sandbox object (same Annex B quirk the other smoke tests // already document) -- re-export what this test needs. const augmentedSrc = src + "\nglobalThis.RUN_DETAIL_RE = RUN_DETAIL_RE;\n"; +// (renderRunNodeListHtml/renderForkBadgeHtml/mountForkPanels are already +// top-level function declarations, so they land in the sandbox globalThis +// directly -- no extra re-export needed for those, unlike the const regex.) new vm.Script(augmentedSrc, { filename: "app.js" }).runInContext(sandbox); let assertionCount = 0; @@ -114,4 +117,34 @@ function check(condition, message) { check(!li.includes("data-run-task-id"), "the old task-switcher click target must be gone"); } +// ---- 4. D (2026-08-06, time-travel debugging): fork panel only on 'done' nodes ---- +{ + const run = { + run_id: "run-1", org_node: "planner", status: "running", + nodes: [ + { node_exec_id: "exec-running", node_id: "prd", subagent_type: "planner", status: "running", retry_count: 0 }, + { node_exec_id: "exec-done", node_id: "sdd", subagent_type: "planner", status: "done", retry_count: 0 }, + ], + }; + const html = sandbox.renderRunNodeListHtml(run); + const doneDetail = html.slice(html.indexOf('id="todo-detail-exec-done"')); + const runningDetail = html.slice( + html.indexOf('id="todo-detail-exec-running"'), html.indexOf('id="todo-detail-exec-done"') + ); + check(doneDetail.includes("btn-fork"), "a 'done' node must offer a fork trigger"); + check(doneDetail.includes("fork-content"), "the fork form's textarea must be present for a 'done' node"); + check(!runningDetail.includes("btn-fork"), "a non-'done' node must not offer a fork trigger (source must be 'done', FR-1)"); +} + +// ---- 5. D: renderForkBadgeHtml -- only shown for an actual fork ---- +{ + const original = { run_id: "run-1", parent_run_id: null, fork_of_node_exec_id: null }; + check(sandbox.renderForkBadgeHtml(original) === "", "a normal (non-forked) run must render no badge"); + + const forked = { run_id: "run-2", parent_run_id: "run-1", fork_of_node_exec_id: "run-1-prd" }; + const badge = sandbox.renderForkBadgeHtml(forked); + check(badge.includes("run-1-prd"), "badge must name which node was forked from"); + check(badge.includes('href="#/runs/run-1"'), "badge must link back to the parent run"); +} + console.log("frontend/test_run_detail_smoke.js: all " + assertionCount + " assertions passed"); diff --git a/src/backend_server.py b/src/backend_server.py index 07b9551..d96f27c 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -39,8 +39,8 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from src import ( artifact_query, budget_guardrail, bug_record, deployment_query, execution_log, org_loop_route, paths, project_query, project_scaffold, schema, settings_config, - skill_query, skill_update, task_create, task_execution, task_query, work_graph_run_query, - ws_subscribe, + skill_query, skill_update, task_create, task_execution, task_query, work_graph_fork, + work_graph_run_query, ws_subscribe, ) _FRONTEND_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend") @@ -131,6 +131,7 @@ _ARTIFACTS_PREFIX = "/api/artifacts/" _SKILLS_PREFIX = "/api/skills/" _DEPLOYMENTS_PREFIX = "/api/deployments/" _WORK_GRAPH_RUNS_PREFIX = "/api/work-graph-runs/" +_FORK_SUFFIX = "/fork" class _Handler(BaseHTTPRequestHandler): @@ -254,6 +255,9 @@ class _Handler(BaseHTTPRequestHandler): elif self.path.startswith(_SKILLS_PREFIX) and self.path.endswith("/rollback"): skill_id = self.path[len(_SKILLS_PREFIX):-len("/rollback")] self._handle_skill_rollback(skill_id) + elif self.path.startswith(_WORK_NODES_PREFIX) and self.path.endswith(_FORK_SUFFIX): + node_exec_id = self.path[len(_WORK_NODES_PREFIX):-len(_FORK_SUFFIX)] + self._handle_work_node_fork(node_exec_id) else: self._send_json(404, {"error": "NOT_FOUND"}) @@ -686,6 +690,53 @@ class _Handler(BaseHTTPRequestHandler): else: self._send_json(200, result) + def _handle_work_node_fork(self, node_exec_id): + """D (2026-08-06, time-travel debugging), POST + /api/work-nodes/:node_exec_id/fork, body {"new_output_payload": {...}}. + Mirrors _handle_task_execute's shape: create_fork() runs + synchronously (fast, DB-only) so the response carries the new + run_id immediately, real execution continues in the background -- + the console navigates straight to #/runs/:run_id and watches it + progress live via the existing work-graph-run endpoints, no new + polling mechanism needed.""" + payload, error = self._read_json_body() + if error: + self._send_json(400, error) + return + new_output_payload = payload.get("new_output_payload") + if not isinstance(new_output_payload, dict): + self._send_json(400, {"error": "INVALID_INPUT", "detail": "new_output_payload must be an object"}) + return + + with self.server.conn_lock: + row = self.server.conn.execute( + "SELECT p.target_folder_path FROM work_node_execution wne " + "JOIN work_graph_run wgr ON wgr.run_id = wne.run_id " + "JOIN task t ON t.task_id = wgr.task_id " + "JOIN project p ON p.project_id = t.project_id " + "WHERE wne.node_exec_id = ?", + (node_exec_id,), + ).fetchone() + if row is None: + self._send_json(404, {"error": "NOT_FOUND"}) + return + target_folder_path = row[0] + + try: + run_id, _thread = work_graph_fork.start_fork_in_background( + self.server.db_path, node_exec_id, new_output_payload, target_folder_path, + ) + except work_graph_fork.ForkSourceNotDoneError as e: + self._send_json(409, {"error": "FORK_SOURCE_NOT_DONE", "detail": str(e)}) + except work_graph_fork.ForkParentRunNotTerminalError as e: + self._send_json(409, {"error": "FORK_PARENT_RUN_NOT_TERMINAL", "detail": str(e)}) + except work_graph_fork.ForkGraphShapeUnsupportedError as e: + self._send_json(422, {"error": "FORK_GRAPH_SHAPE_UNSUPPORTED", "detail": str(e)}) + except ValueError as e: + self._send_json(404, {"error": "NOT_FOUND", "detail": str(e)}) + else: + self._send_json(200, {"run_id": run_id, "started": True}) + def _handle_artifact_get(self, artifact_id): try: with self.server.conn_lock: diff --git a/src/schema.py b/src/schema.py index 556dfdb..3ca156f 100644 --- a/src/schema.py +++ b/src/schema.py @@ -127,7 +127,16 @@ CREATE TABLE work_graph_run ( )), started_at TEXT, completed_at TEXT, - result_artifact_refs TEXT + result_artifact_refs TEXT, + -- D (2026-08-06, time-travel debugging): both NULL for a normal 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. + 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 @@ -398,3 +407,22 @@ def migrate_expand_artifact_types(conn): """ ) conn.commit() + + +def migrate_add_work_graph_fork_columns(conn): + """D (2026-08-06, time-travel debugging): adds parent_run_id/ + fork_of_node_exec_id to work_graph_run on an ALREADY-POPULATED + database. Unlike migrate_expand_artifact_types, this is a plain `ALTER + TABLE ... ADD COLUMN` (no CHECK constraint involved, so no need for the + create-copy-drop-rename dance). Idempotent and safe to call + unconditionally, including against a freshly create_schema()'d database + (which already has both columns).""" + cols = {row[1] for row in conn.execute("PRAGMA table_info(work_graph_run)").fetchall()} + if "parent_run_id" not in cols: + conn.execute("ALTER TABLE work_graph_run ADD COLUMN parent_run_id TEXT REFERENCES work_graph_run(run_id)") + if "fork_of_node_exec_id" not in cols: + conn.execute( + "ALTER TABLE work_graph_run ADD COLUMN fork_of_node_exec_id TEXT " + "REFERENCES work_node_execution(node_exec_id)" + ) + conn.commit() diff --git a/src/task_query.py b/src/task_query.py index b1493ec..5fcc59d 100644 --- a/src/task_query.py +++ b/src/task_query.py @@ -155,7 +155,8 @@ def get_task(conn, task_id): ] runs = conn.execute( - "SELECT run_id, org_node, graph_manifest, status, started_at, completed_at " + "SELECT run_id, org_node, 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() @@ -164,6 +165,9 @@ def get_task(conn, task_id): "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], + # 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]), } for w in runs diff --git a/src/test_backend_server.py b/src/test_backend_server.py index 8f58afe..b4b9091 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -1368,6 +1368,117 @@ class WorkGraphRunListRouteTest(unittest.TestCase): self.assertEqual(body["error"], "NOT_FOUND") +class WorkNodeForkRouteTest(unittest.TestCase): + """D (2026-08-06, time-travel debugging): POST + /api/work-nodes/:node_exec_id/fork. `start_fork_in_background` is + mocked throughout -- same reasoning as TaskExecutionRouteTest just + above: it opens its OWN sqlite connection via schema.connect(db_path), + which for a ':memory:' test db would silently create a second, + disconnected database. Real fork behavior is already covered in + test_work_graph_fork.py (real file db, real threading).""" + + def setUp(self): + self.server = backend_server.create_server(db_path=":memory:") + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + conn = self.server.conn + conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "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')" + ) + conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', 'new', 'active')" + ) + 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, output_payload, status) " + "VALUES ('node-1', 'run-1', 'prd', 'planner', 'skill-1', 'in.json', 'out.json', '{}', '{}', 'done')" + ) + conn.commit() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + + def _post(self, path, payload=None): + host, port = self.server.server_address + conn = http.client.HTTPConnection(host, port, timeout=5) + body = json.dumps(payload).encode("utf-8") if payload is not None else b"" + conn.request("POST", path, body=body) + resp = conn.getresponse() + raw = resp.read() + body = json.loads(raw) if raw else None + conn.close() + return resp.status, body + + def test_fork_returns_run_id_and_calls_through_with_target_folder_path(self): + with patch.object(backend_server.work_graph_fork, "start_fork_in_background") as mock_start: + mock_start.return_value = ("run-fork-1", None) + status, body = self._post( + "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "new prd"}} + ) + self.assertEqual(status, 200) + self.assertEqual(body, {"run_id": "run-fork-1", "started": True}) + mock_start.assert_called_once_with( + self.server.db_path, "node-1", {"content": "new prd"}, "/tmp/proj-1", + ) + + def test_fork_nonexistent_node_returns_404(self): + status, body = self._post( + "/api/work-nodes/no-such-node/fork", {"new_output_payload": {"content": "x"}} + ) + self.assertEqual(status, 404) + self.assertEqual(body["error"], "NOT_FOUND") + + def test_fork_missing_new_output_payload_returns_400(self): + status, body = self._post("/api/work-nodes/node-1/fork", {}) + self.assertEqual(status, 400) + self.assertEqual(body["error"], "INVALID_INPUT") + + def test_fork_source_not_done_returns_409(self): + with patch.object( + backend_server.work_graph_fork, "start_fork_in_background", + side_effect=backend_server.work_graph_fork.ForkSourceNotDoneError("not done"), + ): + status, body = self._post( + "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}} + ) + self.assertEqual(status, 409) + self.assertEqual(body["error"], "FORK_SOURCE_NOT_DONE") + + def test_fork_parent_run_not_terminal_returns_409(self): + with patch.object( + backend_server.work_graph_fork, "start_fork_in_background", + side_effect=backend_server.work_graph_fork.ForkParentRunNotTerminalError("still running"), + ): + status, body = self._post( + "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}} + ) + self.assertEqual(status, 409) + self.assertEqual(body["error"], "FORK_PARENT_RUN_NOT_TERMINAL") + + def test_fork_unsupported_graph_shape_returns_422(self): + with patch.object( + backend_server.work_graph_fork, "start_fork_in_background", + side_effect=backend_server.work_graph_fork.ForkGraphShapeUnsupportedError("nope"), + ): + status, body = self._post( + "/api/work-nodes/node-1/fork", {"new_output_payload": {"content": "x"}} + ) + self.assertEqual(status, 422) + self.assertEqual(body["error"], "FORK_GRAPH_SHAPE_UNSUPPORTED") + + class DeploymentRouteTest(unittest.TestCase): """U-043: c_deployment_list/c_deployment_current/c_deployment_get for v_deployment_status.""" diff --git a/src/test_schema.py b/src/test_schema.py index 9f82479..9fa7994 100644 --- a/src/test_schema.py +++ b/src/test_schema.py @@ -6,6 +6,7 @@ total) and widened `bug_record.root_cause_category` to accept `core_engine`. Run: python3 -m unittest src.test_schema -v """ +import re import sqlite3 import unittest @@ -489,5 +490,65 @@ class DeployHealthSignalTest(ConstraintTestBase): ) +class MigrateAddWorkGraphForkColumnsTest(unittest.TestCase): + """D (2026-08-06): real upgrade path for an already-populated database + whose work_graph_run table predates the fork columns.""" + + def setUp(self): + self.conn = schema.connect(":memory:") + # Strip the fork columns (and their leading comment block) back out + # of work_graph_run's CREATE TABLE to simulate a pre-migration + # database, without hand-maintaining a second copy of the DDL text. + old_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);", + schema.SCHEMA_SQL, + ) + self.assertNotIn("parent_run_id", old_sql) # sanity: the substitution actually matched + 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.commit() + + def tearDown(self): + self.conn.close() + + def test_migration_preserves_existing_rows_and_adds_nullable_columns(self): + schema.migrate_add_work_graph_fork_columns(self.conn) + row = self.conn.execute( + "SELECT status, parent_run_id, fork_of_node_exec_id FROM work_graph_run WHERE run_id = 'run-1'" + ).fetchone() + self.assertEqual(row, ("done", None, None)) + + def test_idempotent_on_a_database_already_migrated(self): + schema.migrate_add_work_graph_fork_columns(self.conn) + schema.migrate_add_work_graph_fork_columns(self.conn) # must not raise + count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run").fetchone()[0] + self.assertEqual(count, 1) + + def test_noop_on_a_freshly_created_already_expanded_database(self): + fresh = schema.connect(":memory:") + try: + schema.create_schema(fresh) + schema.migrate_add_work_graph_fork_columns(fresh) # must not raise + cols = {row[1] for row in fresh.execute("PRAGMA table_info(work_graph_run)").fetchall()} + self.assertIn("parent_run_id", cols) + self.assertIn("fork_of_node_exec_id", cols) + finally: + fresh.close() + + if __name__ == "__main__": unittest.main() diff --git a/src/test_task_query.py b/src/test_task_query.py index 37d0a9d..bb465fb 100644 --- a/src/test_task_query.py +++ b/src/test_task_query.py @@ -143,10 +143,38 @@ class GetTaskTest(TaskQueryTestBase): self.assertEqual(run["graph_manifest"], manifest) self.assertEqual(len(run["nodes"]), 1) self.assertEqual(run["nodes"][0]["node_id"], "n1") - self.assertEqual(run["nodes"][0]["status"], "done") - self.assertEqual(run["nodes"][0]["output_payload"], {"x": 1}) - self.assertEqual(run["nodes"][0]["input_schema_ref"], "in.json") - self.assertEqual(run["nodes"][0]["output_schema_ref"], "out.json") + self.assertIsNone(run["parent_run_id"]) + self.assertIsNone(run["fork_of_node_exec_id"]) + + def test_forked_run_surfaces_parent_and_source_node(self): + # D (2026-08-06): v_task_detail must be able to tell a fork branch + # 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')" + ) + self.conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "VALUES ('skill-1', 'planner', 'skills/skill-1/', 'sha', '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, output_payload, status) " + "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, " + "parent_run_id, fork_of_node_exec_id) " + "VALUES ('run-fork', 'task-1', 'planner', '{}', 'done', 'run-orig', 'node-1')" + ) + self.conn.commit() + + result = task_query.get_task(self.conn, "task-1") + by_run_id = {r["run_id"]: r for r in result["work_graph_runs"]} + self.assertIsNone(by_run_id["run-orig"]["parent_run_id"]) + self.assertEqual(by_run_id["run-fork"]["parent_run_id"], "run-orig") + self.assertEqual(by_run_id["run-fork"]["fork_of_node_exec_id"], "node-1") def test_work_node_includes_quota_wait_fields_for_v_graph_visualizer(self): # U-026: v_graph_visualizer's waiting_for_quota state needs diff --git a/src/test_work_graph_fork.py b/src/test_work_graph_fork.py new file mode 100644 index 0000000..5eb95ef --- /dev/null +++ b/src/test_work_graph_fork.py @@ -0,0 +1,345 @@ +"""D (2026-08-06) tests: work_graph_fork.py, mocked at the claude +subprocess boundary (CLAUDE.md Testing constraint) -- same pattern as +test_work_graph_orchestrate.py. Covers the requirements doc's acceptance +standards 1-3 (docs/todo/work-graph-time-travel-debugging-requirements.md). + +Run: python3 -m unittest src.test_work_graph_fork -v +""" + +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +from src import schema, work_graph_execute, work_graph_fork + + +_BRD_PRD_SDD_BENCHMARK_MANIFEST = { + "nodes": [ + {"node_id": "brd", "artifact_type": "brd", "depends_on": []}, + {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, + {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]}, + {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]}, + ], + "edges": [], +} + + +class ForkTestBase(unittest.TestCase): + def setUp(self): + self.home_tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.home_tmp.cleanup) + self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") + os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name + self.addCleanup(self._restore_graphlemon_home) + + self.conn = schema.connect(":memory:", check_same_thread=False) + schema.create_schema(self.conn) + self.conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', '实现贪吃蛇游戏', 'goal', 'validating', 'planner')" + ) + self.conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "VALUES ('planner-doc-writing', 'planner', 'skills/planner-doc-writing/', 'unknown', 'new', 'active')" + ) + self.conn.commit() + + def _restore_graphlemon_home(self): + if self.old_graphlemon_home is None: + os.environ.pop("GRAPHLEMON_HOME", None) + else: + os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home + + def tearDown(self): + self.conn.close() + + def _seed_completed_run(self, run_id, org_node, graph_manifest, contents, status="done"): + """Simulates an already-finished original run: a work_graph_run + 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), + ) + node_exec_ids = {} + for node in graph_manifest["nodes"]: + node_id = node["node_id"] + node_exec_id = run_id + "-" + node_id + node_exec_ids[node_id] = node_exec_id + 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, output_payload, status) " + "VALUES (?, ?, ?, ?, 'planner-doc-writing', 'in.json', 'out.json', ?, ?, 'done')", + ( + node_exec_id, run_id, node_id, org_node, + json.dumps({"node_id": node_id}), json.dumps(contents[node_id]), + ), + ) + self.conn.commit() + return node_exec_ids + + +class ForkFromPrdAcceptanceCriterion1Test(ForkTestBase): + """Acceptance standard 1: fork from PRD with different content, SDD and + Benchmark re-execute with the new PRD content, BRD is not re-executed.""" + + def test_downstream_nodes_reexecute_upstream_node_reused(self): + contents = { + "brd": {"content": "original brd"}, "prd": {"content": "original prd"}, + "sdd": {"content": "original sdd"}, "benchmark": {"content": "original benchmark"}, + } + node_exec_ids = self._seed_completed_run( + "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents + ) + + execute_calls = [] + + def fake_execute(cwd, name, args, output_schema, timeout=600): + execute_calls.append(name) + return {"content": "regenerated for " + name} + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + result = work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["prd"], {"content": "a completely different prd"}, "/tmp/proj-1", + ) + + # only SDD and Benchmark actually ran claude -p calls -- BRD and PRD did not. + self.assertEqual(len(execute_calls), 2) + + outputs = result["outputs"] + self.assertEqual(outputs["brd"], {"content": "original brd"}) + self.assertEqual(outputs["prd"], {"content": "a completely different prd"}) + self.assertNotEqual(outputs["sdd"], {"content": "original sdd"}) + self.assertNotEqual(outputs["benchmark"], {"content": "original benchmark"}) + + new_run_id = result["run_id"] + rows = { + r[0]: (r[1], json.loads(r[2])) + for r in self.conn.execute( + "SELECT node_id, status, output_payload FROM work_node_execution WHERE run_id = ?", (new_run_id,) + ).fetchall() + } + self.assertEqual(rows["brd"], ("done", {"content": "original brd"})) + self.assertEqual(rows["prd"], ("done", {"content": "a completely different prd"})) + self.assertEqual(rows["sdd"][0], "done") + self.assertEqual(rows["benchmark"][0], "done") + + run_row = self.conn.execute( + "SELECT status, parent_run_id, fork_of_node_exec_id FROM work_graph_run WHERE run_id = ?", + (new_run_id,), + ).fetchone() + self.assertEqual(run_row, ("done", "run-orig", node_exec_ids["prd"])) + + +class OriginalRunUnaffectedTest(ForkTestBase): + """Acceptance standard 2: the original run's records are completely + untouched by a fork.""" + + def test_original_run_and_nodes_unchanged_after_fork(self): + contents = { + "brd": {"content": "original brd"}, "prd": {"content": "original prd"}, + "sdd": {"content": "original sdd"}, "benchmark": {"content": "original benchmark"}, + } + node_exec_ids = self._seed_completed_run( + "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents + ) + before_run = self.conn.execute("SELECT * FROM work_graph_run WHERE run_id = 'run-orig'").fetchone() + before_nodes = self.conn.execute( + "SELECT * FROM work_node_execution WHERE run_id = 'run-orig' ORDER BY node_id" + ).fetchall() + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}): + work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["prd"], {"content": "different"}, "/tmp/proj-1", + ) + + after_run = self.conn.execute("SELECT * FROM work_graph_run WHERE run_id = 'run-orig'").fetchone() + after_nodes = self.conn.execute( + "SELECT * FROM work_node_execution WHERE run_id = 'run-orig' ORDER BY node_id" + ).fetchall() + self.assertEqual(before_run, after_run) + self.assertEqual(before_nodes, after_nodes) + + +class MultipleForksFromSameNodeTest(ForkTestBase): + """Acceptance standard 3: the same historical node can be forked + multiple times, producing independent, separately-traceable branches.""" + + def test_two_forks_from_same_node_are_independent(self): + contents = { + "brd": {"content": "original brd"}, "prd": {"content": "original prd"}, + "sdd": {"content": "original sdd"}, "benchmark": {"content": "original benchmark"}, + } + node_exec_ids = self._seed_completed_run( + "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents + ) + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "branch-a"}): + result_a = work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["prd"], {"content": "prd variant A"}, "/tmp/proj-1", + ) + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "branch-b"}): + result_b = work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["prd"], {"content": "prd variant B"}, "/tmp/proj-1", + ) + + self.assertNotEqual(result_a["run_id"], result_b["run_id"]) + self.assertEqual(result_a["outputs"]["prd"], {"content": "prd variant A"}) + self.assertEqual(result_b["outputs"]["prd"], {"content": "prd variant B"}) + self.assertEqual(result_a["outputs"]["sdd"], {"content": "branch-a"}) + self.assertEqual(result_b["outputs"]["sdd"], {"content": "branch-b"}) + + for run_id in (result_a["run_id"], result_b["run_id"]): + row = self.conn.execute( + "SELECT parent_run_id, fork_of_node_exec_id FROM work_graph_run WHERE run_id = ?", (run_id,) + ).fetchone() + self.assertEqual(row, ("run-orig", node_exec_ids["prd"])) + + +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')", + (json.dumps(_BRD_PRD_SDD_BENCHMARK_MANIFEST),), + ) + 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', 'prd', 'planner', 'planner-doc-writing', 'in.json', 'out.json', '{}', 'running')" + ) + self.conn.commit() + with self.assertRaises(work_graph_fork.ForkSourceNotDoneError): + work_graph_fork.fork_work_graph_run(self.conn, "node-1", {"content": "x"}, "/tmp/proj-1") + + def test_forking_from_a_non_terminal_parent_run_raises(self): + contents = {"brd": {"content": "b"}, "prd": {"content": "p"}, "sdd": {"content": "s"}, "benchmark": {"content": "bm"}} + node_exec_ids = self._seed_completed_run( + "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents, status="running", + ) + with self.assertRaises(work_graph_fork.ForkParentRunNotTerminalError): + work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["prd"], {"content": "x"}, "/tmp/proj-1", + ) + + def test_unsupported_graph_shape_raises(self): + manifest = {"nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": ["a"]}], "edges": []} + node_exec_ids = self._seed_completed_run( + "run-orig", "coder", manifest, {"a": {"content": "x"}, "b": {"content": "y"}}, + ) + with self.assertRaises(work_graph_fork.ForkGraphShapeUnsupportedError): + work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["a"], {"content": "z"}, "/tmp/proj-1", + ) + + +class StartForkInBackgroundTest(unittest.TestCase): + """Real threading (own connection per thread) -- needs a real file + db_path, not ':memory:' (a fresh ':memory:' connection per thread + would be a completely separate, empty database), same reasoning as + test_task_execution.py's own StartTaskExecutionInBackgroundTest.""" + + def setUp(self): + self.home_tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.home_tmp.cleanup) + self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") + os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name + self.addCleanup(self._restore_graphlemon_home) + + self.tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + self.tmp_db.close() + self.db_path = self.tmp_db.name + self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path)) + + conn = schema.connect(self.db_path) + schema.create_schema(conn) + conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', 'g', 'goal', 'validating', 'planner')" + ) + conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "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')", + (json.dumps(_BRD_PRD_SDD_BENCHMARK_MANIFEST),), + ) + contents = { + "brd": {"content": "b"}, "prd": {"content": "p"}, "sdd": {"content": "s"}, "benchmark": {"content": "bm"}, + } + for node in _BRD_PRD_SDD_BENCHMARK_MANIFEST["nodes"]: + node_id = node["node_id"] + 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, output_payload, status) " + "VALUES (?, 'run-orig', ?, 'planner', 'planner-doc-writing', 'in.json', 'out.json', ?, ?, 'done')", + ("run-orig-" + node_id, node_id, json.dumps({"node_id": node_id}), json.dumps(contents[node_id])), + ) + conn.commit() + conn.close() + + def _restore_graphlemon_home(self): + if self.old_graphlemon_home is None: + os.environ.pop("GRAPHLEMON_HOME", None) + else: + os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home + + def test_returns_run_id_immediately_and_completes_in_background(self): + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "regenerated"}): + run_id, thread = work_graph_fork.start_fork_in_background( + self.db_path, "run-orig-prd", {"content": "new prd"}, "/tmp/proj-1", + ) + self.assertTrue(run_id) + thread.join(timeout=10) + + conn = schema.connect(self.db_path) + status = conn.execute("SELECT status FROM work_graph_run WHERE run_id = ?", (run_id,)).fetchone()[0] + sdd_output = conn.execute( + "SELECT output_payload FROM work_node_execution WHERE run_id = ? AND node_id = 'sdd'", (run_id,) + ).fetchone()[0] + conn.close() + self.assertEqual(status, "done") + self.assertEqual(json.loads(sdd_output), {"content": "regenerated"}) + + +class ForkOfLeafNodeTest(ForkTestBase): + """Forking the last node in a chain has nothing downstream -- no real + execution needed, just the substituted output in a new run.""" + + def test_forking_leaf_node_needs_no_execution(self): + contents = { + "brd": {"content": "original brd"}, "prd": {"content": "original prd"}, + "sdd": {"content": "original sdd"}, "benchmark": {"content": "original benchmark"}, + } + node_exec_ids = self._seed_completed_run( + "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents + ) + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = AssertionError("should not be called -- nothing downstream of benchmark") + result = work_graph_fork.fork_work_graph_run( + self.conn, node_exec_ids["benchmark"], {"content": "different benchmark"}, "/tmp/proj-1", + ) + self.assertEqual(result["outputs"]["benchmark"], {"content": "different benchmark"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/test_work_graph_run_query.py b/src/test_work_graph_run_query.py index cc32498..2622435 100644 --- a/src/test_work_graph_run_query.py +++ b/src/test_work_graph_run_query.py @@ -27,11 +27,13 @@ class WorkGraphRunQueryTestBase(unittest.TestCase): self.conn.close() 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): + completed_at=None, task_id="task-1", node_count=0, + parent_run_id=None, fork_of_node_exec_id=None): self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status, " - "started_at, completed_at) VALUES (?, ?, ?, '{}', ?, ?, ?)", - (run_id, task_id, org_node, status, started_at, completed_at), + "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), ) for i in range(node_count): node_exec_id = "{}-node-{}".format(run_id, i) @@ -145,6 +147,36 @@ class GetWorkGraphRunTest(WorkGraphRunQueryTestBase): result = work_graph_run_query.get_work_graph_run(self.conn, "run-1") self.assertEqual(result["nodes"], []) + def test_normal_run_has_null_fork_fields(self): + self._insert_run("run-1", "planner", node_count=0) + result = work_graph_run_query.get_work_graph_run(self.conn, "run-1") + self.assertIsNone(result["parent_run_id"]) + self.assertIsNone(result["fork_of_node_exec_id"]) + + def test_forked_run_surfaces_parent_and_source_node(self): + # D (2026-08-06): a fork must be traceable back to which run and + # which specific node it was forked from. + self._insert_run("run-orig", "planner", node_count=1) + self._insert_run( + "run-fork", "planner", node_count=0, + parent_run_id="run-orig", fork_of_node_exec_id="run-orig-node-0", + ) + result = work_graph_run_query.get_work_graph_run(self.conn, "run-fork") + self.assertEqual(result["parent_run_id"], "run-orig") + self.assertEqual(result["fork_of_node_exec_id"], "run-orig-node-0") + + +class ListWorkGraphRunsForkFieldsTest(WorkGraphRunQueryTestBase): + def test_fork_fields_surfaced_in_list(self): + self._insert_run("run-orig", "planner", node_count=1) + self._insert_run( + "run-fork", "planner", parent_run_id="run-orig", fork_of_node_exec_id="run-orig-node-0", + ) + by_run_id = {r["run_id"]: r for r in work_graph_run_query.list_work_graph_runs(self.conn, "proj-1")} + self.assertIsNone(by_run_id["run-orig"]["parent_run_id"]) + self.assertEqual(by_run_id["run-fork"]["parent_run_id"], "run-orig") + self.assertEqual(by_run_id["run-fork"]["fork_of_node_exec_id"], "run-orig-node-0") + if __name__ == "__main__": unittest.main() diff --git a/src/work_graph_fork.py b/src/work_graph_fork.py new file mode 100644 index 0000000..9393665 --- /dev/null +++ b/src/work_graph_fork.py @@ -0,0 +1,259 @@ +"""D (2026-08-06): Work Graph time-travel debugging -- fork a new +work_graph_run from an already-'done' work_node_execution, substituting a +human-provided output for that node and re-executing everything strictly +downstream of it for real, while every other node reuses the original +run's persisted output_payload untouched. See +docs/todo/work-graph-time-travel-debugging-design.md for the full design. + +**Output substitution, not re-prompting** (design doc §2, resolving the +requirements doc's FR-1 ambiguity): the fork point's own output_payload is +replaced directly with what the human supplies -- it is NOT re-executed +via a new claude -p call. This sidesteps having to reconcile the different +build_prompt/input_payload conventions different callers use (Planner's +fixed artifact_type lookup, E's generic task_prompt field, F's jury) -- +this module only needs the persisted graph_manifest and each node's +output_payload, nothing caller-specific, for the SUBSTITUTED node. The +nodes that DO get genuinely re-executed (the strict downstream subgraph) +still need a real build_prompt, which is reconstructed generically -- see +_reconstruct_build_prompt. + +**v1 restricts forking to a parent run that has already reached a +terminal status** ('done'/'failed'/'blocked') -- design doc §1: this keeps +the new fork completely outside ux_work_graph_run_active's partial unique +index (that index only applies to non-terminal statuses), so no schema/ +constraint change is needed to avoid a collision. Forking from a +still-in-flight parent run is left for a future iteration. + +**Zero changes to the execution engine** (design doc §3): pre-seeding +every non-downstream node as a 'done' work_node_execution row before +calling run_graph() on the UNCHANGED original graph_manifest is enough -- +run_work_node()'s existing REQ-06 breakpoint-recovery short-circuit +(already-'done' nodes return their persisted output without calling +run_fn) does all the "reuse without re-executing" work for free. +""" + +import json +import threading +import time +import uuid + +from src import e2e_pipeline, schema, work_graph_orchestrate, work_graph_plan + +_FORK_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} +_FORK_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]} +_TERMINAL_RUN_STATUSES = ("done", "failed", "blocked") + + +class ForkSourceNotDoneError(Exception): + """FR-1: can only fork from a work_node_execution with status='done' -- + its output_payload has to exist to be a valid substitution point.""" + + +class ForkParentRunNotTerminalError(Exception): + """v1 scope cut (design doc §1): forking from a node inside a + still-in-flight parent run would collide with ux_work_graph_run_active + (same task_id+org_node, both non-terminal) -- not supported yet.""" + + +class ForkGraphShapeUnsupportedError(Exception): + """The original graph_manifest doesn't carry enough self-contained + information to reconstruct a build_prompt generically for the + re-executed downstream nodes -- e.g. F's jury nodes or Coder's single + ad hoc node, which rely on caller-side closures over state that isn't + persisted in graph_manifest itself (design doc §2).""" + + +def _downstream_node_ids(graph_manifest, start_node_id): + """Every node_id reachable from start_node_id via depends_on/edges, + INCLUDING start_node_id itself. Same dependents-graph construction + topological_layers() uses, kept local here rather than imported since + it's a small, self-contained piece not worth threading a shared helper + for across modules.""" + nodes = {n["node_id"]: n for n in graph_manifest["nodes"]} + dependents = {node_id: [] for node_id in nodes} + for node in graph_manifest["nodes"]: + for dep in (node.get("depends_on") or []): + if dep in dependents: + dependents[dep].append(node["node_id"]) + for edge in graph_manifest.get("edges", []): + from_id, to_id = edge.get("from"), edge.get("to") + if from_id in dependents and to_id in nodes: + dependents[from_id].append(to_id) + + seen = set() + frontier = [start_node_id] + while frontier: + node_id = frontier.pop() + if node_id in seen: + continue + seen.add(node_id) + frontier.extend(dependents[node_id]) + return seen + + +def _reconstruct_build_prompt(conn, task_id, org_node, 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 shapes (design doc + §2): Planner's canonical doc chain (every node has 'artifact_type', + prompt reconstructed via e2e_pipeline's own builder + task.goal_text) + and E's generic dynamic path (every node carries its own + self-contained 'task_prompt'). Anything else raises rather than + silently building broken prompts.""" + nodes = graph_manifest["nodes"] + if org_node == "planner" and nodes and all("artifact_type" in n for n in nodes): + goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0] + node_artifact_types = {n["node_id"]: n["artifact_type"] for n in nodes} + return e2e_pipeline._planner_prompt_builder(goal_text, node_artifact_types) + if nodes and all("task_prompt" in n for n in nodes): + return work_graph_plan.generic_node_prompt_builder + raise ForkGraphShapeUnsupportedError( + "graph_manifest nodes must either all have 'artifact_type' (Planner's canonical chain) " + "or all have 'task_prompt' (the generic dynamic-planning path) to reconstruct prompts for a fork" + ) + + +def create_fork(conn, source_node_exec_id, new_output_payload): + """Synchronous, DB-only half of a fork: validates the source node and + parent run, pre-seeds every non-downstream node as 'done' under a + brand-new work_graph_run, and returns everything execute_fork() needs + to actually run the downstream subgraph. Split from execute_fork() so + an HTTP caller can hand the (fast) run_id back immediately and let the + (slow, real claude -p calls) execution happen in the background -- + same split task_execution.start_task_execution_in_background already + 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 " + "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,), + ).fetchone() + if row is None: + 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, + ) = row + if source_status != "done": + raise ForkSourceNotDoneError( + "work_node_execution {} is status={}, must be 'done' to fork from".format( + source_node_exec_id, source_status + ) + ) + if parent_run_status not in _TERMINAL_RUN_STATUSES: + raise ForkParentRunNotTerminalError( + "parent work_graph_run {} is status={}, must be terminal ({}) to fork from -- forking from " + "a still-in-flight run isn't supported yet".format( + source_run_id, parent_run_status, _TERMINAL_RUN_STATUSES + ) + ) + + graph_manifest = json.loads(graph_manifest_json) + build_prompt = _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest) + + downstream = _downstream_node_ids(graph_manifest, source_node_id) - {source_node_id} + original_outputs = { + r[0]: r[1] for r in conn.execute( + "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'", + (source_run_id,), + ).fetchall() + } + + new_run_id = str(uuid.uuid4()) + 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), + ) + conn.commit() + + for node in graph_manifest["nodes"]: + node_id = node["node_id"] + if node_id in downstream: + continue + payload = new_output_payload if node_id == source_node_id else json.loads(original_outputs[node_id]) + 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, output_payload, status) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'done')", + ( + str(uuid.uuid4()), new_run_id, node_id, subagent_type, skill_ref, + "work_graph_node_input.schema.json", "work_graph_node_output.schema.json", + json.dumps({"node_id": node_id}), json.dumps(payload), + ), + ) + conn.commit() + + return { + "run_id": new_run_id, "org_node": org_node, "skill_ref": skill_ref, + "graph_manifest": graph_manifest, "build_prompt": build_prompt, + } + + +def execute_fork(conn, fork_info, target_folder_path, timeout=600, sleep=time.sleep, now=time.time): + """The real (slow) half: runs the downstream subgraph via the + unmodified run_graph() -- every pre-seeded node short-circuits, every + downstream node executes for real (see module docstring). Sets the new + run's status to 'done'/'failed' on completion -- unlike every other + real graph_manifest run in this codebase, which never transitions + work_graph_run.status past its initial value (a pre-existing gap, out + of D's scope to fix generically -- see design doc §3's own note); the + forked run gets this narrowly because FR-4's console visibility needs + SOME real completion signal to show, not because this module is fixing + the wider gap.""" + 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, + fork_info["graph_manifest"], fork_info["build_prompt"], _FORK_INPUT_SCHEMA, _FORK_OUTPUT_SCHEMA, + timeout=timeout, sleep=sleep, now=now, + ) + except Exception: + conn.execute( + "UPDATE work_graph_run SET status = 'failed', completed_at = CURRENT_TIMESTAMP WHERE run_id = ?", + (run_id,), + ) + conn.commit() + raise + conn.execute( + "UPDATE work_graph_run SET status = 'done', completed_at = CURRENT_TIMESTAMP WHERE run_id = ?", (run_id,) + ) + conn.commit() + return outputs + + +def fork_work_graph_run(conn, source_node_exec_id, new_output_payload, target_folder_path, timeout=600): + """create_fork() + execute_fork() in one synchronous call -- what + tests and any fully-synchronous caller should use. HTTP handlers use + the two halves separately (see start_fork_in_background).""" + fork_info = create_fork(conn, source_node_exec_id, new_output_payload) + outputs = execute_fork(conn, fork_info, target_folder_path, timeout=timeout) + return {"run_id": fork_info["run_id"], "outputs": outputs} + + +def start_fork_in_background(db_path, source_node_exec_id, new_output_payload, target_folder_path, timeout=600): + """HTTP entrypoint: create_fork() runs synchronously (fast, DB-only) + so the caller gets run_id back immediately and can navigate the + console straight to it; execute_fork()'s real claude -p calls then run + in a background daemon thread, own connection (check_same_thread=False + -- same reasoning as scheduler.run_goal_to_completion's own conn, see + that module's comment: this can reach run_graph()'s multi-node + concurrent layers). Returns (run_id, thread) -- thread is already + started, callers/tests can .join() it if needed (mirrors + task_execution.start_task_execution_in_background's own Thread + return).""" + conn = schema.connect(db_path, check_same_thread=False) + fork_info = create_fork(conn, source_node_exec_id, new_output_payload) + + def run(): + try: + execute_fork(conn, fork_info, target_folder_path, timeout=timeout) + except Exception: + pass # already reflected in work_graph_run/work_node_execution status + finally: + conn.close() + + thread = threading.Thread(target=run, daemon=True) + thread.start() + return fork_info["run_id"], thread diff --git a/src/work_graph_run_query.py b/src/work_graph_run_query.py index 0e84758..24b1561 100644 --- a/src/work_graph_run_query.py +++ b/src/work_graph_run_query.py @@ -46,7 +46,8 @@ def list_work_graph_runs(conn, project_id, org_node=None, limit=20): query = ( "SELECT wgr.run_id, wgr.task_id, wgr.org_node, 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 COUNT(*) FROM work_node_execution wne WHERE wne.run_id = wgr.run_id) AS node_count, " + "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] @@ -61,6 +62,9 @@ def list_work_graph_runs(conn, project_id, org_node=None, limit=20): { "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], + # 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], } for r in rows ] @@ -72,7 +76,8 @@ class WorkGraphRunNotFoundError(Exception): def get_work_graph_run(conn, run_id): row = conn.execute( - "SELECT run_id, task_id, org_node, graph_manifest, status, started_at, completed_at " + "SELECT run_id, task_id, org_node, 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() @@ -82,6 +87,7 @@ def get_work_graph_run(conn, run_id): "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]), }