宜博 / lemongraph

"""U-014: end-to-end orchestration for architecture doc §11 MVP acceptance
criteria 1-3 (Planner produces validated BRD/PRD/SDD/Benchmark; Coder
implements against the SDD and self-tests pass; Validator runs
benchmark/tests and produces a root-cause-attributed verdict). Criteria 4-7
(Org-level Loop, breakpoint recovery, ...) are explicitly out of this Unit's
scope -- PH2 (U-017/U-021).

**Rewired 2026-08-01 (U-031 regression audit)**: `plan_documents` used to
call `org_node_trigger.trigger_org_node()` directly, bypassing
`work_graph_plan.plan_work_graph()`/`work_graph_execute.execute_work_graph()`
/`work_node_run.run_work_node()` entirely -- those three modules were each
fully unit-tested (and even real-API-validated) in isolation since PH1/PH2,
but nothing in production ever called any of them; this was the one real,
validated closed loop, and it used a completely separate, simpler mechanism.
REQ-02's own coverage line names `c_work_graph_plan`/`c_work_graph_execute`
explicitly, so that gap is now closed: `plan_documents` calls
`plan_work_graph()` for a real graph_manifest, then
`work_graph_orchestrate.run_graph()` executes each node for real (see that
module's docstring for why each node gets its own one-node dynamic-workflow
script rather than one big multi-node script -- REQ-06 breakpoint recovery
needs it). `implement_code` (Coder) goes through the same per-node
mechanism with a synthetic single-node manifest, since REQ-03's own
coverage only names `c_work_node_run`, not graph planning -- there is no
real "plan a Coder module graph" LLM capability built (a much bigger,
separate undertaking; planning docs and planning code modules are
different problems), so a one-node graph is the honest minimum that makes
`c_work_node_run` genuinely load-bearing for Coder too.
`validate_implementation` (Validator) is unchanged -- REQ-04's own
coverage never named the Work Graph controllers at all.

Status-transition bookkeeping (task.status, precondition guards) stays
U-010's job (org_node_lifecycle.trigger_org_node_full) -- called once per
phase with a lightweight "confirm ready" prompt.

**PRD consistency check (2026-08-02, PH4-closure audit)**: REQ-02's
boundary case (contradictory PRD scenario assumptions must be intercepted,
not silently pass validation) had no implementation anywhere -- found while
auditing REQ-01~14 for real coverage before starting PH5.
`_check_prd_consistency` is one extra real `agent()` call (reusing
`agentType='planner'` + a structured-output schema, same mechanism as every
other node) that reviews the PRD's own content for genuine internal
contradiction (not incompleteness/style) right after it's produced, wired
in as `plan_documents`'s `verify_fn` -- see `work_graph_orchestrate.
run_graph`'s docstring for what happens on a failed check (one regeneration
attempt, then `ConsistencyCheckFailedError` if it's still contradictory).
"""

import json
import os
import subprocess
import tempfile
import uuid

from src import (
    artifact_write, org_node_lifecycle, org_node_trigger, paths, task_create,
    work_graph_orchestrate, work_graph_plan, work_node_run,
)

_READY_PROMPT = "Confirm you're ready to begin. Take no other action yet."

_DOC_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
_DOC_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]}

_PRD_CONSISTENCY_SCHEMA = {
    "type": "object",
    "required": ["consistent", "issues"],
    "properties": {
        "consistent": {"type": "boolean"},
        "issues": {"type": "array", "items": {"type": "string"}},
    },
}
_PRD_CONSISTENCY_PROMPT = (
    "Review the following PRD for internally CONTRADICTORY scenario assumptions -- e.g. two "
    "requirements that cannot both hold at once, or the same scenario described two incompatible "
    "ways. Do NOT flag mere incompleteness, ambiguity, or style issues, only genuine "
    "self-contradiction. Reply via the required schema: consistent=false only if you found a real "
    "contradiction, and list each one in issues.\n\nPRD:\n\n{prd_content}"
)

_NO_NARRATION = (
    "The `content` field must contain ONLY the raw document/code text itself -- nothing "
    "else. Do not include any sentence describing what you did, what workflow ran, how "
    "the process went, or that you completed a task. A machine parses `content` directly "
    "into a file with no human review; any narrative text becomes permanently embedded in "
    "a document that's supposed to look like a real, standalone deliverable. Start "
    "directly with the content itself (e.g. a Markdown document starts with its own `#` "
    "heading, not a sentence about producing it)."
)

_DOC_PROMPTS = {
    "brd": (
        "Write a BRD (Business Requirement Document) for this goal, following your own "
        "SOUL.md methodology: {goal_text}\n"
        "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION
    ),
    "prd": (
        "Here is the approved BRD for goal '{goal_text}':\n\n{brd_content}\n\n"
        "Write the PRD (Product Requirement Document) based on this BRD. "
        "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION
    ),
    "sdd": (
        "Here is the approved PRD for goal '{goal_text}':\n\n{prd_content}\n\n"
        "Write the SDD (System Design Document) based on this PRD, including a Development "
        "Unit breakdown. Reply with the document content (Markdown) via the required schema. "
        + _NO_NARRATION
    ),
    "benchmark": (
        "Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n"
        "Write the Benchmark/acceptance-test document based on this SDD: concrete test "
        "scenarios (happy/boundary/failure) with observable pass criteria for each one -- not "
        "just a one-line label per scenario, spell out the actual expected input/behavior/"
        "output for each. Reply with the document content (Markdown) via the required schema. "
        + _NO_NARRATION
    ),
    "test_dataset": (
        "Here is the approved Benchmark for goal '{goal_text}':\n\n{benchmark_content}\n\n"
        "Judge whether this goal has a natural, concrete test dataset (specific inputs with "
        "known-correct outputs, fixture data, or QA pairs the Benchmark's scenarios can be "
        "instantiated against) -- many goals (UI wiring, config/plumbing changes) don't have "
        "one, and that's fine. If it does not apply, reply via the required schema with "
        "content starting exactly with 'N/A: ' followed by a one-sentence reason. If it does "
        "apply, reply with the real dataset itself (e.g. JSON/CSV/a table of concrete cases) "
        "via the required schema. " + _NO_NARRATION
    ),
    "test_script": (
        "Here is the approved Benchmark for goal '{goal_text}':\n\n{benchmark_content}\n\n"
        "Here is the test dataset (may be 'N/A' if this goal has none):\n\n{test_dataset_content}\n\n"
        "Write a real, runnable test script that mechanically checks the Benchmark's "
        "scenarios (using the dataset above if it applies) -- a script a machine can execute "
        "and get a real pass/fail exit code from, not prose describing what should be tested. "
        "Use whatever language/framework fits this specific project's own conventions (don't "
        "invent a new one). If genuinely no mechanical script can meaningfully check this goal "
        "(e.g. a pure judgment call with no automatable check), reply via the required schema "
        "with content starting exactly with 'N/A: ' followed by a one-sentence reason -- this "
        "should be rare, prefer writing a real script whenever a real check is possible. "
        "Reply with the real script content via the required schema. " + _NO_NARRATION
    ),
}

_CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
_CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]}

# 2026-08-12 (Phase 2 §6): retired as the per-node prompt now that Coder
# self-plans (implement_code() no longer runs one synthetic "implement"
# node) -- its guidance is folded into _CODER_PLANNING_CONTEXT below,
# which feeds plan_work_graph()'s planning call instead of a single
# execution call. Kept here, unused, only because deleting it isn't this
# change's job and a future revert would want it back verbatim.
_CODER_PROMPT = (
    "Implement this goal per the approved SDD below. Write real code and real tests, run the "
    "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts "
    "and can explain why not).\n\n"
    "The Planner also produced a Benchmark, a test dataset, and a test script for this goal "
    "(test dataset/script may be 'N/A' if this goal has none -- that's expected, not an error). "
    "If the test script is real (not N/A): materialize it as a real file under this project's "
    "own test conventions (match how existing tests in this repo are named/located), and you "
    "MUST make it pass for real before you're done -- it is not optional extra credit, it's "
    "part of this goal's acceptance bar. Write your own additional tests too if the Planner's "
    "script doesn't cover everything the SDD requires; the Planner's script is a floor, not a "
    "ceiling.\n\n"
    "Reply via the required schema with a `response` field describing what you did and the "
    "real test commands/output (including the Planner's own test script's real output). "
    "Goal: {goal_text}\n\nSDD:\n\n{sdd_content}\n\n"
    "Benchmark:\n\n{benchmark_content}\n\n"
    "Test dataset:\n\n{test_dataset_content}\n\n"
    "Test script:\n\n{test_script_content}"
)

