diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 4bb78f9fd..1e024ba21 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/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" diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py index d9c9d5529..fe892580a 100644 --- a/lib/crewai/tests/utilities/test_events.py +++ b/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] = []