宜博 / 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 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}"
)

_DOC_PROMPTS = {
    "brd": (
        "Write a BRD (Business Requirement Document) for this goal, following your own "
        "SOUL.md methodology: {goal_text}\n"
        "Reply with the document content (Markdown) via the required schema -- no preamble, "
        "no meta-commentary about what you're about to do."
    ),
    "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."
    ),
    "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."
    ),
    "benchmark": (
        "Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n"
        "Write the Benchmark/acceptance-test document based on this SDD: concrete test "
        "scenarios (happy/boundary/failure) with observable pass criteria. "
        "Reply with the document content (Markdown) via the required schema."
    ),
}

_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). Reply via the required schema with a `response` field describing "
    "what you did and the real test commands/output. Goal: {goal_text}\n\nSDD:\n\n{sdd_content}"
)

_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}"
)


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 = {}
        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(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)
            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 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}}.
    """
    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"]}

    outputs = work_graph_orchestrate.run_graph(
        conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
        _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
        verify_fn=_prd_consistency_verify_fn(node_artifact_types, target_folder_path),
    )

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


def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200):
    """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.
    """
    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),
        _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 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.
    """
    response = org_node_trigger.trigger_org_node(
        "validator",
        _VALIDATOR_PROMPT.format(goal_text=goal_text, benchmark_content=benchmark_content),
        target_folder_path, timeout=timeout,
    )
    verdict = "INCONCLUSIVE"
    for line in response.splitlines():
        stripped = line.strip().upper()
        if stripped.startswith("VERDICT:"):
            for candidate in ("PASS", "FAIL", "INCONCLUSIVE"):
                if candidate in stripped:
                    verdict = candidate
                    break
            break
    return {"verdict": verdict, "raw_response": response}


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


def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
    """The body of run_end_to_end() for an ALREADY-CREATED task -- split out
    2026-08-02 so scheduler.py can create the task itself first and still
    have a task_id in scope if this raises partway through (needed to route
    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
    to task.status='blocked_pending_human', which is scheduler.py's job, not
    this module's -- see that module's docstring). Behavior/return shape is
    unchanged from before the split; run_end_to_end() below is now a 2-line
    wrapper over this.
    """
    org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT)
    plan_result = plan_documents(conn, task_id, goal_text, target_folder_path)
    artifact_ids = plan_result["artifact_ids"]
    sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
    benchmark_content = _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"])

    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)

    org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT)
    verdict = validate_implementation(task_id, benchmark_content, goal_text, target_folder_path)

    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()