# 2026-08-12 (Phase 2 §6): fed as `goal_text` into plan_work_graph(conn,
# "coder", ..., append_to_run_id=run_id) -- reuses the existing
# {goal_text} substitution point in _generic_prompt_template() rather than
# adding a new parameter to plan_work_graph() just for Coder's extra
# context (SDD/Benchmark/test dataset/script), which Planner's own
# _PROMPT_TEMPLATE call never needed. Same .replace("{goal_text}", ...)
# convention _PROMPT_TEMPLATE/_generic_prompt_template() already use (not
# .format() -- see work_graph_plan.py's own comment on why), so this
# string's remaining braces (none currently, but keep this in mind if
# edited) must stay literal, not format fields.
_CODER_PLANNING_CONTEXT = (
    "{goal_text}\n\n"
    "Approved SDD:\n\n{sdd_content}\n\n"
    "Benchmark (test dataset/script may be 'N/A' if this goal has none -- that's expected, not "
    "an error):\n\n{benchmark_content}\n\n"
    "Test dataset:\n\n{test_dataset_content}\n\n"
    "Test script:\n\n{test_script_content}\n\n"
    "Guidance: write real code and real tests, run the tests yourself as part of your own "
    "nodes' work, and don't consider the goal done until they pass for real (or you've "
    "exhausted reasonable attempts and a node explains why not). If the Planner's test script "
    "above is real (not N/A), materialize it as a real file under this project's own test "
    "conventions and make it pass for real -- it is not optional, it's part of this goal's "
    "acceptance bar; the Planner's script is a floor, write additional tests of your own if it "
    "doesn't cover everything the SDD requires."
)

_VALIDATOR_PROMPT = (
    "Independently verify the implementation for this goal against the Benchmark below: run "
    "the tests yourself, don't just trust the Coder's claims. Reply with a verdict line "
    "'VERDICT: PASS' or 'VERDICT: FAIL' or 'VERDICT: INCONCLUSIVE' followed by your reasoning "
    "and, if FAIL, a root_cause_category (planning|coder|skill_defect) with detail. "
    "Goal: {goal_text}\n\nBenchmark:\n\n{benchmark_content}"
)

# 2026-08-12 (Phase 2c): retired now that Validator self-plans its own
# verification graph (validate_implementation_with_own_graph() below) via
# plan_work_graph(conn, "validator", ..., append_to_run_id=run_id) instead
# of running this hardcoded fixed chain -- the "held-out eval set, not the
# Coder's own dev set" framing that motivated this dict lives on in
# _VALIDATOR_PLANNING_CONTEXT's guidance text instead. Kept here, unused,
# for the same "a future revert would want it back verbatim" reason
# _CODER_PROMPT above is.
#
# 2026-08-06: Validator gets its OWN benchmark/test_dataset/test_script,
# generated independently from SDD/PRD -- NOT reusing the Planner's/Coder's
# own copies. If the Coder could game the Planner's test_script (e.g.
# hard-code outputs for the specific dataset it developed against without
# solving the general problem), re-running that SAME script proves nothing
# -- this is the "held-out eval set vs. the dev/train set" distinction, not
# a redundant re-check of the same artifact.
_VALIDATOR_DOC_PROMPTS = {
    "validator_benchmark": (
        "You are independently defining YOUR OWN acceptance criteria for this goal, informed "
        "by the SDD/PRD below -- do NOT copy or assume the Planner's own Benchmark, form your "
        "own independent judgment of what 'done' means, from a different angle if you can. "
        "Concrete test scenarios (happy/boundary/failure) with observable pass criteria for "
        "each one -- spell out actual expected input/behavior/output, not one-line labels. "
        "Reply with the document content (Markdown) via the required schema. " + _NO_NARRATION
        + "\n\nGoal: {goal_text}\n\nSDD:\n\n{sdd_content}\n\nPRD:\n\n{prd_content}"
    ),
    "validator_test_dataset": (
        "Here is your own independent Benchmark for goal '{goal_text}':"
        "\n\n{validator_benchmark_content}\n\n"
        "Judge whether this goal has a natural, concrete test dataset. If not, reply via the "
        "required schema with content starting exactly with 'N/A: ' plus a one-sentence reason. "
        "If so, produce your OWN independent dataset -- don't assume what the Coder actually "
        "used, you have no visibility into that and shouldn't need it. Reply via the required "
        "schema. " + _NO_NARRATION
    ),
    "validator_test_script": (
        "Here is your own independent Benchmark:\n\n{validator_benchmark_content}\n\n"
        "Here is your own test dataset (may be 'N/A'):\n\n{validator_test_dataset_content}\n\n"
        "Write a real, runnable test script that mechanically checks your own Benchmark's "
        "scenarios against the ACTUAL current implementation in this repository's working "
        "directory -- run real commands, inspect real files/behavior, don't assume anything "
        "about what was implemented. This script will be executed for real, and if it exits "
        "non-zero the verdict is FAIL regardless of any other judgment -- make sure it actually "
        "tests the real thing, not a strawman. If genuinely no mechanical script is possible for "
        "this goal, reply via the required schema with content starting exactly with 'N/A: ' "
        "plus a one-sentence reason -- this should be rare. " + _NO_NARRATION
    ),
}

# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_DOC_PROMPTS above --
# see that comment. validator_test_script node temporarily disabled
# (2026-08-07, same stopgap as work_graph_plan._PROMPT_TEMPLATE -- see its
# comment for the full rationale/upgrade path): 6/6 real repro rate for
# narrative-instead-of-code output, docs/todo/validator-test-script-
# narrative-output-issue.md.
_VALIDATOR_MANIFEST = {
    "nodes": [
        {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []},
        {
            "node_id": "validator_test_dataset", "artifact_type": "validator_test_dataset",
            "depends_on": ["validator_benchmark"],
        },
    ],
    "edges": [],
}

# 2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above -- see
# that comment.
_VALIDATOR_VERDICT_PROMPT = (
    "Give your final verdict for this goal, informed by your own independent Benchmark below "
    "and the REAL execution result of your own test script (if you wrote one -- 'N/A' means you "
    "judged no mechanical script was possible, form your judgment from the Benchmark alone in "
    "that case). Reply with a verdict line 'VERDICT: PASS' or 'VERDICT: FAIL' or "
    "'VERDICT: INCONCLUSIVE' followed by your reasoning and, if FAIL, a root_cause_category "
    "(planning|coder|skill_defect) with detail.\n\n"
    "Goal: {goal_text}\n\nYour own Benchmark:\n\n{validator_benchmark_content}\n\n"
    "Your own test script's REAL execution result:\n\n{script_execution_result}"
)


class _MissingContentDefault(dict):
    """2026-08-08 (found via a real self-planned run, task involving a
    src/work_node_run.py bug fix): `_DOC_PROMPTS` entries were all written
    for the old fixed brd->prd->sdd->benchmark->... chain and hardcode
    placeholders like {prd_content} assuming that upstream node always
    exists. Work Graph self-planning (FR-B4 "全部自由规划") legitimately
    lets the Planner skip nodes that don't apply -- this real run's SDD
    node correctly had no PRD node at all (a small bug-fix goal doesn't
    need one), so `_DOC_PROMPTS["sdd"]`'s `{prd_content}` placeholder had
    nothing to fill it and `.format()` raised a bare KeyError, landing the
    task on `blocked_pending_human` with the unhelpful generic
    'execution_error' reason. `str.format_map` with this as the mapping
    substitutes a clear placeholder for any content key the self-planned
    graph didn't actually produce, instead of crashing."""
    def __missing__(self, key):
        return "N/A (no corresponding upstream node was included in this Work Graph)"


