宜博 / 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_execute, work_graph_orchestrate, work_graph_plan, work_node_run,
)

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

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

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

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

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

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

_CODER_PROMPT = (
    "Implement this goal per the approved SDD below. Write real code and real tests, run the "
    "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts "
    "and can explain why not).\n\n"
    "The Planner also produced a Benchmark, a test dataset, and a test script for this goal "
    "(test dataset/script may be 'N/A' if this goal has none -- that's expected, not an error). "
    "If the test script is real (not N/A): materialize it as a real file under this project's "
    "own test conventions (match how existing tests in this repo are named/located), and you "
    "MUST make it pass for real before you're done -- it is not optional extra credit, it's "
    "part of this goal's acceptance bar. Write your own additional tests too if the Planner's "
    "script doesn't cover everything the SDD requires; the Planner's script is a floor, not a "
    "ceiling.\n\n"
    "Reply via the required schema with a `response` field describing what you did and the "
    "real test commands/output (including the Planner's own test script's real output). "
    "Goal: {goal_text}\n\nSDD:\n\n{sdd_content}\n\n"
    "Benchmark:\n\n{benchmark_content}\n\n"
    "Test dataset:\n\n{test_dataset_content}\n\n"
    "Test script:\n\n{test_script_content}"
)

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

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

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

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


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


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


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


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


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

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


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


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


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

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

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


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


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


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

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


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

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


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


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

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

    A node whose OWN "-reflect" retry exists (verify_fn rejected the
    first attempt) uses the reflect attempt's content, not the original's
    -- the reflect node_id isn't part of `graph_manifest` so the main loop
    below never visits it directly. `_failed_node_ids(run_graph_error)`
    nodes are excluded outright: their row is schema-valid ('done') but
    verify_fn never accepted either attempt, so neither is trustworthy
    content to persist.
    """
    done_rows = conn.execute(
        "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'",
        (run_id,),
    ).fetchall()
    done_outputs = {node_id: json.loads(output_payload) for node_id, output_payload in done_rows}
    for node_id in list(done_outputs):
        reflect_id = node_id + "-reflect"
        if reflect_id in done_outputs:
            done_outputs[node_id] = done_outputs[reflect_id]
    for excluded in _failed_node_ids(run_graph_error):
        done_outputs.pop(excluded, None)

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


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


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


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

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

    2026-08-08 (Work Graph self-planning + loop-back, FR-B1): `verify_fn`
    is no longer `_combined_planner_verify_fn` -- each self-planned node
    now carries its own `judgment` field (work_graph_plan.py's
    `_PLANNER_NODE_CATALOG` suggests one per node type, the Planner can
    reuse or override it), which `run_graph()` resolves itself per node.
    `_combined_planner_verify_fn`/`_prd_consistency_verify_fn`/
    `_test_script_verify_fn` are NOT deleted -- `validate_implementation_
    with_own_graph()` (Validator's still-hardcoded manifest, unaffected by
    this round) continues to use them (FR-A8). `mechanical_check_registry`
    is still passed unconditionally: a self-planned node declaring
    `judgment.kind == "mechanical"` (e.g. reusing the catalog's suggested
    `check_script_compiles` for a test_script node) needs it registered,
    same as before.
    """
    plan_result = work_graph_plan.plan_work_graph(
        conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout
    )
    run_id = plan_result["run_id"]
    graph_manifest = plan_result["graph_manifest"]
    node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}

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

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

    if run_graph_error is not None:
        raise run_graph_error

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


# 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
            )
            plan_result = {"run_id": None, "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, sdd_content, goal_text, target_folder_path, timeout=1200,
    benchmark_content="N/A: not provided", test_dataset_content="N/A: not provided",
    test_script_content="N/A: not provided",
):
    """Architecture §11 criterion 2: a real Coder call via c_work_node_run
    (work_graph_orchestrate.run_graph with a synthetic single-node
    manifest -- see module docstring for why there's no real Coder graph
    planner), then registers whatever the Coder actually committed as
    code_diff (the diff itself) and test_report (the Coder's own response
    text, which per its migrated SOUL.md is required to include real test
    commands + real output) artifacts.

    2026-08-06: benchmark_content/test_dataset_content/test_script_content
    (default 'N/A: not provided' for callers that predate this, e.g. a
    resumed task from before this change) are threaded into _CODER_PROMPT
    so the Coder must make the Planner's real test script pass for real,
    not just write whatever tests it invents on its own.
    """
    before_sha = _current_commit(target_folder_path)
    run_id = str(uuid.uuid4())
    manifest = {"nodes": [{"node_id": "implement", "depends_on": []}], "edges": []}
    conn.execute(
        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
        "VALUES (?, ?, 'coder', ?, 'running')",
        (run_id, task_id, json.dumps(manifest)),
    )
    conn.commit()

    outputs = work_graph_orchestrate.run_graph(
        conn, run_id, "coder", "coder-implementation", target_folder_path, manifest,
        lambda node, outputs: _CODER_PROMPT.format(
            goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content,
            test_dataset_content=test_dataset_content, test_script_content=test_script_content,
        ),
        _CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout,
    )
    response = outputs["implement"]["response"]

    diff = subprocess.run(
        ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True
    ).stdout

    diff_result = artifact_write.write_artifact(
        conn, target_folder_path, "code_diff", task_id, run_id,
        ".graphlemon/runs/{}/code_diff.patch".format(task_id), diff or "(no diff -- see test_report)",
    )
    report_result = artifact_write.write_artifact(
        conn, target_folder_path, "test_report", task_id, run_id,
        ".graphlemon/runs/{}/test_report.md".format(task_id), response,
    )
    return {"code_diff": diff_result["artifact_id"], "test_report": report_result["artifact_id"]}


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


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

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


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


