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()