graphlemon: auto-commit uncommitted changes for 7055a15f-f35a-4b04-a990-8f3fcfb5…

…437e-main-checkout-drift

graphlemon: auto-commit uncommitted changes for 7055a15f-f35a-4b04-a990-8f3fcfb5…
…437e-main-checkout-drift
graphlemon
1 parent 577d9af0
Showing 6 changed files with 436 additions and 57 deletions
docs/todo/work-graph-partial-artifact-persistence-requirements.md
src/backend_server.py
src/e2e_pipeline.py
src/test_backend_server.py
src/test_e2e_pipeline.py
src/work_graph_orchestrate.py
+# Work Graph 产出应逐节点落盘，而不是整图成功才批量落盘——需求文档
+
+**日期**: 2026-08-07
+**性质**: 需求文档（BRD/PRD 层面），回应一次真实事故的根因排查——任务 `bfd3ee82`（导航栏功能）和 `b243dfe8`（`#/skills` 页面 bug，已有同目标的 `7fc3a7a3` 成功过）都撞上了同一个真实 bug。
+**状态**: ✅ 已实现（2026-08-07）。
+
+## 1. 背景
+
+`src/e2e_pipeline.py::plan_documents()` 和 `validate_implementation_with_own_graph()` 都是同一个模式：先调用 `work_graph_orchestrate.run_graph()` 跑完**整张图的全部节点**，拿到返回的 `outputs` 字典后，再用一个 `for node in topological_order(...)` 循环把每个节点的产出批量写成 `artifact` 行。
+
+问题：如果图里**任何一个节点**（哪怕是最后一个，比如 `test_script`）在 `run_graph()` 内部失败到耗尽重试（`ConsistencyCheckFailedError`），`run_graph()` 会直接抛异常，`outputs` 这个局部变量随着异常传播被整个丢弃——**哪怕 BRD/PRD/SDD/Benchmark 都已经成功生成，一个 artifact 都不会落盘**。
+
+真实复现链路（`bfd3ee82`）：
+1. Planner 的文档链 BRD→PRD→SDD→Benchmark→TestDataset→TestScript，前五个节点都成功，`TestScript` 节点因为另一个已修复的 bug（`check_script_compiles` 返回裸字符串导致 `join()` 逐字符拆解）连续两次验证失败，`run_graph()` 抛 `ConsistencyCheckFailedError`。
+2. 异常一路传到 `scheduler.py`/`task_execution.py` 的兜底处理，`task.status` 变成 `blocked_pending_human`，但此时**没有任何 artifact 被写入**。
+3. 人工点 Resume，`task.status` 恢复到之前记录的 `planning_brd_prd_sdd`（这是 Planner 的 POST-state，语义是"Planner 已完成，该轮到 Coder 了"）。
+4. `task_execution.py::run_or_resume_task_lifecycle()` 的 `planning_brd_prd_sdd` 分支**直接读取 SDD/PRD/Benchmark artifact 内容准备交给 Coder**——但因为第 2 步根本没写任何 artifact，`_artifact_content(..., "sdd")` 查不到行，抛 `TaskExecutionError("task ... has no sdd artifact to resume from")`，任务再次卡死，且这次连"重跑一次 Planner"都做不到（resume 逻辑假设走到这个状态时 Planner 一定已经把 artifact 写好了）。
+
+`b243dfe8` 是同一个根因的更早一次复现（当时判断为"陈旧重复项"，实际是同一个 bug，只是恰好有另一个任务 `7fc3a7a3` 后来重跑成功了，掩盖了这不是任务本身的问题）。
+
+## 2. 目标
+
+1. Work Graph 里每个节点**成功完成后立即落盘为 artifact**，不再等整张图全部成功才批量写。
+2. 一张图里某个节点失败（重试耗尽）时，**已经成功的节点产出必须已经落盘**——任务变成 `blocked_pending_human` 后，人工点 Resume 应该能利用这些已落盘的产出继续，而不是从零开始、也不是像现在这样连"没有 artifact"这种最基础的前提都保证不了。
+3. 覆盖两个已知有这个问题的调用点：`plan_documents()`（Planner 的 BRD→PRD→SDD→Benchmark→TestDataset→TestScript 链）和 `validate_implementation_with_own_graph()`（Validator 自己的 benchmark→test_dataset→test_script 链）——两处是同一个模式复制出来的,应该用同一个修复方式,不要修一处漏一处。
+
+## 3. 非目标
+
+1. **不改 `work_graph_orchestrate.run_graph()` 本身的执行逻辑**——那是 G（并行执行）刚重做过的核心路径,改动面和回归风险都更大。这次修复只动 `plan_documents()`/`validate_implementation_with_own_graph()` 这两个调用方自己怎么处理 `run_graph()` 的返回值/异常,不碰 `run_graph()` 内部。
+2. **不要求"部分失败"之后自动决定要不要继续跑剩下的节点**——那是失败处理策略的问题（比如要不要自动重试整张图剩下的部分），本需求只解决"已经成功的部分有没有被记住"这一件事,继续不继续是既有的 Org-level Loop/人工 Resume 机制的职责,不重新设计。
+3. **不改变任务失败后的对外状态语义**——图里有节点失败,任务依然应该变成 `blocked_pending_human`,不是这次改动想让它"看起来像成功了"。只是失败之后,已经做对的那部分工作不应该被扔掉。
+
+## 4. 功能需求
+
+| # | 需求描述 | 优先级 |
+|---|---|---|
+| FR-1 | `plan_documents()`/`validate_implementation_with_own_graph()` 里"把节点产出写成 artifact"这一步,数据来源要改成直接查 `work_node_execution` 表里 `status='done'` 的行（`output_payload` 已经是 REQ-06 断点续跑机制本来就会持久化的字段,不是新数据源）,而不是依赖 `run_graph()` 成功返回后的内存 `outputs` 字典——这样无论 `run_graph()` 是正常返回还是中途抛异常,只要节点在 DB 里已经是 `done`,就能被正确落盘。 | 必须 |
+| FR-2 | `run_graph()` 抛出的异常（`ConsistencyCheckFailedError` 等）在完成 FR-1 的落盘之后必须原样重新抛出——不能吞掉,任务该进 `blocked_pending_human` 还是要进,这次修复不改变失败语义（见非目标 3）。 | 必须 |
+| FR-3 | 落盘循环要能正确跳过"没跑到/没成功"的节点,不能因为某个节点没有 `output_payload` 就整个循环报错——已有的 `deps` 计算逻辑（"依赖没落盘就不算进 deps 里"）本来就是为这种情况设计的,复用它,不要重新发明。 | 必须 |
+| FR-4 | Planner 和 Validator 两处调用点要复用同一份"按 DB 里的 done 节点落盘"辅助逻辑,不要各写一份——两处现在的写法本来就是复制出来的,统一之后以后再出现第三处类似场景也有地方复用。 | 应该 |
+
+## 5. 验收标准
+
+1. 人为构造一个"前面节点成功、最后一个节点连续验证失败"的场景（复现 `bfd3ee82` 的真实序列），验证 `run_graph()` 抛异常之后，前面成功节点对应的 artifact 确实已经写入 `artifact` 表，且原始异常被正确重新抛出（调用方拿到的还是 `ConsistencyCheckFailedError`，不是别的）。
+2. 验证正常全部成功的路径行为不变——既有测试套件不需要修改断言就应该继续通过。
+3. 验证 Planner 侧修好之后，一个曾经因为这个 bug 卡死的任务，点 Resume 后不会再报"没有 SDD artifact 可恢复"。
+4. Validator 侧同样验证一遍（构造一个 Validator 自己的 benchmark→test_dataset→test_script 链中途失败的场景）。
+
+## 6. 约束与假设
+
+- 复用 REQ-06 已经建立的"`work_node_execution.output_payload` 是节点成功后的权威持久化数据"这个既有约定，不引入新的数据来源或新的持久化机制。
+- 这次修复完全在 `e2e_pipeline.py` 内部（两个调用方函数），不涉及 `work_graph_orchestrate.py`/`work_node_run.py` 的任何改动。
+
+## 7. 已实现
+
+新增 `src/e2e_pipeline.py::_write_completed_node_artifacts()`，`plan_documents()`/`validate_implementation_with_own_graph()` 都改成：`run_graph()` 包一层 `try/except`，无论成功还是失败都调用这个共享辅助函数落盘，失败时原始异常原样重新抛出。
+
+**实现过程中发现的一个比原设计更细的坑，FR-1 原文没写清楚**：`work_node_execution.status='done'` 只代表"通过了 schema 校验"，不代表"通过了 `verify_fn` 的语义检查"——一个节点被 `verify_fn` 拒绝、重试一次后仍然被拒绝（`ConsistencyCheckFailedError`），它自己和它的"-reflect"重试节点在 DB 里其实都是 `status='done'`（只是内容都不合格）。如果单纯按"状态是不是 done"来判断能不能落盘，会把**verify_fn 明确拒绝的内容**也当成"已完成"错误地写成 artifact。修复方式：给 `ConsistencyCheckFailedError` 加了 `node_id` 属性（`work_graph_orchestrate.py`），`_write_completed_node_artifacts` 用它（以及 `WorkGraphLayerError.errors`，覆盖未来多节点并行层的等价场景）排除掉这些节点，不只是排除"根本没跑完"的节点。
+
+同时顺带修了一个只有在这次实现验证时才会暴露的遗留 bug：`validate_implementation_with_own_graph()` 早些时候那次"结束时标记 `work_graph_run.status='done'`"的修复，只覆盖了成功路径——如果 `run_graph()` 本身抛出的不是 `ConsistencyCheckFailedError`（那个走的是 `_verify_and_maybe_regenerate` 自己的 `status='blocked'` 写入，已经安全），而是别的异常类型，`work_graph_run` 还是会卡在 `running`。这次统一走 `_write_completed_node_artifacts` 之后，失败路径已经不需要再单独处理这个——`_verify_and_maybe_regenerate` 自己的 `blocked` 写入本来就覆盖了目前唯一真实会走到这条路径的异常类型，测试里专门加了一条断言确认这一点没有回归。
+
+新增测试：`src/test_e2e_pipeline.py` 的 `PlanDocumentsTest.test_earlier_nodes_survive_a_later_node_failing_verification`、`ValidateImplementationWithOwnGraphTest.test_earlier_nodes_survive_a_later_node_failing_verification`——都验证"前面节点产出确实落盘、失败节点本身不落盘、原始异常原样抛出"。`python3 -m unittest discover -s src`（641 tests）+ `discover -s supervisor -p "test_*.py"`（126 tests）全绿，零回归。
@@ -540,8 +540,16 @@ class _Handler(BaseHTTPRequestHandler):
         except (org_loop_route.TaskNotBlockedError, org_loop_route.TaskNotValidatingError, ValueError) as e:
             self._send_json(400, {"error": "INVALID_PARAM", "detail": str(e)})
             return
