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] = []