diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 0b60067..6322aee 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -48,6 +48,7 @@ attempt, then `ConsistencyCheckFailedError` if it's still contradictory). import json import os import subprocess +import tempfile import uuid from src import ( @@ -76,28 +77,63 @@ _PRD_CONSISTENCY_PROMPT = ( "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 preamble, " - "no meta-commentary about what you're about to do." + "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." + "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." + "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. " - "Reply with the document content (Markdown) via the required schema." + "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 ), } @@ -107,8 +143,21 @@ _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). Reply via the required schema with a `response` field describing " - "what you did and the real test commands/output. Goal: {goal_text}\n\nSDD:\n\n{sdd_content}" + "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 = ( @@ -119,6 +168,77 @@ _VALIDATOR_PROMPT = ( "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_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"], + }, + { + # depends on BOTH -- its own prompt needs + # {validator_benchmark_content} directly, not just what + # survives into validator_test_dataset_content (same reasoning + # as the Planner's own test_script node, see + # work_graph_plan._normalize_graph_manifest). + "node_id": "validator_test_script", "artifact_type": "validator_test_script", + "depends_on": ["validator_test_dataset", "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}" +) + def _planner_prompt_builder(goal_text, node_artifact_types): """node_artifact_types: dict[node_id] -> artifact_type, read off the @@ -137,6 +257,27 @@ def _planner_prompt_builder(goal_text, node_artifact_types): 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 -- @@ -209,6 +350,89 @@ def _prd_consistency_verify_fn(node_artifact_types, target_folder_path): 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 + + +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.""" + def verify_fn(node, output): + if node_artifact_types.get(node["node_id"]) != script_artifact_type: + return True, None + return check_script_compiles(output["content"]) + 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 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 @@ -226,9 +450,16 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): outputs = work_graph_orchestrate.run_graph( conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest, _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, - verify_fn=_prd_consistency_verify_fn(node_artifact_types, target_folder_path), + verify_fn=_combined_planner_verify_fn(node_artifact_types, target_folder_path), ) + # test_dataset/test_script aren't necessarily Markdown (a script is real + # code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately + # generic container; Coder materializes it into a properly-named, + # properly-placed real file under the target project's own test + # conventions when it actually uses it, that's not this write step's job. + _ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"} + artifact_ids = {} for node in work_graph_orchestrate.topological_order(graph_manifest): node_id = node["node_id"] @@ -238,16 +469,21 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): for dep_id in (node.get("depends_on") or []) if node_artifact_types.get(dep_id) in artifact_ids ] + ext = _ARTIFACT_FILE_EXT.get(artifact_type, "md") result = artifact_write.write_artifact( conn, target_folder_path, artifact_type, task_id, run_id, - ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), outputs[node_id]["content"], + ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext), outputs[node_id]["content"], depends_on_artifact_ids=deps, ) artifact_ids[artifact_type] = result["artifact_id"] return {"run_id": run_id, "artifact_ids": artifact_ids} -def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200): +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 @@ -255,6 +491,12 @@ def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, ti 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()) @@ -268,7 +510,10 @@ def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, ti 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), + 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"] @@ -288,6 +533,19 @@ def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, ti 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. @@ -296,22 +554,161 @@ def validate_implementation(task_id, benchmark_content, goal_text, target_folder 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, ) - 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": verdict, "raw_response": response} + 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"]} + 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, + ) + + artifact_ids = {} + for node in work_graph_orchestrate.topological_order(_VALIDATOR_MANIFEST): + node_id = node["node_id"] + artifact_type = node_artifact_types[node_id] + deps = [ + artifact_ids[node_artifact_types[dep_id]] + for dep_id in (node.get("depends_on") or []) + if node_artifact_types.get(dep_id) in artifact_ids + ] + ext = "txt" if artifact_type != "validator_benchmark" else "md" + result = artifact_write.write_artifact( + conn, target_folder_path, artifact_type, task_id, run_id, + ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext), outputs[node_id]["content"], + depends_on_artifact_ids=deps, + ) + artifact_ids[artifact_type] = result["artifact_id"] + + 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, + ) + + 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" + + return { + "verdict": verdict, "raw_response": response, + "artifact_ids": artifact_ids, "script_execution": script_execution, + } def _current_commit(repo_path): @@ -334,13 +731,32 @@ def run_task_lifecycle(conn, task_id, target_folder_path, goal_text): 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) + 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(task_id, benchmark_content, goal_text, target_folder_path) + verdict = validate_implementation_with_own_graph( + conn, task_id, sdd_content, prd_content, goal_text, target_folder_path, + ) return { "task_id": task_id, diff --git a/src/schema.py b/src/schema.py index 1ca4610..556dfdb 100644 --- a/src/schema.py +++ b/src/schema.py @@ -161,7 +161,9 @@ CREATE TABLE work_node_execution ( CREATE TABLE artifact ( artifact_id TEXT PRIMARY KEY, artifact_type TEXT NOT NULL CHECK (artifact_type IN ( - 'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report' + 'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report', + 'test_dataset', 'test_script', + 'validator_benchmark', 'validator_test_dataset', 'validator_test_script' )), task_id TEXT NOT NULL REFERENCES task(task_id), produced_by_run_id TEXT NOT NULL REFERENCES work_graph_run(run_id), @@ -351,3 +353,48 @@ def create_schema(conn): from U-034, PH5) + indexes on an open connection.""" conn.executescript(SCHEMA_SQL) conn.commit() + + +def migrate_expand_artifact_types(conn): + """2026-08-06: adds test_dataset/test_script/validator_benchmark/ + validator_test_dataset/validator_test_script to artifact.artifact_type's + CHECK enum on an ALREADY-POPULATED database -- SQLite has no + `ALTER TABLE ... MODIFY CONSTRAINT`, so this is the standard "create a + new table with the wanted constraint, copy every row, drop the old + table, rename" dance. No other table has a foreign key into `artifact` + (checked), so this doesn't need to touch anything beyond the table + itself and its one index. + + Idempotent and safe to call unconditionally (including against a + freshly create_schema()'d database, which already has the expanded + constraint) -- checks the live constraint text first and no-ops if it's + already expanded, so callers never need to special-case "is this a + fresh or an upgraded database". + """ + row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='artifact'").fetchone() + if row is None or "'test_dataset'" in row[0]: + return + conn.executescript( + """ + CREATE TABLE artifact_new ( + artifact_id TEXT PRIMARY KEY, + artifact_type TEXT NOT NULL CHECK (artifact_type IN ( + 'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report', + 'test_dataset', 'test_script', + 'validator_benchmark', 'validator_test_dataset', 'validator_test_script' + )), + task_id TEXT NOT NULL REFERENCES task(task_id), + produced_by_run_id TEXT NOT NULL REFERENCES work_graph_run(run_id), + file_path TEXT NOT NULL, + git_commit_sha TEXT NOT NULL, + depends_on_artifact_ids TEXT, + validation_status TEXT NOT NULL DEFAULT 'pending' CHECK (validation_status IN ('pending', 'passed', 'failed')), + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ); + INSERT INTO artifact_new SELECT * FROM artifact; + DROP TABLE artifact; + ALTER TABLE artifact_new RENAME TO artifact; + CREATE UNIQUE INDEX ux_artifact_task_type_commit ON artifact(task_id, artifact_type, git_commit_sha); + """ + ) + conn.commit() diff --git a/src/task_execution.py b/src/task_execution.py index d796ef3..f12d6c5 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -74,6 +74,20 @@ def _artifact_content(conn, target_folder_path, task_id, artifact_type): return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0]) +def _artifact_content_or_default(conn, target_folder_path, task_id, artifact_type, default): + """Like _artifact_content, but for the 2026-08-06 test_dataset/ + test_script nodes -- a task resumed from before this feature existed + (or any other reason those two specific artifacts are missing) should + still be resumable, just without them, not hard-fail.""" + row = conn.execute( + "SELECT artifact_id FROM artifact WHERE task_id = ? AND artifact_type = ? ORDER BY created_at DESC LIMIT 1", + (task_id, artifact_type), + ).fetchone() + if row is None: + return default + return e2e_pipeline._read_artifact_content(conn, target_folder_path, row[0]) + + def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): """Same return shape as `e2e_pipeline.run_task_lifecycle` -- continues from wherever `task.status` indicates real work last completed, instead @@ -93,18 +107,33 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text): doc_artifact_ids = _latest_artifact_ids(conn, task_id, _DOC_ARTIFACT_TYPES) sdd_content = _artifact_content(conn, target_folder_path, task_id, "sdd") + prd_content = _artifact_content(conn, target_folder_path, task_id, "prd") benchmark_content = _artifact_content(conn, target_folder_path, task_id, "benchmark") + # 2026-08-06 nodes -- see _artifact_content_or_default's own docstring + # for why these two specifically fall back instead of hard-failing. + test_dataset_content = _artifact_content_or_default( + conn, target_folder_path, task_id, "test_dataset", "N/A: not provided" + ) + test_script_content = _artifact_content_or_default( + conn, target_folder_path, task_id, "test_script", "N/A: not provided" + ) if status == "planning_brd_prd_sdd": _check_not_paused(conn, task_id) org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT) - code_artifact_ids = e2e_pipeline.implement_code(conn, task_id, sdd_content, goal_text, target_folder_path) + code_artifact_ids = e2e_pipeline.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, + ) else: # status == "coding" -- Coder already ran code_artifact_ids = _latest_artifact_ids(conn, task_id, _CODE_ARTIFACT_TYPES) _check_not_paused(conn, task_id) org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, e2e_pipeline._READY_PROMPT) - verdict = e2e_pipeline.validate_implementation(task_id, benchmark_content, goal_text, target_folder_path) + verdict = e2e_pipeline.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": doc_artifact_ids, diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 77195ed..bcee763 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -23,6 +23,7 @@ Run: python3 -m unittest src.test_e2e_pipeline -v """ import json +import os import subprocess import tempfile import unittest @@ -30,7 +31,11 @@ from unittest.mock import patch from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute -_FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"} +_FAKE_DOCS = { + "brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake", + "test_dataset": "N/A: no natural dataset for this fake goal", + "test_script": "assert True # fake but real, compilable script", +} _FAKE_GRAPH_MANIFEST = { "nodes": [ @@ -38,11 +43,15 @@ _FAKE_GRAPH_MANIFEST = { {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]}, {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]}, + {"node_id": "test_dataset", "artifact_type": "test_dataset", "depends_on": ["benchmark"]}, + {"node_id": "test_script", "artifact_type": "test_script", "depends_on": ["test_dataset"]}, ], "edges": [ {"from": "brd", "to": "prd", "edge_type": "needs_judgment"}, {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"}, {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"}, + {"from": "benchmark", "to": "test_dataset", "edge_type": "needs_judgment"}, + {"from": "test_dataset", "to": "test_script", "edge_type": "needs_judgment"}, ], } @@ -71,7 +80,7 @@ class PipelineTestBase(unittest.TestCase): class PlanDocumentsTest(PipelineTestBase): - def test_produces_four_artifacts_with_correct_lineage(self): + def test_produces_six_artifacts_with_correct_lineage(self): self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-1', (SELECT project_id FROM project), '实现贪吃蛇游戏', 'goal', " @@ -86,11 +95,12 @@ class PlanDocumentsTest(PipelineTestBase): {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, + {"content": _FAKE_DOCS["test_dataset"]}, {"content": _FAKE_DOCS["test_script"]}, ] plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name) artifact_ids = plan_result["artifact_ids"] - self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark"}) + self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"}) prd_deps = json.loads( self.conn.execute( "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],) @@ -122,12 +132,21 @@ class RunEndToEndTest(PipelineTestBase): {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, + {"content": _FAKE_DOCS["test_dataset"]}, {"content": _FAKE_DOCS["test_script"]}, {"response": "Implemented. Ran `pytest`: 3 passed."}, + # Validator's OWN independent benchmark/dataset/script (2026-08-06) -- + # deliberately different content from the Planner's, to also prove + # nothing here accidentally reuses the Planner's copies. + {"content": "# Validator's own Benchmark\nfake"}, + {"content": "N/A: validator judged no dataset needed either"}, + {"content": "assert True # validator's own real, compilable script"}, ] result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏") self.assertEqual(result["verdict"]["verdict"], "PASS") - self.assertEqual(set(result["doc_artifact_ids"]), {"brd", "prd", "sdd", "benchmark"}) + self.assertEqual( + set(result["doc_artifact_ids"]), {"brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"} + ) self.assertEqual(set(result["code_artifact_ids"]), {"code_diff", "test_report"}) task_row = self.conn.execute( @@ -138,15 +157,19 @@ class RunEndToEndTest(PipelineTestBase): all_artifacts = self.conn.execute( "SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],) ).fetchone()[0] - self.assertEqual(all_artifacts, 6) # brd, prd, sdd, benchmark, code_diff, test_report + # brd, prd, sdd, benchmark, test_dataset, test_script, code_diff, test_report, + # validator_benchmark, validator_test_dataset, validator_test_script + self.assertEqual(all_artifacts, 11) - # planner and coder each got their own work_graph_run (not sharing one) + # planner, coder, AND validator (2026-08-06: validator now has its own + # real work_graph_run too, not a flat direct call) each got their own + # work_graph_run (not sharing one) run_org_nodes = { row[0] for row in self.conn.execute( "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],) ).fetchall() } - self.assertEqual(run_org_nodes, {"planner", "coder"}) + self.assertEqual(run_org_nodes, {"planner", "coder", "validator"}) class PrdConsistencyVerifyFnTest(unittest.TestCase): @@ -202,6 +225,92 @@ class PrdConsistencyVerifyFnTest(unittest.TestCase): self.assertEqual(args[0], "/tmp/proj-1") +class CheckScriptCompilesTest(unittest.TestCase): + """2026-08-06: the one mechanical (not LLM-judged) verify_fn -- real, + unmocked subprocess calls to py_compile/node --check, no fixtures to + fake since this is testing an actual compiler's real opinion.""" + + def test_na_content_passes_without_compiling(self): + ok, detail = e2e_pipeline.check_script_compiles("N/A: this goal has no automatable check") + self.assertTrue(ok) + self.assertIsNone(detail) + + def test_valid_python_passes(self): + ok, detail = e2e_pipeline.check_script_compiles("def test_thing():\n assert 1 + 1 == 2\n") + self.assertTrue(ok) + + def test_valid_javascript_passes(self): + ok, detail = e2e_pipeline.check_script_compiles("function testThing() { return 1 + 1; }\n") + self.assertTrue(ok) + + def test_narrative_prose_fails_both(self): + # this is the exact failure mode that motivated this check -- a + # narrative sentence about a script, not a real script. + ok, detail = e2e_pipeline.check_script_compiles( + "The graphlemon planner Work Node workflow completed successfully and produced " + "a test script covering 18 scenarios across four areas." + ) + self.assertFalse(ok) + self.assertIn("py_compile", detail) + self.assertIn("node --check", detail) + + def test_syntactically_broken_python_and_js_fails(self): + ok, detail = e2e_pipeline.check_script_compiles("def broken(:\n this is not valid anything\n") + self.assertFalse(ok) + + +class TestScriptVerifyFnTest(unittest.TestCase): + def test_non_test_script_node_passes_through_without_compiling(self): + node_artifact_types = {"brd": "brd"} + verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types) + ok, detail = verify_fn({"node_id": "brd"}, {"content": "not even close to code {{{"}) + self.assertTrue(ok) + self.assertIsNone(detail) + + def test_test_script_node_with_broken_content_fails(self): + node_artifact_types = {"script": "test_script"} + verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types) + ok, detail = verify_fn({"node_id": "script"}, {"content": "this is narrative, not code, no way"}) + self.assertFalse(ok) + + def test_test_script_node_with_real_code_passes(self): + node_artifact_types = {"script": "test_script"} + verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types) + ok, detail = verify_fn({"node_id": "script"}, {"content": "assert True\n"}) + self.assertTrue(ok) + + +class CombinedPlannerVerifyFnTest(unittest.TestCase): + """Dispatches to the right per-node-type check -- both real checks + (PRD consistency and script-compiles) are exercised elsewhere in + isolation; this only tests the dispatch itself.""" + + def test_prd_node_dispatches_to_prd_check(self): + node_artifact_types = {"n1": "prd"} + verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1") + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object( + work_graph_execute, "execute_work_graph", + return_value={"consistent": True, "issues": []}, + ) as mock_execute: + ok, detail = verify_fn({"node_id": "n1"}, {"content": "a prd"}) + self.assertTrue(ok) + mock_execute.assert_called_once() + + def test_test_script_node_dispatches_to_compile_check(self): + node_artifact_types = {"n1": "test_script"} + verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1") + ok, detail = verify_fn({"node_id": "n1"}, {"content": "not code at all, just words"}) + self.assertFalse(ok) + + def test_other_node_types_pass_through(self): + node_artifact_types = {"n1": "test_dataset"} + verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1") + ok, detail = verify_fn({"node_id": "n1"}, {"content": "anything at all"}) + self.assertTrue(ok) + self.assertIsNone(detail) + + class PrdConsistencySelfTargetWorktreeTest(unittest.TestCase): """2026-08-03 regression: a real self-hosting run showed _check_prd_ consistency() calling execute_work_graph() directly against @@ -280,5 +389,131 @@ class ValidateImplementationTest(unittest.TestCase): self.assertEqual(result["verdict"], "INCONCLUSIVE") +class RunTestScriptForRealTest(unittest.TestCase): + """2026-08-06: real, unmocked execution -- these are testing what an + actual interpreter actually does, same reasoning as CheckScriptCompilesTest.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + + def test_na_content_is_not_executed(self): + result = e2e_pipeline.run_test_script_for_real( + "N/A: no automatable check for this goal", self.tmp.name, "task-1", "validator_test_script" + ) + self.assertFalse(result["ran"]) + self.assertIsNone(result["exit_code"]) + + def test_real_passing_python_script(self): + result = e2e_pipeline.run_test_script_for_real( + "import sys\nsys.exit(0)\n", self.tmp.name, "task-1", "validator_test_script" + ) + self.assertTrue(result["ran"]) + self.assertEqual(result["exit_code"], 0) + + def test_real_failing_python_script(self): + result = e2e_pipeline.run_test_script_for_real( + "import sys\nsys.exit(1)\n", self.tmp.name, "task-1", "validator_test_script" + ) + self.assertTrue(result["ran"]) + self.assertEqual(result["exit_code"], 1) + + def test_script_runs_with_real_cwd_so_it_sees_the_real_repo(self): + with open(os.path.join(self.tmp.name, "marker.txt"), "w") as f: + f.write("real file") + script = ( + "import os, sys\n" + "sys.exit(0 if os.path.exists('marker.txt') else 1)\n" + ) + result = e2e_pipeline.run_test_script_for_real(script, self.tmp.name, "task-1", "validator_test_script") + self.assertEqual(result["exit_code"], 0) + + def test_materializes_a_real_file_under_graphlemon_runs(self): + e2e_pipeline.run_test_script_for_real( + "import sys\nsys.exit(0)\n", self.tmp.name, "task-42", "validator_test_script" + ) + expected = os.path.join(self.tmp.name, ".graphlemon", "runs", "task-42", "validator_test_script.py") + self.assertTrue(os.path.exists(expected)) + + +class ValidateImplementationWithOwnGraphTest(unittest.TestCase): + """2026-08-06: the safety-critical piece -- a real non-zero exit from + Validator's own script must force FAIL no matter what the LLM's verdict + line says (user-confirmed hard gate).""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True) + self.conn = schema.connect(":memory:") + schema.create_schema(self.conn) + project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', 'validating', 'validator')" + ) + self.conn.commit() + + def tearDown(self): + self.conn.close() + + def _run(self, script_content, verdict_line): + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + patch.object(org_node_trigger, "trigger_org_node", return_value=verdict_line): + mock_execute.side_effect = [ + {"content": "# Validator's own Benchmark"}, + {"content": "N/A: no dataset needed"}, + {"content": script_content}, + ] + return e2e_pipeline.validate_implementation_with_own_graph( + self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name, + ) + + def test_llm_says_pass_and_script_really_passes_stays_pass(self): + result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good") + self.assertEqual(result["verdict"], "PASS") + self.assertTrue(result["script_execution"]["ran"]) + self.assertEqual(result["script_execution"]["exit_code"], 0) + + def test_llm_says_pass_but_script_really_fails_is_force_overridden_to_fail(self): + # the exact case the user's hard-gate rule exists for: a script + # that objectively fails must not be argued away by the LLM's own + # verdict line. + result = self._run("import sys\nsys.exit(1)\n", "VERDICT: PASS\nlooks good to me") + self.assertEqual(result["verdict"], "FAIL") + + def test_llm_says_fail_and_script_fails_stays_fail(self): + result = self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found") + self.assertEqual(result["verdict"], "FAIL") + + def test_na_script_leaves_llm_judgment_untouched(self): + result = self._run("N/A: no mechanical check possible", "VERDICT: INCONCLUSIVE\nneeds human review") + self.assertEqual(result["verdict"], "INCONCLUSIVE") + self.assertFalse(result["script_execution"]["ran"]) + + def test_writes_three_validator_artifacts_with_lineage(self): + result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good") + self.assertEqual( + set(result["artifact_ids"]), + {"validator_benchmark", "validator_test_dataset", "validator_test_script"}, + ) + script_deps = json.loads( + self.conn.execute( + "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", + (result["artifact_ids"]["validator_test_script"],), + ).fetchone()[0] + ) + self.assertIn(result["artifact_ids"]["validator_benchmark"], script_deps) + self.assertIn(result["artifact_ids"]["validator_test_dataset"], script_deps) + + def test_creates_its_own_work_graph_run_row(self): + result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good") + run_row = self.conn.execute( + "SELECT org_node FROM work_graph_run WHERE task_id = 'task-1'" + ).fetchone() + self.assertEqual(run_row[0], "validator") + + if __name__ == "__main__": unittest.main() diff --git a/src/test_schema.py b/src/test_schema.py index 4d5475c..9f82479 100644 --- a/src/test_schema.py +++ b/src/test_schema.py @@ -206,6 +206,117 @@ class ArtifactUniqueConstraintTest(ConstraintTestBase): with self.assertRaises(sqlite3.IntegrityError): self._insert_artifact("artifact-2") + def test_new_test_artifact_types_accepted(self): + for artifact_type in ( + "test_dataset", "test_script", + "validator_benchmark", "validator_test_dataset", "validator_test_script", + ): + self.conn.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES (?, ?, 'task-1', 'run-1', 'x.md', 'sha-1')", + ("artifact-" + artifact_type, artifact_type), + ) + self.conn.commit() + + +class MigrateExpandArtifactTypesTest(unittest.TestCase): + """2026-08-06: real upgrade path for an already-populated database whose + artifact table still has the pre-expansion CHECK constraint.""" + + def setUp(self): + self.conn = schema.connect(":memory:") + # Simulate a pre-migration database: create the OLD schema by hand + # (not schema.create_schema(), which already has the new constraint). + old_sql = schema.SCHEMA_SQL.replace( + "'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report',\n" + " 'test_dataset', 'test_script',\n" + " 'validator_benchmark', 'validator_test_dataset', 'validator_test_script'", + "'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report'", + ) + self.conn.executescript(old_sql) + self.conn.commit() + self.conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', 'g', 'goal', 'queued', 'planner')" + ) + self.conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES ('run-1', 'task-1', 'planner', '{}', 'planning')" + ) + self.conn.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES ('art-1', 'brd', 'task-1', 'run-1', 'brd.md', 'sha-1')" + ) + self.conn.commit() + + def tearDown(self): + self.conn.close() + + def test_old_check_constraint_rejects_new_type_before_migration(self): + with self.assertRaises(sqlite3.IntegrityError): + self.conn.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES ('art-2', 'test_script', 'task-1', 'run-1', 'x', 'sha-2')" + ) + + def test_migration_preserves_existing_rows_and_accepts_new_types(self): + schema.migrate_expand_artifact_types(self.conn) + + row = self.conn.execute("SELECT artifact_type FROM artifact WHERE artifact_id = 'art-1'").fetchone() + self.assertEqual(row[0], "brd") + + self.conn.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES ('art-2', 'test_script', 'task-1', 'run-1', 'x', 'sha-2')" + ) + self.conn.commit() + count = self.conn.execute("SELECT COUNT(*) FROM artifact").fetchone()[0] + self.assertEqual(count, 2) + + def test_unique_index_still_enforced_after_migration(self): + schema.migrate_expand_artifact_types(self.conn) + self.conn.commit() + with self.assertRaises(sqlite3.IntegrityError): + self.conn.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES ('art-dup', 'brd', 'task-1', 'run-1', 'brd.md', 'sha-1')" + ) + + def test_idempotent_on_a_database_already_migrated(self): + schema.migrate_expand_artifact_types(self.conn) + schema.migrate_expand_artifact_types(self.conn) # must not raise / must not duplicate + count = self.conn.execute("SELECT COUNT(*) FROM artifact").fetchone()[0] + self.assertEqual(count, 1) + + def test_noop_on_a_freshly_created_already_expanded_database(self): + fresh = schema.connect(":memory:") + try: + schema.create_schema(fresh) + schema.migrate_expand_artifact_types(fresh) # must not raise + fresh.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('p', 'P', '/tmp/p', 1, 'active')" + ) + fresh.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('t', 'p', 'g', 'goal', 'queued', 'planner')" + ) + fresh.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES ('r', 't', 'planner', '{}', 'planning')" + ) + fresh.execute( + "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, " + "file_path, git_commit_sha) VALUES ('a', 'test_script', 't', 'r', 'x', 'sha')" + ) + fresh.commit() + finally: + fresh.close() + class ExecutionLogFileUniqueConstraintTest(ConstraintTestBase): def setUp(self): diff --git a/src/test_task_execution.py b/src/test_task_execution.py index ad4610f..1363f1c 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -76,18 +76,27 @@ class ResumeFromCodingTest(TaskExecutionTestBase): def test_skips_to_validator_only(self): self._insert_task("task-1", status="coding", assigned_org_node="coder") + self._write_artifact("task-1", "prd", "# PRD content") self._write_artifact("task-1", "sdd", "# SDD content") self._write_artifact("task-1", "benchmark", "# Benchmark content") self._write_artifact("task-1", "code_diff", "diff content") self._write_artifact("task-1", "test_report", "report content") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger: + # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06) + mock_execute.side_effect = [ + {"content": "# Validator's own Benchmark"}, + {"content": "N/A: no dataset needed"}, + {"content": "import sys\nsys.exit(0)\n"}, + ] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") self.assertEqual(result["verdict"]["verdict"], "PASS") self.assertEqual(result["code_artifact_ids"]["code_diff"], "art-code_diff") - # only the validator's real call happened -- no coder call + # only the validator's final-verdict real call happened -- no coder call mock_trigger.assert_called_once() self.assertEqual(mock_trigger.call_args[0][0], "validator") row = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone() @@ -110,8 +119,15 @@ class ResumeFromPlanningTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"response": "done, tests pass"}), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger: + mock_execute.side_effect = [ + {"response": "done, tests pass"}, # coder's implement node + # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06) + {"content": "# Validator's own Benchmark"}, + {"content": "N/A: no dataset needed"}, + {"content": "import sys\nsys.exit(0)\n"}, + ] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") self.assertEqual(result["verdict"]["verdict"], "PASS") @@ -140,6 +156,7 @@ class PauseMidResumeTest(TaskExecutionTestBase): def test_pause_before_validator_stops_cleanly(self): self._insert_task("task-1", status="coding", assigned_org_node="coder") + self._write_artifact("task-1", "prd", "# PRD content") self._write_artifact("task-1", "sdd", "# SDD content") self._write_artifact("task-1", "benchmark", "# Benchmark content") self._write_artifact("task-1", "code_diff", "diff content") @@ -259,13 +276,21 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): conn.close() def test_real_thread_completes_and_closes_task(self): + self._write_artifact("task-1", "prd", "# PRD") self._write_artifact("task-1", "sdd", "# SDD") self._write_artifact("task-1", "benchmark", "# Benchmark") self._write_artifact("task-1", "code_diff", "diff") self._write_artifact("task-1", "test_report", "report") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"): + mock_execute.side_effect = [ + {"content": "# Validator's own Benchmark"}, + {"content": "N/A: no dataset needed"}, + {"content": "import sys\nsys.exit(0)\n"}, + ] thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", ) @@ -277,13 +302,24 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): self.assertEqual(status, "done") def test_generic_failure_marks_blocked(self): + self._write_artifact("task-1", "prd", "# PRD") self._write_artifact("task-1", "sdd", "# SDD") self._write_artifact("task-1", "benchmark", "# Benchmark") self._write_artifact("task-1", "code_diff", "diff") self._write_artifact("task-1", "test_report", "report") with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", side_effect=RuntimeError("boom")): + # validator's own benchmark/dataset/script graph completes fine -- + # the failure this test cares about is the FINAL verdict call + # (org_node_trigger.trigger_org_node), which happens after. + mock_execute.side_effect = [ + {"content": "# Validator's own Benchmark"}, + {"content": "N/A: no dataset needed"}, + {"content": "import sys\nsys.exit(0)\n"}, + ] thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", ) diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index ec0b7af..5fa7d81 100644 --- a/src/test_work_graph_plan.py +++ b/src/test_work_graph_plan.py @@ -142,6 +142,8 @@ class NormalizeGraphManifestTest(unittest.TestCase): {"node_id": "prd", "artifact_type": "prd"}, {"node_id": "sdd", "artifact_type": "sdd"}, {"node_id": "benchmark", "artifact_type": "benchmark"}, + {"node_id": "test_dataset", "artifact_type": "test_dataset"}, + {"node_id": "test_script", "artifact_type": "test_script"}, ], "edges": [], } @@ -151,6 +153,11 @@ class NormalizeGraphManifestTest(unittest.TestCase): self.assertEqual(by_type["prd"]["depends_on"], ["brd"]) self.assertEqual(by_type["sdd"]["depends_on"], ["prd"]) self.assertEqual(by_type["benchmark"]["depends_on"], ["sdd"]) + self.assertEqual(by_type["test_dataset"]["depends_on"], ["benchmark"]) + # test_script needs both -- its own prompt references + # {benchmark_content} directly, not just what survives into + # test_dataset_content (see work_graph_plan._normalize_graph_manifest). + self.assertEqual(by_type["test_script"]["depends_on"], ["test_dataset", "benchmark"]) def test_canonical_chain_enforcement_uses_real_node_ids_not_artifact_types(self): # node_id and artifact_type can differ (e.g. real LLM output naming @@ -162,6 +169,8 @@ class NormalizeGraphManifestTest(unittest.TestCase): {"node_id": "step-2", "artifact_type": "prd"}, {"node_id": "step-3", "artifact_type": "sdd"}, {"node_id": "step-4", "artifact_type": "benchmark"}, + {"node_id": "step-5", "artifact_type": "test_dataset"}, + {"node_id": "step-6", "artifact_type": "test_script"}, ], "edges": [], } @@ -170,6 +179,8 @@ class NormalizeGraphManifestTest(unittest.TestCase): self.assertEqual(by_type["prd"]["depends_on"], ["step-1"]) self.assertEqual(by_type["sdd"]["depends_on"], ["step-2"]) self.assertEqual(by_type["benchmark"]["depends_on"], ["step-3"]) + self.assertEqual(by_type["test_dataset"]["depends_on"], ["step-4"]) + self.assertEqual(by_type["test_script"]["depends_on"], ["step-5", "step-4"]) def test_partial_canonical_set_does_not_trigger_enforcement(self): # only brd/prd/sdd present, no benchmark -- not a recognized diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index ced125a..a2c5c23 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -41,7 +41,10 @@ GRAPH_MANIFEST_SCHEMA = { "required": ["node_id", "artifact_type", "depends_on"], "properties": { "node_id": {"type": "string"}, - "artifact_type": {"type": "string", "enum": ["brd", "prd", "sdd", "benchmark"]}, + "artifact_type": { + "type": "string", + "enum": ["brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"], + }, "depends_on": {"type": "array", "items": {"type": "string"}}, }, }, @@ -62,13 +65,14 @@ GRAPH_MANIFEST_SCHEMA = { } _PROMPT_TEMPLATE = ( - "Plan (do not execute) the Work Graph needed to produce BRD, PRD, SDD, and " - "Benchmark artifacts for this goal: {goal_text}\n" - "Output a graph_manifest with one node per artifact_type (brd, prd, sdd, " - "benchmark) and edges expressing their dependency chain (sdd depends on prd " - "depends on brd, benchmark depends on sdd). Every edge needs edge_type: " - "'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for anything " - "requiring the agent to actually reason about content.\n" + "Plan (do not execute) the Work Graph needed to produce BRD, PRD, SDD, Benchmark, " + "Test Dataset, and Test Script artifacts for this goal: {goal_text}\n" + "Output a graph_manifest with one node per artifact_type (brd, prd, sdd, benchmark, " + "test_dataset, test_script) and edges expressing their dependency chain (sdd depends " + "on prd depends on brd, benchmark depends on sdd, test_dataset depends on benchmark, " + "test_script depends on test_dataset). Every edge needs edge_type: 'zero_token' for " + "pure aggregation/reduce steps, 'needs_judgment' for anything requiring the agent to " + "actually reason about content.\n" "End your reply with the graph_manifest as a single fenced ```json code block " "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key." ) @@ -102,7 +106,7 @@ def _extract_graph_manifest(text): return None -_CANONICAL_DOC_CHAIN = ("brd", "prd", "sdd", "benchmark") +_CANONICAL_DOC_CHAIN = ("brd", "prd", "sdd", "benchmark", "test_dataset", "test_script") def _normalize_graph_manifest(manifest): @@ -158,6 +162,19 @@ def _normalize_graph_manifest(manifest): if prev_id not in cur_node["depends_on"]: cur_node["depends_on"].append(prev_id) + # 2026-08-06: test_script's own prompt needs {benchmark_content} + # directly (not just what survives into test_dataset_content, which + # can be a terse 'N/A: ...' with none of the original scenario + # detail) -- the plain linear chain above only gives it + # test_dataset as an immediate dependency, so add benchmark as a + # second one. e2e_pipeline._planner_prompt_builder already threads + # every *immediate* dependency's content through generically, this + # is the one canonical node that genuinely needs two. + benchmark_id = by_artifact_type["benchmark"]["node_id"] + script_node = by_artifact_type["test_script"] + if benchmark_id not in script_node["depends_on"]: + script_node["depends_on"].append(benchmark_id) + return manifest