fix: dedupe tool_selection and honor payload on catalog seams

Addresses review findings on the remaining interception points. `TOOL_SELECTION`
is no longer dispatched in `Crew._prepare_tools`; the single dispatch in
`Agent.create_agent_executor` covers both crew and standalone runs, so hooks no
longer fire twice (and additive edits no longer duplicate tools). `PRE_DELEGATION`
now dispatches outside the delegation `try/except`, so a `HookAborted` propagates
instead of being swallowed into a tool-error string. The `PRE_STEP` (flow),
`PRE_CODE_EXECUTION`, and `MCP_CONNECT` seams now apply the returned payload:
flow step params are rebound onto the call, script code is recompiled from the
edited source, and the MCP transport is replaced before connecting.

docs/docs.json

@@ -375,7 +375,8 @@
                       "edge/en/learn/using-annotations",
                       "edge/en/learn/execution-hooks",
                       "edge/en/learn/llm-hooks",
-                      "edge/en/learn/tool-hooks"
+                      "edge/en/learn/tool-hooks",
+                      "edge/en/learn/interception-hooks"
                     ]
                   },
                   {

docs/edge/en/concepts/llms.mdx

@@ -144,6 +144,18 @@ In this section, you'll find detailed examples that help you select, configure,
     )
     ```
 
+    **Custom OpenAI-Compatible Endpoint:**
+    ```python Code
+    from crewai import LLM
+
+    llm = LLM(
+        model="anthropic/claude-sonnet-4-6",
+        custom_openai=True,
+        base_url="https://your-gateway.example.com/v1",
+        api_key="your-gateway-api-key",
+    )
+    ```
+
     **Advanced Configuration:**
     ```python Code
     from crewai import LLM

docs/edge/en/learn/execution-hooks.mdx

@@ -42,6 +42,14 @@ Control and monitor tool execution:
 
 [View Tool Hooks Documentation →](/learn/tool-hooks)
 
+<Note>
+LLM and tool hooks are two points in a larger catalog. See
+[Interception Hooks](/learn/interception-hooks) for every framework-native
+interception point (execution boundaries, steps, memory, knowledge, flow
+transitions, and more) and the shared payload-in/payload-out contract they all
+follow.
+</Note>
+
 ## Hook Registration Methods
 
 ### 1. Decorator-Based Hooks (Recommended)

docs/edge/en/learn/interception-hooks.mdx

@@ -0,0 +1,168 @@
+---
+title: Interception Hooks
+description: The full catalog of framework-native interception points and the payload-in/payload-out contract every hook follows
+mode: "wide"
+---
+
+Interception hooks give you a single, uniform way to observe and modify CrewAI's
+runtime at well-defined points — from the moment an execution starts, through
+every model call, tool call, memory read, and flow transition, down to the final
+output. All points share one contract and one registration API.
+
+The four LLM/tool hooks documented in [LLM Hooks](/learn/llm-hooks) and
+[Tool Hooks](/learn/tool-hooks) are the same mechanism. Their existing
+decorators (`@before_llm_call`, `@before_tool_call`, ...) and `return False`
+semantics keep working unchanged; interception hooks generalize the same engine
+to the rest of the framework.
+
+## The contract
+
+Every hook is a **synchronous** callable that receives a single typed context:
+
+```python
+from crewai.hooks import on, HookAborted, InterceptionPoint
+
+@on(InterceptionPoint.INPUT)
+def add_defaults(ctx):
+    # 1. Observe: read anything off the context.
+    # 2. Mutate in place: change ctx.payload or nested fields directly.
+    ctx.payload.setdefault("locale", "en-US")
+    # 3. Or replace: return a new value to swap ctx.payload.
+    # 4. Or abort: raise HookAborted(reason, source) to stop the operation.
+    return None
+```
+
+A hook may do any of four things:
+
+| Action | How | Effect |
+|--------|-----|--------|
+| **Proceed** | `return None` (or nothing) | Operation continues unchanged |
+| **Mutate** | Change `ctx.payload` / fields in place | Change is visible downstream |
+| **Replace** | `return new_payload` | A non-`None` return replaces `ctx.payload` |
+| **Abort** | `raise HookAborted(reason, source)` | Operation is stopped; the reason propagates |
+
+### Composition, ordering, and fail-open
+
+- Multiple hooks on the same point run in **registration order**, global hooks
+  first, then execution-scoped hooks.
+- The (possibly mutated) payload flows from one hook to the next.
+- `HookAborted` **propagates by design** and stops the chain.
+- Any *other* exception raised by a hook is **swallowed** (fail-open) so a single
+  buggy hook can't crash a run — the same protection the legacy hooks provide.
+- When no hook is registered for a point, dispatch is a single dict lookup
+  (no-op fast path), so unused points cost effectively nothing.
+
+## Registering hooks
+
+Use the `@on` decorator for global hooks. It mirrors the legacy decorators'
+ergonomics, including `agents=` / `tools=` filters:
+
+```python
+from crewai.hooks import on, InterceptionPoint, HookAborted
+
+@on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
+def guard_deletes(ctx):
+    raise HookAborted(reason="file deletion is not allowed", source="policy")
+```
+
+Applied to a method inside a `@CrewBase` class, `@on` registers a crew-scoped
+hook (active only while that crew runs), matching the existing crew-scoped hook
+behavior.
+
+## Interception point catalog
+
+`payload` is the value a hook may mutate or replace at each point.
+
+### Execution boundaries
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `EXECUTION_START` | A crew or flow is about to begin | inputs `dict` |
+| `INPUT` | Resolved inputs for the execution | inputs `dict` |
+| `OUTPUT` | Final result is ready | the output object |
+| `EXECUTION_END` | A crew or flow has finished | the output object |
+
+### Model & tool boundaries (legacy-compatible)
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `PRE_MODEL_CALL` | Before an LLM call | `LLMCallHookContext` |
+| `POST_MODEL_CALL` | After an LLM call | response |
+| `PRE_TOOL_CALL` | Before a tool runs | `ToolCallHookContext` |
+| `POST_TOOL_CALL` | After a tool runs | tool result |
+
+### Step & agent points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `PRE_STEP` | Before a task or flow-method step | step input |
+| `POST_STEP` | After a task or flow-method step | step output |
+| `TOOL_SELECTION` | Tools are offered to an agent | list of tools |
+| `PRE_DELEGATION` | An agent is about to delegate | delegation input |
+| `RETRY_ATTEMPT` | An operation is about to be retried | retry input |
+
+`PRE_STEP` / `POST_STEP` carry `ctx.kind` (`"task"` or `"flow_method"`) and
+`ctx.step_name`.
+
+### Subsystem points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `MEMORY_WRITE` | A value is about to be stored in memory | value |
+| `MEMORY_READ` | A memory query is issued | query |
+| `KNOWLEDGE_RETRIEVAL` | A knowledge query is issued | query |
+| `PRE_CODE_EXECUTION` | Code is about to run (flow `ScriptAction`) | code string |
+| `MCP_CONNECT` | An MCP client is about to connect | connection params |
+| `FILE_ACCESS` | Reserved — no live seam yet | path |
+| `ARTIFACT_OUTPUT` | Reserved — no live seam yet | artifact |
+
+`FILE_ACCESS` and `ARTIFACT_OUTPUT` are part of the frozen catalog but have no
+consumer seam yet: registering against them is accepted and simply never fires,
+the same as any point with no hooks.
+
+### Flow-specific points
+
+| Point | When | `payload` |
+|-------|------|-----------|
+| `FLOW_TRANSITION` | A flow moves to its triggered methods | list of target methods |
+| `ROUTER_DECISION` | A flow router picks a route | route label |
+
+## Aborting an operation
+
+`HookAborted` carries a `reason` and an optional `source`. The `source` defaults
+to the aborting hook when omitted, which is useful for telemetry and failure
+messages:
+
+```python
+@on(InterceptionPoint.EXECUTION_START)
+def enforce_policy(ctx):
+    if not ctx.payload.get("authorized"):
+        raise HookAborted(reason="unauthorized execution", source="access-control")
+```
+
+## Telemetry
+
+Whenever a point actually dispatches to at least one hook, CrewAI emits a
+`HookDispatchedEvent` on the event bus with the point, the outcome
+(`proceeded` / `modified` / `aborted`), the hook count, the duration, and — for
+aborts — the reason and source. The no-op fast path emits nothing.
+
+## Managing hooks in tests
+
+```python
+import pytest
+from crewai.hooks import clear_all_hooks
+
+@pytest.fixture(autouse=True)
+def reset_hooks():
+    clear_all_hooks()
+    yield
+    clear_all_hooks()
+```
+
+## Related documentation
+
+- [Execution Hooks Overview →](/learn/execution-hooks)
+- [LLM Call Hooks →](/learn/llm-hooks)
+- [Tool Call Hooks →](/learn/tool-hooks)
+- [Before and After Kickoff Hooks →](/learn/before-and-after-kickoff-hooks)

docs/edge/en/learn/litellm-removal-guide.mdx

@@ -240,14 +240,15 @@ from crewai import LLM
 
 # After (OpenAI-compatible mode, no LiteLLM needed):
 llm = LLM(
-    model="openai/llama3",
+    model="llama3",
+    custom_openai=True,
     base_url="http://localhost:11434/v1",
     api_key="ollama"  # Ollama doesn't require a real API key
 )
 ```
 
 <Tip>
-  Many local inference servers (Ollama, vLLM, LM Studio, llama.cpp) expose an OpenAI-compatible API. You can use the `openai/` prefix with a custom `base_url` to connect to any of them natively.
+  Many local inference servers (Ollama, vLLM, LM Studio, llama.cpp) expose an OpenAI-compatible API. You can use `custom_openai=True` with a custom `base_url` to connect to any of them natively while keeping the model ID your gateway expects.
 </Tip>
 
 ### Step 4: Update your YAML configs
@@ -295,6 +296,92 @@ crewai run
 uv run pytest
 ```
 
+## Custom OpenAI-Compatible Endpoints
+
+Many providers and local servers (Ollama, vLLM, LM Studio, llama.cpp, LiteLLM proxies, and hosted gateways) expose an **OpenAI-compatible** API. Instead of routing these through LiteLLM, you can talk to them directly with CrewAI's native OpenAI integration by setting `custom_openai=True`.
+
+This is the recommended replacement for any LiteLLM provider that offers an OpenAI-compatible endpoint.
+
+### How it works
+
+- `custom_openai=True` forces CrewAI to use the native OpenAI SDK, regardless of the model name.
+- The model ID is passed to the endpoint without validation against OpenAI's known-model list. This lets you use arbitrary model IDs your gateway expects (for example, `anthropic/claude-sonnet-4-6` served behind an OpenAI-compatible proxy). An optional leading `openai/` routing prefix is stripped.
+- A base URL is **required**. CrewAI resolves it, in order, from:
+  1. `base_url=...`
+  2. `api_base=...`
+  3. `OPENAI_BASE_URL` environment variable
+  4. `OPENAI_API_BASE` environment variable (legacy)
+
+  If none are set, CrewAI raises a `ValueError` so misconfiguration fails fast instead of silently hitting `api.openai.com`.
+
+```python
+from crewai import LLM
+
+llm = LLM(
+    model="anthropic/claude-sonnet-4-6",  # passed through as-is
+    custom_openai=True,
+    base_url="https://your-gateway.example/v1",
+    api_key="your-key",
+)
+```
+
+### Connect to common servers
+
+<Tabs>
+  <Tab title="Ollama">
+    ```python
+    from crewai import LLM
+
+    llm = LLM(
+        model="llama3.2:latest",
+        custom_openai=True,
+        base_url="http://localhost:11434/v1",
+        api_key="ollama",  # Ollama ignores it, but the client requires a value
+    )
+    ```
+  </Tab>
+  <Tab title="vLLM">
+    ```python
+    from crewai import LLM
+
+    llm = LLM(
+        model="meta-llama/Meta-Llama-3.1-8B-Instruct",
+        custom_openai=True,
+        base_url="http://localhost:8000/v1",
+        api_key="not-needed",
+    )
+    ```
+  </Tab>
+  <Tab title="LM Studio">
+    ```python
+    from crewai import LLM
+
+    llm = LLM(
+        model="your-loaded-model",
+        custom_openai=True,
+        base_url="http://localhost:1234/v1",
+        api_key="lm-studio",
+    )
+    ```
+  </Tab>
+  <Tab title="Env vars">
+    ```bash
+    export OPENAI_BASE_URL="https://your-gateway.example/v1"
+    export OPENAI_API_KEY="your-key"
+    ```
+    ```python
+    from crewai import LLM
+
+    # base_url is picked up from OPENAI_BASE_URL / OPENAI_API_BASE
+    llm = LLM(model="anthropic/claude-sonnet-4-6", custom_openai=True)
+    ```
+  </Tab>
+</Tabs>
+
+<Tip>
+  If you use the `openai/` prefix with a model that isn't a known OpenAI model and pass `base_url` or `api_base` directly, CrewAI automatically treats it as a custom OpenAI-compatible endpoint. Environment variables alone do not enable automatic routing for unknown models; set `custom_openai=True` when configuring the endpoint through `OPENAI_BASE_URL` or `OPENAI_API_BASE`.
+</Tip>
+
 ## Quick Reference: Model String Mapping
 
 Here are common migration paths from LiteLLM-dependent providers to native ones:
@@ -321,7 +408,8 @@ llm = LLM(model="anthropic/claude-sonnet-4-20250514")  # High quality
 # Ollama → OpenAI-compatible (keep using local models)
 # llm = LLM(model="ollama/llama3")
 llm = LLM(
-    model="openai/llama3",
+    model="llama3",
+    custom_openai=True,
     base_url="http://localhost:11434/v1",
     api_key="ollama"
 )