+        # 'validating' added 2026-08-07 (real incident): task_execution.
+        # run_or_resume_task_lifecycle has resumed from 'validating' since
+        # the earlier same-day fix (skips straight to
+        # validate_implementation_with_own_graph, the readiness trigger
+        # already having run once) -- but THIS allowlist, a separate code
+        # path, still didn't know that, so a manual Resume on a task stuck
+        # in 'validating' flipped the status back with nothing ever
+        # actually continuing it.
         if pre_resume_reason not in ("budget_exceeded", "loop_limit_reached") and result.get("new_status") in (
-            "queued", "planning_brd_prd_sdd", "coding",
+            "queued", "planning_brd_prd_sdd", "coding", "validating",
         ):
             task_row = self.server.conn.execute(
                 "SELECT project_id, goal_text FROM task WHERE task_id = ?", (task_id,)
@@ -549,9 +557,17 @@ class _Handler(BaseHTTPRequestHandler):
             target_folder_path = self.server.conn.execute(
                 "SELECT target_folder_path FROM project WHERE project_id = ?", (task_row[0],)
             ).fetchone()[0]
-            task_execution.start_task_execution_in_background(
-                self.server.db_path, task_id, target_folder_path, task_row[1],
-            )
+            try:
+                task_execution.start_task_execution_in_background(
+                    self.server.db_path, task_id, target_folder_path, task_row[1],
+                )
+            except task_execution.TaskExecutionError:
+                # Already has an execution in flight (e.g. the
+                # reconciliation loop dispatched it a moment before this
+                # click landed) -- the resume itself already succeeded
+                # (route_org_loop.route_org_loop above), so still report
+                # 200; the in-flight run will get there on its own.
+                pass
         self._send_json(200, result)
     def _handle_task_execute(self, task_id):
@@ -572,7 +588,14 @@ class _Handler(BaseHTTPRequestHandler):
         target_folder_path = self.server.conn.execute(
             "SELECT target_folder_path FROM project WHERE project_id = ?", (project_id,)
         ).fetchone()[0]
-        task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text)
+        try:
+            task_execution.start_task_execution_in_background(self.server.db_path, task_id, target_folder_path, goal_text)
+        except task_execution.TaskExecutionError as e:
+            # 2026-08-07: already has an execution in flight (e.g. a rapid
+            # double-click, or the reconciliation loop won a race against
+            # this click) -- report it instead of a 500.
+            self._send_json(409, {"error": "ALREADY_IN_FLIGHT", "detail": str(e)})
+            return
         self._send_json(200, {"task_id": task_id, "started": True})
     def _handle_task_pause(self, task_id, reason):
