Phase 2b: Coder self-plans its own implementation graph via append_to_run_id

Implements org-graph-unification-phase2-universal-self-planning-design.md §2.3, §3, §6 -- Coder's implement_code() no longer runs one hardcoded synthetic "implement" node; it self-plans via plan_work_graph(conn, "coder", ..., append_to_run_id=run_id), appending its own nodes onto Planner's existing work_graph_run instead of creating a separate one. - work_graph_plan.py: per-org_node suggested node-type catalogs (_CODER_NODE_CATALOG/_VALIDATOR_NODE_CATALOG, "starting points, not requirements" same as _PLANNER_NODE_CATALOG), new judgment/on_fail/ promotes_to_artifact fields in the generic planning prompt. New plan_work_graph(..., append_to_run_id=...) parameter: merges newly planned nodes into an existing run (node_id collision -> raise, never silent overwrite), returns new_node_ids so callers can avoid reprocessing nodes an earlier stage already promoted to artifacts. New shared dispatch_build_prompt() for graphs mixing Planner-shaped (artifact_type) and Coder-shaped (task_prompt) nodes side by side. - e2e_pipeline.py: _write_completed_node_artifacts() now also promotes nodes declaring promotes_to_artifact (not just Planner's artifact_type nodes), with an only_node_ids scope so a merged-graph caller doesn't re-touch artifacts an earlier stage already wrote. implement_code() reworked around self-planning; code_diff stays Python-computed (a real git diff, not any node's own content -- no node can know the diff ahead of running it), test_report is promoted from whatever the self-planned graph declares, falling back to a synthesized concatenation if it declares none. needs_worktree=True now passed to run_graph() for Coder (found missing entirely during this rework -- harmless with the old single-node manifest, a real 03-C §3.6 gap once concurrent code-writing nodes are possible). Also fixed a real bug found while wiring this up: _review_planner_output_with_retries()'s "reuse existing artifacts" branch hardcoded run_id=None, even though every real artifact row already has a valid produced_by_run_id -- Coder had nothing to append into on that resume path. - work_graph_fork.py: _reconstruct_build_prompt now dispatches per-node (via the new shared dispatch_build_prompt) instead of requiring the whole graph to be one uniform shape -- a real regression Phase 2b would otherwise have introduced silently, since a forked merged graph can now genuinely mix both node shapes. Sequencing note (unchanged from Phase 2a's own finding): work_graph_run's org_node column removal and the ux_work_graph_run_active index narrowing (design doc §2.1/§2.2) stay deferred to Phase 2c -- Validator still creates its own separate run until Validator also appends into the shared one, so narrowing the index to task_id-only now would collide with it. python3 -m unittest discover -s src (752 tests) + discover -s supervisor -p "test_*.py" (126 tests) both green, zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Phase 2b: Coder self-plans its own implementation graph via append_to_run_id
Implements org-graph-unification-phase2-universal-self-planning-design.md §2.3, §3, §6 -- Coder's implement_code() no longer runs one hardcoded synthetic "implement" node; it self-plans via plan_work_graph(conn, "coder", ..., append_to_run_id=run_id), appending its own nodes onto Planner's existing work_graph_run instead of creating a separate one. - work_graph_plan.py: per-org_node suggested node-type catalogs (_CODER_NODE_CATALOG/_VALIDATOR_NODE_CATALOG, "starting points, not requirements" same as _PLANNER_NODE_CATALOG), new judgment/on_fail/ promotes_to_artifact fields in the generic planning prompt. New plan_work_graph(..., append_to_run_id=...) parameter: merges newly planned nodes into an existing run (node_id collision -> raise, never silent overwrite), returns new_node_ids so callers can avoid reprocessing nodes an earlier stage already promoted to artifacts. New shared dispatch_build_prompt() for graphs mixing Planner-shaped (artifact_type) and Coder-shaped (task_prompt) nodes side by side. - e2e_pipeline.py: _write_completed_node_artifacts() now also promotes nodes declaring promotes_to_artifact (not just Planner's artifact_type nodes), with an only_node_ids scope so a merged-graph caller doesn't re-touch artifacts an earlier stage already wrote. implement_code() reworked around self-planning; code_diff stays Python-computed (a real git diff, not any node's own content -- no node can know the diff ahead of running it), test_report is promoted from whatever the self-planned graph declares, falling back to a synthesized concatenation if it declares none. needs_worktree=True now passed to run_graph() for Coder (found missing entirely during this rework -- harmless with the old single-node manifest, a real 03-C §3.6 gap once concurrent code-writing nodes are possible). Also fixed a real bug found while wiring this up: _review_planner_output_with_retries()'s "reuse existing artifacts" branch hardcoded run_id=None, even though every real artifact row already has a valid produced_by_run_id -- Coder had nothing to append into on that resume path. - work_graph_fork.py: _reconstruct_build_prompt now dispatches per-node (via the new shared dispatch_build_prompt) instead of requiring the whole graph to be one uniform shape -- a real regression Phase 2b would otherwise have introduced silently, since a forked merged graph can now genuinely mix both node shapes. Sequencing note (unchanged from Phase 2a's own finding): work_graph_run's org_node column removal and the ux_work_graph_run_active index narrowing (design doc §2.1/§2.2) stay deferred to Phase 2c -- Validator still creates its own separate run until Validator also appends into the shared one, so narrowing the index to task_id-only now would collide with it. python3 -m unittest discover -s src (752 tests) + discover -s supervisor -p "test_*.py" (126 tests) both green, zero regressions. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee · 宜博
1 parent bd1404db
Showing 8 changed files with 857 additions and 113 deletions
src/e2e_pipeline.py
src/task_execution.py
src/test_e2e_pipeline.py
src/test_task_execution.py
src/test_work_graph_fork.py
src/test_work_graph_plan.py
src/work_graph_fork.py
src/work_graph_plan.py
@@ -140,6 +140,12 @@ _DOC_PROMPTS = {
 _CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
 _CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]}
+# 2026-08-12 (Phase 2 §6): retired as the per-node prompt now that Coder
+# self-plans (implement_code() no longer runs one synthetic "implement"
+# node) -- its guidance is folded into _CODER_PLANNING_CONTEXT below,
+# which feeds plan_work_graph()'s planning call instead of a single
+# execution call. Kept here, unused, only because deleting it isn't this
+# change's job and a future revert would want it back verbatim.
 _CODER_PROMPT = (
     "Implement this goal per the approved SDD below. Write real code and real tests, run the "
     "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts "
@@ -160,6 +166,32 @@ _CODER_PROMPT = (
     "Test script:\n\n{test_script_content}"
 )
+# 2026-08-12 (Phase 2 §6): fed as `goal_text` into plan_work_graph(conn,
+# "coder", ..., append_to_run_id=run_id) -- reuses the existing
+# {goal_text} substitution point in _generic_prompt_template() rather than
+# adding a new parameter to plan_work_graph() just for Coder's extra
+# context (SDD/Benchmark/test dataset/script), which Planner's own
+# _PROMPT_TEMPLATE call never needed. Same .replace("{goal_text}", ...)
+# convention _PROMPT_TEMPLATE/_generic_prompt_template() already use (not
+# .format() -- see work_graph_plan.py's own comment on why), so this
+# string's remaining braces (none currently, but keep this in mind if
+# edited) must stay literal, not format fields.
+_CODER_PLANNING_CONTEXT = (
+    "{goal_text}\n\n"
+    "Approved SDD:\n\n{sdd_content}\n\n"
+    "Benchmark (test dataset/script may be 'N/A' if this goal has none -- that's expected, not "
+    "an error):\n\n{benchmark_content}\n\n"
+    "Test dataset:\n\n{test_dataset_content}\n\n"
+    "Test script:\n\n{test_script_content}\n\n"
+    "Guidance: write real code and real tests, run the tests yourself as part of your own "
+    "nodes' work, and don't consider the goal done until they pass for real (or you've "
+    "exhausted reasonable attempts and a node explains why not). If the Planner's test script "
+    "above is real (not N/A), materialize it as a real file under this project's own test "
+    "conventions and make it pass for real -- it is not optional, it's part of this goal's "
+    "acceptance bar; the Planner's script is a floor, write additional tests of your own if it "
+    "doesn't cover everything the SDD requires."
+)
+
 _VALIDATOR_PROMPT = (
     "Independently verify the implementation for this goal against the Benchmark below: run "
     "the tests yourself, don't just trust the Coder's claims. Reply with a verdict line "
@@ -486,8 +518,30 @@ def _failed_node_ids(exc):
     return set()
+def _resolve_node_artifact_type(node, node_artifact_types):
+    """A node promotes to an `artifact` row if EITHER: it's in the caller's
+    precomputed `node_artifact_types` dict (Planner's/Validator's fixed
+    catalogs, where every node is a promotable artifact by construction --
+    unchanged behavior), OR it declares `artifact_type` directly on
+    itself, OR (2026-08-12, Phase 2 §2.3) it declares `promotes_to_
+    artifact: {"artifact_type": ...}` (Coder's/Validator's generic
+    self-planned nodes -- most of which should declare NEITHER, meaning
+    "this is an internal graph step, not a final deliverable"). Returns
+    None for a node that isn't promoting anything -- not an error, the
+    expected case for most self-planned nodes."""
+    if node["node_id"] in node_artifact_types:
+        return node_artifact_types[node["node_id"]]
+    if "artifact_type" in node:
+        return node["artifact_type"]
+    promotes = node.get("promotes_to_artifact")
+    if promotes:
+        return promotes.get("artifact_type")
+    return None
+
+
 def _write_completed_node_artifacts(
-    conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, run_graph_error=None,
+    conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for,
+    run_graph_error=None, only_node_ids=None,
 ):
     """Writes an artifact for every node that's genuinely done AND
     verify_fn-accepted, sourced from `work_node_execution` in the DB --
@@ -511,6 +565,15 @@ def _write_completed_node_artifacts(
     nodes are excluded outright: their row is schema-valid ('done') but
     verify_fn never accepted either attempt, so neither is trustworthy
     content to persist.
+
+    2026-08-12 (Phase 2 §2.3/§6): a node with no resolvable artifact_type
+    (see `_resolve_node_artifact_type`) is silently skipped, not an
+    error -- most Coder/Validator self-planned nodes are internal graph
+    steps, not final deliverables. `only_node_ids` (default None = every
+    node in `graph_manifest`, unchanged behavior) restricts processing to
+    a subset -- needed once a caller works over a MERGED graph (Coder's
+    `append_to_run_id`) and must avoid re-processing/re-promoting nodes an
+    EARLIER stage already wrote artifacts for.
     """
     done_rows = conn.execute(
         "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'",
@@ -529,7 +592,11 @@ def _write_completed_node_artifacts(
         node_id = node["node_id"]
         if node_id not in done_outputs:
             continue
-        artifact_type = node_artifact_types[node_id]
+        if only_node_ids is not None and node_id not in only_node_ids:
+            continue
+        artifact_type = _resolve_node_artifact_type(node, node_artifact_types)
+        if artifact_type is None:
+            continue
         deps = [
             artifact_ids[node_artifact_types[dep_id]]
             for dep_id in (node.get("depends_on") or [])
@@ -743,7 +810,19 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_
             artifact_ids = task_execution._latest_artifact_ids(
                 conn, task_id, task_execution._DOC_ARTIFACT_TYPES
             )
-            plan_result = {"run_id": None, "artifact_ids": artifact_ids}
+            # Phase 2 §6.2 (2026-08-12, found while wiring implement_code()'s
+            # new append_to_run_id requirement): this used to hardcode
+            # run_id=None -- harmless for Phase 1 (nothing downstream read
+            # it), but Phase 2's Coder self-planning needs a REAL run_id to
+            # append into. Every real artifact row already has one
+            # (produced_by_run_id is NOT NULL) -- read it off the sdd
+            # artifact (required a few lines below anyway, guaranteed
+            # present whenever `existing` is truthy) rather than inventing
+            # a new lookup path.
+            sdd_run_id = conn.execute(
+                "SELECT produced_by_run_id FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],)
+            ).fetchone()[0]
+            plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids}
         else:
             plan_result = plan_documents(conn, task_id, goal_text + feedback_suffix, target_folder_path)
             artifact_ids = plan_result["artifact_ids"]
@@ -792,57 +871,124 @@ def plan_and_review_documents(conn, task_id, goal_text, target_folder_path, time
 def implement_code(
-    conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200,
+    conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200,
     benchmark_content="N/A: not provided", test_dataset_content="N/A: not provided",
     test_script_content="N/A: not provided",
 ):
-    """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.
+    """Phase 2 §6 (2026-08-12): Coder now self-plans its own implementation
+    graph via `plan_work_graph(..., append_to_run_id=run_id)` -- `run_id`
+    is Planner's own work_graph_run (created by `plan_and_review_
+    documents()`), no longer a fresh row this function creates itself.
+    Coder's self-planned nodes get appended into that SAME run, not a
+    separate one -- see work_graph_plan.py's own append_to_run_id
+    docstring for the merge mechanics. Replaces the old synthetic
+    single-node "implement" manifest (architecture §11 criterion 2's
+    original, pre-self-planning shape).
+
+    `code_diff` stays a Python-computed artifact (a real `git diff` across
+    whatever Coder's whole self-planned graph actually changed on disk,
+    end to end) -- not something any self-planned node's own `content`
+    field could meaningfully hold, since no node knows the diff ahead of
+    running it. `test_report` is promoted from whichever self-planned
+    node(s) declared `promotes_to_artifact: {"artifact_type":
+    "test_report"}` (the catalog's own "integration_check" entry suggests
+    this); if the LLM's plan didn't include one (the catalog is
+    suggestions, not requirements), falls back to concatenating every
+    self-planned node's own content in topological order -- same spirit
+    as the old single-node manifest's `response` field always being
+    Coder's own narrative of what it did, just synthesized instead of
+    directly given by one node.
+
+    `needs_worktree=True` passed to run_graph() (2026-08-12, found while
+    doing this rework -- NOT previously passed here at all, even though
+    Coder is the one org_node that actually writes real code): with a
+    single synthetic node this was harmless (nothing to collide with
+    inside one run), but a genuinely self-planned graph can have 2+
+    code-writing nodes in the SAME topological layer running concurrently
+    -- 03-C §3.6's "并行写代码的节点必须先申请 git_worktree_ref" applies for
+    real now, not just in theory.
     2026-08-06: benchmark_content/test_dataset_content/test_script_content
     (default 'N/A: not provided' for callers that predate this, e.g. a
-    resumed task from before this change) are 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.
+    resumed task from before this change) are folded into
+    _CODER_PLANNING_CONTEXT so Coder's planning call sees the Planner's
+    real test script and must make it pass for real, not just write
+    whatever tests it invents on its own.
     """
     before_sha = _current_commit(target_folder_path)
-    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,
+    planning_context = _CODER_PLANNING_CONTEXT.format(
+        goal_text=goal_text, sdd_content=sdd_content, benchmark_content=benchmark_content,
+        test_dataset_content=test_dataset_content, test_script_content=test_script_content,
+    )
+    plan_result = work_graph_plan.plan_work_graph(
+        conn, "coder", task_id, planning_context, target_folder_path,
+        timeout=300, append_to_run_id=run_id,
+    )
+    graph_manifest = plan_result["graph_manifest"]
+    new_node_ids = set(plan_result["new_node_ids"])
+
+    # Covers BOTH node shapes that can now coexist in this merged graph:
+    # Planner's own nodes (artifact_type) and Coder's just-appended ones
+    # (promotes_to_artifact) -- used for both _planner_prompt_builder's
+    # prompt construction (only its own artifact_type entries are ever
+    # queried) and _write_completed_node_artifacts' artifact-type/deps
+    # resolution below.
+    node_artifact_types = {}
+    for n in graph_manifest["nodes"]:
+        if "artifact_type" in n:
+            node_artifact_types[n["node_id"]] = n["artifact_type"]
+        elif n.get("promotes_to_artifact"):
+            node_artifact_types[n["node_id"]] = n["promotes_to_artifact"]["artifact_type"]
+
+    build_prompt = work_graph_plan.dispatch_build_prompt(
+        _planner_prompt_builder(goal_text, node_artifact_types), work_graph_plan.generic_node_prompt_builder,
     )
-    response = outputs["implement"]["response"]
+    run_graph_error = None
+    outputs = None
+    try:
+        outputs = work_graph_orchestrate.run_graph(
+            conn, run_id, "coder", "coder-implementation", target_folder_path, graph_manifest,
+            build_prompt, _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA, timeout=timeout,
+            needs_worktree=True, mechanical_check_registry={"check_script_compiles": check_script_compiles},
+        )
+    except Exception as e:
+        run_graph_error = e
     diff = subprocess.run(
         ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True
     ).stdout
-
     diff_result = artifact_write.write_artifact(
         conn, target_folder_path, "code_diff", task_id, run_id,
         ".graphlemon/runs/{}/code_diff.patch".format(task_id), diff or "(no diff -- see test_report)",
     )
-    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,
+
+    artifact_ids = _write_completed_node_artifacts(
+        conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
+        _planner_ext_for, run_graph_error, only_node_ids=new_node_ids,
     )
-    return {"code_diff": diff_result["artifact_id"], "test_report": report_result["artifact_id"]}
+
+    if run_graph_error is not None:
+        raise run_graph_error
+
+    if "test_report" not in artifact_ids:
+        # The self-planned graph didn't include a node declaring
+        # promotes_to_artifact: {"artifact_type": "test_report"} --
+        # synthesize one so callers can still rely on this key always
+        # being present (same contract the old single-node manifest gave
+        # unconditionally via its one node's `response` field).
+        report_content = "\n\n---\n\n".join(
+            "### {}\n\n{}".format(node_id, (outputs or {}).get(node_id, {}).get("content", ""))
+            for node_id in new_node_ids
+            if node_id in (outputs or {})
+        ) or "(Coder's self-planned graph produced no synthesizable report content)"
+        report_result = artifact_write.write_artifact(
+            conn, target_folder_path, "test_report", task_id, run_id,
+            ".graphlemon/runs/{}/test_report.md".format(task_id), report_content,
+        )
+        artifact_ids["test_report"] = report_result["artifact_id"]
+
+    return {"code_diff": diff_result["artifact_id"], "test_report": artifact_ids["test_report"]}
 def _parse_verdict_line(response):
@@ -1093,7 +1239,7 @@ def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
     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,
+        conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path,
         benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
         test_script_content=test_script_content,
     )
@@ -165,7 +165,7 @@ def run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text):
         _check_not_paused(conn, task_id)
         org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, e2e_pipeline._READY_PROMPT)
         code_artifact_ids = e2e_pipeline.implement_code(
-            conn, task_id, sdd_content, goal_text, target_folder_path,
+            conn, task_id, plan_result["run_id"], sdd_content, goal_text, target_folder_path,
             benchmark_content=benchmark_content, test_dataset_content=test_dataset_content,
             test_script_content=test_script_content,
         )
@@ -99,6 +99,56 @@ def _fake_plan_subprocess_run(*args, **kwargs):
     return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
+# Phase 2 §6 (2026-08-12): Coder now self-plans too (implement_code() calls
+# plan_work_graph(conn, "coder", ..., append_to_run_id=...), going through
+# the SAME mocked subprocess boundary as Planner's own planning call) --
+# depends_on Planner's sdd node so it lands in a LATER topological layer
+# than Planner's already-'done' nodes, not the same one (both being
+# in-degree-zero would trigger run_graph()'s concurrent ThreadPoolExecutor
+# layer path purely as a mocking artifact, unrelated to what this test is
+# actually checking).
+_FAKE_CODER_GRAPH_MANIFEST = {
+    "nodes": [{
+        "node_id": "impl", "depends_on": ["sdd"], "task_prompt": "implement it",
+        "promotes_to_artifact": {"artifact_type": "test_report"},
+    }],
+    "edges": [],
+}
+
+
+_real_subprocess_run = subprocess.run  # captured before any test patches subprocess.run globally
+
+
+def _fake_plan_subprocess_run_dispatch(*args, **kwargs):
+    """Real plan_work_graph() dispatches by the --agent argv value -- this
+    mock does the same, so both Planner's and Coder's planning calls (which
+    now share the same mocked subprocess.run patch target) each get their
+    own realistic fake manifest instead of one hardcoded response serving
+    both callers (which would hand Coder Planner's own brd/prd/... nodes,
+    colliding with the ones already in the run it's appending onto).
+
+    patch("src.work_graph_plan.subprocess.run", ...) patches subprocess.run
+    GLOBALLY, not just within work_graph_plan.py -- `subprocess` is the same
+    singleton module object everywhere it's imported, so this side_effect
+    also intercepts every OTHER real subprocess.run call this test's own
+    call chain makes (artifact_write.write_artifact's `git add`/`git
+    commit`, e2e_pipeline's `git diff`/`git rev-parse`, etc). Those aren't
+    `claude --agent ... -p ...` calls -- pass them through to the REAL
+    subprocess.run instead of returning a canned fake "claude planning
+    response" CompletedProcess for them too (which happened to not visibly
+    break anything before this dispatch function existed, only because
+    nothing downstream read those specific calls' .stdout in a way that
+    exposed it -- still wrong to keep doing on purpose now that this
+    function actually inspects argv)."""
+    argv = args[0]
+    if not argv or argv[0] != "claude":
+        return _real_subprocess_run(*args, **kwargs)
+    org_node = argv[argv.index("--agent") + 1]
+    manifest = _FAKE_CODER_GRAPH_MANIFEST if org_node == "coder" else _FAKE_GRAPH_MANIFEST
+    payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
+    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
+
+
 def _fake_plan_subprocess_run_with_test_script(*args, **kwargs):
     payload = {
         "is_error": False,
@@ -117,7 +167,11 @@ class PipelineTestBase(unittest.TestCase):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)
         subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
-        self.conn = schema.connect(":memory:")
+        # check_same_thread=False (2026-08-12, Phase 2 §6) -- see
+        # test_task_execution.py's TaskExecutionTestBase for why: a merged
+        # graph's concurrent-layer path can now be reached even in tests
+        # that don't otherwise care about G's parallel execution.
+        self.conn = schema.connect(":memory:", check_same_thread=False)
         schema.create_schema(self.conn)
         project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
@@ -125,6 +179,94 @@ class PipelineTestBase(unittest.TestCase):
         self.conn.close()
+class WriteCompletedNodeArtifactsTest(PipelineTestBase):
+    """Phase 2 §2.3/§6 (2026-08-12) direct coverage: promotes_to_artifact
+    resolution and only_node_ids scoping, tested directly rather than only
+    incidentally through plan_documents()/implement_code() integration
+    tests."""
+
+    def setUp(self):
+        super().setUp()
+        self.conn.execute(
+            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
+            "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', 'coding', 'coder')"
+        )
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES ('run-1', 'task-1', 'coder', '{}', 'running')"
+        )
+        self.conn.execute(
+            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+            "VALUES ('coder-implementation', 'coder', 'skills/coder-implementation/', 'unknown', 'new', 'active')"
+        )
+        self.conn.commit()
+
+    def _mark_done(self, node_id, content):
+        self.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 (?, 'run-1', ?, 'coder', 'coder-implementation', 'in.json', 'out.json', '{}', ?, 'done')",
+            ("node-exec-" + node_id, node_id, json.dumps({"content": content})),
+        )
+        self.conn.commit()
+
+    def test_node_with_no_artifact_type_or_promotes_to_artifact_is_skipped(self):
+        # most self-planned Coder nodes are internal graph steps, not final
+        # deliverables -- must be silently skipped, not an error.
+        manifest = {"nodes": [{"node_id": "explore", "depends_on": []}], "edges": []}
+        self._mark_done("explore", "some intermediate reasoning")
+        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
+            self.conn, self.tmp.name, "task-1", "run-1", manifest, {}, lambda t: "md",
+        )
+        self.assertEqual(artifact_ids, {})
+        count = self.conn.execute("SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(count, 0)
+
+    def test_promotes_to_artifact_node_is_persisted(self):
+        manifest = {
+            "nodes": [{
+                "node_id": "final_report", "depends_on": [],
+                "promotes_to_artifact": {"artifact_type": "test_report"},
+            }],
+            "edges": [],
+        }
+        self._mark_done("final_report", "# Real report content")
+        node_artifact_types = {"final_report": "test_report"}
+        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
+            self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
+        )
+        self.assertEqual(set(artifact_ids), {"test_report"})
+        content = self.conn.execute(
+            "SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_ids["test_report"],)
+        ).fetchone()[0]
+        with open(os.path.join(self.tmp.name, content)) as f:
+            self.assertEqual(f.read(), "# Real report content")
+
+    def test_only_node_ids_scopes_to_new_nodes_skipping_earlier_promotable_ones(self):
+        # a merged graph (Planner's already-artifact'd nodes + Coder's just-
+        # appended ones) must not re-promote/re-process nodes an EARLIER
+        # stage already wrote artifacts for, even though they'd otherwise
+        # be individually promotable.
+        manifest = {
+            "nodes": [
+                {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []},
+                {
+                    "node_id": "impl", "depends_on": ["sdd"],
+                    "promotes_to_artifact": {"artifact_type": "test_report"},
+                },
+            ],
+            "edges": [],
+        }
+        self._mark_done("sdd", "# SDD content")
+        self._mark_done("impl", "# Report content")
+        node_artifact_types = {"sdd": "sdd", "impl": "test_report"}
+        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
+            self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
+            only_node_ids={"impl"},
+        )
+        self.assertEqual(set(artifact_ids), {"test_report"})  # NOT "sdd" -- outside only_node_ids
+
+
 class PlanDocumentsTest(PipelineTestBase):
     def test_produces_five_artifacts_with_correct_lineage(self):
         self.conn.execute(
@@ -228,7 +370,7 @@ class PlanDocumentsTest(PipelineTestBase):
 class RunEndToEndTest(PipelineTestBase):
     def test_full_pipeline_wiring(self):
         with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \
-             patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
+             patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run_dispatch), \
              patch.object(work_graph_execute, "write_workflow_script"), \
              patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
              patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."):
@@ -237,7 +379,11 @@ class RunEndToEndTest(PipelineTestBase):
                 {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                 {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                 {"content": _FAKE_DOCS["test_dataset"]},
-                {"response": "Implemented. Ran `pytest`: 3 passed."},
+                # Phase 2 §6 (2026-08-12): Coder self-plans now -- one
+                # self-planned "impl" node (_FAKE_CODER_GRAPH_MANIFEST),
+                # content-keyed like every other self-planned node, not the
+                # old single synthetic node's response-keyed output.
+                {"content": "Implemented. Ran `pytest`: 3 passed."},
                 # Validator's OWN independent benchmark/dataset (2026-08-06) --
                 # deliberately different content from the Planner's, to also prove
                 # nothing here accidentally reuses the Planner's copies.
@@ -265,15 +411,31 @@ class RunEndToEndTest(PipelineTestBase):
         # (test_script/validator_test_script disabled, see _VALIDATOR_MANIFEST)
         self.assertEqual(all_artifacts, 9)
-        # planner, coder, AND validator (2026-08-06: validator now has its own
-        # real work_graph_run too, not a flat direct call) each got their own
-        # work_graph_run (not sharing one)
-        run_org_nodes = {
+        # Phase 2 §6 (2026-08-12): Coder no longer gets its own work_graph_run
+        # -- implement_code() appends its self-planned nodes onto Planner's
+        # SAME run (plan_work_graph(..., append_to_run_id=...)), so only TWO
+        # rows exist now, not three: the shared Planner+Coder run (still
+        # labeled org_node='planner' -- that column reflects who ORIGINATED
+        # the run, not every org_node that has since contributed nodes to it;
+        # Phase 2c is expected to fold Validator into the same run too, at
+        # which point this column stops being a reliable "who's involved"
+        # signal at all, see design doc §2.1) and Validator's own separate run.
+        run_rows = self.conn.execute(
+            "SELECT run_id, org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
+        ).fetchall()
+        self.assertEqual(len(run_rows), 2)
+        self.assertEqual({row[1] for row in run_rows}, {"planner", "validator"})
+
+        # Coder's self-planned "impl" node genuinely landed in the SAME run
+        # as Planner's nodes, not a separate one.
+        planner_run_id = next(row[0] for row in run_rows if row[1] == "planner")
+        node_ids = {
             row[0] for row in self.conn.execute(
-                "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
+                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (planner_run_id,)
             ).fetchall()
         }
-        self.assertEqual(run_org_nodes, {"planner", "coder", "validator"})
+        self.assertIn("impl", node_ids)
+        self.assertIn("sdd", node_ids)
 class LooksLikeNarrationTest(unittest.TestCase):
@@ -437,7 +599,7 @@ class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase):
     def test_prd_and_benchmark_missing_does_not_raise_key_error(self):
         captured = {}
-        def fake_implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, **kwargs):
+        def fake_implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, **kwargs):
             captured["benchmark_content"] = kwargs.get("benchmark_content")
             return {"code_diff": "diff-1", "test_report": "report-1"}
@@ -14,7 +14,10 @@ import threading
 import unittest
 from unittest.mock import patch
-from src import e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, task_execution, work_graph_execute
+from src import (
+    e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler,
+    task_execution, work_graph_execute, work_graph_plan,
+)
 def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
@@ -22,12 +25,46 @@ def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
     return "Ready."
+# Phase 2 §6 (2026-08-12): implement_code() now self-plans via
+# plan_work_graph(conn, "coder", ..., append_to_run_id=...) instead of
+# running one hardcoded synthetic node -- these tests aren't about self-
+# planning mechanics (that's test_e2e_pipeline.py's job), so this fakes
+# the planning subprocess with the smallest possible one-node manifest,
+# keeping the rest of each test's mock_execute.side_effect shape close to
+# what it was before this change (one node execution slot for Coder,
+# {"content": ...} instead of the old single-node manifest's {"response": ...}).
+_FAKE_CODER_MANIFEST = {
+    "nodes": [{
+        "node_id": "impl", "depends_on": [], "task_prompt": "implement it",
+        "promotes_to_artifact": {"artifact_type": "test_report"},
+    }],
+    "edges": [],
+}
+
+
+def _fake_coder_plan_subprocess_run(*args, **kwargs):
+    payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_CODER_MANIFEST) + "\n```"}
+    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
+
+
 class TaskExecutionTestBase(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)
         subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
-        self.conn = schema.connect(":memory:")
+        # check_same_thread=False (2026-08-12, Phase 2 §6): a merged graph
+        # can legitimately put an already-'done' node (Planner's, short-
+        # circuiting instantly) in the SAME topological layer as a newly-
+        # appended node with no depends_on between them (e.g. Coder's own
+        # self-planned node happens not to depend on Planner's) --
+        # work_graph_orchestrate.run_graph()'s multi-node-layer path always
+        # uses a ThreadPoolExecutor once len(pending) > 1, regardless of how
+        # fast one of them finishes. Real production connections already do
+        # this (scheduler.py/task_execution.py's own start_task_execution_
+        # in_background, same "G" reasoning) -- this test fixture just
+        # hadn't needed it before self-planning could produce concurrent
+        # layers spanning old + new nodes.
+        self.conn = schema.connect(":memory:", check_same_thread=False)
         schema.create_schema(self.conn)
         project = project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
         self.project_id = project["project_id"]
@@ -45,9 +82,18 @@ class TaskExecutionTestBase(unittest.TestCase):
     def _write_artifact(self, task_id, artifact_type, content):
         run_id = "run-" + artifact_type
+        # Phase 2 §6 (2026-08-12): graph_manifest is no longer a throwaway
+        # '{}' -- implement_code() now reads it back (plan_work_graph's
+        # append_to_run_id mode) whenever it appends onto the run the sdd
+        # artifact points at, so it needs a real "nodes"/"edges" shape, not
+        # an empty object real production data would never actually have.
+        manifest = json.dumps({
+            "nodes": [{"node_id": artifact_type, "artifact_type": artifact_type, "depends_on": []}],
+            "edges": [],
+        })
         self.conn.execute(
             "INSERT OR IGNORE INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-            "VALUES (?, ?, 'planner', '{}', 'done')", (run_id, task_id),
+            "VALUES (?, ?, 'planner', ?, 'done')", (run_id, task_id, manifest),
         )
         rel_path = ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type)
         abs_path = os.path.join(self.tmp.name, rel_path)
@@ -122,12 +168,13 @@ class ResumeFromPlanningTest(TaskExecutionTestBase):
         self._write_artifact("task-1", "benchmark", "# Benchmark content")
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
              patch.object(work_graph_execute, "write_workflow_script"), \
              patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
              patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \
              patch.object(e2e_pipeline, "plan_documents") as mock_plan:
             mock_execute.side_effect = [
-                {"response": "done, tests pass"},  # coder's implement node
+                {"content": "done, tests pass"},  # coder's self-planned "impl" node
                 # validator's OWN benchmark/test_dataset/test_script graph (2026-08-06)
                 {"content": "# Validator's own Benchmark"},
                 {"content": "N/A: no dataset needed"},
@@ -169,12 +216,13 @@ class ResumeFromReviewingTest(TaskExecutionTestBase):
             return _fake_ready_trigger(org_node, prompt, target_folder_path, timeout)
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=tracking_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
              patch.object(work_graph_execute, "write_workflow_script"), \
              patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
              patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \
              patch.object(e2e_pipeline, "plan_documents") as mock_plan:
             mock_execute.side_effect = [
-                {"response": "done, tests pass"},  # coder's implement node
+                {"content": "done, tests pass"},  # coder's self-planned "impl" node
                 {"content": "# Validator's own Benchmark"},
                 {"content": "N/A: no dataset needed"},
                 {"content": "import sys\nsys.exit(0)\n"},
@@ -209,16 +257,22 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase):
             # writes a real, validation_status='passed' sdd artifact, since
             # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] checks the real
             # `artifact` table, not just this function's Python return value.
+            # run_id matches what _write_artifact's own test fixture convention
+            # actually creates ("run-" + artifact_type) -- a real plan_documents()
+            # call's return value would likewise point at the real work_graph_run
+            # row it just inserted, which Phase 2's implement_code() now needs to
+            # append into.
             self._write_artifact(task_id, "sdd", "# SDD\nfreshly regenerated")
-            return {"run_id": "run-x", "artifact_ids": {"sdd": "art-sdd"}}
+            return {"run_id": "run-sdd", "artifact_ids": {"sdd": "art-sdd"}}
         with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \
+             patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_coder_plan_subprocess_run), \
              patch.object(work_graph_execute, "write_workflow_script"), \
              patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
              patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"), \
              patch.object(e2e_pipeline, "plan_documents", side_effect=fake_plan_documents) as mock_plan:
             mock_execute.side_effect = [
-                {"response": "done, tests pass"},  # coder's implement node
+                {"content": "done, tests pass"},  # coder's self-planned "impl" node
                 {"content": "# Validator's own Benchmark"},
                 {"content": "N/A: no dataset needed"},
                 {"content": "import sys\nsys.exit(0)\n"},
@@ -139,6 +139,50 @@ class ForkFromPrdAcceptanceCriterion1Test(ForkTestBase):
         self.assertEqual(run_row, ("done", "run-orig", node_exec_ids["prd"]))
+class ForkFromMixedShapeGraphTest(ForkTestBase):
+    """Phase 2 §2.1.2 (2026-08-12) regression: a graph_manifest can now
+    genuinely mix Planner-shaped nodes ('artifact_type') and Coder-shaped
+    nodes ('task_prompt') side by side, once plan_work_graph(...,
+    append_to_run_id=...) appends Coder's self-planned nodes onto
+    Planner's original run. Forking from the artifact_type node must
+    still correctly reconstruct a build_prompt for the DOWNSTREAM
+    task_prompt node -- before this fix, _reconstruct_build_prompt
+    required the WHOLE graph to be one uniform shape and would have
+    raised ForkGraphShapeUnsupportedError here."""
+
+    def test_downstream_generic_node_reexecutes_after_forking_upstream_planner_node(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []},
+                {
+                    "node_id": "impl", "depends_on": ["sdd"], "task_prompt": "implement per the sdd",
+                    "promotes_to_artifact": {"artifact_type": "test_report"},
+                },
+            ],
+            "edges": [],
+        }
+        contents = {"sdd": {"content": "original sdd"}, "impl": {"content": "original impl"}}
+        node_exec_ids = self._seed_completed_run("run-orig", "planner", manifest, contents)
+
+        execute_calls = []
+
+        def fake_execute(cwd, name, args, output_schema, timeout=600):
+            execute_calls.append(name)
+            return {"content": "regenerated for " + name}
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute):
+            result = work_graph_fork.fork_work_graph_run(
+                self.conn, node_exec_ids["sdd"], {"content": "a completely different sdd"}, "/tmp/proj-1",
+            )
+
+        # only "impl" (downstream of the fork point) genuinely re-ran.
+        self.assertEqual(len(execute_calls), 1)
+        outputs = result["outputs"]
+        self.assertEqual(outputs["sdd"], {"content": "a completely different sdd"})
+        self.assertNotEqual(outputs["impl"], {"content": "original impl"})
+
+
 class OriginalRunUnaffectedTest(ForkTestBase):
     """Acceptance standard 2: the original run's records are completely
     untouched by a fork."""
@@ -399,6 +399,129 @@ class GenericPlanningPromptTest(PlanWorkGraphTestBase):
         self.assertNotIn("BRD, PRD, SDD, Benchmark", prompt)
         self.assertIn("task_prompt", prompt)
+    def test_coder_prompt_includes_its_own_suggested_catalog_not_validators(self):
+        # Phase 2 §3 (2026-08-12): each org_node gets its OWN suggested
+        # node-type catalog, not a shared one -- "starting points, not
+        # requirements", same contract _PLANNER_NODE_CATALOG already uses.
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertIn("Suggested node types for this goal", prompt)
+        self.assertIn('"node_type": "unit_test"', prompt)
+        self.assertIn('"node_type": "integration_check"', prompt)
+        self.assertNotIn('"node_type": "independent_benchmark"', prompt)  # validator-only
+        self.assertIn("promotes_to_artifact", prompt)
+        self.assertIn("dynamic_loop_back", prompt)  # explicitly told NOT to use it
+
+    def test_validator_prompt_includes_its_own_catalog_not_coders(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "validator", "task-1", "goal", "/tmp/proj-1")
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertIn('"node_type": "independent_benchmark"', prompt)
+        self.assertIn('"node_type": "test_script"', prompt)
+        self.assertNotIn('"node_type": "unit_test"', prompt)  # coder-only
+
+    def test_unknown_org_node_gets_no_catalog_section_at_all(self):
+        # An org_node this hasn't been extended for yet (Phase 1's
+        # planner_reviewer, for instance) still gets the generic path --
+        # just with no suggested catalog, same as before catalogs existed.
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "planner_reviewer", "task-1", "goal", "/tmp/proj-1")
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertNotIn("Suggested node types for this goal", prompt)
+        self.assertIn("task_prompt", prompt)  # still the generic template itself
+
+
+class AppendToRunIdTest(PlanWorkGraphTestBase):
+    """Phase 2 §6.1 (2026-08-12): plan_work_graph(..., append_to_run_id=...)
+    merges new nodes into an EXISTING work_graph_run instead of INSERTing a
+    fresh row."""
+
+    def _seed_run(self, run_id, manifest):
+        self.conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES (?, 'task-1', 'planner', ?, 'running')",
+            (run_id, json.dumps(manifest)),
+        )
+        self.conn.commit()
+
+    def test_merges_new_nodes_into_existing_run_not_a_new_row(self):
+        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
+        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []}
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
+            )
+            result = work_graph_plan.plan_work_graph(
+                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
+            )
+
+        self.assertEqual(result["run_id"], "run-1")
+        self.assertEqual(result["new_node_ids"], ["impl"])
+        merged_node_ids = {n["node_id"] for n in result["graph_manifest"]["nodes"]}
+        self.assertEqual(merged_node_ids, {"sdd", "impl"})
+
+        # exactly one work_graph_run row for this task -- no second row created.
+        row_count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'").fetchone()[0]
+        self.assertEqual(row_count, 1)
+        persisted = json.loads(
+            self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
+        )
+        self.assertEqual({n["node_id"] for n in persisted["nodes"]}, {"sdd", "impl"})
+
+    def test_existing_node_ids_are_injected_into_the_prompt(self):
+        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
+        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []}
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
+            )
+            work_graph_plan.plan_work_graph(
+                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
+            )
+        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
+        self.assertIn("already has these nodes", prompt)
+        self.assertIn("sdd", prompt)
+        self.assertIn("ADDITIONAL nodes only", prompt)
+
+    def test_node_id_collision_with_existing_node_raises(self):
+        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
+        # the "new" plan reuses "sdd" as its own node_id -- a real collision.
+        colliding_manifest = {"nodes": [{"node_id": "sdd", "depends_on": [], "task_prompt": "oops"}], "edges": []}
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(colliding_manifest)}
+            )
+            with self.assertRaises(work_graph_plan.PlanningFailedError):
+                work_graph_plan.plan_work_graph(
+                    self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
+                )
+        # the original run's manifest must be untouched by the failed append.
+        persisted = json.loads(
+            self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
+        )
+        self.assertEqual([n["node_id"] for n in persisted["nodes"]], ["sdd"])
+
+    def test_append_to_nonexistent_run_id_raises(self):
+        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": [], "task_prompt": "do it"}], "edges": []}
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
+            )
+            with self.assertRaises(ValueError):
+                work_graph_plan.plan_work_graph(
+                    self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="no-such-run",
+                )
+
 class CoderThreeIndependentModulesScenarioTest(PlanWorkGraphTestBase):
     """E acceptance criterion 1: a Coder scenario that naturally splits into
@@ -94,23 +94,46 @@ def _downstream_node_ids(graph_manifest, start_node_id):
 def _reconstruct_build_prompt(conn, task_id, org_node, graph_manifest):
     """Rebuilds a build_prompt(node, outputs) usable for the re-executed
     downstream subgraph, purely from persisted state -- no access to the
