diff --git a/README.md b/README.md
index d4c4013..8aae312 100644
--- a/README.md
+++ b/README.md
@@ -46,16 +46,23 @@ board to a PR — or fork it as a starting point.
   requires a spec, EARS acceptance criteria, and explicit `files_to_modify`.
 - **Escalation (opt-in)** — with a `coders` map of >1 distinct delegate, a capability
   failure climbs a model tier (`fast→smart→reasoning`) and blocks at the top.
-- **coder.solve() board seam (ADR 0064 P2)** — on a fresh build, when the
+- **coder.solve() board seam (ADR 0064 P2/P3)** — on a fresh build, when the
   [`coder`](https://github.com/protoLabsAI/protoAgent/tree/main/plugins/coder) plugin
   is enabled AND the feature has acceptance criteria AND `coder_solve_test_cmd` (or
   `local_gate_cmd`) is set, the loop dispatches through `coder.solve()`'s
-  execution-grounded ladder (greedy → best-of-k → tree-search) instead of a single
-  `delegate_to(acp)` shot — gated on the feature's acceptance tests actually PASSING
-  in a real candidate worktree, never an LLM judge. Composes WITH the tier ladder
-  above (solve() searches *within* a tier; a search that never passes escalates a
-  tier, or blocks, exactly like a no-diff dispatch). Missing coder/acceptance/test
-  command ⇒ honest degrade to the single shot — see `coder_seam.py`.
+  execution-grounded ladder — greedy → best-of-k → tree-search → **fusion** — instead
+  of a single `delegate_to(acp)` shot, gated on the feature's acceptance tests
+  actually PASSING in a real candidate worktree, never an LLM judge. **Fusion** (rung
+  4, opt-in via `coder_solve_fusion_delegate`) is a richer *generator* for the
+  hardest features the cheaper rungs couldn't pass — it can't tool-call (a plain
+  completion, e.g. `protolabs/fusion`, not an ACP session), so `coder_seam.py` hands
+  it the current content of the feature's declared files and writes its reply's
+  files into a fresh worktree itself; the SAME `verify()` oracle judges it. Composes
+  WITH the tier ladder above (solve() searches *within* a tier; a search that never
+  passes escalates a tier, or blocks, exactly like a no-diff dispatch). Missing
+  coder/acceptance/test command ⇒ honest degrade to the single shot; missing
+  `coder_solve_fusion_delegate` ⇒ the ladder simply stops at tree-search — see
+  `coder_seam.py`.
 - **Planning layer** — two reasoning subagents (`decompose` + `antagonist`) driven by
   the `decompose-project` skill: idea → outline → MADR ADRs → epics › milestones ›
   features, hardened by an adversary, with a per-epic human gate.
@@ -114,6 +121,10 @@ project_board:
                              # acceptance criteria + a test command — see "What it does").
   coder_solve_test_cmd: "pytest tests/ -q"  # solve()'s verify() oracle; falls back to
                              # local_gate_cmd if blank, else the seam honest-degrades.
+  coder_solve_fusion_delegate: ""  # rung 4 (ADR 0064 P3), opt-in: an `openai`-type
+                             # delegate name (e.g. protolabs/fusion) for the hardest
+                             # features. Blank (default) = ladder stops at tree-search.
+  coder_solve_fusion_k: 2    # candidates fusion generates when reached
   # webhook_secret: "..."    # set before exposing /webhook/pr publicly
 ```
 
diff --git a/coder_seam.py b/coder_seam.py
index f9edc0a..a8d4c6d 100644
--- a/coder_seam.py
+++ b/coder_seam.py
@@ -1,7 +1,8 @@
 """The P2 board seam (ADR 0064): dispatch a feature's build through the `coder`
 plugin's execution-grounded ``solve()`` ladder instead of a single
-``delegate_to(acp)`` shot — greedy → best-of-k → tree-search, gated on the
-feature's acceptance tests actually PASSING in a real worktree, never an LLM judge.
+``delegate_to(acp)`` shot — greedy → best-of-k → tree-search → fusion, gated on
+the feature's acceptance tests actually PASSING in a real worktree, never an LLM
+judge.
 
 **Composes** `plugins.coder.solve` (a separate, git-URL-installed plugin — imported
 lazily/best-effort so this repo carries no hard dependency on it and no import-time
@@ -28,19 +29,96 @@
 the acceptance criteria as part of its definition of done; this module's ``verify``
 just RUNS whatever tests exist in a candidate's worktree via the configured command
 and gates on its exit code — real execution, no fabricated grounding.
-"""
+
+**Rung 4 — fusion (ADR 0064 P3).** Fusion (e.g. ``protolabs/fusion``) is a strong
+*generator* but, per the ADR, it **can't tool-call** — unlike the ``acp`` coder
+(a real edit/verify session in the worktree), it can only return a plain chat
+completion. So its candidate generation is a DIFFERENT shape from the ACP rungs:
+``_fusion_prompt`` hands it the task + the CURRENT content of the feature's
+declared ``files_to_modify`` (read from the base repo — fusion has no tool access
+to look these up itself) and asks for the complete, final content of every file it
+creates or changes; ``_parse_fusion_files`` extracts ``{path: content}`` from the
+reply; ``_WorktreeSolveAdapter.generate_fusion`` writes those files into a fresh
+worktree (the same throwaway-per-candidate discipline as the ACP rungs) and hands
+the path to the SAME ``verify()`` — real acceptance tests, same oracle, no separate
+judge. Wholesale file replacement (not a unified diff) is deliberate: an LLM
+completion reliably reproduces a full file; a hand-rolled patch with drifted
+context lines is a common failure mode `git apply` doesn't forgive. Only reached
+when a ``fusion_delegate`` is configured (an ``openai``-type Delegate, already
+resolved by the caller) — absent that, ``solve()`` gets ``fusion_generate=None``
+and stops at tree-search exactly as before (honest degrade, unchanged)."""
 
 from __future__ import annotations
 
 import asyncio
 import importlib
 import logging
+import re
+from pathlib import Path
 from typing import Callable
 
 from . import worktree
 
 log = logging.getLogger("protoagent.plugins.project_board")
 
+# ``### path/to/file.py`` header, then a fenced block (any/no language hint) holding
+# that file's COMPLETE new content. Deliberately simple/strict: a fusion completion
+# that doesn't follow the format parses to no files, which just fails verify() like
+# any other empty candidate — never a silent partial/mangled write.
+_FUSION_FILE_RE = re.compile(r"^###\s+(\S.+?)\s*$\n```[^\n]*\n(.*?)```", re.MULTILINE | re.DOTALL)
+
+_FUSION_READ_MAX_CHARS = 20_000  # per-file context cap — fusion sees enough to edit, not a dump
+
+
+def _parse_fusion_files(reply: str) -> dict[str, str]:
+    """Extract ``{relative path: full file content}`` from a fusion completion. No
+    match ⇒ empty dict — the caller writes nothing, and the untouched worktree just
+    fails ``verify()`` like any other candidate that didn't address the task."""
+    return {path.strip(): content for path, content in _FUSION_FILE_RE.findall(reply or "")}
+
+
+def _fusion_prompt(task: str, *, feedback: str | None, repo: str, files_to_modify: list[str]) -> str:
+    """Build fusion's prompt. Fusion can't read the repo itself (no tool-calling), so
+    this hands it the CURRENT content of every file the feature declares — read from
+    the base repo, best-effort (a listed-but-not-yet-created file is noted as new)."""
+    file_blocks = []
+    for rel in files_to_modify:
+        p = Path(repo) / rel
+        try:
+            text = p.read_text(errors="replace")[:_FUSION_READ_MAX_CHARS]
+            file_blocks.append(f"### {rel} (current content)\n```\n{text}\n```")
+        except OSError:
+            file_blocks.append(f"### {rel} (does not exist yet — you are creating it)")
+    files_section = (
+        "\n\n".join(file_blocks) if file_blocks else "(no existing files listed — create what the task needs)"
+    )
+    parts = [
+        "Implement the task below. You have NO tool access — you cannot read or run "
+        "anything else, so work only from what's given here.",
+        "",
+        "## Task",
+        task.strip(),
+        "",
+        "## Current file contents",
+        files_section,
+        "",
+        "## Your reply format — REQUIRED, exactly this shape per file",
+        "For every file you create or change, return its COMPLETE, FINAL content "
+        "(never a partial diff or `...` elisions) as:",
+        "### relative/path/to/file.py",
+        "```",
+        "<the file's entire new content>",
+        "```",
+        "Only include files you're actually creating or changing. No prose outside the file blocks.",
+    ]
+    if feedback:
+        parts += [
+            "",
+            "## Your previous attempt FAILED the acceptance tests — fix exactly this",
+            feedback.strip(),
+        ]
+    return "\n".join(parts)
+
 
 class SolveExhausted(worktree.WorktreeError):
     """``coder.solve()`` spent its whole generation budget against this feature's
@@ -118,6 +196,9 @@ def __init__(
         test_cmd: str,
         test_timeout: float,
         verdict_cls,
+        fusion_delegate=None,
+        files_to_modify: list[str] | None = None,
+        _fusion_dispatch=None,
     ):
         self.repo = repo
         self.base = base
@@ -128,6 +209,12 @@ def __init__(
         self.test_cmd = test_cmd
         self.test_timeout = test_timeout
         self.verdict_cls = verdict_cls  # `plugins.coder.solve.Verdict` — passed in, never imported here
+        self.fusion_delegate = fusion_delegate  # a resolved `openai`-type Delegate, or None
+        self.files_to_modify = files_to_modify or []
+        # Test-injection seam (mirrors `_solve`/`_budget_cls`/`_verdict_cls` on
+        # `dispatch()`): production never passes this — the real lazy
+        # `ADAPTERS["openai"].dispatch` import happens in `generate_fusion` below.
+        self._fusion_dispatch = _fusion_dispatch
         self.candidates: list[tuple[str, str]] = []  # (worktree_path, branch)
         # `git worktree add` against the SAME repo must not run concurrently (best-
         # of-k dispatches `generate()` via asyncio.gather) — serialize just that
@@ -135,15 +222,57 @@ def __init__(
         self._wt_lock = asyncio.Lock()
         self._n = 0
 
-    async def generate(self, task: str, *, feedback: str | None = None) -> str:
+    async def _new_candidate_worktree(self) -> tuple[str, str]:
         self._n += 1
         cid = f"{self.fid}.g{self._n}"
         async with self._wt_lock:
             wt, branch = await worktree.create_worktree(self.repo, self.base, cid, self.root)
         self.candidates.append((wt, branch))
+        return wt, branch
+
+    async def generate(self, task: str, *, feedback: str | None = None) -> str:
+        wt, _branch = await self._new_candidate_worktree()
         await worktree.dispatch_coder(self.coder, wt, _augment_prompt(task, feedback), timeout=self.dispatch_timeout)
         return wt
 
+    async def generate_fusion(self, task: str, *, feedback: str | None = None) -> str:
+        """Rung 4 (ADR 0064 P3): fusion can't tool-call, so instead of dispatching an
+        ACP session into the worktree, get a plain completion and write its files into
+        one. Same candidate bookkeeping (``candidates``/``_wt_lock``) as ``generate``,
+        so promote/reap treats a fusion winner identically to an ACP one."""
+        if self._fusion_dispatch is not None:
+            openai_dispatch = self._fusion_dispatch
+        else:
+            from plugins.delegates.adapters import ADAPTERS
+
+            openai_dispatch = ADAPTERS["openai"].dispatch
+
+        prompt = _fusion_prompt(task, feedback=feedback, repo=self.repo, files_to_modify=self.files_to_modify)
+        reply = await openai_dispatch(self.fusion_delegate, prompt, timeout=self.dispatch_timeout)
+        files = _parse_fusion_files(reply)
+        wt, _branch = await self._new_candidate_worktree()
+        wt_root = Path(wt).resolve()
+        written = 0
+        for rel, content in files.items():
+            # `rel` comes from a model completion — an absolute path or a `../` climb
+            # would otherwise write outside the worktree (Path.__truediv__ with an
+            # absolute right-hand side even discards the left side entirely). Resolve
+            # and require containment; skip (don't crash the candidate) on a miss.
+            dest = (wt_root / rel).resolve()
+            if wt_root not in dest.parents and dest != wt_root:
+                log.warning(
+                    "[project_board] %s fusion tried to write outside its worktree: %r — skipped", self.fid, rel
+                )
+                continue
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            dest.write_text(content)
+            written += 1
+        if not written:
+            log.warning(
+                "[project_board] %s fusion reply parsed to 0 writable files — candidate is unchanged base", self.fid
+            )
+        return wt
+
     async def verify(self, candidate_wt: str):
         Verdict = self.verdict_cls
         try:
@@ -215,9 +344,13 @@ async def dispatch(
     k: int,
     tree_depth: int,
     record_gens: RecordGens | None = None,
+    fusion_delegate=None,
+    fusion_k: int = 2,
+    files_to_modify: list[str] | None = None,
     _solve=None,
     _budget_cls=None,
     _verdict_cls=None,
+    _fusion_dispatch=None,
 ) -> tuple[str, str, str]:
     """Run the execution-grounded ladder for one feature build.
 
