From d114a0f73c2a547cc3d867bf689a6b41ff0edf62 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:15:56 -0300 Subject: [PATCH 1/4] feat: wire execution-boundary interception points Adds the typed interception contexts (`crewai/hooks/contexts.py`) and wires the `execution_start`, `input`, `output`, and `execution_end` points for both crews and flows through the dispatcher. `prepare_kickoff` and `Flow.kickoff_async` fire `execution_start`/`input` so a hook can rewrite resolved inputs before the run, while `Crew._create_crew_output` and the flow tail fire `output`/`execution_end` so the final result can be observed or replaced. Closes the eight critical-path points without touching the legacy hooks. --- lib/crewai/src/crewai/crew.py | 14 ++- lib/crewai/src/crewai/crews/utils.py | 13 +++ .../src/crewai/flow/runtime/__init__.py | 29 ++++++ lib/crewai/src/crewai/hooks/contexts.py | 58 ++++++++++++ lib/crewai/src/crewai/hooks/dispatch.py | 10 ++- .../hooks/test_interception_conformance.py | 89 +++++++++++++++++++ 6 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 lib/crewai/src/crewai/hooks/contexts.py create mode 100644 lib/crewai/tests/hooks/test_interception_conformance.py diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 696d831c80..2179458c8e 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1924,7 +1924,10 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: # Finalization is handled by trace listener (always initialized) # The batch manager checks contextvar to determine if tracing is enabled - return CrewOutput( + from crewai.hooks.contexts import ExecutionEndContext, OutputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + crew_output = CrewOutput( raw=final_task_output.raw, pydantic=final_task_output.pydantic, json_dict=final_task_output.json_dict, @@ -1932,6 +1935,15 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: token_usage=self.token_usage, ) + output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) + dispatch(InterceptionPoint.OUTPUT, output_ctx) + crew_output = output_ctx.payload + + end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + + return crew_output + 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 6faa0e8c28..c096a8fd24 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -278,6 +278,9 @@ def prepare_kickoff( reset_emission_counter() reset_last_event_id() + from crewai.hooks.contexts import ExecutionStartContext, InputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + normalized: dict[str, Any] | None = None if inputs is not None: if not isinstance(inputs, Mapping): @@ -286,11 +289,21 @@ def prepare_kickoff( ) normalized = dict(inputs) + start_ctx = ExecutionStartContext( + crew=crew, inputs=normalized or {}, payload=normalized + ) + dispatch(InterceptionPoint.EXECUTION_START, start_ctx) + normalized = start_ctx.payload + for before_callback in crew.before_kickoff_callbacks: if normalized is None: normalized = {} normalized = before_callback(normalized) + input_ctx = InputContext(crew=crew, inputs=normalized or {}, payload=normalized) + dispatch(InterceptionPoint.INPUT, input_ctx) + normalized = input_ctx.payload + if resuming and crew._kickoff_event_id: if crew.verbose: from crewai.events.utils.console_formatter import ConsoleFormatter diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index a6c73e6979..201ec4c047 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2062,6 +2062,24 @@ async def kickoff_async( self._attach_usage_aggregation_listener() try: + from crewai.hooks.contexts import ( + ExecutionEndContext, + ExecutionStartContext, + InputContext, + OutputContext, + ) + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + start_ctx = ExecutionStartContext( + flow=self, inputs=inputs or {}, payload=inputs + ) + dispatch(InterceptionPoint.EXECUTION_START, start_ctx) + inputs = start_ctx.payload + + input_ctx = InputContext(flow=self, inputs=inputs or {}, payload=inputs) + dispatch(InterceptionPoint.INPUT, input_ctx) + inputs = input_ctx.payload + # Reset flow state for fresh execution unless restoring from persistence is_restoring = ( inputs and "id" in inputs and self.persistence is not None @@ -2297,6 +2315,12 @@ async def kickoff_async( method_outputs = self.method_outputs final_output = method_outputs[-1] if method_outputs else None + output_ctx = OutputContext( + flow=self, output=final_output, payload=final_output + ) + dispatch(InterceptionPoint.OUTPUT, output_ctx) + final_output = output_ctx.payload + if self._event_futures: await asyncio.gather( *[asyncio.wrap_future(f) for f in self._event_futures] @@ -2347,6 +2371,11 @@ async def kickoff_async( else: trace_listener.batch_manager.finalize_batch() + end_ctx = ExecutionEndContext( + flow=self, output=final_output, payload=final_output + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + return final_output finally: # Safety net for the exception path; the success path already diff --git a/lib/crewai/src/crewai/hooks/contexts.py b/lib/crewai/src/crewai/hooks/contexts.py new file mode 100644 index 0000000000..d718903b13 --- /dev/null +++ b/lib/crewai/src/crewai/hooks/contexts.py @@ -0,0 +1,58 @@ +"""Typed contexts for the interception points wired in phases 2-5. + +Each context is a dataclass whose fields are nullable and defaulted, so a field +that is not meaningful for a given runtime (e.g. ``agent_role`` inside a flow) +is simply ``None`` rather than an error. Every context exposes a ``payload`` +field: the interceptable value a hook may mutate in place or replace by +returning a new value. + +The legacy ``pre/post_model_call`` and ``pre/post_tool_call`` points keep using +:class:`~crewai.hooks.llm_hooks.LLMCallHookContext` and +:class:`~crewai.hooks.tool_hooks.ToolCallHookContext` for backwards +compatibility; they are intentionally not redefined here. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class InterceptionContext: + """Base context shared by the framework-native interception points.""" + + payload: Any = None + agent: Any = None + agent_role: str | None = None + task: Any = None + crew: Any = None + flow: Any = None + + +@dataclass +class ExecutionStartContext(InterceptionContext): + """``execution_start``: a crew or flow is about to begin. ``payload`` = inputs.""" + + inputs: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class InputContext(InterceptionContext): + """``input``: resolved inputs for an execution. ``payload`` = inputs.""" + + inputs: dict[str, Any] = field(default_factory=dict) + + +@dataclass +class OutputContext(InterceptionContext): + """``output``: final result of a crew or flow. ``payload`` = the output object.""" + + output: Any = None + + +@dataclass +class ExecutionEndContext(InterceptionContext): + """``execution_end``: a crew or flow has finished. ``payload`` = the output object.""" + + output: Any = None diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index 2822c73dd8..095484220e 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -41,11 +41,15 @@ class InterceptionPoint(str, Enum): """Interception points wired by this layer. New points are introduced alongside the seams that dispatch them, so the - enum only ever lists points with a live consumer. This layer ships the - model- and tool-call boundaries, which back the legacy - ``before/after_llm_call`` and ``before/after_tool_call`` hooks. + enum only ever lists points with a live consumer. """ + # Execution-level boundaries + EXECUTION_START = "execution_start" + INPUT = "input" + OUTPUT = "output" + EXECUTION_END = "execution_end" + # Model / tool boundaries (legacy-compatible) PRE_MODEL_CALL = "pre_model_call" POST_MODEL_CALL = "post_model_call" diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py new file mode 100644 index 0000000000..3593699d52 --- /dev/null +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -0,0 +1,89 @@ +"""Conformance suite for the framework-native interception points. + +For each wired point this suite asserts the shared contract: the probe hook +sees a well-shaped payload, an in-place/returned modification is honored, and a +:class:`HookAborted` interrupts the step. Enterprise / ACS adapters build +against these guarantees. +""" + +from __future__ import annotations + +from crewai.flow.flow import Flow, listen, start +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear_all, + on, +) +import pytest + + +@pytest.fixture(autouse=True) +def clear_dispatch_registry(): + clear_all() + yield + clear_all() + + +class _SimpleFlow(Flow): + @start() + def begin(self): + return "begin" + + @listen(begin) + def finish(self, _): + return "flow-result" + + +class TestFlowExecutionBoundaries: + """execution_start / input / output / execution_end on a flow.""" + + def test_all_boundary_points_fire_once(self): + fired: list[str] = [] + + for point in ( + InterceptionPoint.EXECUTION_START, + InterceptionPoint.INPUT, + InterceptionPoint.OUTPUT, + InterceptionPoint.EXECUTION_END, + ): + + @on(point) + def _probe(ctx, _point=point): + fired.append(_point.value) + + _SimpleFlow().kickoff(inputs={"seed": 1}) + + assert fired == [ + "execution_start", + "input", + "output", + "execution_end", + ] + + def test_output_modification_is_honored(self): + @on(InterceptionPoint.OUTPUT) + def rewrite(ctx): + return "intercepted" + + result = _SimpleFlow().kickoff() + assert result == "intercepted" + + def test_input_payload_carries_inputs(self): + seen: dict = {} + + @on(InterceptionPoint.INPUT) + def capture(ctx): + seen.update(ctx.payload or {}) + + _SimpleFlow().kickoff(inputs={"seed": 42}) + assert seen == {"seed": 42} + + def test_abort_at_execution_start_interrupts(self): + @on(InterceptionPoint.EXECUTION_START) + def block(ctx): + raise HookAborted(reason="not allowed", source="policy") + + with pytest.raises(HookAborted) as exc: + _SimpleFlow().kickoff() + assert exc.value.reason == "not allowed" From 7d3e83ddff4d9578e8303ef61164e4f2645b02b0 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:16:04 -0300 Subject: [PATCH 2/4] fix: correct execution-boundary hook ordering and input aliasing Reworks the crew and flow boundary seams flagged in review. `OUTPUT` and `EXECUTION_END` now run before the completion event (`CrewKickoffCompletedEvent` and `FlowFinishedEvent`) so a `HookAborted` no longer leaves a spurious completed signal and a returned payload replacement is honored on the emitted and returned result. Boundary contexts alias `inputs` to the same object as `payload` instead of a fresh dict from `or`, so in-place edits survive read-back. Flows re-publish the resolved inputs into `flow_inputs` baggage after the `INPUT` hook so trigger-payload injection observes hook rewrites, and a resumed flow now dispatches `OUTPUT`/`EXECUTION_END` on its completion path. --- lib/crewai/src/crewai/crew.py | 40 ++++++++------ lib/crewai/src/crewai/crews/utils.py | 13 ++++- .../src/crewai/flow/runtime/__init__.py | 52 ++++++++++++++++--- 3 files changed, 79 insertions(+), 26 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 2179458c8e..db49b3a1e0 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1906,23 +1906,6 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: final_string_output = final_task_output.raw self._finish_execution(final_string_output) self.token_usage = self.calculate_usage_metrics() - # Ensure background memory saves finish (and emit their - # completed/failed events) before the kickoff-completed event below - # triggers listener teardown/finalization. - self._drain_memory_writes() - crewai_event_bus.flush() - crewai_event_bus.emit( - self, - CrewKickoffCompletedEvent( - crew_name=self.name, - output=final_task_output, - total_tokens=self.token_usage.total_tokens, - started_event_id=self._kickoff_event_id, - ), - ) - - # Finalization is handled by trace listener (always initialized) - # The batch manager checks contextvar to determine if tracing is enabled from crewai.hooks.contexts import ExecutionEndContext, OutputContext from crewai.hooks.dispatch import InterceptionPoint, dispatch @@ -1935,12 +1918,35 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: token_usage=self.token_usage, ) + # OUTPUT/EXECUTION_END run before the kickoff-completed event (mirroring + # the flow OUTPUT-before-FlowFinishedEvent ordering) so a HookAborted + # prevents a spurious completed signal and any payload replacement is + # honored on the returned output. output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) dispatch(InterceptionPoint.OUTPUT, output_ctx) crew_output = output_ctx.payload end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output) dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + crew_output = end_ctx.payload + + # Ensure background memory saves finish (and emit their + # completed/failed events) before the kickoff-completed event below + # triggers listener teardown/finalization. + self._drain_memory_writes() + crewai_event_bus.flush() + crewai_event_bus.emit( + self, + CrewKickoffCompletedEvent( + crew_name=self.name, + output=final_task_output, + total_tokens=self.token_usage.total_tokens, + started_event_id=self._kickoff_event_id, + ), + ) + + # Finalization is handled by trace listener (always initialized) + # The batch manager checks contextvar to determine if tracing is enabled return crew_output diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index c096a8fd24..bfac589ec8 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -289,8 +289,13 @@ def prepare_kickoff( ) normalized = dict(inputs) + # ``inputs`` aliases the same object as ``payload`` (not a fresh ``{}`` from + # ``or``) so in-place edits to either survive read-back, per the context + # contract. ``None`` inputs are preserved rather than coerced to ``{}``. start_ctx = ExecutionStartContext( - crew=crew, inputs=normalized or {}, payload=normalized + crew=crew, + inputs=normalized if normalized is not None else {}, + payload=normalized, ) dispatch(InterceptionPoint.EXECUTION_START, start_ctx) normalized = start_ctx.payload @@ -300,7 +305,11 @@ def prepare_kickoff( normalized = {} normalized = before_callback(normalized) - input_ctx = InputContext(crew=crew, inputs=normalized or {}, payload=normalized) + input_ctx = InputContext( + crew=crew, + inputs=normalized if normalized is not None else {}, + payload=normalized, + ) dispatch(InterceptionPoint.INPUT, input_ctx) normalized = input_ctx.payload diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 201ec4c047..812d83268f 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -1476,6 +1476,22 @@ async def _resume_async_body(self, feedback: str = "") -> Any: else (resumed_method_output if emit else result) ) + # A resumed flow completes here rather than in kickoff_async, so the + # OUTPUT/EXECUTION_END seams must fire on this path too (before + # FlowFinishedEvent) to expose the final result to policy hooks. + from crewai.hooks.contexts import ExecutionEndContext, OutputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + + output_ctx = OutputContext(flow=self, output=final_result, payload=final_result) + dispatch(InterceptionPoint.OUTPUT, output_ctx) + final_result = output_ctx.payload + + end_ctx = ExecutionEndContext( + flow=self, output=final_result, payload=final_result + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + final_result = end_ctx.payload + if self._event_futures: await asyncio.gather( *[ @@ -2037,6 +2053,9 @@ async def kickoff_async( flow_name_token = None flow_defer_trace_finalization_token = None request_id_token = None + # Re-published after the INPUT hook so trigger-payload injection reads + # the hook-rewritten inputs rather than the pre-hook baggage above. + flow_inputs_token = None if current_flow_id.get() is None: flow_id_token = current_flow_id.set(self.flow_id) flow_name_token = current_flow_name.set( @@ -2070,16 +2089,29 @@ async def kickoff_async( ) from crewai.hooks.dispatch import InterceptionPoint, dispatch + # ``inputs`` aliases the same object as ``payload`` (not a fresh + # ``{}`` from ``or``) so in-place edits survive read-back. start_ctx = ExecutionStartContext( - flow=self, inputs=inputs or {}, payload=inputs + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, ) dispatch(InterceptionPoint.EXECUTION_START, start_ctx) inputs = start_ctx.payload - input_ctx = InputContext(flow=self, inputs=inputs or {}, payload=inputs) + input_ctx = InputContext( + flow=self, + inputs=inputs if inputs is not None else {}, + payload=inputs, + ) dispatch(InterceptionPoint.INPUT, input_ctx) inputs = input_ctx.payload + # Publish the resolved inputs so trigger-payload injection and other + # baggage readers observe hook rewrites (the baggage set before the + # hooks carried the pre-hook inputs). + flow_inputs_token = attach(baggage.set_baggage("flow_inputs", inputs or {})) + # Reset flow state for fresh execution unless restoring from persistence is_restoring = ( inputs and "id" in inputs and self.persistence is not None @@ -2321,6 +2353,15 @@ async def kickoff_async( dispatch(InterceptionPoint.OUTPUT, output_ctx) final_output = output_ctx.payload + # EXECUTION_END runs before FlowFinishedEvent so a HookAborted + # prevents a spurious finished signal and payload replacement is + # honored on the emitted result and the returned value. + end_ctx = ExecutionEndContext( + flow=self, output=final_output, payload=final_output + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + final_output = end_ctx.payload + if self._event_futures: await asyncio.gather( *[asyncio.wrap_future(f) for f in self._event_futures] @@ -2371,11 +2412,6 @@ async def kickoff_async( else: trace_listener.batch_manager.finalize_batch() - end_ctx = ExecutionEndContext( - flow=self, output=final_output, payload=final_output - ) - dispatch(InterceptionPoint.EXECUTION_END, end_ctx) - return final_output finally: # Safety net for the exception path; the success path already @@ -2399,6 +2435,8 @@ async def kickoff_async( current_flow_name.reset(flow_name_token) if flow_id_token is not None: current_flow_id.reset(flow_id_token) + if flow_inputs_token is not None: + detach(flow_inputs_token) detach(flow_token) crewai_event_bus._exit_runtime_scope(runtime_scope) From 2781b7106c52a97cd72c4c317989a2a4fed2970d Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 14 Jul 2026 02:29:50 -0300 Subject: [PATCH 3/4] chore: drop redundant seam comments from execution-boundary wiring Removes two inline comments narrating the OUTPUT/EXECUTION_END dispatch ordering in `crew.py` and the flow runtime, plus a stray sentence about enterprise adapters in the conformance-suite docstring. Comment-only cleanup, no behavior change. --- lib/crewai/src/crewai/crew.py | 4 ---- lib/crewai/src/crewai/flow/runtime/__init__.py | 3 --- lib/crewai/tests/hooks/test_interception_conformance.py | 3 +-- 3 files changed, 1 insertion(+), 9 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index db49b3a1e0..ea0272c041 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1918,10 +1918,6 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: token_usage=self.token_usage, ) - # OUTPUT/EXECUTION_END run before the kickoff-completed event (mirroring - # the flow OUTPUT-before-FlowFinishedEvent ordering) so a HookAborted - # prevents a spurious completed signal and any payload replacement is - # honored on the returned output. output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) dispatch(InterceptionPoint.OUTPUT, output_ctx) crew_output = output_ctx.payload diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 812d83268f..6733c46458 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -1476,9 +1476,6 @@ async def _resume_async_body(self, feedback: str = "") -> Any: else (resumed_method_output if emit else result) ) - # A resumed flow completes here rather than in kickoff_async, so the - # OUTPUT/EXECUTION_END seams must fire on this path too (before - # FlowFinishedEvent) to expose the final result to policy hooks. from crewai.hooks.contexts import ExecutionEndContext, OutputContext from crewai.hooks.dispatch import InterceptionPoint, dispatch diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py index 3593699d52..9c2965bd24 100644 --- a/lib/crewai/tests/hooks/test_interception_conformance.py +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -2,8 +2,7 @@ For each wired point this suite asserts the shared contract: the probe hook sees a well-shaped payload, an in-place/returned modification is honored, and a -:class:`HookAborted` interrupts the step. Enterprise / ACS adapters build -against these guarantees. +:class:`HookAborted` interrupts the step. """ from __future__ import annotations From fa0ed74405f64ddb7e08e73ea6d09e7757732cd4 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Tue, 14 Jul 2026 02:37:08 -0300 Subject: [PATCH 4/4] fix: keep crew output typed across boundary hook dispatch `_create_crew_output` reassigned `crew_output` from the hook contexts' `payload`, which is typed `Any`, so mypy flagged `no-any-return` at the function's return. Cast the payload back to `CrewOutput` after each dispatch and split the `ExecutionEndContext` construction to satisfy `ruff format`'s line-length limit. --- lib/crewai/src/crewai/crew.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index ea0272c041..686b08cc40 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1920,11 +1920,13 @@ def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput: output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output) dispatch(InterceptionPoint.OUTPUT, output_ctx) - crew_output = output_ctx.payload + crew_output = cast(CrewOutput, output_ctx.payload) - end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output) + end_ctx = ExecutionEndContext( + crew=self, output=crew_output, payload=crew_output + ) dispatch(InterceptionPoint.EXECUTION_END, end_ctx) - crew_output = end_ctx.payload + crew_output = cast(CrewOutput, end_ctx.payload) # Ensure background memory saves finish (and emit their # completed/failed events) before the kickoff-completed event below