宜博 / lemongraph

"""c_work_graph_plan (03-C §3.4): produces a `graph_manifest`, does not execute.

U-011 scope: a real `claude --agent <org_node> -p` call, recorded as a
`work_graph_run` row with `status='planning'` (04-P `p_work_graph_execution`
start_state).

**Empirical finding (2026-07-31)**: `--json-schema` does NOT constrain output
when combined with `--agent` -- tried it for real, the response came back as
normal prose with an embedded ```json fenced block, `structured_output` was
entirely absent from the payload even though the identity itself came through
correctly (the response referenced "SOUL 原则 #10" by number, i.e. it was
genuinely reasoning as the migrated Planner). So this parses the
graph_manifest out of free-form response text instead of trusting
`structured_output` -- see `_extract_graph_manifest`. Not documented anywhere;
found by actually running it, not assumed.

Does NOT execute the resulting plan or produce artifacts -- that's
U-012 (`c_work_graph_execute` full)/U-013 (`c_work_node_run`), and nothing
today owns `c_artifact_write` either (REQ-02 lists it, no Unit in
development-plan.md builds it -- a real gap, flagged rather than silently
plugged, since it affects more than just this Unit's scope). REQ-02's happy
case (real BRD/PRD/SDD/Benchmark artifacts with lineage) needs all of that;
development-plan.md's U-011 Done line claimed it passes here alone, which
isn't accurate -- corrected in this session's notes, same pattern as
U-009/U-010.
"""

import json
import subprocess
import uuid

from src import execution_log, paths

# node_id reserved for the synthetic work_node_execution _record_planning_node
# creates below -- task_query.py matches on this exact string to recognize
# "this run has a real planning node to display", so keep the two in sync.
PLANNING_NODE_ID = "__planning__"

GRAPH_MANIFEST_SCHEMA = {
    "type": "object",
    "required": ["nodes", "edges"],
    "properties": {
        "nodes": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["node_id", "artifact_type", "depends_on"],
                "properties": {
                    "node_id": {"type": "string"},
                    # "artifact_type" (Planner's fixed-catalog nodes) and "task_prompt"
                    # (Coder/Validator's generic self-planned nodes, _generic_prompt_
                    # template()) are mutually exclusive node shapes in practice --
                    # "required" above only reflects Planner's shape (this schema
                    # predates the generic path); see work_graph_plan.py's own dispatch
                    # helpers for how callers tell the two apart per-node.
                    "artifact_type": {
                        "type": "string",
                        "enum": ["brd", "prd", "sdd", "benchmark", "test_dataset", "test_script"],
                    },
                    "task_prompt": {"type": "string"},
                    "depends_on": {"type": "array", "items": {"type": "string"}},
                    # 2026-08-08 (Work Graph self-planning + loop-back) -- both optional,
                    # documentation only (this schema isn't jsonschema-enforced, see
                    # module docstring's --json-schema/--agent note).
                    "judgment": {
                        "type": "object",
                        "properties": {
                            "kind": {"type": "string", "enum": ["none", "mechanical", "llm"]},
                            "check_ref": {"type": "string"},
                            "judge_prompt": {"type": "string"},
                        },
                    },
                    "on_fail": {
                        "type": "object",
                        "properties": {
                            "action": {
                                "type": "string",
                                "enum": ["retry_self", "loop_back", "hard_fail", "dynamic_loop_back"],
                            },
                            "max_attempts": {"type": "integer"},
                        },
                    },
                    # 2026-08-12 (Phase 2 §2.3/§4): also optional/documentation-only.
                    "promotes_to_artifact": {
                        "type": "object",
                        "properties": {"artifact_type": {"type": "string"}},
                    },
                    "subagent_type": {"type": "string"},
                    "skill_ref": {"type": "string"},
                    # 2026-08-12 (Phase 2c): echoed-back catalog label, e.g.
                    # "test_script"/"final_verdict" -- see _generic_prompt_
                    # template's own comment for why this exists (finding
                    # role-specific nodes among arbitrarily-named
                    # self-planned ones).
                    "node_type": {"type": "string"},
                },
            },
        },
        "edges": {
            "type": "array",
            "items": {
                "type": "object",
                "required": ["from", "to", "edge_type"],
                "properties": {
                    "from": {"type": "string"},
                    "to": {"type": "string"},
                    # "on_failure" (2026-08-08): a node's failure path, expressed the
                    # same way its success path already is -- see _on_failure_target
                    # in work_graph_orchestrate.py.
                    "edge_type": {"type": "string", "enum": ["zero_token", "needs_judgment", "on_failure"]},
                },
            },
        },
    },
}

