Bug: Multiple error handling issues in Crew config, callbacks, and async task execution

Summary

Found 4 related error handling bugs while using CrewAI in production workflows. All are reproducible with minimal examples.

Bug 1: _create_task raises cryptic StopIteration on unknown agent role

File: lib/crewai/src/crewai/crew.py:895

When a task config references an agent role that doesn't exist, next() without a default raises StopIteration — an internal Python iterator protocol error that gives the user zero context about what went wrong.

crew = Crew(config={
    "agents": [{"role": "Researcher", "goal": "Research", "backstory": "..."}],
    "tasks": [{"description": "Write", "expected_output": "Article", "agent": "Writer"}]
})
# StopIteration (no message saying 'Writer' doesn't match any agent)

Expected: ValueError with message like "Task references agent role 'Writer' which doesn't match any agent. Available roles: ['Researcher']"

Bug 2: after_kickoff_callbacks replaces result with None if callback doesn't return

File: lib/crewai/src/crewai/crew.py:1032-1033

for after_callback in self.after_kickoff_callbacks:
    result = after_callback(result)

If a user writes a callback that performs side effects (logging, metrics) without returning the result, result becomes None and downstream code crashes.

def my_callback(output):
    print(f"Done: {output.raw}")  # Forgot to return

crew = Crew(after_kickoff_callbacks=[my_callback], ...)
crew.kickoff()  # AttributeError on NoneType

Bug 3: Async task callback crashes with asyncio.run() inside running event loop

File: lib/crewai/src/crewai/task.py:851

if inspect.iscoroutine(cb_result):
    asyncio.run(cb_result)  # Crashes in Jupyter or kickoff_async()

asyncio.run() cannot be called from within an already-running event loop (Jupyter notebooks, kickoff_async()). This makes async callbacks unusable in common environments.

Bug 4: _create_task mutates input config dict

File: lib/crewai/src/crewai/crew.py:898

del task_config["agent"]

This modifies the caller's dict. If the same config is reused (loop, retry), second use raises KeyError: 'agent'.

Reproduction Script

import asyncio, inspect
from crewai import Crew

# Bug 1
try:
    Crew(config={
        "agents": [{"role": "Researcher", "goal": "R", "backstory": "B"}],
        "tasks": [{"description": "D", "expected_output": "E", "agent": "Writer"}]
    })
except StopIteration:
    print("Bug 1 confirmed: StopIteration with no context")

# Bug 2
def logging_cb(output):
    print(f"Got: {type(output)}")
result = {"raw": "output"}
result = logging_cb(result)
assert result is None, "Bug 2 confirmed: result swallowed"

# Bug 3
async def async_cb(output):
    await asyncio.sleep(0.001)
async def test():
    cb = async_cb("x")
    if inspect.iscoroutine(cb):
        asyncio.run(cb)
try:
    asyncio.run(test())
except RuntimeError as e:
    print(f"Bug 3 confirmed: {e}")

# Bug 4
cfg = {"description": "D", "expected_output": "E", "agent": "Writer"}
del cfg["agent"]
assert "agent" not in cfg, "Bug 4 confirmed: config mutated"

Environment

• crewai 1.15.2a2
• Python 3.13
• macOS

Start with lib/crewai/src/crewai/crew.py:895-898 and 1032-1033, then inspect lib/crewai/src/crewai/task.py:851 and run the reproduction script. Add regression coverage for unknown roles, callback results, async callbacks in running loops, and config reuse; done means these cases produce contextual errors or preserve the expected result without mutating input.