def _planner_prompt_builder(goal_text, node_artifact_types):
    """node_artifact_types: dict[node_id] -> artifact_type, read off the
    real graph_manifest before running it (build_prompt only ever sees one
    node/outputs-so-far pair at a time, not the whole manifest)."""
    def build_prompt(node, outputs):
        node_id = node["node_id"]
        artifact_type = node_artifact_types.get(node_id, node_id)
        content_kwargs = {"goal_text": goal_text}
        for dep_id in (node.get("depends_on") or []):
            dep_output = outputs.get(dep_id)
            if dep_output is not None:
                dep_type = node_artifact_types.get(dep_id, dep_id)
                content_kwargs[dep_type + "_content"] = dep_output["content"]
        return _DOC_PROMPTS[artifact_type].format_map(_MissingContentDefault(content_kwargs))
    return build_prompt


def _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types):
    """2026-08-12 (Phase 2c): retired alongside _VALIDATOR_MANIFEST above --
    see that comment; validate_implementation_with_own_graph() now uses
    dispatch_build_prompt(_planner_prompt_builder(...), generic_node_prompt_builder)
    like every other self-planning org_node.

    Same shape as _planner_prompt_builder, but the FIRST node
    (validator_benchmark) needs sdd_content/prd_content from OUTSIDE
    validator's own tiny graph -- those aren't a dependency within this
    graph, they come from the Planner's already-completed one. Passed
    unconditionally into every node's kwargs; .format() only consumes the
    placeholders each template actually references, extra kwargs are
    harmless."""
    def build_prompt(node, outputs):
        node_id = node["node_id"]
        artifact_type = node_artifact_types.get(node_id, node_id)
        content_kwargs = {"sdd_content": sdd_content, "prd_content": prd_content}
        for dep_id in (node.get("depends_on") or []):
            dep_output = outputs.get(dep_id)
            if dep_output is not None:
                dep_type = node_artifact_types.get(dep_id, dep_id)
                content_kwargs[dep_type + "_content"] = dep_output["content"]
        return _VALIDATOR_DOC_PROMPTS[artifact_type].format(goal_text=goal_text, **content_kwargs)
    return build_prompt


def _coerce_bool(value):
    """A real call (2026-08-02) returned `"consistent": "false"` as a JSON
    STRING, not a boolean, even with `"type": "boolean"` in the schema --
    `if "false":` is truthy in Python, so this silently inverted the check
    until caught by a real test (mocks always return real Python bools, so
    this never showed up until RUN_REAL_AGENT_E2E=1). Don't trust the
    schema's declared type any more than 05 §5.4.1.1 already tells us to
    trust --json-schema itself."""
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() == "true"
    return bool(value)


def _check_prd_consistency(target_folder_path, prd_content, timeout=180):
    """REQ-02 boundary: one real agent() call reviewing the PRD's own
    content for genuine internal contradiction. Returns (ok, issues).

    **Self-target worktree isolation (2026-08-03, real self-hosting run)**:
    this call uses `agentType='planner'` with `memory: project`, which
    writes a real, uncommitted change to `.claude/agent-memory/planner/
    MEMORY.md` in whatever `cwd` it runs with. Every prior real run of this
    check only ever targeted a throwaway target-project clone, so calling
    it directly against `target_folder_path` never caused a problem --
    until a genuine self-hosting run hit it: the memory write landed as an
    uncommitted local change on the *main* checkout (bypassing
    `work_node_run.run_work_node`'s worktree isolation entirely, since this
    call never goes through it), which then made the *next* node's own
    worktree merge fail with "local changes would be overwritten by
    merge". 05 §5.8's unconditional self-target worktree rule has to apply
    here too, not just to `run_work_node`'s own callers."""
    is_self = paths.is_self_target(target_folder_path)
    node_exec_id = "prd-consistency-" + uuid.uuid4().hex[:12]
    worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None
    cwd = worktree_path or target_folder_path
    try:
        prompt = _PRD_CONSISTENCY_PROMPT.format(prd_content=prd_content)
        result = org_node_trigger.trigger_org_node_structured(
            "planner", prompt, cwd, _PRD_CONSISTENCY_SCHEMA, timeout=timeout
        )
        if worktree_path:
            work_node_run.ensure_worktree_committed(worktree_path, node_exec_id)
            # also catch drift on the MAIN checkout (worktrees share project
            # identity, so a real `memory: project` write can land there
            # regardless of `cwd` -- see work_node_run.ensure_worktree_
            # committed's own docstring) before the merge below.
            work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift")
            work_node_run.merge_worktree(target_folder_path, node_exec_id)
        return _coerce_bool(result["consistent"]), result.get("issues", [])
    finally:
        if worktree_path:
            work_node_run.remove_worktree(target_folder_path, worktree_path)


def _prd_consistency_verify_fn(node_artifact_types, target_folder_path):
    """work_graph_orchestrate.run_graph's verify_fn -- only the PRD node
    gets checked, every other node passes through untouched."""
    def verify_fn(node, output):
        if node_artifact_types.get(node["node_id"]) != "prd":
            return True, None
        return _check_prd_consistency(target_folder_path, output["content"])
    return verify_fn


def _is_na_content(content):
    """test_dataset/test_script are allowed to legitimately not apply to a
    given goal -- '_DOC_PROMPTS' asks for content starting with 'N/A: ' in
    that case. That's a pass, not something to compile-check."""
    return content.strip().upper().startswith("N/A")


def _detect_script_language(content):
    """Tries Python first, then Node/JS (the two languages this project's
    own tests are ever written in -- see CLAUDE.md's Testing constraints).
    Returns (language_or_None, detail) -- language is 'python'/'javascript'
    on success, None if neither compiles (detail then holds both compilers'
    real error output). Shared by check_script_compiles (the verify_fn,
    which only needs a yes/no) and run_test_script_for_real (which also
    needs to know HOW to actually invoke the script)."""
    with tempfile.TemporaryDirectory(prefix="graphlemon-script-check-") as tmpdir:
        py_path = os.path.join(tmpdir, "candidate.py")
        with open(py_path, "w", encoding="utf-8") as f:
            f.write(content)
        py_result = subprocess.run(
            ["python3", "-m", "py_compile", py_path], capture_output=True, text=True
        )
        if py_result.returncode == 0:
            return "python", None

        js_path = os.path.join(tmpdir, "candidate.js")
        with open(js_path, "w", encoding="utf-8") as f:
            f.write(content)
        js_result = subprocess.run(["node", "--check", js_path], capture_output=True, text=True)
        if js_result.returncode == 0:
            return "javascript", None

    return None, (
        "content does not compile as Python or Node/JS -- must be a real, runnable script, "
        "not a narrative description.\npython3 -m py_compile:\n{}\nnode --check:\n{}".format(
            py_result.stderr, js_result.stderr
        )
    )


def check_script_compiles(content):
    """Mechanical (not LLM-judged) check that `content` is real, syntactically
    valid code, not narrative prose masquerading as a script -- this is the
    one node type where a genuinely objective check is possible (2026-08-06
    design: see docs/implementation-gap-analysis.md's Benchmark-narration
    incident). Returns (ok, detail)."""
    if _is_na_content(content):
        return True, None
    language, detail = _detect_script_language(content)
    return language is not None, detail


_NARRATION_RETRY_HINT = (
    "If your previous output was a narrative description of an action (e.g. describing a task/"
    "workflow you ran, launched, or will report on) instead of literal script source code, "
    "discard it entirely -- do not describe what you did, are doing, or launched. The `content` "
    "field must contain nothing but the script's real source code, starting from its very first "
    "real line (e.g. an import/shebang), with no leading or trailing prose."
)