# 2026-08-08 (Work Graph self-planning + loop-back, docs/todo/work-graph-
# self-planning-and-loop-back-requirements.md, FR-B4 "全部自由规划"):
# Planner's doc-writing chain is no longer a fixed architectural constant
# (formally reverses docs/done/dynamic-work-graph-design-skill-design.md
# §3's decision) -- the LLM decides which node types this goal needs, in
# what order, and how each one's judgment/on_fail should work. This
# catalog is reference DATA the prompt hands the LLM, not a requirement
# that every entry appears -- FR-B2. test_script (disabled 2026-08-07,
# docs/done/validator-test-script-narrative-output-issue.md) is back in
# this catalog per that decision's §8 item 7: Part B doesn't wait for the
# narrative-output structural fix, that fix is tracked separately
# alongside a future Coder-side test_script module.
_PRD_JUDGE_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: ok=false only if you found a real "
    "contradiction, and list each one in issues.\n\nPRD:\n\n{content}"
)

_PLANNER_NODE_CATALOG = [
    {"artifact_type": "brd", "description": "Business Requirements Document -- why this goal matters, who it's for."},
    {
        "artifact_type": "prd", "description": "Product Requirements Document -- concrete scenarios/behavior.",
        "suggested_judgment": {"kind": "llm", "judge_prompt": _PRD_JUDGE_PROMPT},
    },
    {"artifact_type": "sdd", "description": "Solution Design Document -- how it will be implemented."},
    {"artifact_type": "benchmark", "description": "Concrete acceptance scenarios with observable pass criteria."},
    {
        "artifact_type": "test_dataset",
        "description": "Concrete test data (inputs + expected outputs) if this goal has a natural one -- "
                        "many goals (UI wiring, config changes) don't, that's fine, omit this node then.",
    },
    {
        "artifact_type": "test_script",
        "description": "A real, runnable script that mechanically checks the Benchmark's scenarios -- "
                        "only include this if a genuine mechanical check is possible for this goal.",
        "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
    },
]

# NOTE: no `str.format()` involved in building or filling this template --
# the embedded catalog JSON and the instructional JSON examples below
# (e.g. {"kind": "none"}) are full of their own literal braces, which
# .format() would try to parse as format fields and crash on.
# plan_work_graph() fills in goal_text with a plain
# `.replace("{goal_text}", goal_text)` instead, which treats every other
# brace in this string as inert literal text -- no escaping needed at all.
# 2026-08-12: _generic_prompt_template() (below) now follows this SAME
# convention for the same reason -- its own catalog JSON has the same
# literal-brace problem .format() would choke on, which the original
# _GENERIC_PROMPT_TEMPLATE constant used to dodge by having no embedded
# JSON at all; now that it does (Phase 2 §3's node catalogs), it needed
# the same fix this template already had.
_PROMPT_TEMPLATE = (
    "Plan (do not execute) the Work Graph for this goal: {goal_text}\n"
    "Available node types (use any subset, in any order that makes sense -- this goal may not "
    "need all of them, and 'suggested_judgment' entries are starting points you can reuse or "
    "override, not requirements):\n" + json.dumps(_PLANNER_NODE_CATALOG) + "\n"
    "For each node you include, decide: node_id, artifact_type, depends_on (which earlier nodes' "
    "output it genuinely needs), a judgment config (how to tell if this node's output is "
    "acceptable -- {\"kind\": \"none\"} if nothing needs checking, {\"kind\": \"mechanical\", "
    "\"check_ref\": ...} for a named mechanical check, or {\"kind\": \"llm\", \"judge_prompt\": "
    "\"...{content}...\"} for your own judgment prompt with a {content} placeholder), and an "
    "on_fail config (what happens if judgment fails: {\"action\": \"retry_self\"}, "
    "{\"action\": \"loop_back\"}, or {\"action\": \"hard_fail\"}, each optionally with "
    "\"max_attempts\"). If a node's on_fail.action is 'loop_back', you MUST also add a "
    "corresponding edge in the top-level 'edges' array with edge_type 'on_failure', from that "
    "node to the exact earlier node_id it should return to -- the target must be a real upstream "
    "dependency of this node (something that genuinely produced input this node used), not an "
    "arbitrary node. This mirrors how the success path already works: every node's 'yes, this "
    "succeeded, what's next' is expressed as a normal dependency edge, and 'no, this failed, go "
    "back to X' is expressed the same way, just with edge_type 'on_failure' instead. Every other "
    "edge needs edge_type: 'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for "
    "anything requiring the agent to actually reason about content.\n"
    "End your reply with the graph_manifest as a single fenced ```json code block "
    "shaped exactly like {\"nodes\": [...], \"edges\": [...]} -- no extra wrapper key."
)