-    original call site's Python closures. Two supported shapes (design doc
-    §2): Planner's canonical doc chain (every node has 'artifact_type',
+    original call site's Python closures. Two supported node shapes
+    (design doc §2): Planner's canonical doc chain ('artifact_type',
     prompt reconstructed via e2e_pipeline's own builder + task.goal_text)
-    and E's generic dynamic path (every node carries its own
-    self-contained 'task_prompt'). Anything else raises rather than
-    silently building broken prompts."""
+    and E's generic dynamic path ('task_prompt'). A node with neither
+    raises rather than silently building a broken prompt.
+
+    2026-08-12 (Phase 2 §2.1.2, found while designing append_to_run_id):
+    this used to require the WHOLE graph to be one uniform shape (all
+    nodes 'artifact_type', or all 'task_prompt') -- correct when every
+    graph_manifest genuinely was one shape throughout, which stopped being
+    true the moment Coder's self-planned nodes (task_prompt) can get
+    appended onto Planner's original run (artifact_type) via
+    plan_work_graph(..., append_to_run_id=...). Dispatches per-NODE now
+    (work_graph_plan.dispatch_build_prompt) instead of once for the whole
+    graph -- every pre-Phase-2 graph_manifest still has just one shape
+    throughout, so this is a strict generalization, not a behavior change
+    for anything that isn't a merged graph. `org_node` is unused by this
+    per-node dispatch itself (kept as a parameter -- callers still pass it
+    for other purposes, e.g. run_graph()'s own org_node argument;
+    removing it from work_graph_run/this signature is Phase 2c's job
+    alongside the actual org_node column removal, not this fix's)."""
     nodes = graph_manifest["nodes"]
