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

@@ -157,6 +157,7 @@ class MyCustomCrew:
 - **FlowCreatedEvent**: يُرسل عند إنشاء تدفق
 - **FlowStartedEvent**: يُرسل عند بدء تنفيذ تدفق
 - **FlowFinishedEvent**: يُرسل عند اكتمال تنفيذ تدفق
+- **FlowFailedEvent**: يُرسل عند فشل تنفيذ تدفق. يحتوي على اسم التدفق والاستثناء الذي أنهى التنفيذ.
 - **FlowPausedEvent**: يُرسل عند إيقاف تدفق مؤقتًا بانتظار ملاحظات بشرية
 
 ### أحداث LLM

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

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 메서드가 실행을 시작할 때 발생

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

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",

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",

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

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

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)

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.
 

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,

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",

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