# E (2026-08-06): the generic, cross-Org-Agent path -- used for every
# org_node OTHER than 'planner'. Planner's own doc-writing chain stays on
# _PROMPT_TEMPLATE above unchanged (zero regression risk, see
# docs/done/dynamic-work-graph-design-skill-design.md §3): that chain is a
# fixed architectural constant (brd->prd->sdd->benchmark->...), not
# something that should vary call to call. Each node here carries its own
# self-contained `task_prompt` (see the design-dynamic-work-graph Skill
# this prompt points at) instead of an `artifact_type` that a fixed lookup
# table like _DOC_PROMPTS could resolve.
#
# 2026-08-12 (Phase 2 §3, user-confirmed 2026-08-08 -- reverses this
# comment's own prior "no fixed small vocabulary of node types" framing):
# Coder/Validator's sub-tasks DO get a suggested (not required) catalog
# now, same "starting points, not requirements" contract _PLANNER_NODE_
# CATALOG already uses -- the LLM can ignore every entry and invent its
# own node_ids/task_prompts freely, same as before this change.
_NODE_REVIEW_JUDGE_PROMPT = (
    "Judge whether this node's output is genuinely what it claims to be, not a narrative "
    "description of an action taken (e.g. 'I ran/produced/completed X') standing in for the real "
    "content itself -- this is a known recurring failure mode in nodes that explore with tools "
    "before producing a final answer. ok=false if the content is narration/progress-reporting "
    "prose rather than the real deliverable (code, test output, structured analysis, etc.), or if "
    "it's otherwise clearly incomplete/wrong for what was asked. List each concrete problem in "
    "issues.\n\nContent:\n\n{content}"
)

_CODER_NODE_CATALOG = [
    {"node_type": "impl", "description": "Actual code changes implementing the SDD."},
    {
        "node_type": "unit_test",
        "description": "New/updated unit tests covering the change -- same repo convention as existing tests.",
        "suggested_judgment": {"kind": "llm", "judge_prompt": _NODE_REVIEW_JUDGE_PROMPT},
    },
    {
        "node_type": "integration_check",
        "description": "Actually runs the real test suite/benchmark script against the change and reports "
                        "pass/fail -- not a description of doing so.",
        "suggested_judgment": {"kind": "llm", "judge_prompt": _NODE_REVIEW_JUDGE_PROMPT},
        "suggested_promotes_to_artifact": {"artifact_type": "test_report"},
    },
    {
        "node_type": "test_script",
        "description": "A real, runnable script that mechanically checks the SDD's acceptance scenarios --"
                        "only include this if a genuine mechanical check beyond unit_test/integration_check "
                        "is useful for this goal (e.g. a standalone repro/benchmark script this task's "
                        "deliverable should ship with).",
        "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
    },
]

_VALIDATOR_NODE_CATALOG = [
    {
        "node_type": "independent_benchmark",
        "description": "Validator's own held-out acceptance scenarios -- NOT reused from Planner's/Coder's "
                        "benchmark, a genuine second opinion.",
    },
    {"node_type": "test_dataset", "description": "Concrete test data, if this goal has a natural one."},
    {
        "node_type": "test_script",
        "description": "A real, runnable script mechanically checking the independent benchmark's scenarios. "
                        "It will actually be EXECUTED against the real repository after you plan it -- write "
                        "real commands, not a description of what a script would do.",
        "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
    },
    {
        "node_type": "final_verdict",
        "description": "Exactly ONE node of this type, depending on every other node whose content your "
                        "verdict needs. Its content must be your final judgment: a line 'VERDICT: PASS', "
                        "'VERDICT: FAIL', or 'VERDICT: INCONCLUSIVE', followed by your reasoning (and, if not "
                        "PASS, a root_cause_category of planning|coder|skill_defect with detail). This is the "
                        "ONLY node in a Validator graph allowed to use on_fail.action='dynamic_loop_back' -- "
                        "see the on_fail instructions below for why.",
        "suggested_judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
        "suggested_on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
    },
]

_GENERIC_NODE_CATALOGS = {"coder": _CODER_NODE_CATALOG, "validator": _VALIDATOR_NODE_CATALOG}


