"""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, only_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. `only_node_ids` (default None = every node in `graph_manifest`, unchanged behavior) restricts processing to a subset -- needed once a caller works over a MERGED graph (Coder's `append_to_run_id`) and must avoid re-processing/re-promoting nodes an EARLIER stage already wrote artifacts for. """ done_rows = conn.execute( "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'", (run_id,), ).fetchall() done_outputs = {node_id: json.loads(output_payload) for node_id, output_payload in done_rows} for node_id in list(done_outputs): reflect_id = node_id + "-reflect" if reflect_id in done_outputs: done_outputs[node_id] = done_outputs[reflect_id] for excluded in _failed_node_ids(run_graph_error): done_outputs.pop(excluded, None) artifact_ids = {} for node in work_graph_orchestrate.topological_order(graph_manifest): node_id = node["node_id"] if node_id not in done_outputs: continue if only_node_ids is not None and node_id not in only_node_ids: continue artifact_type = _resolve_node_artifact_type(node, node_artifact_types) if artifact_type is None: 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"] return artifact_ids # test_dataset/test_script aren't necessarily Markdown (a script is real # code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately # generic container; Coder materializes it into a properly-named, # properly-placed real file under the target project's own test # conventions when it actually uses it, that's not this write step's job. _PLANNER_ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"} def _planner_ext_for(artifact_type): return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md") def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): """Architecture §11 criterion 1: a real Planner Work Graph (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark, each written via c_artifact_write with correct lineage. Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}. 2026-08-07: artifacts are now written from whatever's genuinely done in the DB (see _write_completed_node_artifacts), not from run_graph()'s in-memory return value -- so a failure partway through (e.g. the last node in the chain) still persists everything that succeeded before it, instead of discarding it all. The original run_graph() failure is re-raised unchanged after salvaging -- this does not change whether or when a task ends up blocked_pending_human, only what's left behind for a human's Resume to build on when it does. 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. """ plan_result = work_graph_plan.plan_work_graph( conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout ) 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, ) 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 _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. """ from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred feedback_suffix = "" verdict = None 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) artifact_ids = plan_result["artifact_ids"] 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, only_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, only_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()