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

@@ -1924,7 +1924,10 @@ class Crew(FlowTrackable, BaseModel):
         # 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 @@ class Crew(FlowTrackable, BaseModel):
             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]],

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

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

@@ -2062,6 +2062,24 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             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 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             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 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     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

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

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"

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"