def _generic_prompt_template(org_node):
    """Builds _GENERIC_PROMPT_TEMPLATE's replacement, parameterized by
    org_node so each gets its own suggested catalog (or none, for any
    future org_node this hasn't been extended for yet -- catalog_section
    is simply empty then, same as before this change existed).

    2026-08-12 (Phase 2c): `node_type` is now asked to be ECHOED BACK onto
    each planned node (not just a label in the suggestion data) -- e2e_
    pipeline.py's Validator rework needs to find "the test_script node"/
    "the final_verdict node" among self-planned, arbitrarily-named nodes,
    and node_type is the one signal that survives self-planning for this
    (node_id is freeform, task_prompt is prose). `dynamic_loop_back` is
    now allowed, but ONLY for org_node='validator' AND only on a node
    that also declares node_type='final_verdict' -- this reverses the
    Phase 2a/2b instruction ("reserved for the terminal review stage
    outside this graph, not for nodes you plan yourself"), because
    Validator's terminal review IS now planned inside this graph, not a
    flat call after it (see _VALIDATOR_NODE_CATALOG's own 'final_verdict'
    entry for why: only a real graph node can use the loop_back mechanism
    _run_one_node already implements). Every OTHER org_node still can't
    use it -- Coder's catalog has no comparable terminal-judgment role."""
    catalog = _GENERIC_NODE_CATALOGS.get(org_node, [])
    catalog_section = (
        "Suggested node types for this goal (use any subset, in any order, invent your own if "
        "none fit -- these are starting points, not requirements):\n" + json.dumps(catalog) + "\n"
        if catalog else ""
    )
    dynamic_loop_back_rule = (
        "{\"action\": \"dynamic_loop_back\"} ONLY for the node also declaring "
        "node_type='final_verdict' (see the final_verdict catalog entry above) -- no on_failure "
        "edge needed for this one, the target is chosen dynamically at failure time, not declared "
        "now.\n"
        if org_node == "validator" else
        "do NOT use \"dynamic_loop_back\" -- that's reserved for Validator's terminal review node, "
        "not for nodes you plan yourself.\n"
    )
    return (
        "Plan (do not execute) a Work Graph for this goal: {goal_text}\n"
        + catalog_section +
        "Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge "
        "when a node's task_prompt genuinely needs another node's output content -- nodes with "
        "no real data dependency between them will be scheduled to run CONCURRENTLY, so adding "
        "an edge just to make the graph look ordered actively slows execution down. If this goal "
        "is small enough to finish in one call, output a single-node graph_manifest rather than "
        "forcing a split.\n"
        "Output a graph_manifest whose nodes each have: node_id (a short readable identifier), "
        "depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt "
        "for that node's own execution -- assume the agent running it sees nothing except this "
        "task_prompt and its dependencies' outputs, not this planning conversation), plus:\n"
        "  - node_type (optional but STRONGLY suggested when a node matches one of the suggested "
        "types above): echo back the matching node_type string (e.g. \"test_script\") so it's "
        "identifiable later -- omit only for a node that doesn't match any suggested type.\n"
        "  - judgment (optional): how to tell if this node's output is acceptable -- "
        "{\"kind\": \"none\"} if nothing needs checking (the default if omitted), {\"kind\": "
        "\"llm\", \"judge_prompt\": \"...{content}...\"} for your own judgment prompt with a "
        "{content} placeholder.\n"
        "  - on_fail (only meaningful if judgment.kind != \"none\"): {\"action\": \"retry_self\"}, "
        "{\"action\": \"hard_fail\"}, or {\"action\": \"loop_back\"} with a matching on_failure "
        "edge to a genuine upstream dependency (same rule _PLANNER_NODE_CATALOG's own nodes "
        "follow); " + dynamic_loop_back_rule +
        "  - promotes_to_artifact (optional): {\"artifact_type\": \"test_report\"} ONLY for a "
        "node producing this task's actual final deliverable text -- most nodes should omit "
        "this; it marks a node's content for persistence beyond this graph's own execution. "
        "(code_diff, if relevant, is derived separately from real file changes, not from any "
        "node's own content -- never declare artifact_type: \"code_diff\" here. The final_verdict "
        "node, if any, should also NOT declare promotes_to_artifact -- a verdict is a judgment, "
        "not a persisted deliverable.)\n"
        "Every edge needs edge_type: 'zero_token' for pure aggregation/reduce steps, "
        "'needs_judgment' for anything requiring the agent to actually reason about content.\n"
        "End your reply with the graph_manifest as a single fenced ```json code block "
        "shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key."
    )


class PlanningFailedError(Exception):
    """PLANNING_FAILED -- 03-C §3.4."""


def extract_json_payload(text, required_keys=()):
    """--json-schema doesn't constrain output when --agent is set (see module
    docstring) -- pulls a JSON payload out of free-form response text
    instead: try the whole text as JSON, then a real JSON object starting at
    every '{' in the text; accept the first candidate that parses to a dict
    containing every key in required_keys. Generalized (2026-08, work-graph-
    node-execution-deblackbox) from what used to be `_extract_graph_manifest`'s
    own inline algorithm, so `org_node_trigger.trigger_org_node_structured`
    can reuse it for arbitrary output_schemas, not just graph_manifest's
    {nodes, edges} shape. Wrapper-key fallbacks (like _extract_graph_manifest's
    own "graph_manifest" key below) are caller-specific and deliberately left
    out of this shared layer.

    2026-08-15 (real incident, task b243dfe8, diagnosed live with the user --
    they correctly pushed back on "is this over-constrained code" and asked
    why a check step needs to be this strict; the honest answer turned out to
    be the opposite: this used to under-use what's available, approximating
    JSON-object-boundary-finding with a regex, `_JSON_FENCE_RE`'s
    ```(?:json)?\\s*(\\{.*?\\})\\s*``` -- a NON-GREEDY match, so it stops at the
    FIRST '}' anywhere in the text that happens to be followed by whitespace
    and a closing fence. Content containing real code (a test script,
    guaranteed to contain '{'/'}' -- dict literals, f-strings, function
    calls) can easily contain that exact sequence INCIDENTALLY, deep inside
    an otherwise perfectly valid JSON string value -- no malformed input
    required. Confirmed with fully valid, json.dumps()-produced input:
    {"content": "d = {\"a\": 1}   ```\\nmore code\\n"} silently failed to
    extract, because the regex stopped at the '}' right after `{"a": 1}`,
    which happens to be followed by spaces (legal, unescaped JSON string
    content) and then a literal ```. A regex fundamentally can't track
    nested/balanced braces correctly (a classical limitation, not a bug in
    how carefully this one was written) -- the fix isn't a tighter regex, it's
    using the parser Python already ships: json.JSONDecoder.raw_decode()
    walking real JSON grammar (proper string-escaping/nesting awareness) is
    immune to this whole class of failure, for free, with less code than the
    regex approach had."""
    try:
        parsed = json.loads(text)
        if isinstance(parsed, dict) and all(key in parsed for key in required_keys):
            return parsed
    except (json.JSONDecodeError, TypeError):
        pass
    decoder = json.JSONDecoder()
    for idx, ch in enumerate(text):
        if ch != "{":
            continue
        try:
            parsed, _ = decoder.raw_decode(text, idx)
        except json.JSONDecodeError:
            continue
        if isinstance(parsed, dict) and all(key in parsed for key in required_keys):
            return parsed
    return None


