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.

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,75 @@ 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.
+
+        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
+                the resume path gates its lifecycle events on
+                ``suppress_flow_events``; ``kickoff_async`` emits them either
+                way and lets listeners filter.
+        """
+        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)
+
+            # 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"
+                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,15 @@ 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
+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 +564,320 @@ 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_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] = []
+
+    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()