fix(flow): don't double-append the turn reply when a handler trims history

handle_turn() (and stream_turn) decided "did the handler append its
reply?" by snapshotting the assistant-message count before kickoff and
appending the stringified result when the count came back unchanged. A
handler that appends its reply and then trims state.messages to a cap —
a normal bounded-context pattern — left the count unchanged, so the
fallback appended the reply a second time on every turn once trimming
engaged, and the duplicates then crowded real turns out of the capped
window.

Replace the count heuristic with an explicit per-turn flag:
append_assistant_message() sets _assistant_reply_appended, handle_turn
and stream_turn clear it before kickoff and only fall back when no
assistant message was appended during the turn. The now-unused
_assistant_message_count() helper is removed.

Fixes EPD-181.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -133,6 +133,7 @@ class _ConversationalMixin:
         _pending_user_message: str | dict[str, Any] | None
         _pending_intents: Sequence[str] | None
         _pending_intent_llm: str | BaseLLM | None
+        _assistant_reply_appended: bool
 
         def _clear_or_listeners(self) -> None:
             pass
@@ -310,11 +311,11 @@ class _ConversationalMixin:
             if "from_checkpoint" not in kickoff_kwargs:
                 self._reset_turn_execution_state()
 
-            assistant_count = self._assistant_message_count()
+            object.__setattr__(self, "_assistant_reply_appended", False)
             result = self.kickoff(inputs={"id": sid}, **kickoff_kwargs)
             if (
                 result is not None
-                and self._assistant_message_count() == assistant_count
+                and not self._assistant_reply_appended
                 and self._is_public_turn_result(result)
             ):
                 self.append_assistant_message(self._stringify_result(result))
@@ -387,7 +388,7 @@ class _ConversationalMixin:
                 if "from_checkpoint" not in kickoff_kwargs:
                     self._reset_turn_execution_state()
 
-                assistant_count = self._assistant_message_count()
+                object.__setattr__(self, "_assistant_reply_appended", False)
                 original_stream = bool(getattr(self, "stream", False))
                 original_streaming_turn = getattr(
                     self, "_streaming_conversation_turn", False
@@ -403,7 +404,7 @@ class _ConversationalMixin:
                     )
                 if (
                     result is not None
-                    and self._assistant_message_count() == assistant_count
+                    and not self._assistant_reply_appended
                     and self._is_public_turn_result(result)
                 ):
                     self.append_assistant_message(self._stringify_result(result))
@@ -618,6 +619,9 @@ class _ConversationalMixin:
         metadata: dict[str, Any] | None = None,
     ) -> None:
         """Append a final user-visible assistant message."""
+        # Explicit signal for handle_turn's "did the handler reply?" check.
+        # A count heuristic breaks when handlers trim history mid-turn.
+        object.__setattr__(self, "_assistant_reply_appended", True)
         state = cast(ConversationState, self.state)
         state.messages.append(
             ConversationMessage(
@@ -788,6 +792,8 @@ class _ConversationalMixin:
             object.__setattr__(self, "_pending_intent_llm", None)
         if not hasattr(self, "_streaming_conversation_turn"):
             object.__setattr__(self, "_streaming_conversation_turn", False)
+        if not hasattr(self, "_assistant_reply_appended"):
+            object.__setattr__(self, "_assistant_reply_appended", False)
 
     def _create_default_extension_state(self) -> ConversationState | None:
         initial_state_t = getattr(self, "_initial_state_t", None)
@@ -1107,10 +1113,6 @@ class _ConversationalMixin:
             return "public"
         return "private"
 
-    def _assistant_message_count(self) -> int:
-        state = cast(ConversationState, self.state)
-        return sum(1 for message in state.messages if message.role == "assistant")
-
     def _is_public_turn_result(self, result: Any) -> bool:
         if not isinstance(result, str):
             return False

lib/crewai/tests/test_flow_conversation.py

@@ -1555,6 +1555,71 @@ class TestConversationalFlow:
         )
 
 
+class TestHandleTurnReplyFallback:
+    """Regression tests for EPD-181: ``handle_turn()`` decided "did the
+    handler append its reply?" by comparing assistant-message counts. A
+    handler that appends its reply AND trims history to a cap left the count
+    unchanged, so the fallback appended the reply a second time — every turn,
+    once trimming engaged. The check now uses an explicit appended-this-turn
+    flag.
+    """
+
+    MAX_MESSAGES = 4
+
+    def _make_bot(self) -> ConversationalFlow:
+        max_messages = self.MAX_MESSAGES
+
+        class EchoBot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "ECHO"
+
+            @listen("ECHO")
+            def echo(self) -> str:
+                reply = f"echo: {self.state.current_user_message or ''}"
+                self.append_assistant_message(reply)  # handler DOES append
+                if len(self.state.messages) > max_messages:  # ...and trims
+                    self.state.messages = self.state.messages[-max_messages:]
+                return reply
+
+        return EchoBot()
+
+    def test_no_duplicate_reply_when_handler_trims_history(self) -> None:
+        bot = self._make_bot()
+        for i in range(1, 5):
+            bot.handle_turn(f"message {i}")
+            contents = [message.content for message in bot.state.messages]
+            assert len(contents) == len(set(contents)), (
+                f"duplicate reply on turn {i}: {contents}"
+            )
+
+        # The capped window holds the last two full turns, in order.
+        assert [message.content for message in bot.state.messages] == [
+            "message 3",
+            "echo: message 3",
+            "message 4",
+            "echo: message 4",
+        ]
+
+    def test_fallback_still_appends_when_handler_does_not_reply(self) -> None:
+        class SilentBot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "WORK"
+
+            @listen("WORK")
+            def work(self) -> str:
+                return "computed reply"  # returns without appending
+
+        bot = SilentBot()
+        bot.handle_turn("hello")
+
+        assistant_messages = [
+            message.content
+            for message in bot.state.messages
+            if message.role == "assistant"
+        ]
+        assert assistant_messages == ["computed reply"]
+
+
 class TestFlowTracingWhenSuppressed:
     def test_flow_started_emitted_when_panel_events_suppressed(self) -> None:
         class QuietFlow(Flow[ChatState]):