def _extract_graph_manifest(text):
    """Thin wrapper around extract_json_payload: also accepts the payload
    wrapped one level deep under a "graph_manifest" key (models sometimes
    add that wrapper despite being asked not to) -- kept as its own function
    rather than folding the wrapper-key check into extract_json_payload
    itself, since no other caller needs this specific wrapper-key name.

    2026-08-15: own fallback scan uses the same real-parser approach
    extract_json_payload's own docstring explains (not the old regex,
    _JSON_FENCE_RE -- same non-greedy-can't-track-nesting vulnerability
    would otherwise still exist here, just for the wrapper-key shape
    instead of the bare {nodes, edges} shape)."""
    result = extract_json_payload(text, required_keys=["nodes", "edges"])
    if result is not None:
        return result
    try:
        parsed = json.loads(text)
    except (json.JSONDecodeError, TypeError):
        parsed = None
    if isinstance(parsed, dict):
        inner = parsed.get("graph_manifest")
        if isinstance(inner, dict) and "nodes" in inner and "edges" in inner:
            return inner
    decoder = json.JSONDecoder()
    for idx, ch in enumerate(text):
        if ch != "{":
            continue
        try:
            parsed, _ = decoder.raw_decode(text, idx)
        except json.JSONDecodeError:
            continue
        if not isinstance(parsed, dict):
            continue
        inner = parsed.get("graph_manifest")
        if isinstance(inner, dict) and "nodes" in inner and "edges" in inner:
            return inner
    return None


def _transitive_ancestors(graph_manifest, node_id):
    """node_id itself + every node it (directly or transitively) depends
    on, via `depends_on`/normal data-dependency edges (`on_failure` edges
    excluded -- they're control-flow, not data dependency, same reasoning
    as work_graph_orchestrate.py's own `_transitive_dependents`/
    `topological_layers`). Deliberately reimplemented here rather than
    imported from work_graph_orchestrate.py -- the two modules don't
    depend on each other today (work_graph_orchestrate.py is a lower-level
    execution engine, work_graph_plan.py is planning-only) and this is a
    small enough function that duplicating it doesn't justify introducing
    that coupling."""
    deps = {n.get("node_id"): set(n.get("depends_on") or []) for n in graph_manifest.get("nodes", [])}
    for edge in graph_manifest.get("edges", []):
        if edge.get("edge_type") == "on_failure":
            continue
        from_id, to_id = edge.get("from"), edge.get("to")
        if to_id in deps and from_id in deps:
            deps[to_id].add(from_id)

    ancestors, stack = set(), [node_id]
    while stack:
        current = stack.pop()
        if current in ancestors:
            continue
        ancestors.add(current)
        stack.extend(deps.get(current, []))
    return ancestors


def _validate_loop_back_targets(manifest, additional_valid_target_ids=()):
    """2026-08-08 (Work Graph self-planning + loop-back): every node
    declaring `on_fail.action == 'loop_back'` must have exactly one
    `edge_type: 'on_failure'` edge originating from it (see
    work_graph_orchestrate.py's `_on_failure_target`), and that edge's
    target must be a genuine upstream dependency -- looping back to a node
    that never influenced this one has no coherent meaning. Raises
    PlanningFailedError (not a generic ValueError) so this surfaces the
    same way every other planning failure does, before any node ever runs
    -- catching a malformed loop_back here is far cheaper than discovering
    it mid-execution.

    `additional_valid_target_ids` (2026-08-13, real incident task b243dfe8):
    append mode (see plan_work_graph's own docstring) calls this on the
    NEWLY-planned manifest alone, before it's merged with the existing
    persisted one -- a new node legitimately looping back to an
    already-persisted node from an earlier append is completely valid
    (e.g. a new verification node targeting an earlier SDD node), but
    without this, that target genuinely isn't in THIS manifest's own
    node_ids yet, and got wrongly rejected as nonexistent. Ancestor-chain
    validation below is skipped for a target in this set -- an
    already-persisted node's ancestry was already validated when IT was
    planned, and re-deriving `_transitive_ancestors` across a manifest that
    doesn't yet contain those nodes isn't meaningful anyway."""
    node_ids = {n.get("node_id") for n in manifest.get("nodes", [])}
    additional_valid_target_ids = set(additional_valid_target_ids)
    for node in manifest.get("nodes", []):
        if (node.get("on_fail") or {}).get("action") != "loop_back":
            continue
        node_id = node.get("node_id")
        on_failure_edges = [
            e for e in manifest.get("edges", [])
            if e.get("from") == node_id and e.get("edge_type") == "on_failure"
        ]
        if len(on_failure_edges) != 1:
            raise PlanningFailedError(
                "node {} declares on_fail.action='loop_back' but has {} on_failure edge(s), "
                "expected exactly 1".format(node_id, len(on_failure_edges))
            )
        target = on_failure_edges[0].get("to")
        if target in additional_valid_target_ids:
            continue
        if target not in node_ids:
            raise PlanningFailedError(
                "node {}'s on_failure edge targets {!r}, which does not exist in this graph_manifest".format(
                    node_id, target
                )
            )
        if target not in _transitive_ancestors(manifest, node_id):
            raise PlanningFailedError(
                "node {}'s on_failure edge targets {!r}, which is not an upstream dependency -- "
                "loop_back must target a node that genuinely influenced this one".format(node_id, target)
            )