def _test_script_verify_fn(node_artifact_types, script_artifact_type="test_script"):
    """work_graph_orchestrate.run_graph's verify_fn for a test_script node
    -- mechanical compile check, not LLM judgment. On failure the same
    reflecting/retry-once mechanism the PRD-consistency check already uses
    kicks in (see run_graph's own docstring). `script_artifact_type` is
    parameterized so this same function backs both the Planner's
    'test_script' node and the Validator's own 'validator_test_script' node.

    2026-08-07 (docs/todo/validator-test-script-narrative-output-issue.md):
    `check_script_compiles`'s own detail string already says "not a
    narrative description", but that's a generic explanation of what the
    mechanical check looks for, buried ahead of two raw compiler
    tracebacks -- real observed reruns still reproduced near-identical
    narrative content on the very next reflect attempt. `_NARRATION_RETRY_HINT`
    adds an explicit, directive corrective as its own separate feedback
    item (phrased conditionally, so it's a harmless no-op for genuine
    syntax-typo failures that were never narration in the first place)."""
    def verify_fn(node, output):
        if node_artifact_types.get(node["node_id"]) != script_artifact_type:
            return True, None
        ok, detail = check_script_compiles(output["content"])
        # work_graph_orchestrate._verify_and_maybe_regenerate does
        # "; ".join(feedback or []), i.e. it expects a list of messages (the
        # PRD-consistency verify_fn's `issues` shape) -- check_script_compiles
        # returns a single string, which `join` would otherwise iterate
        # character-by-character into a garbled reflect-prompt.
        return ok, ([detail, _NARRATION_RETRY_HINT] if detail else None)
    return verify_fn


def _combined_planner_verify_fn(node_artifact_types, target_folder_path):
    """Dispatches to whichever of the per-node-type verify_fns actually
    applies -- run_graph only accepts one verify_fn, and each node type's
    check is independent (a node is at most one artifact_type)."""
    prd_check = _prd_consistency_verify_fn(node_artifact_types, target_folder_path)
    script_check = _test_script_verify_fn(node_artifact_types)

    def verify_fn(node, output):
        artifact_type = node_artifact_types.get(node["node_id"])
        if artifact_type == "prd":
            return prd_check(node, output)
        if artifact_type == "test_script":
            return script_check(node, output)
        return True, None
    return verify_fn


def _failed_node_ids(exc):
    """2026-08-07: which node_id(s) a run_graph() failure means were
    NEVER verify_fn-accepted, even if their work_node_execution row is
    'done' (schema-valid only -- see ConsistencyCheckFailedError's own
    docstring). Handles both the common case today (a bare
    ConsistencyCheckFailedError -- Planner's/Validator's doc chains are
    currently serial, one node per layer) and WorkGraphLayerError (G's
    multi-node-layer aggregate, for whenever a fan-out graph hits this)."""
    if isinstance(exc, work_graph_orchestrate.WorkGraphLayerError):
        return {node_id for node_id, _ in exc.errors}
    if isinstance(exc, work_graph_orchestrate.ConsistencyCheckFailedError) and exc.node_id:
        return {exc.node_id}
    return set()


def _resolve_node_artifact_type(node, node_artifact_types):
    """A node promotes to an `artifact` row if EITHER: it's in the caller's
    precomputed `node_artifact_types` dict (Planner's/Validator's fixed
    catalogs, where every node is a promotable artifact by construction --
    unchanged behavior), OR it declares `artifact_type` directly on
    itself, OR (2026-08-12, Phase 2 §2.3) it declares `promotes_to_
    artifact: {"artifact_type": ...}` (Coder's/Validator's generic
    self-planned nodes -- most of which should declare NEITHER, meaning
    "this is an internal graph step, not a final deliverable"). Returns
    None for a node that isn't promoting anything -- not an error, the
    expected case for most self-planned nodes."""
    if node["node_id"] in node_artifact_types:
        return node_artifact_types[node["node_id"]]
    if "artifact_type" in node:
        return node["artifact_type"]
    promotes = node.get("promotes_to_artifact")
    if promotes:
        return promotes.get("artifact_type")
    return None


def _write_completed_node_artifacts(
    conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for,
    run_graph_error=None, only_node_ids=None,
):
    """Writes an artifact for every node that's genuinely done AND
    verify_fn-accepted, sourced from `work_node_execution` in the DB --
    not a run_graph() return value, so this works identically whether
    run_graph() returned normally (every node accounted for) or raised
    partway through (only some nodes are).

    2026-08-07 (real incident, docs/todo/work-graph-partial-artifact-
    persistence-requirements.md): run_graph() raising discards its
    would-be return value entirely (a Python exception drops the local
    scope), so previously ANY node failing -- even the very last one in
    an otherwise-fully-successful chain -- meant NOTHING got persisted,
    silently throwing away real, already-generated, already-good content.
    A resumed task then found no BRD/PRD/SDD artifacts to build on at all,
    even though they'd been generated correctly the first time.

    A node whose OWN "-reflect" retry exists (verify_fn rejected the
    first attempt) uses the reflect attempt's content, not the original's
    -- the reflect node_id isn't part of `graph_manifest` so the main loop
    below never visits it directly. `_failed_node_ids(run_graph_error)`
    nodes are excluded outright: their row is schema-valid ('done') but
    verify_fn never accepted either attempt, so neither is trustworthy
    content to persist.

    2026-08-12 (Phase 2 §2.3/§6): a node with no resolvable artifact_type
    (see `_resolve_node_artifact_type`) is silently skipped, not an
    error -- most Coder/Validator self-planned nodes are internal graph
    steps, not final deliverables. `only_node_ids` (default None = every
    node in `graph_manifest`, unchanged behavior) restricts processing to
    a subset -- needed once a caller works over a MERGED graph (Coder's
    `append_to_run_id`) and must avoid re-processing/re-promoting nodes an
    EARLIER stage already wrote artifacts for.
    """
    done_rows = conn.execute(
        "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'",
        (run_id,),
    ).fetchall()
    done_outputs = {node_id: json.loads(output_payload) for node_id, output_payload in done_rows}
    for node_id in list(done_outputs):
        reflect_id = node_id + "-reflect"
        if reflect_id in done_outputs:
            done_outputs[node_id] = done_outputs[reflect_id]
    for excluded in _failed_node_ids(run_graph_error):
        done_outputs.pop(excluded, None)

    artifact_ids = {}
    for node in work_graph_orchestrate.topological_order(graph_manifest):
        node_id = node["node_id"]
        if node_id not in done_outputs:
            continue
        if only_node_ids is not None and node_id not in only_node_ids:
            continue
        artifact_type = _resolve_node_artifact_type(node, node_artifact_types)
        if artifact_type is None:
            continue
        deps = [
            artifact_ids[node_artifact_types[dep_id]]
            for dep_id in (node.get("depends_on") or [])
            if node_artifact_types.get(dep_id) in artifact_ids
        ]
        result = artifact_write.write_artifact(
            conn, target_folder_path, artifact_type, task_id, run_id,
            ".graphlemon/runs/{}/{}.{}".format(task_id, artifact_type, ext_for(artifact_type)),
            done_outputs[node_id]["content"], depends_on_artifact_ids=deps,
        )
        artifact_ids[artifact_type] = result["artifact_id"]
    return artifact_ids


# test_dataset/test_script aren't necessarily Markdown (a script is real
# code, a dataset could be JSON/CSV/anything) -- .txt is a deliberately
# generic container; Coder materializes it into a properly-named,
# properly-placed real file under the target project's own test
# conventions when it actually uses it, that's not this write step's job.
_PLANNER_ARTIFACT_FILE_EXT = {"test_dataset": "txt", "test_script": "txt"}


def _planner_ext_for(artifact_type):
    return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md")


