From d787844a37312a2faefb7a64f66853939a287e0d Mon Sep 17 00:00:00 2001 From: Darvell Long Date: Wed, 22 Apr 2026 15:10:11 -0700 Subject: [PATCH 1/2] fix: preserve CrewAI tool history across turns Keep CrewAI tool events in converted history and replay the full room transcript on each kickoff so multi-step peer workflows retain more execution context. --- src/thenvoi/adapters/crewai.py | 35 ++-- src/thenvoi/converters/crewai.py | 168 +++++++++++++----- tests/adapters/test_crewai_adapter.py | 70 ++++++++ tests/converters/test_crewai.py | 91 +++++++++- tests/framework_configs/converters.py | 1 - .../test_converter_conformance.py | 2 +- 6 files changed, 301 insertions(+), 66 deletions(-) diff --git a/src/thenvoi/adapters/crewai.py b/src/thenvoi/adapters/crewai.py index c01f572c2..9b47592b5 100644 --- a/src/thenvoi/adapters/crewai.py +++ b/src/thenvoi/adapters/crewai.py @@ -305,9 +305,7 @@ async def _process_message( """Internal message processing logic.""" if is_session_bootstrap: if history: - self._message_history[room_id] = [ - {"role": h["role"], "content": h["content"]} for h in history - ] + self._message_history[room_id] = [dict(h) for h in history] logger.info( "Room %s: Loaded %s historical messages", room_id, @@ -317,50 +315,43 @@ async def _process_message( self._message_history[room_id] = [] logger.info("Room %s: No historical messages found", room_id) elif room_id not in self._message_history: - self._message_history[room_id] = [] - - messages = [] - - if is_session_bootstrap and self._message_history.get(room_id): - history_text = "\n".join( - f"{m['role']}: {m['content']}" for m in self._message_history[room_id] - ) - messages.append( - { - "role": "user", - "content": f"[Previous conversation:]\n{history_text}", - } - ) + self._message_history[room_id] = [dict(h) for h in history] if participants_msg: - messages.append( + self._message_history[room_id].append( { "role": "user", "content": f"[System]: {participants_msg}", + "sender": "System", + "sender_type": "System", } ) logger.info("Room %s: Participants updated", room_id) if contacts_msg: - messages.append( + self._message_history[room_id].append( { "role": "user", "content": f"[System]: {contacts_msg}", + "sender": "System", + "sender_type": "System", } ) logger.info("Room %s: Contacts broadcast received", room_id) user_message = msg.format_for_llm() - messages.append({"role": "user", "content": user_message}) - self._message_history[room_id].append( { "role": "user", "content": user_message, + "sender": msg.sender_name, + "sender_type": msg.sender_type, } ) - total_messages = len(self._message_history[room_id]) + messages = [dict(m) for m in self._message_history[room_id]] + + total_messages = len(messages) logger.info( "Room %s: Processing with %s messages (first_msg=%s)", room_id, diff --git a/src/thenvoi/converters/crewai.py b/src/thenvoi/converters/crewai.py index c4895410e..748e7fda1 100644 --- a/src/thenvoi/converters/crewai.py +++ b/src/thenvoi/converters/crewai.py @@ -2,10 +2,14 @@ from __future__ import annotations +import json +import logging from typing import Any from thenvoi.core.protocols import HistoryConverter +logger = logging.getLogger(__name__) + # Type alias for CrewAI messages (simple dict format) CrewAIMessages = list[dict[str, Any]] @@ -14,56 +18,142 @@ class CrewAIHistoryConverter(HistoryConverter[CrewAIMessages]): """ Converts platform history to CrewAI-compatible message format. - Output: [{"role": "user", "content": "...", "sender": "..."}] + Output: [{"role": "user"|"assistant", "content": "...", "sender": "..."}] - Note: - - Only converts text messages (tool_call/tool_result events are skipped) - - User messages include sender name for context - - Other agents' messages are included with role "assistant" - - This agent's messages are skipped (redundant with CrewAI's internal state) + Notes: + - Text messages preserve sender context + - Other agents' text messages remain "assistant" + - This agent's own text messages are skipped (CrewAI already tracks them in-run) + - Tool events are converted to replayable text so CrewAI can see prior actions/results """ def __init__(self, agent_name: str = ""): - """ - Initialize converter. - - Args: - agent_name: Name of this agent. Messages from this agent are skipped - (they're redundant with internal state). Messages from other - agents are included with their sender info. - """ self._agent_name = agent_name def set_agent_name(self, name: str) -> None: - """ - Set agent name so converter knows which messages to skip. - - Args: - name: Name of this agent - """ self._agent_name = name + @staticmethod + def _dump_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, default=str) + + @staticmethod + def _user_content(content: str, sender_name: str) -> str: + return f"[{sender_name}]: {content}" if sender_name else content + + def _convert_tool_call(self, hist: dict[str, Any]) -> dict[str, Any] | None: + content = hist.get("content", "") + sender_name = hist.get("sender_name", self._agent_name) + sender_type = hist.get("sender_type", "Agent") + + try: + event = json.loads(content) + except json.JSONDecodeError: + logger.warning("Failed to parse CrewAI tool_call: %s", repr(content[:100])) + return None + + tool_name = event.get("name") or event.get("tool") + if not tool_name: + logger.warning( + "Skipping CrewAI tool_call with missing tool name: %s", + repr(content[:100]), + ) + return None + + tool_input = event.get("args") + if tool_input is None: + tool_input = event.get("input", {}) + + rendered = f"[Tool Call] {tool_name}" + if tool_input not in ({}, None, ""): + rendered = f"{rendered} {self._dump_json(tool_input)}" + + return { + "role": "assistant", + "content": rendered, + "sender": sender_name, + "sender_type": sender_type, + } + + def _convert_tool_result(self, hist: dict[str, Any]) -> dict[str, Any] | None: + content = hist.get("content", "") + sender_name = hist.get("sender_name", "") + sender_type = hist.get("sender_type", "System") + + try: + event = json.loads(content) + except json.JSONDecodeError: + logger.warning( + "Failed to parse CrewAI tool_result: %s", repr(content[:100]) + ) + return None + + tool_name = event.get("name") or event.get("tool") + if not tool_name: + logger.warning( + "Skipping CrewAI tool_result with missing tool name: %s", + repr(content[:100]), + ) + return None + + is_error = bool(event.get("is_error")) or "error" in event + output = event.get("output") + if output is None: + output = event.get("result") + if output is None and "error" in event: + output = event.get("error") + if output is None: + logger.warning( + "Skipping CrewAI tool_result with missing output/result: %s", + repr(content[:100]), + ) + return None + + rendered_output = output if isinstance(output, str) else self._dump_json(output) + prefix = "[Tool Error]" if is_error else "[Tool Result]" + + return { + "role": "user", + "content": self._user_content( + f"{prefix} {tool_name}: {rendered_output}", + sender_name, + ), + "sender": sender_name, + "sender_type": sender_type, + } + def convert(self, raw: list[dict[str, Any]]) -> CrewAIMessages: """Convert platform history to CrewAI format.""" messages: CrewAIMessages = [] for hist in raw: message_type = hist.get("message_type", "text") - - # Only convert text messages - if message_type != "text": - continue - role = hist.get("role", "user") content = hist.get("content", "") sender_name = hist.get("sender_name", "") sender_type = hist.get("sender_type", "User") + if message_type == "thought": + continue + + if message_type == "tool_call": + tool_call = self._convert_tool_call(hist) + if tool_call is not None: + messages.append(tool_call) + continue + + if message_type == "tool_result": + tool_result = self._convert_tool_result(hist) + if tool_result is not None: + messages.append(tool_result) + continue + + if message_type != "text": + continue + if role == "assistant" and sender_name == self._agent_name: - # Skip THIS agent's text (redundant with CrewAI state) continue - elif role == "assistant": - # Other agents' messages + if role == "assistant": messages.append( { "role": "assistant", @@ -72,17 +162,15 @@ def convert(self, raw: list[dict[str, Any]]) -> CrewAIMessages: "sender_type": sender_type, } ) - else: - # User messages - messages.append( - { - "role": "user", - "content": f"[{sender_name}]: {content}" - if sender_name - else content, - "sender": sender_name, - "sender_type": sender_type, - } - ) + continue + + messages.append( + { + "role": "user", + "content": self._user_content(content, sender_name), + "sender": sender_name, + "sender_type": sender_type, + } + ) return messages diff --git a/tests/adapters/test_crewai_adapter.py b/tests/adapters/test_crewai_adapter.py index 58fae77c2..4926cb1fa 100644 --- a/tests/adapters/test_crewai_adapter.py +++ b/tests/adapters/test_crewai_adapter.py @@ -355,6 +355,76 @@ async def test_loads_existing_history( assert len(adapter._message_history["room-123"]) >= 3 + @pytest.mark.asyncio + async def test_replays_full_history_to_kickoff( + self, CrewAIAdapter, sample_message, mock_tools, mock_crewai_agent + ): + adapter = CrewAIAdapter() + await adapter.on_started("TestBot", "Test bot") + adapter._crewai_agent = mock_crewai_agent + adapter._message_history["room-123"] = [ + { + "role": "user", + "content": "[Bob]: Step 1", + "sender": "Bob", + "sender_type": "User", + }, + { + "role": "assistant", + "content": '[Tool Call] thenvoi_lookup_peers {"page": 1}', + "sender": "TestBot", + "sender_type": "Agent", + }, + { + "role": "user", + "content": '[TestBot]: [Tool Result] thenvoi_lookup_peers: {"peers": []}', + "sender": "TestBot", + "sender_type": "Agent", + }, + ] + + await adapter.on_message( + msg=sample_message, + tools=mock_tools, + history=[], + participants_msg=None, + contacts_msg=None, + is_session_bootstrap=False, + room_id="room-123", + ) + + call_args = mock_crewai_agent.kickoff_async.call_args + messages = call_args[0][0] + + assert messages[0]["content"] == "[Bob]: Step 1" + assert "[Tool Call] thenvoi_lookup_peers" in messages[1]["content"] + assert "[Tool Result] thenvoi_lookup_peers" in messages[2]["content"] + assert messages[3]["content"] == "[Alice]: Hello, agent!" + + @pytest.mark.asyncio + async def test_appends_system_updates_to_room_history( + self, CrewAIAdapter, sample_message, mock_tools, mock_crewai_agent + ): + adapter = CrewAIAdapter() + await adapter.on_started("TestBot", "Test bot") + adapter._crewai_agent = mock_crewai_agent + + await adapter.on_message( + msg=sample_message, + tools=mock_tools, + history=[], + participants_msg="Alice joined the room", + contacts_msg="[Contacts]: @alice is now a contact", + is_session_bootstrap=True, + room_id="room-123", + ) + + history = adapter._message_history["room-123"] + assert history[0]["content"] == "[System]: Alice joined the room" + assert history[1]["content"] == "[System]: [Contacts]: @alice is now a contact" + assert history[0]["sender"] == "System" + assert history[1]["sender"] == "System" + @pytest.mark.asyncio async def test_calls_kickoff_async( self, CrewAIAdapter, sample_message, mock_tools, mock_crewai_agent diff --git a/tests/converters/test_crewai.py b/tests/converters/test_crewai.py index fca3b537a..080e75044 100644 --- a/tests/converters/test_crewai.py +++ b/tests/converters/test_crewai.py @@ -3,8 +3,8 @@ Tests for shared converter behavior (user messages, agent filtering, empty history, edge cases, output shape) live in tests/framework_conformance/test_converter_conformance.py. -This file contains CrewAI-specific assistant message formatting and crew -workflow tests. +This file contains CrewAI-specific assistant message formatting, tool event +replay, and crew workflow tests. """ from thenvoi.converters.crewai import CrewAIHistoryConverter @@ -110,3 +110,90 @@ def test_preserves_sender_type(self): assert result[0]["sender_type"] == "User" assert result[1]["sender_type"] == "Agent" + + +class TestToolEventReplay: + """Tests for CrewAI tool event replay formatting.""" + + def test_converts_tool_call_to_assistant_replay_message(self): + converter = CrewAIHistoryConverter(agent_name="Router") + raw = [ + { + "role": "assistant", + "content": '{"tool": "thenvoi_lookup_peers", "input": {"page": 1}}', + "sender_name": "Router", + "sender_type": "Agent", + "message_type": "tool_call", + } + ] + + result = converter.convert(raw) + + assert len(result) == 1 + assert result[0]["role"] == "assistant" + assert result[0]["content"] == '[Tool Call] thenvoi_lookup_peers {"page": 1}' + assert result[0]["sender"] == "Router" + assert result[0]["sender_type"] == "Agent" + + def test_converts_tool_result_to_user_replay_message(self): + converter = CrewAIHistoryConverter(agent_name="Router") + raw = [ + { + "role": "assistant", + "content": '{"tool": "thenvoi_lookup_peers", "result": {"peers": ["a"]}}', + "sender_name": "Router", + "sender_type": "Agent", + "message_type": "tool_result", + } + ] + + result = converter.convert(raw) + + assert len(result) == 1 + assert result[0]["role"] == "user" + assert ( + result[0]["content"] + == '[Router]: [Tool Result] thenvoi_lookup_peers: {"peers": ["a"]}' + ) + assert result[0]["sender"] == "Router" + assert result[0]["sender_type"] == "Agent" + + def test_converts_tool_error_to_user_replay_message(self): + converter = CrewAIHistoryConverter(agent_name="Router") + raw = [ + { + "role": "assistant", + "content": '{"tool": "thenvoi_add_participant", "error": "No such peer"}', + "sender_name": "Router", + "sender_type": "Agent", + "message_type": "tool_result", + } + ] + + result = converter.convert(raw) + + assert len(result) == 1 + assert result[0]["role"] == "user" + assert ( + result[0]["content"] + == "[Router]: [Tool Error] thenvoi_add_participant: No such peer" + ) + + def test_skips_malformed_tool_events(self): + converter = CrewAIHistoryConverter() + raw = [ + { + "role": "assistant", + "content": "not json", + "message_type": "tool_call", + }, + { + "role": "assistant", + "content": '{"tool": "thenvoi_lookup_peers"}', + "message_type": "tool_result", + }, + ] + + result = converter.convert(raw) + + assert result == [] diff --git a/tests/framework_configs/converters.py b/tests/framework_configs/converters.py index b8f105a6d..e5f255d47 100644 --- a/tests/framework_configs/converters.py +++ b/tests/framework_configs/converters.py @@ -166,7 +166,6 @@ def _build_crewai_config() -> ConverterConfig: display_name="CrewAI", converter_factory=_crewai_factory, empty_result=[], - skips_tool_events=True, empty_sender_behavior=SenderBehavior.CONTENT_AS_IS, missing_sender_behavior=SenderBehavior.CONTENT_AS_IS, has_sender_metadata=True, diff --git a/tests/framework_conformance/test_converter_conformance.py b/tests/framework_conformance/test_converter_conformance.py index 562f8f142..2ae086666 100644 --- a/tests/framework_conformance/test_converter_conformance.py +++ b/tests/framework_conformance/test_converter_conformance.py @@ -438,7 +438,7 @@ class TestToolEventHandling: def test_tool_events_skipped_for_simple_converters( self, converter_config, make_converter, output ): - """CrewAI and Parlant skip tool_call/tool_result messages entirely.""" + """Converters marked as text-only skip tool_call/tool_result messages entirely.""" if not converter_config.skips_tool_events: pytest.skip( f"{converter_config.display_name} processes tool events (tested below)" From e6abfa359a264ee82a777781b4b3cd7ccf2f4b8d Mon Sep 17 00:00:00 2001 From: Darvell Long Date: Wed, 22 Apr 2026 15:31:39 -0700 Subject: [PATCH 2/2] fix: prioritize platform instructions in CrewAI backstory Prepend the Thenvoi platform prompt before custom CrewAI backstory text so the adapter's routing and tool-use rules land earlier in the backstory block. --- src/thenvoi/adapters/crewai.py | 6 +++--- tests/adapters/test_crewai_adapter.py | 18 ++++++++++++++++++ 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/src/thenvoi/adapters/crewai.py b/src/thenvoi/adapters/crewai.py index 9b47592b5..2b2c231fe 100644 --- a/src/thenvoi/adapters/crewai.py +++ b/src/thenvoi/adapters/crewai.py @@ -185,15 +185,15 @@ async def on_started(self, agent_name: str, agent_description: str) -> None: goal = self.goal or agent_description or "Help users accomplish their tasks" if self.backstory: - # User provided full backstory -- append capability-gated platform - # instructions so the LLM knows about memory/contact tools if enabled. + # User provided full backstory -- prepend capability-gated platform + # instructions so Thenvoi rules land before custom backstory text. platform_prompt = render_system_prompt( agent_name=agent_name, agent_description=agent_description, custom_section=self.custom_section or "", features=self.features, ) - backstory = f"{self.backstory}\n\n{platform_prompt}" + backstory = f"{platform_prompt}\n\n{self.backstory}" else: backstory = render_system_prompt( agent_name=agent_name, diff --git a/tests/adapters/test_crewai_adapter.py b/tests/adapters/test_crewai_adapter.py index 4926cb1fa..2c6a6c442 100644 --- a/tests/adapters/test_crewai_adapter.py +++ b/tests/adapters/test_crewai_adapter.py @@ -258,6 +258,24 @@ async def test_uses_custom_role_goal_backstory(self, CrewAIAdapter, crewai_mocks assert call_kwargs["goal"] == "Find information" assert "Expert researcher" in call_kwargs["backstory"] + @pytest.mark.asyncio + async def test_prepends_platform_instructions_before_custom_backstory( + self, CrewAIAdapter, crewai_mocks + ): + crewai_mocks.Agent.reset_mock() + + adapter = CrewAIAdapter(backstory="Custom backstory goes here.") + await adapter.on_started(agent_name="TestBot", agent_description="A test bot") + + call_kwargs = crewai_mocks.Agent.call_args[1] + backstory = call_kwargs["backstory"] + + assert "Multi-participant chat" in backstory + assert "Custom backstory goes here." in backstory + assert backstory.index("Multi-participant chat") < backstory.index( + "Custom backstory goes here." + ) + @pytest.mark.asyncio async def test_uses_agent_name_as_default_role(self, CrewAIAdapter, crewai_mocks): crewai_mocks.Agent.reset_mock()