def _normalize_graph_manifest(manifest, additional_valid_loop_back_target_ids=()):
    """Real LLM output has been observed (2026-08-01, U-031 integration
    validation) to use `id` instead of the documented `node_id` key, and to
    express dependencies purely via `edges` (from/to) without ever
    populating each node's own `depends_on` array -- even though this
    schema requires both. Normalizing here, right after parsing and before
    persisting, means every downstream consumer (work_graph_orchestrate.py's
    topological sort, v_task_detail's to-do tree, v_graph_visualizer's DAG
    layering) gets a manifest that actually matches the documented shape,
    instead of each one needing its own defensive "id or node_id,
    depends_on or edges" fallback logic repeated independently. Mutates and
    returns `manifest`.

    2026-08-08 (Work Graph self-planning + loop-back, FR-B4 "全部自由规划"):
    the canonical-doc-chain forced-order enforcement that used to live here
    (2026-08-03 incident: a real call omitted brd->prd->sdd->benchmark's
    dependency entirely, scheduling nodes out of order) is REMOVED, not
    extended -- Planner's node set/order is no longer a fixed architectural
    constant (formally reverses docs/done/dynamic-work-graph-design-skill-
    design.md §3). Robustness against that same failure mode now rests
    entirely on the edge-derived depends_on backfill below (unchanged) plus
    `_validate_loop_back_targets` (new, below) -- there is no longer a
    hardcoded canonical chain to fall back on if an LLM response omits
    dependency info, so a real reproduction of the 2026-08-03 incident
    shape is a genuine, currently-unmitigated risk this design accepted
    (see docs/done/work-graph-self-planning-and-loop-back-requirements.md
    §7's risk list) -- FR-B3's own acceptance criterion only covers the
    "wrong key names" failure mode this backfill already handles, not a
    complete absence of dependency info with no canonical chain to guess from.
    """
    nodes = manifest.get("nodes", [])
    for node in nodes:
        if "node_id" not in node and "id" in node:
            node["node_id"] = node["id"]
        node.setdefault("depends_on", [])

    node_ids = {n.get("node_id") for n in nodes}
    for edge in manifest.get("edges", []):
        if edge.get("edge_type") == "on_failure":
            continue  # control-flow, not a data dependency -- see _transitive_ancestors
        from_id, to_id = edge.get("from"), edge.get("to")
        if from_id in node_ids and to_id in node_ids:
            for node in nodes:
                if node.get("node_id") == to_id and from_id not in node["depends_on"]:
                    node["depends_on"].append(from_id)

    _validate_loop_back_targets(manifest, additional_valid_loop_back_target_ids)
    return manifest


