PR #5313: feat: add graceful cancellation support (aclose/cancel) for streaming outputs

• Repository: crewAIInc/crewAI
• Author: devin-ai-integration[bot]
• State: open | merged: False
• Link: https://github.com/crewAIInc/crewAI/pull/5313

Description (problem / solution / changelog)

Summary

Adds aclose() and cancel() methods to CrewStreamingOutput and FlowStreamingOutput (via the shared StreamingOutputBase) so callers can abort in-flight streaming and release resources promptly — e.g. when an HTTP client disconnects from a FastAPI StreamingResponse.

Types (types/streaming.py):

• New _cancelled flag, _cancel_event (asyncio), _cancel_thread_event (threading), _background_task, _background_thread attributes on StreamingOutputBase
• aclose() — async cancellation: sets cancel events, cancels the background asyncio.Task, awaits cleanup
• cancel() — sync cancellation: sets cancel events, fires task cancellation (no await), marks completed
• is_cancelled property
• Both are idempotent (no-op on already-completed streams)

Utilities (utilities/streaming.py):

• create_chunk_generator (sync): switched from blocking queue.get() to polling with 0.1s timeout so cancellation can be detected between chunks
• create_async_chunk_generator: switched from await queue.get() to asyncio.wait() racing queue-get vs cancel-event, with proper task cancellation in the finally block
• Both generators wire their cancel primitives to the streaming output object via a _wire_cancel() helper called each iteration

Tests: 14 new tests in TestStreamingCancellation covering async close, sync cancel, background task cancellation, idempotency, no-op on completed streams, FlowStreamingOutput support, and multi-call safety.

Closes #5312

Review & Testing Checklist for Human

• Performance regression in sync streaming: create_chunk_generator now polls with a 0.1s timeout instead of a blocking queue.get(). Verify this doesn't introduce noticeable latency in chunk delivery for real LLM streaming workloads.
• Async generator overhead: create_async_chunk_generator now creates two futures (ensure_future) and calls asyncio.wait per chunk. Confirm this is acceptable overhead vs the original single await queue.get().
• Thread safety of _wire_cancel: The wiring mutates output_holder[0]'s internal attributes on every iteration. Verify no race with concurrent cancel()/aclose() calls from another thread/task.
• cancel() does not join the background thread: By design, sync cancel() only signals and marks complete — the daemon thread may still run briefly. Verify this is acceptable for the intended FastAPI disconnect use case.
• Integration test gap: All new tests use mocked generators. Consider testing aclose() through the full Crew.akickoff(stream=True) → cancel path to validate end-to-end behavior with the real event bus and queue wiring.

Notes

• The _background_thread attribute is stored but not used for active cleanup (Python threads can't be cancelled). Cleanup relies on the polling loop checking the threading.Event.
• _wire_cancel() is called every loop iteration; it could be optimized to wire once, but the cost is minimal (just attribute assignment + truthiness check).

Link to Devin session: https://app.devin.ai/sessions/d15a57dd90ab46c2a89fa4288605d1f4

<!-- CURSOR_SUMMARY -->

---

[!NOTE] Medium Risk Adds new cancellation paths that alter sync/async streaming control flow and background task/thread lifecycle handling; mistakes could cause leaked handlers/tasks or prematurely completed streams under race conditions.

Overview Adds graceful cancellation to streaming outputs via new StreamingOutputBase.aclose() (async) and cancel() (sync), plus an is_cancelled flag, so callers can abort in-flight crew/flow streaming and mark streams completed.

Updates the sync and async chunk generators to support cooperative cancellation by wiring per-stream cancel events into the output object and changing queue waits to be interruptible (sync polling with timeout; async wait() racing queue-get vs cancel-event), with adjusted cleanup logic for background tasks/threads.

Extends test_streaming.py with a new cancellation-focused test suite covering idempotency, completed-stream no-ops, background task cancellation, and both CrewStreamingOutput and FlowStreamingOutput behavior.

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

<!-- /CURSOR_SUMMARY -->

Changed files

• lib/crewai/src/crewai/types/streaming.py (modified, +86/-0)
• lib/crewai/src/crewai/utilities/streaming.py (modified, +55/-4)
• lib/crewai/tests/test_streaming.py (modified, +216/-0)