@@ -439,12 +439,106 @@ def _combined_planner_verify_fn(node_artifact_types, target_folder_path):
     return verify_fn
+def _failed_node_ids(exc):
+    """2026-08-07: which node_id(s) a run_graph() failure means were
+    NEVER verify_fn-accepted, even if their work_node_execution row is
+    'done' (schema-valid only -- see ConsistencyCheckFailedError's own
+    docstring). Handles both the common case today (a bare
+    ConsistencyCheckFailedError -- Planner's/Validator's doc chains are
+    currently serial, one node per layer) and WorkGraphLayerError (G's
+    multi-node-layer aggregate, for whenever a fan-out graph hits this)."""
+    if isinstance(exc, work_graph_orchestrate.WorkGraphLayerError):
+        return {node_id for node_id, _ in exc.errors}
+    if isinstance(exc, work_graph_orchestrate.ConsistencyCheckFailedError) and exc.node_id:
+        return {exc.node_id}
+    return set()
+
+
+def _write_completed_node_artifacts(
+    conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, run_graph_error=None,
+):
+    """Writes an artifact for every node that's genuinely done AND
+    verify_fn-accepted, sourced from `work_node_execution` in the DB --
+    not a run_graph() return value, so this works identically whether
+    run_graph() returned normally (every node accounted for) or raised
+    partway through (only some nodes are).
+
+    2026-08-07 (real incident, docs/todo/work-graph-partial-artifact-
+    persistence-requirements.md): run_graph() raising discards its
+    would-be return value entirely (a Python exception drops the local
+    scope), so previously ANY node failing -- even the very last one in
+    an otherwise-fully-successful chain -- meant NOTHING got persisted,
+    silently throwing away real, already-generated, already-good content.
+    A resumed task then found no BRD/PRD/SDD artifacts to build on at all,
+    even though they'd been generated correctly the first time.
+
+    A node whose OWN "-reflect" retry exists (verify_fn rejected the
+    first attempt) uses the reflect attempt's content, not the original's
+    -- the reflect node_id isn't part of `graph_manifest` so the main loop
+    below never visits it directly. `_failed_node_ids(run_graph_error)`
+    nodes are excluded outright: their row is schema-valid ('done') but
+    verify_fn never accepted either attempt, so neither is trustworthy
+    content to persist.
+    """
+    done_rows = conn.execute(
+        "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'",
+        (run_id,),
+    ).fetchall()
+    done_outputs = {node_id: json.loads(output_payload) for node_id, output_payload in done_rows}
+    for node_id in list(done_outputs):
+        reflect_id = node_id + "-reflect"
+        if reflect_id in done_outputs:
+            done_outputs[node_id] = done_outputs[reflect_id]
+    for excluded in _failed_node_ids(run_graph_error):
+        done_outputs.pop(excluded, None)
+
+    artifact_ids = {}
+    for node in work_graph_orchestrate.topological_order(graph_manifest):
+        node_id = node["node_id"]
+        if node_id not in done_outputs:
+            continue
+        artifact_type = node_artifact_types[node_id]
+        deps = [
+            artifact_ids[node_artifact_types[dep_id]]
+            for dep_id in (node.get("depends_on") or [])
+            if node_artifact_types.get(dep_id) in artifact_ids
+        ]
+        result = artifact_write.write_artifact(
+            conn, target_folder_path, artifact_type, task_id, run_id,
+            ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext_for(artifact_type)),
+            done_outputs[node_id]["content"], depends_on_artifact_ids=deps,
+        )
+        artifact_ids[artifact_type] = result["artifact_id"]
+    return artifact_ids
+
+
+# test_dataset/test_script aren't necessarily Markdown (a script is real
+# code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately
+# generic container; Coder materializes it into a properly-named,
+# properly-placed real file under the target project's own test
+# conventions when it actually uses it, that's not this write step's job.
+_PLANNER_ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"}
+
+
+def _planner_ext_for(artifact_type):
+    return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md")
+
+
 def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
     """Architecture §11 criterion 1: a real Planner Work Graph
     (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
     work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark,
     each written via c_artifact_write with correct lineage.
     Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}.
+
+    2026-08-07: artifacts are now written from whatever's genuinely done in
+    the DB (see _write_completed_node_artifacts), not from run_graph()'s
+    in-memory return value -- so a failure partway through (e.g. the last
+    node in the chain) still persists everything that succeeded before it,
+    instead of discarding it all. The original run_graph() failure is
+    re-raised unchanged after salvaging -- this does not change whether or
+    when a task ends up blocked_pending_human, only what's left behind for
+    a human's Resume to build on when it does.
     """
     plan_result = work_graph_plan.plan_work_graph(
         conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout
@@ -453,35 +547,24 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
     graph_manifest = plan_result["graph_manifest"]
     node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}
