fix: dispatch execution_end hook on failed crew and flow executions (#6607)

* fix: dispatch execution_end hook on failed crew and flow executions

The `execution_end` interception point only fired after a successful
kickoff, so consumers never learned about failed runs. Crew kickoff
paths (`kickoff`/`akickoff`) and the flow runtime (`kickoff_async`,
`resume_async`) now dispatch it on the failure path too, with new
additive `status` ("completed"/"failed") and `error` fields on
`ExecutionEndContext`. Pairing flags guarantee exactly-once dispatch,
keep the start/end pairing invariant, and the original exception
propagates unchanged.

* fix: track execution_end pairing per invocation for reentrant flows

Reentrant kickoffs on the same Flow instance are supported (usage
aggregation already accommodates them), but the instance-level pairing
booleans let an inner kickoff's completion mark the outer execution as
ended, skipping the outer failure's `execution_end`. The pairing state
now lives in each `kickoff_async` invocation's locals, and the resume
path passes a per-invocation holder into `_resume_async_body`. Crew
keeps its instance flags since crew kickoffs are not reentrant on the
same instance (`kickoff_for_each` copies the crew).

lib/crewai/src/crewai/crew.py

@@ -217,6 +217,8 @@ class Crew(FlowTrackable, BaseModel):
         default_factory=TaskOutputStorageHandler
     )
     _kickoff_event_id: str | None = PrivateAttr(default=None)
