D: Work Graph time-travel debugging (fork from a historical node)

Implements docs/todo/work-graph-time-travel-debugging-design.md, backend and console together (user confirmed 2026-08-06). Key finding that shrank the v1 scope: ux_work_graph_run_active is a PARTIAL unique index (only covers non-terminal statuses), not a blanket one -- forking from a parent run that's already reached a terminal status never collides with it at all. v1 restricts forking to exactly that case, so no index/constraint change is needed, only two new nullable columns (parent_run_id, fork_of_node_exec_id). Resolved the requirements doc's FR-1 ambiguity ("override input_payload, re-trigger execution") as OUTPUT SUBSTITUTION instead: the fork point's own output_payload is replaced with the human-supplied content directly, it is never re-executed via a new claude -p call -- only the strict downstream subgraph actually re-runs. This sidesteps reconciling the different build_prompt/input_payload conventions each caller uses (Planner's artifact_type lookup, E's generic task_prompt field, F's jury) -- work_graph_fork.py only needs the persisted graph_manifest and each node's output_payload. Implementation needs ZERO changes to the execution engine: pre-seeding every non-downstream node as a 'done' work_node_execution row before calling the unmodified run_graph() on the unmodified 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. _reconstruct_build_prompt supports two graph shapes generically from persisted state alone (Planner's canonical doc chain via e2e_pipeline._planner_prompt_builder, and E's generic task_prompt field); anything else raises ForkGraphShapeUnsupportedError rather than producing broken prompts. New: src/work_graph_fork.py (create_fork/execute_fork/fork_work_graph_run/ start_fork_in_background, the last splitting fast DB-only setup from slow real execution so an HTTP caller gets run_id back immediately, same pattern task_execution.start_task_execution_in_background already uses). work_graph_run gets parent_run_id/fork_of_node_exec_id (SCHEMA_SQL + migrate_add_work_graph_fork_columns for existing databases). work_graph_run_query.py/task_query.py surface both fields. New endpoint POST /api/work-nodes/:node_exec_id/fork. Console: a "从这里分叉" form on each 'done' node in v_run_detail, a fork badge linking back to the parent run, and a 🔀 marker on v_graph_console's run list -- no diff/merge UI (non-goal #3), read-only branch visibility only. Verified end-to-end for real: started the actual backend, sent a real HTTP fork request, confirmed the new run persisted with correct parent_run_id/fork_of_node_exec_id lineage. Tests cover requirements doc acceptance standards 1-3 (PRD swapped -> SDD/Benchmark genuinely re-execute, BRD untouched; original run completely unaffected; the same node forked twice produces independent branches) plus schema migration idempotency, query-layer field surfacing, the HTTP route, and 3 frontend smoke-test assertions. Full src suite (608 tests) + all 3 frontend smoke tests pass. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

D: Work Graph time-travel debugging (fork from a historical node)
Implements docs/todo/work-graph-time-travel-debugging-design.md, backend and console together (user confirmed 2026-08-06). Key finding that shrank the v1 scope: ux_work_graph_run_active is a PARTIAL unique index (only covers non-terminal statuses), not a blanket one -- forking from a parent run that's already reached a terminal status never collides with it at all. v1 restricts forking to exactly that case, so no index/constraint change is needed, only two new nullable columns (parent_run_id, fork_of_node_exec_id). Resolved the requirements doc's FR-1 ambiguity ("override input_payload, re-trigger execution") as OUTPUT SUBSTITUTION instead: the fork point's own output_payload is replaced with the human-supplied content directly, it is never re-executed via a new claude -p call -- only the strict downstream subgraph actually re-runs. This sidesteps reconciling the different build_prompt/input_payload conventions each caller uses (Planner's artifact_type lookup, E's generic task_prompt field, F's jury) -- work_graph_fork.py only needs the persisted graph_manifest and each node's output_payload. Implementation needs ZERO changes to the execution engine: pre-seeding every non-downstream node as a 'done' work_node_execution row before calling the unmodified run_graph() on the unmodified 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. _reconstruct_build_prompt supports two graph shapes generically from persisted state alone (Planner's canonical doc chain via e2e_pipeline._planner_prompt_builder, and E's generic task_prompt field); anything else raises ForkGraphShapeUnsupportedError rather than producing broken prompts. New: src/work_graph_fork.py (create_fork/execute_fork/fork_work_graph_run/ start_fork_in_background, the last splitting fast DB-only setup from slow real execution so an HTTP caller gets run_id back immediately, same pattern task_execution.start_task_execution_in_background already uses). work_graph_run gets parent_run_id/fork_of_node_exec_id (SCHEMA_SQL + migrate_add_work_graph_fork_columns for existing databases). work_graph_run_query.py/task_query.py surface both fields. New endpoint POST /api/work-nodes/:node_exec_id/fork. Console: a "从这里分叉" form on each 'done' node in v_run_detail, a fork badge linking back to the parent run, and a 🔀 marker on v_graph_console's run list -- no diff/merge UI (non-goal #3), read-only branch visibility only. Verified end-to-end for real: started the actual backend, sent a real HTTP fork request, confirmed the new run persisted with correct parent_run_id/fork_of_node_exec_id lineage. Tests cover requirements doc acceptance standards 1-3 (PRD swapped -> SDD/Benchmark genuinely re-execute, BRD untouched; original run completely unaffected; the same node forked twice produces independent branches) plus schema migration idempotency, query-layer field surfacing, the HTTP route, and 3 frontend smoke-test assertions. Full src suite (608 tests) + all 3 frontend smoke tests pass. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 762ba779
Showing 15 changed files with 1119 additions and 17 deletions
docs/todo/graph-engineering-open-items.md
docs/todo/work-graph-time-travel-debugging-design.md
frontend/app.js
frontend/style.css
frontend/test_run_detail_smoke.js
src/backend_server.py
src/schema.py
src/task_query.py
src/test_backend_server.py
src/test_schema.py
src/test_task_query.py
src/test_work_graph_fork.py
src/test_work_graph_run_query.py
src/work_graph_fork.py
src/work_graph_run_query.py
@@ -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 术语对照 | — | ✅ 已完成 | — | — | — |
+# 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` 非空时显示一个"🔀 从 `<fork_of_node_exec_id 对应的 node_id>` 分叉自 `<parent_run_id 前 8 位>`"的标记，点击可跳转到父 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 节）。
@@ -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 '<div class="hint">该运行尚无节点</div>';
   const items = run.nodes.map((wn) => {
     const status = wn.status;
+    const forkHtml = status === "done" ? (
+      '<div class="fork-panel">'
+      + '<button class="btn btn-fork" type="button">🔀 从这里分叉</button>'
+      + '<div class="fork-form" style="display:none;">'
+      + '<textarea class="fork-content" rows="4" placeholder="新的 content（替换这个节点的产出，下游节点会用这份新内容真正重新执行）"></textarea>'
+      + '<button class="btn btn-fork-submit" type="button">提交分叉</button>'
+      + '<div class="fork-error error-banner" style="display:none;"></div>'
+      + "</div>"
+      + "</div>"
+    ) : "";
     return (
       '<li class="status-' + status + '">'
       + '<div class="todo-row" data-node-exec-id="' + wn.node_exec_id + '">'
@@ -1502,6 +1518,7 @@ function renderRunNodeListHtml(run) {
       + '<div class="todo-detail" id="todo-detail-' + wn.node_exec_id + '" data-status="' + status + '">'
       + (wn.reflection_notes ? "reflection_notes: " + escapeHtml(wn.reflection_notes) + "<br>" : "")
       + '<div class="log-viewer-slot" id="log-slot-' + wn.node_exec_id + '"></div>'
+      + forkHtml
       + "</div>"
       + "</li>"
     );
@@ -1509,6 +1526,58 @@ function renderRunNodeListHtml(run) {
   return '<ul class="todo-tree">' + items + "</ul>";
 }
+// 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 (
+    '<div class="fork-badge">🔀 从节点 <code>' + escapeHtml(run.fork_of_node_exec_id) + '</code> 分叉自 '
+    + '<a href="#/runs/' + encodeURIComponent(run.parent_run_id) + '">' + escapeHtml(run.parent_run_id) + "</a></div>"
+  );
+}
+
+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 = (
       "<h1>" + escapeHtml(run.org_node) + " Work Graph Run</h1>"
+      + renderForkBadgeHtml(run)
       + '<div class="meta-row">'
       + "<div><b>RUN_ID</b>" + escapeHtml(run.run_id) + "</div>"
       + "<div><b>STATUS</b>" + escapeHtml(run.status) + "</div>"
@@ -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 ? '<div class="hint">暂无 Graph 运行记录</div>' : (
     '<ul class="simple-list" id="gc_runs_list">' + runs.slice(0, 10).map((r) =>
       '<li data-run-id="' + r.run_id + '" style="cursor:pointer;"><span>' + escapeHtml(r.org_node) + " · run "
-      + r.run_id.slice(0, 8) + "</span><span>" + escapeHtml(r.status) + " · " + r.node_count + " 节点</span></li>"
+      + r.run_id.slice(0, 8) + (r.parent_run_id ? " 🔀" : "") + "</span><span>"
+      + escapeHtml(r.status) + " · " + r.node_count + " 节点</span></li>"
     ).join("") + "</ul>"
   );
@@ -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; }
@@ -53,6 +53,9 @@ const src = fs.readFileSync(path.join(__dirname, &quot;app.js&quot;), &quot;utf8&quot;);
 // 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");
@@ -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 = &quot;/api/artifacts/&quot;
 _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:
@@ -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()
@@ -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
@@ -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."""
@@ -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()
@@ -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
+"""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()
@@ -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()
+"""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
@@ -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]),
     }