Give Planner/Coder/Validator real, machine-checkable acceptance criteria

Root cause traced (docs/implementation-gap-analysis.md): the Benchmark document Validator relies on for its verdict was frequently a narrative about the work session ("The planner workflow completed and produced a Benchmark defining 18 scenarios...") instead of the actual scenarios -- found via a real skill-detail-page task's benchmark.md, then confirmed present since the earliest real self-hosting runs. Validator's own judgment was, in practice, resting on thin, sometimes-vacuous prose. Planner's work graph (brd->prd->sdd->benchmark) extends to benchmark->test_dataset->test_script: test_dataset/test_script are always present as nodes, but the generating subagent judges per-goal whether a real dataset/script applies (many goals, e.g. UI wiring, legitimately don't have one -- replying 'N/A: <reason>' is a real, expected answer, not a failure). test_script gets a genuinely mechanical verify_fn (real py_compile/node --check, not LLM judgment) -- the one node type where an objective check is actually possible. All four document prompts also gained an explicit anti-narration instruction (only 'brd' had a weaker version of this before). Coder's implementation prompt now receives the real benchmark/dataset/ script and must make the Planner's real script pass for real, not just whatever tests it invents on its own. Validator gets its own 3-node Work Graph (validator_benchmark -> validator_test_dataset -> validator_test_script), generated independently from SDD/PRD -- deliberately NOT reusing Coder's or Planner's own copies, same reasoning as a held-out eval set vs. the dev set a model was tuned against. The resulting script, if real, is actually executed against the current repository state; a non-zero real exit code force-overrides the verdict to FAIL in code (user-confirmed hard gate: the script is not up for LLM debate on a failure, only on a pass can subjective judgment add INCONCLUSIVE/risk notes). schema.py: artifact.artifact_type's CHECK constraint gains the 5 new types, plus migrate_expand_artifact_types() -- SQLite has no ALTER-based constraint migration, so this is the standard rebuild-copy-swap dance, applied for real to the live ~/.graphlemon/graphlemon.db (42 rows, verified byte-identical count before/after, integrity_check clean). skills/qa-task-validation.md and skills/project-sdd-planning.md already gained a standing principle from this same incident: verify cross-task data dependencies against real system state, not seeded fixtures alone (see prior commit d7035fc) -- this commit is the mechanism-level half of the same fix, not just the prose-guidance half. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Give Planner/Coder/Validator real, machine-checkable acceptance criteria
Root cause traced (docs/implementation-gap-analysis.md): the Benchmark document Validator relies on for its verdict was frequently a narrative about the work session ("The planner workflow completed and produced a Benchmark defining 18 scenarios...") instead of the actual scenarios -- found via a real skill-detail-page task's benchmark.md, then confirmed present since the earliest real self-hosting runs. Validator's own judgment was, in practice, resting on thin, sometimes-vacuous prose. Planner's work graph (brd->prd->sdd->benchmark) extends to benchmark->test_dataset->test_script: test_dataset/test_script are always present as nodes, but the generating subagent judges per-goal whether a real dataset/script applies (many goals, e.g. UI wiring, legitimately don't have one -- replying 'N/A: <reason>' is a real, expected answer, not a failure). test_script gets a genuinely mechanical verify_fn (real py_compile/node --check, not LLM judgment) -- the one node type where an objective check is actually possible. All four document prompts also gained an explicit anti-narration instruction (only 'brd' had a weaker version of this before). Coder's implementation prompt now receives the real benchmark/dataset/ script and must make the Planner's real script pass for real, not just whatever tests it invents on its own. Validator gets its own 3-node Work Graph (validator_benchmark -> validator_test_dataset -> validator_test_script), generated independently from SDD/PRD -- deliberately NOT reusing Coder's or Planner's own copies, same reasoning as a held-out eval set vs. the dev set a model was tuned against. The resulting script, if real, is actually executed against the current repository state; a non-zero real exit code force-overrides the verdict to FAIL in code (user-confirmed hard gate: the script is not up for LLM debate on a failure, only on a pass can subjective judgment add INCONCLUSIVE/risk notes). schema.py: artifact.artifact_type's CHECK constraint gains the 5 new types, plus migrate_expand_artifact_types() -- SQLite has no ALTER-based constraint migration, so this is the standard rebuild-copy-swap dance, applied for real to the live ~/.graphlemon/graphlemon.db (42 rows, verified byte-identical count before/after, integrity_check clean). skills/qa-task-validation.md and skills/project-sdd-planning.md already gained a standing principle from this same incident: verify cross-task data dependencies against real system state, not seeded fixtures alone (see prior commit d7035fc) -- this commit is the mechanism-level half of the same fix, not just the prose-guidance half. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee · 宜博
1 parent e3baa491
Showing 8 changed files with 947 additions and 45 deletions
src/e2e_pipeline.py
src/schema.py
src/task_execution.py
src/test_e2e_pipeline.py
src/test_schema.py
src/test_task_execution.py
src/test_work_graph_plan.py
src/work_graph_plan.py
@@ -48,6 +48,7 @@ attempt, then `ConsistencyCheckFailedError` if it&#39;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 = {&quot;type&quot;: &quot;object&quot;, &quot;required&quot;: [&quot;response&quot;]}
 _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/<task_id>/
+    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,
@@ -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()
@@ -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,
@@ -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()
@@ -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):
@@ -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",
             )
@@ -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
@@ -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