fix: harden interception dispatcher against review findings

Corrects several dispatcher edge cases surfaced in review. `_default_reducer`
now reports a modification only when a `payload` is actually applied, the
`agents=` filter falls back to `agent_role` for contexts without an `agent`
object, and `unregister` resolves the filter wrapper stashed by `on` so a
filtered hook can be removed. The tool-hook runners honor the executing
agent's `verbose` flag instead of silently swallowing hook errors, and the
ReAct tool path now runs `POST_TOOL_CALL` on blocked calls to match the
native paths. Also adds abort-telemetry coverage and replaces the flaky
absolute no-op timing budget with a relative one.
This commit is contained in:
Lucas Gomide
2026-07-11 19:12:15 -03:00
parent a85e100bec
commit f7bd240499
4 changed files with 135 additions and 15 deletions

View File

@@ -129,9 +129,17 @@ def register(point: InterceptionPoint, hook: HookFn) -> None:
def unregister(point: InterceptionPoint, hook: HookFn) -> bool:
"""Unregister a specific global hook. Returns True if it was removed."""
"""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:
_global_hooks[point].remove(hook)
hooks.remove(target)
return True
except ValueError:
return False
@@ -235,10 +243,14 @@ def _emit_telemetry(
def _default_reducer(ctx: Any, result: Any) -> bool:
"""Default payload semantics: a non-None return replaces ``ctx.payload``."""
if result is not None:
if hasattr(ctx, "payload"):
ctx.payload = result
"""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
@@ -368,7 +380,10 @@ def _wrap_with_filters(
return None
if agents:
agent = getattr(ctx, "agent", None)
if agent is not None and getattr(agent, "role", None) not in agents:
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)
@@ -412,6 +427,9 @@ def on(
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

View File

@@ -160,6 +160,15 @@ def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> boo
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.
@@ -173,7 +182,7 @@ def run_before_tool_call_hooks(context: ToolCallHookContext) -> bool:
InterceptionPoint.PRE_TOOL_CALL,
context,
reducer=before_tool_call_reducer,
verbose=False,
verbose=_hook_verbose(context),
)
return False
except HookAborted:
@@ -190,7 +199,7 @@ def run_after_tool_call_hooks(context: ToolCallHookContext) -> str | None:
InterceptionPoint.POST_TOOL_CALL,
context,
reducer=after_tool_call_reducer,
verbose=False,
verbose=_hook_verbose(context),
)
return context.tool_result

View File

@@ -104,7 +104,20 @@ async def aexecute_tool_and_check_finality(
blocked_message = (
f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
)
return ToolResult(blocked_message, False)
# 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)
@@ -207,7 +220,20 @@ def execute_tool_and_check_finality(
blocked_message = (
f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
)
return ToolResult(blocked_message, False)
# 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)

View File

@@ -23,6 +23,7 @@ from crewai.hooks.dispatch import (
register,
register_scoped,
scoped_hooks,
unregister as unregister_hook,
)
from crewai.hooks.llm_hooks import (
get_before_llm_call_hooks,
@@ -36,6 +37,7 @@ class _Ctx:
payload: object = None
tool_name: str | None = None
agent: object = None
agent_role: str | None = None
@pytest.fixture(autouse=True)
@@ -143,6 +145,27 @@ class TestOnDecorator:
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."""
@@ -208,22 +231,66 @@ class TestTelemetry:
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_budget(self):
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)
elapsed = time.perf_counter() - start
# Generous CI-safe budget: < 5µs per no-op dispatch on average.
assert elapsed / iterations < 5e-6
noop = time.perf_counter() - start
assert noop < baseline * 50 + 5e-3