refactor: drop speculative interception points from the catalog

Removes `pre_delegation`, `retry_attempt`, `flow_transition`, and
`router_decision` — none has a hard use case yet, and `retry_attempt`
was wired at three unrelated seams (agent task retry, tool parse retry,
guardrail retry) with a different payload at each, so its semantics were
ambiguous. Also casts the `post_step` payload back to `TaskOutput` in
`task.py` so the seam keeps mypy green.

lib/crewai/src/crewai/agent/core.py

@@ -656,22 +656,6 @@ class Agent(BaseAgent):
 
         return result
 
-    def _dispatch_retry_attempt(self, e: Exception, task: Task) -> None:
-        """Fire the ``retry_attempt`` interception point before re-executing a task."""
-        from crewai.hooks.contexts import RetryAttemptContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retry_ctx = RetryAttemptContext(
-            agent=self,
-            agent_role=getattr(self, "role", None),
-            task=task,
-            attempt=self._times_executed,
-            max_attempts=self.max_retry_limit,
-            error=e,
-            payload=e,
-        )
-        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-
     def _check_execution_error(self, e: Exception, task: Task) -> None:
         """Check if an execution error should be re-raised immediately.
 
@@ -725,7 +709,6 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
-        self._dispatch_retry_attempt(e, task)
         return self.execute_task(task, context, tools)
 
     async def _handle_execution_error_async(
@@ -747,7 +730,6 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
-        self._dispatch_retry_attempt(e, task)
         return await self.aexecute_task(task, context, tools)
 
     def message(self, content: str, **kwargs: Any) -> str:

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

@@ -2643,7 +2643,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             updated_params = pre_step_ctx.payload
             if isinstance(updated_params, dict):
                 positional = sorted(
-                    (k for k in updated_params if k.startswith("_") and k[1:].isdigit()),
+                    (
+                        k
+                        for k in updated_params
+                        if k.startswith("_") and k[1:].isdigit()
+                    ),
                     key=lambda k: int(k[1:]),
                 )
                 args = tuple(updated_params[k] for k in positional)
@@ -2886,19 +2890,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     if isinstance(router_result, enum.Enum)
                     else router_result
                 )
-
-                from crewai.hooks.contexts import RouterDecisionContext
-                from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-                router_ctx = RouterDecisionContext(
-                    flow=self,
-                    router_name=str(router_name),
-                    route=router_result,
-                    payload=router_result,
-                )
-                dispatch(InterceptionPoint.ROUTER_DECISION, router_ctx)
-                router_result = router_ctx.payload
-
                 router_result_str = str(router_result)
                 router_result_event = FlowMethodName(router_result_str)
                 router_results.append(router_result_event)
@@ -2927,19 +2918,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     current_trigger, router_only=False
                 )
                 if listeners_triggered:
-                    from crewai.hooks.contexts import FlowTransitionContext
-                    from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-                    transition_ctx = FlowTransitionContext(
-                        flow=self,
-                        from_method=str(trigger_method),
-                        to_methods=[str(name) for name in listeners_triggered],
-                        trigger=str(current_trigger),
-                        payload=listeners_triggered,
-                    )
-                    dispatch(InterceptionPoint.FLOW_TRANSITION, transition_ctx)
-                    listeners_triggered = transition_ctx.payload
-
                     listener_result = router_result_payloads.get(
                         str(current_trigger), result
                     )

lib/crewai/src/crewai/hooks/contexts.py

@@ -78,23 +78,6 @@ class ToolSelectionContext(InterceptionContext):
     tools: list[Any] = field(default_factory=list)
 
 
-@dataclass
-class PreDelegationContext(InterceptionContext):
-    """``pre_delegation``: an agent is about to delegate work. ``payload`` = delegation input."""
-
-    coworker: str | None = None
-    delegate_to: Any = None
-
-
-@dataclass
-class RetryAttemptContext(InterceptionContext):
-    """``retry_attempt``: an operation is about to be retried."""
-
-    attempt: int = 0
-    max_attempts: int | None = None
-    error: Any = None
-
-
 @dataclass
 class MemoryWriteContext(InterceptionContext):
     """``memory_write``: a value is about to be written to memory. ``payload`` = value."""
@@ -132,20 +115,3 @@ class MCPConnectContext(InterceptionContext):
 
     server_name: str | None = None
     server_params: Any = None
-
-
-@dataclass
-class FlowTransitionContext(InterceptionContext):
-    """``flow_transition``: a flow is moving to triggered methods."""
-
-    from_method: str | None = None
-    to_methods: list[str] = field(default_factory=list)
-    trigger: str | None = None
-
-
-@dataclass
-class RouterDecisionContext(InterceptionContext):
-    """``router_decision``: a flow router is choosing a route. ``payload`` = route label."""
-
-    router_name: str | None = None
-    route: Any = None

lib/crewai/src/crewai/hooks/dispatch.py

@@ -60,8 +60,6 @@ class InterceptionPoint(str, Enum):
     PRE_STEP = "pre_step"
     POST_STEP = "post_step"
     TOOL_SELECTION = "tool_selection"
-    PRE_DELEGATION = "pre_delegation"
-    RETRY_ATTEMPT = "retry_attempt"
 
     # Subsystem points
     MEMORY_WRITE = "memory_write"
@@ -70,10 +68,6 @@ class InterceptionPoint(str, Enum):
     PRE_CODE_EXECUTION = "pre_code_execution"
     MCP_CONNECT = "mcp_connect"
 
-    # Flow-specific points
-    FLOW_TRANSITION = "flow_transition"
-    ROUTER_DECISION = "router_decision"
-
 
 class HookAborted(Exception):  # noqa: N818 - public contract name from OSS-86
     """Raised by a hook (or a legacy adapter) to abort the intercepted operation.