-    outputs = work_graph_orchestrate.run_graph(
-        conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
-        _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
-        verify_fn=_combined_planner_verify_fn(node_artifact_types, target_folder_path),
+    run_graph_error = None
+    try:
+        work_graph_orchestrate.run_graph(
+            conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
+            _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
+            verify_fn=_combined_planner_verify_fn(node_artifact_types, target_folder_path),
+        )
+    except Exception as e:
+        run_graph_error = e
+
+    artifact_ids = _write_completed_node_artifacts(
+        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
+        _planner_ext_for, run_graph_error,
     )
-    # test_dataset/test_script aren't necessarily Markdown (a script is real
-    # code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately
-    # generic container; Coder materializes it into a properly-named,
-    # properly-placed real file under the target project's own test
-    # conventions when it actually uses it, that's not this write step's job.
-    _ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"}
+    if run_graph_error is not None:
+        raise run_graph_error
-    artifact_ids = {}
-    for node in work_graph_orchestrate.topological_order(graph_manifest):
-        node_id = node["node_id"]
-        artifact_type = node_artifact_types[node_id]
-        deps = [
-            artifact_ids[node_artifact_types[dep_id]]
-            for dep_id in (node.get("depends_on") or [])
-            if node_artifact_types.get(dep_id) in artifact_ids
-        ]
-        ext = _ARTIFACT_FILE_EXT.get(artifact_type, "md")
-        result = artifact_write.write_artifact(
-            conn, target_folder_path, artifact_type, task_id, run_id,
-            ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext), outputs[node_id]["content"],
-            depends_on_artifact_ids=deps,
-        )
-        artifact_ids[artifact_type] = result["artifact_id"]
     return {"run_id": run_id, "artifact_ids": artifact_ids}