@@ -234,6 +367,14 @@ async def dispatch(
     import happens here, deferred so this module carries no hard dependency on the
     `coder` plugin).
 
+    ``fusion_delegate`` (a resolved ``openai``-type Delegate, or ``None``) gates rung
+    4 (ADR 0064 P3) — the caller resolves it (mirroring how ``coder`` itself is
+    resolved), so this module never does delegate lookup. ``None`` (unconfigured) ⇒
+    ``solve()`` gets ``fusion_generate=None`` and stops at tree-search, unchanged from
+    before this rung existed. ``files_to_modify`` feeds fusion's prompt (it can't read
+    the repo itself, unlike the ACP rungs) — the same list the feature's Ready gate
+    already required.
+
     **``solve()`` itself can raise.** The ladder (`coder`'s own ``solve.py``) has no
     try/except around ``generate``/``verify`` — it assumes a candidate attempt never
     errors, only that it might fail its tests. A REAL dispatch can still raise
@@ -263,6 +404,9 @@ async def dispatch(
         test_cmd=test_cmd,
         test_timeout=test_timeout,
         verdict_cls=Verdict,
+        fusion_delegate=fusion_delegate,
+        files_to_modify=files_to_modify,
+        _fusion_dispatch=_fusion_dispatch,
     )
     try:
         result = await solve(
@@ -272,6 +416,8 @@ async def dispatch(
             budget=Budget(budget),
             k=k,
             tree_depth=tree_depth,
+            fusion_generate=adapter.generate_fusion if fusion_delegate is not None else None,
+            fusion_k=fusion_k,
         )
     except Exception as exc:
         for wt, branch in adapter.candidates:
diff --git a/loop.py b/loop.py
index a05986f..9b1d82c 100644
--- a/loop.py
+++ b/loop.py
@@ -234,6 +234,14 @@ def __init__(self, cfg: dict):
         self.coder_solve_budget = max(1, int(self.cfg.get("coder_solve_budget", 6)))
         self.coder_solve_k = max(1, int(self.cfg.get("coder_solve_k", 3)))
         self.coder_solve_tree_depth = max(0, int(self.cfg.get("coder_solve_tree_depth", 2)))
+        # Rung 4 (ADR 0064 P3): a richer generator for the HARDEST features — reached
+        # only after greedy AND best-of-k AND tree-search all fail their tests. Fusion
+        # can't tool-call (it's a plain completion, not an ACP session), so it's an
+        # `openai`-type delegate name, resolved per-dispatch in `_drive` (mirroring how
+        # `coder`/`reviewer` are resolved) — never here, this is just config plumbing.
+        # Blank ⇒ no fusion rung; the ladder stops at tree-search exactly as before.
+        self.coder_solve_fusion_delegate = str(self.cfg.get("coder_solve_fusion_delegate", "")).strip()
+        self.coder_solve_fusion_k = max(1, int(self.cfg.get("coder_solve_fusion_k", 2)))
         # KG lessons (the flywheel READ half): before dispatching a coder, query the
         # knowledge graph (via graph.sdk) for distilled lessons relevant to THIS feature
         # and inject them into the prompt — so the coder heeds this area's known failure
@@ -673,6 +681,11 @@ async def _drive(self, feature: dict):
                         self._inflight[fid] = (repo, wt, branch)
                         result = await worktree.dispatch_coder(coder, wt, prompt, timeout=self.coder_timeout or None)
                     elif self._use_coder_solve(feature) and not self._ci_feedback.get(fid):
+                        fusion = (
+                            self._resolve_delegate(self.coder_solve_fusion_delegate, "openai")
+                            if self.coder_solve_fusion_delegate
+                            else None
+                        )
                         wt, branch, result = await coder_seam.dispatch(
                             task=prompt,
                             coder=coder,
@@ -687,6 +700,9 @@ async def _drive(self, feature: dict):
                             k=self.coder_solve_k,
                             tree_depth=self.coder_solve_tree_depth,
                             record_gens=lambda n: store.record_gens_spent(fid, n),
+                            fusion_delegate=fusion,
+                            fusion_k=self.coder_solve_fusion_k,
+                            files_to_modify=feature.get("files_to_modify") or [],
                         )
                         self._inflight[fid] = (repo, wt, branch)
                     elif self.max_mode_n > 1 and not self._ci_feedback.get(fid):
diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml
index 73a5918..48668ee 100644
--- a/protoagent.plugin.yaml
+++ b/protoagent.plugin.yaml
@@ -1,6 +1,6 @@
 id: project_board
 name: Project Board (coding orchestration)
-version: 0.27.0
+version: 0.28.0
 description: >-
   A board-driven coding-orchestration plugin: a lean 6-state board (backlog → ready
   → in_progress → in_review → done, + a blocked flag) backed by **beads** (`br`), an
diff --git a/pyproject.toml b/pyproject.toml
index 74670d6..62d084a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "project-board"
-version = "0.27.0"
+version = "0.28.0"
 description = "Board-driven coding-orchestration plugin for protoAgent (beads board + ACP spawn loop + planning layer + console view)."
 requires-python = ">=3.11"
 
diff --git a/tests/test_coder_seam.py b/tests/test_coder_seam.py
index b8c7492..7169337 100644
--- a/tests/test_coder_seam.py
+++ b/tests/test_coder_seam.py
@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from pathlib import Path
 
 from project_board import coder_seam, worktree
 from project_board.coder_seam import SolveExhausted, _WorktreeSolveAdapter, dispatch, should_use_solve
@@ -113,7 +114,7 @@ async def _promote(repo, src_wt, src_branch, fid, root=".worktrees"):
 async def test_dispatch_promotes_the_winner_and_reaps_the_losers(monkeypatch):
     created, removed, promoted = _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         # exercise the adapter for real: two candidates, the second "wins".
         await generate(task, feedback=None)
         c1 = await generate(task, feedback=None)
@@ -149,7 +150,7 @@ async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
 async def test_dispatch_records_gens_even_on_a_single_greedy_win(monkeypatch):
     _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         c0 = await generate(task, feedback=None)
         return _FakeResult(solution=c0, passed=True, rung="greedy", gens_spent=1, candidates_tried=1)
 
@@ -182,7 +183,7 @@ async def test_dispatch_promotes_the_winner_even_if_record_gens_raises(monkeypat
     already-verified winning candidate or leak it un-promoted."""
     created, removed, promoted = _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         c0 = await generate(task, feedback=None)
         return _FakeResult(solution=c0, passed=True, rung="greedy", gens_spent=1, candidates_tried=1)
 
@@ -218,7 +219,7 @@ def _boom_record(n):
 async def test_dispatch_raises_solve_exhausted_and_reaps_every_candidate(monkeypatch):
     created, removed, promoted = _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         await generate(task, feedback=None)
         c1 = await generate(task, feedback="prior failure")
         v = _FakeVerdict(passed=False, total=1, failed=1, output="AssertionError: nope")
@@ -260,7 +261,7 @@ async def test_dispatch_exhausted_with_no_candidates_at_all(monkeypatch):
     dispatch() must still raise cleanly with nothing to reap."""
     _created, removed, promoted = _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         return _FakeResult(
             solution=None, passed=None, rung="none", gens_spent=0, candidates_tried=0, note="budget exhausted"
         )
@@ -296,7 +297,7 @@ async def test_dispatch_still_reaps_and_raises_solve_exhausted_when_record_gens_
     the (honest) `SolveExhausted` the caller needs to see."""
     created, removed, promoted = _stub_worktree(monkeypatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         await generate(task, feedback=None)
         c1 = await generate(task, feedback="prior failure")
         v = _FakeVerdict(passed=False, total=1, failed=1, output="AssertionError: nope")
@@ -361,7 +362,7 @@ async def _dispatch(coder, wt, prompt, *, timeout=None):
 
     monkeypatch.setattr(worktree, "dispatch_coder", _dispatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         await generate(task, feedback=None)  # candidate 1: dispatch succeeds
         await generate(task, feedback=None)  # candidate 2: dispatch raises — uncaught by solve()
 
@@ -414,7 +415,7 @@ async def _dispatch(coder, wt, prompt, *, timeout=None):
 
     monkeypatch.setattr(worktree, "dispatch_coder", _dispatch)
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         await generate(task, feedback=None)
         await generate(task, feedback=None)
 
@@ -458,7 +459,7 @@ async def test_dispatch_raise_with_no_candidates_created_yet_skips_record_gens(m
     class _Boom(RuntimeError):
         pass
 
-    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth):
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
         raise _Boom("blew up before any generation")
 
     gens = []
@@ -616,3 +617,235 @@ async def _boom_wait_for(coro, timeout):
     v = await adapter.verify("/wt/feat-bd-1.g1")
     assert v.passed is False
     assert "timed out" in v.output
+
+
+# ── rung 4: fusion (ADR 0064 P3) — a plain completion, not an ACP session ────────
+
+
+def test_parse_fusion_files_single_file():
+    reply = "### foo/bar.py\n```python\nprint('hi')\n```"
+    assert coder_seam._parse_fusion_files(reply) == {"foo/bar.py": "print('hi')\n"}
+
+
+def test_parse_fusion_files_multiple_files():
+    reply = "### a.py\n```\nAAA\n```\n\nsome prose in between\n\n### b/c.py\n```\nBBB\n```"
+    assert coder_seam._parse_fusion_files(reply) == {"a.py": "AAA\n", "b/c.py": "BBB\n"}
+
+
+def test_parse_fusion_files_no_match_returns_empty():
+    assert coder_seam._parse_fusion_files("I looked at it but didn't change anything.") == {}
+    assert coder_seam._parse_fusion_files("") == {}
+
+
+def test_fusion_prompt_includes_task_and_existing_file_content(tmp_path):
+    (tmp_path / "existing.py").write_text("def old(): pass\n")
+    prompt = coder_seam._fusion_prompt(
+        "fix the thing", feedback=None, repo=str(tmp_path), files_to_modify=["existing.py"]
+    )
+    assert "fix the thing" in prompt
+    assert "def old(): pass" in prompt
+    assert "existing.py" in prompt
+
+
+def test_fusion_prompt_notes_a_not_yet_created_file(tmp_path):
+    prompt = coder_seam._fusion_prompt("add new.py", feedback=None, repo=str(tmp_path), files_to_modify=["new.py"])
+    assert "does not exist yet" in prompt
+
+
+def test_fusion_prompt_includes_feedback_when_refining(tmp_path):
+    prompt = coder_seam._fusion_prompt(
+        "fix it", feedback="2/3 failing: AssertionError", repo=str(tmp_path), files_to_modify=[]
+    )
+    assert "FAILED the acceptance tests" in prompt
+    assert "AssertionError" in prompt
+
+
+async def test_generate_fusion_writes_parsed_files_into_a_fresh_worktree(monkeypatch, tmp_path):
+    created, *_ = _stub_worktree(monkeypatch)
+
+    async def _fake_openai_dispatch(delegate, prompt, *, timeout=None):
+        return "### sub/dir/new.py\n```\nCONTENT\n```"
+
+    adapter = _WorktreeSolveAdapter(
+        repo="/repo",
+        base="main",
+        root=".worktrees",
+        fid="bd-1",
+        coder=object(),
+        dispatch_timeout=None,
+        test_cmd="pytest -q",
+        test_timeout=30,
+        verdict_cls=_FakeVerdict,
+        fusion_delegate=object(),  # any non-None placeholder — resolution is the caller's job
+        files_to_modify=[],
+        _fusion_dispatch=_fake_openai_dispatch,
+    )
+
+    # `_stub_worktree`'s fake `create_worktree` always returns "/wt/feat-<cid>" — redirect
+    # it to a real tmp_path so the write actually lands somewhere we can inspect.
+    async def _create_in_tmp(repo, base, cid, root):
+        d = tmp_path / cid
+        d.mkdir(parents=True, exist_ok=True)
+        return (str(d), f"feat/{cid}")
+
+    monkeypatch.setattr(worktree, "create_worktree", _create_in_tmp)
+
+    wt = await adapter.generate_fusion("do the thing")
+    assert (Path(wt) / "sub" / "dir" / "new.py").read_text() == "CONTENT\n"
+    assert adapter.candidates == [(wt, "feat/bd-1.g1")]  # tracked like any other candidate
+
+
+async def test_generate_fusion_rejects_a_path_traversal_attempt(monkeypatch, tmp_path):
+    _stub_worktree(monkeypatch)
+
+    async def _fake_openai_dispatch(delegate, prompt, *, timeout=None):
+        return (
+            "### ../../etc/passwd\n```\npwned\n```\n\n### /etc/shadow\n```\npwned2\n```\n\n### legit.py\n```\nfine\n```"
+        )
+
+    async def _create_in_tmp(repo, base, cid, root):
+        d = tmp_path / cid
+        d.mkdir(parents=True, exist_ok=True)
+        return (str(d), f"feat/{cid}")
+
+    monkeypatch.setattr(worktree, "create_worktree", _create_in_tmp)
+
+    adapter = _WorktreeSolveAdapter(
+        repo="/repo",
+        base="main",
+        root=".worktrees",
+        fid="bd-1",
+        coder=object(),
+        dispatch_timeout=None,
+        test_cmd="pytest -q",
+        test_timeout=30,
+        verdict_cls=_FakeVerdict,
+        fusion_delegate=object(),
+        files_to_modify=[],
+        _fusion_dispatch=_fake_openai_dispatch,
+    )
+    wt = await adapter.generate_fusion("do the thing")
+    # only the legitimate relative path was written; nothing escaped the worktree
+    assert (Path(wt) / "legit.py").read_text() == "fine\n"
+    assert not (Path(wt).parent / "etc").exists()
+    assert not Path("/etc/shadow_THIS_MUST_NOT_EXIST_pwned2").exists()
+
+
+async def test_generate_fusion_empty_reply_writes_nothing_and_does_not_crash(monkeypatch, tmp_path):
+    _stub_worktree(monkeypatch)
+
+    async def _fake_openai_dispatch(delegate, prompt, *, timeout=None):
+        return "I looked at the task but have no changes."
+
+    async def _create_in_tmp(repo, base, cid, root):
+        d = tmp_path / cid
+        d.mkdir(parents=True, exist_ok=True)
+        return (str(d), f"feat/{cid}")
+
+    monkeypatch.setattr(worktree, "create_worktree", _create_in_tmp)
+
+    adapter = _WorktreeSolveAdapter(
+        repo="/repo",
+        base="main",
+        root=".worktrees",
+        fid="bd-1",
+        coder=object(),
+        dispatch_timeout=None,
+        test_cmd="pytest -q",
+        test_timeout=30,
+        verdict_cls=_FakeVerdict,
+        fusion_delegate=object(),
+        files_to_modify=[],
+        _fusion_dispatch=_fake_openai_dispatch,
+    )
+    wt = await adapter.generate_fusion("do the thing")
+    assert list(Path(wt).iterdir()) == []  # untouched — will just fail verify() like any empty candidate
+
+
+# ── dispatch(): fusion end-to-end + honest degrade ───────────────────────────────
+
+
+async def test_dispatch_reaches_fusion_when_cheaper_rungs_fail(monkeypatch, tmp_path):
+    """A `_fake_solve` standing in for the REAL ladder: simulates greedy/best-of-k/
+    tree-search all failing, then calls `fusion_generate` and wins — proving
+    `dispatch()` wires `fusion_generate`/`fusion_k` through to `solve()` and that a
+    fusion-produced candidate promotes exactly like an ACP one."""
+    created, removed, promoted = _stub_worktree(monkeypatch)
+
+    async def _create_in_tmp(repo, base, cid, root):
+        d = tmp_path / cid
+        d.mkdir(parents=True, exist_ok=True)
+        return (str(d), f"feat/{cid}")
+
+    monkeypatch.setattr(worktree, "create_worktree", _create_in_tmp)
+
+    async def _fake_openai_dispatch(delegate, prompt, *, timeout=None):
+        return "### fixed.py\n```\nfixed content\n```"
+
+    seen_fusion_k = {}
+
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
+        seen_fusion_k["k"] = fusion_k
+        assert fusion_generate is not None  # dispatch() must have wired it through
+        c = await fusion_generate(task, feedback="2/2 failing")
+        return _FakeResult(solution=c, passed=True, rung="fusion", gens_spent=5, candidates_tried=5)
+
+    gens = []
+    wt, branch, result = await dispatch(
+        task="do the thing",
+        coder=object(),
+        repo="/repo",
+        base="main",
+        root=".worktrees",
+        fid="bd-1",
+        dispatch_timeout=None,
+        test_cmd="pytest -q",
+        test_timeout=30,
+        budget=6,
+        k=3,
+        tree_depth=2,
+        record_gens=gens.append,
+        fusion_delegate=object(),
+        fusion_k=4,
+        files_to_modify=[],
+        _solve=_fake_solve,
+        _budget_cls=_FakeBudget,
+        _verdict_cls=_FakeVerdict,
+        _fusion_dispatch=_fake_openai_dispatch,
+    )
+    assert seen_fusion_k["k"] == 4
+    assert "fusion" in result and "gens=5" in result
+    assert promoted and promoted[0][2] == "bd-1"
+    assert gens == [5]
+
+
+async def test_dispatch_without_a_fusion_delegate_passes_none_through(monkeypatch):
+    """Honest degrade (unchanged from before this rung existed): no fusion_delegate
+    configured ⇒ solve() gets fusion_generate=None ⇒ the ladder stops at tree-search."""
+    _stub_worktree(monkeypatch)
+    seen = {}
+
+    async def _fake_solve(task, *, generate, verify, budget, k, tree_depth, fusion_generate=None, fusion_k=2):
+        seen["fusion_generate"] = fusion_generate
+        c = await generate(task, feedback=None)
+        return _FakeResult(solution=c, passed=True, rung="greedy", gens_spent=1, candidates_tried=1)
+
+    await dispatch(
+        task="do the thing",
+        coder=object(),
+        repo="/repo",
+        base="main",
+        root=".worktrees",
+        fid="bd-1",
+        dispatch_timeout=None,
+        test_cmd="pytest -q",
+        test_timeout=30,
+        budget=6,
+        k=3,
+        tree_depth=2,
+        # fusion_delegate omitted — defaults to None
+        _solve=_fake_solve,
+        _budget_cls=_FakeBudget,
+        _verdict_cls=_FakeVerdict,
+    )
+    assert seen["fusion_generate"] is None
diff --git a/tests/test_loop.py b/tests/test_loop.py
index 951db09..b6b03e3 100644
--- a/tests/test_loop.py
+++ b/tests/test_loop.py
@@ -478,6 +478,9 @@ async def _fake_dispatch(
         k,
         tree_depth,
         record_gens=None,
+        fusion_delegate=None,
+        fusion_k=2,
+        files_to_modify=None,
     ):
         seen["fid"] = fid
         seen["test_cmd"] = test_cmd