diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py
index 696d831c80..686b08cc40 100644
--- a/lib/crewai/src/crewai/crew.py
+++ b/lib/crewai/src/crewai/crew.py
@@ -1906,6 +1906,28 @@ 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()
+
+        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,
+            tasks_output=task_outputs,
+            token_usage=self.token_usage,
+        )
+
+        output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output)
+        dispatch(InterceptionPoint.OUTPUT, output_ctx)
+        crew_output = cast(CrewOutput, output_ctx.payload)
+
+        end_ctx = ExecutionEndContext(
+            crew=self, output=crew_output, payload=crew_output
+        )
+        dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
+        crew_output = cast(CrewOutput, end_ctx.payload)
+
         # Ensure background memory saves finish (and emit their
         # completed/failed events) before the kickoff-completed event below
         # triggers listener teardown/finalization.
@@ -1924,13 +1946,7 @@ 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(
-            raw=final_task_output.raw,
-            pydantic=final_task_output.pydantic,
-            json_dict=final_task_output.json_dict,
-            tasks_output=task_outputs,
-            token_usage=self.token_usage,
-        )
+        return crew_output
 
     def _process_async_tasks(
         self,
diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py
index 6faa0e8c28..bfac589ec8 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,30 @@ 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 if normalized is not None else {},
+        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 if normalized is not None else {},
+        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..6733c46458 100644
--- a/lib/crewai/src/crewai/flow/runtime/__init__.py
+++ b/lib/crewai/src/crewai/flow/runtime/__init__.py
@@ -1476,6 +1476,19 @@ async def _resume_async_body(self, feedback: str = "") -> Any:
             else (resumed_method_output if emit else result)
         )
 
+        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 +2050,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(
@@ -2062,6 +2078,37 @@ 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
+
+            # ``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 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 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
@@ -2297,6 +2344,21 @@ 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
+
+            # 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]
@@ -2370,6 +2432,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)
 
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..9c2965bd24
--- /dev/null
+++ b/lib/crewai/tests/hooks/test_interception_conformance.py
@@ -0,0 +1,88 @@
+"""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.
+"""
+
+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"