[rkfshakti]

Fixes #8904.

Follow-up to #7630 (the closed #7634 was the pre-#8192 fix for the same family).

The bug (reproduced on main)

After #8192, every sync Flow method runs in a worker thread (asyncio.to_thread, flow runtime _execute_method). An async native tool's coroutine therefore goes through the bare asyncio.run() inside Tool.run — on a fresh, unrelated event loop per call. Anything bound to the kickoff caller's loop fails deterministically:

RuntimeError: Task ... got Future <Future pending> attached to a different loop

Minimal repro on unpatched main: a Crew with one @tool async def; the tool awaits a Future created on the kickoff_async caller's loop after asyncio.run(crew.kickoff_async()) → the tool result the LLM sees is Error executing tool: ... attached to a different loop. Same shape for any caller-loop-bound resource (httpx.AsyncClient created before kickoff, sockets).

The fix

• The async kickoffs capture the caller's event loop into a ContextVar (Crew.kickoff_async non-stream branch, AgentExecutor.invoke_async). Crew.kickoff_async is await asyncio.to_thread(self.kickoff, ...), so capturing before the thread boundary is what makes the loop visible inside every worker lane — ContextVars propagate through to_thread and the Flow runtime's copy_context (including the ThreadPoolExecutor in execute_native_tool).
• agent_utils.run_native_tool_result(...) executes the tool: when a caller loop is present and the tool has a working async implementation, bridge tool.arun(**args) onto the caller loop via run_coroutine_threadsafe(...).result() (the worker thread blocks; the coroutine runs with caller-loop affinity). Otherwise the exact historical sync behavior — the available_functions mapping stays authoritative, so test doubles and cache wrappers are untouched.
• Async-capability detection: Tool-wrapped functions are async iff the wrapped function is a coroutine function (Tool always overrides _arun, even for sync funcs); class-based BaseTools qualify when their type overrides the NotImplementedError-raising BaseTool._arun.

Tests

New lib/crewai/tests/agents/test_native_async_tool_loop_affinity.py:

• worker-thread bridge onto a caller-loop Future (unit, with and without caller loop),
• full Crew + scripted-LLM kickoff_async path asserting the caller-loop Future resolves instead of the cross-loop error.

Verification

• ruff check clean on all touched files.
• A/B vs pristine main: pytest lib/crewai/tests/agents -k "native or agent_executor or invoke" — identical failure sets (3 pre-existing TestDeferredNativeToolResults / TestResponseFormatWithKickoff failures reproduce on pristine main in this environment; 172 pass on both), plus 4 new green tests.
• With the patch, the repro above returns got:expected-item instead of the error.

---

Note

Medium Risk
Changes core async kickoff and native-tool concurrency (ContextVar bridging, thread pools, blocking waits); behavior is scoped and heavily tested but affects all async kickoff + async tool combinations.

Overview
Fixes async native tools breaking under kickoff_async when sync Flow work runs in worker threads: coroutines no longer spin up a fresh loop per call, so caller-loop-bound resources (e.g. httpx.AsyncClient, Futures) stop failing with attached to a different loop.

Caller-loop capture: Crew.kickoff_async (streaming and non-streaming) and AgentExecutor.invoke_async store the running loop in a ContextVar for the duration of the run, so worker-thread tool paths can see it after to_thread / context copy.

Tool execution: Native tool dispatch in agent_executor and agent_utils now goes through run_native_tool_result, which schedules tool.arun on that loop via run_coroutine_threadsafe when the tool is async; sync tools and missing-loop cases keep the old run() behavior. Bridging skips when already on the target loop (deadlock guard), propagates tool errors without double execution, and times out after CREWAI_NATIVE_TOOL_BRIDGE_TIMEOUT (default 300s) with coroutine cancellation.

Executor pool split: Sync kickoff bodies move from the loop’s default asyncio.to_thread pool to a dedicated run_in_kickoff_thread pool so kickoff threads waiting on bridged tools don’t starve the default pool when tools use asyncio.to_thread inside their coroutines.