def _record_planning_node(conn, run_id, task_id, org_node, prompt, raw_response, graph_manifest):
    """Persists the planning call itself (the `claude --agent <org_node> -p
    <prompt>` subprocess call in plan_work_graph() below, NOT any node it
    planned) as a real, clickable work_node_execution -- previously this
    call's own raw output was discarded the moment _extract_graph_manifest
    pulled the JSON back out of it, and the call itself left no trace
    anywhere the console renders (v_task_detail's to-do tree/DAG only ever
    show the nodes a planning call PRODUCED, never the act of planning
    itself). Deliberately does NOT touch work_graph_run.graph_manifest --
    that column stays the pure, LLM-authored manifest run_graph()/fork/
    append-mode all still execute and merge against unchanged; task_query.py
    synthesizes a display-only node from THIS row (matched by node_id ==
    PLANNING_NODE_ID) when building the console's API response instead, so
    execution/fork/append logic never has to know this row exists.

    Only called from the fresh (non-append) branch of plan_work_graph() --
    one real "how was this run planned" entry per run is enough to answer
    the question; Coder/Validator's own append_to_run_id planning calls
    aren't given one (YAGNI until someone actually needs to inspect those
    too)."""
    project_id = conn.execute("SELECT project_id FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]

    # work_node_execution.skill_ref is a required FK to a real skill row.
    # design-dynamic-work-graph is the actual Skill documenting this
    # self-planning mechanism (skills/design-dynamic-work-graph/SKILL.md) --
    # reused here rather than inventing a placeholder skill_id nothing else
    # would ever reference.
    skill_id = "design-dynamic-work-graph"
    if conn.execute("SELECT 1 FROM skill WHERE skill_id = ?", (skill_id,)).fetchone() is None:
        conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES (?, ?, ?, 'unknown', 'new', 'active')",
            (skill_id, org_node, "skills/" + skill_id + "/"),
        )

    node_exec_id = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
        "input_schema_ref, output_schema_ref, input_payload, output_payload, status) "
        "VALUES (?, ?, ?, ?, ?, 'work_graph_plan.goal_text', 'work_graph_plan.graph_manifest', ?, ?, 'done')",
        (
            node_exec_id, run_id, PLANNING_NODE_ID, org_node, skill_id,
            json.dumps({"prompt": prompt}), json.dumps({"graph_manifest": graph_manifest}),
        ),
    )

    log_file_id = str(uuid.uuid4())
    log_path = paths.log_file_path(project_id, task_id, run_id, node_exec_id)
    conn.execute(
        "INSERT INTO execution_log_file (log_file_id, node_exec_id, project_id, file_path, format, status) "
        "VALUES (?, ?, ?, ?, 'jsonl', 'closed')",
        (log_file_id, node_exec_id, project_id, log_path),
    )
    # execution_log.append_log does its own conn.commit() -- covers the two
    # INSERTs above too, same connection/transaction.
    execution_log.append_log(
        conn, node_exec_id,
        json.dumps({"event": "node_start", "org_node": org_node, "prompt": prompt}, ensure_ascii=False) + "\n"
        + json.dumps(
            {"event": "node_completed", "raw_response": raw_response, "graph_manifest": graph_manifest},
            ensure_ascii=False,
        ) + "\n",
    )


