feat: add memory recall/remember, planning, and collaboration to Flow

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

agents/agent_flow.py

@@ -275,6 +275,36 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
 
     # ── Private helpers ──────────────────────────────────────────────
 
+    _memory = None  # Set by CrewOrchestrator before kickoff
+
+    def _recall_for_agent(self, agent_id: str) -> str:
+        """Recall relevant memories for this agent, formatted as context."""
+        if self._memory is None:
+            return ""
+        try:
+            from config.settings import settings
+            matches = self._memory.recall(
+                self.state.message,
+                scope=f"/agent/{agent_id}",
+                limit=settings.memory.recall_limit,
+                depth=settings.memory.recall_depth,
+            )
+        except Exception:
+            return ""
+        if not matches:
+            return ""
+        lines = [f"- {m.record.content}" for m in matches]
+        return "## Relevant context from prior turns\n" + "\n".join(lines)
+
+    def _remember_response(self, agent_id: str, content: str):
+        """Store an agent's response in its scoped memory."""
+        if self._memory is None:
+            return
+        try:
+            self._memory.remember(content, scope=f"/agent/{agent_id}")
+        except Exception:
+            pass
+
     def _build_llm(self):
         from crewai import LLM
         return LLM(model=self.state.model_str, temperature=settings.temperature)
@@ -345,12 +375,18 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
             llm=llm,
             tools=tools,
             verbose=False,
-            allow_delegation=False,
+            allow_delegation=settings.crew.collaboration and agent_id in ADVISOR_IDS,
             knowledge_sources=knowledge_sources if knowledge_sources else None,
         )
 
-        task_description = (
-            f"{self.state.context}\n\n"
+        memories = self._recall_for_agent(agent_id)
+
+        context_parts = [self.state.context]
+        if memories:
+            context_parts.append(memories)
+
+        task_description = "\n\n".join(context_parts) + "\n\n"
+        task_description += (
             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. "
@@ -415,6 +451,8 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
             agents=[p[0] for p in pairs],
             tasks=[p[1] for p in pairs],
             process=Process.sequential,
+            planning=settings.crew.planning,
+            planning_llm=self._build_llm(),
             verbose=False,
         )
         crew_result = crew.kickoff()
@@ -424,6 +462,7 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
             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()))
+                self._remember_response(agent_id, raw.strip())
         return responses
 
     def _run_single_agent_crew(self, agent_id):
@@ -435,7 +474,14 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
         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 = Crew(
+            agents=[crew_agent],
+            tasks=[task],
+            process=Process.sequential,
+            planning=settings.crew.planning,
+            planning_llm=self._build_llm(),
+            verbose=False,
+        )
         crew_result = crew.kickoff()
         task_outputs = crew_result.tasks_output if hasattr(crew_result, 'tasks_output') else []
         if not task_outputs:
@@ -462,6 +508,9 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
             else:
                 self.state.cad_response = AgentResponse.from_agent("cad", raw_output)
 
+        if self.state.cad_response is not None:
+            self._remember_response("cad", raw_output)
+
     def _run_cam_step(self):
         if "cam" not in self.state.active_agent_ids:
             return
@@ -483,3 +532,6 @@ class AgentDispatchFlow(Flow[AgentFlowState]):
             )
         except (json.JSONDecodeError, ValueError, KeyError):
             self.state.cam_response = AgentResponse.from_agent("cam", raw_output)
+
+        if self.state.cam_response is not None:
+            self._remember_response("cam", raw_output)

tests/test_agent_flow.py

@@ -421,3 +421,90 @@ class TestAgentDispatchFlow:
         )
         assert len(results) >= 2  # advisor + cad
         assert flow.state.cad_code is not None
+
+
+class TestMemoryHelpers:
+    def test_recall_returns_empty_when_no_memory(self):
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="bracket design",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = None
+        result = flow._recall_for_agent("design")
+        assert result == ""
+
+    def test_recall_formats_matches(self):
+        mock_memory = MagicMock()
+        mock_match = MagicMock()
+        mock_match.record.content = "L-bracket with fillets"
+        mock_memory.recall.return_value = [mock_match]
+
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="bracket",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = mock_memory
+        result = flow._recall_for_agent("design")
+        assert "## Relevant context from prior turns" in result
+        assert "L-bracket with fillets" in result
+        mock_memory.recall.assert_called_once()
+
+    def test_recall_returns_empty_when_no_matches(self):
+        mock_memory = MagicMock()
+        mock_memory.recall.return_value = []
+
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="bracket",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = mock_memory
+        result = flow._recall_for_agent("design")
+        assert result == ""
+
+    def test_remember_stores_with_scope(self):
+        mock_memory = MagicMock()
+
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="test",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = mock_memory
+        flow._remember_response("engineering", "Use 3mm walls in aluminum.")
+        mock_memory.remember.assert_called_once_with(
+            "Use 3mm walls in aluminum.",
+            scope="/agent/engineering",
+        )
+
+    def test_remember_noop_when_no_memory(self):
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="test",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = None
+        flow._remember_response("design", "test")  # Should not raise
+
+
+class TestCollaborationFlag:
+    def test_advisors_get_delegation(self):
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="test",
+            context="",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = None
+        from crewai import LLM
+        llm = LLM(model="gemini/gemini-2.5-flash", temperature=0.2)
+        agent, task = flow._build_crew_agent("design", llm)
+        assert agent.allow_delegation is True
+
+    def test_generators_no_delegation(self):
+        flow = AgentDispatchFlow(initial_state=AgentFlowState(
+            message="test",
+            context="",
+            model_str="gemini/gemini-2.5-flash",
+        ))
+        flow._memory = None
+        from crewai import LLM
+        llm = LLM(model="gemini/gemini-2.5-flash", temperature=0.2)
+        agent, task = flow._build_crew_agent("cad", llm)
+        assert agent.allow_delegation is False