From d52d0a162814624095e58116c7bb6f58924e52b2 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Wed, 29 Jul 2026 15:44:55 -0300 Subject: [PATCH] feat: emit FlowFailedEvent when a flow execution fails (#6718) * feat: emit FlowFailedEvent when a flow execution fails A failed flow never emitted a terminal lifecycle event, so the `flow_started` scope stayed open and consumers such as tracing closed the root span with a generic orphaned message instead of the real error. `kickoff_async` and the resume path now emit `FlowFailedEvent`, paired with `flow_started` and carrying the exception, after draining pending handlers and background memory writes. The resume path also emits the `MethodExecutionStartedEvent` it was missing for the method being resumed, so its finished or failed event pairs with its own scope instead of popping the flow's. * fix: skip FlowFailedEvent when the run never opened a scope The `kickoff_async` try block starts before `FlowStartedEvent` is emitted, so an abort in the execution-start hooks, in input handling or in state restore emitted a `flow_failed` with no opener, which pops an unrelated scope and warns about an empty scope stack. The failure event is now gated on the flow scope actually being open, either from this kickoff's `flow_started` or from a restored deferred session scope. --- docs/edge/ar/concepts/event-listener.mdx | 1 + docs/edge/en/concepts/event-listener.mdx | 1 + docs/edge/ko/concepts/event-listener.mdx | 1 + docs/edge/pt-BR/concepts/event-listener.mdx | 1 + lib/crewai/src/crewai/events/__init__.py | 3 + lib/crewai/src/crewai/events/event_context.py | 2 + .../src/crewai/events/event_listener.py | 10 + lib/crewai/src/crewai/events/event_types.py | 2 + .../listeners/tracing/trace_listener.py | 5 + .../src/crewai/events/types/flow_events.py | 18 ++ .../src/crewai/flow/runtime/__init__.py | 155 ++++++++-- .../src/crewai/state/checkpoint_config.py | 1 + lib/crewai/tests/utilities/test_events.py | 272 ++++++++++++++++++ 13 files changed, 447 insertions(+), 25 deletions(-) diff --git a/docs/edge/ar/concepts/event-listener.mdx b/docs/edge/ar/concepts/event-listener.mdx index 41be56cb8..7fa6d671d 100644 --- a/docs/edge/ar/concepts/event-listener.mdx +++ b/docs/edge/ar/concepts/event-listener.mdx @@ -157,6 +157,7 @@ class MyCustomCrew: - **FlowCreatedEvent**: يُرسل عند إنشاء تدفق - **FlowStartedEvent**: يُرسل عند بدء تنفيذ تدفق - **FlowFinishedEvent**: يُرسل عند اكتمال تنفيذ تدفق +- **FlowFailedEvent**: يُرسل عند فشل تنفيذ تدفق. يحتوي على اسم التدفق والاستثناء الذي أنهى التنفيذ. - **FlowPausedEvent**: يُرسل عند إيقاف تدفق مؤقتًا بانتظار ملاحظات بشرية ### أحداث LLM diff --git a/docs/edge/en/concepts/event-listener.mdx b/docs/edge/en/concepts/event-listener.mdx index b3eb33e83..403a2ff59 100644 --- a/docs/edge/en/concepts/event-listener.mdx +++ b/docs/edge/en/concepts/event-listener.mdx @@ -256,6 +256,7 @@ CrewAI provides a wide range of events that you can listen for: - **FlowCreatedEvent**: Emitted when a Flow is created - **FlowStartedEvent**: Emitted when a Flow starts execution - **FlowFinishedEvent**: Emitted when a Flow completes execution +- **FlowFailedEvent**: Emitted when a Flow execution fails. Contains the flow name and the exception that ended the execution. - **FlowPausedEvent**: Emitted when a Flow is paused waiting for human feedback. Contains the flow name, flow ID, method name, current state, message shown when requesting feedback, and optional list of possible outcomes for routing. - **FlowPlotEvent**: Emitted when a Flow is plotted - **MethodExecutionStartedEvent**: Emitted when a Flow method starts execution diff --git a/docs/edge/ko/concepts/event-listener.mdx b/docs/edge/ko/concepts/event-listener.mdx index e2858bf55..aec99523c 100644 --- a/docs/edge/ko/concepts/event-listener.mdx +++ b/docs/edge/ko/concepts/event-listener.mdx @@ -255,6 +255,7 @@ CrewAI는 여러분이 청취할 수 있는 다양한 이벤트를 제공합니 - **FlowCreatedEvent**: Flow가 생성될 때 발생 - **FlowStartedEvent**: Flow가 실행을 시작할 때 발생 - **FlowFinishedEvent**: Flow가 실행을 완료할 때 발생 +- **FlowFailedEvent**: Flow 실행이 실패할 때 발생합니다. Flow 이름과 실행을 종료시킨 예외를 포함합니다. - **FlowPausedEvent**: 사람의 피드백을 기다리며 Flow가 일시 중지될 때 발생합니다. Flow 이름, Flow ID, 메서드 이름, 현재 상태, 피드백 요청 시 표시되는 메시지, 라우팅을 위한 선택적 결과 목록을 포함합니다. - **FlowPlotEvent**: Flow가 플롯될 때 발생 - **MethodExecutionStartedEvent**: Flow 메서드가 실행을 시작할 때 발생 diff --git a/docs/edge/pt-BR/concepts/event-listener.mdx b/docs/edge/pt-BR/concepts/event-listener.mdx index 85cb201a8..412440ab5 100644 --- a/docs/edge/pt-BR/concepts/event-listener.mdx +++ b/docs/edge/pt-BR/concepts/event-listener.mdx @@ -256,6 +256,7 @@ O CrewAI fornece uma ampla variedade de eventos para escuta: - **FlowCreatedEvent**: Emitido ao criar um Flow - **FlowStartedEvent**: Emitido ao iniciar a execução de um Flow - **FlowFinishedEvent**: Emitido ao concluir a execução de um Flow +- **FlowFailedEvent**: Emitido quando a execução de um Flow falha. Contém o nome do flow e a exceção que encerrou a execução. - **FlowPausedEvent**: Emitido quando um Flow é pausado aguardando feedback humano. Contém o nome do flow, ID do flow, nome do método, estado atual, mensagem exibida ao solicitar feedback e lista opcional de resultados possíveis para roteamento. - **FlowPlotEvent**: Emitido ao plotar um Flow - **MethodExecutionStartedEvent**: Emitido ao iniciar a execução de um método do Flow diff --git a/lib/crewai/src/crewai/events/__init__.py b/lib/crewai/src/crewai/events/__init__.py index 8a31e5397..3a9ab7e2b 100644 --- a/lib/crewai/src/crewai/events/__init__.py +++ b/lib/crewai/src/crewai/events/__init__.py @@ -68,6 +68,7 @@ if TYPE_CHECKING: ConversationTurnStartedEvent, FlowCreatedEvent, FlowEvent, + FlowFailedEvent, FlowFinishedEvent, FlowPlotEvent, FlowStartedEvent, @@ -194,6 +195,7 @@ _LAZY_EVENT_MAPPING: dict[str, str] = { "ConversationTurnStartedEvent": "crewai.events.types.flow_events", "FlowCreatedEvent": "crewai.events.types.flow_events", "FlowEvent": "crewai.events.types.flow_events", + "FlowFailedEvent": "crewai.events.types.flow_events", "FlowFinishedEvent": "crewai.events.types.flow_events", "FlowPlotEvent": "crewai.events.types.flow_events", "FlowStartedEvent": "crewai.events.types.flow_events", @@ -329,6 +331,7 @@ __all__ = [ "Depends", "FlowCreatedEvent", "FlowEvent", + "FlowFailedEvent", "FlowFinishedEvent", "FlowPlotEvent", "FlowStartedEvent", diff --git a/lib/crewai/src/crewai/events/event_context.py b/lib/crewai/src/crewai/events/event_context.py index 74e0d86dc..63fe48d90 100644 --- a/lib/crewai/src/crewai/events/event_context.py +++ b/lib/crewai/src/crewai/events/event_context.py @@ -269,6 +269,7 @@ SCOPE_STARTING_EVENTS: frozenset[str] = frozenset( SCOPE_ENDING_EVENTS: frozenset[str] = frozenset( { "flow_finished", + "flow_failed", "flow_paused", "method_execution_finished", "method_execution_failed", @@ -320,6 +321,7 @@ SCOPE_ENDING_EVENTS: frozenset[str] = frozenset( VALID_EVENT_PAIRS: dict[str, str] = { "flow_finished": "flow_started", + "flow_failed": "flow_started", "flow_paused": "flow_started", "method_execution_finished": "method_execution_started", "method_execution_failed": "method_execution_started", diff --git a/lib/crewai/src/crewai/events/event_listener.py b/lib/crewai/src/crewai/events/event_listener.py index a1a771f44..537d5edc4 100644 --- a/lib/crewai/src/crewai/events/event_listener.py +++ b/lib/crewai/src/crewai/events/event_listener.py @@ -43,6 +43,7 @@ from crewai.events.types.env_events import ( from crewai.events.types.flow_events import ( ConversationTurnCompletedEvent, FlowCreatedEvent, + FlowFailedEvent, FlowFinishedEvent, FlowPausedEvent, FlowStartedEvent, @@ -318,6 +319,15 @@ class EventListener(BaseEventListener): source.flow_id, ) + @crewai_event_bus.on(FlowFailedEvent) + def on_flow_failed(source: Any, event: FlowFailedEvent) -> None: + if not getattr(source, "suppress_flow_events", False): + self.formatter.handle_flow_status( + event.flow_name, + source.flow_id, + "failed", + ) + @crewai_event_bus.on(ConversationTurnCompletedEvent) def on_conversation_turn_completed( _: Any, event: ConversationTurnCompletedEvent diff --git a/lib/crewai/src/crewai/events/event_types.py b/lib/crewai/src/crewai/events/event_types.py index 8c589849a..d0b9d8bd8 100644 --- a/lib/crewai/src/crewai/events/event_types.py +++ b/lib/crewai/src/crewai/events/event_types.py @@ -58,6 +58,7 @@ from crewai.events.types.flow_events import ( ConversationTurnCompletedEvent, ConversationTurnFailedEvent, ConversationTurnStartedEvent, + FlowFailedEvent, FlowFinishedEvent, FlowStartedEvent, MethodExecutionFailedEvent, @@ -170,6 +171,7 @@ EventTypes = ( | ConversationTurnStartedEvent | FlowStartedEvent | FlowFinishedEvent + | FlowFailedEvent | MethodExecutionStartedEvent | MethodExecutionFinishedEvent | MethodExecutionFailedEvent diff --git a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py index 789e70e69..0db7bcf03 100644 --- a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py +++ b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py @@ -66,6 +66,7 @@ from crewai.events.types.flow_events import ( ConversationMessageAddedEvent, ConversationRouteSelectedEvent, FlowCreatedEvent, + FlowFailedEvent, FlowFinishedEvent, FlowPlotEvent, FlowStartedEvent, @@ -274,6 +275,10 @@ class TraceCollectionListener(BaseEventListener): def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None: self._handle_trace_event("flow_finished", source, event) + @event_bus.on(FlowFailedEvent) + def on_flow_failed(source: Any, event: FlowFailedEvent) -> None: + self._handle_trace_event("flow_failed", source, event) + @event_bus.on(FlowPlotEvent) def on_flow_plot(source: Any, event: FlowPlotEvent) -> None: self._handle_action_event("flow_plot", source, event) diff --git a/lib/crewai/src/crewai/events/types/flow_events.py b/lib/crewai/src/crewai/events/types/flow_events.py index 8e33b384c..1e7216db1 100644 --- a/lib/crewai/src/crewai/events/types/flow_events.py +++ b/lib/crewai/src/crewai/events/types/flow_events.py @@ -94,6 +94,24 @@ class FlowFinishedEvent(FlowEvent): state: dict[str, Any] | BaseModel +class FlowFailedEvent(FlowEvent): + """Event emitted when a flow execution fails. + + Attributes: + flow_name: Name of the flow that failed. + error: The exception that ended the execution. + """ + + error: Exception + type: Literal["flow_failed"] = "flow_failed" + + model_config = ConfigDict(arbitrary_types_allowed=True) + + @field_serializer("error") + def _serialize_error(self, error: Exception) -> str: + return str(error) + + class FlowPausedEvent(FlowEvent): """Event emitted when a flow is paused waiting for human feedback. diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 1781e3e2e..4bb78f9fd 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -67,6 +67,7 @@ from crewai.events.listeners.tracing.utils import ( ) from crewai.events.types.flow_events import ( FlowCreatedEvent, + FlowFailedEvent, FlowFinishedEvent, FlowPausedEvent, FlowPlotEvent, @@ -1341,6 +1342,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): except Exception as e: if not hook_state["end_dispatched"]: self._dispatch_execution_end_failure(e) + await self._emit_flow_failed(e, respect_suppression=True) raise finally: # Match kickoff_async: drain pending handlers so the resumed @@ -1382,35 +1384,68 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ) emit = context.emit - # The serialized context carries the full LLM config (a dict, or a - # legacy model string) — the single source for cross- and same-process - # resume. - result = await self._finalize_human_feedback( - method_name=context.method_name, - method_output=context.method_output, - raw_feedback=feedback, - emit=emit, - default_outcome=context.default_outcome, - llm=context.llm, - metadata=context.metadata, - ) - collapsed_outcome = result.outcome - resumed_method_output = ( - result.output - if emit and isinstance(result, HumanFeedbackResult) - else result - ) + if not self.suppress_flow_events: + # Opens the scope the finished event below closes; without it that + # event pops the enclosing ``flow_started`` instead. + future = crewai_event_bus.emit( + self, + MethodExecutionStartedEvent( + type="method_execution_started", + flow_name=self._definition.name, + method_name=context.method_name, + state=self._copy_and_serialize_state(), + ), + ) + if future and isinstance(future, Future): + try: + await asyncio.wrap_future(future) + except Exception: + logger.warning( + "MethodExecutionStartedEvent handler failed", exc_info=True + ) - self._completed_methods.add(FlowMethodName(context.method_name)) + try: + # The serialized context carries the full LLM config (a dict, or a + # legacy model string) — the single source for cross- and + # same-process resume. + result = await self._finalize_human_feedback( + method_name=context.method_name, + method_output=context.method_output, + raw_feedback=feedback, + emit=emit, + default_outcome=context.default_outcome, + llm=context.llm, + metadata=context.metadata, + ) + collapsed_outcome = result.outcome + resumed_method_output = ( + result.output + if emit and isinstance(result, HumanFeedbackResult) + else result + ) - await asyncio.to_thread( - self._persist_method_completion, FlowMethodName(context.method_name) - ) + self._completed_methods.add(FlowMethodName(context.method_name)) - self._pending_feedback_context = None + await asyncio.to_thread( + self._persist_method_completion, FlowMethodName(context.method_name) + ) - if self.persistence is not None: - self.persistence.clear_pending_feedback(context.flow_id) + self._pending_feedback_context = None + + if self.persistence is not None: + self.persistence.clear_pending_feedback(context.flow_id) + except Exception as e: + if not self.suppress_flow_events: + crewai_event_bus.emit( + self, + MethodExecutionFailedEvent( + type="method_execution_failed", + flow_name=self._definition.name, + method_name=context.method_name, + error=e, + ), + ) + raise if not self.suppress_flow_events: crewai_event_bus.emit( @@ -2096,6 +2131,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): # EXECUTION_START/EXECUTION_END dispatch independently. execution_start_dispatched = False execution_end_dispatched = False + # Guards the failure event: everything between here and the + # ``flow_started`` emission below (hooks, input handling, state + # restore) can raise, and a ``flow_failed`` with no opener would pop + # an unrelated scope. + flow_scope_open = False try: from crewai.hooks.contexts import ( @@ -2235,6 +2275,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): and get_current_parent_id() is None ): restore_event_scope(((deferred_started_event_id, "flow_started"),)) + flow_scope_open = True elif get_current_parent_id() is None: reset_emission_counter() reset_last_event_id() @@ -2249,6 +2290,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): inputs=inputs, ) future = crewai_event_bus.emit(self, started_event) + flow_scope_open = True if future: try: await asyncio.wrap_future(future) @@ -2440,6 +2482,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): # not (exactly-once per invocation). if execution_start_dispatched and not execution_end_dispatched: self._dispatch_execution_end_failure(e) + if flow_scope_open: + await self._emit_flow_failed(e) raise finally: # Safety net for the exception path; the success path already @@ -2487,6 +2531,67 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): except Exception: # noqa: S110 - aborting an already-failed execution is meaningless pass + async def _emit_flow_failed( + self, error: Exception, *, respect_suppression: bool = False + ) -> None: + """Emit ``FlowFailedEvent`` and close out the trace batch for a failed run. + + Mirrors the terminal block of the success path: drain pending event + handlers and background memory saves so their spans close before the + flow span does, then emit and finalize the trace batch. Never raises, + so the original exception propagates unchanged. + + Args: + error: The exception that ended the execution. + respect_suppression: Skip the emission for suppressed flows. Only + the resume path gates its lifecycle events on + ``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 + + try: + if self._event_futures: + await asyncio.gather( + *[asyncio.wrap_future(f) for f in self._event_futures], + return_exceptions=True, + ) + self._event_futures.clear() + + await asyncio.to_thread(self._drain_memory_writes) + await asyncio.to_thread(crewai_event_bus.flush) + future = crewai_event_bus.emit( + self, + FlowFailedEvent( + type="flow_failed", + flow_name=self._definition.name, + error=error, + ), + ) + if future and isinstance(future, Future): + try: + await asyncio.wrap_future(future) + except Exception: + logger.warning("FlowFailedEvent handler failed", exc_info=True) + + trace_listener = TraceCollectionListener() + 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() + else: + trace_listener.batch_manager.finalize_batch() + except Exception: + logger.warning("Failed to signal flow failure", exc_info=True) + async def akickoff( self, inputs: dict[str, Any] | None = None, diff --git a/lib/crewai/src/crewai/state/checkpoint_config.py b/lib/crewai/src/crewai/state/checkpoint_config.py index 1ddae2983..e9fd5e997 100644 --- a/lib/crewai/src/crewai/state/checkpoint_config.py +++ b/lib/crewai/src/crewai/state/checkpoint_config.py @@ -38,6 +38,7 @@ CheckpointEventType = Literal[ "flow_created", "flow_started", "flow_finished", + "flow_failed", "flow_paused", "method_execution_started", "method_execution_finished", diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py index 8b71747b0..d9c9d5529 100644 --- a/lib/crewai/tests/utilities/test_events.py +++ b/lib/crewai/tests/utilities/test_events.py @@ -23,6 +23,7 @@ from crewai.events.types.crew_events import ( ) from crewai.events.types.flow_events import ( FlowCreatedEvent, + FlowFailedEvent, FlowFinishedEvent, FlowStartedEvent, HumanFeedbackReceivedEvent, @@ -46,8 +47,11 @@ from crewai.events.types.tool_usage_events import ( ToolUsageErrorEvent, ToolUsageFinishedEvent, ) +from crewai.flow.async_feedback.types import PendingFeedbackContext from crewai.flow.flow import Flow, listen, start from crewai.flow.human_feedback import human_feedback +from crewai.flow.persistence.sqlite import SQLiteFlowPersistence +from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on from crewai.llm import LLM from crewai.task import Task from crewai.tools.base_tool import BaseTool @@ -556,6 +560,274 @@ def test_flow_emits_finish_event(): assert result == "completed" +def test_flow_emits_failed_event_paired_with_started_event(): + started: list[FlowStartedEvent] = [] + failed: list[FlowFailedEvent] = [] + + class BoomFlow(Flow[dict]): + @start() + def begin(self): + raise RuntimeError("boom") + + 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) + + with pytest.raises(RuntimeError, match="boom"): + BoomFlow().kickoff() + wait_for_event_handlers() + + assert len(failed) == 1 + assert failed[0].type == "flow_failed" + assert failed[0].flow_name == "BoomFlow" + assert isinstance(failed[0].error, RuntimeError) + assert str(failed[0].error) == "boom" + assert failed[0].started_event_id == started[0].event_id + + +def test_suppressed_flow_failure_matches_finished_event_emission(): + finished: list[FlowFinishedEvent] = [] + failed: list[FlowFailedEvent] = [] + + class SuppressedFlow(Flow): + suppress_flow_events: bool = True + + @start() + def begin(self): + return "ok" + + class SuppressedBoomFlow(Flow): + suppress_flow_events: bool = True + + @start() + def begin(self): + raise RuntimeError("boom") + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowFinishedEvent) + def handle_flow_finished(source, event): + finished.append(event) + + @crewai_event_bus.on(FlowFailedEvent) + def handle_flow_failed(source, event): + failed.append(event) + + SuppressedFlow().kickoff() + with pytest.raises(RuntimeError, match="boom"): + SuppressedBoomFlow().kickoff() + wait_for_event_handlers() + + assert len(finished) == 1 + assert len(failed) == 1 + + +def test_abort_before_flow_started_emits_no_failed_event(): + started: list[FlowStartedEvent] = [] + failed: list[FlowFailedEvent] = [] + + class BlockedFlow(Flow): + @start() + def begin(self) -> str: + return "never runs" + + clear_all() + try: + + @on(InterceptionPoint.EXECUTION_START) + def block(_ctx): + raise HookAborted(reason="blocked by policy") + + 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) + + with pytest.raises(HookAborted): + BlockedFlow().kickoff() + wait_for_event_handlers() + finally: + clear_all() + + assert started == [] + assert failed == [] + + +def test_resume_emits_failed_event_paired_with_resume_started_event(tmp_path): + started: list[FlowStartedEvent] = [] + failed: list[FlowFailedEvent] = [] + + class ResumeBoomFlow(Flow): + @start() + def begin(self) -> str: + return "content" + + @listen(begin) + def after_feedback(self, _feedback): + raise RuntimeError("boom on resume") + + persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db")) + flow_id = "resume-failure-test" + persistence.save_pending_feedback( + flow_uuid=flow_id, + context=PendingFeedbackContext( + flow_id=flow_id, + flow_class="ResumeBoomFlow", + method_name="begin", + method_output="content", + message="Review:", + ), + state_data={"id": flow_id}, + ) + + 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) + + flow = ResumeBoomFlow.from_pending(flow_id, persistence) + with pytest.raises(RuntimeError, match="boom on resume"): + flow.resume("ok") + wait_for_event_handlers() + + assert len(started) == 1 + assert len(failed) == 1 + assert failed[0].flow_name == "ResumeBoomFlow" + assert str(failed[0].error) == "boom on resume" + assert failed[0].started_event_id == started[0].event_id + + +def test_resume_pairs_resumed_method_events_with_their_own_scope(tmp_path): + started: list[FlowStartedEvent] = [] + finished: list[FlowFinishedEvent] = [] + method_started: list[MethodExecutionStartedEvent] = [] + method_finished: list[MethodExecutionFinishedEvent] = [] + + class ResumeFlow(Flow): + @start() + def begin(self) -> str: + return "content" + + @listen(begin) + def after_feedback(self, _feedback): + return "done" + + persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db")) + flow_id = "resume-pairing-test" + persistence.save_pending_feedback( + flow_uuid=flow_id, + context=PendingFeedbackContext( + flow_id=flow_id, + flow_class="ResumeFlow", + method_name="begin", + method_output="content", + message="Review:", + ), + state_data={"id": flow_id}, + ) + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(FlowStartedEvent) + def handle_flow_started(source, event): + started.append(event) + + @crewai_event_bus.on(FlowFinishedEvent) + def handle_flow_finished(source, event): + finished.append(event) + + @crewai_event_bus.on(MethodExecutionStartedEvent) + def handle_method_started(source, event): + method_started.append(event) + + @crewai_event_bus.on(MethodExecutionFinishedEvent) + def handle_method_finished(source, event): + method_finished.append(event) + + ResumeFlow.from_pending(flow_id, persistence).resume("ok") + wait_for_event_handlers() + + resumed_started = next(e for e in method_started if e.method_name == "begin") + resumed_finished = next(e for e in method_finished if e.method_name == "begin") + + assert resumed_finished.started_event_id == resumed_started.event_id + assert finished[0].started_event_id == started[0].event_id + + +def test_resume_failing_before_method_finishes_keeps_flow_pairing(tmp_path): + started: list[FlowStartedEvent] = [] + failed: list[FlowFailedEvent] = [] + method_failed: list[MethodExecutionFailedEvent] = [] + + class ResumeFlow(Flow): + @start() + def begin(self) -> str: + return "content" + + @listen(begin) + def after_feedback(self, _feedback): + return "done" + + persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db")) + flow_id = "resume-finalize-failure-test" + persistence.save_pending_feedback( + flow_uuid=flow_id, + context=PendingFeedbackContext( + flow_id=flow_id, + flow_class="ResumeFlow", + method_name="begin", + method_output="content", + message="Review:", + ), + state_data={"id": flow_id}, + ) + + 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(MethodExecutionFailedEvent) + def handle_method_failed(source, event): + method_failed.append(event) + + flow = ResumeFlow.from_pending(flow_id, persistence) + with patch.object( + Flow, + "_finalize_human_feedback", + side_effect=RuntimeError("feedback collapse failed"), + ): + with pytest.raises(RuntimeError, match="feedback collapse failed"): + flow.resume("ok") + wait_for_event_handlers() + + assert len(method_failed) == 1 + assert method_failed[0].method_name == "begin" + assert len(failed) == 1 + assert failed[0].started_event_id == started[0].event_id + + def test_flow_emits_method_execution_started_event(): received_events = [] lock = threading.Lock()