[BUG] Two async_execution tasks on the same agent crash with "Executor is already running"

[parthiban-sivakumar]

Description

Assigning two or more async_execution=True tasks to the same agent crashes the crew with:

RuntimeError: Executor is already running. Cannot invoke the same executor instance concurrently.

An Agent holds a single agent_executor, and create_agent_executor() mutates that field. Concurrent async tasks race on it and then both invoke it. AgentExecutor.invoke() resets shared state at entry (state.messages.clear(), iterations = 0, current_answer = None), so the guard at experimental/agent_executor.py:2867 rejects the second caller.

This is the pattern shown in the async example in docs/edge/en/concepts/tasks.mdx, where list_ideas and list_important_history are both assigned to researcher. Running that example verbatim crashes.

The guard looks correct — concurrent entry really would corrupt the state. The problem is that Crew schedules concurrent tasks onto one executor in the first place.

Steps to Reproduce

1. Create an agent.
2. Give it two tasks with async_execution=True.
3. Add a third, non-async task so the crew has something to finish on.
4. kickoff().

Or run the async example from docs/edge/en/concepts/tasks.mdx unchanged.

Expected behavior

Two async tasks assigned to the same agent should run concurrently and complete, the same way they do when assigned to different agents.

Screenshots/Code snippets

researcher = Agent(role="Researcher", goal="Research AI.", backstory="You research.", llm=llm)
writer = Agent(role="Writer", goal="Write articles.", backstory="You write.", llm=llm)

list_ideas = Task(description="List of 5 interesting ideas to explore for an article about AI.",
                  expected_output="Bullet point list of 5 ideas.", agent=researcher, async_execution=True)
list_important_history = Task(description="Research the history of AI and give me the 5 most important events.",
                              expected_output="Bullet point list of 5 events.", agent=researcher, async_execution=True)
write_article = Task(description="Write an article about AI, its history, and interesting ideas.",
                     expected_output="A 4 paragraph article about AI.", agent=writer,
                     context=[list_ideas, list_important_history])

Crew(agents=[researcher, writer], tasks=[list_ideas, list_important_history, write_article]).kickoff()

CRASHED: RuntimeError: Executor is already running. Cannot invoke the same executor instance concurrently.

Operating System

Other (specify in additional context)

Python Version

3.12

crewAI Version

1.15.18

crewAI Tools Version

1.15.18

Virtual Environment

Venv

Evidence

Matrix on main @ 7e18abd:

default executor, 2 async, same agent          FAIL  RuntimeError
default executor, 3 async, same agent          FAIL  RuntimeError
CrewAgentExecutor, 2 async, same agent         OK
kickoff_async, 2 async, same agent             FAIL  RuntimeError
hierarchical, 2 async                          OK
CONTROL: 2 async, different agents             OK

This is a regression. The legacy CrewAgentExecutor handles the case; AgentExecutor does not. The default changed in 332263462 (#5745, 2026-05-12), "deprecate CrewAgentExecutor, default Crew agents to AgentExecutor".

Setting executor_class=CrewAgentExecutor works, but emits a DeprecationWarning telling users to switch to the executor that fails, so it isn't a durable workaround.

The error message names an internal object. A user who assigned two async tasks to one agent has nothing in "the same executor instance" to connect back to what they wrote.

Related but distinct: #4389 and #4432 cover this executor's state not resetting between sequential tasks. #4389 notes the experimental executor "correctly resets all execution state at the beginning of invoke()" — that reset is exactly what makes concurrent entry unsafe.

Possible Solution

Build the executor separately from storing it, and give each async task an executor bound to the thread running it. Task.execute_async already copies contextvars per thread, so siblings stay isolated without locks or copying the agent, and the sequential path is unchanged.

Copying the agent per task would be a smaller change, but Agent.copy() excludes _token_process and shallow-copies the llm, so async task token usage would stop being counted.

PR to follow.

Additional context

My OS is macOS Tahoe 26.5.2 and I'm on Python 3.13.13, neither of which is in the dropdowns. Reproduced from a source checkout of main at commit 7e18abd.

This issue was written with AI assistance and should carry the llm-generated label per CONTRIBUTING.md. I can't apply labels myself — could a maintainer add it?