diff --git a/docs/todo/work-graph-partial-artifact-persistence-requirements.md b/docs/todo/work-graph-partial-artifact-persistence-requirements.md new file mode 100644 index 0000000..61c35f4 --- /dev/null +++ b/docs/todo/work-graph-partial-artifact-persistence-requirements.md @@ -0,0 +1,62 @@ +# 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）全绿，零回归。 diff --git a/src/backend_server.py b/src/backend_server.py index 9788236..ca0bd35 100644 --- a/src/backend_server.py +++ b/src/backend_server.py @@ -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): diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 096388f..5613095 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -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, diff --git a/src/test_backend_server.py b/src/test_backend_server.py index b4b9091..eccea7b 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -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 ( diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index edd6177..2ff4d79 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -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__": diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py index 21fab05..d51fbe3 100644 --- a/src/work_graph_orchestrate.py +++ b/src/work_graph_orchestrate.py @@ -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,))