From a68f0adb081b41b0c1e8abcf9bf4d0339dd6e11e Mon Sep 17 00:00:00 2001 From: bboonnee Date: Wed, 12 Aug 2026 22:32:16 +0800 Subject: [PATCH] Org Graph unification Phase 2c: Validator self-plans its verification graph --- docs/todo/org-graph-unification-phase2-universal-self-planning-design.md | 2 +- src/e2e_pipeline.py | 305 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------------------------------------------------- src/task_execution.py | 26 ++++++++++++++++++++++++-- src/test_e2e_pipeline.py | 435 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- src/test_task_execution.py | 195 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------- src/work_graph_plan.py | 59 ++++++++++++++++++++++++++++++++++++++++++++++++++++------- 6 files changed, 663 insertions(+), 359 deletions(-) diff --git a/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md b/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md index be0c1bb..d743639 100644 --- a/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md +++ b/docs/todo/org-graph-unification-phase2-universal-self-planning-design.md @@ -3,7 +3,7 @@ **日期**: 2026-08-08 **性质**: 详细设计，对应 `docs/todo/org-graph-unification-and-universal-self-planning-requirements.md` 的 Phase 2（FR-2/FR-3/FR-4），按 §4/§4.1 已拍板的合图（方案 A）路径设计。 **依赖**: 硬依赖 Phase 1（`docs/done/org-graph-unification-phase1-review-gate-design.md`）——已实现（commit `ea19c7d`，2026-08-08）。§9 接口核对清单已于 2026-08-12 逐条核对完成：三项全部吻合（`plan_and_review_documents()` 经 `_review_planner_output_with_retries()` 透传 `plan_documents()` 的 `{"run_id", "artifact_ids"}` 返回形状，含 `run_id`；`task.status` 新值确认是 `reviewing_planner_output`；`roles/planner_reviewer/role.json` 目前 `skills: ["planner-output-review", "unattended-agent-operations"]`，未预留 §7.1 新 skill 位置，Phase 2 实施时新增一行即可，不是不兼容）。 -**状态**: Phase 2a（commit `86fc0bb`）+ Phase 2b（commit `7c46a2f`）已实现，2026-08-12。Phase 2c（Validator 接入自规划 + `org_node` 列删除 + 精确 loop_back 收口）待实施，详见 §10。**2026-08-08 修订**：写作过程中，需求文档 §7 的四个开放问题已被用户直接拍板（详见该文档 §7 对应条目），其中两条推翻了本文档最初的设计判断——Coder/Validator 自规划改为"建议节点清单"而非完全自由（§3），`work_graph_run.org_node` 改为彻底移除而非保留（§2.1）——本版本已按拍板结果改写，不是原始草稿。**2026-08-08 追加**（同日，见需求文档新增 §3.1）：FR-4 的目标被明确为"失败即进化"完整方法论（判断失败 → Reflection-review 动态诊断 → 系统性缺陷则先 Fix-Skill → 跳回执行），当时 §5 设计的 `dynamic_loop_back` 只覆盖第 1/2/4 步，第 3 步（Fix-Skill）是已知缺口。**2026-08-12 补齐**：§5 已扩写为完整四步设计（§5.4 Reflection-review 现在同时判断目标节点和缺陷性质，§5.5 新增 Fix-Skill，真实调用既有但从未真正投入使用的 `skill_update.py::update_skill()`），过程中发现两个既有事实（`skills:` frontmatter 只按名字动态加载内容、不需要重新部署；`org_level_loop.py` 的 `_ROUTE_FOR_CATEGORY` 从来没有 `skill_defect` 这个 key，`update_skill()` 在这个代码库里投产以来从未被真正调用过）——详见 §5 开头。由哪个身份执笔改 Skill 已通过 `AskUserQuestion` 确认（2026-08-12）：统一用 Validator，见 §5.5。 +**状态**: Phase 2a（commit `86fc0bb`）+ Phase 2b（commit `7c46a2f`）已实现，2026-08-12。Phase 2c 的 Validator 自规划改造（§6 核心部分）已实现，2026-08-12：`validate_implementation_with_own_graph()` 改为 `plan_work_graph(conn, "validator", ..., append_to_run_id=run_id)` 追加进 Planner/Coder 共享的同一个 `work_graph_run`，终审 verdict 变成图内一个真实节点（`node_type='final_verdict'`，`judgment.kind='mechanical'`、`check_ref='verdict_says_pass'`，`on_fail.action='dynamic_loop_back'`——Validator 图里唯一允许用这个 action 的节点），硬门槛（脚本真实非零退出码强制 FAIL，不听 LLM 辩解）通过 `_verdict_node_build_prompt`/`_verdict_says_pass_check` 的共享可变状态闭包保留，Python 强制而非只是 prompt 里说说。`_VALIDATOR_MANIFEST`/`_VALIDATOR_DOC_PROMPTS`/`_VALIDATOR_VERDICT_PROMPT`/`_validator_prompt_builder` 保留但不再使用（同 `_CODER_PROMPT` 先例）。`python3 -m unittest discover -s src`（756 tests）+ `discover -s supervisor`（126 tests）全绿。**尚未实施**：§2.1（`org_node` 列删除）、§2.2（`ux_work_graph_run_active` 收紧到 `UNIQUE(task_id)`）、§7.1（新 skill `work-graph-node-review` + 三个 role.json 追加）、§7.2 的收口验证（见 §10 Phase 2c 小节末尾新增的分析：一旦 verdict 节点自己的 `dynamic_loop_back` 内部消化重试，`validate_implementation_with_own_graph()` 正常返回时 verdict 必为 PASS，`org_level_loop.py` 的旧粗粒度路由对这条新路径自然不可达，不需要显式新旧流程判断——决定保留旧分支代码不删，加注释说明，不做额外改动）。**2026-08-08 修订**：写作过程中，需求文档 §7 的四个开放问题已被用户直接拍板（详见该文档 §7 对应条目），其中两条推翻了本文档最初的设计判断——Coder/Validator 自规划改为"建议节点清单"而非完全自由（§3），`work_graph_run.org_node` 改为彻底移除而非保留（§2.1）——本版本已按拍板结果改写，不是原始草稿。**2026-08-08 追加**（同日，见需求文档新增 §3.1）：FR-4 的目标被明确为"失败即进化"完整方法论（判断失败 → Reflection-review 动态诊断 → 系统性缺陷则先 Fix-Skill → 跳回执行），当时 §5 设计的 `dynamic_loop_back` 只覆盖第 1/2/4 步，第 3 步（Fix-Skill）是已知缺口。**2026-08-12 补齐**：§5 已扩写为完整四步设计（§5.4 Reflection-review 现在同时判断目标节点和缺陷性质，§5.5 新增 Fix-Skill，真实调用既有但从未真正投入使用的 `skill_update.py::update_skill()`），过程中发现两个既有事实（`skills:` frontmatter 只按名字动态加载内容、不需要重新部署；`org_level_loop.py` 的 `_ROUTE_FOR_CATEGORY` 从来没有 `skill_defect` 这个 key，`update_skill()` 在这个代码库里投产以来从未被真正调用过）——详见 §5 开头。由哪个身份执笔改 Skill 已通过 `AskUserQuestion` 确认（2026-08-12）：统一用 Validator，见 §5.5。 --- diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 778e226..af95d26 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -200,6 +200,15 @@ _VALIDATOR_PROMPT = ( "Goal: {goal_text}\n\nBenchmark:\n\n{benchmark_content}" ) +# 2026-08-12 (Phase 2c): retired now that Validator self-plans its own +# verification graph (validate_implementation_with_own_graph() below) via +# plan_work_graph(conn, "validator", ..., append_to_run_id=run_id) instead +# of running this hardcoded fixed chain -- the "held-out eval set, not the +# Coder's own dev set" framing that motivated this dict lives on in +# _VALIDATOR_PLANNING_CONTEXT's guidance text instead. Kept here, unused, +# for the same "a future revert would want it back verbatim" reason +# _CODER_PROMPT above is. +# # 2026-08-06: Validator gets its OWN benchmark/test_dataset/test_script, # generated independently from SDD/PRD -- NOT reusing the Planner's/Coder's # own copies. If the Coder could game the Planner's test_script (e.g. @@ -240,10 +249,12 @@ _VALIDATOR_DOC_PROMPTS = { ), } -# validator_test_script node temporarily disabled (2026-08-07, same -# stopgap as work_graph_plan._PROMPT_TEMPLATE -- see its comment for the -# full rationale/upgrade path): 6/6 real repro rate for narrative-instead- -# of-code output, docs/todo/validator-test-script-narrative-output-issue.md. +# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_DOC_PROMPTS above -- +# see that comment. validator_test_script node temporarily disabled +# (2026-08-07, same stopgap as work_graph_plan._PROMPT_TEMPLATE -- see its +# comment for the full rationale/upgrade path): 6/6 real repro rate for +# narrative-instead-of-code output, docs/todo/validator-test-script- +# narrative-output-issue.md. _VALIDATOR_MANIFEST = { "nodes": [ {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []}, @@ -255,6 +266,8 @@ _VALIDATOR_MANIFEST = { "edges": [], } +# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above -- see +# that comment. _VALIDATOR_VERDICT_PROMPT = ( "Give your final verdict for this goal, informed by your own independent Benchmark below " "and the REAL execution result of your own test script (if you wrote one -- 'N/A' means you " @@ -303,7 +316,12 @@ def _planner_prompt_builder(goal_text, node_artifact_types): def _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types): - """Same shape as _planner_prompt_builder, but the FIRST node + """2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above -- + see that comment; validate_implementation_with_own_graph() now uses + dispatch_build_prompt(_planner_prompt_builder(...), generic_node_prompt_builder) + like every other self-planning org_node. + + Same shape as _planner_prompt_builder, but the FIRST node (validator_benchmark) needs sdd_content/prd_content from OUTSIDE validator's own tiny graph -- those aren't a dependency within this graph, they come from the Planner's already-completed one. Passed @@ -644,12 +662,16 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): now carries its own `judgment` field (work_graph_plan.py's `_PLANNER_NODE_CATALOG` suggests one per node type, the Planner can reuse or override it), which `run_graph()` resolves itself per node. - `_combined_planner_verify_fn`/`_prd_consistency_verify_fn`/ - `_test_script_verify_fn` are NOT deleted -- `validate_implementation_ - with_own_graph()` (Validator's still-hardcoded manifest, unaffected by - this round) continues to use them (FR-A8). `mechanical_check_registry` - is still passed unconditionally: a self-planned node declaring - `judgment.kind == "mechanical"` (e.g. reusing the catalog's suggested + `_combined_planner_verify_fn`/`_prd_consistency_verify_fn` are NOT + deleted -- this function still uses them below (FR-A8). + `_test_script_verify_fn` is ALSO still used below for Planner's own + `test_script` node (2026-08-12: Validator's own equivalent node no + longer needs it -- validate_implementation_with_own_graph() now + self-plans too, and its `final_verdict` node's hard gate is enforced + via a mechanical_check_registry closure instead, see that function's + own docstring). `mechanical_check_registry` is still passed + unconditionally: a self-planned node declaring `judgment.kind == + "mechanical"` (e.g. reusing the catalog's suggested `check_script_compiles` for a test_script node) needs it registered, same as before. """ @@ -1089,105 +1111,214 @@ def _format_script_execution_result(result): ).format(result["exit_code"], result["stdout"][-4000:], result["stderr"][-4000:]) +def _verdict_says_pass_check(shared_state): + """Phase 2c (2026-08-12): mechanical judgment for the final_verdict + node -- `shared_state` is a plain dict the matching build_prompt + wrapper (_verdict_node_build_prompt) populates with the REAL script + execution result as a side effect, read back here since a mechanical + check callable only ever receives `content`, no side channel of its + own. Two ways to fail: + 1. Hard gate (user-confirmed 2026-08-06, preserved verbatim): a real + non-zero exit from Validator's own test script is not up for LLM + debate -- force-reject regardless of what the verdict text says. + 2. The verdict itself isn't PASS -- INCONCLUSIVE is treated the SAME + as FAIL here (ok=False), not specially accepted: this node's job + is to keep trying (bounded by the loop_back attempt budget, same + as everything else on_fail.action='dynamic_loop_back' governs) to + actually resolve the goal, not to settle for genuine uncertainty + on the first attempt. If the budget is exhausted without reaching + PASS, run_graph() raises ConsistencyCheckFailedError same as any + other exhausted loop -- this function then never returns a non-PASS + verdict normally, it either returns PASS or raises (see this + function's own docstring for why that's the right contract).""" + def check(content): + script_execution = shared_state.get("script_execution") + if script_execution and script_execution["ran"] and script_execution["exit_code"] not in (0, None): + return False, "real test script exited non-zero (exit_code={}) -- hard gate, not up for debate".format( + script_execution["exit_code"] + ) + verdict = _parse_verdict_line(content) + if verdict == "PASS": + return True, None + return False, "verdict was {}, not PASS".format(verdict) + return check + + +def _verdict_node_build_prompt(inner_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout): + """Wraps a build_prompt(node, outputs) callable: for the ONE node + declaring node_type='final_verdict', performs REAL test script + execution (if a node_type='test_script' dependency produced real + content) as a side effect before returning the prompt -- embedding + the real result into both the prompt text (so the model sees it) and + `shared_state` (so _verdict_says_pass_check's hard gate can enforce it + in Python, not just ask nicely). Every other node passes through to + `inner_build_prompt` completely unchanged.""" + def build_prompt(node, outputs): + if node.get("node_type") != "final_verdict": + return inner_build_prompt(node, outputs) + script_node_id = next( + (nid for nid, node_type in node_types.items() if node_type == "test_script" and nid in outputs), None + ) + if script_node_id: + script_execution = run_test_script_for_real( + outputs[script_node_id].get("content", ""), target_folder_path, task_id, + "validator_final_verdict_script", timeout=timeout, + ) + else: + script_execution = { + "ran": False, "exit_code": None, "stdout": "", "stderr": "", + "note": "N/A: this self-planned graph has no test_script-typed node", + } + shared_state["script_execution"] = script_execution + return inner_build_prompt(node, outputs) + ( + "\n\nThe REAL execution result of your own test script (if any):\n" + + _format_script_execution_result(script_execution) + ) + return build_prompt + + +_VALIDATOR_PLANNING_CONTEXT = ( + "{goal_text}\n\n" + "Approved SDD:\n\n{sdd_content}\n\n" + "PRD:\n\n{prd_content}\n\n" + "Guidance: you are independently verifying this goal's implementation. Form your OWN " + "acceptance criteria informed by (but not copied from) the SDD/PRD above -- don't assume " + "what the Coder actually did, you have no visibility into that beyond what you inspect " + "yourself. A held-out test script (if you plan one) will be executed for real against the " + "actual current repository state -- write real commands, not a description of what a script " + "would do. Your final verdict must be its own node (see the final_verdict entry in the " + "suggested node types), depending on whatever earlier nodes it genuinely needs." +) + + def validate_implementation_with_own_graph( - conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200, + conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200, ): - """2026-08-06: Validator's own Work Graph (validator_benchmark -> - validator_test_dataset -> validator_test_script), independently - generated from SDD/PRD -- NOT reusing the Planner's/Coder's own - benchmark/dataset/script (see _VALIDATOR_DOC_PROMPTS's module comment - for why: a held-out eval set, not a re-run of the dev set the Coder - already developed against). The resulting test_script, if real (not - N/A), is then actually EXECUTED against the current repository state -- - a non-zero real exit code force-overrides the verdict to FAIL in code, - not just by asking the LLM nicely (user-confirmed 2026-08-06: script is - the hard gate, subjective judgment can only add INCONCLUSIVE/risk notes - on top of a passing script, never override a failing one to PASS). - - Returns {"verdict", "raw_response", "artifact_ids", "script_execution"}. + """Phase 2c (2026-08-12): Validator self-plans its own verification + graph via plan_work_graph(conn, "validator", ..., append_to_run_id= + run_id) -- appended onto the SAME shared run Planner/Coder already + contributed to, replacing the old hardcoded _VALIDATOR_MANIFEST/ + _VALIDATOR_DOC_PROMPTS/_validator_prompt_builder fixed-chain approach + (kept in the module, unused, for the same "a future revert would want + it back verbatim" reason implement_code()'s retired _CODER_PROMPT is). + + The final verdict is now a REAL node inside this graph (node_type= + 'final_verdict', judgment.kind='mechanical', check_ref= + 'verdict_says_pass', on_fail.action='dynamic_loop_back') instead of a + flat call after the graph finishes -- this is what lets a FAIL + genuinely trigger Reflection-review + [Fix-Skill] + a precise loop + back to whichever real ancestor node (possibly Planner's SDD, possibly + one of Coder's own implementation nodes, anywhere in the shared run) + is actually at fault, reusing work_graph_orchestrate.py's existing + dynamic_loop_back machinery (Phase 2a) rather than reimplementing a + parallel Python-level retry loop. The real test-script hard gate + (2026-08-06, user-confirmed: script is the hard gate, subjective + judgment can only add INCONCLUSIVE/risk notes on top of a passing + script, never override a failing one to PASS) is preserved via + _verdict_node_build_prompt/_verdict_says_pass_check's shared mutable + state -- still genuinely Python-enforced, not just prompted. + + **Contract change from the old flat-call version**: because the + final_verdict node's own mechanical judgment treats anything other + than PASS as ok=False (INCONCLUSIVE included, see _verdict_says_pass_ + check's own docstring for why) and retries via dynamic_loop_back up to + its attempt budget, this function can now only ever RETURN normally + with verdict='PASS' -- a goal that can't be resolved within the loop + budget raises ConsistencyCheckFailedError instead (same as any other + exhausted work_graph_orchestrate.py loop) rather than returning a + 'FAIL'/'INCONCLUSIVE' verdict. This is what naturally retires org_ + level_loop.py's involvement for tasks using this path: scheduler.py:: + finish_task_lifecycle's "verdict != PASS -> run_fix_loop()" branch + becomes unreachable here, not because of an explicit new-flow-vs-old- + flow check, but because the condition it's gated on can no longer + occur on this path (resolves design doc §8 point 4 without adding one). + A caller that genuinely needs the old three-way PASS/FAIL/INCONCLUSIVE + contract (e.g. this function failing for a reason unrelated to the + goal itself) still gets that distinction via scheduler.py's existing + exception-based blocked_reason routing, same mechanism prd_consistency_ + check_failed/test_script_verification_failed already use. + + Returns {"verdict": "PASS", "raw_response", "artifact_ids", "script_execution"}. """ - run_id = str(uuid.uuid4()) - conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, 'validator', ?, 'running')", - (run_id, task_id, json.dumps(_VALIDATOR_MANIFEST)), + planning_context = _VALIDATOR_PLANNING_CONTEXT.format( + goal_text=goal_text, sdd_content=sdd_content, prd_content=prd_content, + ) + plan_result = work_graph_plan.plan_work_graph( + conn, "validator", task_id, planning_context, target_folder_path, + timeout=300, append_to_run_id=run_id, + ) + graph_manifest = plan_result["graph_manifest"] + new_node_ids = set(plan_result["new_node_ids"]) + node_types = {n["node_id"]: n.get("node_type") for n in graph_manifest["nodes"]} + + verdict_node_ids = [nid for nid, nt in node_types.items() if nt == "final_verdict" and nid in new_node_ids] + if len(verdict_node_ids) != 1: + raise work_graph_plan.PlanningFailedError( + "validator's self-planned graph must have exactly one node_type='final_verdict' node, " + "found {}: {}".format(len(verdict_node_ids), verdict_node_ids) + ) + verdict_node_id = verdict_node_ids[0] + + node_artifact_types = {} + for n in graph_manifest["nodes"]: + if "artifact_type" in n: + node_artifact_types[n["node_id"]] = n["artifact_type"] + elif n.get("promotes_to_artifact"): + node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"] + + planner_builder = _planner_prompt_builder(goal_text, node_artifact_types) + base_build_prompt = work_graph_plan.dispatch_build_prompt(planner_builder, work_graph_plan.generic_node_prompt_builder) + shared_state = {} + build_prompt = _verdict_node_build_prompt( + base_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout, ) - conn.commit() - node_artifact_types = {n["node_id"]: n["artifact_type"] for n in _VALIDATOR_MANIFEST["nodes"]} - # 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, + conn, run_id, "validator", "validator-independent-verification", target_folder_path, graph_manifest, + build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout, + mechanical_check_registry={ + "check_script_compiles": check_script_compiles, + "verdict_says_pass": _verdict_says_pass_check(shared_state), + }, ) 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, + conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, + _planner_ext_for, run_graph_error, only_node_ids=new_node_ids, ) if run_graph_error is not None: raise run_graph_error - if "validator_test_script" in outputs: - script_content = outputs["validator_test_script"]["content"] - script_execution = run_test_script_for_real( - script_content, target_folder_path, task_id, "validator_test_script", timeout=timeout, - ) - else: - script_execution = { - "ran": False, "exit_code": None, "stdout": "", "stderr": "", - "note": "N/A: validator_test_script node is currently disabled (see _VALIDATOR_MANIFEST)", - } + # Reaching here means the final_verdict node's own mechanical judgment + # passed -- see this function's own docstring for why that means + # verdict is unconditionally 'PASS' at this point, not something that + # needs re-parsing/re-checking. + response = outputs[verdict_node_id]["content"] + script_execution = shared_state.get("script_execution") or { + "ran": False, "exit_code": None, "stdout": "", "stderr": "", + "note": "N/A: no test_script-typed node in this self-planned graph", + } - response = org_node_trigger.trigger_org_node( - "validator", - _VALIDATOR_VERDICT_PROMPT.format( - goal_text=goal_text, - validator_benchmark_content=outputs["validator_benchmark"]["content"], - script_execution_result=_format_script_execution_result(script_execution), - ), - target_folder_path, timeout=timeout, - ) - verdict = _parse_verdict_line(response) - - # Hard gate (user-confirmed 2026-08-06): a real non-zero exit from - # validator's OWN test script is not up for LLM debate -- force FAIL - # regardless of what the verdict line said. A passing/not-run script - # leaves the LLM's own judgment (including INCONCLUSIVE) untouched. - 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. + # 2026-08-07 (real incident, preserved): this run_id is now the SHARED + # Planner+Coder+Validator run (Phase 2c), not a Validator-only one -- + # marking it 'done' here is still correct: reaching this line means the + # goal genuinely resolved to PASS, the task really is done. + # ux_work_graph_run_active's real re-verify collision this comment + # originally described no longer applies the same way once §2.2 lands + # (Phase 2c also narrows that index) -- kept here as the historical + # record of why this UPDATE exists at all, not because the exact old + # failure mode is still reachable. conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,)) conn.commit() return { - "verdict": verdict, "raw_response": response, + "verdict": "PASS", "raw_response": response, "artifact_ids": artifact_ids, "script_execution": script_execution, } @@ -1246,7 +1377,7 @@ def run_task_lifecycle(conn, task_id, target_folder_path, goal_text): org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT) verdict = validate_implementation_with_own_graph( - conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, + conn, task_id, plan_result["run_id"], sdd_content, prd_content, goal_text, target_folder_path, ) return { diff --git a/src/task_execution.py b/src/task_execution.py index b213327..0b884d8 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -64,6 +64,21 @@ def _latest_artifact_ids(conn, task_id, artifact_types): return {r[0]: r[1] for r in rows} +def _shared_run_id_for_task(conn, task_id): + """Phase 2c: recovers the shared Planner+Coder+Validator work_graph_run + id for a task resumed straight from 'coding'/'validating' (no plan_result + in scope at that point, unlike the 'reviewing_planner_output' branch + above) -- same "read it off a real artifact row's produced_by_run_id" + trick e2e_pipeline._review_planner_output_with_retries() already uses, + since every real artifact row has a non-null FK to the run that made it.""" + row = conn.execute( + "SELECT produced_by_run_id FROM artifact WHERE task_id = ? AND artifact_type = 'sdd' " + "ORDER BY created_at DESC LIMIT 1", + (task_id,), + ).fetchone() + return row[0] if row else None + + def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default): """Soft-fail artifact content lookup: returns `default` instead of raising when no such artifact exists for this task yet -- e.g. the @@ -164,8 +179,9 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): _check_not_paused(conn, task_id) org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT) + run_id = plan_result["run_id"] code_artifact_ids = e2e_pipeline.implement_code( - conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path, + conn, task_id, run_id, sdd_content, goal_text, target_folder_path, benchmark_content=benchmark_content, test_dataset_content=test_dataset_content, test_script_content=test_script_content, ) @@ -180,6 +196,12 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): sdd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "sdd", "N/A: not provided") prd_content = _artifact_content_or_default(conn, target_folder_path, task_id, "prd", "N/A: not provided") code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES) + run_id = _shared_run_id_for_task(conn, task_id) + if run_id is None: + raise TaskExecutionError( + "task {} is status={} but has no sdd artifact to recover a work_graph_run id from" + .format(task_id, status) + ) _check_not_paused(conn, task_id) if status != "validating": @@ -188,7 +210,7 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): # hit org_node_lifecycle's own PRE_STATE guard (needs "coding"). org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT) verdict = e2e_pipeline.validate_implementation_with_own_graph( - conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, + conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, ) return { diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index cf7f9d8..06d3bde 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -115,24 +115,45 @@ _FAKE_CODER_GRAPH_MANIFEST = { "edges": [], } +# Phase 2c (2026-08-12): Validator now self-plans too, into the SAME shared +# run as Planner/Coder -- depends_on "sdd" for the same "avoid landing in +# Planner's own already-'done' in-degree-zero layer" reason +# _FAKE_CODER_GRAPH_MANIFEST's own comment gives. A single node_type= +# 'final_verdict' node is the minimal real shape validate_implementation_ +# with_own_graph() accepts (see work_graph_plan._VALIDATOR_NODE_CATALOG's +# own catalog entry) -- no test_script node here, this fixture is about +# pipeline wiring, not the hard-gate mechanism (see VerdictHardGateTest/ +# ValidateImplementationWithOwnGraphTest above for that). +_FAKE_VALIDATOR_GRAPH_MANIFEST = { + "nodes": [{ + "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict", + "node_type": "final_verdict", + "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, + }], + "edges": [], +} + _real_subprocess_run = subprocess.run # captured before any test patches subprocess.run globally def _fake_plan_subprocess_run_dispatch(*args, **kwargs): """Real plan_work_graph() dispatches by the --agent argv value -- this - mock does the same, so both Planner's and Coder's planning calls (which - now share the same mocked subprocess.run patch target) each get their - own realistic fake manifest instead of one hardcoded response serving - both callers (which would hand Coder Planner's own brd/prd/... nodes, - colliding with the ones already in the run it's appending onto). + mock does the same, so Planner's/Coder's/Validator's planning calls + (which now all share the same mocked subprocess.run patch target) each + get their own realistic fake manifest instead of one hardcoded response + serving every caller (which would hand Coder/Validator Planner's own + brd/prd/... nodes, colliding with the ones already in the run they're + appending onto). patch("src.work_graph_plan.subprocess.run", ...) patches subprocess.run GLOBALLY, not just within work_graph_plan.py -- `subprocess` is the same singleton module object everywhere it's imported, so this side_effect also intercepts every OTHER real subprocess.run call this test's own call chain makes (artifact_write.write_artifact's `git add`/`git - commit`, e2e_pipeline's `git diff`/`git rev-parse`, etc). Those aren't + commit`, e2e_pipeline's `git diff`/`git rev-parse`, the final_verdict + node's own REAL test-script execution, etc). Those aren't `claude --agent ... -p ...` calls -- pass them through to the REAL subprocess.run instead of returning a canned fake "claude planning response" CompletedProcess for them too (which happened to not visibly @@ -144,7 +165,12 @@ def _fake_plan_subprocess_run_dispatch(*args, **kwargs): if not argv or argv[0] != "claude": return _real_subprocess_run(*args, **kwargs) org_node = argv[argv.index("--agent") + 1] - manifest = _FAKE_CODER_GRAPH_MANIFEST if org_node == "coder" else _FAKE_GRAPH_MANIFEST + if org_node == "coder": + manifest = _FAKE_CODER_GRAPH_MANIFEST + elif org_node == "validator": + manifest = _FAKE_VALIDATOR_GRAPH_MANIFEST + else: + manifest = _FAKE_GRAPH_MANIFEST payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"} return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") @@ -384,11 +410,10 @@ class RunEndToEndTest(PipelineTestBase): # content-keyed like every other self-planned node, not the # old single synthetic node's response-keyed output. {"content": "Implemented. Ran `pytest`: 3 passed."}, - # Validator's OWN independent benchmark/dataset (2026-08-06) -- - # deliberately different content from the Planner's, to also prove - # nothing here accidentally reuses the Planner's copies. - {"content": "# Validator's own Benchmark\nfake"}, - {"content": "N/A: validator judged no dataset needed either"}, + # Phase 2c (2026-08-12): Validator's own verdict is now a + # self-planned node too (_FAKE_VALIDATOR_GRAPH_MANIFEST), + # not a flat org_node_trigger.trigger_org_node call. + {"content": "VERDICT: PASS\nAll scenarios pass."}, ] result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏") @@ -406,35 +431,38 @@ class RunEndToEndTest(PipelineTestBase): all_artifacts = self.conn.execute( "SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],) ).fetchone()[0] - # brd, prd, sdd, benchmark, test_dataset, code_diff, test_report, - # validator_benchmark, validator_test_dataset - # (test_script/validator_test_script disabled, see _VALIDATOR_MANIFEST) - self.assertEqual(all_artifacts, 9) - - # Phase 2 §6 (2026-08-12): Coder no longer gets its own work_graph_run - # -- implement_code() appends its self-planned nodes onto Planner's - # SAME run (plan_work_graph(..., append_to_run_id=...)), so only TWO - # rows exist now, not three: the shared Planner+Coder run (still - # labeled org_node='planner' -- that column reflects who ORIGINATED - # the run, not every org_node that has since contributed nodes to it; - # Phase 2c is expected to fold Validator into the same run too, at - # which point this column stops being a reliable "who's involved" - # signal at all, see design doc §2.1) and Validator's own separate run. + # brd, prd, sdd, benchmark, test_dataset, code_diff, test_report -- + # Phase 2c: Validator's self-planned "verdict" node deliberately + # does NOT declare promotes_to_artifact (a verdict is a judgment, + # not a persisted deliverable -- see _generic_prompt_template's own + # instruction), so no validator_* artifacts exist anymore. + self.assertEqual(all_artifacts, 7) + + # Phase 2c (2026-08-12): Validator now ALSO appends its self-planned + # nodes onto the SAME shared run (plan_work_graph(..., append_to_ + # run_id=...)) instead of creating its own separate row -- only ONE + # work_graph_run row exists for this task now, still labeled + # org_node='planner' (that column reflects who ORIGINATED the run, + # not every org_node that has since contributed nodes to it -- §2.1 + # is expected to retire this column's meaning once it's no longer + # a reliable "who's involved" signal at all). run_rows = self.conn.execute( - "SELECT run_id, org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],) + "SELECT run_id, org_node, status FROM work_graph_run WHERE task_id = ?", (result["task_id"],) ).fetchall() - self.assertEqual(len(run_rows), 2) - self.assertEqual({row[1] for row in run_rows}, {"planner", "validator"}) + self.assertEqual(len(run_rows), 1) + self.assertEqual(run_rows[0][1], "planner") + self.assertEqual(run_rows[0][2], "done") - # Coder's self-planned "impl" node genuinely landed in the SAME run - # as Planner's nodes, not a separate one. - planner_run_id = next(row[0] for row in run_rows if row[1] == "planner") + # Coder's "impl" AND Validator's "verdict" nodes both genuinely + # landed in the SAME run as Planner's own nodes. + shared_run_id = run_rows[0][0] node_ids = { row[0] for row in self.conn.execute( - "SELECT node_id FROM work_node_execution WHERE run_id = ?", (planner_run_id,) + "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,) ).fetchall() } self.assertIn("impl", node_ids) + self.assertIn("verdict", node_ids) self.assertIn("sdd", node_ids) @@ -603,7 +631,7 @@ class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase): captured["benchmark_content"] = kwargs.get("benchmark_content") return {"code_diff": "diff-1", "test_report": "report-1"} - def fake_validate(conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, **kwargs): + def fake_validate(conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, **kwargs): captured["prd_content"] = prd_content return {"verdict": "PASS", "raw_response": "ok"} @@ -919,205 +947,234 @@ class RunTestScriptForRealTest(unittest.TestCase): self.assertTrue(os.path.exists(expected)) -# validator_test_script disabled by default 2026-08-07 (see -# e2e_pipeline._VALIDATOR_MANIFEST's own comment) -- tests below that need -# to exercise the hard-gate/salvage mechanisms involving that node patch -# _VALIDATOR_MANIFEST back to this 3-node shape for their own duration, so -# that real code (still in the codebase, just not planned by default) stays -# covered instead of silently rotting. -_VALIDATOR_MANIFEST_WITH_TEST_SCRIPT = { +def _fake_validator_plan_subprocess_run(manifest): + # Same real-passthrough-for-non-'claude'-argv requirement as this + # file's other subprocess.run fakes (see _fake_plan_subprocess_run_ + # dispatch's own docstring) -- the final_verdict node's build_prompt + # wrapper actually EXECUTES a real test script via a real subprocess.run + # call of its own (run_test_script_for_real/_detect_script_language), + # which would otherwise also get intercepted by this same globally- + # patched target and handed the fake planning JSON instead of a real + # exit code. + def run(*args, **kwargs): + argv = args[0] if args else kwargs.get("args") + if not argv or argv[0] != "claude": + return _real_subprocess_run(*args, **kwargs) + payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"} + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") + return run + + +# on_fail.action='hard_fail' below (not the real catalog's suggested +# 'dynamic_loop_back') is a deliberate test-fixture choice, not something a +# real self-planned graph would use: it isolates the hard-gate mechanical +# check itself (this file's job) from the separate dynamic_loop_back/ +# Reflection-review retry machinery, which already has its own dedicated +# coverage in test_work_graph_orchestrate.py (DynamicLoopBackContentErrorTest/ +# DynamicLoopBackSkillDefectTest/TallyReflectionTest) -- mocking a 3-perspective +# real reflection jury here would be expensive and wouldn't test anything new. +_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY = { + "nodes": [{ + "node_id": "verdict", "depends_on": [], "task_prompt": "give your final verdict", + "node_type": "final_verdict", + "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "on_fail": {"action": "hard_fail"}, + }], + "edges": [], +} + +_FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT = { "nodes": [ - {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []}, { - "node_id": "validator_test_dataset", "artifact_type": "validator_test_dataset", - "depends_on": ["validator_benchmark"], + "node_id": "test_script", "depends_on": [], "task_prompt": "write a real test script", + "node_type": "test_script", + "judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"}, }, { - "node_id": "validator_test_script", "artifact_type": "validator_test_script", - "depends_on": ["validator_test_dataset", "validator_benchmark"], + "node_id": "verdict", "depends_on": ["test_script"], "task_prompt": "give your final verdict", + "node_type": "final_verdict", + "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "on_fail": {"action": "hard_fail"}, }, ], "edges": [], } +class VerdictHardGateTest(unittest.TestCase): + """Phase 2c (2026-08-12): unit coverage of the two functions carrying + the safety-critical hard gate (2026-08-06, user-confirmed: a real + non-zero exit from Validator's own test script forces FAIL regardless + of what the LLM's verdict line says) forward into the self-planned + final_verdict node design -- direct/isolated rather than through the + full self-planning + run_graph() + work_node_run machinery, which the + integration tests below cover for the wiring, not this specific logic.""" + + def test_real_script_failure_overrides_pass_verdict_text(self): + shared_state = {"script_execution": {"ran": True, "exit_code": 1, "stdout": "", "stderr": "boom"}} + ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good") + self.assertFalse(ok) + self.assertIn("exit_code=1", detail) + + def test_passing_script_and_pass_verdict_is_ok(self): + shared_state = {"script_execution": {"ran": True, "exit_code": 0, "stdout": "", "stderr": ""}} + ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good") + self.assertTrue(ok) + self.assertIsNone(detail) + + def test_no_script_and_pass_verdict_is_ok(self): + ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: PASS\nlooks good") + self.assertTrue(ok) + + def test_fail_verdict_is_not_ok_even_with_no_script(self): + ok, detail = e2e_pipeline._verdict_says_pass_check({})("VERDICT: FAIL\nreal bug found") + self.assertFalse(ok) + self.assertIn("FAIL", detail) + + def test_inconclusive_verdict_is_not_ok(self): + # 2026-08-12 (Phase 2c, design decision): INCONCLUSIVE is treated + # the same as FAIL here, not specially accepted -- see + # validate_implementation_with_own_graph's own docstring for why. + ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: INCONCLUSIVE\nneeds human review") + self.assertFalse(ok) + + def test_build_prompt_passes_through_unchanged_for_non_verdict_nodes(self): + shared_state = {} + inner = lambda node, outputs: "inner prompt for " + node["node_id"] + wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60) + result = wrapped({"node_id": "impl"}, {}) + self.assertEqual(result, "inner prompt for impl") + self.assertEqual(shared_state, {}) + + def test_build_prompt_runs_real_script_for_verdict_node_with_script_dependency(self): + shared_state = {} + inner = lambda node, outputs: "base prompt" + node_types = {"test_script": "test_script"} + outputs = {"test_script": {"content": "import sys\nsys.exit(0)\n"}} + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True) + wrapped = e2e_pipeline._verdict_node_build_prompt(inner, node_types, "task-1", tmp, shared_state, 60) + result = wrapped({"node_id": "verdict", "node_type": "final_verdict"}, outputs) + self.assertTrue(shared_state["script_execution"]["ran"]) + self.assertEqual(shared_state["script_execution"]["exit_code"], 0) + self.assertIn("base prompt", result) + self.assertIn("REAL execution result", result) + + def test_build_prompt_na_stub_for_verdict_node_with_no_script_dependency(self): + shared_state = {} + inner = lambda node, outputs: "base prompt" + wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60) + wrapped({"node_id": "verdict", "node_type": "final_verdict"}, {}) + self.assertFalse(shared_state["script_execution"]["ran"]) + + class ValidateImplementationWithOwnGraphTest(unittest.TestCase): - """2026-08-06: the safety-critical piece -- a real non-zero exit from - Validator's own script must force FAIL no matter what the LLM's verdict - line says (user-confirmed hard gate).""" + """Phase 2c (2026-08-12): Validator now self-plans its own verification + graph, appended onto the SAME shared run Planner/Coder already + contributed to (plan_work_graph(..., append_to_run_id=run_id)) -- see + e2e_pipeline.validate_implementation_with_own_graph's own docstring. + VerdictHardGateTest above covers the hard-gate logic itself in + isolation; these tests cover the wiring: planning dispatch, mechanical + check registration, run_id reuse, and result unwrapping.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True) - self.conn = schema.connect(":memory:") + self.conn = schema.connect(":memory:", check_same_thread=False) schema.create_schema(self.conn) project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False) 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', 'validating', 'validator')" ) + # The shared Planner+Coder run Validator appends onto -- empty on + # purpose (these tests aren't about cross-stage node dependencies, + # see test_task_execution.py for that), just needs to genuinely + # exist for append_to_run_id's own existence check. + self.conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES ('run-shared', 'task-1', 'planner', '{\"nodes\": [], \"edges\": []}', 'running')" + ) self.conn.commit() def tearDown(self): self.conn.close() - def _run(self, script_content, verdict_line, with_test_script=False): - mock_responses = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - ] - if with_test_script: + def _run(self, manifest, verdict_content, script_content=None): + mock_responses = [] + if script_content is not None: mock_responses.append({"content": script_content}) - with contextlib.ExitStack() as stack: - stack.enter_context(patch.object(work_graph_execute, "write_workflow_script")) - mock_execute = stack.enter_context(patch.object(work_graph_execute, "execute_work_graph")) - stack.enter_context(patch.object(org_node_trigger, "trigger_org_node", return_value=verdict_line)) - if with_test_script: - stack.enter_context( - patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT) - ) + mock_responses.append({"content": verdict_content}) + with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_validator_plan_subprocess_run(manifest)), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: mock_execute.side_effect = mock_responses return e2e_pipeline.validate_implementation_with_own_graph( - self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name, + self.conn, "task-1", "run-shared", "sdd content", "prd content", "goal text", self.tmp.name, ) def test_llm_says_pass_and_script_really_passes_stays_pass(self): - result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True) + result = self._run( + _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good", + script_content="import sys\nsys.exit(0)\n", + ) self.assertEqual(result["verdict"], "PASS") self.assertTrue(result["script_execution"]["ran"]) self.assertEqual(result["script_execution"]["exit_code"], 0) - def test_llm_says_pass_but_script_really_fails_is_force_overridden_to_fail(self): + def test_llm_says_pass_but_script_really_fails_raises(self): # the exact case the user's hard-gate rule exists for: a script # that objectively fails must not be argued away by the LLM's own - # verdict line. - result = self._run( - "import sys\nsys.exit(1)\n", "VERDICT: PASS\nlooks good to me", with_test_script=True - ) - self.assertEqual(result["verdict"], "FAIL") - - def test_llm_says_fail_and_script_fails_stays_fail(self): - result = self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found") - self.assertEqual(result["verdict"], "FAIL") + # verdict line -- hard_fail (see the fixture's own comment) makes + # this deterministic without invoking the loop-back/reflection jury. + with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError): + self._run( + _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good to me", + script_content="import sys\nsys.exit(1)\n", + ) - def test_na_script_leaves_llm_judgment_untouched(self): - result = self._run("N/A: no mechanical check possible", "VERDICT: INCONCLUSIVE\nneeds human review") - self.assertEqual(result["verdict"], "INCONCLUSIVE") + def test_no_script_node_leaves_llm_judgment_as_the_only_signal(self): + result = self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good") + self.assertEqual(result["verdict"], "PASS") self.assertFalse(result["script_execution"]["ran"]) - def test_writes_two_validator_artifacts_with_lineage(self): - # test_script disabled by default -- see _VALIDATOR_MANIFEST_WITH_ - # TEST_SCRIPT variant below for the 3-artifact/script-lineage case. - result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good") - self.assertEqual(set(result["artifact_ids"]), {"validator_benchmark", "validator_test_dataset"}) - dataset_deps = json.loads( - self.conn.execute( - "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", - (result["artifact_ids"]["validator_test_dataset"],), - ).fetchone()[0] - ) - self.assertIn(result["artifact_ids"]["validator_benchmark"], dataset_deps) - - def test_writes_three_validator_artifacts_with_lineage_when_test_script_re_enabled(self): - result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True) - self.assertEqual( - set(result["artifact_ids"]), - {"validator_benchmark", "validator_test_dataset", "validator_test_script"}, - ) - script_deps = json.loads( - self.conn.execute( - "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", - (result["artifact_ids"]["validator_test_script"],), - ).fetchone()[0] - ) - self.assertIn(result["artifact_ids"]["validator_benchmark"], script_deps) - self.assertIn(result["artifact_ids"]["validator_test_dataset"], script_deps) - - 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, 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, - ) + def test_appends_into_the_shared_run_without_creating_a_new_row(self): + self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good") + rows = self.conn.execute("SELECT run_id, status FROM work_graph_run WHERE task_id = 'task-1'").fetchall() + self.assertEqual(rows, [("run-shared", "done")]) + node_ids = { + r[0] for r in self.conn.execute( + "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'" + ).fetchall() + } + self.assertIn("verdict", node_ids) + + def test_second_call_appends_fresh_nodes_without_colliding(self): + # A real re-verify plans its OWN new node_ids each time (a fresh + # LLM planning call, not a replay) -- node_id collision is what + # plan_work_graph's append mode actively guards against (raises + # PlanningFailedError), so this uses a different node_id for the + # second call, same as a real second planning call naturally would. + second_manifest = { + "nodes": [{ + "node_id": "verdict_v2", "depends_on": [], "task_prompt": "give your final verdict", + "node_type": "final_verdict", + "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "on_fail": {"action": "hard_fail"}, + }], + "edges": [], + } + self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good") + second = self._run(second_manifest, "VERDICT: PASS\nfixed now") 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. - # validator_test_script itself is disabled by default (see - # _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT above) -- re-enabled here to - # keep the salvage mechanism (generic, not test_script-specific) covered. - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ - patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT): - 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") + node_ids = { + r[0] for r in self.conn.execute( + "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'" + ).fetchall() + } + self.assertEqual(node_ids, {"verdict", "verdict_v2"}) if __name__ == "__main__": diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 12c1b3a..2fd6f54 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -16,7 +16,7 @@ from unittest.mock import patch from src import ( e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, - task_execution, work_graph_execute, work_graph_plan, + task_execution, work_graph_execute, work_graph_orchestrate, work_graph_plan, ) @@ -41,9 +41,45 @@ _FAKE_CODER_MANIFEST = { "edges": [], } +# Phase 2c (2026-08-12): Validator now self-plans too (validate_ +# implementation_with_own_graph() calls plan_work_graph(conn, "validator", +# ..., append_to_run_id=...), through the SAME mocked subprocess boundary +# as Planner's/Coder's own planning calls) -- a single node_type= +# 'final_verdict' node is the minimal real shape this mechanism accepts +# (see work_graph_plan._VALIDATOR_NODE_CATALOG's own catalog entry). +# depends_on ["sdd"] for the same "avoid landing in the same in-degree-zero +# topological layer as an already-'done' node" reason _FAKE_CODER_MANIFEST +# above documents. +_FAKE_VALIDATOR_MANIFEST = { + "nodes": [{ + "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict", + "node_type": "final_verdict", + "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, + }], + "edges": [], +} -def _fake_coder_plan_subprocess_run(*args, **kwargs): - payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_CODER_MANIFEST) + "\n```"} +_real_subprocess_run = subprocess.run # captured before any test patches subprocess.run globally + + +def _fake_plan_subprocess_run_dispatch(*args, **kwargs): + """Real plan_work_graph() dispatches by the --agent argv value -- this + mock does the same, so Planner's/Coder's/Validator's planning calls + (which all now share the same mocked subprocess.run patch target) each + get their own realistic fake manifest instead of one hardcoded response + serving all three callers. Same dispatch-by-argv + real-passthrough-for- + non-'claude'-calls pattern as test_e2e_pipeline.py's helper of the same + name -- see that module's docstring for the full rationale (this file's + own git add/commit/diff calls inside the `with patch(...)` blocks below + would otherwise also get intercepted, since `subprocess` is one shared + module object).""" + argv = args[0] + if not argv or argv[0] != "claude": + return _real_subprocess_run(*args, **kwargs) + org_node = argv[argv.index("--agent") + 1] + manifest = _FAKE_VALIDATOR_MANIFEST if org_node == "validator" else _FAKE_CODER_MANIFEST + payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"} return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") @@ -95,6 +131,24 @@ class TaskExecutionTestBase(unittest.TestCase): "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest), ) + # Phase 2c (2026-08-12, found while adding Validator self-planning + # tests): a real Planner run_graph() call ALSO leaves a real + # work_node_execution row behind for this node -- without one here + # too, a later self-planning append onto this SAME run_id (Coder's/ + # Validator's own plan_work_graph(..., append_to_run_id=run_id)) + # finds no existing (run_id, node_id) row for `artifact_type` and + # spuriously RE-EXECUTES it for real via the mocked execute_work_ + # graph, burning an extra mock_execute.side_effect slot the calling + # test never accounted for (masked before this fix only because + # those tests happened to provide enough spare list items and never + # asserted on the exact content that ended up threaded through). + work_graph_orchestrate.ensure_placeholder_skill(self.conn, "planner-doc-writing", "planner") + self.conn.execute( + "INSERT OR IGNORE 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', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')", + ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})), + ) rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type) abs_path = os.path.join(self.tmp.name, rel_path) os.makedirs(os.path.dirname(abs_path), exist_ok=True) @@ -130,22 +184,17 @@ class ResumeFromCodingTest(TaskExecutionTestBase): self._write_artifact("task-1", "test_report", "report content") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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") as mock_trigger: - # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06) - mock_execute.side_effect = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, - ] + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + # validator's self-planned final_verdict node (Phase 2c) + mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") self.assertEqual(result["verdict"]["verdict"], "PASS") self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff") - # only the validator's final-verdict real call happened -- no coder call - mock_trigger.assert_called_once() - self.assertEqual(mock_trigger.call_args[0][0], "validator") + # only the validator's self-planned node ran -- no coder call + mock_execute.assert_called_once() row = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone() self.assertEqual(row[0], "validating") @@ -168,17 +217,14 @@ class ResumeFromPlanningTest(TaskExecutionTestBase): self._write_artifact("task-1", "benchmark", "# Benchmark content") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ - patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ {"content": "done, tests pass"}, # coder's self-planned "impl" node - # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06) - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, + {"content": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") @@ -187,11 +233,11 @@ class ResumeFromPlanningTest(TaskExecutionTestBase): # existing artifacts were reused -- plan_documents() must NOT have been # called again (that's the whole point of the existence-probing retry loop). mock_plan.assert_not_called() - # two real org_node_trigger.trigger_org_node calls: planner_reviewer's - # judgment, then validator's final verdict -- not just validator anymore. - self.assertEqual(mock_trigger.call_count, 2) + # one real org_node_trigger.trigger_org_node call: planner_reviewer's + # judgment -- Validator's own verdict is now a self-planned node (Phase + # 2c), read via mock_execute above, not a flat org_node_trigger call. + mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer") - self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator") class ResumeFromReviewingTest(TaskExecutionTestBase): @@ -216,16 +262,14 @@ class ResumeFromReviewingTest(TaskExecutionTestBase): return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout) with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=tracking_ready_trigger), \ - patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ {"content": "done, tests pass"}, # coder's self-planned "impl" node - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, + {"content": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") @@ -236,9 +280,11 @@ class ResumeFromReviewingTest(TaskExecutionTestBase): # this task into 'reviewing_planner_output' in the first place). self.assertEqual(ready_calls, ["coder", "validator"]) # but the real review judgment call DOES still run (that's the actual - # gap this status resumes into -- see the module docstring). + # gap this status resumes into -- see the module docstring). Validator's + # own verdict is now a self-planned node (Phase 2c), read via + # mock_execute above, not a second org_node_trigger call. + mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args_list[0][0][0], "planner_reviewer") - self.assertEqual(mock_trigger.call_args_list[1][0][0], "validator") class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): @@ -266,16 +312,14 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): return {"run_id": "run-sdd", "artifact_ids": {"sdd": "art-sdd"}} with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ - patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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"), \ patch.object(e2e_pipeline, "plan_documents", side_effect=fake_plan_documents) as mock_plan: mock_execute.side_effect = [ {"content": "done, tests pass"}, # coder's self-planned "impl" node - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, + {"content": "VERDICT: PASS\nlooks good"}, # validator's self-planned final_verdict node (Phase 2c) ] # must not raise (this is the actual regression: it used to KeyError/ # TaskExecutionError before ever reaching here) @@ -304,14 +348,10 @@ class ResumeFromValidatingTest(TaskExecutionTestBase): self._write_artifact("task-1", "test_report", "report content") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger) as mock_ready, \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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") as mock_trigger: - mock_execute.side_effect = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, - ] + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") self.assertEqual(result["verdict"]["verdict"], "PASS") @@ -321,14 +361,19 @@ class ResumeFromValidatingTest(TaskExecutionTestBase): # get the task into 'validating' in the first place, and calling it # again would hit org_node_lifecycle's own "needs status=coding" guard. mock_ready.assert_not_called() - # only the real work_graph_run's own final call happened - mock_trigger.assert_called_once() - self.assertEqual(mock_trigger.call_args[0][0], "validator") - # a brand-new validator work_graph_run was created (none existed before) - rows = self.conn.execute( - "SELECT run_id FROM work_graph_run WHERE task_id = 'task-1' AND org_node = 'validator'" - ).fetchall() - self.assertEqual(len(rows), 1) + # only the validator's self-planned node ran + mock_execute.assert_called_once() + # Phase 2c: Validator appends its final_verdict node into the SAME + # shared run the sdd artifact points at (via _shared_run_id_for_task), + # not a brand-new work_graph_run row of its own -- confirm the merge + # landed on "run-sdd" and no extra row was created for this task. + row = self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-sdd'").fetchone() + node_ids = {n["node_id"] for n in json.loads(row[0])["nodes"]} + self.assertIn("verdict", node_ids) + total_rows = self.conn.execute( + "SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'" + ).fetchone()[0] + self.assertEqual(total_rows, 5) class ResumeUnresumableStatusTest(TaskExecutionTestBase): @@ -473,9 +518,24 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): def _write_artifact(self, task_id, artifact_type, content): conn = schema.connect(self.db_path) run_id = "run-" + artifact_type + # Real "nodes"/"edges" shape + a matching 'done' work_node_execution + # row -- see TaskExecutionTestBase._write_artifact's own comment + # (same file) for why both are needed once Validator self-plans and + # appends onto this run via plan_work_graph(..., append_to_run_id=...). + manifest = json.dumps({ + "nodes": [{"node_id": artifact_type, "artifact_type": artifact_type, "depends_on": []}], + "edges": [], + }) conn.execute( "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id), + "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest), + ) + work_graph_orchestrate.ensure_placeholder_skill(conn, "planner-doc-writing", "planner") + conn.execute( + "INSERT OR IGNORE 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', 'planner-doc-writing', 'in.json', 'out.json', '{}', ?, 'done')", + ("node-exec-" + run_id, run_id, artifact_type, json.dumps({"content": content})), ) rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type) abs_path = os.path.join(self.repo.name, rel_path) @@ -506,14 +566,10 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): self._write_artifact("task-1", "test_report", "report") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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"): - mock_execute.side_effect = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, - ] + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", ) @@ -561,14 +617,10 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout) with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=blocking_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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"): - mock_execute.side_effect = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, - ] + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", ) @@ -591,17 +643,14 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): self._write_artifact("task-1", "test_report", "report") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ 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", side_effect=RuntimeError("boom")): - # validator's own benchmark/dataset/script graph completes fine -- - # the failure this test cares about is the FINAL verdict call - # (org_node_trigger.trigger_org_node), which happens after. - mock_execute.side_effect = [ - {"content": "# Validator's own Benchmark"}, - {"content": "N/A: no dataset needed"}, - {"content": "import sys\nsys.exit(0)\n"}, - ] + patch.object(work_graph_execute, "execute_work_graph", side_effect=RuntimeError("boom")): + # Phase 2c: Validator's own verdict is now a self-planned node, + # not a flat org_node_trigger.trigger_org_node call -- the + # failure this test cares about (a real exception surfacing as + # blocked_pending_human/execution_error) is injected at its own + # node's real execution instead. thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", ) diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index c4f78bb..fd9d0a8 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -81,6 +81,12 @@ GRAPH_MANIFEST_SCHEMA = { }, "subagent_type": {"type": "string"}, "skill_ref": {"type": "string"}, + # 2026-08-12 (Phase 2c): echoed-back catalog label, e.g. + # "test_script"/"final_verdict" -- see _generic_prompt_ + # template's own comment for why this exists (finding + # role-specific nodes among arbitrarily-named + # self-planned ones). + "node_type": {"type": "string"}, }, }, }, @@ -232,9 +238,22 @@ _VALIDATOR_NODE_CATALOG = [ {"node_type": "test_dataset", "description": "Concrete test data, if this goal has a natural one."}, { "node_type": "test_script", - "description": "A real, runnable script mechanically checking the independent benchmark's scenarios.", + "description": "A real, runnable script mechanically checking the independent benchmark's scenarios. " + "It will actually be EXECUTED against the real repository after you plan it -- write " + "real commands, not a description of what a script would do.", "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"}, }, + { + "node_type": "final_verdict", + "description": "Exactly ONE node of this type, depending on every other node whose content your " + "verdict needs. Its content must be your final judgment: a line 'VERDICT: PASS', " + "'VERDICT: FAIL', or 'VERDICT: INCONCLUSIVE', followed by your reasoning (and, if not " + "PASS, a root_cause_category of planning|coder|skill_defect with detail). This is the " + "ONLY node in a Validator graph allowed to use on_fail.action='dynamic_loop_back' -- " + "see the on_fail instructions below for why.", + "suggested_judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"}, + "suggested_on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, + }, ] _GENERIC_NODE_CATALOGS = {"coder": _CODER_NODE_CATALOG, "validator": _VALIDATOR_NODE_CATALOG} @@ -245,15 +264,37 @@ def _generic_prompt_template(org_node): org_node so each gets its own suggested catalog (or none, for any future org_node this hasn't been extended for yet -- catalog_section is simply empty then, same as before this change existed). - node_type is a label in the SUGGESTION data only, not a required - graph_manifest node field -- node_id/task_prompt stay exactly as - freeform as before.""" + + 2026-08-12 (Phase 2c): `node_type` is now asked to be ECHOED BACK onto + each planned node (not just a label in the suggestion data) -- e2e_ + pipeline.py's Validator rework needs to find "the test_script node"/ + "the final_verdict node" among self-planned, arbitrarily-named nodes, + and node_type is the one signal that survives self-planning for this + (node_id is freeform, task_prompt is prose). `dynamic_loop_back` is + now allowed, but ONLY for org_node='validator' AND only on a node + that also declares node_type='final_verdict' -- this reverses the + Phase 2a/2b instruction ("reserved for the terminal review stage + outside this graph, not for nodes you plan yourself"), because + Validator's terminal review IS now planned inside this graph, not a + flat call after it (see _VALIDATOR_NODE_CATALOG's own 'final_verdict' + entry for why: only a real graph node can use the loop_back mechanism + _run_one_node already implements). Every OTHER org_node still can't + use it -- Coder's catalog has no comparable terminal-judgment role.""" catalog = _GENERIC_NODE_CATALOGS.get(org_node, []) catalog_section = ( "Suggested node types for this goal (use any subset, in any order, invent your own if " "none fit -- these are starting points, not requirements):\n" + json.dumps(catalog) + "\n" if catalog else "" ) + dynamic_loop_back_rule = ( + "{\"action\": \"dynamic_loop_back\"} ONLY for the node also declaring " + "node_type='final_verdict' (see the final_verdict catalog entry above) -- no on_failure " + "edge needed for this one, the target is chosen dynamically at failure time, not declared " + "now.\n" + if org_node == "validator" else + "do NOT use \"dynamic_loop_back\" -- that's reserved for Validator's terminal review node, " + "not for nodes you plan yourself.\n" + ) return ( "Plan (do not execute) a Work Graph for this goal: {goal_text}\n" + catalog_section + @@ -267,6 +308,9 @@ def _generic_prompt_template(org_node): "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt " "for that node's own execution -- assume the agent running it sees nothing except this " "task_prompt and its dependencies' outputs, not this planning conversation), plus:\n" + " - node_type (optional but STRONGLY suggested when a node matches one of the suggested " + "types above): echo back the matching node_type string (e.g. \"test_script\") so it's " + "identifiable later -- omit only for a node that doesn't match any suggested type.\n" " - judgment (optional): how to tell if this node's output is acceptable -- " "{\"kind\": \"none\"} if nothing needs checking (the default if omitted), {\"kind\": " "\"llm\", \"judge_prompt\": \"...{content}...\"} for your own judgment prompt with a " @@ -274,13 +318,14 @@ def _generic_prompt_template(org_node): " - on_fail (only meaningful if judgment.kind != \"none\"): {\"action\": \"retry_self\"}, " "{\"action\": \"hard_fail\"}, or {\"action\": \"loop_back\"} with a matching on_failure " "edge to a genuine upstream dependency (same rule _PLANNER_NODE_CATALOG's own nodes " - "follow) -- do NOT use \"dynamic_loop_back\", that's reserved for the terminal review " - "stage outside this graph, not for nodes you plan yourself.\n" + "follow); " + dynamic_loop_back_rule + " - promotes_to_artifact (optional): {\"artifact_type\": \"test_report\"} ONLY for a " "node producing this task's actual final deliverable text -- most nodes should omit " "this; it marks a node's content for persistence beyond this graph's own execution. " "(code_diff, if relevant, is derived separately from real file changes, not from any " - "node's own content -- never declare artifact_type: \"code_diff\" here.)\n" + "node's own content -- never declare artifact_type: \"code_diff\" here. The final_verdict " + "node, if any, should also NOT declare promotes_to_artifact -- a verdict is a judgment, " + "not a persisted deliverable.)\n" "Every edge needs edge_type: 'zero_token' for pure aggregation/reduce steps, " "'needs_judgment' for anything requiring the agent to actually reason about content.\n" "End your reply with the graph_manifest as a single fenced ```json code block " -- libgit2 0.21.4