@@ -663,30 +746,32 @@ def validate_implementation_with_own_graph(
     conn.commit()
     node_artifact_types = {n["node_id"]: n["artifact_type"] for n in _VALIDATOR_MANIFEST["nodes"]}
-    outputs = work_graph_orchestrate.run_graph(
-        conn, run_id, "validator", "validator-independent-verification", target_folder_path, _VALIDATOR_MANIFEST,
-        _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types),
-        _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
-        verify_fn=_test_script_verify_fn(node_artifact_types, script_artifact_type="validator_test_script"),
-        timeout=timeout,
+    # 2026-08-07: same run_graph()-partial-failure salvage as
+    # plan_documents() -- see _write_completed_node_artifacts's own
+    # docstring and docs/todo/work-graph-partial-artifact-persistence-
+    # requirements.md. `outputs` is only used below on the SUCCESS
+    # continuation (script execution + verdict prompt); the failure path
+    # re-raises before ever touching it.
+    run_graph_error = None
+    outputs = None
+    try:
+        outputs = work_graph_orchestrate.run_graph(
+            conn, run_id, "validator", "validator-independent-verification", target_folder_path, _VALIDATOR_MANIFEST,
+            _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types),
+            _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
+            verify_fn=_test_script_verify_fn(node_artifact_types, script_artifact_type="validator_test_script"),
+            timeout=timeout,
+        )
+    except Exception as e:
+        run_graph_error = e
+
+    artifact_ids = _write_completed_node_artifacts(
+        conn, target_folder_path, task_id, run_id, _VALIDATOR_MANIFEST, node_artifact_types,
+        lambda artifact_type: "txt" if artifact_type != "validator_benchmark" else "md", run_graph_error,
     )