@@ -349,6 +437,9 @@ llm = LLM(
   <Accordion title="What about environment variables like OPENAI_API_KEY?">
     Native providers use the same environment variables you're already familiar with. No changes needed for `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMINI_API_KEY`, etc.
   </Accordion>
+  <Accordion title="How do I connect to Groq, Together AI, or other OpenAI-compatible providers without LiteLLM?">
+    Most of these providers expose an OpenAI-compatible API. Use `custom_openai=True` with their base URL and API key — see [Custom OpenAI-Compatible Endpoints](#custom-openai-compatible-endpoints). For example, Groq: `LLM(model="llama-3.1-70b-versatile", custom_openai=True, base_url="https://api.groq.com/openai/v1", api_key="...")`. The model ID is passed through untouched, so use whatever ID the provider expects.
+  </Accordion>
 </AccordionGroup>
 
 ## Related Resources

lib/cli/src/crewai_cli/crew_run_tui.py

@@ -568,12 +568,32 @@ FooterKey .footer-key--key {
         self._default_inputs: dict[str, Any] | None = None
         self._crew_result: Any = None
         self._crew_json_path: Any = None
+        # Declarative-flow execution state. A flow renders per-method "STEPS"
+        # (built from flow method events) instead of the crew task list.
+        self._flow_inputs: dict[str, Any] | None = None
+        self._flow_method_types: dict[str, str] = {}
+        self._flow_steps: list[dict[str, Any]] = []
+        self._current_method: str | None = None
         self._elapsed_frozen: float | None = None
         self._want_deploy: bool = False
         self._trace_url: str | None = None
         self._consent_screen: TraceConsentScreen | None = None
         self._telemetry: Telemetry | None = None
 
+    @property
+    def _is_flow_run(self) -> bool:
+        """True for a non-conversational declarative flow (the STEPS view).
+
+        Gates every flow-specific rendering branch so crew and conversational
+        paths stay byte-identical.
+        """
+        return self._flow is not None and not self._is_conversational
+
+    @property
+    def _run_noun(self) -> str:
+        """User-facing noun for the run — 'flow' for a declarative flow, else 'crew'."""
+        return "flow" if self._is_flow_run else "crew"
+
     # ── Layout ──────────────────────────────────────────────
 
     def compose(self) -> ComposeResult:
@@ -602,6 +622,8 @@ FooterKey .footer-key--key {
         self._tick_timer = self.set_interval(1 / 8, self._tick)
         if self._is_conversational and self._flow:
             self._start_conversational_session()
+        elif self._flow:
+            self._run_flow_worker()
         elif self._crew:
             self._run_crew_worker()
         elif self._crew_json_path:
@@ -681,6 +703,49 @@ FooterKey .footer-key--key {
         except Exception as e:
             self.call_from_thread(self._on_crew_failed, str(e))
 
+    @work(thread=True, exclusive=True, group="flow")
+    def _run_flow_worker(self) -> None:
+        from crewai.events.listeners.tracing.utils import (
+            set_suppress_tracing_messages,
+            set_tui_mode,
+        )
+
+        set_tui_mode(True)
+        set_suppress_tracing_messages(True)
+        try:
+            # A declarative flow returns either a CrewOutput (has ``.raw``) or a
+            # bare value (str/dict/pydantic); _stringify_output handles both.
+            result = self._flow.kickoff(inputs=self._flow_inputs)
+            output = self._stringify_output(result)
+            with self._lock:
+                self._crew_result = result
+            self.call_from_thread(self._on_crew_done, output)
+        except Exception as e:
+            self.call_from_thread(self._on_crew_failed, str(e))
+
+    def _set_flow_step_status(self, name: str, status: str) -> None:
+        """Update a flow method step's status. Caller must hold ``self._lock``."""
+        for step in self._flow_steps:
+            if step["name"] == name:
+                step["status"] = status
+                return
+
+    def _clear_current_method(self, finished_name: str) -> None:
+        """Drop the header's active method once it ends. Caller holds the lock.
+
+        Falls back to another still-active step (methods can overlap) so the
+        header never keeps spinning a method the STEPS list already shows as
+        done or failed.
+        """
+        if self._current_method != finished_name:
+            return
+        self._current_method = next(
+            (s["name"] for s in self._flow_steps if s["status"] == "active"), None
+        )
+        # The active method changed; drop its agent so the header doesn't show a
+        # stale agent until the next method's agent event arrives.
+        self._current_agent = ""
+
     def _on_crew_done(self, output: str | None) -> None:
         with self._lock:
             self._status = "completed"
@@ -694,13 +759,18 @@ FooterKey .footer-key--key {
             for k in self._task_statuses:
                 if self._task_statuses[k] == "active":
                     self._task_statuses[k] = "done"
+            for step in self._flow_steps:
+                if step["status"] == "active":
+                    step["status"] = "done"
             now = time.time()
             for entry in self._log_entries:
                 if entry["status"] == "running":
                     if entry["tool_name"] == "memory_save":
                         continue
                     entry["status"] = "timeout"
-                    entry["error"] = "No result received before crew completed"
+                    entry["error"] = (
+                        f"No result received before {self._run_noun} completed"
+                    )
                     entry["duration"] = now - entry["start_time"]
         try:
             from crewai.events.listeners.tracing.trace_listener import (
@@ -739,13 +809,18 @@ FooterKey .footer-key--key {
             self._is_streaming = False
             self._current_step = None
             self._elapsed_frozen = time.time() - self._start_time
+            for step in self._flow_steps:
+                if step["status"] == "active":
+                    step["status"] = "failed"
             now = time.time()
             for entry in self._log_entries:
                 if entry["status"] == "running":
                     if entry["tool_name"] == "memory_save":
                         continue
                     entry["status"] = "error"
-                    entry["error"] = "No result received before crew failed"
+                    entry["error"] = (
+                        f"No result received before {self._run_noun} failed"
+                    )
                     entry["duration"] = now - entry["start_time"]
         self._tick()
         self.call_later(self._focus_activity_log)
@@ -1156,6 +1231,45 @@ FooterKey .footer-key--key {
             widget.update(t)
             return
 
+        if self._is_flow_run:
+            t.append("  STEPS\n", style=f"bold {_C_PRIMARY}")
+            t.append("\n")
+            if not self._flow_steps:
+                t.append("  ○ waiting…\n", style=_C_DIM)
+            for step in self._flow_steps:
+                name = step["name"]
+                max_name = sidebar_width - 6
+                if len(name) > max_name:
+                    name = name[: max_name - 1] + "…"
+                status = step.get("status", "pending")
+                if status == "done":
+                    t.append("  ✔ ", style=_C_GREEN)
+                    t.append(name, style=_C_DIM)
+                elif status == "active":
+                    t.append(f"  {self._spinner()} ", style=_C_PRIMARY)
+                    t.append(name, style=f"bold {_C_TEXT}")
+                elif status == "failed":
+                    t.append("  ✘ ", style=_C_RED)
+                    t.append(name, style=_C_RED)
+                elif status == "paused":
+                    t.append("  ⏸ ", style=_C_TEAL)
+                    t.append(name, style=_C_TEAL)
+                else:
+                    t.append("  ○ ", style=_C_DIM)
+                    t.append(name, style=_C_DIM)
+                if step.get("call_type"):
+                    t.append(f"  ({step['call_type']})", style=_C_DIM)
+                t.append("\n")
+
+            t.append("\n")
+            t.append("  TOKENS\n", style=f"bold {_C_PRIMARY}")
+            t.append("\n")
+            out = self._output_tokens + self._live_out_tokens
+            t.append(f"  ↑ {self._input_tokens:,}\n", style=_C_DIM)
+            t.append(f"  ↓ {out:,}\n", style=_C_DIM)
+            widget.update(t)
+            return
+
         t.append("  TASKS\n", style=f"bold {_C_PRIMARY}")
         t.append("\n")
 
@@ -1225,6 +1339,55 @@ FooterKey .footer-key--key {
             widget.update(t)
             return
 
+        if self._is_flow_run:
+            if self._status == "completed":
+                elapsed = self._elapsed_frozen or (time.time() - self._start_time)
+                t.append("✔ ", style=f"bold {_C_GREEN}")
+                t.append("Flow complete", style=f"bold {_C_GREEN}")
+                t.append(f"  {elapsed:.1f}s", style=_C_DIM)
+                out = self._output_tokens + self._live_out_tokens
+                parts = []
+                if self._input_tokens:
+                    parts.append(f"↑{self._input_tokens:,}")
+                if out:
+                    parts.append(f"↓{out:,}")
+                if parts:
+                    t.append(f"  {' '.join(parts)} tokens", style=_C_DIM)
+            elif self._status == "failed":
+                t.append("✘ ", style=f"bold {_C_RED}")
+                t.append("Failed", style=f"bold {_C_RED}")
+                if self._error:
+                    t.append(f"\n{self._error[:120]}", style=_C_RED)
+            elif self._current_method:
+                paused = any(
+                    s["name"] == self._current_method and s["status"] == "paused"
+                    for s in self._flow_steps
+                )
+                if paused:
+                    t.append("⏸ ", style=_C_TEAL)
+                    t.append(self._current_method, style=f"bold {_C_TEAL}")
+                else:
+                    t.append(f"{self._spinner()} ", style=_C_PRIMARY)
+                    t.append(self._current_method, style=f"bold {_C_PRIMARY}")
+                call_type = self._flow_method_types.get(self._current_method)
+                if call_type:
+                    t.append(f"  ({call_type})", style=_C_DIM)
+                if paused:
+                    t.append("  waiting for feedback", style=_C_DIM)
+                elif self._current_agent:
+                    t.append("\nAgent: ", style=_C_DIM)
+                    t.append(self._current_agent, style=f"bold {_C_TEXT}")
+            else:
+                t.append(f"{self._spinner()} ", style=_C_PRIMARY)
+                # "Working…" once a step has run (between/after methods);
+                # "Starting flow…" only before the first method.
+                t.append(
+                    "Working…" if self._flow_steps else "Starting flow…",
+                    style=_C_DIM,
+                )
+            widget.update(t)
+            return
+
         if self._status == "completed":
             elapsed = self._elapsed_frozen or (time.time() - self._start_time)
             t.append("✔ ", style=f"bold {_C_GREEN}")
@@ -1839,6 +2002,13 @@ FooterKey .footer-key--key {
     def _subscribe(self) -> None:
         from crewai.events.event_bus import crewai_event_bus
         from crewai.events.types.crew_events import CrewKickoffStartedEvent
+        from crewai.events.types.flow_events import (
+            FlowStartedEvent,
+            MethodExecutionFailedEvent,
+            MethodExecutionFinishedEvent,
+            MethodExecutionPausedEvent,
+            MethodExecutionStartedEvent,
+        )
         from crewai.events.types.llm_events import (
             LLMCallCompletedEvent,
             LLMCallStartedEvent,
@@ -1872,13 +2042,74 @@ FooterKey .footer-key--key {
         @crewai_event_bus.on(CrewKickoffStartedEvent)
         def on_crew_started(source: Any, event: CrewKickoffStartedEvent) -> None:
             with self._lock:
-                if event.crew_name:
+                # In flow mode the app is named for the flow; a nested crew's
+                # kickoff (a `call: crew` step) must not rename it.
+                if event.crew_name and not self._is_flow_run:
                     self._crew_name = event.crew_name
                     self.title = f"CrewAI — {event.crew_name}"
                 self._status = "working"
 
         self._register_handler(CrewKickoffStartedEvent, on_crew_started)
 
+        # ── Declarative-flow method events → STEPS panel ────────
+        @crewai_event_bus.on(FlowStartedEvent)
+        def on_flow_started(source: Any, event: FlowStartedEvent) -> None:
+            with self._lock:
+                self._status = "working"
+
+        self._register_handler(FlowStartedEvent, on_flow_started)
+
+        @crewai_event_bus.on(MethodExecutionStartedEvent)
+        def on_method_started(source: Any, event: MethodExecutionStartedEvent) -> None:
+            with self._lock:
+                name = event.method_name
+                self._current_method = name
+                # Agent is per-method; clear it so the header doesn't show the
+                # previous method's agent until a new agent event arrives.
+                self._current_agent = ""
+                for step in self._flow_steps:
+                    if step["name"] == name:
+                        step["status"] = "active"
+                        break
+                else:
+                    self._flow_steps.append(
+                        {
+                            "name": name,
+                            "call_type": self._flow_method_types.get(name),
+                            "status": "active",
+                        }
+                    )
+
+        self._register_handler(MethodExecutionStartedEvent, on_method_started)
+
+        @crewai_event_bus.on(MethodExecutionFinishedEvent)
+        def on_method_finished(
+            source: Any, event: MethodExecutionFinishedEvent
+        ) -> None:
+            with self._lock:
+                self._set_flow_step_status(event.method_name, "done")
+                self._clear_current_method(event.method_name)
+
+        self._register_handler(MethodExecutionFinishedEvent, on_method_finished)
+
+        @crewai_event_bus.on(MethodExecutionFailedEvent)
+        def on_method_failed(source: Any, event: MethodExecutionFailedEvent) -> None:
+            with self._lock:
+                self._set_flow_step_status(event.method_name, "failed")
+                self._clear_current_method(event.method_name)
+
+        self._register_handler(MethodExecutionFailedEvent, on_method_failed)
+
+        @crewai_event_bus.on(MethodExecutionPausedEvent)
+        def on_method_paused(source: Any, event: MethodExecutionPausedEvent) -> None:
+            # A @human_feedback method paused; flow status panels are suppressed
+            # in TUI mode, so surface the wait in STEPS/header instead of leaving
+            # a spinner. _current_method stays pointed at it.
+            with self._lock:
+                self._set_flow_step_status(event.method_name, "paused")
+
+        self._register_handler(MethodExecutionPausedEvent, on_method_paused)
+
         @crewai_event_bus.on(TaskStartedEvent)
         def on_task_started(source: Any, event: TaskStartedEvent) -> None:
             with self._lock:

lib/cli/src/crewai_cli/run_declarative_flow.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import json
+import logging
 from pathlib import Path
 import subprocess
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import click
 from crewai_core.project import ProjectDefinitionError, configured_project_definition
@@ -18,6 +19,13 @@ from crewai_cli.input_prompt import (
 from crewai_cli.utils import build_env_with_all_tool_credentials
 
 
+if TYPE_CHECKING:
+    from crewai.flow.flow import Flow
+
+
+logger = logging.getLogger(__name__)
+
+
 def run_declarative_flow_in_project_env(
     definition: str | Path, inputs: str | None = None
 ) -> None:
@@ -66,17 +74,182 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N
     flow = load_declarative_flow(definition)
     resolved_inputs = _resolve_flow_inputs(flow, provided)
 
+    # The TUI is the interactive default. Headless contexts run directly on the
+    # terminal: deploy/CREWAI_DMN, piped output, CI — anything without an
+    # interactive TTY. is_interactive() already folds in the CREWAI_DMN check.
+    # Human-feedback flows also run on the terminal: their methods collect input
+    # via the flow runtime's blocking input()/Rich prompts (and async feedback
+    # returns a pending marker rather than completing), neither of which the
+    # Textual TUI can handle correctly.
+    if is_interactive() and not _flow_uses_human_feedback(flow):
+        _run_declarative_flow_tui(flow, resolved_inputs or None)
+        return
+
     try:
         result = flow.kickoff(inputs=resolved_inputs or None)
     except Exception as exc:
         click.echo(
-            f"An error occurred while running the declarative flow: {exc}", err=True
+            f"An error occurred while running the declarative flow: {exc}",
+            err=True,
         )
         raise SystemExit(1) from exc
-
     click.echo(_format_result(result))
 
 
+def _run_declarative_flow_tui(
+    flow: Flow[Any], resolved_inputs: dict[str, Any] | None
+) -> Any:
+    """Run a declarative flow on the CrewAI TUI (the interactive default).
+
+    Mirrors the declarative-crew TUI contract (``run_crew._run_json_crew``):
+    a failed flow exits non-zero, a user quit ends the process so in-flight LLM
+    work stops, and choosing Deploy chains into the deploy command.
+    """
+    import os
+    import sys
+
+    from crewai.events.event_listener import EventListener
+
+    from crewai_cli.crew_run_tui import CrewRunApp
+
+    # The flow runtime (unlike a Crew constructor) doesn't create the event
+    # listener, and the TUI's trace/telemetry features depend on it.
+    EventListener()
+
+    # The STEPS panel and header are driven by flow method events. A flow may
+    # declare ``config.suppress_flow_events`` (a headless/production
+    # optimization) which would leave STEPS stuck on "waiting…" here — so force
+    # emission on for the interactive TUI run. The headless path never reaches
+    # this and keeps the flow's declared setting.
+    try:
+        flow.suppress_flow_events = False
+    except Exception:
+        logger.debug(
+            "Could not disable suppress_flow_events for the flow TUI", exc_info=True
+        )
+
+    app = CrewRunApp(crew_name=flow.name or type(flow).__name__)
+    app._flow = flow
+    app._flow_inputs = resolved_inputs
+    app._flow_method_types = _flow_method_types(flow)
+
+    app.run()
+
+    _print_flow_post_tui_summary(app)
+
+    if app._status == "failed":
+        raise SystemExit(1)
+
+    if app._status not in ("completed", "failed"):
+        # User quit mid-run. kickoff runs in a thread worker that cannot be
+        # force-cancelled, so end the process to stop in-flight LLM and tool
+        # work instead of letting it burn tokens in the background.
+        click.secho("\n  Run cancelled.", fg="yellow")
+        sys.stdout.flush()
+        os._exit(130)
+
+    if getattr(app, "_want_deploy", False):
+        from crewai_cli.run_crew import _chain_deploy
+
+        _chain_deploy()
+
+    return app._crew_result
+
+
+def _flow_uses_human_feedback(flow: Flow[Any]) -> bool:
+    """True if any declarative method declares ``@human_feedback``.
+
+    Such flows need the flow runtime's interactive stdin / Rich prompts, which
+    don't compose with Textual — so they run on the terminal, not the TUI.
+    """
+    try:
+        return any(
+            method.human_feedback is not None
+            for method in flow._definition.methods.values()
+        )
+    except Exception:
+        logger.debug("Could not inspect flow for human feedback", exc_info=True)
+        return False
+
+
+def _flow_method_types(flow: Flow[Any]) -> dict[str, str]:
+    """Map each declarative method name to its ``call`` type (crew/agent/…).
+
+    Best-effort: the STEPS panel shows this as a dim label. Method events don't
+    carry the call type, so it's read from the flow definition up front.
+    """
+    method_types: dict[str, str] = {}
+    try:
+        for name, method_definition in flow._definition.methods.items():
+            method_types[name] = method_definition.do.call
+    except Exception:
+        logger.debug("Could not derive flow method types", exc_info=True)
+    return method_types
+
+
+def _print_flow_post_tui_summary(app: Any) -> None:
+    """Print a compact result panel after the flow TUI exits."""
+    import time
+
+    from rich.console import Console
+    from rich.markdown import Markdown
+    from rich.padding import Padding
+    from rich.panel import Panel
+    from rich.text import Text
+
+    console = Console()
+    elapsed = (app._elapsed_frozen or (time.time() - app._start_time)) or 0.0
+
+    out_tokens = app._output_tokens + app._live_out_tokens
+    token_parts = []
+    if app._input_tokens:
+        token_parts.append(f"↑{app._input_tokens:,}")
+    if out_tokens:
+        token_parts.append(f"↓{out_tokens:,}")
+    token_str = "  ".join(token_parts)
+    if token_str:
+        token_str += " tokens"
+
+    crewai_red = "#FF5A50"
+    crewai_teal = "#1F7982"
+
+    if app._status == "completed":
+        summary = Text()
+        summary.append("  ✔ Flow complete", style=f"bold {crewai_teal}")
+        summary.append(f" in {elapsed:.1f}s", style="dim")
+        if token_str:
+            summary.append(f"  {token_str}", style="dim")
+        console.print(
+            Panel(
+                summary,
+                title=f" {app._crew_name} ",
+                title_align="left",
+                border_style=crewai_teal,
+                padding=(0, 1),
+            )
+        )
+        if app._final_output:
+            console.print()
+            console.print(Text("  Final Result", style=f"bold {crewai_teal}"))
+            console.print()
+            console.print(Padding(Markdown(app._final_output), (0, 2)))
+    elif app._status == "failed":
+        content = Text()
+        content.append("  ✘ Failed", style=f"bold {crewai_red}")
+        content.append(f" after {elapsed:.1f}s\n", style="dim")
+        if app._error:
+            content.append(f"\n  {app._error}\n", style=crewai_red)
+        console.print(
+            Panel(
+                content,
+                title=f" {app._crew_name} ",
+                title_align="left",
+                border_style=crewai_red,
+                padding=(0, 1),
+            )
+        )
+
+
 def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
     """Resolve kickoff inputs from the flow's state schema.
 

lib/cli/tests/test_crew_run_tui.py

@@ -6,6 +6,14 @@ from unittest.mock import Mock
 import pytest
 
 from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.crew_events import CrewKickoffStartedEvent
+from crewai.events.types.flow_events import (
+    FlowStartedEvent,
+    MethodExecutionFailedEvent,
+    MethodExecutionFinishedEvent,
+    MethodExecutionPausedEvent,
+    MethodExecutionStartedEvent,
+)
 from crewai.events.types.memory_events import (
     MemorySaveCompletedEvent,
     MemorySaveFailedEvent,
@@ -959,6 +967,31 @@ async def test_crew_done_does_not_mark_unfinished_tool_successful() -> None:
     assert app._plan_step_status == {1: "failed", 2: "done", 3: "done"}
 
 
+@pytest.mark.asyncio
+async def test_flow_done_uses_flow_wording_for_unfinished_tool() -> None:
+    # The shared completion handler reports "flow" (not "crew") in flow mode.
+    app = CrewRunApp(crew_name="Demo Flow")
+    app._flow = SimpleNamespace()
+
+    async with app.run_test(size=(100, 40)) as pilot:
+        app._log_entries = [
+            {
+                "tool_name": "search",
+                "status": "running",
+                "args": None,
+                "result": None,
+                "error": None,
+                "start_time": time.time() - 2,
+                "duration": None,
+                "task_idx": 1,
+            }
+        ]
+        app._on_crew_done("final output")
+        await pilot.pause()
+
+    assert app._log_entries[0]["error"] == "No result received before flow completed"
+
+
 @pytest.mark.asyncio
 async def test_crew_done_does_not_timeout_memory_save() -> None:
     app = _app_with_plan()
@@ -1481,3 +1514,210 @@ def test_overlapping_task_logs_keep_their_own_state() -> None:
         assert any(step.get("summary") == "thinking" for step in entry2["steps"])
     finally:
         app._unsubscribe()
+
+
+# ── Declarative-flow (non-conversational) TUI support ───────
+
+
+def test_is_flow_run_gating() -> None:
+    """The flow-render gate must be true only for a non-conversational flow."""
+    crew_app = CrewRunApp(total_tasks=1)
+    crew_app._crew = SimpleNamespace()
+    assert crew_app._is_flow_run is False
+
+    conv_app = CrewRunApp(conversational=True)
+    conv_app._flow = SimpleNamespace()
+    assert conv_app._is_flow_run is False
+
+    flow_app = CrewRunApp()
+    flow_app._flow = SimpleNamespace()
+    assert flow_app._is_flow_run is True
+
+
+def test_flow_method_events_build_steps() -> None:
+    app = CrewRunApp(crew_name="Demo")
+    app._flow = SimpleNamespace()
+    app._flow_method_types = {"research": "crew", "summarize": "agent"}
+    app._subscribe()
+    try:
+        _emit_event(FlowStartedEvent(flow_name="Demo"))
+        assert app._status == "working"
+
+        _emit_event(
+            MethodExecutionStartedEvent(
+                flow_name="Demo", method_name="research", state={}
+            )
+        )
+        assert app._flow_steps == [
+            {"name": "research", "call_type": "crew", "status": "active"}
+        ]
+        assert app._current_method == "research"
+
+        _emit_event(
+            MethodExecutionFinishedEvent(
+                flow_name="Demo", method_name="research", result="ok", state={}
+            )
+        )
+        _emit_event(
+            MethodExecutionStartedEvent(
+                flow_name="Demo", method_name="summarize", state={}
+            )
+        )
+        _emit_event(
+            MethodExecutionFailedEvent(
+                flow_name="Demo",
+                method_name="summarize",
+                error=RuntimeError("boom"),
+            )
+        )
+    finally:
+        app._unsubscribe()
+
+    assert app._flow_steps == [
+        {"name": "research", "call_type": "crew", "status": "done"},
+        {"name": "summarize", "call_type": "agent", "status": "failed"},
+    ]
+    # The header must not keep spinning a method that already ended.
+    assert app._current_method is None
+
+
+def test_current_method_clears_and_falls_back_across_overlap() -> None:
+    app = CrewRunApp(crew_name="Demo")
+    app._flow = SimpleNamespace()
+    app._subscribe()
+    try:
+        _emit_event(
+            MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={})
+        )
+        _emit_event(
+            MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={})
+        )
+        assert app._current_method == "b"
+
+        # 'a' finishes while 'b' is still active → header stays on 'b'.
+        _emit_event(
+            MethodExecutionFinishedEvent(
+                flow_name="Demo", method_name="a", result=None, state={}
+            )
+        )
+        assert app._current_method == "b"
+
+        # 'b' finishes → nothing active left → header clears.
+        _emit_event(
+            MethodExecutionFinishedEvent(
+                flow_name="Demo", method_name="b", result=None, state={}
+            )
+        )
+        assert app._current_method is None
+    finally:
+        app._unsubscribe()
+
+
+def test_flow_method_transitions_clear_current_agent() -> None:
+    app = CrewRunApp(crew_name="Demo")
+    app._flow = SimpleNamespace()
+    app._subscribe()
+    try:
+        _emit_event(
+            MethodExecutionStartedEvent(flow_name="Demo", method_name="a", state={})
+        )
+        app._current_agent = "Researcher"  # an agent ran during method 'a'
+
+        # Starting a new method clears the previous method's agent.
+        _emit_event(
+            MethodExecutionStartedEvent(flow_name="Demo", method_name="b", state={})
+        )
+        assert app._current_agent == ""
+
+        app._current_agent = "Writer"
+        # 'b' ending switches the active method ('a' still active) → agent clears.
+        _emit_event(
+            MethodExecutionFinishedEvent(
+                flow_name="Demo", method_name="b", result=None, state={}
+            )
+        )
+        assert app._current_agent == ""
+    finally:
+        app._unsubscribe()
+
+
+def test_crew_kickoff_does_not_rename_flow_run() -> None:
+    # A `call: crew` step must not relabel the flow with the nested crew's name.
+    app = CrewRunApp(crew_name="My Flow")
+    app._flow = SimpleNamespace()
+    app._subscribe()
+    try:
+        _emit_event(CrewKickoffStartedEvent(crew_name="Nested Crew", inputs=None))
+        assert app._crew_name == "My Flow"
+        assert app._status == "working"
+    finally:
+        app._unsubscribe()
+
+
+def test_crew_kickoff_renames_in_crew_mode() -> None:
+    # Regression: crew runs still adopt the crew name from the event.
+    app = CrewRunApp(crew_name="Crew")
+    app._crew = SimpleNamespace()
+    app._subscribe()
+    try:
+        _emit_event(CrewKickoffStartedEvent(crew_name="Real Crew", inputs=None))
+        assert app._crew_name == "Real Crew"
+    finally:
+        app._unsubscribe()
+
+
+def test_method_paused_marks_step_paused() -> None:
+    app = CrewRunApp(crew_name="Demo")
+    app._flow = SimpleNamespace()
+    app._subscribe()
+    try:
+        _emit_event(
+            MethodExecutionStartedEvent(flow_name="Demo", method_name="ask", state={})
+        )
+        _emit_event(
+            MethodExecutionPausedEvent(
+                flow_name="Demo",
+                method_name="ask",
+                state={},
+                flow_id="flow-1",
+                message="Need your input",
+            )
+        )
+        assert app._flow_steps == [
+            {"name": "ask", "call_type": None, "status": "paused"}
+        ]
+        assert app._current_method == "ask"
+    finally:
+        app._unsubscribe()
+
+
+@pytest.mark.asyncio
+async def test_declarative_flow_runs_on_tui() -> None:
+    """End-to-end: on_mount dispatches _run_flow_worker → flow.kickoff →
+    _on_crew_done, and any still-active step is swept to done on completion."""
+    kicked: dict[str, object] = {}
+
+    class FakeFlow:
+        name = "Demo Flow"
+
+        def kickoff(self, inputs=None):
+            kicked["inputs"] = inputs
+            return "flow result"
+
+    app = CrewRunApp(crew_name="Demo Flow")
+    app._flow = FakeFlow()
+    app._flow_inputs = {"topic": "AI"}
+    # A step left active (no Finished event) must be swept to done by _on_crew_done.
+    app._flow_steps = [{"name": "compute", "call_type": "expression", "status": "active"}]
+
+    async with app.run_test() as pilot:
+        for _ in range(100):
+            await pilot.pause(0.05)
+            if app._status == "completed":
+                break
+
+    assert kicked["inputs"] == {"topic": "AI"}
+    assert app._status == "completed"
+    assert app._final_output == "flow result"
+    assert app._crew_result == "flow result"
+    assert app._flow_steps[0]["status"] == "done"

lib/cli/tests/test_run_declarative_flow.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -9,6 +10,17 @@ import crewai_cli.input_prompt as input_prompt_module
 import crewai_cli.run_declarative_flow as run_declarative_flow_module
 
 
+@pytest.fixture(autouse=True)
+def _headless_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Default these tests to the headless/terminal path.
+
+    ``run_declarative_flow`` now launches the TUI when interactive, which can't
+    run under pytest; tests here assert the terminal/headless contract. Tests
+    that exercise TUI routing override ``is_dmn_mode_enabled`` explicitly.
+    """
+    monkeypatch.setenv("CREWAI_DMN", "true")
+
+
 FLOW_YAML = """\
 schema: crewai.flow/v1
 name: TestFlow
@@ -400,3 +412,202 @@ def test_id_restore_still_drops_unknown_keys(
     assert resolved == {"id": "run-123"}  # id kept, typo dropped
     assert "Ignoring unknown input 'prospect_emai'" in captured.err
     assert "Ignoring unknown input 'id'" not in captured.err
+
+
+# ── TUI vs terminal (headless/deploy) routing ──────────────────────
+
+
+def _install_fake_flow_app(monkeypatch, *, status, want_deploy=False):
+    """Replace CrewRunApp/EventListener/summary so _run_declarative_flow_tui is
+    driven by a controllable fake app."""
+
+    class FakeEventListener:
+        pass
+
+    class FakeApp:
+        def __init__(self, crew_name=""):
+            self._crew_name = crew_name
+            self._status = status
+            self._want_deploy = want_deploy
+            self._crew_result = "result"
+
+        def run(self):
+            pass
+
+    monkeypatch.setattr(
+        "crewai.events.event_listener.EventListener", FakeEventListener
+    )
+    monkeypatch.setattr("crewai_cli.crew_run_tui.CrewRunApp", FakeApp)
+    monkeypatch.setattr(
+        run_declarative_flow_module, "_print_flow_post_tui_summary", lambda app: None
+    )
+
+
+def test_run_declarative_flow_dmn_uses_terminal(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setenv("CREWAI_DMN", "true")
+    monkeypatch.setattr(
+        run_declarative_flow_module,
+        "_run_declarative_flow_tui",
+        lambda *a, **k: pytest.fail("DMN/headless mode must not launch the TUI"),
+    )
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"prospect_email":"a@b.com"}'
+    )
+
+    assert capsys.readouterr().out == "a@b.com\n"
+
+
+def test_run_declarative_flow_interactive_uses_tui(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True)
+    captured: dict[str, object] = {}
+    monkeypatch.setattr(
+        run_declarative_flow_module,
+        "_run_declarative_flow_tui",
+        lambda flow, resolved: captured.update(flow=flow, inputs=resolved),
+    )
+    path = _write(tmp_path, REQUIRED_FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(
+        str(path), '{"prospect_email":"a@b.com"}'
+    )
+
+    assert captured["inputs"] == {"prospect_email": "a@b.com"}
+    assert captured["flow"] is not None
+
+
+def test_run_declarative_flow_tui_failed_exits_nonzero(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_fake_flow_app(monkeypatch, status="failed")
+
+    with pytest.raises(SystemExit) as exc_info:
+        run_declarative_flow_module._run_declarative_flow_tui(
+            SimpleNamespace(name="Flow"), None
+        )
+
+    assert exc_info.value.code == 1
+
+
+def test_run_declarative_flow_tui_user_quit_exits_130(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_fake_flow_app(monkeypatch, status="chatting")
+    exit_calls: list[int] = []
+    monkeypatch.setattr(os, "_exit", lambda code: exit_calls.append(code))
+
+    run_declarative_flow_module._run_declarative_flow_tui(
+        SimpleNamespace(name="Flow"), None
+    )
+
+    assert exit_calls == [130]
+
+
+def test_run_declarative_flow_tui_chains_deploy(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_fake_flow_app(monkeypatch, status="completed", want_deploy=True)
+    deploy_calls: list[bool] = []
+    monkeypatch.setattr(
+        "crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True)
+    )
+
+    run_declarative_flow_module._run_declarative_flow_tui(
+        SimpleNamespace(name="Flow"), None
+    )
+
+    assert deploy_calls == [True]
+
+
+def test_run_declarative_flow_tui_no_deploy_when_not_requested(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _install_fake_flow_app(monkeypatch, status="completed", want_deploy=False)
+    deploy_calls: list[bool] = []
+    monkeypatch.setattr(
+        "crewai_cli.run_crew._chain_deploy", lambda: deploy_calls.append(True)
+    )
+
+    run_declarative_flow_module._run_declarative_flow_tui(
+        SimpleNamespace(name="Flow"), None
+    )
+
+    assert deploy_calls == []
+
+
+def test_run_declarative_flow_tui_enables_flow_events(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # The STEPS panel depends on flow method events; a flow that declared
+    # suppress_flow_events must have it forced off for the interactive TUI run.
+    _install_fake_flow_app(monkeypatch, status="completed")
+    flow = SimpleNamespace(name="Flow", suppress_flow_events=True)
+
+    run_declarative_flow_module._run_declarative_flow_tui(flow, None)
+
+    assert flow.suppress_flow_events is False
+
+
+def test_flow_uses_human_feedback_detection() -> None:
+    hf_flow = SimpleNamespace(
+        _definition=SimpleNamespace(
+            methods={
+                "ask": SimpleNamespace(human_feedback=SimpleNamespace(emit=None)),
+                "plain": SimpleNamespace(human_feedback=None),
+            }
+        )
+    )
+    assert run_declarative_flow_module._flow_uses_human_feedback(hf_flow) is True
+
+    no_hf = SimpleNamespace(
+        _definition=SimpleNamespace(
+            methods={"a": SimpleNamespace(human_feedback=None)}
+        )
+    )
+    assert run_declarative_flow_module._flow_uses_human_feedback(no_hf) is False
+    # No definition → False, no error.
+    assert run_declarative_flow_module._flow_uses_human_feedback(SimpleNamespace()) is False
+
+
+def test_human_feedback_flow_uses_terminal_even_when_interactive(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # A human-feedback flow must run on the terminal (blocking input / Rich
+    # prompts) even in an interactive session, never on the TUI.
+    monkeypatch.setattr(run_declarative_flow_module, "is_interactive", lambda: True)
+    monkeypatch.setattr(
+        run_declarative_flow_module, "_flow_uses_human_feedback", lambda flow: True
+    )
+    monkeypatch.setattr(
+        run_declarative_flow_module,
+        "_run_declarative_flow_tui",
+        lambda *a, **k: pytest.fail("human-feedback flow must run on the terminal"),
+    )
+    path = _write(tmp_path, FLOW_YAML)
+
+    run_declarative_flow_module.run_declarative_flow(str(path), '{"topic":"AI"}')
+
+    assert capsys.readouterr().out == "AI\n"
+
+
+def test_flow_method_types_from_definition() -> None:
+    flow = SimpleNamespace(
+        _definition=SimpleNamespace(
+            methods={
+                "fetch": SimpleNamespace(do=SimpleNamespace(call="expression")),
+                "research": SimpleNamespace(do=SimpleNamespace(call="crew")),
+            }
+        )
+    )
+
+    assert run_declarative_flow_module._flow_method_types(flow) == {
+        "fetch": "expression",
+        "research": "crew",
+    }
+    # No definition → empty map, no error.
+    assert run_declarative_flow_module._flow_method_types(SimpleNamespace()) == {}

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

@@ -86,6 +86,7 @@ from crewai.skills.models import Skill as SkillModel
 from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.callback import SerializableCallable
+from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.agent_utils import (
     get_tool_names,
     is_inside_event_loop,
@@ -400,10 +401,29 @@ class Agent(BaseAgent):
         return self.planning_config is not None or self.planning
 
     def _setup_agent_executor(self) -> None:
-        """Initialize the agent executor with a default cache handler."""
-        if not self.cache_handler:
-            self.cache_handler = CacheHandler()
-        self.set_cache_handler(self.cache_handler)
+        """Initialize the agent's tools handler and optional tool cache.
+
+        Tool-result caching is opt-in: a standalone agent gets a cache only
+        when it was constructed with an explicit ``cache=True`` or a
+        ``cache_handler``. Agents inside a crew additionally receive the
+        crew's shared handler when ``Crew(cache=True)``. Without an opt-in,
+        repeated tool calls with identical arguments always re-execute the
+        tool — the safe default for live-data and state-mutating tools.
+        """
+        # Recorded before any crew can offer its shared handler at kickoff,
+        # so copy() can distinguish a construction-time opt-in from runtime
+        # crew wiring (which must not turn copies into cachers).
+        self._constructor_cache_opt_in = bool(
+            self.cache
+            and (self.cache_handler is not None or "cache" in self.model_fields_set)
+        )
+        opted_in = self.cache_handler is not None or (
+            "cache" in self.model_fields_set and self.cache
+        )
+        if opted_in:
+            if not self.cache_handler:
+                self.cache_handler = CacheHandler()
+            self.set_cache_handler(self.cache_handler)
 
     def set_knowledge(self, crew_embedder: EmbedderConfig | None = None) -> None:
         """Initialize knowledge sources with the agent or crew embedder config."""
@@ -636,6 +656,22 @@ 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.
 
@@ -689,6 +725,7 @@ 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(
@@ -710,6 +747,7 @@ 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:
@@ -1034,6 +1072,21 @@ class Agent(BaseAgent):
             An instance of the CrewAgentExecutor class.
         """
         raw_tools: list[BaseTool] = tools or self.tools or []
+
+        from crewai.hooks.contexts import ToolSelectionContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        selection_ctx = ToolSelectionContext(
+            agent=self,
+            agent_role=getattr(self, "role", None),
+            task=task,
+            crew=self.crew,
+            tools=raw_tools,
+            payload=raw_tools,
+        )
+        dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx)
+        raw_tools = selection_ctx.payload
+
         parsed_tools = parse_tools(raw_tools)
 
         prompt, stop_words, rpm_limit_fn = self._build_execution_prompt(raw_tools)
@@ -1582,9 +1635,18 @@ class Agent(BaseAgent):
                 crewai_event_bus.emit(self, event=started_event)
                 self._kickoff_event_id = started_event.event_id
 
-            output = self._execute_and_build_output(executor, inputs, response_format)
+            usage_baseline = self._current_usage_summary()
+            output = self._execute_and_build_output(
+                executor, inputs, response_format, usage_baseline
+            )
             return self._finalize_kickoff(
-                output, executor, inputs, response_format, messages, agent_info
+                output,
+                executor,
+                inputs,
+                response_format,
+                messages,
+                agent_info,
+                usage_baseline,
             )
 
         except Exception as e:
@@ -1598,6 +1660,7 @@ class Agent(BaseAgent):
         response_format: type[Any] | None,
         messages: str | list[LLMMessage],
         agent_info: dict[str, Any],
+        usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Apply guardrails, save to memory, and emit completion event.
 
@@ -1608,6 +1671,8 @@ class Agent(BaseAgent):
             response_format: Optional response format.
             messages: The original messages.
             agent_info: Agent metadata for events.
+            usage_baseline: Usage snapshot taken at kickoff start, so retries
+                report per-call usage relative to it.
 
         Returns:
             The finalized output.
@@ -1618,6 +1683,7 @@ class Agent(BaseAgent):
                 executor=executor,
                 inputs=inputs,
                 response_format=response_format,
+                usage_baseline=usage_baseline,
             )
 
         self._save_kickoff_to_memory(messages, output.raw)
@@ -1669,11 +1735,24 @@ class Agent(BaseAgent):
         except Exception as e:
             self._logger.log("error", f"Failed to save kickoff result to memory: {e}")
 
+    def _current_usage_summary(self) -> UsageMetrics:
+        """Snapshot the cumulative usage counters backing this agent's LLM.
+
+        The counters live on the LLM instance (or the agent's token process
+        for non-BaseLLM models) and grow for the object's lifetime — across
+        calls and across agents sharing the instance. Per-call usage is the
+        delta between two snapshots.
+        """
+        if isinstance(self.llm, BaseLLM):
+            return self.llm.get_token_usage_summary()
+        return self._token_process.get_summary()
+
     def _build_output_from_result(
         self,
         result: dict[str, Any],
         executor: AgentExecutor,
         response_format: type[Any] | None = None,
+        usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Build a LiteAgentOutput from an executor result dict.
 
@@ -1683,6 +1762,9 @@ class Agent(BaseAgent):
             result: The result dictionary from executor.invoke / invoke_async.
             executor: The executor instance.
             response_format: Optional response format.
+            usage_baseline: Usage snapshot taken at kickoff start. When given,
+                the output carries only this call's usage (the delta) instead
+                of the LLM instance's cumulative lifetime counters.
 
         Returns:
             LiteAgentOutput with raw output, formatted result, and metrics.
@@ -1727,10 +1809,9 @@ class Agent(BaseAgent):
         else:
             raw_output = str(output) if not isinstance(output, str) else output
 
-        if isinstance(self.llm, BaseLLM):
-            usage_metrics = self.llm.get_token_usage_summary()
-        else:
-            usage_metrics = self._token_process.get_summary()
+        usage_metrics = self._current_usage_summary()
+        if usage_baseline is not None:
+            usage_metrics = usage_metrics.delta_since(usage_baseline)
 
         raw_str = (
             raw_output
@@ -1759,20 +1840,26 @@ class Agent(BaseAgent):
         executor: AgentExecutor,
         inputs: dict[str, str],
         response_format: type[Any] | None = None,
+        usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Execute the agent synchronously and build the output object."""
         result = cast(dict[str, Any], executor.invoke(inputs))
-        return self._build_output_from_result(result, executor, response_format)
+        return self._build_output_from_result(
+            result, executor, response_format, usage_baseline
+        )
 
     async def _execute_and_build_output_async(
         self,
         executor: AgentExecutor,
         inputs: dict[str, str],
         response_format: type[Any] | None = None,
+        usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Execute the agent asynchronously and build the output object."""
         result = await executor.invoke_async(inputs)
-        return self._build_output_from_result(result, executor, response_format)
+        return self._build_output_from_result(
+            result, executor, response_format, usage_baseline
+        )
 
     def _process_kickoff_guardrail(
         self,
@@ -1781,6 +1868,7 @@ class Agent(BaseAgent):
         inputs: dict[str, str],
         response_format: type[Any] | None = None,
         retry_count: int = 0,
+        usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Process guardrail for kickoff execution with retry logic.
 
@@ -1790,6 +1878,9 @@ class Agent(BaseAgent):
             inputs: Input dictionary for re-execution.
             response_format: Optional response format.
             retry_count: Current retry count.
+            usage_baseline: Usage snapshot taken at kickoff start, so a
+                retried output reports the whole call's usage, not just the
+                last attempt's.
 
         Returns:
             Validated/updated output.
@@ -1827,7 +1918,9 @@ class Agent(BaseAgent):
                 role="user",
             )
 
-            output = self._execute_and_build_output(executor, inputs, response_format)
+            output = self._execute_and_build_output(
+                executor, inputs, response_format, usage_baseline
+            )
 
             return self._process_kickoff_guardrail(
                 output=output,
@@ -1835,6 +1928,7 @@ class Agent(BaseAgent):
                 inputs=inputs,
                 response_format=response_format,
                 retry_count=retry_count + 1,
+                usage_baseline=usage_baseline,
             )
 
         if guardrail_result.result is not None:
@@ -1897,11 +1991,18 @@ class Agent(BaseAgent):
                 crewai_event_bus.emit(self, event=started_event)
                 self._kickoff_event_id = started_event.event_id
 
+            usage_baseline = self._current_usage_summary()
             output = await self._execute_and_build_output_async(
-                executor, inputs, response_format
+                executor, inputs, response_format, usage_baseline
             )
             return self._finalize_kickoff(
-                output, executor, inputs, response_format, messages, agent_info
+                output,
+                executor,
+                inputs,
+                response_format,
+                messages,
+                agent_info,
+                usage_baseline,
             )
 
         except Exception as e:

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -205,7 +205,11 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
         role (str): Role of the agent.
         goal (str): Objective of the agent.
         backstory (str): Backstory of the agent.
-        cache (bool): Whether the agent should use a cache for tool usage.
+        cache (bool): Whether the agent participates in tool-result caching
+            when a cache is enabled. The default (True) only permits
+            participation — caching activates when the crew sets cache=True
+            or the agent explicitly opts in with cache=True or a
+            cache_handler; cache=False excludes the agent entirely.
         config (dict[str, Any] | None): Configuration for the agent.
         verbose (bool): Verbose mode for the Agent Execution.
         max_rpm (int | None): Maximum number of requests per minute for the agent execution.
@@ -254,6 +258,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     _logger: Logger = PrivateAttr(default_factory=lambda: Logger(verbose=False))
     _rpm_controller: RPMController | None = PrivateAttr(default=None)
     _request_within_rpm_limit: SerializableCallable | None = PrivateAttr(default=None)
+    _constructor_cache_opt_in: bool = PrivateAttr(default=False)
     _original_role: str | None = PrivateAttr(default=None)
     _original_goal: str | None = PrivateAttr(default=None)
     _original_backstory: str | None = PrivateAttr(default=None)
@@ -267,7 +272,14 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
         description="Configuration for the agent", default=None, exclude=True
     )
     cache: bool = Field(
-        default=True, description="Whether the agent should use a cache for tool usage."
+        default=True,
+        description=(
+            "Whether the agent participates in tool-result caching when a "
+            "cache is enabled. Caching itself is opt-in: it activates only "
+            "when the crew sets cache=True or the agent explicitly opts in "
+            "(cache=True or a cache_handler at construction). Set False to "
+            "exclude this agent even when the crew enables caching."
+        ),
     )
     verbose: bool = Field(
         default=False, description="Verbose mode for the Agent Execution"
@@ -716,6 +728,19 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
 
         copied_data = self.model_dump(exclude=exclude)
         copied_data = {k: v for k, v in copied_data.items() if v is not None}
+        # Tool-result caching distinguishes "explicitly enabled" from the
+        # field default via model_fields_set; don't let the dump turn the
+        # default into an explicit opt-in on the copy. An agent that opted
+        # in at construction via an explicit cache_handler (excluded from
+        # the dump) must stay opted in — carry the consent as cache=True so
+        # the copy wires its own fresh handler. A handler merely offered by
+        # a crew at kickoff is runtime wiring, not consent, and must not
+        # opt the copy in; _constructor_cache_opt_in is recorded before any
+        # crew wiring can happen.
+        if "cache" not in self.model_fields_set:
+            copied_data.pop("cache", None)
+            if self._constructor_cache_opt_in:
+                copied_data["cache"] = True
         return type(self)(
             **copied_data,
             llm=existing_llm,

lib/crewai/src/crewai/agents/crew_agent_executor.py

@@ -46,8 +46,8 @@ from crewai.hooks.llm_hooks import (
 )
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    get_after_tool_call_hooks,
-    get_before_tool_call_hooks,
+    run_after_tool_call_hooks,
+    run_before_tool_call_hooks,
 )
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.agent_utils import (
@@ -951,7 +951,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
 
         track_delegation_if_needed(func_name, args_dict or {}, self.task)
 
-        hook_blocked = False
         before_hook_context = ToolCallHookContext(
             tool_name=func_name,
             tool_input=args_dict or {},
@@ -960,19 +959,7 @@ class CrewAgentExecutor(BaseAgentExecutor):
             task=self.task,
             crew=self.crew,
         )
-        before_hooks = get_before_tool_call_hooks()
-        try:
-            for hook in before_hooks:
-                hook_result = hook(before_hook_context)
-                if hook_result is False:
-                    hook_blocked = True
-                    break
-        except Exception as hook_error:
-            if self.agent.verbose:
-                PRINTER.print(
-                    content=f"Error in before_tool_call hook: {hook_error}",
-                    color="red",
-                )
+        hook_blocked = run_before_tool_call_hooks(before_hook_context)
 
         if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -1033,19 +1020,7 @@ class CrewAgentExecutor(BaseAgentExecutor):
             tool_result=result,
             raw_tool_result=raw_tool_result,
         )
-        after_hooks = get_after_tool_call_hooks()
-        try:
-            for after_hook in after_hooks:
-                after_hook_result = after_hook(after_hook_context)
-                if after_hook_result is not None:
-                    result = after_hook_result
-                    after_hook_context.tool_result = result
-        except Exception as hook_error:
-            if self.agent.verbose:
-                PRINTER.print(
-                    content=f"Error in after_tool_call hook: {hook_error}",
-                    color="red",
-                )
+        result = run_after_tool_call_hooks(after_hook_context)
 
         if not error_event_emitted:
             crewai_event_bus.emit(

lib/crewai/src/crewai/crew.py

@@ -168,8 +168,11 @@ class Crew(FlowTrackable, BaseModel):
         manager_agent: Custom agent that will be used as manager.
         memory: Whether the crew should use memory to store memories of it's
             execution.
-        cache: Whether the crew should use a cache to store the results of the
-            tools execution.
+        cache: Whether to cache tool results for the crew's agents. Off by
+            default; when enabled, repeated calls to the same tool with
+            identical arguments reuse the first result without re-executing —
+            avoid enabling for live-data or state-mutating tools unless they
+            gate writes with a cache_function.
         function_calling_llm: The language model that will run the tool calling
             for all the agents.
         process: The process flow that the crew will follow (e.g., sequential,
@@ -216,7 +219,16 @@ class Crew(FlowTrackable, BaseModel):
     _kickoff_event_id: str | None = PrivateAttr(default=None)
 
     name: str | None = Field(default="crew")
-    cache: bool = Field(default=True)
+    cache: bool = Field(
+        default=False,
+        description=(
+            "Whether to cache tool results for the crew's agents. Opt-in: "
+            "when enabled, repeated calls to the same tool with identical "
+            "arguments return the first result without re-executing the "
+            "tool — do not enable for live-data or state-mutating tools "
+            "unless they set a cache_function that prevents caching."
+        ),
+    )
     tasks: list[Task] = Field(default_factory=list)
     agents: Annotated[
         list[BaseAgent],
@@ -1048,8 +1060,9 @@ class Crew(FlowTrackable, BaseModel):
             )
             raise
         finally:
-            if self._memory is not None and hasattr(self._memory, "drain_writes"):
-                self._memory.drain_writes()
+            # Safety net for the exception path; the success path already
+            # drained in _create_crew_output before emitting completion.
+            self._drain_memory_writes()
             clear_files(self.id)
             detach(token)
             crewai_event_bus._exit_runtime_scope(runtime_scope)
@@ -1260,6 +1273,9 @@ class Crew(FlowTrackable, BaseModel):
             )
             raise
         finally:
+            # Safety net for the exception path; the success path already
+            # drained in _create_crew_output before emitting completion.
+            self._drain_memory_writes()
             clear_files(self.id)
             detach(token)
             crewai_event_bus._exit_runtime_scope(runtime_scope)
@@ -1503,6 +1519,11 @@ class Crew(FlowTrackable, BaseModel):
             )
             self.manager_agent = manager
         manager.crew = self
+        # The manager is created outside the agents loop that offers the
+        # crew's cache handler at validation time; offer it here so an
+        # opted-in crew (cache=True) also dedupes the manager's tool calls.
+        if self.cache:
+            manager.set_cache_handler(self._cache_handler)
 
     def _get_execution_start_index(self, tasks: list[Task]) -> int | None:
         if self.checkpoint_kickoff_event_id is None:
@@ -1666,6 +1687,9 @@ class Crew(FlowTrackable, BaseModel):
             if files_needing_tool:
                 tools = self._add_file_tools(tools, files_needing_tool)
 
+        # TOOL_SELECTION is dispatched once, in Agent.create_agent_executor,
+        # which every crew task funnels through. Dispatching here as well would
+        # fire the point twice on a crew run (and duplicate additive edits).
         return tools
 
     def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
@@ -1841,6 +1865,38 @@ class Crew(FlowTrackable, BaseModel):
                 output=output.raw,
             )
 
+    def _drain_memory_writes(self) -> None:
+        """Block until all pending background memory saves have completed.
+
+        Covers the crew memory, per-agent memories, and the manager agent's
+        memory — agents save through ``agent.memory`` when set (see
+        ``BaseAgentExecutor._save_to_memory``), so draining only
+        ``self._memory`` can miss in-flight saves. Scope/slice views are
+        unwrapped to their backing ``Memory`` so each pool is drained once.
+
+        Must run before ``CrewKickoffCompletedEvent`` is emitted: listeners
+        (e.g. telemetry sessions) tear down on that event, and any
+        ``MemorySaveCompletedEvent``/``MemorySaveFailedEvent`` emitted after
+        teardown is lost, leaving the save span orphaned.
+        """
+        seen: set[int] = set()
+        candidates = [
+            self._memory,
+            self.memory,
+            getattr(self.manager_agent, "memory", None),
+            *(getattr(agent, "memory", None) for agent in self.agents),
+        ]
+        for mem in candidates:
+            if mem is None or isinstance(mem, bool):
+                continue
+            backing = getattr(mem, "_memory", None) or mem
+            if id(backing) in seen:
+                continue
+            seen.add(id(backing))
+            drain = getattr(backing, "drain_writes", None)
+            if callable(drain):
+                drain()
+
     def _create_crew_output(self, task_outputs: list[TaskOutput]) -> CrewOutput:
         if not task_outputs:
             raise ValueError("No task outputs available to create crew output.")
@@ -1853,6 +1909,34 @@ class Crew(FlowTrackable, BaseModel):
         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/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,
@@ -1867,13 +1951,7 @@ 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(
-            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,

lib/crewai/src/crewai/crews/crew_output.py

@@ -24,9 +24,23 @@ class CrewOutput(BaseModel):
         description="Output of each task", default_factory=list
     )
     token_usage: UsageMetrics = Field(
-        description="Processed token summary", default_factory=UsageMetrics
+        description=(
+            "Processed token summary; ``usage_metrics`` exposes the same "
+            "data as a plain dict"
+        ),
+        default_factory=UsageMetrics,
     )
 
+    @property
+    def usage_metrics(self) -> dict[str, Any]:
+        """Token usage as a plain dict.
+
+        Same attribute name and shape as ``LiteAgentOutput.usage_metrics``
+        (the ``Agent.kickoff()`` result), so a usage accessor written for one
+        result type works on both.
+        """
+        return self.token_usage.model_dump()
+
     @property
     def json(self) -> str | None:  # type: ignore[override]
         if self.tasks_output[-1].output_format != OutputFormat.JSON:

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

lib/crewai/src/crewai/events/types/hook_events.py

@@ -0,0 +1,19 @@
+from typing import Literal
+
+from crewai.events.base_events import BaseEvent
+
+
+class HookDispatchedEvent(BaseEvent):
+    """Event emitted whenever an interception point dispatches to hooks.
+
+    Only emitted when at least one hook is registered for the point, so the
+    no-op fast path stays free of event overhead.
+    """
+
+    type: Literal["hook_dispatched"] = "hook_dispatched"
+    interception_point: str
+    outcome: Literal["proceeded", "modified", "aborted"]
+    hook_count: int
+    duration_ms: float
+    abort_reason: str | None = None
+    abort_source: str | None = None

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -211,6 +211,13 @@ To enable tracing, do any one of these:
         """Print a panel with consistent formatting if verbose is enabled."""
         panel = self.create_panel(content, title, style)
         if is_flow:
+            # A TUI (e.g. the CLI's CrewRunApp) owns the screen and renders flow
+            # progress in its own STEPS panel; emitting Rich panels here would
+            # interleave with and corrupt the TUI, so suppress them in TUI mode.
+            from crewai.events.listeners.tracing.utils import is_tui_mode
+
+            if is_tui_mode():
+                return
             self.print(panel)
             self.print()
         else:

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -62,8 +62,8 @@ from crewai.hooks.llm_hooks import (
 )
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    get_after_tool_call_hooks,
-    get_before_tool_call_hooks,
+    run_after_tool_call_hooks,
+    run_before_tool_call_hooks,
 )
 from crewai.hooks.types import (
     AfterLLMCallHookCallable,
@@ -1975,7 +1975,6 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
 
         track_delegation_if_needed(func_name, args_dict, self.task)
 
-        hook_blocked = False
         before_hook_context = ToolCallHookContext(
             tool_name=func_name,
             tool_input=args_dict,
@@ -1984,19 +1983,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             task=self.task,
             crew=self.crew,
         )
-        before_hooks = get_before_tool_call_hooks()
-        try:
-            for hook in before_hooks:
-                hook_result = hook(before_hook_context)
-                if hook_result is False:
-                    hook_blocked = True
-                    break
-        except Exception as hook_error:
-            if self.agent.verbose:
-                PRINTER.print(
-                    content=f"Error in before_tool_call hook: {hook_error}",
-                    color="red",
-                )
+        hook_blocked = run_before_tool_call_hooks(before_hook_context)
 
         if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -2060,19 +2047,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             tool_result=result,
             raw_tool_result=raw_tool_result,
         )
-        after_hooks = get_after_tool_call_hooks()
-        try:
-            for after_hook in after_hooks:
-                after_hook_result = after_hook(after_hook_context)
-                if after_hook_result is not None:
-                    result = after_hook_result
-                    after_hook_context.tool_result = result
-        except Exception as hook_error:
-            if self.agent.verbose:
-                PRINTER.print(
-                    content=f"Error in after_tool_call hook: {hook_error}",
-                    color="red",
-                )
+        result = run_after_tool_call_hooks(after_hook_context)
 
         if not error_event_emitted:
             crewai_event_bus.emit(

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -133,6 +133,8 @@ class _ConversationalMixin:
         _pending_user_message: str | dict[str, Any] | None
         _pending_intents: Sequence[str] | None
         _pending_intent_llm: str | BaseLLM | None
+        _turn_classified_intent: str | None
+        _assistant_reply_appended: bool
 
         def _clear_or_listeners(self) -> None:
             pass
@@ -185,12 +187,22 @@ class _ConversationalMixin:
             )
             return configured_route
 
-        if state.last_intent:
+        turn_intent = self._turn_classified_intent
+        if turn_intent:
+            state.last_intent = turn_intent
             self._emit_conversation_route_selected(
-                state.last_intent,
+                turn_intent,
                 previous_intent=previous_intent,
             )
-            return state.last_intent
+            return turn_intent
+
+        if previous_intent:
+            logger.debug(
+                "route_turn() returned no route and no intent was classified "
+                "this turn; ignoring stale last_intent=%r from a previous turn "
+                "and falling back to built-in routing",
+                previous_intent,
+            )
 
         if self.can_answer_from_history(context):
             state.last_intent = "answer_from_history"
@@ -310,11 +322,11 @@ class _ConversationalMixin:
             if "from_checkpoint" not in kickoff_kwargs:
                 self._reset_turn_execution_state()
 
-            assistant_count = self._assistant_message_count()
+            object.__setattr__(self, "_assistant_reply_appended", False)
             result = self.kickoff(inputs={"id": sid}, **kickoff_kwargs)
             if (
                 result is not None
-                and self._assistant_message_count() == assistant_count
+                and not self._assistant_reply_appended
                 and self._is_public_turn_result(result)
             ):
                 self.append_assistant_message(self._stringify_result(result))
@@ -387,7 +399,7 @@ class _ConversationalMixin:
                 if "from_checkpoint" not in kickoff_kwargs:
                     self._reset_turn_execution_state()
 
-                assistant_count = self._assistant_message_count()
+                object.__setattr__(self, "_assistant_reply_appended", False)
                 original_stream = bool(getattr(self, "stream", False))
                 original_streaming_turn = getattr(
                     self, "_streaming_conversation_turn", False
@@ -403,7 +415,7 @@ class _ConversationalMixin:
                     )
                 if (
                     result is not None
-                    and self._assistant_message_count() == assistant_count
+                    and not self._assistant_reply_appended
                     and self._is_public_turn_result(result)
                 ):
                     self.append_assistant_message(self._stringify_result(result))
@@ -550,6 +562,11 @@ class _ConversationalMixin:
         supply per-route descriptions, or change the default/fallback intent.
         Override this method to bypass the LLM router entirely (e.g.,
         permission gates before the LLM decision).
+
+        Returning a falsy value means "no routing decision": the turn falls
+        through to the built-in defaults (``answer_from_history`` when
+        configured, else ``converse``). It never replays a previous turn's
+        intent.
         """
         config = self._conversation_config
         if config is None:
@@ -618,6 +635,9 @@ class _ConversationalMixin:
         metadata: dict[str, Any] | None = None,
     ) -> None:
         """Append a final user-visible assistant message."""
+        # Explicit signal for handle_turn's "did the handler reply?" check.
+        # A count heuristic breaks when handlers trim history mid-turn.
+        object.__setattr__(self, "_assistant_reply_appended", True)
         state = cast(ConversationState, self.state)
         state.messages.append(
             ConversationMessage(
@@ -722,6 +742,7 @@ class _ConversationalMixin:
                     context=self.conversation_messages,
                 )
                 state.last_intent = intent
+                object.__setattr__(self, "_turn_classified_intent", intent)
                 return intent
             return text
 
@@ -788,6 +809,10 @@ class _ConversationalMixin:
             object.__setattr__(self, "_pending_intent_llm", None)
         if not hasattr(self, "_streaming_conversation_turn"):
             object.__setattr__(self, "_streaming_conversation_turn", False)
+        if not hasattr(self, "_turn_classified_intent"):
+            object.__setattr__(self, "_turn_classified_intent", None)
+        if not hasattr(self, "_assistant_reply_appended"):
+            object.__setattr__(self, "_assistant_reply_appended", False)
 
     def _create_default_extension_state(self) -> ConversationState | None:
         initial_state_t = getattr(self, "_initial_state_t", None)
@@ -852,6 +877,7 @@ class _ConversationalMixin:
         self._method_call_counts.clear()
         self._clear_or_listeners()
         self._is_execution_resuming = False
+        object.__setattr__(self, "_turn_classified_intent", None)
 
     def _apply_pending_conversational_turn(self) -> None:
         """Drain the stashed user message + classify if intents configured.
@@ -859,6 +885,7 @@ class _ConversationalMixin:
         Called from ``Flow.kickoff_async`` AFTER persist state restore so
         the appended message survives ``self.persistence.load_state(...)``.
         """
+        object.__setattr__(self, "_turn_classified_intent", None)
         if self._pending_user_message is None:
             return
 
@@ -1107,10 +1134,6 @@ class _ConversationalMixin:
             return "public"
         return "private"
 
-    def _assistant_message_count(self) -> int:
-        state = cast(ConversationState, self.state)
-        return sum(1 for message in state.messages if message.role == "assistant")
-
     def _is_public_turn_result(self, result: Any) -> bool:
         if not isinstance(result, str):
             return False
@@ -1190,6 +1213,15 @@ class _ConversationalMixin:
         )
         from crewai.events.types.flow_events import FlowFinishedEvent
 
+        # Background memory saves must finish (and emit their completed/failed
+        # events) before the session-end flow_finished / batch finalization
+        # below tears down listeners, mirroring the non-deferred kickoff path.
+        # The flush then waits for those events' async bus handlers.
+        drain_memory_writes = getattr(self, "_drain_memory_writes", None)
+        if callable(drain_memory_writes):
+            drain_memory_writes()
+        crewai_event_bus.flush()
+
         # Only emit the session-end event when a deferred flow_started is
         # actually pending. ``_deferred_flow_started_event_id`` is set only by
         # deferred kickoffs; when finalization was not deferred, each per-turn

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

@@ -956,6 +956,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             return self.memory.remember_many(content, **kwargs)
         return self.memory.remember(content, **kwargs)
 
+    def _drain_memory_writes(self) -> None:
+        """Block until pending background memory saves for this flow finish.
+
+        Must run before ``FlowFinishedEvent`` is emitted: listeners (e.g.
+        telemetry sessions) tear down on that event, and any
+        ``MemorySaveCompletedEvent``/``MemorySaveFailedEvent`` emitted after
+        teardown is lost, leaving the save span orphaned.
+        """
+        mem = self.memory
+        if mem is None:
+            return
+        backing = getattr(mem, "_memory", None) or mem
+        drain = getattr(backing, "drain_writes", None)
+        if callable(drain):
+            drain()
+
     def extract_memories(self, content: str) -> list[str]:
         """Extract discrete memories from content. Delegates to this flow's memory.
 
@@ -1460,6 +1476,22 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             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(
                 *[
@@ -1474,6 +1506,14 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             not self.suppress_flow_events
             and not self._should_defer_trace_finalization()
         ):
+            # Background memory saves must finish (and emit their
+            # completed/failed events) before flow-finished triggers
+            # listener teardown/finalization; the flush then waits for those
+            # events' async handlers, mirroring Crew._create_crew_output.
+            # Offloaded to a thread so the blocking waits don't stall other
+            # coroutines on the loop.
+            await asyncio.to_thread(self._drain_memory_writes)
+            await asyncio.to_thread(crewai_event_bus.flush)
             future = crewai_event_bus.emit(
                 self,
                 FlowFinishedEvent(
@@ -2013,6 +2053,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
         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(
@@ -2038,6 +2081,37 @@ 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
+
+            # ``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
@@ -2273,6 +2347,21 @@ 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
+
+            # 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]
@@ -2285,6 +2374,14 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             # flag is read from either the instance attribute or an extension
             # definition.
             if not self._should_defer_trace_finalization():
+                # Background memory saves must finish (and emit their
+                # completed/failed events) before flow-finished triggers
+                # listener teardown/finalization; the flush then waits for
+                # those events' async handlers, mirroring
+                # Crew._create_crew_output. Offloaded to a thread so the
+                # blocking waits don't stall other coroutines on the loop.
+                await asyncio.to_thread(self._drain_memory_writes)
+                await asyncio.to_thread(crewai_event_bus.flush)
                 future = crewai_event_bus.emit(
                     self,
                     FlowFinishedEvent(
@@ -2317,9 +2414,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
 
             return final_output
         finally:
-            # Ensure all background memory saves complete before returning
-            if self.memory is not None and hasattr(self.memory, "drain_writes"):
-                self.memory.drain_writes()
+            # Safety net for the exception path; the success path already
+            # drained before emitting FlowFinishedEvent.
+            self._drain_memory_writes()
             # Drain pending LLMCallCompletedEvent handlers before
             # detaching so `flow.usage_metrics` reflects every call
             # emitted during this kickoff — mirrors `Crew.kickoff()`,
@@ -2338,6 +2435,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 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)
 
@@ -2530,6 +2629,33 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                 if future:
                     self._event_futures.append(future)
 
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="flow_method",
+                step_name=str(method_name),
+                flow=self,
+                payload=dumped_params,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+
+            # Apply hook edits/replacement of the step params back onto the
+            # call. ``dumped_params`` maps positional args to ``_0, _1, ...``
+            # keys and keeps kwargs by name, so reverse that mapping here.
+            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()),
+                    key=lambda k: int(k[1:]),
+                )
+                args = tuple(updated_params[k] for k in positional)
+                kwargs = {
+                    k: v
+                    for k, v in updated_params.items()
+                    if not (k.startswith("_") and k[1:].isdigit())
+                }
+
             # Set method name in context so ask() can read it without
             # stack inspection.  Must happen before copy_context() so the
             # value propagates into the thread pool for sync methods.
@@ -2557,6 +2683,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     method_name, method_definition.human_feedback, result
                 )
 
+            post_step_ctx = StepContext(
+                kind="flow_method",
+                step_name=str(method_name),
+                flow=self,
+                output=result,
+                payload=result,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            result = post_step_ctx.payload
+
             self._method_outputs.append({"method": str(method_name), "output": result})
 
             # For @human_feedback methods with emit, the result is the collapsed outcome
@@ -2753,6 +2889,19 @@ 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)
@@ -2781,6 +2930,19 @@ 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/flow/runtime/_actions.py

@@ -224,7 +224,34 @@ class ScriptAction:
 
     def run(self, *args: Any, **kwargs: Any) -> Any:
         local_context = _pop_local_context(kwargs)
-        return self.handler(
+
+        from crewai.hooks.contexts import PreCodeExecutionContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        code_ctx = PreCodeExecutionContext(
+            flow=self.flow,
+            code=self.definition.code,
+            language="python",
+            payload=self.definition.code,
+        )
+        dispatch(InterceptionPoint.PRE_CODE_EXECUTION, code_ctx)
+
+        # Honor a hook that rewrites the code, via either a returned payload
+        # replacement or an in-place ``ctx.code`` edit. Recompile only when the
+        # source actually changed so the common no-hook path stays free.
+        effective_code = (
+            code_ctx.payload
+            if isinstance(code_ctx.payload, str)
+            and code_ctx.payload != self.definition.code
+            else code_ctx.code
+        )
+        handler = (
+            self.handler
+            if effective_code == self.definition.code
+            else self._compile_handler(effective_code)
+        )
+
+        return handler(
             state=self.flow.state,
             outputs=outputs_by_name(
                 self.flow._method_outputs,
@@ -234,7 +261,7 @@ class ScriptAction:
             item=local_context.get("item") if local_context else None,
         )
 
-    def _compile_handler(self) -> Callable[..., Any]:
+    def _compile_handler(self, code: str | None = None) -> Callable[..., Any]:
         raw = os.environ.get(_ALLOW_SCRIPT_EXECUTION_ENV_VAR, "")
         if raw.strip().lower() not in _TRUSTED_SCRIPT_EXECUTION_VALUES:
             raise FlowScriptExecutionDisabledError(
@@ -243,8 +270,9 @@ class ScriptAction:
                 "trusted flow definitions."
             )
 
+        source = code if code is not None else self.definition.code
         filename = f"crewai.flow.script.{self.flow._definition.name}"
-        module = ast.parse(self.definition.code, filename=filename)
+        module = ast.parse(source, filename=filename)
         function = ast.FunctionDef(
             name="_flow_script",
             args=ast.arguments(

lib/crewai/src/crewai/hooks/__init__.py

@@ -6,6 +6,17 @@ from crewai.hooks.decorators import (
     before_llm_call,
     before_tool_call,
 )
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    clear as clear_hooks,
+    clear_all as clear_all_hooks,
+    dispatch,
+    get_hooks,
+    on,
+    register as register_hook,
+    unregister as unregister_hook,
+)
 from crewai.hooks.llm_hooks import (
     LLMCallHookContext,
     clear_after_llm_call_hooks,
@@ -74,6 +85,8 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]:
 
 
 __all__ = [
+    "HookAborted",
+    "InterceptionPoint",
     "LLMCallHookContext",
     "ToolCallHookContext",
     "after_llm_call",
@@ -83,20 +96,27 @@ __all__ = [
     "clear_after_llm_call_hooks",
     "clear_after_tool_call_hooks",
     "clear_all_global_hooks",
+    "clear_all_hooks",
     "clear_all_llm_call_hooks",
     "clear_all_tool_call_hooks",
     "clear_before_llm_call_hooks",
     "clear_before_tool_call_hooks",
+    "clear_hooks",
+    "dispatch",
     "get_after_llm_call_hooks",
     "get_after_tool_call_hooks",
     "get_before_llm_call_hooks",
     "get_before_tool_call_hooks",
+    "get_hooks",
+    "on",
     "register_after_llm_call_hook",
     "register_after_tool_call_hook",
     "register_before_llm_call_hook",
     "register_before_tool_call_hook",
+    "register_hook",
     "unregister_after_llm_call_hook",
     "unregister_after_tool_call_hook",
     "unregister_before_llm_call_hook",
     "unregister_before_tool_call_hook",
+    "unregister_hook",
 ]

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

@@ -0,0 +1,166 @@
+"""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
+
+
+@dataclass
+class StepContext(InterceptionContext):
+    """``pre_step`` / ``post_step``: a task or flow-method step boundary.
+
+    ``kind`` is ``"task"`` for crew tasks and ``"flow_method"`` for flow methods.
+    ``payload`` is the step input (pre) or step output (post).
+    """
+
+    kind: str | None = None
+    step_name: str | None = None
+    output: Any = None
+
+
+@dataclass
+class ToolSelectionContext(InterceptionContext):
+    """``tool_selection``: the set of tools offered to an agent. ``payload`` = tools list."""
+
+    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."""
+
+    memory_type: str | None = None
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class MemoryReadContext(InterceptionContext):
+    """``memory_read``: a memory query is being issued. ``payload`` = query (pre) / results (post)."""
+
+    memory_type: str | None = None
+    query: str | None = None
+
+
+@dataclass
+class KnowledgeRetrievalContext(InterceptionContext):
+    """``knowledge_retrieval``: a knowledge query. ``payload`` = query / retrieved results."""
+
+    query: Any = None
+
+
+@dataclass
+class PreCodeExecutionContext(InterceptionContext):
+    """``pre_code_execution``: code is about to run. ``payload`` = the code string."""
+
+    code: str | None = None
+    language: str | None = None
+
+
+@dataclass
+class MCPConnectContext(InterceptionContext):
+    """``mcp_connect``: an MCP client is about to connect. ``payload`` = connection params."""
+
+    server_name: str | None = None
+    server_params: Any = None
+
+
+@dataclass
+class FileAccessContext(InterceptionContext):
+    """``file_access``: reserved. No live consumer seam yet."""
+
+    path: str | None = None
+    mode: str | None = None
+
+
+@dataclass
+class ArtifactOutputContext(InterceptionContext):
+    """``artifact_output``: reserved. No live consumer seam yet."""
+
+    artifact: 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

@@ -0,0 +1,436 @@
+"""Generic interception-hook dispatcher.
+
+This module is the single engine behind every CrewAI interception point. A hook
+receives a typed context, may mutate it in place and/or return a replacement
+payload, and may raise :class:`HookAborted` to stop the intercepted operation
+with a reason and source.
+
+The four public hook families (``before/after_llm_call`` and
+``before/after_tool_call``) are adapters registered on this dispatcher, so the
+legacy dialect (``register_*``/decorators/``return False``) and the new dialect
+(``@on(point)`` / ``HookAborted``) share one ordered queue per point.
+
+Design notes:
+- Global registration order is preserved; execution-scoped hooks (via
+  ``contextvars``) run after global ones, mirroring
+  ``events/event_bus.py``'s ``_runtime_state_var`` scoping pattern.
+- ``dispatch`` has a no-op fast path (a single dict lookup) when no hooks are
+  registered for a point.
+- Hooks are synchronous. They may be invoked from async seams, so they must not
+  block on heavy I/O (same restriction as the legacy hooks).
+- ``HookAborted`` propagates by design. Any other exception raised by a hook is
+  swallowed (fail-open) to preserve the framework's protection against a buggy
+  user hook.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+import contextvars
+from enum import Enum
+from functools import wraps
+import inspect
+import time
+from typing import Any
+
+from crewai.utilities.string_utils import sanitize_tool_name
+
+
+class InterceptionPoint(str, Enum):
+    """Catalog of every interception point in the framework.
+
+    The full catalog is frozen from day zero. Points without a live consumer
+    seam yet (``FILE_ACCESS``, ``ARTIFACT_OUTPUT``) can still be registered
+    against; dispatch for them is simply never triggered, which is the same
+    semantics as any point with no hooks.
+    """
+
+    # 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"
+    PRE_TOOL_CALL = "pre_tool_call"
+    POST_TOOL_CALL = "post_tool_call"
+
+    # Step & agent points
+    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"
+    MEMORY_READ = "memory_read"
+    KNOWLEDGE_RETRIEVAL = "knowledge_retrieval"
+    PRE_CODE_EXECUTION = "pre_code_execution"
+    MCP_CONNECT = "mcp_connect"
+    FILE_ACCESS = "file_access"
+    ARTIFACT_OUTPUT = "artifact_output"
+
+    # 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.
+
+    Args:
+        reason: Human-readable explanation of why the operation was aborted.
+        source: Optional identifier of the aborting hook (callable, string, or
+            any object). Used for telemetry and failure messages.
+    """
+
+    def __init__(self, reason: str, source: Any = None) -> None:
+        super().__init__(reason)
+        self.reason = reason
+        self.source = source
+
+
+HookFn = Callable[[Any], Any]
+
+# (ctx, result) -> modified?  A reducer maps a hook's return value onto the
+# context using point-specific semantics. It may raise HookAborted.
+Reducer = Callable[[Any, Any], bool]
+
+
+_global_hooks: dict[InterceptionPoint, list[HookFn]] = {
+    point: [] for point in InterceptionPoint
+}
+
+_scoped_hooks_var: contextvars.ContextVar[
+    dict[InterceptionPoint, list[HookFn]] | None
+] = contextvars.ContextVar("crewai_scoped_hooks", default=None)
+
+
+_TELEMETRY_SOURCE = object()
+
+
+def get_global_hook_list(point: InterceptionPoint) -> list[HookFn]:
+    """Return the live global hook list for a point.
+
+    The returned list object is stable for the lifetime of the process, which
+    lets legacy modules alias their module-level registries to it. Mutate it in
+    place (append/remove/clear); never rebind it.
+    """
+    return _global_hooks[point]
+
+
+def register(point: InterceptionPoint, hook: HookFn) -> None:
+    """Register a global hook for an interception point."""
+    _global_hooks[point].append(hook)
+
+
+def unregister(point: InterceptionPoint, hook: HookFn) -> bool:
+    """Unregister a specific global hook. Returns True if it was removed.
+
+    When ``hook`` was registered through :func:`on` with ``agents``/``tools``
+    filters, the stored callable is a wrapper rather than ``hook`` itself. The
+    wrapper is stashed on ``hook._registered_hook`` at registration time, so it
+    can be resolved and removed here.
+    """
+    hooks = _global_hooks[point]
+    target = hook if hook in hooks else getattr(hook, "_registered_hook", hook)
+    try:
+        hooks.remove(target)
+        return True
+    except ValueError:
+        return False
+
+
+def get_hooks(point: InterceptionPoint) -> list[HookFn]:
+    """Return a copy of the global hooks registered for a point."""
+    return _global_hooks[point].copy()
+
+
+def clear(point: InterceptionPoint) -> int:
+    """Clear all global hooks for a point. Returns the number cleared."""
+    count = len(_global_hooks[point])
+    _global_hooks[point].clear()
+    return count
+
+
+def clear_all() -> None:
+    """Clear all global hooks across every interception point."""
+    for hooks in _global_hooks.values():
+        hooks.clear()
+
+
+@contextmanager
+def scoped_hooks(
+    hooks: dict[InterceptionPoint, list[HookFn]] | None = None,
+) -> Iterator[dict[InterceptionPoint, list[HookFn]]]:
+    """Enter an execution-scoped hook registry.
+
+    Hooks registered inside this context (via :func:`register_scoped`) run after
+    global hooks and are discarded when the context exits. Mirrors the event
+    bus's scoped-handler pattern.
+    """
+    scope: dict[InterceptionPoint, list[HookFn]] = hooks if hooks is not None else {}
+    token = _scoped_hooks_var.set(scope)
+    try:
+        yield scope
+    finally:
+        _scoped_hooks_var.reset(token)
+
+
+def register_scoped(point: InterceptionPoint, hook: HookFn) -> None:
+    """Register a hook scoped to the current :func:`scoped_hooks` context."""
+    scope = _scoped_hooks_var.get()
+    if scope is None:
+        raise RuntimeError(
+            "register_scoped() called outside of a scoped_hooks() context"
+        )
+    scope.setdefault(point, []).append(hook)
+
+
+def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]:
+    """Resolve the ordered hooks for a point: global first, then scoped."""
+    global_hooks = _global_hooks[point]
+    scope = _scoped_hooks_var.get()
+    if scope:
+        scoped = scope.get(point)
+        if scoped:
+            return [*global_hooks, *scoped]
+    return global_hooks
+
+
+def _source_name(source: Any) -> str | None:
+    """Best-effort readable name for a hook source."""
+    if source is None:
+        return None
+    if isinstance(source, str):
+        return source
+    name = getattr(source, "__name__", None)
+    if name:
+        return name
+    return type(source).__name__
+
+
+def _emit_telemetry(
+    point: InterceptionPoint,
+    outcome: str,
+    hook_count: int,
+    duration_ms: float,
+    abort_reason: str | None,
+    abort_source: str | None,
+) -> None:
+    """Emit a HookDispatchedEvent. Never raises."""
+    try:
+        from crewai.events.event_bus import crewai_event_bus
+        from crewai.events.types.hook_events import HookDispatchedEvent
+
+        crewai_event_bus.emit(
+            _TELEMETRY_SOURCE,
+            event=HookDispatchedEvent(
+                interception_point=point.value,
+                outcome=outcome,  # type: ignore[arg-type]
+                hook_count=hook_count,
+                duration_ms=duration_ms,
+                abort_reason=abort_reason,
+                abort_source=abort_source,
+            ),
+        )
+    except Exception:  # noqa: S110 - telemetry must never break dispatch
+        pass
+
+
+def _default_reducer(ctx: Any, result: Any) -> bool:
+    """Default payload semantics: a non-None return replaces ``ctx.payload``.
+
+    Only reports a modification when the payload was actually applied, so a
+    context without a ``payload`` attribute does not produce a misleading
+    ``"modified"`` telemetry outcome.
+    """
+    if result is not None and hasattr(ctx, "payload"):
+        ctx.payload = result
+        return True
+    return False
+
+
+def _invoke_hook(
+    point: InterceptionPoint,
+    hook: HookFn,
+    ctx: Any,
+    reducer: Reducer,
+    verbose: bool,
+) -> bool:
+    """Run a single hook and apply its result via the reducer.
+
+    Returns whether the context was modified. Raises :class:`HookAborted` (with
+    ``source`` populated) to abort; any other exception is swallowed (fail-open).
+    """
+    try:
+        result = hook(ctx)
+        return reducer(ctx, result)
+    except HookAborted as aborted:
+        if aborted.source is None:
+            aborted.source = hook
+        raise
+    except Exception as error:
+        if verbose:
+            from crewai_core.printer import PRINTER
+
+            PRINTER.print(
+                content=f"Error in {point.value} hook: {error}",
+                color="yellow",
+            )
+        return False
+
+
+def run_hooks(
+    point: InterceptionPoint,
+    ctx: Any,
+    hooks: list[HookFn],
+    *,
+    reducer: Reducer | None = None,
+    verbose: bool = True,
+) -> Any:
+    """Execute an explicit list of hooks against a context.
+
+    This is the shared engine used both by :func:`dispatch` (which resolves
+    global + scoped hooks) and by seams that carry a pre-snapshotted hook list
+    (e.g. per-executor LLM hook lists).
+
+    Args:
+        point: The interception point being dispatched.
+        ctx: The typed context passed to each hook (mutated in place).
+        hooks: The ordered hooks to run.
+        reducer: Maps each hook's return value onto ``ctx``. Defaults to
+            :func:`_default_reducer` (payload replacement). May raise
+            :class:`HookAborted`.
+        verbose: Whether to print swallowed-hook-error warnings.
+
+    Returns:
+        The (possibly mutated) context.
+
+    Raises:
+        HookAborted: If a hook or the reducer aborts the operation. Telemetry is
+            still emitted before propagating.
+    """
+    if not hooks:
+        return ctx
+
+    active_reducer = reducer if reducer is not None else _default_reducer
+    start = time.perf_counter()
+    outcome = "proceeded"
+    abort_reason: str | None = None
+    abort_source: str | None = None
+    modified = False
+
+    try:
+        for hook in list(hooks):
+            if _invoke_hook(point, hook, ctx, active_reducer, verbose):
+                modified = True
+        outcome = "modified" if modified else "proceeded"
+        return ctx
+    except HookAborted as aborted:
+        outcome = "aborted"
+        abort_reason = aborted.reason
+        abort_source = _source_name(aborted.source)
+        raise
+    finally:
+        _emit_telemetry(
+            point,
+            outcome,
+            len(hooks),
+            (time.perf_counter() - start) * 1000.0,
+            abort_reason,
+            abort_source,
+        )
+
+
+def dispatch(
+    point: InterceptionPoint,
+    ctx: Any,
+    *,
+    reducer: Reducer | None = None,
+    verbose: bool = True,
+) -> Any:
+    """Dispatch a context to all hooks registered for a point.
+
+    Resolves global then scoped hooks and runs them through :func:`run_hooks`.
+    No-op fast path when nothing is registered.
+    """
+    hooks = _resolve_hooks(point)
+    if not hooks:
+        return ctx
+    return run_hooks(point, ctx, hooks, reducer=reducer, verbose=verbose)
+
+
+def _wrap_with_filters(
+    func: HookFn,
+    agents: list[str] | None,
+    tools: list[str] | None,
+) -> HookFn:
+    """Wrap a hook so it only runs for matching agents/tools (context-shape aware)."""
+
+    @wraps(func)
+    def filtered(ctx: Any) -> Any:
+        if tools:
+            tool_name = getattr(ctx, "tool_name", None)
+            if tool_name is not None and tool_name not in tools:
+                return None
+        if agents:
+            agent = getattr(ctx, "agent", None)
+            role = getattr(agent, "role", None) if agent is not None else None
+            if role is None:
+                role = getattr(ctx, "agent_role", None)
+            if role is not None and role not in agents:
+                return None
+        return func(ctx)
+
+    return filtered
+
+
+def on(
+    point: InterceptionPoint,
+    *,
+    agents: list[str] | None = None,
+    tools: list[str] | None = None,
+) -> Callable[[HookFn], HookFn]:
+    """Register a function as a hook for an interception point.
+
+    Mirrors the legacy decorators' ergonomics: supports ``agents=`` / ``tools=``
+    filters and, when applied to a method inside a ``@CrewBase`` class, defers
+    registration to crew initialization (crew-scoped) instead of registering
+    globally.
+
+    Example:
+        >>> @on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
+        ... def guard(ctx):
+        ...     raise HookAborted("deletion not allowed")
+    """
+    normalized_tools = [sanitize_tool_name(t) for t in tools] if tools else None
+
+    def decorator(func: HookFn) -> HookFn:
+        func._interception_point = point  # type: ignore[attr-defined]
+        if normalized_tools:
+            func._filter_tools = normalized_tools  # type: ignore[attr-defined]
+        if agents:
+            func._filter_agents = agents  # type: ignore[attr-defined]
+
+        params = list(inspect.signature(func).parameters.keys())
+        is_method = len(params) >= 2 and params[0] == "self"
+
+        if not is_method:
+            hook = (
+                _wrap_with_filters(func, agents, normalized_tools)
+                if (agents or normalized_tools)
+                else func
+            )
+            register(point, hook)
+            # Remember the actually-registered callable so unregister_hook(func)
+            # can resolve the filter wrapper.
+            func._registered_hook = hook  # type: ignore[attr-defined]
+
+        return func
+
+    return decorator

lib/crewai/src/crewai/hooks/llm_hooks.py

@@ -5,6 +5,11 @@ from typing import TYPE_CHECKING, Any, cast
 from crewai_core.printer import PRINTER
 
 from crewai.events.event_listener import event_listener
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    get_global_hook_list,
+)
 from crewai.hooks.types import (
     AfterLLMCallHookCallable,
     AfterLLMCallHookType,
@@ -150,8 +155,37 @@ class LLMCallHookContext:
             event_listener.formatter.resume_live_updates()
 
 
-_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = []
-_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = []
+# The legacy registries are aliased to the generic dispatcher's global hook
+# lists for the model-call points, so legacy registrations and new-dialect
+# ``@on(InterceptionPoint.PRE_MODEL_CALL)`` hooks share one ordered queue.
+_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.PRE_MODEL_CALL)
+)
+_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.POST_MODEL_CALL)
+)
+
+
+def before_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``pre_model_call`` hooks.
+
+    A ``False`` return aborts the call (mapped to :class:`HookAborted`); messages
+    are modified in place, so no payload replacement occurs here.
+    """
+    if result is False:
+        raise HookAborted(reason="before_llm_call hook returned False")
+    return False
+
+
+def after_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``post_model_call`` hooks.
+
+    A non-empty string return replaces the response on the context.
+    """
+    if result is not None and isinstance(result, str):
+        context.response = result
+        return True
+    return False
 
 
 def register_before_llm_call_hook(

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any
 from crewai_core.printer import PRINTER
 
 from crewai.events.event_listener import event_listener
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    dispatch,
+    get_global_hook_list,
+)
 from crewai.hooks.types import (
     AfterToolCallHookCallable,
     AfterToolCallHookType,
@@ -121,8 +127,81 @@ class ToolCallHookContext:
             event_listener.formatter.resume_live_updates()
 
 
-_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = []
-_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = []
+# The legacy registries are aliased to the generic dispatcher's global hook
+# lists for the tool-call points, so legacy registrations and new-dialect
+# ``@on(InterceptionPoint.PRE_TOOL_CALL)`` hooks share one ordered queue.
+_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.PRE_TOOL_CALL)
+)
+_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = (
+    get_global_hook_list(InterceptionPoint.POST_TOOL_CALL)
+)
+
+
+def before_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``pre_tool_call`` hooks.
+
+    A ``False`` return blocks the call (mapped to :class:`HookAborted`); tool
+    input is modified in place, so no payload replacement occurs here.
+    """
+    if result is False:
+        raise HookAborted(reason="before_tool_call hook returned False")
+    return False
+
+
+def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
+    """Legacy calling convention for ``post_tool_call`` hooks.
+
+    A non-None return replaces the tool result on the context.
+    """
+    if result is not None:
+        context.tool_result = result
+        return True
+    return False
+
+
+def _hook_verbose(context: ToolCallHookContext) -> bool:
+    """Whether swallowed-hook-error warnings should be printed.
+
+    Mirrors the pre-dispatcher behavior where a failing tool hook surfaced a
+    warning when the executing agent was verbose.
+    """
+    return bool(getattr(context.agent, "verbose", False))
+
+
+def run_before_tool_call_hooks(context: ToolCallHookContext) -> bool:
+    """Run all ``pre_tool_call`` hooks against a context.
+
+    Returns:
+        True if a hook blocked execution (returned False or raised
+        :class:`HookAborted`), False otherwise. Tool input mutations on the
+        context persist regardless.
+    """
+    try:
+        dispatch(
+            InterceptionPoint.PRE_TOOL_CALL,
+            context,
+            reducer=before_tool_call_reducer,
+            verbose=_hook_verbose(context),
+        )
+        return False
+    except HookAborted:
+        return True
+
+
+def run_after_tool_call_hooks(context: ToolCallHookContext) -> str | None:
+    """Run all ``post_tool_call`` hooks against a context.
+
+    Returns:
+        The (possibly modified) tool result carried on the context.
+    """
+    dispatch(
+        InterceptionPoint.POST_TOOL_CALL,
+        context,
+        reducer=after_tool_call_reducer,
+        verbose=_hook_verbose(context),
+    )
+    return context.tool_result
 
 
 def register_before_tool_call_hook(

lib/crewai/src/crewai/knowledge/knowledge.py

@@ -145,6 +145,13 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
+        from crewai.hooks.contexts import KnowledgeRetrievalContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
+        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
+        query = retrieval_ctx.payload
+
         return self.storage.search(
             query,
             limit=results_limit,
@@ -183,6 +190,13 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
+        from crewai.hooks.contexts import KnowledgeRetrievalContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
+        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
+        query = retrieval_ctx.payload
+
         return await self.storage.asearch(
             query,
             limit=results_limit,

lib/crewai/src/crewai/lite_agent_output.py

@@ -6,6 +6,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
+from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.planning_types import TodoItem
 from crewai.utilities.types import LLMMessage
 
@@ -38,7 +39,13 @@ class LiteAgentOutput(BaseModel):
     )
     agent_role: str = Field(description="Role of the agent that produced this output")
     usage_metrics: dict[str, Any] | None = Field(
-        description="Token usage metrics for this execution", default=None
+        description=(
+            "Token usage for this kickoff call only (guardrail retries "
+            "included), not the LLM instance's cumulative totals, as a "
+            "plain dict; ``token_usage`` exposes the same data as a "
+            "UsageMetrics object"
+        ),
+        default=None,
     )
     messages: list[LLMMessage] = Field(
         description="Messages of the agent", default_factory=list
@@ -86,6 +93,19 @@ class LiteAgentOutput(BaseModel):
             return self.pydantic.model_dump()
         return {}
 
+    @property
+    def token_usage(self) -> UsageMetrics:
+        """Token usage as a ``UsageMetrics`` object.
+
+        Same attribute name and type as ``CrewOutput.token_usage``, so a
+        usage accessor written for one result type works on both. Returns
+        zeroed metrics when no usage was captured (``usage_metrics`` is
+        ``None``).
+        """
+        if not self.usage_metrics:
+            return UsageMetrics()
+        return UsageMetrics.model_validate(self.usage_metrics)
+
     @property
     def completed_todos(self) -> list[TodoExecutionResult]:
         """Get only the completed todos."""

lib/crewai/src/crewai/llm.py

@@ -394,19 +394,35 @@ class LLM(BaseLLM):
         """Factory method that routes to native SDK or falls back to LiteLLM.
 
         Routing priority:
-            1. If 'provider' kwarg is present, use that provider with constants
-            2. If only 'model' kwarg, use constants to infer provider
-            3. If "/" in model name:
+            1. If ``custom_openai=True``, force the native OpenAI provider,
+               overriding any explicit provider. A custom endpoint is required.
+            2. If ``provider`` is present, use that provider.
+            3. If "/" is in the model name:
                - Check if prefix is a native provider (openai/anthropic/azure/bedrock/gemini)
                - If yes, validate model against constants
                - If valid, route to native SDK; otherwise route to LiteLLM
+            4. Otherwise, infer the provider from the model name.
         """
         if not model or not isinstance(model, str):
             raise ValueError("Model must be a non-empty string")
 
+        custom_openai = bool(kwargs.pop("custom_openai", False))
+        custom_openai_route = custom_openai
         explicit_provider = kwargs.get("provider")
 
-        if explicit_provider:
+        if custom_openai:
+            if not cls._has_custom_openai_endpoint(kwargs):
+                raise ValueError(
+                    "custom_openai=True requires base_url, api_base, "
+                    "OPENAI_BASE_URL, or OPENAI_API_BASE"
+                )
+            provider = "openai"
+            use_native = True
+            prefix, separator, model_part = model.partition("/")
+            model_string = (
+                model_part if separator and prefix.lower() == "openai" else model
+            )
+        elif explicit_provider:
             provider = explicit_provider
             use_native = True
             model_string = model
@@ -435,9 +451,17 @@ class LLM(BaseLLM):
 
             canonical_provider = provider_mapping.get(prefix.lower())
 
-            if canonical_provider and cls._validate_model_in_constants(
-                model_part, canonical_provider
-            ):
+            valid_native_model = bool(
+                canonical_provider
+                and cls._validate_model_in_constants(model_part, canonical_provider)
+            )
+            custom_openai_route = bool(
+                canonical_provider == "openai"
+                and not valid_native_model
+                and cls._has_custom_openai_base_url(kwargs)
+            )
+
+            if canonical_provider and (valid_native_model or custom_openai_route):
                 provider = canonical_provider
                 use_native = True
                 model_string = model_part
@@ -455,6 +479,8 @@ class LLM(BaseLLM):
             try:
                 # Remove 'provider' from kwargs if it exists to avoid duplicate keyword argument
                 kwargs_copy = {k: v for k, v in kwargs.items() if k != "provider"}
+                if custom_openai_route:
+                    kwargs_copy["custom_openai"] = True
                 return cast(
                     Self,
                     native_class(model=model_string, provider=provider, **kwargs_copy),
@@ -590,6 +616,20 @@ class LLM(BaseLLM):
 
         return cls._matches_provider_pattern(model, provider)
 
+    @staticmethod
+    def _has_custom_openai_base_url(kwargs: dict[str, Any]) -> bool:
+        """Return whether this call explicitly configures a custom endpoint."""
+        return bool(kwargs.get("base_url") or kwargs.get("api_base"))
+
+    @classmethod
+    def _has_custom_openai_endpoint(cls, kwargs: dict[str, Any]) -> bool:
+        """Return whether a custom endpoint is configured explicitly or by env."""
+        return bool(
+            cls._has_custom_openai_base_url(kwargs)
+            or os.getenv("OPENAI_BASE_URL")
+            or os.getenv("OPENAI_API_BASE")
+        )
+
     @classmethod
     def _infer_provider_from_model(cls, model: str) -> str:
         """Infer the provider from the model name.

lib/crewai/src/crewai/llms/base_llm.py

@@ -966,8 +966,15 @@ class BaseLLM(BaseModel, ABC):
     def get_token_usage_summary(self) -> UsageMetrics:
         """Get summary of token usage for this LLM instance.
 
+        The counters are cumulative for the lifetime of this instance: they
+        grow across every call made through it, including calls issued by
+        different agents sharing the instance. For usage scoped to a single
+        call, snapshot before and after and use
+        ``UsageMetrics.delta_since`` (agent kickoff results already report
+        per-call usage this way).
+
         Returns:
-            Dictionary with token usage totals
+            UsageMetrics with this instance's lifetime token usage totals.
         """
         return UsageMetrics(**self._token_usage)
 
@@ -1000,13 +1007,14 @@ class BaseLLM(BaseModel, ABC):
 
         from crewai_core.printer import PRINTER
 
+        from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch
         from crewai.hooks.llm_hooks import (
             LLMCallHookContext,
+            before_llm_call_reducer,
             get_before_llm_call_hooks,
         )
 
-        before_hooks = get_before_llm_call_hooks()
-        if not before_hooks:
+        if not get_before_llm_call_hooks():
             return True
 
         hook_context = LLMCallHookContext(
@@ -1017,24 +1025,19 @@ class BaseLLM(BaseModel, ABC):
             task=None,
             crew=None,
         )
-        verbose = getattr(from_agent, "verbose", True) if from_agent else True
 
         try:
-            for hook in before_hooks:
-                result = hook(hook_context)
-                if result is False:
-                    if verbose:
-                        PRINTER.print(
-                            content="LLM call blocked by before_llm_call hook",
-                            color="yellow",
-                        )
-                    return False
-        except Exception as e:
-            if verbose:
-                PRINTER.print(
-                    content=f"Error in before_llm_call hook: {e}",
-                    color="yellow",
-                )
+            dispatch(
+                InterceptionPoint.PRE_MODEL_CALL,
+                hook_context,
+                reducer=before_llm_call_reducer,
+            )
+        except HookAborted:
+            PRINTER.print(
+                content="LLM call blocked by before_llm_call hook",
+                color="yellow",
+            )
+            return False
 
         return True
 
@@ -1067,15 +1070,14 @@ class BaseLLM(BaseModel, ABC):
         if from_agent is not None or not isinstance(response, str):
             return response
 
-        from crewai_core.printer import PRINTER
-
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
         from crewai.hooks.llm_hooks import (
             LLMCallHookContext,
+            after_llm_call_reducer,
             get_after_llm_call_hooks,
         )
 
-        after_hooks = get_after_llm_call_hooks()
-        if not after_hooks:
+        if not get_after_llm_call_hooks():
             return response
 
         hook_context = LLMCallHookContext(
@@ -1087,20 +1089,11 @@ class BaseLLM(BaseModel, ABC):
             crew=None,
             response=response,
         )
-        verbose = getattr(from_agent, "verbose", True) if from_agent else True
-        modified_response = response
 
-        try:
-            for hook in after_hooks:
-                result = hook(hook_context)
-                if result is not None and isinstance(result, str):
-                    modified_response = result
-                    hook_context.response = modified_response
-        except Exception as e:
-            if verbose:
-                PRINTER.print(
-                    content=f"Error in after_llm_call hook: {e}",
-                    color="yellow",
-                )
+        dispatch(
+            InterceptionPoint.POST_MODEL_CALL,
+            hook_context,
+            reducer=after_llm_call_reducer,
+        )
 
-        return modified_response
+        return hook_context.response if hook_context.response is not None else response

lib/crewai/src/crewai/llms/providers/openai/completion.py

@@ -232,6 +232,7 @@ class OpenAICompletion(BaseLLM):
     auto_chain: bool = False
     auto_chain_reasoning: bool = False
     api_base: str | None = None
+    custom_openai: bool = False
     is_o1_model: bool = False
     is_gpt4_model: bool = False
 
@@ -245,6 +246,20 @@ class OpenAICompletion(BaseLLM):
     def _normalize_openai_fields(cls, data: Any) -> Any:
         if not isinstance(data, dict):
             return data
+        if data.get("custom_openai"):
+            custom_base_url = (
+                data.get("base_url")
+                or data.get("api_base")
+                or os.getenv("OPENAI_BASE_URL")
+                or os.getenv("OPENAI_API_BASE")
+            )
+            if not custom_base_url:
+                raise ValueError(
+                    "custom_openai=True requires base_url, api_base, "
+                    "OPENAI_BASE_URL, or OPENAI_API_BASE"
+                )
+            if not data.get("base_url") and not data.get("api_base"):
+                data["base_url"] = custom_base_url
         if not data.get("provider"):
             data["provider"] = "openai"
         data["api_key"] = data.get("api_key") or os.getenv("OPENAI_API_KEY")
@@ -355,6 +370,15 @@ class OpenAICompletion(BaseLLM):
             config["seed"] = self.seed
         if self.reasoning_effort is not None:
             config["reasoning_effort"] = self.reasoning_effort
+        if self.custom_openai:
+            config["model"] = self.model
+            config["custom_openai"] = True
+            config["base_url"] = (
+                self.base_url
+                or self.api_base
+                or os.getenv("OPENAI_BASE_URL")
+                or os.getenv("OPENAI_API_BASE")
+            )
         return config
 
     def _get_client_params(self) -> dict[str, Any]:
@@ -372,6 +396,7 @@ class OpenAICompletion(BaseLLM):
             "base_url": self.base_url
             or self.api_base
             or os.getenv("OPENAI_BASE_URL")
+            or os.getenv("OPENAI_API_BASE")
             or None,
             "timeout": self.timeout,
             "max_retries": self.max_retries,

lib/crewai/src/crewai/mcp/client.py

@@ -152,6 +152,20 @@ class MCPClient:
         server_name, server_url, transport_type = self._get_server_info()
         is_reconnect = self._was_connected
 
+        from crewai.hooks.contexts import MCPConnectContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        connect_ctx = MCPConnectContext(
+            server_name=server_name,
+            server_params=self.transport,
+            payload=self.transport,
+        )
+        dispatch(InterceptionPoint.MCP_CONNECT, connect_ctx)
+        # Honor a hook that replaces the connection transport/params so the
+        # connection below actually uses the returned value.
+        if connect_ctx.payload is not None:
+            self.transport = connect_ctx.payload
+
         started_at = datetime.now()
         crewai_event_bus.emit(
             self,

lib/crewai/src/crewai/memory/unified_memory.py

@@ -466,6 +466,18 @@ class Memory(BaseModel):
         if self.read_only:
             return None
 
+        from crewai.hooks.contexts import MemoryWriteContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        write_ctx = MemoryWriteContext(
+            agent_role=agent_role,
+            memory_type="unified_memory",
+            metadata=metadata or {},
+            payload=content,
+        )
+        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
+        content = write_ctx.payload
+
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -561,6 +573,18 @@ class Memory(BaseModel):
         if not contents or self.read_only:
             return []
 
+        from crewai.hooks.contexts import MemoryWriteContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        write_ctx = MemoryWriteContext(
+            agent_role=agent_role,
+            memory_type="unified_memory",
+            metadata=metadata or {},
+            payload=contents,
+        )
+        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
+        contents = write_ctx.payload
+
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -712,6 +736,17 @@ class Memory(BaseModel):
         # so that the search sees all persisted records.
         self.drain_writes()
 
+        from crewai.hooks.contexts import MemoryReadContext
+        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+        read_ctx = MemoryReadContext(
+            memory_type="unified_memory",
+            query=query,
+            payload=query,
+        )
+        dispatch(InterceptionPoint.MEMORY_READ, read_ctx)
+        query = read_ctx.payload
+
         effective_scope = scope
         if effective_scope is None and self.root_scope:
             effective_scope = self.root_scope

lib/crewai/src/crewai/task.py

@@ -662,6 +662,21 @@ class Task(BaseModel):
                 crewai_event_bus.emit(
                     self, TaskStartedEvent(context=context, task=self)
                 )
+
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                payload=context,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+            context = pre_step_ctx.payload
+
             result = await agent.aexecute_task(
                 task=self,
                 context=context,
@@ -718,6 +733,18 @@ class Task(BaseModel):
                     guardrail=self._guardrail,
                 )
 
+            post_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                output=task_output,
+                payload=task_output,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            task_output = post_step_ctx.payload
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -787,6 +814,21 @@ class Task(BaseModel):
                 crewai_event_bus.emit(
                     self, TaskStartedEvent(context=context, task=self)
                 )
+
+            from crewai.hooks.contexts import StepContext
+            from crewai.hooks.dispatch import InterceptionPoint, dispatch
+
+            pre_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                payload=context,
+            )
+            dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
+            context = pre_step_ctx.payload
+
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -843,6 +885,18 @@ class Task(BaseModel):
                     guardrail=self._guardrail,
                 )
 
+            post_step_ctx = StepContext(
+                kind="task",
+                step_name=self.name or self.description,
+                agent=agent,
+                agent_role=getattr(agent, "role", None),
+                task=self,
+                output=task_output,
+                payload=task_output,
+            )
+            dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
+            task_output = post_step_ctx.payload
+
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -884,6 +938,32 @@ 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
 
@@ -1317,6 +1397,13 @@ Follow these guidelines:
                     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,
@@ -1427,6 +1514,13 @@ Follow these guidelines:
                     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,6 +108,22 @@ 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/base_tool.py

@@ -5,7 +5,6 @@ import asyncio
 from collections.abc import Awaitable, Callable
 import importlib
 from inspect import Parameter, signature
-import json
 import threading
 from typing import (
     Any,
@@ -37,9 +36,9 @@ from crewai.tools.structured_tool import (
     _infer_result_schema_from_callable,
     _serialize_schema,
     build_schema_hint,
+    format_description_for_llm,
 )
 from crewai.types.callback import SerializableCallable, _resolve_dotted_path
-from crewai.utilities.pydantic_schema_utils import generate_model_description
 from crewai.utilities.string_utils import sanitize_tool_name
 
 
@@ -479,15 +478,27 @@ class BaseTool(BaseModel, ABC):
                 f"{self.__class__.__name__}Schema", **fields
             )
 
+    @property
+    def formatted_description(self) -> str:
+        """LLM-facing composite of name, argument schema, and description.
+
+        Use this when rendering the tool into a prompt; ``description``
+        holds only the authored text.
+        """
+        return format_description_for_llm(self.name, self.args_schema, self.description)
+
     def _generate_description(self) -> None:
-        """Generate the tool description with a JSON schema for arguments."""
-        schema = generate_model_description(self.args_schema)
-        args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
-        self.description = (
-            f"Tool Name: {sanitize_tool_name(self.name)}\n"
-            f"Tool Arguments: {args_json}\n"
-            f"Tool Description: {self.description}"
-        )
+        """Deprecated hook kept for backward compatibility; does nothing.
+
+        Historically this rewrote the public ``description`` field at
+        construction time into the LLM-facing composite (``Tool Name: …\\n
+        Tool Arguments: …\\nTool Description: <authored>``). The authored
+        ``description`` is now preserved as written and the composite is
+        exposed separately via :attr:`formatted_description`.
+
+        ``model_post_init`` still calls this so subclasses that override it
+        (e.g. adapters that customize the composite) keep working.
+        """
 
 
 _BASE_TOOL_CLS = BaseTool

lib/crewai/src/crewai/tools/structured_tool.py

@@ -4,6 +4,7 @@ import asyncio
 from collections.abc import Callable
 import inspect
 import json
+import re
 import textwrap
 from typing import TYPE_CHECKING, Annotated, Any, cast, get_type_hints
 import warnings
@@ -21,7 +22,10 @@ from pydantic import (
 from typing_extensions import Self
 
 from crewai.utilities.logger import Logger
-from crewai.utilities.pydantic_schema_utils import create_model_from_schema
+from crewai.utilities.pydantic_schema_utils import (
+    create_model_from_schema,
+    generate_model_description,
+)
 from crewai.utilities.string_utils import sanitize_tool_name
 
 
@@ -108,6 +112,70 @@ def build_schema_hint(args_schema: type[BaseModel]) -> str:
         return ""
 
 
+# Matches a description that IS a pre-composed LLM block (as written by
+# older versions into the field, and by adapters that still bake it in).
+# Anchored to the full three-line shape so authored prose that merely
+# mentions "Tool Description:" is never mistaken for a composite. Greedy
+# ``.*`` keeps only the text after the LAST marker, matching the historical
+# split behavior for nested pre-baked blocks.
+_COMPOSITE_DESCRIPTION_RE = re.compile(
+    r"^Tool Name:.*\nTool Arguments:.*\nTool Description:\s*",
+    re.DOTALL,
+)
+
+
+def strip_composite_description_prefix(description: str) -> str:
+    """Return the authored text from a pre-composed LLM description block.
+
+    Descriptions that don't start with the composite shape are returned
+    unchanged.
+    """
+    match = _COMPOSITE_DESCRIPTION_RE.match(description)
+    if match:
+        return description[match.end() :]
+    return description
+
+
+def format_description_for_llm(
+    name: str,
+    args_schema: type[BaseModel] | None,
+    description: str,
+) -> str:
+    """Compose the LLM-facing tool description.
+
+    Combines the tool name, its argument JSON schema, and the authored
+    description into the prompt block agents see. The authored
+    ``description`` field itself is never mutated — prompt rendering calls
+    this on demand.
+
+    Idempotent: if ``description`` already *is* a composed block (e.g. a
+    tool deserialized from a checkpoint written by an older version, or an
+    adapter that bakes the composite into the field), only the authored
+    text after the marker is used. The check is anchored to the composite
+    shape, so authored prose that merely mentions ``"Tool Description:"``
+    passes through untouched.
+
+    Args:
+        name: The tool name (sanitized for the prompt).
+        args_schema: The tool's argument schema, if any.
+        description: The authored tool description.
+
+    Returns:
+        The composed, LLM-facing description block.
+    """
+    description = strip_composite_description_prefix(description)
+    if args_schema is not None:
+        schema = generate_model_description(args_schema)
+        args_json = json.dumps(schema["json_schema"]["schema"], indent=2)
+    else:
+        args_json = "{}"
+    return (
+        f"Tool Name: {sanitize_tool_name(name)}\n"
+        f"Tool Arguments: {args_json}\n"
+        f"Tool Description: {description}"
+    )
+
+
 class ToolUsageLimitExceededError(Exception):
     """Exception raised when a tool has reached its maximum usage limit."""
 
@@ -141,6 +209,15 @@ class CrewStructuredTool(BaseModel):
     _logger: Logger = PrivateAttr(default_factory=Logger)
     _original_tool: Any = PrivateAttr(default=None)
 
+    @property
+    def formatted_description(self) -> str:
+        """LLM-facing composite of name, argument schema, and description.
+
+        Use this when rendering the tool into a prompt; ``description``
+        holds only the authored text.
+        """
+        return format_description_for_llm(self.name, self.args_schema, self.description)
+
     @model_validator(mode="after")
     def _validate_func(self) -> Self:
         if self.func is not None:

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

@@ -430,7 +430,7 @@ class ToolUsage:
                         ).format(
                             error=e,
                             tool=sanitize_tool_name(tool.name),
-                            tool_inputs=tool.description,
+                            tool_inputs=tool.formatted_description,
                         )
                         result = ToolUsageError(
                             f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
@@ -670,7 +670,7 @@ class ToolUsage:
                         ).format(
                             error=e,
                             tool=sanitize_tool_name(tool.name),
-                            tool_inputs=tool.description,
+                            tool_inputs=tool.formatted_description,
                         )
                         result = ToolUsageError(
                             f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
@@ -803,7 +803,7 @@ class ToolUsage:
 
     def _render(self) -> str:
         """Render the tool name and description in plain text."""
-        descriptions = [tool.description for tool in self.tools]
+        descriptions = [tool.formatted_description for tool in self.tools]
         return "\n--\n".join(descriptions)
 
     def _function_calling(
@@ -879,6 +879,22 @@ 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/src/crewai/types/usage_metrics.py

@@ -76,6 +76,38 @@ class UsageMetrics(BaseModel):
         self.cache_creation_tokens += usage_metrics.cache_creation_tokens
         self.successful_requests += usage_metrics.successful_requests
 
+    def delta_since(self, baseline: Self) -> Self:
+        """Return the per-call usage accrued since ``baseline`` was captured.
+
+        Both objects must come from the same monotonically increasing
+        accumulator (e.g. an LLM instance's lifetime counters). Differences
+        are clamped at zero so a reset accumulator can't produce negative
+        usage.
+
+        Args:
+            baseline: A snapshot of the same accumulator taken earlier.
+
+        Returns:
+            A new UsageMetrics with the field-wise difference.
+        """
+        return type(self)(
+            total_tokens=max(0, self.total_tokens - baseline.total_tokens),
+            prompt_tokens=max(0, self.prompt_tokens - baseline.prompt_tokens),
+            cached_prompt_tokens=max(
+                0, self.cached_prompt_tokens - baseline.cached_prompt_tokens
+            ),
+            completion_tokens=max(
+                0, self.completion_tokens - baseline.completion_tokens
+            ),
+            reasoning_tokens=max(0, self.reasoning_tokens - baseline.reasoning_tokens),
+            cache_creation_tokens=max(
+                0, self.cache_creation_tokens - baseline.cache_creation_tokens
+            ),
+            successful_requests=max(
+                0, self.successful_requests - baseline.successful_requests
+            ),
+        )
+
     @classmethod
     def from_provider_dict(cls, usage_data: dict[str, Any] | None) -> Self | None:
         """Normalize a provider's raw usage dict into a ``UsageMetrics``.

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -27,7 +27,10 @@ from crewai.agents.parser import (
 from crewai.llms.base_llm import BaseLLM, call_stop_override
 from crewai.tools import BaseTool as CrewAITool
 from crewai.tools.base_tool import BaseTool
-from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.structured_tool import (
+    CrewStructuredTool,
+    strip_composite_description_prefix,
+)
 from crewai.tools.tool_types import ToolResult
 from crewai.utilities.errors import AgentRepositoryError
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
@@ -147,7 +150,14 @@ def render_text_description_and_args(
     Returns:
         Plain text description of tools.
     """
-    tool_strings = [tool.description for tool in tools]
+    # Fall back to the raw description for duck-typed tools (including test
+    # mocks) that don't provide a real formatted_description string.
+    tool_strings = [
+        formatted
+        if isinstance((formatted := getattr(tool, "formatted_description", None)), str)
+        else tool.description
+        for tool in tools
+    ]
     return "\n".join(tool_strings)
 
 
@@ -190,10 +200,10 @@ def convert_tools_to_openai_schema(
             except Exception:
                 parameters = {}
 
-        # BaseTool formats description as "Tool Name: ...\nTool Arguments: ...\nTool Description: {original}"
-        description = tool.description
-        if "Tool Description:" in description:
-            description = description.split("Tool Description:")[-1].strip()
+        # Old checkpoints and some adapters bake the composed LLM block
+        # ("Tool Name: ...\nTool Arguments: ...\nTool Description: {authored}")
+        # into the description field; keep only the authored text here.
+        description = strip_composite_description_prefix(tool.description)
 
         sanitized_name = sanitize_tool_name(tool.name)
 
@@ -1443,8 +1453,8 @@ def execute_single_native_tool_call(
     )
     from crewai.hooks.tool_hooks import (
         ToolCallHookContext,
-        get_after_tool_call_hooks,
-        get_before_tool_call_hooks,
+        run_after_tool_call_hooks,
+        run_before_tool_call_hooks,
     )
 
     info = extract_tool_call_info(tool_call)
@@ -1507,7 +1517,6 @@ def execute_single_native_tool_call(
 
     track_delegation_if_needed(func_name, args_dict, task)
 
-    hook_blocked = False
     before_hook_context = ToolCallHookContext(
         tool_name=func_name,
         tool_input=args_dict,
@@ -1516,13 +1525,7 @@ def execute_single_native_tool_call(
         task=task,
         crew=crew,
     )
-    try:
-        for hook in get_before_tool_call_hooks():
-            if hook(before_hook_context) is False:
-                hook_blocked = True
-                break
-    except Exception:  # noqa: S110
-        pass
+    hook_blocked = run_before_tool_call_hooks(before_hook_context)
 
     error_event_emitted = False
     if hook_blocked:
@@ -1577,14 +1580,7 @@ def execute_single_native_tool_call(
         tool_result=result,
         raw_tool_result=raw_tool_result,
     )
-    try:
-        for after_hook in get_after_tool_call_hooks():
-            hook_result = after_hook(after_hook_context)
-            if hook_result is not None:
-                result = hook_result
-                after_hook_context.tool_result = result
-    except Exception:  # noqa: S110
-        pass
+    result = run_after_tool_call_hooks(after_hook_context)
 
     if not error_event_emitted:
         crewai_event_bus.emit(
@@ -1681,27 +1677,31 @@ def _setup_before_llm_call_hooks(
         True if LLM execution should proceed, False if blocked by a hook.
     """
     if executor_context and executor_context.before_llm_call_hooks:
-        from crewai.hooks.llm_hooks import LLMCallHookContext
+        from crewai.hooks.dispatch import (
+            HookAborted,
+            InterceptionPoint,
+            run_hooks,
+        )
+        from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer
 
         original_messages = executor_context.messages
 
         hook_context = LLMCallHookContext(executor_context)
         try:
-            for hook in executor_context.before_llm_call_hooks:
-                result = hook(hook_context)
-                if result is False:
-                    if verbose:
-                        printer.print(
-                            content="LLM call blocked by before_llm_call hook",
-                            color="yellow",
-                        )
-                    return False
-        except Exception as e:
+            run_hooks(
+                InterceptionPoint.PRE_MODEL_CALL,
+                hook_context,
+                executor_context.before_llm_call_hooks,
+                reducer=before_llm_call_reducer,
+                verbose=verbose,
+            )
+        except HookAborted:
             if verbose:
                 printer.print(
-                    content=f"Error in before_llm_call hook: {e}",
+                    content="LLM call blocked by before_llm_call hook",
                     color="yellow",
                 )
+            return False
 
         if not isinstance(executor_context.messages, list):
             if verbose:
@@ -1739,7 +1739,8 @@ def _setup_after_llm_call_hooks(
         The potentially modified response (string or Pydantic model).
     """
     if executor_context and executor_context.after_llm_call_hooks:
-        from crewai.hooks.llm_hooks import LLMCallHookContext
+        from crewai.hooks.dispatch import InterceptionPoint, run_hooks
+        from crewai.hooks.llm_hooks import LLMCallHookContext, after_llm_call_reducer
 
         original_messages = executor_context.messages
 
@@ -1752,18 +1753,15 @@ def _setup_after_llm_call_hooks(
             hook_response = str(answer)
 
         hook_context = LLMCallHookContext(executor_context, response=hook_response)
-        try:
-            for hook in executor_context.after_llm_call_hooks:
-                modified_response = hook(hook_context)
-                if modified_response is not None and isinstance(modified_response, str):
-                    hook_response = modified_response
-
-        except Exception as e:
-            if verbose:
-                printer.print(
-                    content=f"Error in after_llm_call hook: {e}",
-                    color="yellow",
-                )
+        run_hooks(
+            InterceptionPoint.POST_MODEL_CALL,
+            hook_context,
+            executor_context.after_llm_call_hooks,
+            reducer=after_llm_call_reducer,
+            verbose=verbose,
+        )
+        if hook_context.response is not None:
+            hook_response = hook_context.response
 
         if not isinstance(executor_context.messages, list):
             if verbose:

lib/crewai/src/crewai/utilities/tool_utils.py

@@ -6,15 +6,14 @@ from crewai.agents.parser import AgentAction
 from crewai.agents.tools_handler import ToolsHandler
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    get_after_tool_call_hooks,
-    get_before_tool_call_hooks,
+    run_after_tool_call_hooks,
+    run_before_tool_call_hooks,
 )
 from crewai.security.fingerprint import Fingerprint
 from crewai.tools.structured_tool import CrewStructuredTool
 from crewai.tools.tool_types import ToolResult
 from crewai.tools.tool_usage import ToolUsage, ToolUsageError
 from crewai.utilities.i18n import I18N_DEFAULT
-from crewai.utilities.logger import Logger
 from crewai.utilities.string_utils import sanitize_tool_name
 
 
@@ -57,11 +56,10 @@ async def aexecute_tool_and_check_finality(
         fingerprint_context: Optional context for fingerprinting.
         crew: Optional crew instance for hook context.
 
-    Returns:
+        Returns:
         ToolResult containing the execution result and whether it should be
         treated as a final answer.
     """
-    logger = Logger(verbose=crew.verbose if crew else False)
     tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools}
 
     if agent_key and agent_role and agent:
@@ -102,18 +100,24 @@ async def aexecute_tool_and_check_finality(
             crew=crew,
         )
 
-        before_hooks = get_before_tool_call_hooks()
-        try:
-            for hook in before_hooks:
-                result = hook(hook_context)
-                if result is False:
-                    blocked_message = (
-                        f"Tool execution blocked by hook. "
-                        f"Tool: {tool_calling.tool_name}"
-                    )
-                    return ToolResult(blocked_message, False)
-        except Exception as e:
-            logger.log("error", f"Error in before_tool_call hook: {e}")
+        if run_before_tool_call_hooks(hook_context):
+            blocked_message = (
+                f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
+            )
+            # Run POST_TOOL_CALL even on a blocked call so monitoring hooks
+            # still fire, matching the native tool-call paths.
+            blocked_hook_context = ToolCallHookContext(
+                tool_name=sanitized_tool_name,
+                tool_input=tool_input,
+                tool=tool,
+                agent=agent,
+                task=task,
+                crew=crew,
+                tool_result=blocked_message,
+                raw_tool_result=blocked_message,
+            )
+            modified_result = run_after_tool_call_hooks(blocked_hook_context)
+            return ToolResult(modified_result, False)
 
         tool_result = await tool_usage.ause(tool_calling, agent_action.text)
         raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -129,16 +133,7 @@ async def aexecute_tool_and_check_finality(
             raw_tool_result=raw_tool_result,
         )
 
-        after_hooks = get_after_tool_call_hooks()
-        modified_result: str = tool_result
-        try:
-            for after_hook in after_hooks:
-                hook_result = after_hook(after_hook_context)
-                if hook_result is not None:
-                    modified_result = hook_result
-                    after_hook_context.tool_result = modified_result
-        except Exception as e:
-            logger.log("error", f"Error in after_tool_call hook: {e}")
+        modified_result = run_after_tool_call_hooks(after_hook_context)
 
         return ToolResult(modified_result, tool.result_as_answer)
 
@@ -181,7 +176,6 @@ def execute_tool_and_check_finality(
     Returns:
         ToolResult containing the execution result and whether it should be treated as a final answer
     """
-    logger = Logger(verbose=crew.verbose if crew else False)
     tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools}
 
     if agent_key and agent_role and agent:
@@ -222,18 +216,24 @@ def execute_tool_and_check_finality(
             crew=crew,
         )
 
-        before_hooks = get_before_tool_call_hooks()
-        try:
-            for hook in before_hooks:
-                result = hook(hook_context)
-                if result is False:
-                    blocked_message = (
-                        f"Tool execution blocked by hook. "
-                        f"Tool: {tool_calling.tool_name}"
-                    )
-                    return ToolResult(blocked_message, False)
-        except Exception as e:
-            logger.log("error", f"Error in before_tool_call hook: {e}")
+        if run_before_tool_call_hooks(hook_context):
+            blocked_message = (
+                f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
+            )
+            # Run POST_TOOL_CALL even on a blocked call so monitoring hooks
+            # still fire, matching the native tool-call paths.
+            blocked_hook_context = ToolCallHookContext(
+                tool_name=sanitized_tool_name,
+                tool_input=tool_input,
+                tool=tool,
+                agent=agent,
+                task=task,
+                crew=crew,
+                tool_result=blocked_message,
+                raw_tool_result=blocked_message,
+            )
+            modified_result = run_after_tool_call_hooks(blocked_hook_context)
+            return ToolResult(modified_result, False)
 
         tool_result = tool_usage.use(tool_calling, agent_action.text)
         raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -249,16 +249,7 @@ def execute_tool_and_check_finality(
             raw_tool_result=raw_tool_result,
         )
 
-        after_hooks = get_after_tool_call_hooks()
-        modified_result: str = tool_result
-        try:
-            for after_hook in after_hooks:
-                hook_result = after_hook(after_hook_context)
-                if hook_result is not None:
-                    modified_result = hook_result
-                    after_hook_context.tool_result = modified_result
-        except Exception as e:
-            logger.log("error", f"Error in after_tool_call hook: {e}")
+        modified_result = run_after_tool_call_hooks(after_hook_context)
 
         return ToolResult(modified_result, tool.result_as_answer)
 

lib/crewai/tests/agents/test_lite_agent.py

@@ -1138,3 +1138,160 @@ def test_lite_agent_memory_instance_recall_and_save_called():
     mock_memory.remember_many.assert_called_once_with(
         ["Fact one.", "Fact two."], agent_role="Test"
     )
+
+
+class _FixedUsageLLM(BaseLLM):
+    """Offline BaseLLM that records fixed usage (100/10 tokens) per call."""
+
+    def __init__(self):
+        super().__init__(model="fixed-usage-model")
+
+    def call(
+        self,
+        messages,
+        tools=None,
+        callbacks=None,
+        available_functions=None,
+        from_task=None,
+        from_agent=None,
+        response_model=None,
+    ) -> str:
+        self._track_token_usage_internal(
+            {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}
+        )
+        return "Thought: I know the answer.\nFinal Answer: fake answer"
+
+    def supports_function_calling(self) -> bool:
+        return False
+
+    def supports_stop_words(self) -> bool:
+        return False
+
+    def get_context_window_size(self) -> int:
+        return 4096
+
+
+class TestKickoffUsageMetricsArePerCall:
+    """Regression tests for EPD-177: kickoff results used to expose the LLM
+    instance's cumulative lifetime counters, so counts accumulated across
+    calls and pooled across agents sharing one LLM object.
+    """
+
+    def _make_agent(self, role: str, llm: BaseLLM) -> Agent:
+        return Agent(
+            role=role,
+            goal="Answer questions.",
+            backstory="Test agent.",
+            llm=llm,
+            verbose=False,
+        )
+
+    def test_agents_sharing_one_llm_report_per_call_usage(self):
+        shared = _FixedUsageLLM()
+        r1 = self._make_agent("agent one", shared).kickoff("question one")
+        r2 = self._make_agent("agent two", shared).kickoff("question two")
+
+        assert r1.usage_metrics is not None
+        assert r1.usage_metrics["prompt_tokens"] > 0
+        # The second agent's call must not include the first agent's tokens.
+        assert r2.usage_metrics == r1.usage_metrics
+
+        # The shared LLM instance still exposes cumulative lifetime totals.
+        lifetime = shared.get_token_usage_summary()
+        assert lifetime.prompt_tokens == (
+            r1.usage_metrics["prompt_tokens"] + r2.usage_metrics["prompt_tokens"]
+        )
+        assert lifetime.successful_requests == (
+            r1.usage_metrics["successful_requests"]
+            + r2.usage_metrics["successful_requests"]
+        )
+
+    def test_repeated_kickoffs_on_same_agent_report_per_call_usage(self):
+        agent = self._make_agent("agent", _FixedUsageLLM())
+        r1 = agent.kickoff("question one")
+        r2 = agent.kickoff("question two")
+
+        assert r1.usage_metrics is not None
+        assert r1.usage_metrics["prompt_tokens"] > 0
+        assert r2.usage_metrics == r1.usage_metrics
+
+    @pytest.mark.asyncio
+    async def test_async_kickoff_reports_per_call_usage(self):
+        shared = _FixedUsageLLM()
+        r1 = await self._make_agent("agent one", shared).kickoff_async("question one")
+        r2 = await self._make_agent("agent two", shared).kickoff_async("question two")
+
+        assert r1.usage_metrics is not None
+        assert r1.usage_metrics["prompt_tokens"] > 0
+        assert r2.usage_metrics == r1.usage_metrics
+
+    def test_guardrail_retry_usage_includes_all_attempts(self):
+        """A guardrail retry re-invokes the LLM within the same kickoff, so
+        the result must report the whole call's usage — every attempt — not
+        just the last one."""
+        baseline = (
+            self._make_agent("baseline", _FixedUsageLLM())
+            .kickoff("question one")
+            .usage_metrics
+        )
+
+        attempts: list[str] = []
+
+        def flaky_guardrail(output):
+            attempts.append(output.raw)
+            if len(attempts) == 1:
+                return (False, "Please try again.")
+            return (True, output.raw)
+
+        agent = Agent(
+            role="agent",
+            goal="Answer questions.",
+            backstory="Test agent.",
+            llm=_FixedUsageLLM(),
+            guardrail=flaky_guardrail,
+            verbose=False,
+        )
+        result = agent.kickoff("question one")
+
+        assert len(attempts) == 2
+        assert result.usage_metrics["successful_requests"] == (
+            2 * baseline["successful_requests"]
+        )
+        assert result.usage_metrics["prompt_tokens"] == 2 * baseline["prompt_tokens"]
+        assert result.usage_metrics["total_tokens"] == 2 * baseline["total_tokens"]
+
+
+class TestUsageMetricsDeltaSince:
+    def test_field_wise_difference(self):
+        baseline = UsageMetrics(
+            total_tokens=110,
+            prompt_tokens=100,
+            completion_tokens=10,
+            successful_requests=1,
+        )
+        current = UsageMetrics(
+            total_tokens=330,
+            prompt_tokens=300,
+            completion_tokens=30,
+            cached_prompt_tokens=5,
+            reasoning_tokens=7,
+            cache_creation_tokens=3,
+            successful_requests=3,
+        )
+
+        delta = current.delta_since(baseline)
+
+        assert delta == UsageMetrics(
+            total_tokens=220,
+            prompt_tokens=200,
+            completion_tokens=20,
+            cached_prompt_tokens=5,
+            reasoning_tokens=7,
+            cache_creation_tokens=3,
+            successful_requests=2,
+        )
+
+    def test_clamps_negative_differences_to_zero(self):
+        baseline = UsageMetrics(total_tokens=100, prompt_tokens=90, successful_requests=2)
+        delta = UsageMetrics().delta_since(baseline)
+        assert delta == UsageMetrics()

lib/crewai/tests/hooks/test_dispatch.py

@@ -0,0 +1,296 @@
+"""Unit tests for the generic interception-hook dispatcher.
+
+These cover the new contract (payload-in/payload-out + HookAborted), the shared
+ordered queue between the legacy and new dialects on the four model/tool points,
+execution-scoped hooks, fail-open exception handling, telemetry, and the no-op
+fast-path overhead budget.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import time
+
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.hook_events import HookDispatchedEvent
+from crewai.hooks.dispatch import (
+    HookAborted,
+    InterceptionPoint,
+    clear_all,
+    dispatch,
+    get_hooks,
+    on,
+    register,
+    register_scoped,
+    scoped_hooks,
+    unregister as unregister_hook,
+)
+from crewai.hooks.llm_hooks import (
+    get_before_llm_call_hooks,
+    register_before_llm_call_hook,
+)
+import pytest
+
+
+@dataclass
+class _Ctx:
+    payload: object = None
+    tool_name: str | None = None
+    agent: object = None
+    agent_role: str | None = None
+
+
+@pytest.fixture(autouse=True)
+def clear_dispatch_registry():
+    """Ensure every test starts and ends with an empty global registry."""
+    clear_all()
+    yield
+    clear_all()
+
+
+class TestDispatchContract:
+    """The core payload-in/payload-out + HookAborted contract."""
+
+    def test_noop_fast_path_returns_context_unchanged(self):
+        ctx = _Ctx(payload="original")
+        result = dispatch(InterceptionPoint.INPUT, ctx)
+        assert result is ctx
+        assert ctx.payload == "original"
+
+    def test_return_value_replaces_payload(self):
+        def double(ctx):
+            return ctx.payload * 2
+
+        register(InterceptionPoint.INPUT, double)
+        ctx = _Ctx(payload="ab")
+        dispatch(InterceptionPoint.INPUT, ctx)
+        assert ctx.payload == "abab"
+
+    def test_in_place_mutation_is_honored(self):
+        def mutate(ctx):
+            ctx.payload.append(1)
+            return None
+
+        register(InterceptionPoint.INPUT, mutate)
+        ctx = _Ctx(payload=[])
+        dispatch(InterceptionPoint.INPUT, ctx)
+        assert ctx.payload == [1]
+
+    def test_hooks_run_in_registration_order(self):
+        order: list[int] = []
+        register(InterceptionPoint.INPUT, lambda ctx: order.append(1))
+        register(InterceptionPoint.INPUT, lambda ctx: order.append(2))
+        dispatch(InterceptionPoint.INPUT, _Ctx())
+        assert order == [1, 2]
+
+    def test_hook_aborted_propagates_with_reason_and_source(self):
+        def blocker(ctx):
+            raise HookAborted(reason="nope", source="policy")
+
+        register(InterceptionPoint.INPUT, blocker)
+        with pytest.raises(HookAborted) as exc:
+            dispatch(InterceptionPoint.INPUT, _Ctx())
+        assert exc.value.reason == "nope"
+        assert exc.value.source == "policy"
+
+    def test_ordinary_exception_is_swallowed_and_later_hooks_run(self):
+        ran: list[str] = []
+
+        def boom(ctx):
+            ran.append("boom")
+            raise ValueError("bug in user hook")
+
+        def after(ctx):
+            ran.append("after")
+
+        register(InterceptionPoint.INPUT, boom)
+        register(InterceptionPoint.INPUT, after)
+        dispatch(InterceptionPoint.INPUT, _Ctx(), verbose=False)
+        assert ran == ["boom", "after"]
+
+
+class TestOnDecorator:
+    """The @on decorator registers and filters like the legacy decorators."""
+
+    def test_on_registers_global_hook(self):
+        @on(InterceptionPoint.MEMORY_WRITE)
+        def hook(ctx):
+            return None
+
+        assert hook in get_hooks(InterceptionPoint.MEMORY_WRITE)
+
+    def test_tool_filter_skips_non_matching_tools(self):
+        seen: list[str] = []
+
+        @on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"])
+        def hook(ctx):
+            seen.append(ctx.tool_name)
+
+        dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="other_tool"))
+        dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="allowed_tool"))
+        assert seen == ["allowed_tool"]
+
+    def test_agent_filter_skips_non_matching_agents(self):
+        seen: list[str] = []
+
+        class _Agent:
+            def __init__(self, role):
+                self.role = role
+
+        @on(InterceptionPoint.PRE_MODEL_CALL, agents=["Researcher"])
+        def hook(ctx):
+            seen.append(ctx.agent.role)
+
+        dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Writer")))
+        dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Researcher")))
+        assert seen == ["Researcher"]
+
+    def test_agent_filter_falls_back_to_agent_role(self):
+        seen: list[str] = []
+
+        @on(InterceptionPoint.PRE_STEP, agents=["Researcher"])
+        def hook(ctx):
+            seen.append(ctx.agent_role)
+
+        # No agent object, only the agent_role string (e.g. flow seams).
+        dispatch(InterceptionPoint.PRE_STEP, _Ctx(agent_role="Writer"))
+        dispatch(InterceptionPoint.PRE_STEP, _Ctx(agent_role="Researcher"))
+        assert seen == ["Researcher"]
+
+    def test_unregister_resolves_filtered_wrapper(self):
+        @on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"])
+        def hook(ctx):
+            return None
+
+        assert len(get_hooks(InterceptionPoint.PRE_TOOL_CALL)) == 1
+        assert unregister_hook(InterceptionPoint.PRE_TOOL_CALL, hook) is True
+        assert get_hooks(InterceptionPoint.PRE_TOOL_CALL) == []
+
+
+class TestSharedQueueWithLegacyDialect:
+    """Legacy registrations and @on hooks compose in one ordered queue."""
+
+    def test_on_and_legacy_share_pre_model_call_queue(self):
+        def legacy(ctx):
+            return None
+
+        @on(InterceptionPoint.PRE_MODEL_CALL)
+        def modern(ctx):
+            return None
+
+        register_before_llm_call_hook(legacy)
+
+        queue = get_before_llm_call_hooks()
+        assert modern in queue
+        assert legacy in queue
+        # registration order preserved: modern registered before legacy
+        assert queue.index(modern) < queue.index(legacy)
+
+
+class TestScopedHooks:
+    """Execution-scoped hooks run after globals and are discarded on exit."""
+
+    def test_scoped_runs_after_global_then_cleared(self):
+        order: list[str] = []
+        register(InterceptionPoint.OUTPUT, lambda ctx: order.append("global"))
+
+        with scoped_hooks():
+            register_scoped(InterceptionPoint.OUTPUT, lambda ctx: order.append("scoped"))
+            dispatch(InterceptionPoint.OUTPUT, _Ctx())
+
+        # outside the scope the scoped hook is gone
+        dispatch(InterceptionPoint.OUTPUT, _Ctx())
+
+        assert order == ["global", "scoped", "global"]
+
+
+class TestTelemetry:
+    """dispatch emits a HookDispatchedEvent only when hooks ran."""
+
+    def test_no_event_on_empty_fast_path(self):
+        events: list[HookDispatchedEvent] = []
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(HookDispatchedEvent)
+            def _capture(_source, event):
+                events.append(event)
+
+            dispatch(InterceptionPoint.INPUT, _Ctx())
+
+        assert events == []
+
+    def test_event_reports_outcome(self):
+        events: list[HookDispatchedEvent] = []
+
+        register(InterceptionPoint.INPUT, lambda ctx: "changed")
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(HookDispatchedEvent)
+            def _capture(_source, event):
+                events.append(event)
+
+            dispatch(InterceptionPoint.INPUT, _Ctx())
+            # Telemetry handlers run on the bus's thread pool; flush so the
+            # assertion doesn't race the emit.
+            crewai_event_bus.flush()
+
+        assert len(events) == 1
+        assert events[0].interception_point == "input"
+        assert events[0].outcome == "modified"
+        assert events[0].hook_count == 1
+
+    def test_event_reports_abort_outcome(self):
+        events: list[HookDispatchedEvent] = []
+
+        def blocker(ctx):
+            raise HookAborted(reason="blocked", source="policy")
+
+        register(InterceptionPoint.INPUT, blocker)
+
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(HookDispatchedEvent)
+            def _capture(_source, event):
+                events.append(event)
+
+            with pytest.raises(HookAborted):
+                dispatch(InterceptionPoint.INPUT, _Ctx())
+            crewai_event_bus.flush()
+
+        assert len(events) == 1
+        assert events[0].interception_point == "input"
+        assert events[0].outcome == "aborted"
+        assert events[0].abort_reason == "blocked"
+        assert events[0].abort_source == "policy"
+
+
+class TestNoOpOverhead:
+    """The no-op fast path must stay cheap (a single dict lookup)."""
+
+    def test_noop_dispatch_overhead_is_bounded(self):
+        # Relative (not absolute) budget: the no-op fast path is a dict lookup
+        # plus a guard, so it should stay within a wide multiple of a bare
+        # function call. This catches accidental O(n) regressions without
+        # depending on absolute timing on shared CI runners.
+        ctx = _Ctx()
+        iterations = 100_000
+
+        def _baseline(_c):
+            return _c
+
+        for _ in range(1000):  # warm up both paths
+            dispatch(InterceptionPoint.INPUT, ctx)
+            _baseline(ctx)
+
+        start = time.perf_counter()
+        for _ in range(iterations):
+            _baseline(ctx)
+        baseline = time.perf_counter() - start
+
+        start = time.perf_counter()
+        for _ in range(iterations):
+            dispatch(InterceptionPoint.INPUT, ctx)
+        noop = time.perf_counter() - start
+
+        assert noop < baseline * 50 + 5e-3

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -0,0 +1,178 @@
+"""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, router, 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"
+
+
+class TestFlowStepPoints:
+    """pre_step / post_step for flow methods (kind=flow_method)."""
+
+    def test_pre_and_post_step_fire_per_method(self):
+        kinds: list[tuple[str, str | None]] = []
+
+        @on(InterceptionPoint.PRE_STEP)
+        def pre(ctx):
+            kinds.append(("pre", ctx.step_name))
+
+        @on(InterceptionPoint.POST_STEP)
+        def post(ctx):
+            kinds.append(("post", ctx.step_name))
+
+        _SimpleFlow().kickoff()
+
+        assert ("pre", "begin") in kinds
+        assert ("post", "begin") in kinds
+        assert ("pre", "finish") in kinds
+        assert ("post", "finish") in kinds
+
+    def test_post_step_can_rewrite_method_output(self):
+        @on(InterceptionPoint.POST_STEP)
+        def rewrite(ctx):
+            if ctx.step_name == "finish":
+                return "rewritten"
+            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

lib/crewai/tests/llms/openai/test_openai.py

@@ -30,10 +30,156 @@ def test_openai_completion_is_used_when_no_provider_prefix():
     llm = LLM(model="gpt-4o")
 
     from crewai.llms.providers.openai.completion import OpenAICompletion
-    assert isinstance(llm, OpenAICompletion)
+    assert llm.__class__.__name__ == "OpenAICompletion"
     assert llm.provider == "openai"
     assert llm.model == "gpt-4o"
 
+
+def test_custom_openai_flag_uses_native_openai_without_provider_prefix():
+    """Custom OpenAI-compatible endpoints can serve arbitrary model ids."""
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
+        llm = LLM(
+            model="anthropic/claude-sonnet-4-6",
+            custom_openai=True,
+            base_url="https://gateway.example/v1",
+            is_litellm=False,
+        )
+
+    assert llm.__class__.__name__ == "OpenAICompletion"
+    assert llm.is_litellm is False
+    assert llm.provider == "openai"
+    assert llm.model == "anthropic/claude-sonnet-4-6"
+    assert llm.base_url == "https://gateway.example/v1"
+    assert llm.custom_openai is True
+    assert "custom_openai" not in llm.additional_params
+
+    config = llm.to_config_dict()
+    assert config["model"] == "anthropic/claude-sonnet-4-6"
+    assert config["custom_openai"] is True
+    assert config["base_url"] == "https://gateway.example/v1"
+
+    rebuilt = LLM(**config)
+    assert isinstance(rebuilt, OpenAICompletion)
+    assert rebuilt.model == "anthropic/claude-sonnet-4-6"
+    assert rebuilt.base_url == "https://gateway.example/v1"
+
+
+def test_custom_openai_flag_requires_custom_base_url():
+    """Avoid routing arbitrary custom model ids to api.openai.com by mistake."""
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=True):
+        with pytest.raises(ValueError, match="custom_openai=True requires"):
+            LLM(
+                model="anthropic/claude-sonnet-4-6",
+                custom_openai=True,
+                is_litellm=False,
+            )
+
+
+def test_direct_custom_openai_completion_requires_custom_base_url():
+    """Direct construction must not silently fall back to api.openai.com."""
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=True):
+        with pytest.raises(ValueError, match="custom_openai=True requires"):
+            OpenAICompletion(
+                model="anthropic/claude-sonnet-4-6",
+                custom_openai=True,
+            )
+
+
+def test_custom_openai_flag_strips_openai_routing_prefix():
+    """The openai/ routing prefix is not part of the gateway's model id."""
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
+        llm = LLM(
+            model="openai/anthropic/claude-sonnet-4-6",
+            custom_openai=True,
+            base_url="https://gateway.example/v1",
+            is_litellm=False,
+        )
+
+    assert isinstance(llm, OpenAICompletion)
+    assert llm.model == "anthropic/claude-sonnet-4-6"
+
+
+def test_openai_prefixed_custom_endpoint_uses_native_sdk_for_nested_model_id():
+    """Custom OpenAI-compatible endpoints may serve non-OpenAI model ids."""
+    with patch.dict(os.environ, {"OPENAI_API_KEY": "test-key"}, clear=False):
+        llm = LLM(
+            model="openai/anthropic/claude-sonnet-4-6",
+            base_url="https://gateway.example/v1",
+            is_litellm=False,
+        )
+
+    assert llm.__class__.__name__ == "OpenAICompletion"
+    assert llm.is_litellm is False
+    assert llm.provider == "openai"
+    assert llm.model == "anthropic/claude-sonnet-4-6"
+    assert llm.custom_openai is True
+    assert llm.base_url == "https://gateway.example/v1"
+
+def test_explicit_custom_openai_uses_legacy_api_base_env_var():
+    """Explicit custom routing supports the legacy endpoint environment variable."""
+    with patch.dict(
+        os.environ,
+        {
+            "OPENAI_API_KEY": "test-key",
+            "OPENAI_API_BASE": "https://gateway.example/v1",
+        },
+        clear=False,
+    ):
+        os.environ.pop("OPENAI_BASE_URL", None)
+        llm = LLM(
+            model="openai/anthropic/claude-sonnet-4-6",
+            custom_openai=True,
+            is_litellm=False,
+        )
+
+    assert isinstance(llm, OpenAICompletion)
+    assert llm.is_litellm is False
+    assert llm.provider == "openai"
+    assert llm.model == "anthropic/claude-sonnet-4-6"
+    assert llm.custom_openai is True
+
+
+def test_openai_prefixed_unknown_model_ignores_ambient_base_url_for_routing():
+    """Ambient OpenAI configuration must not opt unknown models into native routing."""
+    with patch.dict(
+        os.environ,
+        {
+            "OPENAI_API_KEY": "test-key",
+            "OPENAI_BASE_URL": "https://gateway.example/v1",
+        },
+        clear=True,
+    ):
+        with (
+            patch("crewai.llm._ensure_litellm", return_value=False),
+            pytest.raises(ImportError, match="LiteLLM fallback package"),
+        ):
+            LLM(model="openai/not-a-real-openai-model")
+
+
+@pytest.mark.parametrize("endpoint_field", ["api_base", "env"])
+def test_custom_openai_config_preserves_resolved_endpoint(endpoint_field):
+    """Serialized custom OpenAI configs can reconstruct the same endpoint."""
+    kwargs = {}
+    env = {"OPENAI_API_KEY": "test-key"}
+    if endpoint_field == "api_base":
+        kwargs["api_base"] = "https://gateway.example/v1"
+    else:
+        env["OPENAI_API_BASE"] = "https://gateway.example/v1"
+
+    with patch.dict(os.environ, env, clear=True):
+        llm = LLM(
+            model="anthropic/claude-sonnet-4-6",
+            custom_openai=True,
+            **kwargs,
+        )
+
+    config = llm.to_config_dict()
+    assert config["base_url"] == "https://gateway.example/v1"
+    rebuilt = LLM(**config)
+    assert isinstance(rebuilt, OpenAICompletion)
+    assert rebuilt.base_url == "https://gateway.example/v1"
+
+
 @pytest.mark.vcr()
 def test_openai_is_default_provider_without_explicit_llm_set_on_agent():
     """
@@ -60,14 +206,13 @@ def test_openai_is_default_provider_without_explicit_llm_set_on_agent():
 
 
 
-def test_openai_completion_module_is_imported():
+def test_openai_completion_module_is_imported(monkeypatch):
     """
     Test that the completion module is properly imported when using OpenAI provider
     """
     module_name = "crewai.llms.providers.openai.completion"
 
-    if module_name in sys.modules:
-        del sys.modules[module_name]
+    monkeypatch.delitem(sys.modules, module_name, raising=False)
 
     LLM(model="gpt-4o")
 
@@ -421,12 +566,25 @@ def test_openai_get_client_params_with_env_var():
         client_params = llm._get_client_params()
         assert client_params["base_url"] == "https://env.openai.com/v1"
 
+def test_openai_get_client_params_with_legacy_api_base_env_var():
+    """
+    Test that _get_client_params uses OPENAI_API_BASE when OPENAI_BASE_URL is absent.
+    """
+    with patch.dict(os.environ, {
+        "OPENAI_API_BASE": "https://legacy-env.openai.com/v1",
+    }, clear=False):
+        os.environ.pop("OPENAI_BASE_URL", None)
+        llm = OpenAICompletion(model="gpt-4o")
+        client_params = llm._get_client_params()
+        assert client_params["base_url"] == "https://legacy-env.openai.com/v1"
+
 def test_openai_get_client_params_priority_order():
     """
-    Test the priority order: base_url > api_base > OPENAI_BASE_URL env var
+    Test the priority order: base_url > api_base > OPENAI_BASE_URL > OPENAI_API_BASE
     """
     with patch.dict(os.environ, {
         "OPENAI_BASE_URL": "https://env.openai.com/v1",
+        "OPENAI_API_BASE": "https://legacy-env.openai.com/v1",
     }):
         llm1 = OpenAICompletion(
             model="gpt-4o",

lib/crewai/tests/test_crew.py

@@ -859,6 +859,7 @@ def test_cache_hitting_between_agents(researcher, writer, ceo):
     crew = Crew(
         agents=[ceo, researcher],
         tasks=tasks,
+        cache=True,
     )
 
     with patch.object(CacheHandler, "read") as read:
@@ -2246,7 +2247,9 @@ def test_tools_with_custom_caching():
         agent=writer2,
     )
 
-    crew = Crew(agents=[writer1, writer2], tasks=[task1, task2, task3, task4])
+    crew = Crew(
+        agents=[writer1, writer2], tasks=[task1, task2, task3, task4], cache=True
+    )
 
     with patch.object(
         CacheHandler, "add", wraps=crew._cache_handler.add
@@ -4598,6 +4601,98 @@ def test_reset_memory_uses_full_unified_memory_reset(researcher):
     reset.assert_not_called()
 
 
+def test_kickoff_drains_pending_memory_saves_before_completion_event(researcher):
+    """Background memory saves must finish (and emit their completion events)
+    before CrewKickoffCompletedEvent, otherwise listeners that tear down on
+    kickoff-completed (e.g. telemetry sessions) see the save span as orphaned."""
+    import time
+
+    from crewai.events.types.crew_events import CrewKickoffCompletedEvent
+
+    order: list[str] = []
+
+    def slow_save():
+        time.sleep(0.3)
+        order.append("save-done")
+
+    crew = Crew(
+        agents=[researcher],
+        process=Process.sequential,
+        tasks=[
+            Task(description="Task 1", expected_output="output", agent=researcher),
+        ],
+        memory=True,
+        task_callback=lambda _output: crew._memory._submit_save(slow_save),
+    )
+
+    completed = threading.Event()
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(CrewKickoffCompletedEvent)
+        def on_completed(_source, _event):
+            order.append("kickoff-completed")
+            completed.set()
+
+        with patch.object(Agent, "execute_task", return_value="ok"):
+            crew.kickoff()
+
+        assert completed.wait(timeout=5)
+
+    assert order.index("save-done") < order.index("kickoff-completed")
+
+
+def test_kickoff_drains_agent_memory_saves_before_completion_event(tmp_path):
+    """Agents save through their own ``agent.memory`` when set; those pools
+    must also be drained before CrewKickoffCompletedEvent."""
+    import time
+
+    from crewai.events.types.crew_events import CrewKickoffCompletedEvent
+
+    agent_memory = Memory(storage=str(tmp_path / "agent-mem"))
+    agent_with_memory = Agent(
+        role="Researcher",
+        goal="Research things",
+        backstory="Experienced researcher",
+        memory=agent_memory,
+    )
+
+    order: list[str] = []
+
+    def slow_save():
+        time.sleep(0.3)
+        order.append("save-done")
+
+    crew = Crew(
+        agents=[agent_with_memory],
+        process=Process.sequential,
+        tasks=[
+            Task(
+                description="Task 1",
+                expected_output="output",
+                agent=agent_with_memory,
+            ),
+        ],
+        task_callback=lambda _output: agent_memory._submit_save(slow_save),
+    )
+
+    completed = threading.Event()
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(CrewKickoffCompletedEvent)
+        def on_completed(_source, _event):
+            order.append("kickoff-completed")
+            completed.set()
+
+        with patch.object(Agent, "execute_task", return_value="ok"):
+            crew.kickoff()
+
+        assert completed.wait(timeout=5)
+
+    assert order.index("save-done") < order.index("kickoff-completed")
+
+
 def test_reset_knowledge_with_only_crew_knowledge(researcher, writer):
     mock_ks = MagicMock(spec=Knowledge)
 

lib/crewai/tests/test_flow.py

@@ -2353,3 +2353,41 @@ def test_locked_dict_proxy_ior():
 def test_locked_dict_proxy_reversed():
     flow = _make_dict_flow()
     assert list(reversed(flow.state.data)) == ["c", "b", "a"]
+
+
+def test_flow_drains_pending_memory_saves_before_finished_event(tmp_path):
+    """Background memory saves must finish (and emit their completion events)
+    before FlowFinishedEvent, otherwise listeners that tear down on
+    flow-finished (e.g. telemetry sessions) see the save span as orphaned."""
+    import time
+
+    from crewai.memory.unified_memory import Memory
+
+    order: list[str] = []
+
+    def slow_save():
+        time.sleep(0.3)
+        order.append("save-done")
+
+    class MemoryFlow(Flow):
+        @start()
+        def step_1(self):
+            self.memory._submit_save(slow_save)
+            return "done"
+
+    flow = MemoryFlow(memory=Memory(storage=str(tmp_path / "flow-mem")))
+
+    finished = threading.Event()
+
+    with crewai_event_bus.scoped_handlers():
+
+        @crewai_event_bus.on(FlowFinishedEvent)
+        def on_finished(_source, _event):
+            order.append("flow-finished")
+            finished.set()
+
+        flow.kickoff()
+
+        assert finished.wait(timeout=5)
+
+    assert order.index("save-done") < order.index("flow-finished")

lib/crewai/tests/test_flow_conversation.py

@@ -1555,6 +1555,180 @@ class TestConversationalFlow:
         )
 
 
+class TestHandleTurnReplyFallback:
+    """Regression tests for EPD-181: ``handle_turn()`` decided "did the
+    handler append its reply?" by comparing assistant-message counts. A
+    handler that appends its reply AND trims history to a cap left the count
+    unchanged, so the fallback appended the reply a second time — every turn,
+    once trimming engaged. The check now uses an explicit appended-this-turn
+    flag.
+    """
+
+    MAX_MESSAGES = 4
+
+    def _make_bot(self) -> ConversationalFlow:
+        max_messages = self.MAX_MESSAGES
+
+        class EchoBot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "ECHO"
+
+            @listen("ECHO")
+            def echo(self) -> str:
+                reply = f"echo: {self.state.current_user_message or ''}"
+                self.append_assistant_message(reply)  # handler DOES append
+                if len(self.state.messages) > max_messages:  # ...and trims
+                    self.state.messages = self.state.messages[-max_messages:]
+                return reply
+
+        return EchoBot()
+
+    def test_no_duplicate_reply_when_handler_trims_history(self) -> None:
+        bot = self._make_bot()
+        for i in range(1, 5):
+            bot.handle_turn(f"message {i}")
+            contents = [message.content for message in bot.state.messages]
+            assert len(contents) == len(set(contents)), (
+                f"duplicate reply on turn {i}: {contents}"
+            )
+
+        # The capped window holds the last two full turns, in order.
+        assert [message.content for message in bot.state.messages] == [
+            "message 3",
+            "echo: message 3",
+            "message 4",
+            "echo: message 4",
+        ]
+
+    def test_fallback_still_appends_when_handler_does_not_reply(self) -> None:
+        class SilentBot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                return "WORK"
+
+            @listen("WORK")
+            def work(self) -> str:
+                return "computed reply"  # returns without appending
+
+        bot = SilentBot()
+        bot.handle_turn("hello")
+
+        assistant_messages = [
+            message.content
+            for message in bot.state.messages
+            if message.role == "assistant"
+        ]
+        assert assistant_messages == ["computed reply"]
+
+
+class TestFalsyRouteTurnFallback:
+    """A falsy ``route_turn()`` must never replay a previous turn's intent.
+
+    Regression tests for EPD-176: an overridden ``route_turn()`` returning
+    ``None`` on an unhandled input used to silently reuse the sticky
+    ``state.last_intent`` from the *previous* turn, running the wrong handler
+    with no error or warning.
+    """
+
+    def test_falsy_route_turn_does_not_replay_previous_turns_intent(self) -> None:
+        ran: list[str] = []
+
+        class Bot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                message = context.get("current_user_message") or ""
+                if "hello" in message.lower():
+                    return "GREETING"
+                return None  # unhandled input -> falsy return
+
+            @listen("GREETING")
+            def greeting(self) -> str:
+                ran.append("GREETING")
+                reply = "Hi! I only do greetings."
+                self.append_assistant_message(reply)
+                return reply
+
+            @listen("WEATHER")
+            def weather(self) -> str:
+                ran.append("WEATHER")
+                reply = "It is sunny."
+                self.append_assistant_message(reply)
+                return reply
+
+        flow = Bot()
+        flow.handle_turn("hello there")
+        assert ran == ["GREETING"]
+        assert flow.state.last_intent == "GREETING"
+
+        flow.handle_turn("what is the meaning of life?")
+        assert ran == ["GREETING"], (
+            "an unhandled turn must not re-run the previous turn's handler"
+        )
+        # With no routing decision the turn falls through to the built-in
+        # 'converse' default instead of replaying the stale intent.
+        assert flow.state.last_intent == "converse"
+        assert flow.state.messages[-1].content != "Hi! I only do greetings."
+
+    def test_stale_intent_ignored_but_route_selected_event_still_emitted(
+        self,
+    ) -> None:
+        class Bot(ConversationalFlow):
+            def route_turn(self, context: dict[str, Any]) -> str | None:
+                message = context.get("current_user_message") or ""
+                return "work" if "work" in message else None
+
+            @listen("work")
+            def do_work(self) -> str:
+                self.append_assistant_message("worked")
+                return "worked"
+
+        flow = Bot()
+        routes: list[ConversationRouteSelectedEvent] = []
+        with crewai_event_bus.scoped_handlers():
+
+            @crewai_event_bus.on(ConversationRouteSelectedEvent)
+            def capture(_: Any, event: ConversationRouteSelectedEvent) -> None:
+                routes.append(event)
+
+            flow.handle_turn("work please")
+            flow.handle_turn("something unrelated")
+            crewai_event_bus.flush()
+
+        assert [event.route for event in routes] == ["work", "converse"]
+        # The fallback decision still reports the prior intent for visibility.
+        assert routes[1].previous_intent == "work"
+
+    def test_fresh_intent_classified_this_turn_still_routes(self) -> None:
+        """The legacy ``default_intents`` path classifies per turn and must
+        keep routing on the freshly classified intent — including when the
+        intent changes between turns."""
+        ran: list[str] = []
+
+        @ConversationConfig(
+            default_intents=["search", "weather"], intent_llm="gpt-4o-mini"
+        )
+        class LegacyFlow(ConversationalFlow):
+            @listen("search")
+            def handle_search(self) -> str:
+                ran.append("search")
+                self.append_assistant_message("searched")
+                return "searched"
+
+            @listen("weather")
+            def handle_weather(self) -> str:
+                ran.append("weather")
+                self.append_assistant_message("sunny")
+                return "sunny"
+
+        flow = LegacyFlow()
+        with patch.object(
+            flow, "_collapse_to_outcome", side_effect=["search", "weather"]
+        ):
+            flow.handle_turn("look up crewai")
+            flow.handle_turn("how is the weather?")
+
+        assert ran == ["search", "weather"]
+        assert flow.state.last_intent == "weather"
+
+
 class TestFlowTracingWhenSuppressed:
     def test_flow_started_emitted_when_panel_events_suppressed(self) -> None:
         class QuietFlow(Flow[ChatState]):

lib/crewai/tests/test_tool_cache_default.py

@@ -0,0 +1,263 @@
+# mypy: ignore-errors
+"""Regression tests for EPD-180: tool-result caching used to be ON by default,
+so an LLM calling the same tool with identical arguments twice in one run got
+the first (possibly stale) result back without the tool executing — silently
+wrong for live-data tools, and silently dropped actions for stateful tools.
+
+Caching is now opt-in: ``Crew(cache=True)`` for crews, ``Agent(cache=True)``
+(or an explicit ``cache_handler``) for standalone agents. The machinery —
+including per-tool ``cache_function`` write gating — is unchanged once opted
+in.
+
+The end-to-end tests run fully offline: a fake OpenAI client scripts two
+identical tool calls followed by a final answer, mirroring the EPD-180
+clean-room repro.
+"""
+
+from openai.types.chat import ChatCompletion
+from pydantic import BaseModel, Field
+
+from crewai import LLM, Agent, Crew, Task
+from crewai.agents.cache.cache_handler import CacheHandler
+from crewai.tools import BaseTool
+
+
+class LookupArgs(BaseModel):
+    city: str = Field(description="City to look up.")
+
+
+def make_live_tool():
+    """A tool returning a different value on every real execution."""
+    executions = []
+
+    class LiveLookupTool(BaseTool):
+        name: str = "live_lookup"
+        description: str = "Returns a live (time-varying) reading for a city."
+        args_schema: type[BaseModel] = LookupArgs
+        # cache_function deliberately NOT set — exercising the default.
+
+        def _run(self, city: str) -> str:
+            executions.append(city)
+            return f"reading #{len(executions)} for {city}"
+
+    return LiveLookupTool(), executions
+
+
+def make_scripted_llm():
+    """An offline LLM whose client scripts two identical tool calls."""
+
+    def tool_call_response(call_id: str):
+        return {
+            "index": 0,
+            "finish_reason": "tool_calls",
+            "message": {
+                "role": "assistant",
+                "content": None,
+                "tool_calls": [
+                    {
+                        "id": call_id,
+                        "type": "function",
+                        "function": {
+                            "name": "live_lookup",
+                            "arguments": '{"city": "paris"}',
+                        },
+                    }
+                ],
+            },
+        }
+
+    scripted = [
+        tool_call_response("call_1"),
+        tool_call_response("call_2"),  # identical name+args, new id
+        {
+            "index": 0,
+            "finish_reason": "stop",
+            "message": {"role": "assistant", "content": "Final answer: done."},
+        },
+    ]
+
+    class FakeCompletions:
+        def __init__(self):
+            self.n = 0
+
+        def create(self, **params):
+            choice = scripted[min(self.n, len(scripted) - 1)]
+            self.n += 1
+            return ChatCompletion.model_validate(
+                {
+                    "id": f"chatcmpl-fake-{self.n}",
+                    "object": "chat.completion",
+                    "created": 1,
+                    "model": params.get("model", "gpt-4o"),
+                    "choices": [choice],
+                    "usage": {
+                        "prompt_tokens": 10,
+                        "completion_tokens": 5,
+                        "total_tokens": 15,
+                    },
+                }
+            )
+
+    class FakeClient:
+        def __init__(self):
+            self.chat = type("Chat", (), {"completions": FakeCompletions()})()
+
+    llm = LLM(model="openai/gpt-4o")
+    llm._client = FakeClient()
+    return llm
+
+
+def run_crew(**crew_kwargs):
+    tool, executions = make_live_tool()
+    agent = Agent(
+        role="reader",
+        goal="Look things up.",
+        backstory="Test agent.",
+        llm=make_scripted_llm(),
+        tools=[tool],
+        verbose=False,
+    )
+    task = Task(
+        description="Look up paris twice and report.",
+        expected_output="A report.",
+        agent=agent,
+    )
+    crew = Crew(agents=[agent], tasks=[task], verbose=False, **crew_kwargs)
+    crew.kickoff()
+    return executions
+
+
+class TestToolCachingIsOptIn:
+    def test_default_reexecutes_identical_tool_calls(self):
+        """EPD-180: with no opt-in, both identical calls must really execute."""
+        executions = run_crew()
+        assert len(executions) == 2
+
+    def test_crew_cache_true_dedupes_identical_tool_calls(self):
+        """Opting in via Crew(cache=True) restores the dedup behavior."""
+        executions = run_crew(cache=True)
+        assert len(executions) == 1
+
+
+class TestAgentCacheWiring:
+    def _agent(self, **kwargs) -> Agent:
+        return Agent(
+            role="reader",
+            goal="Look things up.",
+            backstory="Test agent.",
+            **kwargs,
+        )
+
+    def test_standalone_agent_has_no_cache_by_default(self):
+        agent = self._agent()
+        assert agent.tools_handler.cache is None
+        assert agent.cache_handler is None
+
+    def test_standalone_agent_explicit_cache_true_opts_in(self):
+        agent = self._agent(cache=True)
+        assert agent.tools_handler.cache is not None
+        assert agent.cache_handler is not None
+
+    def test_standalone_agent_explicit_cache_handler_opts_in(self):
+        handler = CacheHandler()
+        agent = self._agent(cache_handler=handler)
+        assert agent.tools_handler.cache is handler
+
+    def test_explicit_cache_false_stays_off_even_with_handler(self):
+        agent = self._agent(cache=False, cache_handler=CacheHandler())
+        assert agent.tools_handler.cache is None
+
+    def test_agents_accept_a_crew_offered_handler_by_default(self):
+        """``Crew(cache=True)`` offers its handler via set_cache_handler at
+        kickoff; agents that didn't explicitly opt out must accept it."""
+        agent = self._agent()
+        assert agent.tools_handler.cache is None
+
+        handler = CacheHandler()
+        agent.set_cache_handler(handler)
+        assert agent.tools_handler.cache is handler
+
+    def test_agents_that_opted_out_refuse_a_crew_offered_handler(self):
+        agent = self._agent(cache=False)
+        agent.set_cache_handler(CacheHandler())
+        assert agent.tools_handler.cache is None
+
+    def test_copy_of_default_agent_does_not_opt_in(self):
+        """copy() rebuilds from model_dump(), which includes the field
+        default cache=True — that must not read as an explicit opt-in on
+        the copy (Bugbot review finding on the original PR)."""
+        copied = self._agent().copy()
+        assert copied.tools_handler.cache is None
+        assert copied.cache_handler is None
+
+    def test_copy_of_opted_in_agent_stays_opted_in(self):
+        copied = self._agent(cache=True).copy()
+        assert copied.tools_handler.cache is not None
+
+    def test_copy_of_handler_opted_in_agent_stays_opted_in(self):
+        """An explicit cache_handler is an opt-in too; copy() excludes the
+        handler itself, but the consent must survive — the copy wires its
+        own fresh handler (Bugbot review finding on the original PR)."""
+        source = self._agent(cache_handler=CacheHandler())
+        copied = source.copy()
+        assert copied.tools_handler.cache is not None
+        assert copied.tools_handler.cache is not source.tools_handler.cache
+
+    def test_copy_of_explicit_cache_false_with_handler_stays_off(self):
+        copied = self._agent(cache=False, cache_handler=CacheHandler()).copy()
+        assert copied.tools_handler.cache is None
+
+    def test_copy_of_crew_wired_agent_does_not_opt_in(self):
+        """A handler offered by a crew at kickoff (set_cache_handler) is
+        runtime wiring, not construction-time consent — copies of such
+        agents must not become standalone cachers (Bugbot review finding
+        on the original PR)."""
+        agent = self._agent()
+        agent.set_cache_handler(CacheHandler())  # what Crew(cache=True) does
+        assert agent.tools_handler.cache is not None
+
+        copied = agent.copy()
+        assert copied.tools_handler.cache is None
+        assert copied.cache_handler is None
+
+
+class TestHierarchicalManagerCacheWiring:
+    """The auto-created hierarchical manager is built outside the agents
+    loop that offers the crew's cache handler; an opted-in crew must wire
+    the manager too (Bugbot review finding on the original PR)."""
+
+    def _crew(self, **crew_kwargs) -> Crew:
+        from crewai.process import Process
+
+        agent = Agent(role="worker", goal="Do work.", backstory="Test agent.")
+        task = Task(description="Do the work.", expected_output="Done.")
+        return Crew(
+            agents=[agent],
+            tasks=[task],
+            process=Process.hierarchical,
+            manager_llm="gpt-4o",
+            **crew_kwargs,
+        )
+
+    def test_manager_gets_crew_handler_when_cache_enabled(self):
+        crew = self._crew(cache=True)
+        crew._create_manager_agent()
+        assert crew.manager_agent.tools_handler.cache is crew._cache_handler
+
+    def test_manager_has_no_cache_when_crew_did_not_opt_in(self):
+        crew = self._crew()
+        crew._create_manager_agent()
+        assert crew.manager_agent.tools_handler.cache is None
+
+    def test_user_provided_manager_with_cache_false_stays_excluded(self):
+        manager = Agent(
+            role="manager",
+            goal="Manage.",
+            backstory="Test manager.",
+            cache=False,
+            allow_delegation=True,
+        )
+        crew = self._crew(cache=True)
+        crew.manager_agent = manager
+        crew._create_manager_agent()
+        assert manager.tools_handler.cache is None

lib/crewai/tests/test_usage_shape_parity.py

@@ -0,0 +1,143 @@
+# mypy: ignore-errors
+"""Regression tests for EPD-178: token usage was exposed in different shapes
+and attribute names per code path — ``Agent.kickoff()`` results carried a
+plain dict at ``.usage_metrics`` (no ``token_usage`` attribute at all), while
+``Crew.kickoff()`` results carried a ``UsageMetrics`` object at
+``.token_usage`` (no ``usage_metrics`` attribute), so any single accessor
+written for one path raised ``AttributeError`` on the other.
+
+Both result types now expose both surfaces: ``.token_usage`` as a
+``UsageMetrics`` object and ``.usage_metrics`` as a plain dict.
+"""
+
+from crewai import Agent, Crew, Task
+from crewai.crews.crew_output import CrewOutput
+from crewai.lite_agent_output import LiteAgentOutput
+from crewai.llms.base_llm import BaseLLM
+from crewai.types.usage_metrics import UsageMetrics
+
+
+class _FixedUsageLLM(BaseLLM):
+    """Offline BaseLLM that records fixed usage (100/10 tokens) per call."""
+
+    def __init__(self):
+        super().__init__(model="fixed-usage-model")
+
+    def call(
+        self,
+        messages,
+        tools=None,
+        callbacks=None,
+        available_functions=None,
+        from_task=None,
+        from_agent=None,
+        response_model=None,
+    ) -> str:
+        self._track_token_usage_internal(
+            {"prompt_tokens": 100, "completion_tokens": 10, "total_tokens": 110}
+        )
+        return "Thought: I know the answer.\nFinal Answer: fake answer"
+
+    def supports_function_calling(self) -> bool:
+        return False
+
+    def supports_stop_words(self) -> bool:
+        return False
+
+    def get_context_window_size(self) -> int:
+        return 4096
+
+
+class TestUsageShapeUnitParity:
+    def test_lite_agent_output_exposes_token_usage_object(self):
+        metrics = UsageMetrics(
+            total_tokens=110,
+            prompt_tokens=100,
+            completion_tokens=10,
+            successful_requests=1,
+        )
+        output = LiteAgentOutput(
+            agent_role="analyst", usage_metrics=metrics.model_dump()
+        )
+
+        assert output.token_usage == metrics
+        assert isinstance(output.token_usage, UsageMetrics)
+
+    def test_lite_agent_output_token_usage_zeroed_when_absent(self):
+        output = LiteAgentOutput(agent_role="analyst")
+
+        assert output.usage_metrics is None
+        assert output.token_usage == UsageMetrics()
+
+    def test_crew_output_exposes_usage_metrics_dict(self):
+        metrics = UsageMetrics(
+            total_tokens=110,
+            prompt_tokens=100,
+            completion_tokens=10,
+            successful_requests=1,
+        )
+        output = CrewOutput(token_usage=metrics)
+
+        assert output.usage_metrics == metrics.model_dump()
+        assert isinstance(output.usage_metrics, dict)
+
+    def test_both_shapes_carry_identical_keys(self):
+        """The dict shape has exactly the UsageMetrics fields on both types."""
+        crew_dict = CrewOutput(token_usage=UsageMetrics()).usage_metrics
+        lite = LiteAgentOutput(
+            agent_role="analyst", usage_metrics=UsageMetrics().model_dump()
+        )
+
+        assert set(crew_dict) == set(UsageMetrics.model_fields)
+        assert set(lite.usage_metrics) == set(UsageMetrics.model_fields)
+
+
+class TestUsageShapeEndToEnd:
+    """Mirror of the EPD-178 clean-room repro, offline via a fake BaseLLM."""
+
+    @staticmethod
+    def _read_via_object(result) -> int:
+        """Single accessor written against the CrewOutput shape."""
+        return result.token_usage.prompt_tokens
+
+    @staticmethod
+    def _read_via_dict(result) -> int:
+        """Single accessor written against the LiteAgentOutput shape."""
+        return result.usage_metrics["prompt_tokens"]
+
+    def test_single_accessor_works_on_both_kickoff_paths(self):
+        agent_a = Agent(
+            role="analyst",
+            goal="Answer questions.",
+            backstory="Test agent.",
+            llm=_FixedUsageLLM(),
+            verbose=False,
+        )
+        result_agent = agent_a.kickoff("a question")
+
+        agent_b = Agent(
+            role="analyst",
+            goal="Answer questions.",
+            backstory="Test agent.",
+            llm=_FixedUsageLLM(),
+            verbose=False,
+        )
+        task = Task(
+            description="Answer: a question",
+            expected_output="A short answer.",
+            agent=agent_b,
+        )
+        crew = Crew(agents=[agent_b], tasks=[task], verbose=False)
+        result_crew = crew.kickoff()
+
+        assert isinstance(result_agent, LiteAgentOutput)
+        assert isinstance(result_crew, CrewOutput)
+
+        # Both accessors work on both result types and agree with each other.
+        for result in (result_agent, result_crew):
+            object_read = self._read_via_object(result)
+            dict_read = self._read_via_dict(result)
+            assert object_read == dict_read
+            assert object_read > 0
+            assert isinstance(result.token_usage, UsageMetrics)
+            assert isinstance(result.usage_metrics, dict)

lib/crewai/tests/tools/test_base_tool.py

@@ -18,11 +18,16 @@ def test_creating_a_tool_using_annotation():
         return question
 
     assert my_tool.name == "Name of my tool"
-    assert "Tool Name: name_of_my_tool" in my_tool.description
-    assert "Tool Arguments:" in my_tool.description
-    assert '"question"' in my_tool.description
-    assert '"type": "string"' in my_tool.description
-    assert "Tool Description: Clear description for what this tool is useful for" in my_tool.description
+    # The authored description is preserved as written; the LLM-facing
+    # composite lives at formatted_description.
+    assert my_tool.description == (
+        "Clear description for what this tool is useful for, your agent will need this information to use it."
+    )
+    assert "Tool Name: name_of_my_tool" in my_tool.formatted_description
+    assert "Tool Arguments:" in my_tool.formatted_description
+    assert '"question"' in my_tool.formatted_description
+    assert '"type": "string"' in my_tool.formatted_description
+    assert "Tool Description: Clear description for what this tool is useful for" in my_tool.formatted_description
     assert my_tool.args_schema.model_json_schema()["properties"] == {
         "question": {"title": "Question", "type": "string"}
     }
@@ -33,9 +38,10 @@ def test_creating_a_tool_using_annotation():
     converted_tool = my_tool.to_structured_tool()
     assert converted_tool.name == "Name of my tool"
 
-    assert "Tool Name: name_of_my_tool" in converted_tool.description
-    assert "Tool Arguments:" in converted_tool.description
-    assert '"question"' in converted_tool.description
+    assert converted_tool.description == my_tool.description
+    assert "Tool Name: name_of_my_tool" in converted_tool.formatted_description
+    assert "Tool Arguments:" in converted_tool.formatted_description
+    assert '"question"' in converted_tool.formatted_description
     assert converted_tool.args_schema.model_json_schema()["properties"] == {
         "question": {"title": "Question", "type": "string"}
     }
@@ -56,11 +62,16 @@ def test_creating_a_tool_using_baseclass():
     my_tool = MyCustomTool()
     assert my_tool.name == "Name of my tool"
 
-    assert "Tool Name: name_of_my_tool" in my_tool.description
-    assert "Tool Arguments:" in my_tool.description
-    assert '"question"' in my_tool.description
-    assert '"type": "string"' in my_tool.description
-    assert "Tool Description: Clear description for what this tool is useful for" in my_tool.description
+    # The authored description is preserved as written; the LLM-facing
+    # composite lives at formatted_description.
+    assert my_tool.description == (
+        "Clear description for what this tool is useful for, your agent will need this information to use it."
+    )
+    assert "Tool Name: name_of_my_tool" in my_tool.formatted_description
+    assert "Tool Arguments:" in my_tool.formatted_description
+    assert '"question"' in my_tool.formatted_description
+    assert '"type": "string"' in my_tool.formatted_description
+    assert "Tool Description: Clear description for what this tool is useful for" in my_tool.formatted_description
     assert my_tool.args_schema.model_json_schema()["properties"] == {
         "question": {"title": "Question", "type": "string"}
     }
@@ -69,9 +80,10 @@ def test_creating_a_tool_using_baseclass():
     converted_tool = my_tool.to_structured_tool()
     assert converted_tool.name == "Name of my tool"
 
-    assert "Tool Name: name_of_my_tool" in converted_tool.description
-    assert "Tool Arguments:" in converted_tool.description
-    assert '"question"' in converted_tool.description
+    assert converted_tool.description == my_tool.description
+    assert "Tool Name: name_of_my_tool" in converted_tool.formatted_description
+    assert "Tool Arguments:" in converted_tool.formatted_description
+    assert '"question"' in converted_tool.formatted_description
     assert converted_tool.args_schema.model_json_schema()["properties"] == {
         "question": {"title": "Question", "type": "string"}
     }
@@ -695,3 +707,88 @@ class TestToolDecoratorArunValidation:
 
         with pytest.raises(ValueError, match="validation failed"):
             await async_execute.arun(wrong_arg="value")
+
+
+class TestAuthoredDescriptionPreserved:
+    """Regression tests for EPD-179: BaseTool.model_post_init silently
+    rewrote the authored ``description`` into the LLM-facing composite
+    (``Tool Name: …\\nTool Arguments: …\\nTool Description: <authored>``).
+    The authored field must survive construction as written, with the
+    composite exposed separately at ``formatted_description``.
+    """
+
+    AUTHORED = "Returns the current temperature for a city."
+
+    def _make_tool(self) -> BaseTool:
+        class TempArgs(BaseModel):
+            city: str = Field(description="City name to look up.")
+
+        class TempTool(BaseTool):
+            name: str = "get_temperature"
+            description: str = TestAuthoredDescriptionPreserved.AUTHORED
+            args_schema: type[BaseModel] = TempArgs
+
+            def _run(self, city: str) -> str:
+                return f"22C in {city}"
+
+        return TempTool()
+
+    def test_description_equals_authored_text(self):
+        tool_instance = self._make_tool()
+        assert tool_instance.description == self.AUTHORED
+
+    def test_formatted_description_contains_composite(self):
+        tool_instance = self._make_tool()
+        formatted = tool_instance.formatted_description
+        assert "Tool Name: get_temperature" in formatted
+        assert "Tool Arguments:" in formatted
+        assert '"city"' in formatted
+        assert formatted.endswith(f"Tool Description: {self.AUTHORED}")
+
+    def test_formatted_description_tracks_later_description_edits(self):
+        tool_instance = self._make_tool()
+        tool_instance.description = "Edited description."
+        assert tool_instance.formatted_description.endswith(
+            "Tool Description: Edited description."
+        )
+
+    def test_prose_mentioning_the_marker_is_not_truncated(self):
+        """Authored text that merely mentions "Tool Description:" must reach
+        the LLM untouched — only descriptions that ARE a pre-composed block
+        (anchored three-line shape) get stripped."""
+        tool_instance = self._make_tool()
+        prose = (
+            "Formats prompts. The output includes a line reading "
+            "'Tool Description:' followed by the tool's summary."
+        )
+        tool_instance.description = prose
+        assert tool_instance.formatted_description.endswith(
+            f"Tool Description: {prose}"
+        )
+
+    def test_composite_is_not_reapplied_to_prebaked_descriptions(self):
+        """A description that already contains a composed block (old
+        checkpoints, adapters that bake the composite into the field) must
+        not be double-wrapped."""
+        tool_instance = self._make_tool()
+        tool_instance.description = (
+            "Tool Name: get_temperature\n"
+            'Tool Arguments: {"city": "str"}\n'
+            f"Tool Description: {self.AUTHORED}"
+        )
+        formatted = tool_instance.formatted_description
+        assert formatted.count("Tool Description:") == 1
+        assert formatted.endswith(f"Tool Description: {self.AUTHORED}")
+
+    def test_prompt_rendering_still_uses_composite(self):
+        from crewai.utilities.agent_utils import render_text_description_and_args
+
+        tool_instance = self._make_tool()
+        structured = tool_instance.to_structured_tool()
+        assert structured.description == self.AUTHORED
+
+        for candidate in (tool_instance, structured):
+            rendered = render_text_description_and_args([candidate])
+            assert "Tool Name: get_temperature" in rendered
+            assert "Tool Arguments:" in rendered
+            assert f"Tool Description: {self.AUTHORED}" in rendered

lib/crewai/tests/utilities/test_console_formatter_tui_mode.py

@@ -0,0 +1,46 @@
+"""Flow panels must be suppressed while a TUI owns the screen."""
+
+from rich.text import Text
+
+from crewai.events.listeners.tracing.utils import set_tui_mode
+from crewai.events.utils.console_formatter import ConsoleFormatter
+
+
+def _make_formatter(monkeypatch):
+    fmt = ConsoleFormatter(verbose=True)
+    calls: list[object] = []
+    monkeypatch.setattr(fmt, "print", lambda *a, **k: calls.append(a))
+    return fmt, calls
+
+
+def test_flow_panel_suppressed_in_tui_mode(monkeypatch):
+    fmt, calls = _make_formatter(monkeypatch)
+    set_tui_mode(True)
+    try:
+        fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True)
+    finally:
+        set_tui_mode(False)
+
+    assert calls == []
+
+
+def test_flow_panel_prints_when_not_tui_mode(monkeypatch):
+    fmt, calls = _make_formatter(monkeypatch)
+    set_tui_mode(False)
+
+    fmt.print_panel(Text("x"), "🌊 Flow Started", "blue", is_flow=True)
+
+    # Panel + trailing blank line.
+    assert len(calls) == 2
+
+
+def test_non_flow_panel_unaffected_by_tui_mode(monkeypatch):
+    # tui_mode only gates flow panels; regular panels still follow verbose.
+    fmt, calls = _make_formatter(monkeypatch)
+    set_tui_mode(True)
+    try:
+        fmt.print_panel(Text("x"), "Task", "blue", is_flow=False)
+    finally:
+        set_tui_mode(False)
+
+    assert len(calls) == 2