def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
    """Architecture §11 criterion 1: a real Planner Work Graph
    (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
    work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark,
    each written via c_artifact_write with correct lineage.
    Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}.

    2026-08-07: artifacts are now written from whatever's genuinely done in
    the DB (see _write_completed_node_artifacts), not from run_graph()'s
    in-memory return value -- so a failure partway through (e.g. the last
    node in the chain) still persists everything that succeeded before it,
    instead of discarding it all. The original run_graph() failure is
    re-raised unchanged after salvaging -- this does not change whether or
    when a task ends up blocked_pending_human, only what's left behind for
    a human's Resume to build on when it does.

    2026-08-08 (Work Graph self-planning + loop-back, FR-B1): `verify_fn`
    is no longer `_combined_planner_verify_fn` -- each self-planned node
    now carries its own `judgment` field (work_graph_plan.py's
    `_PLANNER_NODE_CATALOG` suggests one per node type, the Planner can
    reuse or override it), which `run_graph()` resolves itself per node.
    `_combined_planner_verify_fn`/`_prd_consistency_verify_fn` are NOT
    deleted -- this function still uses them below (FR-A8).
    `_test_script_verify_fn` is ALSO still used below for Planner's own
    `test_script` node (2026-08-12: Validator's own equivalent node no
    longer needs it -- validate_implementation_with_own_graph() now
    self-plans too, and its `final_verdict` node's hard gate is enforced
    via a mechanical_check_registry closure instead, see that function's
    own docstring). `mechanical_check_registry` is still passed
    unconditionally: a self-planned node declaring `judgment.kind ==
    "mechanical"` (e.g. reusing the catalog's suggested
    `check_script_compiles` for a test_script node) needs it registered,
    same as before.
    """
    plan_result = work_graph_plan.plan_work_graph(
        conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout
    )
    run_id = plan_result["run_id"]
    graph_manifest = plan_result["graph_manifest"]
    node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}

    run_graph_error = None
    try:
        work_graph_orchestrate.run_graph(
            conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
            _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
            mechanical_check_registry={"check_script_compiles": check_script_compiles},
        )
    except Exception as e:
        run_graph_error = e

    artifact_ids = _write_completed_node_artifacts(
        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
        _planner_ext_for, run_graph_error,
    )

    if run_graph_error is not None:
        raise run_graph_error

    return {"run_id": run_id, "artifact_ids": artifact_ids}


# ponytail: 子串匹配，从 docs/todo/validator-test-script-narrative-output-issue.md
# 记录的真实复现文案里提炼——"以为在描述自己做了什么"的稳定措辞模式。只作为喂给
# planner_reviewer 的提示信号，不是硬性机械拦截（跟 check_script_compiles 不同：
# 这里没有客观的"编译通过/失败"可判断，误判成本是把合法的、恰好提到这些词的文档
# 打回重来）。上限：这份短语表穷举不完，Phase 2 的结构性修复（把"探索"和"最终作答"
# 拆成两次独立调用，见 validator-test-script-narrative-output-issue.md §8）才是
# 真正的上游修复，这里只是先兜住已经复现过的具体案例。
_NARRATION_PHRASES = (
    "no further action", "launched the", "workflow finished", "workflow completed",
    "it produced a", "you'll be notified", "no files were changed",
)


def looks_like_narration(content):
    lowered = (content or "").lower()
    return any(phrase in lowered for phrase in _NARRATION_PHRASES)


_PLANNER_REVIEW_PROMPT = (
    "Independently review the Planner's output for this goal, before Coder is allowed to "
    "start implementing. Judge two things: (1) is this real, structured document content, "
    "not narrative prose describing an action taken/completed/launched; (2) is it consistent "
    "with the goal and internally consistent across the documents provided. Some document "
    "types below may be 'N/A: not provided' -- that's expected for goals that don't need "
    "them, not itself a defect.\n\n{narration_hint}"
    "Reply with a verdict line 'VERDICT: PASS' or 'VERDICT: FAIL' or 'VERDICT: INCONCLUSIVE' "
    "followed by your reasoning.\n\n"
    "Goal: {goal_text}\n\nBRD:\n\n{brd_content}\n\nPRD:\n\n{prd_content}\n\n"
    "SDD:\n\n{sdd_content}\n\nBenchmark:\n\n{benchmark_content}"
)


def review_planner_output(
    conn, task_id, goal_text, target_folder_path,
    sdd_content, brd_content="N/A: not provided",
    prd_content="N/A: not provided", benchmark_content="N/A: not provided",
    timeout=300,
):
    """The real planner_reviewer judgment call (not trigger_org_node_full's
    ready handshake). Returns {"verdict", "raw_response"}."""
    hints = [
        "Heuristic flag: {} content resembles a narrative description of an action "
        "rather than structured document content -- verify carefully.".format(label)
        for label, content in (("SDD", sdd_content), ("Benchmark", benchmark_content))
        if looks_like_narration(content)
    ]
    response = org_node_trigger.trigger_org_node(
        "planner_reviewer",
        _PLANNER_REVIEW_PROMPT.format(
            goal_text=goal_text, brd_content=brd_content, prd_content=prd_content,
            sdd_content=sdd_content, benchmark_content=benchmark_content,
            narration_hint=("\n".join(hints) + "\n\n" if hints else ""),
        ),
        target_folder_path, timeout=timeout,
    )
    return {"verdict": _parse_verdict_line(response), "raw_response": response}


_PLANNER_REVIEW_MAX_ATTEMPTS = 3


class PlannerReviewLoopLimitReachedError(Exception):
    """_PLANNER_REVIEW_MAX_ATTEMPTS rounds of planner_reviewer all failed to produce PASS."""


def _existing_doc_artifact_contents(conn, target_folder_path, task_id):
    """{artifact_type: content} for whichever of brd/prd/sdd/benchmark
    genuinely already exist for this task -- may be an empty dict (e.g. the
    task-execution-resume-assumes-planner-artifacts-exist.md incident: the
    Planner's Work Graph failed on its very first node, nothing was ever
    persisted). Existence-probing instead of hard-failing is the actual P0
    bug fix -- see _review_planner_output_with_retries's docstring.

    Import deferred to call time (not module level): task_execution.py
    already imports this module AND scheduler.py, and scheduler.py's own
    top level reads e2e_pipeline.PlannerReviewLoopLimitReachedError to
    build _BLOCKED_REASON_FOR_EXCEPTION -- a module-level `from src import
    task_execution` here closes that into a real 3-way import cycle
    (e2e_pipeline -> task_execution -> scheduler -> e2e_pipeline) that
    breaks depending on which module happens to be imported first. Neither
    function that needs task_execution's helpers runs at import time, so a
    local import is a genuine fix, not a workaround."""
    from src import task_execution
    ids = task_execution._latest_artifact_ids(conn, task_id, task_execution._DOC_ARTIFACT_TYPES)
    return {t: _read_artifact_content(conn, target_folder_path, aid) for t, aid in ids.items()}