-    if org_node == "planner" and nodes and all("artifact_type" in n for n in nodes):
-        goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
-        node_artifact_types = {n["node_id"]: n["artifact_type"] for n in nodes}
-        return e2e_pipeline._planner_prompt_builder(goal_text, node_artifact_types)
-    if nodes and all("task_prompt" in n for n in nodes):
-        return work_graph_plan.generic_node_prompt_builder
-    raise ForkGraphShapeUnsupportedError(
-        "graph_manifest nodes must either all have 'artifact_type' (Planner's canonical chain) "
-        "or all have 'task_prompt' (the generic dynamic-planning path) to reconstruct prompts for a fork"
-    )
+    if not nodes:
+        raise ForkGraphShapeUnsupportedError("graph_manifest has no nodes to reconstruct a build_prompt for")
+    goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
+    node_artifact_types = {n["node_id"]: n["artifact_type"] for n in nodes if "artifact_type" in n}
+    planner_builder = e2e_pipeline._planner_prompt_builder(goal_text, node_artifact_types)
+    dispatch = work_graph_plan.dispatch_build_prompt(planner_builder, work_graph_plan.generic_node_prompt_builder)
+
+    def build_prompt(node, outputs):
+        try:
+            return dispatch(node, outputs)
+        except ValueError:
+            raise ForkGraphShapeUnsupportedError(
+                "node {} has neither 'artifact_type' (Planner's canonical chain) nor 'task_prompt' "
+                "(the generic dynamic-planning path) -- can't reconstruct a prompt for a fork".format(
+                    node.get("node_id")
+                )
+            )
+    return build_prompt
 def create_fork(conn, source_node_exec_id, new_output_payload):
