From 7fd1997e8eb34a4458c1b24d55000092e6db98e0 Mon Sep 17 00:00:00 2001
From: Lucas Gomide <lucaslg200@gmail.com>
Date: Wed, 29 Jul 2026 14:57:10 -0300
Subject: [PATCH] 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.
---
 .../src/crewai/flow/runtime/__init__.py       | 10 ++++-
 lib/crewai/tests/utilities/test_events.py     | 37 +++++++++++++++++++
 2 files changed, 46 insertions(+), 1 deletion(-)

diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py
index d897e9364..4bb78f9fd 100644
--- a/lib/crewai/src/crewai/flow/runtime/__init__.py
+++ b/lib/crewai/src/crewai/flow/runtime/__init__.py
@@ -2131,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 (
@@ -2270,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()
@@ -2284,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)
@@ -2475,7 +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)
-            await self._emit_flow_failed(e)
+            if flow_scope_open:
+                await self._emit_flow_failed(e)
             raise
         finally:
             # Safety net for the exception path; the success path already
diff --git a/lib/crewai/tests/utilities/test_events.py b/lib/crewai/tests/utilities/test_events.py
index 42980d7d4..d9c9d5529 100644
--- a/lib/crewai/tests/utilities/test_events.py
+++ b/lib/crewai/tests/utilities/test_events.py
@@ -51,6 +51,7 @@ 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
@@ -627,6 +628,42 @@ def test_suppressed_flow_failure_matches_finished_event_emission():
     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] = []