def _record_planner_review_event(conn, task_id, attempt, verdict, used_existing_artifacts):
    project_id = conn.execute("SELECT project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
    conn.execute(
        "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
        "VALUES (?, ?, ?, 'planner_review_verdict', ?)",
        (
            str(uuid.uuid4()), project_id, task_id,
            json.dumps({"attempt": attempt, "verdict": verdict, "used_existing_artifacts": used_existing_artifacts}),
        ),
    )
    conn.commit()


def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_path, timeout=300):
    """fresh-run (plan_and_review_documents) and resume-from-
    reviewing_planner_output (task_execution.py) both drive this same core
    loop. Attempt 1 prefers reusing whatever doc artifacts already exist
    (the resume case: don't burn a fresh plan_documents() call re-doing
    work that already succeeded) -- if none exist (P0 bug scenario: the
    Planner graph failed before persisting anything), or on any subsequent
    attempt, calls plan_documents() to generate a fresh batch. Every attempt
    records a planner_review_verdict execution_event for audit; Phase 1
    itself never reads it back to make a decision. Returns plan_documents()'s
    return shape from whichever attempt earned PASS.

    ponytail: every rejected attempt triggers a FULL re-plan of BRD/PRD/SDD/
    Benchmark, not a targeted fix of just the rejected document -- coarse
    but correct, same granularity as org_loop_route.py's existing
    full_replan. Phase 2's node-level loop_back (org-graph-unification-and-
    universal-self-planning-requirements.md Phase 2) is the precise-retry
    upgrade path.
    """
    from src import task_execution  # see _existing_doc_artifact_contents's docstring for why deferred

    feedback_suffix = ""
    verdict = None
    for attempt in range(1, _PLANNER_REVIEW_MAX_ATTEMPTS + 1):
        existing = _existing_doc_artifact_contents(conn, target_folder_path, task_id) if attempt == 1 else {}
        if existing:
            artifact_ids = task_execution._latest_artifact_ids(
                conn, task_id, task_execution._DOC_ARTIFACT_TYPES
            )
            # Phase 2 §6.2 (2026-08-12, found while wiring implement_code()'s
            # new append_to_run_id requirement): this used to hardcode
            # run_id=None -- harmless for Phase 1 (nothing downstream read
            # it), but Phase 2's Coder self-planning needs a REAL run_id to
            # append into. Every real artifact row already has one
            # (produced_by_run_id is NOT NULL) -- read it off the sdd
            # artifact (required a few lines below anyway, guaranteed
            # present whenever `existing` is truthy) rather than inventing
            # a new lookup path.
            sdd_run_id = conn.execute(
                "SELECT produced_by_run_id FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],)
            ).fetchone()[0]
            plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids}
        else:
            plan_result = plan_documents(conn, task_id, goal_text + feedback_suffix, target_folder_path)
            artifact_ids = plan_result["artifact_ids"]

        sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
        brd_content = (
            _read_artifact_content(conn, target_folder_path, artifact_ids["brd"])
            if "brd" in artifact_ids else "N/A: not provided"
        )
        prd_content = (
            _read_artifact_content(conn, target_folder_path, artifact_ids["prd"])
            if "prd" in artifact_ids else "N/A: not provided"
        )
        benchmark_content = (
            _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"])
            if "benchmark" in artifact_ids else "N/A: not provided"
        )

        review = review_planner_output(
            conn, task_id, goal_text, target_folder_path,
            sdd_content, brd_content, prd_content, benchmark_content, timeout=timeout,
        )
        _record_planner_review_event(conn, task_id, attempt, review["verdict"], bool(existing))
        verdict = review["verdict"]
        if verdict == "PASS":
            return plan_result
        feedback_suffix = (
            "\n\n[Planner review attempt {} was rejected ({}): {}. Address this "
            "before continuing.]"
        ).format(attempt, verdict, review["raw_response"])

    raise PlannerReviewLoopLimitReachedError(
        "task {} exhausted {} planner-review attempts, last verdict={}".format(
            task_id, _PLANNER_REVIEW_MAX_ATTEMPTS, verdict
        )
    )


def plan_and_review_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
    """fresh-run entry point: two handshakes (planner, planner_reviewer) +
    the shared retry loop. Replaces run_task_lifecycle()'s previous
    trigger_org_node_full(planner) + plan_documents() pair."""
    org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT)
    org_node_lifecycle.trigger_org_node_full(conn, "planner_reviewer", task_id, target_folder_path, _READY_PROMPT)
    return _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_path, timeout=timeout)


def implement_code(
    conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200,
    benchmark_content="N/A: not provided", test_dataset_content="N/A: not provided",
    test_script_content="N/A: not provided",
):
    """Phase 2 §6 (2026-08-12): Coder now self-plans its own implementation
    graph via `plan_work_graph(..., append_to_run_id=run_id)` -- `run_id`
    is Planner's own work_graph_run (created by `plan_and_review_
    documents()`), no longer a fresh row this function creates itself.
    Coder's self-planned nodes get appended into that SAME run, not a
    separate one -- see work_graph_plan.py's own append_to_run_id
    docstring for the merge mechanics. Replaces the old synthetic
    single-node "implement" manifest (architecture §11 criterion 2's
    original, pre-self-planning shape).

    `code_diff` stays a Python-computed artifact (a real `git diff` across
    whatever Coder's whole self-planned graph actually changed on disk,
    end to end) -- not something any self-planned node's own `content`
    field could meaningfully hold, since no node knows the diff ahead of
    running it. `test_report` is promoted from whichever self-planned
    node(s) declared `promotes_to_artifact: {"artifact_type":
    "test_report"}` (the catalog's own "integration_check" entry suggests
    this); if the LLM's plan didn't include one (the catalog is
    suggestions, not requirements), falls back to concatenating every
    self-planned node's own content in topological order -- same spirit
    as the old single-node manifest's `response` field always being
    Coder's own narrative of what it did, just synthesized instead of
    directly given by one node.

    `needs_worktree=True` passed to run_graph() (2026-08-12, found while
    doing this rework -- NOT previously passed here at all, even though
    Coder is the one org_node that actually writes real code): with a
    single synthetic node this was harmless (nothing to collide with
    inside one run), but a genuinely self-planned graph can have 2+
    code-writing nodes in the SAME topological layer running concurrently
    -- 03-C §3.6's "并行写代码的节点必须先申请 git_worktree_ref" applies for
    real now, not just in theory.

    2026-08-06: benchmark_content/test_dataset_content/test_script_content
    (default 'N/A: not provided' for callers that predate this, e.g. a
    resumed task from before this change) are folded into
    _CODER_PLANNING_CONTEXT so Coder's planning call sees the Planner's
    real test script and must make it pass for real, not just write
    whatever tests it invents on its own.
    """
    before_sha = _current_commit(target_folder_path)

    planning_context = _CODER_PLANNING_CONTEXT.format(
        goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content,
        test_dataset_content=test_dataset_content, test_script_content=test_script_content,
    )
    plan_result = work_graph_plan.plan_work_graph(
        conn, "coder", task_id, planning_context, target_folder_path,
        timeout=300, append_to_run_id=run_id,
    )
    graph_manifest = plan_result["graph_manifest"]
    new_node_ids = set(plan_result["new_node_ids"])

    # Covers BOTH node shapes that can now coexist in this merged graph:
    # Planner's own nodes (artifact_type) and Coder's just-appended ones
    # (promotes_to_artifact) -- used for both _planner_prompt_builder's
    # prompt construction (only its own artifact_type entries are ever
    # queried) and _write_completed_node_artifacts' artifact-type/deps
    # resolution below.
    node_artifact_types = {}
    for n in graph_manifest["nodes"]:
        if "artifact_type" in n:
            node_artifact_types[n["node_id"]] = n["artifact_type"]
        elif n.get("promotes_to_artifact"):
            node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"]

    build_prompt = work_graph_plan.dispatch_build_prompt(
        _planner_prompt_builder(goal_text, node_artifact_types), work_graph_plan.generic_node_prompt_builder,
    )
    run_graph_error = None
    outputs = None
    try:
        outputs = work_graph_orchestrate.run_graph(
            conn, run_id, "coder", "coder-implementation", target_folder_path, graph_manifest,
            build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout,
            needs_worktree=True, mechanical_check_registry={"check_script_compiles": check_script_compiles},
        )
    except Exception as e:
        run_graph_error = e

    diff = subprocess.run(
        ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True
    ).stdout
    diff_result = artifact_write.write_artifact(
        conn, target_folder_path, "code_diff", task_id, run_id,
        ".graphlemon/runs/{}/code_diff.patch".format(task_id), diff or "(no diff -- see test_report)",
    )

    artifact_ids = _write_completed_node_artifacts(
        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
        _planner_ext_for, run_graph_error, only_node_ids=new_node_ids,
    )

    if run_graph_error is not None:
        raise run_graph_error

    if "test_report" not in artifact_ids:
        # The self-planned graph didn't include a node declaring
        # promotes_to_artifact: {"artifact_type": "test_report"} --
        # synthesize one so callers can still rely on this key always
        # being present (same contract the old single-node manifest gave
        # unconditionally via its one node's `response` field).
        report_content = "\n\n---\n\n".join(
            "### {}\n\n{}".format(node_id, (outputs or {}).get(node_id, {}).get("content", ""))
            for node_id in new_node_ids
            if node_id in (outputs or {})
        ) or "(Coder's self-planned graph produced no synthesizable report content)"
        report_result = artifact_write.write_artifact(
            conn, target_folder_path, "test_report", task_id, run_id,
            ".graphlemon/runs/{}/test_report.md".format(task_id), report_content,
        )
        artifact_ids["test_report"] = report_result["artifact_id"]

    return {"code_diff": diff_result["artifact_id"], "test_report": artifact_ids["test_report"]}