@@ -41,10 +41,17 @@ GRAPH_MANIFEST_SCHEMA = {
                 "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
@@ -60,10 +67,20 @@ GRAPH_MANIFEST_SCHEMA = {
                     "on_fail": {
                         "type": "object",
                         "properties": {
-                            "action": {"type": "string", "enum": ["retry_self", "loop_back", "hard_fail"]},
+                            "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"},
                 },
             },
         },
@@ -126,13 +143,19 @@ _PLANNER_NODE_CATALOG = [
     },
 ]
-# NOTE: no `str.format()` involved in building or filling this template
-# (unlike _GENERIC_PROMPT_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
+# 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 "
@@ -163,29 +186,107 @@ _PROMPT_TEMPLATE = (
 # _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, whereas Coder/Validator's
-# sub-tasks have no such fixed small vocabulary of node types. 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.
-_GENERIC_PROMPT_TEMPLATE = (
-    "Plan (do not execute) a Work Graph for this goal: {goal_text}\n"
-    "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). 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."
+# 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"},
+    },
+]
+
+_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.",
+        "suggested_judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
+    },
+]
+
+_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).
+    node_type is a label in the SUGGESTION data only, not a required
+    graph_manifest node field -- node_id/task_prompt stay exactly as
+    freeform as before."""
+    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 ""
+    )
+    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"
+        "  - 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) -- do NOT use \"dynamic_loop_back\", that's reserved for the terminal review "
+        "stage outside this graph, not for nodes you plan yourself.\n"
+        "  - 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.)\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."
+    )
+
 _JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)
@@ -333,22 +434,61 @@ def _normalize_graph_manifest(manifest):
     return manifest
-def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300):
-    """Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row
-    (status='planning') -- moving it to 'running'/executing it is out of scope.
+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` -- 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. Function
-    signature/return shape are unchanged either way."""
+    `_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.format(goal_text=goal_text)
+        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(
         [
@@ -374,24 +514,48 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time
     payload = json.loads(result.stdout)
     if payload.get("is_error"):
         raise PlanningFailedError("plan failed for task {}: {}".format(task_id, payload))
-    graph_manifest = _extract_graph_manifest(payload.get("result", ""))
-    if not graph_manifest:
+    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))
-    graph_manifest = _normalize_graph_manifest(graph_manifest)
+    new_manifest = _normalize_graph_manifest(new_manifest)
+
+    if existing_manifest is None:
+        run_id = str(uuid.uuid4())
+        conn.execute(
+            "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
+            "VALUES (?, ?, ?, ?, 'planning')",
+            (run_id, task_id, org_node, json.dumps(new_manifest)),
+        )
+        conn.commit()
+        return {"run_id": run_id, "graph_manifest": new_manifest, "new_node_ids": []}
-    run_id = str(uuid.uuid4())
+    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(
-        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
-        "VALUES (?, ?, ?, ?, 'planning')",
-        (run_id, task_id, org_node, json.dumps(graph_manifest)),
+        "UPDATE work_graph_run SET graph_manifest = ? WHERE run_id = ?",
+        (json.dumps(merged), append_to_run_id),
     )
     conn.commit()
-    return {"run_id": run_id, "graph_manifest": graph_manifest}
+    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
+    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
@@ -399,3 +563,31 @@ def generic_node_prompt_builder(node, outputs):
     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