"""U-014: end-to-end orchestration for architecture doc §11 MVP acceptance criteria 1-3 (Planner produces validated BRD/PRD/SDD/Benchmark; Coder implements against the SDD and self-tests pass; Validator runs benchmark/tests and produces a root-cause-attributed verdict). Criteria 4-7 (Org-level Loop, breakpoint recovery, ...) are explicitly out of this Unit's scope -- PH2 (U-017/U-021). **Rewired 2026-08-01 (U-031 regression audit)**: `plan_documents` used to call `org_node_trigger.trigger_org_node()` directly, bypassing `work_graph_plan.plan_work_graph()`/`work_graph_execute.execute_work_graph()` /`work_node_run.run_work_node()` entirely -- those three modules were each fully unit-tested (and even real-API-validated) in isolation since PH1/PH2, but nothing in production ever called any of them; this was the one real, validated closed loop, and it used a completely separate, simpler mechanism. REQ-02's own coverage line names `c_work_graph_plan`/`c_work_graph_execute` explicitly, so that gap is now closed: `plan_documents` calls `plan_work_graph()` for a real graph_manifest, then `work_graph_orchestrate.run_graph()` executes each node for real (see that module's docstring for why each node gets its own one-node dynamic-workflow script rather than one big multi-node script -- REQ-06 breakpoint recovery needs it). `implement_code` (Coder) goes through the same per-node mechanism with a synthetic single-node manifest, since REQ-03's own coverage only names `c_work_node_run`, not graph planning -- there is no real "plan a Coder module graph" LLM capability built (a much bigger, separate undertaking; planning docs and planning code modules are different problems), so a one-node graph is the honest minimum that makes `c_work_node_run` genuinely load-bearing for Coder too. `validate_implementation` (Validator) is unchanged -- REQ-04's own coverage never named the Work Graph controllers at all. Status-transition bookkeeping (task.status, precondition guards) stays U-010's job (org_node_lifecycle.trigger_org_node_full) -- called once per phase with a lightweight "confirm ready" prompt. **PRD consistency check (2026-08-02, PH4-closure audit)**: REQ-02's boundary case (contradictory PRD scenario assumptions must be intercepted, not silently pass validation) had no implementation anywhere -- found while auditing REQ-01~14 for real coverage before starting PH5. `_check_prd_consistency` is one extra real `agent()` call (reusing `agentType='planner'` + a structured-output schema, same mechanism as every other node) that reviews the PRD's own content for genuine internal contradiction (not incompleteness/style) right after it's produced, wired in as `plan_documents`'s `verify_fn` -- see `work_graph_orchestrate. run_graph`'s docstring for what happens on a failed check (one regeneration attempt, then `ConsistencyCheckFailedError` if it's still contradictory). """ import json import os import subprocess import tempfile import uuid from src import ( artifact_write, org_node_lifecycle, org_node_trigger, paths, task_create, work_graph_orchestrate, work_graph_plan, work_node_run, ) _READY_PROMPT = "Confirm you're ready to begin. Take no other action yet." _DOC_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} _DOC_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]} _PRD_CONSISTENCY_SCHEMA = { "type": "object", "required": ["consistent", "issues"], "properties": { "consistent": {"type": "boolean"}, "issues": {"type": "array", "items": {"type": "string"}}, }, } _PRD_CONSISTENCY_PROMPT = ( "Review the following PRD for internally CONTRADICTORY scenario assumptions -- e.g. two " "requirements that cannot both hold at once, or the same scenario described two incompatible " "ways. Do NOT flag mere incompleteness, ambiguity, or style issues, only genuine " "self-contradiction. Reply via the required schema: consistent=false only if you found a real " "contradiction, and list each one in issues.\n\nPRD:\n\n{prd_content}" ) _NO_NARRATION = ( "The `content` field must contain ONLY the raw document/code text itself -- nothing " "else. Do not include any sentence describing what you did, what workflow ran, how " "the process went, or that you completed a task. A machine parses `content` directly " "into a file with no human review; any narrative text becomes permanently embedded in " "a document that's supposed to look like a real, standalone deliverable. Start " "directly with the content itself (e.g. a Markdown document starts with its own `#` " "heading, not a sentence about producing it)." ) _DOC_PROMPTS = { "brd": ( "Write a BRD (Business Requirement Document) for this goal, following your own " "SOUL.md methodology: {goal_text}\n" "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION ), "prd": ( "Here is the approved BRD for goal '{goal_text}':\n\n{brd_content}\n\n" "Write the PRD (Product Requirement Document) based on this BRD. " "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION ), "sdd": ( "Here is the approved PRD for goal '{goal_text}':\n\n{prd_content}\n\n" "Write the SDD (System Design Document) based on this PRD, including a Development " "Unit breakdown. Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION ), "benchmark": ( "Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n" "Write the Benchmark/acceptance-test document based on this SDD: concrete test " "scenarios (happy/boundary/failure) with observable pass criteria for each one -- not " "just a one-line label per scenario, spell out the actual expected input/behavior/" "output for each. Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION ), "test_dataset": ( "Here is the approved Benchmark for goal '{goal_text}':\n\n{benchmark_content}\n\n" "Judge whether this goal has a natural, concrete test dataset (specific inputs with " "known-correct outputs, fixture data, or QA pairs the Benchmark's scenarios can be " "instantiated against) -- many goals (UI wiring, config/plumbing changes) don't have " "one, and that's fine. If it does not apply, reply via the required schema with " "content starting exactly with 'N/A: ' followed by a one-sentence reason. If it does " "apply, reply with the real dataset itself (e.g. JSON/CSV/a table of concrete cases) " "via the required schema. " + _NO_NARRATION ), "test_script": ( "Here is the approved Benchmark for goal '{goal_text}':\n\n{benchmark_content}\n\n" "Here is the test dataset (may be 'N/A' if this goal has none):\n\n{test_dataset_content}\n\n" "Write a real, runnable test script that mechanically checks the Benchmark's " "scenarios (using the dataset above if it applies) -- a script a machine can execute " "and get a real pass/fail exit code from, not prose describing what should be tested. " "Use whatever language/framework fits this specific project's own conventions (don't " "invent a new one). If genuinely no mechanical script can meaningfully check this goal " "(e.g. a pure judgment call with no automatable check), reply via the required schema " "with content starting exactly with 'N/A: ' followed by a one-sentence reason -- this " "should be rare, prefer writing a real script whenever a real check is possible. " "Reply with the real script content via the required schema. " + _NO_NARRATION ), } _CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} _CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]} # 2026-08-12 (Phase 2 §6): retired as the per-node prompt now that Coder # self-plans (implement_code() no longer runs one synthetic "implement" # node) -- its guidance is folded into _CODER_PLANNING_CONTEXT below, # which feeds plan_work_graph()'s planning call instead of a single # execution call. Kept here, unused, only because deleting it isn't this # change's job and a future revert would want it back verbatim. _CODER_PROMPT = ( "Implement this goal per the approved SDD below. Write real code and real tests, run the " "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts " "and can explain why not).\n\n" "The Planner also produced a Benchmark, a test dataset, and a test script for this goal " "(test dataset/script may be 'N/A' if this goal has none -- that's expected, not an error). " "If the test script is real (not N/A): materialize it as a real file under this project's " "own test conventions (match how existing tests in this repo are named/located), and you " "MUST make it pass for real before you're done -- it is not optional extra credit, it's " "part of this goal's acceptance bar. Write your own additional tests too if the Planner's " "script doesn't cover everything the SDD requires; the Planner's script is a floor, not a " "ceiling.\n\n" "Reply via the required schema with a `response` field describing what you did and the " "real test commands/output (including the Planner's own test script's real output). " "Goal: {goal_text}\n\nSDD:\n\n{sdd_content}\n\n" "Benchmark:\n\n{benchmark_content}\n\n" "Test dataset:\n\n{test_dataset_content}\n\n" "Test script:\n\n{test_script_content}" ) # 2026-08-12 (Phase 2 §6): fed as `goal_text` into plan_work_graph(conn, # "coder", ..., append_to_run_id=run_id) -- reuses the existing # {goal_text} substitution point in _generic_prompt_template() rather than # adding a new parameter to plan_work_graph() just for Coder's extra # context (SDD/Benchmark/test dataset/script), which Planner's own # _PROMPT_TEMPLATE call never needed. Same .replace("{goal_text}", ...) # convention _PROMPT_TEMPLATE/_generic_prompt_template() already use (not # .format() -- see work_graph_plan.py's own comment on why), so this # string's remaining braces (none currently, but keep this in mind if # edited) must stay literal, not format fields. _CODER_PLANNING_CONTEXT = ( "{goal_text}\n\n" "Approved SDD:\n\n{sdd_content}\n\n" "Benchmark (test dataset/script may be 'N/A' if this goal has none -- that's expected, not " "an error):\n\n{benchmark_content}\n\n" "Test dataset:\n\n{test_dataset_content}\n\n" "Test script:\n\n{test_script_content}\n\n" "Guidance: write real code and real tests, run the tests yourself as part of your own " "nodes' work, and don't consider the goal done until they pass for real (or you've " "exhausted reasonable attempts and a node explains why not). If the Planner's test script " "above is real (not N/A), materialize it as a real file under this project's own test " "conventions and make it pass for real -- it is not optional, it's part of this goal's " "acceptance bar; the Planner's script is a floor, write additional tests of your own if it " "doesn't cover everything the SDD requires." ) _VALIDATOR_PROMPT = ( "Independently verify the implementation for this goal against the Benchmark below: run " "the tests yourself, don't just trust the Coder's claims. Reply with a verdict line " "'VERDICT: PASS' or 'VERDICT: FAIL' or 'VERDICT: INCONCLUSIVE' followed by your reasoning " "and, if FAIL, a root_cause_category (planning|coder|skill_defect) with detail. " "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. # hard-code outputs for the specific dataset it developed against without # solving the general problem), re-running that SAME script proves nothing # -- this is the "held-out eval set vs. the dev/train set" distinction, not # a redundant re-check of the same artifact. _VALIDATOR_DOC_PROMPTS = { "validator_benchmark": ( "You are independently defining YOUR OWN acceptance criteria for this goal, informed " "by the SDD/PRD below -- do NOT copy or assume the Planner's own Benchmark, form your " "own independent judgment of what 'done' means, from a different angle if you can. " "Concrete test scenarios (happy/boundary/failure) with observable pass criteria for " "each one -- spell out actual expected input/behavior/output, not one-line labels. " "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION + "\n\nGoal: {goal_text}\n\nSDD:\n\n{sdd_content}\n\nPRD:\n\n{prd_content}" ), "validator_test_dataset": ( "Here is your own independent Benchmark for goal '{goal_text}':" "\n\n{validator_benchmark_content}\n\n" "Judge whether this goal has a natural, concrete test dataset. If not, reply via the " "required schema with content starting exactly with 'N/A: ' plus a one-sentence reason. " "If so, produce your OWN independent dataset -- don't assume what the Coder actually " "used, you have no visibility into that and shouldn't need it. Reply via the required " "schema. " + _NO_NARRATION ), "validator_test_script": ( "Here is your own independent Benchmark:\n\n{validator_benchmark_content}\n\n" "Here is your own test dataset (may be 'N/A'):\n\n{validator_test_dataset_content}\n\n" "Write a real, runnable test script that mechanically checks your own Benchmark's " "scenarios against the ACTUAL current implementation in this repository's working " "directory -- run real commands, inspect real files/behavior, don't assume anything " "about what was implemented. This script will be executed for real, and if it exits " "non-zero the verdict is FAIL regardless of any other judgment -- make sure it actually " "tests the real thing, not a strawman. If genuinely no mechanical script is possible for " "this goal, reply via the required schema with content starting exactly with 'N/A: ' " "plus a one-sentence reason -- this should be rare. " + _NO_NARRATION ), } # 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": []}, { "node_id": "validator_test_dataset", "artifact_type": "validator_test_dataset", "depends_on": ["validator_benchmark"], }, ], "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 " "judged no mechanical script was possible, form your judgment from the Benchmark alone in " "that case). Reply with a verdict line 'VERDICT: PASS' or 'VERDICT: FAIL' or " "'VERDICT: INCONCLUSIVE' followed by your reasoning and, if FAIL, a root_cause_category " "(planning|coder|skill_defect) with detail.\n\n" "Goal: {goal_text}\n\nYour own Benchmark:\n\n{validator_benchmark_content}\n\n" "Your own test script's REAL execution result:\n\n{script_execution_result}" ) class _MissingContentDefault(dict): """2026-08-08 (found via a real self-planned run, task involving a src/work_node_run.py bug fix): `_DOC_PROMPTS` entries were all written for the old fixed brd->prd->sdd->benchmark->... chain and hardcode placeholders like {prd_content} assuming that upstream node always exists. Work Graph self-planning (FR-B4 "全部自由规划") legitimately lets the Planner skip nodes that don't apply -- this real run's SDD node correctly had no PRD node at all (a small bug-fix goal doesn't need one), so `_DOC_PROMPTS["sdd"]`'s `{prd_content}` placeholder had nothing to fill it and `.format()` raised a bare KeyError, landing the task on `blocked_pending_human` with the unhelpful generic 'execution_error' reason. `str.format_map` with this as the mapping substitutes a clear placeholder for any content key the self-planned graph didn't actually produce, instead of crashing.""" def __missing__(self, key): return "N/A (no corresponding upstream node was included in this Work Graph)" def _planner_prompt_builder(goal_text, node_artifact_types): """node_artifact_types: dict[node_id] -> artifact_type, read off the real graph_manifest before running it (build_prompt only ever sees one node/outputs-so-far pair at a time, not the whole manifest).""" def build_prompt(node, outputs): node_id = node["node_id"] artifact_type = node_artifact_types.get(node_id, node_id) content_kwargs = {"goal_text": goal_text} for dep_id in (node.get("depends_on") or []): dep_output = outputs.get(dep_id) if dep_output is not None: dep_type = node_artifact_types.get(dep_id, dep_id) content_kwargs[dep_type + "_content"] = dep_output["content"] return _DOC_PROMPTS[artifact_type].format_map(_MissingContentDefault(content_kwargs)) return build_prompt def _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types): """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 unconditionally into every node's kwargs; .format() only consumes the placeholders each template actually references, extra kwargs are harmless.""" def build_prompt(node, outputs): node_id = node["node_id"] artifact_type = node_artifact_types.get(node_id, node_id) content_kwargs = {"sdd_content": sdd_content, "prd_content": prd_content} for dep_id in (node.get("depends_on") or []): dep_output = outputs.get(dep_id) if dep_output is not None: dep_type = node_artifact_types.get(dep_id, dep_id) content_kwargs[dep_type + "_content"] = dep_output["content"] return _VALIDATOR_DOC_PROMPTS[artifact_type].format(goal_text=goal_text, **content_kwargs) return build_prompt def _coerce_bool(value): """A real call (2026-08-02) returned `"consistent": "false"` as a JSON STRING, not a boolean, even with `"type": "boolean"` in the schema -- `if "false":` is truthy in Python, so this silently inverted the check until caught by a real test (mocks always return real Python bools, so this never showed up until RUN_REAL_AGENT_E2E=1). Don't trust the schema's declared type any more than 05 §5.4.1.1 already tells us to trust --json-schema itself.""" if isinstance(value, bool): return value if isinstance(value, str): return value.strip().lower() == "true" return bool(value) def _check_prd_consistency(target_folder_path, prd_content, timeout=180): """REQ-02 boundary: one real agent() call reviewing the PRD's own content for genuine internal contradiction. Returns (ok, issues). **Self-target worktree isolation (2026-08-03, real self-hosting run)**: this call uses `agentType='planner'` with `memory: project`, which writes a real, uncommitted change to `.claude/agent-memory/planner/ MEMORY.md` in whatever `cwd` it runs with. Every prior real run of this check only ever targeted a throwaway target-project clone, so calling it directly against `target_folder_path` never caused a problem -- until a genuine self-hosting run hit it: the memory write landed as an uncommitted local change on the *main* checkout (bypassing `work_node_run.run_work_node`'s worktree isolation entirely, since this call never goes through it), which then made the *next* node's own worktree merge fail with "local changes would be overwritten by merge". 05 §5.8's unconditional self-target worktree rule has to apply here too, not just to `run_work_node`'s own callers.""" is_self = paths.is_self_target(target_folder_path) node_exec_id = "prd-consistency-" + uuid.uuid4().hex[:12] worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None cwd = worktree_path or target_folder_path try: prompt = _PRD_CONSISTENCY_PROMPT.format(prd_content=prd_content) result = org_node_trigger.trigger_org_node_structured( "planner", prompt, cwd, _PRD_CONSISTENCY_SCHEMA, timeout=timeout ) if worktree_path: work_node_run.ensure_worktree_committed(worktree_path, node_exec_id) # also catch drift on the MAIN checkout (worktrees share project # identity, so a real `memory: project` write can land there # regardless of `cwd` -- see work_node_run.ensure_worktree_ # committed's own docstring) before the merge below. work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift") work_node_run.merge_worktree(target_folder_path, node_exec_id) return _coerce_bool(result["consistent"]), result.get("issues", []) finally: if worktree_path: work_node_run.remove_worktree(target_folder_path, worktree_path) def _prd_consistency_verify_fn(node_artifact_types, target_folder_path): """work_graph_orchestrate.run_graph's verify_fn -- only the PRD node gets checked, every other node passes through untouched.""" def verify_fn(node, output): if node_artifact_types.get(node["node_id"]) != "prd": return True, None return _check_prd_consistency(target_folder_path, output["content"]) return verify_fn def _is_na_content(content): """test_dataset/test_script are allowed to legitimately not apply to a given goal -- '_DOC_PROMPTS' asks for content starting with 'N/A: ' in that case. That's a pass, not something to compile-check.""" return content.strip().upper().startswith("N/A") def _detect_script_language(content): """Tries Python first, then Node/JS (the two languages this project's own tests are ever written in -- see CLAUDE.md's Testing constraints). Returns (language_or_None, detail) -- language is 'python'/'javascript' on success, None if neither compiles (detail then holds both compilers' real error output). Shared by check_script_compiles (the verify_fn, which only needs a yes/no) and run_test_script_for_real (which also needs to know HOW to actually invoke the script).""" with tempfile.TemporaryDirectory(prefix="graphlemon-script-check-") as tmpdir: py_path = os.path.join(tmpdir, "candidate.py") with open(py_path, "w", encoding="utf-8") as f: f.write(content) py_result = subprocess.run( ["python3", "-m", "py_compile", py_path], capture_output=True, text=True ) if py_result.returncode == 0: return "python", None js_path = os.path.join(tmpdir, "candidate.js") with open(js_path, "w", encoding="utf-8") as f: f.write(content) js_result = subprocess.run(["node", "--check", js_path], capture_output=True, text=True) if js_result.returncode == 0: return "javascript", None return None, ( "content does not compile as Python or Node/JS -- must be a real, runnable script, " "not a narrative description.\npython3 -m py_compile:\n{}\nnode --check:\n{}".format( py_result.stderr, js_result.stderr ) ) def check_script_compiles(content): """Mechanical (not LLM-judged) check that `content` is real, syntactically valid code, not narrative prose masquerading as a script -- this is the one node type where a genuinely objective check is possible (2026-08-06 design: see docs/implementation-gap-analysis.md's Benchmark-narration incident). Returns (ok, detail).""" if _is_na_content(content): return True, None language, detail = _detect_script_language(content) return language is not None, detail _NARRATION_RETRY_HINT = ( "If your previous output was a narrative description of an action (e.g. describing a task/" "workflow you ran, launched, or will report on) instead of literal script source code, " "discard it entirely -- do not describe what you did, are doing, or launched. The `content` " "field must contain nothing but the script's real source code, starting from its very first " "real line (e.g. an import/shebang), with no leading or trailing prose." ) def _test_script_verify_fn(node_artifact_types, script_artifact_type="test_script"): """work_graph_orchestrate.run_graph's verify_fn for a test_script node -- mechanical compile check, not LLM judgment. On failure the same reflecting/retry-once mechanism the PRD-consistency check already uses kicks in (see run_graph's own docstring). `script_artifact_type` is parameterized so this same function backs both the Planner's 'test_script' node and the Validator's own 'validator_test_script' node. 2026-08-07 (docs/done/validator-test-script-narrative-output-issue.md): `check_script_compiles`'s own detail string already says "not a narrative description", but that's a generic explanation of what the mechanical check looks for, buried ahead of two raw compiler tracebacks -- real observed reruns still reproduced near-identical narrative content on the very next reflect attempt. `_NARRATION_RETRY_HINT` adds an explicit, directive corrective as its own separate feedback item (phrased conditionally, so it's a harmless no-op for genuine syntax-typo failures that were never narration in the first place).""" def verify_fn(node, output): if node_artifact_types.get(node["node_id"]) != script_artifact_type: return True, None ok, detail = check_script_compiles(output["content"]) # work_graph_orchestrate._verify_and_maybe_regenerate does # "; ".join(feedback or []), i.e. it expects a list of messages (the # PRD-consistency verify_fn's `issues` shape) -- check_script_compiles # returns a single string, which `join` would otherwise iterate # character-by-character into a garbled reflect-prompt. return ok, ([detail, _NARRATION_RETRY_HINT] if detail else None) return verify_fn def _combined_planner_verify_fn(node_artifact_types, target_folder_path): """Dispatches to whichever of the per-node-type verify_fns actually applies -- run_graph only accepts one verify_fn, and each node type's check is independent (a node is at most one artifact_type).""" prd_check = _prd_consistency_verify_fn(node_artifact_types, target_folder_path) script_check = _test_script_verify_fn(node_artifact_types) def verify_fn(node, output): artifact_type = node_artifact_types.get(node["node_id"]) if artifact_type == "prd": return prd_check(node, output) if artifact_type == "test_script": return script_check(node, output) return True, None 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 _resolve_node_artifact_type(node, node_artifact_types): """A node promotes to an `artifact` row if EITHER: it's in the caller's precomputed `node_artifact_types` dict (Planner's/Validator's fixed catalogs, where every node is a promotable artifact by construction -- unchanged behavior), OR it declares `artifact_type` directly on itself, OR (2026-08-12, Phase 2 §2.3) it declares `promotes_to_ artifact: {"artifact_type": ...}` (Coder's/Validator's generic self-planned nodes -- most of which should declare NEITHER, meaning "this is an internal graph step, not a final deliverable"). Returns None for a node that isn't promoting anything -- not an error, the expected case for most self-planned nodes.""" if node["node_id"] in node_artifact_types: return node_artifact_types[node["node_id"]] if "artifact_type" in node: return node["artifact_type"] promotes = node.get("promotes_to_artifact") if promotes: return promotes.get("artifact_type") return None def _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, run_graph_error=None, new_node_ids=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. 2026-08-12 (Phase 2 §2.3/§6): a node with no resolvable artifact_type (see `_resolve_node_artifact_type`) is silently skipped, not an error -- most Coder/Validator self-planned nodes are internal graph steps, not final deliverables. 2026-08-13 (real incident, task b243dfe8 -- diagnosed live with the user, correcting an architectural misunderstanding they caught): the OLD `only_node_ids` param skipped any node not in "this call's own newly-planned nodes", on the assumption that "not new to this call" means "an earlier stage already promoted it". That assumption breaks the instant an earlier append's OWN promotion attempt never actually ran -- e.g. this exact function being reached only via a code path that crashed earlier in THAT call, before ever getting here. The node's real, valid, already-'done' content in work_node_execution. output_payload then becomes permanently unpromotable: every LATER retry (in append mode) kept excluding it as "not new", forever, even though zero artifact of that type had EVER actually been written -- exactly what happened to this task (multiple 'done' brd/prd/sdd/ benchmark nodes across retries, zero artifact rows, zero .graphlemon/runs// files ever created, going back to the task's original 2026-08-04 run). Renamed to `new_node_ids` and its role inverted: a node whose type ALREADY has a real artifact row for this run_id is skipped UNLESS the node itself is in `new_node_ids` (default None -- treated as "nothing is new", the safe default for a caller with no such concept). This covers both real scenarios correctly, which a pure "already promoted -> always skip" rule (tried first, broke the retry-loop's own already-passing regression test -- a rejected attempt's superseding content MUST be able to promote a fresh artifact of the same type) can't distinguish on its own: - an OLD node (not new to this call) whose type was never actually promoted (this task's real bug): not in already_promoted_types -> promoted normally, regardless of new_node_ids. - an OLD node whose type genuinely was already promoted by an earlier, successful call: skipped -- no duplicate. - a NEW node (in new_node_ids) whose type happens to already have an artifact from a REJECTED earlier attempt (the retry-loop case): promoted anyway, correctly superseding the rejected one.""" 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) already_promoted_types = { row[0] for row in conn.execute( "SELECT artifact_type FROM artifact WHERE task_id = ? AND produced_by_run_id = ?", (task_id, run_id) ).fetchall() } new_node_ids = new_node_ids if new_node_ids is not None else set() 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 = _resolve_node_artifact_type(node, node_artifact_types) if artifact_type is None: continue if artifact_type in already_promoted_types and node_id not in new_node_ids: continue 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"] already_promoted_types.add(artifact_type) 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, append_to_run_id=None): """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. 2026-08-08 (Work Graph self-planning + loop-back, FR-B1): `verify_fn` is no longer `_combined_planner_verify_fn` -- each self-planned node 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` 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. `append_to_run_id` (2026-08-13, ux_work_graph_run_active fix -- see `_review_planner_output_with_retries`'s own docstring): default None keeps this a fresh `plan_work_graph()` INSERT, unchanged from before this parameter existed -- the very first Planner call for a task, when no work_graph_run row exists for it yet. Passed a real run_id (a rejected planner_reviewer attempt re-planning into the SAME task's still-non-terminal run, same convention `implement_code()` already uses for Coder), this appends the freshly-planned BRD/PRD/SDD/Benchmark nodes into that existing row instead of INSERTing a second one -- `ux_work_graph_run_active` allows only one non-terminal row per task_id, and the earlier attempt's row is never finalized to 'done'/'failed' just because its content was rejected (Phase 2c: one run spans a task's whole lifecycle). `_write_completed_node_artifacts` itself now decides what to (re-)promote by checking real artifact rows for this run_id (2026-08-13 fix, see its own docstring) rather than by which call planned which node, so a re-plan naturally promotes whatever hasn't been genuinely promoted yet -- including an earlier attempt's own already-'done' nodes if THEIR promotion never actually ran. """ plan_result = work_graph_plan.plan_work_graph( conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout, append_to_run_id=append_to_run_id, ) run_id = plan_result["run_id"] 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"]} 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, mechanical_check_registry={"check_script_compiles": check_script_compiles}, ) 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, new_node_ids=set(plan_result["new_node_ids"]), ) if run_graph_error is not None: raise run_graph_error return {"run_id": run_id, "artifact_ids": artifact_ids} # ponytail: 子串匹配，从 docs/done/validator-test-script-narrative-output-issue.md # 记录的真实复现文案里提炼——"以为在描述自己做了什么"的稳定措辞模式。只作为喂给 # planner_reviewer 的提示信号，不是硬性机械拦截（跟 check_script_compiles 不同： # 这里没有客观的"编译通过/失败"可判断，误判成本是把合法的、恰好提到这些词的文档 # 打回重来）。上限：这份短语表穷举不完，Phase 2 的结构性修复（把"探索"和"最终作答" # 拆成两次独立调用，见 validator-test-script-narrative-output-issue.md §8）才是 # 真正的上游修复，这里只是先兜住已经复现过的具体案例。 _NARRATION_PHRASES = ( "no further action", "launched the", "workflow finished", "workflow completed", "it produced a", "you'll be notified", "no files were changed", ) def looks_like_narration(content): lowered = (content or "").lower() return any(phrase in lowered for phrase in _NARRATION_PHRASES) _PLANNER_REVIEW_PROMPT = ( "Independently review the Planner's output for this goal, before Coder is allowed to " "start implementing. Judge two things: (1) is this real, structured document content, " "not narrative prose describing an action taken/completed/launched; (2) is it consistent " "with the goal and internally consistent across the documents provided. Some document " "types below may be 'N/A: not provided' -- that's expected for goals that don't need " "them, not itself a defect.\n\n{narration_hint}" "Reply with a verdict line 'VERDICT: PASS' or 'VERDICT: FAIL' or 'VERDICT: INCONCLUSIVE' " "followed by your reasoning.\n\n" "Goal: {goal_text}\n\nBRD:\n\n{brd_content}\n\nPRD:\n\n{prd_content}\n\n" "SDD:\n\n{sdd_content}\n\nBenchmark:\n\n{benchmark_content}" ) def review_planner_output( conn, task_id, goal_text, target_folder_path, sdd_content, brd_content="N/A: not provided", prd_content="N/A: not provided", benchmark_content="N/A: not provided", timeout=300, ): """The real planner_reviewer judgment call (not trigger_org_node_full's ready handshake). Returns {"verdict", "raw_response"}.""" hints = [ "Heuristic flag: {} content resembles a narrative description of an action " "rather than structured document content -- verify carefully.".format(label) for label, content in (("SDD", sdd_content), ("Benchmark", benchmark_content)) if looks_like_narration(content) ] response = org_node_trigger.trigger_org_node( "planner_reviewer", _PLANNER_REVIEW_PROMPT.format( goal_text=goal_text, brd_content=brd_content, prd_content=prd_content, sdd_content=sdd_content, benchmark_content=benchmark_content, narration_hint=("\n".join(hints) + "\n\n" if hints else ""), ), target_folder_path, timeout=timeout, ) return {"verdict": _parse_verdict_line(response), "raw_response": response} _PLANNER_REVIEW_MAX_ATTEMPTS = 3 class PlannerReviewLoopLimitReachedError(Exception): """_PLANNER_REVIEW_MAX_ATTEMPTS rounds of planner_reviewer all failed to produce PASS.""" def _existing_doc_artifact_contents(conn, target_folder_path, task_id): """{artifact_type: content} for whichever of brd/prd/sdd/benchmark genuinely already exist for this task -- may be an empty dict (e.g. the task-execution-resume-assumes-planner-artifacts-exist.md incident: the Planner's Work Graph failed on its very first node, nothing was ever persisted). Existence-probing instead of hard-failing is the actual P0 bug fix -- see _review_planner_output_with_retries's docstring. Import deferred to call time (not module level): task_execution.py already imports this module AND scheduler.py, and scheduler.py's own top level reads e2e_pipeline.PlannerReviewLoopLimitReachedError to build _BLOCKED_REASON_FOR_EXCEPTION -- a module-level `from src import task_execution` here closes that into a real 3-way import cycle (e2e_pipeline -> task_execution -> scheduler -> e2e_pipeline) that breaks depending on which module happens to be imported first. Neither function that needs task_execution's helpers runs at import time, so a local import is a genuine fix, not a workaround.""" from src import task_execution ids = task_execution._latest_artifact_ids(conn, task_id, task_execution._DOC_ARTIFACT_TYPES) return {t: _read_artifact_content(conn, target_folder_path, aid) for t, aid in ids.items()} def _record_planner_review_event(conn, task_id, attempt, verdict, used_existing_artifacts): project_id = conn.execute("SELECT project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()[0] conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) " "VALUES (?, ?, ?, 'planner_review_verdict', ?)", ( str(uuid.uuid4()), project_id, task_id, json.dumps({"attempt": attempt, "verdict": verdict, "used_existing_artifacts": used_existing_artifacts}), ), ) conn.commit() def _existing_active_run_id(conn, task_id): """2026-08-13 (real recurrence on task b243dfe8): finds a task_id's existing non-terminal work_graph_run row directly against the table ux_work_graph_run_active actually constrains, NOT via artifact rows (unlike _existing_doc_artifact_contents/_shared_run_id_for_task) -- a run can exist with a real graph_manifest but zero promoted artifacts (e.g. interrupted before any node's content was written), which those artifact-based lookups are blind to. Only one non-terminal row can ever exist per task_id (the index enforces it), so this is unambiguous when it returns non-None.""" row = conn.execute( "SELECT run_id FROM work_graph_run WHERE task_id = ? AND status NOT IN ('done', 'failed') " "ORDER BY rowid DESC LIMIT 1", (task_id,), ).fetchone() return row[0] if row else None def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_path, timeout=300): """fresh-run (plan_and_review_documents) and resume-from- reviewing_planner_output (task_execution.py) both drive this same core loop. Attempt 1 prefers reusing whatever doc artifacts already exist (the resume case: don't burn a fresh plan_documents() call re-doing work that already succeeded) -- if none exist (P0 bug scenario: the Planner graph failed before persisting anything), or on any subsequent attempt, calls plan_documents() to generate a fresh batch. Every attempt records a planner_review_verdict execution_event for audit; Phase 1 itself never reads it back to make a decision. Returns plan_documents()'s return shape from whichever attempt earned PASS. ponytail: every rejected attempt triggers a FULL re-plan of BRD/PRD/SDD/ Benchmark, not a targeted fix of just the rejected document -- coarse but correct, same granularity as org_loop_route.py's existing full_replan. Phase 2's node-level loop_back (org-graph-unification-and- universal-self-planning-requirements.md Phase 2) is the precise-retry upgrade path. 2026-08-13 (ux_work_graph_run_active fix, real prod IntegrityError -- two live tasks crashed 'sqlite3.IntegrityError: UNIQUE constraint failed: work_graph_run.task_id' right after a non-PASS planner_review_ verdict): a rejected attempt used to call `plan_documents()` with no run_id, which always does a fresh `plan_work_graph()` INSERT -- fine before `ux_work_graph_run_active` existed, but that index now allows only ONE non-terminal work_graph_run row per task_id, and the earlier (rejected) attempt's row is never marked 'done'/'failed' just because its content was rejected (Phase 2c: one run spans a task's whole lifecycle, `work_graph_run.status` is not how "rejected" is recorded -- `planner_review_verdict` execution_events are). `prior_run_id` carries whatever run_id the previous attempt used (the fresh-plan branch's own new run_id, or the resume branch's `sdd_run_id`) into the next attempt's `plan_documents(..., append_to_run_id=prior_run_id)` call -- same "append into the still-open run" convention `implement_code()` already uses for Coder, not a new one invented here. The very first attempt (no prior run_id yet) is unaffected: `append_to_run_id=None` still does the original fresh INSERT. 2026-08-13 follow-up (same day, real recurrence on task b243dfe8 after the fix above already landed): the fix above only threads `prior_run_id` across retry ITERATIONS WITHIN one call to this function -- it missed a second, real source of collision: a task can already have an orphaned non-terminal `work_graph_run` row from a PRIOR, separately-interrupted execution (e.g. a Planner call that persisted `graph_manifest` but crashed/got restarted before any node's content was promoted to an `artifact` row) *before this function is even called at all*. `_existing_doc_artifact_contents` only finds a run via real artifact rows, so it's blind to this case -- attempt 1 still started from `prior_run_id = None` and collided with that leftover row exactly like before. Seeding `prior_run_id` from `_existing_active_run_id()` (queries `work_graph_run` directly, the actual table `ux_work_graph_run_active` enforces uniqueness against, independent of whether artifacts exist yet) closes this: since the index allows only one non-terminal row per task_id, ANY such row IS "the" run for this task by construction (Phase 2c merges Planner/Coder/Validator into one shared run), so appending into it is always correct, never a wrong-run guess. """ from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred feedback_suffix = "" verdict = None prior_run_id = _existing_active_run_id(conn, task_id) for attempt in range(1, _PLANNER_REVIEW_MAX_ATTEMPTS + 1): existing = _existing_doc_artifact_contents(conn, target_folder_path, task_id) if attempt == 1 else {} if existing: artifact_ids = task_execution._latest_artifact_ids( conn, task_id, task_execution._DOC_ARTIFACT_TYPES ) # Phase 2 §6.2 (2026-08-12, found while wiring implement_code()'s # new append_to_run_id requirement): this used to hardcode # run_id=None -- harmless for Phase 1 (nothing downstream read # it), but Phase 2's Coder self-planning needs a REAL run_id to # append into. Every real artifact row already has one # (produced_by_run_id is NOT NULL) -- read it off the sdd # artifact (required a few lines below anyway, guaranteed # present whenever `existing` is truthy) rather than inventing # a new lookup path. sdd_run_id = conn.execute( "SELECT produced_by_run_id FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],) ).fetchone()[0] plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids} else: plan_result = plan_documents( conn, task_id, goal_text + feedback_suffix, target_folder_path, append_to_run_id=prior_run_id, ) artifact_ids = plan_result["artifact_ids"] prior_run_id = plan_result["run_id"] sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"]) brd_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["brd"]) if "brd" in artifact_ids else "N/A: not provided" ) prd_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["prd"]) if "prd" in artifact_ids else "N/A: not provided" ) benchmark_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"]) if "benchmark" in artifact_ids else "N/A: not provided" ) review = review_planner_output( conn, task_id, goal_text, target_folder_path, sdd_content, brd_content, prd_content, benchmark_content, timeout=timeout, ) _record_planner_review_event(conn, task_id, attempt, review["verdict"], bool(existing)) verdict = review["verdict"] if verdict == "PASS": return plan_result feedback_suffix = ( "\n\n[Planner review attempt {} was rejected ({}): {}. Address this " "before continuing.]" ).format(attempt, verdict, review["raw_response"]) raise PlannerReviewLoopLimitReachedError( "task {} exhausted {} planner-review attempts, last verdict={}".format( task_id, _PLANNER_REVIEW_MAX_ATTEMPTS, verdict ) ) def plan_and_review_documents(conn, task_id, goal_text, target_folder_path, timeout=300): """fresh-run entry point: two handshakes (planner, planner_reviewer) + the shared retry loop. Replaces run_task_lifecycle()'s previous trigger_org_node_full(planner) + plan_documents() pair.""" org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT) org_node_lifecycle.trigger_org_node_full(conn, "planner_reviewer", task_id, target_folder_path, _READY_PROMPT) return _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_path, timeout=timeout) def implement_code( conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200, benchmark_content="N/A: not provided", test_dataset_content="N/A: not provided", test_script_content="N/A: not provided", ): """Phase 2 §6 (2026-08-12): Coder now self-plans its own implementation graph via `plan_work_graph(..., append_to_run_id=run_id)` -- `run_id` is Planner's own work_graph_run (created by `plan_and_review_ documents()`), no longer a fresh row this function creates itself. Coder's self-planned nodes get appended into that SAME run, not a separate one -- see work_graph_plan.py's own append_to_run_id docstring for the merge mechanics. Replaces the old synthetic single-node "implement" manifest (architecture §11 criterion 2's original, pre-self-planning shape). `code_diff` stays a Python-computed artifact (a real `git diff` across whatever Coder's whole self-planned graph actually changed on disk, end to end) -- not something any self-planned node's own `content` field could meaningfully hold, since no node knows the diff ahead of running it. `test_report` is promoted from whichever self-planned node(s) declared `promotes_to_artifact: {"artifact_type": "test_report"}` (the catalog's own "integration_check" entry suggests this); if the LLM's plan didn't include one (the catalog is suggestions, not requirements), falls back to concatenating every self-planned node's own content in topological order -- same spirit as the old single-node manifest's `response` field always being Coder's own narrative of what it did, just synthesized instead of directly given by one node. `needs_worktree=True` passed to run_graph() (2026-08-12, found while doing this rework -- NOT previously passed here at all, even though Coder is the one org_node that actually writes real code): with a single synthetic node this was harmless (nothing to collide with inside one run), but a genuinely self-planned graph can have 2+ code-writing nodes in the SAME topological layer running concurrently -- 03-C §3.6's "并行写代码的节点必须先申请 git_worktree_ref" applies for real now, not just in theory. 2026-08-06: benchmark_content/test_dataset_content/test_script_content (default 'N/A: not provided' for callers that predate this, e.g. a resumed task from before this change) are folded into _CODER_PLANNING_CONTEXT so Coder's planning call sees the Planner's real test script and must make it pass for real, not just write whatever tests it invents on its own. """ before_sha = _current_commit(target_folder_path) planning_context = _CODER_PLANNING_CONTEXT.format( goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content, test_dataset_content=test_dataset_content, test_script_content=test_script_content, ) plan_result = work_graph_plan.plan_work_graph( conn, "coder", 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"]) # Covers BOTH node shapes that can now coexist in this merged graph: # Planner's own nodes (artifact_type) and Coder's just-appended ones # (promotes_to_artifact) -- used for both _planner_prompt_builder's # prompt construction (only its own artifact_type entries are ever # queried) and _write_completed_node_artifacts' artifact-type/deps # resolution below. 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"] build_prompt = work_graph_plan.dispatch_build_prompt( _planner_prompt_builder(goal_text, node_artifact_types), work_graph_plan.generic_node_prompt_builder, ) run_graph_error = None outputs = None try: outputs = work_graph_orchestrate.run_graph( conn, run_id, "coder", "coder-implementation", target_folder_path, graph_manifest, build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout, needs_worktree=True, mechanical_check_registry={"check_script_compiles": check_script_compiles}, ) except Exception as e: run_graph_error = e diff = subprocess.run( ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True ).stdout diff_result = artifact_write.write_artifact( conn, target_folder_path, "code_diff", task_id, run_id, ".graphlemon/runs/{}/code_diff.patch".format(task_id), diff or "(no diff -- see test_report)", ) 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, new_node_ids=new_node_ids, ) if run_graph_error is not None: raise run_graph_error if "test_report" not in artifact_ids: # The self-planned graph didn't include a node declaring # promotes_to_artifact: {"artifact_type": "test_report"} -- # synthesize one so callers can still rely on this key always # being present (same contract the old single-node manifest gave # unconditionally via its one node's `response` field). report_content = "\n\n---\n\n".join( "### {}\n\n{}".format(node_id, (outputs or {}).get(node_id, {}).get("content", "")) for node_id in new_node_ids if node_id in (outputs or {}) ) or "(Coder's self-planned graph produced no synthesizable report content)" report_result = artifact_write.write_artifact( conn, target_folder_path, "test_report", task_id, run_id, ".graphlemon/runs/{}/test_report.md".format(task_id), report_content, ) artifact_ids["test_report"] = report_result["artifact_id"] return {"code_diff": diff_result["artifact_id"], "test_report": artifact_ids["test_report"]} def _parse_verdict_line(response): verdict = "INCONCLUSIVE" for line in response.splitlines(): stripped = line.strip().upper() if stripped.startswith("VERDICT:"): for candidate in ("PASS", "FAIL", "INCONCLUSIVE"): if candidate in stripped: verdict = candidate break break return verdict def validate_implementation(task_id, benchmark_content, goal_text, target_folder_path, timeout=1200): """Architecture §11 criterion 3: real Validator call, parses the 'VERDICT: PASS|FAIL|INCONCLUSIVE' line this Unit's prompt asks for. Returns {"verdict", "raw_response"}. Unchanged by the 2026-08-01 rewiring -- REQ-04's own coverage never named the Work Graph controllers (c_bug_record_create/p_validator_root_cause_trace only), so the simpler flat trigger_org_node call is already faithful to what this REQ actually requires. **2026-08-06**: kept as-is (still the simpler flat call, no Validator's own Work Graph) for callers that only need this original REQ-04-scoped behavior -- run_task_lifecycle now uses validate_implementation_with_own_graph() instead, which wraps this same prompt/parsing shape but adds the independent benchmark/dataset/script generation and the real-script-execution hard gate. See that function's own docstring and docs/implementation-gap-analysis.md for why (a Benchmark that's only ever prose an LLM narrates over instead of substantive content isn't a real acceptance anchor). """ response = org_node_trigger.trigger_org_node( "validator", _VALIDATOR_PROMPT.format(goal_text=goal_text, benchmark_content=benchmark_content), target_folder_path, timeout=timeout, ) return {"verdict": _parse_verdict_line(response), "raw_response": response} def _script_file_extension(language): return {"python": "py", "javascript": "js"}.get(language, "txt") def run_test_script_for_real(content, target_folder_path, task_id, node_label, timeout=120): """Materializes `content` as a real file under .graphlemon/runs// and actually executes it with cwd=target_folder_path, so it runs against the REAL current state of the repository (whatever the Coder actually did), not a sandboxed guess. Returns {"ran": bool, "exit_code": int|None, "stdout": str, "stderr": str, "note": str}. `ran=False` covers both legitimate N/A content and the (should be rare, since the generating verify_fn already gated this) case where the content still doesn't compile -- both are "nothing to hard-gate on", the caller decides what INCONCLUSIVE-vs-something-else means for each. """ if _is_na_content(content): return {"ran": False, "exit_code": None, "stdout": "", "stderr": "", "note": content} language, detail = _detect_script_language(content) if language is None: return { "ran": False, "exit_code": None, "stdout": "", "stderr": "", "note": "script does not compile, cannot execute: " + detail, } script_path = os.path.join( target_folder_path, ".graphlemon", "runs", task_id, "{}.{}".format(node_label, _script_file_extension(language)), ) os.makedirs(os.path.dirname(script_path), exist_ok=True) with open(script_path, "w", encoding="utf-8") as f: f.write(content) interpreter = "python3" if language == "python" else "node" try: result = subprocess.run( [interpreter, script_path], cwd=target_folder_path, capture_output=True, text=True, timeout=timeout, ) except subprocess.TimeoutExpired as e: return { "ran": True, "exit_code": None, "stdout": e.stdout or "", "stderr": e.stderr or "", "note": "timed out after {}s".format(timeout), } return { "ran": True, "exit_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "note": "exit code {}".format(result.returncode), } def _format_script_execution_result(result): if not result["ran"]: return "Not executed ({}).".format(result["note"]) return ( "Executed for real. Exit code: {}\nstdout:\n{}\nstderr:\n{}" ).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, run_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200, ): """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"}. """ 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, ) run_graph_error = None outputs = None try: outputs = work_graph_orchestrate.run_graph( 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, graph_manifest, node_artifact_types, _planner_ext_for, run_graph_error, new_node_ids=new_node_ids, ) if run_graph_error is not None: raise run_graph_error # 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", } # 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": "PASS", "raw_response": response, "artifact_ids": artifact_ids, "script_execution": script_execution, } def _current_commit(repo_path): return subprocess.run( ["git", "rev-parse", "HEAD"], cwd=repo_path, capture_output=True, text=True ).stdout.strip() def run_task_lifecycle(conn, task_id, target_folder_path, goal_text): """The body of run_end_to_end() for an ALREADY-CREATED task -- split out 2026-08-02 so scheduler.py can create the task itself first and still have a task_id in scope if this raises partway through (needed to route CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError to task.status='blocked_pending_human', which is scheduler.py's job, not this module's -- see that module's docstring). Behavior/return shape is unchanged from before the split; run_end_to_end() below is now a 2-line wrapper over this. """ plan_result = plan_and_review_documents(conn, task_id, goal_text, target_folder_path) artifact_ids = plan_result["artifact_ids"] sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"]) # FIX (2026-08-08, Org Graph unification Phase 1): prd/benchmark used to be # artifact_ids["prd"]/artifact_ids["benchmark"] unconditional subscripts, which # KeyError'd whenever a self-planned graph omitted either type -- not resume-only, # this crashed a fresh first run too. Same "in" guard test_dataset/test_script # already used below. prd_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["prd"]) if "prd" in artifact_ids else "N/A: not provided" ) benchmark_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"]) if "benchmark" in artifact_ids else "N/A: not provided" ) # test_dataset/test_script are new 2026-08-06 nodes -- guard with .get # so a task resumed from an artifact set produced before this change # (no test_dataset/test_script artifact_ids) still works, falling back # to implement_code's own "not provided" defaults. test_dataset_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["test_dataset"]) if "test_dataset" in artifact_ids else "N/A: not provided" ) test_script_content = ( _read_artifact_content(conn, target_folder_path, artifact_ids["test_script"]) if "test_script" in artifact_ids else "N/A: not provided" ) org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, _READY_PROMPT) code_artifact_ids = implement_code( conn, task_id, plan_result["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, ) 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, plan_result["run_id"], sdd_content, prd_content, goal_text, target_folder_path, ) return { "task_id": task_id, "doc_artifact_ids": artifact_ids, "code_artifact_ids": code_artifact_ids, "verdict": verdict, } def run_end_to_end(conn, target_folder_path, goal_text): """Full orchestration: create_task -> plan_documents -> implement_code -> validate_implementation. Returns a dict with every intermediate result, for inspection/debugging rather than hiding it behind a single verdict. """ project_id = conn.execute( "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,) ).fetchone()[0] task_id = task_create.create_task(conn, project_id, goal_text)["task_id"] return run_task_lifecycle(conn, task_id, target_folder_path, goal_text) def _read_artifact_content(conn, target_folder_path, artifact_id): file_path = conn.execute("SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)).fetchone()[0] with open(os.path.join(target_folder_path, file_path), encoding="utf-8") as f: return f.read()