improve convo routing cycle with one less route (#6102)

* improve one less route

* flows in flows, new agent executor causing early trace batch finalization

* addressing comments

* addressing comments pt2

* lint and typecheck fix

lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py

@@ -292,7 +292,7 @@ class TraceCollectionListener(BaseEventListener):
         @event_bus.on(CrewKickoffCompletedEvent)
         def on_crew_completed(source: Any, event: CrewKickoffCompletedEvent) -> None:
             self._handle_trace_event("crew_kickoff_completed", source, event)
-            if self.batch_manager.defer_session_finalization:
+            if self._should_defer_session_finalization():
                 return
             if self._nested_in_flow_execution():
                 return
@@ -306,7 +306,7 @@ class TraceCollectionListener(BaseEventListener):
         @event_bus.on(CrewKickoffFailedEvent)
         def on_crew_failed(source: Any, event: CrewKickoffFailedEvent) -> None:
             self._handle_trace_event("crew_kickoff_failed", source, event)
-            if self.batch_manager.defer_session_finalization:
+            if self._should_defer_session_finalization():
                 return
             if self._nested_in_flow_execution():
                 return
@@ -734,7 +734,7 @@ class TraceCollectionListener(BaseEventListener):
             if not self.batch_manager.is_batch_initialized():
                 return
             # Multi-turn flows defer batch finalization to finalize_session_traces().
-            if self.batch_manager.defer_session_finalization:
+            if self._should_defer_session_finalization():
                 return
             self.batch_manager.finalize_batch()
 
@@ -745,6 +745,15 @@ class TraceCollectionListener(BaseEventListener):
 
         return current_flow_id.get() is not None
 
+    def _should_defer_session_finalization(self) -> bool:
+        """True when the active trace belongs to a deferred flow session."""
+        from crewai.flow.flow_context import current_flow_defer_trace_finalization
+
+        return (
+            self.batch_manager.defer_session_finalization
+            or current_flow_defer_trace_finalization.get()
+        )
+
     def _flow_owns_trace_batch(self) -> bool:
         """True when an in-flight conversational flow already owns the trace batch."""
         if self.batch_manager.batch_owner_type == "flow":
@@ -786,7 +795,11 @@ class TraceCollectionListener(BaseEventListener):
         (``current_flow_id``) to keep LLM/tool events from falling back to an
         implicit crew batch.
         """
-        from crewai.flow.flow_context import current_flow_id, current_flow_name
+        from crewai.flow.flow_context import (
+            current_flow_defer_trace_finalization,
+            current_flow_id,
+            current_flow_name,
+        )
 
         flow_id = current_flow_id.get()
         if flow_id is None:
@@ -802,6 +815,8 @@ class TraceCollectionListener(BaseEventListener):
         }
         self.batch_manager.batch_owner_type = "flow"
         self.batch_manager.batch_owner_id = flow_id
+        if current_flow_defer_trace_finalization.get():
+            self.batch_manager.defer_session_finalization = True
         self._initialize_batch(user_context, execution_metadata)
         return True
 

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

@@ -46,7 +46,9 @@ from crewai.flow.conversation import (
     get_conversation_messages,
     receive_user_message as _receive_user_message,
 )
-from crewai.flow.dsl import listen, router, start
+from crewai.flow.dsl import listen, start
+from crewai.flow.dsl._utils import _set_flow_method_definition
+from crewai.flow.flow_definition import FlowMethodDefinition
 from crewai.utilities.types import LLMMessage
 
 
@@ -72,6 +74,15 @@ def _iter_condition_labels(condition: Any) -> set[str]:
     return set()
 
 
+def _conversation_start_router(func: Callable[..., Any]) -> Any:
+    wrapper = start()(func)
+    _set_flow_method_definition(
+        cast(Any, wrapper),
+        FlowMethodDefinition(start=True, router=True),
+    )
+    return wrapper
+
+
 class _ConversationalMixin:
     """Experimental conversational graph for ``Flow``.
 
@@ -85,10 +96,7 @@ class _ConversationalMixin:
     conversational: ClassVar[bool] = False
     conversational_config: ClassVar[ConversationConfig | None] = None
     builtin_routes: ClassVar[tuple[str, ...]] = ("converse", "end")
-    internal_routes: ClassVar[tuple[str, ...]] = (
-        "answer_from_history",
-        "conversation_start",
-    )
+    internal_routes: ClassVar[tuple[str, ...]] = ("answer_from_history",)
     builtin_route_descriptions: ClassVar[dict[str, str]] = {
         "converse": (
             "Ordinary chat, follow-ups, summaries, clarifications, and "
@@ -138,23 +146,24 @@ class _ConversationalMixin:
         def kickoff(self, *args: Any, **kwargs: Any) -> Any:
             pass
 
-    @start()
-    @_conversational_only
     def conversation_start(self) -> str | None:
-        """Internal Flow entrypoint that hands the user message to the router.
+        """Return the current user message for conversational route selection.
 
-        In conversational mode, ``Flow.kickoff_async`` runs all ``@start``
-        methods sequentially and this one is registered last, so any user
-        ``@start`` methods (e.g. permission loading) have already finished
-        before the returned value triggers ``route_conversation``.
+        This remains as a plain overridable helper for compatibility. It is not
+        registered as a Flow method; ``route_conversation`` is the synthetic
+        built-in start/router that begins a conversational turn.
         """
         state = cast(ConversationState, self.state)
         return state.current_user_message
 
-    @router(conversation_start)
+    @_conversation_start_router
     @_conversational_only
     def route_conversation(self) -> str:
         """Route the current turn to a listener label."""
+        if "conversation_start" not in {
+            str(method_name) for method_name in self._completed_methods
+        }:
+            self.conversation_start()
         state = cast(ConversationState, self.state)
         context = self.build_router_context()
         previous_intent = state.last_intent
@@ -651,16 +660,16 @@ class _ConversationalMixin:
         if not type(self)._is_conversational():
             return start_methods, False
 
-        conversation_start = "conversation_start"
-        if conversation_start not in {str(method) for method in start_methods}:
+        route_conversation = "route_conversation"
+        if route_conversation not in {str(method) for method in start_methods}:
             return start_methods, False
 
         ordered_starts = [
-            method for method in start_methods if str(method) != conversation_start
+            method for method in start_methods if str(method) != route_conversation
         ]
         ordered_starts.append(
             next(
-                method for method in start_methods if str(method) == conversation_start
+                method for method in start_methods if str(method) == route_conversation
             )
         )
         return ordered_starts, True
@@ -1047,12 +1056,15 @@ class _ConversationalMixin:
 
         trace_listener = TraceCollectionListener()
         batch_manager = trace_listener.batch_manager
-        if batch_manager.batch_owner_type == "flow":
-            if trace_listener.first_time_handler.is_first_time:
-                trace_listener.first_time_handler.mark_events_collected()
-                trace_listener.first_time_handler.handle_execution_completion()
-            else:
-                batch_manager.finalize_batch()
+        try:
+            if batch_manager.batch_owner_type == "flow":
+                if trace_listener.first_time_handler.is_first_time:
+                    trace_listener.first_time_handler.mark_events_collected()
+                    trace_listener.first_time_handler.handle_execution_completion()
+                else:
+                    batch_manager.finalize_batch()
+        finally:
+            batch_manager.defer_session_finalization = False
 
 
 __all__ = ["_ConversationalMixin"]

lib/crewai/src/crewai/flow/conversational_definition.py

@@ -39,9 +39,7 @@ class FlowConversationalDefinition(BaseModel):
     visible_agent_outputs: list[str] | Literal["all"] | None = None
     defer_trace_finalization: bool = True
     builtin_routes: list[str] = Field(default_factory=lambda: ["converse", "end"])
-    internal_routes: list[str] = Field(
-        default_factory=lambda: ["answer_from_history", "conversation_start"]
-    )
+    internal_routes: list[str] = Field(default_factory=lambda: ["answer_from_history"])
 
 
 __all__ = [

lib/crewai/src/crewai/flow/dsl/_utils.py

@@ -313,7 +313,7 @@ def _build_conversational_definition(
     internal_routes = getattr(
         flow_class,
         "internal_routes",
-        ("answer_from_history", "conversation_start"),
+        ("answer_from_history",),
     )
     if config is None:
         return FlowConversationalDefinition(

lib/crewai/src/crewai/flow/flow_context.py

@@ -15,6 +15,10 @@ current_flow_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
     "flow_id", default=None
 )
 
+current_flow_defer_trace_finalization: contextvars.ContextVar[bool] = (
+    contextvars.ContextVar("flow_defer_trace_finalization", default=False)
+)
+
 current_flow_method_name: contextvars.ContextVar[str] = contextvars.ContextVar(
     "flow_method_name", default="unknown"
 )

lib/crewai/src/crewai/flow/runtime.py

@@ -85,7 +85,12 @@ from crewai.events.types.flow_events import (
     MethodExecutionStartedEvent,
 )
 from crewai.flow.dsl._utils import build_flow_definition
-from crewai.flow.flow_context import current_flow_id, current_flow_request_id
+from crewai.flow.flow_context import (
+    current_flow_defer_trace_finalization,
+    current_flow_id,
+    current_flow_name,
+    current_flow_request_id,
+)
 from crewai.flow.flow_definition import (
     FlowDefinition,
     FlowDefinitionCondition,
@@ -1514,7 +1519,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             )
             self._event_futures.clear()
 
-        if not self.suppress_flow_events:
+        if (
+            not self.suppress_flow_events
+            and not self._should_defer_trace_finalization()
+        ):
             future = crewai_event_bus.emit(
                 self,
                 FlowFinishedEvent(
@@ -1531,7 +1539,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     logger.warning("FlowFinishedEvent handler failed", exc_info=True)
 
             trace_listener = TraceCollectionListener()
-            if trace_listener.batch_manager.batch_owner_type == "flow":
+            if (
+                trace_listener.batch_manager.batch_owner_type == "flow"
+                and current_flow_id.get() == self.flow_id
+                and not trace_listener.batch_manager.defer_session_finalization
+                and not current_flow_defer_trace_finalization.get()
+            ):
                 if trace_listener.first_time_handler.is_first_time:
                     trace_listener.first_time_handler.mark_events_collected()
                     trace_listener.first_time_handler.handle_execution_completion()
@@ -2020,9 +2033,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow_token = attach(ctx)
 
         flow_id_token = None
+        flow_name_token = None
+        flow_defer_trace_finalization_token = None
         request_id_token = None
         if current_flow_id.get() is None:
             flow_id_token = current_flow_id.set(self.flow_id)
+            flow_name_token = current_flow_name.set(
+                self.name or self.__class__.__name__
+            )
+            flow_defer_trace_finalization_token = (
+                current_flow_defer_trace_finalization.set(
+                    self._should_defer_trace_finalization()
+                )
+            )
         if current_flow_request_id.get() is None:
             request_id_token = current_flow_request_id.set(self.flow_id)
 
@@ -2117,6 +2140,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             should_emit_flow_started = not (
                 defer_trace_finalization and deferred_started_event_id
             )
+            if current_flow_id.get() == self.flow_id:
+                TraceCollectionListener().batch_manager.defer_session_finalization = (
+                    defer_trace_finalization
+                )
 
             if (
                 defer_trace_finalization
@@ -2290,7 +2317,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         )
 
                 trace_listener = TraceCollectionListener()
-                if trace_listener.batch_manager.batch_owner_type == "flow":
+                if (
+                    trace_listener.batch_manager.batch_owner_type == "flow"
+                    and current_flow_id.get() == self.flow_id
+                    and not trace_listener.batch_manager.defer_session_finalization
+                    and not current_flow_defer_trace_finalization.get()
+                ):
                     if trace_listener.first_time_handler.is_first_time:
                         trace_listener.first_time_handler.mark_events_collected()
                         trace_listener.first_time_handler.handle_execution_completion()
@@ -2304,6 +2336,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 self.memory.drain_writes()
             if request_id_token is not None:
                 current_flow_request_id.reset(request_id_token)
+            if flow_defer_trace_finalization_token is not None:
+                current_flow_defer_trace_finalization.reset(
+                    flow_defer_trace_finalization_token
+                )
+            if flow_name_token is not None:
+                current_flow_name.reset(flow_name_token)
             if flow_id_token is not None:
                 current_flow_id.reset(flow_id_token)
             detach(flow_token)

lib/crewai/tests/test_flow_conversation.py

@@ -26,7 +26,11 @@ from crewai.experimental import (
     RouterConfig,
 )
 from crewai.flow import Flow, ChatState, listen, start
-from crewai.flow.flow_context import current_flow_id, current_flow_name
+from crewai.flow.flow_context import (
+    current_flow_defer_trace_finalization,
+    current_flow_id,
+    current_flow_name,
+)
 from crewai.flow.conversation import (
     append_message,
     get_conversation_messages,
@@ -598,9 +602,9 @@ class TestConversationalFlow:
         """Conversational flows: user ``@start`` methods finish before router fires.
 
         Non-chat flows run ``@start`` methods in parallel via ``asyncio.gather``,
-        which would race with ``conversation_start`` and let the router fire
+        which would race with ``route_conversation`` and let the router fire
         before user setup finished. In conversational mode the framework runs
-        them sequentially, with ``conversation_start`` last.
+        them sequentially, with ``route_conversation`` last.
         """
         order: list[str] = []
 
@@ -643,15 +647,10 @@ class TestConversationalFlow:
         assert "attach_bus" in order  # still fires every turn
         assert "route_turn" in order
 
-    def test_subclass_can_override_conversation_start_without_redecorating(
+    def test_subclass_can_override_conversation_start_helper(
         self,
     ) -> None:
-        """Overriding an inherited ``@start`` method must not unregister it.
-
-        Before the metaclass fix, subclasses had to re-apply ``@start()`` on
-        every override or the parent's ``conversation_start`` would silently
-        drop out of the start registry — leaving the flow with nothing to fire.
-        """
+        """The compatibility helper remains overridable without adding a Flow node."""
 
         bootstrap_calls: list[str] = []
 
@@ -672,6 +671,38 @@ class TestConversationalFlow:
         flow = BootstrapFlow()
         flow.handle_turn("hi")
 
+        assert bootstrap_calls == ["ran"]
+        assert "conversation_start" not in BootstrapFlow.flow_definition().methods
+        route_definition = BootstrapFlow.flow_definition().methods["route_conversation"]
+        assert route_definition.start is True
+        assert route_definition.router is True
+        assert flow.state.messages[-1].content == "worked"
+
+    def test_legacy_decorated_conversation_start_runs_once_per_turn(
+        self,
+    ) -> None:
+        """Legacy ``@start`` overrides are not invoked again by the router."""
+
+        bootstrap_calls: list[str] = []
+
+        @ConversationConfig()
+        class BootstrapFlow(ConversationalFlow):
+            @start()
+            def conversation_start(self) -> str | None:
+                bootstrap_calls.append("ran")
+                return super().conversation_start()
+
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "work"
+
+            @listen("work")
+            def do_work(self) -> str:
+                self.append_assistant_message("worked")
+                return "worked"
+
+        flow = BootstrapFlow()
+        flow.handle_turn("hi")
+
         assert bootstrap_calls == ["ran"]
         assert flow.state.messages[-1].content == "worked"
 
@@ -1170,6 +1201,40 @@ class TestConversationalFlow:
                     "finalize_session_traces must finalize the trace batch once"
                 )
 
+    def test_deferred_resume_skips_per_resume_flow_finished_event(self) -> None:
+        """Deferred sessions do not emit terminal events while resuming."""
+        from crewai.events.types.flow_events import FlowFinishedEvent
+        from crewai.flow.async_feedback.types import PendingFeedbackContext
+
+        class DeferredResumeFlow(Flow[ChatState]):
+            defer_trace_finalization = True
+
+            @start()
+            def begin(self) -> str:
+                return "started"
+
+        flow = DeferredResumeFlow()
+        flow._pending_feedback_context = PendingFeedbackContext(
+            flow_id=flow.flow_id,
+            flow_class="DeferredResumeFlow",
+            method_name="begin",
+            method_output="started",
+            message="Review",
+        )
+
+        finished_events: list[FlowFinishedEvent] = []
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(FlowFinishedEvent)
+            def capture(_: Any, event: FlowFinishedEvent) -> None:
+                finished_events.append(event)
+
+            flow.resume("approved")
+            crewai_event_bus.flush()
+
+        assert finished_events == []
+
     def test_finalize_session_traces_restores_event_scope(self, capsys) -> None:
         """No ``empty scope stack`` warning when deferred ``flow_finished`` fires.
 
@@ -1471,6 +1536,44 @@ class TestDeferredFlowLifecycleEvents:
                     listener.batch_manager.finalize_batch()
             mock_finalize.assert_not_called()
 
+    def test_deferred_flow_kickoff_marks_trace_manager_session_deferred(
+        self,
+    ) -> None:
+        class DeferredTraceFlow(Flow[ChatState]):
+            @start()
+            def begin(self) -> str:
+                return "done"
+
+        listener = TraceCollectionListener()
+        listener.batch_manager.defer_session_finalization = False
+
+        flow = DeferredTraceFlow()
+        flow.defer_trace_finalization = True
+
+        with patch.object(listener.batch_manager, "finalize_batch"):
+            flow.kickoff()
+
+        assert listener.batch_manager.defer_session_finalization is True
+
+        flow.finalize_session_traces()
+
+        assert listener.batch_manager.defer_session_finalization is False
+
+    def test_non_deferred_flow_kickoff_clears_stale_trace_manager_flag(
+        self,
+    ) -> None:
+        class PlainTraceFlow(Flow[ChatState]):
+            @start()
+            def begin(self) -> str:
+                return "done"
+
+        listener = TraceCollectionListener()
+        listener.batch_manager.defer_session_finalization = True
+
+        PlainTraceFlow().kickoff()
+
+        assert listener.batch_manager.defer_session_finalization is False
+
 
 class TestNestedCrewTracing:
     def test_is_inside_active_flow_context_when_kickoff_running(self) -> None:
@@ -1524,3 +1627,130 @@ class TestNestedCrewTracing:
             elif listener.batch_manager.batch_owner_type == "crew":
                 listener.batch_manager.finalize_batch()
             mock_finalize.assert_not_called()
+
+    def test_lazy_flow_batch_from_context_preserves_deferred_parent(self) -> None:
+        from crewai.events.listeners.tracing.trace_listener import (
+            TraceCollectionListener,
+        )
+
+        listener = TraceCollectionListener()
+        listener.batch_manager.current_batch = None
+        listener.batch_manager.batch_owner_type = None
+        listener.batch_manager.batch_owner_id = None
+        listener.batch_manager.defer_session_finalization = False
+        listener.batch_manager.event_buffer.clear()
+
+        flow_id_token = current_flow_id.set("parent-flow-id")
+        flow_name_token = current_flow_name.set("ParentChatFlow")
+        defer_token = current_flow_defer_trace_finalization.set(True)
+        try:
+            initialized = listener._try_initialize_flow_batch_from_context(
+                type("Event", (), {"timestamp": None})()
+            )
+
+            assert initialized is True
+            assert listener.batch_manager.batch_owner_type == "flow"
+            assert listener.batch_manager.batch_owner_id == "parent-flow-id"
+            assert listener.batch_manager.defer_session_finalization is True
+            assert listener.batch_manager.current_batch is not None
+            assert (
+                listener.batch_manager.current_batch.execution_metadata[
+                    "execution_type"
+                ]
+                == "flow"
+            )
+            assert (
+                listener.batch_manager.current_batch.execution_metadata["flow_name"]
+                == "ParentChatFlow"
+            )
+        finally:
+            current_flow_defer_trace_finalization.reset(defer_token)
+            current_flow_name.reset(flow_name_token)
+            current_flow_id.reset(flow_id_token)
+            listener.batch_manager.current_batch = None
+            listener.batch_manager.batch_owner_type = None
+            listener.batch_manager.batch_owner_id = None
+            listener.batch_manager.trace_batch_id = None
+            listener.batch_manager.defer_session_finalization = False
+            listener.batch_manager.event_buffer.clear()
+
+    def test_nested_agent_executor_flow_does_not_finalize_parent_batch(
+        self,
+    ) -> None:
+        from crewai import Agent, Crew, Task
+        from crewai.llms.base_llm import BaseLLM
+
+        class StaticLLM(BaseLLM):
+            def __init__(self) -> None:
+                super().__init__(model="debug-static-llm", provider="debug")
+
+            def call(
+                self,
+                messages: Any,
+                tools: Any = None,
+                callbacks: Any = None,
+                available_functions: Any = None,
+                from_task: Any = None,
+                from_agent: Any = None,
+                response_model: Any = None,
+            ) -> str:
+                return (
+                    "Thought: I can answer directly.\n"
+                    "Final Answer: nested crew result"
+                )
+
+        class NestedCrewFlow(Flow[ChatState]):
+            defer_trace_finalization = True
+            tracing = True
+
+            @start()
+            def begin(self) -> str:
+                return "run_nested_crew"
+
+            @listen(begin)
+            def run_nested_crew(self, _: str) -> str:
+                agent = Agent(
+                    role="Debug Agent",
+                    goal="Return a short deterministic result",
+                    backstory="Used only for trace finalization debugging.",
+                    llm=StaticLLM(),
+                    verbose=False,
+                )
+                task = Task(
+                    description="Return the deterministic nested crew result.",
+                    expected_output="nested crew result",
+                    agent=agent,
+                )
+                return Crew(agents=[agent], tasks=[task], verbose=False).kickoff().raw
+
+        listener = TraceCollectionListener()
+        listener.batch_manager.current_batch = None
+        listener.batch_manager.batch_owner_type = None
+        listener.batch_manager.batch_owner_id = None
+        listener.batch_manager.trace_batch_id = None
+        listener.batch_manager.defer_session_finalization = False
+        listener.batch_manager.event_buffer.clear()
+        listener.first_time_handler.is_first_time = False
+
+        def initialize_backend_batch(*_: Any, **__: Any) -> None:
+            listener.batch_manager.trace_batch_id = "debug-trace-batch"
+
+        flow = NestedCrewFlow()
+
+        with (
+            patch.object(
+                listener.batch_manager,
+                "_initialize_backend_batch",
+                side_effect=initialize_backend_batch,
+            ),
+            patch.object(listener.batch_manager, "finalize_batch") as mock_finalize,
+        ):
+            flow.kickoff()
+            crewai_event_bus.flush()
+            flow.kickoff()
+            crewai_event_bus.flush()
+
+            assert mock_finalize.call_count == 0, (
+                "nested AgentExecutor flows inside a deferred parent Flow must "
+                "not finalize the parent trace batch"
+            )

lib/crewai/tests/test_flow_definition.py

@@ -223,10 +223,11 @@ def test_flow_definition_includes_conversational_builtins_when_enabled():
     assert definition.conversational.enabled is True
     assert definition.conversational.defer_trace_finalization is True
     assert definition.conversational.builtin_routes == ["converse", "end"]
-    assert "conversation_start" in methods
+    assert "conversation_start" not in methods
     assert "route_conversation" in methods
     assert "converse_turn" in methods
-    assert methods["conversation_start"].start is True
+    assert methods["route_conversation"].start is True
+    assert methods["route_conversation"].router is True
 
 
 def test_flow_definition_serializes_conversational_config():
@@ -260,7 +261,7 @@ def test_flow_definition_serializes_conversational_config():
     assert conversational.router.fallback_intent == "end"
 
 
-def test_flow_definition_preserves_undecorated_conversational_override():
+def test_flow_definition_uses_collapsed_conversational_router_start():
     class ChatFlow(Flow):
         conversational = True
 
@@ -269,8 +270,10 @@ def test_flow_definition_preserves_undecorated_conversational_override():
 
     methods = ChatFlow.flow_definition().methods
 
-    assert methods["conversation_start"].start is True
+    assert "conversation_start" not in methods
     assert "route_conversation" in methods
+    assert methods["route_conversation"].start is True
+    assert methods["route_conversation"].router is True
 
 
 def test_flow_definition_serializes_human_feedback_metadata():