diff --git a/docs/edge/en/learn/execution-boundary-hooks.mdx b/docs/edge/en/learn/execution-boundary-hooks.mdx index 5b7a2ffe8..80e2b382e 100644 --- a/docs/edge/en/learn/execution-boundary-hooks.mdx +++ b/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" ``` @@ -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: diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index ce3822387..ed9b77ebd 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/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]], diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index bfac589ec..cc155806a 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/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: diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index eaf2b8b38..1781e3e2e 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/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, diff --git a/lib/crewai/src/crewai/hooks/contexts.py b/lib/crewai/src/crewai/hooks/contexts.py index ec1366233..36759a30c 100644 --- a/lib/crewai/src/crewai/hooks/contexts.py +++ b/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 diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py index c9371bcc6..c2caff531 100644 --- a/lib/crewai/tests/hooks/test_interception_conformance.py +++ b/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)