Regression coverage is added in test_native_async_tool_loop_affinity.py (bridge behavior, E2E kickoff_async, streaming path, concurrent kickoffs, inherited _arun).

^{Reviewed by Cursor Bugbot for commit d874b95. Bugbot is set up for automated code reviews on this repo. Configure here.}

[strix-security]

Strix Security Review

Warning

This pull request has 5 commits after the last Strix review (5f948e3). Strix has not reviewed these changes.
Automatic review on push is off for this repository. To review the latest changes, tag @strix-security in a comment, or turn on re-review on push.

To fix before merging (1)

• 🔴 HIGH Unbounded bridge wait in run_native_tool_result permanently deadlocks the caller loop's default executor — lib/crewai/src/crewai/utilities/agent_utils.py:1556-1565

PR summary: This PR makes async native tools run on the event loop that called the async kickoff: a ContextVar captures the caller loop in `Crew.kickoff_async` and `AgentExecutor.invoke_async`, and a new `run_native_tool_result` helper bridges `tool.arun(...)` onto that loop with `run_coroutine_threadsafe(...).result()` while the worker thread waits, keeping the old direct call for sync tools.

Review summary

Reviewed the full diff and traced the new loop-bridging path through the crew entrypoints, AgentExecutor, the flow runtime's worker lanes, and the tool invocation chain. The bridging logic itself is sound on the axes that usually matter: argument validation and usage-limit accounting are enforced identically by run and arun, tool-call hooks, caching and failure policy sit outside the helper and are unaffected, the ContextVar is set/reset token-paired per task and does not leak across concurrent requests, and a scheduling failure cannot re-execute a tool. The real defect is the unbounded wait the bridge introduces: an asynchronous tool whose coroutine offloads blocking work with asyncio.to_thread — the very pattern the new docstring recommends — needs a thread from the same default executor that every in-flight kickoff already occupies, so once in-flight kickoffs reach the executor's worker count the process wedges permanently. The finding is titled "Unbounded bridge wait in run_native_tool_result permanently deadlocks the caller loop's default executor" and carries an A/B reproduction through the public Crew API. Separately, and not as a security matter: the fix is not applied to the stream=True branch of Crew.kickoff_async or to Crew.akickoff, and _has_async_implementation misses _arun overrides inherited from an intermediate base class, so the reported cross-loop bug still reproduces on those paths.

Ruled out (8)

• Does calling tool.arun instead of the mapped available_functions callable skip the tool cache, usage limits, approval hooks or failure policy?
  The mapping holds the same tool's bound run, and `_validate_kwargs`/`_claim_usage` are enforced in both `run` and `arun`, while caching, hooks and failure policy are dispatched around the call site rather than inside it.
• Can a scheduling failure in the bridge re-execute the tool and duplicate its side effects?
  `arun` is an async def, so building the coroutine performs no work, the `coro.close()` fallback discards an unstarted coroutine, and the fallback runs the tool exactly once.
• Can `_has_async_implementation` misclassifications escalate into a security problem?
  The two failure shapes require a developer-authored class that violates the documented `async def _arun` contract, are decided entirely by developer code rather than LLM input, and fail closed as an `Error executing tool` string.
• Does a third-party BaseTool that overrides run() but not arun() have its own gate skipped?
  No in-repo tool overrides `run`, the framework controls are enforced in both base paths, and the only party who could define such a gate is the tool's own author, so no trust boundary is crossed.
• Can the caller-loop ContextVar leak across concurrent requests or tenants and point execution at another run's loop?
  Set and reset are token-paired inside the same asyncio Task, every propagation lane copies the context at submit time only, and no context object is stored in flow state, executors or caches.
• Does a blocking async tool stalling the application loop constitute a vulnerability?
  That consequence is explicitly documented in the new helper and is the tool author's own anti-pattern rather than an attacker capability, matching the semantics of awaiting an async tool directly as the docs already advise.
