fix: dispatch execution_end hook on failed crew and flow executions

docs/edge/en/learn/execution-boundary-hooks.mdx

@@ -18,7 +18,7 @@ Four interception points cover the boundaries:
 | `EXECUTION_START` | A crew or flow is about to begin | inputs `dict` |
 | `INPUT` | Resolved inputs for the execution | inputs `dict` |
 | `OUTPUT` | The final result is ready | the output object |
-| `EXECUTION_END` | The execution has finished | the output object |
+| `EXECUTION_END` | The execution has finished (success or failure) | the output object, or `None` on failure |
 
 For a crew, the output payload is a `CrewOutput`. For a flow, it is the final
 flow-method result.
@@ -68,7 +68,9 @@ class OutputContext(InterceptionContext):
     output: Any             # The output object
 
 class ExecutionEndContext(InterceptionContext):
-    output: Any             # The output object
+    output: Any                    # The output object (None when status == "failed")
+    status: str                    # "completed" or "failed"
+    error: BaseException | None    # The exception when status == "failed"
 ```
 
 <Note>
@@ -136,6 +138,28 @@ def redact_emails(ctx):
 payload from earlier hooks; the final rewritten value is what `kickoff()`
 returns.
 
+### Observing Failures
+
+`EXECUTION_END` fires exactly once per execution, on success and on failure
+alike. When the run raises — a task error, a flow-method exception, or a
+`HookAborted` from an earlier point — the hook receives `status="failed"` with
+the exception in `ctx.error`, and the original exception still propagates out
+of `kickoff()` unchanged:
+
+```python
+@on(InterceptionPoint.EXECUTION_END)
+def report_outcome(ctx):
+    if ctx.status == "failed":
+        notify_policy_engine(status="failed", error=repr(ctx.error))
+    else:
+        notify_policy_engine(status="completed")
+```
+
+Two caveats: `EXECUTION_END` does not fire when `EXECUTION_START` never
+dispatched (an abort at start counts as the execution never beginning), and
+raising `HookAborted` from a failure-path `EXECUTION_END` dispatch is ignored —
+there is nothing left to abort, and the original error wins.
+
 ## Ordering
 
 For a crew run the boundary order is:

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 @@ def run_crew() -> None:
 
             return result
         except Exception as e:
+            self._dispatch_execution_end_failure(e)
             crewai_event_bus.emit(
                 self,
                 CrewKickoffFailedEvent(
@@ -1263,6 +1266,7 @@ async def run_crew() -> None:
 
             return result
         except Exception as e:
+            self._dispatch_execution_end_failure(e)
             crewai_event_bus.emit(
                 self,
                 CrewKickoffFailedEvent(
@@ -1925,6 +1929,9 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput:
         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 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput:
 
         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 @@ async def resume_async(self, feedback: str = "") -> Any:
         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 @@ async def resume_async(self, feedback: str = "") -> Any:
             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 @@ async def _resume_async_body(self, feedback: str = "") -> Any:
         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 @@ async def kickoff_async(
             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 @@ async def kickoff_async(
                 payload=inputs,
             )
             dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
+            execution_start_dispatched = True
             inputs = start_ctx.payload
 
             input_ctx = InputContext(
@@ -2356,6 +2377,9 @@ async def kickoff_async(
             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 @@ async def kickoff_async(
                         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 @@ async def kickoff_async(
             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 @@
 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