def run_test_script_for_real(content, target_folder_path, task_id, node_label, timeout=120):
    """Materializes `content` as a real file under .graphlemon/runs/<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"]}
    # 2026-08-07: same run_graph()-partial-failure salvage as
    # plan_documents() -- see _write_completed_node_artifacts's own
    # docstring and docs/todo/work-graph-partial-artifact-persistence-
    # requirements.md. `outputs` is only used below on the SUCCESS
    # continuation (script execution + verdict prompt); the failure path
    # re-raises before ever touching it.
    run_graph_error = None
    outputs = None
    try:
        outputs = work_graph_orchestrate.run_graph(
            conn, run_id, "validator", "validator-independent-verification", target_folder_path, _VALIDATOR_MANIFEST,
            _validator_prompt_builder(goal_text, sdd_content, prd_content, node_artifact_types),
            _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
            verify_fn=_test_script_verify_fn(node_artifact_types, script_artifact_type="validator_test_script"),
            timeout=timeout,
        )
    except Exception as e:
        run_graph_error = e

    artifact_ids = _write_completed_node_artifacts(
        conn, target_folder_path, task_id, run_id, _VALIDATOR_MANIFEST, node_artifact_types,
        lambda artifact_type: "txt" if artifact_type != "validator_benchmark" else "md", run_graph_error,
    )

    if run_graph_error is not None:
        raise run_graph_error

    if "validator_test_script" in outputs:
        script_content = outputs["validator_test_script"]["content"]
        script_execution = run_test_script_for_real(
            script_content, target_folder_path, task_id, "validator_test_script", timeout=timeout,
        )
    else:
        script_execution = {
            "ran": False, "exit_code": None, "stdout": "", "stderr": "",
            "note": "N/A: validator_test_script node is currently disabled (see _VALIDATOR_MANIFEST)",
        }

    response = org_node_trigger.trigger_org_node(
        "validator",
        _VALIDATOR_VERDICT_PROMPT.format(
            goal_text=goal_text,
            validator_benchmark_content=outputs["validator_benchmark"]["content"],
            script_execution_result=_format_script_execution_result(script_execution),
        ),
        target_folder_path, timeout=timeout,
    )
    verdict = _parse_verdict_line(response)

    # Hard gate (user-confirmed 2026-08-06): a real non-zero exit from
    # validator's OWN test script is not up for LLM debate -- force FAIL
    # regardless of what the verdict line said. A passing/not-run script
    # leaves the LLM's own judgment (including INCONCLUSIVE) untouched.
    if script_execution["ran"] and script_execution["exit_code"] not in (0, None):
        verdict = "FAIL"

    # 2026-08-07 (real incident): this function INSERTs its work_graph_run
    # row as status='running' at the top and, on the happy path (verify_fn
    # returns ok=True, so _verify_and_maybe_regenerate never touches
    # work_graph_run.status at all -- it only writes 'reflecting'/
    # 'verifying'/'blocked' on the RETRY path), never wrote anything past
    # that -- regardless of PASS/FAIL/INCONCLUSIVE, a validation run that
    # got this far genuinely finished executing. `ux_work_graph_run_active`
    # (schema.py) is a unique index on (task_id, org_node) covering exactly
    # this non-terminal status set, so a real re-verify later (org_level_
    # loop.run_fix_loop after a FAIL) collided with this row on INSERT --
    # "UNIQUE constraint failed: work_graph_run.task_id, work_graph_run.
    # org_node" -- found for real via a task reconciliation resume.
    conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = ?", (run_id,))
    conn.commit()

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


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


def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
    """The body of run_end_to_end() for an ALREADY-CREATED task -- split out
    2026-08-02 so scheduler.py can create the task itself first and still
    have a task_id in scope if this raises partway through (needed to route
    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
    to task.status='blocked_pending_human', which is scheduler.py's job, not
    this module's -- see that module's docstring). Behavior/return shape is
    unchanged from before the split; run_end_to_end() below is now a 2-line
    wrapper over this.
    """
    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, sdd_content, goal_text, target_folder_path,
        benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
        test_script_content=test_script_content,
    )

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

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


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


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