[BUG] Cancelled async tasks remain in execution_spans and retain task graphs

Description

Cancelling an async task after TaskStartedEvent leaves the task in the global EventListener.execution_spans dictionary. The dictionary keeps a strong reference to the Task, which can retain its Agent, Crew, LLM clients, tools, and other request-scoped objects.

The completed and ordinary-failure paths remove entries with pop(), following the fix for #4222 / #4161. The cancellation path remains uncovered because Task._aexecute_core() catches Exception, while asyncio.CancelledError inherits directly from BaseException.

Current main at a53ecc17f1829354e2d2ff58bbbad1d84ce568b2 still has this flow:

crewai_event_bus.emit(
    self, TaskStartedEvent(context=context, task=self)
)

try:
    ...
except Exception as e:
    crewai_event_bus.emit(
        self,
        TaskFailedEvent(error=str(e), error_type=type(e), task=self),
    )
    raise e

CancelledError bypasses the TaskFailedEvent, so the listener never runs self.execution_spans.pop(source, None).

Steps to Reproduce

Run this with CrewAI 1.15.2 and Python 3.12. It does not call an LLM because Agent.aexecute_task is replaced with a deterministic cancelled coroutine.

import asyncio
import gc
import weakref
from unittest.mock import patch

from crewai import Agent, Crew, Task
from crewai.events.event_listener import event_listener


async def cancelled(self, *args, **kwargs):
    raise asyncio.CancelledError()


async def main():
    event_listener.execution_spans.clear()
    task_refs = []
    agent_refs = []
    crew_refs = []

    with patch.object(Agent, "aexecute_task", cancelled):
        for _ in range(5):
            agent = Agent(
                role="test",
                goal="test",
                backstory="test",
                llm="gpt-4o-mini",
            )
            task = Task(
                description="test",
                expected_output="test",
                agent=agent,
            )
            crew = Crew(agents=[agent], tasks=[task], tracing=False)
            task_refs.append(weakref.ref(task))
            agent_refs.append(weakref.ref(agent))
            crew_refs.append(weakref.ref(crew))

            try:
                await crew.akickoff()
            except asyncio.CancelledError:
                pass

            del crew, task, agent

    gc.collect()
    print(
        {
            "execution_spans": len(event_listener.execution_spans),
            "alive_tasks": sum(ref() is not None for ref in task_refs),
            "alive_agents": sum(ref() is not None for ref in agent_refs),
            "alive_crews": sum(ref() is not None for ref in crew_refs),
        }
    )


asyncio.run(main())

Observed result:

{'execution_spans': 5, 'alive_tasks': 5, 'alive_agents': 5, 'alive_crews': 5}

The same accumulation occurs under repeated request cancellation: after 4, 8, and 12 cancellations, execution_spans contains 4, 8, and 12 task entries respectively. A full gc.collect() does not remove them. Removing those dictionary entries releases the associated object graphs.

Expected behavior

Every TaskStartedEvent should have a terminal cleanup path. Async cancellation should remove the task from execution_spans, close or mark the telemetry span appropriately, and allow the task object graph to be garbage-collected.

Screenshots/Code snippets

The minimal reproduction and observed output are included above.

Operating System

macOS (Apple Silicon)

Python Version

3.12

crewAI Version

1.15.2; current main source at a53ecc17f1829354e2d2ff58bbbad1d84ce568b2 has the same exception boundary

crewAI Tools Version

Not installed / not required

Virtual Environment

Venv

Evidence

• Successful async executions leave execution_spans at zero.
• Ordinary exceptions leave execution_spans at zero because TaskFailedEvent is emitted.
• CancelledError executions add one retained task entry per cancellation.
• Full garbage collection leaves those entries and object graphs reachable.
• Clearing the entries releases the retained graphs.

This is related to #4222, but it is a separate terminal-path gap. #4222 fixed completed and failed tasks by replacing assignment-to-None with pop(). Async cancellation emits neither of those terminal events.

Possible Solution

Handle asyncio.CancelledError explicitly in _aexecute_core() before except Exception, emit a terminal task event, and immediately re-raise the cancellation. Reusing TaskFailedEvent would be the smallest behavioral change; a dedicated cancellation event could preserve cancellation semantics for telemetry consumers.

Add a regression test asserting that cancellation leaves no entry in event_listener.execution_spans and that repeated cancellations do not retain prior task graphs.

Additional context

This affects long-running async services where client disconnects, request timeouts, shutdown, or orchestration cancellation can propagate into crew.akickoff().

Start at Task._aexecute_core() and follow how TaskStartedEvent, TaskFailedEvent, and EventListener.execution_spans interact. Add a regression test for asyncio.CancelledError, including repeated cancellations, then verify that cancellation is re-raised and execution_spans contains no retained task entries.