Merge branch 'main' into fix/conversational-persist-fallback-append

Call _persist_method_completion via RuntimeFlow cast instead of a mixin
TYPE_CHECKING stub so persistence stays owned by the Flow engine.

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

@@ -330,6 +330,7 @@ class _ConversationalMixin:
                 and self._is_public_turn_result(result)
             ):
                 self.append_assistant_message(self._stringify_result(result))
+                self._persist_state_after_fallback_append()
         except Exception as exc:
             failed_event = ConversationTurnFailedEvent(
                 type="conversation_turn_failed",
@@ -419,6 +420,7 @@ class _ConversationalMixin:
                     and self._is_public_turn_result(result)
                 ):
                     self.append_assistant_message(self._stringify_result(result))
+                    self._persist_state_after_fallback_append()
             except HumanFeedbackPending as exc:
                 return exc
             except Exception as exc:
@@ -879,6 +881,26 @@ class _ConversationalMixin:
         self._is_execution_resuming = False
         object.__setattr__(self, "_turn_classified_intent", None)
 
+    def _persist_state_after_fallback_append(self) -> None:
+        """Re-persist after fallback append so restore sees the assistant reply.
+
+        Per-method snapshots inside ``kickoff`` run before this fallback, so
+        return-only ``@listen`` handlers need one more save for restore to see
+        the assistant message on the next turn.
+        """
+        if not self._method_outputs:
+            return
+        last_entry = self._method_outputs[-1]
+        if not isinstance(last_entry, dict) or not last_entry.get("method"):
+            return
+
+        from crewai.flow.runtime import Flow as RuntimeFlow
+        from crewai.flow.types import FlowMethodName
+
+        cast(RuntimeFlow[Any], self)._persist_method_completion(
+            FlowMethodName(str(last_entry["method"]))
+        )
+
     def _apply_pending_conversational_turn(self) -> None:
         """Drain the stashed user message + classify if intents configured.
 

lib/crewai/tests/test_flow_conversation.py

@@ -1619,6 +1619,54 @@ class TestHandleTurnReplyFallback:
         ]
         assert assistant_messages == ["computed reply"]
 
+    def test_return_only_handler_persists_assistant_for_fresh_instance(
+        self, tmp_path: Any
+    ) -> None:
+        """Return-only handlers must survive @persist restore on a new Flow instance."""
+        from crewai.flow.persistence import SQLiteFlowPersistence, persist
+
+        persistence = SQLiteFlowPersistence(str(tmp_path / "fallback_persist.db"))
+        session_id = str(uuid4())
+
+        @persist(persistence)
+        class ReturnOnlyBot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "WORK"
+
+            @listen("WORK")
+            def work(self) -> str:
+                return "computed reply"
+
+        first_turn = ReturnOnlyBot(persistence=persistence)
+        first_turn.handle_turn("hello", session_id=session_id)
+
+        second_turn = ReturnOnlyBot(persistence=persistence)
+        second_turn.handle_turn("follow up", session_id=session_id)
+
+        roles = [message.role for message in second_turn.state.messages]
+        assert roles == ["user", "assistant", "user", "assistant"]
+        assert [message.content for message in second_turn.state.messages] == [
+            "hello",
+            "computed reply",
+            "follow up",
+            "computed reply",
+        ]
+
+        restored = persistence.load_state(session_id)
+        assert restored is not None
+        assert [message["content"] for message in restored["messages"]] == [
+            "hello",
+            "computed reply",
+            "follow up",
+            "computed reply",
+        ]
+        assert [message["role"] for message in restored["messages"]] == [
+            "user",
+            "assistant",
+            "user",
+            "assistant",
+        ]
+
 
 class TestFalsyRouteTurnFallback:
     """A falsy ``route_turn()`` must never replay a previous turn's intent.