"""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_execute, 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"]} _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}" ) _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-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 ), } # 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": [], } _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): """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: script_name = "graphlemon-prd-consistency-" + uuid.uuid4().hex prompt = _PRD_CONSISTENCY_PROMPT.format(prd_content=prd_content) script_content = ( "export const meta = {{ name: {name}, description: 'graphlemon PRD consistency check' }}\n\n" "const result = await agent({prompt}, {{ agentType: 'planner', schema: {schema} }})\n\n" "return result\n" ).format(name=json.dumps(script_name), prompt=json.dumps(prompt), schema=json.dumps(_PRD_CONSISTENCY_SCHEMA)) work_graph_execute.write_workflow_script(cwd, script_name, script_content) result = work_graph_execute.execute_work_graph( cwd, script_name, {}, _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/todo/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 _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, run_graph_error=None, ): """Writes an artifact for every node that's genuinely done AND verify_fn-accepted, sourced from `work_node_execution` in the DB -- not a run_graph() return value, so this works identically whether run_graph() returned normally (every node accounted for) or raised partway through (only some nodes are). 2026-08-07 (real incident, docs/todo/work-graph-partial-artifact- persistence-requirements.md): run_graph() raising discards its would-be return value entirely (a Python exception drops the local scope), so previously ANY node failing -- even the very last one in an otherwise-fully-successful chain -- meant NOTHING got persisted, silently throwing away real, already-generated, already-good content. A resumed task then found no BRD/PRD/SDD artifacts to build on at all, even though they'd been generated correctly the first time. A node whose OWN "-reflect" retry exists (verify_fn rejected the first attempt) uses the reflect attempt's content, not the original's -- the reflect node_id isn't part of `graph_manifest` so the main loop below never visits it directly. `_failed_node_ids(run_graph_error)` nodes are excluded outright: their row is schema-valid ('done') but verify_fn never accepted either attempt, so neither is trustworthy content to persist. """ done_rows = conn.execute( "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'", (run_id,), ).fetchall() done_outputs = {node_id: json.loads(output_payload) for node_id, output_payload in done_rows} for node_id in list(done_outputs): reflect_id = node_id + "-reflect" if reflect_id in done_outputs: done_outputs[node_id] = done_outputs[reflect_id] for excluded in _failed_node_ids(run_graph_error): done_outputs.pop(excluded, None) artifact_ids = {} for node in work_graph_orchestrate.topological_order(graph_manifest): node_id = node["node_id"] if node_id not in done_outputs: continue artifact_type = node_artifact_types[node_id] deps = [ artifact_ids[node_artifact_types[dep_id]] for dep_id in (node.get("depends_on") or []) if node_artifact_types.get(dep_id) in artifact_ids ] result = artifact_write.write_artifact( conn, target_folder_path, artifact_type, task_id, run_id, ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext_for(artifact_type)), done_outputs[node_id]["content"], depends_on_artifact_ids=deps, ) artifact_ids[artifact_type] = result["artifact_id"] return artifact_ids # test_dataset/test_script aren't necessarily Markdown (a script is real # code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately # generic container; Coder materializes it into a properly-named, # properly-placed real file under the target project's own test # conventions when it actually uses it, that's not this write step's job. _PLANNER_ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"} def _planner_ext_for(artifact_type): return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md") def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): """Architecture §11 criterion 1: a real Planner Work Graph (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark, each written via c_artifact_write with correct lineage. Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}. 2026-08-07: artifacts are now written from whatever's genuinely done in the DB (see _write_completed_node_artifacts), not from run_graph()'s in-memory return value -- so a failure partway through (e.g. the last node in the chain) still persists everything that succeeded before it, instead of discarding it all. The original run_graph() failure is re-raised unchanged after salvaging -- this does not change whether or when a task ends up blocked_pending_human, only what's left behind for a human's Resume to build on when it does. 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`/ `_test_script_verify_fn` are NOT deleted -- `validate_implementation_ with_own_graph()` (Validator's still-hardcoded manifest, unaffected by this round) continues to use them (FR-A8). `mechanical_check_registry` is still passed unconditionally: a self-planned node declaring `judgment.kind == "mechanical"` (e.g. reusing the catalog's suggested `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} def implement_code( conn, task_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", ): """Architecture §11 criterion 2: a real Coder call via c_work_node_run (work_graph_orchestrate.run_graph with a synthetic single-node manifest -- see module docstring for why there's no real Coder graph planner), then registers whatever the Coder actually committed as code_diff (the diff itself) and test_report (the Coder's own response text, which per its migrated SOUL.md is required to include real test commands + real output) artifacts. 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 threaded into _CODER_PROMPT so the Coder must make the Planner's real test script pass for real, not just write whatever tests it invents on its own. """ before_sha = _current_commit(target_folder_path) run_id = str(uuid.uuid4()) manifest = {"nodes": [{"node_id": "implement", "depends_on": []}], "edges": []} conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " "VALUES (?, ?, 'coder', ?, 'running')", (run_id, task_id, json.dumps(manifest)), ) conn.commit() outputs = work_graph_orchestrate.run_graph( conn, run_id, "coder", "coder-implementation", target_folder_path, manifest, lambda node, outputs: _CODER_PROMPT.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, ), _CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout, ) response = outputs["implement"]["response"] 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)", ) report_result = artifact_write.write_artifact( conn, target_folder_path, "test_report", task_id, run_id, ".graphlemon/runs/{}/test_report.md".format(task_id), response, ) return {"code_diff": diff_result["artifact_id"], "test_report": report_result["artifact_id"]} 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 validate_implementation_with_own_graph( conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200, ): """2026-08-06: Validator's own Work Graph (validator_benchmark -> validator_test_dataset -> validator_test_script), independently generated from SDD/PRD -- NOT reusing the Planner's/Coder's own benchmark/dataset/script (see _VALIDATOR_DOC_PROMPTS's module comment for why: a held-out eval set, not a re-run of the dev set the Coder already developed against). The resulting test_script, if real (not N/A), is then actually EXECUTED against the current repository state -- a non-zero real exit code force-overrides the verdict to FAIL in code, not just by asking the LLM nicely (user-confirmed 2026-08-06: script is the hard gate, subjective judgment can only add INCONCLUSIVE/risk notes on top of a passing script, never override a failing one to PASS). Returns {"verdict", "raw_response", "artifact_ids", "script_execution"}. """ run_id = str(uuid.uuid4()) conn.execute( "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " "VALUES (?, ?, 'validator', ?, 'running')", (run_id, task_id, json.dumps(_VALIDATOR_MANIFEST)), ) conn.commit() node_artifact_types = {n["node_id"]: n["artifact_type"] for n in _VALIDATOR_MANIFEST["nodes"]} # 2026-08-07: same run_graph()-partial-failure salvage as # plan_documents() -- see _write_completed_node_artifacts's own # docstring and docs/todo/work-graph-partial-artifact-persistence- # requirements.md. `outputs` is only used below on the SUCCESS # continuation (script execution + verdict prompt); the failure path # re-raises before ever touching it. run_graph_error = None outputs = None try: outputs = work_graph_orchestrate.run_graph( conn, run_id, "validator", "validator-independent-verification", target_folder_path, _VALIDATOR_MANIFEST, _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, verify_fn=_test_script_verify_fn(node_artifact_types, script_artifact_type="validator_test_script"), timeout=timeout, ) except Exception as e: run_graph_error = e artifact_ids = _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, _VALIDATOR_MANIFEST, node_artifact_types, lambda artifact_type: "txt" if artifact_type != "validator_benchmark" else "md", run_graph_error, ) if run_graph_error is not None: raise run_graph_error if "validator_test_script" in outputs: script_content = outputs["validator_test_script"]["content"] script_execution = run_test_script_for_real( script_content, target_folder_path, task_id, "validator_test_script", timeout=timeout, ) else: script_execution = { "ran": False, "exit_code": None, "stdout": "", "stderr": "", "note": "N/A: validator_test_script node is currently disabled (see _VALIDATOR_MANIFEST)", } response = org_node_trigger.trigger_org_node( "validator", _VALIDATOR_VERDICT_PROMPT.format( goal_text=goal_text, validator_benchmark_content=outputs["validator_benchmark"]["content"], script_execution_result=_format_script_execution_result(script_execution), ), target_folder_path, timeout=timeout, ) verdict = _parse_verdict_line(response) # Hard gate (user-confirmed 2026-08-06): a real non-zero exit from # validator's OWN test script is not up for LLM debate -- force FAIL # regardless of what the verdict line said. A passing/not-run script # leaves the LLM's own judgment (including INCONCLUSIVE) untouched. if script_execution["ran"] and script_execution["exit_code"] not in (0, None): verdict = "FAIL" # 2026-08-07 (real incident): this function INSERTs its work_graph_run # row as status='running' at the top and, on the happy path (verify_fn # returns ok=True, so _verify_and_maybe_regenerate never touches # work_graph_run.status at all -- it only writes 'reflecting'/ # 'verifying'/'blocked' on the RETRY path), never wrote anything past # that -- regardless of PASS/FAIL/INCONCLUSIVE, a validation run that # got this far genuinely finished executing. `ux_work_graph_run_active` # (schema.py) is a unique index on (task_id, org_node) covering exactly # this non-terminal status set, so a real re-verify later (org_level_ # loop.run_fix_loop after a FAIL) collided with this row on INSERT -- # "UNIQUE constraint failed: work_graph_run.task_id, work_graph_run. # org_node" -- found for real via a task reconciliation resume. conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,)) conn.commit() return { "verdict": verdict, "raw_response": response, "artifact_ids": artifact_ids, "script_execution": script_execution, } 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. """ org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT) plan_result = plan_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"]) prd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["prd"]) benchmark_content = _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"]) # 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, 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, 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()