+    _execution_start_dispatched: bool = PrivateAttr(default=False)
+    _execution_end_dispatched: bool = PrivateAttr(default=False)
 
     name: str | None = Field(default="crew")
     cache: bool = Field(
@@ -1050,6 +1052,7 @@ class Crew(FlowTrackable, BaseModel):
 
             return result
         except Exception as e:
+            self._dispatch_execution_end_failure(e)
             crewai_event_bus.emit(
                 self,
                 CrewKickoffFailedEvent(
@@ -1263,6 +1266,7 @@ class Crew(FlowTrackable, BaseModel):
 
             return result
         except Exception as e:
+            self._dispatch_execution_end_failure(e)
             crewai_event_bus.emit(
                 self,
                 CrewKickoffFailedEvent(
@@ -1925,6 +1929,9 @@ class Crew(FlowTrackable, BaseModel):
         end_ctx = ExecutionEndContext(
             crew=self, output=crew_output, payload=crew_output
         )
+        # Flag set before dispatching so an EXECUTION_END hook that raises
+        # HookAborted does not trigger a second (failure) dispatch upstream.
+        self._execution_end_dispatched = True
         dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
         crew_output = cast(CrewOutput, end_ctx.payload)
 
@@ -1951,6 +1958,32 @@ class Crew(FlowTrackable, BaseModel):
 
         return crew_output
 
+    def _dispatch_execution_end_failure(self, error: BaseException) -> None:
+        """Dispatch EXECUTION_END with status="failed" for a kickoff that raised.
+
+        No-op when EXECUTION_START never dispatched (pairing invariant) or when
+        EXECUTION_END already fired for this execution (exactly-once). Never
+        raises, so the original kickoff exception propagates unchanged.
+
+        Instance-level flags are sufficient here because crew kickoffs are not
+        reentrant on the same instance (``kickoff_for_each`` copies the crew;
+        unlike Flow, nothing in the crew runtime supports nested kickoffs).
+        """
+        if not self._execution_start_dispatched or self._execution_end_dispatched:
+            return
+        self._execution_end_dispatched = True
+
+        from crewai.hooks.contexts import ExecutionEndContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        try:
+            dispatch(
+                InterceptionPoint.EXECUTION_END,
+                ExecutionEndContext(crew=self, status="failed", error=error),
+            )
+        except Exception:  # noqa: S110 - aborting an already-failed execution is meaningless
+            pass
+
     def _process_async_tasks(
         self,
         futures: list[tuple[Task, Future[TaskOutput], int]],

lib/crewai/src/crewai/crews/utils.py

@@ -297,7 +297,12 @@ def prepare_kickoff(
         inputs=normalized if normalized is not None else {},
         payload=normalized,
     )
+    # Pairing flags: EXECUTION_END fires (once) only for executions whose
+    # EXECUTION_START actually dispatched, including on the failure path.
+    crew._execution_start_dispatched = False
+    crew._execution_end_dispatched = False
     dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+    crew._execution_start_dispatched = True
     normalized = start_ctx.payload
 
     for before_callback in crew.before_kickoff_callbacks:

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -1332,8 +1332,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         if self._flow_match_id is not None:
             flow_id_token = current_flow_id.set(self._flow_match_id)
         self._attach_usage_aggregation_listener()
+        # Per-invocation pairing state: a resumed execution's EXECUTION_START
+        # fired in the original kickoff, so a failure here still owes the
+        # paired EXECUTION_END (unless the body already dispatched it).
+        hook_state = {"end_dispatched": False}
         try:
-            return await self._resume_async_body(feedback)
+            return await self._resume_async_body(feedback, hook_state)
+        except Exception as e:
+            if not hook_state["end_dispatched"]:
+                self._dispatch_execution_end_failure(e)
+            raise
         finally:
             # Match kickoff_async: drain pending handlers so the resumed
             # phase's LLM events all hit `_aggregated_usage_metrics`
@@ -1343,7 +1351,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             if flow_id_token is not None:
                 current_flow_id.reset(flow_id_token)
 
-    async def _resume_async_body(self, feedback: str = "") -> Any:
+    async def _resume_async_body(
+        self, feedback: str = "", hook_state: dict[str, bool] | None = None
+    ) -> Any:
         if get_current_parent_id() is None:
             reset_emission_counter()
             reset_last_event_id()
@@ -1486,6 +1496,10 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         end_ctx = ExecutionEndContext(
             flow=self, output=final_result, payload=final_result
         )
+        # Flag set before dispatching so an EXECUTION_END hook that raises
+        # HookAborted does not trigger a second (failure) dispatch upstream.
+        if hook_state is not None:
+            hook_state["end_dispatched"] = True
         dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
         final_result = end_ctx.payload
 
@@ -2077,6 +2091,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             self._aggregated_usage_metrics = UsageMetrics()
             self._attach_usage_aggregation_listener()
 
+        # Pairing state is local (per invocation) so reentrant kickoffs on the
+        # same instance (see usage aggregation above) each track their own
+        # EXECUTION_START/EXECUTION_END dispatch independently.
+        execution_start_dispatched = False
+        execution_end_dispatched = False
+
         try:
             from crewai.hooks.contexts import (
                 ExecutionEndContext,
@@ -2094,6 +2114,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 payload=inputs,
             )
             dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+            execution_start_dispatched = True
             inputs = start_ctx.payload
 
             input_ctx = InputContext(
@@ -2356,6 +2377,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             end_ctx = ExecutionEndContext(
                 flow=self, output=final_output, payload=final_output
             )
+            # Flag set before dispatching so an EXECUTION_END hook that raises
+            # HookAborted does not trigger a second (failure) dispatch below.
+            execution_end_dispatched = True
             dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
             final_output = end_ctx.payload
 
@@ -2410,6 +2434,13 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                         trace_listener.batch_manager.finalize_batch()
 
             return final_output
+        except Exception as e:
+            # Pairing invariant: only fire the failure EXECUTION_END when this
+            # invocation's EXECUTION_START dispatched and its EXECUTION_END has
+            # not (exactly-once per invocation).
+            if execution_start_dispatched and not execution_end_dispatched:
+                self._dispatch_execution_end_failure(e)
+            raise
         finally:
             # Safety net for the exception path; the success path already
             # drained before emitting FlowFinishedEvent.
@@ -2437,6 +2468,25 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             detach(flow_token)
             crewai_event_bus._exit_runtime_scope(runtime_scope)
 
+    def _dispatch_execution_end_failure(self, error: BaseException) -> None:
+        """Dispatch EXECUTION_END with status="failed" for an execution that raised.
+
+        Callers enforce the pairing invariant (EXECUTION_START dispatched,
+        EXECUTION_END not yet) with per-invocation state, so reentrant kickoffs
+        on the same instance stay exactly-once. Never raises, so the original
+        exception propagates unchanged.
+        """
+        from crewai.hooks.contexts import ExecutionEndContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        try:
+            dispatch(
+                InterceptionPoint.EXECUTION_END,
+                ExecutionEndContext(flow=self, status="failed", error=error),
+            )
+        except Exception:  # noqa: S110 - aborting an already-failed execution is meaningless
+            pass
+
     async def akickoff(
         self,
         inputs: dict[str, Any] | None = None,

lib/crewai/src/crewai/hooks/contexts.py

@@ -15,7 +15,7 @@ compatibility; they are intentionally not redefined here.
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Literal
 
 
 @dataclass
@@ -53,9 +53,17 @@ class OutputContext(InterceptionContext):
 
 @dataclass
 class ExecutionEndContext(InterceptionContext):
-    """``execution_end``: a crew or flow has finished. ``payload`` = the output object."""
+    """``execution_end``: a crew or flow has finished. ``payload`` = the output object.
+
+    Dispatched on both successful and failed executions. ``status`` is
+    ``"completed"`` on success and ``"failed"`` when the execution raised;
+    ``error`` carries the exception on failure (``output``/``payload`` are
+    ``None`` in that case).
+    """
 
     output: Any = None
+    status: Literal["completed", "failed"] = "completed"
+    error: BaseException | None = None
 
 
 @dataclass

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -7,6 +7,7 @@ sees a well-shaped payload, an in-place/returned modification is honored, and a
 
 from __future__ import annotations
 
+import asyncio
 from unittest.mock import patch
 
 from crewai.agent import Agent
@@ -41,6 +42,24 @@ class _SimpleFlow(Flow):
         return "flow-result"
 
 
+class _FailingFlow(Flow):
+    @start()
+    def begin(self):
+        raise RuntimeError("flow boom")
+
+
+class _ReentrantFailingFlow(Flow):
+    """Kicks itself off once from inside a method, then fails in the outer run."""
+
+    @start()
+    async def begin(self):
+        if getattr(self, "_reentered", False):
+            return "inner-ok"
+        self._reentered = True
+        await self.kickoff_async()
+        raise RuntimeError("outer boom")
+
+
 class TestFlowExecutionBoundaries:
     """execution_start / input / output / execution_end on a flow."""
 
@@ -152,6 +171,157 @@ class TestTaskStepPoints:
         assert (tmp_path / "output.txt").read_text() == "sanitized output"
 
 
+class TestExecutionEndOnFailure:
+    """execution_end fires exactly once, on success and on failure alike."""
+
+    @staticmethod
+    def _crew() -> Crew:
+        agent = Agent(role="Writer", goal="Write", backstory="Writes things.")
+        task = Task(
+            description="Write something",
+            expected_output="Some text",
+            agent=agent,
+        )
+        return Crew(agents=[agent], tasks=[task], verbose=False)
+
+    def test_crew_success_fires_completed_once(self):
+        seen: list[tuple[str, BaseException | None]] = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append((ctx.status, ctx.error))
+
+        with patch.object(Agent, "execute_task", return_value="fine"):
+            self._crew().kickoff()
+
+        assert seen == [("completed", None)]
+
+    def test_crew_failure_fires_failed_once_and_reraises(self):
+        seen = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append(ctx)
+
+        error = RuntimeError("crew boom")
+        with patch.object(Agent, "execute_task", side_effect=error):
+            with pytest.raises(RuntimeError, match="crew boom"):
+                self._crew().kickoff()
+
+        assert len(seen) == 1
+        assert seen[0].status == "failed"
+        assert seen[0].error is error
+        assert seen[0].output is None
+
+    def test_crew_kickoff_async_failure_fires_failed_once(self):
+        seen = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append(ctx)
+
+        with patch.object(
+            Agent, "execute_task", side_effect=RuntimeError("crew boom")
+        ):
+            with pytest.raises(RuntimeError, match="crew boom"):
+                asyncio.run(self._crew().kickoff_async())
+
+        assert len(seen) == 1
+        assert seen[0].status == "failed"
+
+    def test_flow_success_fires_completed_once(self):
+        seen: list[tuple[str, BaseException | None]] = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append((ctx.status, ctx.error))
+
+        _SimpleFlow().kickoff()
+
+        assert seen == [("completed", None)]
+
+    def test_flow_failure_fires_failed_once_and_reraises(self):
+        seen = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append(ctx)
+
+        with pytest.raises(RuntimeError, match="flow boom"):
+            _FailingFlow().kickoff()
+
+        assert len(seen) == 1
+        assert seen[0].status == "failed"
+        assert isinstance(seen[0].error, RuntimeError)
+        assert seen[0].output is None
+
+    def test_reentrant_flow_kickoff_pairs_ends_per_invocation(self):
+        seen: list[tuple[str, str | None]] = []
+
+        @on(InterceptionPoint.EXECUTION_START)
+        def capture_start(ctx):
+            seen.append(("start", None))
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture_end(ctx):
+            seen.append(("end", ctx.status))
+
+        with pytest.raises(RuntimeError, match="outer boom"):
+            _ReentrantFailingFlow().kickoff()
+
+        assert seen == [
+            ("start", None),
+            ("start", None),
+            ("end", "completed"),
+            ("end", "failed"),
+        ]
+
+    def test_no_execution_end_when_execution_start_aborts(self):
+        seen = []
+
+        @on(InterceptionPoint.EXECUTION_START)
+        def block(ctx):
+            raise HookAborted(reason="blocked")
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def capture(ctx):
+            seen.append(ctx)
+
+        with pytest.raises(HookAborted):
+            _SimpleFlow().kickoff()
+        with pytest.raises(HookAborted):
+            self._crew().kickoff()
+
+        assert seen == []
+
+    def test_aborting_execution_end_hook_fires_once_for_flow(self):
+        calls: list[str] = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def abort_end(ctx):
+            calls.append(ctx.status)
+            raise HookAborted(reason="no")
+
+        with pytest.raises(HookAborted):
+            _SimpleFlow().kickoff()
+
+        assert calls == ["completed"]
+
+    def test_aborting_execution_end_hook_fires_once_for_crew(self):
+        calls: list[str] = []
+
+        @on(InterceptionPoint.EXECUTION_END)
+        def abort_end(ctx):
+            calls.append(ctx.status)
+            raise HookAborted(reason="no")
+
+        with patch.object(Agent, "execute_task", return_value="fine"):
+            with pytest.raises(HookAborted):
+                self._crew().kickoff()
+
+        assert calls == ["completed"]
+
+
 class TestCrewOutput:
     def test_output_modification_reaches_kickoff_completed_event(self):
         @on(InterceptionPoint.OUTPUT)