def _parse_verdict_line(response):
    verdict = "INCONCLUSIVE"
    for line in response.splitlines():
        stripped = line.strip().upper()
        if stripped.startswith("VERDICT:"):
            for candidate in ("PASS", "FAIL", "INCONCLUSIVE"):
                if candidate in stripped:
                    verdict = candidate
                    break
            break
    return verdict


def validate_implementation(task_id, benchmark_content, goal_text, target_folder_path, timeout=1200):
    """Architecture §11 criterion 3: real Validator call, parses the
    'VERDICT: PASS|FAIL|INCONCLUSIVE' line this Unit's prompt asks for.
    Returns {"verdict", "raw_response"}. Unchanged by the 2026-08-01
    rewiring -- REQ-04's own coverage never named the Work Graph
    controllers (c_bug_record_create/p_validator_root_cause_trace only),
    so the simpler flat trigger_org_node call is already faithful to what
    this REQ actually requires.

    **2026-08-06**: kept as-is (still the simpler flat call, no Validator's
    own Work Graph) for callers that only need this original REQ-04-scoped
    behavior -- run_task_lifecycle now uses
    validate_implementation_with_own_graph() instead, which wraps this same
    prompt/parsing shape but adds the independent benchmark/dataset/script
    generation and the real-script-execution hard gate. See that function's
    own docstring and docs/implementation-gap-analysis.md for why (a
    Benchmark that's only ever prose an LLM narrates over instead of
    substantive content isn't a real acceptance anchor).
    """
    response = org_node_trigger.trigger_org_node(
        "validator",
        _VALIDATOR_PROMPT.format(goal_text=goal_text, benchmark_content=benchmark_content),
        target_folder_path, timeout=timeout,
    )
    return {"verdict": _parse_verdict_line(response), "raw_response": response}


def _script_file_extension(language):
    return {"python": "py", "javascript": "js"}.get(language, "txt")


