fix: emit FlowFailedEvent for deferred conversational sessions

`_emit_flow_failed` returned early whenever `defer_trace_finalization`
was on, mirroring the success path where the terminal event is owed to a
later `finalize_session_traces()`. A failed turn never reaches that call,
so the session's `flow_started` was left open and tracing closed the root
span as orphaned. The event is now emitted for deferred sessions too,
clearing the stashed started-event id so a later finalization does not
emit a second terminal event, while batch finalization stays with
whoever owns the session.

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

@@ -2541,6 +2541,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         flow span does, then emit and finalize the trace batch. Never raises,
         so the original exception propagates unchanged.
 
+        Deferred sessions still get the event — unlike the success path, whose
+        terminal event is owed to a later ``finalize_session_traces()``, a
+        failure ends the session there and then — but the batch finalization
+        stays with whoever owns the session.
+
         Args:
             error: The exception that ended the execution.
             respect_suppression: Skip the emission for suppressed flows. Only
@@ -2548,8 +2553,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 ``suppress_flow_events``; ``kickoff_async`` emits them either
                 way and lets listeners filter.
         """
-        if self._should_defer_trace_finalization():
-            return
         if respect_suppression and self.suppress_flow_events:
             return
 
@@ -2577,6 +2580,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 except Exception:
                     logger.warning("FlowFailedEvent handler failed", exc_info=True)
 
+            # The session scope is closed now, so a later
+            # ``finalize_session_traces()`` must not emit a second terminal
+            # event against it.
+            object.__setattr__(self, "_deferred_flow_started_event_id", None)
+
             trace_listener = TraceCollectionListener()
             if (
                 trace_listener.batch_manager.batch_owner_type == "flow"

lib/crewai/tests/utilities/test_events.py

@@ -47,6 +47,10 @@ from crewai.events.types.tool_usage_events import (
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
 )
+from crewai.experimental.conversational import (
+    ConversationConfig,
+    ConversationState,
+)
 from crewai.flow.async_feedback.types import PendingFeedbackContext
 from crewai.flow.flow import Flow, listen, start
 from crewai.flow.human_feedback import human_feedback
@@ -628,6 +632,52 @@ def test_suppressed_flow_failure_matches_finished_event_emission():
     assert len(failed) == 1
 
 
+def test_deferred_session_failure_emits_a_single_terminal_event():
+    started: list[FlowStartedEvent] = []
+    failed: list[FlowFailedEvent] = []
+    finished: list[FlowFinishedEvent] = []
+
+    @ConversationConfig(defer_trace_finalization=True)
+    class ConvoFlow(Flow[ConversationState]):
+        conversational = True
+
+        @start()
+        def load(self) -> None:
+            return None
+
+        def route_turn(self, context):
+            return "BOOM"
+
+        @listen("BOOM")
+        def boom(self) -> str:
+            raise ConnectionError("backend unreachable")
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowStartedEvent)
+        def handle_flow_started(source, event):
+            started.append(event)
+
+        @crewai_event_bus.on(FlowFailedEvent)
+        def handle_flow_failed(source, event):
+            failed.append(event)
+
+        @crewai_event_bus.on(FlowFinishedEvent)
+        def handle_flow_finished(source, event):
+            finished.append(event)
+
+        flow = ConvoFlow()
+        with pytest.raises(ConnectionError, match="backend unreachable"):
+            flow.handle_turn("go", session_id="session-1")
+        flow.finalize_session_traces()
+        wait_for_event_handlers()
+
+    assert len(started) == 1
+    assert len(failed) == 1
+    assert failed[0].started_event_id == started[0].event_id
+    assert finished == []
+
+
 def test_abort_before_flow_started_emits_no_failed_event():
     started: list[FlowStartedEvent] = []
     failed: list[FlowFailedEvent] = []