• Can an open-but-idle caller loop hang a worker thread during ordinary usage?
  Only a hand-rolled loop that is abandoned or closed after a cancelled kickoff reaches it, while `asyncio.run` keeps servicing the surviving worker during shutdown, so the only sufferer is the application's own teardown.
• Does the streaming or nested-kickoff coverage gap route tool coroutines onto another user's loop?
  The wrong-loop affinity is real but reachable only when application code nests a streaming kickoff in its own call stack, so a single caller affects only itself and no request state or credential crosses a boundary.

_{Fixed the findings? re-run the review, or tag @strix-security in a PR comment to run a fresh review.}

_{Review 1 on this PR · 5f948e3 · requested Nov 1 08:13 UTC when the PR was opened}

---

Reviewed by Strix
_{Re-run review · Configure security review settings}

[cursor]

Stale Bugbot comment from a previous run.

[strix-security]

Strix flagged a new security finding below. See the pinned summary comment for the full PR status.

[rkfshakti]

Thanks @strix-security and Cursor Bugbot — all three findings were real and are fixed in f2e0824c7.

1. 🔴 HIGH CWE-833 — unbounded bridge wait (run_native_tool_result)
The wait is now bounded by NATIVE_TOOL_BRIDGE_TIMEOUT_SECONDS (default 300s, overridable with CREWAI_NATIVE_TOOL_BRIDGE_TIMEOUT). On timeout the scheduled coroutine is cancelled and a TimeoutError explains the cause (tool blocking the loop, or drawing from the caller loop's default executor). It deliberately does not fall back to tool_func — the tool already started and may have had side effects, so a retry would duplicate them.

The root cause is a documentation bug as much as a code one: the previous docstring recommended asyncio.to_thread inside a bridged tool. It now says the opposite — await the library's native async client, and if a blocking SDK is unavoidable, hand it a dedicated ThreadPoolExecutor (loop.run_in_executor(<own pool>, ...)) rather than the caller loop's default pool.

2. 🟡 Medium — inherited _arun treated as sync
_has_async_implementation now resolves _arun through the MRO (getattr(type(tool), "_arun", None)) instead of type(tool).__dict__, so a class-based tool inheriting a real _arun from a parent is bridged instead of falling back to a fresh asyncio.run loop.

3. 🟡 Medium — streaming kickoff skipped the caller loop
run_crew (the stream=True path) now captures/resets the caller loop around its asyncio.to_thread(self.kickoff, ...), exactly like the non-streaming path, so async native tools bridge on streaming kickoffs too.

Three regression tests were added — each fails on the previous head and passes now:

Test	Without fix
test_inherited_arun_is_treated_as_async	returns the sync fallback instead of got:inherited-ok
test_bridge_wait_is_bounded_and_cancels_the_coroutine	worker thread never returns (asserts TimeoutError + cancellation lands)
test_streaming_kickoff_captures_the_caller_loop	captured caller loop is None

9 passed for the file; ruff check/ruff format clean; mypy clean on both source files. The 5 pre-existing test_empty_llm_response_retry failures in this environment reproduce identically on pristine main (f04cce923: 5 failed / 50 passed) — unrelated to this diff.

@strix-security — re-review requested, the HIGH should be closed by the bounded wait.

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Bridge timeout swallows tool TimeoutError

Medium Severity

The new bridge wait handler treats every TimeoutError as a hung run_coroutine_threadsafe wait. On Python 3.11+ concurrent.futures.TimeoutError is the builtin TimeoutError, so a finished tool that raised asyncio.TimeoutError (for example from asyncio.wait_for) is rewritten into the 300s hung-loop error and the original exception is discarded.

 

^{Reviewed by Cursor Bugbot for commit f2e0824. Configure here.}

[rkfshakti]

Structural fix for the CWE-833 finding (afce3f8d4) — beyond the bounded wait, @strix-security's suggested root-cause fix is in.

The deadlock was: each in-flight kickoff_async parked one default executor thread inside run_native_tool_result, while a bridged tool that offloads blocking work with asyncio.to_thread needed a thread from that same pool. Enough concurrent kickoffs and the pool starves — every run wedges, and subsequent to_thread work across the process stalls too.

run_in_kickoff_thread keeps the asyncio.to_thread contract (still off-loop, still gets a contextvars.copy_context() copy so the caller-loop ContextVar — and any other task context — propagates into the worker) but takes its worker from a dedicated bounded pool (_kickoff_executor, threads named crewai-kickoff). The loop's default pool is now free for the tool coroutines that need it. Crew.kickoff_async uses it on both the streaming and non-streaming paths.

The bounded wait stays as defence in depth: lanes we don't own can still occupy default-pool threads (a Flow sync method wrapping a nested kickoff does exactly that), so a starve there surfaces as TimeoutError rather than a permanent hang.

Two more regression tests, both red before this commit:

Test	Before	After
test_concurrent_kickoffs_do_not_starve_the_loop_default_executor — min(32, cpu+4) concurrent kickoffs whose tools call asyncio.to_thread, bridge bound shortened to 5s so a starve fails fast	tools returned did not finish within 5s	all return sdk:q; exactly one offload per crew
test_kickoff_worker_thread_is_not_a_loop_default_executor_thread	worker thread asyncio_0	crewai-kickoff_*

Current head: loop-affinity suite 11/11, test_streaming.py 37/37, ruff check/format and mypy clean. (test_crew_kickoff_streaming_usage_metrics and the test_empty_llm_response_retry set fail identically on pristine main here — no LLM credentials in this environment.)

[cursor]

🔒 Agentic Security Review

Severity: MEDIUM

Description: kickoff_async now pins every class-based native tool that overrides _arun onto the caller event loop. _has_async_implementation treats any _arun other than BaseTool._arun as async, so tools whose _arun only calls blocking _run (ExecuteCommandTool, ExecuteCodeTool, and the other Bedrock code-interpreter tools) run that sync body on the loop thread. NATIVE_TOOL_BRIDGE_TIMEOUT_SECONDS only unblocks the kickoff worker via future.cancel(); cancellation is not delivered until the synchronous section returns, so the tool's side effect continues and the loop stays stalled for the full call.

Impact: A model-selected native tool call during Crew.kickoff_async is supposed to keep blocking work on the kickoff worker and, if bridged, to stop at the bridge timeout. This change moves blocking _arun wrappers onto the shared caller loop. One long code-interpreter invoke (or any similar _arun that blocks before its next await) freezes every other task on that loop — other in-flight kickoff_async runs, streaming chunk delivery, and cancellation — for the whole sync call, which can exceed the 300s agent-visible timeout. The worker is told the tool timed out and can request another call, but that call only queues behind the still-running body, so the first command or code execution is not stopped. Previously tool.run executed _run on the kickoff worker and did not occupy the caller loop.

Remediation: Do not treat an _arun override as proof the body is non-blocking. Bridge tool.arun onto the caller loop only when the implementation actually awaits; run sync _run and blocking _arun wrappers via the existing tool.run path on the kickoff worker or a dedicated executor. A bridge timeout must bound the loop-side work itself — future.cancel() does not interrupt a synchronous section already running on the loop thread. Add a regression where async def _arun calls a blocking _run under kickoff_async and assert the caller loop still makes progress and the sync body is not left running after TimeoutError.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit d874b95. Configure here.}

[cursor]

Default executor starves on invoke_async

High Severity

invoke_async captures the caller loop so tools bridge via run_native_tool_result, but sync Flow methods still run on the loop default executor. Each run parks a default-pool thread in future.result(), so concurrent Agent.kickoff_async or Flow executions deadlock when a bridged tool also uses asyncio.to_thread. Crew.kickoff_async avoids this with run_in_kickoff_thread.

Additional Locations (1)

• lib/crewai/src/crewai/utilities/agent_utils.py#L1597-L1635

 

^{Reviewed by Cursor Bugbot for commit d874b95. Configure here.}