def run_test_script_for_real(content, target_folder_path, task_id, node_label, timeout=120):
    """Materializes `content` as a real file under .graphlemon/runs/<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 _verdict_says_pass_check(shared_state):
    """Phase 2c (2026-08-12): mechanical judgment for the final_verdict
    node -- `shared_state` is a plain dict the matching build_prompt
    wrapper (_verdict_node_build_prompt) populates with the REAL script
    execution result as a side effect, read back here since a mechanical
    check callable only ever receives `content`, no side channel of its
    own. Two ways to fail:
    1. Hard gate (user-confirmed 2026-08-06, preserved verbatim): a real
       non-zero exit from Validator's own test script is not up for LLM
       debate -- force-reject regardless of what the verdict text says.
    2. The verdict itself isn't PASS -- INCONCLUSIVE is treated the SAME
       as FAIL here (ok=False), not specially accepted: this node's job
       is to keep trying (bounded by the loop_back attempt budget, same
       as everything else on_fail.action='dynamic_loop_back' governs) to
       actually resolve the goal, not to settle for genuine uncertainty
       on the first attempt. If the budget is exhausted without reaching
       PASS, run_graph() raises ConsistencyCheckFailedError same as any
       other exhausted loop -- this function then never returns a non-PASS
       verdict normally, it either returns PASS or raises (see this
       function's own docstring for why that's the right contract)."""
    def check(content):
        script_execution = shared_state.get("script_execution")
        if script_execution and script_execution["ran"] and script_execution["exit_code"] not in (0, None):
            return False, "real test script exited non-zero (exit_code={}) -- hard gate, not up for debate".format(
                script_execution["exit_code"]
            )
        verdict = _parse_verdict_line(content)
        if verdict == "PASS":
            return True, None
        return False, "verdict was {}, not PASS".format(verdict)
    return check


def _verdict_node_build_prompt(inner_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout):
    """Wraps a build_prompt(node, outputs) callable: for the ONE node
    declaring node_type='final_verdict', performs REAL test script
    execution (if a node_type='test_script' dependency produced real
    content) as a side effect before returning the prompt -- embedding
    the real result into both the prompt text (so the model sees it) and
    `shared_state` (so _verdict_says_pass_check's hard gate can enforce it
    in Python, not just ask nicely). Every other node passes through to
    `inner_build_prompt` completely unchanged."""
    def build_prompt(node, outputs):
        if node.get("node_type") != "final_verdict":
            return inner_build_prompt(node, outputs)
        script_node_id = next(
            (nid for nid, node_type in node_types.items() if node_type == "test_script" and nid in outputs), None
        )
        if script_node_id:
            script_execution = run_test_script_for_real(
                outputs[script_node_id].get("content", ""), target_folder_path, task_id,
                "validator_final_verdict_script", timeout=timeout,
            )
        else:
            script_execution = {
                "ran": False, "exit_code": None, "stdout": "", "stderr": "",
                "note": "N/A: this self-planned graph has no test_script-typed node",
            }
        shared_state["script_execution"] = script_execution
        return inner_build_prompt(node, outputs) + (
            "\n\nThe REAL execution result of your own test script (if any):\n"
            + _format_script_execution_result(script_execution)
        )
    return build_prompt


_VALIDATOR_PLANNING_CONTEXT = (
    "{goal_text}\n\n"
    "Approved SDD:\n\n{sdd_content}\n\n"
    "PRD:\n\n{prd_content}\n\n"
    "Guidance: you are independently verifying this goal's implementation. Form your OWN "
    "acceptance criteria informed by (but not copied from) the SDD/PRD above -- don't assume "
    "what the Coder actually did, you have no visibility into that beyond what you inspect "
    "yourself. A held-out test script (if you plan one) will be executed for real against the "
    "actual current repository state -- write real commands, not a description of what a script "
    "would do. Your final verdict must be its own node (see the final_verdict entry in the "
    "suggested node types), depending on whatever earlier nodes it genuinely needs."
)


def validate_implementation_with_own_graph(
    conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, timeout=1200,
):
    """Phase 2c (2026-08-12): Validator self-plans its own verification
    graph via plan_work_graph(conn, "validator", ..., append_to_run_id=
    run_id) -- appended onto the SAME shared run Planner/Coder already
    contributed to, replacing the old hardcoded _VALIDATOR_MANIFEST/
    _VALIDATOR_DOC_PROMPTS/_validator_prompt_builder fixed-chain approach
    (kept in the module, unused, for the same "a future revert would want
    it back verbatim" reason implement_code()'s retired _CODER_PROMPT is).

    The final verdict is now a REAL node inside this graph (node_type=
    'final_verdict', judgment.kind='mechanical', check_ref=
    'verdict_says_pass', on_fail.action='dynamic_loop_back') instead of a
    flat call after the graph finishes -- this is what lets a FAIL
    genuinely trigger Reflection-review + [Fix-Skill] + a precise loop
    back to whichever real ancestor node (possibly Planner's SDD, possibly
    one of Coder's own implementation nodes, anywhere in the shared run)
    is actually at fault, reusing work_graph_orchestrate.py's existing
    dynamic_loop_back machinery (Phase 2a) rather than reimplementing a
    parallel Python-level retry loop. The real test-script hard gate
    (2026-08-06, user-confirmed: script is the hard gate, subjective
    judgment can only add INCONCLUSIVE/risk notes on top of a passing
    script, never override a failing one to PASS) is preserved via
    _verdict_node_build_prompt/_verdict_says_pass_check's shared mutable
    state -- still genuinely Python-enforced, not just prompted.

    **Contract change from the old flat-call version**: because the
    final_verdict node's own mechanical judgment treats anything other
    than PASS as ok=False (INCONCLUSIVE included, see _verdict_says_pass_
    check's own docstring for why) and retries via dynamic_loop_back up to
    its attempt budget, this function can now only ever RETURN normally
    with verdict='PASS' -- a goal that can't be resolved within the loop
    budget raises ConsistencyCheckFailedError instead (same as any other
    exhausted work_graph_orchestrate.py loop) rather than returning a
    'FAIL'/'INCONCLUSIVE' verdict. This is what naturally retires org_
    level_loop.py's involvement for tasks using this path: scheduler.py::
    finish_task_lifecycle's "verdict != PASS -> run_fix_loop()" branch
    becomes unreachable here, not because of an explicit new-flow-vs-old-
    flow check, but because the condition it's gated on can no longer
    occur on this path (resolves design doc §8 point 4 without adding one).
    A caller that genuinely needs the old three-way PASS/FAIL/INCONCLUSIVE
    contract (e.g. this function failing for a reason unrelated to the
    goal itself) still gets that distinction via scheduler.py's existing
    exception-based blocked_reason routing, same mechanism prd_consistency_
    check_failed/test_script_verification_failed already use.

    Returns {"verdict": "PASS", "raw_response", "artifact_ids", "script_execution"}.
    """
    planning_context = _VALIDATOR_PLANNING_CONTEXT.format(
        goal_text=goal_text, sdd_content=sdd_content, prd_content=prd_content,
    )
    plan_result = work_graph_plan.plan_work_graph(
        conn, "validator", task_id, planning_context, target_folder_path,
        timeout=300, append_to_run_id=run_id,
    )
    graph_manifest = plan_result["graph_manifest"]
    new_node_ids = set(plan_result["new_node_ids"])
    node_types = {n["node_id"]: n.get("node_type") for n in graph_manifest["nodes"]}

    verdict_node_ids = [nid for nid, nt in node_types.items() if nt == "final_verdict" and nid in new_node_ids]
    if len(verdict_node_ids) != 1:
        raise work_graph_plan.PlanningFailedError(
            "validator's self-planned graph must have exactly one node_type='final_verdict' node, "
            "found {}: {}".format(len(verdict_node_ids), verdict_node_ids)
        )
    verdict_node_id = verdict_node_ids[0]

    node_artifact_types = {}
    for n in graph_manifest["nodes"]:
        if "artifact_type" in n:
            node_artifact_types[n["node_id"]] = n["artifact_type"]
        elif n.get("promotes_to_artifact"):
            node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"]

    planner_builder = _planner_prompt_builder(goal_text, node_artifact_types)
    base_build_prompt = work_graph_plan.dispatch_build_prompt(planner_builder, work_graph_plan.generic_node_prompt_builder)
    shared_state = {}
    build_prompt = _verdict_node_build_prompt(
        base_build_prompt, node_types, task_id, target_folder_path, shared_state, timeout,
    )

    run_graph_error = None
    outputs = None
    try:
        outputs = work_graph_orchestrate.run_graph(
            conn, run_id, "validator", "validator-independent-verification", target_folder_path, graph_manifest,
            build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout,
            mechanical_check_registry={
                "check_script_compiles": check_script_compiles,
                "verdict_says_pass": _verdict_says_pass_check(shared_state),
            },
        )
    except Exception as e:
        run_graph_error = e

    artifact_ids = _write_completed_node_artifacts(
        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
        _planner_ext_for, run_graph_error, only_node_ids=new_node_ids,
    )

    if run_graph_error is not None:
        raise run_graph_error

    # Reaching here means the final_verdict node's own mechanical judgment
    # passed -- see this function's own docstring for why that means
    # verdict is unconditionally 'PASS' at this point, not something that
    # needs re-parsing/re-checking.
    response = outputs[verdict_node_id]["content"]
    script_execution = shared_state.get("script_execution") or {
        "ran": False, "exit_code": None, "stdout": "", "stderr": "",
        "note": "N/A: no test_script-typed node in this self-planned graph",
    }

    # 2026-08-07 (real incident, preserved): this run_id is now the SHARED
    # Planner+Coder+Validator run (Phase 2c), not a Validator-only one --
    # marking it 'done' here is still correct: reaching this line means the
    # goal genuinely resolved to PASS, the task really is done.
    # ux_work_graph_run_active's real re-verify collision this comment
    # originally described no longer applies the same way once §2.2 lands
    # (Phase 2c also narrows that index) -- kept here as the historical
    # record of why this UPDATE exists at all, not because the exact old
    # failure mode is still reachable.
    conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,))
    conn.commit()

    return {
        "verdict": "PASS", "raw_response": response,
        "artifact_ids": artifact_ids, "script_execution": script_execution,
    }


def _current_commit(repo_path):
    return subprocess.run(
        ["git", "rev-parse", "HEAD"], cwd=repo_path, capture_output=True, text=True
    ).stdout.strip()


def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
    """The body of run_end_to_end() for an ALREADY-CREATED task -- split out
    2026-08-02 so scheduler.py can create the task itself first and still
    have a task_id in scope if this raises partway through (needed to route
    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
    to task.status='blocked_pending_human', which is scheduler.py's job, not
    this module's -- see that module's docstring). Behavior/return shape is
    unchanged from before the split; run_end_to_end() below is now a 2-line
    wrapper over this.
    """
    plan_result = plan_and_review_documents(conn, task_id, goal_text, target_folder_path)
    artifact_ids = plan_result["artifact_ids"]
    sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
    # FIX (2026-08-08, Org Graph unification Phase 1): prd/benchmark used to be
    # artifact_ids["prd"]/artifact_ids["benchmark"] unconditional subscripts, which
    # KeyError'd whenever a self-planned graph omitted either type -- not resume-only,
    # this crashed a fresh first run too. Same "in" guard test_dataset/test_script
    # already used below.
    prd_content = (
        _read_artifact_content(conn, target_folder_path, artifact_ids["prd"])
        if "prd" in artifact_ids else "N/A: not provided"
    )
    benchmark_content = (
        _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"])
        if "benchmark" in artifact_ids else "N/A: not provided"
    )
    # test_dataset/test_script are new 2026-08-06 nodes -- guard with .get
    # so a task resumed from an artifact set produced before this change
    # (no test_dataset/test_script artifact_ids) still works, falling back
    # to implement_code's own "not provided" defaults.
    test_dataset_content = (
        _read_artifact_content(conn, target_folder_path, artifact_ids["test_dataset"])
        if "test_dataset" in artifact_ids else "N/A: not provided"
    )
    test_script_content = (
        _read_artifact_content(conn, target_folder_path, artifact_ids["test_script"])
        if "test_script" in artifact_ids else "N/A: not provided"
    )

    org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, _READY_PROMPT)
    code_artifact_ids = implement_code(
        conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path,
        benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
        test_script_content=test_script_content,
    )

    org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT)
    verdict = validate_implementation_with_own_graph(
        conn, task_id, plan_result["run_id"], sdd_content, prd_content, goal_text, target_folder_path,
    )

    return {
        "task_id": task_id,
        "doc_artifact_ids": artifact_ids,
        "code_artifact_ids": code_artifact_ids,
        "verdict": verdict,
    }


def run_end_to_end(conn, target_folder_path, goal_text):
    """Full orchestration: create_task -> plan_documents -> implement_code ->
    validate_implementation. Returns a dict with every intermediate result,
    for inspection/debugging rather than hiding it behind a single verdict.
    """
    project_id = conn.execute(
        "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
    ).fetchone()[0]
    task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
    return run_task_lifecycle(conn, task_id, target_folder_path, goal_text)


def _read_artifact_content(conn, target_folder_path, artifact_id):
    file_path = conn.execute("SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)).fetchone()[0]
    with open(os.path.join(target_folder_path, file_path), encoding="utf-8") as f:
        return f.read()