diff --git a/src/portfolio_optimiser/workflow.py b/src/portfolio_optimiser/workflow.py
new file mode 100644
index 0000000..1dffb6e
--- /dev/null
+++ b/src/portfolio_optimiser/workflow.py
@@ -0,0 +1,83 @@
+"""fresh_workflow isolation factory + maker-checker GroupChat (B7 / B4 / Layer-1 HITL).
+
+Two Fase 1 findings are load-bearing here:
+
+* **B7 (state isolation):** a *reused* built workflow accumulates the MAF conversation thread
+  across ``.run()`` calls, so project N contaminates N+1. The mitigation is a FACTORY that
+  builds a FRESH ``GroupChatBuilder`` with FRESH chat clients **per project run** — zero
+  cross-run bleed. ``fresh_workflow`` is that factory.
+* **B4 (bounded debate):** a never-converging debate does NOT self-terminate. ``with_max_rounds``
+  is the hard cap; the budget middleware (Step 4, wired by the orchestrator) is the external
+  token guard. ``make_termination`` is a higher turn-count safety net, not the sole stop.
+
+**Layer-1 HITL** optionally enables the GA in-run synchronous review gate
+(``with_request_info(agents=[checker])``) — no checkpoint (research 01: durable checkpoint
+resume is fragile; the durable verdict is captured out-of-band in Step 12).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+
+from agent_framework import Agent, BaseChatClient, Message
+from agent_framework.orchestrations import GroupChatBuilder
+
+_MAKER_CHECKER_ROLES = ("proposer", "checker")
+_INSTRUCTIONS = {
+    "proposer": "You propose one concrete cost-saving measure for the project.",
+    "checker": "You critique the proposal and flag any constraint violation.",
+}
+
+
+def make_termination(n_turns: int) -> Callable[[Sequence[Message]], bool]:
+    """A GroupChat ``TerminationCondition`` that stops once ``n_turns`` turns have been taken.
+    Deterministic and endpoint-free; the hard cap is ``with_max_rounds`` (B4)."""
+    if n_turns <= 0:
+        raise ValueError(f"n_turns must be positive, got {n_turns}")
+
+    def terminate(conversation: Sequence[Message]) -> bool:
+        return len(conversation) >= n_turns
+
+    return terminate
+
+
+def maker_checker_agents(client_factory: Callable[[str], BaseChatClient]) -> list[Agent]:
+    """FRESH proposer + checker agents, each backed by a FRESH client (zero cross-run state)."""
+    return [
+        Agent(client_factory(role), _INSTRUCTIONS[role], name=role)
+        for role in _MAKER_CHECKER_ROLES
+    ]
+
+
+def fresh_workflow(
+    client_factory: Callable[[str], BaseChatClient],
+    *,
+    max_rounds: int = 3,
+    enable_layer1_hitl: bool = False,
+) -> object:
+    """Build a FRESH maker-checker GroupChat with FRESH clients per call (B7). Bounded by
+    ``with_max_rounds`` (B4) plus a higher turn-count termination safety net. ``client_factory``
+    is called once per role, so each run owns its own clients — no state survives between runs.
+    """
+    agents = maker_checker_agents(client_factory)
+    names = [a.name for a in agents]
+    counter = {"n": 0}
+
+    def select(_state: object) -> str:
+        choice = names[counter["n"] % len(names)]
+        counter["n"] += 1
+        return choice
+
+    builder = (
+        GroupChatBuilder(
+            participants=agents,
+            selection_func=select,
+            # Safety net well above the hard cap; with_max_rounds is the binding bound (B4).
+            termination_condition=make_termination(max_rounds * len(names) + 1),
+        )
+        .with_max_rounds(max_rounds)
+    )
+    if enable_layer1_hitl:
+        # Layer-1: in-run synchronous review on the checker (no checkpoint — research 01).
+        builder = builder.with_request_info(agents=[agents[-1]])
+    return builder.build()
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
new file mode 100644
index 0000000..bf273b4
--- /dev/null
+++ b/tests/test_workflow.py
@@ -0,0 +1,59 @@
+"""Step 9 tests — fresh_workflow isolation + maker-checker round cap (FakeChatClient, no LLM).
+
+Isolation is asserted on received-message CONTENT (not a call counter — the Fase 1 round-2
+fix): a fresh workflow per run gives each run a clean thread, so a participant sees ONLY its
+own project. The round cap is pinned to the EXACT observed count (retiring the 1-turn=1-round
+off-by-one assumption). Pattern: tests/spikes/test_b_footguns.py + tests/spikes/test_harness.py.
+"""
+
+from collections.abc import Callable
+
+from agent_framework import BaseChatClient
+from spikes._harness import FakeChatClient
+
+from portfolio_optimiser.reference_domain import load_reference_projects
+from portfolio_optimiser.workflow import fresh_workflow
+
+
+def _recording_factory(created: list[FakeChatClient]) -> Callable[[str], BaseChatClient]:
+    def factory(role: str) -> BaseChatClient:
+        client = FakeChatClient(default_reply=f"{role} view")
+        created.append(client)
+        return client
+
+    return factory
+
+
+async def test_participant_sees_only_its_own_project() -> None:
+    ids = [p.id for p in load_reference_projects()]
+    seen_per_run: list[list[str]] = []
+    for pid in ids:
+        created: list[FakeChatClient] = []
+        wf = fresh_workflow(_recording_factory(created), max_rounds=2)
+        await wf.run(f"Evaluate project {pid}.")
+        blob = " ".join(t for c in created for call in c.received_texts for t in call)
+        seen_per_run.append([q for q in ids if q in blob])
+    # A fresh instance per run -> each participant sees ONLY its own project (B7 mitigation).
+    assert seen_per_run == [[pid] for pid in ids]
+
+
+async def test_never_converging_debate_halts_at_exactly_max_rounds() -> None:
+    created: list[FakeChatClient] = []
+
+    def factory(role: str) -> BaseChatClient:
+        client = FakeChatClient(default_reply="still disagreeing, keep debating")
+        created.append(client)
+        return client
+
+    wf = fresh_workflow(factory, max_rounds=3)
+    await wf.run("Debate a never-ending question.")
+    # with_max_rounds(3) is the hard cap; the debate halts at EXACTLY 3 turns, not <=.
+    assert sum(c.call_count for c in created) == 3
+
+
+def test_layer1_hitl_option_builds() -> None:
+    def factory(role: str) -> BaseChatClient:
+        return FakeChatClient(default_reply="ok")
+
+    wf = fresh_workflow(factory, max_rounds=2, enable_layer1_hitl=True)
+    assert wf is not None