From e9d4c57b2a40668a476f1e321acf7a5d6cbd8927 Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Fri, 28 Aug 2026 13:32:09 -0300 Subject: [PATCH] fix: run model call hooks on every path and propagate a deny (#7111) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: let a hook deny reach the caller as a deny A hook that raised `HookAborted` on `pre_model_call` never reached the code making the call: the LLM layer caught it and returned `False`, which providers translated into `ValueError("LLM call blocked by before_llm_call hook")`, dropping the reason and the source and making a policy decision indistinguishable from a provider outage. Every internal model call then absorbed that error through the `except Exception` that keeps a provider hiccup from failing a run, so memory analysis fell back to defaults and the converter and reasoning handler retried the call that was just denied. The abort now propagates out of the LLM layer while the boolean convention keeps its documented `ValueError` via `LegacyHookBlocked`, and the fail-open handlers around internal model calls re-raise it instead of degrading. * fix: dispatch model call hooks on the paths that skipped them A model call was only checked when the executor loop drove it: the `from_agent is not None` short-circuit in `base_llm` silenced the hooks for agent planning and step observation, no provider `acall` dispatched them at all, and `InternalInstructor` bypassed `llm.call` entirely. This replaces that short-circuit with an explicit `model_call_hooks_already_dispatched` window so the enclosing caller claims the dispatch, adds the pre-call dispatch to every provider's `acall`, and runs the hooks around the Instructor client call. A denial now emits a denied event instead of being logged and reported as a provider failure. * fix: report a boolean-convention deny as a deny, not an outage A `before_llm_call` hook that blocks by returning `False` reached the five native providers as a plain `ValueError`, which fell through to their generic `except Exception` and was logged and emitted as `OpenAI API call failed: ...` — the same deny raised as `HookAborted` was already labelled correctly, so the two dialects disagreed on whether a policy decision was a provider outage. The LLM layer now converts it into `LLMCallBlockedError`, still a `ValueError` so the fail-open handlers around internal model calls keep absorbing it, but its own type so a provider can report the decision it is. Since a block is raised rather than returned, the thirteen callers that turned the return flag into a raise by hand drop that line, and `_prepare_llm_call` raises the same type. * fix: keep a denied plan from letting the agent run unplanned `AgentExecutor.generate_plan` wraps `handle_agent_reasoning()` in a bare `except Exception`, so guarding the reasoning handler alone still left the deny absorbed one frame up: the executor logged "Error during planning" and the agent proceeded with no plan. It now re-raises `HookAborted` like the other planning boundaries, and the accompanying test also covers the boolean convention still degrading at a fail-open site. * fix: stop a denied knowledge query from running the task without knowledge `handle_knowledge_retrieval` and its async twin wrap the query rewrite in their own `except Exception`, so guarding `_get_knowledge_search_query` alone still let `execute_task` continue on the unaugmented prompt after a deny. Both now emit the terminal `KnowledgeSearchQueryFailedEvent` and re-raise `HookAborted`, matching the second-frame guard already added to `AgentExecutor.generate_plan`. Also documents the abort contract on `PlannerObserver.observe`. * fix: stop nine callers from re-swallowing a model call deny CodeRabbit caught the replan path re-swallowing a deny, so an AST sweep of every caller of a guarded function found the same defeat in nine places: classic and replan planning, memory recall and memory save on both `Agent` and `LiteAgent`, the base executor's save, and `LLMGuardrail.__call__`, which turned a refused call into validation feedback. Each now re-raises `HookAborted` after emitting whatever terminal event it owes, while every other failure keeps degrading as before — the knowledge guards move to that same idiom instead of duplicating their emit. * fix: pair a denied guardrail with the event it started Re-raising from `LLMGuardrail` left `process_guardrail` between its started and completed events, so a denied validation read as one still in flight rather than a policy decision. It now emits `LLMGuardrailCompletedEvent` with the deny reason before the abort leaves, matching what every other guarded site in this change already does. * fix: stop retrying a task after a hook denied its model call `Agent.execute_task` funnels every exception into `_handle_execution_error`, which re-runs the whole task up to `max_retry_limit` times, so a policy deny read as a transient blip: a crew whose first model call was denied retried and returned a normal answer. `HookAborted` now joins `_passthrough_exceptions`, the tuple already reserved for deliberate stops. The new boundary tests drive the public entry points instead of the frame that makes the call, and count model calls so a deny that gets retried fails the assertion — ten of the twelve fail against `main`. * fix: stop a denied plan step from being reported as a failed step Making model call hooks reachable on agent-bearing calls put a deny inside `StepExecutor.execute`, whose broad `except Exception` turned it into `StepResult(success=False)` and let the plan carry on; `HookAborted` now joins `ToolExecutionFailedError` in the passthrough handlers there, and `execute_todos_parallel` re-raises a deny that `return_exceptions=True` would otherwise record as one failed todo. `_emit_call_denied_event` also renders the source through the now-public `source_name`, so a hook that names itself with a callable reads as its name instead of a repr. --------- Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com> --- lib/crewai/src/crewai/agent/core.py | 26 +- lib/crewai/src/crewai/agent/utils.py | 9 + .../agent_builder/base_agent_executor.py | 4 + .../src/crewai/agents/planner_observer.py | 21 + lib/crewai/src/crewai/agents/step_executor.py | 5 +- .../src/crewai/experimental/agent_executor.py | 9 + .../evaluation/evaluation_display.py | 4 + lib/crewai/src/crewai/flow/human_feedback.py | 8 + .../src/crewai/flow/runtime/__init__.py | 5 + lib/crewai/src/crewai/hooks/dispatch.py | 4 +- lib/crewai/src/crewai/hooks/llm_hooks.py | 45 +- lib/crewai/src/crewai/lite_agent.py | 9 + lib/crewai/src/crewai/llm.py | 28 +- lib/crewai/src/crewai/llms/base_llm.py | 89 ++- .../llms/providers/anthropic/completion.py | 21 +- .../crewai/llms/providers/azure/completion.py | 21 +- .../llms/providers/bedrock/completion.py | 16 +- .../llms/providers/gemini/completion.py | 18 +- .../llms/providers/openai/completion.py | 21 +- lib/crewai/src/crewai/memory/analyze.py | 16 + lib/crewai/src/crewai/memory/recall_flow.py | 4 + lib/crewai/src/crewai/tasks/llm_guardrail.py | 6 + .../src/crewai/utilities/agent_utils.py | 87 +-- lib/crewai/src/crewai/utilities/converter.py | 16 + lib/crewai/src/crewai/utilities/crew_chat.py | 8 + lib/crewai/src/crewai/utilities/guardrail.py | 23 +- .../crewai/utilities/internal_instructor.py | 11 +- .../src/crewai/utilities/reasoning_handler.py | 6 + .../hooks/test_deny_reaches_the_caller.py | 309 +++++++++++ .../hooks/test_hook_abort_propagation.py | 522 ++++++++++++++++++ lib/crewai/tests/hooks/test_llm_hooks.py | 24 +- .../tests/hooks/test_model_call_hook_reach.py | 149 +++++ 32 files changed, 1443 insertions(+), 101 deletions(-) create mode 100644 lib/crewai/tests/hooks/test_deny_reaches_the_caller.py create mode 100644 lib/crewai/tests/hooks/test_hook_abort_propagation.py create mode 100644 lib/crewai/tests/hooks/test_model_call_hook_reach.py diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 0ac21743a..5f5c5a027 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -74,6 +74,7 @@ from crewai.events.types.memory_events import ( ) from crewai.events.types.skill_events import SkillUsedEvent from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import HookAborted from crewai.knowledge.knowledge import Knowledge from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource from crewai.lite_agent_output import LiteAgentOutput @@ -139,7 +140,10 @@ if TYPE_CHECKING: # Deliberate stops, not transient errors: never swallowed into the # max_retry_limit loop. -_passthrough_exceptions: tuple[type[Exception], ...] = (ToolExecutionFailedError,) +_passthrough_exceptions: tuple[type[Exception], ...] = ( + ToolExecutionFailedError, + HookAborted, +) _EXECUTOR_CLASS_MAP: dict[str, type] = { "CrewAgentExecutor": CrewAgentExecutor, @@ -711,6 +715,9 @@ class Agent(BaseAgent): error=str(e), ), ) + # a deny aborts the task; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise return task_prompt @@ -1438,6 +1445,18 @@ class Agent(BaseAgent): ), ) return rewritten_query + except HookAborted as e: + # A deny still owes the started event above its terminal event; only + # the fallback to no query is skipped. + crewai_event_bus.emit( + self, + event=KnowledgeQueryFailedEvent( + error=str(e), + from_task=task, + from_agent=self, + ), + ) + raise except Exception as e: crewai_event_bus.emit( self, @@ -1634,6 +1653,9 @@ class Agent(BaseAgent): error=str(e), ), ) + # a deny aborts the kickoff; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise inputs: dict[str, Any] = { "input": formatted_messages, @@ -1815,6 +1837,8 @@ class Agent(BaseAgent): extracted = agent_memory.extract_memories(raw) if extracted: agent_memory.remember_many(extracted) + except HookAborted: + raise except Exception as e: self._logger.log("error", f"Failed to save kickoff result to memory: {e}") diff --git a/lib/crewai/src/crewai/agent/utils.py b/lib/crewai/src/crewai/agent/utils.py index 47701c8e4..20eb06b47 100644 --- a/lib/crewai/src/crewai/agent/utils.py +++ b/lib/crewai/src/crewai/agent/utils.py @@ -15,6 +15,7 @@ from crewai.events.types.knowledge_events import ( KnowledgeRetrievalStartedEvent, KnowledgeSearchQueryFailedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context from crewai.utilities.pydantic_schema_utils import generate_model_description from crewai.utilities.types import LLMMessage @@ -53,6 +54,8 @@ def handle_reasoning(agent: Agent, task: Task) -> None: planning_handler.handle_agent_reasoning() ) task.description += f"\n\nPlanning:\n{planning_output.plan.plan}" + except HookAborted: + raise except Exception as e: agent._logger.log("error", f"Error during planning: {e!s}") @@ -195,6 +198,9 @@ def handle_knowledge_retrieval( from_agent=agent, ), ) + # a deny aborts the task; any other failure degrades to no knowledge + if isinstance(e, HookAborted): + raise return task_prompt @@ -391,4 +397,7 @@ async def ahandle_knowledge_retrieval( from_agent=agent, ), ) + # a deny aborts the task; any other failure degrades to no knowledge + if isinstance(e, HookAborted): + raise return task_prompt diff --git a/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py b/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py index d251b1d36..39ac025a5 100644 --- a/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py +++ b/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py @@ -30,6 +30,8 @@ class BaseAgentExecutor(BaseModel): def _save_to_memory(self, output: AgentFinish) -> None: """Save task result to unified memory (memory or crew._memory).""" + from crewai.hooks.dispatch import HookAborted + if self.agent is None: return memory = getattr(self.agent, "memory", None) or ( @@ -61,5 +63,7 @@ class BaseAgentExecutor(BaseModel): ) else: memory.remember_many(extracted, agent_role=self.agent.role) + except HookAborted: + raise except Exception as e: self.agent._logger.log("error", f"Failed to save to memory: {e}") diff --git a/lib/crewai/src/crewai/agents/planner_observer.py b/lib/crewai/src/crewai/agents/planner_observer.py index 2882ac780..ce08e6416 100644 --- a/lib/crewai/src/crewai/agents/planner_observer.py +++ b/lib/crewai/src/crewai/agents/planner_observer.py @@ -131,7 +131,13 @@ class PlannerObserver: StepObservation with the Planner's analysis. Any suggested refinements are structured StepRefinement objects ready for direct application — no second LLM call needed. + + Raises: + HookAborted: A `pre_model_call` hook denied the observation call. + Every other failure degrades to a conservative observation. """ + from crewai.hooks.dispatch import HookAborted + agent_role = self.agent.role crewai_event_bus.emit( @@ -188,6 +194,21 @@ class PlannerObserver: return observation + except HookAborted as e: + # A deny still owes the started event above its terminal event; only + # the conservative-replan fallback is skipped. + crewai_event_bus.emit( + self.agent, + event=StepObservationFailedEvent( + agent_role=agent_role, + step_number=completed_step.step_number, + step_description=completed_step.description, + error=str(e), + from_task=self.task, + from_agent=self.agent, + ), + ) + raise except Exception as e: logger.warning( f"Observation LLM call failed: {e}. Defaulting to conservative replan." diff --git a/lib/crewai/src/crewai/agents/step_executor.py b/lib/crewai/src/crewai/agents/step_executor.py index 415586bb9..a25db4e2f 100644 --- a/lib/crewai/src/crewai/agents/step_executor.py +++ b/lib/crewai/src/crewai/agents/step_executor.py @@ -28,6 +28,7 @@ from crewai.events.types.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.tools.tool_failure import ToolExecutionFailedError from crewai.utilities.agent_utils import ( build_text_tool_calling_fallback_message, @@ -181,7 +182,7 @@ class StepExecutor: tool_calls_made=tool_calls_made, execution_time=elapsed, ) - except ToolExecutionFailedError: + except (ToolExecutionFailedError, HookAborted): # A deliberate stop: StepResult(success=False) would let the plan # carry on. raise @@ -224,7 +225,7 @@ class StepExecutor: tool_calls_made=tool_calls_made, execution_time=elapsed, ) - except ToolExecutionFailedError: + except (ToolExecutionFailedError, HookAborted): # Same as the outer handler, reached via the text-tooling # fallback. raise diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 45fffa1e4..d51e7aaca 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -57,6 +57,7 @@ from crewai.events.types.tool_usage_events import ( from crewai.flow.flow import Flow, listen, or_, router, start from crewai.flow.flow_context import current_flow_id from crewai.flow.types import FlowMethodName +from crewai.hooks.dispatch import HookAborted from crewai.hooks.llm_hooks import ( get_after_llm_call_hooks, get_before_llm_call_hooks, @@ -420,6 +421,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): # Do NOT mutate task.description — it's a shared object that # accumulates plan text on re-invoke. + except HookAborted: + raise except Exception as e: if hasattr(self.agent, "_logger"): self.agent._logger.log("error", f"Error during planning: {e!s}") @@ -1295,6 +1298,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): # todo ↔ result (or exception) mapping. step_results: list[tuple[TodoItem, StepResult]] = [] for todo, item in zip(ready, gathered, strict=True): + if isinstance(item, HookAborted): + raise item if isinstance(item, BaseException): error_msg = f"Error: {item!s}" todo.result = error_msg @@ -2552,6 +2557,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): ) return + except HookAborted: + raise except Exception as e: if self.agent and self.agent.verbose: PRINTER.print( @@ -2704,6 +2711,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): color="green", ) + except HookAborted: + raise except Exception as e: if hasattr(self.agent, "_logger"): self.agent._logger.log("error", f"Error during replanning: {e!s}") diff --git a/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py b/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py index 3f2314a36..6d462a007 100644 --- a/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py +++ b/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py @@ -311,6 +311,8 @@ class EvaluationDisplayFormatter: scores: list[float | None], strategy: AggregationStrategy, ) -> str: + from crewai.hooks.dispatch import HookAborted + if len(feedbacks) <= 2 and all(len(fb) < 200 for fb in feedbacks): return "\n\n".join( [f"Feedback {i + 1}: {fb}" for i, fb in enumerate(feedbacks)] @@ -372,6 +374,8 @@ class EvaluationDisplayFormatter: raise ValueError("LLM must be initialized") return llm.call(prompt) + except HookAborted: + raise except Exception: return "Synthesized from multiple tasks: " + "\n\n".join( [f"- {fb[:500]}..." for fb in feedbacks] diff --git a/lib/crewai/src/crewai/flow/human_feedback.py b/lib/crewai/src/crewai/flow/human_feedback.py index a1fabd658..0e0785f99 100644 --- a/lib/crewai/src/crewai/flow/human_feedback.py +++ b/lib/crewai/src/crewai/flow/human_feedback.py @@ -253,6 +253,8 @@ def _pre_review_with_lessons( learn_source: str, learn_strict: bool, ) -> Any: + from crewai.hooks.dispatch import HookAborted + try: mem = flow_instance.memory if mem is None: @@ -282,6 +284,8 @@ def _pre_review_with_lessons( return PreReviewResult.model_validate(response).improved_output reviewed = llm_inst.call(messages) return reviewed if isinstance(reviewed, str) else str(reviewed) + except HookAborted: + raise except Exception: if learn_strict: logger.warning( @@ -308,6 +312,8 @@ def _distill_and_store_lessons( learn_source: str, learn_strict: bool, ) -> None: + from crewai.hooks.dispatch import HookAborted + try: mem = flow_instance.memory if mem is None: @@ -344,6 +350,8 @@ def _distill_and_store_lessons( if lessons: mem.remember_many(lessons, source=learn_source) # type: ignore[union-attr] + except HookAborted: + raise except Exception: if learn_strict: logger.warning( diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 01f645e2e..8c756a39d 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -3847,6 +3847,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): from pydantic import BaseModel, Field + from crewai.hooks.dispatch import HookAborted from crewai.llm import LLM from crewai.llms.base_llm import BaseLLM as BaseLLMClass from crewai.utilities.i18n import I18N_DEFAULT @@ -3902,6 +3903,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): logger.warning(f"Unexpected response type: {type(response)}") return outcomes[0] + except HookAborted: + raise except Exception as e: logger.warning( f"Structured output failed, falling back to simple prompting: {e}" @@ -3933,6 +3936,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ) return outcomes[0] + except HookAborted: + raise except Exception as fallback_err: logger.warning( f"Simple prompting also failed: {fallback_err}. " diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index fd692c900..92826a71f 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -209,7 +209,7 @@ def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]: return global_hooks -def _source_name(source: Any) -> str | None: +def source_name(source: Any) -> str | None: """Best-effort readable name for a hook source.""" if source is None: return None @@ -341,7 +341,7 @@ def run_hooks( except HookAborted as aborted: outcome = "aborted" abort_reason = aborted.reason - abort_source = _source_name(aborted.source) + abort_source = source_name(aborted.source) raise finally: _emit_telemetry( diff --git a/lib/crewai/src/crewai/hooks/llm_hooks.py b/lib/crewai/src/crewai/hooks/llm_hooks.py index 55563e84e..7658bf514 100644 --- a/lib/crewai/src/crewai/hooks/llm_hooks.py +++ b/lib/crewai/src/crewai/hooks/llm_hooks.py @@ -1,5 +1,8 @@ from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar from typing import TYPE_CHECKING, Any, cast from crewai_core.printer import PRINTER @@ -166,14 +169,50 @@ _after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = ( ) +class LegacyHookBlocked(HookAborted): + """A ``before_llm_call`` hook blocked the call by returning ``False``. + + Distinguishes the boolean convention, which the LLM layer keeps surfacing as + the documented ``ValueError``, from a hook that raised :class:`HookAborted` + itself and must reach the caller as the deny it is. Raised by the reducer and + consumed inside the LLM layer, which re-raises it as + :class:`~crewai.llms.base_llm.LLMCallBlockedError`. + """ + + +_model_call_hooks_dispatched: ContextVar[bool] = ContextVar( + "model_call_hooks_dispatched", default=False +) + + +@contextmanager +def model_call_hooks_dispatched() -> Iterator[None]: + """Mark the window where the model-call hooks already ran for a pending call. + + The executor dispatches with its own richer context (executor, task, crew) + and only then reaches the LLM. Without this marker the LLM layer would + dispatch a second time for the same call. + """ + token = _model_call_hooks_dispatched.set(True) + try: + yield + finally: + _model_call_hooks_dispatched.reset(token) + + +def model_call_hooks_already_dispatched() -> bool: + """Whether an enclosing caller already dispatched hooks for the current call.""" + return _model_call_hooks_dispatched.get() + + 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. + A ``False`` return aborts the call (mapped to :class:`LegacyHookBlocked`); + messages are modified in place, so no payload replacement occurs here. """ if result is False: - raise HookAborted(reason="before_llm_call hook returned False") + raise LegacyHookBlocked(reason="before_llm_call hook returned False") return False diff --git a/lib/crewai/src/crewai/lite_agent.py b/lib/crewai/src/crewai/lite_agent.py index 373b774e4..48a130956 100644 --- a/lib/crewai/src/crewai/lite_agent.py +++ b/lib/crewai/src/crewai/lite_agent.py @@ -598,6 +598,8 @@ class LiteAgent(FlowTrackable, BaseModel): def _inject_memory_context(self) -> None: """Recall relevant memories and append to the system message. No-op if _memory is None.""" + from crewai.hooks.dispatch import HookAborted + if self._memory is None: return query = self._get_last_user_content() @@ -641,9 +643,14 @@ class LiteAgent(FlowTrackable, BaseModel): error=str(e), ), ) + # a deny aborts the run; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise def _save_to_memory(self, output_text: str) -> None: """Extract discrete memories from the run and remember each. No-op if _memory is None or read-only.""" + from crewai.hooks.dispatch import HookAborted + if self._memory is None or self._memory.read_only: return input_str = self._get_last_user_content() or "User request" @@ -652,6 +659,8 @@ class LiteAgent(FlowTrackable, BaseModel): extracted = self._memory.extract_memories(raw) if extracted: self._memory.remember_many(extracted, agent_role=self.role) + except HookAborted: + raise except Exception as e: if self.verbose: PRINTER.print( diff --git a/lib/crewai/src/crewai/llm.py b/lib/crewai/src/crewai/llm.py index 371aadf87..1eb52255f 100644 --- a/lib/crewai/src/crewai/llm.py +++ b/lib/crewai/src/crewai/llm.py @@ -31,10 +31,12 @@ from crewai.events.types.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.base_llm import ( BaseLLM, JsonResponseFormat, + LLMCallBlockedError, get_current_call_id, llm_call_context, ) @@ -1039,12 +1041,15 @@ class LLM(BaseLLM): if not tool_calls or not available_functions: if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched + instructor_instance = InternalInstructor( content=full_response, model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() usage_dict = self._usage_to_dict(usage_info) self._handle_emit_call_events( @@ -1241,6 +1246,7 @@ class LLM(BaseLLM): str: The response text """ if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched from crewai.utilities.internal_instructor import InternalInstructor messages = params.get("messages", []) @@ -1256,7 +1262,8 @@ class LLM(BaseLLM): model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() self._handle_emit_call_events( response=structured_response, @@ -1396,6 +1403,7 @@ class LLM(BaseLLM): str: The response text """ if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched from crewai.utilities.internal_instructor import InternalInstructor messages = params.get("messages", []) @@ -1411,7 +1419,8 @@ class LLM(BaseLLM): model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() self._handle_emit_call_events( response=structured_response, @@ -1874,8 +1883,11 @@ class LLM(BaseLLM): msg_role: Literal["assistant"] = "assistant" message["role"] = msg_role - if not self._invoke_before_llm_call_hooks(messages, from_agent): - raise ValueError("LLM call blocked by before_llm_call hook") + try: + self._invoke_before_llm_call_hooks(messages, from_agent) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise with suppress_warnings(): if callbacks and len(callbacks) > 0: @@ -2016,6 +2028,12 @@ class LLM(BaseLLM): msg_role: Literal["assistant"] = "assistant" message["role"] = msg_role + try: + self._invoke_before_llm_call_hooks(messages, from_agent) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise + with suppress_warnings(): if callbacks and len(callbacks) > 0: self.set_callbacks(callbacks) diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index 3b40601bc..9bd964a31 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -72,6 +72,15 @@ class JsonResponseFormat(TypedDict): type: Literal["json_object"] +class LLMCallBlockedError(ValueError): + """A ``before_llm_call`` hook blocked the call by returning ``False``. + + A ``ValueError`` so the fail-open handlers around internal model calls keep + absorbing it, and its own type so a provider can report it as the decision + it is instead of letting it read as a provider outage. + """ + + DEFAULT_CONTEXT_WINDOW_SIZE: Final[int] = 4096 DEFAULT_SUPPORTS_STOP_WORDS: Final[bool] = True _JSON_EXTRACTION_PATTERN: Final[re.Pattern[str]] = re.compile(r"\{.*}", re.DOTALL) @@ -660,6 +669,28 @@ class BaseLLM(BaseModel, ABC): ), ) + def _emit_call_denied_event( + self, + denial: Exception, + from_task: Task | None = None, + from_agent: BaseAgent | None = None, + ) -> None: + """Report a hook deny as a deny rather than as a provider failure. + + The call still owes its started event a terminal one, so the failed event + is emitted — with a message that names the decision instead of blaming + the provider for an outage that never happened. + """ + from crewai.hooks.dispatch import source_name + + source = source_name(getattr(denial, "source", None)) + reason = getattr(denial, "reason", str(denial)) + message = f"LLM call denied by {source or 'hook'}: {reason}" + logging.warning(message) + self._emit_call_failed_event( + error=message, from_task=from_task, from_agent=from_agent + ) + def _emit_stream_chunk_event( self, chunk: str, @@ -990,43 +1021,51 @@ class BaseLLM(BaseModel, ABC): messages: list[LLMMessage], from_agent: BaseAgent | None = None, ) -> bool: - """Invoke before_llm_call hooks for direct LLM calls (no agent context). + """Invoke before_llm_call hooks for an LLM call reaching the provider. This method should be called by native provider implementations before - making the actual LLM call when from_agent is None (direct calls). + making the actual LLM call. It no-ops when an enclosing caller — the + executor — already dispatched the hooks for this same call. Args: messages: The messages being sent to the LLM - from_agent: The agent making the call (None for direct calls) + from_agent: The agent making the call, when there is one Returns: - True if LLM call should proceed, False if blocked by hook + True, so a provider may still guard the call with ``if not ...``. + A block is raised, never returned. + + Raises: + HookAborted: If a hook raised it. The deny reaches the caller intact + instead of being flattened into a provider-style error. + LLMCallBlockedError: If a legacy hook blocked the call by returning + ``False``. A ``ValueError``, so the fail-open handlers around + internal model calls keep absorbing it. Example: >>> # In a native provider's call() method: - >>> if from_agent is None and not self._invoke_before_llm_call_hooks( - ... messages, from_agent - ... ): - ... raise ValueError("LLM call blocked by hook") + >>> self._invoke_before_llm_call_hooks(messages, from_agent) """ - if from_agent is not None: - return True - from crewai_core.printer import PRINTER - from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch + from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, + LegacyHookBlocked, before_llm_call_reducer, + model_call_hooks_already_dispatched, ) + if model_call_hooks_already_dispatched(): + return True + # No early global-list guard: dispatch resolves global + execution-scoped # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, llm=self, - agent=None, + agent=from_agent, task=None, crew=None, ) @@ -1037,12 +1076,14 @@ class BaseLLM(BaseModel, ABC): hook_context, reducer=before_llm_call_reducer, ) - except HookAborted: + except LegacyHookBlocked as blocked: PRINTER.print( content="LLM call blocked by before_llm_call hook", color="yellow", ) - return False + raise LLMCallBlockedError( + "LLM call blocked by before_llm_call hook" + ) from blocked return True @@ -1052,42 +1093,44 @@ class BaseLLM(BaseModel, ABC): response: str, from_agent: BaseAgent | None = None, ) -> str: - """Invoke after_llm_call hooks for direct LLM calls (no agent context). + """Invoke after_llm_call hooks for an LLM call that reached the provider. This method should be called by native provider implementations after - receiving the LLM response when from_agent is None (direct calls). + receiving the LLM response. It no-ops when an enclosing caller — the + executor — already dispatched the hooks for this same call. Args: messages: The messages that were sent to the LLM response: The response from the LLM - from_agent: The agent that made the call (None for direct calls) + from_agent: The agent that made the call, when there is one Returns: The potentially modified response string Example: >>> # In a native provider's call() method: - >>> if from_agent is None and isinstance(result, str): + >>> if isinstance(result, str): ... result = self._invoke_after_llm_call_hooks( ... messages, result, from_agent ... ) """ - if from_agent is not None or not isinstance(response, str): - return response - from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, after_llm_call_reducer, + model_call_hooks_already_dispatched, ) + if model_call_hooks_already_dispatched() or not isinstance(response, str): + return response + # No early global-list guard: dispatch resolves global + execution-scoped # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, llm=self, - agent=None, + agent=from_agent, task=None, crew=None, response=response, diff --git a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py index e6bc4255f..71ce23a58 100644 --- a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py +++ b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py @@ -8,7 +8,13 @@ from typing import Any, Final, Literal, Protocol, TypeGuard, TypedDict, cast from pydantic import BaseModel, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import ( + BaseLLM, + JsonResponseFormat, + LLMCallBlockedError, + llm_call_context, +) from crewai.llms.hooks.base import BaseInterceptor from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport from crewai.llms.providers.utils.common import safe_tool_conversion @@ -401,10 +407,7 @@ class AnthropicCompletion(BaseLLM): self._format_messages_for_anthropic(messages) ) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) completion_params = self._prepare_completion_params( formatted_messages, system_message, tools, available_functions @@ -429,6 +432,9 @@ class AnthropicCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"Anthropic API call failed: {e!s}" logging.error(error_msg) @@ -476,6 +482,8 @@ class AnthropicCompletion(BaseLLM): self._format_messages_for_anthropic(messages) ) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + completion_params = self._prepare_completion_params( formatted_messages, system_message, tools, available_functions ) @@ -499,6 +507,9 @@ class AnthropicCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"Anthropic API call failed: {e!s}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/llms/providers/azure/completion.py b/lib/crewai/src/crewai/llms/providers/azure/completion.py index 0f7cbd4fc..8cdb49cca 100644 --- a/lib/crewai/src/crewai/llms/providers/azure/completion.py +++ b/lib/crewai/src/crewai/llms/providers/azure/completion.py @@ -9,6 +9,7 @@ from urllib.parse import urlparse from pydantic import BaseModel, PrivateAttr, model_validator from typing_extensions import Self +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.hooks.base import BaseInterceptor from crewai.utilities.agent_utils import is_context_length_exceeded @@ -42,7 +43,12 @@ try: ) from crewai.events.types.llm_events import LLMCallType - from crewai.llms.base_llm import BaseLLM, call_stream_override, llm_call_context + from crewai.llms.base_llm import ( + BaseLLM, + LLMCallBlockedError, + call_stream_override, + llm_call_context, + ) except ImportError: raise ImportError( @@ -521,10 +527,7 @@ class AzureCompletion(BaseLLM): formatted_messages = self._format_messages_for_azure(messages) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) completion_params = self._prepare_completion_params( formatted_messages, tools, effective_response_model @@ -547,6 +550,9 @@ class AzureCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: return self._handle_api_error(e, from_task, from_agent) # type: ignore[func-returns-value] @@ -603,6 +609,8 @@ class AzureCompletion(BaseLLM): formatted_messages = self._format_messages_for_azure(messages) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + completion_params = self._prepare_completion_params( formatted_messages, tools, effective_response_model ) @@ -624,6 +632,9 @@ class AzureCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: self._handle_api_error(e, from_task, from_agent) diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index d10fc2ba8..ab05c55da 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py @@ -11,7 +11,8 @@ from pydantic import BaseModel, PrivateAttr, model_validator from typing_extensions import Required from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context from crewai.llms.providers.utils.common import safe_tool_conversion from crewai.utilities.agent_utils import is_context_length_exceeded from crewai.utilities.exceptions.context_window_exceeding_exception import ( @@ -377,10 +378,7 @@ class BedrockCompletion(BaseLLM): messages ) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) body: BedrockConverseRequestBody = { "inferenceConfig": self._get_inference_config(), @@ -447,6 +445,9 @@ class BedrockCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: if is_context_length_exceeded(e): logging.error(f"Context window exceeded: {e}") @@ -510,6 +511,8 @@ class BedrockCompletion(BaseLLM): messages ) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + body: BedrockConverseRequestBody = { "inferenceConfig": self._get_inference_config(), } @@ -575,6 +578,9 @@ class BedrockCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: if is_context_length_exceeded(e): logging.error(f"Context window exceeded: {e}") diff --git a/lib/crewai/src/crewai/llms/providers/gemini/completion.py b/lib/crewai/src/crewai/llms/providers/gemini/completion.py index b099fe237..0bdf8ec0e 100644 --- a/lib/crewai/src/crewai/llms/providers/gemini/completion.py +++ b/lib/crewai/src/crewai/llms/providers/gemini/completion.py @@ -10,7 +10,8 @@ from typing import Any, Literal, cast from pydantic import BaseModel, Field, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context from crewai.llms.hooks.base import BaseInterceptor from crewai.utilities.agent_utils import is_context_length_exceeded from crewai.utilities.exceptions.context_window_exceeding_exception import ( @@ -313,10 +314,7 @@ class GeminiCompletion(BaseLLM): messages_for_hooks = self._convert_contents_to_dict(formatted_content) - if not self._invoke_before_llm_call_hooks( - messages_for_hooks, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent) config = self._prepare_generation_config( system_instruction, tools, effective_response_model @@ -341,6 +339,9 @@ class GeminiCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except APIError as e: error_msg = f"Google Gemini API error: {e.code} - {e.message}" logging.error(error_msg) @@ -397,6 +398,10 @@ class GeminiCompletion(BaseLLM): self._format_messages_for_gemini(messages) ) + messages_for_hooks = self._convert_contents_to_dict(formatted_content) + + self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent) + config = self._prepare_generation_config( system_instruction, tools, effective_response_model ) @@ -420,6 +425,9 @@ class GeminiCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except APIError as e: error_msg = f"Google Gemini API error: {e.code} - {e.message}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/llms/providers/openai/completion.py b/lib/crewai/src/crewai/llms/providers/openai/completion.py index 2b667e168..20234e501 100644 --- a/lib/crewai/src/crewai/llms/providers/openai/completion.py +++ b/lib/crewai/src/crewai/llms/providers/openai/completion.py @@ -36,8 +36,14 @@ from openai.types.responses import ( from pydantic import BaseModel, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id -from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context +from crewai.llms.base_llm import ( + BaseLLM, + JsonResponseFormat, + LLMCallBlockedError, + llm_call_context, +) from crewai.llms.hooks.base import BaseInterceptor from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport from crewai.llms.providers.utils.common import safe_tool_conversion @@ -462,10 +468,7 @@ class OpenAICompletion(BaseLLM): formatted_messages = self._format_messages(messages) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) if self._effective_api() == "responses": return self._call_responses( @@ -486,6 +489,9 @@ class OpenAICompletion(BaseLLM): response_model=response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"OpenAI API call failed: {e!s}" logging.error(error_msg) @@ -600,6 +606,8 @@ class OpenAICompletion(BaseLLM): formatted_messages = self._format_messages(messages) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + if self._effective_api() == "responses": return await self._acall_responses( messages=formatted_messages, @@ -619,6 +627,9 @@ class OpenAICompletion(BaseLLM): response_model=response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"OpenAI API call failed: {e!s}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/memory/analyze.py b/lib/crewai/src/crewai/memory/analyze.py index 65d671d0d..3bb3a92a4 100644 --- a/lib/crewai/src/crewai/memory/analyze.py +++ b/lib/crewai/src/crewai/memory/analyze.py @@ -168,6 +168,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]: Returns: List of short, self-contained memory statements (or [content] on failure). """ + from crewai.hooks.dispatch import HookAborted + if not (content or "").strip(): return [] user = _get_prompt("extract_memories_user").format(content=content) @@ -188,6 +190,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]: data = json.loads(response) return ExtractedMemories.model_validate(data).memories return ExtractedMemories.model_validate(response).memories + except HookAborted: + raise except Exception as e: _logger.warning( "Memory extraction failed, storing full content as single memory: %s", @@ -216,6 +220,8 @@ def analyze_query( Returns: QueryAnalysis with keywords, suggested_scopes, complexity, recall_queries, time_filter. """ + from crewai.hooks.dispatch import HookAborted + scope_desc = "" if scope_info: scope_desc = f"Current scope has {scope_info.record_count} records, categories: {scope_info.categories}" @@ -241,6 +247,8 @@ def analyze_query( data = json.loads(response) return QueryAnalysis.model_validate(data) return QueryAnalysis.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Query analysis failed, using defaults (complexity=simple): %s", @@ -284,6 +292,8 @@ def analyze_for_save( Returns: MemoryAnalysis with suggested_scope, categories, importance, extracted_metadata. """ + from crewai.hooks.dispatch import HookAborted + user = _get_prompt("save_user").format( content=content, existing_scopes=existing_scopes or ["/"], @@ -306,6 +316,8 @@ def analyze_for_save( data = json.loads(response) return MemoryAnalysis.model_validate(data) return MemoryAnalysis.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Memory save analysis failed, using defaults: %s", @@ -336,6 +348,8 @@ def analyze_for_consolidation( Returns: ConsolidationPlan with actions per record and whether to insert the new content. """ + from crewai.hooks.dispatch import HookAborted + if not existing_records: return ConsolidationPlan(actions=[], insert_new=True) records_lines: list[str] = [] @@ -366,6 +380,8 @@ def analyze_for_consolidation( data = json.loads(response) return ConsolidationPlan.model_validate(data) return ConsolidationPlan.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Consolidation analysis failed, defaulting to insert: %s", diff --git a/lib/crewai/src/crewai/memory/recall_flow.py b/lib/crewai/src/crewai/memory/recall_flow.py index 8b2d64651..fbeb5c5c6 100644 --- a/lib/crewai/src/crewai/memory/recall_flow.py +++ b/lib/crewai/src/crewai/memory/recall_flow.py @@ -296,6 +296,8 @@ class RecallFlow(Flow[RecallState]): Decrements the exploration budget so the loop terminates. """ + from crewai.hooks.dispatch import HookAborted + self.state.exploration_budget -= 1 enhanced = [] @@ -321,6 +323,8 @@ class RecallFlow(Flow[RecallState]): "results": finding["results"], } ) + except HookAborted: + raise except Exception: enhanced.append( { diff --git a/lib/crewai/src/crewai/tasks/llm_guardrail.py b/lib/crewai/src/crewai/tasks/llm_guardrail.py index 754596ab7..da3498cc0 100644 --- a/lib/crewai/src/crewai/tasks/llm_guardrail.py +++ b/lib/crewai/src/crewai/tasks/llm_guardrail.py @@ -105,7 +105,11 @@ class LLMGuardrail: Tuple[bool, Any]: A tuple containing: - bool: True if validation passed, False otherwise - Any: The validation result or error message + + Raises: + HookAborted: A `pre_model_call` hook denied the validation call. """ + from crewai.hooks.dispatch import HookAborted try: result = self._validate_output(task_output) @@ -115,5 +119,7 @@ class LLMGuardrail: if result.pydantic.valid: return True, task_output.raw return False, result.pydantic.feedback + except HookAborted: + raise except Exception as e: return False, f"Error while validating the task output: {e!s}" diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index 6c0ea6bdc..b8f664c11 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -24,7 +24,7 @@ from crewai.agents.parser import ( OutputParserError, parse, ) -from crewai.llms.base_llm import BaseLLM, call_stop_override +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, call_stop_override from crewai.tools import BaseTool as CrewAITool from crewai.tools.base_tool import BaseTool from crewai.tools.structured_tool import ( @@ -481,31 +481,41 @@ def enforce_rpm_limit( request_within_rpm_limit() +@contextlib.contextmanager def _prepare_llm_call( executor_context: CrewAgentExecutor | AgentExecutor | LiteAgent | None, messages: list[LLMMessage], printer: Printer, verbose: bool = True, -) -> list[LLMMessage]: +) -> Iterator[list[LLMMessage]]: """Shared pre-call logic: run before hooks and resolve messages. + Yields for the duration of the LLM call so the LLM layer knows the hooks + already ran with this executor's context and does not dispatch them twice. + Args: executor_context: Optional executor context for hook invocation. messages: The messages to send to the LLM. printer: Printer instance for output. verbose: Whether to print output. - Returns: + Yields: The resolved messages list (may come from executor_context). Raises: ValueError: If a before hook blocks the call. """ - if executor_context is not None: - if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose): - raise ValueError("LLM call blocked by before_llm_call hook") - messages = executor_context.messages - return messages + from crewai.hooks.llm_hooks import model_call_hooks_dispatched + + if executor_context is None: + yield messages + return + + if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose): + raise LLMCallBlockedError("LLM call blocked by before_llm_call hook") + + with model_call_hooks_dispatched(): + yield executor_context.messages def _validate_and_finalize_llm_response( @@ -577,17 +587,18 @@ def get_llm_response( Exception: If an error occurs. ValueError: If the response is None or empty. """ - messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose) - - answer = llm.call( - messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - response_model=response_model, - ) + with _prepare_llm_call( + executor_context, messages, printer, verbose=verbose + ) as prepared_messages: + answer = llm.call( + prepared_messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, + response_model=response_model, + ) return _validate_and_finalize_llm_response( answer, executor_context, printer, verbose=verbose @@ -630,17 +641,18 @@ async def aget_llm_response( Exception: If an error occurs. ValueError: If the response is None or empty. """ - messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose) - - answer = await llm.acall( - messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - response_model=response_model, - ) + with _prepare_llm_call( + executor_context, messages, printer, verbose=verbose + ) as prepared_messages: + answer = await llm.acall( + prepared_messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, + response_model=response_model, + ) return _validate_and_finalize_llm_response( answer, executor_context, printer, verbose=verbose @@ -1953,16 +1965,23 @@ def _setup_before_llm_call_hooks( verbose: Whether to print output. Returns: - True if LLM execution should proceed, False if blocked by a hook. + True if LLM execution should proceed, False if a hook blocked it by + returning ``False``. + + Raises: + HookAborted: If a hook raised it, so the deny reaches the caller intact. """ if executor_context: from crewai.hooks.dispatch import ( - HookAborted, InterceptionPoint, get_scoped_hooks, run_hooks, ) - from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer + from crewai.hooks.llm_hooks import ( + LLMCallHookContext, + LegacyHookBlocked, + before_llm_call_reducer, + ) # Executor snapshot first, then execution-scoped hooks — the same # ordering dispatch() applies to global vs scoped hooks. @@ -1984,7 +2003,7 @@ def _setup_before_llm_call_hooks( reducer=before_llm_call_reducer, verbose=verbose, ) - except HookAborted: + except LegacyHookBlocked: if verbose: printer.print( content="LLM call blocked by before_llm_call hook", diff --git a/lib/crewai/src/crewai/utilities/converter.py b/lib/crewai/src/crewai/utilities/converter.py index d31b76f48..f55888589 100644 --- a/lib/crewai/src/crewai/utilities/converter.py +++ b/lib/crewai/src/crewai/utilities/converter.py @@ -93,6 +93,8 @@ class Converter(OutputConverter): Raises: ConverterError: If conversion fails after maximum attempts. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): response = self.llm.call( @@ -108,6 +110,8 @@ class Converter(OutputConverter): raise ConverterError( f"Failed to convert text into a Pydantic model due to validation error: {e}" ) from e + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return self.to_pydantic(current_attempt + 1) @@ -117,6 +121,8 @@ class Converter(OutputConverter): async def ato_pydantic(self, current_attempt: int = 1) -> BaseModel: """Async equivalent of ``to_pydantic`` — uses ``acall`` so the event loop is not blocked.""" + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): response = await self.llm.acall( @@ -132,6 +138,8 @@ class Converter(OutputConverter): raise ConverterError( f"Failed to convert text into a Pydantic model due to validation error: {e}" ) from e + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return await self.ato_pydantic(current_attempt + 1) @@ -152,10 +160,14 @@ class Converter(OutputConverter): ConverterError: If conversion fails after maximum attempts. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): return self._create_instructor().to_json() return json.dumps(self.llm.call(self._build_messages())) + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return self.to_json(current_attempt + 1) @@ -168,10 +180,14 @@ class Converter(OutputConverter): sync-only); we run it via ``asyncio.to_thread`` so the event loop stays free. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): return await asyncio.to_thread(self._create_instructor().to_json) return json.dumps(await self.llm.acall(self._build_messages())) + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return await self.ato_json(current_attempt + 1) diff --git a/lib/crewai/src/crewai/utilities/crew_chat.py b/lib/crewai/src/crewai/utilities/crew_chat.py index 0ae40ea5b..8fa5a685c 100644 --- a/lib/crewai/src/crewai/utilities/crew_chat.py +++ b/lib/crewai/src/crewai/utilities/crew_chat.py @@ -417,6 +417,8 @@ def generate_input_description_with_ai( Returns: A concise description of the input. """ + from crewai.hooks.dispatch import HookAborted + context_texts = [] placeholder_pattern = re.compile(r"\{(.+?)}") @@ -460,6 +462,8 @@ def generate_input_description_with_ai( ) try: response = chat_llm.call(messages=[{"role": "user", "content": prompt}]) + except HookAborted: + raise except Exception as exc: click.secho( f"Warning: failed to generate input description for '{input_name}' " @@ -480,6 +484,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st Returns: A concise description of the crew's purpose (15 words or less). """ + from crewai.hooks.dispatch import HookAborted + context_texts = [] placeholder_pattern = re.compile(r"\{(.+?)}") @@ -514,6 +520,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st ) try: response = chat_llm.call(messages=[{"role": "user", "content": prompt}]) + except HookAborted: + raise except Exception as exc: click.secho( f"Warning: failed to generate crew description ({exc}); using default.", diff --git a/lib/crewai/src/crewai/utilities/guardrail.py b/lib/crewai/src/crewai/utilities/guardrail.py index faf27fa9f..8454ffd92 100644 --- a/lib/crewai/src/crewai/utilities/guardrail.py +++ b/lib/crewai/src/crewai/utilities/guardrail.py @@ -144,6 +144,7 @@ def process_guardrail( Raises: TypeError: If output is not a TaskOutput or LiteAgentOutput ValueError: If guardrail is None + HookAborted: A `pre_model_call` hook denied an LLM-backed guardrail. """ from crewai.lite_agent_output import LiteAgentOutput from crewai.tasks.task_output import TaskOutput @@ -158,6 +159,7 @@ def process_guardrail( LLMGuardrailCompletedEvent, LLMGuardrailStartedEvent, ) + from crewai.hooks.dispatch import HookAborted started_event = LLMGuardrailStartedEvent( guardrail=guardrail, @@ -167,7 +169,26 @@ def process_guardrail( ) crewai_event_bus.emit(event_source, started_event) - result = guardrail(output) + try: + result = guardrail(output) + except HookAborted as e: + # a deny ends the validation, so the started event above still needs a + # terminal one before it leaves + crewai_event_bus.emit( + event_source, + LLMGuardrailCompletedEvent( + success=False, + result=None, + error=str(e), + retry_count=retry_count, + guardrail_type=started_event.guardrail_type, + guardrail_name=started_event.guardrail_name, + from_agent=from_agent, + from_task=from_task, + ), + ) + raise + guardrail_result = GuardrailResult.from_tuple(result) crewai_event_bus.emit( diff --git a/lib/crewai/src/crewai/utilities/internal_instructor.py b/lib/crewai/src/crewai/utilities/internal_instructor.py index 074f3a824..21020275a 100644 --- a/lib/crewai/src/crewai/utilities/internal_instructor.py +++ b/lib/crewai/src/crewai/utilities/internal_instructor.py @@ -143,8 +143,12 @@ class InternalInstructor(Generic[T]): Instance of the specified Pydantic model with structured data Raises: - ValueError: If LLM is not provided or invalid + ValueError: If LLM is not provided or invalid, or if a hook blocked + the call by returning ``False``. + HookAborted: If a hook denied the call. """ + from crewai.llms.base_llm import BaseLLM + messages: list[LLMMessage] = [{"role": "user", "content": self.content}] if not _is_valid_llm(self.llm): @@ -157,6 +161,11 @@ class InternalInstructor(Generic[T]): else: model_name = self.llm.model + if isinstance(self.llm, BaseLLM): + # This reaches the provider client directly, so the hooks the LLM + # layer would have dispatched are dispatched here instead. + self.llm._invoke_before_llm_call_hooks(messages, self.agent) + return self._client.chat.completions.create( # type: ignore[no-any-return] model=model_name, response_model=self.model, messages=messages ) diff --git a/lib/crewai/src/crewai/utilities/reasoning_handler.py b/lib/crewai/src/crewai/utilities/reasoning_handler.py index 1028a3f3d..21c0a1e1d 100644 --- a/lib/crewai/src/crewai/utilities/reasoning_handler.py +++ b/lib/crewai/src/crewai/utilities/reasoning_handler.py @@ -362,6 +362,8 @@ class AgentReasoning: """ self.logger.debug(f"Using function calling for {plan_type} planning") + from crewai.hooks.dispatch import HookAborted + try: system_prompt = self._get_system_prompt() @@ -412,6 +414,8 @@ class AgentReasoning: "READY: I am ready to execute the task." in response_str, ) + except HookAborted: + raise except Exception as e: self.logger.warning( f"Error during function calling: {e!s}. Falling back to text parsing." @@ -435,6 +439,8 @@ class AgentReasoning: [], "READY: I am ready to execute the task." in fallback_str, ) + except HookAborted: + raise except Exception as inner_e: self.logger.error(f"Error during fallback text parsing: {inner_e!s}") return ( diff --git a/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py b/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py new file mode 100644 index 000000000..303b69ee7 --- /dev/null +++ b/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py @@ -0,0 +1,309 @@ +"""A deny raised inside a run has to reach whoever started the run. + +The sibling propagation tests all call the frame that makes the model call +directly, so they prove a deny escapes *that* function and nothing about what +its callers do with it. Every regression in this area has lived one or more +frames up, in a broad ``except Exception`` that turned the deny into a degraded +result. These tests drive the public entry points instead, and count model calls +so a deny that gets retried reads as a failure rather than as a pass. +""" + +from __future__ import annotations + +from typing import Any + +from crewai.agent import Agent +from crewai.agents.step_executor import StepExecutor +from crewai.crew import Crew +from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on +from crewai.lite_agent import LiteAgent +from crewai.llms.base_llm import BaseLLM +from crewai.task import Task +from crewai.utilities.planning_types import TodoItem +from crewai.utilities.step_execution_context import StepExecutionContext +from crewai.utilities.types import LLMMessage +import pytest + +from ..utils import wait_for_event_handlers + + +class StubProviderLLM(BaseLLM): + """Answers without a network, dispatching the before hooks like a provider.""" + + def __init__(self, fail_first_call: bool = False) -> None: + super().__init__(model="stub") + self.fail_first_call = fail_first_call + self.answered = 0 + + def call( + self, + messages: str | list[LLMMessage], + tools: list[dict[str, Any]] | None = None, + callbacks: list[Any] | None = None, + available_functions: dict[str, Any] | None = None, + from_task: Any | None = None, + from_agent: Any | None = None, + response_model: Any | None = None, + **kwargs: Any, + ) -> str: + formatted: list[LLMMessage] = ( + messages + if isinstance(messages, list) + else [{"role": "user", "content": messages}] + ) + self._invoke_before_llm_call_hooks(formatted, from_agent) + self.answered += 1 + if self.fail_first_call and self.answered == 1: + raise RuntimeError("the provider blipped") + return "Thought: done\nFinal Answer: ok" + + def supports_function_calling(self) -> bool: + return False + + +class DenyingMemory: + """Stands in for the memory whose own model call a hook denied.""" + + read_only = False + root_scope = None + + def __init__(self, deny_on: str, error: Exception | None = None) -> None: + self.deny_on = deny_on + self.error = error or HookAborted( + reason="memory is off limits", source="policy" + ) + self.touched: list[str] = [] + + def _step(self, name: str) -> None: + self.touched.append(name) + if name == self.deny_on: + raise self.error + + def drain_writes(self) -> None: + pass + + def recall(self, *args: Any, **kwargs: Any) -> list[Any]: + self._step("recall") + return [] + + def extract_memories(self, *args: Any, **kwargs: Any) -> list[str]: + self._step("extract_memories") + return ["a memory"] + + def remember_many(self, *args: Any, **kwargs: Any) -> None: + self._step("remember_many") + + def search(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + +class StubKnowledge: + def query(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + +class DenyingStepExecutor: + """Stands in for the per-step executor whose own model call was denied.""" + + def __init__(self) -> None: + self.executed = 0 + + def execute(self, *args: Any, **kwargs: Any) -> Any: + self.executed += 1 + raise HookAborted(reason="no model calls allowed", source="policy") + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # A kickoff emits events whose handlers run on a pool; draining them here + # keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +def deny_nth_model_call(n: int) -> list[str]: + """Deny the nth model call of the run, returning the log of attempts.""" + attempts: list[str] = [] + + @on(InterceptionPoint.PRE_MODEL_CALL) + def gate(_ctx: Any) -> None: + attempts.append("attempt") + if len(attempts) == n: + raise HookAborted(reason="no model calls allowed", source="policy") + + return attempts + + +def build_agent(**kwargs: Any) -> Agent: + return Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + **kwargs, + ) + + +def build_crew(agent: Agent, **task_kwargs: Any) -> Crew: + task = Task( + description="Say ok", expected_output="ok", agent=agent, **task_kwargs + ) + return Crew(agents=[agent], tasks=[task]) + + +def run_agent_kickoff() -> Any: + return build_agent().kickoff("say ok") + + +def run_agent_kickoff_with_planning() -> Any: + return build_agent(planning=True).kickoff("say ok") + + +def run_crew_kickoff() -> Any: + return build_crew(build_agent()).kickoff() + + +def run_crew_kickoff_with_planning() -> Any: + return build_crew(build_agent(planning=True)).kickoff() + + +def run_crew_kickoff_with_knowledge() -> Any: + agent = build_agent() + agent.knowledge = StubKnowledge() + return build_crew(agent).kickoff() + + +def run_lite_agent_kickoff() -> Any: + return LiteAgent( + role="Worker", goal="Answer", backstory="You answer.", llm=StubProviderLLM() + ).kickoff("say ok") + + +@pytest.mark.parametrize( + "entry_point", + [ + run_agent_kickoff, + run_agent_kickoff_with_planning, + run_crew_kickoff, + run_crew_kickoff_with_planning, + run_crew_kickoff_with_knowledge, + run_lite_agent_kickoff, + ], + ids=[ + "agent.kickoff", + "agent.kickoff-planning", + "crew.kickoff", + "crew.kickoff-planning", + "crew.kickoff-knowledge", + "lite_agent.kickoff", + ], +) +def test_a_denied_model_call_reaches_the_caller(entry_point): + attempts = deny_nth_model_call(1) + + with pytest.raises(HookAborted): + entry_point() + + assert len(attempts) == 1 + + +def test_a_denied_guardrail_stops_the_crew_instead_of_retrying_the_task(): + attempts = deny_nth_model_call(2) + crew = build_crew(build_agent(), guardrail="The answer must be polite") + + with pytest.raises(HookAborted): + crew.kickoff() + + # the answer, then the denied validation, and nothing after it + assert len(attempts) == 2 + + +@pytest.mark.parametrize("deny_on", ["recall", "extract_memories"]) +def test_a_denied_memory_step_reaches_the_caller(deny_on): + memory = DenyingMemory(deny_on) + agent = build_agent() + agent.memory = memory + + with pytest.raises(HookAborted): + agent.kickoff("say ok") + + assert memory.touched.count(deny_on) == 1 + + +def test_a_denied_memory_save_stops_the_crew_instead_of_retrying_the_task(): + memory = DenyingMemory("extract_memories") + agent = build_agent() + agent.memory = memory + + with pytest.raises(HookAborted): + build_crew(agent).kickoff() + + assert memory.touched.count("extract_memories") == 1 + + +def build_step_executor(llm: StubProviderLLM) -> StepExecutor: + return StepExecutor(llm=llm, tools=[], agent=build_agent()) + + +def a_step() -> tuple[TodoItem, StepExecutionContext]: + return ( + TodoItem(step_number=1, description="Say ok"), + StepExecutionContext(task_description="Say ok", task_goal="ok"), + ) + + +def test_a_denied_step_stops_the_plan_instead_of_reporting_a_failed_step(): + attempts = deny_nth_model_call(1) + todo, context = a_step() + + with pytest.raises(HookAborted): + build_step_executor(StubProviderLLM()).execute(todo, context) + + assert len(attempts) == 1 + + +def test_an_ordinary_step_failure_still_reports_a_failed_step(): + todo, context = a_step() + + result = build_step_executor(StubProviderLLM(fail_first_call=True)).execute( + todo, context + ) + + assert result.success is False + + +@pytest.mark.asyncio +async def test_a_denied_parallel_step_reaches_the_caller(): + agent = build_agent() + executor = AgentExecutor(agent=agent, llm=agent.llm, task=None) + executor.state.todos.items = [ + TodoItem(step_number=1, description="first"), + TodoItem(step_number=2, description="second"), + ] + step_executor = DenyingStepExecutor() + object.__setattr__(executor, "_ensure_step_executor", lambda: step_executor) + + with pytest.raises(HookAborted): + await executor.execute_todos_parallel() + + +def test_an_ordinary_model_failure_is_still_retried(): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(fail_first_call=True), + ) + + assert "ok" in str(build_crew(agent).kickoff()) + + +def test_an_ordinary_memory_failure_still_degrades(): + memory = DenyingMemory("extract_memories", error=RuntimeError("storage is down")) + agent = build_agent() + agent.memory = memory + + assert str(agent.kickoff("say ok")) == "ok" diff --git a/lib/crewai/tests/hooks/test_hook_abort_propagation.py b/lib/crewai/tests/hooks/test_hook_abort_propagation.py new file mode 100644 index 000000000..e498eaa04 --- /dev/null +++ b/lib/crewai/tests/hooks/test_hook_abort_propagation.py @@ -0,0 +1,522 @@ +"""A model-call deny must reach the caller as a deny. + +Two layers used to erase it. The LLM layer caught ``HookAborted`` and returned +``False``, which every provider turned into ``ValueError("LLM call blocked...")`` +— losing the reason, the source, and any way to tell a policy decision from a +provider outage. Downstream, every internal model call is wrapped in +``except Exception`` so a provider hiccup degrades instead of failing the run, +and those handlers then absorbed the flattened deny, often retrying the very +call that was just denied. + +A deny also owes the started event a terminal one, and owes it an honest label: +it is a decision, not an outage the provider caused. + +These use a real ``LLM`` with a real ``pre_model_call`` hook: the deny fires +inside ``dispatch`` before the provider is reached, so nothing here needs a +network or a cassette. +""" + +from __future__ import annotations + +from typing import Any + +from crewai.agent import Agent +from crewai.agent.planning_config import PlanningConfig +from crewai.agent.utils import ( + ahandle_knowledge_retrieval, + handle_knowledge_retrieval, + handle_reasoning, +) +from crewai.agents.planner_observer import PlannerObserver +from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.knowledge_events import ( + KnowledgeQueryFailedEvent, + KnowledgeQueryStartedEvent, +) +from crewai.events.types.llm_events import LLMCallFailedEvent +from crewai.events.types.llm_guardrail_events import ( + LLMGuardrailCompletedEvent, + LLMGuardrailStartedEvent, +) +from crewai.events.types.observation_events import ( + StepObservationFailedEvent, + StepObservationStartedEvent, +) +from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear_all, + on, +) +from crewai.hooks.llm_hooks import register_before_llm_call_hook +from crewai.llm import LLM +from crewai.memory.analyze import ( + analyze_for_consolidation, + analyze_for_save, + analyze_query, + extract_memories_from_content, +) +from crewai.memory.types import MemoryRecord +from crewai.task import Task +from crewai.tasks.llm_guardrail import LLMGuardrail +from crewai.tasks.task_output import TaskOutput +from crewai.utilities.converter import Converter +from crewai.utilities.guardrail import process_guardrail +from crewai.utilities.planning_types import TodoItem +import pytest +from pydantic import BaseModel + +from ..utils import wait_for_event_handlers + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # These calls emit LLM events whose handlers run on a pool; draining them + # here keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +@pytest.fixture +def denied_calls() -> list[Any]: + return [] + + +@pytest.fixture +def denying_llm(denied_calls: list[Any]) -> LLM: + @on(InterceptionPoint.PRE_MODEL_CALL) + def deny(ctx: Any) -> None: + denied_calls.append(ctx) + raise HookAborted(reason="no model calls allowed", source="policy") + + return LLM(model="gpt-4o-mini") + + +class _Person(BaseModel): + name: str + + +def test_a_raised_deny_keeps_its_reason_out_of_the_llm_layer(denying_llm): + with pytest.raises(HookAborted) as exc: + denying_llm.call([{"role": "user", "content": "hi"}]) + + assert exc.value.reason == "no model calls allowed" + assert exc.value.source == "policy" + + +def test_the_boolean_convention_still_blocks_with_the_documented_error(): + register_before_llm_call_hook(lambda _ctx: False) + + with pytest.raises(ValueError, match="LLM call blocked by before_llm_call hook"): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + +def test_the_boolean_convention_is_still_absorbed_by_a_fail_open_handler(): + # Unlike a raised abort, the boolean deny must keep degrading rather than + # failing the run — otherwise adopting this fix breaks existing hooks. + register_before_llm_call_hook(lambda _ctx: False) + + analysis = analyze_query("a query", ["/"], None, LLM(model="gpt-4o-mini")) + + assert analysis.recall_queries == ["a query"] + + +@pytest.mark.parametrize( + "denied_helper", + [ + lambda llm: extract_memories_from_content("some content", llm), + lambda llm: analyze_query("a query", ["/"], None, llm), + lambda llm: analyze_for_save("some content", ["/"], [], llm), + lambda llm: analyze_for_consolidation( + "new content", + [MemoryRecord(id="1", content="old content", scope="/")], + llm, + ), + ], + ids=["extract", "query", "save", "consolidate"], +) +def test_memory_analysis_surfaces_a_deny_instead_of_a_safe_default( + denied_helper, denying_llm, denied_calls +): + with pytest.raises(HookAborted): + denied_helper(denying_llm) + + assert denied_calls, "the helper never reached the model" + + +def test_a_denied_conversion_is_not_retried(denying_llm, denied_calls): + converter = Converter( + text="Name: Ada", + llm=denying_llm, + model=_Person, + instructions="Extract the person", + max_attempts=3, + ) + + with pytest.raises(HookAborted): + converter.to_pydantic() + + assert len(denied_calls) == 1 + + +def test_a_denied_knowledge_query_still_reports_the_failure_it_started(denying_llm): + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + started: list[KnowledgeQueryStartedEvent] = [] + failed: list[KnowledgeQueryFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(KnowledgeQueryStartedEvent) + def _on_started(_source: Any, event: KnowledgeQueryStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(KnowledgeQueryFailedEvent) + def _on_failed(_source: Any, event: KnowledgeQueryFailedEvent) -> None: + failed.append(event) + + with pytest.raises(HookAborted): + agent._get_knowledge_search_query(task.description, task) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(failed) == 1 + assert failed[0].error == "no model calls allowed" + + +def test_a_denied_step_observation_still_reports_the_failure_it_started(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + ) + observer = PlannerObserver(agent=agent, task=None) + step = TodoItem(step_number=1, description="do the thing", result="done") + started: list[StepObservationStartedEvent] = [] + failed: list[StepObservationFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(StepObservationStartedEvent) + def _on_started(_source: Any, event: StepObservationStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(StepObservationFailedEvent) + def _on_failed(_source: Any, event: StepObservationFailedEvent) -> None: + failed.append(event) + + with pytest.raises(HookAborted): + observer.observe(step, "done", [], []) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(failed) == 1 + assert failed[0].error == "no model calls allowed" + assert failed[0].step_number == 1 + + +def test_a_denied_knowledge_retrieval_does_not_fall_back_to_the_plain_prompt( + denying_llm, +): + # the retrieval helper wraps the query rewrite in its own except Exception, + # so guarding the rewrite alone still let the task run without knowledge + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + # only has to be truthy: the deny fires before any knowledge is queried + agent.knowledge = object() + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + + with pytest.raises(HookAborted): + handle_knowledge_retrieval( + agent, + task, + "the task prompt", + {}, + lambda *_args, **_kwargs: [], + lambda *_args, **_kwargs: [], + ) + + +@pytest.mark.asyncio +async def test_a_denied_async_knowledge_retrieval_reaches_the_caller(denying_llm): + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + agent.knowledge = object() + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + + with pytest.raises(HookAborted): + await ahandle_knowledge_retrieval(agent, task, "the task prompt", {}) + + +def test_a_denied_plan_stops_the_legacy_planning_path(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning=True, + ) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + with pytest.raises(HookAborted): + handle_reasoning(agent, task) + + +def test_a_denied_replan_does_not_keep_executing_the_stale_plan(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning_config=PlanningConfig( + reasoning_effort="low", + max_attempts=1, + max_steps=2, + max_replans=1, + max_step_iterations=2, + ), + ) + executor = AgentExecutor(agent=agent, llm=denying_llm, task=None) + executor._kickoff_input = "do the thing" + + with pytest.raises(HookAborted): + executor._trigger_replan("the first plan failed") + + +class _DenyingMemory: + """Stands in for unified memory whose model call was denied upstream.""" + + read_only = False + root_scope = None + + def __init__(self, error: Exception): + self._error = error + + def drain_writes(self) -> None: + pass + + def recall(self, *_args: Any, **_kwargs: Any): + raise self._error + + def extract_memories(self, *_args: Any, **_kwargs: Any): + raise self._error + + def remember_many(self, *_args: Any, **_kwargs: Any) -> None: + pass + + +def test_a_denied_memory_recall_does_not_run_the_task_without_memory(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy")) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + with pytest.raises(HookAborted): + agent._retrieve_memory_context(task, "the task prompt") + + +def test_an_ordinary_memory_recall_failure_still_degrades(): + # the guard must single out a deny: a broken store has always been allowed + # to degrade to no memory, and turning that into a failed run is a break + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(ValueError("vector store is down")) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + assert agent._retrieve_memory_context(task, "the prompt") == "the prompt" + + +def test_a_denied_memory_save_does_not_report_a_clean_kickoff(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy")) + + with pytest.raises(HookAborted): + agent._save_kickoff_to_memory("the input", "the output") + + +def test_an_ordinary_memory_save_failure_still_degrades(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(ValueError("vector store is down")) + + agent._save_kickoff_to_memory("the input", "the output") + + +def test_a_denied_guardrail_does_not_read_as_a_failed_validation(denying_llm): + # returning (False, "Error while validating...") would feed the agent a + # retry prompt built from a call the policy refused + guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm) + task_output = TaskOutput( + description="Say hi", raw="hi", agent="Doer", expected_output="A greeting." + ) + + with pytest.raises(HookAborted): + guardrail(task_output) + + +def test_a_denied_guardrail_still_reports_the_validation_it_started(denying_llm): + guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm) + task_output = TaskOutput( + description="Say hi", raw="hi", agent="Doer", expected_output="A greeting." + ) + started: list[LLMGuardrailStartedEvent] = [] + completed: list[LLMGuardrailCompletedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMGuardrailStartedEvent) + def _on_started(_source: Any, event: LLMGuardrailStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(LLMGuardrailCompletedEvent) + def _on_completed(_source: Any, event: LLMGuardrailCompletedEvent) -> None: + completed.append(event) + + with pytest.raises(HookAborted): + process_guardrail(output=task_output, guardrail=guardrail, retry_count=0) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(completed) == 1 + assert completed[0].success is False + assert "no model calls allowed" in (completed[0].error or "") + + +def test_a_denied_plan_stops_the_executor_instead_of_running_unplanned(denying_llm): + # the executor wraps planning in its own except Exception, so guarding the + # reasoning handler alone still left the agent proceeding with no plan + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning_config=PlanningConfig( + reasoning_effort="low", + max_attempts=1, + max_steps=2, + max_replans=0, + max_step_iterations=2, + ), + ) + executor = AgentExecutor(agent=agent, llm=denying_llm, task=None) + executor._kickoff_input = "do the thing" + + with pytest.raises(HookAborted): + executor.generate_plan() + + +@pytest.mark.asyncio +async def test_an_async_call_is_denied_like_a_sync_one(denying_llm): + with pytest.raises(HookAborted) as exc: + await denying_llm.acall([{"role": "user", "content": "hi"}]) + + assert exc.value.reason == "no model calls allowed" + assert exc.value.source == "policy" + + +def test_a_denied_structured_conversion_reaches_the_caller(denying_llm): + # The function-calling path goes through Instructor, which reaches the + # provider client without passing through ``llm.call``. + assert denying_llm.supports_function_calling() + converter = Converter( + text="Name: Ada", + llm=denying_llm, + model=_Person, + instructions="Extract the person", + max_attempts=1, + ) + + with pytest.raises(HookAborted): + converter.to_json() + + +def test_a_deny_is_not_reported_as_a_provider_failure(denying_llm): + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(HookAborted): + denying_llm.call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == "LLM call denied by policy: no model calls allowed" + + +def test_a_deny_names_the_hook_that_raised_it_rather_than_its_repr(): + def gate_on_approved_models(_ctx: Any) -> None: + raise HookAborted(reason="model not approved", source=gate_on_approved_models) + + on(InterceptionPoint.PRE_MODEL_CALL)(gate_on_approved_models) + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(HookAborted): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == ( + "LLM call denied by gate_on_approved_models: model not approved" + ) + + +def test_the_boolean_convention_is_not_reported_as_a_provider_failure(): + register_before_llm_call_hook(lambda _ctx: False) + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(ValueError): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == ( + "LLM call denied by hook: LLM call blocked by before_llm_call hook" + ) diff --git a/lib/crewai/tests/hooks/test_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index 0c917712d..42c7c9ef4 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -646,11 +646,27 @@ class TestDirectLLMScopedHooks: with scoped_hooks(): register_scoped(InterceptionPoint.PRE_MODEL_CALL, block) - proceed = llm._invoke_before_llm_call_hooks( - [{"role": "user", "content": "hi"}], from_agent=None - ) + with pytest.raises(HookAborted, match="blocked by scoped hook"): + llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) - assert proceed is False + def test_a_scoped_hook_returning_false_blocks_the_call(self): + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import register_scoped, scoped_hooks + from crewai.llms.base_llm import LLMCallBlockedError + + llm = self._stub_llm() + + with scoped_hooks(): + register_scoped(InterceptionPoint.PRE_MODEL_CALL, lambda _ctx: False) + with pytest.raises(LLMCallBlockedError) as blocked: + llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) + + # a ValueError so the fail-open handlers around internal calls still absorb it + assert isinstance(blocked.value, ValueError) def test_scoped_after_hook_modifies_direct_response(self): from crewai.hooks import InterceptionPoint diff --git a/lib/crewai/tests/hooks/test_model_call_hook_reach.py b/lib/crewai/tests/hooks/test_model_call_hook_reach.py new file mode 100644 index 000000000..ad89d9239 --- /dev/null +++ b/lib/crewai/tests/hooks/test_model_call_hook_reach.py @@ -0,0 +1,149 @@ +"""Which model calls the ``pre_model_call`` hooks can see, and how many times. + +The LLM layer used to skip the hooks whenever a call carried an agent, assuming +the executor had already dispatched them. That holds inside the executor loop +and left every other agent-bearing call — step observation, planning, plan +synthesis — invisible. The executor now marks the window where it already +dispatched, which is what keeps a call from being seen twice. + +The stub stands in for a native provider: it invokes the before hooks the way +every provider does and answers without a network. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +from crewai.agent import Agent +from crewai.hooks.dispatch import InterceptionPoint, clear_all, on +from crewai.llms.base_llm import BaseLLM +from crewai.utilities.agent_utils import get_llm_response +from crewai.utilities.types import LLMMessage +from crewai_core.printer import Printer +from pydantic import BaseModel +import pytest + +from ..utils import wait_for_event_handlers + + +class StubProviderLLM(BaseLLM): + def __init__(self) -> None: + super().__init__(model="stub") + + def call( + self, + messages: str | list[LLMMessage], + tools: list[dict[str, Any]] | None = None, + callbacks: list[Any] | None = None, + available_functions: dict[str, Any] | None = None, + from_task: Any | None = None, + from_agent: Any | None = None, + response_model: Any | None = None, + **kwargs: Any, + ) -> str: + formatted: list[LLMMessage] = ( + messages + if isinstance(messages, list) + else [{"role": "user", "content": messages}] + ) + self._invoke_before_llm_call_hooks(formatted, from_agent) + return "Thought: done\nFinal Answer: ok" + + def supports_function_calling(self) -> bool: + return False + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # A kickoff emits events whose handlers run on a pool; draining them here + # keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +@pytest.fixture +def seen_agents() -> list[str | None]: + roles: list[str | None] = [] + + @on(InterceptionPoint.PRE_MODEL_CALL) + def record(ctx: Any) -> None: + agent = getattr(ctx, "agent", None) + roles.append(getattr(agent, "role", None)) + + return roles + + +def test_the_executor_loop_is_seen_exactly_once(seen_agents): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + ) + + assert str(agent.kickoff("say ok")) == "ok" + assert seen_agents == ["Worker"] + + +def test_a_direct_call_carrying_an_agent_is_seen_once(seen_agents): + agent = Agent( + role="Planner", + goal="Plan", + backstory="You plan.", + llm=StubProviderLLM(), + ) + + StubProviderLLM().call([{"role": "user", "content": "hi"}], from_agent=agent) + + assert seen_agents == ["Planner"] + + +def test_a_direct_call_without_an_agent_is_seen_once(seen_agents): + StubProviderLLM().call([{"role": "user", "content": "hi"}]) + + assert seen_agents == [None] + + +def test_a_structured_litellm_call_is_seen_once(seen_agents, monkeypatch): + pytest.importorskip("litellm") + import instructor + + from crewai.llm import LLM + + class Answer(BaseModel): + text: str + + client = SimpleNamespace( + chat=SimpleNamespace( + completions=SimpleNamespace(create=lambda **_: Answer(text="ok")) + ) + ) + monkeypatch.setattr(instructor, "from_litellm", lambda *_, **__: client) + + llm = LLM(model="openai/gpt-4o-mini", is_litellm=True) + llm.call([{"role": "user", "content": "hi"}], response_model=Answer) + + assert seen_agents == [None] + + +def test_an_agent_call_outside_the_executor_loop_is_seen_once(seen_agents): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + ) + + get_llm_response( + llm=StubProviderLLM(), + messages=[{"role": "user", "content": "hi"}], + callbacks=[], + printer=Printer(), + from_agent=agent, + executor_context=None, + ) + + assert seen_agents == ["Worker"]