def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None):
    """Returns {"run_id", "graph_manifest", "new_node_ids"}. Creates one
    `work_graph_run` row (status='planning') -- moving it to 'running'/
    executing it is out of scope.

    2026-08-08: `org_node == 'planner'` gets `_PROMPT_TEMPLATE` (now a
    genuine self-planning prompt over `_PLANNER_NODE_CATALOG`, see that
    template's own comment for why goal_text is filled in via `.replace()`
    rather than `.format()`). Every other org_node gets
    `_generic_prompt_template(org_node)` -- the cross-Org-Agent dynamic
    planning path described in the design-dynamic-work-graph Skill, whose
    nodes carry their own task_prompt instead of a fixed artifact_type.

    2026-08-12 (Phase 2 §6.1, user-confirmed 2026-08-08: a parameter on
    this function, not a separate plan_work_graph_append()): with
    `append_to_run_id=None` (default), behavior is unchanged from before
    this parameter existed -- a fresh INSERT. With `append_to_run_id=
    <existing run_id>`, this call plans ADDITIONAL nodes for an
    already-existing work_graph_run (e.g. Coder appending its own
    implementation nodes onto Planner's run) instead of creating a new
    row: the existing manifest's node_ids are injected into the prompt so
    new nodes can `depends_on` them, a `node_id` collision between the
    newly-planned nodes and the existing ones raises PlanningFailedError
    (never silently renamed or overwritten), and the row is UPDATEd with
    the merged manifest instead of a new row being INSERTed.
    `new_node_ids` (only meaningfully non-empty in append mode, though
    always present for a uniform return shape) is the set of node_ids this
    specific call added -- callers that promote node output to `artifact`
    rows (e2e_pipeline.py's `_write_completed_node_artifacts`) need this
    to avoid re-processing nodes an EARLIER stage already promoted; the
    merged manifest itself doesn't distinguish "planned this call" from
    "already existed" once merged."""
    existing_manifest = None
    if append_to_run_id:
        row = conn.execute(
            "SELECT graph_manifest FROM work_graph_run WHERE run_id = ?", (append_to_run_id,)
        ).fetchone()
        if row is None:
            raise ValueError("no such work_graph_run to append to: " + append_to_run_id)
        existing_manifest = json.loads(row[0])

    if org_node == "planner":
        prompt = _PROMPT_TEMPLATE.replace("{goal_text}", goal_text)
    else:
        prompt = _generic_prompt_template(org_node).replace("{goal_text}", goal_text)

    if existing_manifest is not None:
        existing_ids = [n["node_id"] for n in existing_manifest["nodes"]]
        prompt = prompt.replace(
            "Plan (do not execute)",
            "This graph already has these nodes from earlier stages (you may depends_on them, but do "
            "NOT redeclare or duplicate them -- pick different node_ids for your own new nodes): "
            "{}. Plan (do not execute) ADDITIONAL nodes only".format(existing_ids),
            1,
        )

    result = subprocess.run(
        [
            "claude", "--agent", org_node,
            "-p", prompt,
            "--output-format", "json",
            "--permission-mode", "auto",
        ],
        cwd=target_folder_path,
        stdin=subprocess.DEVNULL,
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    if result.returncode != 0:
        # 2026-07-31 fix (same bug as org_node_trigger.py): include stdout too.
        raise PlanningFailedError(
            "plan failed for task {}\nstderr: {}\nstdout (truncated): {}".format(
                task_id, result.stderr, result.stdout[:4000]
            )
        )

    payload = json.loads(result.stdout)
    if payload.get("is_error"):
        raise PlanningFailedError("plan failed for task {}: {}".format(task_id, payload))
    new_manifest = _extract_graph_manifest(payload.get("result", ""))
    if not new_manifest:
        raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload))
    existing_node_ids = (
        {n["node_id"] for n in existing_manifest["nodes"]} if existing_manifest is not None else ()
    )
    new_manifest = _normalize_graph_manifest(new_manifest, existing_node_ids)

    if existing_manifest is None:
        run_id = str(uuid.uuid4())
        # org_node column removed (Org Graph unification Phase 2c §2.1,
        # 2026-08-12) -- this run's real per-node ownership lives in each
        # node's own work_node_execution.subagent_type (set per node in
        # _run_one_node()), not a single run-level value; a merged run
        # genuinely spans several org_nodes by design (Coder/Validator
        # append their own nodes into this SAME run via append_to_run_id).
        conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, ?, ?, 'planning')",
            (run_id, task_id, json.dumps(new_manifest)),
        )
        conn.commit()
        _record_planning_node(conn, run_id, task_id, org_node, prompt, payload.get("result", ""), new_manifest)
        return {"run_id": run_id, "graph_manifest": new_manifest, "new_node_ids": []}

    new_node_ids = [n["node_id"] for n in new_manifest["nodes"]]
    existing_ids_set = {n["node_id"] for n in existing_manifest["nodes"]}
    collision = existing_ids_set & set(new_node_ids)
    if collision:
        raise PlanningFailedError(
            "append_to_run_id={} node_id collision with existing nodes: {}".format(append_to_run_id, collision)
        )
    merged = {
        "nodes": existing_manifest["nodes"] + new_manifest["nodes"],
        "edges": existing_manifest.get("edges", []) + new_manifest.get("edges", []),
    }
    # _validate_loop_back_targets over the MERGED graph, not just the new
    # nodes -- a new node's on_fail.loop_back could (in principle) target
    # an existing node, and _normalize_graph_manifest already validated
    # new_manifest in isolation above, not against the merged shape.
    _validate_loop_back_targets(merged)
    conn.execute(
        "UPDATE work_graph_run SET graph_manifest = ? WHERE run_id = ?",
        (json.dumps(merged), append_to_run_id),
    )
    conn.commit()
    return {"run_id": append_to_run_id, "graph_manifest": merged, "new_node_ids": new_node_ids}


def generic_node_prompt_builder(node, outputs):
    """E (2026-08-06): the `build_prompt(node, outputs)` run_graph() needs
    for a graph_manifest produced via `_generic_prompt_template()` -- lives
    here, not in e2e_pipeline.py, because it's tightly coupled to the
    `task_prompt` field convention that prompt asks for (keeping both in
    the same module means a field rename can't silently desync producer
    and consumer). No factory/closure needed: every node is self-contained
    per the design-dynamic-work-graph Skill's contract, so `outputs`
    (already-completed dependency content) isn't needed here."""
    return node["task_prompt"]


def dispatch_build_prompt(planner_builder, generic_builder):
    """2026-08-12 (Phase 2 §6.1): a build_prompt(node, outputs) that picks
    per-NODE between planner_builder (Planner's artifact_type-keyed nodes,
    e.g. e2e_pipeline._planner_prompt_builder's return value) and
    generic_builder (task_prompt-keyed nodes, e.g. generic_node_prompt_
    builder above) -- needed once a single graph_manifest can genuinely
    contain BOTH shapes side by side, which append_to_run_id makes real
    for the first time (Planner's original nodes + Coder's/Validator's
    later self-planned ones, merged into one run). Every pre-Phase-2
    caller only ever had ONE shape for a whole graph and dispatched once
    up front; this replaces that with a per-node check so a caller doesn't
    need to know in advance whether ANY given node in a merged graph is
    Planner-shaped or generic-shaped. Shared here (not duplicated in
    e2e_pipeline.py/work_graph_fork.py, its two real callers) so the two
    don't drift on what "neither shape" should do."""
    def build_prompt(node, outputs):
        if "artifact_type" in node:
            return planner_builder(node, outputs)
        if "task_prompt" in node:
            return generic_builder(node, outputs)
        raise ValueError(
            "node {} has neither 'artifact_type' nor 'task_prompt' -- can't dispatch a build_prompt".format(
                node.get("node_id")
            )
        )
    return build_prompt