From 8d2d6b18feb8fd9fd3dc95dd664ab99a7861dd3a Mon Sep 17 00:00:00 2001 From: Joao Moura Date: Thu, 9 Jul 2026 23:59:24 -0700 Subject: [PATCH] fix(flow): don't double-append the turn reply when a handler trims history MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../experimental/conversational_mixin.py | 18 ++--- lib/crewai/tests/test_flow_conversation.py | 65 +++++++++++++++++++ 2 files changed, 75 insertions(+), 8 deletions(-) diff --git a/lib/crewai/src/crewai/experimental/conversational_mixin.py b/lib/crewai/src/crewai/experimental/conversational_mixin.py index 490565e9d..c35bb7e0d 100644 --- a/lib/crewai/src/crewai/experimental/conversational_mixin.py +++ b/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 diff --git a/lib/crewai/tests/test_flow_conversation.py b/lib/crewai/tests/test_flow_conversation.py index c6d3f91c8..e00f96f94 100644 --- a/lib/crewai/tests/test_flow_conversation.py +++ b/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]):