-    artifact_ids = {}
-    for node in work_graph_orchestrate.topological_order(_VALIDATOR_MANIFEST):
-        node_id = node["node_id"]
-        artifact_type = node_artifact_types[node_id]
-        deps = [
-            artifact_ids[node_artifact_types[dep_id]]
-            for dep_id in (node.get("depends_on") or [])
-            if node_artifact_types.get(dep_id) in artifact_ids
-        ]
-        ext = "txt" if artifact_type != "validator_benchmark" else "md"
-        result = artifact_write.write_artifact(
-            conn, target_folder_path, artifact_type, task_id, run_id,
-            ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext), outputs[node_id]["content"],
-            depends_on_artifact_ids=deps,
-        )
-        artifact_ids[artifact_type] = result["artifact_id"]
+    if run_graph_error is not None:
+        raise run_graph_error
     script_content = outputs["validator_test_script"]["content"]
     script_execution = run_test_script_for_real(
@@ -711,6 +796,21 @@ def validate_implementation_with_own_graph(
     if script_execution["ran"] and script_execution["exit_code"] not in (0, None):
         verdict = "FAIL"
+    # 2026-08-07 (real incident): this function INSERTs its work_graph_run
+    # row as status='running' at the top and, on the happy path (verify_fn
+    # returns ok=True, so _verify_and_maybe_regenerate never touches
+    # work_graph_run.status at all -- it only writes 'reflecting'/
+    # 'verifying'/'blocked' on the RETRY path), never wrote anything past
+    # that -- regardless of PASS/FAIL/INCONCLUSIVE, a validation run that
+    # got this far genuinely finished executing. `ux_work_graph_run_active`
+    # (schema.py) is a unique index on (task_id, org_node) covering exactly
+    # this non-terminal status set, so a real re-verify later (org_level_
+    # loop.run_fix_loop after a FAIL) collided with this row on INSERT --
+    # "UNIQUE constraint failed: work_graph_run.task_id, work_graph_run.
+    # org_node" -- found for real via a task reconciliation resume.
+    conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,))
+    conn.commit()
+
     return {
         "verdict": verdict, "raw_response": response,
         "artifact_ids": artifact_ids, "script_execution": script_execution,
@@ -874,6 +874,59 @@ class TaskExecutionRouteTest(unittest.TestCase):
         self.assertEqual(status, 200)
         mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")
+    def test_manual_resume_from_validating_auto_continues(self):
+        # 2026-08-07 real incident: a task stuck in 'validating' (the
+        # readiness trigger already ran once -- see task_execution.py's
+        # 'validating' resume section) got its status flipped back by
+        # manual_resume but nothing ever actually continued it, because
+        # this allowlist hadn't been updated to match task_execution.
+        # run_or_resume_task_lifecycle's own 'validating' support added
+        # earlier the same day.
+        self.server.conn.execute(
+            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'execution_error' "
+            "WHERE task_id = 'task-1'"
+        )
+        self.server.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
+            "'{\"from_status\": \"validating\", \"to_status\": \"blocked_pending_human\"}')"
+        )
+        self.server.conn.commit()
+        with patch.object(backend_server.task_execution, "start_task_execution_in_background") as mock_start:
+            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
+        self.assertEqual(status, 200)
+        mock_start.assert_called_once_with(self.server.db_path, "task-1", "/tmp/proj-1", "the goal")
+
+    def test_execute_already_in_flight_returns_409_not_500(self):
+        with patch.object(
+            backend_server.task_execution, "start_task_execution_in_background",
+            side_effect=backend_server.task_execution.TaskExecutionError("task task-1 already has an execution in flight"),
+        ):
+            status, body = self._post("/api/tasks/task-1/execute")
+        self.assertEqual(status, 409)
+        self.assertEqual(body["error"], "ALREADY_IN_FLIGHT")
+
+    def test_manual_resume_still_returns_200_when_already_in_flight(self):
+        # The route_org_loop transition itself already succeeded; losing a
+        # race to dispatch (e.g. the reconciliation loop got there a moment
+        # earlier) isn't a failure of the resume request itself.
+        self.server.conn.execute(
+            "UPDATE task SET status = 'blocked_pending_human', blocked_reason = 'execution_error' "
+            "WHERE task_id = 'task-1'"
+        )
+        self.server.conn.execute(
+            "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+            "VALUES ('evt-1', 'proj-1', 'task-1', 'task_status_changed', "
+            "'{\"from_status\": \"coding\", \"to_status\": \"blocked_pending_human\"}')"
+        )
+        self.server.conn.commit()
+        with patch.object(
+            backend_server.task_execution, "start_task_execution_in_background",
+            side_effect=backend_server.task_execution.TaskExecutionError("already in flight"),
+        ):
+            status, body = self._post("/api/tasks/task-1/route", {"decided_route": "manual_resume"})
+        self.assertEqual(status, 200)
+
 def _ws_handshake_request(path, key):
     return (
@@ -29,7 +29,7 @@ import tempfile
 import unittest
 from unittest.mock import patch
-from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute
+from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute, work_graph_orchestrate
 _FAKE_DOCS = {
     "brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake",
@@ -120,6 +120,58 @@ class PlanDocumentsTest(PipelineTestBase):
         ).fetchone()
         self.assertEqual(run_row, ("planner", "planning"))
+    def test_earlier_nodes_survive_a_later_node_failing_verification(self):
+        # 2026-08-07 real incident (docs/todo/work-graph-partial-artifact-
+        # persistence-requirements.md): test_script fails verify_fn twice
+        # (original + one regeneration) -- ConsistencyCheckFailedError must
+        # still propagate (failure semantics unchanged), but BRD/PRD/SDD/
+        # Benchmark/TestDataset -- all genuinely produced and accepted --
+        # must be persisted as artifacts, not silently discarded.
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', "
+            "'planning_brd_prd_sdd', 'planner')"
+        )
+        self.conn.commit()
+
+        # _fake_plan_subprocess_run ignores its args and always returns a
+        # canned success payload -- fine for the "claude" planning call it's
+        # meant for, but `patch("src.work_graph_plan.subprocess.run", ...)`
+        # patches the ONE global `subprocess` module every other importer
+        # shares too (work_graph_plan does `import subprocess`, not
+        # `from subprocess import run`), so it would ALSO silently hijack
+        # check_script_compiles's real `py_compile`/`node --check` calls,
+        # making them always report success regardless of actual content --
+        # exactly defeating the failure this test needs to trigger. Only
+        # intercept the "claude" planning call; let everything else
+        # (including this test's real compile checks) run for real.
+        real_subprocess_run = subprocess.run
+
+        def _discriminating_fake(*args, **kwargs):
+            cmd = args[0] if args else kwargs.get("args")
+            if cmd and cmd[0] == "claude":
+                return _fake_plan_subprocess_run(*args, **kwargs)
+            return real_subprocess_run(*args, **kwargs)
+
+        with patch("src.work_graph_plan.subprocess.run", side_effect=_discriminating_fake), \
+             patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [
+                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
+                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
+                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
+                {"content": _FAKE_DOCS["test_dataset"]},
+                {"content": "this is prose, not a real script"},  # test_script: fails compile check
+                {"content": "still not real code either"},  # the one regeneration attempt: also fails
+            ]
+            with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
+                e2e_pipeline.plan_documents(self.conn, "task-1", "goal", self.tmp.name)
+
+        rows = self.conn.execute("SELECT artifact_type FROM artifact WHERE task_id = 'task-1'").fetchall()
+        persisted = {r[0] for r in rows}
+        self.assertEqual(persisted, {"brd", "prd", "sdd", "benchmark", "test_dataset"})
+        self.assertNotIn("test_script", persisted)
+
 class RunEndToEndTest(PipelineTestBase):
     def test_full_pipeline_wiring(self):
@@ -516,9 +568,83 @@ class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
     def test_creates_its_own_work_graph_run_row(self):
         result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
         run_row = self.conn.execute(
-            "SELECT org_node FROM work_graph_run WHERE task_id = 'task-1'"
+            "SELECT org_node, status FROM work_graph_run WHERE task_id = 'task-1'"
         ).fetchone()
         self.assertEqual(run_row[0], "validator")
+        # 2026-08-07 (real incident): this used to stay 'running' forever on
+        # the happy path -- verify_fn returning ok=True means
+        # _verify_and_maybe_regenerate never touches work_graph_run.status
+        # at all, so nothing else in this function's old code ever marked
+        # it terminal. schema.py's `ux_work_graph_run_active` unique index
+        # covers exactly this non-terminal status set, so a stuck 'running'
+        # row silently blocked any later re-verify of the SAME task from
+        # ever creating its own new validator run.
+        self.assertEqual(run_row[1], "done")
+
+    def test_second_validation_run_does_not_collide_with_the_first(self):
+        # The actual regression: org_level_loop.run_fix_loop re-verifying
+        # after a FAIL (or a reconciliation-loop resume retrying) calls this
+        # function again for the SAME task_id -- must not raise
+        # "UNIQUE constraint failed: work_graph_run.task_id,
+        # work_graph_run.org_node" against a first run stuck non-terminal.
+        self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")
+
+        # Benchmark/dataset content must differ from the first call's --
+        # write_artifact does a real `git commit`, which fails with "nothing
+        # to commit" if a second write produces byte-identical file content
+        # (the file path isn't run_id-scoped). Real re-verify runs would
+        # naturally regenerate slightly different content each time; this
+        # just makes that explicit rather than accidentally relying on it.
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
+             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nfixed now"):
+            mock_execute.side_effect = [
+                {"content": "# Validator's own Benchmark (re-verify)"},
+                {"content": "N/A: no dataset needed, still"},
+                {"content": "import sys\nsys.exit(0)\n"},
+            ]
+            second = e2e_pipeline.validate_implementation_with_own_graph(
+                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
+            )
+
+        self.assertEqual(second["verdict"], "PASS")
+        rows = self.conn.execute(
+            "SELECT status FROM work_graph_run WHERE task_id = 'task-1' ORDER BY rowid"
+        ).fetchall()
+        self.assertEqual([r[0] for r in rows], ["done", "done"])
+
+    def test_earlier_nodes_survive_a_later_node_failing_verification(self):
+        # 2026-08-07 (same real incident/fix as PlanDocumentsTest's own
+        # test of this name): validator_test_script fails verify_fn twice
+        # -- ConsistencyCheckFailedError must still propagate, but
+        # validator_benchmark/validator_test_dataset (genuinely produced
+        # and accepted) must be persisted, not discarded along with it.
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            mock_execute.side_effect = [
+                {"content": "# Validator's own Benchmark"},
+                {"content": "N/A: no dataset needed"},
+                {"content": "this is prose, not a real script"},  # fails compile check
+                {"content": "still not real code either"},  # the one regeneration attempt: also fails
+            ]
+            with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
+                e2e_pipeline.validate_implementation_with_own_graph(
+                    self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
+                )
+
+        rows = self.conn.execute("SELECT artifact_type FROM artifact WHERE task_id = 'task-1'").fetchall()
+        persisted = {r[0] for r in rows}
+        self.assertEqual(persisted, {"validator_benchmark", "validator_test_dataset"})
+        self.assertNotIn("validator_test_script", persisted)
+        # the work_graph_run itself must NOT be left stuck 'running' either
+        # (ux_work_graph_run_active would block any later re-verify) --
+        # _verify_and_maybe_regenerate's own 'blocked' write on this path
+        # already covers it; confirming it here catches a regression if
+        # that ever stops being true.
+        run_status = self.conn.execute(
+            "SELECT status FROM work_graph_run WHERE task_id = 'task-1'"
+        ).fetchone()[0]
+        self.assertEqual(run_status, "blocked")
 if __name__ == "__main__":
@@ -82,7 +82,21 @@ def _log_event(conn, node_exec_id, event_type, **payload):
 class ConsistencyCheckFailedError(Exception):
     """`verify_fn` rejected a node's output twice in a row (original +
     one regeneration attempt) -- `work_graph_run.status` is already
-    'blocked' by the time this is raised."""
+    'blocked' by the time this is raised.
+
+    `node_id` (2026-08-07): the rejected node's id, so a caller salvaging
+    OTHER nodes' already-persisted artifacts after a partial run_graph()
+    failure (see e2e_pipeline.py's `_write_completed_node_artifacts`) can
+    tell this one apart from a node that's `work_node_execution.status
+    ='done'` for a genuinely different reason -- 'done' there only means
+    "passed schema validation", not "passed verify_fn". The original AND
+    its "-reflect" retry both end up schema-valid-and-'done' even when
+    THIS error fires (verify_fn rejected both), so status alone can't
+    distinguish a verify_fn-rejected node from a genuinely good one."""
+
+    def __init__(self, message, node_id=None):
+        super().__init__(message)
+        self.node_id = node_id
 class WorkGraphLayerError(Exception):
@@ -464,7 +478,8 @@ def _verify_and_maybe_regenerate(
         conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
         conn.commit()
         raise ConsistencyCheckFailedError(
-            "node {} still failed verification after one regeneration attempt: {}".format(node_id, feedback2)
+            "node {} still failed verification after one regeneration attempt: {}".format(node_id, feedback2),
+            node_id=node_id,
         )
     conn.execute("UPDATE work_graph_run SET status = 'verifying' WHERE run_id = ?", (run_id,))