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/deployment-sizing.mdx

@@ -1,117 +0,0 @@
----
-title: "Deployment Sizing"
-description: Choose the right deployment size for your crew workloads — and know when to scale up.
----
-
-## Overview
-
-Every CrewAI Enterprise deployment runs on a fixed resource tier called an **instance size**. The size controls how much CPU, memory, and — most importantly — how many crew runs can execute simultaneously. Choosing the wrong size is the most common cause of queue build-up, slow run starts, and OOMKilled pods.
-
-This page explains what each size provides, how to read the signals that you've outgrown your current tier, and how to right-size for your workload.
-
----
-
-## Instance Sizes
-
-| # | Name | vCPU | Memory | Max Concurrent Runs | Storage |
-|---|------|------|--------|---------------------|---------|
-| 1 | Small | 1 | 2 GiB | 4 | 20 GiB |
-| 2 | Regular | 2 | 4 GiB | 16 | 20 GiB |
-| 3 | Large | 4 | 8 GiB | 32 | 20 GiB |
-| 4 | Extra Large | 8 | 16 GiB | 64 | 100 GiB |
-| 5 | Extra Extra Large | 16 | 32 GiB | 128 | 100 GiB |
-| 6 | Insane Large | 32 | 64 GiB | 256 | 100 GiB |
-
-**vCPU** and **Memory** are the total resources allocated to the deployment (web server + workers + Redis combined).
-
-**Max Concurrent Runs** is the worker concurrency limit — the number of crew runs that can be actively executing at the same time. Runs submitted beyond this limit are queued and wait for a slot to open.
-
-<Note>
-  Concurrency is per-deployment, not per-crew. If you have 10 crews deployed on a Small instance, all 10 share the same pool of 4 concurrent run slots.
-</Note>
-
----
-
-## What "concurrent runs" actually means
-
-A **concurrent run** is one active kickoff of a crew — from the moment it starts executing until it completes or errors. It does not mean the number of agents running in parallel inside a single crew (that's controlled by your crew's process type and agent configuration).
-
-**Example:** A Small deployment (concurrency = 4) with 20 incoming run requests will execute 4 runs simultaneously and queue the remaining 16. Each queued run starts as soon as a slot frees up.
-
----
-
-## Symptoms of an undersized deployment
-
-| Symptom | Likely cause |
-|---------|-------------|
-| Runs sit in `queued` state for a long time | Concurrency limit reached — all worker slots are occupied |
-| Runs complete slowly even for simple tasks | CPU throttling — workers are competing for the same vCPU budget |
-| Pods restart with `OOMKilled` | Memory limit exceeded — reduce concurrency or upgrade size |
-| Builds fail or time out | Insufficient CPU/memory for the BuildKit image build step |
-| High p95/p99 run latency with normal p50 | Bursty traffic hitting the concurrency ceiling |
-
----
-
-## How to choose a size
-
-### Start with your concurrency requirement
-
-Estimate the peak number of crew runs you expect to have in-flight simultaneously. Add ~25% headroom for bursts.
-
-| Peak concurrent runs | Recommended size |
-|----------------------|-----------------|
-| 1–3 | Small |
-| 4–12 | Regular |
-| 13–25 | Large |
-| 26–50 | Extra Large |
-| 51–100 | Extra Extra Large |
-| 100+ | Insane Large |
-
-### Factor in run duration
-
-Long-running crews (minutes to hours) hold concurrency slots for the full duration. If your crews run for 10 minutes on average and you receive 30 runs per hour, you need at least `30 × (10/60) = 5` concurrent slots — Regular or above.
-
-### Factor in memory per run
-
-Each concurrent run consumes memory proportional to the number of agents, the size of context windows, and any in-memory data processing. If individual runs are memory-heavy (large document processing, many parallel agents), size up even if your concurrency requirement is low.
-
-A rough heuristic: assume **~256 MiB per concurrent run** as a baseline, then add overhead for your specific workload. On a Small instance (2 GiB total, shared with web and Redis), you have roughly 1 GiB available for workers — enough for ~4 lightweight runs, which matches the concurrency limit.
-
----
-
-## Changing your deployment size
-
-Deployment size is configurable from the **Admin Panel → Deployments → [your deployment] → Instance Size**. Changes take effect on the next deployment cycle (a rolling restart of the worker pods).
-
-<Warning>
-  Downsizing a deployment that is actively processing runs will cause in-flight runs to be interrupted when the old pods are replaced. Schedule size changes during low-traffic windows.
-</Warning>
-
----
-
-## Monitoring utilization
-
-Use these signals to track whether your current size is appropriate:
-
-```bash
-# Check current pod resource usage
-kubectl top pods
-
-# Watch for OOMKilled restarts
-kubectl get pods -o wide
-kubectl describe pod <worker-pod-name> | grep -A5 "Last State"
-
-# Check worker queue depth (from a web pod)
-kubectl exec -it deploy/crewai-web -- bin/rails runner \
-  "puts Sidekiq::Queue.all.map { |q| \"#{q.name}: #{q.size}\" }.join(\"\\n\")"
-```
-
-A consistently non-zero queue depth on the default queue is the clearest signal that you need more concurrency (a larger instance size).
-
----
-
-## Related
-
-- [Troubleshooting](/troubleshooting) — OOMKilled, pod restarts, build failures
-- [Factory Health & Debug](/factory-health) — health check endpoint and component status
-- [Aurora Instance Sizing](/deployment-guides/aws-workos-wharf-studio#aurora-instance-sizing) — database sizing to match your deployment tier

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