lib/crewai/src/crewai/task.py

@@ -743,7 +743,7 @@ class Task(BaseModel):
                 payload=task_output,
             )
             dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
-            task_output = post_step_ctx.payload
+            task_output = cast(TaskOutput, post_step_ctx.payload)
 
             self.output = task_output
             self.end_time = datetime.datetime.now()
@@ -895,7 +895,7 @@ class Task(BaseModel):
                 payload=task_output,
             )
             dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
-            task_output = post_step_ctx.payload
+            task_output = cast(TaskOutput, post_step_ctx.payload)
 
             self.output = task_output
             self.end_time = datetime.datetime.now()
@@ -938,32 +938,6 @@ class Task(BaseModel):
             clear_task_files(self.id)
             reset_current_task_id(task_id_token)
 
-    def _dispatch_guardrail_retry_attempt(
-        self,
-        agent: BaseAgent | None,
-        context: str | None,
-        attempt: int,
-        error: Any,
-    ) -> str | None:
-        """Fire ``retry_attempt`` before re-executing a task after a guardrail failure.
-
-        Returns the (possibly hook-modified) retry context.
-        """
-        from crewai.hooks.contexts import RetryAttemptContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retry_ctx = RetryAttemptContext(
-            agent=agent,
-            agent_role=getattr(agent, "role", None),
-            task=self,
-            attempt=attempt,
-            max_attempts=self.guardrail_max_retries,
-            error=error,
-            payload=context,
-        )
-        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-        return retry_ctx.payload
-
     def _post_agent_execution(self, agent: BaseAgent) -> None:
         pass
 
@@ -1396,14 +1370,6 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-
-            context = self._dispatch_guardrail_retry_attempt(
-                agent=agent,
-                context=context,
-                attempt=current_retry_count,
-                error=guardrail_result.error,
-            )
-
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -1513,14 +1479,6 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-
-            context = self._dispatch_guardrail_retry_attempt(
-                agent=agent,
-                context=context,
-                attempt=current_retry_count,
-                error=guardrail_result.error,
-            )
-
             result = await agent.aexecute_task(
                 task=self,
                 context=context,

lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py

@@ -108,22 +108,6 @@ class BaseAgentTool(BaseTool):
             )
 
         selected_agent = agent[0]
-
-        from crewai.hooks.contexts import PreDelegationContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        # Dispatched outside the try/except below so a HookAborted propagates
-        # instead of being swallowed into a tool-error string.
-        delegation_ctx = PreDelegationContext(
-            agent=selected_agent,
-            agent_role=getattr(selected_agent, "role", None),
-            coworker=sanitized_name,
-            delegate_to=selected_agent,
-            payload=task,
-        )
-        dispatch(InterceptionPoint.PRE_DELEGATION, delegation_ctx)
-        task = delegation_ctx.payload
-
         try:
             task_with_assigned_agent = Task(
                 description=task,

lib/crewai/src/crewai/tools/tool_usage.py

@@ -879,22 +879,6 @@ class ToolUsage:
                 return ToolUsageError(
                     f"{I18N_DEFAULT.errors('tool_usage_error').format(error=e)}\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
                 )
-
-            from crewai.hooks.contexts import RetryAttemptContext
-            from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-            retry_ctx = RetryAttemptContext(
-                agent=self.agent,
-                agent_role=getattr(self.agent, "role", None),
-                task=self.task,
-                attempt=self._run_attempts,
-                max_attempts=self._max_parsing_attempts,
-                error=e,
-                payload=tool_string,
-            )
-            dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-            tool_string = retry_ctx.payload
-
             return self._tool_calling(tool_string)
 
     def _validate_tool_input(self, tool_input: str | None) -> dict[str, Any]:

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -7,7 +7,7 @@ sees a well-shaped payload, an in-place/returned modification is honored, and a
 
 from __future__ import annotations
 
-from crewai.flow.flow import Flow, listen, router, start
+from crewai.flow.flow import Flow, listen, start
 from crewai.hooks.dispatch import (
     HookAborted,
     InterceptionPoint,
@@ -117,61 +117,3 @@ class TestFlowStepPoints:
             return None
 
         assert _SimpleFlow().kickoff() == "rewritten"
-
-
-class _RouterFlow(Flow):
-    @start()
-    def begin(self):
-        return "begin"
-
-    @router(begin)
-    def route(self):
-        return "go_left"
-
-    @listen("go_left")
-    def left(self):
-        return "left"
-
-    @listen("go_right")
-    def right(self):
-        return "right"
-
-
-class TestFlowTransitionAndRouter:
-    """flow_transition and router_decision on a routed flow."""
-
-    def test_transition_payload_carries_from_and_to(self):
-        seen: list[tuple[str | None, list[str]]] = []
-
-        @on(InterceptionPoint.FLOW_TRANSITION)
-        def capture(ctx):
-            seen.append((ctx.from_method, list(ctx.to_methods)))
-
-        _RouterFlow().kickoff()
-
-        assert any(to == ["left"] for _from, to in seen)
-
-    def test_router_decision_fires_with_route(self):
-        routes: list[object] = []
-
-        @on(InterceptionPoint.ROUTER_DECISION)
-        def capture(ctx):
-            routes.append(ctx.route)
-
-        _RouterFlow().kickoff()
-        assert "go_left" in routes
-
-    def test_router_decision_can_reroute(self):
-        @on(InterceptionPoint.ROUTER_DECISION)
-        def reroute(ctx):
-            return "go_right"
-
-        landed: list[str] = []
-
-        @on(InterceptionPoint.PRE_STEP)
-        def track(ctx):
-            landed.append(ctx.step_name)
-
-        _RouterFlow().kickoff()
-        assert "right" in landed
-        assert "left" not in landed