docs: add CrewAI Flow refactor implementation plan

8 tasks: AgentResponse model, FlowState, routing, readiness checks,
full AgentDispatchFlow, crew_orchestrator integration, RoutingEngine
removal, and final validation. TDD throughout.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

docs/superpowers/plans/2026-04-12-crewai-flow-refactor.md

@@ -0,0 +1,1469 @@
+# CrewAI Flow Refactor Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace `Crew.kickoff()` and `RoutingEngine` with CrewAI Flow's `@start`/`@listen`/`@router` decorators for agent dispatch, enabling parallel advisory agents and router-gated generation.
+
+**Architecture:** New `agents/agent_flow.py` contains typed models (`AgentResponse`, `AgentFlowState`) and `AgentDispatchFlow` — a Flow with two routers: `route_message` (replaces `RoutingEngine`) and `check_readiness` (gates CAD/CAM). Advisory agents run in one Crew; CAD and CAM run as single-agent Crews gated by readiness. The outer `_run_crew()` in `crew_orchestrator.py` delegates dispatch to the Flow and keeps post-processing.
+
+**Tech Stack:** CrewAI 1.14 (`crewai.flow.flow.Flow`, `@start`, `@listen`, `@router`), Pydantic BaseModel, existing CrewAI Agent/Task/Crew/LLM
+
+**Spec:** `docs/superpowers/specs/2026-04-12-crewai-flow-refactor-design.md`
+
+---
+
+### Task 1: AgentResponse model + tests
+
+**Files:**
+- Create: `agents/agent_flow.py`
+- Create: `tests/test_agent_flow.py`
+
+- [ ] **Step 1: Write failing tests for AgentResponse**
+
+```python
+# tests/test_agent_flow.py
+"""Tests for agents/agent_flow.py — AgentResponse, AgentFlowState, AgentDispatchFlow."""
+
+from agents.agent_flow import AgentResponse
+
+
+class TestAgentResponse:
+    def test_from_agent_populates_all_fields(self):
+        resp = AgentResponse.from_agent("design", "Great bracket idea.")
+        assert resp.agent_id == "design"
+        assert resp.agent_name == "Design Agent"
+        assert resp.color == "#7c3aed"
+        assert resp.avatar == "DA"
+        assert resp.message == "Great bracket idea."
+        assert resp.code is None
+
+    def test_from_agent_with_code(self):
+        resp = AgentResponse.from_agent("cad", "Model generated.", code="import cq")
+        assert resp.code == "import cq"
+        assert resp.agent_id == "cad"
+
+    def test_from_agent_engineering(self):
+        resp = AgentResponse.from_agent("engineering", "Use 3mm walls.")
+        assert resp.agent_name == "Engineering Agent"
+        assert resp.color == "#00b4d8"
+
+    def test_model_dump_matches_format_response(self):
+        resp = AgentResponse.from_agent("cnc", "Looks machinable.")
+        d = resp.model_dump()
+        assert set(d.keys()) == {"agent_id", "agent_name", "message", "color", "avatar", "code"}
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: FAIL — `ModuleNotFoundError: No module named 'agents.agent_flow'`
+
+- [ ] **Step 3: Implement AgentResponse model**
+
+```python
+# agents/agent_flow.py
+"""CrewAI Flow for multi-agent dispatch.
+
+Replaces Crew.kickoff() and RoutingEngine with a Flow that routes messages
+to advisory agents (parallel) and gates generation agents (CAD/CAM) via
+readiness checks.
+"""
+
+from __future__ import annotations
+
+import re
+from pydantic import BaseModel, Field
+
+from agents.definitions import AGENTS
+
+
+class AgentResponse(BaseModel):
+    """Typed agent response — replaces raw dicts from _format_response()."""
+    agent_id: str
+    agent_name: str
+    message: str
+    color: str
+    avatar: str
+    code: str | None = None
+
+    @classmethod
+    def from_agent(cls, agent_id: str, message: str,
+                   code: str | None = None) -> AgentResponse:
+        """Build from agent_id, looking up metadata from AGENTS registry."""
+        agent_def = AGENTS[agent_id]
+        return cls(
+            agent_id=agent_id,
+            agent_name=agent_def.name,
+            message=message,
+            color=agent_def.color,
+            avatar=agent_def.avatar,
+            code=code,
+        )
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: 4 passed
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add agents/agent_flow.py tests/test_agent_flow.py
+git commit -m "feat: add AgentResponse model with from_agent factory"
+```
+
+---
+
+### Task 2: AgentFlowState model + extract_code utility
+
+**Files:**
+- Modify: `agents/agent_flow.py`
+- Modify: `tests/test_agent_flow.py`
+
+- [ ] **Step 1: Write failing tests for AgentFlowState and extract_code**
+
+```python
+# tests/test_agent_flow.py — append to file
+from agents.agent_flow import AgentFlowState, extract_code
+
+
+class TestAgentFlowState:
+    def test_defaults(self):
+        state = AgentFlowState()
+        assert state.message == ""
+        assert state.active_agent_ids == []
+        assert state.advisor_responses == []
+        assert state.cad_response is None
+        assert state.cam_response is None
+        assert state.cad_code is None
+        assert state.cam_plan is None
+        assert state.mentions == []
+        assert state.is_approved_phase is False
+
+    def test_with_inputs(self):
+        state = AgentFlowState(
+            message="build a bracket",
+            model_str="gemini/gemini-2.5-flash",
+            mentions=["design"],
+            is_approved_phase=True,
+        )
+        assert state.message == "build a bracket"
+        assert state.mentions == ["design"]
+        assert state.is_approved_phase is True
+
+
+class TestExtractCode:
+    def test_extracts_fenced_python(self):
+        text = "Here is code:\n```python\nimport cadquery as cq\nresult = cq.Workplane('XY').box(10,10,10)\n```\nDone."
+        code = extract_code(text)
+        assert code is not None
+        assert "import cadquery" in code
+        assert "result =" in code
+
+    def test_extracts_unfenced_cq_code(self):
+        text = "import cadquery as cq\nresult = cq.Workplane('XY').box(5,5,5)"
+        code = extract_code(text)
+        assert code is not None
+        assert "cq.Workplane" in code
+
+    def test_returns_none_for_plain_text(self):
+        text = "NOT READY: I need dimensions."
+        assert extract_code(text) is None
+
+    def test_extracts_generic_fenced_block(self):
+        text = "```\nimport cadquery as cq\nresult = cq.Workplane('XY').box(1,1,1)\n```"
+        code = extract_code(text)
+        assert code is not None
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `pytest tests/test_agent_flow.py::TestAgentFlowState -v && pytest tests/test_agent_flow.py::TestExtractCode -v`
+Expected: FAIL — `ImportError: cannot import name 'AgentFlowState'`
+
+- [ ] **Step 3: Implement AgentFlowState and extract_code**
+
+Add to `agents/agent_flow.py` after the `AgentResponse` class:
+
+```python
+class AgentFlowState(BaseModel):
+    """Orchestration state for a single chat turn. Lives only during Flow execution."""
+    # Input (set by _run_crew before kickoff)
+    message: str = ""
+    context: str = ""
+    model_str: str = ""
+    mentions: list[str] = Field(default_factory=list)
+    is_approved_phase: bool = False
+
+    # Set by route_message router
+    active_agent_ids: list[str] = Field(default_factory=list)
+
+    # Set by prepare_agents
+    knowledge_sources_data: list[str] = Field(default_factory=list)
+
+    # Accumulated output (set by Flow steps)
+    advisor_responses: list[AgentResponse] = Field(default_factory=list)
+    cad_response: AgentResponse | None = None
+    cam_response: AgentResponse | None = None
+
+    # Generation artifacts
+    cad_code: str | None = None
+    cam_plan: "CAMPlan | None" = None
+
+
+def extract_code(text: str) -> str | None:
+    """Extract Python code from LLM output.
+
+    Checks for markdown fenced blocks first, then bare CadQuery markers.
+    """
+    match = re.search(r"```(?:python)?\s*\n(.*?)```", text, re.DOTALL)
+    if match:
+        return match.group(1).strip()
+    if any(marker in text for marker in ["import cadquery", "cq.", "result ="]):
+        return text.strip()
+    return None
+```
+
+Add forward-reference import at top of file for type annotation:
+
+```python
+from __future__ import annotations
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: 10 passed
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add agents/agent_flow.py tests/test_agent_flow.py
+git commit -m "feat: add AgentFlowState model and extract_code utility"
+```
+
+---
+
+### Task 3: route_message logic + tests
+
+**Files:**
+- Modify: `agents/agent_flow.py`
+- Modify: `tests/test_agent_flow.py`
+
+- [ ] **Step 1: Write failing tests for routing logic**
+
+The routing logic will be implemented as a standalone function first (testable without the full Flow), then wired into the Flow's `@router` step in Task 5.
+
+```python
+# tests/test_agent_flow.py — append to file
+from agents.agent_flow import route_agents
+
+class TestRouteAgents:
+    def test_approved_phase_locks_to_config(self):
+        ids = route_agents("anything", mentions=[], is_approved_phase=True)
+        assert ids == ["cad", "cnc"]
+
+    def test_mentions_override_routing(self):
+        ids = route_agents("anything", mentions=["design", "cam"], is_approved_phase=False)
+        assert ids == ["design", "cam"]
+
+    def test_design_keywords(self):
+        ids = route_agents("I want a sleek design with smooth shape", mentions=[], is_approved_phase=False)
+        assert "design" in ids
+
+    def test_engineering_keywords(self):
+        ids = route_agents("Use M6 bolts with 3mm wall thickness in aluminum", mentions=[], is_approved_phase=False)
+        assert "engineering" in ids
+
+    def test_cnc_keywords(self):
+        ids = route_agents("Can this be machined on a 3-axis CNC mill?", mentions=[], is_approved_phase=False)
+        assert "cnc" in ids
+
+    def test_cam_keywords(self):
+        ids = route_agents("Generate a toolpath for this part", mentions=[], is_approved_phase=False)
+        assert "cam" in ids
+
+    def test_default_when_no_match(self):
+        ids = route_agents("hello there", mentions=[], is_approved_phase=False)
+        assert ids == ["design", "engineering"]
+
+    def test_max_three_agents(self):
+        ids = route_agents("design shape in aluminum for CNC machining", mentions=[], is_approved_phase=False)
+        assert len(ids) <= 3
+
+    def test_cad_trigger_adds_cad(self):
+        ids = route_agents("Generate a preview", mentions=[], is_approved_phase=False)
+        assert "cad" in ids
+
+    def test_no_cad_trigger_without_keyword(self):
+        ids = route_agents("hello there", mentions=[], is_approved_phase=False)
+        assert "cad" not in ids
+
+    def test_cad_not_duplicated_when_already_routed(self):
+        ids = route_agents("generate code for this design", mentions=[], is_approved_phase=False)
+        assert ids.count("cad") <= 1
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `pytest tests/test_agent_flow.py::TestRouteAgents -v`
+Expected: FAIL — `ImportError: cannot import name 'route_agents'`
+
+- [ ] **Step 3: Implement route_agents function**
+
+Add to `agents/agent_flow.py`:
+
+```python
+from config.settings import settings
+
+# Agent role sets for routing
+ADVISOR_IDS = frozenset({"design", "engineering", "cnc"})
+GENERATOR_IDS = frozenset({"cad", "cam"})
+
+
+def route_agents(
+    message: str,
+    mentions: list[str],
+    is_approved_phase: bool,
+) -> list[str]:
+    """Select which agents should respond to this message.
+
+    Replaces RoutingEngine.route() + has_cad_trigger() + approved/mentions logic.
+    Reads routing keywords and CAD triggers from config.yaml.
+    """
+    if is_approved_phase:
+        return list(settings.planning.approved_agents)
+
+    if mentions:
+        return list(mentions)
+
+    # Keyword scoring
+    lower = message.lower()
+    keywords: dict[str, list[str]] = settings.routing.keywords
+    max_agents: int = settings.orchestration.max_active_agents
+
+    scores: dict[str, int] = {agent_id: 0 for agent_id in keywords}
+    for agent_id, kws in keywords.items():
+        for kw in kws:
+            if kw in lower:
+                scores[agent_id] += 1
+
+    active = [aid for aid, score in sorted(scores.items(), key=lambda x: -x[1]) if score > 0]
+    if not active:
+        active = ["design", "engineering"]
+    active = active[:max_agents]
+
+    # CAD trigger check
+    cad_triggers: list[str] = settings.routing.cad_trigger_keywords
+    if "cad" not in active and any(kw in lower for kw in cad_triggers):
+        active.append("cad")
+
+    return active
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_agent_flow.py::TestRouteAgents -v`
+Expected: 11 passed
+
+- [ ] **Step 5: Run full test suite**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: 21 passed
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add agents/agent_flow.py tests/test_agent_flow.py
+git commit -m "feat: add route_agents function replacing RoutingEngine"
+```
+
+---
+
+### Task 4: check_readiness logic + collect_results + tests
+
+**Files:**
+- Modify: `agents/agent_flow.py`
+- Modify: `tests/test_agent_flow.py`
+
+- [ ] **Step 1: Write failing tests**
+
+```python
+# tests/test_agent_flow.py — append to file
+from agents.agent_flow import check_readiness, collect_responses
+
+
+class TestCheckReadiness:
+    def test_ready_when_advisors_clean(self):
+        responses = [
+            AgentResponse.from_agent("design", "L-bracket looks good."),
+            AgentResponse.from_agent("engineering", "3mm walls in aluminum."),
+        ]
+        result = check_readiness(responses, active_agent_ids=["design", "engineering", "cad"])
+        assert result == "READY"
+
+    def test_not_ready_when_advisor_flags(self):
+        responses = [
+            AgentResponse.from_agent("design", "L-bracket looks good."),
+            AgentResponse.from_agent("cnc", "NOT READY: Missing dimensions and material."),
+        ]
+        result = check_readiness(responses, active_agent_ids=["design", "cnc", "cad"])
+        assert result == "NOT_READY"
+
+    def test_skip_generation_when_no_generators(self):
+        responses = [
+            AgentResponse.from_agent("design", "Nice shape."),
+        ]
+        result = check_readiness(responses, active_agent_ids=["design", "engineering"])
+        assert result == "SKIP_GENERATION"
+
+    def test_not_ready_case_insensitive(self):
+        responses = [
+            AgentResponse.from_agent("engineering", "not ready: need wall thickness"),
+        ]
+        result = check_readiness(responses, active_agent_ids=["engineering", "cad"])
+        assert result == "NOT_READY"
+
+
+class TestCollectResponses:
+    def test_merges_all_responses(self):
+        advisors = [AgentResponse.from_agent("design", "Shape OK.")]
+        cad = AgentResponse.from_agent("cad", "Model generated.", code="result = cq.box()")
+        cam = AgentResponse.from_agent("cam", "Machining plan ready.")
+        result = collect_responses(advisors, cad, cam)
+        assert len(result) == 3
+        assert result[0].agent_id == "design"
+        assert result[1].agent_id == "cad"
+        assert result[2].agent_id == "cam"
+
+    def test_handles_none_cad_and_cam(self):
+        advisors = [AgentResponse.from_agent("engineering", "Specs look good.")]
+        result = collect_responses(advisors, None, None)
+        assert len(result) == 1
+        assert result[0].agent_id == "engineering"
+
+    def test_handles_empty_advisors(self):
+        cad = AgentResponse.from_agent("cad", "NOT READY: need dimensions")
+        result = collect_responses([], cad, None)
+        assert len(result) == 1
+        assert result[0].agent_id == "cad"
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `pytest tests/test_agent_flow.py::TestCheckReadiness tests/test_agent_flow.py::TestCollectResponses -v`
+Expected: FAIL — `ImportError`
+
+- [ ] **Step 3: Implement check_readiness and collect_responses**
+
+Add to `agents/agent_flow.py`:
+
+```python
+def check_readiness(
+    advisor_responses: list[AgentResponse],
+    active_agent_ids: list[str],
+) -> str:
+    """Inspect advisor responses and active agents to determine generation path.
+
+    Returns:
+        "READY" — advisors OK, generators should run
+        "NOT_READY" — at least one advisor flagged NOT READY
+        "SKIP_GENERATION" — no generators (cad/cam) in active list
+    """
+    has_generators = bool(GENERATOR_IDS & set(active_agent_ids))
+    if not has_generators:
+        return "SKIP_GENERATION"
+
+    for resp in advisor_responses:
+        if resp.message.strip().upper().startswith("NOT READY:"):
+            return "NOT_READY"
+
+    return "READY"
+
+
+def collect_responses(
+    advisor_responses: list[AgentResponse],
+    cad_response: AgentResponse | None,
+    cam_response: AgentResponse | None,
+) -> list[AgentResponse]:
+    """Merge all agent responses into a single ordered list."""
+    result = list(advisor_responses)
+    if cad_response is not None:
+        result.append(cad_response)
+    if cam_response is not None:
+        result.append(cam_response)
+    return result
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: 28 passed
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add agents/agent_flow.py tests/test_agent_flow.py
+git commit -m "feat: add check_readiness and collect_responses helpers"
+```
+
+---
+
+### Task 5: AgentDispatchFlow — full Flow class
+
+**Files:**
+- Modify: `agents/agent_flow.py`
+- Modify: `tests/test_agent_flow.py`
+
+- [ ] **Step 1: Write failing tests for the Flow**
+
+These tests validate the Flow wiring without making LLM calls. They test `prepare_agents`, `route_message`, and `no_agents` paths by mocking the Crew execution.
+
+```python
+# tests/test_agent_flow.py — append to file
+from unittest.mock import patch, MagicMock
+from agents.agent_flow import AgentDispatchFlow
+
+
+class TestAgentDispatchFlow:
+    def test_no_agents_path(self):
+        """Flow with no matching agents completes without error."""
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="xyzzy",
+            context="## User's latest message\nxyzzy",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow.kickoff()
+        assert flow.state.active_agent_ids == ["design", "engineering"]
+
+    def test_approved_phase_sets_generators(self):
+        """Approved phase locks agents to config approved_agents."""
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="build it",
+            context="## APPROVED PLAN",
+            model_str="gemini/gemini-2.5-flash",
+            is_approved_phase=True,
+        ))
+        # Mock _run_single_agent_crew to avoid actual LLM calls
+        with patch.object(flow, '_run_single_agent_crew', return_value="NOT READY: need dims"):
+            flow.kickoff()
+        assert flow.state.active_agent_ids == ["cad", "cnc"]
+
+    def test_mentions_override(self):
+        """Explicit mentions override keyword routing."""
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="check this",
+            context="",
+            model_str="gemini/gemini-2.5-flash",
+            mentions=["cam"],
+        ))
+        with patch.object(flow, '_run_single_agent_crew', return_value="Machining plan ready."):
+            flow.kickoff()
+        assert flow.state.active_agent_ids == ["cam"]
+
+    def test_collect_results_populated(self):
+        """Flow populates collected responses from advisor + cad paths."""
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="design a bracket",
+            context="",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        # Simulate: advisors respond, no generators routed
+        with patch.object(flow, '_run_advisor_crew', return_value=[
+            AgentResponse.from_agent("design", "L-bracket idea."),
+        ]):
+            flow.kickoff()
+        results = collect_responses(
+            flow.state.advisor_responses,
+            flow.state.cad_response,
+            flow.state.cam_response,
+        )
+        assert len(results) >= 1
+```
+
+- [ ] **Step 2: Run tests to verify they fail**
+
+Run: `pytest tests/test_agent_flow.py::TestAgentDispatchFlow -v`
+Expected: FAIL — `ImportError: cannot import name 'AgentDispatchFlow'`
+
+- [ ] **Step 3: Implement AgentDispatchFlow**
+
+Add to `agents/agent_flow.py`:
+
+```python
+import logging
+from pathlib import Path
+
+from crewai.flow.flow import Flow, listen, start, router
+
+logger = logging.getLogger(__name__)
+
+WIKI_DIR = Path(__file__).parent.parent / "docs" / "wiki"
+
+
+class AgentDispatchFlow(Flow[AgentFlowState]):
+    """Flow-based agent dispatch replacing Crew.kickoff() + RoutingEngine.
+
+    Graph:
+        prepare_agents → route_message (router)
+            ├─ HAS_ADVISORS → run_advisors → check_readiness (router)
+            │       ├─ READY → run_cad → run_cam → collect_results
+            │       ├─ NOT_READY → run_cad_not_ready → skip_cam → collect_results
+            │       └─ SKIP_GENERATION → skip_generation → collect_results
+            ├─ GENERATORS_ONLY → run_cad_gen_only → run_cam_gen_only → collect_results
+            └─ NO_AGENTS → no_agents → collect_results
+    """
+
+    # ── Start ────────────────────────────────────────────────────────────
+
+    @start()
+    def prepare_agents(self):
+        """Build LLM config and load wiki knowledge sources."""
+        for filename in ("cutting-parameters.md", "gcode-reference.md"):
+            path = WIKI_DIR / filename
+            if path.exists():
+                self.state.knowledge_sources_data.append(path.read_text())
+
+    # ── Routing ──────────────────────────────────────────────────────────
+
+    @router(prepare_agents)
+    def route_message(self):
+        """Select agents and return path: HAS_ADVISORS | GENERATORS_ONLY | NO_AGENTS."""
+        self.state.active_agent_ids = route_agents(
+            self.state.message,
+            self.state.mentions,
+            self.state.is_approved_phase,
+        )
+        if not self.state.active_agent_ids:
+            return "NO_AGENTS"
+
+        has_advisors = bool(ADVISOR_IDS & set(self.state.active_agent_ids))
+        has_generators = bool(GENERATOR_IDS & set(self.state.active_agent_ids))
+
+        if has_advisors:
+            return "HAS_ADVISORS"
+        if has_generators:
+            return "GENERATORS_ONLY"
+        return "NO_AGENTS"
+
+    # ── Advisory path ────────────────────────────────────────────────────
+
+    @listen("HAS_ADVISORS")
+    def run_advisors(self):
+        """Run advisory agents (design, engineering, cnc) as one Crew."""
+        advisor_ids = [aid for aid in self.state.active_agent_ids if aid in ADVISOR_IDS]
+        if not advisor_ids:
+            return
+
+        responses = self._run_advisor_crew(advisor_ids)
+        self.state.advisor_responses = responses
+
+    @router(run_advisors)
+    def check_readiness_router(self):
+        """Gate generation based on advisor responses."""
+        return check_readiness(self.state.advisor_responses, self.state.active_agent_ids)
+
+    @listen("READY")
+    def run_cad(self):
+        """Run CAD Coder agent — expects code generation."""
+        self._run_cad_step()
+
+    @listen("NOT_READY")
+    def run_cad_not_ready(self):
+        """Run CAD Coder agent — expects NOT READY gap list."""
+        self._run_cad_step()
+
+    @listen("SKIP_GENERATION")
+    def skip_generation(self):
+        """No generators requested — pass through."""
+        pass
+
+    @listen(run_cad)
+    def run_cam(self):
+        """Run CAM agent after successful CAD generation."""
+        self._run_cam_step()
+
+    @listen(run_cad_not_ready)
+    def skip_cam_after_not_ready(self):
+        """CAM skipped — CAD wasn't ready."""
+        pass
+
+    # ── Generators-only path ─────────────────────────────────────────────
+
+    @listen("GENERATORS_ONLY")
+    def run_cad_gen_only(self):
+        """Run CAD Coder directly (no advisors ran)."""
+        self._run_cad_step()
+
+    @listen(run_cad_gen_only)
+    def run_cam_gen_only(self):
+        """Run CAM after CAD in generators-only path."""
+        self._run_cam_step()
+
+    # ── No-agents path ───────────────────────────────────────────────────
+
+    @listen("NO_AGENTS")
+    def no_agents(self):
+        """Defensive — no agents matched. Pass through."""
+        pass
+
+    # ── Collect ──────────────────────────────────────────────────────────
+
+    @listen(run_cam, skip_cam_after_not_ready, skip_generation,
+            run_cam_gen_only, no_agents)
+    def collect_results(self):
+        """Merge all responses — accessible via state after kickoff."""
+        pass  # Responses already on state; caller reads them directly.
+
+    # ── Private helpers ──────────────────────────────────────────────────
+
+    def _build_llm(self):
+        """Build a CrewAI LLM from state.model_str."""
+        from crewai import LLM
+        return LLM(model=self.state.model_str, temperature=settings.temperature)
+
+    def _build_knowledge_sources(self):
+        """Build StringKnowledgeSources from loaded wiki data."""
+        sources = []
+        try:
+            from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
+            for content in self.state.knowledge_sources_data:
+                sources.append(StringKnowledgeSource(content=content))
+        except ImportError:
+            pass
+        return sources
+
+    def _build_crew_agent(self, agent_id: str, llm):
+        """Create a CrewAI Agent + Task for the given agent_id.
+
+        Returns (Agent, Task) tuple with per-agent tools, backstory, and
+        task description matching the current crew_orchestrator.py logic.
+        """
+        from crewai import Agent, Task
+        from agents.tools import (
+            query_design_state_tool, execute_cad_tool,
+            validate_cad_tool, generate_gcode_tool,
+        )
+        from core.cam import CAMPlan
+
+        agent_def = AGENTS[agent_id]
+        tools = [query_design_state_tool]
+        extra_backstory = ""
+        task_output_pydantic = None
+
+        if agent_id == "cad":
+            tools.extend([execute_cad_tool, validate_cad_tool])
+            from core.cadquery_prompts import CADQUERY_SYSTEM_PROMPT
+            extra_backstory = (
+                "\n\nBefore deciding if specs are sufficient, ALWAYS call the "
+                "Query Design State tool first to check what the orchestrator "
+                "already knows (dimensions, material, features). Only say "
+                "NOT READY if the tool confirms information is truly missing.\n\n"
+                "When generating code, use the Execute CadQuery Code tool "
+                "to test your code. If it fails, fix the errors and try again. "
+                "Use the Validate CNC Manufacturability tool to check the result. "
+                "Output ONLY valid CadQuery Python that assigns result to a "
+                "cq.Workplane. Import cadquery as cq.\n\n"
+                f"CadQuery reference:\n{CADQUERY_SYSTEM_PROMPT}"
+            )
+        elif agent_id == "cnc":
+            extra_backstory = (
+                "\n\nBefore deciding if manufacturability info is sufficient, "
+                "ALWAYS call the Query Design State tool first to check what "
+                "the orchestrator already knows (material, dimensions, features, "
+                "constraints, axis). Only say NOT READY if the tool confirms "
+                "information is truly missing."
+            )
+        elif agent_id == "cam":
+            tools.append(generate_gcode_tool)
+            task_output_pydantic = CAMPlan
+        elif agent_id in ("design", "engineering"):
+            extra_backstory = (
+                "\n\nBefore asking clarifying questions, call the Query Design "
+                "State tool to check what is already known. Do NOT ask about "
+                "fields the tool shows as already provided (e.g. do not ask "
+                "'What material?' if the tool returns a known material)."
+            )
+
+        knowledge_sources = self._build_knowledge_sources() if agent_id in ("cnc", "cam") else []
+
+        crew_agent = Agent(
+            role=agent_def.role,
+            goal=agent_def.goal,
+            backstory=agent_def.backstory + extra_backstory,
+            llm=llm,
+            tools=tools,
+            verbose=False,
+            allow_delegation=False,
+            knowledge_sources=knowledge_sources if knowledge_sources else None,
+        )
+
+        task_description = (
+            f"{self.state.context}\n\n"
+            f"As the {agent_def.role}, respond to the user's latest message. "
+            f"Keep your response concise (2-4 sentences). "
+            f"Do NOT repeat anything from the conversation history. "
+            f"Add NEW information from your expertise.\n\n"
+            f"Build on other agents' input — agree, disagree, refine, or add."
+        )
+
+        if agent_id == "cad":
+            task_description += (
+                "\n\nFIRST call the Query Design State tool to check what the "
+                "orchestrator already knows. Use the returned 'known' fields "
+                "as your specs. Only say 'NOT READY:' listing truly missing "
+                "items if the tool shows critical gaps (no shape, no dimensions, "
+                "no features). If enough info exists, generate CadQuery code and "
+                "use the Execute CadQuery Code tool to verify it works."
+            )
+        elif agent_id == "cam":
+            task_description += (
+                "\n\nFIRST call the Query Design State tool to check available "
+                "specs. If there is no CAD model generated yet or the tool shows "
+                "critical gaps (no material, no dimensions), start with "
+                "'NOT READY:' and list only the truly missing items. "
+                "If enough info exists, analyze the part geometry and create an "
+                "optimal machining strategy. Select operations in order (roughing "
+                "before finishing). Use the Generate G-code Toolpath tool to create "
+                "the G-code."
+            )
+        elif agent_id == "cnc":
+            task_description += (
+                "\n\nFIRST call the Query Design State tool to check what the "
+                "orchestrator already knows about dimensions, material, and "
+                "constraints. Only say 'NOT READY:' listing truly missing items "
+                "if the tool confirms critical gaps. If enough info exists, "
+                "provide your manufacturability assessment."
+            )
+        else:
+            task_description += (
+                "\n\nFIRST call the Query Design State tool to see what is "
+                "already known. Only ask clarifying questions about fields "
+                "the tool shows as missing in YOUR domain."
+            )
+
+        if agent_id == "cad":
+            expected_output = "Valid CadQuery Python code or a 'NOT READY:' message."
+        elif agent_id in ("cnc", "cam"):
+            expected_output = "A concise expert assessment or a 'NOT READY:' message listing missing items."
+        else:
+            expected_output = "A concise response from your expert perspective (2-4 sentences)."
+
+        task = Task(
+            description=task_description,
+            expected_output=expected_output,
+            agent=crew_agent,
+            output_pydantic=task_output_pydantic,
+        )
+
+        return crew_agent, task
+
+    def _run_advisor_crew(self, advisor_ids: list[str]) -> list[AgentResponse]:
+        """Run advisory agents as a single sequential Crew. Returns list of AgentResponse."""
+        from crewai import Crew, Process
+
+        llm = self._build_llm()
+        agents_and_tasks = [self._build_crew_agent(aid, llm) for aid in advisor_ids]
+        crew_agents = [at[0] for at in agents_and_tasks]
+        crew_tasks = [at[1] for at in agents_and_tasks]
+
+        crew = Crew(agents=crew_agents, tasks=crew_tasks, process=Process.sequential, verbose=False)
+        crew_result = crew.kickoff()
+
+        responses = []
+        task_outputs = crew_result.tasks_output if hasattr(crew_result, 'tasks_output') else []
+        for i, agent_id in enumerate(advisor_ids):
+            raw = str(task_outputs[i]) if i < len(task_outputs) else (str(crew_result) if i == 0 else "")
+            if raw.strip():
+                responses.append(AgentResponse.from_agent(agent_id, raw.strip()))
+        return responses
+
+    def _run_single_agent_crew(self, agent_id: str) -> str:
+        """Run a single agent as a one-agent Crew. Returns raw output string."""
+        from crewai import Crew, Process
+
+        llm = self._build_llm()
+        crew_agent, task = self._build_crew_agent(agent_id, llm)
+        crew = Crew(agents=[crew_agent], tasks=[task], process=Process.sequential, verbose=False)
+        crew_result = crew.kickoff()
+
+        task_outputs = crew_result.tasks_output if hasattr(crew_result, 'tasks_output') else []
+        return str(task_outputs[0]).strip() if task_outputs else str(crew_result).strip()
+
+    def _run_cad_step(self):
+        """Shared CAD agent execution for READY, NOT_READY, and GENERATORS_ONLY paths."""
+        if "cad" not in self.state.active_agent_ids:
+            return
+
+        raw_output = self._run_single_agent_crew("cad")
+        if not raw_output:
+            return
+
+        if raw_output.upper().startswith("NOT READY:"):
+            self.state.cad_response = AgentResponse.from_agent("cad", raw_output)
+        else:
+            code = extract_code(raw_output)
+            if code:
+                self.state.cad_response = AgentResponse.from_agent("cad", "Model generated.", code=code)
+                self.state.cad_code = code
+            else:
+                self.state.cad_response = AgentResponse.from_agent("cad", raw_output)
+
+    def _run_cam_step(self):
+        """Shared CAM agent execution — only runs if cad_code exists."""
+        if "cam" not in self.state.active_agent_ids:
+            return
+        if self.state.cad_code is None:
+            return
+
+        raw_output = self._run_single_agent_crew("cam")
+        if not raw_output:
+            return
+
+        # Try to parse CAMPlan from pydantic output
+        from core.cam import CAMPlan
+        try:
+            import json
+            plan_data = json.loads(raw_output)
+            cam_plan = CAMPlan(**plan_data)
+            self.state.cam_plan = cam_plan
+            self.state.cam_response = AgentResponse.from_agent(
+                "cam",
+                f"Machining plan: {', '.join(cam_plan.operations)} | "
+                f"{cam_plan.tool_diameter}mm endmill | {cam_plan.post_processor}",
+            )
+        except (json.JSONDecodeError, ValueError, KeyError):
+            self.state.cam_response = AgentResponse.from_agent("cam", raw_output)
+```
+
+- [ ] **Step 4: Run tests to verify they pass**
+
+Run: `pytest tests/test_agent_flow.py -v`
+Expected: 32 passed
+
+- [ ] **Step 5: Run full test suite to check for regressions**
+
+Run: `pytest tests/ -x -q`
+Expected: All tests pass (existing tests unmodified)
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add agents/agent_flow.py tests/test_agent_flow.py
+git commit -m "feat: add AgentDispatchFlow with router-gated agent dispatch"
+```
+
+---
+
+### Task 6: Wire Flow into crew_orchestrator.py
+
+**Files:**
+- Modify: `agents/crew_orchestrator.py`
+
+- [ ] **Step 1: Replace _run_crew internals with Flow**
+
+Replace the entire `_run_crew` method body in `agents/crew_orchestrator.py`. Keep phase manipulation, pre-extraction, context building, and post-processing. Replace the Crew creation + kickoff + response parsing block with Flow instantiation.
+
+Replace `agents/crew_orchestrator.py` content:
+
+```python
+"""CrewAI orchestrator for all LLM backends.
+
+Uses CrewAI Flow for agent dispatch where each specialist agent gets its own
+focused LLM call with tools, structured output, and knowledge sources.
+
+Falls back to MockChatBackend if CrewAI is not installed.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from agents.base import BaseOrchestrator
+from agents.definitions import AGENTS
+from agents.design_state import DesignState, DesignPlan, extract_decisions, compute_score
+from agents.gap_analyzer import analyze_gaps, generate_question_cards
+from agents.orchestrator import _format_response
+from config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_OUTPUT_DIR = Path(__file__).parent.parent / "output"
+
+
+def _build_agent_context(
+    message: str,
+    history: list[dict],
+    design_state: DesignState,
+    max_history: int = 20,
+    approved_plan: DesignPlan | None = None,
+) -> str:
+    """Build a shared context string that each CrewAI agent receives."""
+    parts = []
+
+    if approved_plan:
+        parts.append(approved_plan.render_approved())
+    else:
+        spec = design_state.render()
+        if spec:
+            parts.append(f"## Current Design Spec\n{spec}")
+
+    recent = history[-max_history:] if len(history) > max_history else history
+    if recent:
+        lines = []
+        for msg in recent:
+            if msg.get("role") == "user":
+                lines.append(f"USER: {msg.get('content', '')}")
+            else:
+                aid = msg.get("agent_id", "unknown")
+                name = AGENTS.get(aid, AGENTS["design"]).name
+                lines.append(f"{name.upper()}: {msg.get('content', '')}")
+        parts.append("## Recent conversation\n" + "\n".join(lines))
+
+    parts.append(f"## User's latest message\n{message}")
+    return "\n\n".join(parts)
+
+
+def _is_plan_trigger(message: str) -> bool:
+    """Check if user message is requesting a plan review."""
+    lower = message.lower().strip()
+    for keyword in settings.planning.trigger_keywords:
+        if keyword in lower:
+            return True
+    return False
+
+
+def _get_crewai_model(backend_name: str) -> str:
+    """Get the CrewAI model string for a backend name."""
+    return settings.backends.crewai_models.get(backend_name, "gemini/gemini-2.5-flash")
+
+
+class CrewOrchestrator(BaseOrchestrator):
+    """Multi-call orchestrator using CrewAI Flow for agent dispatch.
+
+    Falls back to MockChatBackend if CrewAI is not installed.
+    """
+
+    def __init__(self, backend_name: str = "gemini", output_dir=None):
+        super().__init__(output_dir=output_dir or DEFAULT_OUTPUT_DIR)
+        self.backend_name = backend_name
+        self._crew_available = self._check_crewai()
+
+    @staticmethod
+    def _check_crewai() -> bool:
+        try:
+            import importlib.util
+            return importlib.util.find_spec("crewai") is not None
+        except (ImportError, ModuleNotFoundError):
+            return False
+
+    def chat_turn(
+        self,
+        message: str,
+        history: list[dict],
+        mentions: list[str] | None = None,
+        max_history: int = 30,
+        design_state: dict | None = None,
+        plan_context: bool = False,
+    ) -> dict:
+        # Phase: manual plan trigger (before crew/fallback dispatch)
+        state = DesignState(**(design_state or {}))
+        if state.phase == "exploring" and _is_plan_trigger(message):
+            score = compute_score(state)
+            plan = DesignPlan.from_state(state, confidence_score=score)
+            state.phase = "planning"
+            state.plan = plan
+            return {
+                "responses": [],
+                "preview": None,
+                "design_state": state.model_dump(),
+                "question_cards": [],
+            }
+
+        if not self._crew_available:
+            return self._fallback(message, history, mentions, max_history, design_state, plan_context)
+
+        try:
+            return self._run_crew(message, history, mentions, max_history, design_state, plan_context)
+        except Exception as exc:
+            logger.warning("CrewAI run failed (%s), falling back", exc, exc_info=True)
+            try:
+                return self._fallback(message, history, mentions, max_history, design_state, plan_context)
+            except Exception as fallback_exc:
+                logger.error("Fallback also failed: %s", fallback_exc, exc_info=True)
+                return {
+                    "responses": [_format_response(
+                        "design",
+                        f"Backend error: {exc}. Fallback also failed: {fallback_exc}. "
+                        f"Please check that your API key is set correctly.",
+                    )],
+                    "preview": None,
+                    "design_state": design_state or {},
+                    "question_cards": [],
+                }
+
+    def _run_crew(
+        self,
+        message: str,
+        history: list[dict],
+        mentions: list[str] | None,
+        max_history: int,
+        design_state_dict: dict | None,
+        plan_context: bool = False,
+    ) -> dict:
+        from agents.tools import set_design_state, get_last_shape
+        from agents.agent_flow import AgentFlowState, AgentDispatchFlow, collect_responses
+        from core.cam import generate_gcode
+
+        state = DesignState(**(design_state_dict or {}))
+
+        # Phase: if in planning and user sends a non-plan message, reset to exploring.
+        if state.phase == "planning" and not plan_context:
+            state.phase = "exploring"
+            state.plan = None
+
+        # Phase: approved — pass flag to Flow for routing
+        approved_plan = None
+        is_approved = state.phase == "approved" and state.plan is not None
+        if is_approved:
+            approved_plan = state.plan
+
+        # Pre-extract decisions from the user's current message
+        state = state.update_from_messages([], user_message=message)
+
+        # Expose design state to the orchestrator tool
+        set_design_state(state.model_dump())
+
+        context = _build_agent_context(message, history, state, max_history, approved_plan=approved_plan)
+
+        # ── Run Flow ──────────────────────────────────────────────────
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message=message,
+            context=context,
+            model_str=_get_crewai_model(self.backend_name),
+            mentions=list(mentions) if mentions else [],
+            is_approved_phase=is_approved,
+        ))
+        flow.kickoff()
+
+        # Read typed results
+        agent_responses = collect_responses(
+            flow.state.advisor_responses,
+            flow.state.cad_response,
+            flow.state.cam_response,
+        )
+        cad_code = flow.state.cad_code
+        cam_plan = flow.state.cam_plan
+
+        # ── Post-processing (unchanged logic) ─────────────────────────
+        # Convert AgentResponse models to dicts for API compatibility
+        responses = [r.model_dump() for r in agent_responses]
+        preview = None
+
+        # CAD execution + export
+        if cad_code:
+            shape = get_last_shape()
+            if shape is not None:
+                from core.executor import export_all
+                from core.validator import validate_for_cnc
+
+                part_name = message[:40].strip().replace(" ", "_").lower()
+                part_name = "".join(c for c in part_name if c.isalnum() or c == "_") or "part"
+                base_path = self.output_dir / part_name
+                try:
+                    export_all(shape, base_path)
+                except Exception:
+                    pass
+
+                execution_data = {"success": True}
+                try:
+                    bb = shape.val().BoundingBox()
+                    execution_data["volume_mm3"] = shape.val().Volume()
+                    execution_data["bounding_box_mm"] = [bb.xlen, bb.ylen, bb.zlen]
+                    execution_data["face_count"] = len(shape.faces().vals())
+                    execution_data["edge_count"] = len(shape.edges().vals())
+                except Exception:
+                    pass
+
+                validation = validate_for_cnc(shape, part_name=part_name)
+                preview = {
+                    "success": True,
+                    "part_name": part_name,
+                    "stl_url": f"/api/models/{part_name}.stl",
+                    "step_url": f"/api/models/{part_name}.step",
+                    "threemf_url": f"/api/models/{part_name}.3mf",
+                    "execution": execution_data,
+                    "validation": validation.model_dump(),
+                }
+
+        # G-code generation
+        if preview and preview.get("success") and cam_plan:
+            shape = get_last_shape()
+            if shape is not None:
+                cam_result = generate_gcode(
+                    shape=shape,
+                    operations=cam_plan.operations,
+                    tool_config=cam_plan.to_tool_config(),
+                    post_processor=cam_plan.post_processor,
+                )
+                preview["cam"] = cam_result.model_dump()
+                if cam_result.success and cam_result.gcode:
+                    part_name = preview["part_name"]
+                    gcode_path = self.output_dir / f"{part_name}.gcode"
+                    gcode_path.write_text(cam_result.gcode)
+                    preview["gcode_url"] = f"/api/models/{part_name}.gcode"
+
+        # Update design state
+        agent_msgs = [{"message": r.get("message", "")} for r in responses]
+        updated_state = extract_decisions(agent_msgs, state, message)
+
+        # Gap analysis
+        gap_result = analyze_gaps(responses)
+        question_cards = []
+        if gap_result.has_gaps:
+            cards = generate_question_cards(gap_result, updated_state, user_message=message)
+            question_cards = [c.model_dump() for c in cards]
+
+        # Auto-trigger plan if score crosses threshold
+        if updated_state.phase == "exploring":
+            score = compute_score(updated_state)
+            if score >= settings.planning.threshold:
+                plan = DesignPlan.from_state(updated_state, confidence_score=score)
+                updated_state.phase = "planning"
+                updated_state.plan = plan
+
+        # If approved and CAD said NOT READY, reset
+        if state.phase == "approved":
+            for r in responses:
+                if r.get("agent_id") == "cad" and r.get("message", "").upper().startswith("NOT READY:"):
+                    updated_state.phase = "exploring"
+                    updated_state.plan = None
+                    break
+
+        return {
+            "responses": responses,
+            "preview": preview,
+            "design_state": updated_state.model_dump(),
+            "question_cards": question_cards,
+        }
+
+    def _fallback(
+        self,
+        message: str,
+        history: list[dict],
+        mentions: list[str] | None,
+        max_history: int,
+        design_state: dict | None,
+        plan_context: bool = False,
+    ) -> dict:
+        """Fall back to MockChatBackend."""
+        from agents.tools import set_design_state
+        from agents.orchestrator import MockChatBackend
+
+        state = DesignState(**(design_state or {}))
+        state = state.update_from_messages([], user_message=message)
+        set_design_state(state.model_dump())
+
+        mock = MockChatBackend(output_dir=self.output_dir)
+        result = mock.chat_turn(message, history, mentions, design_state=state.model_dump(), plan_context=plan_context)
+        if "question_cards" not in result:
+            result_responses = result.get("responses", [])
+            gap_result = analyze_gaps(result_responses)
+            if gap_result.has_gaps:
+                cards = generate_question_cards(gap_result, state, user_message=message)
+                result["question_cards"] = [c.model_dump() for c in cards]
+            else:
+                result["question_cards"] = []
+        return result
+```
+
+- [ ] **Step 2: Run existing tests to verify no regressions**
+
+Run: `pytest tests/test_crew_orchestrator.py -v`
+Expected: All tests pass (fallback path, planning phase, gap analysis)
+
+- [ ] **Step 3: Run full test suite**
+
+Run: `pytest tests/ -x -q`
+Expected: All tests pass
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add agents/crew_orchestrator.py
+git commit -m "refactor: replace Crew.kickoff() with AgentDispatchFlow in _run_crew"
+```
+
+---
+
+### Task 7: Remove RoutingEngine + update MockChatBackend
+
+**Files:**
+- Delete: `agents/routing.py`
+- Modify: `agents/orchestrator.py`
+- Modify: `tests/test_routing.py`
+- Modify: `tests/test_cam_routing.py`
+
+- [ ] **Step 1: Update MockChatBackend to use route_agents**
+
+In `agents/orchestrator.py`, replace the `RoutingEngine` import and `_router` singleton with `route_agents` from `agent_flow`:
+
+Replace:
+```python
+from agents.routing import RoutingEngine
+```
+with:
+```python
+from agents.agent_flow import route_agents
+```
+
+Replace:
+```python
+_router = RoutingEngine()
+```
+with nothing (remove the line).
+
+In `MockChatBackend.chat_turn()`, replace:
+```python
+            active = _router.route(message)
+```
+with:
+```python
+            active = route_agents(message, mentions=[], is_approved_phase=False)
+```
+
+- [ ] **Step 2: Update test_routing.py to test route_agents**
+
+Replace `tests/test_routing.py`:
+
+```python
+"""Tests for agent routing via route_agents function."""
+
+from agents.agent_flow import route_agents
+
+
+class TestRouteAgents:
+    def test_route_design_keywords(self):
+        agents = route_agents("I want a sleek design with smooth shape", [], False)
+        assert "design" in agents
+
+    def test_route_engineering_keywords(self):
+        agents = route_agents("Use M6 bolts with 3mm wall thickness in aluminum", [], False)
+        assert "engineering" in agents
+
+    def test_route_cnc_keywords(self):
+        agents = route_agents("Can this be machined on a 3-axis CNC mill?", [], False)
+        assert "cnc" in agents
+
+    def test_route_default_when_no_match(self):
+        agents = route_agents("hello there", [], False)
+        assert agents == ["design", "engineering"]
+
+    def test_route_max_three_agents(self):
+        agents = route_agents("design shape in aluminum for CNC machining, generate preview", [], False)
+        assert len(agents) <= 3
+
+    def test_has_cad_trigger_true(self):
+        agents = route_agents("Generate a preview", [], False)
+        assert "cad" in agents
+
+    def test_has_cad_trigger_false(self):
+        agents = route_agents("hello there", [], False)
+        assert "cad" not in agents
+```
+
+- [ ] **Step 3: Update test_cam_routing.py to test route_agents**
+
+Replace `tests/test_cam_routing.py`:
+
+```python
+"""Tests for CAM agent routing and definitions."""
+
+from agents.agent_flow import route_agents
+
+
+class TestCAMRouting:
+    def test_route_cam_keywords_toolpath(self):
+        agents = route_agents("Generate a toolpath for this part", [], False)
+        assert "cam" in agents
+
+    def test_route_cam_keywords_gcode(self):
+        agents = route_agents("Create gcode for CNC milling", [], False)
+        assert "cam" in agents
+
+    def test_route_cam_keywords_slicer(self):
+        agents = route_agents("Run the slicer on this model", [], False)
+        assert "cam" in agents
+
+    def test_cam_agent_exists_in_definitions(self):
+        from agents.definitions import AGENTS
+        assert "cam" in AGENTS
+
+    def test_cam_agent_has_color(self):
+        from agents.definitions import AGENT_COLORS
+        assert "cam" in AGENT_COLORS
+
+    def test_cam_agent_has_name(self):
+        from agents.definitions import AGENT_NAMES
+        assert AGENT_NAMES["cam"] == "CAM Agent"
+```
+
+- [ ] **Step 4: Delete agents/routing.py**
+
+```bash
+git rm agents/routing.py
+```
+
+- [ ] **Step 5: Run all tests**
+
+Run: `pytest tests/ -x -q`
+Expected: All tests pass
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add agents/orchestrator.py tests/test_routing.py tests/test_cam_routing.py
+git commit -m "refactor: remove RoutingEngine, use route_agents from agent_flow"
+```
+
+---
+
+### Task 8: Final validation + cleanup
+
+**Files:**
+- Verify all files
+
+- [ ] **Step 1: Run full test suite**
+
+Run: `pytest tests/ -v`
+Expected: All tests pass with no warnings about missing modules
+
+- [ ] **Step 2: Check for stale imports of routing.py**
+
+Run: `grep -r "from agents.routing" --include="*.py" .`
+Expected: No results (all imports removed)
+
+Run: `grep -r "RoutingEngine" --include="*.py" .`
+Expected: No results
+
+- [ ] **Step 3: Verify API contract unchanged**
+
+Run: `python -c "from agents.crew_orchestrator import CrewOrchestrator; o = CrewOrchestrator(); print(type(o))"`
+Expected: `<class 'agents.crew_orchestrator.CrewOrchestrator'>`
+
+Run: `python -c "from agents.orchestrator import get_orchestrator; o = get_orchestrator('mock'); print(type(o))"`
+Expected: `<class 'agents.orchestrator.MockChatBackend'>`
+
+- [ ] **Step 4: Verify Flow import works**
+
+Run: `python -c "from agents.agent_flow import AgentDispatchFlow, AgentResponse, AgentFlowState, route_agents, check_readiness, collect_responses, extract_code; print('All imports OK')"`
+Expected: `All imports OK`
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add -A
+git commit -m "chore: final cleanup after CrewAI Flow refactor"
+```